feat: real-time reactions

Author: SrGaabriel

server/src/main.rs

@@ -69,12 +69,12 @@ async fn main() {
         .route("/ws", get(server::subscribe_chat_handshake))
         .route("/api/users/@me", get(server::rest::user::get_self))
         .route("/api/contacts/@me", get(server::rest::contacts::get_contacts))
-        .route("/api/messages/:contact_id", post(server::rest::messages::create_message))
-        .route("/api/messages/:contact_id", get(server::rest::messages::get_messages))
-        .route("/api/messages/:contact_id", put(server::rest::messages::edit_message))
-        .route("/api/messages/:contact_id", delete(server::rest::messages::delete_message))
-        .route("/api/messages/:contact_id/reactions", post(server::rest::reactions::add_reaction))
-        .route("/api/messages/:contact_id/reactions/:reaction_id", delete(server::rest::reactions::remove_reaction))
+        .route("/api/channels/:context_id/messages", post(server::rest::messages::create_message))
+        .route("/api/channels/:context_id/messages", get(server::rest::messages::get_messages))
+        .route("/api/channels/:context_id/messages/:message_id", put(server::rest::messages::edit_message))
+        .route("/api/channels/:context_id/messages/:message_id", delete(server::rest::messages::delete_message))
+        .route("/api/channels/:context_id/messages/:message_id/reactions", post(server::rest::reactions::add_reaction))
+        .route("/api/channels/:context_id/messages/:message_id/reactions/:reaction_id", delete(server::rest::reactions::remove_reaction))
         .route_layer(
             middleware::from_fn(authorize)
         )

server/src/server/gateway/context.rs

@@ -0,0 +1,17 @@
+use dashmap::DashMap;
+use futures_util::FutureExt;
+use tokio::sync::mpsc::Sender;
+use crate::server::messages::Packet;
+
+pub async fn send_packet_to_context(packet_queue: &mut DashMap<i64, Sender<Box<dyn Packet + Send>>>, context: i64, packet: Box<dyn Packet + Send>) {
+    println!("Sending packet to context: {}", context);
+    if let Some(tx) = packet_queue.get(&context) {
+        println!("Context was found, now sending packet");
+        tx.send(packet).then(|result| {
+            if let Err(e) = result {
+                eprintln!("Failed to send message: {:?}", e);
+            }
+            futures_util::future::ready(())
+        }).await;
+    }
+}

server/src/server/gateway/mod.rs

@@ -7,6 +7,7 @@ use crate::server::messages::{Packet, PacketMessage, PacketStaticId};
 
 pub mod receipts;
 pub mod typing;
+pub mod context;
 
 #[async_trait]
 pub trait GatewayHandler {

server/src/server/messages.rs

@@ -96,4 +96,38 @@ pub struct MessageDeleted {
     pub message_id: i64,
     #[prost(int64, tag = "2")]
     pub context_id: i64
+}
+
+#[derive(Clone, PartialEq, Message)]
+#[packet(id = 7)]
+pub struct ReactionAdded {
+    #[prost(int64, tag = "1")]
+    pub message_id: i64,
+    #[prost(int64, tag = "2")]
+    pub user_id: i64,
+    #[prost(string, tag = "3")]
+    pub emoji: String,
+    #[prost(int32, tag = "4")]
+    pub reaction_count: i32,
+    #[prost(int32, tag = "5")]
+    pub reaction_id: i32,
+    #[prost(int64, tag = "6")]
+    pub context_id: i64
+}
+
+#[derive(Clone, PartialEq, Message)]
+#[packet(id = 8)]
+pub struct ReactionRemoved {
+    #[prost(int64, tag = "1")]
+    pub message_id: i64,
+    #[prost(int64, tag = "2")]
+    pub user_id: i64,
+    #[prost(string, tag = "3")]
+    pub emoji: String,
+    #[prost(int32, tag = "4")]
+    pub reaction_count: i32,
+    #[prost(int32, tag = "5")]
+    pub reaction_id: i32,
+    #[prost(int64, tag = "6")]
+    pub context_id: i64
 }

server/src/server/rest/messages.rs

@@ -3,14 +3,15 @@ use axum::body::Body;
 use axum::extract::Path;
 use axum::http::{Request, StatusCode};
 use diesel::{debug_query, ExpressionMethods, OptionalExtension, QueryDsl, RunQueryDsl, Table};
+use diesel::pg::Pg;
 use diesel::sql_types::BigInt;
 use diesel::prelude::*;
 use futures_util::FutureExt;
 use http_body_util::BodyExt;
 use serde::Deserialize;
 
 use crate::entity::message::{CompleteMessage, Message};
-use crate::entity::message::messages::{content, edited, id as messageId, user_id};
+use crate::entity::message::messages::{content, context, edited, id as messageId, user_id};
 use crate::entity::message::messages::dsl::messages as messagesTable;
 use crate::entity::message::messages::dsl::messages;
 use crate::entity::user::User;
@@ -158,7 +159,6 @@ ORDER BY
     let bilateral_messages = query.load::<CompleteMessage>(connection).expect("Error loading messages");
 
     ok(bilateral_messages.iter().map(|m| {
-        println!("{}", &m.reactions);
         IterablePrivateMessage {
             id: m.id,
             user_id: m.user_id,
@@ -173,7 +173,7 @@ ORDER BY
 }
 
 pub async fn edit_message(
-    Path(message_id): Path<i64>,
+    Path((channel_id, message_id)): Path<(i64, i64)>,
     Extension(state): Extension<SharedState>,
     request: Request<Body>
 ) -> IrisResponse<PrivateMessage> {
@@ -185,6 +185,7 @@ pub async fn edit_message(
     let message: Json<MessageCreationRequest> = message.unwrap();
 
     let query = messages
+        .filter(context.eq(channel_id))
         .filter(messageId.eq(message_id))
         .filter(user_id.eq(user.id));
 
@@ -220,14 +221,15 @@ pub async fn edit_message(
 }
 
 pub async fn delete_message(
-    Path(message_id): Path<i64>,
+    Path((channel_id, message_id)): Path<(i64, i64)>,
     Extension(state): Extension<SharedState>,
     request: Request<Body>
 ) -> IrisResponse<()> {
     let user = request.extensions().get::<User>().cloned().expect("User not found");
 
     let state = &mut state.write().await;
     let query = messages
+        .filter(context.eq(channel_id))
         .filter(messageId.eq(message_id))
         .filter(user_id.eq(user.id));
     let deleted = diesel::delete(query).returning(messagesTable::all_columns()).get_result::<Message>(&mut state.database);

server/src/server/rest/reactions.rs

@@ -16,10 +16,13 @@ use crate::SharedState;
 use http_body_util::BodyExt;
 use diesel::QueryDsl;
 use diesel::ExpressionMethods;
+use futures_util::FutureExt;
 use crate::entity::reactions::reaction_users::user_id;
+use crate::server::gateway::context::send_packet_to_context;
+use crate::server::messages::{ReactionAdded, ReactionRemoved};
 
 pub async fn add_reaction(
-    Path(message_reaction_id): Path<i64>,
+    Path((channel_id, message_identifier)): Path<(i64, i64)>,
     Extension(state): Extension<SharedState>,
     request: Request<Body>
 ) -> IrisResponse<ReactionAddResponse> {
@@ -29,97 +32,121 @@ pub async fn add_reaction(
         return error(StatusCode::BAD_REQUEST, "Invalid reaction");
     }
     let request: ReactionAddRequest = request.unwrap().0;
+    let emoticon = request.reaction_type.clone();
 
-    let database = &mut state.write().await.database;
-    let transaction_result = database.transaction::<_, diesel::result::Error, _>(|connection|  {
-        let reaction_details: Option<(i32, i32)> = match request.reaction_id {
-            Some(message_reaction_id) => {
-                let count = diesel::update(reactionsTable)
-                    .filter(reaction_id.eq(message_reaction_id))
-                    .set(reaction_count.eq(reaction_count + 1))
-                    .returning(reaction_count)
-                    .get_result::<i32>(connection)?;
-                Some((message_reaction_id, count))
-            },
-            None => {
-                let reaction = diesel::update(reactionsTable)
-                    .filter(message_id.eq(message_reaction_id))
-                    .filter(emoji.eq(request.reaction_type.clone()))
-                    .set(reaction_count.eq(reaction_count + 1))
-                    .returning((reaction_id, reaction_count))
-                    .get_result::<(i32, i32)>(connection)
-                    .optional()?;
-
-                if let Some(tuple) = reaction {
-                    println!("Already found a reaction, no need for new one");
-                    Some(tuple)
-                } else {
-                    println!("Inserted");
-                    let new_reaction = ReactionInsert {
-                        message_id: message_reaction_id,
-                        emoji: request.reaction_type
-                    };
-                    let query = diesel::insert_into(reactionsTable)
-                        .values(&new_reaction)
-                        .returning(reaction_id)
+    let state = &mut state.write().await;
+    let transaction_result = {
+        state.database.transaction::<_, diesel::result::Error, _>(|connection| {
+            let reaction_details: Option<(i32, i32)> = match request.reaction_id {
+                Some(message_reaction_id) => {
+                    let count = diesel::update(reactionsTable)
+                        .filter(reaction_id.eq(message_reaction_id))
+                        .set(reaction_count.eq(reaction_count + 1))
+                        .returning(reaction_count)
                         .get_result::<i32>(connection)?;
+                    Some((message_reaction_id, count))
+                }
+                None => {
+                    let reaction = diesel::update(reactionsTable)
+                        .filter(message_id.eq(message_identifier))
+                        .filter(emoji.eq(request.reaction_type.clone()))
+                        .set(reaction_count.eq(reaction_count + 1))
+                        .returning((reaction_id, reaction_count))
+                        .get_result::<(i32, i32)>(connection)
+                        .optional()?;
+
+                    if let Some(tuple) = reaction {
+                        println!("Already found a reaction, no need for new one");
+                        Some(tuple)
+                    } else {
+                        println!("Inserted");
+                        let new_reaction = ReactionInsert {
+                            message_id: message_identifier,
+                            emoji: request.reaction_type,
+                        };
+                        let query = diesel::insert_into(reactionsTable)
+                            .values(&new_reaction)
+                            .returning(reaction_id)
+                            .get_result::<i32>(connection)?;
 
-                    Some((query, 1))
+                        Some((query, 1))
+                    }
                 }
+            };
+            if reaction_details.is_none() {
+                return Err(diesel::result::Error::NotFound);
             }
-        };
-        if reaction_details.is_none() {
-            return Err(diesel::result::Error::NotFound);
-        }
-
-        let (message_reaction_id, count) = reaction_details.unwrap();
-        let reaction_user = ReactionUserInsert {
-            reaction_id: message_reaction_id,
-            user_id: user.id,
-        };
-
-        let user_query = diesel::insert_into(reactionUsersTable)
-            .values(&reaction_user)
-            .execute(connection)?;
-        Ok((message_reaction_id, count))
-    });
+
+            let (message_reaction_id, count) = reaction_details.unwrap();
+            let reaction_user = ReactionUserInsert {
+                reaction_id: message_reaction_id,
+                user_id: user.id,
+            };
+
+            let user_query = diesel::insert_into(reactionUsersTable)
+                .values(&reaction_user)
+                .execute(connection)?;
+            Ok((message_reaction_id, count))
+        })
+    };
 
     if transaction_result.is_err() {
         return error(StatusCode::INTERNAL_SERVER_ERROR, "Failed to add reaction");
     }
     let (message_reaction_id, count) = transaction_result.unwrap();
 
+    send_packet_to_context(&mut state.packet_queue, channel_id.clone(), Box::new(ReactionAdded {
+        message_id: message_identifier,
+        user_id: user.id,
+        emoji: emoticon,
+        reaction_count: count,
+        reaction_id: message_reaction_id,
+        context_id: channel_id
+    })).await;
+
     ok(ReactionAddResponse {
         reaction_id: message_reaction_id,
         reaction_count: count
     })
 }
 
 pub async fn remove_reaction(
-    Path((message_identifier, reaction_identifier)): Path<(i64, i32)>,
+    Path((channel_id, message_identifier, reaction_identifier)): Path<(i64, i64, i32)>,
     Extension(state): Extension<SharedState>,
     request: Request<Body>
 ) -> IrisResponse<()> {
     let user = request.extensions().get::<User>().cloned().expect("User not found");
 
-    let database = &mut state.write().await.database;
-    let transaction_result = database.transaction::<_, diesel::result::Error, _>(|connection|  {
-        // reduce one from reaction count
-        let reaction = diesel::update(reactionsTable)
-            .filter(reaction_id.eq(reaction_identifier))
-            .set(reaction_count.eq(reaction_count - 1))
-            .execute(connection)?;
-
-        let user_query = diesel::delete(reactionUsersTable)
-            .filter(reactionUsersTableReactionId.eq(reaction_identifier))
-            .filter(user_id.eq(user.id))
-            .execute(connection)?;
-        Ok(())
-    });
+    let state = &mut state.write().await;
+    let transaction_result = {
+        state.database.transaction::<_, diesel::result::Error, _>(|connection| {
+            // reduce one from reaction count
+            let count = diesel::update(reactionsTable)
+                .filter(reaction_id.eq(reaction_identifier))
+                .set(reaction_count.eq(reaction_count - 1))
+                .returning(reaction_count)
+                .get_result::<i32>(connection)?;
+
+            diesel::delete(reactionUsersTable)
+                .filter(reactionUsersTableReactionId.eq(reaction_identifier))
+                .filter(user_id.eq(user.id))
+                .execute(connection)?;
+            Ok((count))
+        })
+    };
 
     if transaction_result.is_err() {
         return error(StatusCode::INTERNAL_SERVER_ERROR, "Failed to remove reaction");
     }
 
+    send_packet_to_context(&mut state.packet_queue, channel_id.clone(), Box::new(ReactionRemoved {
+        message_id: message_identifier,
+        user_id: user.id,
+        emoji: String::from(""),
+        reaction_count: transaction_result.unwrap(),
+        reaction_id: reaction_identifier,
+        context_id: channel_id
+    })).await;
+
     no_content()
 }