haematite 0.6.2

//! §11 metadata pins: HBR1 association (§2 M2), snapshot resolution (§2
//! M3), the parents non-pin, and manifest shapes (§3).

use std::error::Error;
use std::fs;
use std::path::{Path, PathBuf};

use serde_json::json;

use crate::branch::refstore::BranchRefStore;
use crate::branch::snapshot::SnapshotRegistry;
use crate::branch::{BranchRefRecord, BranchShardRef};
use crate::db::Database;
use crate::tree::Hash;

use super::error::VacuumError;
use super::report::SweepBlocker;
use super::test_support::{committed_root, config_for, expect_refusal, stats};
use super::{VacuumOptions, vacuum_stats};

/// Two-shard store where BOTH shards materialised, with version garbage.
/// Returns a key routed to each shard.
fn build_two_shards(data_dir: &Path) -> Result<(Vec<u8>, Vec<u8>), Box<dyn Error>> {
    let db = Database::create(config_for(data_dir, 2))?;
    let key_for = |target: usize| -> Result<Vec<u8>, Box<dyn Error>> {
        (0..10_000_u64)
            .map(|candidate| format!("k{candidate}").into_bytes())
            .find(|key| db.shard_for(key) == target)
            .ok_or_else(|| "no key for shard".into())
    };
    let key0 = key_for(0)?;
    let key1 = key_for(1)?;
    for version in 0..3_u64 {
        db.append(
            key0.clone(),
            vec![format!("v{version}").into_bytes()],
            version,
        )?;
        db.append(
            key1.clone(),
            vec![format!("v{version}").into_bytes()],
            version,
        )?;
    }
    drop(db);
    Ok((key0, key1))
}

fn write_record(
    refs_dir: &Path,
    name: &str,
    shards: Vec<BranchShardRef>,
    parents: Vec<Hash>,
) -> Result<(), Box<dyn Error>> {
    let mut store = BranchRefStore::open(refs_dir)?;
    store.create(BranchRefRecord {
        name: name.to_owned(),
        created: 1,
        kind: crate::branch::BranchKind::Work,
        namespace_lineage: None,
        seq: 1,
        timestamp: 1,
        shards,
        parents,
    })?;
    Ok(())
}

fn options_with_refs(data_dir: &Path, refs_dir: &Path) -> VacuumOptions {
    let mut options = VacuumOptions::new(data_dir.to_path_buf());
    options.refs_dirs.push(refs_dir.to_path_buf());
    options
}

/// §2 M2 / §11: a branch record whose `shard_id` ≥ `shard_count` is a typed
/// refusal — the config's model or the record is wrong.
#[test]
fn branch_record_beyond_shard_count_refuses() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    let (_key0, _key1) = build_two_shards(temp.path())?;
    let root = committed_root(temp.path(), 0).ok_or("root")?;
    let refs = tempfile::tempdir()?;
    write_record(
        refs.path(),
        "wild",
        vec![BranchShardRef {
            shard_id: 9,
            fork_anchor: root,
            head: root,
        }],
        Vec::new(),
    )?;

    let error = expect_refusal(vacuum_stats(&options_with_refs(temp.path(), refs.path())))?;
    assert!(
        matches!(
            &error,
            VacuumError::ShardBeyondCount {
                id: 9,
                shard_count: 2
            }
        ),
        "got {error:?}"
    );
    Ok(())
}

/// §2 M2 / §11: an anchor/head that fails to resolve in
/// `shard-{record.shard_id}/store` refuses EVEN WHEN the same hash resolves
/// in a different store — the persisted association is the contract, and
/// cross-store coincidence must not launder it.
#[test]
fn branch_association_is_not_laundered_by_coincidence() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    let (_key0, _key1) = build_two_shards(temp.path())?;
    // Shard 1's root resolves in shard 1 — the record claims shard 0.
    let foreign_root = committed_root(temp.path(), 1).ok_or("root")?;
    let refs = tempfile::tempdir()?;
    write_record(
        refs.path(),
        "misfiled",
        vec![BranchShardRef {
            shard_id: 0,
            fork_anchor: foreign_root,
            head: foreign_root,
        }],
        Vec::new(),
    )?;

    let error = expect_refusal(vacuum_stats(&options_with_refs(temp.path(), refs.path())))?;
    match &error {
        VacuumError::UnresolvableRoot { root, .. } => assert_eq!(*root, foreign_root),
        other => return Err(format!("expected UnresolvableRoot, got {other:?}").into()),
    }
    Ok(())
}

/// Build a single-shard store with a captured SUPERSEDED root: version 0's
/// committed root, no longer referenced by the final WAL marker.
fn build_with_old_root(data_dir: &Path) -> Result<Hash, Box<dyn Error>> {
    let db = Database::create(config_for(data_dir, 1))?;
    db.append(b"kv".to_vec(), vec![b"v0".to_vec()], 0)?;
    drop(db);
    let old_root = committed_root(data_dir, 0).ok_or("v0 root")?;
    let db = Database::open(data_dir)?;
    db.append(b"kv".to_vec(), vec![b"v1".to_vec()], 1)?;
    db.append(b"other".to_vec(), vec![b"v1".to_vec()], 0)?;
    drop(db);
    let new_root = committed_root(data_dir, 0).ok_or("v1 root")?;
    assert_ne!(old_root, new_root, "the old root must be superseded");
    Ok(old_root)
}

/// §2 M3 / §11: an unbound registry root resolves-in-at-least-one-store and
/// marks there — a superseded root the M1 walk would leave unmarked is
/// retained by its snapshot pin (leak-safe over-mark pinned as retention).
#[test]
fn snapshot_pin_retains_superseded_root() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    let old_root = build_with_old_root(temp.path())?;

    let baseline = stats(temp.path())?;
    assert!(
        baseline.totals.unmarked_nodes > 0,
        "the superseded version is garbage without a pin"
    );

    let external = tempfile::tempdir()?;
    let registry_file = external.path().join("snapshots.hsr");
    let mut registry = SnapshotRegistry::open(&registry_file)?;
    registry.name_at("keep-v0", old_root, 1)?;
    drop(registry);

    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.snapshot_files.push(registry_file);
    let pinned = vacuum_stats(&options)?;

    assert_eq!(pinned.marks.snapshot_roots, 1);
    assert!(
        pinned.totals.marked_nodes > baseline.totals.marked_nodes,
        "the snapshot pin must retain nodes the baseline left unmarked"
    );
    Ok(())
}

/// §2 M3 / §11: a registry root resolving in NO store is `UnresolvableRoot`
/// — never silently skipped.
#[test]
fn snapshot_resolving_nowhere_refuses() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_two_shards(temp.path())?;
    let registry_dir = tempfile::tempdir()?;
    let registry_file = registry_dir.path().join("snapshots.hsr");
    let mut registry = SnapshotRegistry::open(&registry_file)?;
    registry.name_at("phantom", Hash::from_bytes([0xAB; 32]), 1)?;
    drop(registry);

    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.snapshot_files.push(registry_file);
    let error = expect_refusal(vacuum_stats(&options))?;
    assert!(
        matches!(&error, VacuumError::UnresolvableRoot { .. }),
        "got {error:?}"
    );
    Ok(())
}

/// §2 non-pin, test-enforced: HBR1 `parents` are lineage metadata and mark
/// NOTHING — a superseded root named only by `parents` stays unmarked.
#[test]
fn branch_parents_are_not_pins() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    let old_root = build_with_old_root(temp.path())?;
    let baseline = stats(temp.path())?;

    let current = committed_root(temp.path(), 0).ok_or("current root")?;
    let refs = tempfile::tempdir()?;
    write_record(
        refs.path(),
        "lineage",
        vec![BranchShardRef {
            shard_id: 0,
            fork_anchor: current,
            head: current,
        }],
        vec![old_root],
    )?;

    let report = vacuum_stats(&options_with_refs(temp.path(), refs.path()))?;
    assert_eq!(
        report.totals.marked_nodes, baseline.totals.marked_nodes,
        "anchor/head duplicate M1's root and parents must add NOTHING"
    );
    Ok(())
}

fn write_manifest_json(data_dir: &Path, value: &serde_json::Value) -> Result<(), Box<dyn Error>> {
    fs::write(
        data_dir.join(super::manifest::MANIFEST_FILE),
        serde_json::to_vec_pretty(value)?,
    )?;
    Ok(())
}

/// §3 ⟨r7⟩ / §11: manifest entries are reported with their durable states;
/// non-terminal states and missing listed sources are sweep blockers; stats
/// completes and still consults readable sources.
#[test]
fn manifest_states_reported_and_blockers_recorded() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_two_shards(temp.path())?;
    // A real refs dir listed as Publishing (readable, unsettled).
    let refs_dir = temp.path().join("listed-refs");
    let root = committed_root(temp.path(), 0).ok_or("root")?;
    write_record(
        &refs_dir,
        "mid-publish",
        vec![BranchShardRef {
            shard_id: 0,
            fork_anchor: root,
            head: root,
        }],
        Vec::new(),
    )?;
    let ghost: PathBuf = temp.path().join("ghost-refs");
    write_manifest_json(
        temp.path(),
        &json!({
            "format_version": 1,
            "generation": 4,
            "entries": [
                {
                    "kind": "refs_dir",
                    "path": refs_dir,
                    "state": "publishing",
                    "reservation_nonce": 7
                },
                {
                    "kind": "refs_dir",
                    "path": ghost,
                    "state": "published",
                    "reservation_nonce": 8
                }
            ]
        }),
    )?;

    let report = stats(temp.path())?;
    let manifest = report.metadata.manifest.as_ref().ok_or("manifest read")?;
    assert_eq!(manifest.generation, 4);
    assert_eq!(manifest.entries.len(), 2);
    assert!(
        report.sweep_blockers.iter().any(|blocker| matches!(
            blocker,
            SweepBlocker::ManifestEntryUnsettled { state, .. } if state == "Publishing"
        )),
        "unsettled entry recorded"
    );
    assert!(
        report
            .sweep_blockers
            .iter()
            .any(|blocker| matches!(blocker, SweepBlocker::ManifestSourceMissing { path } if path == &ghost)),
        "missing listed source recorded"
    );
    // The readable unsettled source WAS consulted: stats reports whatever
    // records are visible.
    assert!(
        report
            .metadata
            .sources
            .iter()
            .any(|source| source.path == refs_dir && source.records == 1),
        "readable listed source consulted"
    );
    // The listed dir is accounted at top level (not uninventoried).
    assert!(
        report
            .uninventoried
            .iter()
            .all(|entry| entry.name != "listed-refs"),
        "manifest-listed in-dir source is accounted"
    );
    Ok(())
}

/// §3 ⟨r4⟩ / §11 stale-manifest detection: canonical metadata present but
/// unlisted in an existing manifest is proof the manifest lies.
#[test]
fn stale_manifest_flagged_on_unlisted_canonical() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_two_shards(temp.path())?;
    let root = committed_root(temp.path(), 0).ok_or("root")?;
    let canonical = temp.path().join(super::CANONICAL_REFS_DIR);
    write_record(
        &canonical,
        "unlisted",
        vec![BranchShardRef {
            shard_id: 0,
            fork_anchor: root,
            head: root,
        }],
        Vec::new(),
    )?;
    write_manifest_json(
        temp.path(),
        &json!({ "format_version": 1, "generation": 1, "entries": [] }),
    )?;

    let report = stats(temp.path())?;
    assert!(
        report
            .sweep_blockers
            .iter()
            .any(|blocker| matches!(blocker, SweepBlocker::StaleManifest { unlisted } if unlisted == &canonical)),
        "unlisted canonical source must be flagged"
    );
    // Still consulted (unbound) — the operator sees its records.
    assert!(
        report
            .metadata
            .sources
            .iter()
            .any(|source| source.path == canonical && source.records == 1)
    );
    Ok(())
}

/// §2 M3 ⟨r4, M3b⟩ / §11: a manifest-bound registry's roots are walked ONLY
/// in the declared store(s) — a root absent from its declared store refuses
/// even when fully present in another. Companion: the same shape UNBOUND
/// marks in the resolving store (the fallback rule, pinned as distinct).
#[test]
fn manifest_bound_registry_walks_only_declared_stores() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_two_shards(temp.path())?;
    let shard1_root = committed_root(temp.path(), 1).ok_or("root")?;
    let registry_file = temp.path().join("bound.hsr");
    let mut registry = SnapshotRegistry::open(&registry_file)?;
    registry.name_at("wrong-store", shard1_root, 1)?;
    drop(registry);
    write_manifest_json(
        temp.path(),
        &json!({
            "format_version": 1,
            "generation": 1,
            "entries": [{
                "kind": "snapshot_registry",
                "path": registry_file,
                "state": "published",
                "reservation_nonce": 1,
                "shards": [0]
            }]
        }),
    )?;

    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::UnresolvableRoot { .. }),
        "got {error:?}"
    );

    // Companion: unbound (manifest removed, registry supplied) resolves in
    // shard 1 and the run completes.
    fs::remove_file(temp.path().join(super::manifest::MANIFEST_FILE))?;
    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.snapshot_files.push(registry_file);
    let report = vacuum_stats(&options)?;
    assert_eq!(report.marks.snapshot_roots, 1);
    Ok(())
}

/// §3: an existing manifest that fails to decode fails LOUD — it is the
/// trust root. A newer format stamp refuses identically.
#[test]
fn undecodable_or_newer_manifest_fails_loud() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_two_shards(temp.path())?;
    fs::write(
        temp.path().join(super::manifest::MANIFEST_FILE),
        b"{ not json",
    )?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::MetadataOpen { .. }),
        "got {error:?}"
    );

    write_manifest_json(
        temp.path(),
        &json!({ "format_version": 2, "generation": 1, "entries": [] }),
    )?;
    let error = expect_refusal(stats(temp.path()))?;
    match &error {
        VacuumError::MetadataOpen { reason, .. } => {
            assert!(reason.contains("format_version"), "{reason}");
        }
        other => return Err(format!("expected MetadataOpen, got {other:?}").into()),
    }
    Ok(())
}