//! Per-instance processing lease primitive (effects-vs-commit §11):
//! mutual exclusion (a live holder blocks others), strict exclusion (no
//! same-worker re-acquire of a live lease), release, takeover after expiry,
//! and the **fence token** (phase 7) that makes a steal explicit so a
//! stolen-from holder — even one reusing the same `worker_id` after a restart
//! — can't masquerade as its own stale lease. Gated on `DATABASE_URL`.

use std::time::Duration;

use mire::EventStore;
use mire_sagas::instance_lease::{AcquireOutcome, acquire, heartbeat, release};
use sqlx::PgPool;

async fn maybe_store() -> Option<EventStore> {
    let url = std::env::var("DATABASE_URL")
        .ok()
        .filter(|s| !s.is_empty())?;
    let pool = PgPool::connect(&url).await.expect("connect");
    let store = EventStore::new(pool);
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");
    Some(store)
}

/// The fence of an `Acquired` outcome, or panic if contended.
fn fence_of(o: AcquireOutcome) -> i64 {
    match o {
        AcquireOutcome::Acquired { fence } => fence,
        AcquireOutcome::Contended => panic!("expected Acquired, got Contended"),
    }
}

#[tokio::test]
async fn lease_excludes_then_allows_takeover() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool();
    let saga = "lease-test";
    let inst = format!("inst-{}", uuid::Uuid::new_v4());
    let ttl = Duration::from_secs(30);

    // A acquires (a fresh lease starts at fence 1).
    let fa = fence_of(acquire(pool, saga, &inst, "worker-A", ttl).await.unwrap());
    assert_eq!(fa, 1, "a fresh lease starts at fence 1");
    // B is excluded while A holds a live lease.
    assert_eq!(
        acquire(pool, saga, &inst, "worker-B", ttl).await.unwrap(),
        AcquireOutcome::Contended
    );
    // STRICT exclusion: even A cannot re-acquire its own *live* lease — this
    // prevents two concurrent calls on one runner (same worker_id) from both
    // processing the instance. A extends its lease via `heartbeat`, not
    // `acquire`.
    assert_eq!(
        acquire(pool, saga, &inst, "worker-A", ttl).await.unwrap(),
        AcquireOutcome::Contended
    );
    // A heartbeats successfully at its fence; a non-holder's heartbeat is a
    // no-op (wrong worker_id), and so is a heartbeat at the wrong fence.
    assert!(
        heartbeat(pool, saga, &inst, "worker-A", fa, ttl)
            .await
            .unwrap()
    );
    assert!(
        !heartbeat(pool, saga, &inst, "worker-B", fa, ttl)
            .await
            .unwrap()
    );
    assert!(
        !heartbeat(pool, saga, &inst, "worker-A", fa + 99, ttl)
            .await
            .unwrap(),
        "a heartbeat at a fence we don't hold must fail"
    );

    // A releases → B can now take over, and the steal BUMPS the fence.
    release(pool, saga, &inst, "worker-A", fa).await.unwrap();
    let fb = fence_of(acquire(pool, saga, &inst, "worker-B", ttl).await.unwrap());
    assert!(fb > fa, "stealing the lease bumps the fence ({fb} > {fa})");

    // Different instance is independent (its own fresh fence).
    let other = format!("inst-{}", uuid::Uuid::new_v4());
    assert_eq!(
        fence_of(acquire(pool, saga, &other, "worker-A", ttl).await.unwrap()),
        1
    );
}

#[tokio::test]
async fn expired_lease_is_stealable() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool();
    let saga = "lease-test";
    let inst = format!("inst-{}", uuid::Uuid::new_v4());

    // A acquires with a short TTL, then "dies" (never releases/heartbeats).
    // TTL is comfortably larger than test-execution latency so the
    // "immediately excluded" check below is not racy.
    let fa = fence_of(
        acquire(pool, saga, &inst, "worker-A", Duration::from_millis(500))
            .await
            .unwrap(),
    );
    // Immediately, B is excluded.
    assert_eq!(
        acquire(pool, saga, &inst, "worker-B", Duration::from_secs(30))
            .await
            .unwrap(),
        AcquireOutcome::Contended
    );
    // After the TTL lapses, B takes over (at-least-once) at a bumped fence.
    tokio::time::sleep(Duration::from_millis(700)).await;
    let fb = fence_of(
        acquire(pool, saga, &inst, "worker-B", Duration::from_secs(30))
            .await
            .unwrap(),
    );
    assert!(fb > fa, "takeover bumps the fence");
    // And A — now the stale holder — can no longer heartbeat its lost lease.
    assert!(
        !heartbeat(pool, saga, &inst, "worker-A", fa, Duration::from_secs(30))
            .await
            .unwrap()
    );
}

/// The fence's reason for existing: a worker that crashes and restarts under
/// the **same `worker_id`** must not be able to act on its old, stolen-from
/// lease. The old in-flight task heartbeats the *old* fence; the fence guard
/// rejects it even though the worker_id matches — which a worker_id-only lease
/// could not do.
#[tokio::test]
async fn fence_rejects_a_stale_same_worker_after_restart() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool();
    let saga = "lease-test";
    let inst = format!("inst-{}", uuid::Uuid::new_v4());

    // The original hold, with a short TTL it then lets lapse (a crash).
    let old = fence_of(
        acquire(pool, saga, &inst, "worker-A", Duration::from_millis(300))
            .await
            .unwrap(),
    );
    tokio::time::sleep(Duration::from_millis(450)).await;

    // The SAME worker_id restarts and re-acquires its own expired lease —
    // a steal, so the fence advances.
    let new = fence_of(
        acquire(pool, saga, &inst, "worker-A", Duration::from_secs(30))
            .await
            .unwrap(),
    );
    assert!(
        new > old,
        "a same-worker re-acquire of an expired lease bumps the fence"
    );

    // The zombie (the old in-flight task at the OLD fence) cannot heartbeat,
    // even though its worker_id matches the current holder. The fresh hold can.
    assert!(
        !heartbeat(pool, saga, &inst, "worker-A", old, Duration::from_secs(30))
            .await
            .unwrap(),
        "the old fence must be rejected — the fence is what makes this safe"
    );
    assert!(
        heartbeat(pool, saga, &inst, "worker-A", new, Duration::from_secs(30))
            .await
            .unwrap(),
        "the current hold's fence still heartbeats"
    );
    // The zombie's release at the old fence is a no-op (doesn't free the live lease).
    release(pool, saga, &inst, "worker-A", old).await.unwrap();
    assert!(
        heartbeat(pool, saga, &inst, "worker-A", new, Duration::from_secs(30))
            .await
            .unwrap(),
        "a stale-fence release must not free the current hold"
    );
}
