//! C4/SAGA-2 regression: a timeout must not run compensation for a saga that
//! a concurrent delivery actually completed.
//!
//! Setup: `hold` completes (its compensation `release` increments a counter),
//! then `confirm` suspends awaiting an external response with a deadline. We
//! race `deliver(ConfirmOk)` against `tick_timeouts(now-in-the-future)` so the
//! deadline is always expired. Exactly one wins via OCC. The invariant: when
//! the delivery wins (saga Succeeded), the timeout's compensation of `hold`
//! must NOT have run — `release` count is 0. Before the fix the cascade ran
//! before the commit, so a lost race still fired the release.
//!
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

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

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

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

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

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

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

// --- hold: completes inline, has a counting compensation ------------------

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "c4.hold-req")]
struct HoldReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "c4.hold-ok")]
struct HoldOk;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Held;

async fn hold_step(State(s): State<St>) -> Emit {
    Emit::event(HoldReq { id: s.id.clone() })
}
async fn handle_hold(Req(_): Req<HoldReq>) -> HoldOk {
    HoldOk
}
async fn release_hold(_: Output<Held>) -> Emit {
    RELEASES.fetch_add(1, Ordering::SeqCst);
    Emit::none()
}

// --- confirm: suspends awaiting an external response (Regime B) -----------

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "c4.confirm-req")]
struct ConfirmReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "c4.confirm-ok")]
struct ConfirmOk;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Confirmed;

async fn confirm_step(Output(_): Output<Held>) -> Emit {
    // No registered handler for ConfirmReq → suspends awaiting ConfirmOk.
    Emit::event(ConfirmReq { id: "x".into() }).timeout(Duration::from_millis(1))
}

struct C4Saga;
impl Saga for C4Saga {
    type State = St;
    fn name() -> &'static str {
        "c4-race-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Booking, _>(|e: &BookingRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "hold",
            task(hold_step)
                .on_response(|_: HoldOk| Held)
                .compensate(comp(release_hold)),
        )
        .step(
            "confirm",
            task(confirm_step)
                .after("hold")
                .terminal_success()
                .on_response(|_: ConfirmOk| Confirmed),
        )
        .handle(handle_hold)
    }
}

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

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn timeout_does_not_compensate_a_delivered_saga() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = C4Saga::build(SagaBuilder::<C4Saga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);

    let mut delivered = 0;
    let mut timed_out = 0;

    for _ in 0..40 {
        let id = format!("c4-{}", uuid::Uuid::new_v4());
        runner
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .expect("trigger");

        // hold completed during trigger; confirm is suspended with an
        // already-expired deadline. Reset the counter, then race.
        RELEASES.store(0, Ordering::SeqCst);
        let now_future = chrono::Utc::now() + chrono::Duration::seconds(3600);

        let (_d, _t) = tokio::join!(
            runner.deliver(&id, &ConfirmOk),
            runner.tick_timeouts(now_future),
        );

        let env = runner.load(&id).await.unwrap().unwrap();
        let releases = RELEASES.load(Ordering::SeqCst);
        match env.status {
            SagaInstanceStatus::SucceededTerminal => {
                delivered += 1;
                assert_eq!(
                    releases, 0,
                    "C4: the delivery won (saga succeeded) but the timeout still released the hold"
                );
            }
            SagaInstanceStatus::FailedTerminal => {
                timed_out += 1;
                assert_eq!(releases, 1, "timeout won → hold released exactly once");
            }
            other => panic!("unexpected terminal status {other:?} (releases={releases})"),
        }
    }

    // Sanity: the loop actually exercised the race (not all one-sided every
    // run). We don't require a specific split — just that the invariant held
    // across 40 races. Print the breakdown for visibility.
    eprintln!("c4 race: delivered={delivered}, timed_out={timed_out}");
    assert_eq!(delivered + timed_out, 40);
}
