verit-core 0.2.0

//! `.verit` — the Veritate **file**: many messages in one self-contained,
//! `mmap`-able, appendable file.
//!
//! Normative definition: `docs/Architecture/VERIT - File Format Specification.md`.
//! The rationale is `docs/decisions/ADR-0002`. This module implements that
//! document; where the two disagree, the document wins.
//!
//! ```text
//! ┌─ header (32 B, written once) ───────────────────────────────────────────┐
//! │ "VRTF" · version · required_features · optional_features                │
//! ├─ generation 1 ──────────────────────────────────────────────────────────┤
//! │  records · schema section (VRSB) · index · footer (64 B)                │
//! ├─ generation 2 (append / removal — appended, nothing above is rewritten) ┤
//! │  new records · schema section · index · footer (64 B)  ← authoritative   │
//! └─────────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! Four properties fall out of that shape:
//!
//! - **Self-contained.** The schema section holds a schema for every id in the
//!   index, so a `.verit` file plus nothing else is readable. Records are
//!   hash-only, so each schema is stored once no matter how many records use it.
//! - **Crash-safe without a journal.** A footer is authoritative only once fully
//!   written and CRC-valid. A crash mid-append leaves the previous footer intact,
//!   so the file reads exactly as it did before. [`FileView::open`] finds the
//!   newest valid footer by scanning back from the end.
//! - **Snapshot reads with no locking.** Committed bytes are never rewritten, so
//!   a reader holding a footer has a stable view while a writer appends. One
//!   writer, many readers.
//! - **Stable identity.** Every record carries a monotonic `u64` **record id**
//!   that is never reused and survives removal and compaction. A position is
//!   not an identity — it shifts under both — so anything that remembers a
//!   record across commits (a consumer checkpoint, a sync cursor) must remember
//!   its id. Because ids ascend with the index, [`FileView::find_by_id`] is a
//!   binary search over bytes already loaded, and
//!   [`FileView::records_after`] gives a tailing reader "everything since my
//!   checkpoint" for free.
//!
//! Two entry points, sharing one commit-image implementation so they cannot
//! drift:
//!
//! - [`FileBuilder`] builds a complete generation-1 file in memory, returning
//!   `Vec<u8>`. No I/O, so it is what the golden-file corpus and the ports'
//!   conformance suites are written against.
//! - [`FileWriter`] owns a [`std::fs::File`] and performs incremental,
//!   crash-safe commits — [`append`](FileWriter::append),
//!   [`remove_id`](FileWriter::remove_id), [`commit`](FileWriter::commit),
//!   [`compact`](FileWriter::compact).
//!
//! Reading stays zero-copy and dependency-free: [`FileView::open`] takes a
//! `&[u8]`, so `mmap` the file with your platform's facility (or
//! [`std::fs::read`] it) and pass the bytes. [`FileView::get`] returns a
//! sub-slice you hand straight to [`Message::parse`].
//!
//! ```
//! # use verit_core::{Dt, SchemaBuilder, Value};
//! # use verit_core::file::{FileBuilder, FileView};
//! let schema = SchemaBuilder::new()
//!     .add_struct("Point", vec![(1, "x", Dt::I32), (2, "y", Dt::I32)])
//!     .build("Point")
//!     .unwrap();
//!
//! let mut b = FileBuilder::new();
//! let first = b.append(&schema, &Value::Struct(vec![(1, Value::I32(3)), (2, Value::I32(4))])).unwrap();
//! b.append(&schema, &Value::Struct(vec![(1, Value::I32(-1)), (2, Value::I32(0))])).unwrap();
//! let bytes = b.finish().unwrap();
//!
//! // From here on, pretend we know nothing but `bytes`.
//! let f = FileView::open(&bytes).unwrap();
//! assert_eq!(f.len(), 2);
//! assert_eq!(f.dump_json(0).unwrap(), r#"{"x":3,"y":4}"#);
//! // A record is found by id, not by position.
//! assert_eq!(f.find_by_id(first), Some(0));
//! ```
//!
//! ## What this file is not
//!
//! One file. Retention — "drop the oldest million records" — costs a full
//! compaction here, and no layout inside a single file avoids that. The answer
//! is **segmented files**: many `.verit` files under a naming convention, whole
//! segments dropped. That belongs in a layer above the format, and is
//! deliberately not built into it.

use std::collections::{HashMap, HashSet};
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

use crate::encode::{encode, SchemaMode};
use crate::error::{Error, Result};
use crate::hash::crc32;
use crate::message::Message;
use crate::registry::SchemaRegistry;
use crate::resolve::Resolver;
use crate::schema::Schema;
use crate::value::Value;

/// File magic: "VRTF". Deliberately not a `VRTC` version bump — a 0.1.0
/// container is rejected here structurally rather than partially misread.
pub const FILE_MAGIC: &[u8; 4] = b"VRTF";
/// File format version this build reads and writes.
pub const FILE_VERSION: u8 = 1;
/// Fixed header length, and the alignment every region starts on.
pub const FILE_HEADER_LEN: usize = 32;
/// Fixed footer length — one cache line, with room reserved for one more field.
pub const FOOTER_LEN: usize = 64;
/// Index entry width: `u64 id`, `u64 offset`, `u64 length`, `u128 schema_id`.
pub const INDEX_ENTRY_LEN: usize = 40;
/// Record ids start at 1, so `0` is available as "no record".
pub const FIRST_RECORD_ID: u64 = 1;
/// `optional_features` bit 0 — the file carries a CRC-32 per record.
///
/// Optional in the strict sense of spec §3.1: a reader that does not implement
/// it reads the file correctly and simply never checks the checksums, because
/// they live in space it already skips. See [`FileView::verify_checksums`].
pub const OPT_RECORD_CRC: u32 = 1;

const ALIGN: u64 = 8;

#[inline]
fn align_up(x: u64) -> Result<u64> {
    x.checked_add(ALIGN - 1)
        .map(|v| v & !(ALIGN - 1))
        .ok_or(Error::BadFile("offset overflow"))
}

/// One index entry: a record's identity, where it lives, and which schema
/// interprets it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Record {
    /// Monotonic record id. Never reused; survives removal and compaction.
    /// **This is the record's identity** — its position is not.
    pub id: u64,
    /// Absolute offset of the record's first byte. Always 8-aligned.
    pub offset: u64,
    /// The record's length in bytes, unpadded.
    pub length: u64,
    /// The record's 128-bit schema id, mirroring the one in its own header.
    pub schema_id: u128,
}

/// A parsed footer — the authoritative statement of a file's committed state.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Footer {
    /// Monotonic commit counter, starting at 1.
    pub generation: u64,
    pub index_offset: u64,
    pub schema_offset: u64,
    pub schema_len: u32,
    pub record_count: u32,
    /// Committed extent. Bytes at or beyond this are uncommitted debris.
    pub file_len: u64,
    /// The id the next appended record will take. Persisted rather than derived
    /// from the index, so removing the highest-id record cannot cause an id to
    /// be handed out twice.
    pub next_record_id: u64,
}

impl Footer {
    fn encode(&self) -> [u8; FOOTER_LEN] {
        let mut out = [0u8; FOOTER_LEN];
        out[0..8].copy_from_slice(&self.generation.to_le_bytes());
        out[8..16].copy_from_slice(&self.index_offset.to_le_bytes());
        out[16..24].copy_from_slice(&self.schema_offset.to_le_bytes());
        out[24..28].copy_from_slice(&self.schema_len.to_le_bytes());
        out[28..32].copy_from_slice(&self.record_count.to_le_bytes());
        out[32..40].copy_from_slice(&self.file_len.to_le_bytes());
        out[40..48].copy_from_slice(&self.next_record_id.to_le_bytes());
        // out[48..56] is reserved and stays zero.
        let crc = crc32(&out[0..56]);
        out[56..60].copy_from_slice(&crc.to_le_bytes());
        out[60..64].copy_from_slice(FILE_MAGIC);
        out
    }

    /// Decode a footer candidate sitting at absolute offset `at`, accepting it
    /// only if the trailing magic, the CRC, and the self-referential
    /// `file_len == at + FOOTER_LEN` all agree (spec §7.2). That last check is
    /// what pins a footer to its own position, so `VRTF` bytes occurring inside
    /// record data cannot be mistaken for one.
    fn decode_at(bytes: &[u8; FOOTER_LEN], at: u64) -> Option<Footer> {
        if &bytes[60..64] != FILE_MAGIC {
            return None;
        }
        if u32::from_le_bytes(bytes[56..60].try_into().ok()?) != crc32(&bytes[0..56]) {
            return None;
        }
        // A nonzero reserved field means these are not v1 footer bytes. The
        // header's version check runs first, so a genuinely newer format is
        // refused there with a clear error rather than skipped here.
        if u64::from_le_bytes(bytes[48..56].try_into().ok()?) != 0 {
            return None;
        }
        let file_len = u64::from_le_bytes(bytes[32..40].try_into().ok()?);
        if file_len != at.checked_add(FOOTER_LEN as u64)? {
            return None;
        }
        Some(Footer {
            generation: u64::from_le_bytes(bytes[0..8].try_into().ok()?),
            index_offset: u64::from_le_bytes(bytes[8..16].try_into().ok()?),
            schema_offset: u64::from_le_bytes(bytes[16..24].try_into().ok()?),
            schema_len: u32::from_le_bytes(bytes[24..28].try_into().ok()?),
            record_count: u32::from_le_bytes(bytes[28..32].try_into().ok()?),
            file_len,
            next_record_id: u64::from_le_bytes(bytes[40..48].try_into().ok()?),
        })
    }
}

fn header_bytes(optional_features: u32) -> [u8; FILE_HEADER_LEN] {
    let mut out = [0u8; FILE_HEADER_LEN];
    out[0..4].copy_from_slice(FILE_MAGIC);
    out[4] = FILE_VERSION;
    // required_features stays zero: nothing this version writes is mandatory
    // for a reader to understand.
    out[12..16].copy_from_slice(&optional_features.to_le_bytes());
    out
}

/// Validate the 32-byte header (spec §3). Feature bits split critical from
/// non-critical: an unknown *required* bit is fatal, an unknown *optional* bit
/// is ignored, which is how the format grows without stranding old readers.
fn check_header(buf: &[u8]) -> Result<()> {
    if buf.len() < FILE_HEADER_LEN {
        return Err(Error::Truncated);
    }
    if &buf[0..4] != FILE_MAGIC {
        return Err(Error::BadFile("bad magic (not a .verit file)"));
    }
    if buf[4] != FILE_VERSION {
        return Err(Error::BadFile("unsupported .verit file version"));
    }
    if buf[5] != 0 || u16::from_le_bytes(buf[6..8].try_into().unwrap()) != 0 {
        return Err(Error::BadFile("nonzero reserved header field"));
    }
    let required = u32::from_le_bytes(buf[8..12].try_into().unwrap());
    if required != 0 {
        return Err(Error::UnsupportedFileFeature(required));
    }
    // buf[12..16] is `optional_features`: unknown bits are ignored by design.
    if u64::from_le_bytes(buf[16..24].try_into().unwrap()) != 0
        || u64::from_le_bytes(buf[24..32].try_into().unwrap()) != 0
    {
        return Err(Error::BadFile("nonzero reserved header field"));
    }
    Ok(())
}

/// Serialize the tail of a commit — schema section, index, footer — for a set
/// of records whose bytes already sit below `tail_start`.
///
/// Shared by [`FileBuilder`] and [`FileWriter`] so the in-memory and on-disk
/// writers cannot produce different bytes for the same logical file.
/// `tail_start` must be 8-aligned. Returns `(schema section + index, footer)`
/// kept separate because the commit protocol (§7.1) must durably flush the
/// former *before* writing the latter.
fn commit_tail(
    index: &[Record],
    registry: &SchemaRegistry,
    tail_start: u64,
    generation: u64,
    next_record_id: u64,
    checksums: Option<&[u32]>,
) -> Result<(Vec<u8>, Footer)> {
    let count = u32::try_from(index.len()).map_err(|_| Error::BadFile("too many records"))?;

    // Only the schemas the live index actually references. A schema whose last
    // record was removed drops out of the file on the next commit.
    let live: HashSet<u128> = index.iter().map(|r| r.schema_id).collect();
    let mut pruned = SchemaRegistry::new();
    for id in live {
        let schema = registry.get(id).ok_or(Error::MissingSchema(id))?;
        pruned.register(schema.clone());
    }

    let section = pruned.to_bundle();
    let schema_len =
        u32::try_from(section.len()).map_err(|_| Error::BadFile("schema section too large"))?;

    let schema_offset = tail_start;
    // With per-record checksums, the array sits in the space between the schema
    // section and the index — space a reader without the feature already skips,
    // which is what makes the feature genuinely optional (spec §5.1).
    let crc_bytes = match checksums {
        Some(c) => {
            debug_assert_eq!(c.len(), index.len());
            (c.len() as u64) * 4
        }
        None => 0,
    };
    let index_offset = align_up(schema_offset + section.len() as u64 + crc_bytes)?;
    let index_bytes = (count as u64)
        .checked_mul(INDEX_ENTRY_LEN as u64)
        .ok_or(Error::BadFile("index size overflow"))?;
    let file_len = index_offset
        .checked_add(index_bytes)
        .and_then(|v| v.checked_add(FOOTER_LEN as u64))
        .ok_or(Error::BadFile("file length overflow"))?;

    let mut tail = Vec::with_capacity((file_len - tail_start) as usize);
    tail.extend_from_slice(&section);
    if let Some(c) = checksums {
        // Placed so the array ends exactly at `index_offset`, which is how a
        // reader locates it without a new footer field.
        tail.resize((index_offset - schema_offset) as usize - c.len() * 4, 0);
        for crc in c {
            tail.extend_from_slice(&crc.to_le_bytes());
        }
    }
    tail.resize((index_offset - schema_offset) as usize, 0); // pad to index alignment
    for r in index {
        tail.extend_from_slice(&r.id.to_le_bytes());
        tail.extend_from_slice(&r.offset.to_le_bytes());
        tail.extend_from_slice(&r.length.to_le_bytes());
        tail.extend_from_slice(&r.schema_id.to_le_bytes());
    }

    Ok((
        tail,
        Footer {
            generation,
            index_offset,
            schema_offset,
            schema_len,
            record_count: count,
            file_len,
            next_record_id,
        },
    ))
}

/// Pull the writer schema out of an inline-schema message, or fail if it is
/// hash-only (in which case the caller must supply the schema).
fn schema_of(bytes: &[u8]) -> Result<Schema> {
    Message::parse(bytes)?
        .writer_schema()?
        .ok_or(Error::NoInlineSchema)
}

// ---------------------------------------------------------------------------
// Reading
// ---------------------------------------------------------------------------

/// A read-only, zero-copy view over a `.verit` file's bytes (typically an
/// `mmap`). [`open`](FileView::open) validates the header, footer, schema
/// section, and **every** index entry up front, so each later
/// [`get`](FileView::get) is a bounds-free slice.
#[derive(Clone, Debug)]
pub struct FileView<'a> {
    buf: &'a [u8],
    footer: Footer,
    registry: SchemaRegistry,
}

impl<'a> FileView<'a> {
    /// Open a file image, recovering the newest valid commit.
    ///
    /// Follows spec §7.2: scan back from the end over 8-aligned offsets for a
    /// footer whose trailing magic, CRC, and self-referential length all agree.
    /// The first hit is the newest generation, so an intact file resolves on the
    /// first candidate and a file torn mid-append transparently opens at the
    /// previous generation — the crash-safety guarantee, exercised as a read.
    pub fn open(buf: &'a [u8]) -> Result<FileView<'a>> {
        check_header(buf)?;
        if buf.len() < FILE_HEADER_LEN + FOOTER_LEN {
            return Err(Error::Truncated);
        }

        let footer = Self::find_footer(buf)?;

        // Region geometry. The index must end exactly where the footer begins:
        // no slack is tolerated, which removes a class of ambiguous files.
        if footer.file_len > buf.len() as u64 {
            return Err(Error::BadFile(
                "footer claims more bytes than the image has",
            ));
        }
        if footer.schema_offset % ALIGN != 0 || footer.index_offset % ALIGN != 0 {
            return Err(Error::BadFile("misaligned schema or index offset"));
        }
        if footer.schema_offset < FILE_HEADER_LEN as u64 {
            return Err(Error::BadFile("schema section overlaps the header"));
        }
        let schema_end = footer
            .schema_offset
            .checked_add(footer.schema_len as u64)
            .ok_or(Error::BadFile("schema section extent overflow"))?;
        if schema_end > footer.index_offset {
            return Err(Error::BadFile("schema section overlaps the index"));
        }
        // With per-record checksums, the array occupies the 4 × record_count
        // bytes ending at `index_offset`; it must not run back into the schema
        // section.
        if u32::from_le_bytes(buf[12..16].try_into().unwrap()) & OPT_RECORD_CRC != 0 {
            let crc_bytes = (footer.record_count as u64)
                .checked_mul(4)
                .ok_or(Error::BadFile("checksum array size overflow"))?;
            if footer.index_offset < crc_bytes || footer.index_offset - crc_bytes < schema_end {
                return Err(Error::BadFile("checksum array overlaps the schema section"));
            }
        }
        // `record_count` is attacker-controlled, so this is proven by
        // arithmetic before anything is allocated or indexed (spec §10).
        let index_bytes = (footer.record_count as u64)
            .checked_mul(INDEX_ENTRY_LEN as u64)
            .ok_or(Error::BadFile("index size overflow"))?;
        let index_end = footer
            .index_offset
            .checked_add(index_bytes)
            .ok_or(Error::BadFile("index extent overflow"))?;
        if index_end != footer.file_len - FOOTER_LEN as u64 {
            return Err(Error::BadFile("index does not end at the footer"));
        }

        let registry =
            SchemaRegistry::from_bundle(&buf[footer.schema_offset as usize..schema_end as usize])?;

        let view = FileView {
            buf,
            footer,
            registry,
        };

        // Validate every entry once, so `get` can never go out of bounds, the
        // self-containment rule (§5) is proven before any record is served, and
        // the strictly-ascending id invariant that `find_by_id`'s binary search
        // relies on is established rather than assumed.
        let mut prev_id = 0u64;
        for i in 0..view.len() {
            let r = view.record(i)?;
            if r.id == 0 {
                return Err(Error::BadFile("record id 0 is reserved"));
            }
            if r.id <= prev_id {
                return Err(Error::BadFile("record ids are not strictly ascending"));
            }
            prev_id = r.id;
            if r.offset % ALIGN != 0 {
                return Err(Error::BadFile("misaligned record"));
            }
            if r.offset < FILE_HEADER_LEN as u64 {
                return Err(Error::BadFile("record overlaps the header"));
            }
            let end = r
                .offset
                .checked_add(r.length)
                .ok_or(Error::BadFile("record extent overflow"))?;
            if end > footer.schema_offset {
                return Err(Error::BadFile("record outside the record region"));
            }
            if !view.registry.contains(r.schema_id) {
                return Err(Error::MissingSchema(r.schema_id));
            }
        }
        if footer.next_record_id <= prev_id {
            return Err(Error::BadFile(
                "next_record_id does not exceed every record id",
            ));
        }

        Ok(view)
    }

    fn find_footer(buf: &[u8]) -> Result<Footer> {
        // Highest 8-aligned position a footer could start at. Descending, so
        // the first valid candidate is the newest generation.
        let mut pos = ((buf.len() - FOOTER_LEN) as u64) & !(ALIGN - 1);
        loop {
            let at = pos as usize;
            let bytes: [u8; FOOTER_LEN] = buf[at..at + FOOTER_LEN]
                .try_into()
                .map_err(|_| Error::Internal("footer slice width"))?;
            if let Some(f) = Footer::decode_at(&bytes, pos) {
                return Ok(f);
            }
            if pos < FILE_HEADER_LEN as u64 + ALIGN {
                return Err(Error::NoValidFooter);
            }
            pos -= ALIGN;
        }
    }

    /// Number of live records.
    pub fn len(&self) -> usize {
        self.footer.record_count as usize
    }

    pub fn is_empty(&self) -> bool {
        self.footer.record_count == 0
    }

    /// The commit counter of the generation this view resolved to. A value
    /// lower than expected after a crash means the torn commit was rolled back.
    pub fn generation(&self) -> u64 {
        self.footer.generation
    }

    /// The authoritative footer.
    pub fn footer(&self) -> Footer {
        self.footer
    }

    /// The committed extent. Bytes at or beyond this are uncommitted debris and
    /// carry no meaning.
    pub fn file_len(&self) -> u64 {
        self.footer.file_len
    }

    /// The id the next appended record will take. Every live record's id is
    /// strictly below this, and no id at or above it has ever been used.
    pub fn next_record_id(&self) -> u64 {
        self.footer.next_record_id
    }

    /// Whether this file carries a CRC-32 per record ([`OPT_RECORD_CRC`]).
    pub fn has_record_checksums(&self) -> bool {
        u32::from_le_bytes(self.buf[12..16].try_into().unwrap()) & OPT_RECORD_CRC != 0
    }

    /// The stored CRC-32 for record `i`, or `None` when the file carries none.
    ///
    /// The array ends exactly at `index_offset`, in the space a reader without
    /// the feature already skips — which is what makes the feature optional
    /// rather than a format change.
    pub fn record_checksum(&self, i: usize) -> Option<u32> {
        if !self.has_record_checksums() || i >= self.len() {
            return None;
        }
        let base = self.footer.index_offset as usize - self.len() * 4 + i * 4;
        Some(u32::from_le_bytes(
            self.buf[base..base + 4].try_into().ok()?,
        ))
    }

    /// Verify every record against its stored checksum.
    ///
    /// Returns the number checked — `Ok(0)` for a file that carries none, which
    /// is not an error: checksums are optional, and their absence is a property
    /// of the file, not a fault. A mismatch is [`Error::ChecksumMismatch`],
    /// naming the record's **id** rather than its position.
    pub fn verify_checksums(&self) -> Result<usize> {
        if !self.has_record_checksums() {
            return Ok(0);
        }
        for i in 0..self.len() {
            let want = self.record_checksum(i).ok_or(Error::BadFile(
                "checksum array is shorter than the record count",
            ))?;
            let got = crc32(self.get(i)?);
            if got != want {
                return Err(Error::ChecksumMismatch {
                    id: self.record(i)?.id,
                    expected: want,
                    found: got,
                });
            }
        }
        Ok(self.len())
    }

    /// The file's schema section, decoded. Every schema needed to read every
    /// record is here — this is what "self-contained" means concretely.
    pub fn schemas(&self) -> &SchemaRegistry {
        &self.registry
    }

    /// Index entry `i`.
    pub fn record(&self, i: usize) -> Result<Record> {
        if i >= self.len() {
            return Err(Error::IndexOutOfBounds);
        }
        let base = self.footer.index_offset as usize + i * INDEX_ENTRY_LEN;
        Ok(Record {
            id: u64::from_le_bytes(self.buf[base..base + 8].try_into().unwrap()),
            offset: u64::from_le_bytes(self.buf[base + 8..base + 16].try_into().unwrap()),
            length: u64::from_le_bytes(self.buf[base + 16..base + 24].try_into().unwrap()),
            schema_id: u128::from_le_bytes(self.buf[base + 24..base + 40].try_into().unwrap()),
        })
    }

    /// The raw message bytes of record `i`, borrowing the file image. Hand the
    /// result straight to [`Message::parse`].
    pub fn get(&self, i: usize) -> Result<&'a [u8]> {
        let r = self.record(i)?;
        // Validated exhaustively in `open`; this cannot go out of bounds.
        Ok(&self.buf[r.offset as usize..(r.offset + r.length) as usize])
    }

    /// The position of the record with this id, or `None` if it is not live.
    ///
    /// A binary search: ids ascend with the index (appends are monotonic and
    /// neither removal nor compaction reorders), and `open` proved it. No
    /// secondary structure, and nothing outside the index bytes is touched.
    pub fn find_by_id(&self, id: u64) -> Option<usize> {
        let (mut lo, mut hi) = (0usize, self.len());
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            // `mid < len`, so this entry was validated in `open`.
            let mid_id = self.record(mid).ok()?.id;
            match mid_id.cmp(&id) {
                std::cmp::Ordering::Equal => return Some(mid),
                std::cmp::Ordering::Less => lo = mid + 1,
                std::cmp::Ordering::Greater => hi = mid,
            }
        }
        None
    }

    /// The bytes of the record with this id.
    pub fn get_by_id(&self, id: u64) -> Result<&'a [u8]> {
        self.get(self.find_by_id(id).ok_or(Error::IndexOutOfBounds)?)
    }

    /// Every record whose id is greater than `id`, in order — a tailing
    /// reader's "everything since my checkpoint".
    ///
    /// Pass the last id you processed; pass `0` for the whole file. The search
    /// for the starting position is logarithmic, so polling a large file is
    /// cheap even when nothing has changed.
    pub fn records_after(&self, id: u64) -> impl Iterator<Item = Record> + '_ {
        // First position whose id exceeds `id`.
        let (mut lo, mut hi) = (0usize, self.len());
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            match self.record(mid) {
                Ok(r) if r.id <= id => lo = mid + 1,
                _ => hi = mid,
            }
        }
        (lo..self.len()).map(move |i| self.record(i).expect("index validated in open"))
    }

    /// The schema id of record `i`, read from the index — no record bytes are
    /// touched, so filtering a large mapped file by type stays in one
    /// contiguous region instead of paging the whole file in.
    pub fn schema_id(&self, i: usize) -> Result<u128> {
        Ok(self.record(i)?.schema_id)
    }

    /// The writer schema of record `i`, from the file's own schema section.
    pub fn schema(&self, i: usize) -> Result<&Schema> {
        let id = self.schema_id(i)?;
        self.registry.get(id).ok_or(Error::MissingSchema(id))
    }

    /// Parse record `i` into a [`Message`], zero-copy over the file image.
    pub fn message(&self, i: usize) -> Result<Message<'a>> {
        Message::parse(self.get(i)?)
    }

    /// A [`Resolver`] reading record `i`'s writer schema into `reader` — the
    /// schema-evolution payoff at rest. A record written years ago under an
    /// older schema resolves into today's type, because the file kept the
    /// writer schema alongside it.
    ///
    /// **Builds a resolver on every call.** Resolution is meant to be paid once
    /// per *schema pair*, not once per record, so calling this inside a loop
    /// over a large file repeats identical work. Use
    /// [`resolvers`](Self::resolvers) there — it resolves each distinct schema
    /// in the file once and hands back a lookup.
    pub fn resolver_for(&self, i: usize, reader: &Schema) -> Result<Resolver> {
        self.registry.resolver_for(self.schema_id(i)?, reader)
    }

    /// Resolve every schema this file's records use into `reader`, **once
    /// each**, and return the lookup to use across the whole file.
    ///
    /// This is the shape the "paid once per schema pair" promise actually needs:
    /// hoist it out of the loop, then ask it per record.
    ///
    /// ```
    /// # use verit_core::{Dt, SchemaBuilder, Value};
    /// # use verit_core::file::{FileBuilder, FileView};
    /// # let schema = SchemaBuilder::new()
    /// #     .add_struct("P", vec![(1, "x", Dt::I32)]).build("P").unwrap();
    /// # let mut b = FileBuilder::new();
    /// # b.append(&schema, &Value::Struct(vec![(1, Value::I32(7))])).unwrap();
    /// # let bytes = b.finish().unwrap();
    /// let file = FileView::open(&bytes).unwrap();
    /// let resolvers = file.resolvers(&schema);          // once
    /// for i in 0..file.len() {
    ///     let resolver = resolvers.for_record(&file, i).unwrap();
    ///     let root = file.message(i).unwrap().root(resolver).unwrap();
    ///     assert_eq!(root.get_i32(1).unwrap(), Some(7));
    /// }
    /// ```
    ///
    /// A file may hold records this reader cannot interpret — a mixed-schema
    /// file read by a type that only covers one of them. Those simply do not
    /// appear in the lookup, and [`Resolvers::for_record`] reports them as
    /// incompatible rather than the whole call failing, so a reader can walk a
    /// mixed file and skip what is not for it.
    pub fn resolvers(&self, reader: &Schema) -> Resolvers {
        let mut by_schema: HashMap<u128, Resolver> = HashMap::new();
        for r in self.records() {
            if by_schema.contains_key(&r.schema_id) {
                continue;
            }
            if let Ok(resolver) = self.registry.resolver_for(r.schema_id, reader) {
                by_schema.insert(r.schema_id, resolver);
            }
        }
        Resolvers {
            by_schema,
            reader_id: reader.id(),
        }
    }

    /// Render record `i` as JSON using only this file's bytes.
    pub fn dump_json(&self, i: usize) -> Result<String> {
        crate::dump::dump_json_with(self.schema(i)?, self.get(i)?)
    }

    /// Iterate over each record's bytes in order.
    pub fn iter(&self) -> impl Iterator<Item = &'a [u8]> + '_ {
        (0..self.len()).map(move |i| self.get(i).expect("index validated in open"))
    }

    /// Iterate over the index entries in order.
    pub fn records(&self) -> impl Iterator<Item = Record> + '_ {
        (0..self.len()).map(move |i| self.record(i).expect("index validated in open"))
    }
}

/// One [`Resolver`] per distinct schema in a file, built once by
/// [`FileView::resolvers`].
///
/// The point is the arithmetic: a 10,000-record file written under one schema
/// needs **one** resolution, not 10,000. Resolution walks both schemas and
/// compiles an access plan, so doing it per record turns an O(1) cost into an
/// O(n) one for no benefit.
#[derive(Clone, Debug)]
pub struct Resolvers {
    by_schema: HashMap<u128, Resolver>,
    reader_id: u128,
}

impl Resolvers {
    /// The resolver for a writer schema id, if this reader can interpret it.
    pub fn get(&self, schema_id: u128) -> Option<&Resolver> {
        self.by_schema.get(&schema_id)
    }

    /// The resolver for record `i` of `view`.
    ///
    /// Errors with [`Error::Incompatible`] when the record's schema cannot be
    /// read into this reader — the mixed-file case, where skipping is often the
    /// right response.
    pub fn for_record<'a>(&self, view: &FileView<'a>, i: usize) -> Result<&Resolver> {
        let schema_id = view.schema_id(i)?;
        self.by_schema.get(&schema_id).ok_or_else(|| {
            Error::Incompatible(format!(
                "record {i} was written under schema {schema_id:#034x}, \
                 which does not resolve into reader schema {:#034x}",
                self.reader_id
            ))
        })
    }

    /// How many distinct writer schemas resolved successfully.
    pub fn len(&self) -> usize {
        self.by_schema.len()
    }

    pub fn is_empty(&self) -> bool {
        self.by_schema.is_empty()
    }
}

// ---------------------------------------------------------------------------
// Reading — owning the bytes
// ---------------------------------------------------------------------------

/// A `.verit` file read into memory, owning its bytes.
///
/// [`FileView`] borrows the image it reads, which means it cannot be stored in
/// the same struct as the buffer it points into — so every caller ends up
/// writing the same `{ bytes: Vec<u8> }` wrapper. This is that wrapper, once.
///
/// ```no_run
/// # use verit_core::file::FileReader;
/// let file = FileReader::open("events.verit")?;
/// let view = file.view()?;
/// for i in 0..view.len() {
///     println!("{}", view.dump_json(i)?);
/// }
/// # Ok::<(), verit_core::Error>(())
/// ```
///
/// For a large file, `mmap` it with your platform's facility and hand the
/// mapped `&[u8]` to [`FileView::open`] directly — the reader never copies a
/// record either way, and a map avoids reading bytes nothing touches.
#[derive(Clone, Debug)]
pub struct FileReader {
    bytes: Vec<u8>,
}

impl FileReader {
    /// Read and validate a file. Structural errors surface here rather than at
    /// first access.
    pub fn open<P: AsRef<Path>>(path: P) -> Result<FileReader> {
        FileReader::from_bytes(std::fs::read(path)?)
    }

    /// Take ownership of an already-read image, validating it.
    pub fn from_bytes(bytes: Vec<u8>) -> Result<FileReader> {
        FileView::open(&bytes)?;
        Ok(FileReader { bytes })
    }

    /// A zero-copy view over the owned bytes.
    pub fn view(&self) -> Result<FileView<'_>> {
        FileView::open(&self.bytes)
    }

    /// The raw image.
    pub fn bytes(&self) -> &[u8] {
        &self.bytes
    }
}

// ---------------------------------------------------------------------------
// Writing — in memory
// ---------------------------------------------------------------------------

/// Build a complete generation-1 `.verit` file in memory.
///
/// No I/O, deterministic output: the same records added in the same order
/// always produce the same bytes. That makes it the reference writer the golden
/// file corpus and the ports' conformance suites are checked against
/// (spec §11.5). For incremental, crash-safe mutation of a file on disk, use
/// [`FileWriter`].
pub struct FileBuilder {
    buf: Vec<u8>,
    index: Vec<Record>,
    registry: SchemaRegistry,
    next_id: u64,
    /// One CRC-32 per record, when [`with_record_checksums`] was called.
    ///
    /// [`with_record_checksums`]: FileBuilder::with_record_checksums
    checksums: Option<Vec<u32>>,
}

impl Default for FileBuilder {
    fn default() -> FileBuilder {
        FileBuilder::new()
    }
}

impl FileBuilder {
    pub fn new() -> FileBuilder {
        FileBuilder {
            buf: header_bytes(0).to_vec(),
            index: Vec::new(),
            registry: SchemaRegistry::new(),
            next_id: FIRST_RECORD_ID,
            checksums: None,
        }
    }

    /// Record a CRC-32 per record, setting [`OPT_RECORD_CRC`] in the header.
    ///
    /// The footer's CRC proves a *commit* was not torn; it says nothing about
    /// the record bytes. For a file meant to be read years from now, this is
    /// the difference between detecting bit rot and trusting it. Costs four
    /// bytes per record, and readers without the feature are unaffected.
    ///
    /// Must be called before the first record.
    pub fn with_record_checksums(mut self) -> FileBuilder {
        debug_assert!(self.index.is_empty(), "call before appending");
        self.buf = header_bytes(OPT_RECORD_CRC).to_vec();
        self.checksums = Some(Vec::new());
        self
    }

    /// Encode `value` against `schema` and append it as the next record,
    /// returning its **record id**. The message is written **hash-only** — the
    /// schema goes to the file's schema section once, however many records use
    /// it.
    pub fn append(&mut self, schema: &Schema, value: &Value) -> Result<u64> {
        let bytes = encode(schema, value, SchemaMode::HashOnly)?;
        self.append_message(schema, &bytes)
    }

    /// Append an already-encoded message together with its writer schema,
    /// returning its record id. Accepts hash-only and inline-schema messages
    /// alike; the schema id in the message must match `schema`.
    pub fn append_message(&mut self, schema: &Schema, bytes: &[u8]) -> Result<u64> {
        let id = self.next_id;
        self.append_message_with_id(schema, bytes, id)
    }

    /// Append a record under an explicit id — the compaction path, where ids
    /// must be **preserved**, not reassigned. Ids must be handed over strictly
    /// ascending.
    pub fn append_message_with_id(
        &mut self,
        schema: &Schema,
        bytes: &[u8],
        id: u64,
    ) -> Result<u64> {
        if id < self.next_id {
            return Err(Error::BadFile("record ids must be strictly ascending"));
        }
        let msg = Message::parse(bytes)?;
        if msg.schema_id() != schema.id() {
            return Err(Error::SchemaIdMismatch {
                message: msg.schema_id(),
                expected: schema.id(),
            });
        }
        let schema_id = self.registry.register(schema.clone());
        self.push_bytes(bytes, schema_id, id)
    }

    /// Append a self-describing (inline-schema) message, lifting its schema out
    /// of the message itself. The record is stored verbatim, inline schema and
    /// all — use [`append_message`](Self::append_message) to store it hash-only.
    pub fn append_self_describing(&mut self, bytes: &[u8]) -> Result<u64> {
        let schema = schema_of(bytes)?;
        let schema_id = self.registry.register(schema);
        let id = self.next_id;
        self.push_bytes(bytes, schema_id, id)
    }

    fn push_bytes(&mut self, bytes: &[u8], schema_id: u128, id: u64) -> Result<u64> {
        let offset = align_up(self.buf.len() as u64)?;
        self.buf.resize(offset as usize, 0); // 8-align the record start
        self.buf.extend_from_slice(bytes);
        if let Some(c) = &mut self.checksums {
            c.push(crc32(bytes));
        }
        self.index.push(Record {
            id,
            offset,
            length: bytes.len() as u64,
            schema_id,
        });
        self.next_id = id
            .checked_add(1)
            .ok_or(Error::BadFile("record id space exhausted"))?;
        Ok(id)
    }

    /// Ensure the finished file's `next_record_id` is at least `n`.
    ///
    /// Compaction uses this to carry a writer's id counter across a rewrite:
    /// removing the highest-id record must not let that id be handed out again.
    pub fn reserve_next_record_id(&mut self, n: u64) -> &mut Self {
        self.next_id = self.next_id.max(n);
        self
    }

    /// Number of records added so far.
    pub fn len(&self) -> usize {
        self.index.len()
    }

    pub fn is_empty(&self) -> bool {
        self.index.is_empty()
    }

    /// Finish the file: append the schema section, the index, and the
    /// generation-1 footer, and return the complete image.
    pub fn finish(mut self) -> Result<Vec<u8>> {
        let tail_start = align_up(self.buf.len() as u64)?;
        self.buf.resize(tail_start as usize, 0);
        let (tail, footer) = commit_tail(
            &self.index,
            &self.registry,
            tail_start,
            1,
            self.next_id,
            self.checksums.as_deref(),
        )?;
        self.buf.extend_from_slice(&tail);
        self.buf.extend_from_slice(&footer.encode());
        debug_assert_eq!(self.buf.len() as u64, footer.file_len);
        Ok(self.buf)
    }
}

/// Synchronise the directory holding `path`, so a rename into it is durable.
///
/// A rename is atomic, but the *directory entry* it creates lives in the
/// filesystem's own metadata and is not covered by `fsync` on the file. Without
/// this, a power loss just after `compact` can leave the old file in place —
/// never a torn mix, but a silently undone compaction.
///
/// Unix only. Windows has no equivalent of opening a directory as a file, and
/// `ReplaceFile`-style durability is not reachable from portable `std`; there
/// this is a no-op, which is why the guarantee is stated per-platform.
fn sync_parent_dir(path: &Path) -> Result<()> {
    #[cfg(unix)]
    {
        let dir = path.parent().unwrap_or_else(|| Path::new("."));
        // A directory opened read-only is enough to fsync it on Unix.
        File::open(dir)?.sync_all()?;
    }
    #[cfg(not(unix))]
    {
        let _ = path;
    }
    Ok(())
}

/// An advisory exclusive lock on a `.verit` file, held for a writer's lifetime.
///
/// The format specifies one writer and many readers (§7.3) and deliberately has
/// no in-format locking scheme — that would drag it toward the complexity
/// ADR-0002 §3 declined. This is the documented protocol instead: a sibling
/// `<path>.lock` file created exclusively, removed on drop.
///
/// **Advisory, not enforced.** It stops a second [`FileWriter`] that also asks
/// for the lock; it cannot stop a process that writes the file directly. A
/// stale lock left by a killed process must be removed by hand, which is the
/// honest trade — silently stealing a lock after a timeout would turn a visible
/// operational problem into a corrupted file.
#[derive(Debug)]
struct WriterLock {
    path: PathBuf,
}

impl WriterLock {
    fn acquire(target: &Path) -> Result<WriterLock> {
        let mut lock = target.as_os_str().to_os_string();
        lock.push(".lock");
        let path = PathBuf::from(lock);
        match OpenOptions::new().write(true).create_new(true).open(&path) {
            Ok(_) => Ok(WriterLock { path }),
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                Err(Error::AlreadyLocked(path.display().to_string()))
            }
            Err(e) => Err(Error::Io(e.to_string())),
        }
    }
}

impl Drop for WriterLock {
    fn drop(&mut self) {
        // Best effort: a failure here leaves a stale lock, which is visible and
        // fixable. Panicking in a destructor would be worse.
        let _ = std::fs::remove_file(&self.path);
    }
}

// ---------------------------------------------------------------------------
// Writing — on disk, incrementally
// ---------------------------------------------------------------------------

/// Where a record's bytes are: already committed at a known offset, or staged
/// in memory awaiting the next commit. Its id is assigned at staging time, so a
/// caller can checkpoint against it before the commit lands.
enum Slot {
    Committed {
        id: u64,
        offset: u64,
        length: u64,
        schema_id: u128,
        /// Carried in memory so a commit never re-reads committed records to
        /// recompute what it already knows.
        crc: u32,
    },
    Staged {
        id: u64,
        staged: usize,
        schema_id: u128,
        crc: u32,
    },
}

impl Slot {
    fn id(&self) -> u64 {
        match self {
            Slot::Committed { id, .. } | Slot::Staged { id, .. } => *id,
        }
    }

    fn crc(&self) -> u32 {
        match self {
            Slot::Committed { crc, .. } | Slot::Staged { crc, .. } => *crc,
        }
    }

    fn schema_id(&self) -> u128 {
        match self {
            Slot::Committed { schema_id, .. } | Slot::Staged { schema_id, .. } => *schema_id,
        }
    }
}

/// A `.verit` file open for reading and writing, mutated by crash-safe
/// append-only commits.
///
/// [`append`](Self::append) and the removal methods stage changes;
/// [`commit`](Self::commit) makes them durable and atomic. Nothing is visible to
/// a reader until a commit completes, and a crash mid-commit rolls the file back
/// to the previous generation — so a batch of appends and removals is a
/// transaction.
///
/// Records are addressed by **id**, not position: `append` returns the id it
/// assigned, and [`remove_id`](Self::remove_id) is the safe removal. Positional
/// [`remove`](Self::remove) exists but renumbers everything after it.
///
/// **One writer at a time.** Concurrent writers to the same path are undefined;
/// mutual exclusion is the caller's job. Concurrent *readers* need no
/// coordination at all.
pub struct FileWriter {
    file: File,
    path: PathBuf,
    /// Held for this writer's lifetime when the caller asked to lock. `None`
    /// keeps the historical behaviour: unlocked, and concurrent writers are the
    /// caller's problem exactly as §7.3 says.
    lock: Option<WriterLock>,
    generation: u64,
    committed_len: u64,
    next_record_id: u64,
    slots: Vec<Slot>,
    staged: Vec<Vec<u8>>,
    registry: SchemaRegistry,
    /// Whether this file carries a CRC-32 per record — fixed at creation, since
    /// the header is written once and never modified.
    checksums: bool,
}

impl FileWriter {
    /// Create a new file, truncating any existing one, and commit an empty
    /// generation 1 so the path is immediately a valid `.verit` file.
    pub fn create<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
        FileWriter::create_with(path, 0)
    }

    /// Create a file that records a CRC-32 per record ([`OPT_RECORD_CRC`]).
    ///
    /// The footer's CRC proves a *commit* was not torn; it says nothing about
    /// the record bytes. For a file meant to be read years from now, this is
    /// the difference between detecting bit rot and trusting it. Readers
    /// without the feature are unaffected.
    pub fn create_checksummed<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
        FileWriter::create_with(path, OPT_RECORD_CRC)
    }

    fn create_with<P: AsRef<Path>>(path: P, optional_features: u32) -> Result<FileWriter> {
        let path = path.as_ref().to_path_buf();
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(&path)?;
        file.write_all(&header_bytes(optional_features))?;

        let mut w = FileWriter {
            file,
            path,
            lock: None,
            generation: 0,
            committed_len: FILE_HEADER_LEN as u64,
            next_record_id: FIRST_RECORD_ID,
            slots: Vec::new(),
            staged: Vec::new(),
            registry: SchemaRegistry::new(),
            checksums: optional_features & OPT_RECORD_CRC != 0,
        };
        w.commit()?;
        Ok(w)
    }

    /// Open an existing file, recovering the newest valid commit (spec §7.2).
    /// Any uncommitted tail left by a crash is left in place until the next
    /// commit truncates it, so opening never destroys evidence.
    pub fn open<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
        let path = path.as_ref().to_path_buf();
        let mut file = OpenOptions::new().read(true).write(true).open(&path)?;

        let mut image = Vec::new();
        file.read_to_end(&mut image)?;
        let view = FileView::open(&image)?;

        let checksums = view.has_record_checksums();
        let slots = (0..view.len())
            .map(|i| {
                let r = view.record(i).expect("index validated in open");
                Slot::Committed {
                    id: r.id,
                    offset: r.offset,
                    length: r.length,
                    schema_id: r.schema_id,
                    // Recovered from the file rather than recomputed, so
                    // reopening never re-reads every record.
                    crc: view.record_checksum(i).unwrap_or(0),
                }
            })
            .collect();

        Ok(FileWriter {
            lock: None,
            generation: view.generation(),
            committed_len: view.file_len(),
            next_record_id: view.next_record_id(),
            registry: view.schemas().clone(),
            slots,
            staged: Vec::new(),
            checksums,
            file,
            path,
        })
    }

    /// Open a file **and take an advisory exclusive lock** on it, refusing if
    /// another locking writer already holds it.
    ///
    /// The format allows one writer and many readers (§7.3), and deliberately
    /// has no in-format locking scheme. This is the documented protocol
    /// instead: a sibling `<path>.lock`, held until this writer is dropped.
    ///
    /// It is **advisory**. It stops another `FileWriter` that also locks; it
    /// cannot stop a process that ignores the convention. A lock left behind by
    /// a killed process must be removed by hand — stealing it after a timeout
    /// would turn a visible operational problem into a corrupted file.
    ///
    /// Readers never need this, and never block.
    pub fn open_locked<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
        let path = path.as_ref();
        let lock = WriterLock::acquire(path)?;
        let mut w = FileWriter::open(path)?;
        w.lock = Some(lock);
        Ok(w)
    }

    /// [`open_locked`](Self::open_locked), creating the file if it is absent.
    pub fn open_or_create_locked<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
        let path = path.as_ref();
        let lock = WriterLock::acquire(path)?;
        let mut w = FileWriter::open_or_create(path)?;
        w.lock = Some(lock);
        Ok(w)
    }

    /// Whether this writer holds the advisory lock.
    pub fn is_locked(&self) -> bool {
        self.lock.is_some()
    }

    /// Open the file if it exists, otherwise create it.
    pub fn open_or_create<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
        let path = path.as_ref();
        if path.exists() {
            FileWriter::open(path)
        } else {
            FileWriter::create(path)
        }
    }

    /// Stage `value` as a new record at the end, returning the **record id** it
    /// was assigned. Takes effect on [`commit`](Self::commit); the id is stable
    /// from this moment and is what a consumer should checkpoint against.
    pub fn append(&mut self, schema: &Schema, value: &Value) -> Result<u64> {
        let bytes = encode(schema, value, SchemaMode::HashOnly)?;
        self.append_message(schema, &bytes)
    }

    /// Stage an already-encoded message with its writer schema.
    pub fn append_message(&mut self, schema: &Schema, bytes: &[u8]) -> Result<u64> {
        let msg = Message::parse(bytes)?;
        if msg.schema_id() != schema.id() {
            return Err(Error::SchemaIdMismatch {
                message: msg.schema_id(),
                expected: schema.id(),
            });
        }
        let schema_id = self.registry.register(schema.clone());
        self.stage(bytes.to_vec(), schema_id)
    }

    /// Stage a self-describing (inline-schema) message, lifting its schema out
    /// of the message itself.
    pub fn append_self_describing(&mut self, bytes: &[u8]) -> Result<u64> {
        let schema = schema_of(bytes)?;
        let schema_id = self.registry.register(schema);
        self.stage(bytes.to_vec(), schema_id)
    }

    fn stage(&mut self, bytes: Vec<u8>, schema_id: u128) -> Result<u64> {
        let id = self.next_record_id;
        self.next_record_id = id
            .checked_add(1)
            .ok_or(Error::BadFile("record id space exhausted"))?;
        let crc = if self.checksums { crc32(&bytes) } else { 0 };
        self.staged.push(bytes);
        self.slots.push(Slot::Staged {
            id,
            staged: self.staged.len() - 1,
            schema_id,
            crc,
        });
        Ok(id)
    }

    /// Stage the removal of the record with this id — the safe removal, since
    /// an id does not shift when its neighbours go away.
    ///
    /// **Removal unlinks; it does not erase** (spec §8.2). The record's bytes
    /// stay in the file and remain fully recoverable with a hex editor until
    /// [`compact`](Self::compact) rewrites it. Use [`purge_ids`](Self::purge_ids)
    /// when the data actually has to go.
    pub fn remove_id(&mut self, id: u64) -> Result<&mut Self> {
        let at = self
            .slots
            .iter()
            .position(|s| s.id() == id)
            .ok_or(Error::IndexOutOfBounds)?;
        self.slots.remove(at);
        Ok(self)
    }

    /// Stage the removal of every listed id, returning how many were live.
    /// Unknown ids are ignored, so this is idempotent and safe to retry.
    pub fn remove_ids(&mut self, ids: &[u64]) -> usize {
        let doomed: HashSet<u64> = ids.iter().copied().collect();
        let before = self.slots.len();
        self.slots.retain(|s| !doomed.contains(&s.id()));
        before - self.slots.len()
    }

    /// Stage the removal of the record at position `i`.
    ///
    /// Prefer [`remove_id`](Self::remove_id): positions shift, so removing
    /// record `i` renumbers everything after it, and a loop over positions is
    /// an off-by-one waiting to happen. Same non-erasure caveat as
    /// [`remove_id`](Self::remove_id).
    pub fn remove(&mut self, i: usize) -> Result<&mut Self> {
        if i >= self.slots.len() {
            return Err(Error::IndexOutOfBounds);
        }
        self.slots.remove(i);
        Ok(self)
    }

    /// Keep only the records for which `keep(id, schema_id)` is true, returning
    /// how many were removed. The bulk removal that cannot drift by one.
    ///
    /// Deciding from a record's *contents* needs its bytes, which this does not
    /// hand you: read what you need with [`read_record`](Self::read_record)
    /// first, collect the doomed ids, then call
    /// [`remove_ids`](Self::remove_ids).
    pub fn retain<F: FnMut(u64, u128) -> bool>(&mut self, mut keep: F) -> usize {
        let before = self.slots.len();
        self.slots.retain(|s| keep(s.id(), s.schema_id()));
        before - self.slots.len()
    }

    /// Remove every listed id **and erase it** — one commit, then one
    /// compaction. Returns how many records were removed.
    ///
    /// This is the call to reach for when the data genuinely has to go
    /// (regulated or secret content), because plain removal only unlinks.
    /// Batched on purpose: compaction rewrites the whole file, so doing it per
    /// removal would be `O(file)` each time.
    ///
    /// Erasure covers *this* file only. Backups, snapshots, and unallocated
    /// disk blocks from before the compaction are outside its reach.
    pub fn purge_ids(&mut self, ids: &[u64]) -> Result<usize> {
        let removed = self.remove_ids(ids);
        self.commit()?;
        self.compact()?;
        Ok(removed)
    }

    /// Live record count, including staged-but-uncommitted changes.
    pub fn len(&self) -> usize {
        self.slots.len()
    }

    pub fn is_empty(&self) -> bool {
        self.slots.is_empty()
    }

    /// Number of staged records not yet committed.
    pub fn pending(&self) -> usize {
        self.staged.len()
    }

    /// The generation of the last completed commit.
    pub fn generation(&self) -> u64 {
        self.generation
    }

    /// The id the next appended record will take.
    pub fn next_record_id(&self) -> u64 {
        self.next_record_id
    }

    /// The ids of every live record, in order.
    pub fn ids(&self) -> impl Iterator<Item = u64> + '_ {
        self.slots.iter().map(|s| s.id())
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Make every staged change durable and atomic, per spec §7.1.
    ///
    /// Truncate the uncommitted tail, append the new records, append the schema
    /// section and index, **synchronise**, then append the footer and
    /// synchronise again. The intermediate sync is load-bearing: it is what
    /// guarantees that a durable footer never points at records that never
    /// reached the disk.
    pub fn commit(&mut self) -> Result<u64> {
        // 1. Discard any uncommitted tail from an earlier crash.
        self.file.set_len(self.committed_len)?;
        self.file.seek(SeekFrom::Start(self.committed_len))?;

        // 2. Place staged records, 8-aligned, after the last committed byte.
        let mut cursor = self.committed_len;
        let mut body = Vec::new();
        let mut index = Vec::with_capacity(self.slots.len());
        for slot in &self.slots {
            match slot {
                Slot::Committed {
                    id,
                    offset,
                    length,
                    schema_id,
                    ..
                } => index.push(Record {
                    id: *id,
                    offset: *offset,
                    length: *length,
                    schema_id: *schema_id,
                }),
                Slot::Staged {
                    id,
                    staged,
                    schema_id,
                    ..
                } => {
                    let bytes = &self.staged[*staged];
                    let offset = align_up(cursor)?;
                    body.resize((offset - self.committed_len) as usize, 0);
                    body.extend_from_slice(bytes);
                    cursor = offset + bytes.len() as u64;
                    index.push(Record {
                        id: *id,
                        offset,
                        length: bytes.len() as u64,
                        schema_id: *schema_id,
                    });
                }
            }
        }

        // 3-4. Schema section and index, both rewritten in full.
        let tail_start = align_up(cursor)?;
        body.resize((tail_start - self.committed_len) as usize, 0);
        let generation = self.generation + 1;
        let crcs: Option<Vec<u32>> = if self.checksums {
            Some(self.slots.iter().map(|s| s.crc()).collect())
        } else {
            None
        };
        let (tail, footer) = commit_tail(
            &index,
            &self.registry,
            tail_start,
            generation,
            self.next_record_id,
            crcs.as_deref(),
        )?;
        body.extend_from_slice(&tail);

        self.file.write_all(&body)?;
        // 5. Everything the footer will point at is durable *before* the footer
        //    exists. Skipping this is the one way this design corrupts a file.
        self.file.sync_all()?;

        // 6-7. The footer is the commit point.
        self.file.write_all(&footer.encode())?;
        self.file.sync_all()?;

        // The commit succeeded: staged records are now committed at known
        // offsets, and the pruned registry is the file's live schema set.
        let kept: Vec<u32> = self.slots.iter().map(|s| s.crc()).collect();
        self.slots = index
            .iter()
            .zip(kept)
            .map(|(r, crc)| Slot::Committed {
                id: r.id,
                offset: r.offset,
                length: r.length,
                schema_id: r.schema_id,
                crc,
            })
            .collect();
        self.staged.clear();
        self.generation = generation;
        self.committed_len = footer.file_len;
        Ok(generation)
    }

    /// Rewrite the file with only its live records, at generation 1 — the only
    /// operation that reclaims space, and the only one that actually **erases**
    /// removed records (spec §8.3).
    ///
    /// Record **ids are preserved**, so a consumer's checkpoint stays valid
    /// across a compaction; positions are not. The writer's id counter is
    /// carried over too, so an id belonging to a purged record is never reissued.
    ///
    /// Performed out of place: a fresh file is written and synchronised, then
    /// atomically renamed over the original, so a crash during compaction leaves
    /// the original intact and readable. Staged-but-uncommitted changes are
    /// committed first.
    ///
    /// The rename is atomic, and on Unix the containing directory is
    /// synchronised afterwards so the rename itself survives a power loss. On
    /// other platforms that step is a no-op and the rename may be lost — the
    /// old file survives in that case, never a torn mix of the two.
    pub fn compact(&mut self) -> Result<()> {
        if !self.staged.is_empty() {
            self.commit()?;
        }

        let mut builder = if self.checksums {
            FileBuilder::new().with_record_checksums()
        } else {
            FileBuilder::new()
        };
        for i in 0..self.slots.len() {
            let bytes = self.read_record(i)?;
            let id = self.slots[i].id();
            let schema_id = self.slots[i].schema_id();
            let schema = self
                .registry
                .get(schema_id)
                .ok_or(Error::MissingSchema(schema_id))?
                .clone();
            builder.append_message_with_id(&schema, &bytes, id)?;
        }
        // Carry the counter across, so a purged record's id is never reissued.
        builder.reserve_next_record_id(self.next_record_id);
        let image = builder.finish()?;

        let mut tmp = self.path.clone().into_os_string();
        tmp.push(".compact");
        let tmp = PathBuf::from(tmp);
        {
            let mut f = OpenOptions::new()
                .read(true)
                .write(true)
                .create(true)
                .truncate(true)
                .open(&tmp)?;
            f.write_all(&image)?;
            f.sync_all()?;
        }
        std::fs::rename(&tmp, &self.path)?;
        sync_parent_dir(&self.path)?;

        // Reopening rebuilds our state from the new file; carry the lock across
        // rather than dropping it mid-compaction and letting another writer in.
        let held = self.lock.take();
        *self = FileWriter::open(&self.path)?;
        self.lock = held;
        Ok(())
    }

    /// Read record `i`'s bytes from disk. Committed records are read at their
    /// offset; staged ones come straight from memory.
    pub fn read_record(&mut self, i: usize) -> Result<Vec<u8>> {
        match self.slots.get(i).ok_or(Error::IndexOutOfBounds)? {
            Slot::Staged { staged, .. } => Ok(self.staged[*staged].clone()),
            Slot::Committed { offset, length, .. } => {
                let (offset, length) = (*offset, *length as usize);
                let mut buf = vec![0u8; length];
                self.file.seek(SeekFrom::Start(offset))?;
                self.file.read_exact(&mut buf)?;
                Ok(buf)
            }
        }
    }

    /// Read the bytes of the record with this id.
    pub fn read_record_by_id(&mut self, id: u64) -> Result<Vec<u8>> {
        let at = self
            .slots
            .iter()
            .position(|s| s.id() == id)
            .ok_or(Error::IndexOutOfBounds)?;
        self.read_record(at)
    }

    /// Read the whole file image back, for handing to [`FileView`].
    pub fn image(&mut self) -> Result<Vec<u8>> {
        let mut buf = Vec::new();
        self.file.seek(SeekFrom::Start(0))?;
        self.file.read_to_end(&mut buf)?;
        Ok(buf)
    }
}