//! LockManagerTest — stress tests ported to Rust.
//!
//! Covers: non-blocking acquire, lock demote, lock release,
//! n_total_locks, get_lock_info, get_stats, sharing registry,
//! stealing, lock timeout, concurrent grant/wait.

use std::sync::Arc;
use std::thread;
use std::time::Duration;

use noxu_txn::{LockGrantType, LockManager, LockStats, LockType, TxnError};

// ─── helpers ──────────────────────────────────────────────────────────────────

fn lm() -> Arc<LockManager> {
    Arc::new(LockManager::with_lock_timeout(500))
}

#[allow(dead_code)]
fn lm_no_timeout() -> Arc<LockManager> {
    Arc::new(LockManager::with_lock_timeout(0))
}

fn acquire(
    lm: &LockManager,
    lsn: u64,
    locker: i64,
    ty: LockType,
) -> LockGrantType {
    lm.lock(lsn, locker, ty, false, false).unwrap()
}

fn acquire_nb(
    lm: &LockManager,
    lsn: u64,
    locker: i64,
    ty: LockType,
) -> Result<LockGrantType, TxnError> {
    lm.lock(lsn, locker, ty, true, false)
}

// ─── 1. Basic grant types ─────────────────────────────────────────────────────

#[test]
fn lock_new_grant_type() {
    let lm = lm();
    assert_eq!(acquire(&lm, 1, 10, LockType::Read), LockGrantType::New);
}

#[test]
fn lock_existing_same_locker_same_type() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    assert_eq!(acquire(&lm, 1, 10, LockType::Read), LockGrantType::Existing);
}

#[test]
fn lock_promotion_read_to_write_solo() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    assert_eq!(acquire(&lm, 1, 10, LockType::Write), LockGrantType::Promotion);
}

#[test]
fn two_readers_both_get_new() {
    let lm = lm();
    assert_eq!(acquire(&lm, 1, 10, LockType::Read), LockGrantType::New);
    assert_eq!(acquire(&lm, 1, 20, LockType::Read), LockGrantType::New);
}

#[test]
fn none_type_returns_none_needed() {
    let lm = lm();
    let r = lm.lock(1, 10, LockType::None, false, false).unwrap();
    assert_eq!(r, LockGrantType::NoneNeeded);
}

// ─── 2. Non-blocking acquire ──────────────────────────────────────────────────

#[test]
fn non_blocking_fails_when_write_held() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Write);
    let r = acquire_nb(&lm, 1, 20, LockType::Read);
    assert!(r.is_err(), "should fail non-blocking when write held");
    assert!(matches!(r.unwrap_err(), TxnError::LockNotAvailable { .. }));
}

#[test]
fn non_blocking_succeeds_for_reader_when_no_conflict() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    let r = acquire_nb(&lm, 1, 20, LockType::Read);
    assert_eq!(r.unwrap(), LockGrantType::New);
}

#[test]
fn non_blocking_fails_when_read_held_by_other_and_write_requested() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    let r = acquire_nb(&lm, 1, 20, LockType::Write);
    assert!(matches!(r.unwrap_err(), TxnError::LockNotAvailable { .. }));
}

// ─── 3. Release and n_total_locks ────────────────────────────────────────────

#[test]
fn release_reduces_lock_count() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    acquire(&lm, 2, 10, LockType::Read);
    assert!(lm.n_total_locks() >= 2);
    lm.release(1, 10).unwrap();
    let after = lm.n_total_locks();
    assert!(after < 2, "n_total_locks should decrease after release");
}

#[test]
fn release_unknown_lock_is_ok() {
    let lm = lm();
    // Releasing a lock that was never acquired should not panic.
    let _ = lm.release(999, 10);
}

#[test]
fn n_total_locks_zero_initially() {
    let lm = lm();
    assert_eq!(lm.n_total_locks(), 0);
}

#[test]
fn multiple_locks_accumulate() {
    let lm = lm();
    for i in 1u64..=5 {
        acquire(&lm, i, 10, LockType::Read);
    }
    assert_eq!(lm.n_total_locks(), 5);
}

// ─── 4. Demote ────────────────────────────────────────────────────────────────

#[test]
fn demote_write_to_read_succeeds() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Write);
    assert!(lm.demote(1, 10).is_ok());
}

#[test]
fn is_owned_write_lock_false_after_demote() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Write);
    lm.demote(1, 10).unwrap();
    assert!(!lm.is_owned_write_lock(1, 10));
}

// ─── 5. Lock stealing ────────────────────────────────────────────────────────

#[test]
fn steal_lock_transfers_ownership() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Write);
    // Locker 20 steals the lock from locker 10 (importunate preemption).
    lm.steal_lock(1, 20).unwrap();
    // After preemption, locker 10 no longer holds a write lock.
    assert!(!lm.is_owned_write_lock(1, 10));
}

// ─── 6. get_lock_info (owners, waiters) ──────────────────────────────────────

#[test]
fn get_lock_info_shows_owner() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    let (owners, waiters) = lm.get_lock_info(1);
    assert_eq!(owners, 1);
    assert_eq!(waiters, 0);
}

#[test]
fn get_lock_info_empty_lsn() {
    let lm = lm();
    let (owners, waiters) = lm.get_lock_info(999);
    assert_eq!(owners, 0);
    assert_eq!(waiters, 0);
}

#[test]
fn get_lock_info_two_readers() {
    let lm = lm();
    acquire(&lm, 5, 10, LockType::Read);
    acquire(&lm, 5, 20, LockType::Read);
    let (owners, _waiters) = lm.get_lock_info(5);
    assert_eq!(owners, 2);
}

// ─── 7. get_stats() ──────────────────────────────────────────────────────────

#[test]
fn stats_request_count_increments() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    acquire(&lm, 2, 10, LockType::Read);
    let stats: LockStats = lm.get_stats();
    assert!(stats.lock_requests >= 2);
}

#[test]
fn stats_write_lock_counted() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Write);
    let stats = lm.get_stats();
    assert!(stats.lock_requests >= 1);
    assert!(stats.n_total_locks >= 1);
}

#[test]
fn stats_read_lock_counted() {
    let lm = lm();
    acquire(&lm, 1, 10, LockType::Read);
    let stats = lm.get_stats();
    assert!(stats.lock_requests >= 1);
    assert!(stats.n_total_locks >= 1);
}

// ─── 8. Sharing registry ─────────────────────────────────────────────────────

#[test]
fn registered_lockers_are_in_same_group() {
    let lm = lm();
    lm.register_locker_sharing(100, 42);
    lm.register_locker_sharing(200, 42);
    assert!(lm.same_share_group(100, 200));
}

#[test]
fn unregistered_lockers_not_in_same_group() {
    let lm = lm();
    lm.register_locker_sharing(100, 42);
    assert!(!lm.same_share_group(100, 999));
}

#[test]
fn unregister_removes_from_group() {
    let lm = lm();
    lm.register_locker_sharing(100, 42);
    lm.register_locker_sharing(200, 42);
    lm.unregister_locker_sharing(100);
    assert!(!lm.same_share_group(100, 200));
}

// ─── 9. Lock timeout ─────────────────────────────────────────────────────────

#[test]
fn lock_times_out_when_blocked() {
    let lm = Arc::new(LockManager::with_lock_timeout(50)); // 50 ms timeout
    // Locker 10 holds a write lock.
    acquire(&lm, 1, 10, LockType::Write);

    // Locker 20 should time out trying to read.
    let lm2 = Arc::clone(&lm);
    let result = thread::spawn(move || {
        lm2.lock_with_timeout(1, 20, LockType::Read, false, false, 50)
    })
    .join()
    .unwrap();

    assert!(
        matches!(result, Err(TxnError::LockTimeout { .. })),
        "expected LockTimeout, got {result:?}"
    );
}

#[test]
fn set_lock_timeout_takes_effect() {
    let lm = lm();
    lm.set_lock_timeout(200);
    assert_eq!(lm.get_lock_timeout_ms(), 200);
}

// ─── 10. Concurrent grant ────────────────────────────────────────────────────

#[test]
fn concurrent_readers_all_granted() {
    let lm = Arc::new(LockManager::new());
    let handles: Vec<_> = (0..8)
        .map(|i| {
            let lm = Arc::clone(&lm);
            thread::spawn(move || {
                lm.lock(42, i, LockType::Read, false, false).unwrap()
            })
        })
        .collect();
    for h in handles {
        let g = h.join().unwrap();
        assert!(matches!(g, LockGrantType::New | LockGrantType::Existing));
    }
}

#[test]
fn writer_blocked_then_released() {
    let lm = Arc::new(LockManager::new());
    let lm2 = Arc::clone(&lm);

    // Reader holds lock.
    acquire(&lm, 10, 1, LockType::Read);

    // Writer waits in background.
    let writer = thread::spawn(move || {
        lm2.lock_with_timeout(10, 2, LockType::Write, false, false, 2_000)
    });

    // Release the reader after a short delay.
    thread::sleep(Duration::from_millis(20));
    lm.release(10, 1).unwrap();

    let result = writer.join().unwrap();
    assert!(
        result.is_ok(),
        "writer should have been granted after reader released"
    );
}

// ──────────────────────────────────────────────────────────────────────────────
// JE LockManagerTest ports
// ──────────────────────────────────────────────────────────────────────────────

// JE: testSR15926LargeNodeIds
//
// Locks a record at a very large LSN (top bit set in the 32-bit half) and
// verifies it succeeds.  The original JE bug was that `lock(0x80000000L, ...)`
// went through a path that mishandled the sign bit of an int conversion.
// Noxu uses `u64` for LSNs so the sign-bit hazard does not strictly apply,
// but the test still exercises that the LockManager handles "high" LSNs
// near the 32-bit boundary correctly.
#[test]
fn je_sr15926_large_node_ids_can_be_locked() {
    let lm = lm();
    let lsn: u64 = 0x80000000u64;
    let r = lm.lock(lsn, 1, LockType::Write, false, false);
    assert!(r.is_ok(), "lock at LSN 0x80000000 must succeed: {:?}", r);

    // Also exercise an even larger LSN.
    let lsn2: u64 = 0xFFFF_FFFFu64;
    let r = lm.lock(lsn2, 1, LockType::Write, false, false);
    assert!(r.is_ok(), "lock at LSN 0xFFFFFFFF must succeed: {:?}", r);

    let lsn3: u64 = 0x1_0000_0000u64;
    let r = lm.lock(lsn3, 1, LockType::Write, false, false);
    assert!(r.is_ok(), "lock at LSN 0x1_0000_0000 must succeed: {:?}", r);
}

// JE: testNegatives — repeated lock by same locker is "Existing"
#[test]
fn je_negatives_repeat_read_returns_existing() {
    let lm = lm();
    let lsn: u64 = 1;
    let g = lm.lock(lsn, 10, LockType::Read, false, false).unwrap();
    assert!(matches!(g, LockGrantType::New));
    let g = lm.lock(lsn, 10, LockType::Read, false, false).unwrap();
    assert_eq!(
        g,
        LockGrantType::Existing,
        "repeated read lock by the same locker must return Existing"
    );
}

// JE: testNegatives — release on unrelated LSN is a no-op
#[test]
fn je_negatives_release_unrelated_lsn_is_noop() {
    let lm = lm();
    let lsn: u64 = 1;
    let other: u64 = 2;
    lm.lock(lsn, 10, LockType::Read, false, false).unwrap();
    // Release a LSN we don't hold — must not affect the held lock.
    let _ = lm.release(other, 10);
    // Lock 1 should still be held: a non-blocking write by someone else
    // must fail.
    let r = lm.lock(lsn, 11, LockType::Write, true, false);
    assert!(
        r.is_err(),
        "concurrent write must fail while reader holds: {:?}",
        r
    );
}

// JE: testNegatives — release by non-owner is a no-op
#[test]
fn je_negatives_release_by_non_owner_is_noop() {
    let lm = lm();
    let lsn: u64 = 1;
    lm.lock(lsn, 10, LockType::Read, false, false).unwrap();
    // txn 11 doesn't own the lock — releasing it must not affect ownership.
    let _ = lm.release(lsn, 11);
    // Owner (10) still holds it: a non-blocking write by 11 still fails.
    let r = lm.lock(lsn, 11, LockType::Write, true, false);
    assert!(
        r.is_err(),
        "concurrent write must still fail after non-owner release: {:?}",
        r
    );
}

// JE: testUpgradeLock — same locker holding read can upgrade to write
#[test]
fn je_upgrade_read_to_write_same_locker() {
    let lm = lm();
    let lsn: u64 = 42;
    lm.lock(lsn, 10, LockType::Read, false, false).unwrap();
    let g = lm.lock(lsn, 10, LockType::Write, false, false).unwrap();
    // Both Promoted and Existing are acceptable depending on internal
    // representation; the invariant is that the lock is now write-held.
    assert!(
        matches!(
            g,
            LockGrantType::Promotion
                | LockGrantType::WaitPromotion
                | LockGrantType::Existing
                | LockGrantType::New
        ),
        "read→write upgrade must succeed for same locker: got {:?}",
        g
    );

    // After upgrade, a concurrent read by another txn must fail (write
    // exclusive).
    let r = lm.lock(lsn, 11, LockType::Read, true, false);
    assert!(r.is_err(), "after upgrade, concurrent read must fail: {:?}", r);
}

// JE: testMultipleReaders — three concurrent readers share a lock
#[test]
fn je_multiple_readers_share() {
    let lm = lm();
    let lsn: u64 = 7;
    for locker in [10i64, 11, 12] {
        let g = lm.lock(lsn, locker, LockType::Read, false, false).unwrap();
        assert!(
            matches!(g, LockGrantType::New | LockGrantType::Existing),
            "reader {} must share, got {:?}",
            locker,
            g
        );
    }
}

// ─── DeadlockTest ports (wave 9-C) ──────────────────────────────────────────
//
// JE invariant (DeadlockTest.testDeadlockBetweenTwoTxns):
//   Txn1 owns L1, waits on L2; Txn2 owns L2, waits on L1.  One of
//   them is chosen as the deadlock victim and surfaces a deadlock
//   exception; the other proceeds.
//
// Noxu adaptation: drive the same scenario via two threads using
// `LockManager::lock` directly and assert that exactly one thread
// observes a `TxnError::Deadlock`, and the other ultimately holds
// both locks.

#[test]
fn je_deadlock_between_two_txns() {
    let lm = lm();
    let lm1 = Arc::clone(&lm);
    let lm2 = Arc::clone(&lm);

    let l1: u64 = 100;
    let l2: u64 = 200;
    let txn1: i64 = 1001;
    let txn2: i64 = 1002;

    // Txn1 grabs L1 first.
    lm.lock(l1, txn1, LockType::Write, false, false).unwrap();
    // Txn2 grabs L2 first.
    lm.lock(l2, txn2, LockType::Write, false, false).unwrap();

    // Spawn one waiter on L2 (txn1) and one on L1 (txn2); one of them
    // must surface a deadlock.
    let h1 = thread::spawn(move || {
        lm1.lock(l2, txn1, LockType::Write, false, false)
    });
    // Give h1 time to enter the wait queue, then start h2.
    thread::sleep(Duration::from_millis(50));
    let h2 = thread::spawn(move || {
        lm2.lock(l1, txn2, LockType::Write, false, false)
    });

    let r1 = h1.join().unwrap();
    let r2 = h2.join().unwrap();

    let r1_dl = matches!(&r1, Err(TxnError::Deadlock(_)));
    let r2_dl = matches!(&r2, Err(TxnError::Deadlock(_)));
    assert!(
        r1_dl ^ r2_dl,
        "exactly one of the two waiters must surface a deadlock; \
         r1={:?} r2={:?}",
        r1,
        r2
    );

    // The non-victim eventually granted its lock; clean up.
    let _ = lm.release(l1, txn1);
    let _ = lm.release(l2, txn1);
    let _ = lm.release(l1, txn2);
    let _ = lm.release(l2, txn2);
}

// JE invariant (DeadlockTest.testDeadlockProducedByTwoLockersOnOneLock):
//   Two lockers contend on a single lock with incompatible modes; if
//   they form a cycle (each holds something the other waits for), one
//   becomes the victim.  The simpler one-lock variant: T1 holds Read,
//   T2 holds Read, T1 then upgrades to Write while T2 also tries to
//   upgrade — one deadlocks.
#[test]
fn je_deadlock_two_lockers_on_one_lock() {
    let lm = lm();
    let l: u64 = 42;
    let t1: i64 = 1;
    let t2: i64 = 2;

    // Both grab read locks.
    lm.lock(l, t1, LockType::Read, false, false).unwrap();
    lm.lock(l, t2, LockType::Read, false, false).unwrap();

    // Both try to upgrade to write — exactly one should deadlock
    // (the other completes after the victim aborts).
    let lm1 = Arc::clone(&lm);
    let lm2 = Arc::clone(&lm);
    let h1 =
        thread::spawn(move || lm1.lock(l, t1, LockType::Write, false, false));
    thread::sleep(Duration::from_millis(20));
    let h2 =
        thread::spawn(move || lm2.lock(l, t2, LockType::Write, false, false));

    let r1 = h1.join().unwrap();
    let r2 = h2.join().unwrap();

    let r1_dl = matches!(&r1, Err(TxnError::Deadlock(_)));
    let r2_dl = matches!(&r2, Err(TxnError::Deadlock(_)));
    assert!(
        r1_dl || r2_dl,
        "at least one upgrade must surface a deadlock; r1={:?} r2={:?}",
        r1,
        r2
    );

    let _ = lm.release(l, t1);
    let _ = lm.release(l, t2);
}

// JE invariant (LockTest.testLockConflicts, narrow port):
//   The lock-mode compatibility matrix is asserted for the basic
//   conflict pairs Noxu supports: Read-Read share, Read-Write
//   conflict, Write-Write conflict.
#[test]
fn je_lock_test_conflicts_matrix() {
    let lm = lm();
    let l: u64 = 7;
    // Read-Read: share.
    lm.lock(l, 1, LockType::Read, false, false).unwrap();
    let g = lm.lock(l, 2, LockType::Read, false, false).unwrap();
    assert!(matches!(g, LockGrantType::New | LockGrantType::Existing));
    let _ = lm.release(l, 1);
    let _ = lm.release(l, 2);

    // Read-Write: conflict (non-blocking).
    lm.lock(l, 1, LockType::Read, false, false).unwrap();
    let r = lm.lock(l, 2, LockType::Write, true, false);
    assert!(
        r.is_err(),
        "Write must conflict with held Read (non-blocking): got {:?}",
        r
    );
    let _ = lm.release(l, 1);

    // Write-Write: conflict (non-blocking).
    lm.lock(l, 1, LockType::Write, false, false).unwrap();
    let r = lm.lock(l, 2, LockType::Write, true, false);
    assert!(
        r.is_err(),
        "Write must conflict with held Write (non-blocking): got {:?}",
        r
    );
    let _ = lm.release(l, 1);

    // Write-Read: conflict (non-blocking).
    lm.lock(l, 1, LockType::Write, false, false).unwrap();
    let r = lm.lock(l, 2, LockType::Read, true, false);
    assert!(
        r.is_err(),
        "Read must conflict with held Write (non-blocking): got {:?}",
        r
    );
    let _ = lm.release(l, 1);
}
