//! Regression tests for compensation-side failure handling (review D5/SAGA-5).
//!
//! Compensation runs to completion over every step — one failed rollback must
//! never block the others (the refund still goes out even if an unbook can't
//! be reversed). If a compensation handler panics, its extractor fails, or it
//! has a build_error, the runner flags that step `CompensationFailed` (a
//! durable lock), keeps advancing the cascade, and finalises the saga into
//! `FailedTerminalNeedsIntervention` so operators are paged — never crashing
//! the host and never auto-retrying. The originating failure cause stays in
//! `envelope.last_error` for diagnostics.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::sync::atomic::{AtomicUsize, Ordering};

use mire::{Aggregate, EventData};
use mire_sagas::{
    Dep, Emit, Output, PgSagaRunner, Req, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus,
    Start, comp, task,
};
use serde::{Deserialize, Serialize, Serializer};

mod common;

// =====================================================================
// Shared fixtures: trigger, two steps where step2 always fails so
// compensation runs over step1.
// =====================================================================

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

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "source.event")]
struct SourceEvent {
    id: String,
}

impl Aggregate for Source {
    type Event = SourceEvent;
    fn stream_category() -> &'static str {
        "source"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

#[derive(Default, Clone, Serialize, Deserialize)]
struct St;

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

async fn step1() -> Step1Done {
    Step1Done
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "step2.req")]
struct Step2Request;

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "step2.ok")]
struct Step2Ok;

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "step2.err")]
struct Step2Err {
    reason: String,
}

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

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

async fn step2(Output(_): Output<Step1Done>) -> Emit {
    Emit::event(Step2Request)
}

async fn handle_step2(Req(_): Req<Step2Request>) -> Result<Step2Ok, Step2Err> {
    Err(Step2Err {
        reason: "step2 always errs so compensation cascade fires".into(),
    })
}

// =====================================================================
// AlwaysFails: Serialize impl deliberately errors, for build_error
// test.
// =====================================================================

#[derive(Debug, Clone)]
struct AlwaysFails;

impl Serialize for AlwaysFails {
    fn serialize<S: Serializer>(&self, _: S) -> Result<S::Ok, S::Error> {
        Err(serde::ser::Error::custom("deliberate serialise failure"))
    }
}

impl<'de> Deserialize<'de> for AlwaysFails {
    fn deserialize<D: serde::de::Deserializer<'de>>(_: D) -> Result<Self, D::Error> {
        Ok(AlwaysFails)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "step1.bad-compensate")]
struct BadCompensate {
    bad: AlwaysFails,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "step1.compensated")]
struct Compensated;

// =====================================================================
// Saga 1: compensation handler that PANICS.
// =====================================================================

async fn comp_panic() -> Emit {
    panic!("compensation handler panicked");
}

struct PanicCompSaga;
impl Saga for PanicCompSaga {
    type State = St;
    fn name() -> &'static str {
        "panic-comp-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St,
            })
        })
        .step("step1", task(step1).compensate(comp(comp_panic)))
        .step(
            "step2",
            task(step2)
                .after("step1")
                .on_response(|_: Step2Ok| Step2Done)
                .terminal_success(),
        )
        .handle(handle_step2)
    }
}

// =====================================================================
// Saga 2: compensation handler whose extractor fails (Dep not registered).
// =====================================================================

#[derive(Debug, Clone)]
struct MissingDep;

async fn comp_missing_dep(Dep(_): Dep<MissingDep>) -> Emit {
    Emit::none()
}

struct ExtractErrCompSaga;
impl Saga for ExtractErrCompSaga {
    type State = St;
    fn name() -> &'static str {
        "extract-err-comp-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St,
            })
        })
        .step("step1", task(step1).compensate(comp(comp_missing_dep)))
        .step(
            "step2",
            task(step2)
                .after("step1")
                .on_response(|_: Step2Ok| Step2Done)
                .terminal_success(),
        )
        .handle(handle_step2)
    }
}

// =====================================================================
// Saga 3: compensation handler whose emit has a build_error.
// =====================================================================

async fn comp_build_err() -> Emit {
    Emit::event(BadCompensate { bad: AlwaysFails })
}

struct BuildErrCompSaga;
impl Saga for BuildErrCompSaga {
    type State = St;
    fn name() -> &'static str {
        "build-err-comp-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St,
            })
        })
        .step(
            "step1",
            task(step1).compensate(comp(comp_build_err).on_response::<Compensated>()),
        )
        .step(
            "step2",
            task(step2)
                .after("step1")
                .on_response(|_: Step2Ok| Step2Done)
                .terminal_success(),
        )
        .handle(handle_step2)
    }
}

// =====================================================================
// Tests
// =====================================================================

#[tokio::test]
async fn compensation_handler_panic_does_not_crash() {
    let Some(store) = common::store().await else {
        return;
    };
    let topology = PanicCompSaga::build(SagaBuilder::<PanicCompSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = runner
        .trigger::<Source>(&SourceEvent {
            id: common::unique("panic-test"),
        })
        .await
        .unwrap()
        .unwrap();

    // D5: a panicking compensation does not crash the host and does not halt
    // the cascade, but it is NOT hidden as a clean `Compensated`. The step is
    // flagged `CompensationFailed` and the saga ends in the faulty
    // needs-intervention terminal so operators are paged.
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::FailedTerminalNeedsIntervention),
        "a panicking compensation must surface as needs-intervention, not a clean FailedTerminal"
    );
    assert_eq!(
        runner.step_status(&id, "step1").await.unwrap(),
        SagaStepStatus::CompensationFailed
    );
    let env = runner.load(&id).await.unwrap().unwrap();
    assert!(env.needs_intervention());
    assert_eq!(env.failed_compensations(), vec!["step1"]);
}

#[tokio::test]
async fn compensation_handler_missing_dep_caught_at_build_time() {
    // Pre-build-validation this scenario survived to runtime and the
    // compensation handler's extract failure was meant to be tolerated
    // by the cascade. Now the validator catches it at PgSagaRunner::new()
    // BEFORE traffic — surfacing a detailed diagnostic so the user
    // can fix the .compensate(...) signature OR add .dep(value).
    //
    // Defense-in-depth for panics and build_errors in compensation
    // handlers is still proven by the other two tests in this file.
    // Topology validation needs no live database — a lazy pool gives the
    // runner constructor a store without connecting.
    let pool = sqlx::postgres::PgPoolOptions::new()
        .connect_lazy("postgres://unused:unused@localhost:1/unused")
        .unwrap();
    let store = mire::EventStore::new(pool);
    let topology = ExtractErrCompSaga::build(SagaBuilder::<ExtractErrCompSaga>::for_saga());
    let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        let _runner = PgSagaRunner::new(topology, store);
    }));
    let err = res.expect_err("expected PgSagaRunner::new to panic on the misconfig");
    let msg = err
        .downcast::<String>()
        .map(|b| *b)
        .or_else(|e| e.downcast::<&'static str>().map(|s| s.to_string()))
        .unwrap();
    assert!(
        msg.contains("MissingDep") && msg.contains("compensation for step `step1`"),
        "expected detailed diagnostic, got: {msg}"
    );
}

#[tokio::test]
async fn compensation_handler_build_error_does_not_halt_cascade() {
    let Some(store) = common::store().await else {
        return;
    };
    let topology = BuildErrCompSaga::build(SagaBuilder::<BuildErrCompSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = runner
        .trigger::<Source>(&SourceEvent {
            id: common::unique("build-err"),
        })
        .await
        .unwrap()
        .unwrap();

    // D5: a comp emit that can't serialise is a compensation failure — the
    // cascade still completes but the step is flagged `CompensationFailed`
    // and the saga ends needs-intervention (not a clean FailedTerminal).
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::FailedTerminalNeedsIntervention),
        "a comp build error must surface as needs-intervention"
    );
    assert_eq!(
        runner.step_status(&id, "step1").await.unwrap(),
        SagaStepStatus::CompensationFailed
    );
    assert!(
        runner
            .load(&id)
            .await
            .unwrap()
            .unwrap()
            .needs_intervention()
    );
}

// =====================================================================
// Saga 4 (D5): multi-step rollback where ONE compensation fails. The
// cascade must still run the others — proving the refund goes out even
// when an unbook can't be reversed.
// =====================================================================

static REFUND_RUNS: AtomicUsize = AtomicUsize::new(0);

#[derive(Debug, Clone, Serialize, Deserialize)]
struct ChargeDone;
async fn charge_step() -> ChargeDone {
    ChargeDone
}
/// Succeeds — and records that it ran, so the test can prove the refund
/// happened despite the unbook failing.
async fn comp_refund() -> Emit {
    REFUND_RUNS.fetch_add(1, Ordering::SeqCst);
    Emit::none()
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BookDone;
async fn book_step(Output(_): Output<ChargeDone>) -> BookDone {
    BookDone
}
/// Cannot be reversed — panics.
async fn comp_unbook() -> Emit {
    panic!("seat cannot be unbooked");
}

/// Terminal step that fails (its request handler errs), triggering the
/// compensation cascade over `charge` and `book`.
async fn settle_step(Output(_): Output<BookDone>) -> Emit {
    Emit::event(Step2Request)
}

struct MultiCompSaga;
impl Saga for MultiCompSaga {
    type State = St;
    fn name() -> &'static str {
        "multi-comp-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St,
            })
        })
        .step("charge", task(charge_step).compensate(comp(comp_refund)))
        .step(
            "book",
            task(book_step)
                .after("charge")
                .compensate(comp(comp_unbook)),
        )
        .step(
            "settle",
            task(settle_step)
                .after("book")
                .on_response(|_: Step2Ok| Step2Done)
                .terminal_success(),
        )
        .handle(handle_step2)
    }
}

#[tokio::test]
async fn failed_compensation_does_not_block_the_others() {
    let Some(store) = common::store().await else {
        return;
    };
    REFUND_RUNS.store(0, Ordering::SeqCst);
    let topology = MultiCompSaga::build(SagaBuilder::<MultiCompSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = runner
        .trigger::<Source>(&SourceEvent {
            id: common::unique("multi-1"),
        })
        .await
        .unwrap()
        .unwrap();

    // `settle` failed → cascade rolls back in reverse: book (unbook panics)
    // then charge (refund succeeds). The unbook failure must NOT prevent the
    // refund from running.
    assert_eq!(
        runner.step_status(&id, "book").await.unwrap(),
        SagaStepStatus::CompensationFailed,
        "unbook panicked, so book is CompensationFailed"
    );
    assert_eq!(
        runner.step_status(&id, "charge").await.unwrap(),
        SagaStepStatus::Compensated,
        "refund must have run to completion despite the earlier unbook failure"
    );
    assert_eq!(
        REFUND_RUNS.load(Ordering::SeqCst),
        1,
        "the refund compensation must actually execute even though unbook failed"
    );

    // Saga locks for manual intervention, flagging the specific failed comp.
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::FailedTerminalNeedsIntervention)
    );
    let env = runner.load(&id).await.unwrap().unwrap();
    assert!(env.needs_intervention());
    assert_eq!(env.failed_compensations(), vec!["book"]);
}
