haematite 0.6.2

//! Sol r1 fold pins: the non-mutating config preflight, symlink strictness
//! at every model boundary, unreadable-source reporting, the Q2 single-store
//! residency bound, and the missing-descendant refusal.

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

use crate::db::Database;
use crate::tree::Node;

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

/// The config preflight: an anchor-less directory with ABSENT config —
/// i.e. pointing stats at the wrong directory — refuses with ZERO
/// filesystem changes. The `writer.lock` anchor is licensed only on a
/// VALID legacy dir, and validity is established before the create-capable
/// open.
#[test]
fn absent_config_without_anchor_changes_nothing() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    fs::write(temp.path().join("innocent-bystander.txt"), b"not a db")?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::ConfigUnreadable { .. }),
        "got {error:?}"
    );
    assert!(
        !temp.path().join("writer.lock").exists(),
        "the refusal must not create the anchor"
    );
    assert_eq!(
        before,
        snapshot_tree(temp.path())?,
        "zero filesystem changes"
    );
    Ok(())
}

/// Preflight, unreadable half: permission-denied config on an anchor-less
/// legacy dir refuses without creating the anchor.
#[cfg(unix)]
#[test]
fn unreadable_config_without_anchor_creates_no_anchor() -> Result<(), Box<dyn Error>> {
    use std::os::unix::fs::PermissionsExt;
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"legacy"], 1)?;
    let lock_path = temp.path().join("writer.lock");
    fs::remove_file(&lock_path)?;
    let config_path = temp.path().join("config.json");
    fs::set_permissions(&config_path, fs::Permissions::from_mode(0o000))?;

    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::ConfigUnreadable { .. }),
        "got {error:?}"
    );
    assert!(
        !lock_path.exists(),
        "the refusal must not create the anchor on an unvalidated dir"
    );
    fs::set_permissions(&config_path, fs::Permissions::from_mode(0o644))?;
    Ok(())
}

/// Symlinks are never followed at model boundaries: a `shard-{id}` symlink
/// pointing at a perfectly valid external shard refuses (`UnexpectedLayout`)
/// instead of inventorying external data, and a DANGLING shard symlink is
/// NOT classified as lazy — only a wholly absent entry is.
#[cfg(unix)]
#[test]
fn shard_symlinks_refuse_resolving_and_dangling() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    // Move the real shard elsewhere and symlink it back: valid target,
    // still refused.
    let external = tempfile::tempdir()?;
    let shard_dir = temp.path().join("shard-0");
    let moved = external.path().join("shard-0");
    fs::rename(&shard_dir, &moved)?;
    std::os::unix::fs::symlink(&moved, &shard_dir)?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::UnexpectedLayout { .. }),
        "resolving shard symlink must refuse, got {error:?}"
    );

    // Dangling: same refusal, never lazy.
    fs::remove_file(&shard_dir)?;
    std::os::unix::fs::symlink(external.path().join("gone"), &shard_dir)?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::UnexpectedLayout { .. }),
        "dangling shard symlink must refuse, got {error:?}"
    );
    Ok(())
}

/// Store and WAL symlinks refuse identically — the future sweep must never
/// be handed a candidate model that reaches through a link.
#[cfg(unix)]
#[test]
fn store_and_wal_symlinks_refuse() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    let external = tempfile::tempdir()?;

    let store = store_dir(temp.path(), 0);
    let moved_store = external.path().join("store");
    fs::rename(&store, &moved_store)?;
    std::os::unix::fs::symlink(&moved_store, &store)?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::UnexpectedLayout { .. }),
        "store symlink must refuse, got {error:?}"
    );
    fs::remove_file(&store)?;
    fs::rename(&moved_store, &store)?;

    let wal = temp.path().join("shard-0/shard.wal");
    let moved_wal = external.path().join("shard.wal");
    fs::rename(&wal, &moved_wal)?;
    std::os::unix::fs::symlink(&moved_wal, &wal)?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::UnexpectedLayout { .. }),
        "wal symlink must refuse, got {error:?}"
    );
    Ok(())
}

/// A symlink or directory wearing the `.node-*.tmp` debris name is
/// MALFORMED, not debris — the kind check precedes the name pattern.
#[cfg(unix)]
#[test]
fn debris_named_non_files_are_malformed() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let store = store_dir(temp.path(), 0);
    let prefix_dir = fs::read_dir(&store)?
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .find(|path| path.is_dir())
        .ok_or("store has a fanout dir")?;
    fs::create_dir(prefix_dir.join(".node-dir.tmp"))?;
    std::os::unix::fs::symlink(store.join("nowhere"), prefix_dir.join(".node-link.tmp"))?;

    let report = stats(temp.path())?;
    let shard = &report.shards[0];
    assert_eq!(shard.temp_debris_files, 0, "neither is debris");
    assert!(
        shard.malformed.len() >= 2,
        "both are malformed: {:?}",
        shard.malformed
    );
    Ok(())
}

/// §3: a manifest-listed source that EXISTS but cannot be read is reported
/// (entry unreadable, `SourceUnreadable` blocker) and stats COMPLETES —
/// fail-closed for sweep, never fail-stuck for the report.
#[cfg(unix)]
#[test]
fn unreadable_listed_source_reported_stats_completes() -> Result<(), Box<dyn Error>> {
    use std::os::unix::fs::PermissionsExt;
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let refs_dir = temp.path().join("locked-refs");
    fs::create_dir(&refs_dir)?;
    fs::write(
        temp.path().join(super::manifest::MANIFEST_FILE),
        serde_json::to_vec_pretty(&serde_json::json!({
            "format_version": 1,
            "generation": 1,
            "entries": [{
                "kind": "refs_dir",
                "path": refs_dir,
                "state": "published",
                "reservation_nonce": 1
            }]
        }))?,
    )?;
    fs::set_permissions(&refs_dir, fs::Permissions::from_mode(0o000))?;

    let report = stats(temp.path())?;
    fs::set_permissions(&refs_dir, fs::Permissions::from_mode(0o755))?;

    let manifest = report.metadata.manifest.as_ref().ok_or("manifest")?;
    assert!(
        !manifest.entries[0].source_readable,
        "the entry must be reported unreadable"
    );
    assert!(
        report
            .sweep_blockers
            .iter()
            .any(|blocker| matches!(blocker, SweepBlocker::SourceUnreadable { path, .. } if path == &refs_dir)),
        "unreadable source must be a sweep blocker"
    );
    assert!(
        report
            .metadata
            .sources
            .iter()
            .all(|source| source.path != refs_dir),
        "an unscanned source is not a consulted source"
    );
    Ok(())
}

/// The same access-failure handling covers supplied sources.
#[cfg(unix)]
#[test]
fn unreadable_supplied_source_reported_stats_completes() -> Result<(), Box<dyn Error>> {
    use std::os::unix::fs::PermissionsExt;
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let external = tempfile::tempdir()?;
    let refs_dir = external.path().join("refs");
    fs::create_dir(&refs_dir)?;
    fs::set_permissions(&refs_dir, fs::Permissions::from_mode(0o000))?;

    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.refs_dirs.push(refs_dir.clone());
    let report = vacuum_stats(&options)?;
    fs::set_permissions(&refs_dir, fs::Permissions::from_mode(0o755))?;

    assert!(
        report
            .sweep_blockers
            .iter()
            .any(|blocker| matches!(blocker, SweepBlocker::SourceUnreadable { path, .. } if path == &refs_dir)),
        "unreadable supplied source must be a sweep blocker"
    );
    Ok(())
}

/// Lens Q2, pinned: peak inventory residency during a multi-shard stats run
/// is exactly ONE store — never all stores at once. (Thread-local gauge;
/// each test's run is single-threaded.)
#[test]
fn stats_holds_at_most_one_store_inventory() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(
        temp.path(),
        8,
        &[b"a", b"b", b"c", b"d", b"e", b"f", b"g", b"h"],
        3,
    )?;

    super::inventory::residency::reset();
    let report = stats(temp.path())?;
    let materialised = report
        .shards
        .iter()
        .filter(|shard| shard.nodes.enumerated_nodes > 0)
        .count();
    assert!(
        materialised >= 2,
        "the pin needs several materialised stores"
    );
    assert_eq!(
        super::inventory::residency::peak(),
        1,
        "peak residency must be one store, not all stores"
    );
    Ok(())
}

/// §11 mark hardening: a marked root whose DESCENDANT is missing (root file
/// present, child deleted) is `UnresolvableRoot` naming the missing child —
/// the walk fails closed mid-tree, not only at the root.
#[test]
fn missing_descendant_refuses_with_the_child_named() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    // Enough entries that the root is an internal node with real children —
    // v1 boundaries fire ~1/4096 entries, so bulk-append well past that.
    let db = Database::create(config_for(temp.path(), 1))?;
    for batch in 0..3_u64 {
        let events: Vec<Vec<u8>> = (0..5000_u64)
            .map(|index| format!("event-{batch}-{index}").into_bytes())
            .collect();
        db.append(b"stream".to_vec(), events, batch * 5000)?;
    }
    drop(db);

    let root = committed_root(temp.path(), 0).ok_or("committed root")?;
    let store = store_dir(temp.path(), 0);
    let reader = crate::store::DiskStore::new(&store)?;
    let node = crate::store::NodeStore::get(&reader, &root)?.ok_or("root readable")?;
    let Node::Internal(internal) = &*node else {
        return Err("fixture must produce an internal root; grow the key count".into());
    };
    let child = internal
        .children()
        .first()
        .map(|(_lower_bound, hash)| *hash)
        .ok_or("internal root has children")?;
    fs::remove_file(super::test_support::node_path(&store, child))?;

    let error = expect_refusal(stats(temp.path()))?;
    match &error {
        VacuumError::UnresolvableRoot {
            root: reported_root,
            missing,
            ..
        } => {
            assert_eq!(*reported_root, root);
            assert_eq!(*missing, child, "the refusal names the missing child");
        }
        other => return Err(format!("expected UnresolvableRoot, got {other:?}").into()),
    }
    Ok(())
}

/// In-range shard names of the wrong TYPE refuse: a regular file named
/// `shard-0` is an observable damaged state, not a shard and not lazy.
#[test]
fn in_range_shard_name_with_wrong_kind_refuses() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 2, &[b"solo"], 1)?;
    let target = (0..2)
        .find(|shard_id| !temp.path().join(format!("shard-{shard_id}")).exists())
        .ok_or("one shard must be unmaterialised")?;
    fs::write(temp.path().join(format!("shard-{target}")), b"not a dir")?;

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

/// The full-scope identity extends to EXTERNAL supplied sources: a present
/// supplied refs dir outside `data_dir` (with debris) is byte- and
/// mtime-identical after stats.
#[test]
fn external_supplied_source_is_untouched() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    let external = tempfile::tempdir()?;
    let refs_dir = external.path().join("refs");
    {
        use crate::branch::refstore::BranchRefStore;
        use crate::branch::{BranchRefRecord, BranchShardRef};
        let root = committed_root(temp.path(), 0).ok_or("root")?;
        let mut store = BranchRefStore::open(&refs_dir)?;
        store.create(BranchRefRecord {
            name: "external".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(),
        })?;
    }
    fs::write(refs_dir.join(".branch-orphan.tmp"), b"crash leftover")?;

    let data_before = snapshot_tree(temp.path())?;
    let external_before = snapshot_tree(external.path())?;
    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.refs_dirs.push(refs_dir.clone());
    let report = vacuum_stats(&options)?;

    assert_eq!(data_before, snapshot_tree(temp.path())?);
    assert_eq!(
        external_before,
        snapshot_tree(external.path())?,
        "the external supplied source must be untouched, debris included"
    );
    assert!(
        report
            .metadata
            .sources
            .iter()
            .any(|source| source.path == refs_dir && source.records == 1)
    );
    Ok(())
}

/// Sol r2: a source nested under an execute-denied PARENT fails at the
/// lookup itself, not at `read_dir` — that shape must also be reported
/// (`SourceUnreadable`) with stats completing, for listed AND supplied
/// sources.
#[cfg(unix)]
#[test]
fn source_under_denied_parent_reported_stats_completes() -> Result<(), Box<dyn Error>> {
    use std::os::unix::fs::PermissionsExt;
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let parent = temp.path().join("closed");
    let listed = parent.join("refs");
    fs::create_dir_all(&listed)?;
    let external = tempfile::tempdir()?;
    let supplied_parent = external.path().join("closed");
    let supplied = supplied_parent.join("refs");
    fs::create_dir_all(&supplied)?;
    fs::write(
        temp.path().join(super::manifest::MANIFEST_FILE),
        serde_json::to_vec_pretty(&serde_json::json!({
            "format_version": 1,
            "generation": 1,
            "entries": [{
                "kind": "refs_dir",
                "path": listed,
                "state": "published",
                "reservation_nonce": 1
            }]
        }))?,
    )?;
    fs::set_permissions(&parent, fs::Permissions::from_mode(0o000))?;
    fs::set_permissions(&supplied_parent, fs::Permissions::from_mode(0o000))?;

    let mut options = VacuumOptions::new(temp.path().to_path_buf());
    options.refs_dirs.push(supplied.clone());
    let result = vacuum_stats(&options);
    fs::set_permissions(&parent, fs::Permissions::from_mode(0o755))?;
    fs::set_permissions(&supplied_parent, fs::Permissions::from_mode(0o755))?;
    let report = result?;

    let manifest = report.metadata.manifest.as_ref().ok_or("manifest")?;
    assert!(
        !manifest.entries[0].source_readable,
        "lookup-denied listed entry reported unreadable"
    );
    for path in [&listed, &supplied] {
        assert!(
            report
                .sweep_blockers
                .iter()
                .any(|blocker| matches!(blocker, SweepBlocker::SourceUnreadable { path: p, .. } if p == path)),
            "lookup-denied source {path:?} must be a SourceUnreadable blocker"
        );
    }
    Ok(())
}

/// Sol r2: if the lock syscall fails AFTER the create-capable open made the
/// anchor, the refusal DISCLOSES the write — the r4-M5 transparency rule
/// holds on error paths too, where no report exists to carry it.
#[test]
fn failed_lock_after_anchor_creation_is_disclosed() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"legacy"], 1)?;
    let lock_path = temp.path().join("writer.lock");
    fs::remove_file(&lock_path)?;

    crate::db::lock::fail_next_try_lock();
    let error = expect_refusal(stats(temp.path()))?;
    match &error {
        VacuumError::LockIo { anchor_created, .. } => {
            assert!(anchor_created, "the disclosure flag must be set");
        }
        other => return Err(format!("expected LockIo, got {other:?}").into()),
    }
    assert!(
        lock_path.exists(),
        "the anchor was created by the failed run"
    );
    let text = error.to_string();
    assert!(
        text.contains("CREATED the empty advisory writer.lock"),
        "the refusal text must disclose the sole delta: {text}"
    );

    // Contrast: with the anchor pre-existing, the same failure discloses
    // nothing (there was no write).
    crate::db::lock::fail_next_try_lock();
    let error = expect_refusal(stats(temp.path()))?;
    match &error {
        VacuumError::LockIo { anchor_created, .. } => {
            assert!(!anchor_created, "no write happened this time");
        }
        other => return Err(format!("expected LockIo, got {other:?}").into()),
    }
    Ok(())
}