diff --git a/crates/bonfire/src/events/impl.rs b/crates/bonfire/src/events/impl.rs index 12113537..96a0f2ab 100644 --- a/crates/bonfire/src/events/impl.rs +++ b/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, diff --git a/crates/bonfire/src/websocket.rs b/crates/bonfire/src/websocket.rs index b5611bd0..ee551289 100644 --- a/crates/bonfire/src/websocket.rs +++ b/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>; type WsWriter = SplitSink, 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( } } } -} \ No newline at end of file +} + +async fn fetch_user_slowmodes(user_id: &str) -> Option> { + let mut conn = get_connection().await.ok()?.into_inner(); + let idx_key = format!("slowmode_idx:{}", user_id); + + let channel_ids: Vec = 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 = 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) +} diff --git a/crates/core/database/src/events/client.rs b/crates/core/database/src/events/client.rs index 38c6f941..31864cee 100644 --- a/crates/core/database/src/events/client.rs +++ b/crates/core/database/src/events/client.rs @@ -3,7 +3,12 @@ use revolt_result::Error; use serde::{Deserialize, Serialize}; use revolt_models::v0::{ - AppendMessage, Channel, ChannelUnread, ChannelVoiceState, Emoji, FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser, FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialEmoji, PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, UserVoiceState, Webhook + AppendMessage, Channel, ChannelSlowmode, ChannelUnread, ChannelVoiceState, Emoji, + FieldsChannel, FieldsMember, FieldsMessage, FieldsRole, FieldsServer, FieldsUser, + FieldsWebhook, Member, MemberCompositeKey, Message, PartialChannel, PartialEmoji, + PartialMember, PartialMessage, PartialRole, PartialServer, PartialUser, PartialUserVoiceState, + PartialWebhook, PolicyChange, RemovalIntention, Report, Server, User, UserSettings, + UserVoiceState, Webhook, }; use crate::Database; @@ -51,9 +56,13 @@ impl Default for ReadyPayloadFields { #[serde(tag = "type")] pub enum EventV1 { /// Multiple events - Bulk { v: Vec }, + Bulk { + v: Vec, + }, /// Error event - Error { data: Error }, + Error { + data: Error, + }, /// Successfully authenticated Authenticated, @@ -84,7 +93,9 @@ pub enum EventV1 { }, /// Ping response - Pong { data: Ping }, + Pong { + data: Ping, + }, /// New message Message(Message), @@ -105,7 +116,10 @@ pub enum EventV1 { }, /// Delete message - MessageDelete { id: String, channel: String }, + MessageDelete { + id: String, + channel: String, + }, /// New reaction to a message MessageReact { @@ -131,7 +145,10 @@ pub enum EventV1 { }, /// Bulk delete messages - BulkMessageDelete { channel: String, ids: Vec }, + BulkMessageDelete { + channel: String, + ids: Vec, + }, /// New server ServerCreate { @@ -139,7 +156,7 @@ pub enum EventV1 { server: Server, channels: Vec, emojis: Vec, - voice_states: Vec + voice_states: Vec, }, /// Update existing server @@ -151,7 +168,9 @@ pub enum EventV1 { }, /// Delete server - ServerDelete { id: String }, + ServerDelete { + id: String, + }, /// Update existing server member ServerMemberUpdate { @@ -187,10 +206,16 @@ pub enum EventV1 { }, /// Server role deleted - ServerRoleDelete { id: String, role_id: String }, + ServerRoleDelete { + id: String, + role_id: String, + }, /// Server roles ranks updated - ServerRoleRanksUpdate { id: String, ranks: Vec }, + ServerRoleRanksUpdate { + id: String, + ranks: Vec, + }, /// Update existing user UserUpdate { @@ -202,9 +227,15 @@ pub enum EventV1 { }, /// Relationship with another user changed - UserRelationship { id: String, user: User }, + UserRelationship { + id: String, + user: User, + }, /// Settings updated remotely - UserSettingsUpdate { id: String, update: UserSettings }, + UserSettingsUpdate { + id: String, + update: UserSettings, + }, /// User has been platform banned or deleted their account /// @@ -215,7 +246,10 @@ pub enum EventV1 { /// - Server Memberships /// /// User flags are specified to explain why a wipe is occurring though not all reasons will necessarily ever appear. - UserPlatformWipe { user_id: String, flags: i32 }, + UserPlatformWipe { + user_id: String, + flags: i32, + }, /// New emoji EmojiCreate(Emoji), @@ -226,7 +260,9 @@ pub enum EventV1 { }, /// Delete emoji - EmojiDelete { id: String }, + EmojiDelete { + id: String, + }, /// New report ReportCreate(Report), @@ -242,19 +278,33 @@ pub enum EventV1 { }, /// Delete channel - ChannelDelete { id: String }, + ChannelDelete { + id: String, + }, /// User joins a group - ChannelGroupJoin { id: String, user: String }, + ChannelGroupJoin { + id: String, + user: String, + }, /// User leaves a group - ChannelGroupLeave { id: String, user: String }, + ChannelGroupLeave { + id: String, + user: String, + }, /// User started typing in a channel - ChannelStartTyping { id: String, user: String }, + ChannelStartTyping { + id: String, + user: String, + }, /// User stopped typing in a channel - ChannelStopTyping { id: String, user: String }, + ChannelStopTyping { + id: String, + user: String, + }, /// User acknowledged message in channel ChannelAck { @@ -274,7 +324,9 @@ pub enum EventV1 { }, /// Delete webhook - WebhookDelete { id: String }, + WebhookDelete { + id: String, + }, /// Auth events Auth(AuthifierEvent), @@ -292,7 +344,7 @@ pub enum EventV1 { user: String, from: String, to: String, - state: UserVoiceState + state: UserVoiceState, }, UserVoiceStateUpdate { id: String, @@ -304,7 +356,11 @@ pub enum EventV1 { from: String, to: String, token: String, - } + }, + /// User's active slowmodes + UserSlowmodes { + slowmodes: Vec, + }, } impl EventV1 { diff --git a/crates/core/models/src/v0/channels.rs b/crates/core/models/src/v0/channels.rs index d9fe8e14..faddee74 100644 --- a/crates/core/models/src/v0/channels.rs +++ b/crates/core/models/src/v0/channels.rs @@ -314,6 +314,12 @@ auto_derived!( /// Only used when the user is the first one connected. pub recipients: Option>, } + + pub struct ChannelSlowmode { + pub channel_id: String, + pub duration: u64, + pub retry_after: u64, + } ); impl Channel { diff --git a/crates/delta/src/routes/channels/message_send.rs b/crates/delta/src/routes/channels/message_send.rs index 7406b605..e6551d0c 100644 --- a/crates/delta/src/routes/channels/message_send.rs +++ b/crates/delta/src/routes/channels/message_send.rs @@ -1,12 +1,14 @@ use std::time::Duration; use redis_kiss::{get_connection, redis, AsyncCommands}; +use revolt_database::events::client::EventV1; use revolt_database::util::permissions::DatabasePermissionQuery; use revolt_database::{ util::idempotency::IdempotencyKey, util::reference::Reference, Database, User, }; use revolt_database::{Channel, Interactions, Message, AMQP}; use revolt_models::v0; +use revolt_models::v0::ChannelSlowmode; use revolt_permissions::PermissionQuery; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; @@ -84,6 +86,16 @@ pub async fn message_send( .await .unwrap_or(None); + if set_result.is_some() { + let idx_key = format!("slowmode_idx:{}", user.id); + conn.sadd::<_, _, ()>(&idx_key, channel_id.as_str()) + .await + .ok(); + conn.expire::<_, ()>(&idx_key, *channel_slowmode as usize) + .await + .ok(); + } + // If `set_result` is None, the `NX` condition failed because the key already exists. // This means the user is currently in slowmode. if set_result.is_none() { @@ -92,10 +104,29 @@ pub async fn message_send( // Redis returns positive integers for valid TTLs if ttl > 0 { + EventV1::UserSlowmodes { + slowmodes: vec![ChannelSlowmode { + channel_id: channel_id.to_string(), + duration: *channel_slowmode, + retry_after: ttl as u64, + }], + } + .private(user.id.clone()) + .await; return Err(create_error!(InSlowmode { retry_after: ttl as u64 })); } + } else { + EventV1::UserSlowmodes { + slowmodes: vec![ChannelSlowmode { + channel_id: channel_id.to_string(), + duration: *channel_slowmode, + retry_after: *channel_slowmode, + }], + } + .private(user.id.clone()) + .await; } } // If Redis connection fails, just skip the slowmode check