iridium-db 0.4.0

use super::*;
use crate::features::storage::sstable;

fn open(base: &std::path::Path) -> StorageHandle {
    open_store(StorageConfig {
        buffer_pool_pages: 8,
        wal_dir: base.join("wal"),
        wal_segment_max_bytes: 1 << 20,
        manifest_path: base.join("ir.manifest"),
        sstable_dir: base.join("sst"),
    })
    .unwrap()
}

/// retire_vector_space succeeds and marks the space retired in the manifest.
#[test]
fn retire_vector_space_marks_space_retired() {
    let base = temp_dir("retire_marks_retired");
    let mut handle = open(&base);
    let space_id = 1_u32;

    // Register the space by writing a vector.
    put_full_node(&mut handle, 1, 1, &[]).unwrap();
    let payload = encode_vector_payload_f32(space_id, VectorMetric::Cosine, &[1.0, 0.0], false);
    put_vector_delta(&mut handle, &encode_delta(1, 2, &payload)).unwrap();

    retire_vector_space(&mut handle, space_id).unwrap();

    assert!(
        handle.manifest.is_space_retired(space_id),
        "space must be retired after retire_vector_space"
    );
}

/// put_vector_delta on a retired space returns SpaceRetired.
#[test]
fn put_vector_delta_rejects_retired_space() {
    let base = temp_dir("retire_rejects_delta");
    let mut handle = open(&base);
    let space_id = 2_u32;

    put_full_node(&mut handle, 1, 1, &[]).unwrap();
    let payload = encode_vector_payload_f32(space_id, VectorMetric::Cosine, &[1.0, 0.0], false);
    put_vector_delta(&mut handle, &encode_delta(1, 2, &payload)).unwrap();

    retire_vector_space(&mut handle, space_id).unwrap();

    let payload2 = encode_vector_payload_f32(space_id, VectorMetric::Cosine, &[0.9, 0.1], false);
    let err = put_vector_delta(&mut handle, &encode_delta(1, 3, &payload2)).unwrap_err();
    assert!(
        matches!(err, StorageError::SpaceRetired(id) if id == space_id),
        "expected SpaceRetired({space_id}), got {err:?}"
    );
}

/// put_vector_delta_batch on a retired space returns SpaceRetired.
#[test]
fn put_vector_delta_batch_rejects_retired_space() {
    let base = temp_dir("retire_rejects_batch");
    let mut handle = open(&base);
    let space_id = 3_u32;

    put_full_node(&mut handle, 1, 1, &[]).unwrap();
    retire_vector_space(&mut handle, space_id).unwrap();

    let payload = encode_vector_payload_f32(space_id, VectorMetric::Cosine, &[1.0, 0.0], false);
    let deltas = vec![encode_delta(1, 2, &payload)];
    let err = put_vector_delta_batch(&mut handle, &deltas).unwrap_err();
    assert!(
        matches!(err, StorageError::SpaceRetired(id) if id == space_id),
        "expected SpaceRetired({space_id}), got {err:?}"
    );
}

/// Retired state survives manifest persist and reload.
#[test]
fn retired_state_survives_manifest_reload() {
    let base = temp_dir("retire_manifest_reload");
    let space_id = 4_u32;

    {
        let mut handle = open(&base);
        put_full_node(&mut handle, 1, 1, &[]).unwrap();
        let payload = encode_vector_payload_f32(space_id, VectorMetric::Cosine, &[1.0, 0.0], false);
        put_vector_delta(&mut handle, &encode_delta(1, 2, &payload)).unwrap();
        retire_vector_space(&mut handle, space_id).unwrap();
        // handle dropped here, releasing the directory lock.
    }

    // Reopen — the manifest is re-read from disk.
    let handle2 = open(&base);
    assert!(
        handle2.manifest.is_space_retired(space_id),
        "retired state must survive manifest reload"
    );
}

/// Compaction removes VectorDelta entries for retired spaces.
#[test]
fn compaction_removes_retired_space_vector_deltas() {
    let base = temp_dir("retire_compaction_cleanup");
    let mut handle = open(&base);
    let space_id = 5_u32;

    // Write a full node + vector delta for space 5, flush to SSTable.
    put_full_node(&mut handle, 10, 1, &[]).unwrap();
    let payload = encode_vector_payload_f32(space_id, VectorMetric::Cosine, &[1.0, 0.0], false);
    put_vector_delta(&mut handle, &encode_delta(10, 2, &payload)).unwrap();
    flush(&mut handle).unwrap();

    // Write a second SSTable (compaction only fires with ≥2 runs).
    put_full_node(&mut handle, 11, 1, &[]).unwrap();
    flush(&mut handle).unwrap();

    assert_eq!(
        handle.l0_runs.len(),
        2,
        "need 2 SSTables to trigger compaction"
    );

    // Retire space 5, then compact.
    retire_vector_space(&mut handle, space_id).unwrap();
    compact(&mut handle, Some(0)).unwrap();

    assert_eq!(handle.l0_runs.len(), 1, "compaction should reduce to 1 run");

    // Read the compacted SSTable and verify no VectorDelta for space 5.
    let compacted_path = handle.l0_runs[0].clone();
    let table = sstable::read_sstable(&compacted_path).unwrap();
    let has_retired_vector_delta = table.entries.iter().any(|e| {
        if e.kind != sstable::EntryKind::VectorDelta {
            return false;
        }
        matches!(
            decode_vector_payload(&e.value),
            Ok(DecodedVectorPayload::Structured(v)) if v.descriptor.space_id == space_id
        )
    });
    assert!(
        !has_retired_vector_delta,
        "compaction must remove VectorDelta entries for retired space {space_id}"
    );
}

/// Non-retired spaces are not affected by retirement of another space.
#[test]
fn non_retired_space_unaffected_by_other_retirement() {
    let base = temp_dir("retire_other_unaffected");
    let mut handle = open(&base);

    put_full_node(&mut handle, 1, 1, &[]).unwrap();
    put_full_node(&mut handle, 2, 1, &[]).unwrap();

    let payload1 = encode_vector_payload_f32(1, VectorMetric::Cosine, &[1.0, 0.0], false);
    let payload2 = encode_vector_payload_f32(2, VectorMetric::Cosine, &[0.0, 1.0], false);
    put_vector_delta(&mut handle, &encode_delta(1, 2, &payload1)).unwrap();
    put_vector_delta(&mut handle, &encode_delta(2, 2, &payload2)).unwrap();

    retire_vector_space(&mut handle, 1).unwrap();

    // Space 2 should still accept writes.
    let payload2b = encode_vector_payload_f32(2, VectorMetric::Cosine, &[0.1, 0.9], false);
    put_vector_delta(&mut handle, &encode_delta(2, 3, &payload2b)).unwrap();
}