From 98c7b1b5a5b9fdac5c0ab83be10f0e23114dbfc9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C4=B0spik?= Date: Sat, 28 Mar 2026 03:02:12 +0300 Subject: [PATCH] feat: implement time based message sweep on user ban (#670) * feat: implement time based message sweep on user ban - Adds `delete_message_seconds` (0 to 7 days in seconds) to the ban request payload. Signed-off-by: arsabutispik * refactor: pass ulid conversion to database Signed-off-by: arsabutispik * fix: use COL constant instead of hardcoded string in error mapper Signed-off-by: arsabutispik * refactor: broadcast bulk delete events during ban sweep Updates the `delete_messages_by_author_since` trait to return a HashMap of deleted message IDs grouped by channel. The MongoDB implementation now uses a two-step process: it first runs a projected `find` query to gather the target `_id` and `channel` fields, then executes the `delete_many` operation. This allows the ban route to loop through the affected channels and dispatch `EventV1::BulkMessageDelete` WebSocket events, ensuring that the swept messages are instantly removed from the UI for all connected clients. Signed-off-by: arsabutispik * refactor: optimize message deletion by using $group and aggregate Signed-off-by: arsabutispik * refactor: use with_type in query Signed-off-by: arsabutispik * refactor: abstract function to Message model and mark attachments as deleted Signed-off-by: arsabutispik --------- Signed-off-by: arsabutispik --- .../database/src/models/messages/model.rs | 31 ++++- .../core/database/src/models/messages/ops.rs | 10 ++ .../src/models/messages/ops/mongodb.rs | 110 ++++++++++++++++++ .../src/models/messages/ops/reference.rs | 63 +++++++++- crates/core/models/src/v0/server_bans.rs | 3 + crates/delta/src/routes/servers/ban_create.rs | 13 ++- 6 files changed, 225 insertions(+), 5 deletions(-) diff --git a/crates/core/database/src/models/messages/model.rs b/crates/core/database/src/models/messages/model.rs index 2db57b12..3db6df49 100644 --- a/crates/core/database/src/models/messages/model.rs +++ b/crates/core/database/src/models/messages/model.rs @@ -1,5 +1,3 @@ -use std::{collections::HashSet, hash::RandomState}; - use indexmap::{IndexMap, IndexSet}; use iso8601_timestamp::Timestamp; use revolt_config::{config, FeaturesLimits}; @@ -9,6 +7,8 @@ use revolt_models::v0::{ }; use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue}; use revolt_result::{ErrorType, Result}; +use std::time::SystemTime; +use std::{collections::HashSet, hash::RandomState}; use ulid::Ulid; use validator::Validate; @@ -497,7 +497,7 @@ impl Message { user_mentions.retain(|m| recipients_hash.contains(m)); role_mentions.clear(); } - Channel::TextChannel { ref server, .. }=> { + Channel::TextChannel { ref server, .. } => { let mentions_vec = Vec::from_iter(user_mentions.iter().cloned()); let valid_members = db.fetch_members(server.as_str(), &mentions_vec[..]).await; @@ -1034,6 +1034,31 @@ impl Message { Ok(()) } + /// Bulk delete messages by an author since a given time + pub async fn bulk_delete_by_author_since( + db: &Database, + channels: &[String], + author: &str, + since: SystemTime, + ) -> Result<()> { + let deleted_groups = db + .delete_messages_by_author_since(channels, author, since) + .await?; + + for (channel_id, message_ids) in deleted_groups { + if !message_ids.is_empty() { + EventV1::BulkMessageDelete { + channel: channel_id.clone(), + ids: message_ids, + } + .p(channel_id) + .await; + } + } + + Ok(()) + } + /// Remove a reaction from a message pub async fn remove_reaction(&self, db: &Database, user: &str, emoji: &str) -> Result<()> { // Check if it actually exists diff --git a/crates/core/database/src/models/messages/ops.rs b/crates/core/database/src/models/messages/ops.rs index 4d54e83c..6ebdf4c3 100644 --- a/crates/core/database/src/models/messages/ops.rs +++ b/crates/core/database/src/models/messages/ops.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; +use std::time::SystemTime; use revolt_result::Result; use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage}; @@ -40,4 +42,12 @@ pub trait AbstractMessages: Sync + Send { /// Delete messages from a channel by their ids and corresponding channel id async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()>; + + /// Delete all messages from a specific author in a server from a certain ULID onwards + async fn delete_messages_by_author_since( + &self, + channels: &[String], + author: &str, + since: SystemTime + ) -> Result>>; } diff --git a/crates/core/database/src/models/messages/ops/mongodb.rs b/crates/core/database/src/models/messages/ops/mongodb.rs index 449e71ef..2a79ed5b 100644 --- a/crates/core/database/src/models/messages/ops/mongodb.rs +++ b/crates/core/database/src/models/messages/ops/mongodb.rs @@ -1,8 +1,12 @@ use bson::{to_bson, Document}; use futures::try_join; +use futures::StreamExt; use mongodb::options::FindOptions; use revolt_models::v0::MessageSort; use revolt_result::Result; +use std::collections::{HashMap, HashSet}; +use std::time::SystemTime; +use ulid::Ulid; use crate::{ AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath, Message, MessageQuery, @@ -306,6 +310,112 @@ impl AbstractMessages for MongoDb { .map(|_| ()) .map_err(|_| create_database_error!("delete_many", COL)) } + + /// Delete all messages from a specific author in a server from a certain ULID onwards + async fn delete_messages_by_author_since( + &self, + channels: &[String], + author: &str, + since: SystemTime, + ) -> Result>> { + let threshold_ulid = Ulid::from_datetime(since).to_string(); + + let filter = doc! { + "author": author, + "channel": { "$in": channels }, + "_id": { "$gte": &threshold_ulid } + }; + + let pipeline = vec![ + doc! { "$match": filter.clone() }, + doc! { + "$project": { + "channel": 1_i32, + "message_id": "$_id", + "attachment_ids": { + "$map": { + "input": { "$ifNull": ["$attachments", Vec::::new()] }, + "as": "a", + "in": "$$a._id" + } + } + } + }, + doc! { + "$group": { + "_id": "$channel", + "message_ids": { "$push": "$message_id" }, + "attachment_ids_nested": { "$push": "$attachment_ids" } + } + }, + doc! { + "$project": { + "message_ids": 1_i32, + "attachment_ids": { + "$reduce": { + "input": "$attachment_ids_nested", + "initialValue": Vec::::new(), + "in": { "$setUnion": ["$$value", "$$this"] } + } + } + } + }, + ]; + + #[derive(serde::Deserialize)] + struct AggregatedChannel { + #[serde(rename = "_id")] + channel: String, + message_ids: Vec, + #[serde(default)] + attachment_ids: Vec, + } + + let mut cursor = self + .col::(COL) + .aggregate(pipeline) + .await + .map_err(|_| create_database_error!("aggregate", COL))? + .with_type::(); + + let mut deleted_messages: HashMap> = HashMap::new(); + let mut attachment_ids: HashSet = HashSet::new(); + + while let Some(result) = cursor.next().await { + if let Ok(item) = result { + for id in item.attachment_ids { + attachment_ids.insert(id); + } + deleted_messages.insert(item.channel, item.message_ids); + } + } + + // Mark attachments as deleted before deleting messages + if !attachment_ids.is_empty() { + self.col::("attachments") + .update_many( + doc! { + "_id": { + "$in": attachment_ids.into_iter().collect::>() + } + }, + doc! { + "$set": { + "deleted": true + } + }, + ) + .await + .map_err(|_| create_database_error!("update_many", "attachments"))?; + } + + self.col::(COL) + .delete_many(filter) + .await + .map_err(|_| create_database_error!("delete_many", COL))?; + + Ok(deleted_messages) + } } impl IntoDocumentPath for FieldsMessage { diff --git a/crates/core/database/src/models/messages/ops/reference.rs b/crates/core/database/src/models/messages/ops/reference.rs index ae5b43f0..6f30de6f 100644 --- a/crates/core/database/src/models/messages/ops/reference.rs +++ b/crates/core/database/src/models/messages/ops/reference.rs @@ -1,7 +1,9 @@ +use std::collections::HashMap; use futures::future::try_join_all; use indexmap::IndexSet; use revolt_result::Result; - +use std::time::SystemTime; +use ulid::Ulid; use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage, ReferenceDb}; use super::AbstractMessages; @@ -286,4 +288,63 @@ impl AbstractMessages for ReferenceDb { Ok(()) } + + /// Delete all messages from a specific author in a list of channels from a certain ULID onwards + async fn delete_messages_by_author_since( + &self, + channels: &[String], + author: &str, + since: SystemTime + ) -> Result>> { + let threshold_ulid = Ulid::from_datetime(since).to_string(); + let mut deleted_messages: HashMap> = HashMap::new(); + let mut attachment_ids: Vec = Vec::new(); + + let messages = self.messages.lock().await; + + // First pass: collect attachment IDs and message IDs to delete + for (id, message) in messages.iter() { + let should_delete = message.author == author + && channels.contains(&message.channel) + && id.as_str() >= threshold_ulid.as_str(); + + if should_delete { + // Collect attachment IDs + if let Some(attachments) = &message.attachments { + for attachment in attachments { + attachment_ids.push(attachment.id.clone()); + } + } + + deleted_messages + .entry(message.channel.clone()) + .or_default() + .push(id.clone()); + } + } + drop(messages); + + // Mark attachments as deleted + if !attachment_ids.is_empty() { + let mut files = self.files.lock().await; + for attachment_id in attachment_ids { + if let Some(file) = files.get_mut(&attachment_id) { + file.deleted = Some(true); + } + } + } + + // Delete the messages + self.messages + .lock() + .await + .retain(|id, message| { + let should_keep = !(message.author == author + && channels.contains(&message.channel) + && id.as_str() >= threshold_ulid.as_str()); + should_keep + }); + + Ok(deleted_messages) + } } diff --git a/crates/core/models/src/v0/server_bans.rs b/crates/core/models/src/v0/server_bans.rs index b6debfcd..940a3738 100644 --- a/crates/core/models/src/v0/server_bans.rs +++ b/crates/core/models/src/v0/server_bans.rs @@ -19,6 +19,9 @@ auto_derived!( /// Ban reason #[cfg_attr(feature = "validator", validate(length(min = 0, max = 1024)))] pub reason: Option, + /// Messages to delete in seconds + #[cfg_attr(feature = "validator", validate(range(min = 0, max = 604800)))] + pub delete_message_seconds: Option, } /// Just enough information to list a ban diff --git a/crates/delta/src/routes/servers/ban_create.rs b/crates/delta/src/routes/servers/ban_create.rs index 4ff10329..a29b4300 100644 --- a/crates/delta/src/routes/servers/ban_create.rs +++ b/crates/delta/src/routes/servers/ban_create.rs @@ -4,13 +4,16 @@ use revolt_database::{ get_user_voice_channel_in_server, remove_user_from_voice_channel, UserVoiceChannel, VoiceClient, }, - Database, RemovalIntention, ServerBan, User, + Database, Message, RemovalIntention, ServerBan, User, }; use revolt_models::v0; +use std::time::{Duration, SystemTime}; +use revolt_database::events::client::EventV1; use revolt_permissions::{calculate_server_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; use rocket::{serde::json::Json, State}; +use ulid::Ulid; use validator::Validate; /// # Ban User @@ -73,7 +76,15 @@ pub async fn ban( .await?; } } + // We do this outside the member check so we can sweep hit-and-run spammers who already left. + if let Some(seconds) = data.delete_message_seconds { + if seconds > 0 { + let threshold_time = SystemTime::now() - Duration::from_secs(seconds as u64); + Message::bulk_delete_by_author_since(db, &server.channels, target.id, threshold_time) + .await?; + } + } ServerBan::create(db, &server, target.id, data.reason) .await .map(Into::into)