Add DB-backed model change polling and startup pruning

crates/yaak-models/migrations/20260216000000_model-changes.sql

@@ -0,0 +1,12 @@
+CREATE TABLE model_changes
+(
+    id            INTEGER PRIMARY KEY AUTOINCREMENT,
+    model         TEXT                                                    NOT NULL,
+    model_id      TEXT                                                    NOT NULL,
+    change        TEXT                                                    NOT NULL,
+    update_source TEXT                                                    NOT NULL,
+    payload       TEXT                                                    NOT NULL,
+    created_at    DATETIME DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')) NOT NULL
+);
+
+CREATE INDEX idx_model_changes_created_at ON model_changes (created_at);

crates/yaak-models/src/db_context.rs

@@ -3,8 +3,8 @@ use crate::error::Error::ModelNotFound;
 use crate::error::Result;
 use crate::models::{AnyModel, UpsertModelInfo};
 use crate::util::{ModelChangeEvent, ModelPayload, UpdateSource};
-use log::error;
-use rusqlite::OptionalExtension;
+use rusqlite::types::Type;
+use rusqlite::{OptionalExtension, params};
 use sea_query::{
     Asterisk, Expr, Func, IntoColumnRef, IntoIden, IntoTableRef, OnConflict, Query, SimpleExpr,
     SqliteQueryBuilder,
@@ -14,10 +14,16 @@ use std::fmt::Debug;
 use std::sync::mpsc;
 
 pub struct DbContext<'a> {
-    pub(crate) events_tx: mpsc::Sender<ModelPayload>,
+    pub(crate) _events_tx: mpsc::Sender<ModelPayload>,
     pub(crate) conn: ConnectionOrTx<'a>,
 }
 
+#[derive(Debug, Clone)]
+pub struct PersistedModelChange {
+    pub id: i64,
+    pub payload: ModelPayload,
+}
+
 impl<'a> DbContext<'a> {
     pub(crate) fn find_one<'s, M>(
         &self,
@@ -180,9 +186,8 @@ impl<'a> DbContext<'a> {
             change: ModelChangeEvent::Upsert { created },
         };
 
-        if let Err(e) = self.events_tx.send(payload.clone()) {
-            error!("Failed to send model change {source:?}: {e:?}");
-        }
+        self.record_model_change(&payload)?;
+        let _ = self._events_tx.send(payload);
 
         Ok(m)
     }
@@ -203,9 +208,159 @@ impl<'a> DbContext<'a> {
             change: ModelChangeEvent::Delete,
         };
 
-        if let Err(e) = self.events_tx.send(payload) {
-            error!("Failed to send model change {source:?}: {e:?}");
-        }
+        self.record_model_change(&payload)?;
+        let _ = self._events_tx.send(payload);
+
         Ok(m.clone())
     }
+
+    fn record_model_change(&self, payload: &ModelPayload) -> Result<()> {
+        let payload_json = serde_json::to_string(payload)?;
+        let source_json = serde_json::to_string(&payload.update_source)?;
+        let change_json = serde_json::to_string(&payload.change)?;
+
+        self.conn.resolve().execute(
+            r#"
+                INSERT INTO model_changes (model, model_id, change, update_source, payload)
+                VALUES (?1, ?2, ?3, ?4, ?5)
+            "#,
+            params![
+                payload.model.model(),
+                payload.model.id(),
+                change_json,
+                source_json,
+                payload_json,
+            ],
+        )?;
+
+        Ok(())
+    }
+
+    pub fn latest_model_change_id(&self) -> Result<i64> {
+        let mut stmt = self.conn.prepare("SELECT COALESCE(MAX(id), 0) FROM model_changes")?;
+        Ok(stmt.query_row([], |row| row.get(0))?)
+    }
+
+    pub fn list_model_changes_after(
+        &self,
+        after_id: i64,
+        limit: usize,
+    ) -> Result<Vec<PersistedModelChange>> {
+        let mut stmt = self.conn.prepare(
+            r#"
+                SELECT id, payload
+                FROM model_changes
+                WHERE id > ?1
+                ORDER BY id ASC
+                LIMIT ?2
+            "#,
+        )?;
+
+        let items = stmt.query_map(params![after_id, limit as i64], |row| {
+            let id: i64 = row.get(0)?;
+            let payload_raw: String = row.get(1)?;
+            let payload = serde_json::from_str::<ModelPayload>(&payload_raw).map_err(|e| {
+                rusqlite::Error::FromSqlConversionFailure(1, Type::Text, Box::new(e))
+            })?;
+            Ok(PersistedModelChange { id, payload })
+        })?;
+
+        Ok(items.collect::<std::result::Result<Vec<_>, rusqlite::Error>>()?)
+    }
+
+    pub fn prune_model_changes_older_than_days(&self, days: i64) -> Result<usize> {
+        let offset = format!("-{days} days");
+        Ok(self.conn.resolve().execute(
+            r#"
+                DELETE FROM model_changes
+                WHERE created_at < STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', ?1)
+            "#,
+            params![offset],
+        )?)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::init_in_memory;
+    use crate::models::Workspace;
+
+    #[test]
+    fn records_model_changes_for_upsert_and_delete() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        let workspace = db
+            .upsert_workspace(
+                &Workspace {
+                    name: "Changes Test".to_string(),
+                    setting_follow_redirects: true,
+                    setting_validate_certificates: true,
+                    ..Default::default()
+                },
+                &UpdateSource::Sync,
+            )
+            .expect("Failed to upsert workspace");
+
+        let created_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(created_changes.len(), 1);
+        assert_eq!(created_changes[0].payload.model.id(), workspace.id);
+        assert_eq!(created_changes[0].payload.model.model(), "workspace");
+        assert!(matches!(
+            created_changes[0].payload.change,
+            ModelChangeEvent::Upsert { created: true }
+        ));
+        assert!(matches!(created_changes[0].payload.update_source, UpdateSource::Sync));
+
+        db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync)
+            .expect("Failed to delete workspace");
+
+        let all_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(all_changes.len(), 2);
+        assert!(matches!(all_changes[1].payload.change, ModelChangeEvent::Delete));
+        assert_eq!(
+            db.latest_model_change_id().expect("Failed to read latest ID"),
+            all_changes[1].id
+        );
+
+        let changes_after_first = db
+            .list_model_changes_after(all_changes[0].id, 10)
+            .expect("Failed to list changes after cursor");
+        assert_eq!(changes_after_first.len(), 1);
+        assert!(matches!(changes_after_first[0].payload.change, ModelChangeEvent::Delete));
+    }
+
+    #[test]
+    fn prunes_old_model_changes() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        db.upsert_workspace(
+            &Workspace {
+                name: "Prune Test".to_string(),
+                setting_follow_redirects: true,
+                setting_validate_certificates: true,
+                ..Default::default()
+            },
+            &UpdateSource::Sync,
+        )
+        .expect("Failed to upsert workspace");
+
+        let changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(changes.len(), 1);
+
+        db.conn
+            .resolve()
+            .execute(
+                "UPDATE model_changes SET created_at = '2000-01-01 00:00:00.000' WHERE id = ?1",
+                params![changes[0].id],
+            )
+            .expect("Failed to age model change row");
+
+        let pruned =
+            db.prune_model_changes_older_than_days(30).expect("Failed to prune model changes");
+        assert_eq!(pruned, 1);
+        assert!(db.list_model_changes_after(0, 10).expect("Failed to list changes").is_empty());
+    }
 }

crates/yaak-models/src/models.rs

@@ -2347,6 +2347,15 @@ macro_rules! define_any_model {
                     )*
                 }
             }
+
+            #[inline]
+            pub fn model(&self) -> &str {
+                match self {
+                    $(
+                        AnyModel::$type(inner) => &inner.model,
+                    )*
+                }
+            }
         }
 
         $(

crates/yaak-models/src/query_manager.rs

@@ -25,7 +25,7 @@ impl QueryManager {
             .expect("Failed to gain lock on DB")
             .get()
             .expect("Failed to get a new DB connection from the pool");
-        DbContext { events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Connection(conn) }
+        DbContext { _events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Connection(conn) }
     }
 
     pub fn with_conn<F, T>(&self, func: F) -> T
@@ -39,8 +39,10 @@ impl QueryManager {
             .get()
             .expect("Failed to get new DB connection from the pool");
 
-        let db_context =
-            DbContext { events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Connection(conn) };
+        let db_context = DbContext {
+            _events_tx: self.events_tx.clone(),
+            conn: ConnectionOrTx::Connection(conn),
+        };
 
         func(&db_context)
     }
@@ -62,8 +64,10 @@ impl QueryManager {
             .transaction_with_behavior(TransactionBehavior::Immediate)
             .expect("Failed to start DB transaction");
 
-        let db_context =
-            DbContext { events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Transaction(&tx) };
+        let db_context = DbContext {
+            _events_tx: self.events_tx.clone(),
+            conn: ConnectionOrTx::Transaction(&tx),
+        };
 
         match func(&db_context) {
             Ok(val) => {