From 9551e906789ca28073e829f43796c197f50861e5 Mon Sep 17 00:00:00 2001 From: Nell Date: Sat, 29 Aug 2026 18:39:28 +0200 Subject: [PATCH] init --- frontend/src/stores/emoji.ts | 20 ++ frontend/src/stores/server.ts | 15 + src/core/mod.rs | 11 +- src/domain/events/channel.rs | 5 +- src/domain/events/emoji.rs | 17 + src/domain/events/mod.rs | 1 + src/routes/conversation/handlers.rs | 8 +- src/routes/emoji/handlers.rs | 30 +- src/routes/gateway/handlers.rs | 2 +- src/routes/gateway/mod.rs | 536 +++++++++++++++++----------- src/routes/gateway/routes.rs | 2 +- src/services/category.rs | 7 +- src/services/channel.rs | 26 +- src/services/permission_sync.rs | 54 ++- src/services/realtime_registry.rs | 149 ++++++-- src/services/server.rs | 16 +- src/utils/mod.rs | 12 +- 17 files changed, 642 insertions(+), 269 deletions(-) create mode 100644 src/domain/events/emoji.rs diff --git a/frontend/src/stores/emoji.ts b/frontend/src/stores/emoji.ts index bc6d228..dad6cb5 100644 --- a/frontend/src/stores/emoji.ts +++ b/frontend/src/stores/emoji.ts @@ -1,5 +1,6 @@ import {defineStore} from "pinia"; import {useApi} from "@/composables/useApi.ts"; +import {onGatewayEvent} from "@/plugins/events.ts"; export interface Emoji { id: string; @@ -65,3 +66,22 @@ export const useEmojiStore = defineStore("emoji", { }, }, }); + +onGatewayEvent("Emoji", (payload) => { + const store = useEmojiStore(); + const emoji = payload.content as Emoji; + if (!emoji?.id) return; + + // The store is scoped to the currently displayed server. Global emojis are + // always relevant; server emojis are relevant only for the active server. + if (emoji.server_id !== null && emoji.server_id !== store.activeServerId) return; + + if (payload.action === "add" || payload.action === "update") { + store.emojis = [ + ...store.emojis.filter(item => item.id !== emoji.id), + emoji, + ]; + } else if (payload.action === "remove") { + store.emojis = store.emojis.filter(item => item.id !== emoji.id); + } +}); diff --git a/frontend/src/stores/server.ts b/frontend/src/stores/server.ts index 71e5762..e08bbb0 100644 --- a/frontend/src/stores/server.ts +++ b/frontend/src/stores/server.ts @@ -2,6 +2,7 @@ import {defineStore} from "pinia"; import {useApi} from "@/composables/useApi.ts"; import {useChannelStore} from "@/stores/channel.ts"; import {useCategoryStore} from "@/stores/category.ts"; +import {onGatewayEvent} from "@/plugins/events.ts"; export interface Server { id: string @@ -153,3 +154,17 @@ export const useServerStore = defineStore("server", { } } }); + +onGatewayEvent("Channel", (payload) => { + const channel = payload.content as { server_id?: string | null }; + if (channel?.server_id) { + void useServerStore().fetchServerTree(channel.server_id); + } +}); + +onGatewayEvent("Category", (payload) => { + const category = payload.content as { server_id?: string | null }; + if (category?.server_id) { + void useServerStore().fetchServerTree(category.server_id); + } +}); diff --git a/src/core/mod.rs b/src/core/mod.rs index 4da4b43..0289708 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -5,7 +5,7 @@ use crate::database::Database; use crate::http::server::HttpServer; use crate::metrics::{AppMetrics, reporter}; use crate::repositories::Repositories; -use crate::routes::gateway::GatewayManager; +use crate::routes::gateway::{GatewayManager, RealtimeRouter}; use crate::services::Services; use crate::udp::server::UdpServer; use event_bus::EventBus; @@ -72,8 +72,13 @@ impl App { services .realtime_registry .start_listening(repositories.clone(), event_bus.clone()); - let gateway = Arc::new(GatewayManager::new(services.clone(), repositories.clone())); - gateway.start(event_bus.clone()); + let gateway = Arc::new(GatewayManager::new()); + Arc::new(RealtimeRouter::new( + gateway.clone(), + services.clone(), + repositories.clone(), + )) + .start(event_bus.clone()); let state = AppState { db, diff --git a/src/domain/events/channel.rs b/src/domain/events/channel.rs index 19deaf2..baf60b2 100644 --- a/src/domain/events/channel.rs +++ b/src/domain/events/channel.rs @@ -1,20 +1,17 @@ use crate::models::channel; -use uuid::Uuid; #[derive(Debug, Clone)] pub struct ChannelCreatedEvent { - pub server_id: Uuid, pub channel: channel::Model, } #[derive(Debug, Clone)] pub struct ChannelUpdatedEvent { - pub server_id: Uuid, + pub previous: channel::Model, pub channel: channel::Model, } #[derive(Debug, Clone)] pub struct ChannelDeletedEvent { - pub server_id: Uuid, pub channel: channel::Model, } diff --git a/src/domain/events/emoji.rs b/src/domain/events/emoji.rs new file mode 100644 index 0000000..c209df0 --- /dev/null +++ b/src/domain/events/emoji.rs @@ -0,0 +1,17 @@ +use crate::models::emoji; + +#[derive(Debug, Clone)] +pub struct EmojiCreatedEvent { + pub emoji: emoji::Model, +} + +#[derive(Debug, Clone)] +pub struct EmojiUpdatedEvent { + pub previous: emoji::Model, + pub emoji: emoji::Model, +} + +#[derive(Debug, Clone)] +pub struct EmojiDeletedEvent { + pub emoji: emoji::Model, +} diff --git a/src/domain/events/mod.rs b/src/domain/events/mod.rs index 0e20bb8..9df5974 100644 --- a/src/domain/events/mod.rs +++ b/src/domain/events/mod.rs @@ -1,3 +1,4 @@ pub mod channel; +pub mod emoji; pub mod message; pub mod server; diff --git a/src/routes/conversation/handlers.rs b/src/routes/conversation/handlers.rs index d0222c0..ab71dcd 100644 --- a/src/routes/conversation/handlers.rs +++ b/src/routes/conversation/handlers.rs @@ -3,6 +3,7 @@ use crate::domain::dto::conversation::{ ConversationParticipantResponse, ConversationResponse, CreateConversationRequest, ForkConversationRequest, }; +use crate::domain::events::channel::ChannelCreatedEvent; use crate::http::context::CurrentUser; use crate::http::error::HTTPError; use crate::models::{channel, channel_user, message, user}; @@ -107,7 +108,12 @@ async fn create_channel(state: &AppState, ids: &[Uuid]) -> Result(); - let mut client = GatewayClient::new(user, tx, state.event_bus.clone()); + let mut client = GatewayClient::new(user, tx); client.on_connect().await; state.gateway.add_client(client.clone()); diff --git a/src/routes/gateway/mod.rs b/src/routes/gateway/mod.rs index 06ac1fc..b1f0372 100644 --- a/src/routes/gateway/mod.rs +++ b/src/routes/gateway/mod.rs @@ -1,37 +1,40 @@ +use crate::domain::events::channel::{ + ChannelCreatedEvent, ChannelDeletedEvent, ChannelUpdatedEvent, +}; +use crate::domain::events::emoji::{EmojiCreatedEvent, EmojiDeletedEvent, EmojiUpdatedEvent}; use crate::domain::events::message::{ MessageCreatedEvent, MessageDeletedEvent, MessageReactionAddedEvent, MessageReactionRemovedEvent, MessageUpdatedEvent, }; -use crate::models::user::Model as User; +use crate::models::{category, server, server_user, user}; +use crate::repositories::Repositories; 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_with_data, - message_model_to_message_response_with_reactions, + message_model_to_message_response_with_data, message_model_to_message_response_with_reactions, reaction_model_to_response, }; use crate::routes::server::mapper::server_model_to_server_response; use crate::services::Services; -use crate::repositories::Repositories; use axum::extract::ws::Message; use event_bus::EventBus; use events::GatewayEvent; use parking_lot::RwLock; -use std::collections::HashMap; +use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QuerySelect}; +use serde::Serialize; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use tokio::sync::mpsc; -use tokio::task::JoinHandle; use uuid::Uuid; pub mod events; pub mod handlers; pub mod routes; +/// Couche de transport WebSocket. Elle ne connaît pas les permissions. #[derive(Debug)] pub struct GatewayManager { pub clients: RwLock>, - services: Arc, - repositories: Arc, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -42,59 +45,119 @@ pub struct ConnectionKey { #[derive(Debug, Clone)] pub struct GatewayClient { - user: User, + user: crate::models::user::Model, connection_id: Uuid, pub sender: mpsc::UnboundedSender, - pub event_bus: Arc, - _event_handles: Vec>>, } impl GatewayManager { - pub fn new(services: Arc, repositories: Arc) -> Self { + pub fn new() -> Self { Self { clients: RwLock::new(HashMap::new()), + } + } + + pub(crate) fn add_client(&self, client: GatewayClient) { + self.clients.write().insert(client.key(), client); + } + + pub(crate) fn remove_client(&self, client: &GatewayClient) { + self.clients.write().remove(&client.key()); + } + + pub(crate) fn send_to_users( + &self, + users: impl IntoIterator, + namespace: &'static str, + action: &'static str, + content: T, + ) { + let Ok(json) = serde_json::to_string(&GatewayEvent { + namespace, + action, + content, + }) else { + return; + }; + let users: HashSet<_> = users.into_iter().collect(); + for (key, client) in self.clients.read().iter() { + if users.contains(&key.user_id) { + let _ = client.sender.send(Message::Text(json.clone().into())); + } + } + } +} + +/// Résout les audiences puis délègue l'envoi au transport WebSocket. +pub struct RealtimeRouter { + gateway: Arc, + services: Arc, + repositories: Arc, +} + +impl RealtimeRouter { + pub fn new( + gateway: Arc, + services: Arc, + repositories: Arc, + ) -> Self { + Self { + gateway, services, repositories, } } - /// Démarre les routeurs centraux des événements de messages. + pub fn start(self: &Arc, event_bus: Arc) { - let manager = Arc::clone(self); + let router = Arc::clone(self); event_bus.on_async::("message_created", move |event| { - let manager = Arc::clone(&manager); + let router = Arc::clone(&router); async move { - let message_id = event.message.id; - let attachments = manager + let attachments = router .repositories .message - .attachments_for_messages(&[message_id]) + .attachments_for_messages(&[event.message.id]) .await .ok() - .and_then(|mut items| items.remove(&message_id)) + .and_then(|mut items| items.remove(&event.message.id)) .unwrap_or_default(); - manager.broadcast_message( - event.channel_id, + let users = router + .services + .realtime_registry + .users_for_channel(event.channel_id); + router.gateway.send_to_users( + users, + "Message", "add", - message_model_to_message_response_with_data(event.message, event.server_id, Vec::new(), attachments), + message_model_to_message_response_with_data( + event.message, + event.server_id, + Vec::new(), + attachments, + ), ); } }); - let manager = Arc::clone(self); + let router = Arc::clone(self); event_bus.on_async::("message_updated", move |event| { - let manager = Arc::clone(&manager); + let router = Arc::clone(&router); async move { - let message_id = event.message.id; - let reactions = manager + let reactions = router .services .message_reaction - .grouped_for_messages(&[message_id]) + .grouped_for_messages(&[event.message.id]) .await .ok() - .and_then(|mut groups| groups.remove(&message_id)) + .and_then(|mut groups| groups.remove(&event.message.id)) .unwrap_or_default(); - manager.broadcast_message( - event.channel_id, + let users = router + .services + .realtime_registry + .users_for_channel(event.channel_id); + router.gateway.send_to_users( + users, + "Message", "update", message_model_to_message_response_with_reactions( event.message, @@ -105,22 +168,33 @@ impl GatewayManager { } }); - let manager = Arc::clone(self); + let router = Arc::clone(self); event_bus.on_async::("message_deleted", move |event| { - let manager = Arc::clone(&manager); + let router = Arc::clone(&router); async move { - manager.broadcast_message(event.channel_id, "remove", event.message.id); + let users = router + .services + .realtime_registry + .users_for_channel(event.channel_id); + router + .gateway + .send_to_users(users, "Message", "remove", event.message.id); } }); - let manager = Arc::clone(self); + let router = Arc::clone(self); event_bus.on_async::( "message_reaction_added", move |event| { - let manager = Arc::clone(&manager); + let router = Arc::clone(&router); async move { - manager.broadcast_reaction( - event.channel_id, + let users = router + .services + .realtime_registry + .users_for_channel(event.channel_id); + router.gateway.send_to_users( + users, + "Reaction", "add", reaction_model_to_response(event.reaction), ); @@ -128,99 +202,248 @@ impl GatewayManager { }, ); - let manager = Arc::clone(self); + let router = Arc::clone(self); event_bus.on_async::( "message_reaction_removed", move |event| { - let manager = Arc::clone(&manager); + let router = Arc::clone(&router); async move { - manager.broadcast_reaction( - event.channel_id, + let users = router + .services + .realtime_registry + .users_for_channel(event.channel_id); + router.gateway.send_to_users( + users, + "Reaction", "remove", reaction_model_to_response(event.reaction), ); } }, ); - } - pub(crate) fn add_client(&self, gateway_client: GatewayClient) { - let key = gateway_client.key(); - self.clients.write().insert(key, gateway_client); - } - - pub(crate) fn remove_client(&self, gateway_client: &GatewayClient) { - let key = gateway_client.key(); - self.clients.write().remove(&key); - } - - 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 users = self - .services - .realtime_registry - .users_for_channel(channel_id); - let clients = self.clients.read(); - for (key, client) in clients.iter() { - if users.contains(&key.user_id) { - let _ = client.sender.send(Message::Text(json.clone().into())); + let router = Arc::clone(self); + event_bus.on_async::("channel_created", move |event| { + let router = Arc::clone(&router); + async move { + if let Err(error) = router + .services + .realtime_registry + .refresh_channel(&router.repositories, event.channel.id) + .await + { + tracing::error!(channel_id = %event.channel.id, ?error, "Unable to refresh channel audience"); + return; + } + let users = router + .services + .realtime_registry + .users_for_channel(event.channel.id); + router.gateway.send_to_users( + users, + "Channel", + "add", + channel_model_to_channel_response(event.channel), + ); } - } + }); + + let router = Arc::clone(self); + event_bus.on_async::("channel_updated", move |event| { + let router = Arc::clone(&router); + async move { + let mut users = router + .services + .realtime_registry + .users_for_channel(event.channel.id); + users.extend( + router + .services + .realtime_registry + .users_for_channel(event.previous.id), + ); + router.gateway.send_to_users( + users, + "Channel", + "update", + channel_model_to_channel_response(event.channel), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("channel_deleted", move |event| { + let router = Arc::clone(&router); + async move { + let users = router + .services + .realtime_registry + .users_for_channel(event.channel.id); + router + .gateway + .send_to_users(users, "Channel", "remove", event.channel.id); + router + .services + .realtime_registry + .remove_channel(event.channel.id); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("category_created", move |category| { + let router = Arc::clone(&router); + async move { + let users = router.server_users(category.server_id).await; + router.gateway.send_to_users( + users, + "Category", + "add", + category_model_to_category_response(category), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("category_updated", move |category| { + let router = Arc::clone(&router); + async move { + let users = router.server_users(category.server_id).await; + router.gateway.send_to_users( + users, + "Category", + "update", + category_model_to_category_response(category), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("category_deleted", move |category| { + let router = Arc::clone(&router); + async move { + let users = router.server_users(category.server_id).await; + router + .gateway + .send_to_users(users, "Category", "remove", category.id); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("server_created", move |server| { + let router = Arc::clone(&router); + async move { + let users = router.server_users(server.id).await; + router.gateway.send_to_users( + users, + "Server", + "add", + server_model_to_server_response(server), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("server_updated", move |server| { + let router = Arc::clone(&router); + async move { + let users = router.server_users(server.id).await; + router.gateway.send_to_users( + users, + "Server", + "update", + server_model_to_server_response(server), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::<(server::Model, Vec), _, _>( + "server_deleted", + move |(server, users)| { + let router = Arc::clone(&router); + async move { + router + .gateway + .send_to_users(users, "Server", "remove", server.id); + } + }, + ); + + let router = Arc::clone(self); + event_bus.on_async::("emoji_created", move |event| { + let router = Arc::clone(&router); + async move { + let users = router.emoji_users(event.emoji.server_id).await; + router.gateway.send_to_users( + users, + "Emoji", + "add", + crate::routes::emoji::mapper::response(event.emoji), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("emoji_updated", move |event| { + let router = Arc::clone(&router); + async move { + let mut users = router.emoji_users(event.emoji.server_id).await; + users.extend(router.emoji_users(event.previous.server_id).await); + router.gateway.send_to_users( + users, + "Emoji", + "update", + crate::routes::emoji::mapper::response(event.emoji), + ); + } + }); + + let router = Arc::clone(self); + event_bus.on_async::("emoji_deleted", move |event| { + let router = Arc::clone(&router); + async move { + let users = router.emoji_users(event.emoji.server_id).await; + router.gateway.send_to_users( + users, + "Emoji", + "remove", + crate::routes::emoji::mapper::response(event.emoji), + ); + } + }); } - fn broadcast_reaction( - &self, - channel_id: Uuid, - action: &'static str, - content: T, - ) { - let event = GatewayEvent { - namespace: "Reaction", - action, - content, - }; - let Ok(json) = serde_json::to_string(&event) else { - return; - }; + async fn server_users(&self, server_id: Uuid) -> Vec { + server_user::Entity::find() + .filter(server_user::Column::ServerId.eq(server_id)) + .select_only() + .column(server_user::Column::UserId) + .into_tuple::() + .all(&self.repositories.server.context.db) + .await + .unwrap_or_default() + } - let users = self - .services - .realtime_registry - .users_for_channel(channel_id); - let clients = self.clients.read(); - for (key, client) in clients.iter() { - if users.contains(&key.user_id) { - let _ = client.sender.send(Message::Text(json.clone().into())); - } + async fn emoji_users(&self, server_id: Option) -> Vec { + match server_id { + Some(server_id) => self.server_users(server_id).await, + None => user::Entity::find() + .select_only() + .column(user::Column::Id) + .into_tuple::() + .all(&self.repositories.server.context.db) + .await + .unwrap_or_default(), } } } impl GatewayClient { - pub fn new( - user: User, - sender: mpsc::UnboundedSender, - event_bus: Arc, - ) -> Self { + pub fn new(user: crate::models::user::Model, sender: mpsc::UnboundedSender) -> Self { Self { user, connection_id: Uuid::new_v4(), sender, - event_bus, - _event_handles: Vec::new(), } } @@ -231,113 +454,16 @@ impl GatewayClient { } } - fn subscribe_event( - &self, - event_name: &'static str, - namespace: &'static str, - action: &'static str, - mapper: F, - ) -> Arc> - where - T: Clone + Send + Sync + 'static, - R: serde::Serialize + Send + 'static, - F: Fn(T) -> R + Send + Sync + 'static, - { - let sender = self.sender.clone(); - Arc::new( - self.event_bus - .on_async::(event_name, move |payload| { - let sender = sender.clone(); - let content = mapper(payload); - async move { - let event = GatewayEvent { - namespace, - action, - content, - }; - if let Ok(json) = serde_json::to_string(&event) { - let _ = sender.send(Message::Text(json.into())); - } - } - }), - ) - } - - pub fn subscribe_to_events(&mut self) { - let mut handles = Vec::new(); - - // Les messages sont routés par GatewayManager selon le channel_id. - - handles.push(self.subscribe_event( - "channel_created", - "Channel", - "add", - channel_model_to_channel_response, - )); - handles.push(self.subscribe_event( - "channel_updated", - "Channel", - "update", - channel_model_to_channel_response, - )); - handles.push(self.subscribe_event("channel_deleted", "Channel", "remove", |id: Uuid| id)); - - handles.push(self.subscribe_event( - "category_created", - "Category", - "add", - category_model_to_category_response, - )); - handles.push(self.subscribe_event( - "category_updated", - "Category", - "update", - category_model_to_category_response, - )); - handles.push(self.subscribe_event("category_deleted", "Category", "remove", |id: Uuid| id)); - - handles.push(self.subscribe_event( - "server_created", - "Server", - "add", - server_model_to_server_response, - )); - handles.push(self.subscribe_event( - "server_updated", - "Server", - "update", - server_model_to_server_response, - )); - handles.push(self.subscribe_event("server_deleted", "Server", "remove", |id: Uuid| id)); - - self._event_handles = handles; - } - - pub fn unsubscribe_all(&mut self) { - for handle in self._event_handles.drain(..) { - handle.abort(); - } - } - async fn on_connect(&mut self) { tracing::info!(user_id = %self.user.id, "Client connected"); - self.subscribe_to_events(); } - async fn on_disconnect(&mut self) { tracing::info!(user_id = %self.user.id, "Client disconnected"); - self.unsubscribe_all(); } async fn on_message(&self, message: Message) { - match message { - Message::Binary(_) => {} - Message::Text(content) => { - tracing::info!(user_id = %self.user.id, "Received text message: {}", content); - } - Message::Ping(_) => {} - Message::Pong(_) => {} - Message::Close(_) => {} + if let Message::Text(content) = message { + tracing::info!(user_id = %self.user.id, "Received text message: {}", content); } } } diff --git a/src/routes/gateway/routes.rs b/src/routes/gateway/routes.rs index 8f05510..c79ff59 100644 --- a/src/routes/gateway/routes.rs +++ b/src/routes/gateway/routes.rs @@ -1,7 +1,7 @@ use super::handlers; use crate::core::AppState; -use axum::routing::get; use axum::Router; +use axum::routing::get; pub fn router() -> Router { Router::new().route("/gateway", get(handlers::ws_handler)) diff --git a/src/services/category.rs b/src/services/category.rs index 864c9a6..d2f081b 100644 --- a/src/services/category.rs +++ b/src/services/category.rs @@ -87,6 +87,11 @@ impl CategoryService { let txn = db.begin().await?; + let existing = category::Entity::find_by_id(id) + .one(&txn) + .await? + .ok_or_else(|| anyhow::anyhow!("Category not found"))?; + self.service_context .services .get() @@ -102,7 +107,7 @@ impl CategoryService { txn.commit().await?; if deleted { - event_bus.emit("category_deleted", id); + event_bus.emit("category_deleted", existing); } Ok(deleted) diff --git a/src/services/channel.rs b/src/services/channel.rs index 1b661ad..2c89cf6 100644 --- a/src/services/channel.rs +++ b/src/services/channel.rs @@ -1,4 +1,7 @@ use crate::domain::dto::channel::{CreateChannelRequest, UpdateChannelRequest}; +use crate::domain::events::channel::{ + ChannelCreatedEvent, ChannelDeletedEvent, ChannelUpdatedEvent, +}; use crate::models::server_item_order::OrderedResourceType; use crate::models::{channel, role}; use crate::permissions::PermissionSet; @@ -88,7 +91,12 @@ impl ChannelService { .await?; // Post-commit event emission - event_bus.emit("channel_created", channel.clone()); + event_bus.emit( + "channel_created", + ChannelCreatedEvent { + channel: channel.clone(), + }, + ); Ok(channel) } @@ -108,6 +116,7 @@ impl ChannelService { .await? .ok_or_else(|| anyhow::anyhow!("Channel not found"))?; + let previous = existing.clone(); let mut active: channel::ActiveModel = existing.into(); active.server_id = Set(payload.server_id); active.category_id = Set(payload.category_id); @@ -132,7 +141,13 @@ impl ChannelService { txn.commit().await?; - event_bus.emit("channel_updated", channel.clone()); + event_bus.emit( + "channel_updated", + ChannelUpdatedEvent { + previous, + channel: channel.clone(), + }, + ); Ok(channel) } @@ -143,6 +158,11 @@ impl ChannelService { let txn = db.begin().await?; + let existing = channel::Entity::find_by_id(id) + .one(&txn) + .await? + .ok_or_else(|| anyhow::anyhow!("Channel not found"))?; + self.service_context .services .get() @@ -158,7 +178,7 @@ impl ChannelService { txn.commit().await?; if deleted { - event_bus.emit("channel_deleted", id); + event_bus.emit("channel_deleted", ChannelDeletedEvent { channel: existing }); } Ok(deleted) diff --git a/src/services/permission_sync.rs b/src/services/permission_sync.rs index 11b86a9..58e8247 100644 --- a/src/services/permission_sync.rs +++ b/src/services/permission_sync.rs @@ -1,3 +1,5 @@ +use crate::domain::events::channel::{ChannelCreatedEvent, ChannelDeletedEvent}; +use crate::models::server; use crate::repositories::Repositories; use crate::services::ServicesContext; use std::sync::Arc; @@ -71,8 +73,8 @@ impl PermissionSyncService { event_bus.on_async_with( "server_created", repositories.clone(), - move |repositories, server_id: Uuid| async move { - Self::sync_server(repositories, server_id).await; + move |repositories, server: server::Model| async move { + Self::sync_server(repositories, server.id).await; }, ); @@ -139,16 +141,20 @@ impl PermissionSyncService { event_bus.on_async_with( "channel_created", repositories.clone(), - move |repositories, server_id: Uuid| async move { - Self::sync_server(repositories, server_id).await; + move |repositories, event: ChannelCreatedEvent| async move { + if let Some(server_id) = event.channel.server_id { + Self::sync_server(repositories, server_id).await; + } }, ); event_bus.on_async_with( "channel_deleted", repositories.clone(), - move |repositories, server_id: Uuid| async move { - Self::sync_server(repositories, server_id).await; + move |repositories, event: ChannelDeletedEvent| async move { + if let Some(server_id) = event.channel.server_id { + Self::sync_server(repositories, server_id).await; + } }, ); @@ -167,6 +173,42 @@ impl PermissionSyncService { Self::sync_user(repositories, user_id, server_id).await; }, ); + + event_bus.on_async_with( + "channel_user_permission_created", + repositories.clone(), + move |repositories, (channel_id, user_id, _permissions): (Uuid, Uuid, u64)| async move { + if let Some(channel) = repositories + .channel + .get_by_id(channel_id) + .await + .ok() + .flatten() + { + if let Some(server_id) = channel.server_id { + Self::sync_user(repositories, user_id, server_id).await; + } + } + }, + ); + + event_bus.on_async_with( + "channel_user_permission_deleted", + repositories, + move |repositories, (channel_id, user_id): (Uuid, Uuid)| async move { + if let Some(channel) = repositories + .channel + .get_by_id(channel_id) + .await + .ok() + .flatten() + { + if let Some(server_id) = channel.server_id { + Self::sync_user(repositories, user_id, server_id).await; + } + } + }, + ); } // ------------------------------------------------------------------------- diff --git a/src/services/realtime_registry.rs b/src/services/realtime_registry.rs index abdeaa8..020c30c 100644 --- a/src/services/realtime_registry.rs +++ b/src/services/realtime_registry.rs @@ -3,10 +3,10 @@ use crate::permissions::ChannelPermission; use crate::repositories::Repositories; use event_bus::EventBus; use parking_lot::RwLock; +use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; use uuid::Uuid; -use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; /// In-memory index of the users that can receive events for each channel. #[derive(Debug, Default)] @@ -16,6 +16,58 @@ pub struct RealtimeRegistry { } impl RealtimeRegistry { + /// Rebuilds one channel audience after a committed structural change. + pub async fn refresh_channel( + &self, + repositories: &Repositories, + channel_id: Uuid, + ) -> anyhow::Result<()> { + let channel = channel::Entity::find_by_id(channel_id) + .one(&repositories.channel.context.db) + .await? + .ok_or_else(|| anyhow::anyhow!("Channel not found"))?; + + if channel.channel_type == channel::ChannelType::DM { + let users = channel_user::Entity::find() + .filter(channel_user::Column::ChannelId.eq(channel_id)) + .all(&repositories.channel.context.db) + .await? + .into_iter() + .map(|member| member.user_id); + self.set_channel_users(channel_id, users); + } else { + let permissions = repositories.computed_permission.get_all().await?; + self.set_channel_users( + channel_id, + permissions.into_iter().filter_map(|permission| { + (permission.scope_type == PermissionScopeType::Channel + && permission.resource_id == channel_id + && ChannelPermission::from_bits_retain(permission.permissions as u64) + .contains(ChannelPermission::READ_CHANNEL)) + .then_some(permission.user_id) + }), + ); + } + Ok(()) + } + + async fn refresh_user(&self, repositories: &Repositories, user_id: Uuid) -> anyhow::Result<()> { + let channels = repositories + .computed_permission + .get_all() + .await? + .into_iter() + .filter(|permission| { + permission.user_id == user_id + && permission.scope_type == PermissionScopeType::Channel + && ChannelPermission::from_bits_retain(permission.permissions as u64) + .contains(ChannelPermission::READ_CHANNEL) + }) + .map(|permission| permission.resource_id); + self.set_user_channels(user_id, channels); + Ok(()) + } + pub async fn initialize(&self, repositories: &Repositories) -> anyhow::Result<()> { let permissions = repositories.computed_permission.get_all().await?; let mut channel_users = HashMap::>::new(); @@ -51,8 +103,14 @@ impl RealtimeRegistry { .all(&repositories.channel.context.db) .await?; for member in members { - channel_users.entry(member.channel_id).or_default().insert(member.user_id); - user_channels.entry(member.user_id).or_default().insert(member.channel_id); + channel_users + .entry(member.channel_id) + .or_default() + .insert(member.user_id); + user_channels + .entry(member.user_id) + .or_default() + .insert(member.channel_id); } } @@ -73,13 +131,17 @@ impl RealtimeRegistry { let users: HashSet<_> = users.into_iter().collect(); let old = { let mut by_channel = self.channel_users.write(); - by_channel.insert(channel_id, users.clone()).unwrap_or_default() + by_channel + .insert(channel_id, users.clone()) + .unwrap_or_default() }; let mut by_user = self.user_channels.write(); for user_id in old.difference(&users) { if let Some(channels) = by_user.get_mut(user_id) { channels.remove(&channel_id); - if channels.is_empty() { by_user.remove(user_id); } + if channels.is_empty() { + by_user.remove(user_id); + } } } for user_id in users { @@ -148,21 +210,36 @@ impl RealtimeRegistry { move |repositories, (_channel_id, user_id, _permissions): (Uuid, Uuid, u64)| { let registry = Arc::clone(®istry); async move { - match repositories.computed_permission.get_all().await { - Ok(all) => registry.set_user_channels( - user_id, - all.into_iter() - .filter(|p| { - p.user_id == user_id - && p.scope_type == PermissionScopeType::Channel - && ChannelPermission::from_bits_retain(p.permissions as u64) - .contains(ChannelPermission::READ_CHANNEL) - }) - .map(|p| p.resource_id), - ), - Err(error) => { - tracing::error!(%user_id, ?error, "Unable to refresh realtime registry") - } + if let Err(error) = registry.refresh_user(&repositories, user_id).await { + tracing::error!(%user_id, ?error, "Unable to refresh realtime registry") + } + } + }, + ); + + let registry = Arc::clone(self); + event_bus.on_async_with( + "channel_user_permission_created", + repositories.clone(), + move |repositories, (_channel_id, user_id, _permissions): (Uuid, Uuid, u64)| { + let registry = Arc::clone(®istry); + async move { + if let Err(error) = registry.refresh_user(&repositories, user_id).await { + tracing::error!(%user_id, ?error, "Unable to refresh realtime registry") + } + } + }, + ); + + let registry = Arc::clone(self); + event_bus.on_async_with( + "channel_user_permission_deleted", + repositories.clone(), + move |repositories, (_channel_id, user_id): (Uuid, Uuid)| { + let registry = Arc::clone(®istry); + async move { + if let Err(error) = registry.refresh_user(&repositories, user_id).await { + tracing::error!(%user_id, ?error, "Unable to refresh realtime registry") } } }, @@ -176,18 +253,8 @@ impl RealtimeRegistry { move |repositories, (_server_id, user_id): (Uuid, Uuid)| { let registry = Arc::clone(®istry); async move { - if let Ok(all) = repositories.computed_permission.get_all().await { - registry.set_user_channels( - user_id, - all.into_iter() - .filter(|p| { - p.user_id == user_id - && p.scope_type == PermissionScopeType::Channel - && ChannelPermission::from_bits_retain(p.permissions as u64) - .contains(ChannelPermission::READ_CHANNEL) - }) - .map(|p| p.resource_id), - ); + if let Err(error) = registry.refresh_user(&repositories, user_id).await { + tracing::error!(%user_id, ?error, "Unable to refresh realtime registry") } } }, @@ -210,8 +277,22 @@ mod tests { registry.set_channel_users(channel_id, [first, second]); assert_eq!(registry.users_for_channel(channel_id).len(), 2); - assert!(registry.user_channels.read().get(&first).unwrap().contains(&channel_id)); - assert!(registry.user_channels.read().get(&second).unwrap().contains(&channel_id)); + assert!( + registry + .user_channels + .read() + .get(&first) + .unwrap() + .contains(&channel_id) + ); + assert!( + registry + .user_channels + .read() + .get(&second) + .unwrap() + .contains(&channel_id) + ); registry.set_channel_users(channel_id, [second]); assert!(!registry.user_channels.read().contains_key(&first)); diff --git a/src/services/server.rs b/src/services/server.rs index 6fab1a2..7fedcef 100644 --- a/src/services/server.rs +++ b/src/services/server.rs @@ -1,4 +1,4 @@ -use crate::models::{role, server}; +use crate::models::{role, server, server_user}; use crate::repositories::Repositories; use crate::services::ServicesContext; use sea_orm::{ @@ -85,6 +85,18 @@ impl ServerService { let txn = db.begin().await?; + let existing = server::Entity::find_by_id(id) + .one(&txn) + .await? + .ok_or_else(|| anyhow::anyhow!("Server not found"))?; + let audience = server_user::Entity::find() + .filter(server_user::Column::ServerId.eq(id)) + .all(&txn) + .await? + .into_iter() + .map(|member| member.user_id) + .collect::>(); + let res = server::Entity::delete_by_id(id).exec(&txn).await?; let deleted = res.rows_affected > 0; @@ -92,7 +104,7 @@ impl ServerService { txn.commit().await?; if deleted { - event_bus.emit("server_deleted", id); + event_bus.emit("server_deleted", (existing, audience)); } Ok(deleted) diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 7e0db85..a79a34c 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -13,10 +13,14 @@ pub struct ScopedGuard<'a, K: Eq + Hash + Clone> { impl<'a, K: Eq + Hash + Clone> Drop for ScopedGuard<'a, K> { fn drop(&mut self) { - // Optionnel : Nettoyage des Mutex orphelins dans la HashMap scopée - let mut map = self.manager.scopes.blocking_lock(); - if let Some(weak) = map.get(&self.key) { - if weak.strong_count() == 0 { + // `Drop` peut être exécuté sur une tâche Tokio. `blocking_lock` y + // panique ; le nettoyage est opportuniste et sera retenté au prochain + // `lock_scope` si le mutex de la table est momentanément occupé. + if let Ok(mut map) = self.manager.scopes.try_lock() { + if map + .get(&self.key) + .is_some_and(|weak| weak.strong_count() <= 1) + { map.remove(&self.key); } }