velesdb-core 5.1.0

//! Tests for sparse index persistence: WAL, compaction, and loading.

#![allow(clippy::cast_possible_truncation, clippy::cast_precision_loss)]

use tempfile::tempdir;

use super::inverted_index::SparseInvertedIndex;
use super::persistence::*;
use super::types::SparseVector;

fn make_vector(pairs: Vec<(u32, f32)>) -> SparseVector {
    SparseVector::new(pairs)
}

#[test]
fn test_wal_write_and_replay() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    let index1 = SparseInvertedIndex::new();
    // Insert 100 vectors and write WAL entries
    for i in 0..100u64 {
        let v = make_vector(vec![(1, 1.0), (2, 0.5 + i as f32 * 0.01)]);
        index1.insert(i, &v);
        wal_append_upsert(&wal_path, i, &v).unwrap();
    }

    // Create fresh index and replay
    let index2 = SparseInvertedIndex::new();
    let count = wal_replay(&wal_path, &index2).unwrap();
    assert_eq!(count, 100);
    assert_eq!(index2.doc_count(), 100);

    // Verify postings match
    let p1 = index1.get_all_postings(1);
    let p2 = index2.get_all_postings(1);
    assert_eq!(p1.len(), p2.len());
    for (a, b) in p1.iter().zip(p2.iter()) {
        assert_eq!(a.doc_id, b.doc_id);
        assert!((a.weight - b.weight).abs() < f32::EPSILON);
    }
}

#[test]
fn test_wal_truncated_entry() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    // Write one valid entry
    let v = make_vector(vec![(1, 1.0)]);
    wal_append_upsert(&wal_path, 42, &v).unwrap();

    // Append 5 random bytes (simulating truncation)
    {
        use std::io::Write;
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&wal_path)
            .unwrap();
        f.write_all(&[0xFF, 0x00, 0xAA, 0xBB, 0xCC]).unwrap();
    }

    // Replay should recover the valid entry
    let index = SparseInvertedIndex::new();
    let count = wal_replay(&wal_path, &index).unwrap();
    assert_eq!(count, 1);
    assert_eq!(index.doc_count(), 1);

    let postings = index.get_all_postings(1);
    assert_eq!(postings.len(), 1);
    assert_eq!(postings[0].doc_id, 42);
}

#[test]
fn test_wal_delete_replay() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    let v = make_vector(vec![(1, 1.0)]);
    wal_append_upsert(&wal_path, 1, &v).unwrap();
    wal_append_upsert(&wal_path, 2, &v).unwrap();
    wal_append_delete(&wal_path, 1).unwrap();

    let index = SparseInvertedIndex::new();
    let count = wal_replay(&wal_path, &index).unwrap();
    assert_eq!(count, 3);
    // doc_count is 1 (2 inserts - 1 delete)
    assert_eq!(index.doc_count(), 1);

    let postings = index.get_all_postings(1);
    assert_eq!(postings.len(), 1);
    assert_eq!(postings[0].doc_id, 2);
}

#[test]
fn test_compaction_round_trip() {
    let dir = tempdir().unwrap();

    let index1 = SparseInvertedIndex::new();
    for i in 0..500u64 {
        let v = make_vector(vec![
            (i as u32 % 50, 1.0 + (i as f32) * 0.001),
            (100 + i as u32 % 20, 0.5),
        ]);
        index1.insert(i, &v);
    }

    // Compact to disk
    compact(dir.path(), &index1).unwrap();

    // Load from disk
    let loaded = load_from_disk(dir.path()).unwrap();
    assert!(loaded.is_some());
    let index2 = loaded.unwrap();

    assert_eq!(index2.doc_count(), 500);

    // Verify search results match for a sample term
    let p1 = index1.get_all_postings(5);
    let p2 = index2.get_all_postings(5);
    assert_eq!(p1.len(), p2.len());
    for (a, b) in p1.iter().zip(p2.iter()) {
        assert_eq!(a.doc_id, b.doc_id);
        assert!((a.weight - b.weight).abs() < f32::EPSILON);
    }
}

#[test]
fn test_empty_directory_returns_none() {
    let dir = tempdir().unwrap();
    let result = load_from_disk(dir.path()).unwrap();
    assert!(result.is_none());
}

#[test]
fn test_full_restart_simulation() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    // Phase 1: Insert and compact some vectors
    let index1 = SparseInvertedIndex::new();
    for i in 0..50u64 {
        let v = make_vector(vec![(1, 1.0), (2, 2.0)]);
        index1.insert(i, &v);
    }
    compact(dir.path(), &index1).unwrap();

    // Phase 2: More inserts via WAL only (simulating in-flight mutations)
    for i in 50..60u64 {
        let v = make_vector(vec![(1, 3.0), (3, 1.0)]);
        wal_append_upsert(&wal_path, i, &v).unwrap();
    }

    // Phase 3: Simulate restart — load from disk + replay WAL
    let loaded = load_from_disk(dir.path()).unwrap();
    assert!(loaded.is_some());
    let index2 = loaded.unwrap();

    // Should have 50 compacted + 10 replayed = 60
    assert_eq!(index2.doc_count(), 60);

    // Term 1: all 60 docs
    let p1 = index2.get_all_postings(1);
    assert_eq!(p1.len(), 60);

    // Term 3: only docs 50..60
    let p3 = index2.get_all_postings(3);
    assert_eq!(p3.len(), 10);
}

#[test]
fn test_meta_contains_correct_values() {
    let dir = tempdir().unwrap();

    let index = SparseInvertedIndex::new();
    for i in 0..25u64 {
        let v = make_vector(vec![(i as u32 % 5, 1.0), (10, 0.5)]);
        index.insert(i, &v);
    }
    compact(dir.path(), &index).unwrap();

    // Read metadata from the slot selected by the durable manifest.
    let active = super::persistence_generation::active_snapshot(dir.path(), "sparse")
        .unwrap()
        .unwrap();
    let meta_data = std::fs::read(active.paths.meta).unwrap();
    let meta: SparseMeta = postcard::from_bytes(&meta_data).unwrap();
    assert_eq!(meta.version, 1);
    assert_eq!(meta.doc_count, 25);
    // 5 terms (0..4) + term 10 = 6 terms
    assert_eq!(meta.term_count, 6);
}

#[test]
fn test_wal_missing_file_returns_zero() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("nonexistent.wal");
    let index = SparseInvertedIndex::new();
    let count = wal_replay(&wal_path, &index).unwrap();
    assert_eq!(count, 0);
}

/// Simulates a crash that left a `.tmp` file behind from a previous interrupted compaction.
///
/// Verifies that `load_from_disk` ignores stale `.tmp` artefacts and correctly recovers
/// state from the WAL alone (no `sparse.meta` present — WAL-only scenario).
#[test]
fn test_partial_compaction_crash_recovery() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    // Insert 5 distinct vectors and record them only in the WAL (no compaction).
    for i in 0..5u64 {
        let v = make_vector(vec![(i as u32, 1.0 + i as f32 * 0.1)]);
        wal_append_upsert(&wal_path, i, &v).unwrap();
    }

    // Simulate an interrupted compaction: the .tmp file exists but the final sparse.idx
    // and sparse.meta files were never atomically renamed into place.
    let tmp_path = dir.path().join("sparse.idx.tmp");
    std::fs::write(&tmp_path, b"garbage partial write").unwrap();

    // sparse.meta must NOT exist so load_from_disk follows the WAL-only path.
    assert!(!dir.path().join("sparse.meta").exists());

    // load_from_disk must ignore the .tmp file and recover from the WAL.
    let loaded = load_from_disk(dir.path()).unwrap();
    assert!(
        loaded.is_some(),
        "WAL-only load should return Some after partial compaction crash"
    );
    let index = loaded.unwrap();

    // All 5 WAL-inserted documents must be present.
    assert_eq!(index.doc_count(), 5);

    // Verify each term has exactly one posting with the correct doc_id.
    for i in 0..5u64 {
        let postings = index.get_all_postings(i as u32);
        assert_eq!(
            postings.len(),
            1,
            "term {i} should have exactly one posting"
        );
        assert_eq!(postings[0].doc_id, i);
    }

    // The stale .tmp file must still be present (load_from_disk must not delete it).
    assert!(
        tmp_path.exists(),
        "load_from_disk must not remove stale .tmp artefacts"
    );
}

#[test]
fn test_compaction_truncates_wal() {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    let index = SparseInvertedIndex::new();
    let v = make_vector(vec![(1, 1.0)]);
    index.insert(0, &v);
    wal_append_upsert(&wal_path, 0, &v).unwrap();

    // WAL should have content
    assert!(std::fs::metadata(&wal_path).unwrap().len() > 0);

    compact(dir.path(), &index).unwrap();

    // The old records are gone; the durable generation header is logically empty.
    let replayed = wal_replay(&wal_path, &SparseInvertedIndex::new()).unwrap();
    assert_eq!(replayed, 0);
}

fn compacted_index_with_pending_wal() -> (
    tempfile::TempDir,
    SparseInvertedIndex,
    std::path::PathBuf,
    Vec<u8>,
) {
    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");
    let index = SparseInvertedIndex::new();
    let first = make_vector(vec![(1, 1.0)]);
    wal_append_upsert(&wal_path, 1, &first).unwrap();
    index.insert(1, &first);
    compact(dir.path(), &index).unwrap();

    let second = make_vector(vec![(2, 2.0)]);
    wal_append_upsert(&wal_path, 2, &second).unwrap();
    index.insert(2, &second);
    let wal_before = std::fs::read(&wal_path).unwrap();
    (dir, index, wal_path, wal_before)
}

fn assert_recovery_after_fault(boundary: PublicationBoundary) {
    let (dir, index, wal_path, wal_before) = compacted_index_with_pending_wal();
    let _fault = PublicationFaultGuard::inject(boundary);
    compact(dir.path(), &index).expect_err("publication boundary must fail");

    assert_eq!(std::fs::read(wal_path).unwrap(), wal_before);
    let loaded = load_from_disk(dir.path()).unwrap().unwrap();
    assert_eq!(loaded.doc_count(), 2);
    assert_eq!(loaded.get_all_postings(1)[0].doc_id, 1);
    assert_eq!(loaded.get_all_postings(2)[0].doc_id, 2);
}

#[test]
fn every_publication_boundary_preserves_a_recoverable_generation() {
    for boundary in [
        PublicationBoundary::IndexPromotion,
        PublicationBoundary::TermsPromotion,
        PublicationBoundary::MetaPromotion,
        PublicationBoundary::CommitPoint,
        PublicationBoundary::WalTruncation,
    ] {
        assert_recovery_after_fault(boundary);
    }
}

#[test]
fn current_layout_loads_without_a_manifest() {
    let dir = tempdir().unwrap();
    let index = SparseInvertedIndex::new();
    index.insert(7, &make_vector(vec![(3, 1.5)]));
    compact(dir.path(), &index).unwrap();

    for extension in ["idx", "terms", "meta"] {
        std::fs::rename(
            dir.path().join(format!(".sparse.next.{extension}")),
            dir.path().join(format!("sparse.{extension}")),
        )
        .unwrap();
    }
    std::fs::remove_file(dir.path().join("sparse.snapshot")).unwrap();
    std::fs::write(dir.path().join("sparse.wal"), []).unwrap();

    let loaded = load_from_disk(dir.path()).unwrap().unwrap();
    assert_eq!(loaded.doc_count(), 1);
    assert_eq!(loaded.get_all_postings(3)[0].doc_id, 7);
}

#[test]
fn manifest_selects_one_complete_snapshot_slot() {
    let dir = tempdir().unwrap();
    let index = SparseInvertedIndex::new();
    index.insert(1, &make_vector(vec![(1, 1.0)]));
    compact(dir.path(), &index).unwrap();
    index.insert(2, &make_vector(vec![(2, 2.0)]));
    compact(dir.path(), &index).unwrap();

    std::fs::write(dir.path().join(".sparse.next.meta"), b"inactive-corruption").unwrap();
    let loaded = load_from_disk(dir.path()).unwrap().unwrap();
    assert_eq!(loaded.doc_count(), 2);
    assert_eq!(loaded.get_all_postings(2)[0].doc_id, 2);
}

/// #897: a compacted `sparse.meta` whose `term_count` disagrees with the decoded
/// term dictionary must be rejected rather than loaded (untrusted-input guard).
#[test]
fn test_load_rejects_term_count_mismatch() {
    let dir = tempdir().unwrap();

    let index = SparseInvertedIndex::new();
    for i in 0..10u64 {
        index.insert(i, &make_vector(vec![(i as u32 % 3, 1.0)]));
    }
    compact(dir.path(), &index).unwrap();

    // Sanity: the valid index still loads.
    assert!(load_from_disk(dir.path()).unwrap().is_some());

    // Overwrite the metadata header with an inflated `term_count` that no longer
    // matches the on-disk term dictionary.
    let meta_path = super::persistence_generation::active_snapshot(dir.path(), "sparse")
        .unwrap()
        .unwrap()
        .paths
        .meta;
    let meta = super::persistence::SparseMeta {
        version: 1,
        doc_count: 10,
        term_count: u32::MAX,
    };
    std::fs::write(&meta_path, postcard::to_allocvec(&meta).unwrap()).unwrap();

    match load_from_disk(dir.path()) {
        Err(e) => assert!(
            e.to_string().contains("term count mismatch"),
            "expected term-count mismatch rejection, got: {e}"
        ),
        Ok(_) => panic!("expected term-count mismatch rejection, got Ok"),
    }
}

/// #897: a sparse WAL upsert header declaring a huge `nnz` (here `u32::MAX`) but a
/// truncated body must be skipped without panicking or pre-allocating gigabytes.
#[test]
fn test_wal_oversized_nnz_is_rejected() {
    use std::io::Write;

    let dir = tempdir().unwrap();
    let wal_path = dir.path().join("sparse.wal");

    // Hand-craft a single upsert entry: [total_len u32][op u8][point_id u64][nnz u32]
    // with total_len covering only the 13-byte header (no pair payload), but nnz
    // set to u32::MAX. The body is truncated relative to the declared nnz.
    let total_len: u32 = 1 + 8 + 4;
    let mut bytes = Vec::new();
    bytes.extend_from_slice(&total_len.to_le_bytes());
    bytes.push(0x01); // WAL_OP_UPSERT
    bytes.extend_from_slice(&7u64.to_le_bytes()); // point_id
    bytes.extend_from_slice(&u32::MAX.to_le_bytes()); // crafted nnz
    {
        let mut f = std::fs::File::create(&wal_path).unwrap();
        f.write_all(&bytes).unwrap();
    }

    let index = SparseInvertedIndex::new();
    // Must not panic, must not OOM, and the truncated entry is skipped (0 replayed).
    let replayed = wal_replay(&wal_path, &index).unwrap();
    assert_eq!(replayed, 0, "crafted oversized-nnz entry must be skipped");
    assert_eq!(index.doc_count(), 0);
}