//! E2E tests for the auto-dispatch path: `.effect(...)` handlers
//! registered on the builder, runner drives the saga to terminal in a
//! single `trigger()` call.
//!
//! These tests exercise the typed handler model end-to-end:
//! - the `Req<R>` extractor pulls the typed request payload
//! - the response is auto-delivered (no manual deliver call)
//! - mixing handlers and external-worker emits works cleanly
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

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

mod common;

#[allow(unused)]
#[path = "../examples/travel_booking.rs"]
mod travel_booking;

use travel_booking::{
    AdminForceCancel, Booking, BookingEvent, BookingId, CustomerId, FlightId, HotelId,
    IssueCarrierTicket, Money, PaymentCaptureFailed, RequestPaymentCapture, TravelBookingSaga,
};

fn make_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),
    }
}

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

#[tokio::test]
async fn trigger_drives_saga_to_terminal_with_handlers() {
    let Some(store) = common::store().await else {
        return;
    };
    // With every handler registered, trigger() should drive the saga to
    // SucceededTerminal without any manual deliver() calls.
    let runner = build_runner_with_all_handlers(&store);
    let id = common::unique("auto-1");
    let instance_id = runner
        .trigger::<Booking>(&make_request(&id))
        .await
        .unwrap()
        .unwrap();

    let env = runner.load(&instance_id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);

    // Every forward step should be Completed.
    for step in ["flight", "hotel", "payment", "confirm"] {
        assert_eq!(
            runner.step_status(&instance_id, step).await.unwrap(),
            SagaStepStatus::Completed,
            "step {step} should be Completed"
        );
    }

    // Four typed emits, one per step.
    assert_eq!(runner.emitted(&instance_id).await.unwrap().len(), 4);
}

#[tokio::test]
async fn missing_handler_leaves_emit_dangling() {
    let Some(store) = common::store().await else {
        return;
    };
    // Strip the payment-capture handler. trigger() activates flight + hotel
    // (which auto-dispatch), then payment activates and emits, but no
    // handler responds — payment suspends Active awaiting an external
    // worker.
    //
    // Deadline pushed far out: `tick_timeouts` scans every instance of
    // this saga name, so a sibling test's tick must not be able to expire
    // this instance while we assert on its Active state.
    let mut topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    topology.remove_request_handler("saga.payment.capture-requested");
    let topology = topology.dep(travel_booking::PaymentDeadline(
        std::time::Duration::from_secs(7 * 24 * 3600),
    ));
    let runner = PgSagaRunner::new(topology, store.clone());

    let id = common::unique("auto-stuck");
    let instance_id = runner
        .trigger::<Booking>(&make_request(&id))
        .await
        .unwrap()
        .unwrap();

    // flight + hotel completed via auto-dispatch.
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Completed
    );
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Completed
    );

    // payment activated but is awaiting a response that won't come.
    assert_eq!(
        runner.step_status(&instance_id, "payment").await.unwrap(),
        SagaStepStatus::Active
    );

    // Saga is still Running (not terminal).
    assert_eq!(
        runner.status(&instance_id).await.unwrap(),
        Some(SagaInstanceStatus::Running)
    );

    // Emits include the dangling payment-capture-requested.
    let emitted = runner.emitted(&instance_id).await.unwrap();
    let emit_types: Vec<&str> = emitted.iter().map(|e| e.event_type.as_str()).collect();
    assert!(emit_types.contains(&"saga.payment.capture-requested"));
}

#[tokio::test]
async fn missing_handler_then_tick_timeouts_triggers_compensation() {
    let Some(store) = common::store().await else {
        return;
    };
    // Same setup, but tick past the payment deadline. The saga should
    // fail, compensation should cascade through the registered release
    // handlers and reach FailedTerminal.
    //
    // `tick_timeouts` is saga-WIDE (it scans every `travel-booking-saga`
    // await), and sibling tests (here and in `travel_booking_e2e`, which
    // shares this saga name) leave instances suspended on payment with the
    // DEFAULT 10-minute deadline. So we give THIS instance a deliberately
    // short (1s) payment deadline and tick only `now + 2s`: our instance
    // ripens, the default-10-minute siblings do not — the wide scan can't
    // sweep another test's in-flight instance.
    let mut topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    topology.remove_request_handler("saga.payment.capture-requested");
    let topology = topology.dep(travel_booking::PaymentDeadline(
        std::time::Duration::from_secs(1),
    ));
    let runner = PgSagaRunner::new(topology, store.clone());

    let id = common::unique("auto-timeout");
    let instance_id = runner
        .trigger::<Booking>(&make_request(&id))
        .await
        .unwrap()
        .unwrap();

    // Don't assert the returned count — even bounded, the scan is saga-wide.
    // Tick just past our own 1s deadline (not the siblings' 10-minute one).
    runner
        .tick_timeouts(chrono::Utc::now() + chrono::Duration::seconds(2))
        .await
        .unwrap();

    // Poll for terminal rather than asserting immediately: because
    // `tick_timeouts` is saga-WIDE, a concurrent tick from a sibling test
    // binary that shares this saga name (e.g. `travel_booking_e2e`) can win
    // our instance's lease and still be mid-cascade (uncommitted) at the
    // instant we load — leaving it transiently `Running`. Whichever tick
    // wins, the instance converges to `FailedTerminal` within a moment.
    let mut env = runner.load(&instance_id).await.unwrap().unwrap();
    for _ in 0..40 {
        if env.status == SagaInstanceStatus::FailedTerminal {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        env = runner.load(&instance_id).await.unwrap().unwrap();
    }
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    // Both holds released via auto-dispatch of the registered release handlers.
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Compensated
    );
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Compensated
    );
}

#[tokio::test]
async fn admin_signal_with_handlers_triggers_full_cascade() {
    let Some(store) = common::store().await else {
        return;
    };
    // Strip carrier-ticket so confirm stays Active while we signal.
    let mut topology = TravelBookingSaga::build(SagaBuilder::<TravelBookingSaga>::for_saga());
    topology.remove_request_handler("saga.carrier.ticket-requested");
    let runner = PgSagaRunner::new(topology, store.clone());

    let id = common::unique("auto-cancel");
    let instance_id = runner
        .trigger::<Booking>(&make_request(&id))
        .await
        .unwrap()
        .unwrap();

    // Flight + hotel + payment all completed via auto-dispatch.
    // confirm is Active waiting for ticket-issued (no handler).
    assert_eq!(
        runner.step_status(&instance_id, "confirm").await.unwrap(),
        SagaStepStatus::Active
    );

    runner
        .signal(
            &instance_id,
            AdminForceCancel {
                reason: "fraud signals".into(),
            },
        )
        .await
        .unwrap();

    // Compensation handlers ARE registered → cascade runs synchronously.
    let env = runner.load(&instance_id).await.unwrap().unwrap();
    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
    );
}

#[tokio::test]
async fn handler_returning_err_fails_step_and_triggers_compensation() {
    let Some(store) = common::store().await else {
        return;
    };
    // The payment-capture handler returns `Err(PaymentCaptureFailed)` when
    // booking_id starts with "fraud-". The runner should:
    //  1. emit the typed err event into the log
    //  2. fail the awaiting `payment` step with StepError::Failed
    //  3. fire compensation through the registered release handlers
    let runner = build_runner_with_all_handlers(&store);
    let id = common::unique("fraud-test");
    let request = 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),
    };
    let instance_id = runner.trigger::<Booking>(&request).await.unwrap().unwrap();

    let env = runner.load(&instance_id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);

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

    // Both holds compensated.
    assert_eq!(
        runner.step_status(&instance_id, "flight").await.unwrap(),
        SagaStepStatus::Compensated,
    );
    assert_eq!(
        runner.step_status(&instance_id, "hotel").await.unwrap(),
        SagaStepStatus::Compensated,
    );

    // The typed err event is in the emitted log.
    let emitted = runner.emitted(&instance_id).await.unwrap();
    let types: Vec<&str> = emitted.iter().map(|e| e.event_type.as_str()).collect();
    assert!(types.contains(&"payment.capture-failed"));

    // And we can decode it back to the typed struct.
    let err_emit = emitted
        .iter()
        .find(|e| e.event_type == "payment.capture-failed")
        .unwrap();
    let typed: PaymentCaptureFailed = serde_json::from_value(err_emit.payload.clone()).unwrap();
    assert_eq!(typed.booking_id.0, id);
    assert_eq!(typed.reason, "fraud check declined");
}

#[tokio::test]
async fn typed_outputs_thread_through_downstream_steps() {
    let Some(store) = common::store().await else {
        return;
    };
    // request_payment uses Output<FlightReserved> + Output<HotelReserved>
    // to populate its emit; confirm uses Output<PaymentSettled>. Both must
    // appear in the emitted payloads, proving the runner correctly
    // serialises step outputs and re-deserialises them for downstream
    // From<T> extractors.
    let runner = build_runner_with_all_handlers(&store);
    let id = common::unique("threading-1");
    let instance_id = runner
        .trigger::<Booking>(&make_request(&id))
        .await
        .unwrap()
        .unwrap();

    let emitted = runner.emitted(&instance_id).await.unwrap();
    let payment_req: RequestPaymentCapture = emitted
        .iter()
        .find(|e| e.event_type == "saga.payment.capture-requested")
        .map(|e| serde_json::from_value(e.payload.clone()).unwrap())
        .expect("payment-capture-requested in journey");
    // The flight + hotel hold IDs the in-process handlers minted should
    // have flowed through the flight + hotel step outputs into the
    // payment request.
    assert!(payment_req.flight_hold.0.starts_with("fh-LH-100"));
    assert!(payment_req.hotel_hold.0.starts_with("hh-HILTON-OSL"));
    assert_eq!(payment_req.booking_id.0, id);

    let ticket_req: IssueCarrierTicket = emitted
        .iter()
        .find(|e| e.event_type == "saga.carrier.ticket-requested")
        .map(|e| serde_json::from_value(e.payload.clone()).unwrap())
        .expect("ticket-requested in journey");
    assert_eq!(ticket_req.booking_id.0, id);
    // The payment_id minted by handle_payment_capture should flow into the
    // ticket request via Output<PaymentSettled>.
    assert_eq!(ticket_req.payment_id.0, format!("p-{id}"));
}
