From d04621b1d6252077ce28a9c82d65c0eddc59bff4 Mon Sep 17 00:00:00 2001 From: Paul Makles Date: Sat, 5 Aug 2023 16:59:58 +0100 Subject: [PATCH] feat(core): implement message db methods --- crates/core/database/src/drivers/reference.rs | 6 +- crates/core/database/src/events/client.rs | 13 +- .../database/src/models/messages/model.rs | 23 +- .../core/database/src/models/messages/ops.rs | 6 +- .../src/models/messages/ops/mongodb.rs | 290 +++++++++++++++--- .../src/models/messages/ops/reference.rs | 287 ++++++++++++++--- crates/core/database/src/models/mod.rs | 1 + 7 files changed, 529 insertions(+), 97 deletions(-) diff --git a/crates/core/database/src/drivers/reference.rs b/crates/core/database/src/drivers/reference.rs index 0f3a8664..6f6dd9c2 100644 --- a/crates/core/database/src/drivers/reference.rs +++ b/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>>, pub channel_webhooks: Arc>>, pub emojis: Arc>>, + pub files: Arc>>, + pub messages: Arc>>, pub user_settings: Arc>>, pub users: Arc>>, pub server_bans: Arc>>, pub server_members: Arc>>, pub servers: Arc>>, - pub files: Arc>>, pub safety_reports: Arc>>, pub safety_snapshots: Arc>>, - pub messages: Arc>>, } ); diff --git a/crates/core/database/src/events/client.rs b/crates/core/database/src/events/client.rs index cebc0fee..3f2e812c 100644 --- a/crates/core/database/src/events/client.rs +++ b/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, - FieldsWebhook, MemberCompositeKey, PartialChannel, PartialMember, PartialRole, PartialServer, - PartialUser, PartialWebhook, Server, UserSettings, Webhook, + AppendMessage, Channel, Emoji, FieldsChannel, FieldsMember, FieldsRole, FieldsServer, + FieldsUser, FieldsWebhook, MemberCompositeKey, Message, PartialChannel, PartialMember, + 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 },*/ + BulkMessageDelete { channel: String, ids: Vec }, + /// New server ServerCreate { id: String, diff --git a/crates/core/database/src/models/messages/model.rs b/crates/core/database/src/models/messages/model.rs index 7d7e1ecf..cd9787bc 100644 --- a/crates/core/database/src/models/messages/model.rs +++ b/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!() } } diff --git a/crates/core/database/src/models/messages/ops.rs b/crates/core/database/src/models/messages/ops.rs index 875b3454..7d5ec2ef 100644 --- a/crates/core/database/src/models/messages/ops.rs +++ b/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 reference; +mod mongodb; +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) -> Result<()>; + async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()>; } diff --git a/crates/core/database/src/models/messages/ops/mongodb.rs b/crates/core/database/src/models/messages/ops/mongodb.rs index 882cc8ec..42c2f270 100644 --- a/crates/core/database/src/models/messages/ops/mongodb.rs +++ b/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::MongoDb; +use crate::{AppendMessage, Message, MessageQuery, MessageTimePeriod, MongoDb, PartialMessage}; -use super::AbstractEmojis; +use super::AbstractMessages; -static COL: &str = "emojis"; +static COL: &str = "messages"; #[async_trait] -impl AbstractEmojis for MongoDb { - /// Insert emoji into database. - async fn insert_emoji(&self, emoji: &Emoji) -> Result<()> { - query!(self, insert_one, COL, &emoji).map(|_| ()) +impl AbstractMessages for MongoDb { + /// Insert a new message into the database + async fn insert_message(&self, message: &Message) -> Result<()> { + query!(self, insert_one, COL, &message).map(|_| ()) } - /// Fetch an emoji by its id - async fn fetch_emoji(&self, id: &str) -> Result { + /// Fetch a message by its id + async fn fetch_message(&self, id: &str) -> Result { query!(self, find_one_by_id, COL, id)?.ok_or_else(|| create_error!(NotFound)) } - /// Fetch emoji by their parent id - async fn fetch_emoji_by_parent_id(&self, parent_id: &str) -> Result> { - query!( - self, - find_one, - COL, - doc! { - "parent.id": parent_id - } - )? - .ok_or_else(|| create_error!(NotFound)) - } + /// Fetch multiple messages by given query + async fn fetch_messages(&self, query: MessageQuery) -> Result> { + let mut filter = doc! {}; - /// Fetch emoji by their parent ids - async fn fetch_emoji_by_parent_ids(&self, parent_ids: &[String]) -> Result> { - query!( - self, - find, - COL, - doc! { - "parent.id": { - "$in": parent_ids + // 1. Apply message filters + if let Some(channel) = query.filter.channel { + filter.insert("channel", channel); + } + + if let Some(author) = query.filter.author { + filter.insert("author", author); + } + + let is_search_query = if let Some(query) = query.filter.query { + filter.insert( + "$text", + doc! { + "$search": query + }, + ); + + 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)) } - ) + } } - /// Detach an emoji by its id - async fn detach_emoji(&self, emoji: &Emoji) -> Result<()> { + /// 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(()); + } + self.col::(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::(COL) + .update_one( + doc! { + "_id": id }, doc! { - "$set": { - "parent": { - "type": "Detached" - } + "$addToSet": { + format!("reactions.{emoji}"): user } }, 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::(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::(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::(COL) + .delete_many( + doc! { + "channel": channel, + "_id": { + "$in": ids + } + }, + None, + ) + .await + .map(|_| ()) + .map_err(|_| create_database_error!("delete_many", COL)) + } } diff --git a/crates/core/database/src/models/messages/ops/reference.rs b/crates/core/database/src/models/messages/ops/reference.rs index 2f0c2a2d..b3ddeda3 100644 --- a/crates/core/database/src/models/messages/ops/reference.rs +++ b/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::EmojiParent; -use crate::ReferenceDb; +use crate::{AppendMessage, Message, MessageQuery, PartialMessage, ReferenceDb}; -use super::AbstractEmojis; +use super::AbstractMessages; #[async_trait] -impl AbstractEmojis for ReferenceDb { - /// Insert emoji into database. - async fn insert_emoji(&self, emoji: &Emoji) -> Result<()> { - let mut emojis = self.emojis.lock().await; - if emojis.contains_key(&emoji.id) { - Err(create_database_error!("insert", "emoji")) +impl AbstractMessages for ReferenceDb { + /// Insert a new message into the database + async fn insert_message(&self, message: &Message) -> Result<()> { + let mut messages = self.messages.lock().await; + if messages.contains_key(&message.id) { + 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 - async fn fetch_emoji(&self, id: &str) -> Result { - let emojis = self.emojis.lock().await; - emojis + /// Fetch a message by its id + async fn fetch_message(&self, id: &str) -> Result { + let messages = self.messages.lock().await; + messages .get(id) .cloned() .ok_or_else(|| create_error!(NotFound)) } - /// Fetch emoji by their parent id - async fn fetch_emoji_by_parent_id(&self, parent_id: &str) -> Result> { - let emojis = self.emojis.lock().await; - Ok(emojis + /// Fetch multiple messages by given query + async fn fetch_messages(&self, query: MessageQuery) -> Result> { + let messages = self.messages.lock().await; + let matched_messages = messages .values() - .filter(|emoji| match &emoji.parent { - EmojiParent::Server { id } => id == parent_id, - _ => false, + .filter(|message| { + if let Some(channel) = &query.filter.channel { + if &message.channel != channel { + return false; + } + } + + if let Some(author) = &query.filter.author { + if &message.author != author { + return false; + } + } + + if let Some(query) = &query.filter.query { + if let Some(content) = &message.content { + if !content.to_lowercase().contains(query) { + return false; + } + } else { + return false; + } + } + + true }) .cloned() - .collect()) + .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)) + } + }*/ } - /// Fetch emoji by their parent ids - async fn fetch_emoji_by_parent_ids(&self, parent_ids: &[String]) -> Result> { - let emojis = self.emojis.lock().await; - Ok(emojis - .values() - .filter(|emoji| match &emoji.parent { - EmojiParent::Server { id } => parent_ids.contains(id), - _ => false, - }) - .cloned() - .collect()) - } - - /// Detach an emoji by its id - async fn detach_emoji(&self, emoji: &Emoji) -> Result<()> { - let mut emojis = self.emojis.lock().await; - if let Some(bot) = emojis.get_mut(&emoji.id) { - bot.parent = EmojiParent::Detached; + /// 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(()) + } } diff --git a/crates/core/database/src/models/mod.rs b/crates/core/database/src/models/mod.rs index e5271dfb..b4f6662f 100644 --- a/crates/core/database/src/models/mod.rs +++ b/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