From 0a5599c60d20ccc2223779eeba5dc91a95ea0fe5 Mon Sep 17 00:00:00 2001
From: Owen Jacobson <owen@grimoire.ca>
Date: Thu, 3 Oct 2024 20:17:07 -0400
Subject: Add endpoints for deleting channels and messages.

It is deliberate that the expire() functions do not use them. To avoid races, the transactions must be committed before events get sent, in both cases, which makes them structurally pretty different.
---
 src/message/app.rs | 61 +++++++++++++++++++++++++++++++++++++++---------------
 1 file changed, 44 insertions(+), 17 deletions(-)

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

diff --git a/src/message/app.rs b/src/message/app.rs
index 51f772e..1d34c14 100644
--- a/src/message/app.rs
+++ b/src/message/app.rs
@@ -2,12 +2,12 @@ use chrono::TimeDelta;
 use itertools::Itertools;
 use sqlx::sqlite::SqlitePool;
 
-use super::{repo::Provider as _, Message};
+use super::{repo::Provider as _, Id, Message};
 use crate::{
     channel::{self, repo::Provider as _},
     clock::DateTime,
     db::NotFound as _,
-    event::{broadcaster::Broadcaster, repo::Provider as _, Sequence},
+    event::{broadcaster::Broadcaster, repo::Provider as _, Event, Sequence},
     login::Login,
 };
 
@@ -27,13 +27,13 @@ impl<'a> Messages<'a> {
         sender: &Login,
         sent_at: &DateTime,
         body: &str,
-    ) -> Result<Message, Error> {
+    ) -> Result<Message, SendError> {
         let mut tx = self.db.begin().await?;
         let channel = tx
             .channels()
             .by_id(channel)
             .await
-            .not_found(|| Error::ChannelNotFound(channel.clone()))?;
+            .not_found(|| SendError::ChannelNotFound(channel.clone()))?;
         let sent = tx.sequence().next(sent_at).await?;
         let message = tx
             .messages()
@@ -41,24 +41,40 @@ impl<'a> Messages<'a> {
             .await?;
         tx.commit().await?;
 
-        for event in message.events() {
-            self.events.broadcast(event);
-        }
+        self.events
+            .broadcast(message.events().map(Event::from).collect::<Vec<_>>());
 
         Ok(message.snapshot())
     }
 
+    pub async fn delete(&self, message: &Id, deleted_at: &DateTime) -> Result<(), DeleteError> {
+        let mut tx = self.db.begin().await?;
+        let deleted = tx.sequence().next(deleted_at).await?;
+        let message = tx.messages().delete(message, &deleted).await?;
+        tx.commit().await?;
+
+        self.events.broadcast(
+            message
+                .events()
+                .filter(Sequence::start_from(deleted.sequence))
+                .map(Event::from)
+                .collect::<Vec<_>>(),
+        );
+
+        Ok(())
+    }
+
     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.messages().expired(&expire_at).await?;
 
+        let expired = tx.messages().expired(&expire_at).await?;
         let mut events = Vec::with_capacity(expired.len());
-        for (channel, message) in expired {
+        for message in expired {
             let deleted = tx.sequence().next(relative_to).await?;
-            let message = tx.messages().delete(&channel, &message, &deleted).await?;
+            let message = tx.messages().delete(&message, &deleted).await?;
             events.push(
                 message
                     .events()
@@ -68,21 +84,32 @@ impl<'a> Messages<'a> {
 
         tx.commit().await?;
 
-        for event in events
-            .into_iter()
-            .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence)
-        {
-            self.events.broadcast(event);
-        }
+        self.events.broadcast(
+            events
+                .into_iter()
+                .kmerge_by(|a, b| a.instant.sequence < b.instant.sequence)
+                .map(Event::from)
+                .collect::<Vec<_>>(),
+        );
 
         Ok(())
     }
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum Error {
+pub enum SendError {
+    #[error("channel {0} not found")]
+    ChannelNotFound(channel::Id),
+    #[error(transparent)]
+    DatabaseError(#[from] sqlx::Error),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum DeleteError {
     #[error("channel {0} not found")]
     ChannelNotFound(channel::Id),
+    #[error("message {0} not found")]
+    NotFound(Id),
     #[error(transparent)]
     DatabaseError(#[from] sqlx::Error),
 }
-- 
cgit v1.2.3