//! `SagaEnvelope<S>` — the framework-owned aggregate that carries any saga's
//! durable state. Users never implement this.
//!
//! The envelope holds:
//! - The user's [`Saga::State`] (carried opaquely as JSON in the aggregate
//!   stream so the envelope's `apply` doesn't need to know `S`'s shape).
//! - The map of completed step outputs, keyed by step name.
//! - The map of per-step statuses (active / completed / failed / compensated).
//! - The overall saga status.

use std::collections::BTreeMap;
use std::marker::PhantomData;

use serde::{Deserialize, Serialize};

use mire::{Aggregate, EventData};

use crate::result::StepError;
use crate::saga::Saga;
use crate::step::StepName;

/// Overall lifecycle of a saga instance.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[non_exhaustive]
pub enum SagaInstanceStatus {
    /// Default: before `Started` has applied.
    #[default]
    NotStarted,
    Running,
    SucceededTerminal,
    /// Failed and fully rolled back — every compensation succeeded.
    FailedTerminal,
    /// Failed, and at least one compensation could not complete (a step is
    /// [`SagaStepStatus::CompensationFailed`]). The forward work that *could*
    /// be rolled back was, but a human must resolve the remaining lock(s)
    /// (e.g. a refund went out but a seat couldn't be unbooked). The saga is
    /// terminal and is **not** auto-re-driven (review D5/SAGA-5).
    FailedTerminalNeedsIntervention,
}

impl SagaInstanceStatus {
    /// Whether the saga has reached a terminal state (succeeded, failed, or
    /// failed-needs-intervention). Terminal sagas activate no further steps,
    /// reject signals, and have their await rows purged. Centralized so a new
    /// terminal variant is handled everywhere at once.
    pub fn is_terminal(&self) -> bool {
        matches!(
            self,
            SagaInstanceStatus::SucceededTerminal
                | SagaInstanceStatus::FailedTerminal
                | SagaInstanceStatus::FailedTerminalNeedsIntervention
        )
    }

    /// Stable string mirror persisted in `es_saga_instances.status` (the
    /// recovery/ops status projection, P3.1). Kept distinct from the Serde
    /// representation so the projection schema doesn't shift if the wire
    /// encoding ever changes. `non-started`/`running` are the *recoverable*
    /// (non-terminal) states the recovery scan looks for.
    pub fn as_db_str(&self) -> &'static str {
        match self {
            SagaInstanceStatus::NotStarted => "not-started",
            SagaInstanceStatus::Running => "running",
            SagaInstanceStatus::SucceededTerminal => "succeeded-terminal",
            SagaInstanceStatus::FailedTerminal => "failed-terminal",
            SagaInstanceStatus::FailedTerminalNeedsIntervention => "failed-needs-intervention",
        }
    }
}

/// Status of one step within a saga instance.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub enum SagaStepStatus {
    Pending,
    Active,
    /// A worker has won the claim to resolve this step (the authorizing
    /// transition committed under OCC) and is running its effect; the
    /// completion has not committed yet. Transient — recovery re-drives a
    /// step stuck here. See `tasks/effects-vs-commit.md`.
    Resolving,
    Completed,
    Failed,
    Compensating,
    Compensated,
    /// The step's compensation was attempted but could not complete (handler
    /// error/panic, unhandled comp-emit). The cascade still rolls back the
    /// other steps; this one is a durable lock requiring manual operator
    /// action and is never auto-retried (review D5/SAGA-5).
    CompensationFailed,
    /// The step's `gated_by` predicate evaluated `false` (or every
    /// predecessor of an `All`-merge step was itself skipped). A skipped
    /// step never activates; downstream steps treat it the same as a
    /// non-completed step (so `after`-deps that include it never trigger).
    Skipped,
    /// The step was still in flight (Active/Resolving — its external
    /// request fired, response pending) when a compensation cascade or
    /// cancellation concluded the saga. Its real-world side effect may
    /// have leaked: no response will be routed, and its compensation did
    /// not run (it wasn't opted in via `compensate_even_if_active`). The
    /// step's `requests_for(step)` payloads stay on the envelope so an
    /// operator can see exactly what was requested (review SAGA-13).
    Abandoned,
}

/// A request a step emitted to an *external* worker (Regime B): the durable
/// record of "what was asked", carried on the saga's own event stream via
/// [`SagaEnvelopeEvent::StepEmitted`]. An external worker discovers these
/// (route by `event_type` through `es_saga_awaits`, then read the payload off
/// the envelope) and delivers the response back via `PgSagaRunner::deliver`.
/// Previously the payload was dropped on suspend, so no worker could ever
/// learn what to do (review C5/SAGA-3/DX-1).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct EmittedRequest {
    pub event_type: String,
    pub payload: serde_json::Value,
}

/// The framework-owned aggregate. Generic over the saga type so its
/// `stream_category` is `S::name()` and its state field is `S::State`.
///
/// Users never construct this directly — the runner does. It's `pub` so
/// callers can inspect status via `EventStore::load`.
#[derive(Debug, Clone)]
pub struct SagaEnvelope<S: Saga> {
    pub state: S::State,
    pub status: SagaInstanceStatus,
    pub steps: BTreeMap<String, SagaStepStatus>,
    /// Outputs of completed steps, serialised. Keyed by step name. The runner
    /// deserialises into the step's typed output when extractors ask for it.
    pub outputs: BTreeMap<String, serde_json::Value>,
    /// The error that *started* the failure path. Set once on the first
    /// `StepFailed` event and never overwritten. Use this for "why is this
    /// saga in a compensation cascade?" diagnostics.
    pub originating_error: Option<StepError>,
    /// The most recent step failure. Set on every `StepFailed` event, so
    /// during a cascade this changes as more steps fail. Use for "what's
    /// the latest thing that went wrong?" diagnostics.
    pub last_error: Option<StepError>,
    /// The original trigger payload (JSON), stashed from the `Started`
    /// event so the runner can reconstruct a step's `Trigger<T>` context
    /// after a suspend/resume across process boundaries.
    pub trigger: Option<serde_json::Value>,
    /// Set when the persisted `Started` state could not be deserialized into
    /// the current `S::State` (schema drift between the version that wrote
    /// the saga and the version replaying it). When set, the durable runner
    /// refuses to act on the instance rather than silently running every
    /// step against `S::State::default()` (review C6/DX-3/SAGA-17).
    pub state_decode_failed: Option<String>,
    /// Outstanding external-worker requests, keyed by the emitting step. A
    /// step's entry is set when it emits to an external worker and suspends
    /// ([`SagaEnvelopeEvent::StepEmitted`]) and cleared when the step resolves
    /// (completed / failed / compensated). Lets a worker recover *what* a
    /// suspended step requested (review C5/SAGA-3).
    pub emitted_requests: BTreeMap<String, Vec<EmittedRequest>>,
    /// How many times each step has been *activated* (`StepActivated`) and
    /// *compensated* (`StepCompensating`). The 1-based count; the 0-based
    /// ordinal `count - 1` is the [`crate::IdempotencyKey`]'s `attempt_seq`,
    /// so the key is **derived from the committed log** (P3.4) — a crash
    /// recovery or OCC retry replays the same committed activations and
    /// reproduces the same key. Today every step activates once (ordinal 0);
    /// a future step-retry feature bumps it naturally.
    pub activation_counts: BTreeMap<String, i64>,
    pub compensation_counts: BTreeMap<String, i64>,
    _marker: PhantomData<S>,
}

impl<S: Saga> Default for SagaEnvelope<S> {
    fn default() -> Self {
        Self {
            state: S::State::default(),
            status: SagaInstanceStatus::NotStarted,
            steps: BTreeMap::new(),
            outputs: BTreeMap::new(),
            originating_error: None,
            last_error: None,
            trigger: None,
            state_decode_failed: None,
            emitted_requests: BTreeMap::new(),
            activation_counts: BTreeMap::new(),
            compensation_counts: BTreeMap::new(),
            _marker: PhantomData,
        }
    }
}

/// Events recorded onto the envelope's stream. The state-transition log.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum SagaEnvelopeEvent {
    Started {
        /// Erased `S::State` payload. The envelope deserialises it via the
        /// concrete `S` at load time.
        state: serde_json::Value,
        /// The triggering domain event payload (JSON), stashed so a
        /// step's `Trigger<T>` extractor can be reconstructed after a
        /// durable suspend/resume. `#[serde(default)]` so streams
        /// written before this field existed still deserialise.
        #[serde(default)]
        trigger: Option<serde_json::Value>,
    },
    StepActivated {
        step: String,
    },
    /// `step` emitted a request to an external worker and suspended (Regime B).
    /// Carries the request payload durably on the stream so a worker can
    /// recover what was asked — recorded in the same commit as `StepActivated`
    /// + the `es_saga_awaits` routing row (review C5/SAGA-3/DX-1).
    StepEmitted {
        step: String,
        event_type: String,
        payload: serde_json::Value,
    },
    /// A worker claimed the right to resolve `step` (won the OCC race);
    /// effect is now running. Authorizing transition for the claim-then-effect
    /// model (see `tasks/effects-vs-commit.md`). `#[serde(default)]` siblings
    /// keep older streams readable.
    StepResolving {
        step: String,
    },
    StepCompleted {
        step: String,
        output: serde_json::Value,
    },
    StepFailed {
        step: String,
        error: StepError,
    },
    /// A compensation for `step` was attempted and could not complete. The
    /// cascade continues for the remaining steps; this step becomes a durable
    /// lock (review D5/SAGA-5).
    StepCompensationFailed {
        step: String,
        error: StepError,
    },
    StepCompensating {
        step: String,
    },
    StepCompensated {
        step: String,
    },
    /// The step's `gated_by` predicate evaluated false, or all of its
    /// predecessors were themselves skipped. The step never runs.
    StepSkipped {
        step: String,
        reason: String,
    },
    /// The step was in flight when a cascade/cancellation concluded the
    /// saga and was not opted into active compensation — its external work
    /// is in limbo (review SAGA-13). Emitted requests stay on the envelope.
    StepAbandoned {
        step: String,
        reason: String,
    },
    Confirmed,
    Cancelled {
        reason: String,
    },
}

impl EventData for SagaEnvelopeEvent {
    fn event_type(&self) -> &'static str {
        match self {
            SagaEnvelopeEvent::Started { .. } => "saga.started",
            SagaEnvelopeEvent::StepActivated { .. } => "saga.step-activated",
            SagaEnvelopeEvent::StepEmitted { .. } => "saga.step-emitted",
            SagaEnvelopeEvent::StepResolving { .. } => "saga.step-resolving",
            SagaEnvelopeEvent::StepCompleted { .. } => "saga.step-completed",
            SagaEnvelopeEvent::StepFailed { .. } => "saga.step-failed",
            SagaEnvelopeEvent::StepCompensationFailed { .. } => "saga.step-compensation-failed",
            SagaEnvelopeEvent::StepCompensating { .. } => "saga.step-compensating",
            SagaEnvelopeEvent::StepCompensated { .. } => "saga.step-compensated",
            SagaEnvelopeEvent::StepSkipped { .. } => "saga.step-skipped",
            SagaEnvelopeEvent::StepAbandoned { .. } => "saga.step-abandoned",
            SagaEnvelopeEvent::Confirmed => "saga.confirmed",
            SagaEnvelopeEvent::Cancelled { .. } => "saga.cancelled",
        }
    }
}

impl<S: Saga> Aggregate for SagaEnvelope<S> {
    type Event = SagaEnvelopeEvent;

    fn stream_category() -> &'static str {
        S::name()
    }

    fn apply(&mut self, event: &SagaEnvelopeEvent) {
        match event {
            SagaEnvelopeEvent::Started { state, trigger } => {
                match serde_json::from_value::<S::State>(state.clone()) {
                    Ok(s) => self.state = s,
                    // `apply` is infallible by signature, so we cannot return
                    // the error here — record it so the runner can refuse to
                    // act on a saga whose state no longer deserializes,
                    // instead of running it against `Default` (C6/DX-3).
                    Err(e) => self.state_decode_failed = Some(e.to_string()),
                }
                self.trigger = trigger.clone();
                self.status = SagaInstanceStatus::Running;
            }
            SagaEnvelopeEvent::StepActivated { step } => {
                self.steps.insert(step.clone(), SagaStepStatus::Active);
                *self.activation_counts.entry(step.clone()).or_insert(0) += 1;
            }
            SagaEnvelopeEvent::StepEmitted {
                step,
                event_type,
                payload,
            } => {
                self.emitted_requests
                    .entry(step.clone())
                    .or_default()
                    .push(EmittedRequest {
                        event_type: event_type.clone(),
                        payload: payload.clone(),
                    });
            }
            SagaEnvelopeEvent::StepResolving { step } => {
                self.steps.insert(step.clone(), SagaStepStatus::Resolving);
            }
            SagaEnvelopeEvent::StepCompensationFailed { step, error } => {
                self.steps
                    .insert(step.clone(), SagaStepStatus::CompensationFailed);
                self.emitted_requests.remove(step);
                if self.originating_error.is_none() {
                    self.originating_error = Some(error.clone());
                }
                self.last_error = Some(error.clone());
            }
            SagaEnvelopeEvent::StepCompleted { step, output } => {
                self.steps.insert(step.clone(), SagaStepStatus::Completed);
                self.outputs.insert(step.clone(), output.clone());
                // The request (if any) is now fulfilled — no longer pending.
                self.emitted_requests.remove(step);
            }
            SagaEnvelopeEvent::StepFailed { step, error } => {
                self.steps.insert(step.clone(), SagaStepStatus::Failed);
                self.emitted_requests.remove(step);
                if self.originating_error.is_none() {
                    self.originating_error = Some(error.clone());
                }
                self.last_error = Some(error.clone());
            }
            SagaEnvelopeEvent::StepCompensating { step } => {
                self.steps
                    .insert(step.clone(), SagaStepStatus::Compensating);
                *self.compensation_counts.entry(step.clone()).or_insert(0) += 1;
            }
            SagaEnvelopeEvent::StepCompensated { step } => {
                self.steps.insert(step.clone(), SagaStepStatus::Compensated);
                self.emitted_requests.remove(step);
            }
            SagaEnvelopeEvent::StepSkipped { step, .. } => {
                self.steps.insert(step.clone(), SagaStepStatus::Skipped);
            }
            SagaEnvelopeEvent::StepAbandoned { step, .. } => {
                self.steps.insert(step.clone(), SagaStepStatus::Abandoned);
                // Deliberately KEEP `emitted_requests[step]`: the in-limbo
                // request payload is exactly what an operator needs to see.
            }
            SagaEnvelopeEvent::Confirmed => {
                self.status = SagaInstanceStatus::SucceededTerminal;
            }
            SagaEnvelopeEvent::Cancelled { .. } => {
                // If any compensation failed, the saga is terminal but
                // *faulty* — it needs manual intervention to clear the
                // remaining lock(s). Derived from step statuses so no
                // separate terminal event is needed (review D5).
                self.status = if self
                    .steps
                    .values()
                    .any(|s| matches!(s, SagaStepStatus::CompensationFailed))
                {
                    SagaInstanceStatus::FailedTerminalNeedsIntervention
                } else {
                    SagaInstanceStatus::FailedTerminal
                };
            }
        }
    }
}

impl<S: Saga> SagaEnvelope<S> {
    /// The decode-failure reason if the persisted `Started` state could not
    /// be deserialized into the current `S::State` (schema drift). `None`
    /// means the state is intact. See [`state_decode_failed`](Self::state_decode_failed).
    pub fn state_decode_error(&self) -> Option<&str> {
        self.state_decode_failed.as_deref()
    }

    pub fn step_status(&self, name: StepName) -> SagaStepStatus {
        self.steps
            .get(name)
            .cloned()
            .unwrap_or(SagaStepStatus::Pending)
    }

    /// The 0-based activation ordinal for `step` — the [`crate::IdempotencyKey`]
    /// `attempt_seq` for a forward effect (P3.4). `count - 1` of the committed
    /// `StepActivated` events; `0` for a step activated once (today's model) or
    /// not yet activated. Because it reads the committed log, a recovery
    /// re-drive or OCC retry reproduces the same ordinal → the same key.
    pub fn activation_seq(&self, name: StepName) -> i64 {
        self.activation_counts
            .get(name)
            .map(|c| c - 1)
            .unwrap_or(0)
            .max(0)
    }

    /// The 0-based compensation ordinal for `step` — the `attempt_seq` for a
    /// compensation effect. `count - 1` of the committed `StepCompensating`
    /// events; stable across a recovery-resumed cascade (the prior, uncommitted
    /// `StepCompensating` isn't on the log, so the re-run sees the same count).
    pub fn compensation_seq(&self, name: StepName) -> i64 {
        self.compensation_counts
            .get(name)
            .map(|c| c - 1)
            .unwrap_or(0)
            .max(0)
    }

    /// `true` if the saga is terminal but at least one compensation could not
    /// complete — a durable lock awaiting manual operator action (D5). Hosts
    /// can poll this to surface stuck rollbacks.
    pub fn needs_intervention(&self) -> bool {
        matches!(
            self.status,
            SagaInstanceStatus::FailedTerminalNeedsIntervention
        )
    }

    /// The names of steps whose compensation failed and still need manual
    /// resolution (empty unless [`needs_intervention`](Self::needs_intervention)).
    pub fn failed_compensations(&self) -> Vec<&str> {
        self.steps
            .iter()
            .filter(|(_, s)| matches!(s, SagaStepStatus::CompensationFailed))
            .map(|(name, _)| name.as_str())
            .collect()
    }

    /// The external-worker request(s) a step emitted and is still awaiting a
    /// response for — the durable record of "what to do" that an external
    /// worker reads to fulfil a suspended step (review C5/SAGA-3). Empty for a
    /// step with no outstanding emit.
    pub fn requests_for(&self, step: StepName) -> &[EmittedRequest] {
        self.emitted_requests
            .get(step)
            .map(Vec::as_slice)
            .unwrap_or(&[])
    }
}
