//! SAGA-4 / TEST-9: concurrent duplicate triggers for the same instance must
//! start it once and run the start-step effect exactly once — not once per
//! racing trigger. The reserve-then-drive claim (commit `Started` under
//! `NoStream` OCC before driving) is what guarantees it: the loser conflicts
//! on the reservation and runs no effects.
//!
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

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

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

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

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

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

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

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "ct.charge-req")]
struct ChargeReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "ct.charge-ok")]
struct ChargeOk;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Charged;

async fn charge_step(State(s): State<St>) -> Emit {
    Emit::event(ChargeReq { id: s.id.clone() })
}

/// The start-step effect — counts how many times it actually runs.
async fn handle_charge(Req(_): Req<ChargeReq>) -> ChargeOk {
    CHARGES.fetch_add(1, Ordering::SeqCst);
    ChargeOk
}

struct CtSaga;
impl Saga for CtSaga {
    type State = St;
    fn name() -> &'static str {
        "ct-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(
            "charge",
            task(charge_step)
                .terminal_success()
                .on_response(|_: ChargeOk| Charged),
        )
        .handle(handle_charge)
    }
}

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 concurrent_duplicate_triggers_run_the_start_effect_once() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = CtSaga::build(SagaBuilder::<CtSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);

    // Race many duplicate-trigger pairs; assert the effect ran once each time.
    for _ in 0..30 {
        CHARGES.store(0, Ordering::SeqCst);
        let id = format!("ct-{}", uuid::Uuid::new_v4());
        let ev = BookingRequested { id: id.clone() };

        let (a, b) = tokio::join!(
            runner.trigger::<Booking>(&ev),
            runner.trigger::<Booking>(&ev),
        );
        // Both calls succeed; both return the same instance id.
        assert_eq!(a.unwrap(), Some(id.clone()));
        assert_eq!(b.unwrap(), Some(id.clone()));

        assert_eq!(
            CHARGES.load(Ordering::SeqCst),
            1,
            "the start-step effect must run exactly once across two racing triggers"
        );

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