kimberlite-storage 0.4.2

//! Unit tests for kmb-storage
//!
//! Tests for the append-only segment storage layer with hash chain integrity.

use bytes::Bytes;
use kimberlite_crypto::ChainHash;
use kimberlite_types::{Offset, StreamId};
use tempfile::TempDir;

use crate::{OffsetIndex, Record, Storage, StorageError};

// ============================================================================
// Record Serialization Tests
// ============================================================================

#[test]
fn record_to_bytes_produces_correct_format() {
    let record = Record::new(Offset::new(42), None, Bytes::from("hello"));
    let bytes = record.to_bytes();

    // Total size (AUDIT-2026-03 M-8 with sentinels):
    // 4 (start_sentinel) + 8 (offset) + 32 (prev_hash) + 1 (kind) + 1 (compression) + 4 (len) + 5 (payload) + 4 (crc) + 4 (end_sentinel) = 63 bytes
    assert_eq!(bytes.len(), 63);

    // First 4 bytes: RECORD_START sentinel (0xBADC0FFE)
    let start_sentinel = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
    assert_eq!(start_sentinel, 0xBADC_0FFE);

    // Next 8 bytes: offset (42 in little-endian)
    let offset = u64::from_le_bytes(bytes[4..12].try_into().unwrap());
    assert_eq!(offset, 42);

    // Next 32 bytes: prev_hash (all zeros for genesis)
    assert_eq!(&bytes[12..44], &[0u8; 32]);

    // Next 1 byte: kind (0 = Data)
    assert_eq!(bytes[44], 0);

    // Next 1 byte: compression (0 = None)
    assert_eq!(bytes[45], 0);

    // Next 4 bytes: length (5 in little-endian)
    let length = u32::from_le_bytes(bytes[46..50].try_into().unwrap());
    assert_eq!(length, 5);

    // Next 5 bytes: payload
    assert_eq!(&bytes[50..55], b"hello");

    // Next 4 bytes: CRC (verify it matches expected)
    let stored_crc = u32::from_le_bytes(bytes[55..59].try_into().unwrap());
    let computed_crc = kimberlite_crypto::crc32(&bytes[0..55]);
    assert_eq!(stored_crc, computed_crc);

    // Last 4 bytes: RECORD_END sentinel (0xC0FFEE42)
    let end_sentinel = u32::from_le_bytes(bytes[59..63].try_into().unwrap());
    assert_eq!(end_sentinel, 0xC0FF_EE42);
}

#[test]
fn record_roundtrip_preserves_data() {
    let original = Record::new(Offset::new(123), None, Bytes::from("test payload"));
    let bytes: Bytes = original.to_bytes().into();

    let (parsed, consumed) = Record::from_bytes(&bytes).unwrap();

    assert_eq!(parsed.offset(), Offset::new(123));
    assert_eq!(parsed.prev_hash(), None);
    assert_eq!(parsed.payload().as_ref(), b"test payload");
    assert_eq!(consumed, bytes.len());
}

#[test]
fn record_roundtrip_with_prev_hash() {
    // Create a chain hash from known bytes
    let prev_hash = ChainHash::from_bytes(&[42u8; 32]);
    let original = Record::new(Offset::new(1), Some(prev_hash), Bytes::from("linked"));
    let bytes: Bytes = original.to_bytes().into();

    let (parsed, _) = Record::from_bytes(&bytes).unwrap();

    assert_eq!(parsed.offset(), Offset::new(1));
    assert_eq!(parsed.prev_hash(), Some(prev_hash));
    assert_eq!(parsed.payload().as_ref(), b"linked");
}

#[test]
fn record_from_bytes_detects_corruption() {
    let record = Record::new(Offset::new(0), None, Bytes::from("data"));
    let mut bytes: Vec<u8> = record.to_bytes();

    // Corrupt one byte in the payload (AUDIT-2026-03 M-8: payload starts at offset 50 with sentinels)
    bytes[50] ^= 0xFF;

    let result = Record::from_bytes(&Bytes::from(bytes));
    assert!(matches!(result, Err(StorageError::CorruptedRecord)));
}

#[test]
fn record_from_bytes_handles_truncated_header() {
    // Less than 50 bytes (AUDIT-2026-03 M-8: minimum header size with sentinels: 4 + 8 + 32 + 1 + 1 + 4)
    let short_data = Bytes::from(vec![0u8; 40]);
    let result = Record::from_bytes(&short_data);
    assert!(matches!(result, Err(StorageError::UnexpectedEof)));
}

#[test]
fn record_from_bytes_handles_truncated_payload() {
    // Create a header claiming 100 bytes of payload (AUDIT-2026-03 M-8: with sentinels)
    let mut data = Vec::new();
    data.extend_from_slice(&0xBADC_0FFE_u32.to_le_bytes()); // RECORD_START sentinel
    data.extend_from_slice(&0u64.to_le_bytes()); // offset
    data.extend_from_slice(&[0u8; 32]); // prev_hash
    data.push(0); // kind: Data
    data.push(0); // compression: None
    data.extend_from_slice(&100u32.to_le_bytes()); // length: 100 bytes
    data.extend_from_slice(&[0u8; 50]); // only 50 bytes of payload (should be 100)

    let result = Record::from_bytes(&Bytes::from(data));
    assert!(matches!(result, Err(StorageError::UnexpectedEof)));
}

#[test]
fn record_empty_payload() {
    let record = Record::new(Offset::new(0), None, Bytes::new());
    let bytes: Bytes = record.to_bytes().into();

    let (parsed, _) = Record::from_bytes(&bytes).unwrap();
    assert!(parsed.payload().is_empty());
}

#[test]
fn record_compute_hash_creates_chain() {
    // Genesis record
    let record0 = Record::new(Offset::new(0), None, Bytes::from("hello"));
    let hash0 = record0.compute_hash();

    // Second record links to first
    let record1 = Record::new(Offset::new(1), Some(hash0), Bytes::from("world"));
    let hash1 = record1.compute_hash();

    // Hashes should be different
    assert_ne!(hash0, hash1);

    // Same input should produce same hash
    let record1_copy = Record::new(Offset::new(1), Some(hash0), Bytes::from("world"));
    assert_eq!(record1_copy.compute_hash(), hash1);
}

// ============================================================================
// Index Integration Tests
// ============================================================================

#[test]
fn test_append_and_lookup() {
    let mut index = OffsetIndex::new();
    index.append(0);
    index.append(100);
    index.append(250);

    assert_eq!(index.lookup(Offset::new(0)), Some(0));
    assert_eq!(index.lookup(Offset::new(1)), Some(100));
    assert_eq!(index.lookup(Offset::new(2)), Some(250));
    assert_eq!(index.lookup(Offset::new(3)), None); // out of bounds
    assert_eq!(index.len(), 3);
    assert!(!index.is_empty());
}

#[test]
fn test_save_and_load_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("test.idx");

    let mut original = OffsetIndex::new();

    original.append(0);
    original.append(100);
    original.append(250);

    original.save(&index_path).expect("save succeeds");

    let loaded = OffsetIndex::load(&index_path).expect("load succeeds");

    // Verify loaded matches original
    assert_eq!(loaded.len(), 3);
    assert_eq!(loaded.lookup(Offset::new(0)), Some(0));
    assert_eq!(loaded.lookup(Offset::new(1)), Some(100));
    assert_eq!(loaded.lookup(Offset::new(2)), Some(250));
}

#[test]
fn test_empty_index_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("empty.idx");

    let original = OffsetIndex::new();
    assert!(original.is_empty());

    original.save(&index_path).expect("save succeeds");

    let loaded = OffsetIndex::load(&index_path).expect("load succeeds");

    assert!(loaded.is_empty());
    assert_eq!(loaded.len(), 0);
}

#[test]
fn test_large_index_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("large.idx");

    let mut original = OffsetIndex::new();

    // Add 10,000 positions with increasing byte offsets
    let mut byte_pos = 0u64;
    for _ in 0..10_000 {
        original.append(byte_pos);
        byte_pos += 53; // Simulate ~53 byte records
    }

    original.save(&index_path).expect("save succeeds");

    let loaded = OffsetIndex::load(&index_path).expect("load succeeds");

    assert_eq!(loaded.len(), 10_000);
    assert_eq!(loaded.lookup(Offset::new(0)), Some(0));
    assert_eq!(loaded.lookup(Offset::new(9999)), Some(9999 * 53));
}

#[test]
fn test_load_detects_invalid_magic() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("bad_magic.idx");

    // Write file with wrong magic bytes
    let mut data = vec![0u8; 20]; // minimum valid size
    data[0..4].copy_from_slice(b"XXXX"); // wrong magic
    std::fs::write(&index_path, &data).unwrap();

    let result = OffsetIndex::load(&index_path);
    assert!(matches!(result, Err(StorageError::InvalidIndexMagic)));
}

#[test]
fn test_load_detects_unsupported_version() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("bad_version.idx");

    // Write file with correct magic but wrong version
    let mut data = vec![0u8; 24];
    data[0..4].copy_from_slice(b"VDXI");
    data[4] = 0xFF; // unsupported version
    // Add a valid CRC for the data
    let crc = kimberlite_crypto::crc32(&data[0..20]);
    data[20..24].copy_from_slice(&crc.to_le_bytes());
    std::fs::write(&index_path, &data).unwrap();

    let result = OffsetIndex::load(&index_path);
    assert!(matches!(
        result,
        Err(StorageError::UnsupportedIndexVersion(0xFF))
    ));
}

#[test]
fn test_load_detects_truncated_file() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("truncated.idx");

    // Write file that's too short (less than header + CRC)
    let data = vec![0u8; 10];
    std::fs::write(&index_path, &data).unwrap();

    let result = OffsetIndex::load(&index_path);
    assert!(matches!(
        result,
        Err(StorageError::IndexTruncated {
            expected: 20,
            actual: 10
        })
    ));
}

#[test]
fn test_load_detects_truncated_positions() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("truncated_pos.idx");

    // Create valid header claiming 10 positions, but only provide space for 5
    let mut data = vec![0u8; 16 + (5 * 8) + 4]; // header + 5 positions + CRC
    data[0..4].copy_from_slice(b"VDXI");
    data[4] = 0x01; // version
    data[8..16].copy_from_slice(&10u64.to_le_bytes()); // claims 10 positions
    // CRC of the truncated data (will be wrong but we check truncation first)
    let crc = kimberlite_crypto::crc32(&data[0..data.len() - 4]);
    let crc_start = data.len() - 4;
    data[crc_start..].copy_from_slice(&crc.to_le_bytes());
    std::fs::write(&index_path, &data).unwrap();

    let result = OffsetIndex::load(&index_path);
    // Expected: header(16) + 10 positions(80) + CRC(4) = 100 bytes
    // Actual: header(16) + 5 positions(40) + CRC(4) = 60 bytes
    assert!(matches!(
        result,
        Err(StorageError::IndexTruncated {
            expected: 100,
            actual: 60
        })
    ));
}

#[test]
fn test_load_detects_corrupted_crc() {
    let temp_dir = TempDir::new().unwrap();
    let index_path = temp_dir.path().join("corrupted.idx");

    // Create a valid index, then corrupt it
    let mut index = OffsetIndex::new();
    index.append(0);
    index.append(100);
    index.save(&index_path).expect("save succeeds");

    // Corrupt a byte in the positions area
    let mut data = std::fs::read(&index_path).unwrap();
    data[16] ^= 0xFF; // flip bits in first position
    std::fs::write(&index_path, &data).unwrap();

    let result = OffsetIndex::load(&index_path);
    assert!(matches!(
        result,
        Err(StorageError::IndexChecksumMismatch { .. })
    ));
}

#[test]
fn test_from_positions_creates_valid_index() {
    let positions = vec![0, 100, 250, 400];
    let index = OffsetIndex::from_positions(positions.clone());

    assert_eq!(index.len(), 4);
    assert_eq!(index.positions(), positions.as_slice());
    assert_eq!(index.lookup(Offset::new(2)), Some(250));
}

#[test]
fn test_index_equality() {
    let mut index1 = OffsetIndex::new();
    index1.append(0);
    index1.append(100);

    let index2 = OffsetIndex::from_positions(vec![0, 100]);

    assert_eq!(index1, index2);
}

#[test]
fn test_index_clone() {
    let mut original = OffsetIndex::new();
    original.append(0);
    original.append(100);

    let cloned = original.clone();

    assert_eq!(original, cloned);
    assert_eq!(cloned.lookup(Offset::new(1)), Some(100));
}

// ============================================================================
// Storage Integration Tests
// ============================================================================

mod integration {
    use super::*;
    use tempfile::TempDir;

    fn setup_storage() -> (Storage, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path());
        (storage, temp_dir)
    }

    fn test_events(count: usize) -> Vec<Bytes> {
        (0..count)
            .map(|i| Bytes::from(format!("event-{i}")))
            .collect()
    }

    #[test]
    fn append_and_read_single_event() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        let (new_offset, _hash) = storage
            .append_batch(stream_id, test_events(1), Offset::new(0), None, false)
            .unwrap();

        assert_eq!(new_offset, Offset::new(1));

        let events = storage
            .read_from(stream_id, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].as_ref(), b"event-0");
    }

    #[test]
    fn append_and_read_multiple_events() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        storage
            .append_batch(stream_id, test_events(5), Offset::new(0), None, false)
            .unwrap();

        let events = storage
            .read_from(stream_id, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(events.len(), 5);
        (0..5).for_each(|i| {
            assert_eq!(events[i].as_ref(), format!("event-{i}").as_bytes());
        });
    }

    #[test]
    fn read_from_middle_offset() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Append 10 events
        storage
            .append_batch(stream_id, test_events(10), Offset::new(0), None, false)
            .unwrap();

        // Read from offset 5
        let events = storage
            .read_from(stream_id, Offset::new(5), u64::MAX)
            .unwrap();

        // Should get events 5-9
        assert_eq!(events.len(), 5);
        assert_eq!(events[0].as_ref(), b"event-5");
        assert_eq!(events[4].as_ref(), b"event-9");
    }

    #[test]
    fn read_respects_max_bytes() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Create events with known sizes
        let events: Vec<Bytes> = (0..10)
            .map(|i| Bytes::from(format!("event-{i:04}"))) // Each ~10 bytes
            .collect();

        storage
            .append_batch(stream_id, events, Offset::new(0), None, false)
            .unwrap();

        // Read with max_bytes that should limit results
        // Each event is ~10 bytes, so max_bytes=25 should give us 2-3 events
        let events = storage.read_from(stream_id, Offset::new(0), 25).unwrap();

        // Should get fewer than all 10 events
        assert!(events.len() < 10);
        assert!(!events.is_empty());
    }

    #[test]
    fn append_multiple_batches_sequential() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Append batch 1 (3 events)
        let (offset_after_batch1, hash_after_batch1) = storage
            .append_batch(stream_id, test_events(3), Offset::new(0), None, false)
            .unwrap();
        assert_eq!(offset_after_batch1, Offset::new(3));

        // Append batch 2 (2 events) starting at offset 3, continuing the chain
        let events2: Vec<Bytes> = vec![Bytes::from("batch2-0"), Bytes::from("batch2-1")];
        let (offset_after_batch2, _) = storage
            .append_batch(
                stream_id,
                events2,
                Offset::new(3),
                Some(hash_after_batch1),
                false,
            )
            .unwrap();
        assert_eq!(offset_after_batch2, Offset::new(5));

        // Read all events
        let events = storage
            .read_from(stream_id, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(events.len(), 5);
        // First 3 from batch 1
        assert_eq!(events[0].as_ref(), b"event-0");
        assert_eq!(events[2].as_ref(), b"event-2");
        // Last 2 from batch 2
        assert_eq!(events[3].as_ref(), b"batch2-0");
        assert_eq!(events[4].as_ref(), b"batch2-1");
    }

    #[test]
    fn append_with_fsync() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Append with fsync=true
        let result = storage.append_batch(stream_id, test_events(1), Offset::new(0), None, true);

        // Should succeed (fsync is just durability, shouldn't change behavior)
        assert!(result.is_ok());
    }

    #[test]
    fn multiple_streams_are_isolated() {
        let (mut storage, _dir) = setup_storage();
        let stream1 = StreamId::new(1);
        let stream2 = StreamId::new(2);

        // Append to stream 1
        storage
            .append_batch(
                stream1,
                vec![Bytes::from("stream1-event")],
                Offset::new(0),
                None,
                false,
            )
            .unwrap();

        // Append to stream 2
        storage
            .append_batch(
                stream2,
                vec![Bytes::from("stream2-event")],
                Offset::new(0),
                None,
                false,
            )
            .unwrap();

        // Read from each stream
        let events1 = storage
            .read_from(stream1, Offset::new(0), u64::MAX)
            .unwrap();
        let events2 = storage
            .read_from(stream2, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(events1.len(), 1);
        assert_eq!(events1[0].as_ref(), b"stream1-event");

        assert_eq!(events2.len(), 1);
        assert_eq!(events2[0].as_ref(), b"stream2-event");
    }

    #[test]
    fn hash_chain_is_built_correctly() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Append 3 events
        let (_, final_hash) = storage
            .append_batch(stream_id, test_events(3), Offset::new(0), None, false)
            .unwrap();

        // Manually compute what the hash chain should be
        let record0 = Record::new(Offset::new(0), None, Bytes::from("event-0"));
        let hash0 = record0.compute_hash();

        let record1 = Record::new(Offset::new(1), Some(hash0), Bytes::from("event-1"));
        let hash1 = record1.compute_hash();

        let record2 = Record::new(Offset::new(2), Some(hash1), Bytes::from("event-2"));
        let hash2 = record2.compute_hash();

        // The final hash from append_batch should match our manual computation
        assert_eq!(final_hash, hash2);
    }

    #[test]
    fn tampered_record_is_detected() {
        let (mut storage, dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Write some valid records
        storage
            .append_batch(stream_id, test_events(3), Offset::new(0), None, false)
            .unwrap();

        // Tamper with the file: corrupt the prev_hash of record 1
        let segment_path = dir
            .path()
            .join(stream_id.to_string())
            .join("segment_000000.log");

        let mut data = std::fs::read(&segment_path).unwrap();

        // Record 0 is at offset 0, its size is 8 + 32 + 1 + 4 + 7 + 4 = 56 bytes
        // Record 1 starts at byte 56, its prev_hash is at bytes 56+8 = 64
        // Flip a bit in the prev_hash of record 1
        data[64] ^= 0xFF;

        // Also need to fix the CRC or we'll get CorruptedRecord instead
        // Actually, let's just verify that ANY corruption is caught
        std::fs::write(&segment_path, &data).unwrap();

        // Reading should fail with either ChainVerificationFailed or CorruptedRecord
        let result = storage.read_from(stream_id, Offset::new(0), u64::MAX);
        assert!(result.is_err());
    }

    #[test]
    fn read_records_returns_full_records() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        storage
            .append_batch(stream_id, test_events(3), Offset::new(0), None, false)
            .unwrap();

        let records = storage
            .read_records_from_genesis(stream_id, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(records.len(), 3);

        // First record should have no prev_hash
        assert_eq!(records[0].prev_hash(), None);
        assert_eq!(records[0].offset(), Offset::new(0));

        // Second record should link to first
        assert_eq!(records[1].prev_hash(), Some(records[0].compute_hash()));
        assert_eq!(records[1].offset(), Offset::new(1));

        // Third record should link to second
        assert_eq!(records[2].prev_hash(), Some(records[1].compute_hash()));
        assert_eq!(records[2].offset(), Offset::new(2));
    }
}

// ============================================================================
// Segment Rotation Tests
// ============================================================================

mod segment_rotation_tests {
    use super::*;
    use kimberlite_types::CheckpointPolicy;
    use tempfile::TempDir;

    fn setup_small_segment_storage() -> (Storage, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        // Set max segment size to 500 bytes to force rotation with a few records
        let storage =
            Storage::with_max_segment_size(temp_dir.path(), CheckpointPolicy::default(), 500);
        (storage, temp_dir)
    }

    fn test_events(count: usize) -> Vec<Bytes> {
        (0..count)
            .map(|i| Bytes::from(format!("event-{i}")))
            .collect()
    }

    #[test]
    fn segment_rotates_when_size_exceeded() {
        let (mut storage, _dir) = setup_small_segment_storage();
        let stream_id = StreamId::new(1);

        // Each record is ~56 bytes (8+32+1+4+7+4), so ~10 records = ~560 bytes
        // Should rotate after first batch since 560 > 500
        let (offset1, hash1) = storage
            .append_batch(stream_id, test_events(10), Offset::new(0), None, false)
            .unwrap();

        assert_eq!(offset1, Offset::new(10));

        // After rotation, we should have 2 segments
        assert!(storage.segment_count(stream_id) >= 2);

        // Append more events (these should go to the new segment)
        let (_offset2, _hash2) = storage
            .append_batch(stream_id, test_events(5), offset1, Some(hash1), false)
            .unwrap();

        // Read all events across segments
        let events = storage
            .read_from(stream_id, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(events.len(), 15);
        assert_eq!(events[0].as_ref(), b"event-0");
        assert_eq!(events[9].as_ref(), b"event-9");
        assert_eq!(events[10].as_ref(), b"event-0"); // Second batch starts over
        assert_eq!(events[14].as_ref(), b"event-4");
    }

    #[test]
    fn read_from_middle_across_segments() {
        let (mut storage, _dir) = setup_small_segment_storage();
        let stream_id = StreamId::new(1);

        // Append events that will trigger rotation
        let (offset1, hash1) = storage
            .append_batch(stream_id, test_events(10), Offset::new(0), None, false)
            .unwrap();

        let (_offset2, _hash2) = storage
            .append_batch(stream_id, test_events(5), offset1, Some(hash1), false)
            .unwrap();

        // Read from an offset that's in the second batch (after rotation)
        let events = storage
            .read_from(stream_id, Offset::new(12), u64::MAX)
            .unwrap();

        assert_eq!(events.len(), 3);
        assert_eq!(events[0].as_ref(), b"event-2");
    }

    #[test]
    fn hash_chain_integrity_across_segments() {
        let (mut storage, _dir) = setup_small_segment_storage();
        let stream_id = StreamId::new(1);

        // Append events forcing multiple rotations
        let (offset1, hash1) = storage
            .append_batch(stream_id, test_events(10), Offset::new(0), None, false)
            .unwrap();

        let (offset2, hash2) = storage
            .append_batch(stream_id, test_events(10), offset1, Some(hash1), false)
            .unwrap();

        let (_offset3, _hash3) = storage
            .append_batch(stream_id, test_events(10), offset2, Some(hash2), false)
            .unwrap();

        // Full genesis verification should pass across all segments
        let records = storage
            .read_records_from_genesis(stream_id, Offset::new(0), u64::MAX)
            .unwrap();

        assert_eq!(records.len(), 30);
    }

    #[test]
    fn completed_segments_are_listed() {
        let (mut storage, _dir) = setup_small_segment_storage();
        let stream_id = StreamId::new(1);

        // Force rotation
        let (offset1, hash1) = storage
            .append_batch(stream_id, test_events(10), Offset::new(0), None, false)
            .unwrap();
        let (_offset2, _hash2) = storage
            .append_batch(stream_id, test_events(5), offset1, Some(hash1), false)
            .unwrap();

        let completed = storage.completed_segments(stream_id);
        // Segment 0 should be completed, active segment should not be in the list
        assert!(!completed.is_empty());
        assert!(completed.contains(&0));
    }

    #[test]
    fn no_rotation_when_below_threshold() {
        let temp_dir = TempDir::new().unwrap();
        // Large segment size — no rotation
        let mut storage = Storage::with_max_segment_size(
            temp_dir.path(),
            CheckpointPolicy::default(),
            1024 * 1024 * 1024, // 1GB
        );
        let stream_id = StreamId::new(1);

        storage
            .append_batch(stream_id, test_events(100), Offset::new(0), None, false)
            .unwrap();

        // Should still be on segment 0
        assert_eq!(storage.segment_count(stream_id), 1);
    }
}

// ============================================================================
// Checkpoint Tests
// ============================================================================

mod checkpoint_tests {
    use super::*;
    use kimberlite_types::CheckpointPolicy;
    use tempfile::TempDir;

    fn setup_storage() -> (Storage, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path());
        (storage, temp_dir)
    }

    fn test_events(count: usize) -> Vec<Bytes> {
        (0..count)
            .map(|i| Bytes::from(format!("event-{i}")))
            .collect()
    }

    #[test]
    fn create_and_read_checkpoint() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Append some data records first
        let (next_offset, last_hash) = storage
            .append_batch(stream_id, test_events(5), Offset::new(0), None, false)
            .unwrap();

        assert_eq!(next_offset, Offset::new(5));

        // Create a checkpoint
        let (cp_next_offset, _cp_hash) = storage
            .create_checkpoint(stream_id, next_offset, Some(last_hash), 5, false)
            .unwrap();

        assert_eq!(cp_next_offset, Offset::new(6));

        // Verify checkpoint was recorded
        let last_cp = storage.last_checkpoint(stream_id).unwrap();
        assert_eq!(last_cp, Some(Offset::new(5)));
    }

    #[test]
    fn read_with_checkpoint_verification() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        // Append 10 events
        let (offset1, hash1) = storage
            .append_batch(stream_id, test_events(5), Offset::new(0), None, false)
            .unwrap();

        // Create checkpoint at offset 5
        let (offset2, hash2) = storage
            .create_checkpoint(stream_id, offset1, Some(hash1), 5, false)
            .unwrap();

        // Append 5 more events
        storage
            .append_batch(stream_id, test_events(5), offset2, Some(hash2), false)
            .unwrap();

        // Read using checkpoint-optimized verification
        let records = storage
            .read_records_verified(stream_id, Offset::new(7), u64::MAX)
            .unwrap();

        // Should get events 7, 8, 9, 10 (checkpoint at 5 is skipped, events 6-10 after checkpoint)
        assert_eq!(records.len(), 4);
        assert_eq!(records[0].offset(), Offset::new(7));
    }

    #[test]
    fn checkpoint_policy_triggers() {
        let temp_dir = TempDir::new().unwrap();
        let policy = CheckpointPolicy::every(3);
        let storage = Storage::with_checkpoint_policy(temp_dir.path(), policy);

        // Verify policy is set
        assert_eq!(storage.checkpoint_policy().every_n_records, 3);
        assert!(
            storage
                .checkpoint_policy()
                .should_checkpoint(Offset::new(2))
        ); // 3rd record
        assert!(
            !storage
                .checkpoint_policy()
                .should_checkpoint(Offset::new(3))
        ); // 4th record
    }

    #[test]
    fn empty_stream_has_no_checkpoints() {
        let (mut storage, _dir) = setup_storage();
        let stream_id = StreamId::new(1);

        let last_cp = storage.last_checkpoint(stream_id).unwrap();
        assert_eq!(last_cp, None);
    }
}

// ============================================================================
// Property-Based Tests
// ============================================================================

mod proptests {
    use super::*;
    use proptest::prelude::*;

    proptest! {
        #[test]
        fn record_roundtrip_any_payload(payload in prop::collection::vec(any::<u8>(), 0..1000)) {
            let record = Record::new(Offset::new(42), None, Bytes::from(payload.clone()));
            let bytes: Bytes = record.to_bytes().into();
            let (parsed, consumed) = Record::from_bytes(&bytes).unwrap();

            prop_assert_eq!(parsed.offset(), Offset::new(42));
            prop_assert_eq!(parsed.prev_hash(), None);
            prop_assert_eq!(parsed.payload().as_ref(), payload.as_slice());
            prop_assert_eq!(consumed, bytes.len());
        }

        #[test]
        fn record_roundtrip_any_offset(offset in 0u64..u64::MAX) {
            let record = Record::new(Offset::new(offset), None, Bytes::from("test"));
            let bytes: Bytes = record.to_bytes().into();
            let (parsed, _) = Record::from_bytes(&bytes).unwrap();

            prop_assert_eq!(parsed.offset().as_u64(), offset);
        }

        #[test]
        fn record_roundtrip_with_any_prev_hash(hash_bytes in prop::collection::vec(any::<u8>(), 32..=32)) {
            let hash_arr: [u8; 32] = hash_bytes.try_into().unwrap();
            let prev_hash = ChainHash::from_bytes(&hash_arr);
            let record = Record::new(Offset::new(1), Some(prev_hash), Bytes::from("data"));
            let bytes: Bytes = record.to_bytes().into();
            let (parsed, _) = Record::from_bytes(&bytes).unwrap();

            prop_assert_eq!(parsed.prev_hash(), Some(prev_hash));
        }

        #[test]
        fn corruption_is_detected(
            payload in prop::collection::vec(any::<u8>(), 1..100),
            flip_pos in 0usize..1000
        ) {
            let record = Record::new(Offset::new(0), None, Bytes::from(payload));
            let mut bytes = record.to_bytes();

            // Flip a bit at a valid position (excluding CRC itself)
            let max_pos = bytes.len().saturating_sub(4); // Exclude CRC bytes
            if max_pos > 0 {
                let actual_pos = flip_pos % max_pos;
                bytes[actual_pos] ^= 1;

                let result = Record::from_bytes(&Bytes::from(bytes));
                // Any error is acceptable - could be CorruptedRecord (CRC mismatch)
                // or UnexpectedEof (if length field was corrupted to claim more bytes)
                prop_assert!(result.is_err());
            }
        }

        /// Verifies that hash chain tampering is detected.
        /// Modifying any record in the chain should invalidate all subsequent hashes.
        #[test]
        fn hash_chain_detects_tampering(
            payloads in prop::collection::vec(prop::collection::vec(any::<u8>(), 10..100), 3..10),
            tamper_index in 0usize..9
        ) {
            // Build a hash chain
            let mut records = Vec::new();
            let mut prev_hash = None;

            for (i, payload) in payloads.iter().enumerate() {
                let record = Record::new(Offset::new(i as u64), prev_hash, Bytes::from(payload.clone()));
                prev_hash = Some(record.compute_hash());
                records.push(record);
            }

            // Tamper with a record (but not the last one, so we can verify propagation)
            let actual_tamper_index = tamper_index % records.len().saturating_sub(1);
            if actual_tamper_index < records.len() {
                // Create a modified version of the record
                let mut tampered_payload = records[actual_tamper_index].payload().to_vec();
                if !tampered_payload.is_empty() {
                    tampered_payload[0] ^= 1; // Flip a bit
                }

                let tampered_record = Record::new(
                    records[actual_tamper_index].offset(),
                    records[actual_tamper_index].prev_hash(),
                    Bytes::from(tampered_payload),
                );

                // Verify that the hash changed
                prop_assert_ne!(
                    tampered_record.compute_hash(),
                    records[actual_tamper_index].compute_hash(),
                    "Tampering should change record hash"
                );

                // Verify that subsequent records in the chain would be invalid
                // (their prev_hash no longer matches)
                if actual_tamper_index + 1 < records.len() {
                    let next_record = &records[actual_tamper_index + 1];
                    prop_assert_ne!(
                        next_record.prev_hash(),
                        Some(tampered_record.compute_hash()),
                        "Next record's prev_hash should not match tampered hash"
                    );
                }
            }
        }

        /// Verifies that partial writes are detected.
        /// Truncating a serialized record should fail to parse.
        #[test]
        fn partial_write_detected(
            payload in prop::collection::vec(any::<u8>(), 64..1024),
            truncate_bytes in 1usize..1023
        ) {
            let record = Record::new(Offset::new(0), None, Bytes::from(payload));
            let bytes = record.to_bytes();

            // Truncate the record
            let truncate_at = truncate_bytes % bytes.len();
            if truncate_at > 0 && truncate_at < bytes.len() {
                let truncated = Bytes::from(bytes[..truncate_at].to_vec());

                let result = Record::from_bytes(&truncated);
                // Should fail with UnexpectedEof or CorruptedRecord
                prop_assert!(result.is_err(), "Truncated record should fail to parse");
            }
        }

        /// Verifies that multiple bit flips are also detected.
        /// This ensures the CRC isn't accidentally weak for certain patterns.
        #[test]
        fn multiple_bit_flips_detected(
            payload in prop::collection::vec(any::<u8>(), 32..256),
            flip_positions in prop::collection::vec(0usize..300, 2..5)
        ) {
            let record = Record::new(Offset::new(0), None, Bytes::from(payload));
            let mut bytes = record.to_bytes();

            let max_pos = bytes.len().saturating_sub(4); // Exclude CRC bytes
            if max_pos > 0 {
                // Deduplicate flip positions to avoid canceling out
                let unique_positions: Vec<usize> = flip_positions.iter()
                    .map(|p| p % max_pos)
                    .collect::<std::collections::HashSet<_>>()
                    .into_iter()
                    .collect();

                // Only test if we have at least 2 unique positions
                if unique_positions.len() >= 2 {
                    // Flip multiple bits
                    for actual_pos in &unique_positions {
                        bytes[*actual_pos] ^= 1;
                    }

                    let result = Record::from_bytes(&Bytes::from(bytes));
                    prop_assert!(result.is_err(), "Multiple bit flips should be detected");
                }
            }
        }

        /// Verifies that byte substitution (not just bit flips) is detected.
        #[test]
        fn byte_substitution_detected(
            payload in prop::collection::vec(any::<u8>(), 32..256),
            substitute_pos in 0usize..300,
            new_value in any::<u8>()
        ) {
            let record = Record::new(Offset::new(0), None, Bytes::from(payload));
            let mut bytes = record.to_bytes();

            let max_pos = bytes.len().saturating_sub(4); // Exclude CRC bytes
            if max_pos > 0 {
                let actual_pos = substitute_pos % max_pos;
                let old_value = bytes[actual_pos];

                // Only test if we're actually changing the byte
                if new_value != old_value {
                    bytes[actual_pos] = new_value;

                    let result = Record::from_bytes(&Bytes::from(bytes));
                    prop_assert!(result.is_err(), "Byte substitution should be detected");
                }
            }
        }
    }
}

// ============================================================================
// Torn Write Detection Tests (AUDIT-2026-03 M-8)
// ============================================================================

#[test]
fn torn_write_missing_record_start_sentinel() {
    let record = Record::new(Offset::new(42), None, Bytes::from("test"));
    let mut bytes = record.to_bytes();

    // Corrupt the RECORD_START sentinel (first 4 bytes)
    bytes[0] = 0xFF;
    bytes[1] = 0xFF;
    bytes[2] = 0xFF;
    bytes[3] = 0xFF;

    let result = Record::from_bytes(&Bytes::from(bytes));
    assert!(matches!(result, Err(StorageError::TornWrite { .. })));

    if let Err(StorageError::TornWrite { reason }) = result {
        assert!(reason.contains("RECORD_START"));
    }
}

#[test]
fn torn_write_missing_record_end_sentinel() {
    let record = Record::new(Offset::new(42), None, Bytes::from("test"));
    let mut bytes = record.to_bytes();

    // Corrupt the RECORD_END sentinel (last 4 bytes)
    let len = bytes.len();
    bytes[len - 4] = 0xFF;
    bytes[len - 3] = 0xFF;
    bytes[len - 2] = 0xFF;
    bytes[len - 1] = 0xFF;

    let result = Record::from_bytes(&Bytes::from(bytes));
    assert!(matches!(result, Err(StorageError::TornWrite { .. })));

    if let Err(StorageError::TornWrite { reason }) = result {
        assert!(reason.contains("RECORD_END"));
    }
}

#[test]
fn torn_write_truncated_record_missing_end_sentinel() {
    let record = Record::new(Offset::new(42), None, Bytes::from("test"));
    let mut bytes = record.to_bytes();

    // Truncate the record before the RECORD_END sentinel
    bytes.truncate(bytes.len() - 4);

    let result = Record::from_bytes(&Bytes::from(bytes));
    // Should return UnexpectedEof since we don't have enough bytes for the end sentinel
    assert!(matches!(result, Err(StorageError::UnexpectedEof)));
}

#[test]
fn torn_write_truncated_record_mid_payload() {
    let record = Record::new(Offset::new(42), None, Bytes::from("hello world"));
    let mut bytes = record.to_bytes();

    // Truncate in the middle of the payload (before CRC and end sentinel)
    bytes.truncate(55);

    let result = Record::from_bytes(&Bytes::from(bytes));
    // Should return UnexpectedEof since we don't have enough bytes
    assert!(matches!(result, Err(StorageError::UnexpectedEof)));
}

#[test]
fn torn_write_detection_with_valid_record() {
    let record = Record::new(Offset::new(42), None, Bytes::from("test"));
    let bytes = record.to_bytes();

    // Valid record should pass all checks
    let result = Record::from_bytes(&Bytes::from(bytes));
    assert!(result.is_ok());

    let (parsed, _) = result.unwrap();
    assert_eq!(parsed.offset(), Offset::new(42));
    assert_eq!(parsed.payload(), &Bytes::from("test"));
}

#[test]
fn torn_write_detection_preserves_hash_chain() {
    let prev_hash = Some(kimberlite_crypto::chain_hash(None, b"previous record"));
    let record = Record::new(Offset::new(100), prev_hash, Bytes::from("data"));
    let bytes = record.to_bytes();

    let (parsed, _) = Record::from_bytes(&Bytes::from(bytes)).unwrap();
    assert_eq!(parsed.prev_hash(), prev_hash);
    assert_eq!(parsed.offset(), Offset::new(100));
}

#[test]
fn torn_write_large_payload() {
    // Test with a large payload to ensure sentinels work with various sizes
    let large_payload = vec![0xAB; 10000];
    let record = Record::new(Offset::new(999), None, Bytes::from(large_payload.clone()));
    let bytes = record.to_bytes();

    // Corrupt RECORD_END sentinel
    let mut corrupted = bytes.clone();
    let len = corrupted.len();
    corrupted[len - 1] = 0xFF;

    let result = Record::from_bytes(&Bytes::from(corrupted));
    assert!(matches!(result, Err(StorageError::TornWrite { .. })));

    // Valid record should work
    let (parsed, _) = Record::from_bytes(&Bytes::from(bytes)).unwrap();
    assert_eq!(parsed.payload(), &Bytes::from(large_payload));
}

#[test]
fn torn_write_empty_payload() {
    // Test with empty payload
    let record = Record::new(Offset::new(1), None, Bytes::from(""));
    let bytes = record.to_bytes();

    // Corrupt RECORD_START
    let mut corrupted = bytes.clone();
    corrupted[0] = 0x00;

    let result = Record::from_bytes(&Bytes::from(corrupted));
    assert!(matches!(result, Err(StorageError::TornWrite { .. })));

    // Valid record should work
    let result = Record::from_bytes(&Bytes::from(bytes));
    assert!(result.is_ok());
}

#[test]
fn latest_chain_hash_empty_stream_returns_none() {
    let temp_dir = TempDir::new().expect("temp dir");
    let mut storage = Storage::new(temp_dir.path());
    let stream_id = StreamId::new(1);
    let result = storage
        .latest_chain_hash(stream_id)
        .expect("latest_chain_hash should not error on unknown stream");
    assert!(result.is_none(), "empty stream must return None");
}

#[test]
fn latest_chain_hash_recovers_after_restart() {
    // Simulates the chain_heads recovery case: write records via one
    // Storage instance, drop it, open a fresh instance on the same dir,
    // and verify that the fresh instance can still recover the tail
    // hash. Before this recovery path, the next append would write
    // prev_hash=None into a non-empty stream and corrupt the chain.
    let temp_dir = TempDir::new().expect("temp dir");
    let stream_id = StreamId::new(1);

    let tail_hash_before_restart = {
        let mut storage = Storage::new(temp_dir.path());
        let events = vec![
            Bytes::from_static(b"event1"),
            Bytes::from_static(b"event2"),
            Bytes::from_static(b"event3"),
        ];
        let (_new_offset, new_hash) = storage
            .append_batch(stream_id, events, Offset::ZERO, None, true)
            .expect("initial append must succeed");
        new_hash
    };

    // Drop the storage, open a fresh instance on the same directory.
    let mut storage = Storage::new(temp_dir.path());
    let recovered = storage
        .latest_chain_hash(stream_id)
        .expect("latest_chain_hash must succeed after restart");
    assert_eq!(
        recovered,
        Some(tail_hash_before_restart),
        "recovered hash must match the last hash from before restart"
    );

    // Append one more record using the recovered hash — this is the
    // operation that would have produced a chain break without recovery.
    let (next_offset, _) = storage
        .append_batch(
            stream_id,
            vec![Bytes::from_static(b"post_restart")],
            Offset::new(3),
            recovered,
            true,
        )
        .expect("append after restart must succeed");
    assert_eq!(next_offset, Offset::new(4));

    // Reading the whole stream must verify cleanly end-to-end.
    // `read_from` re-walks the chain with verify=true, so if
    // `recovered` had been wrong on the N+1-th append, this read
    // would return a `ChainVerificationFailed` error.
    //
    // AUDIT-2026-04 M-9: this read-side verification is what closes
    // the restart gap — `append_batch` trusts the caller's
    // `prev_hash`, so a broken chain is visible only at read.
    // Together with `latest_chain_hash` populating correctly on
    // reopen, restart-time chain continuity is observable and
    // tested.
    let records = storage
        .read_from(stream_id, Offset::ZERO, 1024 * 1024)
        .expect("verified read must succeed");
    assert_eq!(records.len(), 4, "all four records must read back");
}