Merge origin/main into issue-83-plan-replicator-sync-sqlite

taariq · taariq · commit 7b6b633bb5da · 2025-12-12T16:45:56.000-08:00
diff --git a/sqlite-watcher/src/change.rs b/sqlite-watcher/src/change.rs
@@ -0,0 +1,49 @@
+use serde_json::Value;
+
+use crate::queue::{ChangeOperation, NewChange};
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct RowChange {
+    pub table_name: String,
+    pub operation: ChangeOperation,
+    pub primary_key: String,
+    pub payload: Option<Value>,
+    pub wal_frame: Option<String>,
+    pub cursor: Option<String>,
+}
+
+impl RowChange {
+    pub fn into_new_change(self) -> NewChange {
+        let payload = self
+            .payload
+            .map(|value| serde_json::to_vec(&value).expect("row change payload serializes"));
+        NewChange {
+            table_name: self.table_name,
+            operation: self.operation,
+            primary_key: self.primary_key,
+            payload,
+            wal_frame: self.wal_frame,
+            cursor: self.cursor,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn converts_to_new_change() {
+        let row = RowChange {
+            table_name: "prices".into(),
+            operation: ChangeOperation::Update,
+            primary_key: "pk1".into(),
+            payload: Some(serde_json::json!({"foo": "bar"})),
+            wal_frame: Some("frame-1".into()),
+            cursor: Some("cursor".into()),
+        };
+        let change = row.into_new_change();
+        assert_eq!(change.table_name, "prices");
+        assert!(change.payload.unwrap().contains(&b'b'));
+    }
+}
diff --git a/sqlite-watcher/src/decoder.rs b/sqlite-watcher/src/decoder.rs
@@ -0,0 +1,48 @@
+use crate::change::RowChange;
+use crate::queue::ChangeOperation;
+use crate::wal::WalEvent;
+use serde_json::json;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+/// Temporary decoder that turns WAL growth bytes into placeholder RowChange events.
+/// Placeholder until row-level decoding is implemented.
+#[derive(Debug, Default, Clone)]
+pub struct WalGrowthDecoder;
+
+impl WalGrowthDecoder {
+    pub fn decode(&self, event: &WalEvent) -> Vec<RowChange> {
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .expect("clock should be >= UNIX epoch");
+        vec![RowChange {
+            table_name: "__wal__".to_string(),
+            operation: ChangeOperation::Insert,
+            primary_key: now.as_nanos().to_string(),
+            payload: Some(json!({
+                "kind": "wal_growth",
+                "bytes_added": event.bytes_added,
+                "current_size": event.current_size,
+                "recorded_at": now.as_secs_f64(),
+            })),
+            wal_frame: None,
+            cursor: None,
+        }]
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn produces_placeholder_row_change() {
+        let decoder = WalGrowthDecoder::default();
+        let rows = decoder.decode(&WalEvent {
+            bytes_added: 1024,
+            current_size: 2048,
+        });
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].table_name, "__wal__");
+        assert_eq!(rows[0].operation, ChangeOperation::Insert);
+    }
+}
diff --git a/sqlite-watcher/src/wal.rs b/sqlite-watcher/src/wal.rs
@@ -0,0 +1,236 @@
+use std::ffi::OsString;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::Sender;
+use std::sync::Arc;
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+
+use anyhow::{Context, Result};
+use tracing::{debug, warn};
+
+#[derive(Debug, Clone, Copy)]
+pub struct WalWatcherConfig {
+    pub poll_interval: Duration,
+    pub min_event_bytes: u64,
+}
+
+impl Default for WalWatcherConfig {
+    fn default() -> Self {
+        Self {
+            poll_interval: Duration::from_millis(500),
+            min_event_bytes: 0,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WalEvent {
+    pub bytes_added: u64,
+    pub current_size: u64,
+}
+
+pub struct WalWatcherHandle {
+    stop: Arc<AtomicBool>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl Drop for WalWatcherHandle {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::SeqCst);
+        if let Some(handle) = self.thread.take() {
+            let _ = handle.join();
+        }
+    }
+}
+
+pub fn start_wal_watcher<P: AsRef<Path>>(
+    db_path: P,
+    options: WalWatcherConfig,
+    sender: Sender<WalEvent>,
+) -> Result<WalWatcherHandle> {
+    let db_path = db_path.as_ref().canonicalize().with_context(|| {
+        format!(
+            "failed to canonicalize database path {}",
+            db_path.as_ref().display()
+        )
+    })?;
+    if !db_path.is_file() {
+        anyhow::bail!("database path {} is not a file", db_path.display());
+    }
+
+    let wal_path = wal_file_path(&db_path);
+    let poll_interval = options.poll_interval;
+    let min_event_bytes = options.min_event_bytes;
+    let stop_flag = Arc::new(AtomicBool::new(false));
+    let thread_stop = Arc::clone(&stop_flag);
+
+    let handle = thread::spawn(move || {
+        let mut last_len = wal_file_size(&wal_path).unwrap_or(0);
+        debug!(
+            wal = %wal_path.display(),
+            last_len,
+            "wal watcher started"
+        );
+        while !thread_stop.load(Ordering::SeqCst) {
+            match wal_file_size(&wal_path) {
+                Ok(len) => {
+                    if len < last_len {
+                        debug!(
+                            wal = %wal_path.display(),
+                            prev = last_len,
+                            current = len,
+                            "wal truncated; resetting baseline"
+                        );
+                        last_len = len;
+                    } else if len > last_len {
+                        let delta = len - last_len;
+                        last_len = len;
+                        if delta >= min_event_bytes {
+                            let event = WalEvent {
+                                bytes_added: delta,
+                                current_size: len,
+                            };
+                            if sender.send(event).is_err() {
+                                debug!("wal watcher stopping because receiver closed");
+                                break;
+                            }
+                        }
+                    }
+                }
+                Err(err) => {
+                    if err.kind() == std::io::ErrorKind::NotFound {
+                        last_len = 0;
+                    } else {
+                        warn!(
+                            wal = %wal_path.display(),
+                            error = %err,
+                            "failed to read wal metadata"
+                        );
+                    }
+                }
+            }
+
+            thread::sleep(poll_interval);
+        }
+
+        debug!("wal watcher exiting");
+    });
+
+    Ok(WalWatcherHandle {
+        stop: stop_flag,
+        thread: Some(handle),
+    })
+}
+
+fn wal_file_path(db_path: &Path) -> PathBuf {
+    let mut os_string = OsString::from(db_path.as_os_str());
+    os_string.push("-wal");
+    PathBuf::from(os_string)
+}
+
+fn wal_file_size(path: &Path) -> std::io::Result<u64> {
+    std::fs::metadata(path).map(|m| m.len())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rusqlite::Connection;
+    use std::sync::mpsc::channel;
+    use std::time::{Duration, Instant};
+    use tempfile::tempdir;
+
+    #[test]
+    fn emits_event_when_wal_grows() {
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("watch.sqlite");
+        let writer = Connection::open(&db_path).unwrap();
+        writer.pragma_update(None, "journal_mode", &"wal").unwrap();
+        writer
+            .pragma_update(None, "wal_autocheckpoint", &0i64)
+            .unwrap();
+        writer
+            .execute(
+                "CREATE TABLE changes(id INTEGER PRIMARY KEY, value TEXT)",
+                [],
+            )
+            .unwrap();
+
+        let (tx, rx) = channel();
+        let handle = start_wal_watcher(
+            &db_path,
+            WalWatcherConfig {
+                poll_interval: Duration::from_millis(50),
+                min_event_bytes: 1,
+            },
+            tx,
+        )
+        .unwrap();
+
+        for i in 0..50 {
+            writer
+                .execute(
+                    "INSERT INTO changes(value) VALUES (?1)",
+                    [format!("value-{i}")],
+                )
+                .unwrap();
+        }
+
+        let event = rx.recv_timeout(Duration::from_secs(5)).unwrap();
+        assert!(event.bytes_added > 0);
+        assert!(event.current_size >= event.bytes_added);
+
+        drop(handle);
+    }
+
+    #[test]
+    fn handles_wal_truncation() {
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("truncate.sqlite");
+        let writer = Connection::open(&db_path).unwrap();
+        writer.pragma_update(None, "journal_mode", &"wal").unwrap();
+        writer
+            .pragma_update(None, "wal_autocheckpoint", &0i64)
+            .unwrap();
+        writer
+            .execute("CREATE TABLE stuff(id INTEGER PRIMARY KEY, value TEXT)", [])
+            .unwrap();
+
+        let (tx, rx) = channel();
+        let handle = start_wal_watcher(
+            &db_path,
+            WalWatcherConfig {
+                poll_interval: Duration::from_millis(25),
+                min_event_bytes: 1,
+            },
+            tx,
+        )
+        .unwrap();
+
+        for i in 0..10 {
+            writer
+                .execute("INSERT INTO stuff(value) VALUES (?1)", [format!("row-{i}")])
+                .unwrap();
+        }
+
+        rx.recv_timeout(Duration::from_secs(5)).unwrap();
+
+        writer
+            .execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")
+            .unwrap();
+
+        // Ensure watcher does not send negative deltas (would panic or overflow)
+        let start = Instant::now();
+        loop {
+            if rx.recv_timeout(Duration::from_millis(100)).is_ok() {
+                break;
+            }
+            if start.elapsed() > Duration::from_millis(500) {
+                break;
+            }
+        }
+
+        drop(handle);
+    }
+}
diff --git a/sqlite-watcher/tests/queue_tests.rs b/sqlite-watcher/tests/queue_tests.rs
@@ -0,0 +1,59 @@
+use sqlite_watcher::queue::{ChangeOperation, ChangeQueue, NewChange, QueueState};
+use tempfile::tempdir;
+
+fn new_change(table: &str, id: &str, op: ChangeOperation) -> NewChange {
+    NewChange {
+        table_name: table.to_string(),
+        operation: op,
+        primary_key: id.to_string(),
+        payload: Some(format!("{{\"id\":\"{id}\"}}").into_bytes()),
+        wal_frame: Some("frame1".to_string()),
+        cursor: None,
+    }
+}
+
+#[test]
+fn durable_enqueue_and_ack_flow() {
+    let dir = tempdir().unwrap();
+    let queue_path = dir.path().join("changes.db");
+    let queue = ChangeQueue::open(&queue_path).unwrap();
+
+    let mut ids = Vec::new();
+    for i in 0..3 {
+        let change = new_change("vaults", &format!("pk-{i}"), ChangeOperation::Insert);
+        ids.push(queue.enqueue(&change).unwrap());
+    }
+
+    let batch = queue.fetch_batch(10).unwrap();
+    assert_eq!(batch.len(), 3);
+
+    queue.ack_up_to(ids[1]).unwrap();
+    queue.purge_acked().unwrap();
+
+    drop(queue);
+
+    let reopened = ChangeQueue::open(&queue_path).unwrap();
+    let remaining = reopened.fetch_batch(10).unwrap();
+    assert_eq!(remaining.len(), 1);
+    assert_eq!(remaining[0].change_id, ids[2]);
+}
+
+#[test]
+fn state_round_trip() {
+    let dir = tempdir().unwrap();
+    let queue_path = dir.path().join("state.db");
+    let queue = ChangeQueue::open(&queue_path).unwrap();
+
+    assert!(queue.get_state("prices").unwrap().is_none());
+
+    let state = QueueState {
+        table_name: "prices".into(),
+        last_change_id: 42,
+        last_wal_frame: Some("frame-42".into()),
+        cursor: Some("cursor-data".into()),
+    };
+    queue.set_state(&state).unwrap();
+
+    let fetched = queue.get_state("prices").unwrap().unwrap();
+    assert_eq!(fetched, state);
+}
diff --git a/sqlite-watcher/tests/wal_watch_tests.rs b/sqlite-watcher/tests/wal_watch_tests.rs
