feat(core): implement message db methods

crates/core/database/src/drivers/reference.rs

@@ -4,7 +4,7 @@ use futures::lock::Mutex;
 
 use crate::{
     Bot, Channel, ChannelCompositeKey, ChannelUnread, Emoji, File, Invite, Member,
-    MemberCompositeKey, Server, ServerBan, User, UserSettings, Webhook,
+    MemberCompositeKey, Message, Server, ServerBan, User, UserSettings, Webhook,
 };
 
 database_derived!(
@@ -17,14 +17,14 @@ database_derived!(
         pub channel_unreads: Arc<Mutex<HashMap<ChannelCompositeKey, ChannelUnread>>>,
         pub channel_webhooks: Arc<Mutex<HashMap<String, Webhook>>>,
         pub emojis: Arc<Mutex<HashMap<String, Emoji>>>,
+        pub files: Arc<Mutex<HashMap<String, File>>>,
+        pub messages: Arc<Mutex<HashMap<String, Message>>>,
         pub user_settings: Arc<Mutex<HashMap<String, UserSettings>>>,
         pub users: Arc<Mutex<HashMap<String, User>>>,
         pub server_bans: Arc<Mutex<HashMap<MemberCompositeKey, ServerBan>>>,
         pub server_members: Arc<Mutex<HashMap<MemberCompositeKey, Member>>>,
         pub servers: Arc<Mutex<HashMap<String, Server>>>,
-        pub files: Arc<Mutex<HashMap<String, File>>>,
         pub safety_reports: Arc<Mutex<HashMap<String, ()>>>,
         pub safety_snapshots: Arc<Mutex<HashMap<String, ()>>>,
-        pub messages: Arc<Mutex<HashMap<String, ()>>>,
     }
 );

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

@@ -2,9 +2,10 @@ use authifier::AuthifierEvent;
 use serde::{Deserialize, Serialize};
 
 use revolt_models::v0::{
-    Channel, Emoji, FieldsChannel, FieldsMember, FieldsRole, FieldsServer, FieldsUser,
+    AppendMessage, Channel, Emoji, FieldsChannel, FieldsMember, FieldsRole, FieldsServer,
-    FieldsWebhook, MemberCompositeKey, PartialChannel, PartialMember, PartialRole, PartialServer,
+    FieldsUser, FieldsWebhook, MemberCompositeKey, Message, PartialChannel, PartialMember,
-    PartialUser, PartialWebhook, Server, UserSettings, Webhook,
+    PartialMessage, PartialRole, PartialServer, PartialUser, PartialWebhook, Server, UserSettings,
+    Webhook,
 };
 use revolt_result::Error;
 
@@ -57,8 +58,7 @@ pub enum EventV1 {
     },
 
     /// Ping response
-    Pong { data: Ping },
+    Pong { data: Ping }, */
-
     /// New message
     Message(Message),
 
@@ -103,7 +103,8 @@ pub enum EventV1 {
     },
 
     /// Bulk delete messages
-    BulkMessageDelete { channel: String, ids: Vec<String> },*/
+    BulkMessageDelete { channel: String, ids: Vec<String> },
+
     /// New server
     ServerCreate {
         id: String,

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

@@ -1,9 +1,9 @@
 use indexmap::{IndexMap, IndexSet};
 use iso8601_timestamp::Timestamp;
 use revolt_models::v0::{Embed, MessageSort, MessageWebhook};
-use revolt_result::{create_error, Result};
+use revolt_result::Result;
 
-use crate::{Database, File};
+use crate::{events::client::EventV1, Database, File};
 
 auto_derived_partial!(
     /// Message
@@ -170,11 +170,26 @@ auto_derived!(
 impl Message {
     /// Send a message without any notifications
     pub async fn send_without_notifications(&mut self, db: &Database) -> Result<()> {
-        todo!()
+        db.insert_message(self).await?;
+
+        // Fan out events
+        EventV1::Message(self.clone().into())
+            .p(self.channel.to_string())
+            .await;
+
+        // TODO: update last_message_id
+        // TODO: add mentions for affected users
+        // TODO: generate embeds
+
+        Ok(())
     }
 
     /// Send a message
-    pub async fn send(&mut self) -> Result<()> {
+    pub async fn send(&mut self, db: &Database) -> Result<()> {
+        self.send_without_notifications(db).await?;
+
+        // TODO: web push / FCM
+
         todo!()
     }
 }

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

@@ -2,8 +2,8 @@ use revolt_result::Result;
 
 use crate::{AppendMessage, Message, MessageQuery, PartialMessage};
 
-// mod mongodb;
+mod mongodb;
-// mod reference;
+mod reference;
 
 #[async_trait]
 pub trait AbstractMessages: Sync + Send {
@@ -35,5 +35,5 @@ pub trait AbstractMessages: Sync + Send {
     async fn delete_message(&self, id: &str) -> Result<()>;
 
     /// Delete messages from a channel by their ids and corresponding channel id
-    async fn delete_messages(&self, channel: &str, ids: Vec<String>) -> Result<()>;
+    async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()>;
 }

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

@@ -1,64 +1,214 @@
-use bson::Document;
+use bson::{to_bson, Document};
+use futures::try_join;
+use mongodb::options::FindOptions;
+use revolt_models::v0::MessageSort;
 use revolt_result::Result;
 
-use crate::Emoji;
+use crate::{AppendMessage, Message, MessageQuery, MessageTimePeriod, MongoDb, PartialMessage};
-use crate::MongoDb;
 
-use super::AbstractEmojis;
+use super::AbstractMessages;
 
-static COL: &str = "emojis";
+static COL: &str = "messages";
 
 #[async_trait]
-impl AbstractEmojis for MongoDb {
+impl AbstractMessages for MongoDb {
-    /// Insert emoji into database.
+    /// Insert a new message into the database
-    async fn insert_emoji(&self, emoji: &Emoji) -> Result<()> {
+    async fn insert_message(&self, message: &Message) -> Result<()> {
-        query!(self, insert_one, COL, &emoji).map(|_| ())
+        query!(self, insert_one, COL, &message).map(|_| ())
     }
 
-    /// Fetch an emoji by its id
+    /// Fetch a message by its id
-    async fn fetch_emoji(&self, id: &str) -> Result<Emoji> {
+    async fn fetch_message(&self, id: &str) -> Result<Message> {
         query!(self, find_one_by_id, COL, id)?.ok_or_else(|| create_error!(NotFound))
     }
 
-    /// Fetch emoji by their parent id
+    /// Fetch multiple messages by given query
-    async fn fetch_emoji_by_parent_id(&self, parent_id: &str) -> Result<Vec<Emoji>> {
+    async fn fetch_messages(&self, query: MessageQuery) -> Result<Vec<Message>> {
-        query!(
+        let mut filter = doc! {};
-            self,
+
-            find_one,
+        // 1. Apply message filters
-            COL,
+        if let Some(channel) = query.filter.channel {
-            doc! {
+            filter.insert("channel", channel);
-                "parent.id": parent_id
-            }
-        )?
-        .ok_or_else(|| create_error!(NotFound))
         }
 
-    /// Fetch emoji by their parent ids
+        if let Some(author) = query.filter.author {
-    async fn fetch_emoji_by_parent_ids(&self, parent_ids: &[String]) -> Result<Vec<Emoji>> {
+            filter.insert("author", author);
-        query!(
+        }
-            self,
+
-            find,
+        let is_search_query = if let Some(query) = query.filter.query {
-            COL,
+            filter.insert(
+                "$text",
                 doc! {
-                "parent.id": {
+                    "$search": query
-                    "$in": parent_ids
+                },
-                }
+            );
-            }
+
+            true
+        } else {
+            false
+        };
+
+        // 2. Find query limit
+        let limit = query.limit.unwrap_or(50);
+
+        // 3. Apply message time period
+        match query.time_period {
+            MessageTimePeriod::Relative { nearby } => {
+                // 3.1. Prepare filters
+                let mut older_message_filter = filter.clone();
+                let mut newer_message_filter = filter;
+
+                older_message_filter.insert(
+                    "_id",
+                    doc! {
+                        "$lt": &nearby
+                    },
+                );
+
+                newer_message_filter.insert(
+                    "_id",
+                    doc! {
+                        "$gte": &nearby
+                    },
+                );
+
+                // 3.2. Execute in both directions
+                let (a, b) = try_join!(
+                    self.find_with_options::<_, Message>(
+                        COL,
+                        newer_message_filter,
+                        FindOptions::builder()
+                            .limit(limit / 2 + 1)
+                            .sort(doc! {
+                                "_id": 1_i32
+                            })
+                            .build(),
+                    ),
+                    self.find_with_options::<_, Message>(
+                        COL,
+                        older_message_filter,
+                        FindOptions::builder()
+                            .limit(limit / 2)
+                            .sort(doc! {
+                                "_id": -1_i32
+                            })
+                            .build(),
                     )
+                )
+                .map_err(|_| create_database_error!("find", COL))?;
+
+                Ok([a, b].concat())
+            }
+            MessageTimePeriod::Absolute {
+                before,
+                after,
+                sort,
+            } => {
+                // 3.1. Apply message ID filter
+                if let Some(doc) = match (before, after) {
+                    (Some(before), Some(after)) => Some(doc! {
+                        "$lt": before,
+                        "$gt": after
+                    }),
+                    (Some(before), _) => Some(doc! {
+                        "$lt": before
+                    }),
+                    (_, Some(after)) => Some(doc! {
+                        "$gt": after
+                    }),
+                    _ => None,
+                } {
+                    filter.insert("_id", doc);
+                }
+
+                // 3.2. Execute with given message sort
+                self.find_with_options(
+                    COL,
+                    filter,
+                    FindOptions::builder()
+                        .limit(limit)
+                        .sort(match sort.unwrap_or(MessageSort::Latest) {
+                            // Sort by relevance, fallback to latest
+                            MessageSort::Relevance => {
+                                if is_search_query {
+                                    doc! {
+                                        "score": {
+                                            "$meta": "textScore"
+                                        }
+                                    }
+                                } else {
+                                    doc! {
+                                        "_id": -1_i32
+                                    }
+                                }
+                            }
+                            // Sort by latest first
+                            MessageSort::Latest => doc! {
+                                "_id": -1_i32
+                            },
+                            // Sort by oldest first
+                            MessageSort::Oldest => doc! {
+                                "_id": 1_i32
+                            },
+                        })
+                        .build(),
+                )
+                .await
+                .map_err(|_| create_database_error!("find", COL))
+            }
+        }
+    }
+
+    /// Update a given message with new information
+    async fn update_message(&self, id: &str, message: &PartialMessage) -> Result<()> {
+        query!(self, update_one_by_id, COL, id, message, vec![], None).map(|_| ())
+    }
+
+    /// Append information to a given message
+    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
+        let mut query = doc! {};
+
+        if let Some(embeds) = &append.embeds {
+            if !embeds.is_empty() {
+                query.insert(
+                    "$push",
+                    doc! {
+                        "embeds": {
+                            "$each": to_bson(embeds)
+                                .map_err(|_| create_database_error!("to_bson", "embeds"))?
+                        }
+                    },
+                );
+            }
+        }
+
+        if query.is_empty() {
+            return Ok(());
         }
 
-    /// Detach an emoji by its id
-    async fn detach_emoji(&self, emoji: &Emoji) -> Result<()> {
         self.col::<Document>(COL)
             .update_one(
                 doc! {
-                    "_id": &emoji.id
+                    "_id": id
+                },
+                query,
+                None,
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_one", COL))
+    }
+
+    /// Add a new reaction to a message
+    async fn add_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()> {
+        self.col::<Document>(COL)
+            .update_one(
+                doc! {
+                    "_id": id
                 },
                 doc! {
-                    "$set": {
+                    "$addToSet": {
-                        "parent": {
+                        format!("reactions.{emoji}"): user
-                            "type": "Detached"
-                        }
                     }
                 },
                 None,
@@ -67,4 +217,64 @@ impl AbstractEmojis for MongoDb {
             .map(|_| ())
             .map_err(|_| create_database_error!("update_one", COL))
     }
+
+    /// Remove a reaction from a message
+    async fn remove_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()> {
+        self.col::<Document>(COL)
+            .update_one(
+                doc! {
+                    "_id": id
+                },
+                doc! {
+                    "$pull": {
+                        format!("reactions.{emoji}"): user
+                    }
+                },
+                None,
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_one", COL))
+    }
+
+    /// Remove reaction from a message
+    async fn clear_reaction(&self, id: &str, emoji: &str) -> Result<()> {
+        self.col::<Document>(COL)
+            .update_one(
+                doc! {
+                    "_id": id
+                },
+                doc! {
+                    "$unset": {
+                        format!("reactions.{emoji}"): 1
+                    }
+                },
+                None,
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_one", COL))
+    }
+
+    /// Delete a message from the database by its id
+    async fn delete_message(&self, id: &str) -> Result<()> {
+        query!(self, delete_one_by_id, COL, id).map(|_| ())
+    }
+
+    /// 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,
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("delete_many", COL))
+    }
 }

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

@@ -1,67 +1,272 @@
+use indexmap::IndexSet;
 use revolt_result::Result;
 
-use crate::Emoji;
+use crate::{AppendMessage, Message, MessageQuery, PartialMessage, ReferenceDb};
-use crate::EmojiParent;
-use crate::ReferenceDb;
 
-use super::AbstractEmojis;
+use super::AbstractMessages;
 
 #[async_trait]
-impl AbstractEmojis for ReferenceDb {
+impl AbstractMessages for ReferenceDb {
-    /// Insert emoji into database.
+    /// Insert a new message into the database
-    async fn insert_emoji(&self, emoji: &Emoji) -> Result<()> {
+    async fn insert_message(&self, message: &Message) -> Result<()> {
-        let mut emojis = self.emojis.lock().await;
+        let mut messages = self.messages.lock().await;
-        if emojis.contains_key(&emoji.id) {
+        if messages.contains_key(&message.id) {
-            Err(create_database_error!("insert", "emoji"))
+            Err(create_database_error!("insert", "message"))
         } else {
-            emojis.insert(emoji.id.to_string(), emoji.clone());
+            messages.insert(message.id.to_string(), message.clone());
             Ok(())
         }
     }
 
-    /// Fetch an emoji by its id
+    /// Fetch a message by its id
-    async fn fetch_emoji(&self, id: &str) -> Result<Emoji> {
+    async fn fetch_message(&self, id: &str) -> Result<Message> {
-        let emojis = self.emojis.lock().await;
+        let messages = self.messages.lock().await;
-        emojis
+        messages
             .get(id)
             .cloned()
             .ok_or_else(|| create_error!(NotFound))
     }
 
-    /// Fetch emoji by their parent id
+    /// Fetch multiple messages by given query
-    async fn fetch_emoji_by_parent_id(&self, parent_id: &str) -> Result<Vec<Emoji>> {
+    async fn fetch_messages(&self, query: MessageQuery) -> Result<Vec<Message>> {
-        let emojis = self.emojis.lock().await;
+        let messages = self.messages.lock().await;
-        Ok(emojis
+        let matched_messages = messages
             .values()
-            .filter(|emoji| match &emoji.parent {
+            .filter(|message| {
-                EmojiParent::Server { id } => id == parent_id,
+                if let Some(channel) = &query.filter.channel {
-                _ => false,
+                    if &message.channel != channel {
-            })
+                        return false;
-            .cloned()
+                    }
-            .collect())
                 }
 
-    /// Fetch emoji by their parent ids
+                if let Some(author) = &query.filter.author {
-    async fn fetch_emoji_by_parent_ids(&self, parent_ids: &[String]) -> Result<Vec<Emoji>> {
+                    if &message.author != author {
-        let emojis = self.emojis.lock().await;
+                        return false;
-        Ok(emojis
+                    }
-            .values()
-            .filter(|emoji| match &emoji.parent {
-                EmojiParent::Server { id } => parent_ids.contains(id),
-                _ => false,
-            })
-            .cloned()
-            .collect())
                 }
 
-    /// Detach an emoji by its id
+                if let Some(query) = &query.filter.query {
-    async fn detach_emoji(&self, emoji: &Emoji) -> Result<()> {
+                    if let Some(content) = &message.content {
-        let mut emojis = self.emojis.lock().await;
+                        if !content.to_lowercase().contains(query) {
-        if let Some(bot) = emojis.get_mut(&emoji.id) {
+                            return false;
-            bot.parent = EmojiParent::Detached;
+                        }
+                    } else {
+                        return false;
+                    }
+                }
+
+                true
+            })
+            .cloned()
+            .collect();
+
+        // TODO: sorting, etc
+
+        Ok(matched_messages)
+
+        /*
+        // 2. Find query limit
+        let limit = query.limit.unwrap_or(50);
+
+        // 3. Apply message time period
+        match query.time_period {
+            MessageTimePeriod::Relative { nearby } => {
+                // 3.1. Prepare filters
+                let mut older_message_filter = filter.clone();
+                let mut newer_message_filter = filter;
+
+                older_message_filter.insert(
+                    "_id",
+                    doc! {
+                        "$lt": &nearby
+                    },
+                );
+
+                newer_message_filter.insert(
+                    "_id",
+                    doc! {
+                        "$gte": &nearby
+                    },
+                );
+
+                // 3.2. Execute in both directions
+                let (a, b) = try_join!(
+                    self.find_with_options::<_, Message>(
+                        COL,
+                        newer_message_filter,
+                        FindOptions::builder()
+                            .limit(limit / 2 + 1)
+                            .sort(doc! {
+                                "_id": 1_i32
+                            })
+                            .build(),
+                    ),
+                    self.find_with_options::<_, Message>(
+                        COL,
+                        older_message_filter,
+                        FindOptions::builder()
+                            .limit(limit / 2)
+                            .sort(doc! {
+                                "_id": -1_i32
+                            })
+                            .build(),
+                    )
+                )
+                .map_err(|_| create_database_error!("find", COL))?;
+
+                Ok([a, b].concat())
+            }
+            MessageTimePeriod::Absolute {
+                before,
+                after,
+                sort,
+            } => {
+                // 3.1. Apply message ID filter
+                if let Some(doc) = match (before, after) {
+                    (Some(before), Some(after)) => Some(doc! {
+                        "$lt": before,
+                        "$gt": after
+                    }),
+                    (Some(before), _) => Some(doc! {
+                        "$lt": before
+                    }),
+                    (_, Some(after)) => Some(doc! {
+                        "$gt": after
+                    }),
+                    _ => None,
+                } {
+                    filter.insert("_id", doc);
+                }
+
+                // 3.2. Execute with given message sort
+                self.find_with_options(
+                    COL,
+                    filter,
+                    FindOptions::builder()
+                        .limit(limit)
+                        .sort(match sort.unwrap_or(MessageSort::Latest) {
+                            // Sort by relevance, fallback to latest
+                            MessageSort::Relevance => {
+                                if is_search_query {
+                                    doc! {
+                                        "score": {
+                                            "$meta": "textScore"
+                                        }
+                                    }
+                                } else {
+                                    doc! {
+                                        "_id": -1_i32
+                                    }
+                                }
+                            }
+                            // Sort by latest first
+                            MessageSort::Latest => doc! {
+                                "_id": -1_i32
+                            },
+                            // Sort by oldest first
+                            MessageSort::Oldest => doc! {
+                                "_id": 1_i32
+                            },
+                        })
+                        .build(),
+                )
+                .await
+                .map_err(|_| create_database_error!("find", COL))
+            }
+        }*/
+    }
+
+    /// Update a given message with new information
+    async fn update_message(&self, id: &str, message: &PartialMessage) -> Result<()> {
+        let mut messages = self.messages.lock().await;
+        if let Some(message_data) = messages.get_mut(id) {
+            message_data.apply_options(message.to_owned());
             Ok(())
         } else {
             Err(create_error!(NotFound))
         }
     }
+
+    /// Append information to a given message
+    async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
+        let mut messages = self.messages.lock().await;
+        if let Some(message_data) = messages.get_mut(id) {
+            if let Some(embeds) = &append.embeds {
+                if !embeds.is_empty() {
+                    if let Some(embeds_data) = &mut message_data.embeds {
+                        embeds_data.extend(embeds.clone());
+                    } else {
+                        message_data.embeds = Some(embeds.clone());
+                    }
+                }
+            }
+
+            Ok(())
+        } else {
+            Err(create_error!(NotFound))
+        }
+    }
+
+    /// Add a new reaction to a message
+    async fn add_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()> {
+        let mut messages = self.messages.lock().await;
+        if let Some(message) = messages.get_mut(id) {
+            if let Some(users) = message.reactions.get_mut(emoji) {
+                users.insert(user.to_string());
+            } else {
+                message
+                    .reactions
+                    .insert(emoji.to_string(), IndexSet::from([user.to_string()]));
+            }
+
+            Ok(())
+        } else {
+            Err(create_error!(NotFound))
+        }
+    }
+
+    /// Remove a reaction from a message
+    async fn remove_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()> {
+        let mut messages = self.messages.lock().await;
+        if let Some(message) = messages.get_mut(id) {
+            if let Some(users) = message.reactions.get_mut(emoji) {
+                users.remove(&user.to_string());
+            }
+
+            Ok(())
+        } else {
+            Err(create_error!(NotFound))
+        }
+    }
+
+    /// Remove reaction from a message
+    async fn clear_reaction(&self, id: &str, emoji: &str) -> Result<()> {
+        let mut messages = self.messages.lock().await;
+        if let Some(message) = messages.get_mut(id) {
+            message.reactions.remove(emoji);
+            Ok(())
+        } else {
+            Err(create_error!(NotFound))
+        }
+    }
+
+    /// Delete a message from the database by its id
+    async fn delete_message(&self, id: &str) -> Result<()> {
+        let mut messages = self.messages.lock().await;
+        if messages.remove(id).is_some() {
+            Ok(())
+        } else {
+            Err(create_error!(NotFound))
+        }
+    }
+
+    /// Delete messages from a channel by their ids and corresponding channel id
+    async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()> {
+        self.messages
+            .lock()
+            .await
+            .retain(|id, message| message.channel != channel && !ids.contains(id));
+
+        Ok(())
+    }
 }

crates/core/database/src/models/mod.rs

@@ -43,6 +43,7 @@ pub trait AbstractDatabase:
     + channel_webhooks::AbstractWebhooks
     + emojis::AbstractEmojis
     + files::AbstractAttachments
+    + messages::AbstractMessages
     + ratelimit_events::AbstractRatelimitEvents
     + server_bans::AbstractServerBans
     + server_members::AbstractServerMembers