feat: Add Mass Mentions to the backend (#394)

* feat: create base of push daemon Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * Add outbound senders * Make web_push send to rabbit instead (temp stuff) * feat: stability and friend requests * make vapid fr stuff not suck * swap naming of queue * move pushd into daemons folder * fix cargo file for move into daemons folder * feat: probably working fcm push notifs * comment out fcm webpush stuff since the config keys dont exist * fix fcm, name queues according to their prod status and configure routing keys * add pushd to docker * mix: Remove old code, add stuff to pushd * fix: lockfile * feat: update rocket to 5.0.1 * fix: fix queues and ack bugs * Move rabbit messsage processing into ack queue * chore: update readme * chore: optimizations for ack database hits * pushd flowchart * misc: update flowchart * exit dependancy hell * add rocket_impl flag to authifier * make the tests file of delta actually compile * fix: don't silence every push message * fix: don't silence all messages * add debug logging for sending data to rabbit from message events * validate mentions at a server membership level * put back that import that was actually important * minor fix to lockfile * update delta authifier * feat: proper permissions for push notifications * add unit test for mention sanitization * remove local file dependancy on authifier * update ports to proper defaults * fixTM the node bindings * Theoretically configure docker releases for pushd Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * declare exchange in pushd and delta * fix createbuckets script Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: reference db implementation Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: remove finally redundant code Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: changes Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: other changes Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: make channel name return result Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * Add role mention parsing * feat: update to mongo 3.1, add member generator. * integrate mass mentions into pushd * patch redis-rs with updated versions * feat: chunk role mentions * move permission bits to 37/38 to avoid livekit conflict * change role mention format to <%id> * fix the lockfile from merge * fix: PR change requests * feat: add tests * fix: i am a dumbass * fix: tests, again --------- Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>
2025-04-05 10:45:04 -07:00
parent 93a5de769d
commit 2540860129
51 changed files with 4083 additions and 2553 deletions
--- a/crates/core/database/src/models/messages/model.rs
+++ b/crates/core/database/src/models/messages/model.rs
@@ -5,9 +5,10 @@ use iso8601_timestamp::Timestamp;
 use revolt_config::{config, FeaturesLimits};
 use revolt_models::v0::{
    self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort,
-    MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text, RE_MENTION,
+    MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text, RE_BLOCKS, RE_MENTION,
+    RE_ROLE_MENTION,
 };
-use revolt_permissions::{ChannelPermission, PermissionValue};
+use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue};
 use revolt_result::{ErrorType, Result};
 use ulid::Ulid;
 use validator::Validate;
@@ -15,7 +16,10 @@ use validator::Validate;
 use crate::{
    events::client::EventV1,
    tasks::{self, ack::AckEvent},
-    util::{bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey},
+    util::{
+        bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
+        permissions::DatabasePermissionQuery,
+    },
    Channel, Database, Emoji, File, User, AMQP,
 };

@@ -53,6 +57,9 @@ auto_derived_partial!(
        /// Array of user ids mentioned in this message
        #[serde(skip_serializing_if = "Option::is_none")]
        pub mentions: Option<Vec<String>>,
+        /// Array of role ids mentioned in this message
+        #[serde(skip_serializing_if = "Option::is_none")]
+        pub role_mentions: Option<Vec<String>>,
        /// Array of message ids this message is replying to
        #[serde(skip_serializing_if = "Option::is_none")]
        pub replies: Option<Vec<String>>,
@@ -71,7 +78,7 @@ auto_derived_partial!(

        /// Bitfield of message flags
        #[serde(skip_serializing_if = "Option::is_none")]
-        pub flags: Option<i32>,
+        pub flags: Option<u32>,
    },
    "PartialMessage"
 );
@@ -199,6 +206,30 @@ auto_derived!(
    }
 );

+pub struct MessageFlagsValue(pub u32);
+
+impl MessageFlagsValue {
+    pub fn has(&self, flag: MessageFlags) -> bool {
+        self.has_value(flag as u32)
+    }
+    pub fn has_value(&self, bit: u32) -> bool {
+        let mask = 1 << bit;
+        self.0 & mask == mask
+    }
+
+    pub fn set(&mut self, flag: MessageFlags, toggle: bool) -> &mut Self {
+        self.set_value(flag as u32, toggle)
+    }
+    pub fn set_value(&mut self, bit: u32, toggle: bool) -> &mut Self {
+        if toggle {
+            self.0 |= 1 << bit;
+        } else {
+            self.0 &= !(1 << bit);
+        }
+        self
+    }
+}
+
 #[allow(clippy::derivable_impls)]
 impl Default for Message {
    fn default() -> Self {
@@ -214,6 +245,7 @@ impl Default for Message {
            edited: None,
            embeds: None,
            mentions: None,
+            role_mentions: None,
            replies: None,
            reactions: Default::default(),
            interactions: Default::default(),
@@ -262,13 +294,43 @@ impl Message {
            return Err(create_error!(EmptyMessage));
        }

-        // Ensure flags are either not set or have permissible values
-        if let Some(flags) = &data.flags {
-            if flags != &0 && flags != &1 {
+        let allow_mass_mentions = allow_mentions && config.features.mass_mentions_enabled;
+
+        let mut mentions_everyone = false;
+        let mut mentions_online = false;
+        let mut suppress_notifications = false;
+
+        if let Some(raw_flags) = &data.flags {
+            if raw_flags > &7 {
+                // quick path to failure: bigger than all the bits combined
                return Err(create_error!(InvalidProperty));
            }
+
+            // First step of mass mention resolution
+            let flags = MessageFlagsValue(*raw_flags);
+            suppress_notifications = flags.has(MessageFlags::SuppressNotifications);
+            mentions_everyone = allow_mentions && flags.has(MessageFlags::MentionsEveryone);
+            mentions_online = allow_mentions && flags.has(MessageFlags::MentionsOnline);
+
+            // Not a bot, and attempting to set mention flags
+            if user.as_ref().is_some_and(|u| u.bot.as_ref().is_none())
+                && (mentions_everyone || mentions_online)
+            {
+                return Err(create_error!(IsNotBot));
+            }
+
+            if mentions_everyone && mentions_online {
+                return Err(create_error!(InvalidFlagValue));
+            }
        }

+        let server_id = match channel {
+            Channel::TextChannel { ref server, .. } | Channel::VoiceChannel { ref server, .. } => {
+                Some(server.clone())
+            }
+            _ => None,
+        };
+
        // Ensure restrict_reactions is not specified without reactions list
        if let Some(interactions) = &data.interactions {
            if interactions.restrict_reactions {
@@ -302,19 +364,99 @@ impl Message {
                .unwrap_or_default(),
            author: author_id,
            webhook: webhook.map(|w| w.into()),
-            flags: data.flags.map(|v| v as i32),
+            flags: data.flags,
            ..Default::default()
        };

        // Parse mentions in message.
        let mut mentions = HashSet::new();
-        if allow_mentions {
-            if let Some(content) = &data.content {
-                for capture in RE_MENTION.captures_iter(content) {
-                    if let Some(mention) = capture.get(1) {
-                        mentions.insert(mention.as_str().to_string());
+        let mut role_mentions = HashSet::new();
+        let mut role_model_mentions = vec![];
+
+        if let Some(raw_content) = &data.content {
+            // remove all codeblocks before running the mention regex
+            let modified_content = RE_BLOCKS.replace_all(raw_content, "");
+
+            for capture in RE_MENTION.captures_iter(&modified_content).flatten() {
+                if let Some(mention) = capture.get(1) {
+                    mentions.insert(mention.as_str().to_string());
+                }
+            }
+            if allow_mass_mentions && server_id.is_some() {
+                // Second step of mass mention resolution.
+                for capture in RE_ROLE_MENTION.captures_iter(&modified_content).flatten() {
+                    let raw = capture.get(0).expect("No capture?").as_str();
+                    if raw == "@everyone" {
+                        mentions_everyone = true;
+                        // Can't break early here since we want to capture any role ids
+                    } else if raw == "@online" {
+                        mentions_online = true;
+                    } else if let Some(role_match) = capture.get(1) {
+                        role_mentions.insert(role_match.as_str().to_string());
                    }
                }
+
+                if !role_mentions.is_empty() {
+                    let server_data = db
+                        .fetch_server(server_id.unwrap().as_str())
+                        .await
+                        .expect("Failed to fetch server");
+
+                    role_mentions = role_mentions
+                        .iter()
+                        .filter(|role_id| server_data.roles.contains_key(*role_id))
+                        .cloned()
+                        .collect();
+
+                    role_model_mentions.extend(
+                        role_mentions
+                            .iter()
+                            .map(|role_id| server_data.roles.get(role_id).to_owned()),
+                    );
+                }
+            }
+        }
+
+        // Validate the user can perform a mass mention
+        if !config.features.mass_mentions_enabled
+            && (mentions_everyone || mentions_online || !role_mentions.is_empty())
+        {
+            mentions_everyone = false;
+            mentions_online = false;
+            role_mentions.clear();
+        } else if mentions_everyone || mentions_online || !role_mentions.is_empty() {
+            debug!(
+                "Mentioned everyone: {}, mentioned online: {}, mentioned roles: {:?}",
+                mentions_everyone, mentions_online, &role_mentions
+            );
+            if let Some(user) = match author {
+                MessageAuthor::User(user) => Some(Ok(user)),
+                MessageAuthor::System { .. } => Some(Err(())), // DISALLOWED
+                MessageAuthor::Webhook(..) => None,            // Bypass check
+            } {
+                if user.is_err() {
+                    return Err(create_error!(InvalidProperty));
+                }
+                let owned_user: User = user.unwrap().to_owned().into();
+
+                let mut query = DatabasePermissionQuery::new(db, &owned_user).channel(&channel);
+                let perms = calculate_channel_permissions(&mut query).await;
+
+                if (mentions_everyone || mentions_online)
+                    && !perms.has_channel_permission(ChannelPermission::MentionEveryone)
+                {
+                    return Err(create_error!(MissingPermission {
+                        permission: ChannelPermission::MentionEveryone.to_string()
+                    }));
+                }
+
+                if !role_mentions.is_empty()
+                    && !perms.has_channel_permission(ChannelPermission::MentionRoles)
+                {
+                    return Err(create_error!(MissingPermission {
+                        permission: ChannelPermission::MentionRoles.to_string()
+                    }));
+                }
            }
        }

@@ -363,6 +505,7 @@ impl Message {
                    let recipients_hash: HashSet<&String, RandomState> =
                        HashSet::from_iter(recipients);
                    mentions.retain(|m| recipients_hash.contains(m));
+                    role_mentions.clear();
                }
                Channel::TextChannel { ref server, .. }
                | Channel::VoiceChannel { ref server, .. } => {
@@ -403,12 +546,27 @@ impl Message {
            message.mentions.replace(mentions.into_iter().collect());
        }

+        if !role_mentions.is_empty() {
+            message
+                .role_mentions
+                .replace(role_mentions.into_iter().collect());
+        }
+
        if !replies.is_empty() {
            message
                .replies
                .replace(replies.into_iter().collect::<Vec<String>>());
        }

+        // Calculate final message flags
+        let mut flag_value = MessageFlagsValue(0);
+        flag_value
+            .set(MessageFlags::SuppressNotifications, suppress_notifications)
+            .set(MessageFlags::MentionsEveryone, mentions_everyone)
+            .set(MessageFlags::MentionsOnline, mentions_online);
+
+        message.flags = Some(flag_value.0);
+
        // Add attachments to message.
        let mut attachments = vec![];
        if data
@@ -487,7 +645,12 @@ impl Message {
                tasks::ack::queue_message(
                    self.channel.to_string(),
                    AckEvent::ProcessMessage {
-                        messages: vec![(None, self.clone(), mentions.clone(), true)],
+                        messages: vec![(
+                            None,
+                            self.clone(),
+                            mentions.clone(),
+                            self.has_suppressed_notifications(),
+                        )],
                    },
                )
                .await;
@@ -531,7 +694,9 @@ impl Message {
        )
        .await?;

-        if !self.has_suppressed_notifications() {
+        if !self.has_suppressed_notifications()
+            && (self.mentions.is_some() || self.contains_mass_push_mention())
+        {
            // send Push notifications
            tasks::ack::queue_message(
                self.channel.to_string(),
@@ -554,7 +719,7 @@ impl Message {
                            }
                            _ => vec![],
                        },
-                        self.has_suppressed_notifications(),
+                        false, // branch already dictates this
                    )],
                },
            )
@@ -591,13 +756,24 @@ impl Message {
    /// Whether this message has suppressed notifications
    pub fn has_suppressed_notifications(&self) -> bool {
        if let Some(flags) = self.flags {
-            flags & MessageFlags::SuppressNotifications as i32
-                == MessageFlags::SuppressNotifications as i32
+            flags & MessageFlags::SuppressNotifications as u32
+                == MessageFlags::SuppressNotifications as u32
        } else {
            false
        }
    }

+    pub fn contains_mass_push_mention(&self) -> bool {
+        let ping = if let Some(flags) = self.flags {
+            let flags = MessageFlagsValue(flags);
+            flags.has(MessageFlags::MentionsEveryone)
+        } else {
+            false
+        };
+
+        ping || self.role_mentions.is_some()
+    }
+
    /// Update message data
    pub async fn update(
        &mut self,
--- a/crates/core/database/src/models/messages/ops/mongodb.rs
+++ b/crates/core/database/src/models/messages/ops/mongodb.rs
@@ -228,7 +228,6 @@ impl AbstractMessages for MongoDb {
                    "_id": id
                },
                query,
-                None,
            )
            .await
            .map(|_| ())
@@ -247,7 +246,6 @@ impl AbstractMessages for MongoDb {
                        format!("reactions.{emoji}"): user
                    }
                },
-                None,
            )
            .await
            .map(|_| ())
@@ -266,7 +264,6 @@ impl AbstractMessages for MongoDb {
                        format!("reactions.{emoji}"): user
                    }
                },
-                None,
            )
            .await
            .map(|_| ())
@@ -285,7 +282,6 @@ impl AbstractMessages for MongoDb {
                        format!("reactions.{emoji}"): 1
                    }
                },
-                None,
            )
            .await
            .map(|_| ())
@@ -300,15 +296,12 @@ impl AbstractMessages for MongoDb {
    /// Delete messages from a channel by their ids and corresponding channel id
    async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()> {
        self.col::<Document>(COL)
-            .delete_many(
-                doc! {
-                    "channel": channel,
-                    "_id": {
-                        "$in": ids
-                    }
-                },
-                None,
-            )
+            .delete_many(doc! {
+                "channel": channel,
+                "_id": {
+                    "$in": ids
+                }
+            })
            .await
            .map(|_| ())
            .map_err(|_| create_database_error!("delete_many", COL))
@@ -362,7 +355,6 @@ impl MongoDb {
                            "deleted": true
                        }
                    },
-                    None,
                )
                .await
                .map_err(|_| create_database_error!("update_many", "attachments"))?;
@@ -370,7 +362,7 @@ impl MongoDb {

        // And then delete said messages.
        self.col::<Document>(COL)
-            .delete_many(projection, None)
+            .delete_many(projection)
            .await
            .map(|_| ())
            .map_err(|_| create_database_error!("delete_many", COL))