//! P3.4 — `IdempotencyKey.attempt_seq` is derived from the committed
//! `StepActivated` (and `StepCompensating`) ordinal on the envelope, not a
//! hardcoded `0`. So a crash-recovery re-drive or OCC retry that replays the
//! same committed activations reproduces the same key — the documented dedupe
//! contract. The activation/compensation derivation is also the seam a future
//! step-retry feature bumps naturally.
//!
//! The unit derivation test needs no DB; the recovery test is gated on
//! `DATABASE_URL`.

#![allow(dead_code)]

use std::sync::Mutex;

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

mod common;

#[derive(Default, Clone)]
struct Trigger;
#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "as.requested")]
struct Requested {
    id: String,
}
impl Aggregate for Trigger {
    type Event = Requested;
    fn stream_category() -> &'static str {
        "as-trigger"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "as.work-req")]
struct WorkReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "as.work-ok")]
struct WorkOk;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct WorkDone;

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

async fn work(State(s): State<St>) -> Emit {
    Emit::event(WorkReq { id: s.id.clone() })
}
async fn handle_work(Req(_): Req<WorkReq>, key: IdempotencyKey) -> WorkOk {
    CAPTURED.lock().unwrap().push(key);
    WorkOk
}

struct AsSaga;
impl Saga for AsSaga {
    type State = St;
    fn name() -> &'static str {
        "attempt-seq"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Trigger, _>(|e: &Requested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "work",
            task(work)
                .on_response(|_: WorkOk| WorkDone)
                .terminal_success(),
        )
        .handle(handle_work)
    }
}

/// No DB: the ordinal is read off the committed-event count, and it bumps with
/// each re-activation/-compensation — proving it isn't a hardcoded constant.
#[test]
fn activation_and_compensation_ordinals_are_derived_and_bump() {
    let mut root = AggregateRoot::<SagaEnvelope<AsSaga>>::new("x");
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(St { id: "x".into() }).unwrap(),
        trigger: None,
    });
    // Never activated → ordinal 0 (and clamped, never negative).
    assert_eq!(root.state.activation_seq("work"), 0);

    root.record(SagaEnvelopeEvent::StepActivated {
        step: "work".into(),
    });
    assert_eq!(
        root.state.activation_seq("work"),
        0,
        "first activation → ordinal 0"
    );

    // A second activation (the future step-retry case) bumps the ordinal — the
    // key would differ, exactly so a retry isn't deduped against attempt 0.
    root.record(SagaEnvelopeEvent::StepActivated {
        step: "work".into(),
    });
    assert_eq!(
        root.state.activation_seq("work"),
        1,
        "re-activation → ordinal 1"
    );

    // Compensation ordinal is tracked independently.
    assert_eq!(root.state.compensation_seq("work"), 0);
    root.record(SagaEnvelopeEvent::StepCompensating {
        step: "work".into(),
    });
    assert_eq!(
        root.state.compensation_seq("work"),
        0,
        "first comp → ordinal 0"
    );
    root.record(SagaEnvelopeEvent::StepCompensating {
        step: "work".into(),
    });
    assert_eq!(
        root.state.compensation_seq("work"),
        1,
        "re-comp → ordinal 1"
    );
}

/// The normal trigger path hands the handler a key whose `attempt_seq` equals
/// the ordinal derivable from the committed envelope — i.e. it's read from the
/// log, not hardcoded.
#[tokio::test]
async fn handler_key_matches_the_committed_envelope_ordinal() {
    let Some(store) = common::store().await else {
        return;
    };
    // No clear(): tests run in parallel and share CAPTURED; each finds its own
    // key by unique instance_id, so clearing would race-wipe a sibling's push.
    let runner = PgSagaRunner::new(
        AsSaga::build(SagaBuilder::<AsSaga>::for_saga()),
        store.clone(),
    );
    let id = common::unique("asnorm");
    runner
        .trigger::<Trigger>(&Requested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    runner.deliver(&id, &WorkOk).await.unwrap();

    let key = CAPTURED
        .lock()
        .unwrap()
        .iter()
        .find(|k| k.instance_id == id)
        .cloned()
        .expect("handler ran");
    assert_eq!(key.step, "work");
    // The key the handler saw equals the ordinal on the committed log.
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(
        key.attempt_seq,
        env.activation_seq("work"),
        "the handler's key is derived from the committed StepActivated ordinal"
    );
    assert_eq!(key.attempt_seq, 0);
}

/// Recovery (P3.3) re-drives a crash-stranded instance and runs the step
/// handler — the key it hands over carries the *derived* ordinal (0 for this
/// single committed activation), proving recovery reuses the log-derived key
/// rather than inventing a fresh counter (P3.4 acceptance).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn recovery_redrive_reuses_the_derived_key() {
    let Some(store) = common::store().await else {
        return;
    };
    // No clear(): see the sibling test — find-by-instance_id isolates us.
    let id = common::unique("asrec");

    // Crash after `Started`, before any drive: only the Started event, no
    // activation, no await. (store.save bypasses the runner's commit, so no
    // await/projection rows — exactly the post-Started-commit-crash shape.)
    let mut root = AggregateRoot::<SagaEnvelope<AsSaga>>::new(&id);
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(St { id: id.clone() }).unwrap(),
        trigger: Some(serde_json::to_value(Requested { id: id.clone() }).unwrap()),
    });
    store.save(&mut root).await.unwrap();
    sqlx::query(
        "INSERT INTO es_saga_instances (saga_name, instance_id, status, shard_id, updated_at) \
         VALUES ($1, $2, 'running', 0, now() - interval '1 hour')",
    )
    .bind(AsSaga::name())
    .bind(&id)
    .execute(store.pool())
    .await
    .unwrap();

    let runner = PgSagaRunner::new(
        AsSaga::build(SagaBuilder::<AsSaga>::for_saga()),
        store.clone(),
    );
    runner.recover().await.unwrap();

    // Recovery activated `work` and ran its handler (in-process) → completed.
    let key = CAPTURED
        .lock()
        .unwrap()
        .iter()
        .find(|k| k.instance_id == id)
        .cloned()
        .expect("recovery should have re-driven the step handler");
    assert_eq!(key.step, "work");
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(
        key.attempt_seq,
        env.activation_seq("work"),
        "recovery's key carries the committed-log ordinal, not a fresh counter"
    );
    assert_eq!(key.attempt_seq, 0);
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}
