[sqlite-watcher] add tonic-based watcher server

Author: taariq

Cargo.lock

@@ -7,15 +7,24 @@ description = "SQLite WAL tailer that streams change events to database-replicat
 license = "Apache-2.0"
 repository = "https://github.com/serenorg/database-replicator"
 readme = "README.md"
+build = "build.rs"
+
+[build-dependencies]
+tonic-build = "0.11"
 
 [dependencies]
 anyhow = "1.0"
 clap = { version = "4.4", features = ["derive", "env"] }
 dirs = "5.0"
+prost = "0.12"
 rusqlite = "0.30"
 serde_json = "1.0"
+tokio = { version = "1.35", features = ["macros", "rt-multi-thread", "signal", "fs"] }
+tonic = { version = "0.11", features = ["transport"] }
+tokio-stream = { version = "0.1", features = ["net"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [dev-dependencies]
 tempfile = "3.8"
+tokio = { version = "1.35", features = ["rt", "macros"] }

sqlite-watcher/Cargo.toml

@@ -32,6 +32,7 @@ Flag summary:
 - `--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.
+- `--listen` + `--token-file` now control the embedded gRPC server. Clients must send `Authorization: Bearer <token>` metadata when calling the `Watcher` service (see `proto/watcher.proto`). Unix sockets/pipes are placeholders until Ticket D is completed; TCP listens on `127.0.0.1:<port>`.
 
 ## Cross-platform notes
 

sqlite-watcher/README.md

@@ -0,0 +1,8 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tonic_build::configure()
+        .build_client(true)
+        .build_server(true)
+        .compile(&["proto/watcher.proto"], &["proto"])?;
+    println!("cargo:rerun-if-changed=proto/watcher.proto");
+    Ok(())
+}

sqlite-watcher/build.rs

@@ -0,0 +1,62 @@
+syntax = "proto3";
+
+package sqlitewatcher;
+
+message HealthCheckRequest {}
+message HealthCheckResponse {
+  string status = 1;
+}
+
+message ListChangesRequest {
+  uint32 limit = 1;
+}
+
+message Change {
+  int64 change_id = 1;
+  string table_name = 2;
+  string op = 3;
+  string primary_key = 4;
+  bytes payload = 5;
+  string wal_frame = 6;
+  string cursor = 7;
+}
+
+message ListChangesResponse {
+  repeated Change changes = 1;
+}
+
+message AckChangesRequest {
+  int64 up_to_change_id = 1;
+}
+
+message AckChangesResponse {
+  uint64 acknowledged = 1;
+}
+
+message GetStateRequest {
+  string table_name = 1;
+}
+
+message GetStateResponse {
+  bool exists = 1;
+  int64 last_change_id = 2;
+  string last_wal_frame = 3;
+  string cursor = 4;
+}
+
+message SetStateRequest {
+  string table_name = 1;
+  int64 last_change_id = 2;
+  string last_wal_frame = 3;
+  string cursor = 4;
+}
+
+message SetStateResponse {}
+
+service Watcher {
+  rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse);
+  rpc ListChanges(ListChangesRequest) returns (ListChangesResponse);
+  rpc AckChanges(AckChangesRequest) returns (AckChangesResponse);
+  rpc GetState(GetStateRequest) returns (GetStateResponse);
+  rpc SetState(SetStateRequest) returns (SetStateResponse);
+}

sqlite-watcher/proto/watcher.proto

@@ -1,4 +1,9 @@
 pub mod change;
 pub mod decoder;
 pub mod queue;
+pub mod server;
 pub mod wal;
+
+pub mod watcher_proto {
+    tonic::include_proto!("sqlitewatcher");
+}

sqlite-watcher/src/lib.rs

@@ -1,4 +1,6 @@
 use std::fmt;
+use std::fs;
+use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
 use std::sync::mpsc;
@@ -8,6 +10,7 @@ use anyhow::{anyhow, bail, Context, Result};
 use clap::Parser;
 use sqlite_watcher::decoder::WalGrowthDecoder;
 use sqlite_watcher::queue::ChangeQueue;
+use sqlite_watcher::server::TcpServerHandle;
 use sqlite_watcher::wal::{start_wal_watcher, WalWatcherConfig as TailConfig};
 use tracing_subscriber::EnvFilter;
 
@@ -206,6 +209,7 @@ fn main() -> Result<()> {
     let cli = Cli::parse();
     init_tracing(&cli.log_filter)?;
     let config = WatcherConfig::try_from(cli)?;
+    let auth_token = read_token_file(&config.token_file)?;
 
     tracing::info!(
         db = %config.database_path.display(),
@@ -219,6 +223,7 @@ fn main() -> Result<()> {
 
     let queue = ChangeQueue::open(&config.queue_path)?;
     let decoder = WalGrowthDecoder::default();
+    let server_handle = start_grpc_server(&config.listen, &config.queue_path, &auth_token)?;
     let (event_tx, event_rx) = mpsc::channel();
     let _wal_handle = start_wal_watcher(
         &config.database_path,
@@ -246,6 +251,7 @@ fn main() -> Result<()> {
         }
     }
 
+    drop(server_handle);
     Ok(())
 }
 
@@ -261,6 +267,43 @@ fn process_wal_event(
     Ok(ids)
 }
 
+fn read_token_file(path: &Path) -> Result<String> {
+    let contents = fs::read_to_string(path)
+        .with_context(|| format!("failed to read token file {}", path.display()))?;
+    let token = contents.trim().to_string();
+    if token.is_empty() {
+        bail!("token file {} is empty", path.display());
+    }
+    Ok(token)
+}
+
+fn start_grpc_server(
+    listen: &ListenAddress,
+    queue_path: &Path,
+    token: &str,
+) -> Result<Option<TcpServerHandle>> {
+    match listen {
+        ListenAddress::Tcp { host, port } => {
+            let addr: SocketAddr = format!("{}:{}", host, port)
+                .parse()
+                .with_context(|| format!("invalid tcp listen address {host}:{port}"))?;
+            let handle = TcpServerHandle::spawn(addr, queue_path.to_path_buf(), token.to_string())?;
+            Ok(Some(handle))
+        }
+        ListenAddress::Unix(path) => {
+            tracing::warn!(
+                path = %path.display(),
+                "unix socket gRPC transport is not yet implemented"
+            );
+            Ok(None)
+        }
+        ListenAddress::Pipe(name) => {
+            tracing::warn!(pipe = name, "named pipe transport is not yet implemented");
+            Ok(None)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;