//! Phase 5 (review SAGA-10/DX-8/SAGA-14): early-signal buffering.
//!
//! A `RoutesToStep` signal that arrives BEFORE its target step suspends is
//! persisted and consumed the moment the step suspends — the caller gets
//! `SignalOutcome::Buffered` (never a "not awaiting, retry" outcome — that
//! variant was removed once buffering made it unreachable, P5.4).
//!
//! To arrange "signal arrives before suspension" deterministically, the
//! saga has a `gate` step that suspends awaiting an external event; the
//! `approve` step (downstream) awaits a `RoutesToStep` signal. While the
//! saga is parked at `gate`, `approve` is still Pending — so a signal for
//! it lands early. Delivering `gate`'s event then activates `approve`,
//! which suspends and drains the buffered signal in the same turn.
//!
//! See `tasks/early-signal-buffering.md`. pg tests gated on `DATABASE_URL`.

#![allow(dead_code)]

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

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

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "esig.review-needed")]
struct ReviewNeeded {
    id: String,
}

impl Aggregate for Account {
    type Event = ReviewNeeded;
    fn stream_category() -> &'static str {
        "esig-account"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

// gate: suspends awaiting an external GateOpened (no in-process handler).
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "esig.gate-requested")]
struct GateRequested {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "esig.gate-opened")]
struct GateOpened;

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

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

// approve: awaits the operator's Decision signal (RoutesToStep).
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "esig.decision")]
struct Decision {
    approved: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct ApproveOut {
    approved: bool,
}

async fn approve_step(Output(_g): Output<GateOut>) -> Emit {
    Emit::none()
}

struct ReviewSaga;
impl Saga for ReviewSaga {
    type State = St;
    fn name() -> &'static str {
        "esig-review"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Account, _>(|e: &ReviewNeeded| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step("gate", task(gate_step).on_response(|_: GateOpened| GateOut))
        .step(
            "approve",
            task(approve_step)
                .after("gate")
                .terminal_success()
                .on_response(|d: Decision| ApproveOut {
                    approved: d.approved,
                }),
        )
        .signal::<Decision>()
        .routes_to_step("approve")
    }
}

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

// ---------------------------------------------------------------------
// pg runner
// ---------------------------------------------------------------------

#[tokio::test]
async fn pg_signal_before_suspend_is_buffered_then_consumed() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = PgSagaRunner::new(
        ReviewSaga::build(SagaBuilder::<ReviewSaga>::for_saga()),
        store,
    );
    let id = format!("esig-pg-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Account>(&ReviewNeeded { id: id.clone() })
        .await
        .expect("trigger");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.step_status("approve"), SagaStepStatus::Pending);

    let outcome = runner
        .signal(&id, Decision { approved: true })
        .await
        .expect("signal ok");
    assert_eq!(outcome, SignalOutcome::Buffered);

    runner
        .deliver(&id, &GateOpened)
        .await
        .expect("deliver gate");
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
    assert_eq!(env.step_status("approve"), SagaStepStatus::Completed);
    let out: ApproveOut =
        serde_json::from_value(env.outputs.get("approve").unwrap().clone()).unwrap();
    assert!(out.approved);
}

#[tokio::test]
async fn pg_buffered_signal_survives_restart_and_is_marked_consumed() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = format!("esig-pg-restart-{}", uuid::Uuid::new_v4());

    // Runner A: trigger + early signal, then dropped.
    {
        let runner_a = PgSagaRunner::new(
            ReviewSaga::build(SagaBuilder::<ReviewSaga>::for_saga()),
            store.clone(),
        );
        runner_a
            .trigger::<Account>(&ReviewNeeded { id: id.clone() })
            .await
            .expect("trigger");
        let outcome = runner_a
            .signal(&id, Decision { approved: true })
            .await
            .expect("signal ok");
        assert_eq!(outcome, SignalOutcome::Buffered);
    }

    // Pending signal row persisted (consumed_at IS NULL).
    let pending: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'esig-review' AND instance_id = $1 AND consumed_at IS NULL",
    )
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(pending, 1, "early signal must persist as pending");

    // Runner B (fresh): drive the gate; the persisted signal drains.
    let runner_b = PgSagaRunner::new(
        ReviewSaga::build(SagaBuilder::<ReviewSaga>::for_saga()),
        store.clone(),
    );
    runner_b
        .deliver(&id, &GateOpened)
        .await
        .expect("deliver gate");
    let env = runner_b.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);

    // The row is marked consumed, not deleted (audit; SAGA-14 retention).
    let consumed: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'esig-review' AND instance_id = $1 AND consumed_at IS NOT NULL",
    )
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(consumed, 1, "consumed signal must be marked, not deleted");
}

/// Edge case: a `RoutesToStep` signal sent to an ALREADY-TERMINAL instance.
/// The `TriggersCompensation` path rejects this with `SagaTerminal`; the
/// `RoutesToStep` path must not silently *buffer* it — that would leave an
/// unconsumable pending row (no step will ever await it) that also can't be
/// pruned (only consumed rows are) and would block any later signal of the
/// type with `DuplicateBufferedSignal`. A signal to a done saga is a caller
/// error; reject it loudly.
#[tokio::test]
async fn pg_signal_to_terminal_instance_is_rejected_not_buffered() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = PgSagaRunner::new(
        ReviewSaga::build(SagaBuilder::<ReviewSaga>::for_saga()),
        store.clone(),
    );
    let id = format!("esig-pg-terminal-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Account>(&ReviewNeeded { id: id.clone() })
        .await
        .expect("trigger");
    // Drive to terminal: open the gate, then approve via the decision signal.
    runner
        .deliver(&id, &GateOpened)
        .await
        .expect("deliver gate");
    runner
        .signal(&id, Decision { approved: true })
        .await
        .expect("decision");
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::SucceededTerminal),
    );

    // Now a late decision arrives for the finished saga. It must be rejected,
    // not buffered into a permanently-unconsumed row.
    let err = runner
        .signal(&id, Decision { approved: false })
        .await
        .expect_err("a signal to a terminal instance must be rejected");
    assert!(
        matches!(err, PgRunnerError::SagaTerminal { .. }),
        "expected SagaTerminal, got: {err}"
    );

    // And no stray pending row was left behind.
    let pending: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'esig-review' AND instance_id = $1 AND consumed_at IS NULL",
    )
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(pending, 0, "a rejected signal must not leave a pending row");
}

#[tokio::test]
async fn pg_duplicate_buffered_signal_is_rejected() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = PgSagaRunner::new(
        ReviewSaga::build(SagaBuilder::<ReviewSaga>::for_saga()),
        store,
    );
    let id = format!("esig-pg-dup-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Account>(&ReviewNeeded { id: id.clone() })
        .await
        .expect("trigger");

    let first = runner
        .signal(&id, Decision { approved: true })
        .await
        .expect("first ok");
    assert_eq!(first, SignalOutcome::Buffered);

    let err = runner
        .signal(&id, Decision { approved: false })
        .await
        .expect_err("duplicate early signal must be rejected");
    assert!(
        matches!(err, PgRunnerError::DuplicateBufferedSignal { .. }),
        "expected DuplicateBufferedSignal, got: {err}"
    );
}
