haematite 0.6.1

//! §11 stats-scope pins: reconciliation, the full-scope mutate-nothing
//! identity (both lock-anchor cases), lazy-shard classification, and the
//! ⟨r15, C1⟩ top-level inventory.

use std::error::Error;
use std::fs;

use super::report::{ShardPresence, SweepBlocker, VacuumMode};
use super::test_support::{build_store, snapshot_tree, stats};
use super::{VacuumOptions, vacuum_stats};

/// The headline reconciliation: `enumerated = marked + unmarked` per shard
/// and in total; repeated versions of the same keys leave version garbage the
/// report must show as unmarked; every marked node was hash-verified.
#[test]
fn stats_reconciles_and_reports_version_garbage() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 2, &[b"alpha", b"beta", b"gamma"], 5)?;

    let report = stats(temp.path())?;
    assert_eq!(report.mode, VacuumMode::Stats);
    assert_eq!(report.shard_count, 2);
    assert!(!report.lock_anchor_created, "create() made the anchor");

    let totals = report.totals;
    assert!(totals.enumerated_nodes > 0, "store must hold nodes");
    assert_eq!(
        totals.enumerated_nodes,
        totals.marked_nodes + totals.unmarked_nodes,
        "node counts must reconcile"
    );
    assert_eq!(
        totals.enumerated_bytes,
        totals.marked_bytes + totals.unmarked_bytes,
        "byte counts must reconcile"
    );
    assert!(totals.marked_nodes > 0, "live roots must mark something");
    assert!(
        totals.unmarked_nodes > 0,
        "five versions of each key must leave superseded nodes"
    );
    for shard in &report.shards {
        assert_eq!(
            shard.nodes.enumerated_nodes,
            shard.nodes.marked_nodes + shard.nodes.unmarked_nodes,
            "per-shard counts must reconcile"
        );
    }
    // ⟨r2, B2⟩ every marked node's hash was recomputed before trusting it.
    assert!(
        report.verified_nodes >= totals.marked_nodes,
        "each marked node requires at least one verification"
    );
    // No manifest, no attestation: stats completes but records the sweep
    // refusal it foresees.
    assert!(
        report
            .sweep_blockers
            .iter()
            .any(|blocker| matches!(blocker, SweepBlocker::MetadataUnattested)),
        "missing manifest+attestation must be reported as a sweep blocker"
    );
    assert!(
        !report.trust_boundary.is_empty(),
        "§6: the trust boundary is named in every report"
    );
    Ok(())
}

/// ⟨r2, B3⟩ a WHOLLY absent `shard-{id}` directory is a never-materialised
/// lazy shard: the run proceeds and classifies it, refusing nothing.
#[test]
fn lazy_shards_classified_not_refused() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    // One key over many shards: most shard dirs never materialise.
    build_store(temp.path(), 8, &[b"solo"], 2)?;

    let report = stats(temp.path())?;
    let materialised = report.shards.len();
    let lazy = report.never_materialised_shards;
    assert_eq!(materialised + lazy, 8);
    assert!(materialised >= 1, "the touched shard materialised");
    assert!(
        lazy >= 1,
        "an 8-shard db with one key must have lazy shards"
    );
    // The per-shard vec holds ONLY materialised shards — lazy ids are
    // aggregated so the report stays O(materialised), never O(shard_count).
    for shard in &report.shards {
        assert_eq!(shard.presence, ShardPresence::Materialised);
    }
    Ok(())
}

/// §11 "stats mutates nothing, full-scope": a recursive byte+mtime snapshot
/// over the ENTIRE data dir is identical before and after `vacuum_stats`,
/// including with `.branch-*.tmp` debris present (the read-only decoder must
/// not unlink it) and a configured-but-absent refs dir (the decoder must not
/// create it, which `BranchRefStore::open` would).
#[test]
fn stats_mutates_nothing_full_scope() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 2, &[b"key-a", b"key-b"], 3)?;

    // A canonical refs dir with one real record AND orphaned temp debris.
    let refs_dir = temp.path().join(super::CANONICAL_REFS_DIR);
    write_branch_record(&refs_dir, temp.path(), 0)?;
    fs::write(refs_dir.join(".branch-orphan.tmp"), b"crash leftover")?;

    let absent_refs = temp.path().join("nonexistent-refs");
    let before = snapshot_tree(temp.path())?;

    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.refs_dirs.push(absent_refs.clone());
    let report = vacuum_stats(&options)?;

    let after = snapshot_tree(temp.path())?;
    assert_eq!(before, after, "stats must not change one byte or mtime");
    assert!(!absent_refs.exists(), "the reader must never create dirs");
    assert!(
        report.metadata.supplied_missing.contains(&absent_refs),
        "the absent supplied path is reported, not silently skipped"
    );
    assert!(!report.lock_anchor_created);
    let consulted = report
        .metadata
        .sources
        .iter()
        .find(|source| source.path == refs_dir)
        .ok_or("canonical refs dir must be consulted")?;
    assert_eq!(consulted.records, 1);
    assert_eq!(
        consulted.temp_debris_files, 1,
        "debris counted, not cleaned"
    );
    Ok(())
}

/// ⟨r4, M5⟩ the single named report-only write exception: on a valid legacy
/// dir with no `writer.lock`, the created anchor is the SOLE delta and the
/// report names it.
#[test]
fn legacy_dir_lock_anchor_is_sole_delta() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"legacy"], 2)?;
    let lock_path = temp.path().join("writer.lock");
    fs::remove_file(&lock_path)?;

    let mut before = snapshot_tree(temp.path())?;
    let report = stats(temp.path())?;
    let after = snapshot_tree(temp.path())?;

    assert!(report.lock_anchor_created, "the report names the anchor");
    // The sole delta: the anchor file, empty, at the lock path.
    let anchor = after
        .get(std::path::Path::new("writer.lock"))
        .ok_or("anchor must exist after the run")?;
    let bytes = anchor.0.as_ref().ok_or("anchor is a file")?;
    assert!(bytes.is_empty(), "the anchor is content-free");
    before.insert(std::path::PathBuf::from("writer.lock"), anchor.clone());
    assert_eq!(before, after, "the anchor must be the SOLE delta");
    Ok(())
}

/// ⟨r15, C1⟩ top-level inventory: foreign entries are enumerated (name,
/// kind, size) and stats completes; a `shard-{id}` dir at `id >=
/// shard_count` is a reported sweep refusal while stats completes.
#[test]
fn top_level_inventory_enumerates_and_flags() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 2, &[b"key"], 2)?;

    fs::write(temp.path().join("stray-notes.txt"), b"operator scribble")?;
    fs::create_dir(temp.path().join("foreign-dir"))?;
    fs::write(temp.path().join("foreign-dir/inner"), b"12345")?;
    // A shard DIR beyond shard_count 2, and a coincidentally named FILE.
    let beyond = temp.path().join("shard-7");
    fs::create_dir(&beyond)?;
    fs::write(beyond.join("shard.wal"), b"junk")?;
    fs::write(temp.path().join("shard-9"), b"just a file")?;

    let report = stats(temp.path())?;
    let names: Vec<&str> = report
        .uninventoried
        .iter()
        .map(|entry| entry.name.as_str())
        .collect();
    assert!(names.contains(&"stray-notes.txt"));
    assert!(names.contains(&"foreign-dir"));
    let dir_entry = report
        .uninventoried
        .iter()
        .find(|entry| entry.name == "foreign-dir")
        .ok_or("foreign dir enumerated")?;
    assert_eq!(dir_entry.size_bytes, Some(5), "recursive size accounted");
    assert!(
        report.sweep_blockers.iter().any(|blocker| matches!(
            blocker,
            SweepBlocker::ShardBeyondCount {
                id: 7,
                shard_count: 2
            }
        )),
        "beyond-count shard dir is a reported sweep refusal; stats completes"
    );
    // The beyond-count dir is ALSO enumerated with kind and size — its
    // bytes must not disappear from the inventory (§4).
    let beyond_entry = report
        .uninventoried
        .iter()
        .find(|entry| entry.name == "shard-7")
        .ok_or("beyond-count dir enumerated with kind and size")?;
    assert_eq!(beyond_entry.size_bytes, Some(4));
    // A FILE named like an out-of-range shard is an ordinary uninventoried
    // entry, never a ShardBeyondCount escalation.
    assert!(names.contains(&"shard-9"));
    assert!(
        !report
            .sweep_blockers
            .iter()
            .any(|blocker| matches!(blocker, SweepBlocker::ShardBeyondCount { id: 9, .. })),
        "a file is not a shard"
    );
    assert_eq!(
        report.shards.len() + report.never_materialised_shards,
        2,
        "beyond-count dirs never join the shard model"
    );
    Ok(())
}

/// Non-canonical shard-dir spellings are foreign entries, not shards.
#[test]
fn non_canonical_shard_names_are_uninventoried() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 2, &[b"key"], 1)?;
    fs::create_dir(temp.path().join("shard-01"))?;
    fs::create_dir(temp.path().join("shard-x"))?;

    let report = stats(temp.path())?;
    let names: Vec<&str> = report
        .uninventoried
        .iter()
        .map(|entry| entry.name.as_str())
        .collect();
    assert!(names.contains(&"shard-01"), "leading zero is not canonical");
    assert!(names.contains(&"shard-x"));
    Ok(())
}

/// Write one HBR1 record into `refs_dir` whose anchor/head are shard
/// `shard_id`'s committed root — a valid record pointing at live data.
fn write_branch_record(
    refs_dir: &std::path::Path,
    data_dir: &std::path::Path,
    shard_id: usize,
) -> Result<(), Box<dyn Error>> {
    use crate::branch::refstore::BranchRefStore;
    use crate::branch::{BranchRefRecord, BranchShardRef};

    let root = super::test_support::committed_root(data_dir, shard_id)
        .ok_or("shard must have a committed root")?;
    let mut store = BranchRefStore::open(refs_dir)?;
    store.create(BranchRefRecord {
        name: "pinned".to_owned(),
        created: 1,
        kind: crate::branch::BranchKind::Work,
        namespace_lineage: None,
        seq: 1,
        timestamp: 1,
        shards: vec![BranchShardRef {
            shard_id,
            fork_anchor: root,
            head: root,
        }],
        parents: Vec::new(),
    })?;
    Ok(())
}