feat: chat websocket

backend/api/Cargo.toml

@@ -16,6 +16,9 @@ hyper = { version = "0.14.24", features = ["full"] }
 common = { path = "../../common" }
 sqlx = { version = "0.6.2", features = ["postgres", "runtime-tokio-rustls", "json", "chrono"] }
 routerify = "3.0.0"
+chrono = { version = "0.4.23", default-features = false }
+futures = "0.3.26"
+hyper-tungstenite = "0.9.0"
 serde_json = "1.0.93"
 
 [dev-dependencies]

backend/api/src/api/macros.rs

@@ -0,0 +1,10 @@
+#[macro_export]
+macro_rules! make_response {
+    ($status:expr, $body:expr) => {
+        Response::builder()
+            .status($status)
+            .header("Content-Type", "application/json")
+            .body(Body::from($body.to_string()))
+            .expect("failed to build response")
+    };
+}

backend/api/src/api/mod.rs

@@ -6,6 +6,7 @@ use tokio::{net::TcpSocket, select};
 
 use crate::global::GlobalState;
 
+mod macros;
 mod v1;
 
 pub fn routes(global: Arc<GlobalState>) -> Router<Body, Infallible> {
@@ -46,7 +47,7 @@ pub async fn run(global: Arc<GlobalState>) -> Result<()> {
                 tokio::spawn(Http::new().serve_connection(
                     socket,
                     request_service.build(addr),
-                ));
+                ).with_upgrades());
             },
         }
     }

backend/api/src/api/tests.rs

@@ -1,8 +1,10 @@
-use std::time::Duration;
+use std::{collections::HashMap, time::Duration};
 
 use common::{context::Context, logging};
 use hyper::{Client, StatusCode};
 
+use tokio::sync::Mutex;
+
 use crate::config::AppConfig;
 
 use super::*;
@@ -27,6 +29,7 @@ async fn test_api_v6() {
         },
         ctx,
         db,
+        chats: Mutex::new(HashMap::new()),
     });
 
     let handle = tokio::spawn(run(global));
@@ -82,6 +85,7 @@ async fn test_api_v4() {
         },
         ctx,
         db,
+        chats: Mutex::new(HashMap::new()),
     });
 
     let handle = tokio::spawn(run(global));
@@ -132,6 +136,7 @@ async fn test_api_bad_bind() {
         },
         ctx,
         db,
+        chats: Mutex::new(HashMap::new()),
     });
 
     assert!(run(global).await.is_err());

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

@@ -0,0 +1,51 @@
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use sqlx::PgPool;
+use std::collections::HashMap;
+
+#[derive(Deserialize)]
+pub enum ChatMessage {
+    Outgoing(OutgoingChatMessage),
+    Incoming(IncomingChatMessage),
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct OutgoingChatMessage {
+    // message or info
+    pub action: String,
+    pub username: String,
+    pub message: String,
+    // nickname color, badges etc
+    pub metadata: HashMap<String, String>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct IncomingChatMessage {
+    // auth or message
+    pub action: String,
+    pub message: String,
+}
+
+impl IncomingChatMessage {
+    pub fn to_outgoing(&self, username: String) -> String {
+        serde_json::to_string(&OutgoingChatMessage {
+            action: "message".to_string(),
+            username,
+            message: self.message.to_string(),
+            metadata: HashMap::new(),
+        })
+        .unwrap()
+    }
+
+    pub async fn save_to_db(&self, chat_id: i64, user_id: i64, pool: &PgPool) {
+        let _ = sqlx::query!(
+            "INSERT INTO chat_messages (chat_room_id, author_id, message, created_at) VALUES ($1, $2, $3, $4)",
+            chat_id,
+            user_id,
+            self.message,
+            Utc::now()
+        )
+        .execute(pool)
+        .await;
+    }
+}

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

@@ -0,0 +1,17 @@
+use hyper::Body;
+use routerify::Router;
+use std::convert::Infallible;
+
+mod messages;
+mod validators;
+mod ws;
+
+pub fn routes() -> Router<Body, Infallible> {
+    Router::builder()
+        .any_method("/:chatId", ws::upgrade_request)
+        .build()
+        .unwrap()
+}
+
+#[cfg(test)]
+mod tests;

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

@@ -0,0 +1 @@
+

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

@@ -0,0 +1,43 @@
+use super::messages::IncomingChatMessage;
+use common::types::{chat_room, user};
+use sqlx::PgPool;
+
+const MAX_MESSAGE_LENGTH: usize = 255;
+
+pub fn valid_chat_message(message: String) -> Option<IncomingChatMessage> {
+    let final_message = serde_json::from_str::<IncomingChatMessage>(&message);
+
+    if let Ok(message_result) = final_message {
+        if message_result.message.len() <= MAX_MESSAGE_LENGTH {
+            return Some(message_result);
+        }
+    }
+
+    None
+}
+
+pub async fn valid_user(session_id: i64, pool: &PgPool) -> Option<user::Model> {
+    let user = sqlx::query_as!(
+        user::Model,
+        "SELECT * FROM users WHERE id = (SELECT user_id FROM sessions WHERE id = $1)",
+        session_id
+    )
+    .fetch_optional(pool)
+    .await
+    .unwrap_or(None);
+
+    user
+}
+
+pub async fn valid_chat(chat_id: i64, pool: &PgPool) -> Option<chat_room::Model> {
+    let chat = sqlx::query_as!(
+        chat_room::Model,
+        "SELECT * FROM chat_rooms WHERE id = $1",
+        chat_id
+    )
+    .fetch_optional(pool)
+    .await
+    .unwrap_or(None);
+
+    chat
+}