Prune stale payjoin sessions on DB open

Delete payjoin sessions older than 30 days when the payjoin database is accessed. Remove related event rows in the same cleanup pass.

diff --git a/src/payjoin/db.rs b/src/payjoin/db.rs
index 39089e144b0007f74e6db4f4066811b1e6d5cda2..69ead104a5d0c9580bfe2c00fbde826b72a814e3 100644 (file)
--- a/src/payjoin/db.rs
+++ b/src/payjoin/db.rs
@@ -60,6 +60,7 @@ impl From<Error> for payjoin::ImplementationError {
 
 /// Default filename for the payjoin database
 pub const DB_FILENAME: &str = "payjoin.sqlite";
+const SESSION_RETENTION_SECS: i64 = 30 * 24 * 60 * 60;
 
 pub fn open_payjoin_db(
     datadir: Option<PathBuf>,
@@ -67,7 +68,9 @@ pub fn open_payjoin_db(
 ) -> std::result::Result<Arc<Database>, BDKCliError> {
     let wallet_dir = prepare_home_dir(datadir)?.join(wallet_name);
     std::fs::create_dir_all(&wallet_dir).map_err(|e| BDKCliError::Generic(e.to_string()))?;
-    Ok(Arc::new(Database::create(wallet_dir.join(DB_FILENAME))?))
+    let db = Arc::new(Database::create(wallet_dir.join(DB_FILENAME))?);
+    db.prune_expired_sessions()?;
+    Ok(db)
 }
 
 /// Returns the current Unix timestamp in seconds
@@ -162,6 +165,83 @@ impl Database {
         Ok(was_seen_before)
     }
 
+    /// Removes old completed sessions and stale incomplete sessions plus their event logs.
+    pub fn prune_expired_sessions(&self) -> Result<()> {
+        let cutoff = now() - SESSION_RETENTION_SECS;
+        let mut conn = self.conn();
+        let tx = conn.transaction()?;
+        let stale_send_session_ids = {
+            let mut stmt = tx.prepare(
+                "SELECT session_id FROM send_sessions
+                 WHERE (completed_at IS NOT NULL AND completed_at < ?1)
+                    OR (
+                        completed_at IS NULL
+                        AND session_id IN (
+                            SELECT session_id FROM send_session_events
+                            GROUP BY session_id
+                            HAVING MAX(created_at) < ?1
+                        )
+                    )",
+            )?;
+            let rows = stmt.query_map(params![cutoff], |row| row.get::<_, i64>(0))?;
+            let mut ids = Vec::new();
+            for row in rows {
+                ids.push(row?);
+            }
+            ids
+        };
+        let stale_receive_session_ids = {
+            let mut stmt = tx.prepare(
+                "SELECT session_id FROM receive_sessions
+                 WHERE (completed_at IS NOT NULL AND completed_at < ?1)
+                    OR (
+                        completed_at IS NULL
+                        AND session_id IN (
+                            SELECT session_id FROM receive_session_events
+                            GROUP BY session_id
+                            HAVING MAX(created_at) < ?1
+                        )
+                    )",
+            )?;
+            let rows = stmt.query_map(params![cutoff], |row| row.get::<_, i64>(0))?;
+            let mut ids = Vec::new();
+            for row in rows {
+                ids.push(row?);
+            }
+            ids
+        };
+        let deleted_any =
+            !stale_send_session_ids.is_empty() || !stale_receive_session_ids.is_empty();
+
+        for session_id in stale_send_session_ids {
+            tx.execute(
+                "DELETE FROM send_session_events WHERE session_id = ?1",
+                params![session_id],
+            )?;
+            tx.execute(
+                "DELETE FROM send_sessions WHERE session_id = ?1",
+                params![session_id],
+            )?;
+        }
+
+        for session_id in stale_receive_session_ids {
+            tx.execute(
+                "DELETE FROM receive_session_events WHERE session_id = ?1",
+                params![session_id],
+            )?;
+            tx.execute(
+                "DELETE FROM receive_sessions WHERE session_id = ?1",
+                params![session_id],
+            )?;
+        }
+
+        tx.commit()?;
+        if deleted_any {
+            conn.execute("VACUUM", [])?;
+        }
+        Ok(())
+    }
+
     /// Returns IDs of all active (incomplete) receive sessions
     pub fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {
         let conn = self.conn();