//! Travel-booking saga — the conformance example for `mire-sagas`.
//!
//! The saga's DAG is declared in `Saga::build`; the in-process worker that
//! services each emitted request is declared on the *same* builder via
//! `.handle::<RequestT>(handler)`. No bare strings, no string matching,
//! no separate worker struct: every request type → response type pairing
//! is registered with `.handle::<R>(...)` and the runner auto-dispatches.
//!
//! Each handler is async, with axum-style extractors. The `Req<R>` extractor
//! pulls the typed request payload.

use std::sync::Arc;
use std::time::Duration;

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

// ---------------------------------------------------------------------
// 1. IDs (newtype wrappers — typed cross-step).
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct BookingId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct CustomerId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct FlightId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct FlightHoldId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct HotelId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct HotelHoldId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct PaymentId(pub String);
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct TicketId(pub String);
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct Money(pub i64);

impl BookingId {
    pub fn new(s: impl Into<String>) -> Self {
        let s = s.into();
        assert!(!s.is_empty(), "BookingId must not be empty");
        Self(s)
    }
}
impl CustomerId {
    pub fn new(s: impl Into<String>) -> Self {
        let s = s.into();
        assert!(!s.is_empty(), "CustomerId must not be empty");
        Self(s)
    }
}
impl FlightId {
    pub fn new(s: impl Into<String>) -> Self {
        let s = s.into();
        assert!(!s.is_empty(), "FlightId must not be empty");
        Self(s)
    }
}
impl HotelId {
    pub fn new(s: impl Into<String>) -> Self {
        let s = s.into();
        assert!(!s.is_empty(), "HotelId must not be empty");
        Self(s)
    }
}
impl Money {
    pub fn from_minor(n: i64) -> Self {
        assert!(n >= 0, "Money must be non-negative");
        assert!(n <= i64::MAX / 2, "Money cap exceeded");
        Self(n)
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum EmailTemplate {
    BookingConfirmed,
    BookingCancelled,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum CancelReason {
    Timeout,
    PaymentFailed(String),
    AdminOverride(String),
}

// ---------------------------------------------------------------------
// 2. The `Booking` aggregate — the trigger source.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Default)]
pub struct Booking {
    pub customer_id: CustomerId,
    pub flight_id: FlightId,
    pub hotel_id: HotelId,
    pub amount: Money,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum BookingEvent {
    Requested {
        booking_id: BookingId,
        customer_id: CustomerId,
        flight_id: FlightId,
        hotel_id: HotelId,
        amount: Money,
    },
    Confirmed,
    Cancelled {
        reason: CancelReason,
    },
}

impl EventData for BookingEvent {
    fn event_type(&self) -> &'static str {
        match self {
            BookingEvent::Requested { .. } => "booking.requested",
            BookingEvent::Confirmed => "booking.confirmed",
            BookingEvent::Cancelled { .. } => "booking.cancelled",
        }
    }
}

impl Aggregate for Booking {
    type Event = BookingEvent;
    fn stream_category() -> &'static str {
        "booking"
    }
    fn apply(&mut self, _event: &BookingEvent) {}
}

// ---------------------------------------------------------------------
// 3. Request events the saga emits.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.flight.hold-requested")]
pub struct RequestFlightHold {
    pub booking_id: BookingId,
    pub flight_id: FlightId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.hotel.hold-requested")]
pub struct RequestHotelHold {
    pub booking_id: BookingId,
    pub hotel_id: HotelId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.payment.capture-requested")]
pub struct RequestPaymentCapture {
    pub booking_id: BookingId,
    pub amount: Money,
    /// The flight hold the payment is for. Threaded through from the
    /// flight step's typed output.
    pub flight_hold: FlightHoldId,
    /// The hotel hold the payment is for.
    pub hotel_hold: HotelHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.carrier.ticket-requested")]
pub struct IssueCarrierTicket {
    pub booking_id: BookingId,
    pub customer_id: CustomerId,
    /// The payment we settled — threaded through from the payment step's
    /// typed output. Lets the carrier service correlate ticket issuance
    /// with the confirmed payment.
    pub payment_id: PaymentId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.flight.hold-release-requested")]
pub struct ReleaseFlightHold {
    pub hold_id: FlightHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.hotel.hold-release-requested")]
pub struct ReleaseHotelHold {
    pub hold_id: HotelHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.payment.refund-requested")]
pub struct RefundPayment {
    pub payment_id: PaymentId,
    pub reason: CancelReason,
}

// ---------------------------------------------------------------------
// 4. Response events — what each registered handler returns.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "flight.held")]
pub struct FlightHeld {
    pub booking_id: BookingId,
    pub hold_id: FlightHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "hotel.held")]
pub struct HotelHeld {
    pub booking_id: BookingId,
    pub hold_id: HotelHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "payment.captured")]
pub struct PaymentCaptured {
    pub booking_id: BookingId,
    pub payment_id: PaymentId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "carrier.ticket-issued")]
pub struct TicketIssued {
    pub booking_id: BookingId,
    pub ticket_id: TicketId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "flight.released")]
pub struct FlightReleased {
    pub hold_id: FlightHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "hotel.released")]
pub struct HotelReleased {
    pub hold_id: HotelHoldId,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "payment.refunded")]
pub struct PaymentRefunded {
    pub payment_id: PaymentId,
}

/// Typed failure event the payment-capture handler can return as the Err
/// branch of `Result<PaymentCaptured, PaymentCaptureFailed>`. When the
/// handler returns Err, the runner emits this event AND fails the
/// awaiting step with `StepError::Failed`, which fires the compensation
/// cascade automatically.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "payment.capture-failed")]
pub struct PaymentCaptureFailed {
    pub booking_id: BookingId,
    pub reason: String,
}

impl std::fmt::Display for PaymentCaptureFailed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "payment capture failed for {}: {}",
            self.booking_id.0, self.reason
        )
    }
}

// ---------------------------------------------------------------------
// 5. The saga's state.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct TravelBooking {
    pub booking_id: BookingId,
    pub customer_id: CustomerId,
    pub flight_id: FlightId,
    pub hotel_id: HotelId,
    pub amount: Money,
}

// ---------------------------------------------------------------------
// 6. Per-step typed outputs.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FlightReserved {
    pub hold_id: FlightHoldId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HotelReserved {
    pub hold_id: HotelHoldId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PaymentSettled {
    pub payment_id: PaymentId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Confirmed {
    pub ticket: TicketId,
}

// ---------------------------------------------------------------------
// 7. Out-of-band signal type.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "saga.admin.force-cancel")]
pub struct AdminForceCancel {
    pub reason: String,
}

// ---------------------------------------------------------------------
// 8. Deps.
// ---------------------------------------------------------------------

#[derive(Debug, Clone, Copy)]
pub struct PaymentDeadline(pub Duration);

#[derive(Debug, Default)]
pub struct CarrierClient;

// ---------------------------------------------------------------------
// 9. STEP HANDLERS — each emits a typed request. Which response event
// completes the step (and how it maps to the step's typed output) is
// declared on the topology in section 12 via `.on_response(...)`.
// ---------------------------------------------------------------------

pub fn reserve_flight(State(state): State<TravelBooking>) -> Emit {
    Emit::event(RequestFlightHold {
        booking_id: state.booking_id.clone(),
        flight_id: state.flight_id.clone(),
    })
}

pub fn reserve_hotel(State(state): State<TravelBooking>) -> Emit {
    Emit::event(RequestHotelHold {
        booking_id: state.booking_id.clone(),
        hotel_id: state.hotel_id.clone(),
    })
}

pub fn request_payment(
    Output(flight): Output<FlightReserved>,
    Output(hotel): Output<HotelReserved>,
    State(state): State<TravelBooking>,
    Dep(deadline): Dep<PaymentDeadline>,
) -> Emit {
    // The deadline is runtime data (a Dep), so it rides on the Emit —
    // overriding the saga's default for this activation.
    Emit::event(RequestPaymentCapture {
        booking_id: state.booking_id.clone(),
        amount: state.amount,
        flight_hold: flight.hold_id,
        hotel_hold: hotel.hold_id,
    })
    .timeout(deadline.0)
}

pub fn confirm(
    Output(payment): Output<PaymentSettled>,
    State(state): State<TravelBooking>,
) -> Emit {
    Emit::event(IssueCarrierTicket {
        booking_id: state.booking_id.clone(),
        customer_id: state.customer_id.clone(),
        payment_id: payment.payment_id,
    })
}

// ---------------------------------------------------------------------
// 10. COMPENSATION HANDLERS — emit the undo request; the completing
// response is declared at registration via `comp(...).on_response::<E>()`.
// ---------------------------------------------------------------------

pub fn release_flight(Output(r): Output<FlightReserved>) -> Emit {
    Emit::event(ReleaseFlightHold { hold_id: r.hold_id })
}

pub fn release_hotel(Output(r): Output<HotelReserved>) -> Emit {
    Emit::event(ReleaseHotelHold { hold_id: r.hold_id })
}

pub fn refund_payment(Output(p): Output<PaymentSettled>, Cause(reason): Cause) -> Emit {
    Emit::event(RefundPayment {
        payment_id: p.payment_id,
        reason: match reason {
            mire_sagas::StepError::Timeout { .. } => CancelReason::Timeout,
            mire_sagas::StepError::DeadLetter { reason, .. } => CancelReason::PaymentFailed(reason),
            mire_sagas::StepError::Cancelled { reason, .. } => CancelReason::AdminOverride(reason),
            mire_sagas::StepError::Failed { reason, .. } => CancelReason::PaymentFailed(reason),
        },
    })
}

// ---------------------------------------------------------------------
// 11. REQUEST HANDLERS — these live on the builder; the runner
// auto-invokes them when a step emits the matching request type.
// No bare strings, no manual driving.
// ---------------------------------------------------------------------

pub async fn handle_flight_hold(Req(r): Req<RequestFlightHold>) -> FlightHeld {
    FlightHeld {
        hold_id: FlightHoldId(format!("fh-{}-{}", r.flight_id.0, r.booking_id.0)),
        booking_id: r.booking_id,
    }
}

pub async fn handle_hotel_hold(Req(r): Req<RequestHotelHold>) -> HotelHeld {
    HotelHeld {
        hold_id: HotelHoldId(format!("hh-{}-{}", r.hotel_id.0, r.booking_id.0)),
        booking_id: r.booking_id,
    }
}

/// Fallible payment-capture handler. Returns `Result<Ok, Err>`:
///  - `Ok(PaymentCaptured)` for a normal capture
///  - `Err(PaymentCaptureFailed)` when the booking_id starts with `"fraud-"`
///
/// On Err, the runner emits the typed `PaymentCaptureFailed` event AND
/// fails the awaiting step → compensation cascade runs without any
/// explicit error handling at the call site.
pub async fn handle_payment_capture(
    Req(r): Req<RequestPaymentCapture>,
) -> Result<PaymentCaptured, PaymentCaptureFailed> {
    if r.booking_id.0.starts_with("fraud-") {
        return Err(PaymentCaptureFailed {
            booking_id: r.booking_id,
            reason: "fraud check declined".into(),
        });
    }
    Ok(PaymentCaptured {
        payment_id: PaymentId(format!("p-{}", r.booking_id.0)),
        booking_id: r.booking_id,
    })
}

pub async fn handle_carrier_ticket(Req(r): Req<IssueCarrierTicket>) -> TicketIssued {
    TicketIssued {
        ticket_id: TicketId(format!("tk-{}", r.booking_id.0)),
        booking_id: r.booking_id,
    }
}

pub async fn handle_release_flight(Req(r): Req<ReleaseFlightHold>) -> FlightReleased {
    FlightReleased { hold_id: r.hold_id }
}

pub async fn handle_release_hotel(Req(r): Req<ReleaseHotelHold>) -> HotelReleased {
    HotelReleased { hold_id: r.hold_id }
}

pub async fn handle_refund_payment(Req(r): Req<RefundPayment>) -> PaymentRefunded {
    PaymentRefunded {
        payment_id: r.payment_id,
    }
}

// ---------------------------------------------------------------------
// 12. The saga definition. ONE builder, both steps and handlers.
// ---------------------------------------------------------------------

pub struct TravelBookingSaga;

impl Saga for TravelBookingSaga {
    type State = TravelBooking;

    fn name() -> &'static str {
        "travel-booking-saga"
    }

    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Booking, _>(|e| match e {
            BookingEvent::Requested {
                booking_id,
                customer_id,
                flight_id,
                hotel_id,
                amount,
            } => Some(Start {
                instance_id: booking_id.0.clone(),
                state: TravelBooking {
                    booking_id: booking_id.clone(),
                    customer_id: customer_id.clone(),
                    flight_id: flight_id.clone(),
                    hotel_id: hotel_id.clone(),
                    amount: *amount,
                },
            }),
            _ => None,
        })
        // Steps + dependencies. Each step bundles its `decide` (the sync
        // `task(handler)` that emits a request), the `effect` that fulfils it
        // (`.effect(...)`, an async request handler), the response→output map
        // (`.on_response(...)`), and its compensation (`.compensate(...)`) —
        // everything about the step, do and undo, stays visually nested under
        // the step's argument list. The effect is keyed by the emitted
        // request's type; R/Args/Resp are inferred from the handler signature.
        .step(
            "flight",
            task(reserve_flight)
                .effect(handle_flight_hold)
                .on_response(|held: FlightHeld| FlightReserved {
                    hold_id: held.hold_id,
                })
                .compensate(
                    comp(release_flight)
                        .effect(handle_release_flight)
                        .on_response::<FlightReleased>(),
                ),
        )
        .step(
            "hotel",
            task(reserve_hotel)
                .effect(handle_hotel_hold)
                .on_response(|held: HotelHeld| HotelReserved {
                    hold_id: held.hold_id,
                })
                .compensate(
                    comp(release_hotel)
                        .effect(handle_release_hotel)
                        .on_response::<HotelReleased>(),
                ),
        )
        .step(
            "payment",
            task(request_payment)
                .effect(handle_payment_capture)
                .after(["flight", "hotel"])
                .on_response(|cap: PaymentCaptured| PaymentSettled {
                    payment_id: cap.payment_id,
                })
                .compensate(
                    comp(refund_payment)
                        .effect(handle_refund_payment)
                        .on_response::<PaymentRefunded>(),
                ),
        )
        .step(
            "confirm",
            task(confirm)
                .effect(handle_carrier_ticket)
                .after("payment")
                .on_response(|t: TicketIssued| Confirmed {
                    ticket: t.ticket_id,
                })
                .terminal_success(),
        )
        // Signals + deps.
        .signal::<AdminForceCancel>()
        .triggers_compensation()
        .dep(PaymentDeadline(Duration::from_secs(10 * 60)))
        .dep(Arc::new(CarrierClient))
    }
}

// ---------------------------------------------------------------------
// 13. main — drives each scenario via runner.trigger() against a real
// Postgres (DATABASE_URL, default the compose.yaml instance). The runner
// auto-dispatches all emits through the registered handlers, so a saga
// runs all the way to terminal in a single trigger() call — every
// transition durably committed to the saga's event stream.
// ---------------------------------------------------------------------

fn main() -> anyhow::Result<()> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    rt.block_on(async {
        let url = std::env::var("DATABASE_URL")
            .unwrap_or_else(|_| "postgres://mire:mire@localhost:5434/mire".to_string());
        let pool = sqlx::PgPool::connect(&url).await?;
        let store = mire::EventStore::new(pool);
        store.migrate().await?;
        mire_sagas::migrate(&store).await?;

        scenario_happy_path(&store).await?;
        println!();
        scenario_payment_timeout(&store).await?;
        println!();
        scenario_admin_cancel(&store).await?;
        println!();
        scenario_handler_returns_err(&store).await?;
        Ok::<(), anyhow::Error>(())
    })?;
    Ok(())
}

/// Unique booking ids per run — instance ids are durable now, so a rerun
/// against the same database must not collide with a previous run's saga
/// (trigger is idempotent per instance id).
fn unique_id(prefix: &str) -> String {
    format!("{prefix}-{}", uuid::Uuid::new_v4().simple())
}

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

async fn scenario_happy_path(store: &mire::EventStore) -> anyhow::Result<()> {
    println!("=== Scenario 1: Happy Path ===");
    let runner = build_runner(store);
    let instance_id = runner
        .trigger::<Booking>(&build_booking_request(&unique_id("b-happy")))
        .await?
        .expect("saga started");
    println!("  saga started: {instance_id}");
    print_journey(&runner, &instance_id).await?;

    let env = runner.load(&instance_id).await?.expect("instance exists");
    println!("  status: {:?}", env.status);
    assert!(matches!(env.status, SagaInstanceStatus::SucceededTerminal));
    Ok(())
}

async fn scenario_payment_timeout(store: &mire::EventStore) -> anyhow::Result<()> {
    println!("=== Scenario 2: Payment Timeout ===");

    // Build a runner WITHOUT the payment-capture handler registered. The
    // emit suspends awaiting an external worker; nobody responds; the
    // saga times out.
    let topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    let topology = strip_handler::<TravelBookingSaga>(topology, "saga.payment.capture-requested");
    let runner = PgSagaRunner::new(topology, store.clone());

    let instance_id = runner
        .trigger::<Booking>(&build_booking_request(&unique_id("b-timeout")))
        .await?
        .expect("saga started");
    println!("  saga started: {instance_id} (payment-capture handler is absent)");

    // Holds completed during trigger() (their handlers are registered).
    // The payment await row carries its deadline; tick past it.
    println!("  ticking timeouts past the payment deadline …");
    let future_now = chrono::Utc::now() + chrono::Duration::minutes(11);
    let timed_out = runner.tick_timeouts(future_now).await?;
    assert!(timed_out >= 1, "payment await should have expired");
    print_journey(&runner, &instance_id).await?;

    let env = runner.load(&instance_id).await?.expect("instance exists");
    println!("  status: {:?}", env.status);
    assert!(matches!(env.status, SagaInstanceStatus::FailedTerminal));
    Ok(())
}

async fn scenario_handler_returns_err(store: &mire::EventStore) -> anyhow::Result<()> {
    println!("=== Scenario 4: Handler Returns Err ===");
    let runner = build_runner(store);

    // booking_id starts with "fraud-" → handle_payment_capture returns Err.
    let request = BookingEvent::Requested {
        booking_id: BookingId::new(unique_id("fraud-001")),
        customer_id: CustomerId::new("alice@example.com"),
        flight_id: FlightId::new("LH-100"),
        hotel_id: HotelId::new("HILTON-OSL"),
        amount: Money::from_minor(150_000),
    };
    let instance_id = runner
        .trigger::<Booking>(&request)
        .await?
        .expect("saga started");
    println!("  saga started: {instance_id} (payment handler will return Err)");
    print_journey(&runner, &instance_id).await?;

    let env = runner.load(&instance_id).await?.expect("instance exists");
    println!("  status: {:?}", env.status);
    // The Err response failed the awaiting step → compensation cascade
    // ran via the registered release handlers.
    assert!(matches!(env.status, SagaInstanceStatus::FailedTerminal));
    Ok(())
}

async fn scenario_admin_cancel(store: &mire::EventStore) -> anyhow::Result<()> {
    println!("=== Scenario 3: Admin Cancel Mid-Confirm ===");

    // Strip the carrier-ticket handler so `confirm` stalls and we can fire
    // the signal at the right moment.
    let topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    let topology = strip_handler::<TravelBookingSaga>(topology, "saga.carrier.ticket-requested");
    let runner = PgSagaRunner::new(topology, store.clone());

    let instance_id = runner
        .trigger::<Booking>(&build_booking_request(&unique_id("b-admin")))
        .await?
        .expect("saga started");
    println!("  saga started: {instance_id} (carrier-ticket handler is absent)");

    // At this point: flight + hotel + payment all completed via auto-dispatch.
    // confirm activated and is awaiting carrier.ticket-issued.
    assert_eq!(
        runner.step_status(&instance_id, "confirm").await?,
        SagaStepStatus::Active,
        "confirm should be Active before signal"
    );

    println!("  admin pulls the plug while confirm is in-flight …");
    runner
        .signal(
            &instance_id,
            AdminForceCancel {
                reason: "fraud signals".into(),
            },
        )
        .await?;
    print_journey(&runner, &instance_id).await?;

    let env = runner.load(&instance_id).await?.expect("instance exists");
    println!("  status: {:?}", env.status);
    assert!(matches!(env.status, SagaInstanceStatus::FailedTerminal));
    Ok(())
}

// ---------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------

fn build_runner(store: &mire::EventStore) -> PgSagaRunner<TravelBookingSaga> {
    let topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    PgSagaRunner::new(topology, store.clone())
}

/// Print every emitted event in stream order — a human-readable trace of
/// the saga's full request/response journey, replayed from the durable
/// `StepEmitted` events on the instance's stream.
async fn print_journey(
    runner: &PgSagaRunner<TravelBookingSaga>,
    instance_id: &str,
) -> anyhow::Result<()> {
    for emit in runner.emitted(instance_id).await? {
        println!(
            "    [seq={:>2}] {:<40} from `{}`",
            emit.sequence, emit.event_type, emit.source_step,
        );
    }
    Ok(())
}

/// Test helper: remove a registered handler from a topology so the example
/// can simulate "external worker not responding" scenarios.
fn strip_handler<S: Saga>(
    mut topology: SagaBuilder<S>,
    request_event_type: &'static str,
) -> SagaBuilder<S> {
    topology.remove_request_handler(request_event_type);
    topology
}
