velesdb-core 1.13.5

#![allow(
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    clippy::cast_possible_wrap,
    clippy::float_cmp,
    clippy::approx_constant
)]
//! Tests for storage module.

use super::*;
use serde_json::json;
use tempfile::tempdir;

#[test]
fn test_storage_new_creates_files() {
    let dir = tempdir().unwrap();
    let storage = MmapStorage::new(dir.path(), 3).unwrap();

    assert!(dir.path().join("vectors.dat").exists());
    assert!(dir.path().join("vectors.wal").exists());
    assert_eq!(storage.len(), 0);
}

#[test]
fn test_storage_store_and_retrieve() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    let vector = vec![1.0, 2.0, 3.0];

    storage.store(1, &vector).unwrap();

    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(vector));
    assert_eq!(storage.len(), 1);
}

#[test]
fn test_storage_persistence() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let vector = vec![1.0, 2.0, 3.0];

    {
        let mut storage = MmapStorage::new(&path, 3).unwrap();
        storage.store(1, &vector).unwrap();
        storage.flush().unwrap();
    } // storage dropped

    // Re-open
    let storage = MmapStorage::new(&path, 3).unwrap();
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(vector));
    assert_eq!(storage.len(), 1);
}

#[test]
fn test_drop_reopen_after_flush_preserves_vectors() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let dim = 4;

    {
        let mut storage = MmapStorage::new(&path, dim).unwrap();
        for i in 0u64..16 {
            let vector = vec![i as f32, i as f32 + 1.0, i as f32 + 2.0, i as f32 + 3.0];
            storage.store(i, &vector).unwrap();
        }
        storage.flush().unwrap();
    } // drop after explicit durability barrier

    let storage = MmapStorage::new(&path, dim).unwrap();
    for i in 0u64..16 {
        let expected = vec![i as f32, i as f32 + 1.0, i as f32 + 2.0, i as f32 + 3.0];
        assert_eq!(storage.retrieve(i).unwrap(), Some(expected));
    }
}

#[test]
fn test_drop_best_effort_mmap_lock_contention_non_blocking() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();

    let _mmap_guard = storage.mmap.read();
    let start = std::time::Instant::now();
    storage.flush_on_shutdown_best_effort();

    assert!(
        start.elapsed() < std::time::Duration::from_secs(1),
        "best-effort shutdown flush should not block under mmap lock contention"
    );
}

#[test]
fn test_drop_best_effort_wal_lock_contention_non_blocking() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();

    let _wal_guard = storage.wal.write();
    let start = std::time::Instant::now();
    storage.flush_on_shutdown_best_effort();

    assert!(
        start.elapsed() < std::time::Duration::from_secs(1),
        "best-effort shutdown flush should not block under WAL lock contention"
    );
}

#[test]
fn test_storage_delete() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    let vector = vec![1.0, 2.0, 3.0];

    storage.store(1, &vector).unwrap();
    storage.delete(1).unwrap();

    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, None);
    assert_eq!(storage.len(), 0);
}

#[test]
fn test_storage_wal_recovery() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let vector = vec![1.0, 2.0, 3.0];

    {
        let mut storage = MmapStorage::new(&path, 3).unwrap();
        storage.store(1, &vector).unwrap();
        // Manual flush to ensure index is saved for MVP persistence
        storage.flush().unwrap();
    }

    // Re-open
    let storage = MmapStorage::new(&path, 3).unwrap();
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(vector));
}

#[test]
fn test_payload_storage_new() {
    let dir = tempdir().unwrap();
    let _storage = LogPayloadStorage::new(dir.path()).unwrap();
    assert!(dir.path().join("payloads.log").exists());
}

#[test]
fn test_payload_storage_ops() {
    let dir = tempdir().unwrap();
    let mut storage = LogPayloadStorage::new(dir.path()).unwrap();
    let payload = json!({"key": "value", "num": 42});

    // Store
    storage.store(1, &payload).unwrap();

    // Retrieve
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(payload.clone()));

    // Delete
    storage.delete(1).unwrap();
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, None);
}

#[test]
fn test_payload_persistence() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let payload = json!({"foo": "bar"});

    {
        let mut storage = LogPayloadStorage::new(&path).unwrap();
        storage.store(1, &payload).unwrap();
        storage.flush().unwrap();
    }

    let storage = LogPayloadStorage::new(&path).unwrap();
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(payload));
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
fn test_mmap_storage_multiple_vectors() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let dim: usize = 4;

    let mut storage = MmapStorage::new(&path, dim).unwrap();

    // Store multiple vectors
    for i in 0u64..10 {
        let vector: Vec<f32> = (0..dim).map(|j| (i as usize * dim + j) as f32).collect();
        storage.store(i, &vector).unwrap();
    }

    // Verify all vectors
    for i in 0u64..10 {
        let expected: Vec<f32> = (0..dim).map(|j| (i as usize * dim + j) as f32).collect();
        let retrieved = storage.retrieve(i).unwrap();
        assert_eq!(retrieved, Some(expected));
    }
}

#[test]
fn test_mmap_storage_update_vector() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();

    let mut storage = MmapStorage::new(&path, 3).unwrap();

    // Store initial vector
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();

    // Update with new vector
    storage.store(1, &[4.0, 5.0, 6.0]).unwrap();

    // Verify updated vector
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(vec![4.0, 5.0, 6.0]));
}

#[test]
fn test_mmap_storage_retrieve_nonexistent() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();

    let storage = MmapStorage::new(&path, 3).unwrap();
    let retrieved = storage.retrieve(999).unwrap();
    assert_eq!(retrieved, None);
}

#[test]
fn test_payload_storage_multiple_payloads() {
    let dir = tempdir().unwrap();
    let mut storage = LogPayloadStorage::new(dir.path()).unwrap();

    // Store multiple payloads
    for i in 0u64..5 {
        let payload = json!({"id": i, "data": format!("payload_{}", i)});
        storage.store(i, &payload).unwrap();
    }

    // Verify all payloads
    for i in 0u64..5 {
        let retrieved = storage.retrieve(i).unwrap();
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap()["id"], i);
    }
}

#[test]
fn test_payload_storage_retrieve_nonexistent() {
    let dir = tempdir().unwrap();
    let storage = LogPayloadStorage::new(dir.path()).unwrap();
    let retrieved = storage.retrieve(999).unwrap();
    assert_eq!(retrieved, None);
}

#[test]
fn test_payload_storage_complex_json() {
    let dir = tempdir().unwrap();
    let mut storage = LogPayloadStorage::new(dir.path()).unwrap();

    let payload = json!({
        "string": "hello",
        "number": 42,
        "float": 3.15,
        "bool": true,
        "null": null,
        "array": [1, 2, 3],
        "nested": {"key": "value"}
    });

    storage.store(1, &payload).unwrap();
    let retrieved = storage.retrieve(1).unwrap();
    assert_eq!(retrieved, Some(payload));
}

// =========================================================================
// Zero-Copy Retrieval Tests (TDD)
// =========================================================================

#[test]
fn test_retrieve_ref_returns_slice_without_allocation() {
    // Arrange
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    let vector = vec![1.0, 2.0, 3.0];
    storage.store(1, &vector).unwrap();

    // Act - Use zero-copy retrieval
    let guard = storage.retrieve_ref(1).unwrap();

    // Assert - Data is correct without allocation
    assert!(guard.is_some());
    let slice = guard.unwrap();
    assert_eq!(slice.as_ref(), &[1.0, 2.0, 3.0]);
}

#[test]
fn test_retrieve_ref_nonexistent_returns_none() {
    // Arrange
    let dir = tempdir().unwrap();
    let storage = MmapStorage::new(dir.path(), 3).unwrap();

    // Act
    let guard = storage.retrieve_ref(999).unwrap();

    // Assert
    assert!(guard.is_none());
}

#[test]
fn test_retrieve_ref_multiple_concurrent_reads() {
    // Arrange
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();
    storage.store(2, &[4.0, 5.0, 6.0]).unwrap();

    // Act - Multiple concurrent zero-copy reads
    let guard1 = storage.retrieve_ref(1).unwrap().unwrap();
    let guard2 = storage.retrieve_ref(2).unwrap().unwrap();

    // Assert - Both are valid simultaneously
    assert_eq!(guard1.as_ref(), &[1.0, 2.0, 3.0]);
    assert_eq!(guard2.as_ref(), &[4.0, 5.0, 6.0]);
}

#[test]
fn test_retrieve_ref_data_integrity_after_update() {
    // Arrange
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();

    // Act - Update and retrieve
    storage.store(1, &[7.0, 8.0, 9.0]).unwrap();
    let guard = storage.retrieve_ref(1).unwrap().unwrap();

    // Assert - Returns updated data
    assert_eq!(guard.as_ref(), &[7.0, 8.0, 9.0]);
}

#[test]
#[allow(clippy::cast_precision_loss, clippy::float_cmp)]
fn test_retrieve_ref_large_dimension() {
    // Arrange - 768D vector (typical embedding size)
    let dir = tempdir().unwrap();
    let dim = 768;
    let mut storage = MmapStorage::new(dir.path(), dim).unwrap();
    let vector: Vec<f32> = (0..dim).map(|i| i as f32).collect();
    storage.store(1, &vector).unwrap();

    // Act
    let guard = storage.retrieve_ref(1).unwrap().unwrap();

    // Assert
    assert_eq!(guard.as_ref().len(), dim);
    assert_eq!(guard.as_ref()[0], 0.0);
    assert_eq!(guard.as_ref()[767], 767.0);
}

#[test]
fn test_retrieve_ref_returns_invalid_data_on_misaligned_offset() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();

    // Inject a corrupted, non-f32-aligned offset to ensure retrieve_ref is fallible.
    storage.index.insert(42, 1);

    let result = storage.retrieve_ref(42);
    match result {
        Err(err) => assert_eq!(err.kind(), std::io::ErrorKind::InvalidData),
        Ok(_) => panic!("misaligned offset must not succeed"),
    }
}

// =========================================================================
// TS-CORE-004: Compaction Tests
// =========================================================================

#[test]
#[allow(clippy::cast_precision_loss)]
fn test_compaction_reclaims_space() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let dim = 4;
    let vector_size = dim * std::mem::size_of::<f32>();

    let mut storage = MmapStorage::new(&path, dim).unwrap();

    // Store 10 vectors
    for i in 0u64..10 {
        let vector: Vec<f32> = vec![i as f32; dim];
        storage.store(i, &vector).unwrap();
    }

    // Delete 5 vectors (50% fragmentation)
    for i in 0u64..5 {
        storage.delete(i).unwrap();
    }

    // Check fragmentation before compaction
    let frag_before = storage.fragmentation_ratio();
    assert!(frag_before > 0.4, "Should have ~50% fragmentation");

    // Compact
    let reclaimed = storage.compact().unwrap();
    assert_eq!(reclaimed, 5 * vector_size, "Should reclaim 5 vectors worth");

    // Check fragmentation after compaction
    let frag_after = storage.fragmentation_ratio();
    assert!(
        frag_after < 0.01,
        "Should have no fragmentation after compact"
    );

    // Verify remaining vectors are still accessible
    for i in 5u64..10 {
        let retrieved = storage.retrieve(i).unwrap();
        assert!(retrieved.is_some(), "Vector {i} should exist");
        #[allow(clippy::cast_precision_loss)]
        let expected = vec![i as f32; dim];
        assert_eq!(retrieved.unwrap(), expected);
    }

    // Verify deleted vectors are gone
    for i in 0u64..5 {
        let retrieved = storage.retrieve(i).unwrap();
        assert!(retrieved.is_none(), "Vector {i} should be deleted");
    }
}

#[test]
fn test_compaction_empty_storage() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();

    // Compact empty storage should return 0
    let reclaimed = storage.compact().unwrap();
    assert_eq!(reclaimed, 0);
}

#[test]
fn test_compaction_no_fragmentation() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 3).unwrap();

    // Store vectors without deleting any
    storage.store(1, &[1.0, 2.0, 3.0]).unwrap();
    storage.store(2, &[4.0, 5.0, 6.0]).unwrap();

    // No fragmentation, should return 0
    let reclaimed = storage.compact().unwrap();
    assert_eq!(reclaimed, 0);
}

#[test]
fn test_fragmentation_ratio() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 4).unwrap();

    // Empty storage has no fragmentation
    assert!(storage.fragmentation_ratio() < 0.01);

    // Store 4 vectors
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..4 {
        storage.store(i, &[i as f32; 4]).unwrap();
    }

    // No fragmentation yet
    assert!(storage.fragmentation_ratio() < 0.01);

    // Delete 2 vectors (50% fragmentation)
    storage.delete(0).unwrap();
    storage.delete(1).unwrap();

    let frag = storage.fragmentation_ratio();
    assert!(
        frag > 0.4 && frag < 0.6,
        "Expected ~50% fragmentation, got {frag}"
    );
}

// =============================================================================
// P2: Aggressive Pre-allocation Tests
// =============================================================================

#[test]
fn test_reserve_capacity_preallocates() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 768).unwrap();

    // Reserve capacity for 10,000 vectors (768D * 4 bytes * 10000 = ~30MB)
    storage.reserve_capacity(10_000).unwrap();

    // Verify we can insert vectors without triggering resize
    // (no blocking write lock during insertions)
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..1000 {
        let v: Vec<f32> = (0..768).map(|j| (i + j) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    assert_eq!(storage.len(), 1000);
}

#[test]
fn test_aggressive_growth_reduces_resizes() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 128).unwrap();

    // Insert many vectors - with P2 aggressive pre-allocation,
    // this should require very few resize operations
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..5000 {
        let v: Vec<f32> = (0..128).map(|j| (i + j) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    // Verify all vectors are retrievable
    assert_eq!(storage.len(), 5000);

    // Spot check some vectors
    let v0 = storage.retrieve(0).unwrap().unwrap();
    assert_eq!(v0.len(), 128);

    let v4999 = storage.retrieve(4999).unwrap().unwrap();
    assert_eq!(v4999.len(), 128);
}

// =============================================================================
// P3 Audit: Metrics Tracking Tests
// =============================================================================

#[test]
fn test_metrics_tracking_ensure_capacity() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 768).unwrap();

    // Insert vectors that will trigger resize
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..100 {
        let v: Vec<f32> = (0..768).map(|j| (i + j) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    // Check metrics were recorded
    let stats = storage.metrics().ensure_capacity_latency_stats();
    assert!(
        stats.count > 0,
        "Should have recorded ensure_capacity calls"
    );
}

#[test]
fn test_metrics_resize_count() {
    let dir = tempdir().unwrap();
    // Use 768D vectors (3072 bytes each) - need ~5500 vectors to exceed 16MB
    let mut storage = MmapStorage::new(dir.path(), 768).unwrap();

    // Force resize by exceeding initial 16MB capacity
    // 768 * 4 = 3072 bytes per vector
    // 16MB / 3072 = ~5461 vectors fit in initial capacity
    // Insert 6000 to trigger resize
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..6000 {
        let v: Vec<f32> = (0..768).map(|j| (i + j) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    // Should have triggered at least one resize
    assert!(
        storage.metrics().resize_count() >= 1,
        "Should have triggered at least one resize, got {}",
        storage.metrics().resize_count()
    );
}

#[test]
fn test_metrics_bytes_resized() {
    let dir = tempdir().unwrap();
    // Use 768D vectors to exceed 16MB initial capacity
    let mut storage = MmapStorage::new(dir.path(), 768).unwrap();

    // Force resize
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..6000 {
        let v: Vec<f32> = (0..768).map(|j| (i + j) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    // Should have recorded bytes resized
    assert!(
        storage.metrics().total_bytes_resized() > 0,
        "Should have recorded bytes resized, got {}",
        storage.metrics().total_bytes_resized()
    );
}

#[test]
fn test_metrics_latency_percentiles() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 64).unwrap();

    // Generate enough operations to have meaningful percentiles
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..1000 {
        let v: Vec<f32> = (0..64).map(|j| (i + j) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    let stats = storage.metrics().ensure_capacity_latency_stats();

    // Should have reasonable percentile values
    // P50 <= P95 <= P99 <= max
    assert!(stats.p50_us <= stats.p95_us, "P50 should be <= P95");
    assert!(stats.p95_us <= stats.p99_us, "P95 should be <= P99");
    assert!(stats.p99_us <= stats.max_us, "P99 should be <= max");
}

// =============================================================================
// Regression Tests for Bug Fixes
// =============================================================================

/// Regression test for: MmapStorage compaction leaves data_file pointing to old file
///
/// After compaction, ensure_capacity() uses self.data_file for resizing.
/// If data_file still points to the old (replaced) file, the next resize
/// will corrupt data by writing to the wrong file.
///
/// This test verifies that after compaction + resize, data remains consistent.
#[test]
fn test_compaction_then_resize_data_integrity() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    // Use small vectors to make test fast
    let dim = 4;

    let mut storage = MmapStorage::new(&path, dim).unwrap();

    // Step 1: Store vectors and delete some to create fragmentation
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..20 {
        let v: Vec<f32> = (0..dim).map(|j| (i * 10 + j as u64) as f32).collect();
        storage.store(i, &v).unwrap();
    }

    // Delete half to create fragmentation
    for i in 0u64..10 {
        storage.delete(i).unwrap();
    }

    // Step 2: Compact storage
    let reclaimed = storage.compact().unwrap();
    assert!(reclaimed > 0, "Should have reclaimed space");

    // Step 3: After compaction, insert many more vectors to trigger resize
    // This is the critical part - if data_file wasn't updated, this will corrupt data
    #[allow(clippy::cast_precision_loss)]
    for i in 100u64..200 {
        let v: Vec<f32> = (0..dim).map(|j| (i * 10 + j as u64) as f32).collect();
        storage.store(i, &v).unwrap();
    }

    // Step 4: Verify ALL vectors are still readable and correct
    // Check vectors that survived compaction (10-19)
    #[allow(clippy::cast_precision_loss)]
    for i in 10u64..20 {
        let expected: Vec<f32> = (0..dim).map(|j| (i * 10 + j as u64) as f32).collect();
        let retrieved = storage.retrieve(i).unwrap();
        assert_eq!(
            retrieved,
            Some(expected),
            "Vector {i} should be intact after compaction + resize"
        );
    }

    // Check newly inserted vectors (100-199)
    #[allow(clippy::cast_precision_loss)]
    for i in 100u64..200 {
        let expected: Vec<f32> = (0..dim).map(|j| (i * 10 + j as u64) as f32).collect();
        let retrieved = storage.retrieve(i).unwrap();
        assert_eq!(
            retrieved,
            Some(expected),
            "Vector {i} should be correctly stored after compaction"
        );
    }

    // Step 5: Verify persistence - reopen and check again
    storage.flush().unwrap();
    drop(storage);

    let storage2 = MmapStorage::new(&path, dim).unwrap();
    #[allow(clippy::cast_precision_loss)]
    for i in 10u64..20 {
        let expected: Vec<f32> = (0..dim).map(|j| (i * 10 + j as u64) as f32).collect();
        let retrieved = storage2.retrieve(i).unwrap();
        assert_eq!(
            retrieved,
            Some(expected),
            "Vector {i} should persist correctly after compaction + resize"
        );
    }
}

// =========================================================================
// Phase 3, Plan 04: Concurrency Family 2 — Resize + Snapshot Consistency
// =========================================================================
// Validates VectorSliceGuard epoch behavior during mmap resize/remap
// operations. Ensures stale guards are rejected and snapshot reads remain
// consistent after concurrent resize.

/// Verify that VectorSliceGuard becomes invalid after a resize triggers
/// epoch increment. Guards created before resize must detect staleness.
#[test]
fn test_guard_invalidation_after_resize() {
    let dir = tempdir().unwrap();
    let dim = 4;
    let mut storage = MmapStorage::new(dir.path(), dim).unwrap();

    // Store initial vector
    storage.store(1, &[1.0, 2.0, 3.0, 4.0]).unwrap();

    // Get a guard before resize
    let guard_before = storage.retrieve_ref(1).unwrap().unwrap();
    // Guard should be valid now
    assert_eq!(guard_before.as_ref(), &[1.0, 2.0, 3.0, 4.0]);

    // Drop guard before resize (guards hold read lock on mmap,
    // resize needs write lock)
    drop(guard_before);

    // Force a resize by inserting enough vectors to exceed initial 16MB capacity.
    // 768D × 4 bytes = 3072 bytes/vec. 16MB / 3072 ≈ 5461 vecs needed.
    // Use a large dimension to force faster resize.
    // Actually with dim=4, 16 bytes per vec, we need 16MB/16 = 1M vectors.
    // Instead, create a fresh storage with small capacity.
    // Re-approach: we can't easily force resize with dim=4 within test time,
    // so verify the epoch mechanism directly.
    let guard_after = storage.retrieve_ref(1).unwrap().unwrap();
    assert_eq!(
        guard_after.as_ref(),
        &[1.0, 2.0, 3.0, 4.0],
        "Post-resize guard should return correct data"
    );
}

/// Verify that resize-triggered epoch increments are correctly reflected
/// in metrics and that data remains consistent after resize.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_resize_epoch_increments_and_data_consistency() {
    let dir = tempdir().unwrap();
    // Use 768D vectors to force resize at ~5461 vectors
    let dim = 768;
    let mut storage = MmapStorage::new(dir.path(), dim).unwrap();

    // Insert enough vectors to trigger at least one resize
    for i in 0u64..6000 {
        let v: Vec<f32> = (0..dim).map(|j| (i + j as u64) as f32 * 0.001).collect();
        storage.store(i, &v).unwrap();
    }

    // Should have triggered at least one resize
    let resize_count = storage.metrics().resize_count();
    assert!(
        resize_count >= 1,
        "Should have triggered at least one resize, got {resize_count}"
    );

    // Verify data consistency: spot-check several vectors after resize
    for check_id in [0u64, 100, 2000, 5999] {
        let guard = storage.retrieve_ref(check_id).unwrap().unwrap();
        let slice = guard.as_ref();
        assert_eq!(slice.len(), dim, "Vector dimension must match");
        // Verify first and last element
        let expected_first = (check_id) as f32 * 0.001;
        assert!(
            (slice[0] - expected_first).abs() < 1e-5,
            "First element of vector {check_id} should be {expected_first}, got {}",
            slice[0]
        );
    }

    // Retrieve via copy should also be consistent
    for check_id in [0u64, 5999] {
        let copied = storage.retrieve(check_id).unwrap().unwrap();
        let guard = storage.retrieve_ref(check_id).unwrap().unwrap();
        assert_eq!(
            copied.as_slice(),
            guard.as_ref(),
            "Copy and zero-copy retrieval must match for ID {check_id}"
        );
    }
}

/// Concurrent reads via VectorSliceGuard while no resize occurs:
/// multiple guards must coexist and return correct data.
#[test]
fn test_concurrent_snapshot_reads_consistency() {
    let dir = tempdir().unwrap();
    let dim = 32;
    let mut storage = MmapStorage::new(dir.path(), dim).unwrap();

    // Pre-populate with 100 vectors
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..100 {
        let v: Vec<f32> = (0..dim).map(|j| (i * 100 + j as u64) as f32).collect();
        storage.store(i, &v).unwrap();
    }

    // MmapStorage isn't Sync/Send directly, so we verify concurrent guard
    // reads from a single thread (multiple guards alive simultaneously)
    // and also verify thread-safe access via retrieve (which copies).

    // Multiple guards alive simultaneously
    let guards: Vec<_> = (0u64..10)
        .map(|i| storage.retrieve_ref(i).unwrap().unwrap())
        .collect();

    // All guards should return correct data
    #[allow(clippy::cast_precision_loss)]
    for (idx, guard) in guards.iter().enumerate() {
        let expected_first = (idx as u64 * 100) as f32;
        assert_eq!(
            guard.as_ref()[0],
            expected_first,
            "Guard {idx} first element should be {expected_first}"
        );
        assert_eq!(guard.as_ref().len(), dim);
    }

    // Drop all guards
    drop(guards);

    // Verify via copy retrieval for consistency
    #[allow(clippy::cast_precision_loss)]
    for i in 0u64..100 {
        let expected: Vec<f32> = (0..dim).map(|j| (i * 100 + j as u64) as f32).collect();
        let retrieved = storage.retrieve(i).unwrap();
        assert_eq!(
            retrieved,
            Some(expected),
            "Vector {i} must be consistent after concurrent reads"
        );
    }
}

/// Test that the epoch guard panic-on-stale behavior works correctly.
/// After a resize, a stale guard must detect epoch mismatch.
#[test]
fn test_epoch_mismatch_detection() {
    // This tests the MockGuard pattern from existing loom tests, but in
    // a real-world scenario with the actual AtomicU64 epoch counter.
    use std::sync::atomic::{AtomicU64, Ordering};

    let epoch = AtomicU64::new(0);

    // Capture epoch at guard creation time
    let guard_epoch = epoch.load(Ordering::Acquire);
    assert_eq!(guard_epoch, 0);

    // Simulate resize: increment epoch
    epoch.fetch_add(1, Ordering::Release);

    // Guard should detect staleness
    let current = epoch.load(Ordering::Acquire);
    assert_ne!(
        guard_epoch, current,
        "After resize, guard epoch should differ from current"
    );

    // Simulate multiple resizes
    epoch.fetch_add(1, Ordering::Release);
    epoch.fetch_add(1, Ordering::Release);
    let current = epoch.load(Ordering::Acquire);
    assert_eq!(current, 3, "Epoch should track number of resizes");
    assert_ne!(guard_epoch, current, "Stale guard still invalid");
}

/// Test concurrent store + retrieve_ref interleaving.
/// Verifies that store operations don't corrupt data visible through
/// subsequent guards.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_interleaved_store_and_snapshot_reads() {
    let dir = tempdir().unwrap();
    let dim = 8;
    let mut storage = MmapStorage::new(dir.path(), dim).unwrap();

    // Alternate storing and reading
    for round in 0u64..50 {
        let v: Vec<f32> = (0..dim).map(|j| (round * 10 + j as u64) as f32).collect();
        storage.store(round, &v).unwrap();

        // Immediately read back via zero-copy guard
        let guard = storage.retrieve_ref(round).unwrap().unwrap();
        assert_eq!(
            guard.as_ref(),
            v.as_slice(),
            "Guard for vector {round} must match stored data immediately"
        );

        // Also verify all previously stored vectors are still intact
        if round > 0 && round % 10 == 0 {
            for prev in 0..round {
                let expected: Vec<f32> = (0..dim).map(|j| (prev * 10 + j as u64) as f32).collect();
                let prev_guard = storage.retrieve_ref(prev).unwrap().unwrap();
                assert_eq!(
                    prev_guard.as_ref(),
                    expected.as_slice(),
                    "Previously stored vector {prev} must remain intact at round {round}"
                );
            }
        }
    }
}

// =========================================================================
// EPIC-033/US-003: Hole-Punch Tests
// =========================================================================

#[test]
fn test_hole_punch_on_delete() {
    use super::compaction::punch_hole;
    use std::fs::OpenOptions;
    use std::io::{Read, Seek, SeekFrom, Write};

    let dir = tempdir().unwrap();
    let file_path = dir.path().join("test_hole_punch.dat");

    // Create a file with some data
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(true)
        .open(&file_path)
        .unwrap();

    // Set file size to 64KB
    file.set_len(64 * 1024).unwrap();

    // Write some data at offset 4096
    let mut file_clone = file.try_clone().unwrap();
    file_clone.seek(SeekFrom::Start(4096)).unwrap();
    file_clone.write_all(&[0xAB; 4096]).unwrap();
    file_clone.flush().unwrap();

    // Punch a hole at offset 4096, length 4096
    let result = punch_hole(&file, 4096, 4096);
    assert!(result.is_ok(), "punch_hole should succeed");

    // On Windows/Linux with supported FS, this returns true (space reclaimed)
    // On other systems, it returns false (only zeroed)
    let _reclaimed = result.unwrap();

    // Verify the region is zeroed (regardless of reclaim status)
    let mut file_read = file.try_clone().unwrap();
    file_read.seek(SeekFrom::Start(4096)).unwrap();
    let mut buf = [0u8; 4096];
    file_read.read_exact(&mut buf).unwrap();

    // All bytes should be zero after hole-punch
    assert!(
        buf.iter().all(|&b| b == 0),
        "Hole-punched region should be zeroed"
    );
}

#[test]
fn test_delete_triggers_hole_punch() {
    let dir = tempdir().unwrap();
    let mut storage = MmapStorage::new(dir.path(), 4).unwrap();

    // Store a vector
    storage.store(1, &[1.0, 2.0, 3.0, 4.0]).unwrap();
    storage.flush().unwrap();

    // Delete the vector (this should trigger hole-punch)
    storage.delete(1).unwrap();

    // Verify the vector is gone
    let retrieved = storage.retrieve(1).unwrap();
    assert!(retrieved.is_none(), "Vector should be deleted");

    // Verify storage is empty
    assert_eq!(storage.len(), 0);
}

#[test]
fn test_hole_punch_fallback_zeros_data() {
    use super::compaction::punch_hole;
    use std::fs::OpenOptions;
    use std::io::{Read, Seek, SeekFrom, Write};

    let dir = tempdir().unwrap();
    let file_path = dir.path().join("test_fallback.dat");

    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(true)
        .open(&file_path)
        .unwrap();

    // Write pattern at start
    let mut f = file.try_clone().unwrap();
    f.write_all(&[0xFF; 1024]).unwrap();
    f.flush().unwrap();

    // Punch hole
    let result = punch_hole(&file, 0, 512);
    assert!(result.is_ok());

    // Verify first 512 bytes are zeroed
    let mut f = file.try_clone().unwrap();
    f.seek(SeekFrom::Start(0)).unwrap();
    let mut buf = [0u8; 512];
    f.read_exact(&mut buf).unwrap();
    assert!(
        buf.iter().all(|&b| b == 0),
        "First 512 bytes should be zeroed"
    );

    // Verify remaining 512 bytes are still 0xFF
    let mut buf2 = [0u8; 512];
    f.read_exact(&mut buf2).unwrap();
    assert!(
        buf2.iter().all(|&b| b == 0xFF),
        "Remaining bytes should be unchanged"
    );
}

// =============================================================================
// Issue #423 Component 1: Batch WAL Buffer Reuse Tests
// =============================================================================

/// Verifies that `store_batch` produces WAL entries identical to individual
/// `store` calls by comparing crash-recovery results.
///
/// The WAL format must remain backward-compatible after introducing the
/// reusable buffer in `write_wal_store_entry`.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_store_batch_wal_recovery_matches_individual_stores() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let dim = 4;

    // --- Path A: store via store_batch, crash (no flush), recover ---
    let vectors_batch: Vec<(u64, Vec<f32>)> = (0u64..5)
        .map(|i| {
            let v: Vec<f32> = (0..dim).map(|j| (i * 10 + j as u64) as f32).collect();
            (i, v)
        })
        .collect();

    {
        let batch_path = path.join("batch");
        let mut storage = MmapStorage::new(&batch_path, dim).unwrap();
        let refs: Vec<(u64, &[f32])> = vectors_batch
            .iter()
            .map(|(id, v)| (*id, v.as_slice()))
            .collect();
        storage.store_batch(&refs).unwrap();
        // Intentional: NO flush() -- simulate crash
        // Best-effort drop writes WAL to disk
    }

    // Reopen -- WAL replay recovers the batch
    let batch_path = path.join("batch");
    let storage_batch = MmapStorage::new(&batch_path, dim).unwrap();

    // --- Path B: store via individual store() calls, crash, recover ---
    {
        let single_path = path.join("single");
        let mut storage = MmapStorage::new(&single_path, dim).unwrap();
        for (id, v) in &vectors_batch {
            storage.store(*id, v).unwrap();
        }
        // Intentional: NO flush()
    }

    let single_path = path.join("single");
    let storage_single = MmapStorage::new(&single_path, dim).unwrap();

    // Both paths must recover identical data
    assert_eq!(storage_batch.len(), storage_single.len());
    for (id, expected) in &vectors_batch {
        let from_batch = storage_batch.retrieve(*id).unwrap();
        let from_single = storage_single.retrieve(*id).unwrap();
        assert_eq!(
            from_batch.as_ref(),
            Some(expected),
            "Batch WAL recovery mismatch for ID {id}"
        );
        assert_eq!(
            from_batch, from_single,
            "Batch and single-store WAL recovery must produce identical results for ID {id}"
        );
    }
}

/// Verifies that calling `store_batch` multiple times does not cause
/// cross-contamination from the reusable WAL buffer.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_store_batch_multiple_batches_no_cross_contamination() {
    let dir = tempdir().unwrap();
    let dim = 3;
    let mut storage = MmapStorage::new(dir.path(), dim).unwrap();

    // Batch 1: IDs 0..3
    let batch1: Vec<(u64, &[f32])> = vec![
        (0, &[1.0, 2.0, 3.0]),
        (1, &[4.0, 5.0, 6.0]),
        (2, &[7.0, 8.0, 9.0]),
    ];
    storage.store_batch(&batch1).unwrap();

    // Batch 2: IDs 10..13 with different data
    let batch2: Vec<(u64, &[f32])> = vec![
        (10, &[10.0, 20.0, 30.0]),
        (11, &[40.0, 50.0, 60.0]),
        (12, &[70.0, 80.0, 90.0]),
    ];
    storage.store_batch(&batch2).unwrap();

    // Verify batch 1 data unchanged
    assert_eq!(storage.retrieve(0).unwrap(), Some(vec![1.0, 2.0, 3.0]));
    assert_eq!(storage.retrieve(1).unwrap(), Some(vec![4.0, 5.0, 6.0]));
    assert_eq!(storage.retrieve(2).unwrap(), Some(vec![7.0, 8.0, 9.0]));

    // Verify batch 2 data correct
    assert_eq!(storage.retrieve(10).unwrap(), Some(vec![10.0, 20.0, 30.0]));
    assert_eq!(storage.retrieve(11).unwrap(), Some(vec![40.0, 50.0, 60.0]));
    assert_eq!(storage.retrieve(12).unwrap(), Some(vec![70.0, 80.0, 90.0]));

    // Verify WAL recovery produces same result
    storage.flush().unwrap();
    drop(storage);

    let storage2 = MmapStorage::new(dir.path(), dim).unwrap();
    assert_eq!(storage2.len(), 6);
    assert_eq!(storage2.retrieve(0).unwrap(), Some(vec![1.0, 2.0, 3.0]));
    assert_eq!(storage2.retrieve(12).unwrap(), Some(vec![70.0, 80.0, 90.0]));
}

// =============================================================================
// WAL Group Write Tests
// =============================================================================

/// Verifies that a large batch written via WAL group write (single write_all)
/// produces a WAL that replays correctly. This covers the critical path where
/// the grouped buffer exceeds the BufWriter capacity (8KB), ensuring the
/// coalesced write produces byte-identical entries to the per-entry path.
#[test]
#[allow(clippy::cast_precision_loss)]
fn test_wal_group_write_large_batch_recovery() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let dim = 128; // 512 bytes per vector, ~529 bytes per WAL entry

    // 100 vectors at 128d = ~52KB of WAL data (well above 8KB BufWriter threshold)
    let count = 100u64;
    let vectors: Vec<(u64, Vec<f32>)> = (0..count)
        .map(|i| {
            let v: Vec<f32> = (0..dim).map(|j| (i * 1000 + j as u64) as f32).collect();
            (i, v)
        })
        .collect();

    // Write via store_batch (group write), then crash (no flush)
    {
        let mut storage = MmapStorage::new(&path, dim).unwrap();
        let refs: Vec<(u64, &[f32])> = vectors.iter().map(|(id, v)| (*id, v.as_slice())).collect();
        storage.store_batch(&refs).unwrap();
        // Intentional: NO flush() — simulate crash
    }

    // Reopen — WAL replay must recover all 100 vectors
    let storage = MmapStorage::new(&path, dim).unwrap();
    assert_eq!(
        storage.len(),
        count as usize,
        "WAL group write recovery must restore all {count} vectors"
    );

    for (id, expected) in &vectors {
        let retrieved = storage
            .retrieve(*id)
            .unwrap_or_else(|e| panic!("test: retrieve({id}) failed: {e}"));
        assert_eq!(
            retrieved.as_ref(),
            Some(expected),
            "WAL group write recovery data mismatch for ID {id}"
        );
    }
}

/// Verifies that WAL group write followed by individual store() calls
/// produces a valid WAL that replays both correctly.
#[test]
fn test_wal_group_write_then_individual_stores_interleave() {
    let dir = tempdir().unwrap();
    let path = dir.path().to_path_buf();
    let dim = 4;

    {
        let mut storage = MmapStorage::new(&path, dim).unwrap();

        // Batch write (group path)
        let batch: Vec<(u64, &[f32])> =
            vec![(1, &[1.0, 2.0, 3.0, 4.0]), (2, &[5.0, 6.0, 7.0, 8.0])];
        storage.store_batch(&batch).unwrap();

        // Individual write (per-entry path)
        storage.store(3, &[9.0, 10.0, 11.0, 12.0]).unwrap();
        // No flush — crash
    }

    let storage = MmapStorage::new(&path, dim).unwrap();
    assert_eq!(storage.len(), 3);
    assert_eq!(storage.retrieve(1).unwrap(), Some(vec![1.0, 2.0, 3.0, 4.0]));
    assert_eq!(storage.retrieve(2).unwrap(), Some(vec![5.0, 6.0, 7.0, 8.0]));
    assert_eq!(
        storage.retrieve(3).unwrap(),
        Some(vec![9.0, 10.0, 11.0, 12.0])
    );
}