//! Chunk 4c.2: the durable suspend/resume path.
//!
//! A step emits a request with NO in-process handler. The runner
//! records `StepActivated`, persists the await to `es_saga_awaits`,
//! commits, and suspends — `trigger()` returns with the saga still
//! Running. Later an external worker calls `deliver(...)`, which
//! routes via the awaits projection, re-runs the (pure) step handler
//! to recover its map closure, applies it, records `StepCompleted`,
//! and drives the saga forward.
//!
//! This is what makes the durable runner actually durable: the saga
//! state survives across the gap between emit and response, which may
//! span process restarts.
//!
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

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

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

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

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

// Step 1: emits a request that NO in-process handler serves — so the
// saga suspends awaiting FlightHeld from an external worker.
#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-sr.flight-hold-requested")]
struct FlightHoldRequest {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-sr.flight-held")]
struct FlightHeld {
    hold_id: String,
}

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

async fn reserve_flight(State(s): State<BookingState>) -> Emit {
    Emit::event(FlightHoldRequest { id: s.id.clone() })
}

// Step 2: downstream of flight, reads its output, also immediate.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Confirmed {
    hold_id: String,
}

async fn confirm(Output(f): Output<FlightReserved>) -> Confirmed {
    Confirmed { hold_id: f.hold_id }
}

struct SuspendSaga;
impl Saga for SuspendSaga {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-suspend-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Booking, _>(|e: &BookingRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: BookingState { id: e.id.clone() },
            })
        })
        .step(
            "flight",
            task(reserve_flight).on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id }),
        )
        .step("confirm", task(confirm).after("flight").terminal_success())
        // No .handle(...) for FlightHoldRequest — it's external.
    }
}

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

fn build_runner(store: EventStore) -> PgSagaRunner<SuspendSaga> {
    let topology = SuspendSaga::build(SagaBuilder::<SuspendSaga>::for_saga());
    PgSagaRunner::new(topology, store)
}

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

    // 1. Trigger — the saga should suspend at `flight` awaiting the
    //    external FlightHeld response.
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .expect("trigger ok");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(
        env.status,
        SagaInstanceStatus::Running,
        "saga should be suspended (Running), not terminal"
    );
    assert_eq!(
        env.step_status("flight"),
        SagaStepStatus::Active,
        "flight should be Active (awaiting external response)"
    );
    assert_eq!(
        env.step_status("confirm"),
        SagaStepStatus::Pending,
        "confirm shouldn't have activated yet"
    );

    // The await row exists in the routing projection.
    let await_count: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_awaits \
         WHERE saga_name = 'pg-suspend-saga' AND instance_id = $1",
    )
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(await_count, 1, "one await row should be persisted");

    // 2. External worker delivers the FlightHeld response.
    let consumed = runner
        .deliver(
            &id,
            &FlightHeld {
                hold_id: format!("fh-{id}"),
            },
        )
        .await
        .expect("deliver ok");
    assert!(consumed, "deliver should report a step consumed the event");

    // 3. Saga resumed: flight Completed, confirm ran, terminal-success.
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
    assert_eq!(env.step_status("flight"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("confirm"), SagaStepStatus::Completed);

    // The flight output was produced by the map closure recovered on
    // resume (re-running the pure handler).
    let flight: FlightReserved =
        serde_json::from_value(env.outputs.get("flight").unwrap().clone()).unwrap();
    assert_eq!(flight.hold_id, format!("fh-{id}"));

    // confirm read flight's output via Output<FlightReserved>.
    let confirmed: Confirmed =
        serde_json::from_value(env.outputs.get("confirm").unwrap().clone()).unwrap();
    assert_eq!(confirmed.hold_id, format!("fh-{id}"));

    // 4. The await row was deleted on resume.
    let await_count: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_awaits \
         WHERE saga_name = 'pg-suspend-saga' AND instance_id = $1",
    )
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(await_count, 0, "await row should be deleted after resume");
}

#[tokio::test]
async fn suspend_persists_emitted_request_payload_durably() {
    // Review C5/SAGA-3/DX-1: when a step emits a request that no
    // in-process handler serves, the runner suspends — and must record
    // *what* was requested durably, so an external worker (in another
    // process, with no shared memory) can discover the request from the
    // event store alone. Previously the payload was dropped and only the
    // awaited `event_type` survived in `es_saga_awaits`.
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = format!("suspend-emit-{}", uuid::Uuid::new_v4());

    // Runner A triggers + suspends, then is dropped.
    {
        let runner_a = build_runner(store.clone());
        runner_a
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .expect("trigger ok");
    }

    // Runner B (fresh instance) recovers the emitted request purely from
    // the persisted envelope — the StepEmitted event committed in the same
    // transaction as StepActivated.
    let runner_b = build_runner(store.clone());
    let env = runner_b.load(&id).await.unwrap().unwrap();
    assert_eq!(env.step_status("flight"), SagaStepStatus::Active);

    let requests = env.requests_for("flight");
    assert_eq!(
        requests.len(),
        1,
        "the emitted FlightHoldRequest should be durably recorded"
    );
    assert_eq!(requests[0].event_type, "pg-sr.flight-hold-requested");
    let recovered: FlightHoldRequest =
        serde_json::from_value(requests[0].payload.clone()).expect("payload round-trips");
    assert_eq!(recovered.id, id, "the request payload survives durably");

    // Once the await resolves, the recorded request is cleared.
    runner_b
        .deliver(
            &id,
            &FlightHeld {
                hold_id: format!("fh-{id}"),
            },
        )
        .await
        .unwrap();
    let env = runner_b.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
    assert!(
        env.requests_for("flight").is_empty(),
        "the emitted request is cleared once the step completes"
    );
}

#[tokio::test]
async fn deliver_to_unknown_event_type_is_noop() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = build_runner(store);
    let id = format!("suspend-noop-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .unwrap();

    // Deliver an event nothing awaits — should be a no-op (false).
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sr.unrelated")]
    struct Unrelated;

    let consumed = runner.deliver(&id, &Unrelated).await.unwrap();
    assert!(!consumed, "delivering an unawaited event must be a no-op");

    // Saga still suspended at flight.
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.step_status("flight"), SagaStepStatus::Active);
}

#[tokio::test]
async fn deliver_survives_a_fresh_runner_instance() {
    // The whole point of durability: a deliver() from a DIFFERENT
    // runner instance (simulating a different process / replica)
    // resumes the saga, because all state is in Postgres.
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = format!("suspend-crossproc-{}", uuid::Uuid::new_v4());

    // Runner A triggers + suspends.
    {
        let runner_a = build_runner(store.clone());
        runner_a
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .unwrap();
    }

    // Runner B (fresh instance, no shared memory) delivers + resumes.
    let runner_b = build_runner(store.clone());
    let consumed = runner_b
        .deliver(
            &id,
            &FlightHeld {
                hold_id: format!("fh-{id}"),
            },
        )
        .await
        .unwrap();
    assert!(consumed);

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