use tower::{Layer, Service, ServiceBuilder};

use crate::adapter::BlockingServiceAdapter;
use crate::service::BlockingService;
use tokio_util::sync::CancellationToken;

mod private {
    pub trait Sealed {}
    impl<L> Sealed for tower::ServiceBuilder<L> {}
}

/// Extension methods on [`tower::ServiceBuilder`].
///
/// This trait is sealed; downstream crates cannot implement it.
pub trait ServiceBuilderExt<L>: private::Sealed {
    /// Terminate the builder chain with a blocking service.
    ///
    /// The `name` identifies this instance in traces and metrics.
    ///
    /// # Cancellation
    ///
    /// If the response 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 to implement graceful shutdown.
    /// If this token is triggered, it cancels _all_ ongoing work.
    ///
    /// # Example
    ///
    /// ```
    /// use apollo_blocking::{CancellationToken, ServiceBuilderExt};
    /// use std::convert::Infallible;
    /// use tower::ServiceBuilder;
    ///
    /// let _svc = ServiceBuilder::new()
    ///     .blocking_service("expensive_op", None, |req: u32, _ct: CancellationToken| {
    ///         Ok::<_, Infallible>(req * 2)
    ///     });
    /// ```
    ///
    /// # Example — with a graceful-shutdown token
    ///
    /// ```
    /// use apollo_blocking::{CancellationToken, ServiceBuilderExt};
    /// use std::convert::Infallible;
    /// use tower::ServiceBuilder;
    ///
    /// let shutdown = CancellationToken::new();
    /// let _svc = ServiceBuilder::new()
    ///     .blocking_service(
    ///         "expensive_op",
    ///         Some(shutdown.clone()),
    ///         |req: u32, ct: CancellationToken| -> Result<u32, Infallible> {
    ///             if ct.is_cancelled() {
    ///                 // cooperative early return
    ///             }
    ///             Ok(req * 2)
    ///         },
    ///     );
    /// ```
    fn blocking_service<B, Req>(
        self,
        name: &'static str,
        parent_token: Option<CancellationToken>,
        svc: B,
    ) -> <L as Layer<BlockingServiceAdapter<B>>>::Service
    where
        L: Layer<BlockingServiceAdapter<B>>,
        <L as Layer<BlockingServiceAdapter<B>>>::Service: Service<Req>,
        B: BlockingService<Req> + Clone + Send + 'static,
        Req: Send + 'static,
        B::Response: Send + 'static,
        B::Error: Send + 'static;
}

impl<L> ServiceBuilderExt<L> for ServiceBuilder<L> {
    fn blocking_service<B, Req>(
        self,
        name: &'static str,
        parent_token: Option<CancellationToken>,
        svc: B,
    ) -> <L as Layer<BlockingServiceAdapter<B>>>::Service
    where
        L: Layer<BlockingServiceAdapter<B>>,
        <L as Layer<BlockingServiceAdapter<B>>>::Service: Service<Req>,
        B: BlockingService<Req> + Clone + Send + 'static,
        Req: Send + 'static,
        B::Response: Send + 'static,
        B::Error: Send + 'static,
    {
        self.service(BlockingServiceAdapter::new(name, svc, parent_token))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::convert::Infallible;
    use tower::{ServiceBuilder, ServiceExt};

    #[tokio::test]
    async fn terminal_combinator_runs_blocking_service() {
        let svc = ServiceBuilder::new().blocking_service(
            "double",
            None,
            |req: u32, _ct: CancellationToken| -> Result<u32, Infallible> { Ok(req * 2) },
        );

        let response = svc.oneshot(7).await.unwrap();

        assert_eq!(response, 14);
    }

    #[tokio::test]
    async fn applies_outer_layers() {
        // tower::filter wraps the inner service so requests pass through it before reaching the
        // adapter — proves the builder's accumulated layers are applied.
        use tower::filter::FilterLayer;

        let svc = ServiceBuilder::new()
            .layer(FilterLayer::new(|req: u32| {
                if req >= 10 { Ok(req) } else { Err("too small") }
            }))
            .blocking_service(
                "test",
                None,
                |req: u32, _ct: CancellationToken| -> Result<u32, Infallible> { Ok(req + 1) },
            );

        assert_eq!(svc.clone().oneshot(20).await.unwrap(), 21);

        let err = svc.oneshot(2).await.unwrap_err();
        // Filter layer rejected before reaching the adapter — error is the filter's.
        assert_eq!(err.to_string(), "too small");
    }

    #[tokio::test]
    async fn parent_token_propagates_through_builder() {
        let parent = CancellationToken::new();
        parent.cancel();

        let svc = ServiceBuilder::new().blocking_service(
            "test",
            Some(parent),
            |_req: (), ct: CancellationToken| -> Result<bool, Infallible> { Ok(ct.is_cancelled()) },
        );

        // Pre-cancelled parent → child observes is_cancelled = true.
        let observed = svc.oneshot(()).await.unwrap();
        assert!(observed);
    }

    #[tokio::test]
    async fn surfaces_blocking_error_through_chain() {
        let svc = ServiceBuilder::new().blocking_service(
            "test",
            None,
            |_req: (), _ct: CancellationToken| -> Result<(), &'static str> { Err("boom") },
        );

        let err = svc.oneshot(()).await.unwrap_err();
        assert_eq!(err, "boom");
    }
}
