//! effects-vs-commit §11.4: the per-instance processing lease guarantees a
//! saga instance's effects run **at-least-once but never concurrently**. We
//! fire many concurrent deliveries of the same response at one instance; the
//! downstream effect records the peak number of simultaneous executions, which
//! must never exceed 1. (Without the lease, the racing deliveries would all
//! drive the downstream effect at once and the peak would exceed 1.)
//!
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

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

static ACTIVE: AtomicUsize = AtomicUsize::new(0);
static PEAK: AtomicUsize = AtomicUsize::new(0);
static RUNS: AtomicUsize = AtomicUsize::new(0);

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

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

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

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

// `confirm`: suspends awaiting an external ConfirmOk (Regime B).
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "nc.confirm-req")]
struct ConfirmReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "nc.confirm-ok")]
struct ConfirmOk;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Confirmed;

fn confirm_step(State(s): State<St>) -> Emit {
    Emit::event(ConfirmReq { id: s.id.clone() })
}

// `finalize`: the downstream effect, activated when `confirm` completes. Its
// in-process handler is what must never run concurrently for one instance.
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "nc.finalize-req")]
struct FinalizeReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "nc.finalize-ok")]
struct FinalizeOk;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Finalized;

fn finalize_step(State(s): State<St>) -> Emit {
    Emit::event(FinalizeReq { id: s.id.clone() })
}

async fn handle_finalize(Req(_): Req<FinalizeReq>) -> FinalizeOk {
    RUNS.fetch_add(1, Ordering::SeqCst);
    let cur = ACTIVE.fetch_add(1, Ordering::SeqCst) + 1;
    // Record the running peak.
    let mut peak = PEAK.load(Ordering::SeqCst);
    while cur > peak {
        match PEAK.compare_exchange(peak, cur, Ordering::SeqCst, Ordering::SeqCst) {
            Ok(_) => break,
            Err(observed) => peak = observed,
        }
    }
    // Widen the window so a genuine concurrency bug is caught reliably.
    tokio::time::sleep(Duration::from_millis(25)).await;
    ACTIVE.fetch_sub(1, Ordering::SeqCst);
    FinalizeOk
}

struct NcSaga;
impl Saga for NcSaga {
    type State = St;
    fn name() -> &'static str {
        "nc-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(
            "confirm",
            task(confirm_step).on_response(|_: ConfirmOk| Confirmed),
        )
        .step(
            "finalize",
            task(finalize_step)
                .after("confirm")
                .terminal_success()
                .on_response(|_: FinalizeOk| Finalized),
        )
        .effect(handle_finalize)
    }
}

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 = 8)]
async fn concurrent_deliveries_never_run_the_effect_concurrently() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    PEAK.store(0, Ordering::SeqCst);
    ACTIVE.store(0, Ordering::SeqCst);
    RUNS.store(0, Ordering::SeqCst);

    let runner = Arc::new(PgSagaRunner::new(
        NcSaga::build(SagaBuilder::<NcSaga>::for_saga()),
        store,
    ));

    let id = format!("nc-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .expect("trigger");
    // `confirm` is now suspended awaiting ConfirmOk.

    // Fire 12 concurrent deliveries of the SAME response at the instance.
    let mut handles = Vec::new();
    for _ in 0..12 {
        let runner = runner.clone();
        let id = id.clone();
        handles.push(tokio::spawn(async move {
            runner.deliver(&id, &ConfirmOk).await
        }));
    }
    let mut consumed = 0;
    for h in handles {
        if h.await.unwrap().expect("deliver must not error") {
            consumed += 1;
        }
    }

    // Exactly one delivery consumed the response; the rest were no-ops.
    assert_eq!(consumed, 1, "exactly one deliver consumes the response");
    // The downstream effect ran (at-least-once) ...
    assert!(RUNS.load(Ordering::SeqCst) >= 1, "finalize effect must run");
    // ... but NEVER concurrently — the lease serialized the racing deliveries.
    assert_eq!(
        PEAK.load(Ordering::SeqCst),
        1,
        "the downstream effect must never run concurrently (peak was {})",
        PEAK.load(Ordering::SeqCst)
    );

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}
