//! P3.1 — the `es_saga_instances` status projection is maintained in the same
//! transaction as every commit, mirroring the envelope's status and shard so
//! recovery (P3.3) and ops can find non-terminal instances without replaying
//! streams.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::time::Duration;

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

mod common;

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

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "isp.requested")]
struct Requested {
    id: String,
}

impl Aggregate for Trigger {
    type Event = Requested;
    fn stream_category() -> &'static str {
        "isp-trigger"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

// "work" suspends awaiting an external response → instance stays Running.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "isp.work-req")]
struct WorkReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "isp.work-done")]
struct WorkDone;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct StepDone;

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

struct StatusSaga;
impl Saga for StatusSaga {
    type State = St;
    fn name() -> &'static str {
        "instance-status-test"
    }
    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(|_: WorkDone| StepDone)
                .terminal_success(),
        )
    }
}

fn runner(store: &mire::EventStore, shards: u32) -> PgSagaRunner<StatusSaga> {
    PgSagaRunner::new(
        StatusSaga::build(SagaBuilder::<StatusSaga>::for_saga()),
        store.clone(),
    )
    .with_num_shards(shards)
}

async fn projected(store: &mire::EventStore, id: &str) -> Option<(String, i32)> {
    sqlx::query_as::<_, (String, i32)>(
        "SELECT status, shard_id FROM es_saga_instances WHERE saga_name = $1 AND instance_id = $2",
    )
    .bind(StatusSaga::name())
    .bind(id)
    .fetch_optional(store.pool())
    .await
    .unwrap()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn status_projection_tracks_running_then_terminal() {
    let Some(store) = common::store().await else {
        return;
    };
    let run = runner(&store, 16);
    let id = common::unique("isp");

    // Trigger → suspends on `work` → Running, with the row written in the
    // trigger's commit.
    run.trigger::<Trigger>(&Requested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    let (status, shard) = projected(&store, &id)
        .await
        .expect("row upserted on trigger commit");
    assert_eq!(status, "running", "suspended instance is Running");
    assert!(
        (0..16).contains(&shard),
        "shard within configured range: {shard}"
    );

    // Deliver the response → terminal success → the row flips, same shard.
    run.deliver(&id, &WorkDone).await.unwrap();
    let (status2, shard2) = projected(&store, &id).await.expect("row still present");
    assert_eq!(
        status2, "succeeded-terminal",
        "delivered instance is terminal"
    );
    assert_eq!(shard2, shard, "shard is stable across commits");
}

// A compensating saga reaches failed-terminal — the projection must mirror it.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "isp.hold-req")]
struct HoldReq;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "isp.held")]
struct Held;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HoldDone;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "isp.rel-req")]
struct RelReq;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "isp.rel")]
struct Rel;

async fn hold(State(_): State<St>) -> Emit {
    Emit::event(HoldReq)
}
async fn handle_hold(mire_sagas::Req(_): mire_sagas::Req<HoldReq>) -> Held {
    Held
}
async fn pay(State(_): State<St>) -> Emit {
    Emit::event(WorkReq { id: "x".into() }).timeout(Duration::from_millis(50))
}
async fn release(Output(_): Output<HoldDone>) -> Emit {
    Emit::event(RelReq)
}
async fn handle_release(mire_sagas::Req(_): mire_sagas::Req<RelReq>) -> Rel {
    Rel
}

struct CompSaga;
impl Saga for CompSaga {
    type State = St;
    fn name() -> &'static str {
        "instance-status-comp-test"
    }
    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(
            "hold",
            task(hold)
                .on_response(|_: Held| HoldDone)
                .compensate(comp(release).on_response::<Rel>()),
        )
        .step(
            "pay",
            task(pay)
                .after("hold")
                .on_response(|_: WorkDone| StepDone)
                .terminal_success(),
        )
        .handle(handle_hold)
        .handle(handle_release)
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn status_projection_mirrors_failed_terminal() {
    let Some(store) = common::store().await else {
        return;
    };
    let run = PgSagaRunner::new(
        CompSaga::build(SagaBuilder::<CompSaga>::for_saga()),
        store.clone(),
    )
    .with_num_shards(8);
    let id = common::unique("ispc");

    run.trigger::<Trigger>(&Requested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    // hold completes in-process, pay suspends with a deadline → Running.
    let row: (String, i32) = sqlx::query_as(
        "SELECT status, shard_id FROM es_saga_instances WHERE saga_name = $1 AND instance_id = $2",
    )
    .bind(CompSaga::name())
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(row.0, "running");

    // Force the pay deadline to expire and tick → fail + compensate hold →
    // failed-terminal.
    tokio::time::sleep(Duration::from_millis(80)).await;
    run.tick_timeouts(chrono::Utc::now()).await.unwrap();

    let status: String = sqlx::query_scalar(
        "SELECT status FROM es_saga_instances WHERE saga_name = $1 AND instance_id = $2",
    )
    .bind(CompSaga::name())
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(
        status, "failed-terminal",
        "projection mirrors the failed-terminal end state"
    );
}
