feat(core): implement channel unreads model

closes #269

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

@@ -3,7 +3,8 @@ use std::{collections::HashMap, sync::Arc};
 use futures::lock::Mutex;
 
 use crate::{
-    Bot, Channel, File, Invite, Member, MemberCompositeKey, Server, User, UserSettings, Webhook,
+    Bot, Channel, ChannelCompositeKey, ChannelUnread, File, Invite, Member, MemberCompositeKey,
+    Server, User, UserSettings, Webhook,
 };
 
 database_derived!(
@@ -13,6 +14,7 @@ database_derived!(
         pub bots: Arc<Mutex<HashMap<String, Bot>>>,
         pub channels: Arc<Mutex<HashMap<String, Channel>>>,
         pub channel_invites: Arc<Mutex<HashMap<String, Invite>>>,
+        pub channel_unreads: Arc<Mutex<HashMap<ChannelCompositeKey, ChannelUnread>>>,
         pub channel_webhooks: Arc<Mutex<HashMap<String, Webhook>>>,
         pub user_settings: Arc<Mutex<HashMap<String, UserSettings>>>,
         pub users: Arc<Mutex<HashMap<String, User>>>,
@@ -24,6 +26,5 @@ database_derived!(
         pub safety_snapshots: Arc<Mutex<HashMap<String, ()>>>,
         pub emoji: Arc<Mutex<HashMap<String, ()>>>,
         pub messages: Arc<Mutex<HashMap<String, ()>>>,
-        pub channel_unreads: Arc<Mutex<HashMap<String, ()>>>,
     }
 );

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

@@ -0,0 +1,5 @@
+mod model;
+mod ops;
+
+pub use model::*;
+pub use ops::*;

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

@@ -0,0 +1,24 @@
+auto_derived!(
+    /// Channel Unread
+    pub struct ChannelUnread {
+        /// Composite key pointing to a user's view of a channel
+        #[serde(rename = "_id")]
+        pub id: ChannelCompositeKey,
+
+        /// Id of the last message read in this channel by a user
+        #[serde(skip_serializing_if = "Option::is_none")]
+        pub last_id: Option<String>,
+        /// Array of message ids that mention the user
+        #[serde(skip_serializing_if = "Option::is_none")]
+        pub mentions: Option<Vec<String>>,
+    }
+
+    /// Composite primary key consisting of channel and user id
+    #[derive(Hash)]
+    pub struct ChannelCompositeKey {
+        /// Channel Id
+        pub channel: String,
+        /// User Id
+        pub user: String,
+    }
+);

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

@@ -0,0 +1,31 @@
+use revolt_result::Result;
+
+use crate::ChannelUnread;
+
+mod mongodb;
+mod reference;
+
+#[async_trait]
+pub trait AbstractChannelUnreads: Sync + Send {
+    /// Acknowledge a message.
+    async fn acknowledge_message(
+        &self,
+        channel_id: &str,
+        user_id: &str,
+        message_id: &str,
+    ) -> Result<()>;
+
+    /// Acknowledge many channels.
+    async fn acknowledge_channels(&self, user_id: &str, channel_ids: &[String]) -> Result<()>;
+
+    /// Add a mention.
+    async fn add_mention_to_unread<'a>(
+        &self,
+        channel_id: &str,
+        user_id: &str,
+        message_ids: &[String],
+    ) -> Result<()>;
+
+    /// Fetch all channel unreads for a user.
+    async fn fetch_unreads(&self, user_id: &str) -> Result<Vec<ChannelUnread>>;
+}

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

@@ -0,0 +1,119 @@
+use bson::Document;
+use mongodb::options::UpdateOptions;
+use revolt_result::Result;
+use ulid::Ulid;
+
+use crate::ChannelUnread;
+use crate::MongoDb;
+
+use super::AbstractChannelUnreads;
+
+static COL: &str = "channel_unreads";
+
+#[async_trait]
+impl AbstractChannelUnreads for MongoDb {
+    /// Acknowledge a message.
+    async fn acknowledge_message(
+        &self,
+        channel_id: &str,
+        user_id: &str,
+        message_id: &str,
+    ) -> Result<()> {
+        self.col::<Document>(COL)
+            .update_one(
+                doc! {
+                    "_id.channel": channel_id,
+                    "_id.user": user_id,
+                },
+                doc! {
+                    "$unset": {
+                        "mentions": 1_i32
+                    },
+                    "$set": {
+                        "last_id": message_id
+                    }
+                },
+                UpdateOptions::builder().upsert(true).build(),
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_one", COL))
+    }
+
+    /// Acknowledge many channels.
+    async fn acknowledge_channels(&self, user_id: &str, channel_ids: &[String]) -> Result<()> {
+        let current_time = Ulid::new().to_string();
+
+        self.col::<Document>(COL)
+            .delete_many(
+                doc! {
+                    "_id.channel": {
+                        "$in": channel_ids
+                    },
+                    "_id.user": user_id
+                },
+                None,
+            )
+            .await
+            .map_err(|_| create_database_error!("delete_many", COL))?;
+
+        self.col::<Document>(COL)
+            .insert_many(
+                channel_ids
+                    .iter()
+                    .map(|channel_id| {
+                        doc! {
+                            "_id": {
+                                "channel": channel_id,
+                                "user": user_id
+                            },
+                            "last_id": &current_time
+                        }
+                    })
+                    .collect::<Vec<Document>>(),
+                None,
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_many", COL))
+    }
+
+    /// Add a mention.
+    async fn add_mention_to_unread<'a>(
+        &self,
+        channel_id: &str,
+        user_id: &str,
+        message_ids: &[String],
+    ) -> Result<()> {
+        self.col::<Document>(COL)
+            .update_one(
+                doc! {
+                    "_id.channel": channel_id,
+                    "_id.user": user_id,
+                },
+                doc! {
+                    "$push": {
+                        "mentions": {
+                            "$each": message_ids
+                        }
+                    }
+                },
+                UpdateOptions::builder().upsert(true).build(),
+            )
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("update_one", COL))
+    }
+
+    /// Fetch all channel unreads for a user.
+    async fn fetch_unreads(&self, user_id: &str) -> Result<Vec<ChannelUnread>> {
+        query!(
+            self,
+            find,
+            COL,
+            doc! {
+                "_id.user": user_id
+            }
+        )
+    }
+}

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

@@ -0,0 +1,89 @@
+use revolt_result::Result;
+use ulid::Ulid;
+
+use crate::{ChannelCompositeKey, ChannelUnread, ReferenceDb};
+
+use super::AbstractChannelUnreads;
+
+#[async_trait]
+impl AbstractChannelUnreads for ReferenceDb {
+    /// Acknowledge a message.
+    async fn acknowledge_message(
+        &self,
+        channel_id: &str,
+        user_id: &str,
+        message_id: &str,
+    ) -> Result<()> {
+        let mut unreads = self.channel_unreads.lock().await;
+        let key = ChannelCompositeKey {
+            channel: channel_id.to_string(),
+            user: user_id.to_string(),
+        };
+
+        if let Some(mut unread) = unreads.get_mut(&key) {
+            unread.mentions = None;
+            unread.last_id.replace(message_id.to_string());
+        } else {
+            unreads.insert(
+                key.clone(),
+                ChannelUnread {
+                    id: key,
+                    last_id: Some(message_id.to_string()),
+                    mentions: None,
+                },
+            );
+        }
+
+        Ok(())
+    }
+
+    /// Acknowledge many channels.
+    async fn acknowledge_channels(&self, user_id: &str, channel_ids: &[String]) -> Result<()> {
+        let current_time = Ulid::new().to_string();
+        for channel_id in channel_ids {
+            self.acknowledge_message(channel_id, user_id, &current_time)
+                .await?;
+        }
+
+        Ok(())
+    }
+
+    /// Add a mention.
+    async fn add_mention_to_unread<'a>(
+        &self,
+        channel_id: &str,
+        user_id: &str,
+        message_ids: &[String],
+    ) -> Result<()> {
+        let mut unreads = self.channel_unreads.lock().await;
+        let key = ChannelCompositeKey {
+            channel: channel_id.to_string(),
+            user: user_id.to_string(),
+        };
+
+        if let Some(unread) = unreads.get_mut(&key) {
+            unread.mentions.replace(message_ids.to_vec());
+        } else {
+            unreads.insert(
+                key.clone(),
+                ChannelUnread {
+                    id: key,
+                    last_id: None,
+                    mentions: Some(message_ids.to_vec()),
+                },
+            );
+        }
+
+        Ok(())
+    }
+
+    /// Fetch all channel unreads for a user.
+    async fn fetch_unreads(&self, user_id: &str) -> Result<Vec<ChannelUnread>> {
+        let unreads = self.channel_unreads.lock().await;
+        Ok(unreads
+            .values()
+            .filter(|unread| unread.id.user == user_id)
+            .cloned()
+            .collect())
+    }
+}

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

@@ -1,6 +1,7 @@
 mod admin_migrations;
 mod bots;
 mod channel_invites;
+mod channel_unreads;
 mod channel_webhooks;
 mod channels;
 mod files;
@@ -13,6 +14,7 @@ mod users;
 pub use admin_migrations::*;
 pub use bots::*;
 pub use channel_invites::*;
+pub use channel_unreads::*;
 pub use channel_webhooks::*;
 pub use channels::*;
 pub use files::*;
@@ -31,6 +33,7 @@ pub trait AbstractDatabase:
     + bots::AbstractBots
     + channels::AbstractChannels
     + channel_invites::AbstractChannelInvites
+    + channel_unreads::AbstractChannelUnreads
     + channel_webhooks::AbstractWebhooks
     + files::AbstractAttachments
     + ratelimit_events::AbstractRatelimitEvents

crates/core/database/src/util/bridge/v0.rs

@@ -61,6 +61,25 @@ impl From<crate::Invite> for Invite {
     }
 }
 
+impl From<crate::ChannelUnread> for ChannelUnread {
+    fn from(value: crate::ChannelUnread) -> Self {
+        ChannelUnread {
+            id: value.id.into(),
+            last_id: value.last_id,
+            mentions: value.mentions.unwrap_or_default(),
+        }
+    }
+}
+
+impl From<crate::ChannelCompositeKey> for ChannelCompositeKey {
+    fn from(value: crate::ChannelCompositeKey) -> Self {
+        ChannelCompositeKey {
+            channel: value.channel,
+            user: value.user,
+        }
+    }
+}
+
 impl From<crate::Webhook> for Webhook {
     fn from(value: crate::Webhook) -> Self {
         Webhook {

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

@@ -0,0 +1,27 @@
+auto_derived!(
+    /// Channel Unread
+    pub struct ChannelUnread {
+        /// Composite key pointing to a user's view of a channel
+        #[serde(rename = "_id")]
+        pub id: ChannelCompositeKey,
+
+        /// Id of the last message read in this channel by a user
+        #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
+        pub last_id: Option<String>,
+        /// Array of message ids that mention the user
+        #[cfg_attr(
+            feature = "serde",
+            serde(skip_serializing_if = "Vec::is_empty", default)
+        )]
+        pub mentions: Vec<String>,
+    }
+
+    /// Composite primary key consisting of channel and user id
+    #[derive(Hash)]
+    pub struct ChannelCompositeKey {
+        /// Channel Id
+        pub channel: String,
+        /// User Id
+        pub user: String,
+    }
+);

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

@@ -1,5 +1,6 @@
 mod bots;
 mod channel_invites;
+mod channel_unreads;
 mod channel_webhooks;
 mod channels;
 mod files;
@@ -9,6 +10,7 @@ mod users;
 
 pub use bots::*;
 pub use channel_invites::*;
+pub use channel_unreads::*;
 pub use channel_webhooks::*;
 pub use channels::*;
 pub use files::*;