diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 510535f..fca6a59 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -1,7 +1,14 @@ diff --git a/frontend/src/stores/message.ts b/frontend/src/stores/message.ts index 4b60238..82cc5a7 100644 --- a/frontend/src/stores/message.ts +++ b/frontend/src/stores/message.ts @@ -1,6 +1,9 @@ import {defineStore} from "pinia"; import {useApi} from "@/composables/useApi.ts"; import {onGatewayEvent} from "@/plugins/events.ts"; +import {useServerStore} from "@/stores/server.ts"; +import {useAuthStore} from "@/stores/auth.ts"; +import {useNotificationStore} from "@/stores/notification.ts"; // Change this value to adjust the maximum number of messages kept in the DOM. // Directional loads automatically use half of this window. @@ -9,6 +12,7 @@ export const MESSAGE_SHIFT_SIZE = Math.max(1, Math.floor(MESSAGE_WINDOW_SIZE / 2 export interface Message { id: string; + server_id: string | null; channel_id: string; user_id: string; content: string; @@ -85,6 +89,7 @@ export const useMessageStore = defineStore("message", { isAtBottom: true, scrollToBottomRequested: false, requestVersion: 0, + seenRealtimeMessageIds: new Set(), }), actions: { @@ -242,8 +247,29 @@ export const useMessageStore = defineStore("message", { return await response.json() as ReadStateResponse; }, - addRealtimeMessage(message: Message) { - if (message.channel_id !== this.activeChannelId) return; + addRealtimeMessage(message: Message, fromGateway = false) { + const serverStore = useServerStore(); + const authStore = useAuthStore(); + const notificationStore = useNotificationStore(); + + if (fromGateway) { + if (this.seenRealtimeMessageIds.has(message.id)) return; + this.seenRealtimeMessageIds.add(message.id); + if (this.seenRealtimeMessageIds.size > 1000) { + const oldest = this.seenRealtimeMessageIds.values().next().value; + if (oldest) this.seenRealtimeMessageIds.delete(oldest); + } + } + + const isOwnMessage = authStore.currentUser?.id === message.user_id; + const isActiveChannel = message.channel_id === this.activeChannelId; + if (!isActiveChannel) { + if (fromGateway && !isOwnMessage) { + serverStore.applyIncomingMessage(message.server_id, message.channel_id); + notificationStore.show("Nouveau message", "Un nouveau message est arrivé dans un autre canal."); + } + return; + } const existingIndex = this.messages.findIndex(current => current.id === message.id); if (existingIndex !== -1) { @@ -253,6 +279,9 @@ export const useMessageStore = defineStore("message", { if (!this.isAtBottom && this.newestId && message.id > this.newestId) { this.hasMoreAfter = true; + if (fromGateway && !isOwnMessage) { + serverStore.applyIncomingMessage(message.server_id, message.channel_id); + } return; } @@ -261,6 +290,8 @@ export const useMessageStore = defineStore("message", { this.hasMoreAfter = false; if (this.isAtBottom) { this.scrollToBottomRequested = true; + } else if (fromGateway && !isOwnMessage) { + serverStore.applyIncomingMessage(message.server_id, message.channel_id); } }, @@ -300,6 +331,7 @@ export const useMessageStore = defineStore("message", { this.loadingAfter = false; this.isAtBottom = true; this.scrollToBottomRequested = false; + this.seenRealtimeMessageIds.clear(); }, }, }); @@ -309,7 +341,7 @@ onGatewayEvent("Message", (payload) => { switch (payload.action) { case "add": - store.addRealtimeMessage(payload.content as Message); + store.addRealtimeMessage(payload.content as Message, true); break; case "update": store.updateMessage(payload.content as Message); diff --git a/frontend/src/stores/notification.ts b/frontend/src/stores/notification.ts new file mode 100644 index 0000000..a6e6a60 --- /dev/null +++ b/frontend/src/stores/notification.ts @@ -0,0 +1,19 @@ +import {defineStore} from "pinia"; + +export const useNotificationStore = defineStore("notification", { + state: () => ({ + visible: false, + title: "", + message: "", + }), + actions: { + show(title: string, message: string) { + this.title = title; + this.message = message; + this.visible = true; + }, + hide() { + this.visible = false; + }, + }, +}); diff --git a/frontend/src/stores/server.ts b/frontend/src/stores/server.ts index 7f410c0..71e5762 100644 --- a/frontend/src/stores/server.ts +++ b/frontend/src/stores/server.ts @@ -131,6 +131,21 @@ export const useServerStore = defineStore("server", { ); } }, + applyIncomingMessage(serverId: string | null, channelId: string) { + if (!serverId) return; + + for (const item of this.currentTree) { + const channels = "Category" in item ? item.Category[1] : "Channel" in item ? [item.Channel] : []; + const channel = channels.find((candidate: { id: string }) => candidate.id === channelId); + if (channel) { + channel.unread_count = (channel.unread_count ?? 0) + 1; + break; + } + } + + const server = this.servers.find(candidate => candidate.id === serverId); + if (server) server.unread_count = (server.unread_count ?? 0) + 1; + }, reset() { this.servers = []; this.loading = false; diff --git a/src/core/mod.rs b/src/core/mod.rs index 6f42a52..93bc6a8 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -36,6 +36,7 @@ impl App { // Initialize gateway manager let gateway = Arc::new(GatewayManager::default()); + gateway.start(event_bus.clone()); // Init one server if no one exist let default_server = match repositories.server.get_default().await? { diff --git a/src/domain/dto/message.rs b/src/domain/dto/message.rs index 7ed5333..9b632ed 100644 --- a/src/domain/dto/message.rs +++ b/src/domain/dto/message.rs @@ -6,6 +6,7 @@ use uuid::Uuid; #[derive(Debug, Serialize, Deserialize, ToSchema)] pub struct MessageResponse { pub id: Uuid, + pub server_id: Option, pub channel_id: Uuid, pub user_id: Uuid, pub content: String, diff --git a/src/routes/gateway/handlers.rs b/src/routes/gateway/handlers.rs index 530eb58..3c14c23 100644 --- a/src/routes/gateway/handlers.rs +++ b/src/routes/gateway/handlers.rs @@ -1,17 +1,19 @@ use crate::core::AppState; use crate::http::context::CurrentUser; use crate::models::user::Model as User; +use crate::permissions::ChannelPermission; use crate::routes::gateway::GatewayClient; use axum::{ extract::{ - ws::{Message, WebSocket, WebSocketUpgrade}, State, + ws::{Message, WebSocket, WebSocketUpgrade}, }, response::IntoResponse, }; use futures_util::{sink::SinkExt, stream::StreamExt}; use serde::Deserialize; use tokio::sync::mpsc; +use uuid::Uuid; #[derive(Deserialize)] pub struct WsQuery { @@ -41,9 +43,17 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) { let (mut sender, mut receiver) = socket.split(); let (tx, mut rx) = mpsc::unbounded_channel::(); + let channel_ids = match accessible_channel_ids(&state, &user).await { + Ok(channel_ids) => channel_ids, + Err(error) => { + tracing::error!(user_id = %user.id, ?error, "Unable to resolve gateway channel access"); + return; + } + }; + let mut client = GatewayClient::new(user, tx, state.event_bus.clone()); client.on_connect().await; - state.gateway.add_client(client.clone()); + state.gateway.add_client(client.clone(), channel_ids); // Task pour envoyer les messages du canal mpsc vers le WebSocket let mut send_task = tokio::spawn(async move { @@ -56,7 +66,6 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) { // Task pour recevoir les messages du WebSocket let client_clone = client.clone(); - let state_clone = state.clone(); let mut recv_task = tokio::spawn(async move { while let Some(Ok(message)) = receiver.next().await { client_clone.on_message(message).await; @@ -69,7 +78,47 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) { _ = (&mut recv_task) => send_task.abort(), }; - state.gateway.remove_client(client.clone()); + state.gateway.remove_client(&client); // // Déconnexion (Disconnect) client.on_disconnect().await; } + +async fn accessible_channel_ids(state: &AppState, user: &User) -> Result, anyhow::Error> { + if user.is_superuser { + return Ok(state + .repositories + .channel + .get_all() + .await? + .into_iter() + .map(|channel| channel.id) + .collect()); + } + + let mut channel_ids = Vec::new(); + for server in state.repositories.server.get_all().await? { + if state + .repositories + .server + .get_user(server.id, user.id) + .await? + .is_none() + { + continue; + } + + let tree = state + .repositories + .server_tree + .get_for_user(server.id, user.id) + .await?; + channel_ids.extend(tree.channels.into_iter().filter_map(|channel| { + channel + .permissions + .filter(|permissions| permissions.contains(ChannelPermission::READ_CHANNEL)) + .map(|_| channel.channel.id) + })); + } + + Ok(channel_ids) +} diff --git a/src/routes/gateway/mod.rs b/src/routes/gateway/mod.rs index 4ddc941..ad6517f 100644 --- a/src/routes/gateway/mod.rs +++ b/src/routes/gateway/mod.rs @@ -1,17 +1,16 @@ -use crate::models::category; -use crate::models::channel; -use crate::models::message; -use crate::models::server; +use crate::domain::events::message::{ + MessageCreatedEvent, MessageDeletedEvent, MessageUpdatedEvent, +}; use crate::models::user::Model as User; use crate::routes::category::mapper::category_model_to_category_response; use crate::routes::channel::mapper::channel_model_to_channel_response; -use crate::routes::message::mapper::message_model_to_message_response; +use crate::routes::message::mapper::message_model_to_message_response_with_server_id; use crate::routes::server::mapper::server_model_to_server_response; use axum::extract::ws::Message; use event_bus::EventBus; use events::GatewayEvent; use parking_lot::RwLock; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use tokio::sync::mpsc; use tokio::task::JoinHandle; @@ -23,8 +22,15 @@ pub mod routes; #[derive(Debug, Default)] pub struct GatewayManager { - // {UserID: {connection_id: GatewayClient}} - pub clients: RwLock>>, + // Chaque connexion est inscrite dans les groupes des canaux accessibles. + pub clients: RwLock>, + channel_subscribers: RwLock>>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ConnectionKey { + pub user_id: Uuid, + pub connection_id: Uuid, } #[derive(Debug, Clone)] @@ -37,19 +43,98 @@ pub struct GatewayClient { } impl GatewayManager { - fn add_client(&self, gateway_client: GatewayClient) { - let mut clients = self.clients.write(); - let user_id = gateway_client.user.id; - clients - .entry(user_id) - .or_insert_with(HashMap::new) - .insert(gateway_client.connection_id, gateway_client); + /// Démarre les routeurs centraux des événements de messages. + pub fn start(self: &Arc, event_bus: Arc) { + let manager = Arc::clone(self); + event_bus.on_async::("message_created", move |event| { + let manager = Arc::clone(&manager); + async move { + manager.broadcast_message( + event.channel_id, + "add", + message_model_to_message_response_with_server_id( + event.message, + event.server_id, + ), + ); + } + }); + + let manager = Arc::clone(self); + event_bus.on_async::("message_updated", move |event| { + let manager = Arc::clone(&manager); + async move { + manager.broadcast_message( + event.channel_id, + "update", + message_model_to_message_response_with_server_id( + event.message, + event.server_id, + ), + ); + } + }); + + let manager = Arc::clone(self); + event_bus.on_async::("message_deleted", move |event| { + let manager = Arc::clone(&manager); + async move { + manager.broadcast_message(event.channel_id, "remove", event.message.id); + } + }); } - fn remove_client(&self, gateway_client: GatewayClient) { - let mut clients = self.clients.write(); - if let Some(client_list) = clients.get_mut(&gateway_client.user.id) { - client_list.remove(&gateway_client.connection_id); + pub(crate) fn add_client( + &self, + gateway_client: GatewayClient, + channel_ids: impl IntoIterator, + ) { + let key = gateway_client.key(); + self.clients.write().insert(key, gateway_client); + + let mut subscribers = self.channel_subscribers.write(); + for channel_id in channel_ids { + subscribers.entry(channel_id).or_default().insert(key); + } + } + + pub(crate) fn remove_client(&self, gateway_client: &GatewayClient) { + let key = gateway_client.key(); + self.clients.write().remove(&key); + + let mut subscribers = self.channel_subscribers.write(); + for channel_subscribers in subscribers.values_mut() { + channel_subscribers.remove(&key); + } + subscribers.retain(|_, values| !values.is_empty()); + } + + fn broadcast_message( + &self, + channel_id: Uuid, + action: &'static str, + content: T, + ) { + let event = GatewayEvent { + namespace: "Message", + action, + content, + }; + let Ok(json) = serde_json::to_string(&event) else { + return; + }; + + let keys = self + .channel_subscribers + .read() + .get(&channel_id) + .cloned() + .unwrap_or_default(); + let clients = self.clients.read(); + for key in keys { + if let Some(client) = clients.get(&key) { + let _ = client.sender.send(Message::Text(json.clone().into())); + } } } } @@ -60,16 +145,22 @@ impl GatewayClient { sender: mpsc::UnboundedSender, event_bus: Arc, ) -> Self { - let connection_id = Uuid::new_v4(); Self { user, - connection_id, + connection_id: Uuid::new_v4(), sender, event_bus, _event_handles: Vec::new(), } } + pub fn key(&self) -> ConnectionKey { + ConnectionKey { + user_id: self.user.id, + connection_id: self.connection_id, + } + } + fn subscribe_event( &self, event_name: &'static str, @@ -105,22 +196,8 @@ impl GatewayClient { pub fn subscribe_to_events(&mut self) { let mut handles = Vec::new(); - // Message - handles.push(self.subscribe_event( - "message_created", - "Message", - "add", - message_model_to_message_response, - )); - handles.push(self.subscribe_event( - "message_updated", - "Message", - "update", - message_model_to_message_response, - )); - handles.push(self.subscribe_event("message_deleted", "Message", "remove", |id: Uuid| id)); + // Les messages sont routés par GatewayManager selon le channel_id. - // Channel handles.push(self.subscribe_event( "channel_created", "Channel", @@ -135,7 +212,6 @@ impl GatewayClient { )); handles.push(self.subscribe_event("channel_deleted", "Channel", "remove", |id: Uuid| id)); - // Category handles.push(self.subscribe_event( "category_created", "Category", @@ -150,7 +226,6 @@ impl GatewayClient { )); handles.push(self.subscribe_event("category_deleted", "Category", "remove", |id: Uuid| id)); - // Server handles.push(self.subscribe_event( "server_created", "Server", @@ -175,20 +250,20 @@ impl GatewayClient { } async fn on_connect(&mut self) { - tracing::info!("Client connected: {:?}", self.user); + tracing::info!(user_id = %self.user.id, "Client connected"); self.subscribe_to_events(); } async fn on_disconnect(&mut self) { - tracing::info!("Client disconnected: {:?}", self.user); + tracing::info!(user_id = %self.user.id, "Client disconnected"); self.unsubscribe_all(); } async fn on_message(&self, message: Message) { match message { - Message::Binary(content) => {} + Message::Binary(_) => {} Message::Text(content) => { - tracing::info!("Received text message: {}", content); + tracing::info!(user_id = %self.user.id, "Received text message: {}", content); } Message::Ping(_) => {} Message::Pong(_) => {} diff --git a/src/routes/message/handlers.rs b/src/routes/message/handlers.rs index 8d1aadb..1c05769 100644 --- a/src/routes/message/handlers.rs +++ b/src/routes/message/handlers.rs @@ -104,7 +104,7 @@ pub async fn create( Json(payload): Json, ) -> Result<(StatusCode, Json), HTTPError> { // Vérifier que le canal existe - state + let channel = state .repositories .channel .get_by_id(payload.channel_id) @@ -130,7 +130,10 @@ pub async fn create( .await?; Ok(( StatusCode::CREATED, - Json(mapper::message_model_to_message_response(message)), + Json(mapper::message_model_to_message_response_with_server_id( + message, + channel.server_id, + )), )) } diff --git a/src/routes/message/mapper.rs b/src/routes/message/mapper.rs index ddaa670..a2cb4fb 100644 --- a/src/routes/message/mapper.rs +++ b/src/routes/message/mapper.rs @@ -8,8 +8,16 @@ use sea_orm::Set; use uuid::Uuid; pub fn message_model_to_message_response(model: message::Model) -> MessageResponse { + message_model_to_message_response_with_server_id(model, None) +} + +pub fn message_model_to_message_response_with_server_id( + model: message::Model, + server_id: Option, +) -> MessageResponse { MessageResponse { id: model.id, + server_id, channel_id: model.channel_id, user_id: model.user_id, content: model.content, diff --git a/src/services/message.rs b/src/services/message.rs index a88c811..e2d416e 100644 --- a/src/services/message.rs +++ b/src/services/message.rs @@ -1,7 +1,10 @@ -use crate::services::ServicesContext; +use crate::domain::events::message::{ + MessageCreatedEvent, MessageDeletedEvent, MessageUpdatedEvent, +}; use crate::models::{channel, message}; +use crate::services::ServicesContext; use event_bus::Scope; -use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, TransactionTrait, Set}; +use sea_orm::{ActiveModelTrait, EntityTrait, QuerySelect, Set, TransactionTrait}; use std::sync::Arc; use uuid::Uuid; @@ -51,7 +54,15 @@ impl MessageService { scopes.push(Scope::uuid("server", server_id)); } - event_bus.emit_scoped("message_created", scopes, msg.clone()); + event_bus.emit_scoped( + "message_created", + scopes, + MessageCreatedEvent { + server_id, + channel_id: msg.channel_id, + message: msg.clone(), + }, + ); Ok(msg) } @@ -78,7 +89,27 @@ impl MessageService { txn.commit().await?; - event_bus.emit("message_updated", msg.clone()); + let server_id = channel::Entity::find_by_id(msg.channel_id) + .select_only() + .column(channel::Column::ServerId) + .into_tuple::>() + .one(db) + .await? + .flatten(); + + let mut scopes = vec![Scope::uuid("channel", msg.channel_id)]; + if let Some(server_id) = server_id { + scopes.push(Scope::uuid("server", server_id)); + } + event_bus.emit_scoped( + "message_updated", + scopes, + MessageUpdatedEvent { + server_id, + channel_id: msg.channel_id, + message: msg.clone(), + }, + ); Ok(msg) } @@ -89,16 +120,39 @@ impl MessageService { let txn = db.begin().await?; - let res = message::Entity::delete_by_id(id) - .exec(&txn) - .await?; + let existing = message::Entity::find_by_id(id).one(db).await?; + let Some(existing) = existing else { + return Ok(false); + }; + + let server_id = channel::Entity::find_by_id(existing.channel_id) + .select_only() + .column(channel::Column::ServerId) + .into_tuple::>() + .one(db) + .await? + .flatten(); + + let res = message::Entity::delete_by_id(id).exec(&txn).await?; let deleted = res.rows_affected > 0; txn.commit().await?; if deleted { - event_bus.emit("message_deleted", id); + let mut scopes = vec![Scope::uuid("channel", existing.channel_id)]; + if let Some(server_id) = server_id { + scopes.push(Scope::uuid("server", server_id)); + } + event_bus.emit_scoped( + "message_deleted", + scopes, + MessageDeletedEvent { + server_id, + channel_id: existing.channel_id, + message: existing, + }, + ); } Ok(deleted)