//! [`SagaWorker`] autonomously drives timeouts (and signal retention)
//! without the host calling `tick_timeouts` itself.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::time::Duration;

use mire::{Aggregate, EventData};
use mire_sagas::{
    Emit, Output, PgSagaRunner, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus, SagaWorker,
    Start, State, comp, task,
};
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;

mod common;

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

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

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

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

// "hold" completes in-process so there's something to compensate when the
// downstream step times out.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "sw.hold-req")]
struct HoldReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "sw.held")]
struct Held;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HoldDone;

fn hold(State(s): State<St>) -> Emit {
    Emit::event(HoldReq { id: s.id.clone() })
}
async fn handle_hold(mire_sagas::Req(_): mire_sagas::Req<HoldReq>) -> Held {
    Held
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "sw.release-req")]
struct ReleaseReq;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "sw.released")]
struct Released;

fn release_hold(Output(_): Output<HoldDone>) -> Emit {
    Emit::event(ReleaseReq)
}
async fn handle_release(mire_sagas::Req(_): mire_sagas::Req<ReleaseReq>) -> Released {
    Released
}

// "pay" emits to an EXTERNAL worker (no handler) → suspends with a short
// deadline. Nobody answers, so it must time out.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "sw.pay-req")]
struct PayReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "sw.paid")]
struct Paid;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PayDone;

fn pay(State(s): State<St>) -> Emit {
    // Short, runtime-chosen deadline rides on the Emit.
    Emit::event(PayReq { id: s.id.clone() }).timeout(Duration::from_millis(200))
}

struct WorkerSaga;
impl Saga for WorkerSaga {
    type State = St;
    fn name() -> &'static str {
        "saga-worker-test"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Booking, _>(|e: &BookingRequested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "hold",
            task(hold)
                .on_response(|_: Held| HoldDone)
                .compensate(comp(release_hold).on_response::<Released>()),
        )
        .step(
            "pay",
            task(pay)
                .after("hold")
                .on_response(|_: Paid| PayDone)
                .terminal_success(),
        )
        .effect(handle_hold)
        .effect(handle_release)
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn worker_drives_timeout_to_compensation_without_host_ticks() {
    let Some(store) = common::store().await else {
        return;
    };

    // Trigger on a plain runner: hold completes in-process; pay suspends
    // with a 200ms deadline awaiting an external response that never comes.
    let runner = PgSagaRunner::new(
        WorkerSaga::build(SagaBuilder::<WorkerSaga>::for_saga()),
        store.clone(),
    );
    let id = common::unique("sw");
    runner
        .trigger::<Booking>(&BookingRequested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        runner.step_status(&id, "hold").await.unwrap(),
        SagaStepStatus::Completed
    );
    assert_eq!(
        runner.step_status(&id, "pay").await.unwrap(),
        SagaStepStatus::Active
    );
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::Running)
    );

    // Start a worker (separate runner, same store) — and DO NOT call
    // tick_timeouts ourselves. The worker must expire the deadline, fail
    // `pay`, and compensate `hold`, all on its own.
    let cancel = CancellationToken::new();
    let worker = SagaWorker::new(PgSagaRunner::new(
        WorkerSaga::build(SagaBuilder::<WorkerSaga>::for_saga()),
        store.clone(),
    ))
    .poll_interval(Duration::from_millis(50));
    let handle = {
        let cancel = cancel.clone();
        tokio::spawn(async move { worker.run(cancel).await })
    };

    // Poll for the instance to reach terminal, purely via the worker.
    let mut terminal = false;
    for _ in 0..100 {
        if runner.is_terminal(&id).await.unwrap() {
            terminal = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    cancel.cancel();
    handle.await.unwrap().unwrap();

    assert!(
        terminal,
        "worker should have driven the timed-out saga to terminal"
    );
    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
    assert_eq!(
        runner.step_status(&id, "pay").await.unwrap(),
        SagaStepStatus::Failed
    );
    assert_eq!(
        runner.step_status(&id, "hold").await.unwrap(),
        SagaStepStatus::Compensated,
        "the completed upstream must be compensated by the worker-driven cascade"
    );
}
