feat: user slowmode events (#760)

* feat: user slowmode events

Signed-off-by: ispik <ispik@ispik.dev>

* fix: remove debug print statement for slowmodes

Signed-off-by: ispik <ispik@ispik.dev>

* refactor: Send user slowmodes as websocket connects instead of trying to send it in ready payload

Signed-off-by: ispik <ispik@ispik.dev>

* refactor: optimize user slowmode handling with bulk operations

Signed-off-by: ispik <ispik@ispik.dev>

* chore: specify release version

Release-As: 0.13.6

---------

Signed-off-by: ispik <ispik@ispik.dev>

crates/bonfire/src/events/impl.rs

@@ -1,6 +1,7 @@
 use std::collections::{HashMap, HashSet};
 
 use futures::future::join_all;
+use redis_kiss::AsyncCommands;
 use revolt_database::{
     events::client::{EventV1, ReadyPayloadFields},
     util::permissions::DatabasePermissionQuery,

crates/bonfire/src/websocket.rs

@@ -13,7 +13,7 @@ use futures::{
     stream::{SplitSink, SplitStream},
     FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
-use redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
+use redis_kiss::{get_connection, AsyncCommands, PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI};
 use revolt_config::report_internal_error;
 use revolt_database::{
     events::{client::EventV1, server::ClientMessage},
@@ -32,6 +32,7 @@ use sentry::Level;
 
 use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
 use crate::events::state::{State, SubscriptionStateChange};
+use revolt_models::v0;
 
 type WsReader = SplitStream<WebSocketStream<TcpStream>>;
 type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
@@ -128,6 +129,14 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
         return;
     }
 
+    let slowmodes = fetch_user_slowmodes(&user_id).await.unwrap_or_default();
+    if !slowmodes.is_empty() {
+        let event = EventV1::UserSlowmodes { slowmodes };
+        if report_internal_error!(write.send(config.encode(&event)).await).is_err() {
+            return;
+        }
+    }
+
     // Create presence session.
     let (first_session, session_id) = create_session(&user_id, 0).await;
 
@@ -527,4 +536,43 @@ async fn worker(
             }
         }
     }
-}
+}
+
+async fn fetch_user_slowmodes(user_id: &str) -> Option<Vec<v0::ChannelSlowmode>> {
+    let mut conn = get_connection().await.ok()?.into_inner();
+    let idx_key = format!("slowmode_idx:{}", user_id);
+
+    let channel_ids: Vec<String> = conn.smembers(&idx_key).await.unwrap_or_default();
+    if channel_ids.is_empty() {
+        return Some(vec![]);
+    }
+
+    // Bulk fetch all TTLs in one round trip
+    let mut pipe = redis_kiss::redis::pipe();
+    for channel_id in &channel_ids {
+        pipe.ttl(format!("slowmode:{}:{}", user_id, channel_id));
+    }
+    let ttls: Vec<i64> = pipe.query_async(&mut conn).await.unwrap_or_default();
+
+    // Partition into alive/expired in one pass
+    let mut slowmodes = vec![];
+    let mut expired = vec![];
+    for (channel_id, ttl) in channel_ids.iter().zip(ttls.iter()) {
+        if *ttl > 0 {
+            slowmodes.push(v0::ChannelSlowmode {
+                channel_id: channel_id.clone(),
+                duration: *ttl as u64,
+                retry_after: *ttl as u64,
+            });
+        } else {
+            expired.push(channel_id.as_str());
+        }
+    }
+
+    // Bulk remove all expired members in one SREM call
+    if !expired.is_empty() {
+        conn.srem::<_, _, ()>(&idx_key, expired).await.ok();
+    }
+
+    Some(slowmodes)
+}