// Snapshot consistency tests
// Tests that reads see consistent point-in-time snapshots
// Critical for correctness: concurrent writes must not affect in-progress reads

use seerdb::{DBOptions, DB};
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::Duration;
use tempfile::TempDir;

#[test]
fn test_read_isolation_during_writes() {
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(DB::open(temp_dir.path()).unwrap());

    // Initial state: key exists with value "v1"
    db.put(b"key", b"v1").unwrap();

    let barrier = Arc::new(Barrier::new(2));

    // Reader thread: reads repeatedly, should always see "v1" or "v2" (never partial state)
    let db_read = db.clone();
    let barrier_read = barrier.clone();
    let reader = thread::spawn(move || {
        barrier_read.wait();
        let mut values = Vec::new();
        for _ in 0..200 {
            if let Some(v) = db_read.get(b"key").unwrap() {
                values.push(v);
            }
            thread::sleep(Duration::from_millis(1));
        }
        values
    });

    // Writer thread: updates value to "v2"
    let db_write = db.clone();
    let barrier_write = barrier.clone();
    let writer = thread::spawn(move || {
        barrier_write.wait();
        thread::sleep(Duration::from_millis(50));
        db_write.put(b"key", b"v2").unwrap();
    });

    let values = reader.join().unwrap();
    writer.join().unwrap();

    // All values must be either "v1" or "v2" (no corruption/partial reads)
    for v in &values {
        assert!(
            v.as_ref() == b"v1" || v.as_ref() == b"v2",
            "Invalid value read"
        );
    }

    // Should have seen both values
    let has_v1 = values.iter().any(|v| v.as_ref() == b"v1");
    let has_v2 = values.iter().any(|v| v.as_ref() == b"v2");
    assert!(has_v1, "Should see old value");
    assert!(has_v2, "Should see new value");
}

#[test]
#[ignore = "Flaky in CI due to timing sensitivity"]
fn test_read_isolation_across_flush() {
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(
        DBOptions::default()
            .memtable_capacity(1024 * 1024)
            .open(temp_dir.path())
            .unwrap(),
    );

    // Populate memtable
    for i in 0..100 {
        db.put(format!("key_{:03}", i).as_bytes(), b"value")
            .unwrap();
    }

    let barrier = Arc::new(Barrier::new(2));

    // Reader thread: reads all keys repeatedly
    let db_read = db.clone();
    let barrier_read = barrier.clone();
    let reader = thread::spawn(move || {
        barrier_read.wait();
        let mut read_count = 0;
        for _ in 0..10 {
            for i in 0..100 {
                if db_read
                    .get(format!("key_{:03}", i).as_bytes())
                    .unwrap()
                    .is_some()
                {
                    read_count += 1;
                }
            }
            thread::sleep(Duration::from_millis(10));
        }
        read_count
    });

    // Flusher thread: flushes memtable during reads
    let db_flush = db.clone();
    let barrier_flush = barrier.clone();
    let flusher = thread::spawn(move || {
        barrier_flush.wait();
        thread::sleep(Duration::from_millis(25));
        db_flush.flush().unwrap();
    });

    let read_count = reader.join().unwrap();
    flusher.join().unwrap();

    // All reads should succeed (data present before/after flush)
    assert_eq!(read_count, 1000, "All keys should be readable during flush");
}

#[test]
fn test_snapshot_isolation_multiple_keys() {
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(DB::open(temp_dir.path()).unwrap());

    // Initial state: keys 0-99 exist
    for i in 0..100 {
        db.put(format!("key_{:03}", i).as_bytes(), b"v1").unwrap();
    }

    let barrier = Arc::new(Barrier::new(2));

    // Reader thread: reads all keys, checks consistency
    let db_read = db.clone();
    let barrier_read = barrier.clone();
    let reader = thread::spawn(move || {
        barrier_read.wait();
        thread::sleep(Duration::from_millis(25)); // Let writer update half

        // Read all keys - should be consistent snapshot
        let mut v1_count = 0;
        let mut v2_count = 0;
        for i in 0..100 {
            if let Some(v) = db_read.get(format!("key_{:03}", i).as_bytes()).unwrap() {
                if v.as_ref() == b"v1" {
                    v1_count += 1;
                } else if v.as_ref() == b"v2" {
                    v2_count += 1;
                }
            }
        }
        (v1_count, v2_count)
    });

    // Writer thread: updates keys 0-49 to "v2"
    let db_write = db.clone();
    let barrier_write = barrier.clone();
    let writer = thread::spawn(move || {
        barrier_write.wait();
        for i in 0..50 {
            db_write
                .put(format!("key_{:03}", i).as_bytes(), b"v2")
                .unwrap();
            thread::sleep(Duration::from_micros(100));
        }
    });

    let (v1_count, v2_count) = reader.join().unwrap();
    writer.join().unwrap();

    // Should see all 100 keys (no missing keys)
    assert_eq!(v1_count + v2_count, 100, "All keys should be present");
}

#[test]
#[ignore] // TODO(0.0.2): Requires Snapshot API for multi-operation consistency
          // Current isolation: Read Committed (per-operation snapshot)
          // This test requires: Snapshot Isolation (multi-operation snapshot)
          //
          // Context: Each get() call captures a separate point-in-time snapshot. Between
          // calls, database state can change (flush, compaction), causing keys to move
          // between memtable/immutable/SSTables. A reader iterating 100 keys may see
          // inconsistent state if a flush happens mid-iteration.
          //
          // Fix: Implement Snapshot API (capture state once, read multiple times)
          // Deferred to 0.0.2+ because vector databases don't require snapshot isolation
          // (Milvus, Qdrant, Weaviate all use eventual consistency for ANN search).
          // See: ai/research/LSM_MVCC_CONCURRENCY_RESEARCH.md
fn test_concurrent_reads_consistent() {
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(DB::open(temp_dir.path()).unwrap());

    // Write initial data
    for i in 0..100 {
        db.put(
            format!("key_{:03}", i).as_bytes(),
            format!("value_{:03}", i).as_bytes(),
        )
        .unwrap();
    }

    let barrier = Arc::new(Barrier::new(4));

    // Spawn 3 concurrent readers
    let handles: Vec<_> = (0..3)
        .map(|reader_id| {
            let db_clone = db.clone();
            let barrier_clone = barrier.clone();
            thread::spawn(move || {
                barrier_clone.wait();
                let mut read_values = Vec::new();
                for i in 0..100 {
                    if let Some(v) = db_clone.get(format!("key_{:03}", i).as_bytes()).unwrap() {
                        read_values.push((i, v));
                    }
                }
                (reader_id, read_values)
            })
        })
        .collect();

    // Writer: updates some keys during reads
    let db_write = db.clone();
    let barrier_write = barrier.clone();
    let writer = thread::spawn(move || {
        barrier_write.wait();
        for i in 0..50 {
            db_write
                .put(format!("key_{:03}", i).as_bytes(), b"updated")
                .unwrap();
        }
    });

    // Collect results
    let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
    writer.join().unwrap();

    // All readers should see valid data (either old or new values)
    for (reader_id, values) in results {
        assert_eq!(
            values.len(),
            100,
            "Reader {} should see all keys",
            reader_id
        );

        // Each value should be either original or "updated"
        for (i, value) in values {
            let expected_original = format!("value_{:03}", i);
            assert!(
                value.as_ref() == expected_original.as_bytes() || value.as_ref() == b"updated",
                "Reader {} saw invalid value for key_{:03}",
                reader_id,
                i
            );
        }
    }
}

#[test]
fn test_delete_snapshot_consistency() {
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(DB::open(temp_dir.path()).unwrap());

    // Write keys
    for i in 0..100 {
        db.put(format!("key_{:03}", i).as_bytes(), b"value")
            .unwrap();
    }

    let barrier = Arc::new(Barrier::new(2));

    // Reader: scans for existing keys
    let db_read = db.clone();
    let barrier_read = barrier.clone();
    let reader = thread::spawn(move || {
        barrier_read.wait();
        thread::sleep(Duration::from_millis(10)); // Let some deletes happen

        let mut found_count = 0;
        for i in 0..100 {
            if db_read
                .get(format!("key_{:03}", i).as_bytes())
                .unwrap()
                .is_some()
            {
                found_count += 1;
            }
        }
        found_count
    });

    // Deleter: deletes keys 0-49
    let db_delete = db.clone();
    let barrier_delete = barrier.clone();
    let deleter = thread::spawn(move || {
        barrier_delete.wait();
        for i in 0..50 {
            db_delete
                .delete(format!("key_{:03}", i).as_bytes())
                .unwrap();
            thread::sleep(Duration::from_micros(100));
        }
    });

    let found_count = reader.join().unwrap();
    deleter.join().unwrap();

    // Reader should see consistent snapshot (all keys present or some deleted, but consistent)
    assert!(
        found_count >= 50 && found_count <= 100,
        "Found {} keys, expected 50-100 (snapshot consistency)",
        found_count
    );
}

#[test]
fn test_memtable_atomicity() {
    // Test that memtable updates are atomic - readers see full update or none
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(DB::open(temp_dir.path()).unwrap());

    db.put(b"counter", b"0").unwrap();

    let barrier = Arc::new(Barrier::new(2));

    // Reader: reads counter value repeatedly
    let db_read = db.clone();
    let barrier_read = barrier.clone();
    let reader = thread::spawn(move || {
        barrier_read.wait();
        let mut values = Vec::new();
        for _ in 0..1000 {
            if let Some(v) = db_read.get(b"counter").unwrap() {
                values.push(String::from_utf8(v.to_vec()).unwrap());
            }
        }
        values
    });

    // Writer: increments counter
    let db_write = db.clone();
    let barrier_write = barrier.clone();
    let writer = thread::spawn(move || {
        barrier_write.wait();
        for i in 1..=10 {
            db_write.put(b"counter", i.to_string().as_bytes()).unwrap();
            thread::sleep(Duration::from_micros(100));
        }
    });

    let values = reader.join().unwrap();
    writer.join().unwrap();

    // All read values should be valid counter values (0-10)
    for v in values {
        let num: u32 = v.parse().unwrap();
        assert!(num <= 10, "Invalid counter value: {}", num);
    }
}

#[test]
fn test_no_stale_reads_after_flush() {
    let temp_dir = TempDir::new().unwrap();
    let db = DB::open(temp_dir.path()).unwrap();

    // Write and flush
    db.put(b"key", b"v1").unwrap();
    db.flush().unwrap();

    // Update in memtable (not flushed yet)
    db.put(b"key", b"v2").unwrap();

    // Read should see latest value (v2 from memtable, not v1 from SSTable)
    let value = db.get(b"key").unwrap().unwrap();
    assert_eq!(
        value.as_ref(),
        b"v2",
        "Should see latest memtable value, not stale SSTable value"
    );
}

#[test]
fn test_no_stale_reads_after_delete() {
    let temp_dir = TempDir::new().unwrap();
    let db = DB::open(temp_dir.path()).unwrap();

    // Write and flush to SSTable
    db.put(b"key", b"value").unwrap();
    db.flush().unwrap();

    // Delete (tombstone in memtable)
    db.delete(b"key").unwrap();

    // Read should return None (tombstone masks SSTable value)
    assert!(
        db.get(b"key").unwrap().is_none(),
        "Tombstone should mask flushed value"
    );

    // Flush tombstone
    db.flush().unwrap();

    // Read should still return None
    assert!(
        db.get(b"key").unwrap().is_none(),
        "Tombstone should persist after flush"
    );
}

#[test]
#[ignore] // Flaky test - race condition between reader/writer can cause spurious failures
fn test_point_in_time_consistency() {
    // Test that a sequence of operations sees consistent point-in-time state
    let temp_dir = TempDir::new().unwrap();
    let db = Arc::new(DB::open(temp_dir.path()).unwrap());

    // Initial state: account balances
    db.put(b"account_a", b"100").unwrap();
    db.put(b"account_b", b"50").unwrap();

    let barrier = Arc::new(Barrier::new(2));

    // Reader: reads both accounts, checks total is constant
    let db_read = db.clone();
    let barrier_read = barrier.clone();
    let reader = thread::spawn(move || {
        barrier_read.wait();
        let mut totals = Vec::new();
        for _ in 0..100 {
            let a = String::from_utf8(db_read.get(b"account_a").unwrap().unwrap().to_vec())
                .unwrap()
                .parse::<u32>()
                .unwrap();
            let b = String::from_utf8(db_read.get(b"account_b").unwrap().unwrap().to_vec())
                .unwrap()
                .parse::<u32>()
                .unwrap();
            totals.push(a + b);
            thread::sleep(Duration::from_micros(100));
        }
        totals
    });

    // Writer: transfers money from A to B (should maintain total)
    let db_write = db.clone();
    let barrier_write = barrier.clone();
    let writer = thread::spawn(move || {
        barrier_write.wait();
        for _ in 0..10 {
            // Transfer 10 from A to B
            let a = String::from_utf8(db_write.get(b"account_a").unwrap().unwrap().to_vec())
                .unwrap()
                .parse::<u32>()
                .unwrap();
            let b = String::from_utf8(db_write.get(b"account_b").unwrap().unwrap().to_vec())
                .unwrap()
                .parse::<u32>()
                .unwrap();

            db_write
                .put(b"account_a", (a - 10).to_string().as_bytes())
                .unwrap();
            thread::sleep(Duration::from_micros(50)); // Small delay between updates
            db_write
                .put(b"account_b", (b + 10).to_string().as_bytes())
                .unwrap();
            thread::sleep(Duration::from_millis(1));
        }
    });

    let totals = reader.join().unwrap();
    writer.join().unwrap();

    // NOTE: Without transactions, reader may see inconsistent state during transfer
    // (e.g., total = 140 if it reads A after decrement but B before increment)
    // This is expected behavior for a non-transactional key-value store
    // We just check that values are reasonable (not corrupted)
    for total in totals {
        assert!(
            total >= 50 && total <= 150,
            "Total out of reasonable range: {}",
            total
        );
    }
}
