//! The `IdempotencyKey` extractor (review D1/D2): a request handler can pull a
//! stable `(instance_id, step, attempt_seq)` key to dedupe non-replayable
//! effects. This test proves the durable runner populates it with the right
//! instance id and the *source step* that emitted the request.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

use std::sync::Mutex;

use mire::{Aggregate, EventData};
use mire_sagas::{Emit, IdempotencyKey, PgSagaRunner, Req, Saga, SagaBuilder, Start, State, task};
use serde::{Deserialize, Serialize};

mod common;

static CAPTURED: Mutex<Option<IdempotencyKey>> = Mutex::new(None);

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

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "idem.source")]
struct SourceEvent {
    id: String,
}

impl Aggregate for Source {
    type Event = SourceEvent;
    fn stream_category() -> &'static str {
        "idem-source"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

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

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "idem.charge-ok")]
struct ChargeOk;

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Charged;

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

/// The effect handler — captures the idempotency key it was handed.
async fn handle_charge(Req(_): Req<ChargeReq>, key: IdempotencyKey) -> ChargeOk {
    *CAPTURED.lock().unwrap() = Some(key);
    ChargeOk
}

struct IdemSaga;
impl Saga for IdemSaga {
    type State = St;
    fn name() -> &'static str {
        "idem-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "charge",
            task(charge_step)
                .terminal_success()
                .on_response(|_: ChargeOk| Charged),
        )
        .effect(handle_charge)
    }
}

#[tokio::test]
async fn request_handler_receives_a_stable_idempotency_key() {
    let Some(store) = common::store().await else {
        return;
    };
    *CAPTURED.lock().unwrap() = None;
    let topology = IdemSaga::build(SagaBuilder::<IdemSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let instance = common::unique("inst-42");
    let id = runner
        .trigger::<Source>(&SourceEvent {
            id: instance.clone(),
        })
        .await
        .unwrap()
        .unwrap();

    let key = CAPTURED
        .lock()
        .unwrap()
        .clone()
        .expect("handler should have run and captured the key");

    assert_eq!(key.instance_id, id);
    assert_eq!(key.instance_id, instance);
    assert_eq!(
        key.step, "charge",
        "key carries the source step that emitted the request"
    );
    assert_eq!(key.attempt_seq, 0);
    assert_eq!(key.token(), format!("{instance}/charge/0"));
}
