diff --git a/src/core/mod.rs b/src/core/mod.rs index 93bc6a8..3712199 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -35,8 +35,6 @@ impl App { let repositories = Arc::new(Repositories::new(db.clone())); // 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? { @@ -69,6 +67,11 @@ impl App { let metrics = AppMetrics::new(); let services = Arc::new(Services::new(repositories.clone(), event_bus.clone())); + services.permission_sync.start_listen_event().await; + services.realtime_registry.initialize(&repositories).await?; + services.realtime_registry.start_listening(repositories.clone(), event_bus.clone()); + let gateway = Arc::new(GatewayManager::new(services.clone())); + gateway.start(event_bus.clone()); let state = AppState { db, diff --git a/src/routes/gateway/handlers.rs b/src/routes/gateway/handlers.rs index 3c14c23..40a1f0d 100644 --- a/src/routes/gateway/handlers.rs +++ b/src/routes/gateway/handlers.rs @@ -1,7 +1,6 @@ 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::{ @@ -13,7 +12,6 @@ use axum::{ use futures_util::{sink::SinkExt, stream::StreamExt}; use serde::Deserialize; use tokio::sync::mpsc; -use uuid::Uuid; #[derive(Deserialize)] pub struct WsQuery { @@ -43,17 +41,9 @@ 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(), channel_ids); + state.gateway.add_client(client.clone()); // Task pour envoyer les messages du canal mpsc vers le WebSocket let mut send_task = tokio::spawn(async move { @@ -82,43 +72,3 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) { // // 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 ad6517f..ac91991 100644 --- a/src/routes/gateway/mod.rs +++ b/src/routes/gateway/mod.rs @@ -10,21 +10,21 @@ use axum::extract::ws::Message; use event_bus::EventBus; use events::GatewayEvent; use parking_lot::RwLock; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::sync::Arc; use tokio::sync::mpsc; use tokio::task::JoinHandle; use uuid::Uuid; +use crate::services::Services; pub mod events; pub mod handlers; pub mod routes; -#[derive(Debug, Default)] +#[derive(Debug)] pub struct GatewayManager { - // Chaque connexion est inscrite dans les groupes des canaux accessibles. pub clients: RwLock>, - channel_subscribers: RwLock>>, + services: Arc, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -43,6 +43,9 @@ pub struct GatewayClient { } impl GatewayManager { + pub fn new(services: Arc) -> Self { + Self { clients: RwLock::new(HashMap::new()), services } + } /// Démarre les routeurs centraux des événements de messages. pub fn start(self: &Arc, event_bus: Arc) { let manager = Arc::clone(self); @@ -84,29 +87,15 @@ impl GatewayManager { }); } - pub(crate) fn add_client( - &self, - gateway_client: GatewayClient, - channel_ids: impl IntoIterator, - ) { + pub(crate) fn add_client(&self, gateway_client: GatewayClient) { 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( @@ -124,15 +113,10 @@ impl GatewayManager { return; }; - let keys = self - .channel_subscribers - .read() - .get(&channel_id) - .cloned() - .unwrap_or_default(); + let users = self.services.realtime_registry.users_for_channel(channel_id); let clients = self.clients.read(); - for key in keys { - if let Some(client) = clients.get(&key) { + for (key, client) in clients.iter() { + if users.contains(&key.user_id) { let _ = client.sender.send(Message::Text(json.clone().into())); } } diff --git a/src/services/mod.rs b/src/services/mod.rs index 8d2a44e..4f21b09 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -16,6 +16,7 @@ pub mod channel; pub mod message; mod permission; pub mod permission_sync; +pub mod realtime_registry; pub mod role; pub mod server; mod server_order; @@ -30,6 +31,7 @@ pub struct ServicesContext { #[derive(Debug, Clone)] pub struct Services { + pub realtime_registry: Arc, pub permission_sync: Arc, pub server_order: Arc, pub channel: Arc, @@ -49,6 +51,7 @@ impl Services { services: OnceLock::new(), }); let permission_sync = Arc::new(PermissionSyncService::new(service_context.clone())); + let realtime_registry = Arc::new(realtime_registry::RealtimeRegistry::default()); let server_order = Arc::new(ServerOrderService::new(service_context.clone())); let channel = Arc::new(ChannelService::new(service_context.clone())); let server = Arc::new(ServerService::new(service_context.clone())); @@ -59,6 +62,7 @@ impl Services { let permission = Arc::new(PermissionService::new(service_context.clone())); let services = Self { + realtime_registry, permission_sync, server_order, channel, diff --git a/src/services/permission_sync.rs b/src/services/permission_sync.rs index 218786b..11b86a9 100644 --- a/src/services/permission_sync.rs +++ b/src/services/permission_sync.rs @@ -62,8 +62,8 @@ impl PermissionSyncService { /// Enregistre les listeners sur l'EventBus pour mettre à jour le cache /// des permissions calculées lors des modifications de structure ou de droits. pub async fn start_listen_event(&self) { - let event_bus = self.service_context.event_bus.clone(); let repositories = self.service_context.repositories.clone(); + let event_bus = self.service_context.event_bus.clone(); // --------------------------------------------------------------------- // Événements Serveur & Membres Serveur diff --git a/src/services/realtime_registry.rs b/src/services/realtime_registry.rs new file mode 100644 index 0000000..0e0d49c --- /dev/null +++ b/src/services/realtime_registry.rs @@ -0,0 +1,103 @@ +use crate::models::computed_permission::PermissionScopeType; +use crate::permissions::ChannelPermission; +use crate::repositories::Repositories; +use event_bus::EventBus; +use parking_lot::RwLock; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use uuid::Uuid; + +/// In-memory index of the users that can receive events for each channel. +#[derive(Debug, Default)] +pub struct RealtimeRegistry { + channel_users: RwLock>>, + user_channels: RwLock>>, +} + +impl RealtimeRegistry { + pub async fn initialize(&self, repositories: &Repositories) -> anyhow::Result<()> { + let permissions = repositories.computed_permission.get_all().await?; + let mut channel_users = HashMap::>::new(); + let mut user_channels = HashMap::>::new(); + + for permission in permissions { + if permission.scope_type != PermissionScopeType::Channel + || !ChannelPermission::from_bits_retain(permission.permissions as u64) + .contains(ChannelPermission::READ_CHANNEL) + { + continue; + } + channel_users.entry(permission.resource_id).or_default().insert(permission.user_id); + user_channels.entry(permission.user_id).or_default().insert(permission.resource_id); + } + + *self.channel_users.write() = channel_users; + *self.user_channels.write() = user_channels; + Ok(()) + } + + pub fn users_for_channel(&self, channel_id: Uuid) -> HashSet { + self.channel_users.read().get(&channel_id).cloned().unwrap_or_default() + } + + pub fn set_user_channels(&self, user_id: Uuid, channels: impl IntoIterator) { + let channels: HashSet<_> = channels.into_iter().collect(); + let old = self.user_channels.write().insert(user_id, channels.clone()).unwrap_or_default(); + let mut by_channel = self.channel_users.write(); + for channel_id in old.difference(&channels) { + if let Some(users) = by_channel.get_mut(channel_id) { + users.remove(&user_id); + if users.is_empty() { by_channel.remove(channel_id); } + } + } + for channel_id in channels { by_channel.entry(channel_id).or_default().insert(user_id); } + } + + pub fn remove_user(&self, user_id: Uuid) { + if let Some(channels) = self.user_channels.write().remove(&user_id) { + let mut by_channel = self.channel_users.write(); + for channel_id in channels { + if let Some(users) = by_channel.get_mut(&channel_id) { + users.remove(&user_id); + if users.is_empty() { by_channel.remove(&channel_id); } + } + } + } + } + + pub fn remove_channel(&self, channel_id: Uuid) { + if let Some(users) = self.channel_users.write().remove(&channel_id) { + let mut by_user = self.user_channels.write(); + for user_id in users { + if let Some(channels) = by_user.get_mut(&user_id) { + channels.remove(&channel_id); + if channels.is_empty() { by_user.remove(&user_id); } + } + } + } + } + + pub fn start_listening(self: &Arc, repositories: Arc, event_bus: Arc) { + let registry = Arc::clone(self); + event_bus.on_async_with("channel_user_permission_updated", repositories.clone(), 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"), + } + } + }); + + let registry = Arc::clone(self); + let repositories = repositories.clone(); + event_bus.on_async_with("server_user_permission_updated", repositories, 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)); + } + } + }); + } +}