blockworx_store/

manifest.rs

//! `manifest.jsonl`: one row per rev, and one per tag.
//! Rationale: `docs/log-vs-snapshot.md` §10.1 and §14.2.
//!
//! The document is `revs/{head}`; this file is the audit trail beside it.
//! **A row carries names and circumstances, never document values** — who
//! wrote it, when, where they were standing, what they were looking at,
//! and which entities they named. Nothing folds a row, so nothing here can
//! drift out from under a later build the way a replayed op could.
//!
//! The chain survives the log it was written for: each row links to the
//! canonical bytes of the one before it, and stamps the blake3 of the rev
//! file it names. Both are hashes over bytes somebody else wrote, so the
//! whole check costs one pass over a small file
//! (`TUNING.md`, Finding 9).
//!
//! Target-independent: a scratch session builds the same rows in memory,
//! and the browser reads the same text out of origin storage.

use blockworx_doc::{
    id::EntityRef,
    rev::Rev,
    trail::{JournalAs, Trail},
};
use serde::{Deserialize, Serialize};

use super::record::{Camera, Digest, Identity, ScopePath, WallTime};
use super::tags::{Tagging, Tags};

/// How many entities a row names before it gives up and says so. Beyond it
/// the reader's recourse is a diff of the two revs, which is an acceptable
/// answer for a grep index and would not have been for a camera.
pub const TOUCHED: usize = 64;

/// Why a row was written. An undo is a forward record, so the manifest is
/// a complete audit trail rather than an erasure — and this says which of
/// its neighbours it took back.
///
/// Tags are pinned: serde ties them to variant names, so a rename would
/// silently rewrite the durable format.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
pub enum RowKind {
    #[serde(rename = "edit")]
    Edit,
    #[serde(rename = "undo")]
    Undo { of: Rev },
    #[serde(rename = "redo")]
    Redo { of: Rev },
    /// A human-readable name for the rev this row's `rev` field
    /// *names* — the one kind whose `rev` is a reference rather than a
    /// position. It consumes no rev and never reaches the trail.
    #[serde(rename = "tag")]
    Tag,
    /// The same reference, taking one name back off. A rev has a set of
    /// tags, so removal names which one.
    #[serde(rename = "untag")]
    Untag,
}

/// What replaying one row does. One method rather than an `Option` per
/// question, so a row cannot answer that it both moves the trail and
/// names a rev.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Replays {
    /// A rev row: it takes a position and moves the trail.
    Trail(JournalAs),
    /// A tag row: it consumes no rev and moves the vocabulary.
    Vocabulary(Tagging),
}

impl RowKind {
    /// What replaying this row does — the boundary the durable spelling
    /// stops at, so [`Trail`] never hears about a file.
    ///
    /// A tag is *about* history rather than part of it, so `Ctrl+Z` after
    /// tagging still takes back the last edit.
    pub fn replays(self) -> Replays {
        match self {
            RowKind::Edit => Replays::Trail(JournalAs::Edit),
            RowKind::Undo { of } => Replays::Trail(JournalAs::Undo { of }),
            RowKind::Redo { of } => Replays::Trail(JournalAs::Redo { of }),
            RowKind::Tag => Replays::Vocabulary(Tagging::Added),
            RowKind::Untag => Replays::Vocabulary(Tagging::Removed),
        }
    }

    /// Whether this row takes a rev of its own, or names one already
    /// taken.
    pub fn takes_a_rev(self) -> bool {
        matches!(self.replays(), Replays::Trail(_))
    }

    /// The rev this row steps, where it steps one.
    pub fn steps(self) -> Option<Rev> {
        match self {
            RowKind::Undo { of } | RowKind::Redo { of } => Some(of),
            RowKind::Edit | RowKind::Tag | RowKind::Untag => None,
        }
    }
}

/// One line of `manifest.jsonl`.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct Row {
    pub rev: Rev,
    pub kind: RowKind,
    pub wall_time: WallTime,
    pub author: Identity,
    pub label: String,
    /// Where the author was standing: every scope is a coordinate space of
    /// its own, so a camera without one means nothing.
    pub scope: ScopePath,
    /// The same path as its author saw it spelled, outermost first.
    /// Recorded at seal beside the ids rather than resolved at read time,
    /// so a row still reads its own path after the blocks along it are
    /// renamed or deleted. Empty at the root.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub scope_names: Vec<String>,
    pub camera: Camera,
    /// The entities the act named, in `EntityRef`'s narration spelling —
    /// the grep index, advisory in the same sense `scope` has always been.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub touched: Vec<EntityRef>,
    /// Whether [`TOUCHED`] cut the list short.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub truncated: bool,
    /// The blake3 of the rev file this row names. A tag has no rev file of
    /// its own and stamps the head it was appended under, which is what
    /// makes a tag written against another history detectable.
    pub hash: Digest,
    /// The digest of the previous row's canonical bytes, or
    /// [`Digest::genesis`] for the first.
    pub parent: Digest,
}

impl Row {
    /// The bytes this row is written as, and the bytes its successor's
    /// `parent` hashes over: `serde_json` with object keys sorted and no
    /// whitespace. One definition for both, so a round trip through the
    /// file cannot break the chain that verifies it.
    // A row is plain data: no map keys that are not strings, which is the
    // only way `to_value` refuses a value.
    #[expect(clippy::expect_used, clippy::missing_panics_doc)]
    pub fn canonical_bytes(&self) -> Vec<u8> {
        let value = serde_json::to_value(self).expect("a manifest row serializes infallibly");
        let mut bytes = Vec::new();
        write_canonical(&value, &mut bytes).expect("writing to a vector is infallible");
        bytes
    }

    /// The link the next row carries as its `parent`.
    pub fn digest(&self) -> Digest {
        Digest::of(&self.canonical_bytes())
    }

    /// The same link, computed against the line a file actually holds, so
    /// a row an older build spelled differently still links as it did.
    // Re-serialization is infallible for a value that just parsed.
    #[expect(clippy::expect_used, clippy::missing_panics_doc)]
    pub fn digest_as_written(&self, line: &str) -> Digest {
        let canonical = self.canonical_bytes();
        if canonical == line.as_bytes() {
            return Digest::of(&canonical);
        }
        let value: serde_json::Value =
            serde_json::from_str(line).expect("the line parsed as a row");
        let mut bytes = Vec::new();
        write_canonical(&value, &mut bytes).expect("writing to a vector is infallible");
        Digest::of(&bytes)
    }

    /// The names of the entities `touched` holds, cut to [`TOUCHED`].
    pub fn naming(mut touched: Vec<EntityRef>) -> (Vec<EntityRef>, bool) {
        let truncated = touched.len() > TOUCHED;
        touched.truncate(TOUCHED);
        (touched, truncated)
    }
}

/// Sorted-key, whitespace-free JSON. Written out rather than delegated to
/// `serde_json::to_vec` over a `Value`, whose key order is the `Map`
/// backing `serde_json` happens to be compiled with — a cargo feature
/// elsewhere in the tree must not be able to change what a manifest hashes
/// to.
pub(crate) fn write_canonical(
    value: &serde_json::Value,
    out: &mut Vec<u8>,
) -> serde_json::Result<()> {
    match value {
        serde_json::Value::Object(entries) => {
            let mut keys: Vec<&String> = entries.keys().collect();
            keys.sort_unstable();
            out.push(b'{');
            for (position, key) in keys.into_iter().enumerate() {
                if position > 0 {
                    out.push(b',');
                }
                serde_json::to_writer(&mut *out, key)?;
                out.push(b':');
                write_canonical(&entries[key], out)?;
            }
            out.push(b'}');
        }
        serde_json::Value::Array(items) => {
            out.push(b'[');
            for (position, item) in items.iter().enumerate() {
                if position > 0 {
                    out.push(b',');
                }
                write_canonical(item, out)?;
            }
            out.push(b']');
        }
        leaf => serde_json::to_writer(out, leaf)?,
    }
    Ok(())
}

/// Where in the file something went wrong — enough to render a span-style
/// report without re-reading it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Located {
    /// 1-based.
    pub line: usize,
    /// 1-based, in bytes from the start of the line.
    pub column: usize,
    /// Byte offset of the line's first byte from the start of the file.
    pub offset: usize,
    /// The line's length in bytes, its newline excluded.
    pub len: usize,
}

impl std::fmt::Display for Located {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "line {}, column {}", self.line, self.column)
    }
}

/// What a row failed on.
#[derive(Debug)]
pub enum Fault {
    Malformed(serde_json::Error),
    /// The row does not link to the one before it, so the file has
    /// been rewritten, reordered, or spliced at this point.
    Chained {
        expected: Digest,
        found: Digest,
    },
    /// The row claims a position the manifest did not reach.
    Misnumbered {
        expected: Rev,
        found: Rev,
    },
    /// A step names a rev no row before it holds.
    Unreached {
        of: Rev,
    },
    /// The rev file this row names is not there, or does not hash to what
    /// the row says it does.
    Unwitnessed {
        at: Rev,
        why: String,
    },
}

impl std::fmt::Display for Fault {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Fault::Malformed(error) => write!(f, "the row does not parse: {error}"),
            Fault::Chained { expected, found } => write!(
                f,
                "the row links to {found}, but the row before it hashes to {expected} — \
                 the manifest has been rewritten here",
            ),
            Fault::Misnumbered { expected, found } => write!(
                f,
                "the row claims rev {found}, but the manifest reaches {expected}",
                found = found.get(),
                expected = expected.get(),
            ),
            Fault::Unreached { of } => write!(
                f,
                "the row steps rev {of}, which no row before it holds",
                of = of.get(),
            ),
            Fault::Unwitnessed { at, why } => write!(
                f,
                "the row names rev {at}, whose file this container cannot show: {why}",
                at = at.get(),
            ),
        }
    }
}

/// A fault and where it is. Kept apart from the rows it interrupted so a
/// container can hold on to the report as its read-only reason without
/// also holding the history it came with.
#[derive(Debug)]
pub struct BreakReport {
    pub at: Located,
    pub fault: Fault,
}

impl std::fmt::Display for BreakReport {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.at, self.fault)
    }
}

/// A final row the writer never finished. Appends are one `write_all` of
/// the line and its newline followed by an fsync, so a manifest that does
/// not end in a newline ends in a write that was never made durable: the
/// row is dropped and the file truncated back to the last whole line.
#[derive(Clone, Copy, Debug)]
pub struct DroppedTail {
    pub at: Located,
    /// Where the whole manifest ends.
    pub good_bytes: u64,
}

/// Whether the manifest ended where it should have. Not a bool: a dropped
/// tail is something the user is told about.
#[derive(Clone, Copy, Debug)]
pub enum Tail {
    Whole,
    Dropped(DroppedTail),
}

/// One row and where the file holds it.
pub struct Verified {
    pub row: Row,
    pub at: Located,
    /// The link the next row must name as its `parent`.
    pub digest: Digest,
}

/// How the row sequence stopped.
pub enum End {
    Whole,
    Truncated(DroppedTail),
    Broken(BreakReport),
}

/// Parse and link, stopping at the first row that does not belong and
/// keeping everything before it.
pub struct Scan {
    pub rows: Vec<Verified>,
    pub end: End,
}

/// The history a prefix of rows makes: what the panel lists, what the
/// tags call the revs, and where the session that wrote them stood.
pub struct History {
    /// One per rev, in rev order — tag rows are projected into `tags`
    /// instead, since they consume no rev.
    pub rows: Vec<Row>,
    pub trail: Trail,
    pub tags: Tags,
    /// The digest the next appended row must name as its parent.
    pub head: Digest,
}

impl History {
    /// The rev the last row named, or [`Rev::ZERO`] for a manifest with no
    /// revs in it.
    pub fn rev(&self) -> Rev {
        self.rows.last().map_or(Rev::ZERO, |row| row.rev)
    }

    pub fn row(&self, rev: Rev) -> Option<&Row> {
        let at = usize::try_from(rev.get().checked_sub(1)?).ok()?;
        self.rows.get(at)
    }
}

/// Read `text` as a chain of rows, verifying every link, every position
/// and every step as it goes.
///
/// The chain is over the bytes the file holds — a hash of one line — so
/// the whole check survives the log, and it is cheap.
pub fn scan(text: &str) -> Scan {
    let mut rows: Vec<Verified> = Vec::new();
    let mut parent = Digest::genesis();
    let mut reached = Rev::ZERO;
    let mut offset = 0;

    for (number, chunk) in text.split_inclusive('\n').enumerate() {
        let line = chunk.strip_suffix('\n');
        let body = line.unwrap_or(chunk);
        let at = Located {
            line: number + 1,
            column: 1,
            offset,
            len: body.len(),
        };
        if line.is_none() {
            return Scan {
                rows,
                end: End::Truncated(DroppedTail {
                    at,
                    good_bytes: offset as u64,
                }),
            };
        }
        offset += chunk.len();

        let row: Row = match serde_json::from_str(body) {
            Ok(row) => row,
            Err(error) => {
                let at = Located {
                    column: error.column().max(1),
                    ..at
                };
                return broken(rows, at, Fault::Malformed(error));
            }
        };
        if row.parent != parent {
            let fault = Fault::Chained {
                expected: parent,
                found: row.parent,
            };
            return broken(rows, at, fault);
        }
        if let Some(fault) = misplaced(&row, reached) {
            return broken(rows, at, fault);
        }
        if row.kind.takes_a_rev() {
            reached = row.rev;
        }
        parent = row.digest_as_written(body);
        rows.push(Verified {
            row,
            at,
            digest: parent,
        });
    }

    Scan {
        rows,
        end: End::Whole,
    }
}

/// Whether `row` sits where the manifest can hold it: a rev row takes the
/// next position, a tag names one already taken, and a step names a rev
/// the trail can have reached.
fn misplaced(row: &Row, reached: Rev) -> Option<Fault> {
    if let Some(of) = row.kind.steps()
        && of > reached
    {
        return Some(Fault::Unreached { of });
    }
    if !row.kind.takes_a_rev() {
        // A tag names a rev the manifest already holds rather than taking
        // one, so its `rev` is checked as a reference instead.
        return (row.rev == Rev::ZERO || row.rev > reached).then_some(Fault::Misnumbered {
            expected: reached,
            found: row.rev,
        });
    }
    let expected = reached.next();
    (row.rev != expected).then_some(Fault::Misnumbered {
        expected,
        found: row.rev,
    })
}

fn broken(rows: Vec<Verified>, at: Located, fault: Fault) -> Scan {
    Scan {
        rows,
        end: End::Broken(BreakReport { at, fault }),
    }
}

/// The history `rows` make: the trail replayed through its own policy, the
/// tags projected, and the rows themselves in rev order.
///
/// One derivation, whether the rows were written this session or read off
/// a disk — which is what keeps a reopened document standing where it was
/// closed.
pub fn history(rows: &[Verified]) -> History {
    let mut trail = Trail::default();
    let mut tags = Tags::default();
    let mut kept = Vec::new();
    for verified in rows {
        let row = &verified.row;
        match row.kind.replays() {
            Replays::Vocabulary(how) => tags.apply(row.rev, &row.label, how),
            Replays::Trail(journal) => {
                trail.record(row.rev, journal);
                kept.push(row.clone());
            }
        }
    }
    History {
        rows: kept,
        trail,
        tags,
        head: rows.last().map_or_else(Digest::genesis, |last| last.digest),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use blockworx_doc::fixtures::{block_id, rev};

    fn row(rev: Rev, kind: RowKind, parent: Digest) -> Row {
        Row {
            rev,
            kind,
            wall_time: WallTime::from_unix_millis(1_756_000_000_000),
            author: Identity::new("ada"),
            label: "Added a block".to_owned(),
            scope: ScopePath::default(),
            scope_names: Vec::new(),
            camera: Camera::UNSEEN,
            touched: vec![EntityRef::Block(block_id(1))],
            truncated: false,
            hash: Digest::of(b"a rev file"),
            parent,
        }
    }

    /// `kinds` as a properly chained manifest.
    fn chained(kinds: &[(Rev, RowKind)]) -> String {
        let mut parent = Digest::genesis();
        let mut text = String::new();
        for (at, kind) in kinds {
            let row = row(*at, *kind, parent);
            parent = row.digest();
            text.push_str(&String::from_utf8(row.canonical_bytes()).expect("utf-8"));
            text.push('\n');
        }
        text
    }

    #[test]
    fn a_row_round_trips_through_its_own_canonical_bytes() {
        let written = row(rev(1), RowKind::Edit, Digest::genesis());
        let bytes = written.canonical_bytes();
        let parsed: Row = serde_json::from_slice(&bytes).expect("the row parses back");
        assert_eq!(parsed, written);
        assert_eq!(
            parsed.canonical_bytes(),
            bytes,
            "and re-canonicalizes to the same bytes, which is what the chain rests on",
        );
    }

    /// The chain is over the bytes a file holds, and canonicalization is
    /// what lets a line that says the same thing differently still link.
    #[test]
    fn a_reformatted_line_links_exactly_as_the_line_it_reformats() {
        let written = row(rev(1), RowKind::Edit, Digest::genesis());
        let canonical = String::from_utf8(written.canonical_bytes()).expect("utf-8");
        let pretty = serde_json::to_string_pretty(&written).expect("it re-serializes");
        assert_ne!(pretty, canonical, "precondition: different bytes");
        assert_eq!(
            written.digest_as_written(&pretty),
            written.digest_as_written(&canonical),
        );
        assert_eq!(written.digest_as_written(&canonical), written.digest());
    }

    /// The tags a rename must not be allowed to move.
    #[test]
    fn row_kinds_are_spelled_as_the_format_names_them() {
        let spelling = |kind: RowKind| serde_json::to_string(&kind).expect("a kind serializes");
        assert_eq!(spelling(RowKind::Edit), "\"edit\"");
        assert_eq!(
            spelling(RowKind::Undo { of: rev(3) }),
            r#"{"undo":{"of":3}}"#
        );
        assert_eq!(
            spelling(RowKind::Redo { of: rev(4) }),
            r#"{"redo":{"of":4}}"#
        );
        assert_eq!(spelling(RowKind::Tag), "\"tag\"");
        assert_eq!(spelling(RowKind::Untag), "\"untag\"");
        for kind in [RowKind::Tag, RowKind::Untag] {
            assert!(!kind.takes_a_rev(), "a tag row moves no stack: {kind:?}");
        }
    }

    /// Through the file: the rows replay into the trail the session that
    /// wrote them closed with, and a tag row leaves it alone.
    #[test]
    fn rows_replay_into_the_trail_and_the_tags() {
        let text = chained(&[
            (rev(1), RowKind::Edit),
            (rev(2), RowKind::Edit),
            (rev(2), RowKind::Tag),
            (rev(3), RowKind::Undo { of: rev(2) }),
        ]);
        let scanned = scan(&text);
        assert!(matches!(scanned.end, End::Whole), "the manifest chains");

        let history = history(&scanned.rows);
        assert_eq!(history.rev(), rev(3));
        assert_eq!(history.rows.len(), 3, "a tag consumes no rev");
        assert_eq!(history.tags.of(rev(2)), ["Added a block"]);
        assert_eq!(history.trail.standing(), rev(1), "the undo stepped back");
        assert_eq!(history.trail.next_redo(), Some(rev(3)));
        assert_eq!(history.row(rev(2)).map(|row| row.rev), Some(rev(2)));
    }

    #[test]
    fn a_rewritten_row_breaks_the_chain_at_its_successor() {
        let text = chained(&[(rev(1), RowKind::Edit), (rev(2), RowKind::Edit)]);
        let mut lines: Vec<String> = text.lines().map(str::to_owned).collect();
        lines[0] = lines[0].replace("Added a block", "Added a bloke");
        let scanned = scan(&(lines.join("\n") + "\n"));

        assert_eq!(scanned.rows.len(), 1, "the prefix before the break stands");
        let End::Broken(report) = scanned.end else {
            panic!("a rewritten row was accepted");
        };
        assert!(matches!(report.fault, Fault::Chained { .. }));
        assert_eq!(report.at.line, 2, "the successor is what detects it");
    }

    #[test]
    fn a_step_naming_a_rev_no_row_holds_is_refused() {
        let text = chained(&[
            (rev(1), RowKind::Edit),
            (rev(2), RowKind::Undo { of: rev(9) }),
        ]);
        let End::Broken(report) = scan(&text).end else {
            panic!("a step into nothing was accepted");
        };
        assert!(
            matches!(report.fault, Fault::Unreached { of } if of == rev(9)),
            "{}",
            report.fault,
        );
    }

    #[test]
    fn a_row_out_of_sequence_is_refused() {
        let text = chained(&[(rev(1), RowKind::Edit), (rev(3), RowKind::Edit)]);
        let End::Broken(report) = scan(&text).end else {
            panic!("a gap in the revs was accepted");
        };
        assert!(matches!(report.fault, Fault::Misnumbered { .. }));

        let unheld = chained(&[(rev(1), RowKind::Edit), (rev(4), RowKind::Tag)]);
        let End::Broken(report) = scan(&unheld).end else {
            panic!("a tag on a rev the manifest does not hold was accepted");
        };
        assert!(matches!(report.fault, Fault::Misnumbered { .. }));
    }

    /// An append is one write of the line and its newline, then an fsync,
    /// so a file that does not end in one ends in a write nobody made
    /// durable.
    #[test]
    fn a_partial_trailing_line_is_dropped_and_the_file_cut_back() {
        let whole = chained(&[(rev(1), RowKind::Edit), (rev(2), RowKind::Edit)]);
        let cut = whole.len() - 20;
        let scanned = scan(&whole[..cut]);
        assert_eq!(scanned.rows.len(), 1);
        let End::Truncated(dropped) = scanned.end else {
            panic!("a half-written row was taken");
        };
        assert_eq!(
            dropped.good_bytes as usize,
            whole.lines().next().expect("a first line").len() + 1,
        );
    }

    /// The cap on the touched list, and the flag that says it bit.
    #[test]
    fn a_touched_list_past_the_cap_is_cut_and_marked() {
        let many: Vec<EntityRef> = (1..=200).map(|n| EntityRef::Block(block_id(n))).collect();
        let (cut, truncated) = Row::naming(many);
        assert_eq!(cut.len(), TOUCHED);
        assert!(truncated);
        assert_eq!(cut[0], EntityRef::Block(block_id(1)), "the cut is a prefix");

        let (few, truncated) = Row::naming(vec![EntityRef::Document]);
        assert_eq!(few, [EntityRef::Document]);
        assert!(!truncated);
    }

    /// The invariant the line is held on: a row is names and
    /// circumstances. `{"x": 12}` in one and the log is back.
    #[test]
    fn a_row_carries_no_document_values() {
        let written = row(rev(1), RowKind::Edit, Digest::genesis());
        let text = String::from_utf8(written.canonical_bytes()).expect("utf-8");
        let fields: serde_json::Map<String, serde_json::Value> =
            serde_json::from_str(&text).expect("the row parses as an object");
        let mut keys: Vec<&str> = fields.keys().map(String::as_str).collect();
        keys.sort_unstable();
        assert_eq!(
            keys,
            [
                "author",
                "camera",
                "hash",
                "kind",
                "label",
                "parent",
                "rev",
                "scope",
                "touched",
                "wall_time",
            ],
        );
        assert_eq!(
            fields["touched"],
            serde_json::json!(["block b1"]),
            "the index is names, not values",
        );
    }
}