[sqlite-watcher] scaffold crate with wal watcher + queue

Author: taariq

Cargo.lock

@@ -1,3 +1,9 @@
+[workspace]
+resolver = "2"
+members = [
+    "sqlite-watcher",
+]
+
 [package]
 name = "database-replicator"
 version = "7.0.14"

Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "sqlite-watcher"
+version = "0.1.0"
+edition = "2021"
+authors = ["SerenAI <[email protected]>"]
+description = "SQLite WAL tailer that streams change events to database-replicator."
+license = "Apache-2.0"
+repository = "https://github.com/serenorg/database-replicator"
+readme = "README.md"
+
+[dependencies]
+anyhow = "1.0"
+clap = { version = "4.4", features = ["derive", "env"] }
+dirs = "5.0"
+rusqlite = "0.30"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+
+[dev-dependencies]
+tempfile = "3.8"

sqlite-watcher/Cargo.toml

@@ -0,0 +1,41 @@
+# sqlite-watcher
+
+`sqlite-watcher` tails SQLite WAL files and exposes change streams that `database-replicator` can consume for incremental syncs. The current milestone focuses on a CLI skeleton plus a WAL-growth watcher loop so we can exercise configuration, logging, and packaging before wiring in the change queue + gRPC service described in `docs/plans/sqlite-watcher-plan.md`.
+
+## Building
+
+```bash
+cargo build -p sqlite-watcher
+```
+
+This crate participates in the main workspace, so `cargo build --workspace` or `cargo test --workspace` will also compile it.
+
+## CLI usage
+
+```bash
+sqlite-watcher \
+  --db /path/to/database.db \
+  --listen unix:/tmp/sqlite-watcher.sock \
+  --token-file ~/.seren/sqlite-watcher/token \
+  --log-level info \
+  --poll-interval-ms 250 \
+  --min-event-bytes 4096
+```
+
+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`).
+- `--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.
+
+## Cross-platform notes
+
+- **Linux/macOS**: Default listener is a Unix domain socket at `/tmp/sqlite-watcher.sock`. Ensure the target SQLite database enables WAL journaling.
+- **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.
+
+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,2 @@
+pub mod queue;
+pub mod wal;

sqlite-watcher/src/lib.rs

@@ -0,0 +1,261 @@
+use std::fmt;
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
+use std::sync::mpsc;
+use std::time::Duration;
+
+use anyhow::{anyhow, bail, Context, Result};
+use clap::Parser;
+use sqlite_watcher::wal::{start_wal_watcher, WalWatcherConfig as TailConfig};
+use tracing_subscriber::EnvFilter;
+
+#[cfg(unix)]
+const DEFAULT_LISTEN: &str = "unix:/tmp/sqlite-watcher.sock";
+#[cfg(not(unix))]
+const DEFAULT_LISTEN: &str = "tcp:50051";
+
+/// Command-line interface definition for sqlite-watcher.
+#[derive(Debug, Clone, Parser)]
+#[command(
+    name = "sqlite-watcher",
+    version,
+    about = "Tails SQLite WAL files and exposes change streams.",
+    long_about = None
+)]
+struct Cli {
+    /// Path to the SQLite database the watcher should monitor.
+    #[arg(long = "db", value_name = "PATH")]
+    db_path: PathBuf,
+
+    /// Listener binding. Accepts unix:/path, tcp:<port>, or pipe:<name>.
+    #[arg(long, value_name = "ENDPOINT", default_value = DEFAULT_LISTEN)]
+    listen: String,
+
+    /// Shared-secret token file used to authenticate RPC clients.
+    #[arg(long = "token-file", value_name = "PATH")]
+    token_file: Option<PathBuf>,
+
+    /// Tracing filter (info,warn,debug,trace). Can also be provided via SQLITE_WATCHER_LOG.
+    #[arg(
+        long = "log-level",
+        value_name = "FILTER",
+        default_value = "info",
+        env = "SQLITE_WATCHER_LOG"
+    )]
+    log_filter: String,
+
+    /// Interval in milliseconds between WAL file polls.
+    #[arg(
+        long = "poll-interval-ms",
+        default_value_t = 500,
+        value_parser = clap::value_parser!(u64).range(50..=60_000)
+    )]
+    poll_interval_ms: u64,
+
+    /// Minimum WAL byte growth required before emitting an event.
+    #[arg(
+        long = "min-event-bytes",
+        default_value_t = 1,
+        value_parser = clap::value_parser!(u64).range(1..=10_000_000)
+    )]
+    min_event_bytes: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum ListenAddress {
+    Unix(PathBuf),
+    Tcp { host: String, port: u16 },
+    Pipe(String),
+}
+
+impl fmt::Display for ListenAddress {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ListenAddress::Unix(path) => write!(f, "unix:{}", path.display()),
+            ListenAddress::Tcp { host, port } => write!(f, "tcp:{}:{}", host, port),
+            ListenAddress::Pipe(name) => write!(f, "pipe:{}", name),
+        }
+    }
+}
+
+impl FromStr for ListenAddress {
+    type Err = anyhow::Error;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        if let Some(path) = value.strip_prefix("unix:") {
+            if cfg!(windows) {
+                bail!("unix sockets are not supported on Windows");
+            }
+            if path.is_empty() {
+                bail!("unix listen path cannot be empty");
+            }
+            return Ok(ListenAddress::Unix(PathBuf::from(path)));
+        }
+
+        if let Some(port) = value.strip_prefix("tcp:") {
+            let port: u16 = port
+                .parse()
+                .map_err(|_| anyhow!("tcp listener must specify a numeric port"))?;
+            return Ok(ListenAddress::Tcp {
+                host: "127.0.0.1".to_string(),
+                port,
+            });
+        }
+
+        if let Some(name) = value.strip_prefix("pipe:") {
+            if cfg!(not(windows)) {
+                bail!("named pipes are only valid on Windows");
+            }
+            if name.is_empty() {
+                bail!("pipe name cannot be empty");
+            }
+            return Ok(ListenAddress::Pipe(name.to_string()));
+        }
+
+        bail!("listen endpoint must start with unix:/, tcp:, or pipe:");
+    }
+}
+
+#[derive(Debug, Clone)]
+struct WatcherConfig {
+    database_path: PathBuf,
+    listen: ListenAddress,
+    token_file: PathBuf,
+    poll_interval: Duration,
+    min_event_bytes: u64,
+}
+
+impl TryFrom<Cli> for WatcherConfig {
+    type Error = anyhow::Error;
+
+    fn try_from(args: Cli) -> Result<Self> {
+        let database_path = ensure_sqlite_file(&args.db_path)?;
+        let listen = ListenAddress::from_str(args.listen.trim())?;
+        let token_file = match args.token_file {
+            Some(path) => expand_home(path)?,
+            None => default_token_path()?,
+        };
+
+        Ok(Self {
+            database_path,
+            listen,
+            token_file,
+            poll_interval: Duration::from_millis(args.poll_interval_ms),
+            min_event_bytes: args.min_event_bytes,
+        })
+    }
+}
+
+fn ensure_sqlite_file(path: &Path) -> Result<PathBuf> {
+    if !path.exists() {
+        bail!("database path {} does not exist", path.display());
+    }
+    if !path.is_file() {
+        bail!("database path {} is not a file", path.display());
+    }
+    Ok(path
+        .canonicalize()
+        .with_context(|| format!("failed to canonicalize {}", path.display()))?)
+}
+
+fn default_token_path() -> Result<PathBuf> {
+    let home = dirs::home_dir().ok_or_else(|| anyhow!("unable to determine home directory"))?;
+    Ok(home.join(".seren/sqlite-watcher/token"))
+}
+
+fn expand_home(path: PathBuf) -> Result<PathBuf> {
+    let as_str = path.to_string_lossy();
+    if let Some(stripped) = as_str.strip_prefix("~/") {
+        let home = dirs::home_dir().ok_or_else(|| anyhow!("unable to determine home directory"))?;
+        return Ok(home.join(stripped));
+    }
+    if as_str == "~" {
+        let home = dirs::home_dir().ok_or_else(|| anyhow!("unable to determine home directory"))?;
+        return Ok(home);
+    }
+    Ok(path)
+}
+
+fn init_tracing(filter: &str) -> Result<()> {
+    let env_filter = EnvFilter::try_new(filter).or_else(|_| EnvFilter::try_new("info"))?;
+    tracing_subscriber::fmt()
+        .with_env_filter(env_filter)
+        .with_target(false)
+        .try_init()
+        .map_err(|err| anyhow!("failed to init tracing subscriber: {err}"))
+}
+
+fn main() -> Result<()> {
+    let cli = Cli::parse();
+    init_tracing(&cli.log_filter)?;
+    let config = WatcherConfig::try_from(cli)?;
+
+    tracing::info!(
+        db = %config.database_path.display(),
+        listen = %config.listen,
+        token = %config.token_file.display(),
+        poll_ms = config.poll_interval.as_millis(),
+        min_event_bytes = config.min_event_bytes,
+        "sqlite-watcher starting"
+    );
+
+    let (event_tx, event_rx) = mpsc::channel();
+    let _wal_handle = start_wal_watcher(
+        &config.database_path,
+        TailConfig {
+            poll_interval: config.poll_interval,
+            min_event_bytes: config.min_event_bytes,
+        },
+        event_tx,
+    )?;
+
+    for event in event_rx {
+        tracing::info!(
+            bytes_added = event.bytes_added,
+            wal_size = event.current_size,
+            "wal file grew"
+        );
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::Parser;
+
+    #[test]
+    fn parses_tcp_listener() {
+        let tmp = tempfile::NamedTempFile::new().unwrap();
+        let cli = Cli::try_parse_from([
+            "sqlite-watcher",
+            "--db",
+            tmp.path().to_str().unwrap(),
+            "--listen",
+            "tcp:6000",
+            "--token-file",
+            "./token",
+            "--log-level",
+            "debug",
+        ])
+        .expect("cli parsing failed");
+
+        let config = WatcherConfig::try_from(cli).expect("config conversion failed");
+        assert!(matches!(
+            config.listen,
+            ListenAddress::Tcp { host, port } if host == "127.0.0.1" && port == 6000
+        ));
+        assert!(config.token_file.ends_with("token"));
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn parses_unix_listener_default() {
+        let tmp = tempfile::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(_)));
+    }
+}