#![cfg(feature = "loom")]
#![allow(clippy::unwrap_used)]

use loom::sync::Arc;
use loom::thread;

use armdb::DiskLoc;
use armdb::skiplist::iter::Iter;
use armdb::skiplist::node::{ConstNode, SkipNode};
use armdb::skiplist::{InsertResult, SkipList};

type TestNode = ConstNode<[u8; 4], 8>;
type TestList = SkipList<TestNode>;

fn disk() -> DiskLoc {
    DiskLoc::new(0, 0, 0)
}

fn alloc_node(key: [u8; 4], value: [u8; 8]) -> *mut TestNode {
    TestNode::alloc(key, value, disk(), 1)
}

/// Wrapper to move raw pointers across loom thread boundaries.
/// SAFETY: SkipNode types implement Send+Sync; the pointer is freshly
/// allocated and exclusively owned until inserted into the skip list.
struct SendPtr<T>(*mut T);
unsafe impl<T> Send for SendPtr<T> {}

// ---------------------------------------------------------------------------
// Test 1: get || insert
// Reader must see either None or the fully-inserted node, never partial state.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_get_insert() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            l1.get(&[0, 0, 0, 1], &guard).map(|n| n.read_value())
        });

        let node = SendPtr(alloc_node([0, 0, 0, 1], [1u8; 8]));
        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.insert(node.0, &guard);
        });

        let read = t1.join().unwrap();
        t2.join().unwrap();

        if let Some(v) = read {
            assert_eq!(v, [1u8; 8]);
        }
    });
}

// ---------------------------------------------------------------------------
// Test 2: get || remove
// Reader must see either the original value or None (after logical deletion).
// ---------------------------------------------------------------------------
#[test]
fn concurrent_get_remove() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        {
            let guard = list.collector().enter();
            let node = alloc_node([0, 0, 0, 1], [1u8; 8]);
            list.insert(node, &guard);
        }

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            l1.get(&[0, 0, 0, 1], &guard).map(|n| n.read_value())
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 1], &guard)
        });

        let read = t1.join().unwrap();
        t2.join().unwrap();

        if let Some(v) = read {
            assert_eq!(v, [1u8; 8]);
        }
    });
}

// ---------------------------------------------------------------------------
// Test 3: insert || remove on the same pre-existing key
// Both are serialized by write_lock. Loom verifies both orderings are correct.
// ---------------------------------------------------------------------------
#[test]
fn insert_remove_same_key() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        {
            let guard = list.collector().enter();
            let node = alloc_node([0, 0, 0, 1], [1u8; 8]);
            list.insert(node, &guard);
        }

        let l1 = list.clone();
        let node = SendPtr(alloc_node([0, 0, 0, 1], [2u8; 8]));
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            let inserted = matches!(l1.insert(node.0, &guard), InsertResult::Inserted);
            if !inserted {
                unsafe { TestNode::dealloc_node(node.0) };
            }
            inserted
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 1], &guard).is_some()
        });

        let was_inserted = t1.join().unwrap();
        let was_removed = t2.join().unwrap();

        // Both hold write_lock → serialized. The key exists at start, so:
        // Order A: remove first (true), then insert (true — key gone, re-inserts)
        // Order B: insert first (false — key exists), then remove (true)
        assert!(was_removed);
        if was_inserted {
            assert_eq!(list.len(), 1);
        }
    });
}

// ---------------------------------------------------------------------------
// Test 4: iter || insert
// Iterator must not panic; weakly-consistent — may or may not see new node.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_iter_insert() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        {
            let guard = list.collector().enter();
            let node = alloc_node([0, 0, 0, 1], [1u8; 8]);
            list.insert(node, &guard);
        }

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            let iter = Iter::all(&l1, &guard);
            iter.map(|n| n.read_value()).collect::<Vec<_>>()
        });

        let node = SendPtr(alloc_node([0, 0, 0, 2], [2u8; 8]));
        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.insert(node.0, &guard);
        });

        let vals = t1.join().unwrap();
        t2.join().unwrap();

        assert!(vals.contains(&[1u8; 8]));
        assert!(vals.len() == 1 || vals.len() == 2);
    });
}

// ---------------------------------------------------------------------------
// Test 5: iter || remove
// Iterator must skip marked nodes; no panic.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_iter_remove() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        {
            let guard = list.collector().enter();
            let n1 = alloc_node([0, 0, 0, 1], [1u8; 8]);
            list.insert(n1, &guard);
            let n2 = alloc_node([0, 0, 0, 2], [2u8; 8]);
            list.insert(n2, &guard);
        }

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            let iter = Iter::all(&l1, &guard);
            iter.map(|n| n.read_value()).collect::<Vec<_>>()
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 1], &guard);
        });

        let vals = t1.join().unwrap();
        t2.join().unwrap();

        // Node 2 is never removed — always visible
        assert!(vals.contains(&[2u8; 8]));
        assert!(vals.len() == 1 || vals.len() == 2);
    });
}

// ---------------------------------------------------------------------------
// Test 6: write_data || read_value (ConstNode SeqLock)
// Reader sees either old or new data, never a torn value.
//
// SeqLock uses loom::thread::yield_now() under loom (instead of spin_loop hint)
// and loom::cell::UnsafeCell for data access tracking.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_write_read_data() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        {
            let guard = list.collector().enter();
            let node = alloc_node([0, 0, 0, 1], [1u8; 8]);
            list.insert(node, &guard);
        }

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            let node = l1.get(&[0, 0, 0, 1], &guard).unwrap();
            node.read_value()
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            let node = l2.get(&[0, 0, 0, 1], &guard).unwrap();
            node.write_data(disk(), &[2u8; 8]);
        });

        let value = t1.join().unwrap();
        t2.join().unwrap();

        assert!(value == [1u8; 8] || value == [2u8; 8]);
    });
}

// ---------------------------------------------------------------------------
// Test 8: insert || insert of adjacent keys sharing a common pred
// Both threads presearch independently, both see the same predecessor (1).
// The loser detects stale frame via validate_link and retries with fallback;
// ultimately both nodes are linked and the list contains all four keys.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_insert_insert_adjacent() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));
        {
            let guard = list.collector().enter();
            list.insert(alloc_node([0, 0, 0, 1], [1u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 9], [9u8; 8]), &guard);
        }

        let n4 = SendPtr(alloc_node([0, 0, 0, 4], [4u8; 8]));
        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            matches!(l1.insert(n4.0, &guard), InsertResult::Inserted)
        });

        let n5 = SendPtr(alloc_node([0, 0, 0, 5], [5u8; 8]));
        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            matches!(l2.insert(n5.0, &guard), InsertResult::Inserted)
        });

        assert!(t1.join().unwrap());
        assert!(t2.join().unwrap());

        let guard = list.collector().enter();
        for k in [1u8, 4, 5, 9] {
            assert!(list.get(&[0, 0, 0, k], &guard).is_some(), "key {k} lost");
        }
        assert_eq!(list.len(), 4);
    });
}

// ---------------------------------------------------------------------------
// Test 9: insert || insert of the SAME key (duplicate-protection contract)
// The same-key contract: no external serialization required — the loser
// always fails validate_link at level 0 (succ already set) and returns Exists.
// Exactly one thread wins (XOR); the loser must dealloc its unlinked node.
// After join: key is present, len == 1 (no duplicate linked).
// ---------------------------------------------------------------------------
#[test]
fn concurrent_insert_same_key_one_wins() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));

        let n_a = SendPtr(alloc_node([0, 0, 0, 3], [1u8; 8]));
        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            match l1.insert(n_a.0, &guard) {
                InsertResult::Inserted => true,
                InsertResult::Exists(_) => {
                    // Loser must free its unlinked node to avoid a memory leak
                    unsafe { TestNode::dealloc_node(n_a.0) };
                    false
                }
            }
        });

        let n_b = SendPtr(alloc_node([0, 0, 0, 3], [2u8; 8]));
        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            match l2.insert(n_b.0, &guard) {
                InsertResult::Inserted => true,
                InsertResult::Exists(_) => {
                    unsafe { TestNode::dealloc_node(n_b.0) };
                    false
                }
            }
        });

        let won1 = t1.join().unwrap();
        let won2 = t2.join().unwrap();
        assert!(won1 ^ won2, "exactly one insert must win");

        let guard = list.collector().enter();
        assert!(list.get(&[0, 0, 0, 3], &guard).is_some());
        assert_eq!(list.len(), 1, "duplicate was linked — lost node");
    });
}

// ---------------------------------------------------------------------------
// Test 10: insert || remove of the insert's predecessor
// T1 inserts key 4 (presearch finds pred=3); T2 concurrently removes 3.
// validate_link detects the marked pred and causes T1 to retry via fallback.
// Both operations succeed: 4 is present, 3 is absent, 1 and 9 are intact.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_insert_remove_pred() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));
        {
            let guard = list.collector().enter();
            list.insert(alloc_node([0, 0, 0, 1], [1u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 3], [3u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 9], [9u8; 8]), &guard);
        }

        let n4 = SendPtr(alloc_node([0, 0, 0, 4], [4u8; 8]));
        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            matches!(l1.insert(n4.0, &guard), InsertResult::Inserted)
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 3], &guard).is_some()
        });

        assert!(t1.join().unwrap());
        assert!(t2.join().unwrap());

        let guard = list.collector().enter();
        assert!(list.get(&[0, 0, 0, 4], &guard).is_some(), "key 4 lost");
        assert!(
            list.get(&[0, 0, 0, 3], &guard).is_none(),
            "key 3 should be removed"
        );
        assert!(list.get(&[0, 0, 0, 1], &guard).is_some(), "key 1 lost");
        assert!(list.get(&[0, 0, 0, 9], &guard).is_some(), "key 9 lost");
        assert_eq!(list.len(), 3);
    });
}

// ---------------------------------------------------------------------------
// Test 11: insert || remove of the insert's successor
// T1 inserts key 3 (presearch finds succ=5); T2 concurrently removes 5.
// The null-aware marked-succ check in validate_link triggers a fallback if 5
// is already marked when T1 attempts to link. Both ops succeed: 3 present,
// 5 absent, 1 intact, len == 2.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_insert_remove_succ() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));
        {
            let guard = list.collector().enter();
            list.insert(alloc_node([0, 0, 0, 1], [1u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 5], [5u8; 8]), &guard);
        }

        let n3 = SendPtr(alloc_node([0, 0, 0, 3], [3u8; 8]));
        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            matches!(l1.insert(n3.0, &guard), InsertResult::Inserted)
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 5], &guard).is_some()
        });

        assert!(t1.join().unwrap());
        assert!(t2.join().unwrap());

        let guard = list.collector().enter();
        assert!(list.get(&[0, 0, 0, 3], &guard).is_some(), "key 3 lost");
        assert!(
            list.get(&[0, 0, 0, 5], &guard).is_none(),
            "key 5 should be removed"
        );
        assert!(list.get(&[0, 0, 0, 1], &guard).is_some(), "key 1 lost");
        assert_eq!(list.len(), 2);
    });
}

// ---------------------------------------------------------------------------
// Test 12: remove || remove of the SAME key (exactly one mark/len-- wins)
// Pre-fill {1, 3}. Both threads race to remove 3. The mark() CAS ensures
// exactly one succeeds; the other sees the already-marked node and returns
// None. XOR on is_some() results must hold. len == 1, get(3) == None.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_remove_same_key_one_wins() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));
        {
            let guard = list.collector().enter();
            list.insert(alloc_node([0, 0, 0, 1], [1u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 3], [3u8; 8]), &guard);
        }

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            l1.remove(&[0, 0, 0, 3], &guard).is_some()
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 3], &guard).is_some()
        });

        let removed1 = t1.join().unwrap();
        let removed2 = t2.join().unwrap();
        assert!(removed1 ^ removed2, "exactly one remove must succeed");

        let guard = list.collector().enter();
        assert!(
            list.get(&[0, 0, 0, 3], &guard).is_none(),
            "key 3 should be absent"
        );
        assert!(list.get(&[0, 0, 0, 1], &guard).is_some(), "key 1 lost");
        assert_eq!(list.len(), 1);
    });
}

// ---------------------------------------------------------------------------
// Test 13: remove || remove of adjacent keys sharing a common pred
// Pre-fill {1, 3, 4, 9}. T1 removes 3, T2 removes 4 (both share pred=1 after
// the first unlink). Each thread's unlink-CAS must succeed independently even
// when the other has already modified the pred's next pointer. Both return
// Some; 3 and 4 are absent; 1 and 9 are intact; len == 2.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_remove_remove_adjacent() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));
        {
            let guard = list.collector().enter();
            list.insert(alloc_node([0, 0, 0, 1], [1u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 3], [3u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 4], [4u8; 8]), &guard);
            list.insert(alloc_node([0, 0, 0, 9], [9u8; 8]), &guard);
        }

        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            l1.remove(&[0, 0, 0, 3], &guard).is_some()
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 4], &guard).is_some()
        });

        assert!(t1.join().unwrap());
        assert!(t2.join().unwrap());

        let guard = list.collector().enter();
        assert!(
            list.get(&[0, 0, 0, 3], &guard).is_none(),
            "key 3 should be absent"
        );
        assert!(
            list.get(&[0, 0, 0, 4], &guard).is_none(),
            "key 4 should be absent"
        );
        assert!(list.get(&[0, 0, 0, 1], &guard).is_some(), "key 1 lost");
        assert!(list.get(&[0, 0, 0, 9], &guard).is_some(), "key 9 lost");
        assert_eq!(list.len(), 2);
    });
}

// ---------------------------------------------------------------------------
// Test 14: remove races a TALLER same-key insert (PR #43 review, R1).
// The remover's pre-lock search_height load can be stale: the inserter bumps
// the tree height (outside the lock) and links a node with more levels than
// the remover's presearch frame covers. The remove must detect
// found_height > search_height, fall back, and search with a fresh height
// loaded under the lock — never dereferencing unpopulated upper-level preds.
// ---------------------------------------------------------------------------
#[test]
fn concurrent_remove_taller_insert_same_key() {
    loom::model(|| {
        let list = Arc::new(TestList::new(false));
        {
            let guard = list.collector().enter();
            list.insert(alloc_node([0, 0, 0, 1], [1u8; 8]), &guard);
        }

        // Height-2 node: taller than the initial tree height of 1.
        let n3 = SendPtr(TestNode::alloc([0, 0, 0, 3], [3u8; 8], disk(), 2));
        let l1 = list.clone();
        let t1 = thread::spawn(move || {
            let guard = l1.collector().enter();
            matches!(l1.insert(n3.0, &guard), InsertResult::Inserted)
        });

        let l2 = list.clone();
        let t2 = thread::spawn(move || {
            let guard = l2.collector().enter();
            l2.remove(&[0, 0, 0, 3], &guard).is_some()
        });

        let inserted = t1.join().unwrap();
        let removed = t2.join().unwrap();
        assert!(inserted, "insert of a fresh key must succeed");

        // remove may linearize before the insert (None) or after (Some).
        let guard = list.collector().enter();
        if removed {
            assert!(list.get(&[0, 0, 0, 3], &guard).is_none());
            assert_eq!(list.len(), 1);
        } else {
            assert!(list.get(&[0, 0, 0, 3], &guard).is_some());
            assert_eq!(list.len(), 2);
        }
        assert!(list.get(&[0, 0, 0, 1], &guard).is_some(), "key 1 lost");
    });
}
