crates/oxidetalis/src/extensions.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 
 use chrono::Utc;
 use oxidetalis_config::Config;
+use oxidetalis_core::types::PublicKey;
 use rayon::iter::{IntoParallelRefMutIterator, ParallelIterator};
 use salvo::Depot;
 use sea_orm::DatabaseConnection;
@@ -25,7 +26,7 @@ use uuid::Uuid;
 
 use crate::{
     nonce::NonceCache,
-    websocket::{OnlineUsers, ServerEvent, SocketUserData},
+    websocket::{OnlineUsers, ServerEvent, SocketUserData, Unsigned},
 };
 
 /// Extension trait for the Depot.
@@ -54,6 +55,12 @@ pub trait OnlineUsersExt {
 
     /// Disconnect inactive users (who not respond for the ping event)
     async fn disconnect_inactive_users(&self);
+
+    /// Returns the connection id of the user, if it is online
+    async fn is_online(&self, public_key: &PublicKey) -> Option<Uuid>;
+
+    /// Send an event to user by connection id
+    async fn send(&self, conn_id: &Uuid, event: ServerEvent<Unsigned>);
 }
 
 impl DepotExt for Depot {
@@ -113,4 +120,20 @@ impl OnlineUsersExt for OnlineUsers {
             true
         });
     }
+
+    async fn is_online(&self, public_key: &PublicKey) -> Option<Uuid> {
+        self.read()
+            .await
+            .iter()
+            .find(|(_, u)| &u.public_key == public_key)
+            .map(|(c, _)| *c)
+    }
+
+    async fn send(&self, conn_id: &Uuid, event: ServerEvent<Unsigned>) {
+        if let Some((_, user)) = self.read().await.iter().find(|(c, _)| *c == conn_id) {
+            let _ = user
+                .sender
+                .unbounded_send(Ok(event.sign(&user.shared_secret).as_ref().into()));
+        }
+    }
 }