//! Behavioural pin for the node-local Lamport clock surfaced on
//! `AutomergeBackend` (peat-mesh#192 — replaces the `peat-cli` wall-clock
//! Lamport-proxy in `peat-node/crates/peat-cli/src/cli/delete.rs` with a
//! peat-mesh-managed source).
//!
//! What this test pins:
//! - `next_lamport()` returns strictly monotonic values on a single node,
//!   even under concurrent callers.
//! - `current_lamport()` is a non-advancing read.
//! - `observe_lamport(observed)` jumps the clock so a subsequent
//!   `next_lamport()` is strictly greater than `observed` — the Lamport
//!   receive-side rule `L_local := max(L_local, observed) + 1`.
//! - `observe_lamport` with a smaller `observed` than the local clock
//!   is a no-op (never regresses).
//! - The clock seeds high enough from `SystemTime::now()` nanos that
//!   restarts of the same node resume past any previously-issued
//!   value with overwhelming probability.

#![cfg(feature = "automerge-backend")]

use std::collections::HashSet;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;

use peat_mesh::security::FormationKey;
use peat_mesh::sync::{AutomergeBackend, AutomergeBackendConfig};
use tempfile::TempDir;

async fn build_backend() -> (Arc<AutomergeBackend>, TempDir) {
    let dir = TempDir::new().expect("tempdir");
    let shared_key = FormationKey::generate_secret();
    let mut cfg = AutomergeBackendConfig::default();
    cfg.data_dir = dir.path().to_path_buf();
    cfg.formation_id = "lamport-test".to_string();
    cfg.base64_shared_key = shared_key;
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    let backend = AutomergeBackend::with_iroh(cfg).await.expect("with_iroh");
    (backend, dir)
}

/// Rebuild a backend against an existing data_dir — the persistence-test
/// fixture. Reuses `dir` so the redb file lives across the prior
/// backend's teardown; binds to a fresh ephemeral iroh port so the
/// rebind can't collide with the prior endpoint.
///
/// Single-attempt construction: the caller must have run
/// `prior_backend.shutdown_and_release().await` before dropping the
/// prior backend, which awaits iroh's Router shutdown future and
/// guarantees the redb file lock is released by the time `drop`
/// returns. Without that sequence the rebuild can hit a same-process
/// lock-contention race; with it, this function never needs a retry.
async fn rebuild_backend_at(dir_path: std::path::PathBuf) -> Arc<AutomergeBackend> {
    let shared_key = FormationKey::generate_secret();
    let mut cfg = AutomergeBackendConfig::default();
    cfg.data_dir = dir_path;
    cfg.formation_id = "lamport-test".to_string();
    cfg.base64_shared_key = shared_key;
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    AutomergeBackend::with_iroh(cfg)
        .await
        .expect("with_iroh (rebuild)")
}

#[tokio::test]
async fn next_lamport_is_strictly_monotonic_on_single_caller() {
    let (backend, _dir) = build_backend().await;

    let a = backend.next_lamport();
    let b = backend.next_lamport();
    let c = backend.next_lamport();
    assert!(a < b, "next_lamport must advance: got {a} then {b}");
    assert!(b < c, "next_lamport must advance: got {b} then {c}");
    // Sanity: the increment is exactly 1 between consecutive single-
    // caller calls (no concurrent advancers in this test).
    assert_eq!(b, a + 1);
    assert_eq!(c, b + 1);
}

#[tokio::test]
async fn current_lamport_does_not_advance() {
    let (backend, _dir) = build_backend().await;

    let initial = backend.current_lamport();
    let again = backend.current_lamport();
    let third = backend.current_lamport();
    assert_eq!(initial, again, "current_lamport must not advance the clock");
    assert_eq!(again, third);

    // A `next_lamport` call between two `current_lamport` reads DOES
    // change the read value — proves `current_lamport` reads live state,
    // not a stale cached snapshot.
    let before_next = backend.current_lamport();
    let issued = backend.next_lamport();
    let after_next = backend.current_lamport();
    assert!(
        issued > before_next,
        "next_lamport must return a value greater than the prior current"
    );
    assert!(
        after_next >= issued,
        "current_lamport must reflect the advance from next_lamport \
         (got before={before_next} issued={issued} after={after_next})"
    );
}

#[tokio::test]
async fn observe_lamport_jumps_clock_past_observed() {
    let (backend, _dir) = build_backend().await;

    // Pick a value far enough above the wall-clock seed that we can
    // unambiguously observe the jump — adding a billion is well within
    // u64 range and avoids any wraparound concern with the nanos seed.
    let observed = backend.current_lamport().saturating_add(1_000_000_000);
    backend.observe_lamport(observed);

    let after = backend.next_lamport();
    assert!(
        after > observed,
        "next_lamport after observe_lamport({observed}) must be strictly greater; got {after}"
    );
}

#[tokio::test]
async fn observe_lamport_does_not_regress_clock() {
    let (backend, _dir) = build_backend().await;

    // Advance the local clock to a known high-water mark.
    let high = backend.next_lamport();

    // Observe something LOWER than the current clock — must be a no-op.
    backend.observe_lamport(high.saturating_sub(100));

    // Subsequent next_lamport must still be strictly above `high`,
    // not regressed back near the lower observed value.
    let after = backend.next_lamport();
    assert!(
        after > high,
        "observe_lamport with a lower value must NOT regress the clock \
         (high={high}, after={after})"
    );
}

#[tokio::test]
async fn observe_lamport_u64_max_is_capped_visible_via_current_lamport() {
    // peat-mesh#198 QA hardening (round-2 [SUGGESTION]): close the
    // loop on the public read-side accessor. A consumer calling
    // `observe_lamport(u64::MAX)` followed by `current_lamport()`
    // must see the capped value `u64::MAX - 1`, not `u64::MAX`.
    // Pairs with the in-tree `observe_atomic_caps_advance_at_u64_max_minus_one`
    // unit pin (which exercises the helper directly) and the
    // `tombstone_receive_caps_wired_clock_against_u64_max` E2E pin
    // (which exercises the wire-up); this test pins the public
    // consumer-facing read path.
    let (backend, _dir) = build_backend().await;

    backend.observe_lamport(u64::MAX);
    let current = backend.current_lamport();
    assert_eq!(
        current,
        u64::MAX - 1,
        "observe_lamport(u64::MAX) must leave current_lamport at u64::MAX - 1, not u64::MAX"
    );

    // And the next next_lamport returns u64::MAX (NOT 0 — the
    // wrap-to-zero pathology the cap forecloses).
    let next = backend.next_lamport();
    assert_eq!(
        next,
        u64::MAX,
        "next_lamport after capped observe must return u64::MAX, not wrap to 0"
    );
}

#[tokio::test]
async fn next_lamport_is_unique_under_concurrent_callers() {
    let (backend, _dir) = build_backend().await;

    // Spawn N concurrent callers and assert every returned value is unique.
    // Each thread calls next_lamport() M times; the union must have N*M
    // distinct entries.
    const TASKS: usize = 8;
    const PER_TASK: usize = 1024;

    let mut handles = Vec::with_capacity(TASKS);
    for _ in 0..TASKS {
        let backend = Arc::clone(&backend);
        handles.push(tokio::spawn(async move {
            let mut local = Vec::with_capacity(PER_TASK);
            for _ in 0..PER_TASK {
                local.push(backend.next_lamport());
            }
            local
        }));
    }

    let mut all = HashSet::with_capacity(TASKS * PER_TASK);
    for h in handles {
        let local = h.await.expect("task join");
        for v in local {
            assert!(
                all.insert(v),
                "next_lamport must return unique values across concurrent callers; \
                 duplicate observed: {v}"
            );
        }
    }
    assert_eq!(all.len(), TASKS * PER_TASK);
}

#[tokio::test]
async fn lamport_seeded_above_zero_from_wall_clock() {
    let (backend, _dir) = build_backend().await;

    // The construction-time seed is `SystemTime::now()` nanos, with a
    // floor of 1. The hard contract this pin enforces is the floor:
    // `next_lamport` must never return 0, which the CLI tombstone path
    // relies on (zero would be ambiguous with "unset"). The seed value
    // itself is *not* asserted with a high threshold here — a tactical
    // device booted with no time source (battery-less RTC, no NTP/GPS
    // sync yet) can legitimately have `SystemTime::now()` near epoch
    // on first boot, and a tighter threshold would spuriously fail
    // there in a way that would mask whatever a real regression looks
    // like. The persistence + cross-restart properties are tracked as
    // follow-ups (see field doc-comment).
    let issued = backend.next_lamport();
    assert!(
        issued >= 1,
        "next_lamport must never return 0 — the wall-clock seed has a \
         floor of 1 and the increment yields ≥1; got {issued}"
    );
}

#[tokio::test]
async fn lamport_highwater_persists_across_backend_restart() {
    // Closes peat-mesh#195. End-to-end pin on the redb-keyed
    // `lamport_highwater` round-trip: build a backend, advance the
    // clock past its wall-clock seed, force a flush, drop the backend,
    // rebuild against the same data dir, assert the resumed seed is
    // strictly greater than the prior high-water.

    let dir = TempDir::new().expect("tempdir");
    let dir_path = dir.path().to_path_buf();

    // First backend incarnation: build, advance, flush.
    let shared_key = FormationKey::generate_secret();
    let mut cfg = AutomergeBackendConfig::default();
    cfg.data_dir = dir_path.clone();
    cfg.formation_id = "persist-test".to_string();
    cfg.base64_shared_key = shared_key;
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    let backend = AutomergeBackend::with_iroh(cfg)
        .await
        .expect("with_iroh (initial)");

    // Advance the clock by a large jump so the rebuild's wall-clock
    // seed cannot accidentally pass the assertion. 1e10 is ~10 seconds
    // of nanoseconds — strictly above any plausible sub-test wall-
    // clock advance during the test's runtime.
    backend.observe_lamport(backend.current_lamport().saturating_add(10_000_000_000));
    let high_before_drop = backend.next_lamport();

    // peat-mesh#197 shutdown sequence: explicit flush + await iroh
    // Router shutdown so the redb file lock releases deterministically
    // before we attempt the rebuild. Without this, the iroh Router's
    // background tokio tasks briefly outlive `AutomergeBackend::drop`
    // and hold the chain of Arcs that keeps the redb Database open;
    // the rebuild then hits a same-process lock-contention race. The
    // explicit `shutdown_and_release` awaits the Router's shutdown
    // future and guarantees the chain is fully released by the time
    // it returns. Pairs with the bounded shutdown timeout
    // (`NetworkedIrohBlobStore::shutdown_timeout`, 5s default).
    backend
        .shutdown_and_release()
        .await
        .expect("shutdown_and_release awaits iroh Router shutdown cleanly");
    drop(backend);

    let resumed = rebuild_backend_at(dir_path).await;
    let resumed_current = resumed.current_lamport();

    assert!(
        resumed_current >= high_before_drop,
        "rebuilt backend's resumed seed must be ≥ the prior run's high-water: \
         resumed_current={resumed_current}, high_before_drop={high_before_drop}"
    );

    let next = resumed.next_lamport();
    assert!(
        next > high_before_drop,
        "first next_lamport after restart must exceed prior high-water: \
         next={next}, high_before_drop={high_before_drop}"
    );

    // Clean shutdown of the resumed backend so the test exits without
    // racing the TempDir cleanup against any still-running iroh
    // tasks. Pairs with the prior `shutdown_and_release` on the first
    // incarnation — the persistence test now demonstrates the full
    // build → shutdown → rebuild → shutdown lifecycle.
    resumed
        .shutdown_and_release()
        .await
        .expect("shutdown_and_release on resumed backend");
    // Drop the resumed backend explicitly so the temp dir's cleanup
    // doesn't race the still-spawned flush task on test exit.
    drop(resumed);
}

#[tokio::test]
async fn shutdown_and_release_is_idempotent() {
    // peat-mesh#197 QA round-5: the public API documents idempotent
    // semantics — the first call runs the full teardown; subsequent
    // calls short-circuit with Ok(()) without touching the closed
    // Router. Pins the defensive AtomicBool guard so a future refactor
    // that drops it (and re-introduces the second-call-misleading-error
    // sharp edge) fails this test.
    let (backend, _dir) = build_backend().await;

    // First call performs the teardown.
    backend
        .shutdown_and_release()
        .await
        .expect("first shutdown_and_release succeeds");

    // Second call must short-circuit cleanly — no panic, no error,
    // no second Router::shutdown attempt against an already-closed
    // endpoint.
    backend
        .shutdown_and_release()
        .await
        .expect("second shutdown_and_release short-circuits cleanly");

    // And a third for good measure.
    backend
        .shutdown_and_release()
        .await
        .expect("third shutdown_and_release short-circuits cleanly");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shutdown_and_release_is_safe_under_concurrent_callers() {
    // peat-mesh#197 QA round-6: the CAS-on-AtomicBool guard's *whole
    // purpose* is the concurrent case — signal-handling races, two
    // shutdown hooks racing each other, etc. Sequential idempotency
    // would have worked with a simpler `Mutex<bool>` or
    // let-second-call-fail. What makes the `compare_exchange` the
    // right primitive is that exactly one caller can win and the
    // others must short-circuit cleanly without invoking another
    // `Router::shutdown` against the in-flight teardown.
    //
    // Runtime flavor (peat-mesh#197 QA round-7): `multi_thread`
    // with 2 workers. Under the default `current_thread` flavor,
    // `tokio::join!` polls both futures sequentially on the single
    // worker — the first hits the CAS synchronously, wins, and the
    // second sees the flag set and short-circuits without ever
    // racing. That captures the sequential-idempotency contract
    // (already covered by `shutdown_and_release_is_idempotent`) but
    // not the genuine concurrent-CAS-race contract this test is
    // supposed to pin. `multi_thread` gives both arms real
    // OS-thread parallelism, so the CAS actually races on every
    // run.
    //
    // Non-determinism note: which caller wins the CAS is the OS
    // scheduler's choice. The pin asserts that BOTH arms return
    // `Ok(())` regardless of who wins, since that's the contract
    // the guard makes. If a future regression turned the CAS into a
    // racing pair where the loser tried to re-shutdown, the loser's
    // call would error out and this test would catch it on at least
    // some fraction of runs.
    let (backend, _dir) = build_backend().await;

    let backend2 = Arc::clone(&backend);
    let (a, b) = tokio::join!(
        async move { backend.shutdown_and_release().await },
        async move { backend2.shutdown_and_release().await },
    );

    a.expect("concurrent arm A must return Ok regardless of CAS winner");
    b.expect("concurrent arm B must return Ok regardless of CAS winner");
}

#[tokio::test]
async fn flush_lamport_highwater_is_idempotent_and_safe_on_unchanged_clock() {
    let (backend, _dir) = build_backend().await;
    backend
        .flush_lamport_highwater()
        .expect("first flush succeeds");
    backend
        .flush_lamport_highwater()
        .expect("second flush of the same value succeeds");
    let _ = backend.next_lamport();
    backend
        .flush_lamport_highwater()
        .expect("flush after advance succeeds");
}

#[tokio::test]
async fn write_lamport_highwater_does_not_regress_against_lower_writes() {
    // QA review on #197 surfaced a race between the periodic flush
    // task and an explicit `flush_lamport_highwater` call: if the
    // periodic load happened first (T1 → V), an explicit flush
    // committed `V+N` at T4, then the periodic write committed `V`
    // at T5 — regressing the persisted high-water below an
    // already-issued value. This pin closes the loop on the fix:
    // `AutomergeStore::write_lamport_highwater` reads existing
    // inside the same redb transaction and commits only when
    // `value > existing`, so any lower-valued write is a no-op at
    // the storage layer regardless of call ordering at the caller.
    //
    // Note on naming: the redb write-lock serialises writers, so
    // there's no actual concurrent interleaving in this test — the
    // storage layer's lock is what enforces the read-then-compare
    // pair's atomicity. The invariant being pinned is
    // "lower-valued writes do not regress the persisted value",
    // which is exactly what defends against the documented race
    // pattern.
    use peat_mesh::storage::AutomergeStore;

    let dir = TempDir::new().expect("tempdir");
    let store = AutomergeStore::open(dir.path().join("automerge")).expect("open");

    // Simulate the regression sequence directly at the store layer:
    // write a high value, then write a lower one. The lower write
    // must be a no-op against the persisted state.
    store
        .write_lamport_highwater(1_000_000)
        .expect("write high");
    assert_eq!(
        store.read_lamport_highwater().expect("read after high"),
        Some(1_000_000)
    );

    store
        .write_lamport_highwater(500_000)
        .expect("write low must succeed (Ok), but not regress");
    assert_eq!(
        store.read_lamport_highwater().expect("read after low"),
        Some(1_000_000),
        "monotonic write: lower value must not regress the persisted high-water"
    );

    // And a higher subsequent write advances correctly — the
    // monotonic clamp only blocks regression, not advancement.
    store
        .write_lamport_highwater(2_000_000)
        .expect("write higher");
    assert_eq!(
        store.read_lamport_highwater().expect("read after higher"),
        Some(2_000_000)
    );
}
