spreadsheet_kit/core/
binlog.rs

1//! Append-only JSONL binlog writer/reader for OpEvent streams.
2//!
3//! The binlog is stored as `events.jsonl` within a session directory.
4//! Each line is a self-contained JSON-serialized [`OpEvent`].
5
6use crate::core::events::OpEvent;
7use anyhow::{Context, Result, anyhow, bail};
8use std::fs::{self, File, OpenOptions};
9use std::io::{BufRead, BufReader, Write};
10use std::path::{Path, PathBuf};
11
12// ---------------------------------------------------------------------------
13// BinlogWriter
14// ---------------------------------------------------------------------------
15
16/// Append-only writer for the session event log.
17pub struct BinlogWriter {
18    path: PathBuf,
19}
20
21impl BinlogWriter {
22    /// Open or create a binlog file at the given path.
23    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
24        let path = path.into();
25        if let Some(parent) = path.parent() {
26            fs::create_dir_all(parent).with_context(|| {
27                format!("failed to create binlog directory: {}", parent.display())
28            })?;
29        }
30        Ok(Self { path })
31    }
32
33    /// Append a single event to the binlog with fsync.
34    pub fn append(&self, event: &OpEvent) -> Result<()> {
35        let line = serde_json::to_string(event).context("failed to serialize OpEvent to JSON")?;
36
37        let mut file = OpenOptions::new()
38            .create(true)
39            .append(true)
40            .open(&self.path)
41            .with_context(|| format!("failed to open binlog: {}", self.path.display()))?;
42
43        writeln!(file, "{}", line).context("failed to write event to binlog")?;
44
45        file.sync_data().context("failed to fsync binlog")?;
46
47        Ok(())
48    }
49
50    /// Return the path to the binlog file.
51    pub fn path(&self) -> &Path {
52        &self.path
53    }
54}
55
56// ---------------------------------------------------------------------------
57// BinlogReader
58// ---------------------------------------------------------------------------
59
60/// Reader for replaying events from a binlog file.
61pub struct BinlogReader {
62    path: PathBuf,
63}
64
65impl BinlogReader {
66    /// Open an existing binlog file for reading.
67    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
68        let path = path.into();
69        if !path.exists() {
70            bail!("binlog not found: {}", path.display());
71        }
72        Ok(Self { path })
73    }
74
75    /// Read all events from the binlog.
76    pub fn read_all(&self) -> Result<Vec<OpEvent>> {
77        let file = File::open(&self.path)
78            .with_context(|| format!("failed to open binlog: {}", self.path.display()))?;
79        let reader = BufReader::new(file);
80        let mut events = Vec::new();
81
82        for (line_number, line_result) in reader.lines().enumerate() {
83            let line = line_result
84                .with_context(|| format!("failed to read binlog line {}", line_number + 1))?;
85
86            let trimmed = line.trim();
87            if trimmed.is_empty() {
88                continue;
89            }
90
91            let event: OpEvent = serde_json::from_str(trimmed).with_context(|| {
92                format!("corrupted binlog at line {}: invalid JSON", line_number + 1)
93            })?;
94
95            events.push(event);
96        }
97
98        Ok(events)
99    }
100
101    /// Read events starting after a given op_id (for replay from a snapshot).
102    pub fn read_after(&self, after_op_id: &str) -> Result<Vec<OpEvent>> {
103        let all = self.read_all()?;
104        let start_index = all
105            .iter()
106            .position(|e| e.op_id == after_op_id)
107            .map(|i| i + 1)
108            .unwrap_or(0);
109        Ok(all.into_iter().skip(start_index).collect())
110    }
111
112    /// Read events up to and including a given op_id.
113    pub fn read_until(&self, until_op_id: &str) -> Result<Vec<OpEvent>> {
114        let all = self.read_all()?;
115        let end_index = all
116            .iter()
117            .position(|e| e.op_id == until_op_id)
118            .ok_or_else(|| anyhow!("op_id '{}' not found in binlog", until_op_id))?;
119        Ok(all.into_iter().take(end_index + 1).collect())
120    }
121
122    /// Validate the parent_id chain: each event's parent_id should match the
123    /// previous event's op_id (except the first event).
124    pub fn validate_lineage(&self) -> Result<Vec<String>> {
125        let events = self.read_all()?;
126        let mut warnings = Vec::new();
127
128        for (i, event) in events.iter().enumerate() {
129            if i == 0 {
130                if event.parent_id.is_some() {
131                    warnings.push(format!(
132                        "first event '{}' has parent_id but is the first in the log",
133                        event.op_id
134                    ));
135                }
136                continue;
137            }
138
139            let expected_parent = &events[i - 1].op_id;
140            match &event.parent_id {
141                Some(parent) if parent != expected_parent => {
142                    warnings.push(format!(
143                        "event '{}' has parent_id '{}' but expected '{}' (branch point or corruption)",
144                        event.op_id, parent, expected_parent
145                    ));
146                }
147                None => {
148                    warnings.push(format!(
149                        "event '{}' at position {} has no parent_id",
150                        event.op_id, i
151                    ));
152                }
153                _ => {}
154            }
155        }
156
157        Ok(warnings)
158    }
159
160    /// Validate hash chain integrity (if events have event_hash/prev_event_hash).
161    pub fn validate_hash_chain(&self) -> Result<Vec<String>> {
162        let events = self.read_all()?;
163        let mut warnings = Vec::new();
164
165        for (i, event) in events.iter().enumerate() {
166            if i == 0 {
167                continue;
168            }
169
170            if let (Some(prev_hash), Some(expected_prev)) =
171                (&events[i - 1].event_hash, &event.prev_event_hash)
172                && prev_hash != expected_prev
173            {
174                warnings.push(format!(
175                    "hash chain broken at event '{}': prev_event_hash '{}' != previous event_hash '{}'",
176                    event.op_id, expected_prev, prev_hash
177                ));
178            }
179        }
180
181        Ok(warnings)
182    }
183
184    /// Return the last event in the binlog (the current tip).
185    pub fn tip(&self) -> Result<Option<OpEvent>> {
186        let events = self.read_all()?;
187        Ok(events.into_iter().last())
188    }
189
190    /// Count events in the binlog.
191    pub fn count(&self) -> Result<usize> {
192        let file = File::open(&self.path)
193            .with_context(|| format!("failed to open binlog: {}", self.path.display()))?;
194        let reader = BufReader::new(file);
195        let count = reader
196            .lines()
197            .map_while(Result::ok)
198            .filter(|l| !l.trim().is_empty())
199            .count();
200        Ok(count)
201    }
202}
203
204// ---------------------------------------------------------------------------
205// Snapshot manifest
206// ---------------------------------------------------------------------------
207
208/// Index entry for a materialized snapshot.
209#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
210pub struct SnapshotEntry {
211    /// The op_id at which this snapshot was taken.
212    pub op_id: String,
213    /// Relative path to the snapshot file within the session directory.
214    pub file_name: String,
215    /// SHA-256 hash of the snapshot file.
216    pub file_hash: String,
217    /// Timestamp of snapshot creation.
218    pub created_at: chrono::DateTime<chrono::Utc>,
219    /// Number of events from base to this snapshot.
220    pub event_count: usize,
221}
222
223/// Manifest tracking all snapshots for a session.
224#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
225pub struct SnapshotManifest {
226    pub session_id: String,
227    pub entries: Vec<SnapshotEntry>,
228}
229
230impl SnapshotManifest {
231    /// Create a new empty manifest.
232    pub fn new(session_id: String) -> Self {
233        Self {
234            session_id,
235            entries: Vec::new(),
236        }
237    }
238
239    /// Load a manifest from disk.
240    pub fn load(path: &Path) -> Result<Self> {
241        let content = fs::read_to_string(path)
242            .with_context(|| format!("failed to read snapshot manifest: {}", path.display()))?;
243        serde_json::from_str(&content).context("failed to parse snapshot manifest")
244    }
245
246    /// Save the manifest to disk.
247    pub fn save(&self, path: &Path) -> Result<()> {
248        let content =
249            serde_json::to_string_pretty(self).context("failed to serialize snapshot manifest")?;
250        if let Some(parent) = path.parent() {
251            fs::create_dir_all(parent)?;
252        }
253        fs::write(path, content)
254            .with_context(|| format!("failed to write snapshot manifest: {}", path.display()))
255    }
256
257    /// Add a new snapshot entry.
258    pub fn add_entry(&mut self, entry: SnapshotEntry) {
259        self.entries.push(entry);
260    }
261
262    /// Find the nearest snapshot at or before the given op_id.
263    /// Returns the entry whose op_id appears earliest in the event order
264    /// but is closest to the target.
265    pub fn nearest_snapshot(
266        &self,
267        target_op_id: &str,
268        event_order: &[String],
269    ) -> Option<&SnapshotEntry> {
270        let target_pos = event_order.iter().position(|id| id == target_op_id)?;
271        self.entries
272            .iter()
273            .filter_map(|entry| {
274                event_order
275                    .iter()
276                    .position(|id| id == &entry.op_id)
277                    .filter(|&pos| pos <= target_pos)
278                    .map(|pos| (pos, entry))
279            })
280            .max_by_key(|(pos, _)| *pos)
281            .map(|(_, entry)| entry)
282    }
283}
284
285// ---------------------------------------------------------------------------
286// Branch metadata
287// ---------------------------------------------------------------------------
288
289/// Branch pointer within a session.
290#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
291pub struct BranchInfo {
292    /// Branch name (e.g. "main", "alt-scenario").
293    pub name: String,
294    /// The op_id at the tip of this branch.
295    pub tip_op_id: Option<String>,
296    /// The op_id where this branch forked from its parent.
297    #[serde(skip_serializing_if = "Option::is_none")]
298    pub fork_point: Option<String>,
299    /// Optional human-readable label.
300    #[serde(skip_serializing_if = "Option::is_none")]
301    pub label: Option<String>,
302    /// Creation timestamp.
303    pub created_at: chrono::DateTime<chrono::Utc>,
304}
305
306/// Container for all branch metadata in a session.
307#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
308pub struct BranchesFile {
309    pub branches: Vec<BranchInfo>,
310}
311
312impl BranchesFile {
313    pub fn new() -> Self {
314        Self {
315            branches: vec![BranchInfo {
316                name: "main".to_string(),
317                tip_op_id: None,
318                fork_point: None,
319                label: None,
320                created_at: chrono::Utc::now(),
321            }],
322        }
323    }
324
325    pub fn load(path: &Path) -> Result<Self> {
326        let content = fs::read_to_string(path)
327            .with_context(|| format!("failed to read branches file: {}", path.display()))?;
328        serde_json::from_str(&content).context("failed to parse branches file")
329    }
330
331    pub fn save(&self, path: &Path) -> Result<()> {
332        let content = serde_json::to_string_pretty(self)?;
333        fs::write(path, content)
334            .with_context(|| format!("failed to write branches file: {}", path.display()))
335    }
336
337    pub fn get_branch(&self, name: &str) -> Option<&BranchInfo> {
338        self.branches.iter().find(|b| b.name == name)
339    }
340
341    pub fn get_branch_mut(&mut self, name: &str) -> Option<&mut BranchInfo> {
342        self.branches.iter_mut().find(|b| b.name == name)
343    }
344
345    pub fn add_branch(&mut self, info: BranchInfo) {
346        self.branches.push(info);
347    }
348}
349
350impl Default for BranchesFile {
351    fn default() -> Self {
352        Self::new()
353    }
354}
355
356// ---------------------------------------------------------------------------
357// Tests
358// ---------------------------------------------------------------------------
359
360#[cfg(test)]
361mod tests {
362    use super::*;
363    use crate::core::events::{Actor, OpEvent, OpKind};
364    use serde_json::json;
365    use tempfile::TempDir;
366
367    fn test_event(session_id: &str, parent_id: Option<&str>) -> OpEvent {
368        OpEvent::new(
369            session_id.to_string(),
370            parent_id.map(|s| s.to_string()),
371            Actor {
372                id: "test".to_string(),
373                run_id: None,
374                source: "test".to_string(),
375            },
376            OpKind::edit_batch(),
377            json!({"cell": "A1", "value": 42}),
378        )
379    }
380
381    #[test]
382    fn binlog_append_and_read() {
383        let tmp = TempDir::new().unwrap();
384        let log_path = tmp.path().join("events.jsonl");
385
386        let writer = BinlogWriter::open(&log_path).unwrap();
387        let event1 = test_event("sess1", None);
388        let event2 = test_event("sess1", Some(&event1.op_id));
389
390        writer.append(&event1).unwrap();
391        writer.append(&event2).unwrap();
392
393        let reader = BinlogReader::open(&log_path).unwrap();
394        let events = reader.read_all().unwrap();
395        assert_eq!(events.len(), 2);
396        assert_eq!(events[0].op_id, event1.op_id);
397        assert_eq!(events[1].op_id, event2.op_id);
398    }
399
400    #[test]
401    fn binlog_read_after() {
402        let tmp = TempDir::new().unwrap();
403        let log_path = tmp.path().join("events.jsonl");
404
405        let writer = BinlogWriter::open(&log_path).unwrap();
406        let e1 = test_event("sess1", None);
407        let e2 = test_event("sess1", Some(&e1.op_id));
408        let e3 = test_event("sess1", Some(&e2.op_id));
409
410        writer.append(&e1).unwrap();
411        writer.append(&e2).unwrap();
412        writer.append(&e3).unwrap();
413
414        let reader = BinlogReader::open(&log_path).unwrap();
415        let after = reader.read_after(&e1.op_id).unwrap();
416        assert_eq!(after.len(), 2);
417        assert_eq!(after[0].op_id, e2.op_id);
418    }
419
420    #[test]
421    fn binlog_lineage_validation() {
422        let tmp = TempDir::new().unwrap();
423        let log_path = tmp.path().join("events.jsonl");
424
425        let writer = BinlogWriter::open(&log_path).unwrap();
426        let e1 = test_event("sess1", None);
427        let e2 = test_event("sess1", Some(&e1.op_id));
428        // e3 has wrong parent (simulates branch or corruption)
429        let e3 = test_event("sess1", Some("wrong_parent"));
430
431        writer.append(&e1).unwrap();
432        writer.append(&e2).unwrap();
433        writer.append(&e3).unwrap();
434
435        let reader = BinlogReader::open(&log_path).unwrap();
436        let warnings = reader.validate_lineage().unwrap();
437        assert_eq!(warnings.len(), 1);
438        assert!(warnings[0].contains("wrong_parent"));
439    }
440
441    #[test]
442    fn snapshot_manifest_nearest() {
443        let mut manifest = SnapshotManifest::new("sess1".to_string());
444        manifest.add_entry(SnapshotEntry {
445            op_id: "op_001".to_string(),
446            file_name: "snap_001.xlsx".to_string(),
447            file_hash: "sha256:aaa".to_string(),
448            created_at: chrono::Utc::now(),
449            event_count: 1,
450        });
451        manifest.add_entry(SnapshotEntry {
452            op_id: "op_005".to_string(),
453            file_name: "snap_005.xlsx".to_string(),
454            file_hash: "sha256:bbb".to_string(),
455            created_at: chrono::Utc::now(),
456            event_count: 5,
457        });
458
459        let order: Vec<String> = (1..=10).map(|i| format!("op_{:03}", i)).collect();
460
461        // Target at op_007 → nearest snapshot is op_005
462        let nearest = manifest.nearest_snapshot("op_007", &order);
463        assert_eq!(nearest.unwrap().op_id, "op_005");
464
465        // Target at op_003 → nearest snapshot is op_001
466        let nearest = manifest.nearest_snapshot("op_003", &order);
467        assert_eq!(nearest.unwrap().op_id, "op_001");
468
469        // Target at op_001 → exact match
470        let nearest = manifest.nearest_snapshot("op_001", &order);
471        assert_eq!(nearest.unwrap().op_id, "op_001");
472    }
473
474    #[test]
475    fn branches_file_roundtrip() {
476        let tmp = TempDir::new().unwrap();
477        let path = tmp.path().join("branches.json");
478
479        let mut bf = BranchesFile::new();
480        bf.add_branch(BranchInfo {
481            name: "alt-scenario".to_string(),
482            tip_op_id: Some("op_abc".to_string()),
483            fork_point: Some("op_005".to_string()),
484            label: Some("Alternative Scenario".to_string()),
485            created_at: chrono::Utc::now(),
486        });
487
488        bf.save(&path).unwrap();
489        let loaded = BranchesFile::load(&path).unwrap();
490        assert_eq!(loaded.branches.len(), 2);
491        assert_eq!(
492            loaded
493                .get_branch("alt-scenario")
494                .unwrap()
495                .tip_op_id
496                .as_deref(),
497            Some("op_abc")
498        );
499    }
500}
spreadsheet_kit/core/binlog.rs

spreadsheet_kit/core/
binlog.rs
