init

event_bus/src/bus.rs

@@ -4,9 +4,10 @@ use std::sync::Arc;
 
 use parking_lot::RwLock;
 use std::collections::HashMap;
+use std::iter;
 use tokio::sync::broadcast;
 use tokio::task::JoinHandle;
-use tracing::log::kv::{Key, Value};
+// use tracing::log::kv::{Key, Value};
 use tracing::{debug, trace, warn};
 use uuid::Uuid;
 
@@ -17,7 +18,7 @@ pub type AnyEvent = Arc<dyn Any + Send + Sync>;
 const DEFAULT_CAPACITY: usize = 64;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
-enum ScopeValue {
+pub enum ScopeValue {
     String(String),
     Uuid(Uuid),
 }
@@ -29,10 +30,36 @@ impl ScopeValue {
         }
     }
 }
+
 #[derive(Debug, Clone, PartialEq, Eq)]
-struct Scope {
-    key: String,
-    value: ScopeValue,
+pub struct Scope {
+    pub key: String,
+    pub value: ScopeValue,
+}
+
+impl Scope {
+    pub fn new(key: impl Into<String>, value: ScopeValue) -> Self {
+        Self {
+            key: key.into(),
+            value,
+        }
+    }
+
+    pub fn uuid(key: impl Into<String>, value: Uuid) -> Self {
+        Self::new(key, ScopeValue::Uuid(value))
+    }
+
+    pub fn string(key: impl Into<String>, value: impl Into<String>) -> Self {
+        Self::new(key, ScopeValue::String(value.into()))
+    }
+}
+impl IntoIterator for Scope {
+    type Item = Scope;
+    type IntoIter = iter::Once<Scope>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        iter::once(self)
+    }
 }
 
 /// The central event bus.
@@ -157,14 +184,7 @@ impl EventBus {
         trace!(topic, "Emitting event");
         let event: AnyEvent = Arc::new(event);
 
-        if let Some(tx) = self.channels.read().get(topic) {
-            let receiver_count = tx.receiver_count();
-            let _ = tx.send(Arc::clone(&event));
-            trace!(
-                topic,
-                receiver_count, "Event delivered to exact-topic channel"
-            );
-        }
+        self.emit_arc(topic, event);
     }
 
     // todo : undocumented...

event_bus/src/lib.rs

@@ -43,7 +43,7 @@ macro_rules! match_event {
 }
 
 mod bus;
-pub use bus::{AnyEvent, EventBus};
+pub use bus::{AnyEvent, EventBus, Scope, ScopeValue};
 
 #[cfg(test)]
 mod tests;

src/core/permission_sync.rs

@@ -41,4 +41,6 @@ impl PermissionSyncService {
             event_bus,
         }
     }
+
+    pub fn listen(&self) {}
 }

src/domain/events/channel.rs

@@ -0,0 +1,17 @@
+use crate::models::prelude::Channel;
+use uuid::Uuid;
+
+pub struct ChannelCreated {
+    server_id: Uuid,
+    channel: Channel,
+}
+
+pub struct ChannelUpdated {
+    server_id: Uuid,
+    channel: Channel,
+}
+
+pub struct ChannelDeleted {
+    server_id: Uuid,
+    channel: Channel,
+}

src/domain/events/message.rs

@@ -0,0 +1,20 @@
+use crate::models::prelude::Message;
+use uuid::Uuid;
+
+pub struct MessageCreated {
+    server_id: Option<Uuid>,
+    channel_id: Uuid,
+    message: Message,
+}
+
+pub struct MessageUpdated {
+    server_id: Option<Uuid>,
+    channel_id: Uuid,
+    message: Message,
+}
+
+pub struct MessageDeleted {
+    server_id: Option<Uuid>,
+    channel_id: Uuid,
+    message: Message,
+}

src/domain/events/mod.rs

@@ -0,0 +1,3 @@
+pub mod channel;
+pub mod message;
+pub mod server;

src/domain/events/server.rs

@@ -0,0 +1,13 @@
+use crate::models::prelude::Server;
+
+pub struct ServerCreated {
+    server: Server,
+}
+
+pub struct ServerUpdated {
+    server: Server,
+}
+
+pub struct ServerDeleted {
+    server: Server,
+}

src/domain/mod.rs

@@ -0,0 +1 @@
+pub mod events;

src/lib.rs

@@ -11,3 +11,5 @@ pub mod udp;
 
 pub mod auth;
 pub mod metrics;
+
+pub mod domain;

src/repositories/message.rs

@@ -1,6 +1,7 @@
 use super::types::MessageFilter;
-use crate::models::message;
+use crate::models::{channel, message};
 use crate::repositories::{AnyResult, RepositoryContext};
+use event_bus::Scope;
 use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
 use std::sync::Arc;
 
@@ -53,7 +54,32 @@ impl MessageRepository {
 
     pub async fn create(&self, active: message::ActiveModel) -> AnyResult<message::Model> {
         let message = active.insert(&self.context.db).await?;
-        self.context.events.emit("message_created", message.clone());
+
+        // self.context.events.emit("message_created", message.clone());
+
+        // todo : test
+        //  Ici l'évènement est déclencher sur les topic suivant :
+        //  message_created
+        //  channel:_channel_uuid_:message_created
+        //  si server : server:_server_uuid_:message_created
+        // scoped event
+        let mut scopes: Vec<Scope> = Vec::new();
+        scopes.push(Scope::uuid("channel", message.channel_id));
+        // retrieve related channel and server
+        let server_id: Option<uuid::Uuid> = channel::Entity::find_by_id(message.channel_id)
+            .select_only()
+            .column(channel::Column::ServerId)
+            .into_tuple::<Option<uuid::Uuid>>()
+            .one(&self.context.db)
+            .await?
+            .flatten();
+
+        if let Some(server_id) = server_id {
+            scopes.push(Scope::uuid("server", server_id));
+        }
+        self.context
+            .events
+            .emit_scoped("message_created", scopes, message.clone());
         Ok(message)
     }