use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use async_trait::async_trait;
use folk_api::{Executor, Plugin, PluginContext, ServerPlugin, ServerPluginWrapper};
use tokio::sync::watch;
use tokio::time::timeout;

struct EchoPlugin;

#[async_trait]
impl ServerPlugin for EchoPlugin {
    fn name(&self) -> &'static str {
        "echo"
    }

    async fn run(&self, mut ctx: PluginContext) -> Result<()> {
        if let Err(e) = ctx.shutdown.changed().await {
            tracing::error!(error = %e, "shutdown sender dropped unexpectedly");
        }
        Ok(())
    }
}

struct NoopExecutor;

#[async_trait]
impl Executor for NoopExecutor {
    async fn execute_method(&self, _: &str, _: bytes::Bytes) -> Result<bytes::Bytes> {
        unimplemented!("not used in this test")
    }
}

#[tokio::test]
async fn server_plugin_wrapper_boots_and_shuts_down() {
    let mut wrapper = ServerPluginWrapper::new(EchoPlugin);

    let (tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();
    tokio::time::sleep(Duration::from_millis(50)).await;
    tx.send(true).unwrap();

    wrapper.shutdown().await.unwrap();
}

/// Regression test for issue #54: boot() must return an error when called
/// a second time while the plugin is already running.
#[tokio::test]
async fn server_plugin_wrapper_double_boot_returns_error() {
    let mut wrapper = ServerPluginWrapper::new(EchoPlugin);
    let (_tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx.clone(),
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };
    let ctx2 = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();
    let err = wrapper.boot(ctx2).await;
    assert!(err.is_err(), "second boot() must return Err");
    assert!(
        err.unwrap_err().to_string().contains("already running"),
        "error message must mention 'already running'"
    );
}

/// Regression test for issue #68: shutdown() must return Ok(()) even when
/// the run loop returns an error — log and continue, do not propagate.
#[tokio::test]
async fn server_plugin_wrapper_run_error_does_not_propagate_shutdown() {
    struct FailingPlugin;

    #[async_trait]
    impl ServerPlugin for FailingPlugin {
        fn name(&self) -> &'static str {
            "failing"
        }

        async fn run(&self, _ctx: PluginContext) -> Result<()> {
            anyhow::bail!("simulated run-loop failure")
        }
    }

    let mut wrapper = ServerPluginWrapper::new(FailingPlugin);
    let (_tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();
    tokio::time::sleep(Duration::from_millis(50)).await;

    let result = wrapper.shutdown().await;
    assert!(
        result.is_ok(),
        "shutdown() must return Ok(()) even when run() returned Err: {result:?}"
    );
}

/// Regression test for issue #68: shutdown() must return Ok(()) even when
/// the run loop panics — log and continue, do not propagate.
#[tokio::test]
async fn server_plugin_wrapper_run_panic_does_not_propagate_shutdown() {
    struct PanickingPlugin;

    #[async_trait]
    impl ServerPlugin for PanickingPlugin {
        fn name(&self) -> &'static str {
            "panicking"
        }

        async fn run(&self, _ctx: PluginContext) -> Result<()> {
            panic!("simulated run-loop panic");
        }
    }

    let mut wrapper = ServerPluginWrapper::new(PanickingPlugin);
    let (_tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();
    tokio::time::sleep(Duration::from_millis(50)).await;

    let result = wrapper.shutdown().await;
    assert!(
        result.is_ok(),
        "shutdown() must return Ok(()) even when run() panicked: {result:?}"
    );
}

/// Regression test for issue #55: shutdown() must not deadlock when a plugin
/// task ignores ctx.shutdown and never exits on its own.
#[tokio::test]
async fn server_plugin_wrapper_shutdown_aborts_stuck_task() {
    struct StubbornPlugin;

    #[async_trait]
    impl ServerPlugin for StubbornPlugin {
        fn name(&self) -> &'static str {
            "stubborn"
        }

        async fn run(&self, _ctx: PluginContext) -> Result<()> {
            // Ignores ctx.shutdown — would block forever without abort().
            std::future::pending::<()>().await;
            Ok(())
        }
    }

    let mut wrapper = ServerPluginWrapper::new(StubbornPlugin);
    let (_tx, rx) = watch::channel(false);
    let ctx = PluginContext {
        executor: Arc::new(NoopExecutor),
        shutdown: rx,
        rpc_registrar: None,
        health_registry: None,
        metrics_registry: None,
    };

    wrapper.boot(ctx).await.unwrap();

    // shutdown() must return quickly; a deadlock would cause a test timeout.
    let result = timeout(Duration::from_secs(5), wrapper.shutdown()).await;
    assert!(result.is_ok(), "shutdown() deadlocked");
    result.unwrap().unwrap();
}
