//! [`StepHandler`] and [`CompensationHandler`] traits — auto-implemented for
//! any `async fn` whose parameters all implement [`Extract`] and whose return
//! type matches one of the handler kinds:
//!
//! - **Emit handlers** return [`Emit`]: they dispatch typed events; the
//!   await + response→output map is declared on the topology via
//!   [`task(...).on_response(...)`](crate::TaskBuilder::on_response).
//! - **Value handlers** return any `O: Serialize + DeserializeOwned`:
//!   the step completes immediately with `O` as its typed output (pure
//!   compute / join steps).
//!
//! Inspired by the axum `Handler<T, S>` pattern: marker types record the
//! tuple of extractor types AND the handler kind, so both shapes register
//! through one `task(handler)` call with no turbofish.

use std::any::TypeId;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::Arc;

use serde::Serialize;
use serde::de::DeserializeOwned;

pub use crate::extract::ExtractError;
// IntoResponse + ResponseValue live in `mire` (alongside EventData /
// EventTypeStatic) so the derive macro can emit per-type impls without
// depending on mire-sagas, and so the `Result<Ok, Err>` impl can coexist
// without coherence conflict.
pub use mire::{IntoResponse, ResponseValue};

use crate::extract::Extract;
use crate::result::{DispatchedAction, Emit, StepBuildError};
use crate::saga::Saga;
use crate::step::StepContext;

/// Boxed future type used to erase concrete handler return types into a
/// uniform shape the runner can hold in a registry.
pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Which shape a step handler has. Resolved at registration time from
/// the handler's return type and stored on the `StepEntry` so the
/// validator can cross-check it against the task's declarations.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StepKind {
    /// Returns [`Emit`] — dispatches events; may await per the topology.
    Emit,
    /// Returns a serialisable value — completes immediately with it.
    Value,
}

/// Marker type selecting the [`Emit`]-returning `StepHandler` impls.
pub struct EmitStep;

/// Marker type selecting the value-returning `StepHandler` impls.
pub struct ValueStep<O>(PhantomData<O>);

/// The type-erased outcome of invoking a step handler.
pub enum ErasedStepOutcome {
    /// An emit handler ran: dispatch list + optional per-activation
    /// timeout + stashed build error.
    Emit {
        dispatch: Vec<DispatchedAction>,
        timeout: Option<std::time::Duration>,
        build_error: Option<StepBuildError>,
    },
    /// A value handler ran: the JSON-serialised output (or the
    /// serialisation failure).
    Value(Result<serde_json::Value, StepBuildError>),
}

/// The erased form of a step handler the runner stores. Knows how to satisfy
/// its extractors from a [`StepContext`], call the user's function, and
/// return the erased outcome.
pub type ErasedStepHandler<S> = Arc<
    dyn for<'a> Fn(StepContext<'a, S>) -> BoxFuture<'a, Result<ErasedStepOutcome, ExtractError>>
        + Send
        + Sync,
>;

/// Trait implemented by step handler functions. `Args` is a tuple of
/// extractor types and `Marker` selects the handler kind ([`EmitStep`] or
/// [`ValueStep`]) — both are inference-only markers so the trait is
/// "selected" by the function's signature.
pub trait StepHandler<S: Saga, Args, Marker>: Send + Sync + 'static {
    fn call<'a>(
        &'a self,
        ctx: StepContext<'a, S>,
    ) -> BoxFuture<'a, Result<ErasedStepOutcome, ExtractError>>;

    /// Topology requirements this handler imposes — one per extractor in
    /// its parameter list. The validator collects these at registration
    /// time and checks them against the rest of the DAG, so problems like
    /// `Output<Foo>` with no upstream producer are caught at
    /// `PgSagaRunner::new()` rather than at first activation.
    fn requirements() -> Vec<crate::extract::ExtractRequirement> {
        Vec::new()
    }

    /// Which handler shape this is (resolved from the return type).
    fn kind() -> StepKind;

    /// For value handlers: the output type's identity, used as the step's
    /// typed output. `None` for emit handlers (their output comes from
    /// the task's `.on_response(...)` map).
    fn value_output() -> Option<(TypeId, &'static str)> {
        None
    }
}

// =====================================================================
// Compensation handlers — same split, but compensations have no typed
// output: they return `Emit` (possibly `Emit::none()`); their optional
// await is declared via `comp(handler).on_response::<E>()`.
// =====================================================================

/// The erased counterpart for compensation handlers.
pub type ErasedCompensationHandler<S> = Arc<
    dyn for<'a> Fn(
            StepContext<'a, S>,
        ) -> BoxFuture<'a, Result<ErasedCompensationResult, ExtractError>>
        + Send
        + Sync,
>;

pub struct ErasedCompensationResult {
    pub dispatch: Vec<DispatchedAction>,
    pub timeout: Option<std::time::Duration>,
    pub build_error: Option<StepBuildError>,
}

/// Trait implemented by compensation handler functions: any `async fn`
/// whose parameters all implement [`Extract`] and that returns [`Emit`].
pub trait CompensationHandler<S: Saga, Args>: Send + Sync + 'static {
    fn call<'a>(
        &'a self,
        ctx: StepContext<'a, S>,
    ) -> BoxFuture<'a, Result<ErasedCompensationResult, ExtractError>>;

    fn requirements() -> Vec<crate::extract::ExtractRequirement> {
        Vec::new()
    }
}

// =====================================================================
// Tuple impls. axum trick: one impl per arity, twice per handler kind.
// The two kinds cannot overlap: `Emit` does not implement `Serialize`,
// so a function's return type selects exactly one marker.
// =====================================================================

macro_rules! impl_step_handler {
    ( $( $ty:ident ),* ) => {
        impl<S, F, $($ty,)*> StepHandler<S, ($($ty,)*), EmitStep> for F
        where
            S: Saga,
            // `decide` steps are SYNCHRONOUS — they fold already-committed
            // state/deps into an `Emit` and must be freely re-runnable on
            // recovery. The sync signature is what keeps an effect (an
            // `.await` on I/O) from being smuggled into a step that can't
            // survive a crash; effects live on `.effect(...)` handlers.
            F: Fn($($ty,)*) -> Emit + Send + Sync + 'static,
            $( $ty: Extract<S> + Send + 'static, )*
        {
            #[allow(non_snake_case, unused_variables)]
            fn call<'a>(
                &'a self,
                ctx: StepContext<'a, S>,
            ) -> BoxFuture<'a, Result<ErasedStepOutcome, ExtractError>> {
                Box::pin(async move {
                    $( let $ty = <$ty as Extract<S>>::extract(&ctx)?; )*
                    let emit = (self)($($ty,)*);
                    Ok(ErasedStepOutcome::Emit {
                        dispatch: emit.dispatch,
                        timeout: emit.timeout,
                        build_error: emit.build_error,
                    })
                })
            }

            fn requirements() -> Vec<crate::extract::ExtractRequirement> {
                vec![ $( <$ty as Extract<S>>::requirements(), )* ]
            }

            fn kind() -> StepKind {
                StepKind::Emit
            }
        }

        impl<S, F, Out, $($ty,)*> StepHandler<S, ($($ty,)*), ValueStep<Out>> for F
        where
            S: Saga,
            // Pure compute/join `decide` steps — sync fold of committed deps
            // into a serialisable output. Same crash-rerun safety rationale
            // as the emit variant: no `async`, so no inline effect.
            F: Fn($($ty,)*) -> Out + Send + Sync + 'static,
            Out: Serialize + DeserializeOwned + Send + 'static,
            $( $ty: Extract<S> + Send + 'static, )*
        {
            #[allow(non_snake_case, unused_variables)]
            fn call<'a>(
                &'a self,
                ctx: StepContext<'a, S>,
            ) -> BoxFuture<'a, Result<ErasedStepOutcome, ExtractError>> {
                Box::pin(async move {
                    $( let $ty = <$ty as Extract<S>>::extract(&ctx)?; )*
                    let out = (self)($($ty,)*);
                    Ok(ErasedStepOutcome::Value(
                        serde_json::to_value(&out).map_err(StepBuildError::SerialiseOutput),
                    ))
                })
            }

            fn requirements() -> Vec<crate::extract::ExtractRequirement> {
                vec![ $( <$ty as Extract<S>>::requirements(), )* ]
            }

            fn kind() -> StepKind {
                StepKind::Value
            }

            fn value_output() -> Option<(TypeId, &'static str)> {
                Some((TypeId::of::<Out>(), std::any::type_name::<Out>()))
            }
        }
    };
}

impl_step_handler!();
impl_step_handler!(T1);
impl_step_handler!(T1, T2);
impl_step_handler!(T1, T2, T3);
impl_step_handler!(T1, T2, T3, T4);
impl_step_handler!(T1, T2, T3, T4, T5);
impl_step_handler!(T1, T2, T3, T4, T5, T6);
impl_step_handler!(T1, T2, T3, T4, T5, T6, T7);
impl_step_handler!(T1, T2, T3, T4, T5, T6, T7, T8);

macro_rules! impl_comp_handler {
    ( $( $ty:ident ),* ) => {
        impl<S, F, $($ty,)*> CompensationHandler<S, ($($ty,)*)> for F
        where
            S: Saga,
            // Compensations are `decide` too — sync `fn` returning the undo
            // `Emit`; the release *effect* lives on `comp(...).effect(...)`.
            F: Fn($($ty,)*) -> Emit + Send + Sync + 'static,
            $( $ty: Extract<S> + Send + 'static, )*
        {
            #[allow(non_snake_case, unused_variables)]
            fn call<'a>(
                &'a self,
                ctx: StepContext<'a, S>,
            ) -> BoxFuture<'a, Result<ErasedCompensationResult, ExtractError>> {
                Box::pin(async move {
                    $( let $ty = <$ty as Extract<S>>::extract(&ctx)?; )*
                    let emit = (self)($($ty,)*);
                    Ok(ErasedCompensationResult {
                        dispatch: emit.dispatch,
                        timeout: emit.timeout,
                        build_error: emit.build_error,
                    })
                })
            }

            fn requirements() -> Vec<crate::extract::ExtractRequirement> {
                vec![ $( <$ty as Extract<S>>::requirements(), )* ]
            }
        }
    };
}

impl_comp_handler!();
impl_comp_handler!(T1);
impl_comp_handler!(T1, T2);
impl_comp_handler!(T1, T2, T3);
impl_comp_handler!(T1, T2, T3, T4);
impl_comp_handler!(T1, T2, T3, T4, T5);
impl_comp_handler!(T1, T2, T3, T4, T5, T6);
impl_comp_handler!(T1, T2, T3, T4, T5, T6, T7);
impl_comp_handler!(T1, T2, T3, T4, T5, T6, T7, T8);

// =====================================================================
// RequestHandler — `.effect(handler)` registers one of these.
//
// Same axum-style shape as StepHandler: any async fn whose parameters all
// implement `Extract` and that returns `Resp` (typed response).
// The Req<R> extractor is what makes the request payload itself available
// inside the handler. UNCHANGED by phase 2 — request handlers carry no
// await closures.
// =====================================================================

/// A typed request handler.
///
/// Convention: **the first parameter of a request handler is always
/// `Req<R>`**, the typed payload. Any further parameters are ordinary
/// extractors (`Dep<T>`, `State<S>`, ...).
///
/// Because `Req<R>` is the first arg, the request type `R` shows up in the
/// handler's signature — and the trait's tuple impls reflect that
/// signature shape, so `R` is inferred at `.effect(handler)` with no
/// turbofish required. `Resp` is inferred from the handler's return type.
pub trait RequestHandler<S: Saga, R, Args, Resp>: Send + Sync + 'static {
    fn call<'a>(&'a self, ctx: StepContext<'a, S>) -> BoxFuture<'a, Result<Resp, ExtractError>>;
}

/// Erased handler the runner stores in its registry, keyed by the request
/// event_type. Takes a synthesised StepContext (with the JSON request
/// payload) and returns the JSON-serialised response payload + its
/// event_type, so the runner can deliver it via the existing `deliver_raw`
/// path.
pub type ErasedRequestHandler<S> = std::sync::Arc<
    dyn for<'a> Fn(StepContext<'a, S>) -> BoxFuture<'a, Result<ErasedRequestResponse, ExtractError>>
        + Send
        + Sync,
>;

#[derive(Debug, Clone)]
pub struct ErasedRequestResponse {
    pub event_type: &'static str,
    pub payload: serde_json::Value,
    /// `true` if the handler returned a `Result::Err` — the runner uses
    /// this to surface `StepError::Failed` on the awaiting step rather
    /// than completing it normally.
    pub is_error: bool,
    /// Display of the Err value when `is_error` is true.
    pub error_reason: Option<String>,
}

macro_rules! impl_request_handler {
    ( $( $ty:ident ),* ) => {
        impl<S, F, Fut, R, Resp, $($ty,)*> RequestHandler<S, R, ($($ty,)*), Resp> for F
        where
            S: Saga,
            F: Fn(crate::extract::Req<R>, $($ty,)*) -> Fut + Send + Sync + 'static,
            Fut: Future<Output = Resp> + Send,
            R: DeserializeOwned + Send + 'static,
            Resp: IntoResponse,
            $( $ty: Extract<S> + Send + 'static, )*
        {
            #[allow(non_snake_case, unused_variables)]
            fn call<'a>(
                &'a self,
                ctx: StepContext<'a, S>,
            ) -> BoxFuture<'a, Result<Resp, ExtractError>> {
                Box::pin(async move {
                    let req = <crate::extract::Req<R> as Extract<S>>::extract(&ctx)?;
                    $( let $ty = <$ty as Extract<S>>::extract(&ctx)?; )*
                    Ok((self)(req, $($ty,)*).await)
                })
            }
        }
    };
}

impl_request_handler!();
impl_request_handler!(T1);
impl_request_handler!(T1, T2);
impl_request_handler!(T1, T2, T3);
impl_request_handler!(T1, T2, T3, T4);
impl_request_handler!(T1, T2, T3, T4, T5);
impl_request_handler!(T1, T2, T3, T4, T5, T6);
impl_request_handler!(T1, T2, T3, T4, T5, T6, T7);
