Merge pull request #15 from JrTimha/development

JrTimha · web-flow · commit bc0bf6c2d13a · 2026-02-08T11:33:14.000+01:00
Merge development branch to main
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,40 +1,40 @@
 [package]
 name = "ism"
-version = "0.7.2"
+version = "0.7.5"
 edition = "2024"
 
 [dependencies]
-log = "0.4.28"
-axum = { version = "0.8.6", features = ["multipart"] }
-tokio = {version = "1.48.0", features = ["full"]}
-tower = "0.5.2"
+log = "0.4.29"
+axum = { version = "0.8.8", features = ["multipart"] }
+tokio = {version = "1.49.0", features = ["full"]}
+tower = "0.5.3"
 config = "0.15.18"
 serde = "1.0.228"
-scylla = { version = "1.3.1", features = ["chrono-04"] }
+scylla = { version = "1.4.1", features = ["chrono-04"] }
 futures = "0.3.31"
-uuid = { version = "1.18.1", features = ["v4", "serde", "v7"] }
-chrono = { version = "0.4.42", features = ["serde"] }
-tower-http = { version = "0.6.6", features = ["cors", "trace"] }
-tracing = "0.1.41"
-tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
+uuid = { version = "1.20.0", features = ["v4", "serde", "v7"] }
+chrono = { version = "0.4.43", features = ["serde"] }
+tower-http = { version = "0.6.8", features = ["cors", "trace"] }
+tracing = "0.1.44"
+tracing-subscriber = { version = "0.3.22", features = ["env-filter"] }
 sqlx = {version = "0.8.6", features = ["runtime-tokio", "postgres", "chrono", "uuid", "macros"]}
 dotenv = "0.15.0"
-serde_json = "1.0.145"
-tokio-stream = { version = "0.1.17", features = ["sync"] }
-rdkafka = { version = "0.38.0", features = ["cmake-build", "tokio"] }
+serde_json = "1.0.149"
+tokio-stream = { version = "0.1.18", features = ["sync"] }
+rdkafka = { version = "0.39.0", features = ["cmake-build", "tokio"] }
 minio = { version = "0.3.0", features = ["default"] }
-image = { version = "0.25.8"}
+image = { version = "0.25.9"}
 bytes = "1.10.1"
 base64 = "0.22.1"
 validator = { version = "0.20.0", features = ["derive"] }
-redis = { version = "1.0.0-rc.3", features = ["tokio-comp", "connection-manager"] }
+redis = { version = "1.0.3", features = ["tokio-comp", "connection-manager"] }
 
 
 #keycloak:
 atomic-time = "0.1.5"
 educe = { version = "0.6.0", default-features = false, features = ["Debug"] }
-http = "1.3.1"
-jsonwebtoken = { version = "10.1.0", features = ["aws_lc_rs"] }
+http = "1.4.0"
+jsonwebtoken = { version = "10.3.0", features = ["rust_crypto"] }
 nonempty = { version = "0.12.0", features = ["std"] }
 reqwest = { version = "0.12.24", features = ["json"], default-features = false }
 serde-querystring = "0.3.0"
@@ -45,7 +45,7 @@ try-again = "0.2.2"
 typed-builder = "0.23.0"
 url = "2.5.7"
 async-trait = "0.1.89"
-thiserror = "2.0.9"
+thiserror = "2.0.18"
 
 [features]
 default = ["default-tls", "reqwest/charset", "reqwest/http2", "reqwest/macos-system-configuration"]
@@ -55,7 +55,4 @@ rustls-tls = ["reqwest/rustls-tls"]
 
 [dev-dependencies]
 assertr = "0.4.2"
-tower-http = { version = "0.6.6", features = ["trace"] }
-tracing-subscriber = "0.3.20"
-uuid = { version = "1.18.1", features = ["v7", "serde"] }
 sqlx-cli =  { version = "0.8.6", features = ["postgres", "rustls"] }
diff --git a/Dockerfile b/Dockerfile
@@ -12,6 +12,7 @@ ENV SQLX_OFFLINE=true
 RUN apt-get update && apt-get install -y --no-install-recommends \
     build-essential \
     libssl-dev \
+    libcurl4-openssl-dev \
     pkg-config \
     cmake
 
diff --git a/src/broadcast/event_broadcast.rs b/src/broadcast/event_broadcast.rs
@@ -35,6 +35,7 @@ pub struct BroadcastChannel {
     push_notification_producer: PushNotificationProducer
 }
 
+
 type UserConnectionMap = RwLock<HashMap<Uuid, Sender<Notification>>>;
 
 
@@ -144,5 +145,48 @@ impl BroadcastChannel {
         }
     }
 
-
 }
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::cache::redis_cache::NoOpCache;
+    use crate::kafka::PushNotificationProducer;
+    use crate::core::KafkaConfig;
+    use crate::broadcast::Notification;
+    use crate::broadcast::NotificationEvent::UserReadChat;
+    use serde_json;
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn send_event_to_subscribed_user_delivers_notification() {
+        // initialize broadcast channel singleton with NoOpCache and logger producer
+        let cache: Arc<dyn crate::cache::redis_cache::Cache> = Arc::new(NoOpCache);
+        let kafka_cfg = KafkaConfig { bootstrap_host: String::from(""), bootstrap_port: 0, topic: String::from(""), client_id: String::from(""), partition: vec![], consumer_group: String::from("") };
+        BroadcastChannel::init(cache, PushNotificationProducer::new(false, kafka_cfg)).await;
+
+        let bc = BroadcastChannel::get();
+
+        let user_id = uuid::Uuid::new_v4();
+        // subscribe
+        let mut rx = bc.subscribe_to_user_events(user_id).await;
+
+        let notification = Notification {
+            body: UserReadChat { user_id, room_id: uuid::Uuid::new_v4() },
+            created_at: chrono::Utc::now()
+        };
+
+        // send to all (only this user)
+        bc.send_event_to_all(vec![user_id], notification.clone()).await;
+
+        // receive
+        let received = rx.recv().await.expect("Should receive notification");
+
+        let sent_json = serde_json::to_string(&notification).expect("serialize sent");
+        let recv_json = serde_json::to_string(&received).expect("serialize recv");
+        println!("Sent: {}", sent_json);
+        println!("Received: {}", recv_json);
+        assert_eq!(sent_json, recv_json);
+    }
+}
diff --git a/src/broadcast/notification.rs b/src/broadcast/notification.rs
@@ -50,7 +50,13 @@ pub enum NotificationEvent {
     * Sending this event to all users in a room where a member has left
     */
     #[serde(rename_all = "camelCase")]
-    RoomChangeEvent {message: MessageDTO, room_preview_text: LastMessagePreviewText}
+    RoomChangeEvent {message: MessageDTO, room_preview_text: LastMessagePreviewText},
+
+    /**
+    * Sending this event to all users in a room when a user has read the latest message
+    */
+    #[serde(rename_all = "camelCase")]
+    UserReadChat {user_id: Uuid, room_id: Uuid}
 }
 
 
diff --git a/src/database/message_database.rs b/src/database/message_database.rs
@@ -38,25 +38,22 @@ impl MessageDatabase {
         if config.with_db_init {
             repository.create_keyspace_with_tables().await;
         }
-        
-        if let Err(err) = repository.change_keyspace(&config.db_keyspace).await {
-            panic!("Failed to use keyspace {:?}", err);
-        }
+
         repository
     }
 
     pub async fn fetch_data(&self, timestamp: DateTime<Utc>, room_id: Uuid) -> Result<Vec<Message>,  Box<dyn Error + Send + Sync>> {
         let session = self.session.clone();
-        let mut iter: TypedRowStream<Message> = session.query_iter("SELECT chat_room_id, message_id, sender_id, msg_body, created_at, msg_type FROM chat_messages WHERE chat_room_id = ? AND created_at < ? ORDER BY created_at DESC LIMIT 25", (room_id, timestamp))
+        let mut iter: TypedRowStream<Message> = session.query_iter("SELECT chat_room_id, message_id, sender_id, msg_body, created_at, msg_type FROM messaging.chat_messages WHERE chat_room_id = ? AND created_at < ? ORDER BY created_at DESC LIMIT 25", (room_id, timestamp))
             .await?.rows_stream::<Message>()?;
         let mut messages: Vec<Message> = Vec::new();
         while let Some(next) = iter.try_next().await? { messages.push(next) }
         Ok(messages)
     }
 
-    pub async fn fetch_specific_message(&self, message_id: &Uuid, room_id: &Uuid, created: &DateTime<Utc>) -> Result<Message, Box<dyn std::error::Error>> {
+    pub async fn fetch_specific_message(&self, message_id: &Uuid, room_id: &Uuid, created: &DateTime<Utc>) -> Result<Message, Box<dyn Error>> {
         let session = self.session.clone();
-        let mut iter = session.query_iter("SELECT chat_room_id, message_id, sender_id, msg_body, created_at, msg_type FROM chat_messages WHERE chat_room_id = ? AND created_at = ? AND message_id = ?", (room_id, created, message_id))
+        let mut iter = session.query_iter("SELECT chat_room_id, message_id, sender_id, msg_body, created_at, msg_type FROM messaging.chat_messages WHERE chat_room_id = ? AND created_at = ? AND message_id = ?", (room_id, created, message_id))
             .await?.rows_stream::<Message>()?;
         match iter.try_next().await? {
             Some(message) => Ok(message),
@@ -67,7 +64,7 @@ impl MessageDatabase {
     pub async fn insert_data(&self, message: Message) -> Result<QueryResult, ExecutionError> {
        let session = self.session.clone();
        session.query_unpaged(
-            "INSERT INTO chat_messages (chat_room_id, message_id, sender_id, msg_body, msg_type, created_at) VALUES (?, ?, ?, ?, ?, ?)",
+            "INSERT INTO messaging.chat_messages (chat_room_id, message_id, sender_id, msg_body, msg_type, created_at) VALUES (?, ?, ?, ?, ?, ?)",
             (message.chat_room_id, message.message_id, message.sender_id, message.msg_body, message.msg_type, message.created_at)
        ).await
     }
@@ -96,7 +93,7 @@ impl MessageDatabase {
     
     pub async fn clear_chat_room_messages(&self, room_id: &Uuid) -> Result<(), ExecutionError> {
         let session = self.session.clone();
-        session.query_unpaged("DELETE FROM chat_messages WHERE chat_room_id = ?", (room_id,)).await?;
+        session.query_unpaged("DELETE FROM messaging.chat_messages WHERE chat_room_id = ?", (room_id,)).await?;
         Ok(())
     }
 
diff --git a/src/keycloak/action.rs b/src/keycloak/action.rs
@@ -144,6 +144,7 @@ impl<I: ActionInput, O: ActionOutput> Action<I, O> {
 }
 
 #[cfg(test)]
+#[allow(unused)]
 mod test {
     use assertr::prelude::*;
 
diff --git a/src/keycloak/layer.rs b/src/keycloak/layer.rs
@@ -21,6 +21,7 @@ extern crate alloc;
 /// Authentication happens by looking for the `Authorization` header on requests and parsing the contained JWT bearer token.
 /// See the crate level documentation for how this layer can be created and used.
 #[derive(Clone, TypedBuilder)]
+#[allow(unused_variables)]
 pub struct KeycloakAuthLayer<R, Extra = ProfileAndEmail>
 where
     R: Role,
diff --git a/src/main.rs b/src/main.rs
@@ -12,7 +12,7 @@ use ism::welcome::welcome;
 //learn to code rust axum here:
 //https://gitlab.com/famedly/conduit/-/tree/next?ref_type=heads
 //https://github.com/AarambhDevHub/rust-backend-axum
-//https://github.com/rust-lang/crates.io/
+//https://github.com/rust-lang/crates.io/ <---- THE BEST!
 #[tokio::main(flavor = "multi_thread")]
 async fn main() {
     
diff --git a/src/rooms/handler.rs b/src/rooms/handler.rs
@@ -200,4 +200,14 @@ pub async fn handle_save_room_image(
     } else {
         Err(AppError::ValidationError("Required field 'image' not found in the upload.".to_string()))
     }
+}
+
+pub async fn handle_get_read_states(
+    Extension(token): Extension<KeycloakToken<String>>,
+    State(state): State<Arc<AppState>>,
+    Path(room_id): Path<Uuid>
+) -> Result<Json<Vec<RoomMember>>, AppError> {
+    check_user_in_room(&state, &token.subject, &room_id).await?;
+    let read_states = RoomService::get_read_states(state, room_id).await?;
+    Ok(Json(read_states))
 }
diff --git a/src/rooms/room_service.rs b/src/rooms/room_service.rs
@@ -4,7 +4,7 @@ use chrono::Utc;
 use log::{error};
 use uuid::Uuid;
 use crate::broadcast::{BroadcastChannel, Notification};
-use crate::broadcast::NotificationEvent::{LeaveRoom, RoomChangeEvent};
+use crate::broadcast::NotificationEvent::{LeaveRoom, RoomChangeEvent, UserReadChat};
 use crate::core::AppState;
 use crate::errors::{AppError};
 use crate::messaging::model::{Message, MessageBody, RoomChangeBody};
@@ -44,9 +44,45 @@ impl RoomService {
     pub async fn mark_room_as_read(state: Arc<AppState>, client_id: Uuid, room_id: Uuid) -> Result<(), AppError> {
         let pl = state.room_repository.get_connection();
         state.room_repository.update_user_read_status(pl, &room_id, &client_id).await?;
+
+        let room = state.room_repository.select_room(&room_id).await?;
+        if let Some(latest_msg_time) = room.latest_message {
+            let user = state.room_repository.select_joined_user_by_id(&room_id, &client_id).await?;
+            if let Some(read_time) = user.last_message_read_at {
+                if read_time >= latest_msg_time {
+                    let users_in_room = state.room_repository.select_room_participants_ids(&room_id).await?;
+                    BroadcastChannel::get().send_event_to_all(
+                        users_in_room,
+                        Notification {
+                            body: UserReadChat { user_id: client_id, room_id },
+                            created_at: Utc::now()
+                        }
+                    ).await;
+                }
+            }
+        } else {
+            let users_in_room = state.room_repository.select_room_participants_ids(&room_id).await?;
+            BroadcastChannel::get().send_event_to_all(
+                users_in_room,
+                Notification {
+                    body: UserReadChat { user_id: client_id, room_id },
+                    created_at: Utc::now()
+                }
+            ).await;
+        }
+
         Ok(())
     }
 
+    pub async fn get_read_states(state: Arc<AppState>, room_id: Uuid) -> Result<Vec<RoomMember>, AppError> {
+        let users = state.room_repository.select_joined_user_in_room(&room_id).await?;
+        let room = state.room_repository.select_room(&room_id).await?;
+        let read_users: Vec<RoomMember> = users.into_iter().filter(|user| {
+            user_has_read(user, room.latest_message)
+        }).collect();
+        Ok(read_users)
+    }
+
     pub async fn create_room(state: Arc<AppState>, client_id: Uuid, new_room: NewRoom) -> Result<ChatRoomDto, AppError> {
         let room_entity = state.room_repository.insert_room(new_room.clone()).await?;
         let users = new_room.invited_users;
@@ -204,6 +240,16 @@ impl RoomService {
 
 }
 
+// Helper used by `get_read_states` — extracted for easier unit testing of the read logic.
+fn user_has_read(user: &RoomMember, room_latest: Option<chrono::DateTime<chrono::Utc>>) -> bool {
+    match (room_latest, user.last_message_read_at) {
+        (Some(latest_msg_time), Some(read_time)) => read_time >= latest_msg_time,
+        (Some(_), None) => false,
+        (None, _) => true,
+    }
+}
+
+
 async fn handle_leave_private_room(state: Arc<AppState>, room: ChatRoomEntity, users: Vec<RoomMember>) -> Result<(), AppError> {
     let mut tx = state.room_repository.start_transaction().await?;
     state.room_repository.delete_room(&mut *tx, &room.id).await?;
@@ -294,4 +340,54 @@ async fn save_room_change_message_and_broadcast(message: Message, state: &Arc<Ap
 
     BroadcastChannel::get().send_event_to_all(to_users, notification).await;
     Ok(())
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::{Utc, Duration};
+    use uuid::Uuid;
+    use crate::model::room_member::{RoomMember, MembershipStatus};
+
+    fn make_member(read_at: Option<chrono::DateTime<Utc>>) -> RoomMember {
+        RoomMember {
+            id: Uuid::new_v4(),
+            display_name: "test".to_string(),
+            profile_picture: None,
+            joined_at: Utc::now(),
+            last_message_read_at: read_at,
+            membership_status: MembershipStatus::Joined
+        }
+    }
+
+    #[test]
+    fn user_has_read_when_no_latest_message() {
+        let user = make_member(None);
+        let result = user_has_read(&user, None);
+        assert!(result, "When room has no latest message, every user should be considered read");
+    }
+
+    #[test]
+    fn user_has_read_when_read_time_ge_latest() {
+        let latest = Utc::now();
+        let read_time = latest + Duration::seconds(1);
+        let user = make_member(Some(read_time));
+        assert!(user_has_read(&user, Some(latest)));
+    }
+
+    #[test]
+    fn user_has_not_read_when_read_time_before_latest() {
+        let latest = Utc::now();
+        let read_time = latest - Duration::seconds(10);
+        let user = make_member(Some(read_time));
+        assert!(!user_has_read(&user, Some(latest)));
+    }
+
+    #[test]
+    fn user_has_not_read_when_no_read_time_and_latest_present() {
+        let latest = Utc::now();
+        let user = make_member(None);
+        assert!(!user_has_read(&user, Some(latest)));
+    }
 }
diff --git a/src/rooms/routes.rs b/src/rooms/routes.rs
@@ -2,7 +2,7 @@ use std::sync::Arc;
 use axum::Router;
 use axum::routing::{get, post};
 use crate::core::AppState;
-use crate::rooms::handler::{handle_create_room, handle_get_joined_rooms, handle_get_room_list_item_by_id, handle_get_room_with_details, handle_get_users_in_room, handle_invite_to_room, handle_leave_room, handle_save_room_image, handle_scroll_chat_timeline, handle_search_existing_single_room, mark_room_as_read};
+use crate::rooms::handler::{handle_create_room, handle_get_joined_rooms, handle_get_room_list_item_by_id, handle_get_room_with_details, handle_get_users_in_room, handle_invite_to_room, handle_leave_room, handle_save_room_image, handle_scroll_chat_timeline, handle_search_existing_single_room, mark_room_as_read, handle_get_read_states};
 
 
 pub fn create_room_routes() -> Router<Arc<AppState>> {
@@ -11,11 +11,12 @@ pub fn create_room_routes() -> Router<Arc<AppState>> {
         .route("/api/rooms/{room_id}/users", get(handle_get_users_in_room))
         .route("/api/rooms/{room_id}/detailed", get(handle_get_room_with_details))
         .route("/api/rooms/{room_id}/timeline", get(handle_scroll_chat_timeline))
-        .route("/api/rooms/{room_id}/mark-read", post(mark_room_as_read))
         .route("/api/rooms/{room_id}", get(handle_get_room_list_item_by_id))
         .route("/api/rooms/{room_id}/leave", post(handle_leave_room))
         .route("/api/rooms/search", get(handle_search_existing_single_room))
         .route("/api/rooms/{room_id}/invite/{user_id}", post(handle_invite_to_room))
         .route("/api/rooms/{room_id}/upload-img", post(handle_save_room_image))
         .route("/api/rooms", get(handle_get_joined_rooms))
+        .route("/api/rooms/{room_id}/mark-read", post(mark_room_as_read))
+        .route("/api/rooms/{room_id}/read-states", get(handle_get_read_states))
 }

Original file line number	Diff line number	Diff line change
`@@ -144,6 +144,7 @@ impl<I: ActionInput, O: ActionOutput> Action<I, O> {`
`144`	`144`	`}`
`145`	`145`
`146`	`146`	`#[cfg(test)]`
	`147`	`+#[allow(unused)]`
`147`	`148`	`mod test {`
`148`	`149`	`use assertr::prelude::*;`
`149`	`150`