//! End-to-end / stress test for the `travel_booking_service` example.
//!
//! The saga orchestrates real (in-process) axum providers over HTTP: inline
//! effects (flight/hotel) call providers directly; payment is async (emit →
//! durable outbox → publisher POSTs the provider and delivers the result back);
//! a never-settling payment times out and the cascade releases the holds.
//! Providers fail every 4th request with a transient 503 — the client and the
//! outbox drain retry, and the idempotency key keeps every effect to exactly
//! once, even across a worker crash.
//!
//! The scenarios cover the spread of a real deployment: a flaky-provider fleet,
//! a concurrent live-worker pool (exactly-once via idempotency at
//! `num_shards == 1`), the sharded scale-out path (work partitioned, redundant
//! publishes → 0), single- and repeated-crash recovery, a pushed cancellation,
//! a refund-on-late-failure, a failed compensation → needs-intervention, and a
//! duplicate callback no-op.
//!
//! Every scenario runs on the **one** saga name, and `tick_timeouts`/
//! outbox-drain are saga-WIDE, so they're serialized via `SERIAL` (the
//! saga-test-isolation rule). Tests that spawn live workers must gracefully
//! cancel AND **join** them before releasing `SERIAL`: a worker torn down with
//! `abort()` can run one more saga-wide timeout tick after the lock is freed
//! and expire the next test's await — join closes that window.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

#[path = "../examples/travel_booking_service.rs"]
mod svc;

use std::time::Duration;

use mire::EventStore;
use mire_sagas::{
    EmitPublisher, OutboxMessage, PgSagaRunner, Saga, SagaBuilder, SagaInstanceStatus,
    SagaStepStatus,
};
use svc::{Booking, BookingRequested, Harness, TravelBooking};
use tokio_util::sync::CancellationToken;

// Every WORKER-spawning test runs the SAME travel-booking topology under its
// OWN saga name. Recovery / timeout / outbox scans are saga-WIDE, so a shared
// name lets one test's workers recover another's instances (against the wrong,
// torn-down backend) — cross-test contamination that surfaces as phantom
// double-charges, dead-backend compensation panics, and stuck bookings. A
// distinct name per test is the isolation (the saga-test-isolation rule;
// mirrors stress_pg_chaos's per-test sagas). Non-worker tests (pushed-cancel,
// failed-compensation, duplicate-callback, the publish-time TOCTOU) run no
// recovery loop, so they safely share the default `TravelBookingSaga` name.
macro_rules! test_saga {
    ($ty:ident, $name:literal) => {
        struct $ty;
        impl Saga for $ty {
            type State = TravelBooking;
            fn name() -> &'static str {
                $name
            }
            fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
                svc::build_travel_topology(b)
            }
        }
    };
}
test_saga!(FleetSaga, "travel-booking-service-fleet");
test_saga!(CrashSaga, "travel-booking-service-crash");
test_saga!(PoolSaga, "travel-booking-service-pool");
test_saga!(SharderSaga, "travel-booking-service-sharded");
test_saga!(StormSaga, "travel-booking-service-storm");
test_saga!(CarrierSaga, "travel-booking-service-carrier");
test_saga!(ChaosSaga, "travel-booking-service-chaos");
test_saga!(PropSaga, "travel-booking-service-prop");

async fn await_terminal<S: Saga>(
    runner: &PgSagaRunner<S>,
    id: &str,
    budget: Duration,
) -> Option<SagaInstanceStatus> {
    let start = std::time::Instant::now();
    loop {
        if let Ok(Some(env)) = runner.load(id).await
            && env.status.is_terminal()
        {
            return Some(env.status);
        }
        if start.elapsed() > budget {
            return None;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }
}

// `tick_timeouts` + the outbox drain are saga-WIDE, so two tests each running
// a fleet on this one saga name would cross-process each other's work. Run
// them serially (each uses unique ids and drains its outbox before finishing,
// so serial execution is fully isolated). The Mutex is runtime-agnostic, so it
// serializes across the per-test tokio runtimes.
static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

// Migrate EXACTLY ONCE per test binary (principle 10). `store()` runs before
// the SERIAL lock, so every test calls it concurrently at startup; migrating
// per-call races DDL (`ALTER`/`CREATE INDEX` take locks even on the no-op
// IF-NOT-EXISTS path) and trips `lock_timeout`. This guard runs the migrations
// once; later callers await the same result and skip the DDL.
static MIGRATED: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();

async fn store() -> Option<EventStore> {
    let url = std::env::var("DATABASE_URL")
        .ok()
        .filter(|s| !s.is_empty())?;
    // Size the pool to the fleet: these stress scenarios fire dozens of
    // concurrent triggers and run worker pools whose drains/delivers/heartbeats
    // all draw connections. sqlx's default max (10) is far too small — under the
    // load it saturates and operations block on the 30s acquire timeout, which
    // shows up as a booking "stuck" until well past its budget (it isn't stuck,
    // it's starved for a connection). A real deployment sizes its pool to its
    // worker count the same way.
    let pool = sqlx::postgres::PgPoolOptions::new()
        .max_connections(40)
        .connect(&url)
        .await
        .expect("connect");
    let store = EventStore::new(pool);
    MIGRATED
        .get_or_init(|| async {
            store.migrate().await.expect("mire migrate");
            mire_sagas::migrate(&store).await.expect("saga migrate");
        })
        .await;
    Some(store)
}

fn booking(id: &str) -> BookingRequested {
    BookingRequested {
        booking_id: id.to_string(),
        flight_id: "LH-100".into(),
        hotel_id: "HILTON-OSL".into(),
        amount_cents: 150_000,
    }
}

/// Diagnostic dump of everything that explains an instance's (lack of)
/// progress: its status projection, any live/stale lease (with fence), pending
/// awaits, and outbox rows. Used to tell a genuine recovery stall from mere
/// slowness when a booking misses its budget.
async fn dump_instance(store: &EventStore, saga: &str, id: &str) -> String {
    let pool = store.pool();
    let inst: Option<(String, i32, String)> = sqlx::query_as(
        "SELECT status, shard_id, updated_at::text FROM es_saga_instances \
         WHERE saga_name=$1 AND instance_id=$2",
    )
    .bind(saga)
    .bind(id)
    .fetch_optional(pool)
    .await
    .unwrap();
    let leases: Vec<(String, String, i64, bool)> = sqlx::query_as(
        "SELECT worker_id, leased_until::text, fence, (leased_until > now()) \
         FROM es_saga_leases WHERE saga_name=$1 AND instance_id=$2",
    )
    .bind(saga)
    .bind(id)
    .fetch_all(pool)
    .await
    .unwrap();
    let awaits: Vec<(String, String)> = sqlx::query_as(
        "SELECT step, event_type FROM es_saga_awaits WHERE saga_name=$1 AND instance_id=$2",
    )
    .bind(saga)
    .bind(id)
    .fetch_all(pool)
    .await
    .unwrap();
    let outbox: Vec<(String, String, Option<String>)> = sqlx::query_as(
        "SELECT step, event_type, published_at::text FROM es_saga_outbox \
         WHERE saga_name=$1 AND instance_id=$2",
    )
    .bind(saga)
    .bind(id)
    .fetch_all(pool)
    .await
    .unwrap();
    format!(
        "instance={inst:?}\n  leases(worker,until,fence,live)={leases:?}\n  awaits={awaits:?}\n  outbox(step,type,published)={outbox:?}"
    )
}

/// The real sanity case: a fleet that, against flaky providers and under
/// concurrency, books the healthy reservations exactly once, compensates a
/// doomed one, and never double-books.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn fleet_books_compensates_and_survives_flaky_providers_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    // fail_every = 4 → ~25% of provider requests get a transient 503; the next
    // request (the retry) is not a multiple of 4, so it succeeds —
    // deterministically transient, so the test never flakes on the flakiness.
    let h = Harness::<FleetSaga>::start_with(store, 4).await;

    const M: usize = 16;
    let run = uuid::Uuid::new_v4().simple().to_string();
    let healthy = format!("tbs-ok-{run}");
    let doomed = format!("doomed-{run}");
    let stress: Vec<String> = (0..M).map(|i| format!("tbs-stress-{run}-{i}")).collect();

    // Fire everything concurrently: the healthy booking, the doomed booking
    // (payment never settles), and M stress bookings.
    let mut evs = vec![booking(&healthy), booking(&doomed)];
    evs.extend(stress.iter().map(|id| booking(id)));
    let triggers = evs.iter().map(|ev| h.trigger.trigger::<Booking>(ev));
    for r in futures::future::join_all(triggers).await {
        r.unwrap();
    }

    // Healthy + stress all reach terminal success despite the flakiness.
    assert_eq!(
        h.await_terminal(&healthy, Duration::from_secs(30)).await,
        Some(SagaInstanceStatus::SucceededTerminal),
    );
    for id in &stress {
        assert_eq!(
            h.await_terminal(id, Duration::from_secs(45)).await,
            Some(SagaInstanceStatus::SucceededTerminal),
            "stress booking {id} should survive flaky providers"
        );
    }

    // The doomed booking times out on payment and compensates its holds.
    assert_eq!(
        h.await_terminal(&doomed, Duration::from_secs(20)).await,
        Some(SagaInstanceStatus::FailedTerminal),
    );
    let env = h.trigger.load(&doomed).await.unwrap().unwrap();
    assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    assert_eq!(env.step_status("hotel"), SagaStepStatus::Compensated);

    // Flakiness genuinely fired (retries happened)...
    assert!(
        h.backend.failures() > 0,
        "the run should have hit transient 503s"
    );

    // ...yet every provider effect happened EXACTLY ONCE. M+2 bookings placed
    // flight/hotel holds; only the M+1 non-doomed captured payment + ticketed;
    // the doomed booking's two holds were released.
    let (flight, hotel, payment, carrier) = h.backend.effects();
    let n = (M + 2) as u64;
    assert_eq!(
        (flight, hotel),
        (n, n),
        "every booking held flight+hotel once"
    );
    assert_eq!(
        payment,
        (M + 1) as u64,
        "every non-doomed booking captured once"
    );
    assert_eq!(
        carrier,
        (M + 1) as u64,
        "every non-doomed booking ticketed once"
    );
    assert_eq!(
        h.backend.releases(),
        (1, 1),
        "the doomed booking's holds were released"
    );

    // Re-triggering the healthy booking is a no-op.
    h.trigger
        .trigger::<Booking>(&booking(&healthy))
        .await
        .unwrap();
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert_eq!(
        h.backend.effects(),
        (n, n, (M + 1) as u64, (M + 1) as u64),
        "a re-trigger must not produce a second hold/charge/ticket"
    );

    h.shutdown().await;
}

/// Crash recovery: a worker is hard-aborted mid-flight, then a fresh worker
/// takes over and completes every booking — exactly once. The durable outbox
/// (not an in-memory queue) is what survives the crash: an un-marked payment
/// row is re-published by the new worker, which re-POSTs (the provider dedupes)
/// and re-delivers (idempotent), so no booking is lost or double-charged.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_crashed_worker_is_replaced_and_bookings_complete_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let trigger = svc::new_runner_sharded::<CrashSaga>(store.clone(), &providers, 1);

    const M: usize = 12;
    let run = uuid::Uuid::new_v4().simple().to_string();
    let ids: Vec<String> = (0..M).map(|i| format!("tbs-crash-{run}-{i}")).collect();

    // Trigger all bookings with NO worker running yet: flight+hotel run inline,
    // payment suspends and its request lands in the durable outbox.
    let evs: Vec<_> = ids.iter().map(|id| booking(id)).collect();
    for r in futures::future::join_all(evs.iter().map(|ev| trigger.trigger::<Booking>(ev))).await {
        r.unwrap();
    }

    // Worker #1 starts draining the outbox, then is HARD-ABORTED mid-flight
    // (a crash — not a graceful cancel), likely between provider call and the
    // outbox row being marked published.
    let c1 = CancellationToken::new();
    let w1 = svc::spawn_worker_sharded::<CrashSaga>(store.clone(), &providers, c1.clone(), 1);
    tokio::time::sleep(Duration::from_millis(120)).await;
    w1.abort();
    let _ = w1.await; // confirm the crashed worker is fully stopped (no lingering)

    // Worker #2 takes over and finishes everything from the durable outbox.
    let c2 = CancellationToken::new();
    let w2 = svc::spawn_worker_sharded::<CrashSaga>(store.clone(), &providers, c2.clone(), 1);

    for id in &ids {
        assert_eq!(
            // Phase 7: recovery is lease-expiry-driven (~1.5s TTL), so the one
            // instance the crash abandoned mid-step is reclaimed in a couple of
            // seconds, not tens. Generous budget for a loaded suite.
            await_terminal(&trigger, id, Duration::from_secs(45)).await,
            Some(SagaInstanceStatus::SucceededTerminal),
            "booking {id} should complete after the worker crash"
        );
    }
    // Exactly once despite the crash + takeover re-publish.
    assert_eq!(
        backend.effects(),
        (M as u64, M as u64, M as u64, M as u64),
        "no booking lost or double-charged across the crash"
    );

    c2.cancel();
    let _ = w2.await;
    server.abort();
}

/// Horizontal scale-out: a POOL of live workers all running at once against
/// the same store, providers, and saga name — the real production shape (you
/// run N replicas, not one). At the default `num_shards == 1` the outbox drain
/// is an unpartitioned `WHERE published_at IS NULL` scan with no
/// `FOR UPDATE SKIP LOCKED`, so multiple workers genuinely contend: two can
/// select the same unpublished capture row and both POST it. That is fine *by
/// design* — the idempotency key dedupes the provider call and the second
/// `deliver` is a no-op against the already-consumed await — so the redundant
/// publish costs an extra HTTP round-trip but never a second charge. This test
/// pins that guarantee: with a fleet of competing workers and flaky providers,
/// every booking still converges exactly once. (Shard to *eliminate* the
/// redundancy; rely on idempotency for *correctness* either way.)
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_pool_of_concurrent_live_workers_books_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend_with(4).await; // flaky too
    let providers = svc::Providers::new(&base);
    let trigger = svc::new_runner_sharded::<PoolSaga>(store.clone(), &providers, 1);

    // A pool of WORKERS live workers, all started up front and NEVER aborted —
    // unlike the chaos test (one worker, churned with gaps), here they all run
    // simultaneously and compete for every outbox row, timeout, and recovery.
    const WORKERS: usize = 4;
    const HEALTHY: usize = 20;
    const DOOMED: usize = 4;
    let run = uuid::Uuid::new_v4().simple().to_string();
    let healthy: Vec<String> = (0..HEALTHY)
        .map(|i| format!("tbs-pool-{run}-{i}"))
        .collect();
    let doomed: Vec<String> = (0..DOOMED)
        .map(|i| format!("doomed-pool-{run}-{i}"))
        .collect();

    let cancel = CancellationToken::new();
    let workers: Vec<_> = (0..WORKERS)
        .map(|_| {
            svc::spawn_worker_sharded::<PoolSaga>(store.clone(), &providers, cancel.clone(), 1)
        })
        .collect();

    // Fire every booking concurrently into the contending fleet.
    let all: Vec<String> = healthy.iter().chain(doomed.iter()).cloned().collect();
    let evs: Vec<_> = all.iter().map(|id| booking(id)).collect();
    for r in futures::future::join_all(evs.iter().map(|ev| trigger.trigger::<Booking>(ev))).await {
        r.unwrap();
    }

    for id in &healthy {
        assert_eq!(
            await_terminal(&trigger, id, Duration::from_secs(60)).await,
            Some(SagaInstanceStatus::SucceededTerminal),
            "healthy booking {id} must succeed under a competing worker pool"
        );
    }
    for id in &doomed {
        assert_eq!(
            await_terminal(&trigger, id, Duration::from_secs(60)).await,
            Some(SagaInstanceStatus::FailedTerminal),
            "doomed booking {id} must compensate under a competing worker pool"
        );
    }

    // The headline guarantee: even with WORKERS workers racing on every row,
    // every provider effect happened EXACTLY ONCE.
    let n = (HEALTHY + DOOMED) as u64;
    let (flight, hotel, payment, carrier) = backend.effects();
    assert_eq!(
        (flight, hotel),
        (n, n),
        "every booking held flight+hotel once"
    );
    assert_eq!(
        payment, HEALTHY as u64,
        "only healthy bookings charged, once each"
    );
    assert_eq!(
        carrier, HEALTHY as u64,
        "only healthy bookings ticketed, once each"
    );
    assert_eq!(
        backend.releases(),
        (DOOMED, DOOMED),
        "every doomed booking's holds released, once each"
    );

    // Make the at-least-once-delivery / exactly-once-effect distinction
    // visible: total POSTs the providers received vs the distinct effects they
    // performed. The surplus is redundant work — competing workers re-publishing
    // the same outbox row and the client retrying transient 503s — all absorbed
    // by idempotency. (Not asserted as a hard lower bound: the exact overlap is
    // timing-dependent; the point proven by the assertions above is that the
    // surplus never produced a second real effect.)
    let (rf, rh, rp, rc) = backend.requests();
    let (ef, eh, ep, ec) = backend.effects();
    println!(
        "pool of {WORKERS} workers — provider requests {:?} vs distinct effects {:?} \
         (surplus {} absorbed by idempotency; {} transient 503s retried)",
        (rf, rh, rp, rc),
        (ef, eh, ep, ec),
        (rf + rh + rp + rc).saturating_sub(ef + eh + ep + ec),
        backend.failures(),
    );

    // Gracefully stop AND join every worker before this test releases SERIAL.
    // `tick_timeouts` is saga-WIDE, so a worker that outlived the test by even
    // one poll could expire the *next* test's payment await (2s) before its own
    // worker captures it — a cross-test flake. `cancel` makes each worker exit
    // its run loop; awaiting the handle guarantees it's fully stopped here.
    cancel.cancel();
    for w in workers {
        let _ = w.await;
    }
    server.abort();
}

/// The sharded scale-out path — the fix for the redundancy the test above
/// documents. With `num_shards == N`, the trigger hashes each booking into a
/// shard and every worker owns a converging *subset* of shards (via shard
/// leases), draining only its own outbox rows. So the same competing pool that
/// re-published rows at `num_shards == 1` now partitions the work: once
/// ownership converges, each capture row is drained by exactly one worker, and
/// the redundant publishes go away — while exactly-once still holds.
///
/// Runs under its OWN saga name (`SharderSaga`), so its `num_shards == 4`
/// stamping can't trip another test's `num_shards == 1` `verify_shard_config`
/// (recovery/scan config is saga-wide) — full isolation, no drain-confirm dance.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_sharded_worker_pool_partitions_work_and_drops_redundant_publishes() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    // Non-flaky backend: with the providers reliable, any surplus of provider
    // requests over distinct effects is attributable purely to worker
    // contention — which is exactly what sharding is supposed to remove.
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);

    const SHARDS: u32 = 4;
    const WORKERS: usize = 4;
    const HEALTHY: usize = 24;
    let run = uuid::Uuid::new_v4().simple().to_string();
    let healthy: Vec<String> = (0..HEALTHY)
        .map(|i| format!("tbs-shard-{run}-{i}"))
        .collect();

    // All runners agree on num_shards: the trigger stamps shards, the workers
    // own subsets of them. Own saga name → fully isolated from the other tests.
    let trigger = svc::new_runner_sharded::<SharderSaga>(store.clone(), &providers, SHARDS);
    let cancel = CancellationToken::new();
    let workers: Vec<_> = (0..WORKERS)
        .map(|_| {
            svc::spawn_worker_sharded::<SharderSaga>(
                store.clone(),
                &providers,
                cancel.clone(),
                SHARDS,
            )
        })
        .collect();

    // Let shard ownership converge *before* triggering, so the steady-state
    // (one owner per shard) is what processes the bookings — not the transient
    // mid-rebalance window where a shard may briefly be scanned by two workers.
    // shard_lease_ttl is 2s; a few TTLs is plenty for 4 workers to settle.
    tokio::time::sleep(Duration::from_secs(3)).await;

    let evs: Vec<_> = healthy.iter().map(|id| booking(id)).collect();
    for r in futures::future::join_all(evs.iter().map(|ev| trigger.trigger::<Booking>(ev))).await {
        r.unwrap();
    }

    for id in &healthy {
        assert_eq!(
            await_terminal(&trigger, id, Duration::from_secs(60)).await,
            Some(SagaInstanceStatus::SucceededTerminal),
            "booking {id} must succeed under the sharded pool"
        );
    }

    // Exactly-once still holds (the non-negotiable guarantee).
    let n = HEALTHY as u64;
    let (flight, hotel, payment, carrier) = backend.effects();
    assert_eq!(
        (flight, hotel),
        (n, n),
        "every booking held flight+hotel once"
    );
    assert_eq!(payment, n, "every booking charged once");
    assert_eq!(carrier, n, "every booking ticketed once");

    // The payoff: with the work partitioned by shard, the payment outbox row
    // is drained by a single owner — so provider requests match effects with at
    // most a tiny slack for any row that straddled a lease renewal. Contrast the
    // `num_shards == 1` pool test, where 4 workers turned 20 charges into ~49
    // POSTs. (Slack = SHARDS: an upper bound on rows that could be double-drained
    // during a single ownership handoff; in practice this is 0 after convergence.)
    let (_, _, rp, _) = backend.requests();
    assert!(
        rp <= payment + SHARDS as u64,
        "sharding should drop redundant payment publishes to ~0; got {rp} POSTs for {payment} charges"
    );
    println!(
        "sharded pool ({SHARDS} shards, {WORKERS} workers) — payment POSTs {rp} vs charges {payment} \
         (redundant publishes: {})",
        rp.saturating_sub(payment),
    );

    // Gracefully stop AND join every worker before releasing SERIAL so none
    // outlives the test (its recovery scan is saga-wide).
    cancel.cancel();
    for w in workers {
        let _ = w.await;
    }
    server.abort();
}

/// The full chaos sanity case: flaky providers + concurrent bookings (mixed
/// healthy and doomed) + a worker pool that is repeatedly hard-aborted and
/// respawned mid-flight. No matter how the crashes interleave with the
/// flakiness, every booking must converge to its correct terminal exactly
/// once — healthy → succeeded (one charge, one ticket), doomed → compensated
/// (holds released, never charged).
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn chaos_flaky_providers_and_repeated_worker_crashes_converge_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend_with(4).await; // flaky
    let providers = svc::Providers::new(&base);
    // Own saga name → recovery can't touch other tests' instances.
    let trigger = svc::new_runner_sharded::<ChaosSaga>(store.clone(), &providers, 1);

    const HEALTHY: usize = 10;
    const DOOMED: usize = 3;
    let run = uuid::Uuid::new_v4().simple().to_string();
    let healthy: Vec<String> = (0..HEALTHY)
        .map(|i| format!("tbs-chaos-{run}-{i}"))
        .collect();
    let doomed: Vec<String> = (0..DOOMED)
        .map(|i| format!("doomed-chaos-{run}-{i}"))
        .collect();
    let all: Vec<String> = healthy.iter().chain(doomed.iter()).cloned().collect();

    // Fire every booking concurrently.
    let evs: Vec<_> = all.iter().map(|id| booking(id)).collect();
    for r in futures::future::join_all(evs.iter().map(|ev| trigger.trigger::<Booking>(ev))).await {
        r.unwrap();
    }

    // Chaos: churn a worker — spawn, run briefly, hard-abort, repeat. Gaps
    // (no worker) and the abort timing exercise crash-recovery + the durable
    // outbox under sustained churn. A tiny LCG keeps the timing varied but
    // deterministic.
    let mut seed: u64 = 0xC0FFEE ^ (run.len() as u64);
    let mut rng = || {
        seed = seed
            .wrapping_mul(6364136223846793005)
            .wrapping_add(1442695040888963407);
        seed >> 33
    };
    for _ in 0..10 {
        let c = CancellationToken::new();
        let w = svc::spawn_worker_sharded::<ChaosSaga>(store.clone(), &providers, c, 1);
        tokio::time::sleep(Duration::from_millis(40 + rng() % 110)).await;
        w.abort(); // crash
        let _ = w.await; // join the crashed worker so it can't run a saga-wide
        // recovery/timeout tick after SERIAL is freed and perturb the next test
        tokio::time::sleep(Duration::from_millis(rng() % 40)).await; // gap with no worker
    }

    // A stable worker finishes whatever the chaos left in flight.
    let c = CancellationToken::new();
    let w = svc::spawn_worker_sharded::<ChaosSaga>(store.clone(), &providers, c.clone(), 1);

    for id in &healthy {
        assert_eq!(
            await_terminal(&trigger, id, Duration::from_secs(60)).await,
            Some(SagaInstanceStatus::SucceededTerminal),
            "healthy booking {id} must converge to success through the chaos"
        );
    }
    for id in &doomed {
        assert_eq!(
            await_terminal(&trigger, id, Duration::from_secs(60)).await,
            Some(SagaInstanceStatus::FailedTerminal),
            "doomed booking {id} must compensate through the chaos"
        );
    }

    // Exactly once, despite flakiness + repeated crashes: every booking held
    // flight+hotel; only the healthy ones charged + ticketed; the doomed ones'
    // holds were released.
    let n = (HEALTHY + DOOMED) as u64;
    let (flight, hotel, payment, carrier) = backend.effects();
    assert_eq!(
        (flight, hotel),
        (n, n),
        "every booking held flight+hotel once"
    );
    assert_eq!(
        payment, HEALTHY as u64,
        "only healthy bookings charged, once each"
    );
    assert_eq!(
        carrier, HEALTHY as u64,
        "only healthy bookings ticketed, once each"
    );
    assert_eq!(
        backend.releases(),
        (DOOMED, DOOMED),
        "every doomed booking's holds released"
    );
    assert!(backend.failures() > 0, "flakiness should have fired");

    c.cancel();
    let _ = w.await;
    server.abort();
}

/// The push / event direction: an unsolicited `CancelBooking` arrives mid-
/// flight (flight+hotel held, payment suspended) and rolls the booking back.
/// No worker is needed — `signal` drives the compensation cascade synchronously
/// — so this is deterministic. The cancel, not a timeout, is what fails it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_pushed_cancellation_rolls_back_the_in_flight_booking() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let runner = svc::new_runner(store.clone(), &providers);

    let id = format!("tbs-cancel-{}", uuid::Uuid::new_v4().simple());
    // trigger drives flight+hotel inline and suspends on payment. With no
    // worker running, it stays parked there — a clean window to cancel.
    runner.trigger::<Booking>(&booking(&id)).await.unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.step_status("flight"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("hotel"), SagaStepStatus::Completed);
    assert_eq!(env.status, SagaInstanceStatus::Running);

    // Push the cancellation — the saga rolls back, releasing both holds.
    runner
        .signal(
            &id,
            svc::CancelBooking {
                reason: "customer changed plans".into(),
            },
        )
        .await
        .unwrap();

    // No worker is running, so nothing else could have failed it — the pushed
    // cancel is what rolled it back.
    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("hotel"), SagaStepStatus::Compensated);
    assert_eq!(
        backend.releases(),
        (1, 1),
        "both holds released by the cancellation"
    );
    // Never charged — payment was still suspended when the cancel arrived.
    assert_eq!(backend.effects().2, 0, "a cancelled booking is not charged");

    server.abort();
}

/// Concurrency guard: two operators hit cancel on the SAME instance at the same
/// instant. The per-instance lease serializes them, and the terminal re-check
/// under the lease (`pg_runner.rs:886` fast-path + a re-load before the effect)
/// lets EXACTLY ONE run the compensation cascade — the loser is rejected, never
/// a second cascade or corrupted state. No worker runs, so the two cancels are
/// the only actors and the outcome is purely about same-instance serialization.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_duplicate_cancels_compensate_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    // Two independent runner handles (separate "operators"), shared store.
    let r1 = svc::new_runner(store.clone(), &providers);
    let r2 = svc::new_runner(store.clone(), &providers);

    let id = format!("tbs-dualcancel-{}", uuid::Uuid::new_v4().simple());
    r1.trigger::<Booking>(&booking(&id)).await.unwrap();
    // flight+hotel held, payment suspended, no worker → parked, ripe for cancel.

    // Race two cancels on the one instance.
    let (a, b) = tokio::join!(
        r1.signal(
            &id,
            svc::CancelBooking {
                reason: "op A".into()
            }
        ),
        r2.signal(
            &id,
            svc::CancelBooking {
                reason: "op B".into()
            }
        ),
    );

    // Exactly one wins; the other is rejected (terminal) — never both, never a
    // panic. This is the whole point: same-instance effects don't run twice.
    let oks = [a.is_ok(), b.is_ok()].into_iter().filter(|ok| *ok).count();
    assert_eq!(oks, 1, "exactly one cancel must win: a={a:?}, b={b:?}");
    let loser = if a.is_ok() { &b } else { &a };
    assert!(
        matches!(
            loser,
            Err(mire_sagas::PgRunnerError::SagaTerminal { .. })
                | Err(mire_sagas::PgRunnerError::Compensating { .. })
        ),
        "the losing cancel must be cleanly rejected, got: {loser:?}"
    );

    let env = r1.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    assert_eq!(env.step_status("hotel"), SagaStepStatus::Compensated);
    assert_eq!(backend.releases(), (1, 1), "both holds released");
    assert_eq!(backend.effects().2, 0, "never charged");

    server.abort();
}

/// Compensating a COMPLETED async step: the carrier permanently rejects the
/// ticket *after* payment has settled, so the inline confirm effect fails and
/// the cascade must REFUND the captured payment (and release the holds). This
/// exercises rollback of a step that completed via the async deliver path.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_carrier_rejection_after_payment_refunds_and_releases() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let trigger = svc::new_runner_sharded::<CarrierSaga>(store.clone(), &providers, 1);
    let cancel = CancellationToken::new();
    let worker =
        svc::spawn_worker_sharded::<CarrierSaga>(store.clone(), &providers, cancel.clone(), 1);

    // "noticket-" → flight+hotel held, payment captured (async), then the
    // carrier rejects the ticket (422) → confirm fails → cascade.
    let id = format!("noticket-{}", uuid::Uuid::new_v4().simple());
    trigger.trigger::<Booking>(&booking(&id)).await.unwrap();

    assert_eq!(
        await_terminal(&trigger, &id, Duration::from_secs(30)).await,
        Some(SagaInstanceStatus::FailedTerminal),
        "a rejected ticket fails the booking"
    );
    let env = trigger.load(&id).await.unwrap().unwrap();
    assert_eq!(
        env.step_status("payment"),
        SagaStepStatus::Compensated,
        "payment refunded"
    );
    assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    assert_eq!(env.step_status("hotel"), SagaStepStatus::Compensated);

    // The payment was captured once and then refunded; both holds released.
    assert_eq!(
        backend.effects().2,
        1,
        "payment was captured before the failure"
    );
    assert_eq!(backend.refunds(), 1, "the captured payment was refunded");
    assert_eq!(backend.releases(), (1, 1), "both holds released");

    cancel.cancel();
    let _ = worker.await;
    server.abort();
}

/// When a compensation ITSELF fails, the saga must not silently swallow it: a
/// release that can't complete (a permanent 422) becomes a durable lock and
/// the instance ends `FailedTerminalNeedsIntervention` — an operator must
/// resolve it. The rest of the cascade still runs (hotel is released).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_failed_compensation_ends_needs_intervention() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let runner = svc::new_runner(store.clone(), &providers);

    // "stuckhold-" → the flight hold can't be released. Trigger (flight+hotel
    // held, payment suspended, no worker), then cancel → cascade.
    let id = format!("stuckhold-{}", uuid::Uuid::new_v4().simple());
    runner.trigger::<Booking>(&booking(&id)).await.unwrap();
    runner
        .signal(
            &id,
            svc::CancelBooking {
                reason: "rollback test".into(),
            },
        )
        .await
        .unwrap();

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(
        env.status,
        SagaInstanceStatus::FailedTerminalNeedsIntervention,
        "a failed compensation must surface as needs-intervention"
    );
    assert!(env.needs_intervention());
    assert_eq!(
        env.step_status("flight"),
        SagaStepStatus::CompensationFailed,
        "the flight release failed → a durable lock"
    );
    assert_eq!(
        env.step_status("hotel"),
        SagaStepStatus::Compensated,
        "the rest of the cascade still ran"
    );
    // Flight hold was NOT released (the release 422'd); hotel was.
    assert_eq!(backend.releases(), (0, 1));

    server.abort();
}

/// Duplicate / out-of-order callbacks are the norm with real webhooks: a
/// provider may deliver the same `PaymentCaptured` twice. The second delivery
/// must be a safe no-op — the booking confirms once and the carrier is
/// ticketed once. (We deliver by hand to simulate the provider's retried
/// callback; no worker needed.)
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_duplicate_payment_callback_is_a_safe_no_op() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let runner = svc::new_runner(store.clone(), &providers);

    // trigger drives flight+hotel inline and suspends on payment.
    let id = format!("tbs-dup-{}", uuid::Uuid::new_v4().simple());
    runner.trigger::<Booking>(&booking(&id)).await.unwrap();

    // First callback resumes payment → confirm runs inline → terminal success.
    let captured = svc::PaymentCaptured {
        payment_id: "pay-dup-1".into(),
    };
    let first = runner.deliver(&id, &captured).await.unwrap();
    assert!(
        first,
        "the first callback resumes the suspended payment step"
    );
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::SucceededTerminal),
    );

    // Duplicate callback: the await is already consumed → no-op, no re-confirm.
    let second = runner.deliver(&id, &captured).await.unwrap();
    assert!(!second, "a duplicate callback is a no-op");
    assert_eq!(
        backend.effects().3,
        1,
        "the carrier was ticketed exactly once despite the duplicate callback"
    );

    server.abort();
}

/// The CONCURRENT variant of the duplicate-callback case: real webhooks deliver
/// the same event twice *at once*, not neatly one-after-another. Two deliveries
/// of the same `PaymentCaptured` race on the one instance — the per-instance
/// lease serializes them, so EXACTLY ONE consumes the await (resumes payment →
/// confirm) and the other finds it already gone (`Ok(false)`). The carrier is
/// ticketed exactly once; no double-resume, no double-confirm. Completes the
/// same-instance concurrency triad: concurrent triggers (storm), concurrent
/// signals (cancel race), concurrent deliveries (here).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_duplicate_deliveries_resume_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let r1 = svc::new_runner(store.clone(), &providers);
    let r2 = svc::new_runner(store.clone(), &providers);

    // trigger drives flight+hotel inline and suspends on payment (no worker).
    let id = format!("tbs-dupdeliver-{}", uuid::Uuid::new_v4().simple());
    r1.trigger::<Booking>(&booking(&id)).await.unwrap();

    // Race two identical callbacks on the one instance.
    let captured = svc::PaymentCaptured {
        payment_id: "pay-concurrent".into(),
    };
    let (a, b) = tokio::join!(r1.deliver(&id, &captured), r2.deliver(&id, &captured));
    let (a, b) = (a.unwrap(), b.unwrap());
    assert_eq!(
        [a, b].into_iter().filter(|x| *x).count(),
        1,
        "exactly one concurrent delivery resumes the step: a={a}, b={b}"
    );

    assert_eq!(
        r1.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::SucceededTerminal),
    );
    // confirm ran once → carrier ticketed exactly once, despite the race.
    assert_eq!(
        backend.effects().3,
        1,
        "carrier ticketed exactly once despite concurrent duplicate delivery"
    );

    server.abort();
}

/// Forward progress racing rollback on the SAME instance: a `PaymentCaptured`
/// delivery (which would complete payment → confirm → success) races a
/// `CancelBooking` signal (which would compensate → failure). The per-instance
/// lease serializes them, so the instance lands in ONE consistent terminal —
/// never a half-applied mix, never a double effect. Whichever wins:
///   deliver wins  → SucceededTerminal, carrier ticketed once, no release;
///   cancel  wins  → FailedTerminal,   not ticketed,          both holds released;
/// and the loser is a clean no-op / rejection. (Payment is delivered directly,
/// so the backend's capture count stays 0 either way.)
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn deliver_racing_cancel_lands_in_one_consistent_terminal() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let r1 = svc::new_runner(store.clone(), &providers);
    let r2 = svc::new_runner(store.clone(), &providers);

    let id = format!("tbs-delivervscancel-{}", uuid::Uuid::new_v4().simple());
    r1.trigger::<Booking>(&booking(&id)).await.unwrap();
    // flight+hotel held, payment suspended, no worker.

    // Race forward delivery against rollback signal on the one instance.
    let captured = svc::PaymentCaptured {
        payment_id: "pay-race".into(),
    };
    let (_d, _s) = tokio::join!(
        r1.deliver(&id, &captured),
        r2.signal(
            &id,
            svc::CancelBooking {
                reason: "race".into()
            }
        ),
    );

    let env = r1.load(&id).await.unwrap().unwrap();
    assert!(env.status.is_terminal(), "must reach a terminal state");
    let (_, _, payment, carrier) = backend.effects();
    assert_eq!(
        payment, 0,
        "payment delivered directly, never captured via backend"
    );
    match env.status {
        SagaInstanceStatus::SucceededTerminal => {
            // Delivery won: confirmed once, never rolled back.
            assert_eq!(carrier, 1, "deliver-wins → ticketed exactly once");
            assert_eq!(env.step_status("confirm"), SagaStepStatus::Completed);
            assert_eq!(backend.releases(), (0, 0), "deliver-wins → no compensation");
        }
        SagaInstanceStatus::FailedTerminal => {
            // Cancel won: rolled back, never ticketed.
            assert_eq!(carrier, 0, "cancel-wins → never ticketed");
            assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
            assert_eq!(env.step_status("hotel"), SagaStepStatus::Compensated);
            assert_eq!(
                backend.releases(),
                (1, 1),
                "cancel-wins → both holds released once"
            );
        }
        other => panic!("unexpected terminal status under the race: {other:?}"),
    }

    server.abort();
}

/// The publish-time TOCTOU, deterministically. The dangerous window is a
/// pushed `CancelBooking` that lands *while the payment capture is in flight* —
/// after the publisher's pre-charge terminal check passed, before the money
/// settles. The booking rolls back (flight+hotel compensated), but the payment
/// step's await was never consumed, so its `refund_payment` compensation can't
/// run — naively that orphans the charge. The hardened publisher re-checks
/// after the charge settles and refunds it.
///
/// We drive this by hand for determinism: a `slowcap-` booking holds the charge
/// for 300ms, and we race `publish()` against a cancel fired at 100ms (so it's
/// reliably inside that window) via `tokio::join!` — no worker, no timeouts, no
/// guessing.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_cancel_during_payment_publish_refunds_the_orphan_charge() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);
    let runner = svc::new_runner(store.clone(), &providers);

    // trigger drives flight+hotel inline and suspends on payment. No worker runs,
    // so the booking parks there until we act — a clean, race-free setup.
    let id = format!("slowcap-{}", uuid::Uuid::new_v4().simple());
    runner.trigger::<Booking>(&booking(&id)).await.unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.step_status("flight"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("hotel"), SagaStepStatus::Completed);
    assert_eq!(env.status, SagaInstanceStatus::Running);

    // Build the publisher and the payment outbox message by hand — exactly what
    // the worker's outbox drain would hand to `publish`.
    let publisher = svc::HttpPublisher::new(
        providers.clone(),
        std::sync::Arc::new(svc::new_runner(store.clone(), &providers)),
    );
    let payload = serde_json::to_value(svc::RequestPaymentCapture {
        booking_id: id.clone(),
        amount_cents: 150_000,
    })
    .unwrap();
    let msg = OutboxMessage {
        saga_name: "travel-booking-service",
        instance_id: &id,
        step: "payment",
        event_type: "tbs.payment.capture-request",
        payload: &payload,
    };

    // Race: publish (charge held 300ms) vs a cancel fired at 100ms — landing
    // after the pre-charge check, before the charge settles.
    let publish_fut = publisher.publish(msg);
    let cancel_fut = async {
        tokio::time::sleep(Duration::from_millis(100)).await;
        runner
            .signal(
                &id,
                svc::CancelBooking {
                    reason: "customer cancelled mid-capture".into(),
                },
            )
            .await
    };
    let (pub_res, sig_res) = tokio::join!(publish_fut, cancel_fut);
    pub_res.expect("publish should succeed (and refund the orphan)");
    sig_res.expect("signal should succeed");

    // The cancel rolled the booking back.
    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("hotel"), SagaStepStatus::Compensated);
    assert_eq!(backend.releases(), (1, 1), "both holds released");

    // The charge settled mid-cancel — but was NOT orphaned: the publisher saw
    // the booking had gone terminal and refunded it. Captured once, refunded
    // once → the customer is made whole.
    assert_eq!(
        backend.effects().2,
        1,
        "the capture settled while cancelling"
    );
    assert_eq!(backend.refunds(), 1, "the orphan charge was refunded");

    server.abort();
}

/// At-least-once domain-event delivery means the SAME booking event can reach
/// several runner replicas at once — each calls `trigger` with the same id.
/// The start path fences this with OCC (`NoStream` on the instance stream,
/// C3/SAGA-4): exactly one trigger reserves the instance and runs the start
/// effects; the rest conflict and return the existing id WITHOUT running a
/// step. This pins that claim under true concurrency (the fleet test only does
/// a *sequential* re-trigger): a storm of simultaneous duplicate triggers must
/// still book each provider exactly once.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_storm_of_concurrent_duplicate_triggers_books_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    let (base, backend, server) = svc::spawn_backend().await;
    let providers = svc::Providers::new(&base);

    // One worker drives the async payment to completion (graceful-join on teardown).
    let cancel = CancellationToken::new();
    let worker =
        svc::spawn_worker_sharded::<StormSaga>(store.clone(), &providers, cancel.clone(), 1);

    const REPLICAS: usize = 8;
    let id = format!("tbs-dup-{}", uuid::Uuid::new_v4().simple());

    // Separate runner instances stand in for separate replicas all receiving
    // the same domain event. Spawn each `trigger` as its own task so they race
    // in PARALLEL on the multi-thread runtime (not cooperatively interleaved) —
    // maximizing the chance multiple pass the fast-path load check and collide
    // at the OCC reserve, which is the fence actually under test.
    let handles: Vec<_> = (0..REPLICAS)
        .map(|_| {
            let r = svc::new_runner_sharded::<StormSaga>(store.clone(), &providers, 1);
            let ev = booking(&id);
            tokio::spawn(async move { r.trigger::<Booking>(&ev).await })
        })
        .collect();

    // Every replica observes the same instance id — winner via reserve, losers
    // via the idempotent conflict path. None errors or panics.
    for h in futures::future::join_all(handles).await {
        assert_eq!(
            h.expect("trigger task should not panic").unwrap(),
            Some(id.clone()),
            "every duplicate trigger resolves to the one instance"
        );
    }

    let observer = svc::new_runner_sharded::<StormSaga>(store.clone(), &providers, 1);
    assert_eq!(
        await_terminal(&observer, &id, Duration::from_secs(30)).await,
        Some(SagaInstanceStatus::SucceededTerminal),
    );

    // The OCC start-fence let exactly one driver run the start effects, so
    // despite REPLICAS concurrent triggers each provider effect happened once.
    assert_eq!(
        backend.effects(),
        (1, 1, 1, 1),
        "a duplicate-trigger storm books flight/hotel/payment/carrier exactly once"
    );

    cancel.cancel();
    let _ = worker.await;
    server.abort();
}

// =====================================================================
// Property / fuzz soak: a randomized high-volume workload under worker-pool
// chaos must always converge to the same per-booking terminal with
// exactly-once effects — for ANY mix, ANY flakiness, ANY crash schedule.
//
// This is "property-based" in the sense that matters here: a randomized input
// (the workload + chaos schedule, from a seeded LCG — no proptest dependency,
// matching the stress_pg_chaos pattern) checked against invariants that hold
// regardless of timing. A failure prints the seed, which replays the exact
// case deterministically. The invariants are timing-independent precisely
// because each booking KIND has a fixed terminal:
//   healthy  → succeeded: charged once, ticketed once
//   doomed   → failed:    never charged, both holds released
//   noticket → failed:    charged once then refunded, both holds released
// so the global effect totals are a pure function of the kind counts — the
// chaos must not perturb them. This is the strongest exactly-once stress in
// the suite: it composes flaky providers + crash recovery (phase-7 lease-
// expiry) + compensation + refund under a fuzzed, churning fleet.
// =====================================================================

/// A tiny SplitMix64-ish LCG — deterministic per seed, no RNG dependency.
fn lcg(state: &mut u64) -> u64 {
    *state = state
        .wrapping_mul(6364136223846793005)
        .wrapping_add(1442695040888963407);
    *state >> 33
}

async fn run_property_case(store: &EventStore, seed: u64) {
    let mut s = seed;
    // Random flakiness: 0 (reliable), or every 4th/7th request a transient 503.
    let fail_every = [0u64, 4, 7][(lcg(&mut s) % 3) as usize];
    let (base, backend, server) = svc::spawn_backend_with(fail_every).await;
    let providers = svc::Providers::new(&base);
    // Own saga name → this fuzz can't recover (or be recovered by) other tests.
    let trigger = svc::new_runner_sharded::<PropSaga>(store.clone(), &providers, 1);

    // The seed fixes the workload SHAPE (counts, kinds, chaos schedule) so a
    // failure reproduces; a per-run uuid keeps the durable instance ids unique
    // so a rerun doesn't collide with a prior run's completed sagas (which
    // would make `trigger` an idempotent no-op and the fresh backend see no
    // effects).
    let run = format!("{seed:x}-{}", uuid::Uuid::new_v4().simple());
    // 20..36 bookings, each randomly one of the three deterministic-terminal
    // kinds (weighted toward healthy).
    let n = 20 + (lcg(&mut s) % 16) as usize;
    let (mut healthy, mut doomed, mut noticket) = (Vec::new(), Vec::new(), Vec::new());
    for i in 0..n {
        match lcg(&mut s) % 5 {
            0 => doomed.push(format!("doomed-prop-{run}-{i}")),
            1 => noticket.push(format!("noticket-prop-{run}-{i}")),
            _ => healthy.push(format!("tbs-prop-{run}-{i}")),
        }
    }
    let all: Vec<String> = healthy
        .iter()
        .chain(&doomed)
        .chain(&noticket)
        .cloned()
        .collect();

    // Fire everything concurrently with NO worker yet: flight+hotel run inline,
    // payment lands in the durable outbox — exercising crash recovery from the
    // first instant. Concurrent (not sequential): the payment step has a 2s
    // timeout, and triggering 30 bookings one-at-a-time would age the first
    // ones' awaits past that deadline before any worker spawns, flipping a
    // healthy booking to a (correct, but unwanted-here) timeout-compensation.
    let evs: Vec<_> = all.iter().map(|id| booking(id)).collect();
    for r in futures::future::join_all(evs.iter().map(|ev| trigger.trigger::<Booking>(ev))).await {
        r.unwrap();
    }

    // Chaos: churn a worker — spawn, run briefly, hard-abort, gap — so crash
    // recovery (phase-7 lease-expiry steal) and the durable outbox carry the
    // workload across repeated worker death.
    let rounds = 4 + lcg(&mut s) % 6;
    for _ in 0..rounds {
        let c = CancellationToken::new();
        let w = svc::spawn_worker_sharded::<PropSaga>(store.clone(), &providers, c, 1);
        tokio::time::sleep(Duration::from_millis(30 + lcg(&mut s) % 130)).await;
        w.abort(); // crash
        let _ = w.await; // join the crashed worker so it can't run a saga-wide
        // recovery/timeout tick after SERIAL is freed and perturb the next test
        tokio::time::sleep(Duration::from_millis(lcg(&mut s) % 40)).await; // worker-less gap
    }
    // A stable worker finishes whatever the chaos left in flight.
    let cancel = CancellationToken::new();
    let worker =
        svc::spawn_worker_sharded::<PropSaga>(store.clone(), &providers, cancel.clone(), 1);

    // Every booking converges to its KIND's terminal, regardless of the chaos.
    // On a miss, dump the instance's row state (status / lease / await / outbox)
    // into the panic so a failure is diagnosable without re-instrumenting.
    for id in &healthy {
        let status = await_terminal(&trigger, id, Duration::from_secs(90)).await;
        if status != Some(SagaInstanceStatus::SucceededTerminal) {
            panic!(
                "seed {seed:x}: healthy {id} must succeed, got {status:?}\n  {}",
                dump_instance(store, PropSaga::name(), id).await
            );
        }
    }
    for id in doomed.iter().chain(&noticket) {
        let status = await_terminal(&trigger, id, Duration::from_secs(90)).await;
        if status != Some(SagaInstanceStatus::FailedTerminal) {
            panic!(
                "seed {seed:x}: {id} must fail+compensate, got {status:?}\n  {}",
                dump_instance(store, PropSaga::name(), id).await
            );
        }
    }

    // The invariants: effect totals are a pure function of the kind counts.
    let (h, d, k) = (
        healthy.len() as u64,
        doomed.len() as u64,
        noticket.len() as u64,
    );
    let total = h + d + k;
    let (flight, hotel, payment, carrier) = backend.effects();
    assert_eq!(
        (flight, hotel),
        (total, total),
        "seed {seed:x}: every booking held flight+hotel exactly once"
    );
    assert_eq!(
        payment,
        h + k,
        "seed {seed:x}: healthy + noticket charged exactly once each"
    );
    assert_eq!(
        carrier, h,
        "seed {seed:x}: only healthy ticketed, once each"
    );
    assert_eq!(
        backend.refunds(),
        k as usize,
        "seed {seed:x}: every noticket charge was refunded"
    );
    assert_eq!(
        backend.releases(),
        ((d + k) as usize, (d + k) as usize),
        "seed {seed:x}: doomed + noticket released both holds"
    );

    cancel.cancel();
    let _ = worker.await;
    server.abort();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn property_random_workload_survives_chaos_exactly_once() {
    let Some(store) = store().await else {
        return;
    };
    let _serial = SERIAL.lock().await;
    // A couple of distinct seeds — each a different workload + chaos schedule.
    // Bump the count (or add seeds) locally to soak harder; each is fully
    // reproducible. Like all the stress scenarios this assumes a FRESH DB
    // (principle 10) — repeated runs without `docker compose down -v` bloat the
    // shared churn tables and autovacuum can transiently blow the time budget.
    for seed in [0xA11CEu64, 0xB0BB1E] {
        run_property_case(&store, seed).await;
    }
}
