laburnum 1.17.1

// Copyright Two Neutron Stars Incorporated and contributors
// SPDX-License-Identifier: BlueOak-1.0.0

//! ADR0005 Index Entry Tests
//!
//! This module tests the core ADR0005 concepts:
//! - Separation of CAS Records from Index Entries
//! - `store()` vs `index()` vs `insert()` methods
//! - `IndexEntry` trait and `HandleEntry<P>` operations
//! - Index-free partitions (`IndexEntry = ()`)
//! - Partition store index operations

use crate::{
  ContentHash, Ident,
  database::{
    Database, HasPartition, Partition, PartitionKey,
    RecordHandle,
    chunk::RecordWriter,
    partitions::{HandleEntry, IndexEntry, PartitionStore},
  },
  record::Record,
};

use super::storage::{TestPartition, TestPartitions, TestRecordData};

// =============================================================================
// 1. store() and index() Separation Tests
// =============================================================================

/// Test that `store()` creates a CAS record without an index entry.
///
/// ADR0005 requires that `store()` only adds to CAS storage, not the index.
#[test]
fn test_store_creates_cas_without_index() {
  let mut writer: RecordWriter<TestPartitions> =
    RecordWriter::new(Ident::new("test_task"));

  let record = TestRecordData::Function {
    params: vec![Ident::new("x")],
    return_type: "i32".to_string(),
  };
  let expected_hash = record.content_hash();

  // Use store() - should NOT create index entry
  let handle = writer.store::<TestPartition>(record);

  // Verify handle has correct hash
  assert_eq!(
    handle.content_hash(),
    expected_hash,
    "store() should return handle with correct content hash"
  );

  let chunk = writer.build();

  // Verify record count includes the stored record
  assert!(!chunk.is_empty(), "Chunk should not be empty after store()");

  // Verify NO index entry was created
  let index_entry = chunk.get_hash(&TestPartition::KEY, "any_key");
  assert!(
    index_entry.is_none(),
    "store() should NOT create index entries"
  );

  // Verify the record IS in CAS storage
  let storage = chunk.storage();
  let store: &PartitionStore<TestPartition> =
    <TestPartitions as crate::database::Partitions>::Stores::store(storage);
  let record_ref = store.get(&expected_hash);
  assert!(
    record_ref.is_some(),
    "store() should add record to CAS storage"
  );
}

/// Test storing the same record multiple times (deduplication).
#[test]
fn test_store_deduplicates_identical_records() {
  let mut writer: RecordWriter<TestPartitions> =
    RecordWriter::new(Ident::new("test_task"));

  let record1 = TestRecordData::Function {
    params: vec![Ident::new("a")],
    return_type: "bool".to_string(),
  };
  let record2 = TestRecordData::Function {
    params: vec![Ident::new("a")],
    return_type: "bool".to_string(),
  };

  let hash1 = record1.content_hash();
  let hash2 = record2.content_hash();
  assert_eq!(hash1, hash2, "Identical records should have same hash");

  let handle1 = writer.store::<TestPartition>(record1);
  let handle2 = writer.store::<TestPartition>(record2);

  // Both handles should point to same content
  assert_eq!(
    handle1.content_hash(),
    handle2.content_hash(),
    "Duplicate stores should return handles to same content"
  );
}

/// Test multiple index entries pointing to the same CAS record.
#[test]
fn test_multiple_index_entries_same_record() {
  let mut writer: RecordWriter<TestPartitions> =
    RecordWriter::new(Ident::new("test_task"));

  let record = TestRecordData::Module {
    exports: vec![Ident::new("shared")],
  };
  let hash = record.content_hash();

  // Insert the same record under multiple sort keys. Because records are
  // content-addressed, all three handles should resolve to the same CAS entry.
  let _h1 = writer.insert::<TestPartition, _>("alias::one", record.clone());
  let _h2 = writer.insert::<TestPartition, _>("alias::two", record.clone());
  let _h3 = writer.insert::<TestPartition, _>("alias::three", record);

  let chunk = writer.build();

  // All entries should point to same hash
  assert_eq!(
    chunk.get_hash(&TestPartition::KEY, "alias::one"),
    Some(hash)
  );
  assert_eq!(
    chunk.get_hash(&TestPartition::KEY, "alias::two"),
    Some(hash)
  );
  assert_eq!(
    chunk.get_hash(&TestPartition::KEY, "alias::three"),
    Some(hash)
  );
}

// =============================================================================
// 2. IndexEntry::primary_hash() Tests
// =============================================================================

/// Test that `()` unit type returns `None` from `primary_hash()`.
///
/// ADR0005 uses `()` as IndexEntry for index-free partitions.
#[test]
fn test_unit_type_primary_hash_returns_none() {
  let unit: () = ();
  assert_eq!(
    IndexEntry::primary_hash(&unit),
    None,
    "() IndexEntry should return None from primary_hash()"
  );
}

/// Test that `HandleEntry<P>` returns the handle's hash from `primary_hash()`.
#[test]
fn test_handle_entry_primary_hash_returns_hash() {
  let hash = ContentHash::new(&[10, 20, 30]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);
  let entry: HandleEntry<TestPartition> = handle.into();

  assert_eq!(
    entry.primary_hash(),
    Some(hash),
    "HandleEntry::primary_hash() should return Some(hash)"
  );
}

/// Test that `HandleEntry::content_hash()` matches `primary_hash()`.
#[test]
fn test_handle_entry_content_hash_matches_primary_hash() {
  let hash = ContentHash::new(&[1, 2, 3, 4]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);
  let entry: HandleEntry<TestPartition> = handle.into();

  assert_eq!(
    entry.content_hash(),
    hash,
    "HandleEntry::content_hash() should return the hash"
  );
  assert_eq!(
    entry.primary_hash(),
    Some(entry.content_hash()),
    "primary_hash() should wrap content_hash() in Some"
  );
}

// =============================================================================
// 3. HandleEntry<P> Operations Tests
// =============================================================================

/// Test `HandleEntry<P>` Copy implementation.
#[test]
fn test_handle_entry_copy() {
  let hash = ContentHash::new(&[9, 10, 11, 12]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);
  let entry: HandleEntry<TestPartition> = handle.into();

  // Copy via assignment
  let copied = entry;

  // Original should still be valid (Copy, not Move)
  assert_eq!(entry.content_hash(), hash);
  assert_eq!(copied.content_hash(), hash);
}

/// Test `HandleEntry<P>` Eq implementation.
#[test]
fn test_handle_entry_equality() {
  let hash1 = ContentHash::new(&[1, 2, 3]);
  let hash2 = ContentHash::new(&[1, 2, 3]); // Same content
  let hash3 = ContentHash::new(&[4, 5, 6]); // Different content

  let entry1: HandleEntry<TestPartition> = RecordHandle::new(hash1).into();
  let entry2: HandleEntry<TestPartition> = RecordHandle::new(hash2).into();
  let entry3: HandleEntry<TestPartition> = RecordHandle::new(hash3).into();

  assert_eq!(entry1, entry2, "Entries with same hash should be equal");
  assert_ne!(
    entry1, entry3,
    "Entries with different hash should not be equal"
  );
}

/// Test `HandleEntry<P>` Hash implementation consistency.
#[test]
fn test_handle_entry_hash_consistency() {
  use std::collections::hash_map::DefaultHasher;
  use std::hash::{Hash, Hasher};

  let content_hash = ContentHash::new(&[7, 8, 9]);
  let entry1: HandleEntry<TestPartition> =
    RecordHandle::new(content_hash).into();
  let entry2: HandleEntry<TestPartition> =
    RecordHandle::new(content_hash).into();

  let mut hasher1 = DefaultHasher::new();
  let mut hasher2 = DefaultHasher::new();
  entry1.hash(&mut hasher1);
  entry2.hash(&mut hasher2);

  assert_eq!(
    hasher1.finish(),
    hasher2.finish(),
    "Equal HandleEntries should have equal hashes"
  );
}

/// Test `HandleEntry<P>` in a HashSet (verifies Hash + Eq).
#[test]
fn test_handle_entry_in_hash_set() {
  use std::collections::HashSet;

  let hash1 = ContentHash::new(&[1, 1, 1]);
  let hash2 = ContentHash::new(&[2, 2, 2]);

  let entry1: HandleEntry<TestPartition> = RecordHandle::new(hash1).into();
  let entry1_dup: HandleEntry<TestPartition> = RecordHandle::new(hash1).into();
  let entry2: HandleEntry<TestPartition> = RecordHandle::new(hash2).into();

  let mut set: HashSet<HandleEntry<TestPartition>> = HashSet::new();
  set.insert(entry1);
  set.insert(entry1_dup); // Duplicate, should not increase size
  set.insert(entry2);

  assert_eq!(set.len(), 2, "HashSet should deduplicate HandleEntries");
  assert!(set.contains(&entry1));
  assert!(set.contains(&entry2));
}

/// Test `HandleEntry<P>::new()` constructor.
#[test]
fn test_handle_entry_new_constructor() {
  let hash = ContentHash::new(&[100, 101, 102]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);
  let entry = HandleEntry::new(handle);

  assert_eq!(
    entry.content_hash(),
    hash,
    "HandleEntry::new() should preserve the handle's hash"
  );
}

/// Test `HandleEntry<P>::handle()` accessor.
#[test]
fn test_handle_entry_handle_accessor() {
  let hash = ContentHash::new(&[200, 201, 202]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);
  let entry: HandleEntry<TestPartition> = handle.into();

  let retrieved_handle = entry.handle();
  assert_eq!(
    retrieved_handle.content_hash(),
    hash,
    "HandleEntry::handle() should return the original handle"
  );
}

/// Test `From<RecordHandle<P>>` for `HandleEntry<P>`.
#[test]
fn test_handle_entry_from_record_handle() {
  let hash = ContentHash::new(&[50, 51, 52]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);

  // Use From trait
  let entry: HandleEntry<TestPartition> = HandleEntry::from(handle);

  assert_eq!(entry.content_hash(), hash);

  // Use Into trait
  let entry2: HandleEntry<TestPartition> = handle.into();
  assert_eq!(entry2.content_hash(), hash);
}

// =============================================================================
// 4. Index-Free Partition Tests
// =============================================================================

/// A partition with no index (IndexEntry = ()).
///
/// Records are stored in CAS but not indexed. They're kept alive only
/// by references from other records.
#[derive(Debug)]
pub struct CasOnlyPartition;

impl PartitionKey for CasOnlyPartition {
  const KEY: Ident = Ident::new("test::cas_only");
}

impl Partition for CasOnlyPartition {
  type Record = TestRecordData;
  type IndexEntry = (); // No index!
  // SortKey still needs Display even for index-free partitions
  // In practice, this won't be used since IndexEntry = ()
  type SortKey = String;

  fn index_entry_from_handle(
    _handle: crate::database::RecordHandle<Self>,
  ) -> Self::IndexEntry {
    // No-op for CAS-only partitions
  }
}

/// Verify that CasOnlyPartition compiles with IndexEntry = ().
#[test]
fn test_cas_only_partition_compiles() {
  // This test verifies the type system accepts IndexEntry = ()
  fn _assert_partition<P: Partition>() {}
  _assert_partition::<CasOnlyPartition>();
}

/// Test that () implements IndexEntry with primary_hash() -> None.
#[test]
fn test_unit_index_entry_trait_impl() {
  fn check_index_entry<T: IndexEntry>(entry: T) -> Option<ContentHash> {
    entry.primary_hash()
  }

  let unit: () = ();
  assert_eq!(
    check_index_entry(unit),
    None,
    "() should implement IndexEntry with primary_hash() -> None"
  );
}

// =============================================================================
// 5. PartitionStore Index Operation Tests
// =============================================================================

/// Test `index_get()` returns None for non-existent keys.
#[test]
fn test_partition_store_index_get_missing() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let result = store.index_get("nonexistent::key");
  assert!(
    result.is_none(),
    "index_get() should return None for missing keys"
  );
}

/// Test `index_get()` returns the entry for existing keys.
#[test]
fn test_partition_store_index_get_existing() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash = ContentHash::new(&[1, 2, 3]);
  let handle: RecordHandle<TestPartition> = RecordHandle::new(hash);
  let entry: HandleEntry<TestPartition> = handle.into();

  store.index_insert("test::key".to_string(), entry);

  let result = store.index_get("test::key");
  assert!(
    result.is_some(),
    "index_get() should return Some for existing keys"
  );
  assert_eq!(
    result.map(|e| e.content_hash()),
    Some(hash),
    "Retrieved entry should have correct hash"
  );
}

/// Test `index_insert()` returns None for new keys.
#[test]
fn test_partition_store_index_insert_new_key() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash = ContentHash::new(&[10, 20, 30]);
  let entry: HandleEntry<TestPartition> = RecordHandle::new(hash).into();

  let old = store.index_insert("new::key".to_string(), entry);
  assert!(
    old.is_none(),
    "index_insert() should return None for new keys"
  );
}

/// Test `index_insert()` returns old entry when overwriting.
#[test]
fn test_partition_store_index_insert_overwrite() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash1 = ContentHash::new(&[1, 1, 1]);
  let hash2 = ContentHash::new(&[2, 2, 2]);

  let entry1: HandleEntry<TestPartition> = RecordHandle::new(hash1).into();
  let entry2: HandleEntry<TestPartition> = RecordHandle::new(hash2).into();

  // First insert
  let old1 = store.index_insert("key".to_string(), entry1);
  assert!(old1.is_none());

  // Overwrite
  let old2 = store.index_insert("key".to_string(), entry2);
  assert!(
    old2.is_some(),
    "index_insert() should return old entry on overwrite"
  );
  assert_eq!(
    old2.map(|e| e.content_hash()),
    Some(hash1),
    "Returned old entry should have original hash"
  );

  // Verify new entry is stored
  let current = store.index_get("key");
  assert_eq!(current.map(|e| e.content_hash()), Some(hash2));
}

/// Test `index_range()` returns entries matching prefix.
#[test]
fn test_partition_store_index_range_matching() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  // Insert entries with different prefixes
  let hash_a1 = ContentHash::new(&[1, 0, 0]);
  let hash_a2 = ContentHash::new(&[1, 0, 1]);
  let hash_b1 = ContentHash::new(&[2, 0, 0]);

  store.index_insert(
    "prefix_a::one".to_string(),
    RecordHandle::new(hash_a1).into(),
  );
  store.index_insert(
    "prefix_a::two".to_string(),
    RecordHandle::new(hash_a2).into(),
  );
  store.index_insert(
    "prefix_b::one".to_string(),
    RecordHandle::new(hash_b1).into(),
  );

  // Query with prefix
  let results = store.index_range("prefix_a::");

  assert_eq!(
    results.len(),
    2,
    "Should find 2 entries with prefix 'prefix_a::'"
  );

  let keys: Vec<&str> = results.iter().map(|(k, _)| k.as_str()).collect();
  assert!(keys.contains(&"prefix_a::one"));
  assert!(keys.contains(&"prefix_a::two"));
}

/// Test `index_range()` returns empty for non-matching prefix.
#[test]
fn test_partition_store_index_range_no_match() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash = ContentHash::new(&[1, 2, 3]);
  store.index_insert("foo::bar".to_string(), RecordHandle::new(hash).into());

  let results = store.index_range("nonexistent::");
  assert!(
    results.is_empty(),
    "index_range() should return empty for non-matching prefix"
  );
}

/// Test `index_range()` with empty prefix returns all entries.
#[test]
fn test_partition_store_index_range_empty_prefix() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash1 = ContentHash::new(&[1, 1, 1]);
  let hash2 = ContentHash::new(&[2, 2, 2]);
  let hash3 = ContentHash::new(&[3, 3, 3]);

  store.index_insert("a".to_string(), RecordHandle::new(hash1).into());
  store.index_insert("b".to_string(), RecordHandle::new(hash2).into());
  store.index_insert("c".to_string(), RecordHandle::new(hash3).into());

  let results = store.index_range("");
  assert_eq!(results.len(), 3, "Empty prefix should match all entries");
}

/// Test `index_remove_prefix()` removes matching entries.
#[test]
fn test_partition_store_index_remove_prefix() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash_a1 = ContentHash::new(&[1, 0, 0]);
  let hash_a2 = ContentHash::new(&[1, 0, 1]);
  let hash_b1 = ContentHash::new(&[2, 0, 0]);

  store
    .index_insert("remove::one".to_string(), RecordHandle::new(hash_a1).into());
  store
    .index_insert("remove::two".to_string(), RecordHandle::new(hash_a2).into());
  store
    .index_insert("keep::one".to_string(), RecordHandle::new(hash_b1).into());

  // Remove with prefix
  let removed = store.index_remove_prefix("remove::");

  assert_eq!(removed.len(), 2, "Should remove 2 entries");

  // Verify they're gone
  assert!(store.index_get("remove::one").is_none());
  assert!(store.index_get("remove::two").is_none());

  // Verify non-matching entry remains
  assert!(store.index_get("keep::one").is_some());
}

/// Test `index_remove_prefix()` returns removed entries for GC.
#[test]
fn test_partition_store_index_remove_prefix_returns_entries() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash1 = ContentHash::new(&[10, 10, 10]);
  let hash2 = ContentHash::new(&[20, 20, 20]);

  store.index_insert("del::a".to_string(), RecordHandle::new(hash1).into());
  store.index_insert("del::b".to_string(), RecordHandle::new(hash2).into());

  let removed = store.index_remove_prefix("del::");

  // Verify returned entries have correct hashes (for GC refcount decrements)
  let hashes: Vec<ContentHash> =
    removed.iter().map(|e| e.content_hash()).collect();
  assert!(
    hashes.contains(&hash1),
    "Removed entries should include hash1"
  );
  assert!(
    hashes.contains(&hash2),
    "Removed entries should include hash2"
  );
}

/// Test `index_remove_prefix()` with non-matching prefix removes nothing.
#[test]
fn test_partition_store_index_remove_prefix_no_match() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash = ContentHash::new(&[1, 2, 3]);
  store.index_insert("keep::this".to_string(), RecordHandle::new(hash).into());

  let removed = store.index_remove_prefix("nonexistent::");

  assert!(
    removed.is_empty(),
    "Should remove nothing with non-matching prefix"
  );
  assert!(
    store.index_get("keep::this").is_some(),
    "Existing entry should remain"
  );
}

/// Test `index_hashes()` collects primary hashes from all entries.
#[test]
fn test_partition_store_index_hashes() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash1 = ContentHash::new(&[1, 1, 1]);
  let hash2 = ContentHash::new(&[2, 2, 2]);
  let hash3 = ContentHash::new(&[3, 3, 3]);

  store.index_insert("a".to_string(), RecordHandle::new(hash1).into());
  store.index_insert("b".to_string(), RecordHandle::new(hash2).into());
  store.index_insert("c".to_string(), RecordHandle::new(hash3).into());

  let hashes = store.index_hashes();

  assert_eq!(hashes.len(), 3, "Should collect 3 hashes");
  assert!(hashes.contains(&hash1));
  assert!(hashes.contains(&hash2));
  assert!(hashes.contains(&hash3));
}

/// Test `index_len()` and `index_is_empty()`.
#[test]
fn test_partition_store_index_len_and_empty() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  assert!(store.index_is_empty(), "New store should have empty index");
  assert_eq!(store.index_len(), 0);

  let hash = ContentHash::new(&[1, 2, 3]);
  store.index_insert("key".to_string(), RecordHandle::new(hash).into());

  assert!(!store.index_is_empty());
  assert_eq!(store.index_len(), 1);
}

/// Test `index_values()` returns all entries.
#[test]
fn test_partition_store_index_values() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash1 = ContentHash::new(&[1, 1, 1]);
  let hash2 = ContentHash::new(&[2, 2, 2]);

  store.index_insert("x".to_string(), RecordHandle::new(hash1).into());
  store.index_insert("y".to_string(), RecordHandle::new(hash2).into());

  let values = store.index_values();

  assert_eq!(values.len(), 2);
  let hashes: Vec<ContentHash> =
    values.iter().map(|e| e.content_hash()).collect();
  assert!(hashes.contains(&hash1));
  assert!(hashes.contains(&hash2));
}

/// Test `index_entries()` returns all (key, entry) pairs.
#[test]
fn test_partition_store_index_entries() {
  let store: PartitionStore<TestPartition> = PartitionStore::new();

  let hash1 = ContentHash::new(&[1, 1, 1]);
  let hash2 = ContentHash::new(&[2, 2, 2]);

  store.index_insert("key1".to_string(), RecordHandle::new(hash1).into());
  store.index_insert("key2".to_string(), RecordHandle::new(hash2).into());

  let entries = store.index_entries();

  assert_eq!(entries.len(), 2);

  let keys: Vec<&str> = entries.iter().map(|(k, _)| k.as_str()).collect();
  assert!(keys.contains(&"key1"));
  assert!(keys.contains(&"key2"));
}

// =============================================================================
// 6. Integration: Index Entries with GC
// =============================================================================

/// Test that index entries are GC roots (their referenced records are kept alive).
#[test]
fn test_index_entries_are_gc_roots() {
  let db: Database<TestPartitions> = Database::new();
  let source_cache = crate::source::cache::reporter::SourceCacheReader::new_empty_for_test();

  // Create and commit a record with an index entry
  let mut writer: RecordWriter<TestPartitions> =
    RecordWriter::new(Ident::new("test_task"));
  let record = TestRecordData::Function {
    params: vec![Ident::new("x")],
    return_type: "i32".to_string(),
  };
  let hash = record.content_hash();
  writer.insert::<TestPartition, _>("func::main", record);
  let chunk = writer.build();
  let _ = db.commit_chunk(chunk, &source_cache);

  // Collect index hashes (GC roots)
  let roots = db.collect_index_hashes();

  assert!(
    roots.iter().any(|r| r.hash == hash),
    "Index entry hash should be a GC root"
  );
}

/// Test that overwriting an index entry changes the GC root.
#[test]
fn test_overwriting_index_entry_changes_gc_root() {
  let db: Database<TestPartitions> = Database::new();
  let source_cache = crate::source::cache::reporter::SourceCacheReader::new_empty_for_test();

  // Commit first record
  let mut writer1: RecordWriter<TestPartitions> =
    RecordWriter::new(Ident::new("task1"));
  let record1 = TestRecordData::Function {
    params: vec![Ident::new("a")],
    return_type: "u32".to_string(),
  };
  let hash1 = record1.content_hash();
  writer1.insert::<TestPartition, _>("func::target", record1);
  let chunk1 = writer1.build();
  let _ = db.commit_chunk(chunk1, &source_cache);

  // Verify first hash is root
  let roots1 = db.collect_index_hashes();
  assert!(roots1.iter().any(|r| r.hash == hash1));

  // Overwrite with different record
  let mut writer2: RecordWriter<TestPartitions> =
    RecordWriter::new(Ident::new("task2"));
  let record2 = TestRecordData::Function {
    params: vec![Ident::new("b"), Ident::new("c")],
    return_type: "i64".to_string(),
  };
  let hash2 = record2.content_hash();
  writer2.insert::<TestPartition, _>("func::target", record2);
  let chunk2 = writer2.build();
  let _ = db.commit_chunk(chunk2, &source_cache);

  // Verify new hash is root, old is not
  let roots2 = db.collect_index_hashes();
  assert!(
    roots2.iter().any(|r| r.hash == hash2),
    "New hash should be GC root after overwrite"
  );
  // Note: hash1 may still be in CAS but should not be a root
}