init

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<HashMap<Uuid, HashSet<Uuid>>>,
+    user_channels: RwLock<HashMap<Uuid, HashSet<Uuid>>>,
+}
+
+impl RealtimeRegistry {
+    pub async fn initialize(&self, repositories: &Repositories) -> anyhow::Result<()> {
+        let permissions = repositories.computed_permission.get_all().await?;
+        let mut channel_users = HashMap::<Uuid, HashSet<Uuid>>::new();
+        let mut user_channels = HashMap::<Uuid, HashSet<Uuid>>::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<Uuid> {
+        self.channel_users.read().get(&channel_id).cloned().unwrap_or_default()
+    }
+
+    pub fn set_user_channels(&self, user_id: Uuid, channels: impl IntoIterator<Item = Uuid>) {
+        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<Self>, repositories: Arc<Repositories>, event_bus: Arc<EventBus>) {
+        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(&registry);
+            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(&registry);
+            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));
+                }
+            }
+        });
+    }
+}