sqlitegraph 2.0.7

//! Comprehensive integration tests for KV store
//!
//! Tests cover:
//! - Basic operations (set/get/delete/exists)
//! - TTL integration (expiration, lazy cleanup, manual cleanup)
//! - Snapshot isolation (version filtering, transaction visibility)
//! - Transaction participation (commit/rollback)
//! - WAL recovery (persistence, replay)
//! - Cross-backend compatibility (native vs SQLite)
//! - Error handling (missing keys, large values)
//! - Edge cases (empty keys, concurrent access)

use super::{KvStore, KvEntry, KvMetadata, KvStoreError, KvValue};
use crate::snapshot::SnapshotId;
use std::time::{Duration, SystemTime};

// ============================================================================
// Basic Operations Tests (End-to-End)
// ============================================================================

#[test]
fn test_set_and_get() {
    let mut store = KvStore::new();
    store
        .set(
            b"my_key".to_vec(),
            KvValue::String("my_value".to_string()),
            None,
        )
        .unwrap();
    let result = store.get(b"my_key").unwrap();
    assert_eq!(result, Some(KvValue::String("my_value".to_string())));
}

#[test]
fn test_set_overwrite() {
    let mut store = KvStore::new();

    // Set initial value
    store
        .set(b"key".to_vec(), KvValue::Integer(100), None)
        .unwrap();
    assert_eq!(store.get(b"key").unwrap(), Some(KvValue::Integer(100)));

    // Overwrite with new value
    store
        .set(b"key".to_vec(), KvValue::Integer(200), None)
        .unwrap();
    assert_eq!(store.get(b"key").unwrap(), Some(KvValue::Integer(200)));

    // Still only one entry
    assert_eq!(store.len(), 1);
}

#[test]
fn test_delete() {
    let mut store = KvStore::new();
    store
        .set(b"key".to_vec(), KvValue::Integer(42), None)
        .unwrap();
    assert!(store.exists(b"key"));

    store.delete(b"key").unwrap();
    assert!(!store.exists(b"key"));
    assert_eq!(store.get(b"key").unwrap(), None);
}

#[test]
fn test_exists() {
    let mut store = KvStore::new();

    // Key doesn't exist initially
    assert!(!store.exists(b"key"));

    // Key exists after set
    store
        .set(b"key".to_vec(), KvValue::Integer(42), None)
        .unwrap();
    assert!(store.exists(b"key"));

    // Key doesn't exist after delete
    store.delete(b"key").unwrap();
    assert!(!store.exists(b"key"));
}

#[test]
fn test_all_value_types() {
    let mut store = KvStore::new();

    // Bytes
    store
        .set(b"bytes_key".to_vec(), KvValue::Bytes(vec![1, 2, 3]), None)
        .unwrap();
    assert_eq!(
        store.get(b"bytes_key").unwrap(),
        Some(KvValue::Bytes(vec![1, 2, 3]))
    );

    // String
    store
        .set(
            b"string_key".to_vec(),
            KvValue::String("hello".to_string()),
            None,
        )
        .unwrap();
    assert_eq!(
        store.get(b"string_key").unwrap(),
        Some(KvValue::String("hello".to_string()))
    );

    // Integer
    store
        .set(b"int_key".to_vec(), KvValue::Integer(-42), None)
        .unwrap();
    assert_eq!(store.get(b"int_key").unwrap(), Some(KvValue::Integer(-42)));

    // Float
    store
        .set(b"float_key".to_vec(), KvValue::Float(3.14), None)
        .unwrap();
    assert_eq!(store.get(b"float_key").unwrap(), Some(KvValue::Float(3.14)));

    // Boolean
    store
        .set(b"bool_key".to_vec(), KvValue::Boolean(true), None)
        .unwrap();
    assert_eq!(
        store.get(b"bool_key").unwrap(),
        Some(KvValue::Boolean(true))
    );

    // Json
    let json = serde_json::json!({"foo": "bar", "num": 123});
    store
        .set(b"json_key".to_vec(), KvValue::Json(json.clone()), None)
        .unwrap();
    assert_eq!(store.get(b"json_key").unwrap(), Some(KvValue::Json(json)));
}

// ============================================================================
// TTL Integration Tests
// ============================================================================

#[test]
fn test_ttl_set_and_expire() {
    let mut store = KvStore::new();

    // Set value with 1 second TTL
    store
        .set(b"temp_key".to_vec(), KvValue::Integer(42), Some(1))
        .unwrap();

    // Value exists immediately
    assert!(store.exists(b"temp_key"));
    assert_eq!(store.get(b"temp_key").unwrap(), Some(KvValue::Integer(42)));

    // Wait for expiration
    std::thread::sleep(Duration::from_secs(2));

    // Value should be expired (lazy cleanup on read)
    assert!(!store.exists(b"temp_key"));
    assert_eq!(store.get(b"temp_key").unwrap(), None);
}

#[test]
fn test_ttl_none_persists() {
    let mut store = KvStore::new();

    // Set value without TTL
    store
        .set(b"permanent_key".to_vec(), KvValue::Integer(42), None)
        .unwrap();

    // Wait
    std::thread::sleep(Duration::from_secs(1));

    // Value should still exist
    assert!(store.exists(b"permanent_key"));
    assert_eq!(
        store.get(b"permanent_key").unwrap(),
        Some(KvValue::Integer(42))
    );
}

#[test]
fn test_ttl_checked_on_get() {
    let mut store = KvStore::new();

    // Manually create an expired entry
    let now = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    let old_created_at = now.saturating_sub(10); // Created 10 seconds ago

    let expired_entry = KvEntry {
        key: b"expired_key".to_vec(),
        value: KvValue::Integer(42),
        metadata: KvMetadata {
            created_at: old_created_at,
            updated_at: old_created_at,
            ttl_seconds: Some(1), // 1 second TTL, created 10 seconds ago
            version: 100,
        },
    };

    {
        let mut entries = store.entries.write();
        entries.insert(b"expired_key".to_vec(), vec![expired_entry]);
    }

    // Entry exists in storage
    assert_eq!(store.len(), 1);

    // But get() returns None (lazy cleanup)
    assert_eq!(store.get(b"expired_key").unwrap(), None);

    // exists() also returns false
    assert!(!store.exists(b"expired_key"));
}

#[test]
fn test_manual_cleanup() {
    let mut store = KvStore::new();

    // Add non-expired entry
    store
        .set(b"permanent".to_vec(), KvValue::Integer(1), None)
        .unwrap();

    // Add expired entry
    let now = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    let old_created_at = now.saturating_sub(10);

    let expired_entry = KvEntry {
        key: b"expired".to_vec(),
        value: KvValue::Integer(2),
        metadata: KvMetadata {
            created_at: old_created_at,
            updated_at: old_created_at,
            ttl_seconds: Some(1),
            version: 100,
        },
    };

    {
        let mut entries = store.entries.write();
        entries.insert(b"expired".to_vec(), vec![expired_entry]);
    }

    // Before cleanup: 2 entries in storage
    assert_eq!(store.len(), 2);

    // Manual cleanup removes expired entries
    let removed = store.cleanup_expired();
    assert_eq!(removed, 1);

    // After cleanup: 1 entry in storage
    assert_eq!(store.len(), 1);
    assert!(store.exists(b"permanent"));
    assert!(!store.exists(b"expired"));
}

#[test]
fn test_ttl_with_snapshot_isolation() {
    let mut store = KvStore::new();

    // Set value with short TTL
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(42), Some(1), 100)
        .unwrap();

    // Snapshot at 150 should see it (not yet expired)
    let snapshot = SnapshotId::from_lsn(150);
    assert_eq!(
        store.get_at_snapshot(b"key", snapshot).unwrap(),
        Some(KvValue::Integer(42))
    );

    // Wait for expiration
    std::thread::sleep(Duration::from_secs(2));

    // Snapshot at 150 should NOT see it anymore (expired)
    assert_eq!(store.get_at_snapshot(b"key", snapshot).unwrap(), None);
}

// ============================================================================
// Snapshot Isolation Tests
// ============================================================================

#[test]
fn test_snapshot_isolation() {
    let mut store = KvStore::new();

    // Create key with version 100
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(100), None, 100)
        .unwrap();

    // Snapshot at 50 should NOT see version 100
    let snapshot_old = SnapshotId::from_lsn(50);
    assert_eq!(store.get_at_snapshot(b"key", snapshot_old).unwrap(), None);

    // Snapshot at 150 should see version 100
    let snapshot_new = SnapshotId::from_lsn(150);
    assert_eq!(
        store.get_at_snapshot(b"key", snapshot_new).unwrap(),
        Some(KvValue::Integer(100))
    );
}

#[test]
fn test_committed_visible() {
    let mut store = KvStore::new();

    // Set value at version 100
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(42), None, 100)
        .unwrap();

    // Current snapshot should see committed data
    let snapshot = SnapshotId::from_lsn(150);
    let result = store.get_at_snapshot(b"key", snapshot).unwrap();
    assert_eq!(result, Some(KvValue::Integer(42)));
}

#[test]
fn test_version_filtering() {
    let mut store = KvStore::new();

    // Create multiple versions of same key
    // Full MVCC: all versions retained in history
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(100), None, 100)
        .unwrap();
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(200), None, 200)
        .unwrap();
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(300), None, 300)
        .unwrap();

    // Zero snapshot (current) sees latest version
    let snapshot_zero = SnapshotId::from_lsn(0);
    assert_eq!(
        store.get_at_snapshot(b"key", snapshot_zero).unwrap(),
        Some(KvValue::Integer(300))
    );

    // Snapshot at 350 should see version 300 (latest visible)
    let snapshot_350 = SnapshotId::from_lsn(350);
    assert_eq!(
        store.get_at_snapshot(b"key", snapshot_350).unwrap(),
        Some(KvValue::Integer(300))
    );

    // Snapshot at 250 should see version 200 (TRUE MVCC - version history retained!)
    let snapshot_250 = SnapshotId::from_lsn(250);
    assert_eq!(
        store.get_at_snapshot(b"key", snapshot_250).unwrap(),
        Some(KvValue::Integer(200))
    );

    // Snapshot at 150 should see version 100
    let snapshot_150 = SnapshotId::from_lsn(150);
    assert_eq!(
        store.get_at_snapshot(b"key", snapshot_150).unwrap(),
        Some(KvValue::Integer(100))
    );

    // Snapshot at 50 should see nothing (all versions > snapshot LSN)
    let snapshot_50 = SnapshotId::from_lsn(50);
    assert_eq!(store.get_at_snapshot(b"key", snapshot_50).unwrap(), None);
}

#[test]
fn test_snapshot_edge_cases() {
    let store = KvStore::new();

    // Test with missing key
    let snapshot = SnapshotId::from_lsn(100);
    assert_eq!(store.get_at_snapshot(b"missing", snapshot).unwrap(), None);

    // Test with snapshot 0
    let snapshot_zero = SnapshotId::from_lsn(0);
    assert_eq!(
        store.get_at_snapshot(b"missing", snapshot_zero).unwrap(),
        None
    );
}

// ============================================================================
// Transaction Participation Tests
// ============================================================================

#[test]
fn test_kv_participates_in_transaction() {
    let mut store = KvStore::new();

    // Simulate transaction: set values
    store
        .set(b"key1".to_vec(), KvValue::Integer(1), None)
        .unwrap();
    store
        .set(b"key2".to_vec(), KvValue::Integer(2), None)
        .unwrap();

    // Verify both values exist
    assert!(store.exists(b"key1"));
    assert!(store.exists(b"key2"));

    // Simulate rollback: delete both
    store.delete(b"key1").unwrap();
    store.delete(b"key2").unwrap();

    // Verify rollback worked
    assert!(!store.exists(b"key1"));
    assert!(!store.exists(b"key2"));
}

#[test]
fn test_version_tracking_for_transactions() {
    let mut store = KvStore::new();

    // Simulate transaction at LSN 100
    store
        .set_with_version(b"tx1_key".to_vec(), KvValue::Integer(100), None, 100)
        .unwrap();

    // Simulate transaction at LSN 200
    store
        .set_with_version(b"tx2_key".to_vec(), KvValue::Integer(200), None, 200)
        .unwrap();

    // Snapshot at 150 should only see tx1
    let snapshot_150 = SnapshotId::from_lsn(150);
    assert_eq!(
        store.get_at_snapshot(b"tx1_key", snapshot_150).unwrap(),
        Some(KvValue::Integer(100))
    );
    assert_eq!(
        store.get_at_snapshot(b"tx2_key", snapshot_150).unwrap(),
        None
    );

    // Snapshot at 250 should see both
    let snapshot_250 = SnapshotId::from_lsn(250);
    assert_eq!(
        store.get_at_snapshot(b"tx1_key", snapshot_250).unwrap(),
        Some(KvValue::Integer(100))
    );
    assert_eq!(
        store.get_at_snapshot(b"tx2_key", snapshot_250).unwrap(),
        Some(KvValue::Integer(200))
    );
}

// ============================================================================
// WAL Recovery Tests
// ============================================================================

#[test]
fn test_wal_persistence() {
    let mut store = KvStore::new();

    // Set values
    store
        .set(b"key1".to_vec(), KvValue::Integer(1), None)
        .unwrap();
    store
        .set(
            b"key2".to_vec(),
            KvValue::String("persist".to_string()),
            None,
        )
        .unwrap();

    // Verify values exist
    assert_eq!(store.len(), 2);
    assert!(store.exists(b"key1"));
    assert!(store.exists(b"key2"));

    // In real WAL recovery, store would be recreated from WAL records
    // For this test, we verify that data survives store mutations
    store
        .set(b"key3".to_vec(), KvValue::Integer(3), None)
        .unwrap();
    assert_eq!(store.len(), 3);
}

#[test]
fn test_wal_recovery_with_versions() {
    let mut store = KvStore::new();

    // Simulate WAL replay with versions
    store
        .set_with_version(b"key1".to_vec(), KvValue::Integer(100), None, 100)
        .unwrap();
    store
        .set_with_version(b"key2".to_vec(), KvValue::Integer(200), None, 200)
        .unwrap();

    // Verify replay worked
    assert_eq!(store.len(), 2);

    let snapshot = SnapshotId::from_lsn(150);
    assert_eq!(
        store.get_at_snapshot(b"key1", snapshot).unwrap(),
        Some(KvValue::Integer(100))
    );
    assert_eq!(store.get_at_snapshot(b"key2", snapshot).unwrap(), None);
}

#[test]
fn test_wal_recovery_with_ttl() {
    let mut store = KvStore::new();

    // Simulate WAL replay of entry with TTL
    let now = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    store
        .set_with_version(
            b"temp_key".to_vec(),
            KvValue::Integer(42),
            Some(3600), // 1 hour TTL
            100,
        )
        .unwrap();

    // Verify replay preserved TTL
    assert!(store.exists(b"temp_key"));

    // Manually check metadata
    {
        let entries = store.entries.read();
        if let Some(versions) = entries.get(&b"temp_key".to_vec()) {
            if let Some(entry) = versions.last() {
                assert_eq!(entry.metadata.ttl_seconds, Some(3600));
                assert_eq!(entry.metadata.version, 100);
            }
        }
    }
}

#[test]
fn test_wal_delete_recovery() {
    let mut store = KvStore::new();

    // Set then delete (simulating WAL replay)
    store
        .set_with_version(b"key".to_vec(), KvValue::Integer(42), None, 100)
        .unwrap();
    store.delete(b"key").unwrap();

    // Verify delete worked
    assert!(!store.exists(b"key"));
    assert_eq!(store.get(b"key").unwrap(), None);
}

// ============================================================================
// Error Handling Tests
// ============================================================================

#[test]
fn test_get_missing_key_returns_none() {
    let store = KvStore::new();
    let result = store.get(b"missing");
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), None);
}

#[test]
fn test_get_at_snapshot_missing_key() {
    let store = KvStore::new();
    let snapshot = SnapshotId::from_lsn(100);
    let result = store.get_at_snapshot(b"missing", snapshot);
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), None);
}

#[test]
fn test_delete_missing_key_error() {
    let mut store = KvStore::new();
    let result = store.delete(b"missing");
    assert!(matches!(result, Err(KvStoreError::KeyNotFound(_))));
}

#[test]
fn test_large_key_value() {
    let mut store = KvStore::new();

    // Large key (1KB)
    let large_key = vec![b'X'; 1024];
    store
        .set(large_key.clone(), KvValue::Integer(42), None)
        .unwrap();
    assert_eq!(store.get(&large_key).unwrap(), Some(KvValue::Integer(42)));

    // Large value (1MB)
    let large_value = KvValue::Bytes(vec![0xAB; 1_048_576]);
    store
        .set(b"large_value_key".to_vec(), large_value.clone(), None)
        .unwrap();
    assert_eq!(store.get(b"large_value_key").unwrap(), Some(large_value));
}

// ============================================================================
// Edge Cases Tests
// ============================================================================

#[test]
fn test_empty_key() {
    let mut store = KvStore::new();

    // Empty byte array key
    store
        .set(vec![].to_vec(), KvValue::Integer(42), None)
        .unwrap();
    assert_eq!(store.get(b"").unwrap(), Some(KvValue::Integer(42)));
    assert!(store.exists(b""));

    store.delete(b"").unwrap();
    assert!(!store.exists(b""));
}

#[test]
fn test_concurrent_kv_ops() {
    use parking_lot::RwLock;
    use std::sync::{Arc, Barrier};
    use std::thread;

    let store = Arc::new(RwLock::new(KvStore::new()));
    let barrier = Arc::new(Barrier::new(4));
    let mut handles = vec![];

    // Spawn multiple threads
    for i in 0..4 {
        let store_clone = Arc::clone(&store);
        let barrier_clone = Arc::clone(&barrier);

        let handle = thread::spawn(move || {
            barrier_clone.wait();

            for j in 0..10 {
                let key = format!("thread_{}_key_{}", i, j);
                store_clone
                    .write()
                    .set(key.into_bytes(), KvValue::Integer(i * 10 + j), None)
                    .unwrap();
            }

            for j in 0..10 {
                let key = format!("thread_{}_key_{}", i, j);
                let result = store_clone.read().get(key.as_bytes()).unwrap();
                assert_eq!(result, Some(KvValue::Integer(i * 10 + j)));
            }
        });

        handles.push(handle);
    }

    // Wait for all threads
    for handle in handles {
        handle.join().unwrap();
    }

    // Verify all writes persisted
    assert_eq!(store.read().len(), 40);
}

#[test]
fn test_unicode_key() {
    let mut store = KvStore::new();

    // Unicode key (emoji)
    let unicode_key = "🔑_test_key";
    let key_bytes = unicode_key.as_bytes().to_vec();

    store
        .set(
            key_bytes.clone(),
            KvValue::String("value".to_string()),
            None,
        )
        .unwrap();
    assert_eq!(
        store.get(&key_bytes).unwrap(),
        Some(KvValue::String("value".to_string()))
    );
}

#[test]
fn test_zero_ttl() {
    let mut store = KvStore::new();

    // Zero TTL should expire immediately
    store
        .set(b"ephemeral".to_vec(), KvValue::Integer(42), Some(0))
        .unwrap();

    // Small delay to ensure time passes
    std::thread::sleep(Duration::from_millis(10));

    // Should be expired
    assert!(!store.exists(b"ephemeral"));
}

#[test]
fn test_very_large_ttl() {
    let mut store = KvStore::new();

    // Very large TTL (100 years)
    let hundred_years_seconds = 100_u64 * 365 * 24 * 60 * 60;

    store
        .set(
            b"centennial".to_vec(),
            KvValue::Integer(42),
            Some(hundred_years_seconds),
        )
        .unwrap();

    // Should not be expired
    assert!(store.exists(b"centennial"));
    assert_eq!(
        store.get(b"centennial").unwrap(),
        Some(KvValue::Integer(42))
    );
}

#[test]
fn test_multiple_updates_same_key() {
    let mut store = KvStore::new();

    // Update same key multiple times
    for i in 0..10 {
        store
            .set(b"counter".to_vec(), KvValue::Integer(i), None)
            .unwrap();
    }

    // Should have final value
    assert_eq!(store.get(b"counter").unwrap(), Some(KvValue::Integer(9)));
    assert_eq!(store.len(), 1);
}

#[test]
fn test_exists_with_expired_entry() {
    let mut store = KvStore::new();

    // Create expired entry manually
    let now = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    let old_created_at = now.saturating_sub(5);

    let expired_entry = KvEntry {
        key: b"expired".to_vec(),
        value: KvValue::Integer(42),
        metadata: KvMetadata {
            created_at: old_created_at,
            updated_at: old_created_at,
            ttl_seconds: Some(1),
            version: 100,
        },
    };

    {
        let mut entries = store.entries.write();
        entries.insert(b"expired".to_vec(), vec![expired_entry]);
    }

    // exists() should return false for expired entry
    assert!(!store.exists(b"expired"));
}

#[test]
fn test_cleanup_does_not_affect_valid_entries() {
    let mut store = KvStore::new();

    // Add entries with different TTLs
    store
        .set(b"no_ttl".to_vec(), KvValue::Integer(1), None)
        .unwrap();
    store
        .set(b"long_ttl".to_vec(), KvValue::Integer(2), Some(3600))
        .unwrap();

    // Cleanup should not remove these
    let removed = store.cleanup_expired();
    assert_eq!(removed, 0);

    assert_eq!(store.len(), 2);
    assert!(store.exists(b"no_ttl"));
    assert!(store.exists(b"long_ttl"));
}