//! The instance-lease heartbeat (effects-vs-commit §11 / Phase 3).
//!
//! A step/request handler is an arbitrary user `async fn`. One that runs
//! longer than `lease_ttl` would, without a heartbeat, let the instance
//! lease lapse — and a second worker could then steal it and run the same
//! effects concurrently. OCC fences the log, not the side effects. The
//! runner heartbeats the lease while an effect runs, so a slow-but-alive
//! worker keeps exclusivity.
//!
//! This test drives a slow in-process handler under `deliver`'s lease and
//! proves a competing worker can never acquire the same instance's lease
//! across several TTLs, and that the effect runs exactly once.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use mire::{Aggregate, EventData};
use mire_sagas::{Emit, PgSagaRunner, Req, Saga, SagaBuilder, Start, State, task};
use serde::{Deserialize, Serialize};

mod common;

static WORK_RUNS: AtomicU64 = AtomicU64::new(0);

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

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "hb.source")]
struct SourceEvent {
    id: String,
}

impl Aggregate for Source {
    type Event = SourceEvent;
    fn stream_category() -> &'static str {
        "hb-source"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

// Step "gate" emits to an EXTERNAL worker (no handler) → suspends awaiting
// GateOpened. So trigger() returns quickly with the saga parked.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "hb.gate-req")]
struct GateReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "hb.gate-opened")]
struct GateOpened;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct GateOut;

async fn gate(State(s): State<St>) -> Emit {
    Emit::event(GateReq { id: s.id.clone() })
}

// Step "work" runs after gate. Its in-process handler is SLOW (sleeps well
// past the lease TTL), simulating a slow external call. It runs under
// deliver()'s lease — the heartbeat must keep that lease alive.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "hb.work-req")]
struct WorkReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "hb.work-done")]
struct WorkDone;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct WorkOut;

async fn work(State(s): State<St>) -> Emit {
    Emit::event(WorkReq { id: s.id.clone() })
}

/// The slow effect: holds for ~3× the lease TTL used by the test.
async fn handle_work(Req(_): Req<WorkReq>) -> WorkDone {
    WORK_RUNS.fetch_add(1, Ordering::SeqCst);
    tokio::time::sleep(Duration::from_millis(3000)).await;
    WorkDone
}

struct HbSaga;
impl Saga for HbSaga {
    type State = St;
    fn name() -> &'static str {
        "hb-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step("gate", task(gate).on_response(|_: GateOpened| GateOut))
        .step(
            "work",
            task(work)
                .after("gate")
                .on_response(|_: WorkDone| WorkOut)
                .terminal_success(),
        )
        .handle(handle_work)
    }
}

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

    // 1s TTL (renewal at ~333ms, ~667ms margin) — generous enough that the
    // heartbeat keeps up even when the full parallel suite saturates the
    // runtime and delays the renewal tick. Smaller TTLs flaked under load.
    let ttl = Duration::from_millis(1000);
    let runner = PgSagaRunner::new(
        HbSaga::build(SagaBuilder::<HbSaga>::for_saga()),
        store.clone(),
    )
    .with_worker_id("worker-A")
    .with_lease_ttl(ttl);
    let id = common::unique("hb");

    // Trigger: gate emits to an external worker and suspends. Quick.
    runner
        .trigger::<Source>(&SourceEvent { id: id.clone() })
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        runner.step_status(&id, "gate").await.unwrap(),
        mire_sagas::SagaStepStatus::Active
    );

    // Deliver the gate response on worker A in the background. This holds
    // the instance lease while `work`'s slow handler runs (~3s, 3× TTL),
    // heartbeating it the whole time.
    let deliver = {
        let runner = PgSagaRunner::new(
            HbSaga::build(SagaBuilder::<HbSaga>::for_saga()),
            store.clone(),
        )
        .with_worker_id("worker-A")
        .with_lease_ttl(ttl);
        let id = id.clone();
        tokio::spawn(async move { runner.deliver(&id, &GateOpened).await })
    };

    // Give the deliver a moment to acquire the lease and enter the slow
    // handler, then a competing worker (B) tries to steal the SAME
    // instance's lease repeatedly across >2 TTLs. The heartbeat must keep
    // A's lease alive, so every attempt is Contended.
    tokio::time::sleep(Duration::from_millis(300)).await;
    let mut stolen = false;
    // ~2.4s of probing (24 × 100ms) from t≈300ms — spans >2 TTLs, well within
    // the 3s slow handler.
    for _ in 0..24 {
        let outcome =
            mire_sagas::instance_lease::acquire(store.pool(), HbSaga::name(), &id, "worker-B", ttl)
                .await
                .unwrap();
        if matches!(
            outcome,
            mire_sagas::instance_lease::AcquireOutcome::Acquired { .. }
        ) {
            stolen = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    assert!(
        !stolen,
        "a competing worker stole the lease during a slow handler — heartbeat failed"
    );

    // The deliver completes; the saga succeeds and the slow effect ran once.
    let consumed = deliver.await.unwrap().unwrap();
    assert!(consumed, "gate response should have been consumed");
    assert_eq!(
        WORK_RUNS.load(Ordering::SeqCst),
        1,
        "the slow effect must run exactly once"
    );
    assert!(runner.is_succeeded(&id).await.unwrap());
}
