[sqlite-watcher] add RowChange abstraction and queue wiring

Author: taariq

Cargo.lock

@@ -13,6 +13,7 @@ anyhow = "1.0"
 clap = { version = "4.4", features = ["derive", "env"] }
 dirs = "5.0"
 rusqlite = "0.30"
+serde_json = "1.0"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 

sqlite-watcher/Cargo.toml

@@ -18,6 +18,7 @@ sqlite-watcher \
   --listen unix:/tmp/sqlite-watcher.sock \
   --token-file ~/.seren/sqlite-watcher/token \
   --log-level info \
+  --queue-db ~/.seren/sqlite-watcher/changes.db \
   --poll-interval-ms 250 \
   --min-event-bytes 4096
 ```
@@ -27,6 +28,7 @@ Flag summary:
 - `--db` (required): SQLite file to monitor; must exist and be accessible in WAL mode.
 - `--listen`: Listener endpoint; accepts `unix:/path`, `tcp:<port>`, or `pipe:<name>`.
 - `--token-file`: Shared-secret used to authenticate gRPC clients (defaults to `~/.seren/sqlite-watcher/token`).
+- `--queue-db`: SQLite file used to persist change events + checkpoints (defaults to `~/.seren/sqlite-watcher/changes.db`).
 - `--log-level`: Tracing filter (also settable via `SQLITE_WATCHER_LOG`).
 - `--poll-interval-ms`: How often to check the WAL file for growth (default 500 ms). Lower values react faster but cost more syscalls.
 - `--min-event-bytes`: Minimum WAL byte growth before emitting an event. Use larger values to avoid spam when very small transactions occur.
@@ -37,5 +39,10 @@ Flag summary:
 - **Windows**: Unix sockets are disabled; pass `--listen tcp:50051` or `--listen pipe:SerenWatcher`. Named pipes allow local service accounts without opening TCP ports.
 - All platforms expect the token file to live under `~/.seren/sqlite-watcher/token` by default; create the directory with `0700` permissions so the watcher refuses to start if the secret is world-readable.
 - The current WAL watcher polls the `*.sqlite-wal` file for byte growth. To keep WAL history available, configure your writers with `PRAGMA journal_mode=WAL;` and raise `wal_autocheckpoint` (or disable it) so the SQLite engine does not aggressively truncate the log.
+- Change queue data is stored under `~/.seren/sqlite-watcher/changes.db`. The binary enforces owner-only permissions on that directory to keep tokens + change data private.
+
+### Placeholder change format
+
+Until the WAL decoder lands, each growth event is recorded as a `RowChange` with `table_name="__wal__"`, `operation=insert`, and a JSON payload describing the byte delta + timestamp. Downstream components can treat these as heartbeats while we finish Tickets B–D.
 
 Additional design constraints and follow-up work items live in `docs/plans/sqlite-watcher-plan.md` and `docs/plans/sqlite-watcher-tickets.md`.

sqlite-watcher/README.md

@@ -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'));
+    }
+}

sqlite-watcher/src/change.rs

@@ -1,2 +1,3 @@
+pub mod change;
 pub mod queue;
 pub mod wal;

sqlite-watcher/src/lib.rs

@@ -2,11 +2,14 @@ use std::fmt;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
 use std::sync::mpsc;
-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use anyhow::{anyhow, bail, Context, Result};
 use clap::Parser;
-use sqlite_watcher::wal::{start_wal_watcher, WalWatcherConfig as TailConfig};
+use serde_json::json;
+use sqlite_watcher::change::RowChange;
+use sqlite_watcher::queue::{ChangeOperation, ChangeQueue};
+use sqlite_watcher::wal::{start_wal_watcher, WalEvent, WalWatcherConfig as TailConfig};
 use tracing_subscriber::EnvFilter;
 
 #[cfg(unix)]
@@ -35,6 +38,10 @@ struct Cli {
     #[arg(long = "token-file", value_name = "PATH")]
     token_file: Option<PathBuf>,
 
+    /// Path to the durable change queue database.
+    #[arg(long = "queue-db", value_name = "PATH")]
+    queue_db: Option<PathBuf>,
+
     /// Tracing filter (info,warn,debug,trace). Can also be provided via SQLITE_WATCHER_LOG.
     #[arg(
         long = "log-level",
@@ -121,6 +128,7 @@ struct WatcherConfig {
     database_path: PathBuf,
     listen: ListenAddress,
     token_file: PathBuf,
+    queue_path: PathBuf,
     poll_interval: Duration,
     min_event_bytes: u64,
 }
@@ -135,11 +143,16 @@ impl TryFrom<Cli> for WatcherConfig {
             Some(path) => expand_home(path)?,
             None => default_token_path()?,
         };
+        let queue_path = match args.queue_db {
+            Some(path) => expand_home(path)?,
+            None => default_queue_path()?,
+        };
 
         Ok(Self {
             database_path,
             listen,
             token_file,
+            queue_path,
             poll_interval: Duration::from_millis(args.poll_interval_ms),
             min_event_bytes: args.min_event_bytes,
         })
@@ -163,6 +176,11 @@ fn default_token_path() -> Result<PathBuf> {
     Ok(home.join(".seren/sqlite-watcher/token"))
 }
 
+fn default_queue_path() -> Result<PathBuf> {
+    let home = dirs::home_dir().ok_or_else(|| anyhow!("unable to determine home directory"))?;
+    Ok(home.join(".seren/sqlite-watcher/changes.db"))
+}
+
 fn expand_home(path: PathBuf) -> Result<PathBuf> {
     let as_str = path.to_string_lossy();
     if let Some(stripped) = as_str.strip_prefix("~/") {
@@ -194,11 +212,13 @@ fn main() -> Result<()> {
         db = %config.database_path.display(),
         listen = %config.listen,
         token = %config.token_file.display(),
+        queue = %config.queue_path.display(),
         poll_ms = config.poll_interval.as_millis(),
         min_event_bytes = config.min_event_bytes,
         "sqlite-watcher starting"
     );
 
+    let queue = ChangeQueue::open(&config.queue_path)?;
     let (event_tx, event_rx) = mpsc::channel();
     let _wal_handle = start_wal_watcher(
         &config.database_path,
@@ -210,24 +230,55 @@ fn main() -> Result<()> {
     )?;
 
     for event in event_rx {
-        tracing::info!(
-            bytes_added = event.bytes_added,
-            wal_size = event.current_size,
-            "wal file grew"
-        );
+        match persist_wal_event(&queue, &event) {
+            Ok(change_id) => {
+                tracing::info!(
+                    bytes_added = event.bytes_added,
+                    wal_size = event.current_size,
+                    change_id,
+                    "queued wal growth event"
+                );
+            }
+            Err(err) => {
+                tracing::warn!(error = %err, "failed to persist wal event to queue");
+            }
+        }
     }
 
     Ok(())
 }
 
+fn persist_wal_event(queue: &ChangeQueue, event: &WalEvent) -> Result<i64> {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(|err| anyhow!("system clock drift: {err}"))?;
+    let row = 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,
+    };
+    queue.enqueue(&row.into_new_change())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use clap::Parser;
+    use sqlite_watcher::queue::ChangeQueue;
+    use sqlite_watcher::wal::WalEvent;
+    use tempfile::{tempdir, NamedTempFile};
 
     #[test]
     fn parses_tcp_listener() {
-        let tmp = tempfile::NamedTempFile::new().unwrap();
+        let tmp = NamedTempFile::new().unwrap();
         let cli = Cli::try_parse_from([
             "sqlite-watcher",
             "--db",
@@ -247,15 +298,33 @@ mod tests {
             ListenAddress::Tcp { host, port } if host == "127.0.0.1" && port == 6000
         ));
         assert!(config.token_file.ends_with("token"));
+        assert!(config.queue_path.ends_with("changes.db"));
     }
 
     #[test]
     #[cfg(unix)]
     fn parses_unix_listener_default() {
-        let tmp = tempfile::NamedTempFile::new().unwrap();
+        let tmp = NamedTempFile::new().unwrap();
         let cli =
             Cli::try_parse_from(["sqlite-watcher", "--db", tmp.path().to_str().unwrap()]).unwrap();
         let config = WatcherConfig::try_from(cli).unwrap();
         assert!(matches!(config.listen, ListenAddress::Unix(_)));
     }
+
+    #[test]
+    fn persist_wal_events_into_queue() {
+        let dir = tempdir().unwrap();
+        let queue_path = dir.path().join("queue.db");
+        let queue = ChangeQueue::open(&queue_path).unwrap();
+
+        let event = WalEvent {
+            bytes_added: 2048,
+            current_size: 4096,
+        };
+        let change_id = persist_wal_event(&queue, &event).unwrap();
+        let batch = queue.fetch_batch(10).unwrap();
+        assert_eq!(batch.len(), 1);
+        assert_eq!(batch[0].change_id, change_id);
+        assert_eq!(batch[0].table_name, "__wal__");
+    }
 }