//! Chunk 4c: PgSagaRunner emits a typed request, registered in-process
//! handler responds, the map closure produces the step's typed output,
//! and the saga drives through to terminal — through real Postgres.
//!
//! This is the durable counterpart to the in-memory `auto_dispatch`
//! tests. Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

// =====================================================================
// Trigger
// =====================================================================

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

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

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

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

// =====================================================================
// Step 1: emits FlightHoldRequest, awaits FlightHeld, maps to
// FlightReserved.
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-emit.flight-hold-requested")]
struct FlightHoldRequest {
    id: String,
}

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

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

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

async fn handle_flight_hold(Req(r): Req<FlightHoldRequest>) -> FlightHeld {
    FlightHeld {
        hold_id: format!("fh-{}", r.id),
    }
}

// =====================================================================
// Step 2: reads upstream Output<FlightReserved>, emits payment.
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-emit.payment-requested")]
struct PaymentRequested {
    hold_id: String,
    customer: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-emit.payment-settled")]
struct PaymentSettled {
    payment_id: String,
}

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

fn pay(State(s): State<BookingState>, Output(flight): Output<FlightReserved>) -> Emit {
    Emit::event(PaymentRequested {
        hold_id: flight.hold_id,
        customer: s.customer.clone(),
    })
}

async fn handle_payment(Req(r): Req<PaymentRequested>) -> PaymentSettled {
    PaymentSettled {
        payment_id: format!("pay-{}", r.hold_id),
    }
}

// =====================================================================
// Saga
// =====================================================================

struct PgEmitSaga;
impl Saga for PgEmitSaga {
    type State = BookingState;
    fn name() -> &'static str {
        "pg-emit-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(),
                    customer: e.customer.clone(),
                },
            })
        })
        .step(
            "flight",
            task(reserve_flight).on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id }),
        )
        .step(
            "pay",
            task(pay)
                .after("flight")
                .terminal_success()
                .on_response(|p: PaymentSettled| PaymentCompleted {
                    payment_id: p.payment_id,
                }),
        )
        .effect(handle_flight_hold)
        .effect(handle_payment)
    }
}

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

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

#[tokio::test]
async fn two_step_emit_chain_runs_through_to_terminal_via_postgres() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = PgEmitSaga::build(SagaBuilder::<PgEmitSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("emit-{}", uuid::Uuid::new_v4());

    runner
        .trigger::<Booking>(&BookingRequested {
            id: id.clone(),
            customer: "alice".into(),
        })
        .await
        .expect("trigger ok")
        .expect("instance created");

    let env = runner.load(&id).await.unwrap().unwrap();

    // Both steps Completed, saga terminal-success.
    assert_eq!(env.step_status("flight"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("pay"), SagaStepStatus::Completed);
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);

    // The flight step's typed output was produced via the map closure
    // applied to the FlightHeld response.
    let flight = env.outputs.get("flight").expect("flight output recorded");
    let flight: FlightReserved = serde_json::from_value(flight.clone()).unwrap();
    assert_eq!(flight.hold_id, format!("fh-{id}"));

    // The payment step read Output<FlightReserved> from the envelope
    // and threaded the hold_id into its own request.
    let payment = env.outputs.get("pay").expect("pay output recorded");
    let payment: PaymentCompleted = serde_json::from_value(payment.clone()).unwrap();
    assert_eq!(payment.payment_id, format!("pay-fh-{id}"));
}

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

    // The flight step emits FlightHoldRequest but no in-process handler
    // is registered → the emit is destined for an external worker, so
    // the saga SUSPENDS awaiting FlightHeld (Regime B). It does not
    // error — a later `deliver(...)` resumes it.
    use mire_sagas::{SagaInstanceStatus, SagaStepStatus};

    struct UnhandledSaga;
    impl Saga for UnhandledSaga {
        type State = BookingState;
        fn name() -> &'static str {
            "pg-emit-unhandled-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(),
                        customer: e.customer.clone(),
                    },
                })
            })
            .step(
                "flight",
                task(reserve_flight)
                    .terminal_success()
                    .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id }),
            )
            // no .handle(...) → external
        }
    }

    let topology = UnhandledSaga::build(SagaBuilder::<UnhandledSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);

    let id = format!("unhandled-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Booking>(&BookingRequested {
            id: id.clone(),
            customer: "bob".into(),
        })
        .await
        .expect("trigger should succeed, suspending the saga");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(
        env.status,
        SagaInstanceStatus::Running,
        "saga should be suspended awaiting the external FlightHeld"
    );
    assert_eq!(env.step_status("flight"), SagaStepStatus::Active);

    // And an external worker can resume it.
    runner
        .deliver(
            &id,
            &FlightHeld {
                hold_id: format!("fh-{id}"),
            },
        )
        .await
        .unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}
