chore: bump version to 0.7.7

Cargo.lock

@@ -138,11 +138,23 @@ version = "1.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2114d64672151c0c5eaa5e131ec84a74f06e1e559830dabba01ca30605d66319"
 dependencies = [
- "concurrent-queue",
+ "concurrent-queue 1.2.2",
- "event-listener",
+ "event-listener 2.5.2",
  "futures-core",
 ]
 
+[[package]]
+name = "async-channel"
+version = "2.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "89b47800b0be77592da0afd425cc03468052844aff33b84e33cc696f64e77b6a"
+dependencies = [
+ "concurrent-queue 2.5.0",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite 0.2.13",
+]
+
 [[package]]
 name = "async-executor"
 version = "1.4.1"
@@ -150,7 +162,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "871f9bb5e0a22eeb7e8cf16641feb87c9dc67032ccf8ff49e772eb9941d3a965"
 dependencies = [
  "async-task",
- "concurrent-queue",
+ "concurrent-queue 1.2.2",
  "fastrand",
  "futures-lite",
  "once_cell",
@@ -163,7 +175,7 @@ version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fd8b508d585e01084059b60f06ade4cb7415cd2e4084b71dd1cb44e7d3fb9880"
 dependencies = [
- "async-channel",
+ "async-channel 1.6.1",
  "async-executor",
  "async-io",
  "async-lock",
@@ -180,7 +192,7 @@ version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e5e18f61464ae81cde0a23e713ae8fd299580c54d697a35820cfd0625b8b0e07"
 dependencies = [
- "concurrent-queue",
+ "concurrent-queue 1.2.2",
  "futures-lite",
  "libc",
  "log",
@@ -199,7 +211,7 @@ version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "287272293e9d8c41773cec55e365490fe034813a2f172f502d6ddcf75b2f582b"
 dependencies = [
- "event-listener",
+ "event-listener 2.5.2",
 ]
 
 [[package]]
@@ -211,7 +223,7 @@ dependencies = [
  "async-io",
  "blocking",
  "cfg-if",
- "event-listener",
+ "event-listener 2.5.2",
  "futures-lite",
  "libc",
  "once_cell",
@@ -237,7 +249,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "62565bb4402e926b29953c785397c6dc0391b7b446e45008b0049eb43cec6f5d"
 dependencies = [
  "async-attributes",
- "async-channel",
+ "async-channel 1.6.1",
  "async-global-executor",
  "async-io",
  "async-lock",
@@ -531,7 +543,7 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c6ccb65d468978a086b69884437ded69a90faab3bbe6e67f242173ea728acccc"
 dependencies = [
- "async-channel",
+ "async-channel 1.6.1",
  "async-task",
  "atomic-waker",
  "fastrand",
@@ -729,6 +741,15 @@ dependencies = [
  "cache-padded",
 ]
 
+[[package]]
+name = "concurrent-queue"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "config"
 version = "0.13.3"
@@ -833,13 +854,9 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-utils"
-version = "0.8.8"
+version = "0.8.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bf124c720b7686e3c2663cf54062ab0f68a88af2fb6a030e87e30bf721fcb38"
+checksum = "22ec99545bb0ed0ea7bb9b8e1e9122ea386ff8a48c0922e43f36d45ab09e0e80"
-dependencies = [
- "cfg-if",
- "lazy_static",
-]
 
 [[package]]
 name = "crypto-bigint"
@@ -1295,6 +1312,27 @@ version = "2.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77f3309417938f28bf8228fcff79a4a37103981e3e186d2ccd19c74b38f4eb71"
 
+[[package]]
+name = "event-listener"
+version = "5.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6032be9bd27023a771701cc49f9f053c751055f71efb2e0ae5c15809093675ba"
+dependencies = [
+ "concurrent-queue 2.5.0",
+ "parking",
+ "pin-project-lite 0.2.13",
+]
+
+[[package]]
+name = "event-listener-strategy"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f214dc438f977e6d4e3500aaa277f5ad94ca83fbbd9b1a15713ce2344ccc5a1"
+dependencies = [
+ "event-listener 5.3.1",
+ "pin-project-lite 0.2.13",
+]
+
 [[package]]
 name = "fake-simd"
 version = "0.1.2"
@@ -1976,13 +2014,13 @@ version = "1.7.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "334e04b4d781f436dc315cb1e7515bd96826426345d498149e4bde36b67f8ee9"
 dependencies = [
- "async-channel",
+ "async-channel 1.6.1",
  "castaway",
  "crossbeam-utils",
  "curl",
  "curl-sys",
  "encoding_rs",
- "event-listener",
+ "event-listener 2.5.2",
  "futures-lite",
  "http",
  "log",
@@ -2266,6 +2304,12 @@ dependencies = [
  "linked-hash-map",
 ]
 
+[[package]]
+name = "lru_time_cache"
+version = "0.11.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9106e1d747ffd48e6be5bb2d97fa706ed25b144fbee4d5c02eae110cd8d6badd"
+
 [[package]]
 name = "maplit"
 version = "1.0.2"
@@ -3408,8 +3452,9 @@ dependencies = [
 
 [[package]]
 name = "revolt-bonfire"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
+ "async-channel 2.3.1",
  "async-std",
  "async-tungstenite",
  "authifier",
@@ -3418,6 +3463,7 @@ dependencies = [
  "futures",
  "log",
  "lru 0.7.6",
+ "lru_time_cache",
  "once_cell",
  "querystring",
  "redis-kiss",
@@ -3436,7 +3482,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-config"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "async-std",
  "cached",
@@ -3452,7 +3498,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-database"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "async-lock",
  "async-recursion",
@@ -3498,9 +3544,9 @@ dependencies = [
 
 [[package]]
 name = "revolt-delta"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
- "async-channel",
+ "async-channel 1.6.1",
  "async-std",
  "authifier",
  "bitfield",
@@ -3544,7 +3590,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-models"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "indexmap",
  "iso8601-timestamp 0.2.11",
@@ -3561,7 +3607,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-permissions"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "async-std",
  "async-trait",
@@ -3576,7 +3622,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-presence"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "async-std",
  "log",
@@ -3587,7 +3633,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-result"
-version = "0.7.5"
+version = "0.7.7"
 dependencies = [
  "revolt_okapi",
  "revolt_rocket_okapi",
@@ -4362,7 +4408,7 @@ version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6d7400c0eff44aa2fcb5e31a5f24ba9716ed90138769e4977a2ba6014ae63eb5"
 dependencies = [
- "async-channel",
+ "async-channel 1.6.1",
  "futures-core",
  "futures-io",
 ]

crates/bonfire/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-bonfire"
-version = "0.7.5"
+version = "0.7.7"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 
@@ -14,6 +14,8 @@ lru = "0.7.6"
 ulid = "0.5.0"
 once_cell = "1.9.0"
 redis-kiss = "0.1.4"
+lru_time_cache = "0.11.11"
+async-channel = "2.3.1"
 
 # parsing
 querystring = "1.1.0"
@@ -39,7 +41,7 @@ revolt-result = { path = "../core/result" }
 revolt-models = { path = "../core/models" }
 revolt-config = { path = "../core/config" }
 revolt-database = { path = "../core/database" }
-revolt-permissions = { version = "0.7.5", path = "../core/permissions" }
+revolt-permissions = { version = "0.7.7", path = "../core/permissions" }
 revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
 
 # redis

crates/bonfire/src/events/impl.rs

@@ -92,6 +92,7 @@ impl State {
     /// Generate a Ready packet for the current user
     pub async fn generate_ready_payload(&mut self, db: &Database) -> Result<EventV1> {
         let user = self.clone_user();
+        self.cache.is_bot = user.bot.is_some();
 
         // Find all relationships to the user.
         let mut user_ids: HashSet<String> = user
@@ -191,6 +192,10 @@ impl State {
 
         for server in &servers {
             self.insert_subscription(server.id.clone()).await;
+
+            if self.cache.is_bot {
+                self.insert_subscription(format!("{}u", server.id)).await;
+            }
         }
 
         for channel in &channels {
@@ -397,6 +402,11 @@ impl State {
                 emojis: _,
             } => {
                 self.insert_subscription(id.clone()).await;
+
+                if self.cache.is_bot {
+                    self.insert_subscription(format!("{}u", id)).await;
+                }
+
                 self.cache.servers.insert(id.clone(), server.clone().into());
                 let member = Member {
                     id: MemberCompositeKey {

crates/bonfire/src/events/state.rs

@@ -1,10 +1,12 @@
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
+    time::Duration,
 };
 
-use async_std::sync::RwLock;
+use async_std::sync::{Mutex, RwLock};
 use lru::LruCache;
+use lru_time_cache::{LruCache as LruTimeCache, TimedEntry};
 use revolt_database::{Channel, Member, Server, User};
 
 /// Enumeration representing some change in subscriptions
@@ -34,6 +36,7 @@ pub enum SubscriptionStateChange {
 #[derive(Debug)]
 pub struct Cache {
     pub user_id: String,
+    pub is_bot: bool,
 
     pub users: HashMap<String, User>,
     pub channels: HashMap<String, Channel>,
@@ -47,6 +50,7 @@ impl Default for Cache {
     fn default() -> Self {
         Cache {
             user_id: Default::default(),
+            is_bot: false,
 
             users: Default::default(),
             channels: Default::default(),
@@ -65,7 +69,9 @@ pub struct State {
     pub session_id: String,
     pub private_topic: String,
     pub state: SubscriptionStateChange,
+
     pub subscribed: Arc<RwLock<HashSet<String>>>,
+    pub active_servers: Arc<Mutex<LruTimeCache<String, ()>>>,
 }
 
 impl State {
@@ -86,6 +92,10 @@ impl State {
         State {
             cache,
             subscribed: Arc::new(RwLock::new(subscribed)),
+            active_servers: Arc::new(Mutex::new(LruTimeCache::with_expiry_duration_and_capacity(
+                Duration::from_secs(900),
+                5,
+            ))),
             session_id,
             private_topic,
             state: SubscriptionStateChange::Reset,
@@ -94,6 +104,39 @@ impl State {
 
     /// Apply currently queued state
     pub async fn apply_state(&mut self) -> SubscriptionStateChange {
+        // Check if we need to change subscriptions to member event topics
+        if !self.cache.is_bot {
+            enum Server {
+                Subscribe(String),
+                Unsubscribe(String),
+            }
+
+            let active_server_changes: Vec<Server> = {
+                let mut active_servers = self.active_servers.lock().await;
+                active_servers
+                    .notify_iter()
+                    .map(|e| match e {
+                        TimedEntry::Valid(k, _) => Server::Subscribe(format!("{}u", k)),
+                        TimedEntry::Expired(k, _) => Server::Unsubscribe(format!("{}u", k)),
+                    })
+                    .collect()
+                // It is bad practice to open more than one Mutex at once and could
+                // lead to a deadlock, so instead we choose to collect the changes.
+            };
+
+            for entry in active_server_changes {
+                match entry {
+                    Server::Subscribe(k) => {
+                        self.insert_subscription(k).await;
+                    }
+                    Server::Unsubscribe(k) => {
+                        self.remove_subscription(&k).await;
+                    }
+                }
+            }
+        }
+
+        // Flush changes to subscriptions
         let state = std::mem::replace(&mut self.state, SubscriptionStateChange::None);
         let mut subscribed = self.subscribed.write().await;
         if let SubscriptionStateChange::Change { add, remove } = &state {

crates/bonfire/src/websocket.rs

@@ -8,7 +8,7 @@ use fred::{
 };
 use futures::{
     channel::oneshot,
-    pin_mut, select,
+    join, pin_mut, select,
     stream::{SplitSink, SplitStream},
     FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
@@ -22,6 +22,7 @@ use revolt_presence::{create_session, delete_session};
 use async_std::{
     net::TcpStream,
     sync::{Mutex, RwLock},
+    task::spawn,
 };
 use revolt_result::create_error;
 use sentry::Level;
@@ -127,22 +128,44 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     }
 
     {
+        // Setup channels and mutexes
         let write = Mutex::new(write);
         let subscribed = state.subscribed.clone();
+        let active_servers = state.active_servers.clone();
+        let (topic_signal_s, topic_signal_r) = async_channel::unbounded();
+
+        // TODO: this needs to be rewritten
+        // Create channels through which the tasks can signal to each other they need to clean up
+        let (kill_signal_1_s, kill_signal_1_r) = async_channel::bounded(1);
+        let (kill_signal_2_s, kill_signal_2_r) = async_channel::bounded(1);
 
         // Create a PubSub connection to poll on.
-        let listener = listener(db, &mut state, addr, &config, &write).fuse();
+        let listener = listener_with_kill_signal(
-        // Read from WebSocket stream.
+            db,
-        let worker = worker(addr, subscribed, user_id.clone(), &config, read, &write).fuse();
+            &mut state,
-
+            addr,
-        // Pin both tasks.
+            &config,
-        pin_mut!(listener, worker);
+            topic_signal_r,
-
+            kill_signal_1_r,
-        // Wait for either disconnect or for listener to die.
+            &write,
-        select!(
+            kill_signal_2_s,
-            () = listener => {},
-            () = worker => {}
         );
+
+        // Read from WebSocket stream.
+        let worker = worker_with_kill_signal(
+            addr,
+            subscribed,
+            active_servers,
+            user_id.clone(),
+            &config,
+            topic_signal_s,
+            kill_signal_2_r,
+            read,
+            &write,
+            kill_signal_1_s,
+        );
+
+        join!(listener, worker);
     }
     // Clean up presence session.
     let last_session = delete_session(&user_id, session_id).await;
@@ -153,15 +176,40 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     }
 }
 
+#[allow(clippy::too_many_arguments)]
+async fn listener_with_kill_signal(
+    db: &'static Database,
+    state: &mut State,
+    addr: SocketAddr,
+    config: &ProtocolConfiguration,
+    topic_signal_r: async_channel::Receiver<()>,
+    kill_signal_r: async_channel::Receiver<()>,
+    write: &Mutex<WsWriter>,
+    kill_signal_s: async_channel::Sender<()>,
+) {
+    listener(
+        db,
+        state,
+        addr,
+        config,
+        topic_signal_r,
+        kill_signal_r,
+        write,
+    )
+    .await;
+    kill_signal_s.send(()).await.ok();
+}
+
 async fn listener(
     db: &'static Database,
     state: &mut State,
     addr: SocketAddr,
     config: &ProtocolConfiguration,
+    topic_signal_r: async_channel::Receiver<()>,
+    kill_signal_r: async_channel::Receiver<()>,
     write: &Mutex<WsWriter>,
 ) {
     let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
-
     let subscriber = match fred::types::Builder::from_config(redis_config).build_subscriber_client()
     {
         Ok(subscriber) => subscriber,
@@ -178,15 +226,26 @@ async fn listener(
         return;
     };
 
+    // Handle Redis connection dropping
+    let (clean_up_s, clean_up_r) = async_channel::bounded(1);
+    let clean_up_s = Arc::new(Mutex::new(clean_up_s));
+    subscriber.on_error(move |_| {
+        let clean_up_s = clean_up_s.clone();
+        spawn(async move {
+            clean_up_s.lock().await.send(()).await.ok();
+        });
+        Ok(())
+    });
+
     let mut message_rx = subscriber.message_rx();
-    loop {
+    'out: loop {
         // Check for state changes for subscriptions.
         match state.apply_state().await {
             SubscriptionStateChange::Reset => {
                 if let Err(err) = subscriber.unsubscribe_all().await {
                     error!("Unsubscribe all failed: {err:?}");
                     sentry::capture_error(&err);
-                    return;
+                    break 'out;
                 }
 
                 let subscribed = state.subscribed.read().await;
@@ -194,7 +253,7 @@ async fn listener(
                     if let Err(err) = subscriber.subscribe(id).await {
                         error!("Subscribe failed: {err:?}");
                         sentry::capture_error(&err);
-                        return;
+                        break 'out;
                     }
                 }
 
@@ -209,7 +268,7 @@ async fn listener(
                     if let Err(err) = subscriber.unsubscribe(id).await {
                         error!("Unsubscribe failed: {err:?}");
                         sentry::capture_error(&err);
-                        return;
+                        break 'out;
                     }
                 }
 
@@ -220,165 +279,241 @@ async fn listener(
                     if let Err(err) = subscriber.subscribe(id).await {
                         error!("Subscribe failed: {err:?}");
                         sentry::capture_error(&err);
-                        return;
+                        break 'out;
                     }
                 }
             }
             SubscriptionStateChange::None => {}
         }
 
-        // Handle incoming events.
+        let t1 = message_rx.recv().fuse();
-        let message = match message_rx.recv().await {
+        let t2 = topic_signal_r.recv().fuse();
-            Ok(message) => message,
+        let t3 = kill_signal_r.recv().fuse();
-            Err(e) => {
+        let t4 = clean_up_r.recv().fuse();
-                error!("Error while consuming pub/sub messages: {e:?}");
-                sentry::capture_error(&e);
-                return;
-            }
-        };
 
-        let event = match *REDIS_PAYLOAD_TYPE {
+        pin_mut!(t1, t2, t3, t4);
-            PayloadType::Json => message
-                .value
-                .as_str()
-                .and_then(|s| serde_json::from_str::<EventV1>(s.as_ref()).ok()),
-            PayloadType::Msgpack => message
-                .value
-                .as_bytes()
-                .and_then(|b| rmp_serde::from_slice::<EventV1>(b).ok()),
-            PayloadType::Bincode => message
-                .value
-                .as_bytes()
-                .and_then(|b| bincode::deserialize::<EventV1>(b).ok()),
-        };
 
-        let Some(mut event) = event else {
+        select! {
-            let err = format!(
+            _ = t4 => {
-                "Failed to deserialise an event for {}! Introspection: `{:?}`",
+                break 'out;
-                message.channel,
+            },
-                message
+            _ = t3 => {
-                    .value
+                break 'out;
-                    .as_string()
+            },
-                    .map(|x| x.chars().take(32).collect::<String>())
+            _ = t2 => {},
-            );
+            message = t1 => {
+                // Handle incoming events.
+                let message = match message {
+                    Ok(message) => message,
+                    Err(e) => {
+                        error!("Error while consuming pub/sub messages: {e:?}");
+                        sentry::capture_error(&e);
+                        break 'out;
+                    }
+                };
 
-            error!("{}", err);
+                let event = match *REDIS_PAYLOAD_TYPE {
-            sentry::capture_message(&err, Level::Error);
+                    PayloadType::Json => message
-            return;
+                        .value
-        };
+                        .as_str()
+                        .and_then(|s| serde_json::from_str::<EventV1>(s.as_ref()).ok()),
+                    PayloadType::Msgpack => message
+                        .value
+                        .as_bytes()
+                        .and_then(|b| rmp_serde::from_slice::<EventV1>(b).ok()),
+                    PayloadType::Bincode => message
+                        .value
+                        .as_bytes()
+                        .and_then(|b| bincode::deserialize::<EventV1>(b).ok()),
+                };
 
-        if let EventV1::Auth(auth) = &event {
+                let Some(mut event) = event else {
-            if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
+                    let err = format!(
-                if &state.session_id == session_id {
+                        "Failed to deserialise an event for {}! Introspection: `{:?}`",
-                    event = EventV1::Logout;
+                        message.channel,
-                }
+                        message
-            } else if let AuthifierEvent::DeleteAllSessions {
+                            .value
-                exclude_session_id, ..
+                            .as_string()
-            } = auth
+                            .map(|x| x.chars().take(32).collect::<String>())
-            {
+                    );
-                if let Some(excluded) = exclude_session_id {
+
-                    if &state.session_id != excluded {
+                    error!("{}", err);
-                        event = EventV1::Logout;
+                    sentry::capture_message(&err, Level::Error);
+                    break 'out;
+                };
+
+                if let EventV1::Auth(auth) = &event {
+                    if let AuthifierEvent::DeleteSession { session_id, .. } = auth {
+                        if &state.session_id == session_id {
+                            event = EventV1::Logout;
+                        }
+                    } else if let AuthifierEvent::DeleteAllSessions {
+                        exclude_session_id, ..
+                    } = auth
+                    {
+                        if let Some(excluded) = exclude_session_id {
+                            if &state.session_id != excluded {
+                                event = EventV1::Logout;
+                            }
+                        } else {
+                            event = EventV1::Logout;
+                        }
                     }
                 } else {
-                    event = EventV1::Logout;
+                    let should_send = state.handle_incoming_event_v1(db, &mut event).await;
+                    if !should_send {
+                        continue;
+                    }
+                }
+
+                let result = write.lock().await.send(config.encode(&event)).await;
+                if let Err(e) = result {
+                    use async_tungstenite::tungstenite::Error;
+                    if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
+                        let err = format!("Error while sending an event to {addr:?}: {e:?}");
+                        warn!("{}", err);
+                        sentry::capture_message(&err, Level::Warning);
+                    }
+
+                    break 'out;
+                }
+
+                if let EventV1::Logout = event {
+                    info!("User {addr:?} received log out event!");
+                    break 'out;
                 }
             }
-        } else {
-            let should_send = state.handle_incoming_event_v1(db, &mut event).await;
-            if !should_send {
-                continue;
-            }
         }
+    }
 
-        let result = write.lock().await.send(config.encode(&event)).await;
+    if let Err(err) = subscriber.quit().await {
-        if let Err(e) = result {
+        error!("{}", err);
-            use async_tungstenite::tungstenite::Error;
+        sentry::capture_error(&err);
-            if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
-                let err = format!("Error while sending an event to {addr:?}: {e:?}");
-                warn!("{}", err);
-                sentry::capture_message(&err, Level::Warning);
-            }
-
-            return;
-        }
-
-        if let EventV1::Logout = event {
-            info!("User {addr:?} received log out event!");
-            return;
-        }
     }
 }
 
+#[allow(clippy::too_many_arguments)]
+async fn worker_with_kill_signal(
+    addr: SocketAddr,
+    subscribed: Arc<RwLock<HashSet<String>>>,
+    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
+    user_id: String,
+    config: &ProtocolConfiguration,
+    topic_signal_s: async_channel::Sender<()>,
+    kill_signal_r: async_channel::Receiver<()>,
+    read: WsReader,
+    write: &Mutex<WsWriter>,
+    kill_signal_s: async_channel::Sender<()>,
+) {
+    worker(
+        addr,
+        subscribed,
+        active_servers,
+        user_id,
+        config,
+        topic_signal_s,
+        kill_signal_r,
+        read,
+        write,
+    )
+    .await;
+    kill_signal_s.send(()).await.ok();
+}
+
+#[allow(clippy::too_many_arguments)]
 async fn worker(
     addr: SocketAddr,
     subscribed: Arc<RwLock<HashSet<String>>>,
+    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
     user_id: String,
     config: &ProtocolConfiguration,
+    topic_signal_s: async_channel::Sender<()>,
+    kill_signal_r: async_channel::Receiver<()>,
     mut read: WsReader,
     write: &Mutex<WsWriter>,
 ) {
     loop {
-        let result = read.try_next().await;
+        let t1 = read.try_next().fuse();
-        let msg = match result {
+        let t2 = kill_signal_r.recv().fuse();
-            Ok(Some(msg)) => msg,
+
-            Ok(None) => {
+        pin_mut!(t1, t2);
-                warn!("Received a None message!");
+
-                sentry::capture_message("Received a None message!", Level::Warning);
+        select! {
+            _ = t2 => {
                 return;
-            }
+            },
-            Err(e) => {
+            result = t1 => {
-                use async_tungstenite::tungstenite::Error;
+                let msg = match result {
-                if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
+                    Ok(Some(msg)) => msg,
-                    let err = format!("Error while reading an event from {addr:?}: {e:?}");
+                    Ok(None) => {
-                    warn!("{}", err);
+                        warn!("Received a None message!");
-                    sentry::capture_message(&err, Level::Warning);
+                        sentry::capture_message("Received a None message!", Level::Warning);
-                }
+                        return;
+                    }
+                    Err(e) => {
+                        use async_tungstenite::tungstenite::Error;
+                        if !matches!(e, Error::AlreadyClosed | Error::ConnectionClosed) {
+                            let err = format!("Error while reading an event from {addr:?}: {e:?}");
+                            warn!("{}", err);
+                            sentry::capture_message(&err, Level::Warning);
+                        }
 
-                return;
+                        return;
-            }
+                    }
-        };
+                };
 
-        let Ok(payload) = config.decode(&msg) else {
+                let Ok(payload) = config.decode(&msg) else {
-            continue;
-        };
-
-        match payload {
-            ClientMessage::BeginTyping { channel } => {
-                if !subscribed.read().await.contains(&channel) {
                     continue;
-                }
+                };
 
-                EventV1::ChannelStartTyping {
+                match payload {
-                    id: channel.clone(),
+                    ClientMessage::BeginTyping { channel } => {
-                    user: user_id.clone(),
+                        if !subscribed.read().await.contains(&channel) {
-                }
+                            continue;
-                .p(channel.clone())
+                        }
-                .await;
-            }
-            ClientMessage::EndTyping { channel } => {
-                if !subscribed.read().await.contains(&channel) {
-                    continue;
-                }
 
-                EventV1::ChannelStopTyping {
+                        EventV1::ChannelStartTyping {
-                    id: channel.clone(),
+                            id: channel.clone(),
-                    user: user_id.clone(),
+                            user: user_id.clone(),
-                }
+                        }
-                .p(channel.clone())
+                        .p(channel.clone())
-                .await;
+                        .await;
-            }
+                    }
-            ClientMessage::Ping { data, responded } => {
+                    ClientMessage::EndTyping { channel } => {
-                if responded.is_none() {
+                        if !subscribed.read().await.contains(&channel) {
-                    write
+                            continue;
-                        .lock()
+                        }
-                        .await
+
-                        .send(config.encode(&EventV1::Pong { data }))
+                        EventV1::ChannelStopTyping {
-                        .await
+                            id: channel.clone(),
-                        .ok();
+                            user: user_id.clone(),
+                        }
+                        .p(channel.clone())
+                        .await;
+                    }
+                    ClientMessage::Subscribe { server_id } => {
+                        let mut servers = active_servers.lock().await;
+                        let has_item = servers.contains_key(&server_id);
+                        servers.insert(server_id, ());
+
+                        if !has_item {
+                            // Poke the listener to adjust subscriptions
+                            topic_signal_s.send(()).await.ok();
+                        }
+                    }
+                    ClientMessage::Ping { data, responded } => {
+                        if responded.is_none() {
+                            write
+                                .lock()
+                                .await
+                                .send(config.encode(&EventV1::Pong { data }))
+                                .await
+                                .ok();
+                        }
+                    }
+                    _ => {}
                 }
             }
-            _ => {}
         }
     }
 }

crates/core/config/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-config"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/database/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -23,13 +23,13 @@ default = ["mongodb", "async-std-runtime", "tasks"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.7.5", path = "../config" }
+revolt-config = { version = "0.7.7", path = "../config" }
-revolt-result = { version = "0.7.5", path = "../result" }
+revolt-result = { version = "0.7.7", path = "../result" }
-revolt-models = { version = "0.7.5", path = "../models", features = [
+revolt-models = { version = "0.7.7", path = "../models", features = [
     "validator",
 ] }
-revolt-presence = { version = "0.7.5", path = "../presence" }
+revolt-presence = { version = "0.7.7", path = "../presence" }
-revolt-permissions = { version = "0.7.5", path = "../permissions", features = [
+revolt-permissions = { version = "0.7.7", path = "../permissions", features = [
     "serde",
     "bson",
 ] }

crates/core/database/src/events/client.rs

@@ -249,7 +249,7 @@ impl EventV1 {
         // TODO: this should be captured by member list in the future and not immediately fanned out to users
         if let Ok(members) = db.fetch_all_memberships(&id).await {
             for member in members {
-                self.clone().p(member.id.server).await;
+                self.clone().server(member.id.server).await;
             }
         }
     }
@@ -259,6 +259,11 @@ impl EventV1 {
         self.p(format!("{id}!")).await;
     }
 
+    /// Publish server member event
+    pub async fn server(self, id: String) {
+        self.p(format!("{id}u")).await;
+    }
+
     /// Publish internal global event
     pub async fn global(self) {
         self.p("global".to_string()).await;

crates/core/database/src/events/server.rs

@@ -8,5 +8,6 @@ pub enum ClientMessage {
     Authenticate { token: String },
     BeginTyping { channel: String },
     EndTyping { channel: String },
+    Subscribe { server_id: String },
     Ping { data: Ping, responded: Option<()> },
 }

crates/core/models/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-models"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -19,8 +19,8 @@ default = ["serde", "partials", "rocket"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.7.5", path = "../config" }
+revolt-config = { version = "0.7.7", path = "../config" }
-revolt-permissions = { version = "0.7.5", path = "../permissions" }
+revolt-permissions = { version = "0.7.7", path = "../permissions" }
 
 # Utility
 regex = "1"

crates/core/permissions/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-permissions"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -21,7 +21,7 @@ async-std = { version = "1.8.0", features = ["attributes"] }
 
 [dependencies]
 # Core
-revolt-result = { version = "0.7.5", path = "../result" }
+revolt-result = { version = "0.7.7", path = "../result" }
 
 # Utility
 auto_ops = "0.3.0"

crates/core/presence/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-presence"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/result/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-result"
-version = "0.7.5"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/delta/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-delta"
-version = "0.7.5"
+version = "0.7.7"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"