use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;

use anyhow::Result;
use async_trait::async_trait;
use bytes::Bytes;
use folk_api::{Executor, PluginContext, ServerPlugin};
use folk_plugin_jobs::Driver;
use folk_plugin_jobs::MemoryDriver;
use folk_plugin_jobs::config::{DriverKind, JobsConfig, QueueConfig};
use folk_plugin_jobs::plugin::JobsPlugin;

struct CountingExecutor {
    count: AtomicU32,
}

impl CountingExecutor {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            count: AtomicU32::new(0),
        })
    }
}

#[async_trait]
impl Executor for CountingExecutor {
    async fn execute(&self, _payload: Bytes) -> Result<Bytes> {
        self.count.fetch_add(1, Ordering::SeqCst);
        Ok(Bytes::new())
    }
}

struct FailingExecutor {
    call_count: AtomicU32,
    fail_until: u32,
}

impl FailingExecutor {
    fn new(fail_until: u32) -> Arc<Self> {
        Arc::new(Self {
            call_count: AtomicU32::new(0),
            fail_until,
        })
    }
}

#[async_trait]
impl Executor for FailingExecutor {
    async fn execute(&self, _payload: Bytes) -> Result<Bytes> {
        let n = self.call_count.fetch_add(1, Ordering::SeqCst) + 1;
        if n <= self.fail_until {
            anyhow::bail!("simulated failure #{n}");
        }
        Ok(Bytes::new())
    }
}

fn make_ctx(
    executor: Arc<dyn Executor>,
    shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> PluginContext {
    PluginContext {
        executor,
        shutdown: shutdown_rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    }
}

#[tokio::test]
async fn jobs_plugin_dispatches_via_executor() {
    let executor = CountingExecutor::new();
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
    let _ctx = make_ctx(executor.clone(), shutdown_rx);

    let plugin = JobsPlugin::new(JobsConfig {
        driver: DriverKind::Memory,
        redis_url: String::new(),
        queues: vec![QueueConfig {
            name: "default".into(),
            concurrency: 1,
            max_retries: 3,
        }],
    });

    let driver = MemoryDriver::new();
    driver
        .push("default", Bytes::from("test-job"))
        .await
        .unwrap();

    // Spawn the consume loop directly
    let d = driver.clone();
    let exec = executor.clone();
    let handle = tokio::spawn(async move {
        // Replicate what the plugin does internally
        let queue = QueueConfig {
            name: "default".into(),
            concurrency: 1,
            max_retries: 3,
        };

        // Pop one job and dispatch
        let payload = d.pop(&queue.name).await.unwrap().unwrap();
        let envelope = rmpv::Value::Map(vec![
            (
                rmpv::Value::String("method".into()),
                rmpv::Value::String("jobs.process".into()),
            ),
            (
                rmpv::Value::String("payload".into()),
                rmpv::Value::Binary(payload.to_vec()),
            ),
        ]);
        let encoded = Bytes::from(rmp_serde::to_vec(&envelope).unwrap());
        exec.execute(encoded).await.unwrap();
    });

    tokio::time::timeout(Duration::from_millis(500), handle)
        .await
        .expect("timed out")
        .expect("task panicked");

    assert_eq!(executor.count.load(Ordering::SeqCst), 1);
    shutdown_tx.send(true).ok();

    // Verify the plugin's name
    assert_eq!(plugin.name(), "jobs");
}

#[tokio::test]
async fn retry_exhaustion_discards_job() {
    let executor = FailingExecutor::new(100); // always fails
    let driver = MemoryDriver::new();
    driver
        .push("default", Bytes::from("doomed-job"))
        .await
        .unwrap();

    let d = driver.clone();
    let exec: Arc<dyn Executor> = executor.clone();
    let (shutdown_tx, _shutdown_rx) = tokio::sync::watch::channel(false);

    let handle = tokio::spawn(async move {
        let queue = QueueConfig {
            name: "default".into(),
            concurrency: 1,
            max_retries: 2,
        };

        let payload = d.pop(&queue.name).await.unwrap().unwrap();
        let envelope = rmpv::Value::Map(vec![
            (
                rmpv::Value::String("method".into()),
                rmpv::Value::String("jobs.process".into()),
            ),
            (
                rmpv::Value::String("payload".into()),
                rmpv::Value::Binary(payload.to_vec()),
            ),
        ]);
        let encoded = Bytes::from(rmp_serde::to_vec(&envelope).unwrap());

        let mut retries = 0u32;
        loop {
            match exec.execute(encoded.clone()).await {
                Ok(_) => break,
                Err(_) => {
                    retries += 1;
                    if retries >= queue.max_retries {
                        break;
                    }
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
            }
        }
    });

    tokio::time::timeout(Duration::from_secs(2), handle)
        .await
        .expect("timed out")
        .expect("task panicked");

    // Should have been called exactly max_retries times (2)
    assert_eq!(executor.call_count.load(Ordering::SeqCst), 2);
    shutdown_tx.send(true).ok();
}
