haematite 0.7.0

//! §11 sweep refusal-totality pins: one per typed refusal, each asserting the
//! refusal is TOTAL — ZERO deletions ANYWHERE, not per-store best-effort.
//!
//! Every fixture builds a store that HOLDS real version garbage (so a clean
//! sweep would delete something), plants the refusal shape, runs
//! `vacuum_sweep` with `execute = true`, asserts the named `VacuumError`, and
//! asserts the whole-tree snapshot is byte+mtime identical afterwards. If a
//! blocker anywhere let even one unlink through, the snapshot comparison
//! fails.

use std::error::Error;
use std::fs;
use std::path::Path;

use serde_json::json;

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

use super::error::VacuumError;
use super::test_support::{
    attested, build_store, committed_root, expect_refusal, node_path, snapshot_tree, store_dir,
};
use super::{VacuumOptions, vacuum_sweep};

/// Build a single-shard store carrying version garbage (a clean sweep of it
/// would delete the superseded nodes), then assert `refusal` left every byte
/// and mtime untouched.
fn assert_total_refusal(
    build: impl FnOnce(&Path) -> Result<VacuumOptions, Box<dyn Error>>,
    check: impl FnOnce(&VacuumError),
) -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    let options = build(temp.path())?;
    let before = snapshot_tree(temp.path())?;

    let error = expect_refusal(vacuum_sweep(&options, true))?;
    check(&error);
    assert_eq!(
        before,
        snapshot_tree(temp.path())?,
        "a refused sweep must delete NOTHING anywhere (refusal is total)"
    );
    Ok(())
}

fn write_record(refs_dir: &Path, name: &str, root: 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: vec![BranchShardRef {
            shard_id: 0,
            fork_anchor: root,
            head: root,
        }],
        parents: Vec::new(),
    })?;
    Ok(())
}

fn write_manifest(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: no manifest and no attestation ⇒ `MetadataUnattested`, zero deletions —
/// a report cannot reveal an omitted path, so the deleting operation refuses.
#[test]
fn unattested_refuses_totally() -> Result<(), Box<dyn Error>> {
    assert_total_refusal(
        |data_dir| Ok(VacuumOptions::new(data_dir.to_path_buf())),
        |error| {
            assert!(
                matches!(
                    error,
                    VacuumError::MetadataUnattested {
                        unlisted_canonical: None
                    }
                ),
                "got {error:?}"
            );
        },
    )
}

/// ⟨r2, B2⟩ a planted hash mismatch ⇒ `NodeHashMismatch`, and NO store is
/// swept — corruption evidence anywhere freezes every deletion.
#[test]
fn hash_mismatch_refuses_totally() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    let root = committed_root(temp.path(), 0).ok_or("root")?;
    let store = store_dir(temp.path(), 0);
    let root_path = node_path(&store, root);
    let donor = fs::read_dir(&store)?
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .filter(|path| path.is_dir())
        .flat_map(|dir| {
            fs::read_dir(dir)
                .into_iter()
                .flatten()
                .filter_map(Result::ok)
        })
        .map(|entry| entry.path())
        .find(|path| path.is_file() && path != &root_path)
        .ok_or("a donor node exists")?;
    fs::write(&root_path, fs::read(&donor)?)?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(vacuum_sweep(&attested(temp.path()), true))?;
    assert!(
        matches!(error, VacuumError::NodeHashMismatch { .. }),
        "got {error:?}"
    );
    assert_eq!(before, snapshot_tree(temp.path())?, "no store swept");
    Ok(())
}

/// §2 M1: a shard store without a decodable WAL ⇒ `StoreWithoutWal`, total.
#[test]
fn store_without_wal_refuses_totally() -> Result<(), Box<dyn Error>> {
    assert_total_refusal(
        |data_dir| {
            fs::remove_file(data_dir.join("shard-0/shard.wal"))?;
            Ok(attested(data_dir))
        },
        |error| {
            assert!(
                matches!(error, VacuumError::StoreWithoutWal { shard_id: 0, .. }),
                "got {error:?}"
            );
        },
    )
}

/// ⟨r15, C1⟩ absent `config.json` ⇒ `ConfigUnreadable` from sweep too.
#[test]
fn config_unreadable_refuses_totally() -> Result<(), Box<dyn Error>> {
    assert_total_refusal(
        |data_dir| {
            fs::remove_file(data_dir.join("config.json"))?;
            Ok(attested(data_dir))
        },
        |error| {
            assert!(
                matches!(error, VacuumError::ConfigUnreadable { .. }),
                "got {error:?}"
            );
        },
    )
}

/// ⟨r15, C1⟩ a `shard-{id}` dir with `id >= shard_count` ⇒ `ShardBeyondCount`
/// — and the in-range store's real garbage is NOT swept (totality).
#[test]
fn shard_beyond_count_refuses_totally() -> Result<(), Box<dyn Error>> {
    assert_total_refusal(
        |data_dir| {
            fs::create_dir(data_dir.join("shard-5"))?;
            Ok(attested(data_dir))
        },
        |error| {
            assert!(
                matches!(
                    error,
                    VacuumError::ShardBeyondCount {
                        id: 5,
                        shard_count: 1
                    }
                ),
                "got {error:?}"
            );
        },
    )
}

/// §4 malformed store entry ⇒ `UnexpectedLayout`, and the live garbage in the
/// SAME store is not swept — the wrong-model rule freezes the whole run.
#[test]
fn malformed_store_entry_refuses_totally() -> Result<(), Box<dyn Error>> {
    assert_total_refusal(
        |data_dir| {
            let store = store_dir(data_dir, 0);
            let prefix = fs::read_dir(&store)?
                .filter_map(Result::ok)
                .map(|entry| entry.path())
                .find(|path| path.is_dir())
                .ok_or("a fanout dir")?;
            fs::write(prefix.join("tooshort"), b"foreign")?;
            Ok(attested(data_dir))
        },
        |error| {
            assert!(
                matches!(error, VacuumError::UnexpectedLayout { .. }),
                "got {error:?}"
            );
        },
    )
}

/// §3 ⟨r7⟩ a non-terminal manifest entry ⇒ `MetadataSourceUnsettled`, total.
#[test]
fn unsettled_manifest_entry_refuses_totally() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    let root = committed_root(temp.path(), 0).ok_or("root")?;
    let refs_dir = temp.path().join("listed-refs");
    write_record(&refs_dir, "mid-publish", root)?;
    write_manifest(
        temp.path(),
        &json!({
            "format_version": 1,
            "generation": 4,
            "entries": [{
                "kind": "refs_dir",
                "path": refs_dir,
                "state": "publishing",
                "reservation_nonce": 7
            }]
        }),
    )?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(vacuum_sweep(
        &VacuumOptions::new(temp.path().to_path_buf()),
        true,
    ))?;
    match &error {
        VacuumError::MetadataSourceUnsettled { state, .. } => assert_eq!(state, "Publishing"),
        other => return Err(format!("expected MetadataSourceUnsettled, got {other:?}").into()),
    }
    assert_eq!(before, snapshot_tree(temp.path())?, "total refusal");
    Ok(())
}

/// §3 ⟨r5, B5⟩ a manifest-listed source that is absent ⇒ `MetadataPathMissing`
/// — any missing listed source refuses, always.
#[test]
fn missing_manifest_source_refuses_totally() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    let ghost = temp.path().join("ghost-refs");
    write_manifest(
        temp.path(),
        &json!({
            "format_version": 1,
            "generation": 1,
            "entries": [{
                "kind": "refs_dir",
                "path": ghost,
                "state": "published",
                "reservation_nonce": 1
            }]
        }),
    )?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(vacuum_sweep(
        &VacuumOptions::new(temp.path().to_path_buf()),
        true,
    ))?;
    assert!(
        matches!(&error, VacuumError::MetadataPathMissing { path } if path == &ghost),
        "got {error:?}"
    );
    assert_eq!(before, snapshot_tree(temp.path())?, "total refusal");
    Ok(())
}

/// §3 ⟨r4, B4⟩ a typo'd supplied path under attestation ⇒ `MetadataPathMissing`
/// — attestation attests completeness, not existence.
#[test]
fn supplied_path_missing_refuses_totally() -> Result<(), Box<dyn Error>> {
    let missing = std::path::PathBuf::from("/definitely/not/here/refs");
    let target = missing.clone();
    assert_total_refusal(
        move |data_dir| {
            let mut options = attested(data_dir);
            options.refs_dirs.push(missing);
            Ok(options)
        },
        move |error| {
            assert!(
                matches!(error, VacuumError::MetadataPathMissing { path } if path == &target),
                "got {error:?}"
            );
        },
    )
}

/// §3 ⟨r4⟩ stale manifest (canonical source present but unlisted) ⇒
/// `MetadataUnattested` naming the unlisted source, total.
#[test]
fn stale_manifest_refuses_totally() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    let root = committed_root(temp.path(), 0).ok_or("root")?;
    let canonical = temp.path().join(super::CANONICAL_REFS_DIR);
    write_record(&canonical, "unlisted", root)?;
    write_manifest(
        temp.path(),
        &json!({ "format_version": 1, "generation": 1, "entries": [] }),
    )?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(vacuum_sweep(
        &VacuumOptions::new(temp.path().to_path_buf()),
        true,
    ))?;
    assert!(
        matches!(&error, VacuumError::MetadataUnattested { unlisted_canonical: Some(path) } if path == &canonical),
        "got {error:?}"
    );
    assert_eq!(before, snapshot_tree(temp.path())?, "total refusal");
    Ok(())
}

/// §2 M3 a snapshot root resolving in NO store ⇒ `UnresolvableRoot`, total.
#[test]
fn unresolvable_snapshot_root_refuses_totally() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    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 = attested(temp.path());
    options.snapshot_files.push(registry_file);
    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(vacuum_sweep(&options, true))?;
    assert!(
        matches!(&error, VacuumError::UnresolvableRoot { .. }),
        "got {error:?}"
    );
    assert_eq!(before, snapshot_tree(temp.path())?, "total refusal");
    Ok(())
}

/// ⟨r9⟩⟨r10⟩ where no real directory-entry durability barrier exists, EVERY
/// sweep mode refuses with `UnsupportedDurability`. Compiles on both platform
/// classes; only runs on the barrier-less one.
#[test]
#[cfg(not(unix))]
fn unsupported_durability_refuses_every_sweep_mode() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"key-a", b"key-b"], 3)?;
    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(vacuum_sweep(&attested(temp.path()), true))?;
    assert!(
        matches!(error, VacuumError::UnsupportedDurability),
        "got {error:?}"
    );
    assert_eq!(before, snapshot_tree(temp.path())?, "total refusal");
    Ok(())
}