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.

Author: Mshehu5

src/payjoin/db.rs

@@ -16,6 +16,7 @@ pub type Result<T> = std::result::Result<T, Error>;
 
 /// Default filename for the payjoin database
 pub const DB_FILENAME: &str = "payjoin.sqlite";
+const SESSION_RETENTION_SECS: i64 = 30 * 24 * 60 * 60;
 
 /// Returns the current Unix timestamp in seconds
 #[inline]
@@ -109,6 +110,90 @@ 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
+                        )
+                    )
+                    OR (
+                        completed_at IS NULL
+                        AND session_id NOT IN (
+                            SELECT session_id FROM send_session_events
+                        )
+                    )",
+            )?;
+            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
+                        )
+                    )
+                    OR (
+                        completed_at IS NULL
+                        AND session_id NOT IN (
+                            SELECT session_id FROM receive_session_events
+                        )
+                    )",
+            )?;
+            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
+        };
+
+        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()?;
+        Ok(())
+    }
+
     /// Returns IDs of all active (incomplete) receive sessions
     pub fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {
         let conn = self.conn();

src/utils.rs

@@ -142,9 +142,9 @@ pub fn open_payjoin_db(
     use crate::payjoin::db::{DB_FILENAME, Database};
     let wallet_dir = prepare_home_dir(datadir)?.join(wallet_name);
     std::fs::create_dir_all(&wallet_dir).map_err(|e| Error::Generic(e.to_string()))?;
-    Ok(std::sync::Arc::new(Database::create(
-        wallet_dir.join(DB_FILENAME),
-    )?))
+    let db = std::sync::Arc::new(Database::create(wallet_dir.join(DB_FILENAME))?);
+    db.prune_expired_sessions()?;
+    Ok(db)
 }
 
 #[cfg(any(