//! Await-routing guarantees (review SAGA-8 / C7).
//!
//! 1. **Terminal instances reject deliveries.** The runner purges all
//!    await rows when an instance goes terminal, so a late `deliver` is a
//!    no-op — a stale await on a Confirmed/Cancelled saga must not
//!    complete a step post-terminally.
//! 2. **Ambiguous awaits are a config error.** Routing is by event-type
//!    string. If two steps of one instance await the same event type, the
//!    runner cannot know which step the response belongs to — an
//!    unordered `LIMIT 1` would silently feed one step's response into
//!    the other. The runner refuses loudly. (Correlation keys are the
//!    long-term fix.)
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

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

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

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "route.order-placed")]
struct OrderPlaced {
    id: String,
}

impl Aggregate for Order {
    type Event = OrderPlaced;
    fn stream_category() -> &'static str {
        "route-order"
    }
    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 = "route.work-requested")]
struct WorkRequested {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "route.work-done")]
struct WorkDone {
    note: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct WorkOutput {
    note: String,
}

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

// ---------------------------------------------------------------------
// Terminal-rejection saga: `ext` suspends awaiting an external WorkDone;
// `finish` (parallel, no deps) completes immediately with terminal
// success. After trigger the saga is Confirmed while `ext`'s await is
// stale.
// ---------------------------------------------------------------------

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

async fn finish_step() -> Noted {
    Noted
}

struct TerminalRace;
impl Saga for TerminalRace {
    type State = St;
    fn name() -> &'static str {
        "route-terminal-race"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Order, _>(|e: &OrderPlaced| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "ext",
            task(ext_step).on_response(|d: WorkDone| WorkOutput { note: d.note }),
        )
        .step("finish", task(finish_step).terminal_success())
    }
}

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

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

    let consumed = runner
        .deliver(
            &id,
            &WorkDone {
                note: "late".into(),
            },
        )
        .await
        .expect("deliver must not error");
    assert!(!consumed, "terminal instance must not consume deliveries");
}

// ---------------------------------------------------------------------
// Ambiguity saga: two parallel steps both await the SAME event type.
// Routing by event-type string cannot distinguish them.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Serialize, Deserialize)]
struct OutA {
    note: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct OutB {
    note: String,
}

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

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

struct Ambiguous;
impl Saga for Ambiguous {
    type State = St;
    fn name() -> &'static str {
        "route-ambiguous"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Order, _>(|e: &OrderPlaced| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "branch_a",
            task(await_a).on_response(|d: WorkDone| OutA { note: d.note }),
        )
        .step(
            "branch_b",
            task(await_b)
                .terminal_success()
                .on_response(|d: WorkDone| OutB { note: d.note }),
        )
    }
}

#[tokio::test]
async fn pg_ambiguous_await_is_a_config_error() {
    // Phase 2 moved this from a runtime delivery refusal to a BUILD-TIME
    // topology error: awaits are declared on the topology, so two
    // concurrently-activatable steps sharing a response event type are
    // detected by `validate()` before any traffic. (The runtime
    // `PgRunnerError::AmbiguousAwait` refusal remains as a backstop for
    // gated topologies the static check exempts.)
    let topology = Ambiguous::build(SagaBuilder::<Ambiguous>::for_saga());
    let err = topology
        .validate()
        .expect_err("two unordered awaits on one event type must fail validation");
    assert!(
        matches!(
            err,
            mire_sagas::TopologyError::AmbiguousAwaitTypes {
                event_type: "route.work-done",
                ..
            }
        ),
        "expected AmbiguousAwaitTypes, got: {err}"
    );
}

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