//! Step context: what a running step sees.

use std::any::TypeId;
use std::collections::BTreeMap;

use serde_json::Value;

use crate::saga::Saga;

/// Stable name for a step within a saga. Used as the durable identifier in the
/// envelope's recorded events.
pub type StepName = &'static str;

/// A stable idempotency key for the effect a handler is about to run:
/// `(instance_id, step, attempt_seq)`.
///
/// The runner guarantees the **same** key across replays of the same
/// authorized effect — an OCC retry or a crash-recovery re-drive reproduce it
/// exactly (`attempt_seq` is derived from the committed `StepActivated`
/// ordinal, not a wall clock or a fresh counter). A handler performing a
/// non-replayable side effect can therefore dedupe on it: run once, then on a
/// matching key choose to skip, return the prior result, or fail. The handler
/// owns that policy — the runner never silently dedupes. See
/// `tasks/effects-vs-commit.md` (D1/D2).
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct IdempotencyKey {
    pub instance_id: String,
    pub step: String,
    /// The activation ordinal for this step within the instance. `0` for the
    /// (current) single-activation model; distinguishes re-activations once
    /// step retries exist.
    pub attempt_seq: i64,
}

impl IdempotencyKey {
    /// A flat string form suitable as a dedupe-table primary key.
    pub fn token(&self) -> String {
        format!("{}/{}/{}", self.instance_id, self.step, self.attempt_seq)
    }
}

impl std::fmt::Display for IdempotencyKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.token())
    }
}

/// Context handed to a step handler at invocation time. Carries the saga's
/// state, the trigger event (if any), the typed outputs of completed upstream
/// steps, the injected dependencies, and the cause (for compensation handlers).
///
/// Extractors pull from this context — handlers never touch it directly.
pub struct StepContext<'a, S: Saga> {
    pub(crate) state: &'a S::State,

    /// The original trigger payload that started this saga, if extracting it
    /// is valid for this step. Stored as JSON because the trigger type is
    /// erased by the time the runner builds the context.
    pub(crate) trigger: Option<&'a Value>,

    /// The request payload being handled, if this context is for a typed
    /// request handler invocation. Present iff the runner is calling a
    /// `.handle::<R>(...)` handler.
    pub(crate) request: Option<&'a Value>,

    /// Completed step outputs, keyed by the output type's `TypeId`. Each step
    /// produces exactly one output type within a saga.
    pub(crate) outputs: &'a BTreeMap<TypeId, Value>,

    /// Dependencies registered on the builder, keyed by `TypeId`.
    pub(crate) deps: &'a BTreeMap<TypeId, std::sync::Arc<dyn std::any::Any + Send + Sync>>,

    /// Present in compensation handlers; carries the originating step error.
    pub(crate) cause: Option<&'a crate::result::StepError>,

    /// Idempotency context for the effect this handler is about to run.
    /// Populated by the runner via [`with_idempotency`](Self::with_idempotency);
    /// defaults are empty/`0` for bare test contexts. Surfaced to handlers via
    /// the [`IdempotencyKey`] extractor.
    pub(crate) instance_id: &'a str,
    pub(crate) step: StepName,
    pub(crate) attempt_seq: i64,
}

impl<'a, S: Saga> StepContext<'a, S> {
    /// Construct a `StepContext` for tests. Production code receives one from
    /// the runner; this is the only documented way to synthesise one outside.
    #[doc(hidden)]
    pub fn for_testing(
        state: &'a S::State,
        trigger: Option<&'a Value>,
        outputs: &'a BTreeMap<TypeId, Value>,
        deps: &'a BTreeMap<TypeId, std::sync::Arc<dyn std::any::Any + Send + Sync>>,
        cause: Option<&'a crate::result::StepError>,
    ) -> Self {
        Self {
            state,
            trigger,
            request: None,
            outputs,
            deps,
            cause,
            instance_id: "",
            step: "",
            attempt_seq: 0,
        }
    }

    /// Like [`for_testing`] but with a request payload (for testing
    /// request handlers, not step handlers).
    ///
    /// This is on the production hot path for in-process request
    /// handlers, so the empty `outputs` slot reads from a single
    /// `OnceLock`-backed static — no per-call allocation.
    #[doc(hidden)]
    pub fn for_testing_with_request(
        state: &'a S::State,
        request: &'a Value,
        deps: &'a BTreeMap<TypeId, std::sync::Arc<dyn std::any::Any + Send + Sync>>,
    ) -> Self {
        use std::sync::OnceLock;
        static EMPTY_OUTPUTS: OnceLock<BTreeMap<TypeId, Value>> = OnceLock::new();
        let outputs = EMPTY_OUTPUTS.get_or_init(BTreeMap::new);
        Self {
            state,
            trigger: None,
            request: Some(request),
            outputs,
            deps,
            cause: None,
            instance_id: "",
            step: "",
            attempt_seq: 0,
        }
    }

    /// Attach the idempotency context (instance id, step name, activation
    /// ordinal) the runner has for this invocation. Returns `self` so it
    /// chains onto a `for_testing*` constructor on the production path.
    pub(crate) fn with_idempotency(
        mut self,
        instance_id: &'a str,
        step: StepName,
        attempt_seq: i64,
    ) -> Self {
        self.instance_id = instance_id;
        self.step = step;
        self.attempt_seq = attempt_seq;
        self
    }

    /// The stable idempotency key for the effect this handler is about to run.
    /// See [`IdempotencyKey`].
    pub fn idempotency_key(&self) -> IdempotencyKey {
        IdempotencyKey {
            instance_id: self.instance_id.to_string(),
            step: self.step.to_string(),
            attempt_seq: self.attempt_seq,
        }
    }

    pub fn state(&self) -> &'a S::State {
        self.state
    }

    pub fn trigger_json(&self) -> Option<&'a Value> {
        self.trigger
    }

    pub fn request_json(&self) -> Option<&'a Value> {
        self.request
    }

    pub fn output_json(&self, type_id: TypeId) -> Option<&'a Value> {
        self.outputs.get(&type_id)
    }

    pub fn dep_any(&self, type_id: TypeId) -> Option<&'a (dyn std::any::Any + Send + Sync)> {
        self.deps.get(&type_id).map(|arc| arc.as_ref())
    }

    pub fn cause(&self) -> Option<&'a crate::result::StepError> {
        self.cause
    }
}
