diff --git a/event_bus/src/bus.rs b/event_bus/src/bus.rs index 883ccc1..551f2b6 100644 --- a/event_bus/src/bus.rs +++ b/event_bus/src/bus.rs @@ -331,6 +331,48 @@ impl EventBus { }) } + // todo : Undocumented + pub fn on_async_with(&self, topic: &str, context: C, handler: F) -> JoinHandle<()> + where + T: Any + Send + Sync + Clone + 'static, + C: Clone + Send + Sync + 'static, + F: Fn(C, T) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + let mut rx = self.get_or_create_sender(topic).subscribe(); + let topic_owned = topic.to_string(); + + debug!(topic, "Async subscriber registered"); + + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(evt) => { + if let Some(typed) = evt.downcast_ref::() { + trace!(topic = topic_owned, "Async handler invoked"); + + handler(context.clone(), typed.clone()).await; + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + warn!( + topic = topic_owned, + skipped = n, + "Subscriber lagged, messages dropped" + ); + } + Err(broadcast::error::RecvError::Closed) => { + debug!( + topic = topic_owned, + "Channel closed, async subscriber exiting" + ); + break; + } + } + } + }) + } + // ───────────────────────────────────────────────────────────────────────── // Subscription — low-level access (advanced use cases) // ───────────────────────────────────────────────────────────────────────── diff --git a/src/core/permission_sync.rs b/src/core/permission_sync.rs index 641575a..674f5f9 100644 --- a/src/core/permission_sync.rs +++ b/src/core/permission_sync.rs @@ -1,7 +1,8 @@ +use crate::domain::events::server::ServerCreatedEvent; use crate::repositories::Repositories; use event_bus::EventBus; use std::sync::Arc; - +use uuid::Uuid; // list of all events : // server_user_created // server_user_deleted @@ -42,5 +43,21 @@ impl PermissionSyncService { } } - pub fn listen(&self) {} + pub async fn start_listen_event(&self) { + // todo : compléter la liste de permission avec le déclenchement des évènements + self.event_bus.on_async_with( + "server_created", + self.repositories.clone(), + move |repositories, payload: ServerCreatedEvent| { + Self::sync_server_permissions(repositories, payload.server.id) + }, + ); + } + + async fn sync_server_permissions(repositories: Arc, server_id: Uuid) { + let _ = repositories + .computed_permission + .full_sync_server(server_id) + .await; + } } diff --git a/src/domain/events/channel.rs b/src/domain/events/channel.rs index 00522a0..19deaf2 100644 --- a/src/domain/events/channel.rs +++ b/src/domain/events/channel.rs @@ -1,17 +1,20 @@ -use crate::models::prelude::Channel; +use crate::models::channel; use uuid::Uuid; -pub struct ChannelCreated { - server_id: Uuid, - channel: Channel, +#[derive(Debug, Clone)] +pub struct ChannelCreatedEvent { + pub server_id: Uuid, + pub channel: channel::Model, } -pub struct ChannelUpdated { - server_id: Uuid, - channel: Channel, +#[derive(Debug, Clone)] +pub struct ChannelUpdatedEvent { + pub server_id: Uuid, + pub channel: channel::Model, } -pub struct ChannelDeleted { - server_id: Uuid, - channel: Channel, +#[derive(Debug, Clone)] +pub struct ChannelDeletedEvent { + pub server_id: Uuid, + pub channel: channel::Model, } diff --git a/src/domain/events/message.rs b/src/domain/events/message.rs index 19829b9..bd52602 100644 --- a/src/domain/events/message.rs +++ b/src/domain/events/message.rs @@ -1,20 +1,23 @@ -use crate::models::prelude::Message; +use crate::models::message; use uuid::Uuid; -pub struct MessageCreated { - server_id: Option, - channel_id: Uuid, - message: Message, +#[derive(Debug, Clone)] +pub struct MessageCreatedEvent { + pub server_id: Option, + pub channel_id: Uuid, + pub message: message::Model, } -pub struct MessageUpdated { - server_id: Option, - channel_id: Uuid, - message: Message, +#[derive(Debug, Clone)] +pub struct MessageUpdatedEvent { + pub server_id: Option, + pub channel_id: Uuid, + pub message: message::Model, } -pub struct MessageDeleted { - server_id: Option, - channel_id: Uuid, - message: Message, +#[derive(Debug, Clone)] +pub struct MessageDeletedEvent { + pub server_id: Option, + pub channel_id: Uuid, + pub message: message::Model, } diff --git a/src/domain/events/server.rs b/src/domain/events/server.rs index 8325137..65c4931 100644 --- a/src/domain/events/server.rs +++ b/src/domain/events/server.rs @@ -1,13 +1,16 @@ -use crate::models::prelude::Server; +use crate::models::server; -pub struct ServerCreated { - server: Server, +#[derive(Debug, Clone)] +pub struct ServerCreatedEvent { + pub server: server::Model, } -pub struct ServerUpdated { - server: Server, +#[derive(Debug, Clone)] +pub struct ServerUpdatedEvent { + pub server: server::Model, } -pub struct ServerDeleted { - server: Server, +#[derive(Debug, Clone)] +pub struct ServerDeletedEvent { + pub server: server::Model, } diff --git a/src/repositories/computed_permission.rs b/src/repositories/computed_permission.rs index b5e9c9a..558dc21 100644 --- a/src/repositories/computed_permission.rs +++ b/src/repositories/computed_permission.rs @@ -7,11 +7,17 @@ use crate::repositories::{AnyResult, RepositoryContext}; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set, TransactionTrait}; use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; +use tokio::sync::Mutex; use uuid::Uuid; use crate::models::computed_permission::PermissionScopeType; +static SYNC_LOCK: OnceLock> = OnceLock::new(); +fn _sync_lock() -> &'static Mutex<()> { + SYNC_LOCK.get_or_init(Mutex::default) +} + #[derive(Clone, Debug)] pub struct ComputedPermissionRepository { pub context: Arc, @@ -50,6 +56,7 @@ impl ComputedPermissionRepository { /// - permissions de canal accordées aux rôles de l'utilisateur ; /// - permissions directes de l'utilisateur dans les canaux. pub async fn full_sync_user(&self, user_id: Uuid, server_id: Uuid) -> AnyResult<()> { + let _guard = _sync_lock().lock().await; // --------------------------------------------------------------------- // Rôles de l'utilisateur // --------------------------------------------------------------------- diff --git a/src/repositories/server.rs b/src/repositories/server.rs index 64c2e82..e92eb4d 100644 --- a/src/repositories/server.rs +++ b/src/repositories/server.rs @@ -2,7 +2,7 @@ use super::types::{ServerExplorerItem, ServerTree}; use super::{AnyResult, RepositoryContext}; use crate::models::{category, channel, role, server, server_user}; use sea_orm::prelude::*; -use sea_orm::{ActiveModelTrait, Set}; +use sea_orm::{ActiveModelTrait, QuerySelect, Set}; use std::sync::Arc; use uuid::Uuid; @@ -72,6 +72,16 @@ impl ServerRepository { .insert(&self.context.db) .await?; + let role_id: Uuid = role::Entity::find() + .filter(role::Column::ServerId.eq(server_id)) + .filter(role::Column::IsDefault.eq(true)) + .select_only() + .column(role::Column::Id) + .into_tuple::() + .one(&self.context.db) + .await? + .ok_or_else(|| anyhow::anyhow!("Rôle par défaut introuvable"))?; + self.context .events .emit("server_user_created", (server_id, user_id));