[retention] Add setting for auto removing notifications

ntfy-daemon/src/message_repo/migrations/03.sql

@@ -0,0 +1 @@
+ALTER TABLE subscription ADD COLUMN retention_hours INTEGER DEFAULT 0;

ntfy-daemon/src/message_repo/mod.rs

@@ -31,6 +31,7 @@ impl Db {
         conn.execute_batch(include_str!("./migrations/00.sql"))?;
         conn.execute_batch(include_str!("./migrations/01.sql"))?;
         conn.execute_batch(include_str!("./migrations/02.sql"))?;
+        conn.execute_batch(include_str!("./migrations/03.sql"))?;
         Ok(())
     }
     fn get_or_insert_server(&mut self, server: &str) -> Result<i64> {
@@ -56,9 +57,14 @@ impl Db {
         tx.commit()?;
         res
     }
-    pub fn insert_message(&mut self, server: &str, json_data: &str) -> Result<(), Error> {
+    pub fn insert_message(
+        &mut self,
+        server: &str,
+        topic: &str,
+        json_data: &str,
+    ) -> Result<(), Error> {
         let server_id = self.get_or_insert_server(server)?;
-        let res = self.conn.read().unwrap().execute(
+        let res = self.conn.write().unwrap().execute(
             "INSERT INTO message (server, data) VALUES (?1, ?2)",
             params![server_id, json_data],
         );
@@ -69,9 +75,48 @@ impl Db {
                 Err(Error::DuplicateMessage)
             }
             Err(e) => Err(Error::Db(e)),
-            Ok(_) => Ok(()),
+            Ok(_) => {
+                self.cleanup_messages(server, topic)?;
+                Ok(())
+            }
         }
     }
+
+    fn server_id_from_endpoint(conn: &Connection, endpoint: &str) -> Result<i64> {
+        conn.query_row(
+            "SELECT id FROM server WHERE endpoint = ?1",
+            params![endpoint],
+            |row| row.get(0),
+        )
+    }
+
+    pub fn cleanup_messages(&self, server: &str, topic: &str) -> Result<(), Error> {
+        let cutoff = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
+
+        let conn = self.conn.read().unwrap();
+        let mut stmt = conn.prepare(
+            "SELECT retention_hours FROM subscription sub
+             JOIN server s ON sub.server = s.id
+             WHERE s.endpoint = ?1 AND sub.topic = ?2 AND sub.retention_hours > 0",
+        )?;
+
+        let retention_hours: Option<u32> = stmt
+            .query_row(params![server, topic], |row| row.get(0))
+            .ok();
+
+        if let Some(hours) = retention_hours {
+            let cutoff = cutoff.saturating_sub((hours as u64) * 3600);
+            conn.execute(
+                "DELETE FROM message WHERE server = ?1 AND topic = ?2 AND timestamp < ?3",
+                params![Self::server_id_from_endpoint(&conn, server)?, topic, cutoff],
+            )?;
+        }
+        Ok(())
+    }
+
     pub fn list_messages(
         &self,
         server: &str,
@@ -117,14 +162,15 @@ impl Db {
     pub fn insert_subscription(&mut self, sub: models::Subscription) -> Result<(), Error> {
         let server_id = self.get_or_insert_server(&sub.server)?;
         self.conn.read().unwrap().execute(
-            "INSERT INTO subscription (server, topic, display_name, reserved, muted, archived) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+            "INSERT INTO subscription (server, topic, display_name, reserved, muted, archived, retention_hours) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
             params![
                 server_id,
                 sub.topic,
                 sub.display_name,
                 sub.reserved,
                 sub.muted,
-                sub.archived
+                sub.archived,
+                sub.retention_hours
             ],
         )?;
         Ok(())
@@ -144,7 +190,7 @@ impl Db {
     pub fn list_subscriptions(&mut self) -> Result<Vec<models::Subscription>, Error> {
         let conn = self.conn.read().unwrap();
         let mut stmt = conn.prepare(
-            "SELECT server.endpoint, sub.topic, sub.display_name, sub.reserved, sub.muted, sub.archived, sub.symbolic_icon, sub.read_until
+            "SELECT server.endpoint, sub.topic, sub.display_name, sub.reserved, sub.muted, sub.archived, sub.symbolic_icon, sub.read_until, sub.retention_hours
             FROM subscription sub
             JOIN server ON server.id = sub.server
             ORDER BY server.endpoint, sub.display_name, sub.topic
@@ -160,6 +206,7 @@ impl Db {
                 archived: row.get(5)?,
                 symbolic_icon: row.get(6)?,
                 read_until: row.get(7)?,
+                retention_hours: row.get(8)?,
             })
         })?;
         let subs: Result<Vec<_>, rusqlite::Error> = rows.collect();
@@ -170,14 +217,15 @@ impl Db {
         let server_id = self.get_or_insert_server(&sub.server)?;
         let res = self.conn.read().unwrap().execute(
             "UPDATE subscription
-            SET display_name = ?1, reserved = ?2, muted = ?3, archived = ?4, read_until = ?5
-            WHERE server = ?6 AND topic = ?7",
+            SET display_name = ?1, reserved = ?2, muted = ?3, archived = ?4, read_until = ?5, retention_hours = ?6
+            WHERE server = ?7 AND topic = ?8",
             params![
                 sub.display_name,
                 sub.reserved,
                 sub.muted,
                 sub.archived,
                 sub.read_until,
+                sub.retention_hours,
                 server_id,
                 sub.topic,
             ],

ntfy-daemon/src/models.rs

@@ -177,6 +177,7 @@ pub struct Subscription {
     pub reserved: bool,
     pub symbolic_icon: Option<String>,
     pub read_until: u64,
+    pub retention_hours: u32,
 }
 
 impl Subscription {
@@ -225,6 +226,7 @@ pub struct SubscriptionBuilder {
     reserved: bool,
     symbolic_icon: Option<String>,
     display_name: String,
+    retention_hours: u32,
 }
 
 impl SubscriptionBuilder {
@@ -237,6 +239,7 @@ impl SubscriptionBuilder {
             reserved: false,
             symbolic_icon: None,
             display_name: String::new(),
+            retention_hours: 0,
         }
     }
 
@@ -270,6 +273,11 @@ impl SubscriptionBuilder {
         self
     }
 
+    pub fn retention_hours(mut self, retention_hours: u32) -> Self {
+        self.retention_hours = retention_hours;
+        self
+    }
+
     pub fn build(self) -> Result<Subscription, Error> {
         let res = Subscription {
             server: self.server,
@@ -280,6 +288,7 @@ impl SubscriptionBuilder {
             symbolic_icon: self.symbolic_icon,
             display_name: self.display_name,
             read_until: 0,
+            retention_hours: self.retention_hours,
         };
         res.validate()
     }

ntfy-daemon/src/subscription.rs

@@ -232,7 +232,11 @@ impl SubscriptionActor {
         // Store in database
         let already_stored: bool = {
             let json_ev = &serde_json::to_string(&msg).unwrap();
-            match self.env.db.insert_message(&self.model.server, json_ev) {
+            match self
+                .env
+                .db
+                .insert_message(&self.model.server, &self.model.topic, json_ev)
+            {
                 Err(Error::DuplicateMessage) => {
                     warn!(topic=?self.model.topic, "received duplicate message");
                     true