haematite 0.6.1

//! §11 differential codec corpus (⟨r4, M4⟩): fixtures written by the
//! production `BranchRefStore`/`SnapshotRegistry` plus malformed variants —
//! the vacuum's read-only readers and the mutating stores accept/reject
//! IDENTICALLY, asserted case-by-case so a future fork of the code path
//! fails this test. (They already call the same shared codec functions by
//! construction; this pins the equivalence behaviourally.)

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

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

use super::metadata::{read_refs_dir, read_snapshot_registry};

/// Copy a fixture dir so the PRODUCTION open (which unlinks temp debris)
/// cannot contaminate the vacuum reader's view of the same fixture.
fn copy_dir(from: &Path, to: &Path) -> Result<(), Box<dyn Error>> {
    fs::create_dir_all(to)?;
    for entry in fs::read_dir(from)? {
        let entry = entry?;
        fs::copy(entry.path(), to.join(entry.file_name()))?;
    }
    Ok(())
}

/// Outcome digest: accept/reject plus the exact decoded record set.
fn production_refs_outcome(dir: &Path) -> Result<String, String> {
    BranchRefStore::open(dir)
        .map(|store| format!("{:?}", store.list().collect::<Vec<_>>()))
        .map_err(|_error| "reject".to_owned())
}

fn vacuum_refs_outcome(dir: &Path) -> Result<String, String> {
    read_refs_dir(dir)
        .map(|scan| format!("{:?}", scan.records.iter().collect::<Vec<_>>()))
        .map_err(|_error| "reject".to_owned())
}

/// Run both readers over copies of one fixture and require the same verdict
/// and, on acceptance, the same records.
fn assert_refs_equivalent(fixture: &Path, case: &str) -> Result<(), Box<dyn Error>> {
    let (_production, _vacuum) = refs_outcomes(fixture, case)?;
    Ok(())
}

/// Equivalence AND rejection: for structurally malformed fixtures the claim
/// is not just "same verdict" but "both refuse" — an equivalence-only
/// assertion would let a future permissive codec pass (Sol r2).
fn assert_refs_both_reject(fixture: &Path, case: &str) -> Result<(), Box<dyn Error>> {
    let (production, vacuum) = refs_outcomes(fixture, case)?;
    assert!(production.is_err(), "case {case}: production must reject");
    assert!(vacuum.is_err(), "case {case}: the vacuum must reject");
    Ok(())
}

type Outcome = Result<String, String>;

fn refs_outcomes(fixture: &Path, case: &str) -> Result<(Outcome, Outcome), Box<dyn Error>> {
    let production_copy = tempfile::tempdir()?;
    let vacuum_copy = tempfile::tempdir()?;
    copy_dir(fixture, production_copy.path())?;
    copy_dir(fixture, vacuum_copy.path())?;
    let production = production_refs_outcome(production_copy.path());
    let vacuum = vacuum_refs_outcome(vacuum_copy.path());
    assert_eq!(
        production, vacuum,
        "case {case}: the two readers must accept/reject identically"
    );
    Ok((production, vacuum))
}

fn valid_refs_fixture(dir: &Path) -> Result<(), Box<dyn Error>> {
    let mut store = BranchRefStore::open(dir)?;
    store.create(BranchRefRecord {
        name: "fixture".to_owned(),
        created: 7,
        kind: crate::branch::BranchKind::Work,
        namespace_lineage: None,
        seq: 3,
        timestamp: 11,
        shards: vec![BranchShardRef {
            shard_id: 1,
            fork_anchor: Hash::from_bytes([0x11; 32]),
            head: Hash::from_bytes([0x22; 32]),
        }],
        parents: vec![Hash::from_bytes([0x33; 32])],
    })?;
    Ok(())
}

fn sole_ref_file(dir: &Path) -> Result<std::path::PathBuf, Box<dyn Error>> {
    fs::read_dir(dir)?
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .find(|path| path.extension().and_then(|ext| ext.to_str()) == Some("ref"))
        .ok_or_else(|| "fixture holds one .ref file".into())
}

#[test]
fn hbr1_corpus_accepts_and_rejects_identically() -> Result<(), Box<dyn Error>> {
    // Valid record: both accept, same records.
    let valid = tempfile::tempdir()?;
    valid_refs_fixture(valid.path())?;
    assert_refs_equivalent(valid.path(), "valid")?;

    // Truncated bytes: both reject.
    let truncated = tempfile::tempdir()?;
    valid_refs_fixture(truncated.path())?;
    let ref_file = sole_ref_file(truncated.path())?;
    let bytes = fs::read(&ref_file)?;
    fs::write(&ref_file, &bytes[..bytes.len() - 5])?;
    assert_refs_equivalent(truncated.path(), "truncated")?;

    // Trailing bytes: both reject.
    let trailing = tempfile::tempdir()?;
    valid_refs_fixture(trailing.path())?;
    let ref_file = sole_ref_file(trailing.path())?;
    let mut bytes = fs::read(&ref_file)?;
    bytes.extend_from_slice(b"xyz");
    fs::write(&ref_file, bytes)?;
    assert_refs_equivalent(trailing.path(), "trailing")?;

    // Corrupt magic: both reject.
    let magic = tempfile::tempdir()?;
    valid_refs_fixture(magic.path())?;
    let ref_file = sole_ref_file(magic.path())?;
    let mut bytes = fs::read(&ref_file)?;
    bytes[0] = b'X';
    fs::write(&ref_file, bytes)?;
    assert_refs_equivalent(magic.path(), "magic")?;

    // Misfiled: a valid record under a file name its branch name does not
    // hash to — both reject (§2.2: a silently dropped record is a silently
    // dropped pin).
    let misfiled = tempfile::tempdir()?;
    valid_refs_fixture(misfiled.path())?;
    let ref_file = sole_ref_file(misfiled.path())?;
    fs::rename(
        &ref_file,
        misfiled.path().join("00000000000000000000000000000000.ref"),
    )?;
    assert_refs_equivalent(misfiled.path(), "misfiled")?;

    // Foreign non-.ref entry: both ignore it (the vacuum counts it; the
    // record sets stay identical).
    let foreign = tempfile::tempdir()?;
    valid_refs_fixture(foreign.path())?;
    fs::write(foreign.path().join("README.txt"), b"not a record")?;
    assert_refs_equivalent(foreign.path(), "foreign")?;

    // Temp debris: production unlinks it, the vacuum counts it — the
    // decoded record sets are identical either way.
    let debris = tempfile::tempdir()?;
    valid_refs_fixture(debris.path())?;
    fs::write(debris.path().join(".branch-crash.tmp"), b"leftover")?;
    assert_refs_equivalent(debris.path(), "debris")?;

    // Corrupt STRUCTURAL byte — the name-length prefix becomes impossible,
    // so decode must fail: both reject, asserted as rejection.
    let corrupt_length = tempfile::tempdir()?;
    valid_refs_fixture(corrupt_length.path())?;
    let ref_file = sole_ref_file(corrupt_length.path())?;
    let mut bytes = fs::read(&ref_file)?;
    bytes[10] = 0xFF; // name length (u64 LE at offset 4) now ≫ remaining
    fs::write(&ref_file, bytes)?;
    assert_refs_both_reject(corrupt_length.path(), "corrupt-length-prefix")?;

    // Arbitrary PAYLOAD byte mutation (here: inside a fixed-width hash) is
    // UNDETECTABLE by a checksumless codec — that is a property of the
    // format, stated honestly: the pin is that both readers accept the SAME
    // mutated record, so a divergence (one adding validation the other
    // lacks) still fails.
    let mutated = tempfile::tempdir()?;
    valid_refs_fixture(mutated.path())?;
    let ref_file = sole_ref_file(mutated.path())?;
    let mut bytes = fs::read(&ref_file)?;
    let middle = bytes.len() / 2;
    bytes[middle] ^= 0xFF;
    fs::write(&ref_file, bytes)?;
    assert_refs_equivalent(mutated.path(), "payload-mutation-equivalent")?;

    // Huge SHARD COUNT (u64::MAX): must fail as corrupt when the bytes run
    // out — never abort at the allocator (the PERSIST-003 discipline in the
    // shared codec; this exact fixture aborted the process before the
    // remaining-bytes capacity cap). Hand-encoded.
    let huge_count = tempfile::tempdir()?;
    let mut bytes = Vec::new();
    bytes.extend_from_slice(b"HBR1");
    crate::branch::persist::push_bytes(&mut bytes, b"x");
    crate::branch::persist::push_u64(&mut bytes, 1); // created
    crate::branch::persist::push_u64(&mut bytes, 1); // seq
    crate::branch::persist::push_u64(&mut bytes, 1); // timestamp
    crate::branch::persist::push_u64(&mut bytes, u64::MAX); // shard count
    fs::write(
        huge_count
            .path()
            .join("00000000000000000000000000000000.ref"),
        bytes,
    )?;
    assert_refs_both_reject(huge_count.path(), "huge-shard-count")?;

    // Non-UTF8 branch NAME inside an otherwise well-formed record: both
    // reject at the UTF-8 check. Hand-encoded — `encode_record` cannot
    // produce this shape. (A non-UTF8 FILENAME cannot be created on APFS,
    // so the name-in-record variant carries that §11 case.)
    let non_utf8 = tempfile::tempdir()?;
    let mut bytes = Vec::new();
    bytes.extend_from_slice(b"HBR1");
    crate::branch::persist::push_bytes(&mut bytes, &[0xFF, 0xFE, 0xFD]);
    crate::branch::persist::push_u64(&mut bytes, 1); // created
    crate::branch::persist::push_u64(&mut bytes, 1); // seq
    crate::branch::persist::push_u64(&mut bytes, 1); // timestamp
    crate::branch::persist::push_u64(&mut bytes, 0); // shards
    crate::branch::persist::push_u64(&mut bytes, 0); // parents
    fs::write(
        non_utf8.path().join("00000000000000000000000000000000.ref"),
        bytes,
    )?;
    assert_refs_equivalent(non_utf8.path(), "non-utf8-name")?;

    Ok(())
}

fn production_registry_outcome(path: &Path) -> Result<String, String> {
    SnapshotRegistry::open(path)
        .map(|registry| format!("{:?}", registry.list_snapshots()))
        .map_err(|_error| "reject".to_owned())
}

fn vacuum_registry_outcome(path: &Path) -> Result<String, String> {
    read_snapshot_registry(path)
        .map(|entries| {
            format!(
                "{:?}",
                entries
                    .iter()
                    .map(|entry| (entry.name.clone(), entry.root_hash, entry.timestamp))
                    .collect::<Vec<_>>()
            )
        })
        .map_err(|_error| "reject".to_owned())
}

fn assert_registry_equivalent(path: &Path, case: &str) {
    assert_eq!(
        production_registry_outcome(path),
        vacuum_registry_outcome(path),
        "case {case}: the two readers must accept/reject identically"
    );
}

fn assert_registry_both_reject(path: &Path, case: &str) {
    let production = production_registry_outcome(path);
    let vacuum = vacuum_registry_outcome(path);
    assert_eq!(
        production, vacuum,
        "case {case}: the two readers must accept/reject identically"
    );
    assert!(production.is_err(), "case {case}: production must reject");
    assert!(vacuum.is_err(), "case {case}: the vacuum must reject");
}

#[test]
fn hsr1_corpus_accepts_and_rejects_identically() -> Result<(), Box<dyn Error>> {
    let temp = tempfile::tempdir()?;
    let path = temp.path().join("snapshots.hsr");
    let mut registry = SnapshotRegistry::open(&path)?;
    registry.name_at("one", Hash::from_bytes([0x44; 32]), 5)?;
    registry.name_at("two", Hash::from_bytes([0x55; 32]), 6)?;
    drop(registry);
    let valid_bytes = fs::read(&path)?;

    assert_registry_equivalent(&path, "valid");

    fs::write(&path, &valid_bytes[..valid_bytes.len() - 3])?;
    assert_registry_equivalent(&path, "truncated");

    let mut trailing = valid_bytes.clone();
    trailing.extend_from_slice(b"zz");
    fs::write(&path, trailing)?;
    assert_registry_equivalent(&path, "trailing");

    let mut magic = valid_bytes.clone();
    magic[0] = b'X';
    fs::write(&path, magic)?;
    assert_registry_equivalent(&path, "magic");

    // Corrupt STRUCTURAL byte — the entry count becomes impossible: both
    // reject, asserted as rejection.
    let mut corrupt_count = valid_bytes.clone();
    corrupt_count[10] = 0xFF; // count (u64 LE at offset 4) now ≫ remaining
    fs::write(&path, corrupt_count)?;
    assert_registry_both_reject(&path, "corrupt-count-prefix");

    // Payload-byte mutation in fixed-width data is undetectable by a
    // checksumless codec (format property, stated honestly): the pin is
    // identical acceptance of the same mutated bytes.
    let mut mutated = valid_bytes;
    let middle = mutated.len() / 2;
    mutated[middle] ^= 0xFF;
    fs::write(&path, mutated)?;
    assert_registry_equivalent(&path, "payload-mutation-equivalent");

    // Wrong-length hash: an entry whose "hash" field is 20 bytes, so the
    // reader consumes into the next field and the record cannot finish
    // cleanly — both reject. Hand-encoded.
    let mut short_hash = Vec::new();
    short_hash.extend_from_slice(b"HSR1");
    crate::branch::persist::push_u64(&mut short_hash, 1); // count
    crate::branch::persist::push_bytes(&mut short_hash, b"stub");
    short_hash.extend_from_slice(&[0xAA; 20]); // 20 bytes where 32 belong
    crate::branch::persist::push_u64(&mut short_hash, 5); // timestamp
    fs::write(&path, short_hash)?;
    assert_registry_equivalent(&path, "wrong-length-hash");

    Ok(())
}