init

src/routes/gateway/mod.rs

@@ -1,10 +1,14 @@
 use crate::domain::events::message::{
-    MessageCreatedEvent, MessageDeletedEvent, MessageUpdatedEvent,
+    MessageCreatedEvent, MessageDeletedEvent, MessageReactionAddedEvent,
+    MessageReactionRemovedEvent, MessageUpdatedEvent,
 };
 use crate::models::user::Model as User;
 use crate::routes::category::mapper::category_model_to_category_response;
 use crate::routes::channel::mapper::channel_model_to_channel_response;
-use crate::routes::message::mapper::message_model_to_message_response_with_server_id;
+use crate::routes::message::mapper::{
+    message_model_to_message_response_with_reactions,
+    message_model_to_message_response_with_server_id, reaction_model_to_response,
+};
 use crate::routes::server::mapper::server_model_to_server_response;
 use crate::services::Services;
 use axum::extract::ws::Message;
@@ -70,12 +74,22 @@ impl GatewayManager {
         event_bus.on_async::<MessageUpdatedEvent, _, _>("message_updated", move |event| {
             let manager = Arc::clone(&manager);
             async move {
+                let message_id = event.message.id;
+                let reactions = manager
+                    .services
+                    .message_reaction
+                    .grouped_for_messages(&[message_id])
+                    .await
+                    .ok()
+                    .and_then(|mut groups| groups.remove(&message_id))
+                    .unwrap_or_default();
                 manager.broadcast_message(
                     event.channel_id,
                     "update",
-                    message_model_to_message_response_with_server_id(
+                    message_model_to_message_response_with_reactions(
                         event.message,
                         event.server_id,
+                        reactions,
                     ),
                 );
             }
@@ -88,6 +102,36 @@ impl GatewayManager {
                 manager.broadcast_message(event.channel_id, "remove", event.message.id);
             }
         });
+
+        let manager = Arc::clone(self);
+        event_bus.on_async::<MessageReactionAddedEvent, _, _>(
+            "message_reaction_added",
+            move |event| {
+                let manager = Arc::clone(&manager);
+                async move {
+                    manager.broadcast_reaction(
+                        event.channel_id,
+                        "add",
+                        reaction_model_to_response(event.reaction),
+                    );
+                }
+            },
+        );
+
+        let manager = Arc::clone(self);
+        event_bus.on_async::<MessageReactionRemovedEvent, _, _>(
+            "message_reaction_removed",
+            move |event| {
+                let manager = Arc::clone(&manager);
+                async move {
+                    manager.broadcast_reaction(
+                        event.channel_id,
+                        "remove",
+                        reaction_model_to_response(event.reaction),
+                    );
+                }
+            },
+        );
     }
 
     pub(crate) fn add_client(&self, gateway_client: GatewayClient) {
@@ -126,6 +170,33 @@ impl GatewayManager {
             }
         }
     }
+
+    fn broadcast_reaction<T: serde::Serialize>(
+        &self,
+        channel_id: Uuid,
+        action: &'static str,
+        content: T,
+    ) {
+        let event = GatewayEvent {
+            namespace: "Reaction",
+            action,
+            content,
+        };
+        let Ok(json) = serde_json::to_string(&event) else {
+            return;
+        };
+
+        let users = self
+            .services
+            .realtime_registry
+            .users_for_channel(channel_id);
+        let clients = self.clients.read();
+        for (key, client) in clients.iter() {
+            if users.contains(&key.user_id) {
+                let _ = client.sender.send(Message::Text(json.clone().into()));
+            }
+        }
+    }
 }
 
 impl GatewayClient {