From 794b18e6ab5df7d65f4e6e501ed5dfeaca6bee12 Mon Sep 17 00:00:00 2001 From: Stephen Waits <1+swaits@noreply.git.swaits.com> Date: Sun, 11 Oct 2026 01:37:13 +0000 Subject: [PATCH] First slice: ledger-store (journal+blobs+sqlite) and ledger-mcp (18 tools, receipt policy) --- crates/ledger-store/src/journal.rs | 213 +++++++++++++++++++++++++++++ 1 file changed, 213 insertions(+) create mode 100644 crates/ledger-store/src/journal.rs diff --git a/crates/ledger-store/src/journal.rs b/crates/ledger-store/src/journal.rs new file mode 100644 index 0000000..82179a1 --- /dev/null +++ b/crates/ledger-store/src/journal.rs @@ -0,0 +1,213 @@ +//! The journal: the source of truth. +//! +//! An append-only file of JSON objects, one per line (JSONL), hash-chained: +//! each line carries the SHA-256 of the previous line's canonical bytes. +//! Bit rot, torn writes, and casual editing are detected on read — never +//! silently absorbed. +//! +//! Canonical form is keys-sorted, whitespace-free JSON. The writer always +//! emits canonical form, so parse → re-serialize round-trips byte-identically +//! and the chain verifies. + +use crate::error::StoreError; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::fs::{File, OpenOptions}; +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; + +/// `prev` value of the first journal line: 64 zeros. +pub const GENESIS_PREV: &str = "0000000000000000000000000000000000000000000000000000000000000000"; + +/// One journal line: a sequenced, timestamped, hash-chained domain event. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct JournalLine { + /// 1-based, gapless sequence number. + pub seq: u64, + /// UTC timestamp, RFC 3339. Recorded, never trusted for ordering. + pub ts: String, + /// Event type: `account.added`, `entry.created`, `entry.posted`, + /// `receipt.attached`, `receipt.linked`, `document.filed`. + #[serde(rename = "type")] + pub line_type: String, + /// The full domain facts. Everything a replay needs. + pub payload: Value, + /// Hex SHA-256 of the previous line's canonical bytes. + pub prev: String, +} + +impl JournalLine { + /// Canonical bytes of this line: keys sorted recursively, no whitespace. + pub fn to_canonical(&self) -> String { + canonical_json(&serde_json::to_value(self).expect("JournalLine serializes")) + } +} + +/// Recursively key-sorted, whitespace-free JSON. +pub fn canonical_json(value: &Value) -> String { + serde_json::to_string(&sort_value(value)).expect("JSON serialization cannot fail") +} + +fn sort_value(v: &Value) -> Value { + match v { + Value::Object(map) => { + let mut keys: Vec<&String> = map.keys().collect(); + keys.sort(); + let mut sorted = serde_json::Map::with_capacity(map.len()); + for k in keys { + sorted.insert(k.clone(), sort_value(&map[k])); + } + Value::Object(sorted) + } + Value::Array(arr) => Value::Array(arr.iter().map(sort_value).collect()), + _ => v.clone(), + } +} + +/// Hex SHA-256 of bytes. +pub fn sha256_hex(bytes: &[u8]) -> String { + let mut h = Sha256::new(); + h.update(bytes); + hex::encode(h.finalize()) +} + +/// Append-only writer. Assigns `seq`, stamps `ts`, chains `prev`, fsyncs +/// every append. The sequence only advances after a successful fsync, so a +/// torn tail can never exist in a file written by this writer. +pub struct JournalWriter { + path: PathBuf, + file: File, + next_seq: u64, + prev_hash: String, +} + +impl JournalWriter { + /// Open an existing journal (verifying its chain) or create a new one. + pub fn open(path: &Path) -> Result { + let lines = if path.exists() { + read_and_verify_journal(path)? + } else { + Vec::new() + }; + Self::from_lines(path, &lines) + } + + /// Build a writer from already-verified lines (avoids re-scanning). + pub(crate) fn from_lines(path: &Path, lines: &[JournalLine]) -> Result { + let (next_seq, prev_hash) = match lines.last() { + Some(last) => (last.seq + 1, sha256_hex(last.to_canonical().as_bytes())), + None => (1, GENESIS_PREV.to_string()), + }; + if !path.exists() + && let Some(parent) = path.parent() + { + std::fs::create_dir_all(parent)?; + } + let file = OpenOptions::new().create(true).append(true).open(path)?; + Ok(Self { + path: path.to_path_buf(), + file, + next_seq, + prev_hash, + }) + } + + /// Append one event. Returns the journaled line. + pub fn append(&mut self, line_type: &str, payload: Value) -> Result { + let line = JournalLine { + seq: self.next_seq, + ts: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true), + line_type: line_type.to_string(), + payload, + prev: self.prev_hash.clone(), + }; + let bytes = line.to_canonical(); + self.file.write_all(bytes.as_bytes())?; + self.file.write_all(b"\n")?; + self.file.sync_all()?; + self.prev_hash = sha256_hex(bytes.as_bytes()); + self.next_seq += 1; + Ok(line) + } + + /// The seq the next append will use. + pub fn next_seq(&self) -> u64 { + self.next_seq + } + + /// Path of the journal file. + pub fn path(&self) -> &Path { + &self.path + } +} + +/// Read and fully verify a journal: every line parses, seqs are gapless from +/// 1, and the hash chain is intact. Any failure names the offending seq. +pub fn read_and_verify_journal(path: &Path) -> Result, StoreError> { + let file = File::open(path)?; + let reader = BufReader::new(file); + let mut lines = Vec::new(); + let mut expected_prev = GENESIS_PREV.to_string(); + for (i, raw) in reader.lines().enumerate() { + let raw = raw?; + let line: JournalLine = serde_json::from_str(&raw).map_err(|e| { + StoreError::JournalCorrupt(format!("seq {}: unparseable line: {e}", i + 1)) + })?; + let want_seq = i as u64 + 1; + if line.seq != want_seq { + return Err(StoreError::JournalCorrupt(format!( + "seq gap: expected {want_seq}, found {}", + line.seq + ))); + } + if line.prev != expected_prev { + return Err(StoreError::JournalCorrupt(format!( + "hash chain broken at seq {}", + line.seq + ))); + } + expected_prev = sha256_hex(line.to_canonical().as_bytes()); + lines.push(line); + } + Ok(lines) +} + +/// Truncate a journal file to its last good line. For recovering *copies* +/// with a torn tail (e.g. a backup taken mid-write): the projection never +/// saw the torn line, so dropping it is safe. Returns the last good seq. +/// The live journal written by [`JournalWriter`] cannot have a torn tail. +pub fn truncate_to_last_good_seq(path: &Path) -> Result { + let raw = std::fs::read(path)?; + let mut good_len = 0usize; + let mut last_good_seq = 0u64; + let mut expected_prev = GENESIS_PREV.to_string(); + // Walk raw bytes: a line is bytes up to and including b'\n'. Trailing + // bytes without a newline are a torn tail and are dropped. + let mut start = 0usize; + for (i, &b) in raw.iter().enumerate() { + if b != b'\n' { + continue; + } + let line_bytes = &raw[start..i]; + start = i + 1; + let line_str = std::str::from_utf8(line_bytes).map_err(|_| { + StoreError::JournalCorrupt(format!("seq {}: invalid utf-8", last_good_seq + 1)) + })?; + if line_str.trim().is_empty() { + break; // blank line: our writer never emits these + } + match serde_json::from_str::(line_str) { + Ok(line) if line.seq == last_good_seq + 1 && line.prev == expected_prev => { + expected_prev = sha256_hex(line.to_canonical().as_bytes()); + last_good_seq = line.seq; + good_len = i + 1; // include the newline + } + _ => break, + } + } + let file = OpenOptions::new().write(true).open(path)?; + file.set_len(good_len as u64)?; + file.sync_all()?; + Ok(last_good_seq) +}