Fix sqlite watcher clippy warnings

Author: taariq

PR_BODY.md

@@ -0,0 +1,8 @@
+## Summary
+- fix sqlite sync change-state handling by treating watcher wal_frame/cursor fields as optional strings and cleaning up unused code
+- implement FromStr for sqlite ChangeOperation and resolve needless borrow lints in queue/server modules
+- keep clippy happy by applying the suggested clamp change and ensuring proto tests build
+
+## Testing
+- cargo clippy
+- cargo test

sqlite-watcher/src/queue.rs

@@ -1,5 +1,6 @@
 use std::fs;
 use std::path::{Path, PathBuf};
+use std::str::FromStr;
 
 use anyhow::{anyhow, Context, Result};
 use rusqlite::{params, Connection, OptionalExtension, Row};
@@ -41,8 +42,12 @@ impl ChangeOperation {
             ChangeOperation::Delete => "delete",
         }
     }
+}
+
+impl FromStr for ChangeOperation {
+    type Err = anyhow::Error;
 
-    pub fn from_str(value: &str) -> Result<Self> {
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
         match value {
             "insert" => Ok(ChangeOperation::Insert),
             "update" => Ok(ChangeOperation::Update),
@@ -98,8 +103,8 @@ impl ChangeQueue {
         }
         let conn = Connection::open(path)
             .with_context(|| format!("failed to open queue database {}", path.display()))?;
-        conn.pragma_update(None, "journal_mode", &"wal").ok();
-        conn.pragma_update(None, "synchronous", &"normal").ok();
+        conn.pragma_update(None, "journal_mode", "wal").ok();
+        conn.pragma_update(None, "synchronous", "normal").ok();
         conn.execute_batch(SCHEMA)
             .context("failed to initialize change queue schema")?;
         Ok(Self {
@@ -132,7 +137,7 @@ impl ChangeQueue {
         let mut rows = stmt.query([limit as i64])?;
         let mut results = Vec::new();
         while let Some(row) = rows.next()? {
-            results.push(row_to_change(&row)?);
+            results.push(row_to_change(row)?);
         }
         Ok(results)
     }

sqlite-watcher/src/server.rs

@@ -193,7 +193,7 @@ impl Watcher for WatcherService {
         &self,
         request: Request<ListChangesRequest>,
     ) -> Result<Response<ListChangesResponse>, Status> {
-        let limit = request.get_ref().limit.max(1).min(10_000) as usize;
+        let limit = request.get_ref().limit.clamp(1, 10_000) as usize;
         let queue = self.queue().map_err(internal_err)?;
         let rows = queue.fetch_batch(limit).map_err(internal_err)?;
         let changes = rows.into_iter().map(change_to_proto).collect();

src/commands/sync_sqlite.rs

@@ -1,13 +1,11 @@
-use std::collections::HashMap;
-use std::path::{Path, PathBuf};
-use std::time::Duration;
-
 use anyhow::{anyhow, bail, Context, Result};
 use clap::ValueEnum;
 use sqlite_watcher::watcher_proto::watcher_client::WatcherClient;
 use sqlite_watcher::watcher_proto::{
     AckChangesRequest, GetStateRequest, HealthCheckRequest, ListChangesRequest, SetStateRequest,
 };
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
 use tokio_postgres::Client;
 use tonic::codegen::InterceptedService;
 use tonic::service::Interceptor;
@@ -17,7 +15,6 @@ use tower::service_fn;
 
 use crate::jsonb::writer::{delete_jsonb_rows, insert_jsonb_batch, upsert_jsonb_rows};
 
-const DEFAULT_BATCH_LIMIT: u32 = 500;
 const GLOBAL_STATE_KEY: &str = "_global";
 
 #[derive(Debug, Clone, Copy, ValueEnum, PartialEq, Eq)]
@@ -63,7 +60,7 @@ pub async fn run(opts: SyncSqliteOptions) -> Result<()> {
     let mut processed_any = false;
 
     loop {
-        let mut req = Request::new(ListChangesRequest {
+        let req = Request::new(ListChangesRequest {
             limit: opts.batch_size.max(1),
         });
         let changes = watcher
@@ -160,8 +157,8 @@ async fn apply_changes(
             change.table_name.clone(),
             TableState {
                 last_change_id: change.change_id,
-                wal_frame: change.wal_frame.clone(),
-                cursor: change.cursor.clone(),
+                wal_frame: non_empty_string(&change.wal_frame),
+                cursor: non_empty_string(&change.cursor),
             },
         );
     }
@@ -235,6 +232,14 @@ fn mode_string(mode: IncrementalMode) -> &'static str {
     }
 }
 
+fn non_empty_string(value: &str) -> Option<String> {
+    if value.is_empty() {
+        None
+    } else {
+        Some(value.to_owned())
+    }
+}
+
 fn load_token(path: Option<&Path>) -> Result<String> {
     let token_path = path
         .map(|p| p.to_path_buf())
@@ -399,17 +404,17 @@ mod tests {
                 op: "insert".into(),
                 primary_key: "1".into(),
                 payload: serde_json::to_vec(&serde_json::json!({"a":1})).unwrap(),
-                wal_frame: None,
-                cursor: None,
+                wal_frame: String::new(),
+                cursor: String::new(),
             },
             Change {
                 change_id: 2,
                 table_name: "foo".into(),
                 op: "delete".into(),
                 primary_key: "2".into(),
                 payload: Vec::new(),
-                wal_frame: None,
-                cursor: None,
+                wal_frame: String::new(),
+                cursor: String::new(),
             },
         ];
         let mut per_table: HashMap<String, TableBatch> = HashMap::new();