//! Chunk 4a integration: trigger + commit through real Postgres.
//!
//! Asserts that `PgSagaRunner::trigger` against a saga whose steps
//! use `...` (a) writes envelope events to the
//! store and (b) reloading via `EventStore::load` reconstructs the
//! same envelope state. This is the architectural-shape proof for
//! the durable runner — all subsequent slices add behaviours on top.
//!
//! Gated on `DATABASE_URL`, like the rest of the mire integration
//! suite. Returns silently when unset.

#![allow(dead_code)]

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

// =====================================================================
// Trigger aggregate
// =====================================================================

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

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

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

// =====================================================================
// Saga: every step is a value (compute) handler — 4a-compatible.
// =====================================================================

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

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

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

async fn step_one(State(s): State<BookingState>) -> StepOneOutput {
    StepOneOutput {
        note: format!("touched {}", s.id),
    }
}

async fn step_two(State(_): State<BookingState>) -> StepTwoOutput {
    StepTwoOutput { finished: true }
}

struct PgImmediateSaga;
impl Saga for PgImmediateSaga {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-immediate-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("step1", task(step_one))
        .step("step2", task(step_two).after("step1").terminal_success())
    }
}

// =====================================================================
// Helpers
// =====================================================================

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<PgImmediateSaga> {
    let topology = PgImmediateSaga::build(SagaBuilder::<PgImmediateSaga>::for_saga());
    PgSagaRunner::new(topology, store)
}

// =====================================================================
// Tests
// =====================================================================

#[tokio::test]
async fn trigger_runs_two_step_immediate_saga_to_terminal() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    let runner = build_runner(store.clone());
    let id = format!("immediate-{}", uuid::Uuid::new_v4());

    let returned = runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .expect("trigger ok")
        .expect("instance created");
    assert_eq!(returned, id);

    // Reload via the runner's own load() — same path the durable
    // driver loop will use.
    let env = runner
        .load(&id)
        .await
        .expect("load ok")
        .expect("instance must exist after save");

    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
    assert_eq!(env.step_status("step1"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("step2"), SagaStepStatus::Completed);

    // The recorded output JSON contains the typed payloads.
    let one = env.outputs.get("step1").expect("step1 output");
    let one_typed: StepOneOutput = serde_json::from_value(one.clone()).unwrap();
    assert!(one_typed.note.contains(&id));
    let two = env.outputs.get("step2").expect("step2 output");
    let two_typed: StepTwoOutput = serde_json::from_value(two.clone()).unwrap();
    assert!(two_typed.finished);
}

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

    let first = runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    let second = runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    assert_eq!(first, second, "same instance_id must collapse");

    // Stream version should not have advanced past the original 4
    // events (Started + StepActivated + StepCompleted ×2 + Confirmed = 5).
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}

#[tokio::test]
async fn trigger_returns_none_for_unrelated_aggregate() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let runner = build_runner(store);

    #[derive(Default, Clone)]
    struct OtherAgg;
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "other.ev")]
    struct OtherEv;
    impl Aggregate for OtherAgg {
        type Event = OtherEv;
        fn stream_category() -> &'static str {
            "unrelated"
        }
        fn apply(&mut self, _: &Self::Event) {}
    }

    let res = runner.trigger::<OtherAgg>(&OtherEv).await.unwrap();
    assert_eq!(res, None, "wrong aggregate must not trigger the saga");
}

// =====================================================================
// Negative: a step that emits returns Unsupported for now (Chunk 4a)
// =====================================================================

mod emit_suspends {
    use super::*;
    use mire::EventData;
    use mire_sagas::{SagaInstanceStatus, SagaStepStatus};

    #[derive(Debug, Clone, Serialize, Deserialize, EventData)]
    #[mire(event_type = "emit.req")]
    struct SomeRequest;

    #[derive(Debug, Clone, Serialize, Deserialize, EventData)]
    #[mire(event_type = "emit.ok")]
    struct SomeOk;

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

    async fn emitting_step(_: State<BookingState>) -> Emit {
        Emit::event(SomeRequest)
    }

    struct EmittingSaga;
    impl Saga for EmittingSaga {
        type State = BookingState;
        fn name() -> &'static str {
            "pg-emitting-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(
                "emit",
                task(emitting_step)
                    .terminal_success()
                    .on_response(|_: SomeOk| EmitOut),
            )
        }
    }

    // A step that emits with no in-process handler suspends awaiting the
    // external response (Regime B), then a `deliver` resumes it.
    #[tokio::test]
    async fn emitting_step_suspends_then_deliver_resumes() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = EmittingSaga::build(SagaBuilder::<EmittingSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);
        let id = format!("emit-{}", uuid::Uuid::new_v4());

        runner
            .trigger::<Booking>(&BookingRequested { id: id.clone() })
            .await
            .expect("trigger should suspend, not error");

        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.status, SagaInstanceStatus::Running);
        assert_eq!(env.step_status("emit"), SagaStepStatus::Active);

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