fix: pr comments

backend/api/Cargo.toml

@@ -26,7 +26,7 @@ serde = { version = "1.0.152", features = ["derive"] }
 hyper = { version = "0.14.24", features = ["full"] }
 common = { path = "../../common" }
 sqlx = { git="https://github.com/launchbadge/sqlx", branch="main", features = ["postgres", "runtime-tokio-native-tls", "json", "chrono"] }
-fred = { version = "6.0.0-beta.3", features = ["subscriber-client", "enable-rustls"] }
+fred = { version = "6.0.0-beta.3", features = ["subscriber-client", "enable-native-tls"] }
 routerify = "3.0.0"
 serde_json = "1.0.93"
 reqwest = { version = "0.11.14", features = ["json"] }

backend/api/src/api/v1/gql/chat.rs

@@ -6,6 +6,7 @@ use super::models::message::Message;
 use async_graphql::{futures_util::Stream, Context, Object, Subscription};
 use async_stream::stream;
 use chrono::Utc;
+use common::types::chat_message;
 use fred::prelude::PubsubInterface;
 use std::collections::HashMap;
 
@@ -23,7 +24,7 @@ impl ChatMutation {
         ctx: &Context<'_>,
         #[graphql(desc = "ID of chat room where the message will be send.")] chat_id: i64,
         #[graphql(desc = "Message content that will be published.")] content: String,
-    ) -> Result<String> {
+    ) -> Result<Message> {
         let global = ctx.get_global();
         let request_context = ctx.get_session();
         const MAX_MESSAGE_LENGTH: usize = 500;
@@ -36,7 +37,7 @@ impl ChatMutation {
 
         let user_id = session
             .as_ref()
-            .ok_or_else(|| GqlError::InvalidInput.with_message("You need to be logged in"))?
+            .ok_or_else(|| GqlError::Unauthorized.with_message("You need to be logged in"))?
             .user_id;
 
         let user = global
@@ -53,20 +54,32 @@ impl ChatMutation {
             .map_err_gql("Failed to fetch chat")?
             .ok_or(GqlError::InvalidInput.with_message("Chat not found"))?;
 
-        let message = Message {
+        let db_message = sqlx::query_as!(
+            chat_message::Model,
+            "INSERT INTO chat_messages (chat_room_id, author_id, message, created_at) VALUES ($1, $2, $3, $4) RETURNING *",
+            chat_id,
+            user_id,
+            content,
+            Utc::now()
+        )
+        .fetch_one(&*global.db)
+        .await
+        .map_err_gql("Failed to send message")?;
+
+        let message_to_send = Message {
+            message_id: db_message.id,
             chat_id,
             username: user.username,
             content: content.to_string(),
             message_type: "message".to_string(),
             metadata: HashMap::from([("color".to_string(), "teal".to_string())]),
         };
 
-        let message_to_send = serde_json::json!(message);
         match global
             .redis_pool
             .publish::<String, String, String>(
                 format!("chat:{}", chat_id),
-                message_to_send.to_string(),
+                serde_json::json!(message_to_send).to_string(),
             )
             .await
         {
@@ -76,17 +89,7 @@ impl ChatMutation {
             }
         };
 
-        let _ = sqlx::query!(
-            "INSERT INTO chat_messages (chat_room_id, author_id, message, created_at) VALUES ($1, $2, $3, $4)",
-            chat_id,
-            user_id,
-            content,
-            Utc::now()
-        )
-        .execute(&*global.db)
-        .await;
-
-        Ok(message_to_send.to_string())
+        Ok(message_to_send)
     }
 }
 
@@ -97,38 +100,29 @@ impl ChatSubscription {
         &self,
         ctx: &'ctx Context<'_>,
         #[graphql(desc = "Chat to subscribe to.")] chat_id: i64,
-    ) -> impl Stream<Item = Message> + 'ctx {
+    ) -> Result<impl Stream<Item = Message> + 'ctx> {
         let global = ctx.get_global();
 
         let welcome_message = Message {
+            message_id: 0,
             chat_id,
             username: "".to_string(),
             metadata: HashMap::new(),
             message_type: "info".to_string(),
             content: "Connected to chat".to_string(),
         };
 
-        let error_message = Message {
-            chat_id,
-            username: "".to_string(),
-            metadata: HashMap::new(),
-            message_type: "info".to_string(),
-            content: "Failed to connect to chat room".to_string(),
+        let _chat = match global.chat_room_by_id_loader.load_one(chat_id).await {
+            Ok(Some(chat)) => chat,
+            Ok(None) => {
+                return Err(GqlError::InvalidInput.with_message("Chat not found"));
+            }
+            Err(_) => {
+                return Err(GqlError::InternalServerError.with_message("Failed to fetch chat"));
+            }
         };
 
-        stream! {
-            let _chat = match global.chat_room_by_id_loader.load_one(chat_id).await {
-                Ok(Some(chat)) => chat,
-                Ok(None) => {
-                    yield error_message;
-                    return;
-                }
-                Err(_) => {
-                    yield error_message;
-                    return;
-                }
-            };
-
+        Ok(stream! {
             let mut message_stream = global.redis_sub_client.on_message();
             let _ = global
                 .redis_sub_client
@@ -144,6 +138,6 @@ impl ChatSubscription {
                     }
                 }
             }
-        }
+        })
     }
 }

backend/api/src/api/v1/gql/mod.rs

@@ -47,7 +47,6 @@ impl Noop {
 }
 
 #[derive(Default, MergedSubscription)]
-/// The root subscription type which contains root level fields.
 pub struct Subscription(chat::ChatSubscription, Noop);
 
 #[ComplexObject]

backend/api/src/api/v1/gql/models/message.rs

@@ -4,7 +4,7 @@ use std::collections::HashMap;
 
 #[derive(SimpleObject, Deserialize, Serialize)]
 pub struct Message {
-    #[graphql(skip)]
+    pub message_id: i64,
     pub chat_id: i64,
     pub username: String,
     pub content: String,