Refactor messages propagation

matrixgw_backend/src/broadcast_messages.rs

@@ -8,6 +8,13 @@ use matrix_sdk::sync::SyncResponse;
 
 pub type BroadcastSender = tokio::sync::broadcast::Sender<BroadcastMessage>;
 
+#[derive(Debug, Clone)]
+pub struct BxRoomEvent<E> {
+    pub user: UserEmail,
+    pub event: Box<E>,
+    pub room: Room,
+}
+
 /// Broadcast messages
 #[derive(Debug, Clone)]
 pub enum BroadcastMessage {
@@ -20,23 +27,11 @@ pub enum BroadcastMessage {
     /// Matrix sync thread has been interrupted
     SyncThreadStopped(MatrixSyncTaskID),
     /// New room message
-    RoomMessageEvent {
-        user: UserEmail,
-        event: Box<OriginalSyncRoomMessageEvent>,
-        room: Room,
-    },
+    RoomMessageEvent(BxRoomEvent<OriginalSyncRoomMessageEvent>),
     /// New reaction message
-    ReactionEvent {
-        user: UserEmail,
-        event: Box<OriginalSyncReactionEvent>,
-        room: Room,
-    },
+    ReactionEvent(BxRoomEvent<OriginalSyncReactionEvent>),
     /// New room redaction
-    RoomRedactionEvent {
-        user: UserEmail,
-        event: Box<OriginalSyncRoomRedactionEvent>,
-        room: Room,
-    },
+    RoomRedactionEvent(BxRoomEvent<OriginalSyncRoomRedactionEvent>),
     /// Raw Matrix sync response
     MatrixSyncResponse { user: UserEmail, sync: SyncResponse },
 }

matrixgw_backend/src/controllers/ws_controller.rs

@@ -5,6 +5,7 @@ use crate::extractors::auth_extractor::{AuthExtractor, AuthenticatedMethod};
 use crate::extractors::matrix_client_extractor::MatrixClientExtractor;
 use crate::matrix_connection::matrix_client::MatrixClient;
 use crate::matrix_connection::matrix_manager::MatrixManagerMsg;
+use crate::users::UserEmail;
 use actix_web::dev::Payload;
 use actix_web::{FromRequest, HttpRequest, HttpResponse, web};
 use actix_ws::Message;
@@ -19,27 +20,50 @@ use tokio::sync::broadcast;
 use tokio::sync::broadcast::Receiver;
 use tokio::time::interval;
 
+#[derive(Debug, serde::Serialize)]
+pub struct WsRoomEvent<E> {
+    pub event: Box<E>,
+    pub room_id: OwnedRoomId,
+}
+
 /// Messages sent to the client
 #[derive(Debug, serde::Serialize)]
 #[serde(tag = "type")]
 pub enum WsMessage {
     /// Room message event
-    RoomMessageEvent {
-        event: Box<RoomMessageEventContent>,
-        room_id: OwnedRoomId,
-    },
+    RoomMessageEvent(WsRoomEvent<RoomMessageEventContent>),
 
     /// Room reaction event
-    RoomReactionEvent {
-        event: Box<ReactionEventContent>,
-        room_id: OwnedRoomId,
-    },
+    RoomReactionEvent(WsRoomEvent<ReactionEventContent>),
 
     /// Room reaction event
-    RoomRedactionEvent {
-        event: Box<RoomRedactionEventContent>,
-        room_id: OwnedRoomId,
-    },
+    RoomRedactionEvent(WsRoomEvent<RoomRedactionEventContent>),
+}
+
+impl WsMessage {
+    pub fn from_bx_message(msg: &BroadcastMessage, user: &UserEmail) -> Option<Self> {
+        match msg {
+            BroadcastMessage::RoomMessageEvent(evt) if &evt.user == user => {
+                Some(Self::RoomMessageEvent(WsRoomEvent {
+                    event: Box::new(evt.event.content.clone()),
+                    room_id: evt.room.room_id().to_owned(),
+                }))
+            }
+            BroadcastMessage::ReactionEvent(evt) if &evt.user == user => {
+                Some(Self::RoomReactionEvent(WsRoomEvent {
+                    event: Box::new(evt.event.content.clone()),
+                    room_id: evt.room.room_id().to_owned(),
+                }))
+            }
+            BroadcastMessage::RoomRedactionEvent(evt) if &evt.user == user => {
+                Some(Self::RoomRedactionEvent(WsRoomEvent {
+                    event: Box::new(evt.event.content.clone()),
+                    room_id: evt.room.room_id().to_owned(),
+                }))
+            }
+            _ => None,
+        }
+    }
 }
 
 /// Main WS route
@@ -108,8 +132,8 @@ pub async fn ws_handler(
                     Err(broadcast::error::RecvError::Lagged(_)) => continue,
                 };
 
-                match msg {
-                    BroadcastMessage::APITokenDeleted(t)  => {
+                match (&msg, WsMessage::from_bx_message(&msg, &auth.user.email)) {
+                    (BroadcastMessage::APITokenDeleted(t), _)  => {
                         match &auth.method{
                             AuthenticatedMethod::Token(tok) if tok.id == t.id => {
                                 log::info!(
@@ -123,39 +147,17 @@ pub async fn ws_handler(
                         }
 
                     },
-                    BroadcastMessage::UserDisconnectedFromMatrix(mail) if mail == auth.user.email => {
+                    (BroadcastMessage::UserDisconnectedFromMatrix(mail), _) if mail == &auth.user.email => {
                         log::info!(
                             "closing WS session of user {mail:?} as user was disconnected from Matrix"
                         );
                         break None;
                     }
 
-                    BroadcastMessage::RoomMessageEvent{user, event, room} if user == auth.user.email => {
+                    (_, Some(message)) => {
                         // Send the message to the websocket
-                        if let Ok(msg) = serde_json::to_string(&WsMessage::RoomMessageEvent {
-                            event: Box::new(event.content),
-                            room_id: room.room_id().to_owned(),
-                        }) && let Err(e) = session.text(msg).await {
-                            log::error!("Failed to send SyncEvent: {e}");
-                        }
-                    }
-
-                    BroadcastMessage::ReactionEvent{user, event, room} if user == auth.user.email => {
-                        // Send the message to the websocket
-                        if let Ok(msg) = serde_json::to_string(&WsMessage::RoomReactionEvent {
-                            event: Box::new(event.content),
-                            room_id: room.room_id().to_owned(),
-                        }) && let Err(e) = session.text(msg).await {
-                            log::error!("Failed to send SyncEvent: {e}");
-                        }
-                    }
-
-                    BroadcastMessage::RoomRedactionEvent{user, event, room} if user == auth.user.email => {
-                        // Send the message to the websocket
-                        if let Ok(msg) = serde_json::to_string(&WsMessage::RoomRedactionEvent {
-                            event: Box::new(event.content),
-                            room_id: room.room_id().to_owned(),
-                        }) && let Err(e) = session.text(msg).await {
+                        if let Ok(msg) = serde_json::to_string(&message)
+                            && let Err(e) = session.text(msg).await {
                             log::error!("Failed to send SyncEvent: {e}");
                         }
                     }

matrixgw_backend/src/matrix_connection/sync_thread.rs

@@ -2,7 +2,7 @@
 //!
 //! This file contains the logic performed by the threads that synchronize with Matrix account.
 
-use crate::broadcast_messages::{BroadcastMessage, BroadcastSender};
+use crate::broadcast_messages::{BroadcastMessage, BroadcastSender, BxRoomEvent};
 use crate::matrix_connection::matrix_client::MatrixClient;
 use crate::matrix_connection::matrix_manager::MatrixManagerMsg;
 use futures_util::StreamExt;
@@ -56,11 +56,11 @@ async fn sync_thread_task(
     let user_msg_handle = client.email.clone();
     handlers.push(client.add_event_handler(
         async move |event: OriginalSyncRoomMessageEvent, room: Room| {
-            if let Err(e) = tx_msg_handle.send(BroadcastMessage::RoomMessageEvent {
+            if let Err(e) = tx_msg_handle.send(BroadcastMessage::RoomMessageEvent(BxRoomEvent {
                 user: user_msg_handle.clone(),
                 event: Box::new(event),
                 room,
-            }) {
+            })) {
                 log::warn!("Failed to forward room message event! {e}");
             }
         },
@@ -70,11 +70,11 @@ async fn sync_thread_task(
     let user_reac_handle = client.email.clone();
     handlers.push(client.add_event_handler(
         async move |event: OriginalSyncReactionEvent, room: Room| {
-            if let Err(e) = tx_reac_handle.send(BroadcastMessage::ReactionEvent {
+            if let Err(e) = tx_reac_handle.send(BroadcastMessage::ReactionEvent(BxRoomEvent {
                 user: user_reac_handle.clone(),
                 event: Box::new(event),
                 room,
-            }) {
+            })) {
                 log::warn!("Failed to forward reaction event! {e}");
             }
         },
@@ -84,11 +84,13 @@ async fn sync_thread_task(
     let user_redac_handle = client.email.clone();
     handlers.push(client.add_event_handler(
         async move |event: OriginalSyncRoomRedactionEvent, room: Room| {
-            if let Err(e) = tx_redac_handle.send(BroadcastMessage::RoomRedactionEvent {
-                user: user_redac_handle.clone(),
-                event: Box::new(event),
-                room,
-            }) {
+            if let Err(e) =
+                tx_redac_handle.send(BroadcastMessage::RoomRedactionEvent(BxRoomEvent {
+                    user: user_redac_handle.clone(),
+                    event: Box::new(event),
+                    room,
+                }))
+            {
                 log::warn!("Failed to forward reaction event! {e}");
             }
         },