//! Chunk 4d timeouts: a suspended step whose deadline elapses is
//! failed with `StepError::Timeout` and compensation cascades.
//!
//! `tick_timeouts(now)` takes the current time explicitly so tests can
//! force expiry without real-time waits (the durable analogue of the
//! in-memory runner's `advance_time`).
//!
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

use std::time::Duration;

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

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

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

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

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

// Step 1: flight, completes inline (has handler), registers comp.
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-to.flight-req")]
struct FlightReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-to.flight-held")]
struct FlightHeld {
    hold_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct FlightReserved {
    hold_id: String,
}

fn reserve_flight(State(s): State<BookingState>) -> Emit {
    Emit::event(FlightReq { id: s.id.clone() })
}
async fn handle_flight(Req(r): Req<FlightReq>) -> FlightHeld {
    FlightHeld {
        hold_id: format!("fh-{}", r.id),
    }
}
fn release_flight(Output(_): Output<FlightReserved>) -> Emit {
    Emit::none()
}

// Step 2: payment, emits to an EXTERNAL worker (no handler) with a
// timeout. Suspends; if no response arrives the deadline expires.
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-to.payment-req")]
struct PaymentReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-to.payment-settled")]
struct PaymentSettled;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PaymentDone;

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

// `tick_timeouts` is saga-WIDE (a worker expires every instance of a
// saga name). So two parallel tests must use DISTINCT saga names, or
// each one's tick would stomp the other's instances. A generic build
// helper keeps the topology shared while the name differs per test.
fn build_timeout_topology<S>(b: SagaBuilder<S>) -> SagaBuilder<S>
where
    S: Saga<State = BookingState>,
{
    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(
        "payment",
        task(pay)
            .after("flight")
            .on_response(|_: PaymentSettled| PaymentDone)
            .timeout(Duration::from_secs(600))
            .terminal_success(),
    )
    .effect(handle_flight)
    // No handler for PaymentReq → payment suspends with a deadline.
}

struct TimeoutSagaA;
impl Saga for TimeoutSagaA {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-timeout-saga-a"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        build_timeout_topology(b)
    }
}

struct TimeoutSagaB;
impl Saga for TimeoutSagaB {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-timeout-saga-b"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        build_timeout_topology(b)
    }
}

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

fn build_runner_a(store: EventStore) -> PgSagaRunner<TimeoutSagaA> {
    PgSagaRunner::new(
        TimeoutSagaA::build(SagaBuilder::<TimeoutSagaA>::for_saga()),
        store,
    )
}
fn build_runner_b(store: EventStore) -> PgSagaRunner<TimeoutSagaB> {
    PgSagaRunner::new(
        TimeoutSagaB::build(SagaBuilder::<TimeoutSagaB>::for_saga()),
        store,
    )
}

#[tokio::test]
async fn expired_deadline_times_out_step_and_compensates() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = build_runner_a(store.clone());
    let id = format!("to-{}", uuid::Uuid::new_v4());

    // Trigger: flight completes inline, payment suspends with a 600s
    // deadline awaiting the external PaymentSettled.
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.step_status("flight"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("payment"), SagaStepStatus::Active);
    assert_eq!(env.status, SagaInstanceStatus::Running);

    // Tick with a `now` BEFORE the deadline → nothing expires.
    let before = chrono::Utc::now();
    let n = runner.tick_timeouts(before).await.unwrap();
    assert_eq!(n, 0, "deadline 600s out shouldn't expire yet");
    assert_eq!(
        runner.load(&id).await.unwrap().unwrap().status,
        SagaInstanceStatus::Running
    );

    // Tick with a `now` PAST the deadline → payment times out.
    let after = chrono::Utc::now() + chrono::Duration::seconds(700);
    let n = runner.tick_timeouts(after).await.unwrap();
    assert_eq!(n, 1, "one instance should have timed out");

    // payment Failed, flight Compensated, saga FailedTerminal.
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(env.step_status("payment"), SagaStepStatus::Failed);
    assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);

    // P5.4: the timeout error reports the CONFIGURED budget (~600s), derived
    // from the await's `deadline_at - created_at` — not the old `timeout_ms: 0`.
    match env.originating_error {
        Some(StepError::Timeout {
            ref step,
            timeout_ms,
        }) => {
            assert_eq!(step, "payment");
            assert!(
                (599_000..=601_000).contains(&timeout_ms),
                "timeout_ms should report the ~600_000ms budget, got {timeout_ms}"
            );
        }
        ref other => panic!("expected a Timeout originating_error, got {other:?}"),
    }

    // The await row was consumed.
    let await_count: i64 =
        sqlx::query_scalar("SELECT count(*) FROM es_saga_awaits WHERE instance_id = $1")
            .bind(&id)
            .fetch_one(store.pool())
            .await
            .unwrap();
    assert_eq!(await_count, 0);
}

#[tokio::test]
async fn deliver_before_deadline_wins_no_timeout() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = build_runner_b(store);
    let id = format!("to-win-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .unwrap();

    // External worker responds in time.
    runner.deliver(&id, &PaymentSettled).await.unwrap();

    // Saga reached terminal-success; a later tick finds nothing.
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);

    let future = chrono::Utc::now() + chrono::Duration::seconds(700);
    let n = runner.tick_timeouts(future).await.unwrap();
    assert_eq!(n, 0, "no await remains after successful delivery");
}

// =====================================================================
// Parallel-suspend timeout: two branches each suspend with a deadline.
// When they time out, the instance must reach FailedTerminal exactly
// once and leave NO leaked await rows — the sibling await of the branch
// that triggered the timeout must be cleaned up too.
// =====================================================================

mod parallel_timeout {
    use super::*;

    // Two independent external branches, then a terminal join.
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pt.a-req")]
    struct AReq {
        id: String,
    }
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pt.a-ok")]
    struct AOk;
    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct ADone;
    fn step_a(State(s): State<BookingState>) -> Emit {
        Emit::event(AReq { id: s.id.clone() })
    }

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pt.b-req")]
    struct BReq {
        id: String,
    }
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pt.b-ok")]
    struct BOk;
    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct BDone;
    fn step_b(State(s): State<BookingState>) -> Emit {
        Emit::event(BReq { id: s.id.clone() })
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct Joined;
    fn join(Output(_a): Output<ADone>, Output(_b): Output<BDone>) -> Joined {
        Joined
    }

    struct ParallelTimeoutSaga;
    impl Saga for ParallelTimeoutSaga {
        type State = BookingState;
        fn name() -> &'static str {
            "pg-parallel-timeout-saga"
        }
        fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
            b.default_step_timeout(Duration::from_secs(300))
                .start_on::<Booking, _>(|e: &BookingRequested| {
                    Some(Start {
                        instance_id: e.id.clone(),
                        state: BookingState { id: e.id.clone() },
                    })
                })
                .step("a", task(step_a).on_response(|_: AOk| ADone))
                .step("b", task(step_b).on_response(|_: BOk| BDone))
                .step("join", task(join).after(["a", "b"]).terminal_success())
            // No handlers → both a and b suspend with deadlines.
        }
    }

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

        runner
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .unwrap();

        // Both branches suspended → two await rows.
        let awaits: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM es_saga_awaits \
             WHERE saga_name = 'pg-parallel-timeout-saga' AND instance_id = $1",
        )
        .bind(&id)
        .fetch_one(store.pool())
        .await
        .unwrap();
        assert_eq!(awaits, 2, "both parallel branches should have await rows");

        // Tick past the deadline. Both awaits are expired, but the first
        // to time out cancels the whole instance — so exactly ONE
        // instance times out, not one-per-branch.
        let after = chrono::Utc::now() + chrono::Duration::seconds(400);
        let n = runner.tick_timeouts(after).await.unwrap();
        assert_eq!(
            n, 1,
            "a parallel-suspend instance must count as ONE timeout, not one per branch"
        );

        // Instance is FailedTerminal.
        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(
            env.status,
            SagaInstanceStatus::FailedTerminal,
            "timed-out parallel saga must be FailedTerminal"
        );

        // CRITICAL: no await rows leak. The sibling of whichever branch
        // triggered the timeout must be cleaned up too.
        let leaked: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM es_saga_awaits \
             WHERE saga_name = 'pg-parallel-timeout-saga' AND instance_id = $1",
        )
        .bind(&id)
        .fetch_one(store.pool())
        .await
        .unwrap();
        assert_eq!(
            leaked, 0,
            "no await rows may leak after a parallel-suspend timeout"
        );

        // A second tick must be a clean no-op (instance already terminal).
        let n = runner.tick_timeouts(after).await.unwrap();
        assert_eq!(n, 0, "second tick on a terminal instance must do nothing");
    }
}

// =====================================================================
// The dangerous case: one branch COMPLETES (with a compensation), two
// siblings are abandoned and time out. The completed branch must be
// compensated EXACTLY ONCE — not once per timed-out sibling.
// =====================================================================

mod parallel_timeout_double_comp {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU64, Ordering};

    // Branch A completes in-process and registers a compensation that
    // counts how many times it runs.
    static COMP_RUNS: AtomicU64 = AtomicU64::new(0);

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "ptd.a-req")]
    struct AReq {
        id: String,
    }
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "ptd.a-ok")]
    struct AOk;
    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct ADone;
    fn step_a(State(s): State<BookingState>) -> Emit {
        Emit::event(AReq { id: s.id.clone() })
    }
    async fn handle_a(Req(_): Req<AReq>) -> AOk {
        AOk
    }
    fn comp_a(Output(_): Output<ADone>) -> Emit {
        COMP_RUNS.fetch_add(1, Ordering::SeqCst);
        Emit::none()
    }

    // Branches B and C are abandoned (external, no handler) → time out.
    macro_rules! abandoned_branch {
        ($req:ident, $rev:literal, $ok:ident, $okev:literal, $done:ident, $stepfn:ident) => {
            #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
            #[mire(event_type = $rev)]
            struct $req {
                id: String,
            }
            #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
            #[mire(event_type = $okev)]
            struct $ok;
            #[derive(Debug, Clone, Serialize, Deserialize)]
            struct $done;
            fn $stepfn(State(s): State<BookingState>) -> Emit {
                Emit::event($req { id: s.id.clone() })
            }
        };
    }
    abandoned_branch!(BReq, "ptd.b-req", BOk, "ptd.b-ok", BDone, step_b);
    abandoned_branch!(CReq, "ptd.c-req", COk, "ptd.c-ok", CDone, step_c);

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct Joined;
    fn join(
        Output(_a): Output<ADone>,
        Output(_b): Output<BDone>,
        Output(_c): Output<CDone>,
    ) -> Joined {
        Joined
    }

    struct DoomedParallelSaga;
    impl Saga for DoomedParallelSaga {
        type State = BookingState;
        fn name() -> &'static str {
            "pg-ptd-saga"
        }
        fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
            b.default_step_timeout(Duration::from_secs(300))
                .start_on::<Booking, _>(|e: &BookingRequested| {
                    Some(Start {
                        instance_id: e.id.clone(),
                        state: BookingState { id: e.id.clone() },
                    })
                })
                .step(
                    "a",
                    task(step_a)
                        .on_response(|_: AOk| ADone)
                        .compensate(comp(comp_a)),
                )
                .step("b", task(step_b).on_response(|_: BOk| BDone))
                .step("c", task(step_c).on_response(|_: COk| CDone))
                .step("join", task(join).after(["a", "b", "c"]).terminal_success())
                .effect(handle_a) // only A has a handler → A completes, B+C abandoned
        }
    }

    #[tokio::test]
    async fn timed_out_siblings_do_not_double_compensate() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        COMP_RUNS.store(0, Ordering::SeqCst);
        let topology = DoomedParallelSaga::build(SagaBuilder::<DoomedParallelSaga>::for_saga());
        let runner = Arc::new(PgSagaRunner::new(topology, store.clone()));
        let id = format!("ptd-{}", uuid::Uuid::new_v4());

        // Trigger: A completes inline; B and C suspend (abandoned).
        runner
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .unwrap();
        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.step_status("a"), SagaStepStatus::Completed);
        assert_eq!(env.step_status("b"), SagaStepStatus::Active);
        assert_eq!(env.step_status("c"), SagaStepStatus::Active);

        // Both B and C time out in the same tick.
        let after = chrono::Utc::now() + chrono::Duration::seconds(400);
        runner.tick_timeouts(after).await.unwrap();

        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
        assert_eq!(
            env.step_status("a"),
            SagaStepStatus::Compensated,
            "A must be compensated"
        );

        // THE ASSERTION: A's compensation ran exactly once, even though
        // two siblings (B and C) both timed out.
        assert_eq!(
            COMP_RUNS.load(Ordering::SeqCst),
            1,
            "completed branch A must be compensated exactly once, not once per timed-out sibling"
        );

        // No await leak.
        let leaked: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM es_saga_awaits \
             WHERE saga_name = 'pg-ptd-saga' AND instance_id = $1",
        )
        .bind(&id)
        .fetch_one(store.pool())
        .await
        .unwrap();
        assert_eq!(leaked, 0, "no await rows may leak");
    }
}
