Allow deleting chat messages

crates/collab/src/db/queries/messages.rs

@@ -171,4 +171,44 @@ impl Database {
         })
         .await
     }
+
+    pub async fn remove_channel_message(
+        &self,
+        channel_id: ChannelId,
+        message_id: MessageId,
+        user_id: UserId,
+    ) -> Result<Vec<ConnectionId>> {
+        self.transaction(|tx| async move {
+            let mut rows = channel_chat_participant::Entity::find()
+                .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
+                .stream(&*tx)
+                .await?;
+
+            let mut is_participant = false;
+            let mut participant_connection_ids = Vec::new();
+            while let Some(row) = rows.next().await {
+                let row = row?;
+                if row.user_id == user_id {
+                    is_participant = true;
+                }
+                participant_connection_ids.push(row.connection());
+            }
+            drop(rows);
+
+            if !is_participant {
+                Err(anyhow!("not a chat participant"))?;
+            }
+
+            let result = channel_message::Entity::delete_by_id(message_id)
+                .filter(channel_message::Column::SenderId.eq(user_id))
+                .exec(&*tx)
+                .await?;
+            if result.rows_affected == 0 {
+                Err(anyhow!("no such message"))?;
+            }
+
+            Ok(participant_connection_ids)
+        })
+        .await
+    }
 }

crates/collab/src/rpc.rs

@@ -265,6 +265,7 @@ impl Server {
             .add_request_handler(join_channel_chat)
             .add_message_handler(leave_channel_chat)
             .add_request_handler(send_channel_message)
+            .add_request_handler(remove_channel_message)
             .add_request_handler(get_channel_messages)
             .add_request_handler(follow)
             .add_message_handler(unfollow)
@@ -2696,6 +2697,25 @@ async fn send_channel_message(
     Ok(())
 }
 
+async fn remove_channel_message(
+    request: proto::RemoveChannelMessage,
+    response: Response<proto::RemoveChannelMessage>,
+    session: Session,
+) -> Result<()> {
+    let channel_id = ChannelId::from_proto(request.channel_id);
+    let message_id = MessageId::from_proto(request.message_id);
+    let connection_ids = session
+        .db()
+        .await
+        .remove_channel_message(channel_id, message_id, session.user_id)
+        .await?;
+    broadcast(Some(session.connection_id), connection_ids, |connection| {
+        session.peer.send(connection, request.clone())
+    });
+    response.send(proto::Ack {})?;
+    Ok(())
+}
+
 async fn join_channel_chat(
     request: proto::JoinChannelChat,
     response: Response<proto::JoinChannelChat>,

crates/collab/src/tests/channel_message_tests.rs

@@ -1,5 +1,5 @@
 use crate::{rpc::RECONNECT_TIMEOUT, tests::TestServer};
-use channel::ChannelChat;
+use channel::{ChannelChat, ChannelMessageId};
 use gpui::{executor::Deterministic, ModelHandle, TestAppContext};
 use std::sync::Arc;
 
@@ -123,15 +123,92 @@ async fn test_rejoin_channel_chat(
     assert_messages(&channel_chat_b, expected_messages, cx_b);
 }
 
+#[gpui::test]
+async fn test_remove_channel_message(
+    deterministic: Arc<Deterministic>,
+    cx_a: &mut TestAppContext,
+    cx_b: &mut TestAppContext,
+    cx_c: &mut TestAppContext,
+) {
+    deterministic.forbid_parking();
+    let mut server = TestServer::start(&deterministic).await;
+    let client_a = server.create_client(cx_a, "user_a").await;
+    let client_b = server.create_client(cx_b, "user_b").await;
+    let client_c = server.create_client(cx_c, "user_c").await;
+
+    let channel_id = server
+        .make_channel(
+            "the-channel",
+            (&client_a, cx_a),
+            &mut [(&client_b, cx_b), (&client_c, cx_c)],
+        )
+        .await;
+
+    let channel_chat_a = client_a
+        .channel_store()
+        .update(cx_a, |store, cx| store.open_channel_chat(channel_id, cx))
+        .await
+        .unwrap();
+    let channel_chat_b = client_b
+        .channel_store()
+        .update(cx_b, |store, cx| store.open_channel_chat(channel_id, cx))
+        .await
+        .unwrap();
+
+    // Client A sends some messages.
+    channel_chat_a
+        .update(cx_a, |c, cx| c.send_message("one".into(), cx).unwrap())
+        .await
+        .unwrap();
+    channel_chat_a
+        .update(cx_a, |c, cx| c.send_message("two".into(), cx).unwrap())
+        .await
+        .unwrap();
+    channel_chat_a
+        .update(cx_a, |c, cx| c.send_message("three".into(), cx).unwrap())
+        .await
+        .unwrap();
+
+    // Clients A and B see all of the messages.
+    deterministic.run_until_parked();
+    let expected_messages = &["one", "two", "three"];
+    assert_messages(&channel_chat_a, expected_messages, cx_a);
+    assert_messages(&channel_chat_b, expected_messages, cx_b);
+
+    // Client A deletes one of their messages.
+    channel_chat_a
+        .update(cx_a, |c, cx| {
+            let ChannelMessageId::Saved(id) = c.message(1).id else {
+                panic!("message not saved")
+            };
+            c.remove_message(id, cx)
+        })
+        .await
+        .unwrap();
+
+    // Client B sees that the message is gone.
+    deterministic.run_until_parked();
+    let expected_messages = &["one", "three"];
+    assert_messages(&channel_chat_a, expected_messages, cx_a);
+    assert_messages(&channel_chat_b, expected_messages, cx_b);
+
+    // Client C joins the channel chat, and does not see the deleted message.
+    let channel_chat_c = client_c
+        .channel_store()
+        .update(cx_c, |store, cx| store.open_channel_chat(channel_id, cx))
+        .await
+        .unwrap();
+    assert_messages(&channel_chat_c, expected_messages, cx_c);
+}
+
 #[track_caller]
 fn assert_messages(chat: &ModelHandle<ChannelChat>, messages: &[&str], cx: &mut TestAppContext) {
-    chat.update(cx, |chat, _| {
-        assert_eq!(
-            chat.messages()
-                .iter()
-                .map(|m| m.body.as_str())
-                .collect::<Vec<_>>(),
-            messages
-        );
-    })
+    assert_eq!(
+        chat.read_with(cx, |chat, _| chat
+            .messages()
+            .iter()
+            .map(|m| m.body.clone())
+            .collect::<Vec<_>>(),),
+        messages
+    );
 }