use std::fmt;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, ready};

use pin_project_lite::pin_project;
use tokio::task::JoinHandle;
use tokio_util::sync::{CancellationToken, DropGuard};
use tower::Service;

use crate::instruments::{
    EVENT_CANCELLED, InstrumentationState, Instruments, RESULT_CANCELLED, RESULT_SUCCESS,
};
use crate::service::BlockingService;

/// Tower [`Service`] adapter that runs a [`BlockingService`] on the tokio blocking thread pool.
///
/// [`ServiceBuilderExt::blocking_service`](crate::ServiceBuilderExt::blocking_service) is the
/// preferred way to set up a blocking service, as it allows layering on middleware.
///
/// # Cancellation
///
/// If the future is dropped before the blocking task completes, the task's cancellation token
/// is triggered. Tasks that may be long-running must implement cooperative cancellation by
/// regularly checking the token.
///
/// An optional parent cancellation token can be supplied at construction to implement graceful
/// shutdown. If this token is triggered, it cancels _all_ ongoing work.
///
/// # Example
///
/// ```
/// use apollo_blocking::{BlockingServiceAdapter, CancellationToken};
///
/// let adapter = BlockingServiceAdapter::new(
///     "expensive_op",
///     |req: u32, _ct: CancellationToken| -> Result<u32, std::convert::Infallible> {
///         Ok(req * 2)
///     },
///     None,
/// );
/// # let _: BlockingServiceAdapter<_> = adapter;
/// ```
pub struct BlockingServiceAdapter<B> {
    name: &'static str,
    inner: B,
    parent: Option<CancellationToken>,
    instruments: Arc<Instruments>,
}

impl<B> BlockingServiceAdapter<B> {
    /// Construct an adapter.
    ///
    /// The `name` identifies this adapter instance in traces and metrics so operators can
    /// distinguish multiple blocking services running in the same process.
    ///
    /// Pass a `parent` cancellation token to cancel _all_ ongoing work. This way you can
    /// implement graceful shutdown.
    pub fn new(name: &'static str, inner: B, parent: Option<CancellationToken>) -> Self {
        Self {
            name,
            inner,
            parent,
            instruments: Instruments::new(),
        }
    }
}

impl<B: Clone> Clone for BlockingServiceAdapter<B> {
    fn clone(&self) -> Self {
        Self {
            name: self.name,
            inner: self.inner.clone(),
            parent: self.parent.clone(),
            instruments: Arc::clone(&self.instruments),
        }
    }
}

impl<B> fmt::Debug for BlockingServiceAdapter<B> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BlockingServiceAdapter")
            .field("name", &self.name)
            .field("has_parent_token", &self.parent.is_some())
            .finish()
    }
}

impl<B, Req> Service<Req> for BlockingServiceAdapter<B>
where
    B: BlockingService<Req> + Clone + Send + 'static,
    Req: Send + 'static,
    B::Response: Send + 'static,
    B::Error: Send + 'static,
{
    type Response = B::Response;
    type Error = B::Error;
    type Future = BlockingFuture<B::Response, B::Error>;

    /// Always [`Poll::Ready`]. The adapter does not back-pressure on its own; compose with a
    /// concurrency-limit layer when bounded admission is required.
    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: Req) -> Self::Future {
        let child = match &self.parent {
            Some(parent) => parent.child_token(),
            None => CancellationToken::new(),
        };
        let drop_guard = child.clone().drop_guard();

        let instruments = self.instruments.start(self.name);

        let inner = self.inner.clone();
        let join =
            tokio::task::spawn_blocking(move || run_blocking(inner, req, child, instruments));

        BlockingFuture {
            join,
            _drop_guard: drop_guard,
        }
    }
}

/// Run the blocking task on a worker thread with instrumentation co-located with the work.
///
/// The instrumentation state lives in this stack frame, not in the adapter future, so the
/// span and metric guards drop only when the blocking work actually exits — even if the
/// async caller dropped its future seconds or minutes earlier. The recorded duration always
/// reflects the wall-clock time the worker thread spent on the task, and a ghost thread that
/// ignores cancellation surfaces in metrics with its true duration when it eventually returns.
///
/// Panics in the user closure unwind through this frame and are caught by the
/// `spawn_blocking` task wrapper, which surfaces them on the [`JoinHandle`] as a panic-flavoured
/// [`tokio::task::JoinError`]. [`InstrumentationState::Drop`] detects the unwind via
/// [`std::thread::panicking`] and records the `panicked` outcome on its way out.
fn run_blocking<B, Req>(
    inner: B,
    req: Req,
    child: CancellationToken,
    mut instruments: InstrumentationState,
) -> Result<B::Response, B::Error>
where
    B: BlockingService<Req>,
{
    let result = inner.call(req, child.clone());
    if child.is_cancelled() {
        instruments.record_event_and_finalize(EVENT_CANCELLED, RESULT_CANCELLED);
    } else {
        instruments.finalize(RESULT_SUCCESS);
    }
    result
}

pin_project! {
    /// Future returned by [`BlockingServiceAdapter`].
    pub struct BlockingFuture<R, E> {
        #[pin]
        join: JoinHandle<Result<R, E>>,
        // The drop guard cancels the child token when this future is dropped before completing.
        // Held by the future, not the adapter, so cancellation tracks the lifetime of one call
        // even though the adapter is cloned and reused across many calls.
        _drop_guard: DropGuard,
    }
}

impl<R, E> std::future::Future for BlockingFuture<R, E> {
    type Output = Result<R, E>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        match ready!(this.join.poll(cx)) {
            Ok(result) => Poll::Ready(result),
            // A JoinError is either a user-code panic or a runtime-abort marker. The abort
            // branch is unreachable through our public API (we don't expose the JoinHandle);
            // it only fires when the tokio runtime is dropped while a blocking task is still
            // queued. Resume the unwind on a panic so the payload propagates; surface a
            // descriptive panic on the rare abort path — `into_panic` alone would itself
            // panic here with an unhelpful message.
            Err(join_err) => match join_err.try_into_panic() {
                Ok(payload) => std::panic::resume_unwind(payload),
                Err(join_err) => panic!("apollo-blocking task aborted: {join_err}"),
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;
    use tower::ServiceExt;

    #[tokio::test]
    async fn returns_inner_response_on_success() {
        let adapter = BlockingServiceAdapter::new(
            "test",
            |req: u32, _ct: CancellationToken| -> Result<u32, std::convert::Infallible> {
                Ok(req + 1)
            },
            None,
        );

        let response = adapter.oneshot(41).await.unwrap();

        assert_eq!(response, 42);
    }

    #[tokio::test]
    async fn surfaces_inner_error() {
        let adapter = BlockingServiceAdapter::new(
            "test",
            |_req: (), _ct: CancellationToken| -> Result<(), &'static str> { Err("nope") },
            None,
        );

        let err = adapter.oneshot(()).await.unwrap_err();

        assert_eq!(err, "nope");
    }

    #[tokio::test]
    async fn panic_in_blocking_task_propagates() {
        let adapter = BlockingServiceAdapter::new(
            "test",
            |_req: (), _ct: CancellationToken| -> Result<(), std::convert::Infallible> {
                panic!("kaboom");
            },
            None,
        );

        // Isolate the panic on a spawned task so it doesn't kill the test runner.
        let join = tokio::spawn(adapter.oneshot(())).await;

        let join_err = join.expect_err("blocking task panic should propagate");
        assert!(join_err.is_panic());

        // The payload must survive the resume_unwind chain — a regression here would silently
        // eat the panic message.
        let payload = join_err.into_panic();
        let message = payload
            .downcast_ref::<&'static str>()
            .expect("panic!(\"literal\") payload is &'static str");
        assert_eq!(*message, "kaboom");
    }

    #[tokio::test]
    async fn poll_ready_is_immediate() {
        let mut adapter = BlockingServiceAdapter::new(
            "test",
            |req: u32, _ct: CancellationToken| -> Result<u32, std::convert::Infallible> { Ok(req) },
            None,
        );

        // ready_oneshot expects Poll::Ready on the first call.
        let ready = adapter.ready().await.unwrap();
        let response = ready.call(7).await.unwrap();
        assert_eq!(response, 7);
    }

    #[tokio::test]
    async fn dropping_future_cancels_child_token() {
        let started = Arc::new(AtomicBool::new(false));
        let saw_cancel = Arc::new(AtomicBool::new(false));

        let svc = {
            let started = Arc::clone(&started);
            let saw_cancel = Arc::clone(&saw_cancel);
            move |_req: (), ct: CancellationToken| -> Result<(), std::convert::Infallible> {
                started.store(true, Ordering::SeqCst);
                // Spin until the token fires (would-be CPU-bound work).
                while !ct.is_cancelled() {
                    std::thread::sleep(Duration::from_millis(1));
                }
                saw_cancel.store(true, Ordering::SeqCst);
                Ok(())
            }
        };

        let mut adapter = BlockingServiceAdapter::new("test", svc, None);
        let future = adapter.call(());

        // Wait until the blocking thread is actually running before we drop the future,
        // otherwise the test could race past the spin loop.
        while !started.load(Ordering::SeqCst) {
            tokio::task::yield_now().await;
        }

        drop(future);

        // The blocking thread keeps running; tokio cannot abort it. We assert the token fires
        // by waiting for the blocking thread to observe it. Time out generously.
        for _ in 0..1000 {
            if saw_cancel.load(Ordering::SeqCst) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("blocking task did not observe child token cancellation");
    }

    #[tokio::test]
    async fn parent_token_cancels_child() {
        let parent = CancellationToken::new();
        let observed = Arc::new(AtomicBool::new(false));

        let svc = {
            let observed = Arc::clone(&observed);
            move |_req: (), ct: CancellationToken| -> Result<(), std::convert::Infallible> {
                while !ct.is_cancelled() {
                    std::thread::sleep(Duration::from_millis(1));
                }
                observed.store(true, Ordering::SeqCst);
                Ok(())
            }
        };

        let adapter = BlockingServiceAdapter::new("test", svc, Some(parent.clone()));
        let task = tokio::spawn(adapter.oneshot(()));

        // Give the blocking thread time to enter its spin loop.
        tokio::time::sleep(Duration::from_millis(20)).await;
        parent.cancel();

        task.await.unwrap().unwrap();
        assert!(observed.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn parent_cancellation_propagates_to_all_concurrent_children() {
        let parent = CancellationToken::new();
        let started_a = Arc::new(AtomicBool::new(false));
        let started_b = Arc::new(AtomicBool::new(false));
        let observed_a = Arc::new(AtomicBool::new(false));
        let observed_b = Arc::new(AtomicBool::new(false));

        let svc = {
            let started_a = Arc::clone(&started_a);
            let started_b = Arc::clone(&started_b);
            let observed_a = Arc::clone(&observed_a);
            let observed_b = Arc::clone(&observed_b);
            move |req: u32, ct: CancellationToken| -> Result<(), std::convert::Infallible> {
                if req == 1 {
                    started_a.store(true, Ordering::SeqCst);
                } else {
                    started_b.store(true, Ordering::SeqCst);
                }
                while !ct.is_cancelled() {
                    std::thread::sleep(Duration::from_millis(1));
                }
                if req == 1 {
                    observed_a.store(true, Ordering::SeqCst);
                } else {
                    observed_b.store(true, Ordering::SeqCst);
                }
                Ok(())
            }
        };

        let mut adapter = BlockingServiceAdapter::new("test", svc, Some(parent.clone()));
        let task_a = tokio::spawn(adapter.call(1));
        let task_b = tokio::spawn(adapter.call(2));

        // Wait for both blocking threads to enter their spin loops before firing the parent,
        // otherwise the test could race past the loops.
        while !started_a.load(Ordering::SeqCst) || !started_b.load(Ordering::SeqCst) {
            tokio::task::yield_now().await;
        }

        parent.cancel();

        task_a.await.unwrap().unwrap();
        task_b.await.unwrap().unwrap();

        assert!(observed_a.load(Ordering::SeqCst));
        assert!(observed_b.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn clone_independent_of_inner_state() {
        let adapter = BlockingServiceAdapter::new(
            "test",
            |req: u32, _ct: CancellationToken| -> Result<u32, std::convert::Infallible> {
                Ok(req * 10)
            },
            None,
        );
        let cloned = adapter.clone();

        let a = adapter.oneshot(1).await.unwrap();
        let b = cloned.oneshot(2).await.unwrap();

        assert_eq!(a, 10);
        assert_eq!(b, 20);
    }

    #[test]
    fn debug_does_not_require_inner_debug() {
        struct NotDebug;
        impl BlockingService<()> for NotDebug {
            type Response = ();
            type Error = std::convert::Infallible;
            fn call(
                &self,
                _req: (),
                _ct: CancellationToken,
            ) -> Result<Self::Response, Self::Error> {
                Ok(())
            }
        }

        let adapter = BlockingServiceAdapter::new("named", NotDebug, None);
        let s = format!("{adapter:?}");
        assert!(s.contains("BlockingServiceAdapter"));
        assert!(s.contains("named"));
    }

    // ---------- Instrumentation ----------
    //
    // The three outcomes (success, cancelled, panicked) must each emit a span with the right
    // attributes and a duration histogram measurement labelled with the outcome.
    // apollo_opentelemetry_test isolates the global tracer and meter providers per
    // TelemetryContext.

    use apollo_opentelemetry_test::{TelemetryContext, assert_metric, assert_span};

    #[tokio::test]
    async fn instrumentation_records_success_outcome() {
        let ctx = TelemetryContext::new();

        let adapter = BlockingServiceAdapter::new(
            "query_planning",
            |req: u32, _ct: CancellationToken| -> Result<u32, std::convert::Infallible> {
                Ok(req + 1)
            },
            None,
        );
        adapter.oneshot(7).await.unwrap();

        assert_span!(
            ctx,
            "apollo.blocking.task",
            "apollo.blocking.name" = "query_planning",
            "apollo.blocking.result" = "success",
        );
        assert_metric!(
            ctx,
            "apollo.blocking.task.duration",
            count: 1,
            "name" = "query_planning",
            "result" = "success",
        );
    }

    #[tokio::test]
    async fn instrumentation_records_cancelled_outcome_when_future_dropped() {
        let ctx = TelemetryContext::new();

        let started = Arc::new(AtomicBool::new(false));
        let task_finished = Arc::new(AtomicBool::new(false));
        let svc = {
            let started = Arc::clone(&started);
            let task_finished = Arc::clone(&task_finished);
            move |_req: (), ct: CancellationToken| -> Result<(), std::convert::Infallible> {
                started.store(true, Ordering::SeqCst);
                while !ct.is_cancelled() {
                    std::thread::sleep(Duration::from_millis(1));
                }
                task_finished.store(true, Ordering::SeqCst);
                Ok(())
            }
        };

        let mut adapter = BlockingServiceAdapter::new("query_planning", svc, None);
        let future = adapter.call(());

        while !started.load(Ordering::SeqCst) {
            tokio::task::yield_now().await;
        }
        drop(future);

        // Instrumentation is finalised on the blocking thread once the user closure exits;
        // wait for that signal, then give the wrapper a moment to drop the metric guards.
        while !task_finished.load(Ordering::SeqCst) {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert_span!(
            ctx,
            "apollo.blocking.task",
            "apollo.blocking.name" = "query_planning",
            "apollo.blocking.result" = "cancelled",
        );
        assert_metric!(
            ctx,
            "apollo.blocking.task.duration",
            count: 1,
            "name" = "query_planning",
            "result" = "cancelled",
        );
    }

    #[tokio::test]
    async fn instrumentation_records_panicked_outcome() {
        let ctx = TelemetryContext::new();

        let adapter = BlockingServiceAdapter::new(
            "query_planning",
            |_req: (), _ct: CancellationToken| -> Result<(), std::convert::Infallible> {
                panic!("kaboom");
            },
            None,
        );
        // The panic propagates to the awaiting task; isolate it via tokio::spawn so it doesn't
        // kill the test runner.
        let _ = tokio::spawn(adapter.oneshot(())).await;

        assert_span!(
            ctx,
            "apollo.blocking.task",
            "apollo.blocking.name" = "query_planning",
            "apollo.blocking.result" = "panicked",
        );
        assert_metric!(
            ctx,
            "apollo.blocking.task.duration",
            count: 1,
            "name" = "query_planning",
            "result" = "panicked",
        );
    }
}
