deduplicate message sending

crates/delta/src/routes/channels/message_send.rs

@@ -1,8 +1,6 @@
-use std::collections::HashSet;
-
 use revolt_quark::{
     models::{
-        message::{Interactions, Masquerade, Reply, SendableEmbed},
+        message::DataMessageSend,
         Message, User,
     },
     perms,
@@ -10,45 +8,9 @@ use revolt_quark::{
     Db, Error, Permission, Ref, Result, types::push::MessageAuthor,
 };
 
-use regex::Regex;
 use rocket::serde::json::Json;
-use serde::{Deserialize, Serialize};
-use ulid::Ulid;
 use validator::Validate;
 
-#[derive(Validate, Serialize, Deserialize, JsonSchema)]
-pub struct DataMessageSend {
-    /// Unique token to prevent duplicate message sending
-    ///
-    /// **This is deprecated and replaced by `Idempotency-Key`!**
-    #[validate(length(min = 1, max = 64))]
-    pub nonce: Option<String>,
-
-    /// Message content to send
-    #[validate(length(min = 0, max = 2000))]
-    pub content: Option<String>,
-    /// Attachments to include in message
-    #[validate(length(min = 1, max = 128))]
-    pub attachments: Option<Vec<String>>,
-    /// Messages to reply to
-    pub replies: Option<Vec<Reply>>,
-    /// Embeds to include in message
-    ///
-    /// Text embed content contributes to the content length cap
-    #[validate(length(min = 1, max = 10))]
-    pub embeds: Option<Vec<SendableEmbed>>,
-    /// Masquerade to apply to this message
-    #[validate]
-    pub masquerade: Option<Masquerade>,
-    /// Information about how this message should be interacted with
-    pub interactions: Option<Interactions>,
-}
-
-lazy_static! {
-    // ignoring I L O and U is intentional
-    pub static ref RE_MENTION: Regex = Regex::new(r"<@([0-9A-HJKMNP-TV-Z]{26})>").unwrap();
-}
-
 /// # Send Message
 ///
 /// Sends a message to the given channel.
@@ -59,71 +21,22 @@ pub async fn message_send(
     user: User,
     target: Ref,
     data: Json<DataMessageSend>,
-    mut idempotency: IdempotencyKey,
+    idempotency: IdempotencyKey,
 ) -> Result<Json<Message>> {
     let data = data.into_inner();
     data.validate()
         .map_err(|error| Error::FailedValidation { error })?;
 
-    // Validate Message is within reasonable length limits
-    Message::validate_sum(&data.content, &data.embeds)?;
-
-    // Ensure the request is unique
-    idempotency.consume_nonce(data.nonce).await?;
-
     // Ensure we have permissions to send a message
     let channel = target.as_channel(db).await?;
+
     let mut permissions = perms(&user).channel(&channel);
     permissions
         .throw_permission_and_view_channel(db, Permission::SendMessage)
         .await?;
 
-    // Check the message is not empty
+    // Verify permissions for masquerade.
-    if (data.content.as_ref().map_or(true, |v| v.is_empty()))
+    if let Some(masq) = &data.masquerade {
-        && (data.attachments.as_ref().map_or(true, |v| v.is_empty()))
-        && (data.embeds.as_ref().map_or(true, |v| v.is_empty()))
-    {
-        return Err(Error::EmptyMessage);
-    }
-
-    // Ensure restrict_reactions is not specified without reactions list
-    if let Some(interactions) = &data.interactions {
-        if interactions.restrict_reactions {
-            let disallowed = if let Some(list) = &interactions.reactions {
-                list.len() == 0
-            } else {
-                true
-            };
-
-            if disallowed {
-                return Err(Error::InvalidProperty);
-            }
-        }
-    }
-
-    // Start constructing the message
-    let message_id = Ulid::new().to_string();
-    let mut message = Message {
-        id: message_id.clone(),
-        channel: channel.id().to_string(),
-        author: Some(user.id.clone()),
-        masquerade: data.masquerade,
-        interactions: data.interactions.unwrap_or_default(),
-        ..Default::default()
-    };
-
-    // 1. Parse mentions in message.
-    let mut mentions = HashSet::new();
-    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());
-            }
-        }
-    }
-
-    // 2. Verify permissions for masquerade.
-    if let Some(masq) = &message.masquerade {
         permissions
             .throw_permission(db, Permission::Masquerade)
             .await?;
@@ -135,97 +48,8 @@ pub async fn message_send(
         }
     }
 
-    // 3. Ensure interactions information is correct
+    // Create the message
-    message.interactions.validate(db, &mut permissions).await?;
+    let message = channel.send_message(db, data, MessageAuthor::User(&user), idempotency).await?;
-
-    // 4. Verify replies are valid.
-    let mut replies = HashSet::new();
-    if let Some(entries) = data.replies {
-        if entries.len() > 5 {
-            return Err(Error::TooManyReplies);
-        }
-
-        for Reply { id, mention } in entries {
-            let message = Ref::from_unchecked(id).as_message(db).await?;
-
-            if mention {
-                mentions.insert(message.author_id().to_owned());
-            }
-
-            replies.insert(message.id);
-        }
-    }
-
-    if !mentions.is_empty() {
-        message.mentions.replace(
-            mentions
-                .into_iter()
-                .filter(|id| !user.has_blocked(id))
-                .collect::<Vec<String>>(),
-        );
-    }
-
-    if !replies.is_empty() {
-        message
-            .replies
-            .replace(replies.into_iter().collect::<Vec<String>>());
-    }
-
-    // 5. Process included embeds.
-    let mut embeds = vec![];
-    if let Some(sendable_embeds) = data.embeds {
-        for sendable_embed in sendable_embeds {
-            embeds.push(sendable_embed.into_embed(db, message_id.clone()).await?)
-        }
-    }
-
-    if !embeds.is_empty() {
-        message.embeds.replace(embeds);
-    }
-
-    // 6. Add attachments to message.
-    let mut attachments = vec![];
-    if let Some(ids) = &data.attachments {
-        if !ids.is_empty() {
-            permissions
-                .throw_permission(db, Permission::UploadFiles)
-                .await?;
-        }
-
-        // ! FIXME: move this to app config
-        if ids.len() > 5 {
-            return Err(Error::TooManyAttachments);
-        }
-
-        for attachment_id in ids {
-            attachments.push(
-                db.find_and_use_attachment(attachment_id, "attachments", "message", &message_id)
-                    .await?,
-            );
-        }
-    }
-
-    if !attachments.is_empty() {
-        message.attachments.replace(attachments);
-    }
-
-    // 7. Set content
-    message.content = data.content;
-
-    // 8. Pass-through nonce value for clients
-    message.nonce = Some(idempotency.into_key());
-
-    message.create(db, &channel, Some(MessageAuthor::User(&user))).await?;
-
-    // Queue up a task for processing embeds
-    if let Some(content) = &message.content {
-        revolt_quark::tasks::process_embeds::queue(
-            channel.id().to_string(),
-            message.id.to_string(),
-            content.clone(),
-        )
-        .await;
-    }
 
     Ok(Json(message))
 }

crates/delta/src/routes/webhooks/webhook_execute.rs

@@ -1,9 +1,6 @@
-use std::collections::HashSet;
+use revolt_quark::{Db, Ref, Result, Error, models::message::{Message, DataMessageSend}, web::idempotency::IdempotencyKey, types::push::MessageAuthor};
-
-use revolt_quark::{Db, Ref, Result, Error, models::{Message, message::Reply}, web::idempotency::IdempotencyKey, types::push::MessageAuthor};
 use rocket::serde::json::Json;
-use ulid::Ulid;
+
-use crate::routes::channels::message_send::{DataMessageSend, RE_MENTION};
 use validator::Validate;
 
 /// # Executes a webhook
@@ -11,7 +8,7 @@ use validator::Validate;
 /// executes a webhook and sends a message
 #[openapi(tag = "Webhooks")]
 #[post("/<target>/<token>", data="<data>")]
-pub async fn req(db: &Db, target: Ref, token: String, data: Json<DataMessageSend>, mut idempotency: IdempotencyKey) -> Result<Json<Message>> {
+pub async fn req(db: &Db, target: Ref, token: String, data: Json<DataMessageSend>, idempotency: IdempotencyKey) -> Result<Json<Message>> {
     let data = data.into_inner();
     data.validate()
         .map_err(|error| Error::FailedValidation { error })?;
@@ -22,117 +19,8 @@ pub async fn req(db: &Db, target: Ref, token: String, data: Json<DataMessageSend
         .then_some(())
         .ok_or(Error::InvalidCredentials)?;
 
-    Message::validate_sum(&data.content, &data.embeds)?;
-
-    idempotency.consume_nonce(data.nonce).await?;
-
     let channel = Ref::from_unchecked(webhook.channel.clone()).as_channel(db).await?;
-
+    let message = channel.send_message(db, data, MessageAuthor::Webhook(&webhook), idempotency).await?;
-    if (data.content.as_ref().map_or(true, |v| v.is_empty()))
-        && (data.attachments.as_ref().map_or(true, |v| v.is_empty()))
-        && (data.embeds.as_ref().map_or(true, |v| v.is_empty()))
-    {
-        return Err(Error::EmptyMessage);
-    }
-
-    let message_id = Ulid::new().to_string();
-    let mut message = Message {
-        id: message_id.clone(),
-        channel: channel.id().to_string(),
-        webhook: Some(webhook.id.clone()),
-        masquerade: data.masquerade,
-        interactions: data.interactions.unwrap_or_default(),
-        ..Default::default()
-    };
-
-    // 1. Parse mentions in message.
-    let mut mentions = HashSet::new();
-    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());
-            }
-        }
-    }
-
-    // 4. Verify replies are valid.
-    let mut replies = HashSet::new();
-    if let Some(entries) = data.replies {
-        if entries.len() > 5 {
-            return Err(Error::TooManyReplies);
-        }
-
-        for Reply { id, mention } in entries {
-            let message = Ref::from_unchecked(id).as_message(db).await?;
-
-            if mention {
-                mentions.insert(message.author_id().to_owned());
-            }
-
-            replies.insert(message.id);
-        }
-    }
-
-    if !mentions.is_empty() {
-        message.mentions.replace(mentions.into_iter().collect());
-    }
-
-    if !replies.is_empty() {
-        message
-            .replies
-            .replace(replies.into_iter().collect::<Vec<String>>());
-    }
-
-    // 5. Process included embeds.
-    let mut embeds = vec![];
-    if let Some(sendable_embeds) = data.embeds {
-        for sendable_embed in sendable_embeds {
-            embeds.push(sendable_embed.into_embed(db, message_id.clone()).await?)
-        }
-    }
-
-    if !embeds.is_empty() {
-        message.embeds.replace(embeds);
-    }
-
-    // 6. Add attachments to message.
-    let mut attachments = vec![];
-    if let Some(ids) = &data.attachments {
-        // ! FIXME: move this to app config
-        if ids.len() > 5 {
-            return Err(Error::TooManyAttachments);
-        }
-
-        for attachment_id in ids {
-            attachments.push(
-                db.find_and_use_attachment(attachment_id, "attachments", "message", &message_id)
-                    .await?,
-            );
-        }
-    }
-
-    if !attachments.is_empty() {
-        message.attachments.replace(attachments);
-    }
-
-    // 7. Set content
-    message.content = data.content;
-
-    // 8. Pass-through nonce value for clients
-    message.nonce = Some(idempotency.into_key());
-
-    message.create(db, &channel, Some(MessageAuthor::Webhook(&webhook))).await?;
-
-    // Queue up a task for processing embeds
-    if let Some(content) = &message.content {
-        revolt_quark::tasks::process_embeds::queue(
-            channel.id().to_string(),
-            message.id.to_string(),
-            content.clone(),
-        )
-        .await;
-    }
 
     Ok(Json(message))
-
 }

crates/quark/src/impl/generic/channels/channel.rs

@@ -1,12 +1,16 @@
+use std::collections::HashSet;
+
+use ulid::Ulid;
+
 use crate::{
     events::client::EventV1,
     models::{
         channel::{FieldsChannel, PartialChannel},
-        message::SystemMessage,
+        message::{SystemMessage, Message, DataMessageSend, Reply, RE_MENTION},
         Channel,
     },
-    tasks::ack::AckEvent,
+    tasks::{ack::AckEvent, process_embeds},
-    Database, Error, OverrideField, Result,
+    Database, Error, OverrideField, Result, types::push::MessageAuthor, web::idempotency::IdempotencyKey, Ref,
 };
 
 impl Channel {
@@ -394,4 +398,139 @@ impl Channel {
             _ => Err(Error::InvalidOperation),
         }
     }
+
+    /// Creates a message in a channel
+    pub async fn send_message(&self, db: &Database, data: DataMessageSend, author: MessageAuthor<'_>, mut idempotency: IdempotencyKey) -> Result<Message> {
+        Message::validate_sum(&data.content, &data.embeds)?;
+
+        idempotency.consume_nonce(data.nonce).await?;
+
+        // Check the message is not empty
+        if (data.content.as_ref().map_or(true, |v| v.is_empty()))
+            && (data.attachments.as_ref().map_or(true, |v| v.is_empty()))
+            && (data.embeds.as_ref().map_or(true, |v| v.is_empty()))
+        {
+            return Err(Error::EmptyMessage);
+        }
+
+        // Ensure restrict_reactions is not specified without reactions list
+        if let Some(interactions) = &data.interactions {
+            if interactions.restrict_reactions {
+                let disallowed = if let Some(list) = &interactions.reactions {
+                    list.is_empty()
+                } else {
+                    true
+                };
+
+                if disallowed {
+                    return Err(Error::InvalidProperty);
+                }
+            }
+        }
+
+        // Start constructing the message
+        let message_id = Ulid::new().to_string();
+        let mut message = Message {
+            id: message_id.clone(),
+            channel: self.id().to_string(),
+            masquerade: data.masquerade,
+            interactions: data.interactions.unwrap_or_default(),
+            ..Default::default()
+        };
+
+        match &author {
+            MessageAuthor::User(user) => message.author = Some(user.id.clone()),
+            MessageAuthor::Webhook(webhook) => message.webhook = Some(webhook.id.clone()),
+        }
+
+        // Parse mentions in message.
+        let mut mentions = HashSet::new();
+        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());
+                }
+            }
+        }
+
+        // Verify replies are valid.
+        let mut replies = HashSet::new();
+        if let Some(entries) = data.replies {
+            if entries.len() > 5 {
+                return Err(Error::TooManyReplies);
+            }
+
+            for Reply { id, mention } in entries {
+                let message = Ref::from_unchecked(id).as_message(db).await?;
+
+                if mention {
+                    mentions.insert(message.author_id().to_owned());
+                }
+
+                replies.insert(message.id);
+            }
+        }
+
+        if !mentions.is_empty() {
+            message.mentions.replace(mentions.into_iter().collect());
+        }
+
+        if !replies.is_empty() {
+            message
+                .replies
+                .replace(replies.into_iter().collect::<Vec<String>>());
+        }
+
+        // Process included embeds.
+        let mut embeds = vec![];
+        if let Some(sendable_embeds) = data.embeds {
+            for sendable_embed in sendable_embeds {
+                embeds.push(sendable_embed.into_embed(db, message_id.clone()).await?)
+            }
+        }
+
+        if !embeds.is_empty() {
+            message.embeds.replace(embeds);
+        }
+
+        // Add attachments to message.
+        let mut attachments = vec![];
+        if let Some(ids) = &data.attachments {
+            // ! FIXME: move this to app config
+            if ids.len() > 5 {
+                return Err(Error::TooManyAttachments);
+            }
+
+            for attachment_id in ids {
+                attachments.push(
+                    db.find_and_use_attachment(attachment_id, "attachments", "message", &message_id)
+                        .await?,
+                );
+            }
+        }
+
+        if !attachments.is_empty() {
+            message.attachments.replace(attachments);
+        }
+
+        // Set content
+        message.content = data.content;
+
+        // Pass-through nonce value for clients
+        message.nonce = Some(idempotency.into_key());
+
+        message.create(db, self, Some(author)).await?;
+
+        // Queue up a task for processing embeds
+        if let Some(content) = &message.content {
+            process_embeds::queue(
+                self.id().to_string(),
+                message.id.to_string(),
+                content.clone(),
+            )
+            .await;
+        }
+
+        Ok(message)
+    }
 }

crates/quark/src/models/channels/message.rs

@@ -2,6 +2,7 @@ use crate::util::regex::RE_COLOUR;
 
 use indexmap::{IndexMap, IndexSet};
 use iso8601_timestamp::Timestamp;
+use regex::Regex;
 use serde::{Deserialize, Serialize};
 use validator::Validate;
 
@@ -205,3 +206,36 @@ pub struct AppendMessage {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub embeds: Option<Vec<Embed>>,
 }
+
+#[derive(Validate, Serialize, Deserialize, JsonSchema)]
+pub struct DataMessageSend {
+    /// Unique token to prevent duplicate message sending
+    ///
+    /// **This is deprecated and replaced by `Idempotency-Key`!**
+    #[validate(length(min = 1, max = 64))]
+    pub nonce: Option<String>,
+
+    /// Message content to send
+    #[validate(length(min = 0, max = 2000))]
+    pub content: Option<String>,
+    /// Attachments to include in message
+    #[validate(length(min = 1, max = 128))]
+    pub attachments: Option<Vec<String>>,
+    /// Messages to reply to
+    pub replies: Option<Vec<Reply>>,
+    /// Embeds to include in message
+    ///
+    /// Text embed content contributes to the content length cap
+    #[validate(length(min = 1, max = 10))]
+    pub embeds: Option<Vec<SendableEmbed>>,
+    /// Masquerade to apply to this message
+    #[validate]
+    pub masquerade: Option<Masquerade>,
+    /// Information about how this message should be interacted with
+    pub interactions: Option<Interactions>,
+}
+
+lazy_static! {
+    // ignoring I L O and U is intentional
+    pub static ref RE_MENTION: Regex = Regex::new(r"<@([0-9A-HJKMNP-TV-Z]{26})>").unwrap();
+}