First slice: ledger-store (journal+blobs+sqlite) and ledger-mcp (18 tools, receipt policy)

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<Self, StoreError> {
+        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<Self, StoreError> {
+        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<JournalLine, StoreError> {
+        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<Vec<JournalLine>, 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<u64, StoreError> {
+    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::<JournalLine>(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)
+}