From 60b711c844f8624348d5d1dac3a625532a8e2a82 Mon Sep 17 00:00:00 2001
From: Owen Jacobson <owen@grimoire.ca>
Date: Fri, 27 Sep 2024 23:03:46 -0400
Subject: Delete expired messages out of band.

Trying to reliably do expiry mid-request was causing some anomalies:

* Creating a channel with a dup name would fail, then succeed after listing channels.

It was very hard to reason about which operations needed to trigger expiry, to fix this "correctly," so now expiry runs on every request.
---
 src/events/app.rs | 40 +++++++++++++++++++++++++---------------
 1 file changed, 25 insertions(+), 15 deletions(-)

(limited to 'src/events/app.rs')

diff --git a/src/events/app.rs b/src/events/app.rs
index 134e86a..03f3ee6 100644
--- a/src/events/app.rs
+++ b/src/events/app.rs
@@ -55,14 +55,35 @@ impl<'a> Events<'a> {
         Ok(event)
     }
 
+    pub async fn expire(&self, relative_to: &DateTime) -> Result<(), sqlx::Error> {
+        // Somewhat arbitrarily, expire after 90 days.
+        let expire_at = relative_to.to_owned() - TimeDelta::days(90);
+
+        let mut tx = self.db.begin().await?;
+        let expired = tx.message_events().expired(&expire_at).await?;
+
+        let mut events = Vec::with_capacity(expired.len());
+        for (channel, message) in expired {
+            let event = tx
+                .message_events()
+                .delete_expired(&channel, &message, relative_to)
+                .await?;
+            events.push(event);
+        }
+
+        tx.commit().await?;
+
+        for event in events {
+            self.broadcaster.broadcast(&event);
+        }
+
+        Ok(())
+    }
+
     pub async fn subscribe(
         &self,
-        subscribed_at: &DateTime,
         resume_at: ResumePoint,
     ) -> Result<impl Stream<Item = types::ResumableEvent> + std::fmt::Debug, sqlx::Error> {
-        // Somewhat arbitrarily, expire after 90 days.
-        let expire_at = subscribed_at.to_owned() - TimeDelta::days(90);
-
         let mut tx = self.db.begin().await?;
         let channels = tx.channels().all().await?;
 
@@ -81,8 +102,6 @@ impl<'a> Events<'a> {
         // querying the DB. We'll prune out duplicates later.
         let live_messages = self.broadcaster.subscribe();
 
-        tx.message_events().expire(&expire_at).await?;
-
         let mut replays = BTreeMap::new();
         let mut resume_live_at = resume_at.clone();
         for channel in channels {
@@ -107,9 +126,6 @@ impl<'a> Events<'a> {
         // * resume is redundant with the resume_at argument to
         //   `tx.broadcasts().replay(…)`.
         let live_messages = live_messages
-            // Sure, it's temporally improbable that we'll ever skip a message
-            // that's 90 days old, but there's no reason not to be thorough.
-            .filter(Self::skip_expired(&expire_at))
             // Filtering on the broadcast resume point filters out messages
             // before resume_at, and filters out messages duplicated from
             // stored_messages.
@@ -134,12 +150,6 @@ impl<'a> Events<'a> {
     ) -> impl for<'m> FnMut(&'m types::ChannelEvent) -> future::Ready<bool> {
         move |event| future::ready(resume_at.not_after(event))
     }
-    fn skip_expired(
-        expire_at: &DateTime,
-    ) -> impl for<'m> FnMut(&'m types::ChannelEvent) -> future::Ready<bool> {
-        let expire_at = expire_at.to_owned();
-        move |event| future::ready(expire_at < event.at)
-    }
 }
 
 #[derive(Debug, thiserror::Error)]
-- 
cgit v1.2.3