//! Chunk 4e first slice: PgSagaRunner compensation cascade.
//!
//! When a request handler returns `Result::Err`, the runner records
//! StepFailed and walks completed steps in reverse declaration order,
//! invoking their `.compensate(...)` handlers (immediate-complete in
//! this slice — Emit::none()). Saga terminates with
//! FailedTerminal.
//!
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

// =====================================================================
// Trigger
// =====================================================================

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

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-comp.booking-requested")]
struct BookingRequested {
    id: String,
}

impl Aggregate for Booking {
    type Event = BookingRequested;
    fn stream_category() -> &'static str {
        "pg-comp-booking"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

// =====================================================================
// Step 1: flight reserve (succeeds, registers compensation)
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-comp.flight-hold-requested")]
struct FlightHoldRequest {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-comp.flight-held")]
struct FlightHeld {
    hold_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct FlightReserved {
    hold_id: String,
}

async fn reserve_flight(State(s): State<BookingState>) -> Emit {
    Emit::event(FlightHoldRequest { id: s.id.clone() })
}

async fn handle_flight_hold(Req(r): Req<FlightHoldRequest>) -> FlightHeld {
    FlightHeld {
        hold_id: format!("fh-{}", r.id),
    }
}

async fn release_flight(Output(_f): Output<FlightReserved>) -> Emit {
    // First slice: noop compensation. Real release would emit
    // ReleaseHold + await ReleaseDone; that's a future slice.
    Emit::none()
}

// =====================================================================
// Step 2: payment (handler ERRORS) — triggers the cascade
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-comp.payment-requested")]
struct PaymentRequested {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-comp.payment-settled")]
struct PaymentSettled {
    payment_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-comp.payment-rejected")]
struct PaymentRejected {
    reason: String,
}

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

#[derive(Debug, Clone, Serialize, Deserialize)]
struct PaymentCompleted {
    payment_id: String,
}

async fn pay(State(s): State<BookingState>) -> Emit {
    Emit::event(PaymentRequested { id: s.id.clone() })
}

async fn handle_payment(Req(_): Req<PaymentRequested>) -> Result<PaymentSettled, PaymentRejected> {
    Err(PaymentRejected {
        reason: "card declined".into(),
    })
}

// =====================================================================
// Saga
// =====================================================================

struct PgCompSaga;
impl Saga for PgCompSaga {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-comp-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Booking, _>(|e: &BookingRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: BookingState { id: e.id.clone() },
            })
        })
        .step(
            "flight",
            task(reserve_flight)
                .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id })
                .compensate(comp(release_flight)),
        )
        .step(
            "pay",
            task(pay)
                .after("flight")
                .terminal_success()
                .on_response(|p: PaymentSettled| PaymentCompleted {
                    payment_id: p.payment_id,
                }),
        )
        .handle(handle_flight_hold)
        .handle(handle_payment)
    }
}

// A payment handler that PANICS instead of returning Err — exercises the
// pg runner's panic catching (SAGA-6).
async fn handle_payment_panics(
    Req(_): Req<PaymentRequested>,
) -> Result<PaymentSettled, PaymentRejected> {
    panic!("payment gateway client panicked");
}

struct PgPanicSaga;
impl Saga for PgPanicSaga {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-panic-comp-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Booking, _>(|e: &BookingRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: BookingState { id: e.id.clone() },
            })
        })
        .step(
            "flight",
            task(reserve_flight)
                .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id })
                .compensate(comp(release_flight)),
        )
        .step(
            "pay",
            task(pay)
                .after("flight")
                .terminal_success()
                .on_response(|p: PaymentSettled| PaymentCompleted {
                    payment_id: p.payment_id,
                }),
        )
        .handle(handle_flight_hold)
        .handle(handle_payment_panics)
    }
}

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)
}

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

#[tokio::test]
async fn payment_failure_compensates_flight_and_terminates_failed() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = PgCompSaga::build(SagaBuilder::<PgCompSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("comp-{}", uuid::Uuid::new_v4());

    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .expect("trigger");
    let env = runner.load(&id).await.unwrap().unwrap();

    // Final state: saga is failed-terminal, flight is Compensated,
    // pay is Failed.
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    assert_eq!(env.step_status("pay"), SagaStepStatus::Failed);

    // The originating error is preserved on the envelope (Chunk's I4
    // semantics still hold over the durable runner).
    assert!(env.originating_error.is_some());
    let orig = env.originating_error.as_ref().unwrap();
    assert!(format!("{orig}").contains("card declined"));
}

/// SAGA-6/SAGA-7: a panicking forward request handler must NOT crash the host
/// or abort the call uncommitted. The runner catches the panic, fails the
/// step, and runs the compensation cascade — same as if the handler returned
/// Err. (Parity with the in-memory runner.)
#[tokio::test]
async fn panicking_forward_handler_compensates_instead_of_crashing() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = PgPanicSaga::build(SagaBuilder::<PgPanicSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("panic-{}", uuid::Uuid::new_v4());

    // The panic must be caught — trigger returns Ok, the host survives.
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .expect("a panicking handler must not crash the runner (SAGA-6)");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    assert_eq!(env.step_status("pay"), SagaStepStatus::Failed);
    let orig = env.originating_error.as_ref().unwrap();
    assert!(
        format!("{orig}").contains("panicked"),
        "originating error should reflect the panic: {orig}"
    );
}

#[tokio::test]
async fn no_completed_steps_means_no_compensations_invoked() {
    // A saga where the very first step fails (no prior completions).
    // Compensation queue is empty; saga still reaches FailedTerminal.

    struct ImmediateFailSaga;
    impl Saga for ImmediateFailSaga {
        type State = BookingState;
        fn name() -> &'static str {
            "pg-immediate-fail-saga"
        }
        fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
            b.start_on::<Booking, _>(|e: &BookingRequested| {
                Some(Start {
                    instance_id: e.id.clone(),
                    state: BookingState { id: e.id.clone() },
                })
            })
            .step(
                "pay",
                task(pay)
                    .terminal_success()
                    .on_response(|p: PaymentSettled| PaymentCompleted {
                        payment_id: p.payment_id,
                    }),
            )
            .handle(handle_payment)
        }
    }

    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = ImmediateFailSaga::build(SagaBuilder::<ImmediateFailSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("fail-first-{}", uuid::Uuid::new_v4());

    runner
        .trigger::<Booking>(&BookingRequested { 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("pay"), SagaStepStatus::Failed);
}

// =====================================================================
// 4e second slice: compensation handler that emits + awaits.
// =====================================================================

mod comp_with_emit {
    use super::*;

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-comp.release-hold-requested")]
    struct ReleaseHoldRequest {
        hold_id: String,
    }

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-comp.hold-released")]
    struct HoldReleased {
        hold_id: String,
    }

    async fn release_flight_for_real(Output(f): Output<FlightReserved>) -> Emit {
        // Real-world shape: emit "release this hold" + await the
        // released-ack event. The runner drives this through the
        // same in-process handler dispatch path as forward steps.
        Emit::event(ReleaseHoldRequest { hold_id: f.hold_id })
    }

    async fn handle_release(Req(r): Req<ReleaseHoldRequest>) -> HoldReleased {
        HoldReleased { hold_id: r.hold_id }
    }

    struct EmittingCompSaga;
    impl Saga for EmittingCompSaga {
        type State = BookingState;
        fn name() -> &'static str {
            "pg-emitting-comp-saga"
        }
        fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
            b.start_on::<Booking, _>(|e: &BookingRequested| {
                Some(Start {
                    instance_id: e.id.clone(),
                    state: BookingState { id: e.id.clone() },
                })
            })
            .step(
                "flight",
                task(reserve_flight)
                    .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id })
                    .compensate(comp(release_flight_for_real).on_response::<HoldReleased>()),
            )
            .step(
                "pay",
                task(pay)
                    .after("flight")
                    .terminal_success()
                    .on_response(|p: PaymentSettled| PaymentCompleted {
                        payment_id: p.payment_id,
                    }),
            )
            .handle(handle_flight_hold)
            .handle(handle_payment)
            .handle(handle_release)
        }
    }

    #[tokio::test]
    async fn comp_handler_can_emit_and_await_response() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = EmittingCompSaga::build(SagaBuilder::<EmittingCompSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);
        let id = format!("emit-comp-{}", uuid::Uuid::new_v4());

        runner
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .expect("trigger ok");
        let env = runner.load(&id).await.unwrap().unwrap();

        // Payment failed → flight compensated via the emit-style
        // comp handler. Saga reaches FailedTerminal.
        assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
        assert_eq!(env.step_status("pay"), SagaStepStatus::Failed);
        assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    }

    #[tokio::test]
    async fn comp_emit_with_no_registered_handler_flags_compensation_failed() {
        // D5: a compensation emit with no in-process handler can't be
        // dispatched — but that no longer aborts the whole transition (which
        // left the saga stuck and re-ran every tick — SAGA-5). Instead the
        // step is flagged `CompensationFailed` and the saga ends in the
        // needs-intervention terminal so an operator resolves it.

        async fn comp_no_handler(_: Output<FlightReserved>) -> Emit {
            Emit::event(ReleaseHoldRequest {
                hold_id: "x".into(),
            })
        }

        struct NoHandlerCompSaga;
        impl Saga for NoHandlerCompSaga {
            type State = BookingState;
            fn name() -> &'static str {
                "pg-no-handler-comp-saga"
            }
            fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
                b.start_on::<Booking, _>(|e: &BookingRequested| {
                    Some(Start {
                        instance_id: e.id.clone(),
                        state: BookingState { id: e.id.clone() },
                    })
                })
                .step(
                    "flight",
                    task(reserve_flight)
                        .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id })
                        .compensate(comp(comp_no_handler).on_response::<HoldReleased>()),
                )
                .step(
                    "pay",
                    task(pay).after("flight").terminal_success().on_response(
                        |p: PaymentSettled| PaymentCompleted {
                            payment_id: p.payment_id,
                        },
                    ),
                )
                .handle(handle_flight_hold)
                .handle(handle_payment)
                // No .handle(handle_release) — comp will fail.
            }
        }

        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = NoHandlerCompSaga::build(SagaBuilder::<NoHandlerCompSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);

        let id = format!("no-handler-{}", uuid::Uuid::new_v4());
        // `pay` fails (terminal), triggering compensation of `flight`; that
        // comp emits with no handler → CompensationFailed. The trigger call
        // itself succeeds (the failure is recorded, not propagated).
        let res = runner
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .expect("trigger must not abort on a compensation failure (D5)");
        assert!(res.is_some());

        let env = runner
            .load(&id)
            .await
            .unwrap()
            .expect("instance should exist");
        assert_eq!(
            env.step_status("flight"),
            mire_sagas::SagaStepStatus::CompensationFailed,
            "an undispatchable comp emit flags the step CompensationFailed"
        );
        assert_eq!(
            env.status,
            mire_sagas::SagaInstanceStatus::FailedTerminalNeedsIntervention
        );
        assert!(env.needs_intervention());
        assert_eq!(env.failed_compensations(), vec!["flight"]);
    }
}
