haematite 0.6.2

//! §11 refusal pins (⟨r3⟩): each test constructs the exact on-disk shape,
//! asserts the named `VacuumError` variant, and asserts the run changed
//! nothing — refusal is total, never per-store best-effort.

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

use crate::db::Database;

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

/// §2 M1: store present, `shard.wal` deleted ⇒ `StoreWithoutWal`, the text
/// treats a present store as forensic evidence, and nothing changes on disk.
#[test]
fn store_without_wal_is_forensic_refusal() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    fs::remove_file(temp.path().join("shard-0/shard.wal"))?;

    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(
            &error,
            VacuumError::StoreWithoutWal {
                shard_id: 0,
                store_present: true,
                ..
            }
        ),
        "got {error:?}"
    );
    let text = error.to_string();
    assert!(text.contains("forensic"), "present-store text names damage");
    assert_eq!(
        before,
        snapshot_tree(temp.path())?,
        "refusal changed nothing"
    );
    Ok(())
}

/// ⟨r15, A2⟩ the crash transient (dir present, no store, no WAL) refuses
/// with a remedy-naming text, the SAME dir wholly absent proceeds as lazy
/// (⟨r2, B3⟩ contrast half), and the named remedy is proven honest
/// end-to-end: reopen, retry the interrupted write, re-run ⇒ clean.
#[test]
fn crash_transient_names_remedy_and_remedy_works() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 4, &[b"solo"], 2)?;

    // Contrast half FIRST: with the untouched shard dirs wholly absent, the
    // run proceeds — the B3 boundary sits exactly where §2 draws it.
    stats(temp.path())?;

    // The crash transient: the router's create_dir_all ran, boot never did.
    let target = (0..4)
        .find(|shard_id| !temp.path().join(format!("shard-{shard_id}")).exists())
        .ok_or("one shard must be unmaterialised")?;
    fs::create_dir(temp.path().join(format!("shard-{target}")))?;

    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(
            &error,
            VacuumError::StoreWithoutWal {
                store_present: false,
                ..
            }
        ),
        "got {error:?}"
    );
    let text = error.to_string();
    assert!(
        text.contains("reopen") && text.contains("re-run"),
        "the refusal must NAME its remedy (fail-closed, not fail-stuck): {text}"
    );

    // The remedy, exactly as named: reopen, retry the interrupted operation.
    let db = Database::open(temp.path())?;
    let key = (0..10_000_u64)
        .map(|candidate| format!("cure:{candidate}").into_bytes())
        .find(|key| db.shard_for(key) == target)
        .ok_or("a key routing to the target shard must exist")?;
    db.append(key, vec![b"cure".to_vec()], 0)?;
    drop(db);

    stats(temp.path())?;
    Ok(())
}

/// §2 M1: a present-but-corrupt WAL fails closed (`CorruptWal`), consistent
/// with the recovery-hardening design.
#[test]
fn corrupt_wal_fails_closed() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    let wal_path = temp.path().join("shard-0/shard.wal");
    let mut bytes = fs::read(&wal_path)?;
    assert!(bytes.len() > 12, "the WAL holds at least one frame");
    // Flip a byte inside the first frame's payload: CRC now mismatches.
    bytes[8] ^= 0xFF;
    fs::write(&wal_path, bytes)?;

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

/// ⟨r15, C1⟩ absent and unreadable `config.json` both refuse
/// (`ConfigUnreadable`) — there is no shard iteration without a trusted
/// `shard_count`.
#[test]
fn config_unreadable_refuses_both_shapes() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let config_path = temp.path().join("config.json");

    // Unreadable: permission-denied.
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(&config_path, fs::Permissions::from_mode(0o000))?;
        let error = expect_refusal(stats(temp.path()))?;
        assert!(
            matches!(&error, VacuumError::ConfigUnreadable { .. }),
            "got {error:?}"
        );
        fs::set_permissions(&config_path, fs::Permissions::from_mode(0o644))?;
    }

    // Absent.
    fs::remove_file(&config_path)?;
    let before = snapshot_tree(temp.path())?;
    let error = expect_refusal(stats(temp.path()))?;
    assert!(
        matches!(&error, VacuumError::ConfigUnreadable { .. }),
        "got {error:?}"
    );
    let text = error.to_string();
    assert!(
        text.contains("restore") && text.contains("re-run"),
        "the refusal names what the operator must decide: {text}"
    );
    assert_eq!(
        before,
        snapshot_tree(temp.path())?,
        "refusal changed nothing"
    );
    Ok(())
}

/// ⟨r2, B2⟩ / §11: a canonically-named node file whose bytes are a DIFFERENT
/// valid node ⇒ `NodeHashMismatch` naming the file, and the mismatched file
/// still exists byte-identical after the failed run — corruption evidence is
/// preserved, never swept around.
#[test]
fn node_hash_mismatch_refuses_and_preserves_evidence() -> 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("shard 0 committed")?;
    let store = store_dir(temp.path(), 0);
    let root_path = node_path(&store, root);
    // Replace the root's bytes with ANOTHER stored node's bytes: valid zstd,
    // valid node, wrong name — the misplaced-copy shape.
    let donor = find_other_node_file(&store, &root_path)?;
    fs::write(&root_path, fs::read(donor)?)?;
    let planted = fs::read(&root_path)?;

    let error = expect_refusal(stats(temp.path()))?;
    match &error {
        VacuumError::NodeHashMismatch { path, .. } => {
            assert_eq!(path, &root_path, "the refusal names the file");
        }
        other => return Err(format!("expected NodeHashMismatch, got {other:?}").into()),
    }
    assert_eq!(
        fs::read(&root_path)?,
        planted,
        "the evidence file must survive byte-identical"
    );
    Ok(())
}

/// §4: `shard.wal` present with the store directory absent is damage —
/// `UnexpectedLayout`, never a silent empty inventory.
#[test]
fn wal_without_store_is_unexpected_layout() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    fs::remove_dir_all(store_dir(temp.path(), 0))?;

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

/// §4: malformed store entries (foreign files, symlinks, wrong-length
/// names) are reported, that store's sweep is blocked, and stats still
/// completes; `.node-*.tmp` is debris, not malformed.
#[test]
fn malformed_store_entries_reported_stats_completes() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 2)?;
    let store = store_dir(temp.path(), 0);

    fs::write(store.join("not-a-fanout-dir"), b"foreign")?;
    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::write(prefix_dir.join("tooshort"), b"bad name")?;
    fs::write(prefix_dir.join(".node-orphan.tmp"), b"install debris")?;
    #[cfg(unix)]
    std::os::unix::fs::symlink(store.join("not-a-fanout-dir"), prefix_dir.join("a-symlink"))?;

    let report = stats(temp.path())?;
    let shard = &report.shards[0];
    assert!(
        shard.malformed.len() >= 2,
        "foreign + bad-name (+ symlink) reported: {:?}",
        shard.malformed
    );
    assert_eq!(shard.temp_debris_files, 1, "debris counted separately");
    assert!(
        report.sweep_blockers.iter().any(|blocker| matches!(
            blocker,
            SweepBlocker::MalformedStoreEntries { shard_id: 0, .. }
        )),
        "malformed entries block that store's sweep"
    );
    // The reconciliation still holds: malformed entries are not nodes.
    assert_eq!(
        report.totals.enumerated_nodes,
        report.totals.marked_nodes + report.totals.unmarked_nodes
    );
    Ok(())
}

/// A foreign entry at the shard-directory level (outside the store) is
/// reported and blocks that store's sweep.
#[test]
fn foreign_shard_dir_entry_reported() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    fs::write(temp.path().join("shard-0/unexpected.bin"), b"???")?;

    let report = stats(temp.path())?;
    assert!(
        report.shards[0]
            .malformed
            .iter()
            .any(|entry| entry.path.ends_with("unexpected.bin")),
        "shard-level foreign entry must be reported"
    );
    Ok(())
}

/// The A4 exclusion: a live writer holds the lock ⇒ `Locked`, never queued.
#[test]
fn live_writer_lock_refuses() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    build_store(temp.path(), 1, &[b"data"], 1)?;
    let db = Database::open(temp.path())?;

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

/// Find a node file in `store` other than `excluded`.
fn find_other_node_file(
    store: &std::path::Path,
    excluded: &std::path::Path,
) -> Result<std::path::PathBuf, Box<dyn Error>> {
    for prefix in fs::read_dir(store)? {
        let prefix = prefix?.path();
        if !prefix.is_dir() {
            continue;
        }
        for entry in fs::read_dir(&prefix)? {
            let path = entry?.path();
            if path != excluded && path.is_file() {
                return Ok(path);
            }
        }
    }
    Err("store must hold more than one node".into())
}