//! Phase 5 (review SAGA-13): in-flight steps in a compensation cascade.
//!
//! When a cascade fires while a step is still **Active** (its external
//! request already went out, the response is pending), the step's
//! real-world side effect is in limbo. Decided semantics (2026-06-12):
//!
//! - **Default**: the step is marked `Abandoned` — a visible, durable
//!   record that external work may have leaked (its `requests_for(step)`
//!   payloads stay on the envelope so an operator can see *what* leaked).
//!   Its compensation does NOT run (compensations usually need the step's
//!   `Output<T>`, which never came to exist).
//! - **Opt-in** via `.compensate_even_if_active(step, handler)`: the
//!   compensation runs in the cascade even from `Active` — valid only for
//!   handlers that don't read the step's own output (e.g. release-by-id
//!   from saga `State`).
//!
//! Both runners must agree. The pg tests are gated on `DATABASE_URL`.

#![allow(dead_code)]

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

use mire::{Aggregate, EventStore};
use mire_sagas::{
    Dep, Emit, PgSagaRunner, Req, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus, Start,
    State, comp, task,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

#[derive(Default, Clone)]
struct Trip;

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "abandon.trip-requested")]
struct TripRequested {
    id: String,
}

impl Aggregate for Trip {
    type Event = TripRequested;
    fn stream_category() -> &'static str {
        "abandon-trip"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

#[derive(Default, Clone, Serialize, Deserialize)]
struct St {
    id: String,
}

// `hold` emits an external request and stays Active (no in-process
// handler for HoldRequested).
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "abandon.hold-requested")]
struct HoldRequested {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "abandon.hold-confirmed")]
struct HoldConfirmed;

#[derive(Debug, Clone, Serialize, Deserialize)]
struct HoldDone;

async fn hold_step(State(s): State<St>) -> Emit {
    Emit::event(HoldRequested { id: s.id.clone() })
}

// `pay` fails via its in-process handler returning Err — fires the
// cascade while `hold` is still Active.
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "abandon.pay-requested")]
struct PayRequested {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "abandon.pay-ok")]
struct PayOk;

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "abandon.pay-declined")]
struct PayDeclined {
    reason: String,
}

impl std::fmt::Display for PayDeclined {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.reason)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct PayDone;

async fn pay_step(State(s): State<St>) -> Emit {
    Emit::event(PayRequested { id: s.id.clone() })
}

async fn handle_pay(Req(_): Req<PayRequested>) -> Result<PayOk, PayDeclined> {
    Err(PayDeclined {
        reason: "card declined".into(),
    })
}

/// Counts opt-in compensation invocations so tests can assert it ran.
#[derive(Clone, Default)]
struct CompProbe(Arc<AtomicU32>);

// Opt-in compensation that deliberately needs NO Output<T> — it releases
// by the id carried in saga state (the documented constraint).
async fn release_hold_by_id(State(_s): State<St>, Dep(probe): Dep<CompProbe>) -> Emit {
    probe.0.fetch_add(1, Ordering::SeqCst);
    Emit::none()
}

// ---------------------------------------------------------------------
// Saga A: no opt-in → `hold` must end Abandoned.
// ---------------------------------------------------------------------

struct AbandonDefault;
impl Saga for AbandonDefault {
    type State = St;
    fn name() -> &'static str {
        "abandon-default"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Trip, _>(|e: &TripRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "hold",
            task(hold_step).on_response(|_: HoldConfirmed| HoldDone),
        )
        .step(
            "pay",
            task(pay_step)
                .terminal_success()
                .on_response(|_: PayOk| PayDone),
        )
        .handle(handle_pay)
    }
}

// ---------------------------------------------------------------------
// Saga B: `.compensate_even_if_active("hold", ...)` → comp runs from
// Active; `hold` ends Compensated.
// ---------------------------------------------------------------------

struct AbandonOptIn;
impl Saga for AbandonOptIn {
    type State = St;
    fn name() -> &'static str {
        "abandon-opt-in"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Trip, _>(|e: &TripRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "hold",
            task(hold_step)
                .on_response(|_: HoldConfirmed| HoldDone)
                .compensate(comp(release_hold_by_id).even_if_active()),
        )
        .step(
            "pay",
            task(pay_step)
                .terminal_success()
                .on_response(|_: PayOk| PayDone),
        )
        .handle(handle_pay)
    }
}

#[tokio::test]
async fn pg_active_step_is_marked_abandoned_by_default() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = PgSagaRunner::new(
        AbandonDefault::build(SagaBuilder::<AbandonDefault>::for_saga()),
        store,
    );
    let id = format!("ab-pg-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Trip>(&TripRequested { id: id.clone() })
        .await
        .expect("trigger");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(env.step_status("hold"), SagaStepStatus::Abandoned);
    assert_eq!(env.requests_for("hold").len(), 1);
}

#[tokio::test]
async fn pg_opt_in_compensation_runs_from_active() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let probe = CompProbe::default();
    let topology = AbandonOptIn::build(SagaBuilder::<AbandonOptIn>::for_saga()).dep(probe.clone());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("ab-pg-opt-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Trip>(&TripRequested { id: id.clone() })
        .await
        .expect("trigger");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(env.step_status("hold"), SagaStepStatus::Compensated);
    assert_eq!(probe.0.load(Ordering::SeqCst), 1);
}

async fn maybe_store() -> Option<EventStore> {
    let url = std::env::var("DATABASE_URL")
        .ok()
        .filter(|s| !s.is_empty())?;
    let pool = PgPool::connect(&url).await.expect("connect");
    let store = EventStore::new(pool);
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");
    Some(store)
}
