feat: search for links and author type

Signed-off-by: Zomatree <me@zomatree.live>

Cargo.lock

@@ -4710,6 +4710,15 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "linkify"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1dfa36d52c581e9ec783a7ce2a5e0143da6237be5811a0b3153fedfdbe9f780"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.11.0"
@@ -7270,12 +7279,13 @@ dependencies = [
 
 [[package]]
 name = "revolt-search"
-version = "0.11.5"
+version = "0.12.0"
 dependencies = [
  "elasticsearch",
  "elasticsearch-dsl",
  "futures",
  "iso8601-timestamp",
+ "linkify 0.10.0",
  "revolt-config",
  "revolt-database",
  "revolt-models",
@@ -7286,7 +7296,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-searchd"
-version = "0.11.5"
+version = "0.12.0"
 dependencies = [
  "amqprs",
  "async-trait",

crates/core/database/src/amqp/amqp.rs

@@ -326,10 +326,14 @@ impl AMQP {
             .await
     }
 
-    pub async fn new_message_search(&self, message: Message) -> Result<(), AMQPError> {
+    pub async fn new_message_search(
+        &self,
+        message: Message,
+        user: Option<User>,
+    ) -> Result<(), AMQPError> {
         let config = revolt_config::config().await;
 
-        let payload = to_string(&message).unwrap();
+        let payload = to_string(&MessageCreatePayload { message, user }).unwrap();
 
         debug!(
             "Sending new message search payload on channel {}: {}",
@@ -351,10 +355,14 @@ impl AMQP {
             .await
     }
 
-    pub async fn edit_message_search(&self, message: Message) -> Result<(), AMQPError> {
+    pub async fn edit_message_search(
+        &self,
+        message: Message,
+        user: Option<User>,
+    ) -> Result<(), AMQPError> {
         let config = revolt_config::config().await;
 
-        let payload = to_string(&message).unwrap();
+        let payload = to_string(&MessageEditPayload { message, user }).unwrap();
 
         debug!(
             "Sending edit message search payload on channel {}: {}",

crates/core/database/src/events/rabbit.rs

@@ -3,7 +3,7 @@ use std::collections::HashMap;
 use revolt_models::v0::PushNotification;
 use serde::{Deserialize, Serialize};
 
-use crate::User;
+use crate::{Message, User};
 
 #[derive(Serialize, Deserialize)]
 pub struct MessageSentPayload {
@@ -81,10 +81,22 @@ pub struct AckPayload {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct MessageDeletePayload {
-    pub message_id: String
+    pub message_id: String,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ChannelDeletePayload {
-    pub channel_id: String
+    pub channel_id: String,
-}
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MessageCreatePayload {
+    pub message: Message,
+    pub user: Option<User>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MessageEditPayload {
+    pub message: Message,
+    pub user: Option<User>,
+}

crates/core/database/src/models/messages/model.rs

@@ -1,6 +1,6 @@
 use indexmap::{IndexMap, IndexSet};
 use iso8601_timestamp::Timestamp;
-use revolt_config::{FeaturesLimits, capture_error, config};
+use revolt_config::{capture_error, config, FeaturesLimits};
 use revolt_models::v0::{
     self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort,
     MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text,
@@ -18,7 +18,7 @@ use crate::{
         bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
         permissions::DatabasePermissionQuery,
     },
-    Channel, Database, Emoji, File, User, AMQP,
+    Channel, Database, Emoji, File, Member, User, AMQP,
 };
 
 #[cfg(feature = "tasks")]
@@ -210,6 +210,13 @@ auto_derived!(
     pub enum FieldsMessage {
         Pinned,
     }
+
+    /// Message along with the user for the author
+    pub struct MessageWithUser {
+        #[serde(flatten)]
+        pub message: Message,
+        pub user: Option<User>,
+    }
 );
 
 pub struct MessageFlagsValue(pub u32);
@@ -272,8 +279,8 @@ impl Message {
         channel: Channel,
         data: DataMessageSend,
         author: MessageAuthor<'_>,
-        user: Option<v0::User>,
+        user: Option<User>,
-        member: Option<v0::Member>,
+        member: Option<Member>,
         limits: FeaturesLimits,
         mut idempotency: IdempotencyKey,
         generate_embeds: bool,
@@ -610,10 +617,10 @@ impl Message {
 
     /// Send a message without any notifications
     pub async fn send_without_notifications(
-        &mut self,
+        &self,
         db: &Database,
-        user: Option<v0::User>,
+        user: Option<User>,
-        member: Option<v0::Member>,
+        member: Option<Member>,
         is_dm: bool,
         generate_embeds: bool,
         // This determines if this function should queue the mentions task or if somewhere else will.
@@ -623,9 +630,18 @@ impl Message {
         db.insert_message(self).await?;
 
         // Fan out events
-        EventV1::Message(self.clone().into_model(user, member))
+        EventV1::Message(self.clone().into_model(
-            .p(self.channel.to_string())
+            match user {
-            .await;
+                Some(user) => {
+                    let is_online = revolt_presence::is_online(&user.id).await;
+                    Some(user.into_known_static(is_online).await)
+                }
+                None => None,
+            },
+            member.map(Into::into),
+        ))
+        .p(self.channel.to_string())
+        .await;
 
         // Update last_message_id
         #[cfg(feature = "tasks")]
@@ -673,8 +689,8 @@ impl Message {
         db: &Database,
         amqp: Option<&AMQP>, // this is optional mostly for tests.
         author: MessageAuthor<'_>,
-        user: Option<v0::User>,
+        user: Option<User>,
-        member: Option<v0::Member>,
+        member: Option<Member>,
         channel: &Channel,
         generate_embeds: bool,
     ) -> Result<()> {
@@ -704,7 +720,17 @@ impl Message {
                     messages: vec![(
                         Some(
                             PushNotification::from(
-                                self.clone().into_model(user, member),
+                                self.clone().into_model(
+                                    match user.clone() {
+                                        Some(user) => {
+                                            let is_online =
+                                                revolt_presence::is_online(&user.id).await;
+                                            Some(user.into_known_static(is_online).await)
+                                        }
+                                        None => None,
+                                    },
+                                    member.map(Into::into),
+                                ),
                                 Some(author),
                                 channel.to_owned().into(),
                             )
@@ -727,7 +753,7 @@ impl Message {
         }
 
         if let Some(amqp) = amqp {
-            if let Err(e) = amqp.new_message_search(self.clone()).await {
+            if let Err(e) = amqp.new_message_search(self.clone(), user).await {
                 log::error!("Error pushing message to RabbitMQ: {e}");
                 capture_error(&e);
             }
@@ -785,6 +811,7 @@ impl Message {
     pub async fn update(
         &mut self,
         db: &Database,
+        amqp: Option<&AMQP>,
         partial: PartialMessage,
         remove: Vec<FieldsMessage>,
     ) -> Result<()> {
@@ -806,9 +833,74 @@ impl Message {
         .p(self.channel.clone())
         .await;
 
+        if let Some(amqp) = amqp {
+            if let Err(e) = amqp
+                .edit_message_search(self.clone(), self.fetch_author(db).await)
+                .await
+            {
+                log::error!("Error pushing message to RabbitMQ: {e}");
+                capture_error(&e);
+            }
+        }
+
         Ok(())
     }
 
+    pub async fn fetch_users(
+        db: &Database,
+        messages: &[Message],
+        server_id: Option<&str>,
+    ) -> Result<(Vec<User>, Vec<Member>)> {
+        let user_ids = messages
+            .iter()
+            .flat_map(|m| {
+                let mut users = vec![m.author.clone()];
+                if let Some(system) = &m.system {
+                    match system {
+                        SystemMessage::ChannelDescriptionChanged { by } => users.push(by.clone()),
+                        SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
+                        SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
+                            users.push(from.clone());
+                            users.push(to.clone())
+                        }
+                        SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
+                        SystemMessage::UserAdded { by, id, .. }
+                        | SystemMessage::UserRemove { by, id, .. } => {
+                            users.push(by.clone());
+                            users.push(id.clone());
+                        }
+                        SystemMessage::UserBanned { id, .. }
+                        | SystemMessage::UserKicked { id, .. }
+                        | SystemMessage::UserJoined { id, .. }
+                        | SystemMessage::UserLeft { id, .. } => {
+                            users.push(id.clone());
+                        }
+                        SystemMessage::Text { .. } => {}
+                        SystemMessage::MessagePinned { by, .. } => {
+                            users.push(by.clone());
+                        }
+                        SystemMessage::MessageUnpinned { by, .. } => {
+                            users.push(by.clone());
+                        }
+                        SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
+                    }
+                }
+                users
+            })
+            .collect::<HashSet<String>>()
+            .into_iter()
+            .collect::<Vec<String>>();
+
+        let users = db.fetch_users(&user_ids).await?;
+        let members = if let Some(server_id) = server_id {
+            db.fetch_members(server_id, &user_ids).await?
+        } else {
+            Vec::new()
+        };
+
+        Ok((users, members))
+    }
+
     /// Helper function to fetch many messages with users
     pub async fn fetch_with_users(
         db: &Database,
@@ -817,74 +909,26 @@ impl Message {
         include_users: Option<bool>,
         server_id: Option<&str>,
     ) -> Result<BulkMessageResponse> {
-        let messages: Vec<v0::Message> = db
+        let messages = db.fetch_messages(query).await?;
-            .fetch_messages(query)
-            .await?
-            .into_iter()
-            .map(|msg| msg.into_model(None, None))
-            .collect();
 
         if let Some(true) = include_users {
-            let user_ids = messages
+            let (users, members) = Message::fetch_users(db, &messages, server_id).await?;
-                .iter()
-                .flat_map(|m| {
-                    let mut users = vec![m.author.clone()];
-                    if let Some(system) = &m.system {
-                        match system {
-                            v0::SystemMessage::ChannelDescriptionChanged { by } => {
-                                users.push(by.clone())
-                            }
-                            v0::SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
-                            v0::SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
-                                users.push(from.clone());
-                                users.push(to.clone())
-                            }
-                            v0::SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
-                            v0::SystemMessage::UserAdded { by, id, .. }
-                            | v0::SystemMessage::UserRemove { by, id, .. } => {
-                                users.push(by.clone());
-                                users.push(id.clone());
-                            }
-                            v0::SystemMessage::UserBanned { id, .. }
-                            | v0::SystemMessage::UserKicked { id, .. }
-                            | v0::SystemMessage::UserJoined { id, .. }
-                            | v0::SystemMessage::UserLeft { id, .. } => {
-                                users.push(id.clone());
-                            }
-                            v0::SystemMessage::Text { .. } => {}
-                            v0::SystemMessage::MessagePinned { by, .. } => {
-                                users.push(by.clone());
-                            }
-                            v0::SystemMessage::MessageUnpinned { by, .. } => {
-                                users.push(by.clone());
-                            }
-                            v0::SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
-                        }
-                    }
-                    users
-                })
-                .collect::<HashSet<String>>()
-                .into_iter()
-                .collect::<Vec<String>>();
-            let users = User::fetch_many_ids_as_mutuals(db, perspective, &user_ids).await?;
 
             Ok(BulkMessageResponse::MessagesAndUsers {
-                messages,
+                messages: messages
-                users,
+                    .into_iter()
-                members: if let Some(server_id) = server_id {
+                    .map(|msg| msg.into_model(None, None))
-                    Some(
+                    .collect(),
-                        db.fetch_members(server_id, &user_ids)
+                users: User::into_mutuals(perspective, users).await,
-                            .await?
+                members: Some(members.into_iter().map(Into::into).collect()),
-                            .into_iter()
-                            .map(Into::into)
-                            .collect(),
-                    )
-                } else {
-                    None
-                },
             })
         } else {
-            Ok(BulkMessageResponse::JustMessages(messages))
+            Ok(BulkMessageResponse::JustMessages(
+                messages
+                    .into_iter()
+                    .map(|msg| msg.into_model(None, None))
+                    .collect(),
+            ))
         }
     }
 
@@ -898,7 +942,9 @@ impl Message {
     ) -> Result<()> {
         if let Some(message) = db.append_message(&id, &append).await? {
             if let Some(amqp) = amqp {
-                if let Err(e) = amqp.edit_message_search(message).await {
+                let author = message.fetch_author(db).await;
+
+                if let Err(e) = amqp.edit_message_search(message, author).await {
                     log::error!("Error pushing message to RabbitMQ: {e}");
                     capture_error(&e);
                 }
@@ -1136,6 +1182,14 @@ impl Message {
             FieldsMessage::Pinned => self.pinned = None,
         }
     }
+
+    pub async fn fetch_author(&self, db: &Database) -> Option<User> {
+        if self.webhook.is_some() {
+            None
+        } else {
+            db.fetch_user(&self.author).await.ok()
+        }
+    }
 }
 
 impl SystemMessage {

crates/core/database/src/models/messages/ops.rs

@@ -1,8 +1,11 @@
+use revolt_result::Result;
 use std::collections::HashMap;
 use std::time::SystemTime;
-use revolt_result::Result;
 
-use crate::{AppendMessage, FieldsMessage, Message, MessageQuery, PartialMessage, util::ChunkedDatabaseGenerator};
+use crate::{
+    util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
+    MessageWithUser, PartialMessage,
+};
 
 #[cfg(feature = "mongodb")]
 mod mongodb;
@@ -23,7 +26,12 @@ pub trait AbstractMessages: Sync + Send {
     async fn fetch_messages_by_id(&self, ids: &[String]) -> Result<Vec<Message>>;
 
     /// Update a given message with new information
-    async fn update_message(&self, id: &str, message: &PartialMessage, remove: Vec<FieldsMessage>) -> Result<()>;
+    async fn update_message(
+        &self,
+        id: &str,
+        message: &PartialMessage,
+        remove: Vec<FieldsMessage>,
+    ) -> Result<()>;
 
     /// Append information to a given message
     async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>>;
@@ -48,7 +56,9 @@ pub trait AbstractMessages: Sync + Send {
         &self,
         channels: &[String],
         author: &str,
-        since: SystemTime
+        since: SystemTime,
     ) -> Result<HashMap<String, Vec<String>>>;
-    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<Message>>;
+
+    /// Fetches all messages along with their author from every message in decending order
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>>;
 }

crates/core/database/src/models/messages/ops/mongodb.rs

@@ -9,8 +9,8 @@ use std::time::SystemTime;
 use ulid::Ulid;
 
 use crate::{
-    AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath, Message, MessageQuery,
+    util::ChunkedDatabaseGenerator, AppendMessage, DocumentId, FieldsMessage, IntoDocumentPath,
-    MessageTimePeriod, MongoDb, PartialMessage, util::ChunkedDatabaseGenerator,
+    Message, MessageQuery, MessageTimePeriod, MessageWithUser, MongoDb, PartialMessage,
 };
 
 use super::AbstractMessages;
@@ -417,7 +417,8 @@ impl AbstractMessages for MongoDb {
         Ok(deleted_messages)
     }
 
-    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<Message>> {
+    /// Fetches all messages along with their author from every message in decending order
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>> {
         let mut session = self
             .start_session()
             .await
@@ -431,13 +432,34 @@ impl AbstractMessages for MongoDb {
 
         let cursor = self
             .col::<Message>(COL)
-            .find(doc! {})
+            .aggregate([
-            .sort(doc ! { "_id": -1i32 })
+                doc! {
+                    "$lookup": {
+                        "from": "users",
+                        "localField": "author",
+                        "foreignField": "_id",
+                        "as": "user"
+                    }
+                },
+                doc! {
+                    "$set": {
+                        "user": {
+                            "$first": "$user"
+                        }
+                    }
+                },
+                doc! {
+                    "$sort": {
+                        "_id": -1
+                    }
+                },
+            ])
+            .with_type::<MessageWithUser>()
             .session(&mut session)
             .batch_size(1000)
             .await
             .inspect_err(|e| log::error!("{e}"))
-            .map_err(|_| create_database_error!("find", COL))?;
+            .map_err(|_| create_database_error!("aggregate", COL))?;
 
         Ok(ChunkedDatabaseGenerator::new_mongo(session, cursor))
     }

crates/core/database/src/models/messages/ops/reference.rs

@@ -1,13 +1,13 @@
-use std::collections::HashMap;
 use futures::future::try_join_all;
 use indexmap::IndexSet;
 use revolt_result::Result;
+use std::collections::HashMap;
 use std::time::SystemTime;
 use ulid::Ulid;
 
 use crate::{
     util::ChunkedDatabaseGenerator, AppendMessage, FieldsMessage, Message, MessageQuery,
-    PartialMessage, ReferenceDb,
+    MessageWithUser, PartialMessage, ReferenceDb,
 };
 
 use super::AbstractMessages;
@@ -64,7 +64,7 @@ impl AbstractMessages for ReferenceDb {
 
                 if let Some(pinned) = query.filter.pinned {
                     if message.pinned.unwrap_or_default() == pinned {
-                        return false
+                        return false;
                     }
                 }
 
@@ -195,7 +195,12 @@ impl AbstractMessages for ReferenceDb {
     }
 
     /// Update a given message with new information
-    async fn update_message(&self, id: &str, message: &PartialMessage, remove: Vec<FieldsMessage>) -> Result<()> {
+    async fn update_message(
+        &self,
+        id: &str,
+        message: &PartialMessage,
+        remove: Vec<FieldsMessage>,
+    ) -> Result<()> {
         let mut messages = self.messages.lock().await;
         if let Some(message_data) = messages.get_mut(id) {
             message_data.apply_options(message.to_owned());
@@ -300,7 +305,7 @@ impl AbstractMessages for ReferenceDb {
         &self,
         channels: &[String],
         author: &str,
-        since: SystemTime
+        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();
@@ -341,22 +346,31 @@ impl AbstractMessages for ReferenceDb {
         }
 
         // Delete the messages
-        self.messages
+        self.messages.lock().await.retain(|id, message| {
-            .lock()
+            let should_keep = !(message.author == author
-            .await
+                && channels.contains(&message.channel)
-            .retain(|id, message| {
+                && id.as_str() >= threshold_ulid.as_str());
-                let should_keep = !(message.author == author
+            should_keep
-                    && channels.contains(&message.channel)
+        });
-                    && id.as_str() >= threshold_ulid.as_str());
-                should_keep
-            });
 
         Ok(deleted_messages)
     }
 
-    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<Message>> {
+    /// Fetches all messages along with their author from every message in decending order
+    async fn fetch_all_messages(&self) -> Result<ChunkedDatabaseGenerator<MessageWithUser>> {
+        let users = self.users.lock().await;
+
         Ok(ChunkedDatabaseGenerator::new_reference(
-            self.messages.lock().await.values().cloned().collect(),
+            self.messages
+                .lock()
+                .await
+                .values()
+                .cloned()
+                .map(|message| MessageWithUser {
+                    user: users.get(&message.author).cloned(),
+                    message,
+                })
+                .collect(),
         ))
     }
 }

crates/core/database/src/models/users/model.rs

@@ -362,6 +362,17 @@ impl User {
         )
     }
 
+    pub async fn into_mutuals(perspective: &User, users: Vec<User>) -> Vec<v0::User> {
+        let online_ids =
+            filter_online(&users.iter().map(|user| user.id.clone()).collect::<Vec<_>>()).await;
+
+        join_all(users.into_iter().map(|user| async {
+            let is_online = online_ids.contains(&user.id);
+            user.into_known(perspective, is_online).await
+        }))
+        .await
+    }
+
     /// Find a free discriminator for a given username
     pub async fn find_discriminator(
         db: &Database,

crates/core/database/src/util/chunked.rs

@@ -1,5 +1,6 @@
 #[cfg(feature = "mongodb")]
 use ::mongodb::{ClientSession, SessionCursor};
+use revolt_result::{Result, ToRevoltError};
 use serde::Deserialize;
 
 #[derive(Debug)]
@@ -12,7 +13,7 @@ pub enum ChunkedDatabaseGenerator<T> {
     },
 
     Reference {
-        offset: i32,
+        offset: usize,
         data: Vec<T>,
     },
 }
@@ -20,33 +21,25 @@ pub enum ChunkedDatabaseGenerator<T> {
 impl<T: for<'d> Deserialize<'d> + Clone> ChunkedDatabaseGenerator<T> {
     #[cfg(feature = "mongodb")]
     pub fn new_mongo(session: ClientSession, cursor: SessionCursor<T>) -> Self {
-        Self::MongoDb {
+        Self::MongoDb { session, cursor }
-            session,
-            cursor,
-        }
     }
 
     pub fn new_reference(data: Vec<T>) -> Self {
-        Self::Reference {
+        Self::Reference { offset: 0, data }
-            offset: 0,
-            data,
-        }
     }
 
-    pub async fn next(&mut self) -> Option<T> {
+    pub async fn next(&mut self) -> Result<Option<T>> {
         match self {
             #[cfg(feature = "mongodb")]
             Self::MongoDb { session, cursor } => {
-                let value = cursor.next(session).await;
+                cursor.next(session).await.transpose().to_internal_error()
-                value.map(|val| val.expect("Failed to fetch the next message"))
             }
             Self::Reference { offset, data } => {
-                if data.len() as i32 >= *offset {
+                if let Some(value) = data.get(*offset) {
-                    None
-                } else {
-                    let resp = &data[*offset as usize];
                     *offset += 1;
-                    Some(resp.clone())
+                    Ok(Some(value.clone()))
+                } else {
+                    Ok(None)
                 }
             }
         }

crates/core/models/src/v0/search.rs

@@ -2,44 +2,68 @@ use iso8601_timestamp::Timestamp;
 use std::collections::HashSet;
 
 auto_derived!(
+    /// Options for searching messages in a server or channel
     pub struct DataChannelMessagesSearch {
+        /// Channel to search in
         pub channel: Option<String>,
+        /// Server to search in
         pub server: Option<String>,
-        pub offset: Option<u64>,
+
-        pub limit: Option<u64>,
+        /// Filter options
         pub filters: Option<DataChannelMessagesSearchFilters>,
+
+        /// What index to start the search at
+        pub offset: Option<u64>,
+        /// Max amount of messages to return
+        pub limit: Option<u64>,
+        /// Sort order
         pub sort: Option<SortOrder>,
-        pub include_users: Option<bool>,
     }
 
+    /// Message search filters
     pub struct DataChannelMessagesSearchFilters {
+        /// Message content
         pub content: Option<String>,
+        /// Specific user
         pub author: Option<HashSet<String>>,
+
+        /// Mentions a user
         pub mentions: Option<HashSet<String>>,
+        /// Mentions a role
         pub role_mentions: Option<HashSet<String>>,
+
+        /// Send before a specific date
         pub before_date: Option<Timestamp>,
+        /// Sent after a specific date
         pub after_date: Option<Timestamp>,
+
+        /// What type of user sent the message
         pub author_type: Option<HashSet<AuthorType>>,
+        /// Whether the message is pinned or not
         pub pinned: Option<bool>,
+        /// Require message to have a specific component type
         pub components: Option<HashSet<MessageComponent>>,
     }
 
+    /// Message author type
     #[derive(Copy, Hash)]
     pub enum AuthorType {
         User,
-        // Bot,
+        Bot,
         Webhook,
     }
 
+    /// Message component
     #[derive(Copy, Hash)]
     pub enum MessageComponent {
         Image,
         Video,
-        // Link,
+        Link,
         File,
         Embed,
     }
 
+    /// Message sort order
     #[derive(Copy, Default)]
     #[cfg_attr(feature = "serde", serde(rename_all = "lowercase"))]
     pub enum SortOrder {

crates/core/search/Cargo.toml

@@ -1,14 +1,14 @@
 [package]
 name = "revolt-search"
-version = "0.11.5"
+version = "0.12.0"
 edition = "2024"
 license = "MIT"
 publish = false
 
 [dependencies]
-revolt-config = { version = "0.11.5", path = "../config" }
+revolt-config = { version = "0.12.0", path = "../config" }
-revolt-database = { version = "0.11.5", path = "../database" }
+revolt-database = { version = "0.12.0", path = "../database" }
-revolt-models = { version = "0.11.5", path = "../models" }
+revolt-models = { version = "0.12.0", path = "../models" }
 elasticsearch = "9.1.0-alpha.1"
 serde_json = "1"
 ulid = "1.2.1"
@@ -16,3 +16,4 @@ iso8601-timestamp = { version = "0.2.10", features = ["serde", "bson"] }
 futures = "0.3.32"
 elasticsearch-dsl = "0.4"
 serde = "1"
+linkify = "0.10.0"

crates/core/search/src/client.rs

@@ -1,19 +1,25 @@
 use std::fmt::Display;
 
 use elasticsearch::{
-    BulkOperation, BulkParts, CreateParts, DeleteByQueryParts, DeleteParts, Elasticsearch, IndexParts, SearchParts, auth::Credentials, http::{
+    BulkOperation, BulkParts, CreateParts, DeleteByQueryParts, DeleteParts, Elasticsearch,
+    IndexParts, SearchParts,
+    auth::Credentials,
+    http::{
         response::Exception,
         transport::{SingleNodeConnectionPool, TransportBuilder},
-    }, indices::{IndicesCreateParts, IndicesDeleteParts}
+    },
+    indices::{IndicesCreateParts, IndicesDeleteParts},
 };
 use elasticsearch_dsl::{FieldSort, Query, Search, SearchResponse, Sort};
-use revolt_database::{Database, Message};
+use linkify::{LinkFinder, LinkKind};
-use serde_json::{Map, Value, json};
+use revolt_database::{Database, Message, MessageWithUser, User};
+use serde_json::{Map, Value, json, to_value};
 
 pub use elasticsearch;
 
-use crate::{MessageComponent, MetadataFile, SearchTerms};
+use crate::{AuthorType, MessageComponent, SearchTerms};
 
+/// Elasticsearch errors
 #[derive(Debug)]
 pub enum Error {
     Http(elasticsearch::Error),
@@ -42,6 +48,7 @@ impl Display for Error {
     }
 }
 
+/// Higher level elasticsearch API more fit for our specific usecase
 #[derive(Debug, Clone)]
 pub struct ElasticsearchClient {
     pub inner: Elasticsearch,
@@ -61,6 +68,7 @@ impl ElasticsearchClient {
         Self { inner }
     }
 
+    /// Delete messages index along with all documents
     pub async fn delete_indexes(&self) -> Result<(), Error> {
         let exception = self
             .inner
@@ -78,6 +86,7 @@ impl ElasticsearchClient {
         }
     }
 
+    /// Create the messages index
     pub async fn setup_indexes(&self) -> Result<(), Error> {
         let exception = self
             .inner
@@ -98,13 +107,14 @@ impl ElasticsearchClient {
                         },
                         "attachments": {
                             "type": "nested",
+                            "dynamic": false,
                             "properties": {
                                 "metadata.type": {
                                     "type": "keyword"
                                 }
                             }
                         },
-                        // TODO: links
+                        "has_link": { "type": "boolean" },
                     }
                 }
             }))
@@ -120,6 +130,7 @@ impl ElasticsearchClient {
         }
     }
 
+    /// Performs a search for messages, returns a vec of message ids
     pub async fn search(&self, terms: SearchTerms) -> Result<Vec<String>, Error> {
         let mut query = Query::bool().filter(Query::terms("channel", terms.channels));
 
@@ -149,17 +160,32 @@ impl ElasticsearchClient {
 
         if let Some(components) = terms.filters.components {
             let mut components_query = Query::bool();
+            let mut attachments_query = Query::bool();
 
             for component in components {
                 match component {
-                    MessageComponent::Image => components_query = components_query.should(Query::term("attachments.metadata.type", "Image")),
+                    MessageComponent::Image => {
-                    MessageComponent::Video => components_query = components_query.should(Query::term("attachments.metadata.type", "Video")),
+                        attachments_query = attachments_query
-                    MessageComponent::File => components_query = components_query.should(Query::exists("attachments.file._id")),
+                            .should(Query::term("attachments.metadata.type", "Image"))
-                    MessageComponent::Embed => query = query.filter(Query::exists("embeds")),
+                    }
+                    MessageComponent::Video => {
+                        attachments_query = attachments_query
+                            .should(Query::term("attachments.metadata.type", "Video"))
+                    }
+                    MessageComponent::Link => {
+                        components_query = components_query.should(Query::exists("has_link"))
+                    }
+                    MessageComponent::File => {
+                        attachments_query = attachments_query.should(Query::exists("attachments"))
+                    }
+                    MessageComponent::Embed => {
+                        components_query = components_query.should(Query::exists("embeds"))
+                    }
                 };
             }
 
-            query = query.filter(Query::nested("attachments", components_query));
+            query = query
+                .filter(components_query.should(Query::nested("attachments", attachments_query)));
         }
 
         let search = Search::new()
@@ -191,7 +217,13 @@ impl ElasticsearchClient {
         }
     }
 
-    fn create_message_source(&self, _db: &Database, message: Message) -> Value {
+    /// Creates a source for a message which can be stored and indexed into elasticsearch
+    fn create_message_source(
+        &self,
+        _db: &Database,
+        message: Message,
+        author: Option<User>,
+    ) -> Value {
         let mut map = Map::new();
 
         map.insert("channel".to_string(), Value::String(message.channel));
@@ -199,25 +231,25 @@ impl ElasticsearchClient {
         map.insert("author".to_string(), Value::String(message.author));
 
         if let Some(content) = message.content {
+            // Is there a better way to handle this? can elasticsearch index links itself?
+            // Maybe in the future store the domains and be able to filter by that as well
+            let mut finder = LinkFinder::new();
+            finder.kinds(&[LinkKind::Url]);
+
+            if finder.links(&content).next().is_some() {
+                map.insert("has_link".to_string(), Value::Bool(true));
+            }
+
             map.insert("content".to_string(), Value::String(content));
         }
 
         if let Some(attachments) = message.attachments {
-            let mut files = Vec::new();
+            // TODO: fetch the file metadata from FileHash because of File.metadata deprecation
-
+            // let metadata = attachment.as_hash(db).await.expect("Failed to fetch FileHash").metadata;
-            for attachment in attachments {
-                // TODO: swap this out for fetching the metadata from FileHash because of deprecation
-                // let metadata = attachment.as_hash(db).await.expect("Failed to fetch FileHash").metadata;
-
-                files.push(MetadataFile {
-                    metadata: attachment.metadata.clone(),
-                    file: attachment,
-                })
-            }
 
             map.insert(
                 "attachments".to_string(),
-                serde_json::to_value(files).unwrap(),
+                serde_json::to_value(attachments).unwrap(),
             );
         }
 
@@ -243,31 +275,38 @@ impl ElasticsearchClient {
             map.insert("pinned".to_string(), Value::Bool(pinned));
         }
 
-        // TODO: bot
+        // This will turn bot author type to user author type if this is ran on a deleted message,
+        // due to the author not existing anymore so fetching will fail, this is probably niche enough
+        // to not really matter, might try fix in the future.
         map.insert(
             "author_type".to_string(),
-            Value::String(
+            to_value(if message.webhook.is_some() {
-                if message.webhook.is_some() {
+                AuthorType::Webhook
-                    "webhook"
+            } else if author.is_some_and(|user| user.bot.is_some()) {
-                } else {
+                AuthorType::Bot
-                    "user"
+            } else {
-                }
+                AuthorType::User
-                .to_string(),
+            })
-            ),
+            .unwrap(),
         );
 
         Value::Object(map)
     }
 
-    pub async fn bulk_index_messages(&self, db: &Database, messages: Vec<Message>) -> Result<(), Error> {
+    /// Bulk uploads and indexes messages to elasticsearch
+    pub async fn bulk_index_messages(
+        &self,
+        db: &Database,
+        messages: Vec<MessageWithUser>,
+    ) -> Result<(), Error> {
         let mut ops = Vec::<BulkOperation<Value>>::new();
 
         for message in messages {
-            let id = message.id.clone();
+            let id = message.message.id.clone();
-            let source = self.create_message_source(db, message);
+            let source = self.create_message_source(db, message.message, message.user);
 
             ops.push(BulkOperation::create(source).id(id).into());
-        };
+        }
 
         let exception = self
             .inner
@@ -285,9 +324,15 @@ impl ElasticsearchClient {
         }
     }
 
-    pub async fn index_message(&self, db: &Database, message: Message) -> Result<(), Error> {
+    /// Uploads and indexes a single message to elasticsearch
+    pub async fn index_message(
+        &self,
+        db: &Database,
+        message: Message,
+        author: Option<User>,
+    ) -> Result<(), Error> {
         let id = message.id.clone();
-        let source = self.create_message_source(db, message);
+        let source = self.create_message_source(db, message, author);
 
         let exception = self
             .inner
@@ -305,9 +350,15 @@ impl ElasticsearchClient {
         }
     }
 
-    pub async fn edit_message(&self, db: &Database, message: Message) -> Result<(), Error> {
+    /// Updates or upserts an existing message to elasticsearch
+    pub async fn edit_message(
+        &self,
+        db: &Database,
+        message: Message,
+        author: Option<User>,
+    ) -> Result<(), Error> {
         let id = message.id.clone();
-        let source = self.create_message_source(db, message);
+        let source = self.create_message_source(db, message, author);
 
         let exception = self
             .inner
@@ -325,6 +376,7 @@ impl ElasticsearchClient {
         }
     }
 
+    /// Deletes a message from elasticsearch
     pub async fn delete_message(&self, message_id: &str) -> Result<(), Error> {
         let exception = self
             .inner
@@ -341,6 +393,7 @@ impl ElasticsearchClient {
         }
     }
 
+    /// Deletes all messages in a channel from elasticsearch
     pub async fn delete_channel(&self, channel_id: &str) -> Result<(), Error> {
         let exception = self
             .inner

crates/core/search/src/types.rs

@@ -1,39 +1,54 @@
 use std::collections::HashSet;
 
 use iso8601_timestamp::Timestamp;
-use revolt_database::{File, Metadata};
 use revolt_models::v0;
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 
+/// Message author type
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Hash)]
 pub enum AuthorType {
     User,
-    // Bot,
+    Bot,
     Webhook,
 }
 
+/// Message component
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
 pub enum MessageComponent {
     Image,
     Video,
-    // Link,
+    Link,
     File,
     Embed,
 }
 
+/// Message search filters
 #[derive(Debug, Clone, Default, PartialEq)]
 pub struct SearchFilters {
+    /// Message content
     pub content: Option<String>,
+    /// Specific user
     pub author: Option<HashSet<String>>,
+
+    /// Mentions a user
     pub mentions: Option<HashSet<String>>,
+    /// Mentions a role
     pub role_mentions: Option<HashSet<String>>,
+
+    /// Send before a specific date
     pub before_date: Option<Timestamp>,
+    /// Sent after a specific date
     pub after_date: Option<Timestamp>,
+
+    /// What type of user sent the message
     pub author_type: Option<HashSet<AuthorType>>,
+    /// Whether the message is pinned or not
     pub pinned: Option<bool>,
+    /// Require message to have a specific component type
     pub components: Option<HashSet<MessageComponent>>,
 }
 
+/// Message sort order
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub enum SortOrder {
     Asc,
@@ -41,26 +56,28 @@ pub enum SortOrder {
     Desc,
 }
 
+/// Options for searching messages in a server or channel
 #[derive(Debug, Clone, PartialEq)]
 pub struct SearchTerms {
+    /// Channels to search in
     pub channels: Vec<String>,
-    pub filters: SearchFilters,
-    pub offset: Option<u64>,
-    pub limit: Option<u64>,
-    pub sort: Option<SortOrder>,
-}
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    /// Filter options
-pub struct MetadataFile {
+    pub filters: SearchFilters,
-    pub file: File,
+
-    pub metadata: Metadata,
+    /// What index to start the search at
+    pub offset: Option<u64>,
+    /// Max amount of messages to return
+    pub limit: Option<u64>,
+    /// Sort order
+    pub sort: Option<SortOrder>,
 }
 
 impl From<v0::AuthorType> for AuthorType {
     fn from(value: v0::AuthorType) -> Self {
         match value {
             v0::AuthorType::User => AuthorType::User,
-            // v0::AuthorType::Bot => AuthorType::Bot,
+            v0::AuthorType::Bot => AuthorType::Bot,
             v0::AuthorType::Webhook => AuthorType::Webhook,
         }
     }
@@ -71,7 +88,7 @@ impl From<v0::MessageComponent> for MessageComponent {
         match value {
             v0::MessageComponent::Image => MessageComponent::Image,
             v0::MessageComponent::Video => MessageComponent::Video,
-            // v0::MessageComponent::Link => MessageComponent::Link,
+            v0::MessageComponent::Link => MessageComponent::Link,
             v0::MessageComponent::File => MessageComponent::File,
             v0::MessageComponent::Embed => MessageComponent::Embed,
         }

crates/daemons/searchd/Cargo.toml

@@ -1,15 +1,15 @@
 [package]
 name = "revolt-searchd"
-version = "0.11.5"
+version = "0.12.0"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 publish = false
 
 [dependencies]
 tokio = { version = "1.50.0", features = ["macros"] }
-revolt-database = { version = "0.11.5", path = "../../core/database" }
+revolt-database = { version = "0.12.0", path = "../../core/database" }
-revolt-config = { version = "0.11.5", path = "../../core/config" }
+revolt-config = { version = "0.12.0", path = "../../core/config" }
-revolt-search = { version = "0.11.5", path = "../../core/search" }
+revolt-search = { version = "0.12.0", path = "../../core/search" }
 amqprs = "1.7.0"
 async-trait = "0.1.89"
 log = "0.4.29"

crates/daemons/searchd/src/consumers/channel_delete.rs

@@ -1,11 +1,16 @@
-use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use amqprs::{
+    BasicProperties, Deliver,
+    channel::{BasicAckArguments, BasicRejectArguments, Channel},
+    consumer::AsyncConsumer,
+};
 use async_trait::async_trait;
 use revolt_database::{Database, events::rabbit::ChannelDeletePayload};
 use revolt_search::ElasticsearchClient;
 
+#[allow(unused)]
 pub struct ChannelDeleteConsumer {
     client: ElasticsearchClient,
-    database: Database
+    database: Database,
 }
 
 impl ChannelDeleteConsumer {
@@ -23,13 +28,25 @@ impl AsyncConsumer for ChannelDeleteConsumer {
         _basic_properties: BasicProperties,
         content: Vec<u8>,
     ) {
-        let payload = serde_json::from_slice::<ChannelDeletePayload>(&content).expect("Failed to decode message");
+        let payload = serde_json::from_slice::<ChannelDeletePayload>(&content)
+            .expect("Failed to decode message");
         log::debug!("Received channel delete {payload:?}");
 
-        if self.client.delete_channel(&payload.channel_id).await.is_ok() {
+        if self
-            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+            .client
+            .delete_channel(&payload.channel_id)
+            .await
+            .is_ok()
+        {
+            channel
+                .basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false))
+                .await
+                .expect("Failed to ack");
         } else {
-            // todo requeue
+            channel
+                .basic_reject(BasicRejectArguments::new(deliver.delivery_tag(), true))
+                .await
+                .expect("Failed to reject");
         }
     }
-}
+}

crates/daemons/searchd/src/consumers/message.rs

@@ -1,11 +1,15 @@
-use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use amqprs::{
+    BasicProperties, Deliver,
+    channel::{BasicAckArguments, BasicRejectArguments, Channel},
+    consumer::AsyncConsumer,
+};
 use async_trait::async_trait;
-use revolt_database::{Database, Message};
+use revolt_database::{Database, events::rabbit::MessageCreatePayload};
 use revolt_search::ElasticsearchClient;
 
 pub struct MessageConsumer {
     client: ElasticsearchClient,
-    database: Database
+    database: Database,
 }
 
 impl MessageConsumer {
@@ -23,13 +27,25 @@ impl AsyncConsumer for MessageConsumer {
         _basic_properties: BasicProperties,
         content: Vec<u8>,
     ) {
-        let message = serde_json::from_slice::<Message>(&content).expect("Failed to decode message");
+        let payload = serde_json::from_slice::<MessageCreatePayload>(&content)
-        log::debug!("Received message {message:?}");
+            .expect("Failed to decode message");
+        log::debug!("Received message {payload:?}");
 
-        if self.client.index_message(&self.database, message).await.is_ok() {
+        if self
-            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+            .client
+            .index_message(&self.database, payload.message, payload.user)
+            .await
+            .is_ok()
+        {
+            channel
+                .basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false))
+                .await
+                .expect("Failed to ack");
         } else {
-            // todo requeue
+            channel
+                .basic_reject(BasicRejectArguments::new(deliver.delivery_tag(), true))
+                .await
+                .expect("Failed to reject");
         }
     }
-}
+}

crates/daemons/searchd/src/consumers/message_delete.rs

@@ -1,11 +1,16 @@
-use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use amqprs::{
+    BasicProperties, Deliver,
+    channel::{BasicAckArguments, BasicRejectArguments, Channel},
+    consumer::AsyncConsumer,
+};
 use async_trait::async_trait;
 use revolt_database::{Database, events::rabbit::MessageDeletePayload};
 use revolt_search::ElasticsearchClient;
 
+#[allow(unused)]
 pub struct MessageDeleteConsumer {
     client: ElasticsearchClient,
-    database: Database
+    database: Database,
 }
 
 impl MessageDeleteConsumer {
@@ -23,13 +28,25 @@ impl AsyncConsumer for MessageDeleteConsumer {
         _basic_properties: BasicProperties,
         content: Vec<u8>,
     ) {
-        let payload = serde_json::from_slice::<MessageDeletePayload>(&content).expect("Failed to decode message");
+        let payload = serde_json::from_slice::<MessageDeletePayload>(&content)
+            .expect("Failed to decode message");
         log::debug!("Received message delete {payload:?}");
 
-        if self.client.delete_message(&payload.message_id).await.is_ok() {
+        if self
-            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+            .client
+            .delete_message(&payload.message_id)
+            .await
+            .is_ok()
+        {
+            channel
+                .basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false))
+                .await
+                .expect("Failed to ack");
         } else {
-            // todo requeue
+            channel
+                .basic_reject(BasicRejectArguments::new(deliver.delivery_tag(), true))
+                .await
+                .expect("Failed to reject");
         }
     }
-}
+}

crates/daemons/searchd/src/consumers/message_edit.rs

@@ -1,11 +1,15 @@
-use amqprs::{BasicProperties, Deliver, channel::{BasicAckArguments, Channel}, consumer::AsyncConsumer};
+use amqprs::{
+    BasicProperties, Deliver,
+    channel::{BasicAckArguments, BasicRejectArguments, Channel},
+    consumer::AsyncConsumer,
+};
 use async_trait::async_trait;
-use revolt_database::{Database, Message};
+use revolt_database::{Database, events::rabbit::MessageEditPayload};
 use revolt_search::ElasticsearchClient;
 
 pub struct MessageEditConsumer {
     client: ElasticsearchClient,
-    database: Database
+    database: Database,
 }
 
 impl MessageEditConsumer {
@@ -23,13 +27,25 @@ impl AsyncConsumer for MessageEditConsumer {
         _basic_properties: BasicProperties,
         content: Vec<u8>,
     ) {
-        let message = serde_json::from_slice::<Message>(&content).expect("Failed to decode message");
+        let payload = serde_json::from_slice::<MessageEditPayload>(&content)
-        log::debug!("Received edit message {message:?}");
+            .expect("Failed to decode message");
+        log::debug!("Received edit message {payload:?}");
 
-        if self.client.edit_message(&self.database, message).await.is_ok() {
+        if self
-            channel.basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false)).await.expect("Failed to ack");
+            .client
+            .edit_message(&self.database, payload.message, payload.user)
+            .await
+            .is_ok()
+        {
+            channel
+                .basic_ack(BasicAckArguments::new(deliver.delivery_tag(), false))
+                .await
+                .expect("Failed to ack");
         } else {
-            // todo requeue
+            channel
+                .basic_reject(BasicRejectArguments::new(deliver.delivery_tag(), true))
+                .await
+                .expect("Failed to reject");
         }
     }
-}
+}

crates/daemons/searchd/src/index.rs

@@ -12,7 +12,7 @@ pub async fn index_existing_messages(db: Database, client: ElasticsearchClient)
 
     let mut chunk = Vec::new();
 
-    while let Some(message) = generator.next().await {
+    while let Some(message) = generator.next().await.expect("Failed to fetch batch") {
         chunk.push(message);
 
         if chunk.len() >= 1000

crates/daemons/searchd/src/main.rs

@@ -1,4 +1,5 @@
 use amqprs::{
+    FieldTable,
     channel::{
         BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments,
         QueueDeclareArguments,
@@ -26,7 +27,10 @@ async fn _main() {
         config.elasticsearch.api_key.clone(),
     );
 
-    if std::env::var("REMAKE_MESSAGES_INDEX").as_deref().is_ok_and(|v| v == "1") {
+    if std::env::var("REMAKE_MESSAGES_INDEX")
+        .as_deref()
+        .is_ok_and(|v| v == "1")
+    {
         if client.delete_indexes().await.is_err() {
             log::info!("Index does not existing, skipping.")
         };
@@ -87,7 +91,10 @@ async fn _main() {
 
     let mut task = None;
 
-    if std::env::var("INDEX_ALL_MESSAGES").as_deref().is_ok_and(|v| v == "1") {
+    if std::env::var("INDEX_ALL_MESSAGES")
+        .as_deref()
+        .is_ok_and(|v| v == "1")
+    {
         task = Some(spawn(index::index_existing_messages(db, client.clone())));
     }
 
@@ -119,8 +126,16 @@ async fn make_queue_and_consume<F: AsyncConsumer + Send + 'static>(
         .await
         .expect("Failed to declare pushd exchange");
 
+    let mut table = FieldTable::new();
+    table.insert("x-queue-type".try_into().unwrap(), "quorum".into());
+
     _ = channel
-        .queue_declare(QueueDeclareArguments::new(queue).durable(true).finish())
+        .queue_declare(
+            QueueDeclareArguments::new(queue)
+                .durable(true)
+                .arguments(table)
+                .finish(),
+        )
         .await
         .unwrap()
         .unwrap();

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

@@ -1,6 +1,6 @@
 use revolt_database::{
     util::{permissions::DatabasePermissionQuery, reference::Reference},
-    Database, PartialMessage, User,
+    Database, PartialMessage, User, AMQP,
 };
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::Result;
@@ -16,6 +16,7 @@ use rocket_empty::EmptyResponse;
 #[delete("/<target>/messages/<msg>/reactions")]
 pub async fn clear_reactions(
     db: &State<Database>,
+    amqp: &State<AMQP>,
     user: User,
     target: Reference<'_>,
     msg: Reference<'_>,
@@ -33,11 +34,12 @@ pub async fn clear_reactions(
     message
         .update(
             db,
+            Some(amqp),
             PartialMessage {
                 reactions: Some(Default::default()),
                 ..Default::default()
             },
-            vec![]
+            vec![],
         )
         .await
         .map(|_| EmptyResponse)

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

@@ -1,7 +1,9 @@
 use iso8601_timestamp::Timestamp;
 use revolt_config::capture_error;
 use revolt_database::{
-    AMQP, Database, Message, PartialMessage, User, tasks, util::{permissions::DatabasePermissionQuery, reference::Reference}
+    tasks,
+    util::{permissions::DatabasePermissionQuery, reference::Reference},
+    Database, Message, PartialMessage, User, AMQP,
 };
 use revolt_models::v0::{self, Embed};
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
@@ -82,7 +84,7 @@ pub async fn edit(
 
     partial.embeds = Some(new_embeds);
 
-    message.update(db, partial, vec![]).await?;
+    message.update(db, Some(amqp), partial, vec![]).await?;
 
     // Queue up a task for processing embeds if the we have sufficient permissions
     if permissions.has_channel_permission(ChannelPermission::SendEmbeds) {
@@ -96,10 +98,5 @@ pub async fn edit(
         }
     }
 
-    if let Err(e) = amqp.edit_message_search(message.clone()).await {
-        log::error!("Error pushing message to RabbitMQ: {e}");
-        capture_error(&e);
-    }
-
     Ok(Json(message.into_model(None, None)))
 }

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

@@ -39,6 +39,7 @@ pub async fn message_pin(
     message
         .update(
             db,
+            Some(amqp),
             PartialMessage {
                 pinned: Some(true),
                 ..Default::default()
@@ -47,7 +48,10 @@ pub async fn message_pin(
         )
         .await?;
 
-    if let Err(e) = amqp.edit_message_search(message.clone()).await {
+    if let Err(e) = amqp
+        .edit_message_search(message.clone(), message.fetch_author(db).await)
+        .await
+    {
         log::error!("Error pushing message to RabbitMQ: {e}");
         capture_error(&e);
     }
@@ -122,8 +126,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.into()),
+            Some(member),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,

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

@@ -126,16 +126,10 @@ pub async fn message_send(
     // Make sure we have server member (edge case if server owner)
     query.are_we_a_member().await;
 
-    // Create model user / members
+    let member = query
-    let model_user = user
-        .clone()
-        .into_known_static(revolt_presence::is_online(&user.id).await)
-        .await;
-
-    let model_member: Option<v0::Member> = query
         .member_ref()
         .as_ref()
-        .map(|member| member.clone().into_owned().into());
+        .map(|member| member.clone().into_owned());
 
     Ok(Json(
         Message::create_from_api(
@@ -144,15 +138,22 @@ pub async fn message_send(
             channel,
             data,
             v0::MessageAuthor::User(&author),
-            Some(model_user.clone()),
+            Some(user.clone()),
-            model_member.clone(),
+            member.clone(),
             user.limits().await,
             idempotency,
             permissions.has_channel_permission(ChannelPermission::SendEmbeds),
             allow_mentions,
         )
         .await?
-        .into_model(Some(model_user), model_member),
+        .into_model(
+            Some(
+                user.clone()
+                    .into_known_static(revolt_presence::is_online(&user.id).await)
+                    .await,
+            ),
+            member.map(Into::into),
+        ),
     ))
 }
 
@@ -262,8 +263,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.clone().into()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,
@@ -302,8 +303,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.clone().into()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -350,8 +351,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.clone().into()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("2".to_string()),
             false,
@@ -396,8 +397,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.clone().into()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -447,8 +448,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.clone().into()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("3".to_string()),
             false,
@@ -484,7 +485,7 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
             Some(member.clone().into()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("4".to_string()),
@@ -515,7 +516,7 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
             Some(member.clone().into()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("4".to_string()),
@@ -570,13 +571,8 @@ mod test {
                     .into(&harness.db, Some(&other_user))
                     .await,
             ),
-            Some(
+            Some(other_user.clone()),
-                other_user
+            Some(other_member.clone()),
-                    .clone()
-                    .into(&harness.db, Some(&other_user))
-                    .await,
-            ),
-            Some(other_member.clone().into()),
             other_user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -615,13 +611,8 @@ mod test {
                     .into(&harness.db, Some(&other_user))
                     .await,
             ),
-            Some(
+            Some(other_user.clone()),
-                other_user
+            Some(other_member.clone()),
-                    .clone()
-                    .into(&harness.db, Some(&other_user))
-                    .await,
-            ),
-            Some(other_member.clone().into()),
             other_user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,
@@ -684,13 +675,8 @@ mod test {
                     .into(&harness.db, Some(&other_user))
                     .await,
             ),
-            Some(
+            Some(other_user.clone()),
-                other_user
+            Some(other_member.clone()),
-                    .clone()
-                    .into(&harness.db, Some(&other_user))
-                    .await,
-            ),
-            Some(other_member.clone().into()),
             other_user.limits().await,
             IdempotencyKey::unchecked_from_string("1".to_string()),
             false,

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

@@ -37,10 +37,18 @@ pub async fn message_unpin(
     }
 
     message
-        .update(db, PartialMessage::default(), vec![FieldsMessage::Pinned])
+        .update(
+            db,
+            Some(amqp),
+            PartialMessage::default(),
+            vec![FieldsMessage::Pinned],
+        )
         .await?;
 
-    if let Err(e) = amqp.edit_message_search(message.clone()).await {
+    if let Err(e) = amqp
+        .edit_message_search(message.clone(), message.fetch_author(db).await)
+        .await
+    {
         log::error!("Error pushing message to RabbitMQ: {e}");
         capture_error(&e);
     }
@@ -120,8 +128,8 @@ mod test {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&harness.db, Some(&user)).await),
-            Some(user.clone().into(&harness.db, Some(&user)).await),
+            Some(user.clone()),
-            Some(member.into()),
+            Some(member),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,

crates/delta/src/routes/search/search_messages.rs

@@ -1,6 +1,4 @@
-use std::collections::HashSet;
+use revolt_database::{util::permissions::DatabasePermissionQuery, Database, Message, User};
-
-use revolt_database::{util::permissions::DatabasePermissionQuery, Database, User};
 use revolt_models::v0;
 use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
 use revolt_result::{create_error, Result, ToRevoltError};
@@ -66,12 +64,7 @@ pub async fn search_messages(
 
     let message_ids = es.search(terms).await.to_internal_error()?;
 
-    let mut messages = db
+    let mut messages = db.fetch_messages_by_id(&message_ids).await?;
-        .fetch_messages_by_id(&message_ids)
-        .await?
-        .into_iter()
-        .map(|msg| msg.into_model(None, None))
-        .collect::<Vec<_>>();
 
     // Fetching the messages looses the order so resort
     let sort_order = data.sort.unwrap_or_default();
@@ -80,68 +73,15 @@ pub async fn search_messages(
         v0::SortOrder::Desc => b.id.cmp(&a.id),
     });
 
-    // TODO: abstract the user and member fetching logic out
+    let (users, members) = Message::fetch_users(db, &messages, server_id.as_deref()).await?;
-    let response = if let Some(true) = data.include_users {
-        let user_ids = messages
-            .iter()
-            .flat_map(|m| {
-                let mut users = vec![m.author.clone()];
-                if let Some(system) = &m.system {
-                    match system {
-                        v0::SystemMessage::ChannelDescriptionChanged { by } => {
-                            users.push(by.clone())
-                        }
-                        v0::SystemMessage::ChannelIconChanged { by } => users.push(by.clone()),
-                        v0::SystemMessage::ChannelOwnershipChanged { from, to, .. } => {
-                            users.push(from.clone());
-                            users.push(to.clone())
-                        }
-                        v0::SystemMessage::ChannelRenamed { by, .. } => users.push(by.clone()),
-                        v0::SystemMessage::UserAdded { by, id, .. }
-                        | v0::SystemMessage::UserRemove { by, id, .. } => {
-                            users.push(by.clone());
-                            users.push(id.clone());
-                        }
-                        v0::SystemMessage::UserBanned { id, .. }
-                        | v0::SystemMessage::UserKicked { id, .. }
-                        | v0::SystemMessage::UserJoined { id, .. }
-                        | v0::SystemMessage::UserLeft { id, .. } => {
-                            users.push(id.clone());
-                        }
-                        v0::SystemMessage::Text { .. } => {}
-                        v0::SystemMessage::MessagePinned { by, .. } => {
-                            users.push(by.clone());
-                        }
-                        v0::SystemMessage::MessageUnpinned { by, .. } => {
-                            users.push(by.clone());
-                        }
-                        v0::SystemMessage::CallStarted { by, .. } => users.push(by.clone()),
-                    }
-                }
-                users
-            })
-            .collect::<HashSet<String>>()
-            .into_iter()
-            .collect::<Vec<String>>();
-        let users = User::fetch_many_ids_as_mutuals(db, &user, &user_ids).await?;
 
-        v0::BulkMessageResponse::MessagesAndUsers {
+    let response = v0::BulkMessageResponse::MessagesAndUsers {
-            messages,
+        messages: messages
-            users,
+            .into_iter()
-            members: if let Some(server_id) = server_id {
+            .map(|msg| msg.into_model(None, None))
-                Some(
+            .collect(),
-                    db.fetch_members(&server_id, &user_ids)
+        users: User::into_mutuals(&user, users).await,
-                        .await?
+        members: Some(members.into_iter().map(Into::into).collect()),
-                        .into_iter()
-                        .map(Into::into)
-                        .collect(),
-                )
-            } else {
-                None
-            },
-        }
-    } else {
-        v0::BulkMessageResponse::JustMessages(messages)
     };
 
     Ok(Json(response))

crates/delta/src/util/test.rs

@@ -203,8 +203,8 @@ impl TestHarness {
                 flags: None,
             },
             v0::MessageAuthor::User(&user.clone().into(&self.db, Some(user)).await),
-            Some(user.clone().into(&self.db, Some(user)).await),
+            Some(user.clone()),
-            Some(member.clone().into()),
+            Some(member.clone()),
             user.limits().await,
             IdempotencyKey::unchecked_from_string("0".to_string()),
             false,