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 <ispik@ispik.dev>

* refactor: pass ulid conversion to database

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

* fix: use COL constant instead of hardcoded string in error mapper

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

* 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 <ispik@ispik.dev>

* refactor: optimize message deletion by using $group and aggregate

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

* refactor: use with_type in query

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

* refactor: abstract function to Message model and mark attachments as deleted

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

---------

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

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

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<HashMap<String, Vec<String>>>;
 }

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<HashMap<String, Vec<String>>> {
+        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::<bson::Bson>::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::<bson::Bson>::new(),
+                            "in": { "$setUnion": ["$$value", "$$this"] }
+                        }
+                    }
+                }
+            },
+        ];
+
+        #[derive(serde::Deserialize)]
+        struct AggregatedChannel {
+            #[serde(rename = "_id")]
+            channel: String,
+            message_ids: Vec<String>,
+            #[serde(default)]
+            attachment_ids: Vec<String>,
+        }
+
+        let mut cursor = self
+            .col::<Document>(COL)
+            .aggregate(pipeline)
+            .await
+            .map_err(|_| create_database_error!("aggregate", COL))?
+            .with_type::<AggregatedChannel>();
+
+        let mut deleted_messages: HashMap<String, Vec<String>> = HashMap::new();
+        let mut attachment_ids: HashSet<String> = 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::<Document>("attachments")
+                .update_many(
+                    doc! {
+                        "_id": {
+                            "$in": attachment_ids.into_iter().collect::<Vec<String>>()
+                        }
+                    },
+                    doc! {
+                        "$set": {
+                            "deleted": true
+                        }
+                    },
+                )
+                .await
+                .map_err(|_| create_database_error!("update_many", "attachments"))?;
+        }
+
+        self.col::<Document>(COL)
+            .delete_many(filter)
+            .await
+            .map_err(|_| create_database_error!("delete_many", COL))?;
+
+        Ok(deleted_messages)
+    }
 }
 
 impl IntoDocumentPath for FieldsMessage {

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<HashMap<String, Vec<String>>> {
+        let threshold_ulid = Ulid::from_datetime(since).to_string();
+        let mut deleted_messages: HashMap<String, Vec<String>> = HashMap::new();
+        let mut attachment_ids: Vec<String> = 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)
+    }
 }

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<String>,
+        /// Messages to delete in seconds
+        #[cfg_attr(feature = "validator", validate(range(min = 0, max = 604800)))]
+        pub delete_message_seconds: Option<i64>,
     }
 
     /// Just enough information to list a ban

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)