verit 0.2.0

//! The `.verit` file format — conformance for
//! `docs/Architecture/VERIT - File Format Specification.md`.
//!
//! Four obligations, in order of how much they matter:
//!
//! 1. **Self-contained** (§5): a file plus nothing else is fully readable.
//! 2. **Crash-safe** (§7): a commit torn at *any* byte rolls back cleanly.
//! 3. **Honest about deletion** (§8.2): removal unlinks, and the bytes are
//!    provably still there until compaction — asserted, not just documented.
//! 4. **Safe on hostile bytes** (§10): typed errors, never a panic.

use std::path::PathBuf;

use verit::{
    encode, Dt, Error, FileBuilder, FileView, FileWriter, Message, Scalars, Schema, SchemaBuilder,
    SchemaMode, Value,
};

// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------

fn point_schema() -> Schema {
    SchemaBuilder::new()
        .add_struct("Point", vec![(1, "x", Dt::I32), (2, "y", Dt::I32)])
        .build("Point")
        .unwrap()
}

fn person_schema() -> Schema {
    SchemaBuilder::new()
        .add_struct("Person", vec![(1, "name", Dt::Str), (2, "age", Dt::U8)])
        .build("Person")
        .unwrap()
}

fn point(x: i32, y: i32) -> Value {
    Value::Struct(vec![(1, Value::I32(x)), (2, Value::I32(y))])
}

fn person(name: &str, age: u8) -> Value {
    Value::Struct(vec![(1, Value::str(name)), (2, Value::U8(age))])
}

/// A unique scratch path. No `tempfile` dependency — the library is zero-dep
/// and its tests keep the same discipline.
fn tmp(name: &str) -> PathBuf {
    let mut p = std::env::temp_dir();
    p.push(format!(
        "verit-file-test-{}-{}.verit",
        std::process::id(),
        name
    ));
    let _ = std::fs::remove_file(&p);
    p
}

struct Scratch(PathBuf);

impl Drop for Scratch {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.0);
        let mut compact = self.0.clone().into_os_string();
        compact.push(".compact");
        let _ = std::fs::remove_file(PathBuf::from(compact));
    }
}

// ---------------------------------------------------------------------------
// 1. Self-contained (§5)
// ---------------------------------------------------------------------------

#[test]
fn file_is_readable_from_its_own_bytes_alone() {
    let schema = person_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &person("Ada", 36)).unwrap();
    b.append(&schema, &person("Grace", 45)).unwrap();
    let bytes = b.finish().unwrap();

    // From here on, pretend we know nothing but `bytes` — no schema, no
    // registry. This is the whole promise of the format.
    let f = FileView::open(&bytes).unwrap();
    assert_eq!(f.len(), 2);
    assert_eq!(f.generation(), 1);
    assert_eq!(f.dump_json(0).unwrap(), r#"{"name":"Ada","age":36}"#);
    assert_eq!(f.dump_json(1).unwrap(), r#"{"name":"Grace","age":45}"#);
    assert_eq!(f.schemas().len(), 1);
}

#[test]
fn schema_is_stored_once_however_many_records_use_it() {
    // A wide schema, so its canonical bytes are large enough that duplicating
    // them per record would visibly dominate the file.
    let fields: Vec<(u16, &str, Dt)> = (1..=24u16)
        .map(|i| (i, FIELD_NAMES[i as usize - 1], Dt::U32))
        .collect();
    let schema = SchemaBuilder::new()
        .add_struct("Wide", fields)
        .build("Wide")
        .unwrap();
    let canonical = schema.canonical_bytes().len();
    let value = Value::Struct((1..=24u16).map(|i| (i, Value::U32(i as u32))).collect());

    const N: usize = 64;

    // Hash-only records: the schema section carries the schema once.
    let mut lean = FileBuilder::new();
    for _ in 0..N {
        lean.append(&schema, &value).unwrap();
    }
    let lean = lean.finish().unwrap();

    // Inline-schema records: the 0.1.0 way to be self-contained, and the thing
    // the schema section exists to avoid.
    let inline = encode(&schema, &value, SchemaMode::Inline).unwrap();
    let mut fat = FileBuilder::new();
    for _ in 0..N {
        fat.append_self_describing(&inline).unwrap();
    }
    let fat = fat.finish().unwrap();

    // Both are self-contained and read identically; only the size differs.
    assert_eq!(FileView::open(&lean).unwrap().schemas().len(), 1);
    assert_eq!(FileView::open(&fat).unwrap().schemas().len(), 1);
    assert_eq!(FileView::open(&lean).unwrap().len(), N);

    let saved = fat.len() - lean.len();
    let expected = (N - 1) * canonical;
    assert!(
        saved >= expected,
        "storing the schema once saved {saved} B; duplicating it {N}× should have \
         cost at least {expected} B ({canonical} B per canonical schema)"
    );
}

const FIELD_NAMES: [&str; 24] = [
    "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india", "juliett",
    "kilo", "lima", "mike", "november", "oscar", "papa", "quebec", "romeo", "sierra", "tango",
    "uniform", "victor", "whiskey", "xray",
];

#[test]
fn one_file_holds_records_of_different_schemas() {
    let points = point_schema();
    let people = person_schema();

    let mut b = FileBuilder::new();
    b.append(&points, &point(3, 4)).unwrap();
    b.append(&people, &person("Ada", 36)).unwrap();
    b.append(&points, &point(-1, 0)).unwrap();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    assert_eq!(f.schemas().len(), 2);
    assert_eq!(f.dump_json(0).unwrap(), r#"{"x":3,"y":4}"#);
    assert_eq!(f.dump_json(1).unwrap(), r#"{"name":"Ada","age":36}"#);
    assert_eq!(f.dump_json(2).unwrap(), r#"{"x":-1,"y":0}"#);

    // Schema ids are readable from the index without touching record bytes.
    assert_eq!(f.schema_id(0).unwrap(), points.id());
    assert_eq!(f.schema_id(1).unwrap(), people.id());
    assert_eq!(f.schema_id(2).unwrap(), points.id());
}

#[test]
fn old_records_resolve_into_todays_reader_schema() {
    // A file written under v1, still holding v1 records, read by a v2 reader.
    let v1 = SchemaBuilder::new()
        .add_struct(
            "Event",
            vec![(1, "service", Dt::Str), (4, "latency", Dt::U16)],
        )
        .build("Event")
        .unwrap();
    let v2 = SchemaBuilder::new()
        .add_struct(
            "Event",
            vec![
                (1, "service", Dt::Str),
                (3, "message", Dt::Str), // added
                (4, "latency", Dt::U32), // widened
            ],
        )
        .build("Event")
        .unwrap();

    let mut b = FileBuilder::new();
    b.append(
        &v1,
        &Value::Struct(vec![(1, Value::str("auth")), (4, Value::U16(150))]),
    )
    .unwrap();
    let bytes = b.finish().unwrap();

    // The file kept the writer schema, so evolution works at rest with no
    // registry and no sidecar.
    let f = FileView::open(&bytes).unwrap();
    let resolver = f.resolver_for(0, &v2).unwrap();
    let msg = f.message(0).unwrap();
    let root = msg.root(&resolver).unwrap();
    assert_eq!(root.get_str(1).unwrap(), Some("auth"));
    assert_eq!(root.get_u32(4).unwrap(), Some(150)); // widened on load
    assert_eq!(root.get_str(3).unwrap(), None); // added field: absent
}

#[test]
fn empty_file_is_valid() {
    let bytes = FileBuilder::new().finish().unwrap();
    let f = FileView::open(&bytes).unwrap();
    assert_eq!(f.len(), 0);
    assert!(f.is_empty());
    assert_eq!(f.generation(), 1);
    assert_eq!(f.file_len(), bytes.len() as u64);
}

#[test]
fn every_record_is_eight_byte_aligned() {
    let schema = person_schema();
    let mut b = FileBuilder::new();
    for n in 0..16 {
        // Deliberately odd, varying lengths to force padding between records.
        b.append(&schema, &person(&"x".repeat(n), n as u8)).unwrap();
    }
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    for r in f.records() {
        assert_eq!(r.offset % 8, 0, "record at {} is not 8-aligned", r.offset);
    }
    assert_eq!(bytes.len() % 8, 0, "file length must be a multiple of 8");
}

#[test]
fn builder_output_is_deterministic() {
    let schema = point_schema();
    let build = || {
        let mut b = FileBuilder::new();
        b.append(&schema, &point(1, 2)).unwrap();
        b.append(&schema, &point(3, 4)).unwrap();
        b.finish().unwrap()
    };
    assert_eq!(build(), build());
}

#[test]
fn accepts_self_describing_messages_and_lifts_their_schema() {
    let schema = person_schema();
    let inline = encode(&schema, &person("Ada", 36), SchemaMode::Inline).unwrap();

    let mut b = FileBuilder::new();
    b.append_self_describing(&inline).unwrap();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    assert_eq!(f.dump_json(0).unwrap(), r#"{"name":"Ada","age":36}"#);
    assert_eq!(f.schema_id(0).unwrap(), schema.id());
}

// ---------------------------------------------------------------------------
// 2. Crash safety (§7)
// ---------------------------------------------------------------------------

#[test]
fn append_commits_bump_the_generation() {
    let path = tmp("generations");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    assert_eq!(w.generation(), 1); // create commits an empty generation 1
    assert!(w.is_empty());

    w.append(&schema, &point(1, 2)).unwrap();
    assert_eq!(w.pending(), 1);
    assert_eq!(w.commit().unwrap(), 2);
    assert_eq!(w.pending(), 0);

    w.append(&schema, &point(3, 4)).unwrap();
    w.append(&schema, &point(5, 6)).unwrap();
    assert_eq!(w.commit().unwrap(), 3); // one footer commits both

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.len(), 3);
    assert_eq!(f.generation(), 3);
    assert_eq!(f.dump_json(2).unwrap(), r#"{"x":5,"y":6}"#);
}

#[test]
fn uncommitted_appends_are_invisible() {
    let path = tmp("uncommitted");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &point(1, 2)).unwrap();
    w.commit().unwrap();

    w.append(&schema, &point(9, 9)).unwrap(); // staged, never committed
    assert_eq!(w.len(), 2);

    let f_image = std::fs::read(&path).unwrap();
    let f = FileView::open(&f_image).unwrap();
    assert_eq!(f.len(), 1, "a staged record must not be visible on disk");
}

#[test]
fn a_commit_torn_at_any_byte_rolls_back_to_the_previous_generation() {
    let path = tmp("torn");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    // Generation 2: the state a crash must preserve.
    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &point(1, 1)).unwrap();
    w.append(&schema, &point(2, 2)).unwrap();
    w.commit().unwrap();
    let good = std::fs::read(&path).unwrap();
    let good_len = good.len();
    let good_view = FileView::open(&good).unwrap();
    assert_eq!((good_view.len(), good_view.generation()), (2, 2));

    // Generation 3: a large append, so the torn window covers records, the
    // schema section, the index, and the footer.
    for n in 0..8 {
        w.append(&schema, &point(100 + n, -n)).unwrap();
    }
    w.commit().unwrap();
    let torn_source = std::fs::read(&path).unwrap();
    assert_eq!(FileView::open(&torn_source).unwrap().generation(), 3);

    // Truncate at EVERY byte offset of that commit. Each prefix is a possible
    // crash point; every one must open cleanly at generation 2.
    for cut in good_len..torn_source.len() {
        let view = FileView::open(&torn_source[..cut]).unwrap_or_else(|e| {
            panic!(
                "truncation at {cut} of {} left an unopenable file: {e}",
                torn_source.len()
            )
        });
        assert_eq!(
            (view.len(), view.generation()),
            (2, 2),
            "truncation at {cut} did not roll back to generation 2"
        );
        // The rolled-back view is not merely structurally valid — it still
        // reads its records correctly.
        assert_eq!(view.dump_json(0).unwrap(), r#"{"x":1,"y":1}"#);
        assert_eq!(view.dump_json(1).unwrap(), r#"{"x":2,"y":2}"#);
    }

    // And the untruncated file is generation 3, with everything.
    let full = FileView::open(&torn_source).unwrap();
    assert_eq!((full.len(), full.generation()), (10, 3));
}

#[test]
fn a_writer_reopening_a_torn_file_recovers_and_moves_on() {
    let path = tmp("recover");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &point(1, 1)).unwrap();
    w.commit().unwrap();
    let committed_len = std::fs::read(&path).unwrap().len();

    // Simulate a crash mid-commit: garbage appended past the committed extent.
    {
        use std::io::Write;
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        f.write_all(&[0xAB; 137]).unwrap();
    }

    // Reopening recovers generation 2 and ignores the debris.
    let mut w = FileWriter::open(&path).unwrap();
    assert_eq!(w.generation(), 2);
    assert_eq!(w.len(), 1);

    // The next commit truncates the debris away rather than building on it.
    w.append(&schema, &point(2, 2)).unwrap();
    w.commit().unwrap();
    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!((f.len(), f.generation()), (2, 3));
    assert_eq!(f.file_len(), image.len() as u64, "debris was not truncated");
    assert!(image.len() < committed_len + 137 + 1024);
}

#[test]
fn committed_record_bytes_are_never_rewritten() {
    // The property everything else rests on: an append must not move or alter
    // an existing record, which is why concurrent readers need no lock.
    let path = tmp("immutable");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &point(7, 7)).unwrap();
    w.commit().unwrap();
    let before = std::fs::read(&path).unwrap();
    let first = FileView::open(&before).unwrap().record(0).unwrap();

    w.append(&schema, &point(8, 8)).unwrap();
    w.commit().unwrap();
    let after = std::fs::read(&path).unwrap();

    let end = (first.offset + first.length) as usize;
    assert_eq!(
        &before[first.offset as usize..end],
        &after[first.offset as usize..end],
        "an append rewrote committed record bytes"
    );
    assert_eq!(FileView::open(&after).unwrap().record(0).unwrap(), first);
}

// ---------------------------------------------------------------------------
// 3. Removal unlinks; compaction erases (§8)
// ---------------------------------------------------------------------------

#[test]
fn remove_unlinks_but_the_bytes_remain_until_compaction() {
    let path = tmp("remove");
    let _s = Scratch(path.clone());
    let schema = person_schema();
    let secret = "correct-horse-battery-staple";

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &person("public", 1)).unwrap();
    let secret_id = w.append(&schema, &person(secret, 2)).unwrap();
    w.commit().unwrap();

    w.remove_id(secret_id).unwrap();
    w.commit().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.len(), 1);
    assert_eq!(f.dump_json(0).unwrap(), r#"{"name":"public","age":1}"#);

    // The documented, deliberately-uncomfortable part: the removed record is
    // gone from the index but still sitting in the file. Spec §8.2 requires
    // this to be stated, so it is also asserted.
    assert!(
        image.windows(secret.len()).any(|w| w == secret.as_bytes()),
        "removal unexpectedly erased data — spec §8.2 says it must not"
    );

    // Compaction is what actually erases it.
    w.compact().unwrap();
    let compacted = std::fs::read(&path).unwrap();
    assert!(
        !compacted
            .windows(secret.len())
            .any(|w| w == secret.as_bytes()),
        "compaction failed to erase a removed record"
    );

    let f = FileView::open(&compacted).unwrap();
    assert_eq!(f.len(), 1);
    assert_eq!(f.generation(), 1, "compaction resets to generation 1");
    assert_eq!(f.dump_json(0).unwrap(), r#"{"name":"public","age":1}"#);
}

#[test]
fn compaction_reclaims_space_and_drops_unreferenced_schemas() {
    let path = tmp("compact");
    let _s = Scratch(path.clone());
    let points = point_schema();
    let people = person_schema();

    let mut w = FileWriter::create(&path).unwrap();
    for i in 0..32 {
        w.append(&points, &point(i, i)).unwrap();
    }
    w.append(&people, &person("Ada", 36)).unwrap();
    w.commit().unwrap();
    let fat = std::fs::read(&path).unwrap().len();

    // Remove every Point, leaving only the one Person — by predicate, so there
    // is no shifting-position loop to get wrong.
    let removed = w.retain(|_, schema_id| schema_id == people.id());
    assert_eq!(removed, 32);
    w.commit().unwrap();
    w.compact().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.len(), 1);
    assert_eq!(f.dump_json(0).unwrap(), r#"{"name":"Ada","age":36}"#);
    assert!(image.len() < fat / 2, "compaction reclaimed too little");
    assert_eq!(
        f.schemas().len(),
        1,
        "the Point schema had no live records and should have dropped out"
    );
    assert!(f.schemas().contains(people.id()));
    assert!(!f.schemas().contains(points.id()));
}

#[test]
fn a_compacted_file_matches_the_reference_builder_byte_for_byte() {
    // Spec §11.5: the on-disk writer and the in-memory reference writer must
    // not be able to drift, so a generation-1 file has exactly one encoding.
    let path = tmp("equivalence");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &point(1, 2)).unwrap();
    w.append(&schema, &point(3, 4)).unwrap();
    w.commit().unwrap();
    w.compact().unwrap();
    let on_disk = std::fs::read(&path).unwrap();

    let mut b = FileBuilder::new();
    b.append(&schema, &point(1, 2)).unwrap();
    b.append(&schema, &point(3, 4)).unwrap();
    let in_memory = b.finish().unwrap();

    assert_eq!(on_disk, in_memory);
}

// ---------------------------------------------------------------------------
// 4. Hostile and malformed input (§10)
// ---------------------------------------------------------------------------

#[test]
#[allow(deprecated)] // the point of the test is that `.vertc` is not `.verit`
fn rejects_a_vertc_container_structurally() {
    let mut c = verit::ContainerWriter::new();
    c.add(b"\x00message-a");
    let container = c.finish();
    assert!(
        matches!(FileView::open(&container), Err(Error::BadFile(_))),
        "a 0.1.0 .vertc container must be refused, not partially misread"
    );
}

#[test]
fn rejects_bad_magic_version_and_reserved_fields() {
    let bytes = FileBuilder::new().finish().unwrap();

    let mut bad_magic = bytes.clone();
    bad_magic[0] = b'X';
    assert!(matches!(FileView::open(&bad_magic), Err(Error::BadFile(_))));

    let mut bad_version = bytes.clone();
    bad_version[4] = 2;
    assert!(matches!(
        FileView::open(&bad_version),
        Err(Error::BadFile(_))
    ));

    let mut bad_reserved = bytes.clone();
    bad_reserved[5] = 1;
    assert!(matches!(
        FileView::open(&bad_reserved),
        Err(Error::BadFile(_))
    ));
}

#[test]
fn rejects_unknown_required_features_and_ignores_unknown_optional_ones() {
    let schema = point_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &point(1, 2)).unwrap();
    let bytes = b.finish().unwrap();

    // A required bit this build does not implement is fatal, and names itself.
    let mut required = bytes.clone();
    required[8..12].copy_from_slice(&0x0000_0004u32.to_le_bytes());
    assert!(matches!(
        FileView::open(&required),
        Err(Error::UnsupportedFileFeature(4))
    ));

    // An *unknown* optional bit is ignored — this is what lets the format grow
    // without stranding readers built today. Bit 0 is deliberately cleared:
    // it is `OPT_RECORD_CRC`, which this build knows, and a known optional bit
    // that adds a region is only ignorable by a reader that does *not*
    // implement it. One that does must validate the region, or it would read
    // checksums out of bytes that hold something else.
    let mut optional = bytes.clone();
    optional[12..16].copy_from_slice(&0xDEAD_BEEEu32.to_le_bytes());
    let f = FileView::open(&optional).unwrap();
    assert_eq!(f.dump_json(0).unwrap(), r#"{"x":1,"y":2}"#);

    // And a file that *claims* checksums without room for the array is
    // malformed, not silently accepted.
    let mut lying = bytes.clone();
    lying[12..16].copy_from_slice(&verit::OPT_RECORD_CRC.to_le_bytes());
    assert!(matches!(FileView::open(&lying), Err(Error::BadFile(_))));
}

#[test]
fn rejects_a_file_whose_index_references_a_missing_schema() {
    let schema = point_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &point(1, 2)).unwrap();
    let mut bytes = b.finish().unwrap();

    // Flip a bit in the index entry's schema id: the section no longer covers
    // it, so the file is not self-contained and must be refused.
    let f = FileView::open(&bytes).unwrap();
    let entry = f.footer().index_offset as usize;
    drop(f);
    bytes[entry + 24] ^= 0x01;
    assert!(matches!(
        FileView::open(&bytes),
        Err(Error::MissingSchema(_))
    ));
}

#[test]
fn rejects_out_of_bounds_and_misaligned_index_entries() {
    let schema = point_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &point(1, 2)).unwrap();
    let bytes = b.finish().unwrap();
    let entry = FileView::open(&bytes).unwrap().footer().index_offset as usize;

    let mut past_end = bytes.clone();
    past_end[entry..entry + 8].copy_from_slice(&u64::MAX.to_le_bytes());
    assert!(matches!(FileView::open(&past_end), Err(Error::BadFile(_))));

    let mut misaligned = bytes.clone();
    misaligned[entry..entry + 8].copy_from_slice(&33u64.to_le_bytes());
    assert!(matches!(
        FileView::open(&misaligned),
        Err(Error::BadFile(_))
    ));

    let mut into_header = bytes.clone();
    into_header[entry..entry + 8].copy_from_slice(&0u64.to_le_bytes());
    assert!(matches!(
        FileView::open(&into_header),
        Err(Error::BadFile(_))
    ));

    let mut huge_len = bytes.clone();
    huge_len[entry + 8..entry + 16].copy_from_slice(&u64::MAX.to_le_bytes());
    assert!(matches!(FileView::open(&huge_len), Err(Error::BadFile(_))));
}

#[test]
fn a_forged_record_count_fails_on_arithmetic_not_on_allocation() {
    // `record_count` is attacker-controlled. A file claiming four billion
    // records must be refused by the geometry checks, without ever allocating.
    let schema = point_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &point(1, 2)).unwrap();
    let mut bytes = b.finish().unwrap();

    let footer = bytes.len() - 64;
    bytes[footer + 28..footer + 32].copy_from_slice(&u32::MAX.to_le_bytes());
    // Repair the CRC so the forgery gets past the commit check and is caught by
    // the structural rules, which is the path that actually matters.
    let crc = verit::hash::crc32(&bytes[footer..footer + 56]);
    bytes[footer + 56..footer + 60].copy_from_slice(&crc.to_le_bytes());

    assert!(matches!(FileView::open(&bytes), Err(Error::BadFile(_))));
}

#[test]
fn rejects_truncation_below_any_valid_commit() {
    let bytes = FileBuilder::new().finish().unwrap();
    // Nothing recoverable is left once the only footer is gone.
    assert!(FileView::open(&bytes[..bytes.len() - 8]).is_err());
    assert!(FileView::open(&bytes[..4]).is_err());
    assert!(FileView::open(&[]).is_err());
}

#[test]
fn never_panics_on_arbitrary_mutations() {
    // A cheap deterministic sweep — the real fuzz target lives in `fuzz/`.
    let schema = person_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &person("Ada", 36)).unwrap();
    b.append(&schema, &person("Grace", 45)).unwrap();
    let bytes = b.finish().unwrap();

    for i in 0..bytes.len() {
        for patch in [0x00u8, 0xFF, 0x80] {
            let mut mutated = bytes.clone();
            mutated[i] = patch;
            // Whatever it decides, it must decide it without unwinding.
            let result = std::panic::catch_unwind(|| {
                if let Ok(f) = FileView::open(&mutated) {
                    for i in 0..f.len() {
                        let _ = f.get(i);
                        let _ = f.dump_json(i);
                    }
                }
            });
            assert!(
                result.is_ok(),
                "panicked on byte {i} patched to {patch:#04x}"
            );
        }
    }
}

#[test]
fn record_bytes_survive_the_round_trip_verbatim() {
    let schema = person_schema();
    let raw = encode(&schema, &person("Ada", 36), SchemaMode::HashOnly).unwrap();

    let mut b = FileBuilder::new();
    b.append_message(&schema, &raw).unwrap();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    assert_eq!(f.get(0).unwrap(), &raw[..]);
    // And the slice is directly parseable — no copy, no fixup.
    assert_eq!(
        Message::parse(f.get(0).unwrap()).unwrap().schema_id(),
        schema.id()
    );
}

#[test]
fn add_message_refuses_a_mismatched_schema() {
    let points = point_schema();
    let people = person_schema();
    let raw = encode(&points, &point(1, 2), SchemaMode::HashOnly).unwrap();

    let mut b = FileBuilder::new();
    assert!(matches!(
        b.append_message(&people, &raw),
        Err(Error::SchemaIdMismatch { .. })
    ));
}

#[test]
fn out_of_range_access_is_a_typed_error() {
    let bytes = FileBuilder::new().finish().unwrap();
    let f = FileView::open(&bytes).unwrap();
    assert!(matches!(f.get(0), Err(Error::IndexOutOfBounds)));
    assert!(matches!(f.record(7), Err(Error::IndexOutOfBounds)));

    let path = tmp("oob");
    let _s = Scratch(path.clone());
    let mut w = FileWriter::create(&path).unwrap();
    assert!(matches!(w.remove(0), Err(Error::IndexOutOfBounds)));
    assert!(matches!(w.remove_id(1), Err(Error::IndexOutOfBounds)));
}

// ---------------------------------------------------------------------------
// 5. Stable record identity (§6, §8.1)
// ---------------------------------------------------------------------------

#[test]
fn ids_are_assigned_monotonically_and_returned_on_append() {
    let path = tmp("ids-monotonic");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    assert_eq!(w.next_record_id(), 1);

    let a = w.append(&schema, &point(1, 1)).unwrap();
    let b = w.append(&schema, &point(2, 2)).unwrap();
    assert_eq!((a, b), (1, 2));
    // The id is stable from staging time, before the commit lands — which is
    // what lets a caller checkpoint against it.
    w.commit().unwrap();
    let c = w.append(&schema, &point(3, 3)).unwrap();
    w.commit().unwrap();
    assert_eq!(c, 3);

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.records().map(|r| r.id).collect::<Vec<_>>(), vec![1, 2, 3]);
    assert_eq!(f.next_record_id(), 4);
}

#[test]
fn removal_does_not_disturb_other_records_ids() {
    let path = tmp("ids-stable-removal");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    let ids: Vec<u64> = (0..5)
        .map(|i| w.append(&schema, &point(i, i)).unwrap())
        .collect();
    w.commit().unwrap();

    w.remove_id(ids[1]).unwrap();
    w.remove_id(ids[3]).unwrap();
    w.commit().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    // Positions shifted; ids did not. This is the whole point.
    assert_eq!(f.records().map(|r| r.id).collect::<Vec<_>>(), vec![1, 3, 5]);
    assert_eq!(
        f.dump_json(f.find_by_id(5).unwrap()).unwrap(),
        r#"{"x":4,"y":4}"#
    );
    assert_eq!(f.find_by_id(ids[1]), None);
}

#[test]
fn ids_and_the_id_counter_survive_compaction() {
    // The failure this prevents: a consumer checkpoints "processed through
    // record 7", the writer compacts, and record 7 silently becomes a different
    // record. Ids must be preserved, and a purged id must never be reissued.
    let path = tmp("ids-compaction");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    let ids: Vec<u64> = (0..4)
        .map(|i| w.append(&schema, &point(i, i)).unwrap())
        .collect();
    w.commit().unwrap();

    // Remove the first and, critically, the *highest*-id record.
    w.remove_ids(&[ids[0], ids[3]]);
    w.commit().unwrap();
    w.compact().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.generation(), 1, "compaction resets the generation");
    assert_eq!(f.records().map(|r| r.id).collect::<Vec<_>>(), vec![2, 3]);
    // Content still reachable by the id a consumer remembered.
    assert_eq!(
        f.dump_json(f.find_by_id(3).unwrap()).unwrap(),
        r#"{"x":2,"y":2}"#
    );
    // The counter carried across: id 4 is retired even though its record is gone.
    assert_eq!(f.next_record_id(), 5);

    let mut w = FileWriter::open(&path).unwrap();
    assert_eq!(w.append(&schema, &point(9, 9)).unwrap(), 5);
}

#[test]
fn find_by_id_is_a_binary_search_over_a_sorted_index() {
    let schema = point_schema();
    let mut b = FileBuilder::new();
    let ids: Vec<u64> = (0..256)
        .map(|i| b.append(&schema, &point(i, -i)).unwrap())
        .collect();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    // The invariant the search rests on, asserted rather than assumed.
    let read: Vec<u64> = f.records().map(|r| r.id).collect();
    assert!(
        read.windows(2).all(|w| w[0] < w[1]),
        "index is not sorted by id"
    );

    for (position, id) in ids.iter().enumerate() {
        assert_eq!(f.find_by_id(*id), Some(position));
    }
    assert_eq!(f.find_by_id(0), None);
    assert_eq!(f.find_by_id(9_999), None);
    assert_eq!(
        verit::Message::parse(f.get_by_id(ids[42]).unwrap())
            .unwrap()
            .schema_id(),
        schema.id()
    );
}

#[test]
fn records_after_gives_a_tailing_reader_everything_since_its_checkpoint() {
    let path = tmp("tailing");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    for i in 0..4 {
        w.append(&schema, &point(i, i)).unwrap();
    }
    w.commit().unwrap();

    // A consumer processes the first batch and checkpoints the last id it saw.
    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    let seen: Vec<u64> = f.records_after(0).map(|r| r.id).collect();
    assert_eq!(seen, vec![1, 2, 3, 4]);
    let checkpoint = *seen.last().unwrap();
    drop(f);

    // The writer appends more; the consumer re-opens and resumes from its mark.
    for i in 4..7 {
        w.append(&schema, &point(i, i)).unwrap();
    }
    w.commit().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    let fresh: Vec<u64> = f.records_after(checkpoint).map(|r| r.id).collect();
    assert_eq!(fresh, vec![5, 6, 7], "resumed batch is wrong");
    assert_eq!(f.records_after(u64::MAX).count(), 0);
}

#[test]
fn purge_ids_removes_and_erases_in_one_pass() {
    let path = tmp("purge");
    let _s = Scratch(path.clone());
    let schema = person_schema();
    let secret = "regulated-personal-data-42";

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &person("keep", 1)).unwrap();
    let doomed = w.append(&schema, &person(secret, 2)).unwrap();
    w.append(&schema, &person("keep-too", 3)).unwrap();
    w.commit().unwrap();

    assert_eq!(w.purge_ids(&[doomed]).unwrap(), 1);

    let image = std::fs::read(&path).unwrap();
    assert!(
        !image.windows(secret.len()).any(|w| w == secret.as_bytes()),
        "purge_ids left the data recoverable — it must erase, not just unlink"
    );
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.len(), 2);
    assert_eq!(f.records().map(|r| r.id).collect::<Vec<_>>(), vec![1, 3]);

    // Idempotent: purging an id that is already gone is a no-op, not an error.
    assert_eq!(w.purge_ids(&[doomed]).unwrap(), 0);
}

#[test]
fn retain_and_remove_ids_are_bulk_and_order_independent() {
    let path = tmp("bulk-removal");
    let _s = Scratch(path.clone());
    let points = point_schema();
    let people = person_schema();

    let mut w = FileWriter::create(&path).unwrap();
    let mut point_ids = Vec::new();
    for i in 0..10 {
        point_ids.push(w.append(&points, &point(i, i)).unwrap());
        w.append(&people, &person("p", i as u8)).unwrap();
    }
    w.commit().unwrap();
    assert_eq!(w.len(), 20);

    // Removing ten records by shifting position would be an off-by-one; by id
    // it is one call, and the order of the list does not matter.
    let mut shuffled = point_ids.clone();
    shuffled.reverse();
    assert_eq!(w.remove_ids(&shuffled), 10);
    // Unknown ids are ignored rather than erroring, so a retry is safe.
    assert_eq!(w.remove_ids(&[9_999, point_ids[0]]), 0);
    w.commit().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert_eq!(f.len(), 10);
    assert!(f.records().all(|r| r.schema_id == people.id()));
}

#[test]
fn rejects_files_whose_ids_break_the_index_invariants() {
    let schema = point_schema();
    let mut b = FileBuilder::new();
    b.append(&schema, &point(1, 1)).unwrap();
    b.append(&schema, &point(2, 2)).unwrap();
    let bytes = b.finish().unwrap();
    let entry = FileView::open(&bytes).unwrap().footer().index_offset as usize;
    let footer = bytes.len() - 64;

    let reseal = |mut image: Vec<u8>| {
        let crc = verit::hash::crc32(&image[footer..footer + 56]);
        image[footer + 56..footer + 60].copy_from_slice(&crc.to_le_bytes());
        image
    };

    // Descending ids would silently break `find_by_id`'s binary search, so the
    // invariant is validated on open rather than trusted.
    let mut descending = bytes.clone();
    descending[entry..entry + 8].copy_from_slice(&9u64.to_le_bytes());
    assert!(matches!(
        FileView::open(&descending),
        Err(Error::BadFile(_))
    ));

    // Duplicate ids: two records answering to one identity.
    let mut duplicate = bytes.clone();
    duplicate[entry + INDEX_ENTRY..entry + INDEX_ENTRY + 8].copy_from_slice(&1u64.to_le_bytes());
    assert!(matches!(FileView::open(&duplicate), Err(Error::BadFile(_))));

    // Id 0 is reserved as "no record".
    let mut zero = bytes.clone();
    zero[entry..entry + 8].copy_from_slice(&0u64.to_le_bytes());
    assert!(matches!(FileView::open(&zero), Err(Error::BadFile(_))));

    // A counter that does not exceed every live id would reissue an id.
    let stale = reseal({
        let mut image = bytes.clone();
        image[footer + 40..footer + 48].copy_from_slice(&1u64.to_le_bytes());
        image
    });
    assert!(matches!(FileView::open(&stale), Err(Error::BadFile(_))));
}

const INDEX_ENTRY: usize = 40;

#[test]
fn builder_refuses_out_of_order_explicit_ids() {
    let schema = point_schema();
    let mut b = FileBuilder::new();
    b.append_message_with_id(
        &schema,
        &encode(&schema, &point(1, 1), SchemaMode::HashOnly).unwrap(),
        10,
    )
    .unwrap();
    assert!(matches!(
        b.append_message_with_id(
            &schema,
            &encode(&schema, &point(2, 2), SchemaMode::HashOnly).unwrap(),
            5,
        ),
        Err(Error::BadFile(_))
    ));
}

// ---------------------------------------------------------------------------
// 6. Bulk scalar lists, resolver reuse, and owning reads
//    (the five findings from the UDSV integration, 2026-08-22)
// ---------------------------------------------------------------------------

fn vector_schema() -> Schema {
    SchemaBuilder::new()
        .add_struct(
            "Embedding",
            vec![(1, "label", Dt::Str), (2, "vector", Dt::list(Dt::F32))],
        )
        .build("Embedding")
        .unwrap()
}

#[test]
fn a_scalar_run_encodes_byte_identically_to_a_value_list() {
    // The whole safety argument for `Value::Scalars`: it is a cost choice, never
    // a wire-format one. If these ever diverge, the bulk path is a new encoding.
    let schema = vector_schema();
    let floats: Vec<f32> = (0..257).map(|i| i as f32 * 0.5 - 3.25).collect();

    let per_element = encode(
        &schema,
        &Value::Struct(vec![
            (1, Value::str("v")),
            (
                2,
                Value::List(floats.iter().map(|f| Value::F32(*f)).collect()),
            ),
        ]),
        SchemaMode::HashOnly,
    )
    .unwrap();

    let bulk = encode(
        &schema,
        &Value::Struct(vec![(1, Value::str("v")), (2, Value::f32_list(&floats))]),
        SchemaMode::HashOnly,
    )
    .unwrap();

    assert_eq!(per_element, bulk, "the bulk writer changed the bytes");
}

#[test]
fn every_scalar_run_type_matches_its_value_list_encoding() {
    // Same obligation as above, across every element type — including the
    // narrow ones where a stride or sign-extension slip would be invisible in
    // f32 alone.
    macro_rules! same {
        ($dt:expr, $elem:expr, $values:expr, $wrap:expr, $bulk:expr) => {{
            let schema = SchemaBuilder::new()
                .add_struct("L", vec![(1, "xs", Dt::list($dt))])
                .build("L")
                .unwrap();
            let listed = Value::List($values.iter().map(|x| $wrap(*x)).collect());
            let a = encode(
                &schema,
                &Value::Struct(vec![(1, listed)]),
                SchemaMode::HashOnly,
            )
            .unwrap();
            let b = encode(
                &schema,
                &Value::Struct(vec![(1, Value::Scalars($bulk($values.to_vec())))]),
                SchemaMode::HashOnly,
            )
            .unwrap();
            assert_eq!(a, b, "bulk encoding differs for {}", $elem);
        }};
    }

    same!(
        Dt::Bool,
        "bool",
        [true, false, true],
        Value::Bool,
        Scalars::Bool
    );
    same!(Dt::U8, "u8", [0u8, 7, 255], Value::U8, Scalars::U8);
    same!(Dt::U16, "u16", [0u16, 513, 65535], Value::U16, Scalars::U16);
    same!(
        Dt::U32,
        "u32",
        [0u32, 70000, u32::MAX],
        Value::U32,
        Scalars::U32
    );
    same!(
        Dt::U64,
        "u64",
        [0u64, 1 << 40, u64::MAX],
        Value::U64,
        Scalars::U64
    );
    same!(Dt::I8, "i8", [-128i8, 0, 127], Value::I8, Scalars::I8);
    same!(
        Dt::I16,
        "i16",
        [-32768i16, 0, 32767],
        Value::I16,
        Scalars::I16
    );
    same!(
        Dt::I32,
        "i32",
        [i32::MIN, 0, i32::MAX],
        Value::I32,
        Scalars::I32
    );
    same!(
        Dt::I64,
        "i64",
        [i64::MIN, 0, i64::MAX],
        Value::I64,
        Scalars::I64
    );
    same!(
        Dt::F32,
        "f32",
        [-1.5f32, 0.0, 3.25],
        Value::F32,
        Scalars::F32
    );
    same!(
        Dt::F64,
        "f64",
        [-1.5f64, 0.0, 3.25],
        Value::F64,
        Scalars::F64
    );
}

#[test]
fn bulk_list_reads_return_exactly_the_written_values() {
    let schema = vector_schema();
    let floats: Vec<f32> = (0..1536).map(|i| (i as f32).sin()).collect();

    let mut b = FileBuilder::new();
    b.append(
        &schema,
        &Value::Struct(vec![(1, Value::str("e")), (2, Value::f32_list(&floats))]),
    )
    .unwrap();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    let resolver = f.resolver_for(0, &schema).unwrap();
    let msg = f.message(0).unwrap();
    let root = msg.root(&resolver).unwrap();
    let list = root.get_list(2).unwrap().unwrap();

    assert_eq!(list.len() as usize, floats.len());
    assert_eq!(list.to_vec_f32().unwrap(), floats, "to_vec_f32 lost data");

    // Element-by-element and bulk must agree exactly — bulk is a fast path, not
    // a different reading of the bytes.
    let mut out = vec![0f32; floats.len()];
    assert_eq!(list.copy_f32(&mut out).unwrap(), floats.len());
    assert_eq!(out, floats);
    for (i, expected) in floats.iter().enumerate() {
        match list.get(i as u32).unwrap() {
            verit::Ref::F32(x) => assert_eq!(x, *expected, "element {i}"),
            other => panic!("element {i} is {}", other.kind()),
        }
    }
}

#[test]
fn a_short_destination_reads_a_prefix_rather_than_failing() {
    let schema = vector_schema();
    let floats: Vec<f32> = (0..64).map(|i| i as f32).collect();
    let mut b = FileBuilder::new();
    b.append(
        &schema,
        &Value::Struct(vec![(1, Value::str("e")), (2, Value::f32_list(&floats))]),
    )
    .unwrap();
    let bytes = b.finish().unwrap();
    let f = FileView::open(&bytes).unwrap();
    let resolver = f.resolver_for(0, &schema).unwrap();
    let msg = f.message(0).unwrap();
    let list = msg.root(&resolver).unwrap().get_list(2).unwrap().unwrap();

    let mut out = [0f32; 8];
    assert_eq!(list.copy_f32(&mut out).unwrap(), 8);
    assert_eq!(out, [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0]);

    // Longer than the list: copies what exists, reports how many.
    let mut big = vec![-1f32; 100];
    assert_eq!(list.copy_f32(&mut big).unwrap(), 64);
    assert_eq!(big[63], 63.0);
    assert_eq!(big[64], -1.0, "past the end must be untouched");
}

#[test]
fn bulk_reads_still_widen_across_schema_evolution() {
    // A list written as u16 and read by a schema that widened it to u32 must go
    // through the conversion path and still be correct — the fast path is only
    // valid when the writer's type matches.
    let v1 = SchemaBuilder::new()
        .add_struct("L", vec![(1, "xs", Dt::list(Dt::U16))])
        .build("L")
        .unwrap();
    let v2 = SchemaBuilder::new()
        .add_struct("L", vec![(1, "xs", Dt::list(Dt::U32))])
        .build("L")
        .unwrap();

    let source: Vec<u16> = vec![0, 1, 300, 65535];
    let mut b = FileBuilder::new();
    b.append(
        &v1,
        &Value::Struct(vec![(1, Value::Scalars(Scalars::U16(source.clone())))]),
    )
    .unwrap();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    let resolver = f.resolver_for(0, &v2).unwrap();
    let msg = f.message(0).unwrap();
    let list = msg.root(&resolver).unwrap().get_list(1).unwrap().unwrap();

    let widened: Vec<u32> = source.iter().map(|x| *x as u32).collect();
    assert_eq!(list.to_vec_u32().unwrap(), widened);

    // Asking for the writer's own width is a type error, because the reader
    // schema says u32 — the plan, not the bytes, decides.
    assert!(matches!(list.to_vec_u16(), Err(Error::TypeMismatch { .. })));
}

#[test]
fn a_byte_list_can_be_borrowed_without_copying() {
    let schema = SchemaBuilder::new()
        .add_struct("B", vec![(1, "xs", Dt::list(Dt::U8))])
        .build("B")
        .unwrap();
    let source: Vec<u8> = (0..=255).collect();
    let mut b = FileBuilder::new();
    b.append(
        &schema,
        &Value::Struct(vec![(1, Value::Scalars(Scalars::U8(source.clone())))]),
    )
    .unwrap();
    let bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    let resolver = f.resolver_for(0, &schema).unwrap();
    let msg = f.message(0).unwrap();
    let list = msg.root(&resolver).unwrap().get_list(1).unwrap().unwrap();

    let borrowed = list.as_u8_slice().unwrap();
    assert_eq!(borrowed, &source[..]);
    // It really is a view into the file image, not a copy.
    let base = bytes.as_ptr() as usize;
    let at = borrowed.as_ptr() as usize;
    assert!(at >= base && at < base + bytes.len(), "as_u8_slice copied");
}

#[test]
fn asking_a_list_for_the_wrong_element_type_is_a_typed_error() {
    let schema = vector_schema();
    let mut b = FileBuilder::new();
    b.append(
        &schema,
        &Value::Struct(vec![
            (1, Value::str("e")),
            (2, Value::f32_list(&[1.0, 2.0])),
        ]),
    )
    .unwrap();
    let bytes = b.finish().unwrap();
    let f = FileView::open(&bytes).unwrap();
    let resolver = f.resolver_for(0, &schema).unwrap();
    let msg = f.message(0).unwrap();
    let list = msg.root(&resolver).unwrap().get_list(2).unwrap().unwrap();

    assert!(matches!(list.to_vec_u32(), Err(Error::TypeMismatch { .. })));
    assert!(matches!(
        list.as_u8_slice(),
        Err(Error::TypeMismatch { .. })
    ));
    // The error names what it actually found, not just what it wanted.
    match list.to_vec_u32() {
        Err(Error::TypeMismatch { got, .. }) => assert_eq!(got, "list<f32>"),
        other => panic!("unexpected: {other:?}"),
    }
}

#[test]
fn resolvers_resolves_each_schema_once_and_handles_mixed_files() {
    let points = point_schema();
    let people = person_schema();

    let mut b = FileBuilder::new();
    for i in 0..10 {
        b.append(&points, &point(i, i)).unwrap();
    }
    b.append(&people, &person("Ada", 36)).unwrap();
    let bytes = b.finish().unwrap();
    let f = FileView::open(&bytes).unwrap();

    // Eleven records, two schemas, and a reader that only understands Point:
    // one resolution, and the Person record reported as incompatible rather
    // than failing the whole call.
    let resolvers = f.resolvers(&points);
    assert_eq!(resolvers.len(), 1, "only Point should resolve into Point");
    assert!(resolvers.get(points.id()).is_some());
    assert!(resolvers.get(people.id()).is_none());

    for i in 0..10 {
        let resolver = resolvers.for_record(&f, i).unwrap();
        let root = f.message(i).unwrap().root(resolver).unwrap();
        assert_eq!(root.get_i32(1).unwrap(), Some(i as i32));
    }
    assert!(matches!(
        resolvers.for_record(&f, 10),
        Err(Error::Incompatible(_))
    ));
}

#[test]
fn file_reader_owns_its_bytes_and_validates_on_open() {
    let path = tmp("owned");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create(&path).unwrap();
    w.append(&schema, &point(4, 5)).unwrap();
    w.commit().unwrap();

    let reader = verit::FileReader::open(&path).unwrap();
    let view = reader.view().unwrap();
    assert_eq!(view.len(), 1);
    assert_eq!(view.dump_json(0).unwrap(), r#"{"x":4,"y":5}"#);

    // A malformed image is refused at open, not at first access.
    let mut broken = std::fs::read(&path).unwrap();
    broken[0] = b'X';
    assert!(matches!(
        verit::FileReader::from_bytes(broken),
        Err(Error::BadFile(_))
    ));
}

// ---------------------------------------------------------------------------
// 7. Operational hardening (roadmap Phase 5)
// ---------------------------------------------------------------------------

#[test]
fn a_locking_writer_refuses_a_second_one() {
    let path = tmp("locked");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::open_or_create_locked(&path).unwrap();
    assert!(w.is_locked());
    w.append(&schema, &point(1, 1)).unwrap();
    w.commit().unwrap();

    // A second locking writer is refused rather than racing.
    match FileWriter::open_locked(&path) {
        Err(Error::AlreadyLocked(p)) => assert!(p.ends_with(".lock")),
        Err(e) => panic!("wrong error for a second writer: {e}"),
        Ok(_) => panic!("a second locking writer should have been refused"),
    }

    // Readers are never blocked by a writer's lock.
    let image = std::fs::read(&path).unwrap();
    assert_eq!(FileView::open(&image).unwrap().len(), 1);

    // Dropping the writer releases it.
    drop(w);
    let mut again = FileWriter::open_locked(&path).unwrap();
    assert_eq!(again.len(), 1);
    again.append(&schema, &point(2, 2)).unwrap();
    again.commit().unwrap();
}

#[test]
fn the_lock_survives_a_compaction() {
    // compact() reopens the file internally; if that dropped the lock, another
    // writer could slip in during the one operation that rewrites everything.
    let path = tmp("lock-compact");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::open_or_create_locked(&path).unwrap();
    let ids: Vec<u64> = (0..4)
        .map(|i| w.append(&schema, &point(i, i)).unwrap())
        .collect();
    w.commit().unwrap();
    w.purge_ids(&[ids[0]]).unwrap();

    assert!(w.is_locked(), "compaction dropped the advisory lock");
    assert!(matches!(
        FileWriter::open_locked(&path),
        Err(Error::AlreadyLocked(_))
    ));
    assert_eq!(w.len(), 3);
}

#[test]
fn an_unlocked_writer_stays_unlocked() {
    // The lock is opt-in: the historical constructors must not start creating
    // lock files, or every existing caller changes behaviour.
    let path = tmp("unlocked");
    let _s = Scratch(path.clone());
    let mut w = FileWriter::create(&path).unwrap();
    w.append(&point_schema(), &point(1, 1)).unwrap();
    w.commit().unwrap();

    assert!(!w.is_locked());
    let mut lock_path = path.clone().into_os_string();
    lock_path.push(".lock");
    assert!(
        !std::path::Path::new(&lock_path).exists(),
        "an unlocked writer must not create a lock file"
    );
    // And a locking writer can take the file while an unlocked one holds it —
    // advisory means advisory, and that is documented rather than pretended.
    assert!(FileWriter::open_locked(&path).is_ok());
}

#[test]
fn per_record_checksums_detect_bit_rot() {
    let schema = person_schema();
    let mut b = FileBuilder::new().with_record_checksums();
    b.append(&schema, &person("Ada", 36)).unwrap();
    b.append(&schema, &person("Grace", 45)).unwrap();
    let mut bytes = b.finish().unwrap();

    let f = FileView::open(&bytes).unwrap();
    assert!(f.has_record_checksums());
    assert_eq!(f.verify_checksums().unwrap(), 2);
    let target = f.record(1).unwrap();
    drop(f);

    // Flip a bit inside record 1's payload — the footer CRC still passes,
    // because it only ever proved the commit was not torn.
    bytes[target.offset as usize + 34] ^= 0x01;
    let f = FileView::open(&bytes).expect("the file is still structurally valid");
    match f.verify_checksums() {
        Err(Error::ChecksumMismatch { id, .. }) => assert_eq!(id, target.id),
        other => panic!("bit rot went undetected: {other:?}"),
    }
}

#[test]
fn checksums_are_optional_to_readers_and_absent_by_default() {
    let schema = point_schema();

    // Default files carry none, and verifying is a no-op rather than an error.
    let mut plain = FileBuilder::new();
    plain.append(&schema, &point(1, 2)).unwrap();
    let plain = plain.finish().unwrap();
    let f = FileView::open(&plain).unwrap();
    assert!(!f.has_record_checksums());
    assert_eq!(f.verify_checksums().unwrap(), 0, "absence is not a fault");
    assert_eq!(f.record_checksum(0), None);

    // A checksummed file reads identically to one without — the array lives in
    // space a reader that ignores the feature already skips.
    let mut summed = FileBuilder::new().with_record_checksums();
    summed.append(&schema, &point(1, 2)).unwrap();
    let summed = summed.finish().unwrap();
    let a = FileView::open(&plain).unwrap();
    let b = FileView::open(&summed).unwrap();
    assert_eq!(a.dump_json(0).unwrap(), b.dump_json(0).unwrap());
    assert_eq!(a.record(0).unwrap().id, b.record(0).unwrap().id);
    assert!(summed.len() > plain.len(), "the array should cost bytes");
}

#[test]
fn checksums_survive_appends_and_compaction() {
    let path = tmp("checksummed");
    let _s = Scratch(path.clone());
    let schema = point_schema();

    let mut w = FileWriter::create_checksummed(&path).unwrap();
    let ids: Vec<u64> = (0..5)
        .map(|i| w.append(&schema, &point(i, -i)).unwrap())
        .collect();
    w.commit().unwrap();
    w.append(&schema, &point(99, 99)).unwrap();
    w.commit().unwrap();

    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert!(f.has_record_checksums());
    assert_eq!(
        f.verify_checksums().unwrap(),
        6,
        "a later commit lost checksums"
    );

    // Compaction must carry the feature and the values across, not silently
    // drop to an unchecksummed file.
    w.purge_ids(&[ids[0]]).unwrap();
    let image = std::fs::read(&path).unwrap();
    let f = FileView::open(&image).unwrap();
    assert!(f.has_record_checksums(), "compaction dropped the feature");
    assert_eq!(f.verify_checksums().unwrap(), 5);
    assert_eq!(f.generation(), 1);
}