//! End-to-end test harness: drive the travel-booking saga through the
//! durable runner in **external-worker mode** — every forward-step
//! response is delivered explicitly via `deliver`, proving cross-call
//! resumption from the persisted awaits. Scenarios:
//!
//! - **Happy path**: trigger → both holds → payment → ticket → confirmed.
//! - **Payment timeout**: trigger → both holds → tick past deadline
//!   → compensation cascade releases hotel + flight (no refund since
//!   payment never captured) → cancelled.
//! - **Admin signal**: trigger → flight + hotel held + payment captured →
//!   admin cancel signal mid-confirm → compensation cascade → cancelled.
//!
//! Forward request handlers are stripped (external-worker mode); the
//! *compensation* request handlers stay registered because the durable
//! runner requires in-process handlers for compensation emits until the
//! outbox lands (tasks/phase-3-worker-loop.md P3.2).
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::time::Duration;

use mire_sagas::{PgSagaRunner, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus};

mod common;

// Pull in the example's saga definition as a module so the test exercises
// the exact same topology the conformance example builds. The example's
// `main()` becomes an unused function in this crate — harmless.
#[allow(unused)]
#[path = "../examples/travel_booking.rs"]
mod travel_booking;

use mire_sagas::PgRunnerError;

use travel_booking::{
    AdminForceCancel, BookingEvent, BookingId, CustomerId, FlightHeld, FlightHoldId, FlightId,
    HotelHeld, HotelHoldId, HotelId, Money, PaymentCaptured, PaymentDeadline, PaymentId,
    RequestFlightHold, RequestHotelHold, RequestPaymentCapture, TicketId, TicketIssued,
    TravelBookingSaga,
};

/// Build a runner whose topology has no FORWARD request handlers — the
/// test drives every forward response explicitly via `deliver`
/// (external-worker mode). Compensation request handlers (releases,
/// refund) stay registered: the durable runner dispatches compensation
/// emits in-process only, until the outbox ships.
///
/// `payment_deadline` is injected per test: `tick_timeouts` sweeps every
/// instance of this saga name, so tests that don't exercise the timeout
/// use a far-future deadline that a sibling's tick can never expire.
fn build_runner(
    store: &mire::EventStore,
    payment_deadline: Duration,
) -> PgSagaRunner<TravelBookingSaga> {
    let mut topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    for et in [
        "saga.flight.hold-requested",
        "saga.hotel.hold-requested",
        "saga.payment.capture-requested",
        "saga.carrier.ticket-requested",
    ] {
        topology.remove_request_handler(et);
    }
    let topology = topology.dep(PaymentDeadline(payment_deadline));
    PgSagaRunner::new(topology, store.clone())
}

const FAR_FUTURE: Duration = Duration::from_secs(7 * 24 * 3600);

fn make_request(id: &str) -> BookingEvent {
    BookingEvent::Requested {
        booking_id: BookingId(id.into()),
        customer_id: CustomerId("alice".into()),
        flight_id: FlightId("LH-100".into()),
        hotel_id: HotelId("HILTON-OSL".into()),
        amount: Money(150_000),
    }
}

// =====================================================================
// Happy path
// =====================================================================

#[tokio::test]
async fn happy_path_runs_to_confirmed() {
    let Some(store) = common::store().await else {
        return;
    };
    let runner = build_runner(&store, FAR_FUTURE);

    // 1. Trigger: a Booking::Requested event lands.
    let id = common::unique("b-happy");
    let request = make_request(&id);
    let instance_id = runner
        .trigger::<travel_booking::Booking>(&request)
        .await
        .expect("trigger ok")
        .expect("instance started");
    assert_eq!(instance_id, id);

    // 2. The saga activated `flight` and `hotel` in parallel; each emitted
    // its hold-request event (recorded durably on the stream).
    let emits = runner.emitted(&instance_id).await.unwrap();
    assert_eq!(emits.len(), 2, "expected one emit per parallel step");

    let event_types: Vec<&str> = emits.iter().map(|e| e.event_type.as_str()).collect();
    assert!(event_types.contains(&"saga.flight.hold-requested"));
    assert!(event_types.contains(&"saga.hotel.hold-requested"));

    // The flight + hotel emits decode back to typed events with the right ids.
    let flight_req: RequestFlightHold = emits
        .iter()
        .find(|e| e.event_type == "saga.flight.hold-requested")
        .map(|e| serde_json::from_value(e.payload.clone()).unwrap())
        .unwrap();
    assert_eq!(flight_req.flight_id.0, "LH-100");

    let hotel_req: RequestHotelHold = emits
        .iter()
        .find(|e| e.event_type == "saga.hotel.hold-requested")
        .map(|e| serde_json::from_value(e.payload.clone()).unwrap())
        .unwrap();
    assert_eq!(hotel_req.hotel_id.0, "HILTON-OSL");

    // Both steps are Active.
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Active
    );
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Active
    );
    // Payment hasn't started yet.
    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Pending
    );

    // 3. Deliver the flight-held response. Flight step completes.
    let delivered = runner
        .deliver(
            &instance_id,
            &FlightHeld {
                booking_id: BookingId(id.clone()),
                hold_id: FlightHoldId("fh-1".into()),
            },
        )
        .await
        .expect("deliver ok");
    assert!(delivered, "step should have consumed the event");
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Completed
    );
    // Payment still pending (hotel hasn't completed).
    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Pending
    );

    // 4. Deliver hotel-held → both reservations done → payment activates.
    let delivered = runner
        .deliver(
            &instance_id,
            &HotelHeld {
                booking_id: BookingId(id.clone()),
                hold_id: HotelHoldId("hh-1".into()),
            },
        )
        .await
        .expect("deliver ok");
    assert!(delivered);
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Completed
    );
    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Active
    );

    // Payment step emitted the capture-request.
    let emits = runner.emitted(&instance_id).await.unwrap();
    let last_emit = emits.last().unwrap();
    assert_eq!(last_emit.event_type, "saga.payment.capture-requested");
    let pay_req: RequestPaymentCapture = serde_json::from_value(last_emit.payload.clone()).unwrap();
    assert_eq!(pay_req.amount.0, 150_000);

    // 5. Deliver payment-captured → payment completes → confirm activates.
    runner
        .deliver(
            &instance_id,
            &PaymentCaptured {
                booking_id: BookingId(id.clone()),
                payment_id: PaymentId("p-1".into()),
            },
        )
        .await
        .expect("deliver ok");
    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Completed
    );
    assert_eq!(
        runner.step_status(&instance_id, "confirm").await.unwrap(),
        SagaStepStatus::Active
    );

    // 6. Deliver ticket-issued → confirm completes → saga succeeded.
    runner
        .deliver(
            &instance_id,
            &TicketIssued {
                booking_id: BookingId(id.clone()),
                ticket_id: TicketId("tk-1".into()),
            },
        )
        .await
        .expect("deliver ok");

    let env = runner.load(&instance_id).await.unwrap().expect("envelope");
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
    assert_eq!(
        runner.step_status(&instance_id, "confirm").await.unwrap(),
        SagaStepStatus::Completed
    );
}

// =====================================================================
// Payment timeout → compensation cascade
// =====================================================================

#[tokio::test]
async fn payment_timeout_triggers_compensation() {
    let Some(store) = common::store().await else {
        return;
    };
    // This test uses the real 10-minute deadline and ticks past it.
    let runner = build_runner(&store, Duration::from_secs(10 * 60));
    let id = common::unique("b-timeout");
    let request = make_request(&id);
    let instance_id = runner
        .trigger::<travel_booking::Booking>(&request)
        .await
        .unwrap()
        .unwrap();

    // Move past holds.
    runner
        .deliver(
            &instance_id,
            &FlightHeld {
                booking_id: BookingId(id.clone()),
                hold_id: FlightHoldId("fh-1".into()),
            },
        )
        .await
        .unwrap();
    runner
        .deliver(
            &instance_id,
            &HotelHeld {
                booking_id: BookingId(id.clone()),
                hold_id: HotelHoldId("hh-1".into()),
            },
        )
        .await
        .unwrap();

    // Payment is active with a 10-minute deadline. Tick well past it; the
    // compensation cascade runs through the registered release handlers.
    runner
        .tick_timeouts(chrono::Utc::now() + chrono::Duration::minutes(11))
        .await
        .unwrap();

    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Failed
    );

    // Both holds released; payment never captured so no refund emit.
    let emits: Vec<String> = runner
        .emitted(&instance_id)
        .await
        .unwrap()
        .iter()
        .map(|e| e.event_type.clone())
        .collect();
    assert!(emits.contains(&"saga.hotel.hold-release-requested".to_string()));
    assert!(emits.contains(&"saga.flight.hold-release-requested".to_string()));
    assert!(!emits.contains(&"saga.payment.refund-requested".to_string()));

    let env = runner.load(&instance_id).await.unwrap().expect("envelope");
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Compensated
    );
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Compensated
    );
}

// =====================================================================
// Admin signal mid-flow
// =====================================================================

#[tokio::test]
async fn admin_cancel_triggers_compensation() {
    let Some(store) = common::store().await else {
        return;
    };
    let runner = build_runner(&store, FAR_FUTURE);
    let id = common::unique("b-admin");
    let request = make_request(&id);
    let instance_id = runner
        .trigger::<travel_booking::Booking>(&request)
        .await
        .unwrap()
        .unwrap();

    runner
        .deliver(
            &instance_id,
            &FlightHeld {
                booking_id: BookingId(id.clone()),
                hold_id: FlightHoldId("fh-1".into()),
            },
        )
        .await
        .unwrap();
    runner
        .deliver(
            &instance_id,
            &HotelHeld {
                booking_id: BookingId(id.clone()),
                hold_id: HotelHoldId("hh-1".into()),
            },
        )
        .await
        .unwrap();
    // Capture payment so the refund branch is exercised too.
    runner
        .deliver(
            &instance_id,
            &PaymentCaptured {
                booking_id: BookingId(id.clone()),
                payment_id: PaymentId("p-1".into()),
            },
        )
        .await
        .unwrap();

    // Mid-confirm, admin pulls the plug. The cascade runs through the
    // registered compensation handlers: refund (most recent completed
    // step is payment), then hotel release, then flight release.
    runner
        .signal(
            &instance_id,
            AdminForceCancel {
                reason: "fraud detected".into(),
            },
        )
        .await
        .unwrap();

    let emits: Vec<String> = runner
        .emitted(&instance_id)
        .await
        .unwrap()
        .iter()
        .map(|e| e.event_type.clone())
        .collect();
    assert!(emits.contains(&"saga.payment.refund-requested".to_string()));
    assert!(emits.contains(&"saga.hotel.hold-release-requested".to_string()));
    assert!(emits.contains(&"saga.flight.hold-release-requested".to_string()));

    let env = runner.load(&instance_id).await.unwrap().expect("envelope");
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Compensated
    );
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Compensated
    );
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Compensated
    );
}

// =====================================================================
// Bug fix regression: signal on a terminal saga must be rejected.
// =====================================================================

#[tokio::test]
async fn signal_on_terminal_saga_is_rejected() {
    let Some(store) = common::store().await else {
        return;
    };
    let runner = build_runner(&store, FAR_FUTURE);
    let id = common::unique("b-late-signal");
    let request = make_request(&id);
    let instance_id = runner
        .trigger::<travel_booking::Booking>(&request)
        .await
        .unwrap()
        .unwrap();

    // Drive saga to succeed.
    runner
        .deliver(
            &instance_id,
            &FlightHeld {
                booking_id: BookingId(id.clone()),
                hold_id: FlightHoldId("fh-1".into()),
            },
        )
        .await
        .unwrap();
    runner
        .deliver(
            &instance_id,
            &HotelHeld {
                booking_id: BookingId(id.clone()),
                hold_id: HotelHoldId("hh-1".into()),
            },
        )
        .await
        .unwrap();
    runner
        .deliver(
            &instance_id,
            &PaymentCaptured {
                booking_id: BookingId(id.clone()),
                payment_id: PaymentId("p-1".into()),
            },
        )
        .await
        .unwrap();
    runner
        .deliver(
            &instance_id,
            &TicketIssued {
                booking_id: BookingId(id.clone()),
                ticket_id: TicketId("t-1".into()),
            },
        )
        .await
        .unwrap();

    assert!(runner.is_succeeded(&instance_id).await.unwrap());

    // Now signal — should be rejected with SagaTerminal.
    let result = runner
        .signal(
            &instance_id,
            AdminForceCancel {
                reason: "too late".into(),
            },
        )
        .await;
    assert!(
        matches!(result, Err(PgRunnerError::SagaTerminal { .. })),
        "expected SagaTerminal error, got {result:?}"
    );

    // Status unchanged.
    assert_eq!(
        runner.status(&instance_id).await.unwrap(),
        Some(SagaInstanceStatus::SucceededTerminal)
    );
}
