fix: use trust_cloudflare config value instead of env var

Signed-off-by: ElfFlu <228066338+ElfFlu@users.noreply.github.com>

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

@@ -144,7 +144,13 @@ pub enum EventV1 {
     },
 
     /// User joins server
-    ServerMemberJoin { id: String, user: String },
+    ServerMemberJoin {
+        id: String,
+        // Deprecated: use member.id.user
+        #[deprecated = "Use member.id.user instead"]
+        user: String,
+        member: Member,
+    },
 
     /// User left server
     ServerMemberLeave {

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

@@ -1,8 +1,8 @@
-use serde::Deserialize;
+use serde::{Serialize, Deserialize};
 
 use super::client::Ping;
 
-#[derive(Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug)]
 #[serde(tag = "type")]
 pub enum ClientMessage {
     Authenticate { token: String },

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

@@ -30,6 +30,9 @@ auto_derived_partial!(
         /// Timestamp this member is timed out until
         #[serde(skip_serializing_if = "Option::is_none")]
         pub timeout: Option<Timestamp>,
+        // This value only exists in the database, not the models.
+        // If it is not-None, the database layer should return None to member fetching queries.
+        // pub pending_deletion_at: Option<Timestamp>
     },
     "PartialMember"
 );
@@ -50,6 +53,7 @@ auto_derived!(
         Avatar,
         Roles,
         Timeout,
+        JoinedAt,
     }
 
     /// Member removal intention
@@ -90,7 +94,7 @@ impl Member {
             return Err(create_error!(AlreadyInServer));
         }
 
-        let member = Member {
+        let mut member = Member {
             id: MemberCompositeKey {
                 server: server.id.to_string(),
                 user: user.id.to_string(),
@@ -98,7 +102,9 @@ impl Member {
             ..Default::default()
         };
 
-        db.insert_member(&member).await?;
+        if let Some(updated) = db.insert_or_merge_member(&member).await? {
+            member = updated;
+        }
 
         let should_fetch = channels.is_none();
         let mut channels = channels.unwrap_or_default();
@@ -124,6 +130,7 @@ impl Member {
         EventV1::ServerMemberJoin {
             id: server.id.clone(),
             user: user.id.clone(),
+            member: member.clone().into(),
         }
         .p(server.id.clone())
         .await;
@@ -186,6 +193,7 @@ impl Member {
 
     pub fn remove_field(&mut self, field: &FieldsMember) {
         match field {
+            FieldsMember::JoinedAt => (),
             FieldsMember::Avatar => self.avatar = None,
             FieldsMember::Nickname => self.nickname = None,
             FieldsMember::Roles => self.roles.clear(),
@@ -224,7 +232,7 @@ impl Member {
         intention: RemovalIntention,
         silent: bool,
     ) -> Result<()> {
-        db.delete_member(&self.id).await?;
+        db.soft_delete_member(&self.id).await?;
 
         EventV1::ServerMemberLeave {
             id: self.id.server.to_string(),
@@ -260,3 +268,74 @@ impl Member {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use iso8601_timestamp::{Duration, Timestamp};
+    use revolt_models::v0::DataCreateServer;
+
+    use crate::{Member, PartialMember, RemovalIntention, Server, User};
+
+    #[async_std::test]
+    async fn muted_member_rejoin() {
+        database_test!(|db| async move {
+            match db {
+                crate::Database::Reference(_) => return,
+                crate::Database::MongoDb(_) => (),
+            }
+            let owner = User::create(&db, "Server Owner".to_string(), None, None)
+                .await
+                .unwrap();
+
+            let kickable_user = User::create(&db, "Member".to_string(), None, None)
+                .await
+                .unwrap();
+
+            let server = Server::create(
+                &db,
+                DataCreateServer {
+                    name: "Server".to_string(),
+                    description: None,
+                    nsfw: None,
+                },
+                &owner,
+                false,
+            )
+            .await
+            .unwrap()
+            .0;
+
+            Member::create(&db, &server, &owner, None).await.unwrap();
+            let mut kickable_member = Member::create(&db, &server, &kickable_user, None)
+                .await
+                .unwrap()
+                .0;
+
+            kickable_member
+                .update(
+                    &db,
+                    PartialMember {
+                        timeout: Some(Timestamp::now_utc() + Duration::minutes(5)),
+                        ..Default::default()
+                    },
+                    vec![],
+                )
+                .await
+                .unwrap();
+
+            assert!(kickable_member.in_timeout());
+
+            kickable_member
+                .remove(&db, &server, RemovalIntention::Kick, false)
+                .await
+                .unwrap();
+
+            let kickable_member = Member::create(&db, &server, &kickable_user, None)
+                .await
+                .unwrap()
+                .0;
+
+            assert!(kickable_member.in_timeout())
+        });
+    }
+}

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

@@ -73,13 +73,13 @@ impl ChunkedServerMembersGenerator {
 #[async_trait]
 pub trait AbstractServerMembers: Sync + Send {
     /// Insert a new server member into the database
-    async fn insert_member(&self, member: &Member) -> Result<()>;
+    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>>;
 
     /// Fetch a server member by their id
     async fn fetch_member(&self, server_id: &str, user_id: &str) -> Result<Member>;
 
     /// Fetch all members in a server
-    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>>;
+    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>>;
 
     /// Fetch all members in a server as an iterator
     async fn fetch_all_members_chunked(
@@ -100,10 +100,10 @@ pub trait AbstractServerMembers: Sync + Send {
     ) -> Result<ChunkedServerMembersGenerator>;
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>>;
+    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>>;
 
     /// Fetch multiple members by their ids
-    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>>;
+    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>>;
 
     /// Fetch member count of a server
     async fn fetch_member_count(&self, server_id: &str) -> Result<usize>;
@@ -119,6 +119,14 @@ pub trait AbstractServerMembers: Sync + Send {
         remove: Vec<FieldsMember>,
     ) -> Result<()>;
 
-    /// Delete a server member by their id
-    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
+    /// Marks a user as no longer a member of a server, while retaining the database value.
+    /// This is used to keep information such as timeouts in place, but will remove information such as join date and applied roles.
+    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
+
+    /// Forcibly delete a server member by their id.
+    /// This will cancel any pending timeouts or other longer term actions, and they will not be reapplied on rejoin.
+    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()>;
+
+    /// Fetch all members who have been marked for deletion.
+    async fn remove_dangling_members(&self) -> Result<()>;
 }

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

@@ -1,4 +1,6 @@
+use bson::Document;
 use futures::StreamExt;
+use iso8601_timestamp::Timestamp;
 use mongodb::options::ReadConcern;
 use revolt_result::Result;
 
@@ -11,9 +13,42 @@ static COL: &str = "server_members";
 
 #[async_trait]
 impl AbstractServerMembers for MongoDb {
-    /// Insert a new server member into the database
-    async fn insert_member(&self, member: &Member) -> Result<()> {
-        query!(self, insert_one, COL, &member).map(|_| ())
+    /// Insert a new server member (or use the existing member if one is found)
+    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
+        let existing: Result<Option<Document>> = query!(
+            self,
+            find_one,
+            COL,
+            doc! {
+                "_id.server": &member.id.server,
+                "_id.user": &member.id.user,
+                "pending_deletion_at": {"$exists": true}
+            }
+        );
+        // Update the existing record if it exist, otherwise make a new record
+        if existing.is_ok_and(|x| x.is_some()) {
+            self.col::<Member>(COL)
+                .find_one_and_update(
+                    doc! {
+                        "_id.server": &member.id.server,
+                        "_id.user": &member.id.user,
+                    },
+                    doc! {
+                        "$set": {
+                            "joined_at": member.joined_at.duration_since(Timestamp::UNIX_EPOCH).whole_seconds(),
+                        },
+                        "$unset": {
+                            "pending_deletion_at": ""
+                        }
+                    },
+                )
+                .return_document(mongodb::options::ReturnDocument::After)
+                .await
+                .map_err(|_| create_database_error!("update_one", COL))
+        } else {
+            query!(self, insert_one, COL, &member).map(|_| ())?;
+            Ok(None)
+        }
     }
 
     /// Fetch a server member by their id
@@ -24,18 +59,20 @@ impl AbstractServerMembers for MongoDb {
             COL,
             doc! {
                 "_id.server": server_id,
-                "_id.user": user_id
+                "_id.user": user_id,
+                "pending_deletion_at": {"$exists": false}
             }
         )?
         .ok_or_else(|| create_error!(NotFound))
     }
 
     /// Fetch all members in a server
-    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
-                "_id.server": server_id
+                "_id.server": server_id,
+                "pending_deletion_at": {"$exists": false}
             })
             .await
             .map_err(|_| create_database_error!("find", COL))?
@@ -139,11 +176,12 @@ impl AbstractServerMembers for MongoDb {
     }
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
-                "_id.user": user_id
+                "_id.user": user_id,
+                "pending_deletion_at": {"$exists": false}
             })
             .await
             .map_err(|_| create_database_error!("find", COL))?
@@ -159,11 +197,12 @@ impl AbstractServerMembers for MongoDb {
     }
 
     /// Fetch multiple members by their ids
-    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
+    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
         Ok(self
             .col::<Member>(COL)
             .find(doc! {
                 "_id.server": server_id,
+                "pending_deletion_at": {"$exists": false},
                 "_id.user": {
                     "$in": ids
                 }
@@ -185,7 +224,8 @@ impl AbstractServerMembers for MongoDb {
     async fn fetch_member_count(&self, server_id: &str) -> Result<usize> {
         self.col::<Member>(COL)
             .count_documents(doc! {
-                "_id.server": server_id
+                "_id.server": server_id,
+                "pending_deletion_at": {"$exists": false}
             })
             .await
             .map(|c| c as usize)
@@ -196,7 +236,8 @@ impl AbstractServerMembers for MongoDb {
     async fn fetch_server_count(&self, user_id: &str) -> Result<usize> {
         self.col::<Member>(COL)
             .count_documents(doc! {
-                "_id.user": user_id
+                "_id.user": user_id,
+                "pending_deletion_at": {"$exists": false}
             })
             .await
             .map(|c| c as usize)
@@ -225,8 +266,42 @@ impl AbstractServerMembers for MongoDb {
         .map(|_| ())
     }
 
+    /// Marks a member for deletion.
+    /// This will remove the record if the user has no pending actions (eg. timeout),
+    /// otherwise will slate the record for deletion by revolt_crond once the actions expire.
+    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+        let member = self.fetch_member(&id.server, &id.user).await;
+        if let Ok(member) = member {
+            if member.in_timeout() {
+                self.col::<Document>(COL)
+                    .update_many(
+                        doc! {
+                            "_id.server": &id.server,
+                            "_id.user": &id.user,
+                        },
+                        doc! {
+                            "$set": {"pending_deletion_at": format!("{}", member.timeout.unwrap().format())},
+                            "$unset": {
+                                "joined_at": "",
+                                "avatar": "",
+                                "nickname": "",
+                                "roles": ""
+                            }
+                        },
+                    )
+                    .await
+                    .map(|_| ())
+                    .map_err(|_| create_database_error!("update_many", COL))
+            } else {
+                self.force_delete_member(id).await
+            }
+        } else {
+            Err(create_database_error!("fetch_member", COL))
+        }
+    }
+
     /// Delete a server member by their id
-    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
         query!(
             self,
             delete_one,
@@ -238,11 +313,25 @@ impl AbstractServerMembers for MongoDb {
         )
         .map(|_| ())
     }
+
+    async fn remove_dangling_members(&self) -> Result<()> {
+        let now = Timestamp::now_utc();
+        let date = bson::to_bson(&now).expect("Failed to serialize timestamp");
+
+        self.col::<Document>(COL)
+            .delete_many(doc! {
+                "pending_deletion_at": {"$lt": date}
+            })
+            .await
+            .map(|_| ())
+            .map_err(|_| create_database_error!("count_documents", COL))
+    }
 }
 
 impl IntoDocumentPath for FieldsMember {
     fn as_path(&self) -> Option<&'static str> {
         Some(match self {
+            FieldsMember::JoinedAt => "joined_at",
             FieldsMember::Avatar => "avatar",
             FieldsMember::Nickname => "nickname",
             FieldsMember::Roles => "roles",

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

@@ -8,13 +8,13 @@ use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
 #[async_trait]
 impl AbstractServerMembers for ReferenceDb {
     /// Insert a new server member into the database
-    async fn insert_member(&self, member: &Member) -> Result<()> {
+    async fn insert_or_merge_member(&self, member: &Member) -> Result<Option<Member>> {
         let mut server_members = self.server_members.lock().await;
         if server_members.contains_key(&member.id) {
             Err(create_database_error!("insert", "member"))
         } else {
             server_members.insert(member.id.clone(), member.clone());
-            Ok(())
+            Ok(None)
         }
     }
 
@@ -31,7 +31,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch all members in a server
-    async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_members(&self, server_id: &str) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(server_members
             .values()
@@ -105,7 +105,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch all memberships for a user
-    async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
+    async fn fetch_all_memberships(&self, user_id: &str) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(server_members
             .values()
@@ -115,7 +115,7 @@ impl AbstractServerMembers for ReferenceDb {
     }
 
     /// Fetch multiple members by their ids
-    async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
+    async fn fetch_members(&self, server_id: &str, ids: &[String]) -> Result<Vec<Member>> {
         let server_members = self.server_members.lock().await;
         Ok(ids
             .iter()
@@ -169,8 +169,26 @@ impl AbstractServerMembers for ReferenceDb {
         }
     }
 
+    /// Soft delete a member
+    async fn soft_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+        let mut server_members = self.server_members.lock().await;
+
+        let member = server_members.get_mut(id);
+        if let Some(member) = member {
+            if member.in_timeout() {
+                panic!("Soft deletion is not implemented.")
+            } else if server_members.remove(id).is_some() {
+                Ok(())
+            } else {
+                Err(create_error!(NotFound))
+            }
+        } else {
+            Err(create_error!(NotFound))
+        }
+    }
+
     /// Delete a server member by their id
-    async fn delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
+    async fn force_delete_member(&self, id: &MemberCompositeKey) -> Result<()> {
         let mut server_members = self.server_members.lock().await;
         if server_members.remove(id).is_some() {
             Ok(())
@@ -178,4 +196,8 @@ impl AbstractServerMembers for ReferenceDb {
             Err(create_error!(NotFound))
         }
     }
+
+    async fn remove_dangling_members(&self) -> Result<()> {
+        todo!()
+    }
 }

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

@@ -1,14 +1,21 @@
-use axum::{extract::FromRequestParts, http::request::Parts};
+use axum::{
+    extract::{FromRef, FromRequestParts},
+    http::request::Parts,
+};
 
 use revolt_result::{create_error, Error, Result};
 
 use crate::{Database, User};
 
 #[async_trait::async_trait]
-impl FromRequestParts<Database> for User {
+impl<S: Send + Sync> FromRequestParts<S> for User
+where
+    Database: FromRef<S>,
+{
     type Rejection = Error;
 
-    async fn from_request_parts(parts: &mut Parts, db: &Database) -> Result<User> {
+    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<User> {
+        let db = Database::from_ref(state);
         if let Some(Ok(bot_token)) = parts.headers.get("x-bot-token").map(|v| v.to_str()) {
             let bot = db.fetch_bot_by_token(bot_token).await?;
             db.fetch_user(&bot.id).await

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

@@ -707,6 +707,7 @@ impl From<crate::FieldsMember> for FieldsMember {
             crate::FieldsMember::Nickname => FieldsMember::Nickname,
             crate::FieldsMember::Roles => FieldsMember::Roles,
             crate::FieldsMember::Timeout => FieldsMember::Timeout,
+            crate::FieldsMember::JoinedAt => FieldsMember::JoinedAt,
         }
     }
 }
@@ -718,6 +719,7 @@ impl From<FieldsMember> for crate::FieldsMember {
             FieldsMember::Nickname => crate::FieldsMember::Nickname,
             FieldsMember::Roles => crate::FieldsMember::Roles,
             FieldsMember::Timeout => crate::FieldsMember::Timeout,
+            FieldsMember::JoinedAt => crate::FieldsMember::JoinedAt,
         }
     }
 }

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

@@ -39,7 +39,9 @@ pub async fn load_fixture(db: &Database, input: &str) -> HashMap<String, String>
             LoadedFixture::User(user) => db.insert_user(&user).await.unwrap(),
             LoadedFixture::Channel(channel) => db.insert_channel(&channel).await.unwrap(),
             LoadedFixture::Server(server) => db.insert_server(&server).await.unwrap(),
-            LoadedFixture::ServerMember(member) => db.insert_member(&member).await.unwrap(),
+            LoadedFixture::ServerMember(member) => {
+                db.insert_or_merge_member(&member).await.unwrap();
+            }
         }
     }
 

crates/core/files/src/lib.rs

@@ -80,6 +80,9 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result<V
         .decrypt_in_place(nonce, b"", &mut buf)
         .map_err(|_| create_error!(InternalError))?;
 
+    // Remove the authentication tag bytes that were added during encryption
+    buf.truncate(buf.len() - AUTHENTICATION_TAG_SIZE_BYTES);
+
     Ok(buf)
 }
 

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

@@ -77,6 +77,7 @@ auto_derived!(
         Avatar,
         Roles,
         Timeout,
+        JoinedAt,
     }
 
     /// Member removal intention

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

@@ -255,12 +255,14 @@ auto_derived!(
         pub flags: i32,
     }
 
-    /// Mutual friends and servers response
+    /// Mutual friends, servers, groups and DMs response
     pub struct MutualResponse {
         /// Array of mutual user IDs that both users are friends with
         pub users: Vec<String>,
         /// Array of mutual server IDs that both users are in
         pub servers: Vec<String>,
+        /// Array of mutual group and dm IDs that both users are in
+        pub channels: Vec<String>,
     }
 
     /// Bot information for if the user is a bot

crates/core/parser/Cargo.toml

@@ -2,7 +2,8 @@
 name = "revolt-parser"
 version = "0.8.8"
 edition = "2021"
-license = "AGPL-3.0-or-later"
+license = "MIT"
+authors = ["Zomatree <me@zomatree.live>", "Paul Makles <me@insrt.uk>"]
 description = "Revolt Backend: Message Parser"
 
 [dependencies]

crates/core/parser/LICENSE

@@ -0,0 +1,9 @@
+MIT License
+
+Copyright (c) 2024 Pawel Makles
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

crates/core/parser/src/lib.rs

@@ -4,15 +4,18 @@ use logos::Logos;
 
 #[derive(Debug, Clone, Logos, PartialEq)]
 #[logos(skip "\n")]
-pub enum MessageToken {
+#[logos(subpattern id="[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}")]
+pub enum MessageToken<'a> {
     #[token("\\")]
     Escape,
-    #[regex("(```[^`\n]*)|(``)|`", |lex| lex.slice().to_owned().chars().filter(|&c| c == '`').count())]
+    #[regex("```[^`\n]*", |_| 3)]
+    #[regex("``", |_| 2)]
+    #[regex("`", |_| 1)]
     CodeblockMarker(usize),
-    #[regex("<@[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}>", |lex| lex.slice()[2..lex.slice().len() - 1].to_owned())]
-    UserMention(String),
-    #[regex("<%[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}>", |lex| lex.slice()[2..lex.slice().len() - 1].to_owned())]
-    RoleMention(String),
+    #[regex("<@(?&id)>", |lex| &lex.slice()[2..lex.slice().len() - 1])]
+    UserMention(&'a str),
+    #[regex("<%(?&id)>", |lex| &lex.slice()[2..lex.slice().len() - 1],)]
+    RoleMention(&'a str),
     #[token("@everyone")]
     MentionEveryone,
     #[token("@online")]
@@ -27,13 +30,13 @@ pub struct MessageResults {
     pub mentions_online: bool
 }
 
-struct MessageParserIterator<I> {
+struct MessageParserIterator<'a, I> {
     inner: I,
-    temp: VecDeque<MessageToken>
+    temp: VecDeque<MessageToken<'a>>
 }
 
-impl<I: Iterator<Item = MessageToken>> Iterator for MessageParserIterator<I> {
-    type Item = MessageToken;
+impl<'a, I: Iterator<Item = MessageToken<'a>>> Iterator for MessageParserIterator<'a, I> {
+    type Item = MessageToken<'a>;
 
     fn next(&mut self) -> Option<Self::Item> {
         if !self.temp.is_empty() {
@@ -80,8 +83,8 @@ pub fn parse_message(text: &str) -> MessageResults {
         match token {
             MessageToken::Escape => {}
             MessageToken::CodeblockMarker(_) => {},
-            MessageToken::UserMention(id) => { results.user_mentions.insert(id); },
-            MessageToken::RoleMention(id) => { results.role_mentions.insert(id); },
+            MessageToken::UserMention(id) => { results.user_mentions.insert(id.to_string()); },
+            MessageToken::RoleMention(id) => { results.role_mentions.insert(id.to_string()); },
             MessageToken::MentionEveryone => results.mentions_everyone = true,
             MessageToken::MentionOnline => results.mentions_online = true,
         };
@@ -106,7 +109,7 @@ mod tests {
         let output = parse_message_iter("Hello <@01FD58YK5W7QRV5H3D64KTQYX3>.").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 1);
-        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
     }
 
     #[test]
@@ -114,7 +117,7 @@ mod tests {
         let output = parse_message_iter("Hello <%01FD58YK5W7QRV5H3D64KTQYX3>.").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 1);
-        assert_eq!(output[0], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[0], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
     }
 
     #[test]
@@ -138,8 +141,8 @@ mod tests {
         let output = parse_message_iter("Hello <@01FD58YK5W7QRV5H3D64KTQYX3>, <%01FD58YK5W7QRV5H3D64KTQYX3>, @everyone and @online.").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 4);
-        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
-        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
         assert_eq!(output[2], MessageToken::MentionEveryone);
         assert_eq!(output[3], MessageToken::MentionOnline);
     }
@@ -149,8 +152,8 @@ mod tests {
         let output = parse_message_iter("<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 4);
-        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
-        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[0], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[1], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
         assert_eq!(output[2], MessageToken::MentionEveryone);
         assert_eq!(output[3], MessageToken::MentionOnline);
     }
@@ -170,15 +173,14 @@ mod tests {
 
         assert_eq!(output.len(), 5);
         assert_eq!(output[0], MessageToken::CodeblockMarker(3));
-        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
-        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
         assert_eq!(output[3], MessageToken::MentionEveryone);
         assert_eq!(output[4], MessageToken::MentionOnline);
     }
 
     #[test]
     fn test_inline_codeblock_no_mentions() {
-
         let output = parse_message_iter("`<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online`").collect::<Vec<_>>();
 
         assert_eq!(output.len(), 2);
@@ -192,12 +194,21 @@ mod tests {
 
         assert_eq!(output.len(), 5);
         assert_eq!(output[0], MessageToken::CodeblockMarker(1));
-        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
-        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3".to_string()));
+        assert_eq!(output[1], MessageToken::UserMention("01FD58YK5W7QRV5H3D64KTQYX3"));
+        assert_eq!(output[2], MessageToken::RoleMention("01FD58YK5W7QRV5H3D64KTQYX3"));
         assert_eq!(output[3], MessageToken::MentionEveryone);
         assert_eq!(output[4], MessageToken::MentionOnline);
     }
 
+    #[test]
+    fn test_codeblock_with_language_no_mentions() {
+        let output = parse_message_iter("```rust\n<@01FD58YK5W7QRV5H3D64KTQYX3><%01FD58YK5W7QRV5H3D64KTQYX3>@everyone@online```").collect::<Vec<_>>();
+
+        assert_eq!(output.len(), 2);
+        assert_eq!(output[0], MessageToken::CodeblockMarker(3));
+        assert_eq!(output[1], MessageToken::CodeblockMarker(3));
+    }
+
     #[test]
     fn test_double_inline_codeblock() {
         let output = parse_message_iter("``this should not ping @everyone``").collect::<Vec<_>>();

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

@@ -8,10 +8,18 @@ pub use server::*;
 pub use user::*;
 
 /// Holds a permission value to manipulate.
-#[derive(Clone, Debug)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub struct PermissionValue(u64);
 
 impl PermissionValue {
+    pub fn from_raw(value: u64) -> Self {
+        Self(value)
+    }
+
+    pub fn into_raw(self) -> u64 {
+        self.0
+    }
+
     /// Apply a given override to this value
     pub fn apply(&mut self, v: Override) {
         self.allow(v.allow);

crates/core/ratelimits/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "revolt-ratelimits"
+version = "0.8.8"
+edition = "2024"
+
+[features]
+rocket = ["dep:rocket", "dep:revolt_rocket_okapi", "revolt-database/rocket-impl"]
+axum = ["dep:axum", "revolt-database/axum-impl"]
+
+default = ["rocket", "axum"]
+
+[dependencies]
+revolt-database = { version = "0.8.8", path = "../database"}
+revolt-result = { version = "0.8.8", path = "../result" }
+revolt-config = { version = "0.8.8", path = "../config" }
+
+rocket = { version = "0.5.1", optional = true }
+revolt_rocket_okapi = { version = "0.10.0", optional = true }
+
+axum = { version = "0.7.5", optional = true, features = ["macros"] }
+
+serde = { version = "1", features = ["derive"] }
+authifier = { version = "1.0.15" }
+dashmap = "5.2.0"
+async-trait = "0.1.81"
+log = "0.4"

crates/core/ratelimits/src/axum.rs

@@ -0,0 +1,194 @@
+use std::net::SocketAddr;
+
+use async_trait::async_trait;
+use axum::{
+    Json, RequestPartsExt, Router,
+    body::Body,
+    extract::{ConnectInfo, FromRef, FromRequestParts, State},
+    http::{HeaderValue, Request, StatusCode, request::Parts},
+    middleware::Next,
+    response::{IntoResponse, Response},
+    routing::get,
+};
+use revolt_database::{Database, User};
+use revolt_config::config;
+
+use crate::ratelimiter::{RatelimitInformation, Ratelimiter, RequestKind};
+
+#[derive(Clone, Copy)]
+pub struct AxumRequestKind;
+
+impl RequestKind for AxumRequestKind {
+    type R<'a> = Parts;
+}
+
+pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<AxumRequestKind>;
+
+fn to_ip(parts: &Parts) -> String {
+    parts
+        .extensions
+        .get::<ConnectInfo<SocketAddr>>()
+        .map(|info| info.ip().to_string())
+        .unwrap_or_default()
+}
+
+async fn to_real_ip(parts: &Parts) -> String {
+    if config().await.api.security.trust_cloudflare {
+        parts
+            .headers
+            .get("CF-Connecting-IP")
+            .map(|x| x.to_str().unwrap().to_string())
+            .unwrap_or_else(|| to_ip(parts))
+    } else {
+        to_ip(parts)
+    }
+}
+
+#[async_trait]
+impl<S: Send + Sync> FromRequestParts<S> for Ratelimiter
+where
+    Database: FromRef<S>,
+    RatelimitStorage: FromRef<S>,
+{
+    type Rejection = Json<Ratelimiter>;
+
+    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
+        if parts
+            .extensions
+            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
+            .is_none()
+        {
+            let storage = RatelimitStorage::from_ref(state);
+
+            let identifier = if let Ok(user) = parts.extract_with_state::<User, _>(state).await {
+                user.id
+            } else {
+                to_real_ip(parts).await
+            };
+
+            let (bucket, resource) = storage.resolver.resolve_bucket(parts);
+            let limit = storage.resolver.resolve_bucket_limit(bucket);
+
+            let ratelimiter =
+                Ratelimiter::from(&storage.map, &identifier, limit, (bucket, resource));
+
+            parts.extensions.insert(ratelimiter.map_err(Json));
+        };
+
+        *parts
+            .extensions
+            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
+            .unwrap()
+    }
+}
+
+#[async_trait]
+impl<S: Send + Sync> FromRequestParts<S> for RatelimitInformation
+where
+    Database: FromRef<S>,
+    RatelimitStorage: FromRef<S>,
+{
+    type Rejection = Json<RatelimitInformation>;
+
+    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
+        if parts
+            .extensions
+            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
+            .is_none()
+        {
+            let ratelimiter = parts.extract_with_state::<Ratelimiter, S>(state).await;
+
+            parts.extensions.insert(ratelimiter);
+        };
+
+        let ratelimiter = *parts
+            .extensions
+            .get::<Result<Ratelimiter, Json<Ratelimiter>>>()
+            .unwrap();
+
+        match ratelimiter {
+            Ok(ratelimter) => Ok(RatelimitInformation::Success(ratelimter)),
+            Err(ratelimiter) => Err(Json(RatelimitInformation::Failure {
+                retry_after: ratelimiter.reset,
+            })),
+        }
+    }
+}
+
+pub async fn ratelimit_middleware(
+    State(database): State<Database>,
+    State(ratelimit_storage): State<RatelimitStorage>,
+    request: Request<Body>,
+    next: Next,
+) -> Response {
+    #[derive(axum::extract::FromRef)]
+    struct TempState {
+        database: Database,
+        ratelimit_storage: RatelimitStorage,
+    }
+
+    let state = TempState {
+        database,
+        ratelimit_storage,
+    };
+
+    let (mut parts, body) = request.into_parts();
+
+    let res = Ratelimiter::from_request_parts(&mut parts, &state).await;
+
+    let (Ok(ratelimiter) | Err(Json(ratelimiter))) = &res;
+
+    let mut response = if res.is_ok() {
+        let request = Request::from_parts(parts, body);
+
+        next.run(request).await
+    } else {
+        let ratelimit_info = RatelimitInformation::from_request_parts(&mut parts, &state).await;
+
+        ratelimit_info.map(Json).into_response()
+    };
+
+    let Ratelimiter {
+        key,
+        limit,
+        remaining,
+        reset,
+    } = ratelimiter;
+
+    let headers = response.headers_mut();
+
+    headers.insert(
+        "X-RateLimit-Limit",
+        HeaderValue::from_str(&limit.to_string()).unwrap(),
+    );
+    headers.insert(
+        "X-RateLimit-Bucket",
+        HeaderValue::from_str(&key.to_string()).unwrap(),
+    );
+    headers.insert(
+        "X-RateLimit-Remaining",
+        HeaderValue::from_str(&remaining.to_string()).unwrap(),
+    );
+    headers.insert(
+        "X-RateLimit-Reset-After",
+        HeaderValue::from_str(&reset.to_string()).unwrap(),
+    );
+
+    if res.is_err() {
+        *response.status_mut() = StatusCode::TOO_MANY_REQUESTS;
+    };
+
+    response
+}
+
+async fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
+    Json(info)
+}
+
+pub fn routes<S: Clone + Send + Sync + 'static>() -> Router<S>
+where
+    Database: FromRef<S>,
+    RatelimitStorage: FromRef<S>,
+{
+    Router::new().route("/ratelimit", get(ratelimit_info))
+}

crates/core/ratelimits/src/lib.rs

@@ -0,0 +1,7 @@
+pub mod ratelimiter;
+
+#[cfg(feature = "rocket")]
+pub mod rocket;
+
+#[cfg(feature = "axum")]
+pub mod axum;

crates/core/ratelimits/src/ratelimiter.rs

@@ -0,0 +1,145 @@
+use std::collections::hash_map::DefaultHasher;
+use std::hash::Hasher;
+use std::ops::Add;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use serde::Serialize;
+
+use dashmap::DashMap;
+
+pub trait RequestKind {
+    type R<'a>;
+}
+
+pub trait RatelimitResolver<R>: Send + Sync {
+    fn resolve_bucket<'a>(&self, request: &'a R) -> (&'a str, Option<&'a str>);
+    fn resolve_bucket_limit(&self, bucket: &str) -> u32;
+}
+
+#[derive(Clone)]
+pub struct RatelimitStorage<K: RequestKind> {
+    pub resolver: Arc<dyn for<'a> RatelimitResolver<K::R<'a>>>,
+    pub map: Arc<DashMap<u64, Entry>>,
+}
+
+impl<K: RequestKind> RatelimitStorage<K> {
+    pub fn new<R: for<'a> RatelimitResolver<K::R<'a>> + 'static>(resolver: R) -> Self {
+        Self {
+            resolver: Arc::new(resolver),
+            map: Arc::new(DashMap::new()),
+        }
+    }
+}
+
+/// Ratelimit Bucket
+#[derive(Clone, Copy, Debug)]
+pub struct Entry {
+    used: u32,
+    reset: u128,
+}
+
+/// Get the current time from Unix Epoch as a Duration
+fn now() -> Duration {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("Time went backwards...")
+}
+
+impl Entry {
+    /// Find bucket by its key
+    pub fn from(map: &DashMap<u64, Entry>, key: u64) -> Entry {
+        map.get(&key).map(|x| *x).unwrap_or_else(|| Entry {
+            used: 0,
+            reset: now().add(Duration::from_secs(10)).as_millis(),
+        })
+    }
+
+    /// Deduct one unit from the bucket and save
+    pub fn deduct(&mut self) {
+        let current_time = now().as_millis();
+        if current_time > self.reset {
+            self.used = 1;
+            self.reset = now().add(Duration::from_secs(10)).as_millis();
+        } else {
+            self.used += 1;
+        }
+    }
+
+    /// Save information
+    pub fn save(self, map: &DashMap<u64, Entry>, key: u64) {
+        map.insert(key, self);
+    }
+
+    /// Get remaining units in the bucket
+    pub fn get_remaining(&self, limit: u32) -> u32 {
+        if now().as_millis() > self.reset {
+            limit
+        } else {
+            limit - self.used
+        }
+    }
+
+    /// Get how long bucket has until reset
+    pub fn left_until_reset(&self) -> u128 {
+        let current_time = now().as_millis();
+        self.reset.saturating_sub(current_time)
+    }
+}
+
+/// Ratelimit Guard
+#[derive(Serialize, Clone, Copy, Debug)]
+#[allow(dead_code)]
+pub struct Ratelimiter {
+    pub key: u64,
+    pub limit: u32,
+    pub remaining: u32,
+    pub reset: u128,
+}
+
+impl Ratelimiter {
+    /// Generate guard from identifier and target bucket
+    pub fn from(
+        map: &DashMap<u64, Entry>,
+        identifier: &str,
+        limit: u32,
+        (bucket, resource): (&str, Option<&str>),
+    ) -> Result<Ratelimiter, Ratelimiter> {
+        let mut key = DefaultHasher::new();
+        key.write(identifier.as_bytes());
+        key.write(bucket.as_bytes());
+
+        if let Some(id) = resource {
+            key.write(id.as_bytes());
+        }
+
+        let key = key.finish();
+        let mut entry = Entry::from(map, key);
+
+        let remaining = entry.get_remaining(limit);
+        let reset = entry.left_until_reset();
+        let mut ratelimiter = Ratelimiter {
+            key,
+            limit,
+            remaining,
+            reset,
+        };
+        if remaining == 0 {
+            return Err(ratelimiter);
+        }
+
+        entry.deduct();
+        entry.save(map, key);
+        ratelimiter.remaining -= 1;
+        ratelimiter.reset = entry.left_until_reset();
+
+        Ok(ratelimiter)
+    }
+}
+
+#[derive(Serialize)]
+#[serde(untagged)]
+pub enum RatelimitInformation {
+    Success(Ratelimiter),
+    Failure { retry_after: u128 },
+}

crates/core/ratelimits/src/rocket.rs

@@ -0,0 +1,163 @@
+use async_trait::async_trait;
+use log::info;
+use rocket::fairing::{Fairing, Info, Kind};
+use rocket::http::uri::Origin;
+use rocket::http::{Method, Status};
+use rocket::request::{FromRequest, Outcome};
+use rocket::serde::json::Json;
+use rocket::{Data, Request, Response, State};
+use revolt_config::config;
+
+use revolt_rocket_okapi::r#gen::OpenApiGenerator;
+use revolt_rocket_okapi::request::{OpenApiFromRequest, RequestHeaderInput};
+
+use authifier::models::Session;
+
+use crate::ratelimiter::RequestKind;
+use crate::ratelimiter::{RatelimitInformation, Ratelimiter};
+
+#[derive(Clone, Copy)]
+pub struct RocketRequestKind;
+
+impl RequestKind for RocketRequestKind {
+    type R<'a> = Request<'a>;
+}
+
+pub type RatelimitStorage = crate::ratelimiter::RatelimitStorage<RocketRequestKind>;
+
+/// Find the remote IP of the client
+fn to_ip(request: &'_ rocket::Request<'_>) -> String {
+    request
+        .remote()
+        .map(|x| x.ip().to_string())
+        .unwrap_or_default()
+}
+
+/// Find the actual IP of the client
+async fn to_real_ip(request: &'_ rocket::Request<'_>) -> String {
+    if config().await.api.security.trust_cloudflare {
+        request
+            .headers()
+            .get_one("CF-Connecting-IP")
+            .map(|x| x.to_string())
+            .unwrap_or_else(|| to_ip(request))
+    } else {
+        to_ip(request)
+    }
+}
+
+#[async_trait]
+impl<'r> FromRequest<'r> for Ratelimiter {
+    type Error = Ratelimiter;
+
+    async fn from_request<'a>(request: &'r rocket::Request<'a>) -> Outcome<Self, Self::Error> {
+        let ratelimiter = request
+            .local_cache_async(async {
+                use rocket::outcome::Outcome;
+
+                let storage = request.guard::<&State<RatelimitStorage>>().await.unwrap();
+
+                let identifier = if let Outcome::Success(session) = request.guard::<Session>().await
+                {
+                    session.id
+                } else {
+                    to_real_ip(request).await
+                };
+
+                let (bucket, resource) = storage.resolver.resolve_bucket(request);
+                let limit = storage.resolver.resolve_bucket_limit(bucket);
+
+                Ratelimiter::from(&storage.map, &identifier, limit, (bucket, resource))
+            })
+            .await;
+
+        match ratelimiter {
+            Ok(ratelimiter) => Outcome::Success(*ratelimiter),
+            Err(ratelimiter) => Outcome::Error((Status::TooManyRequests, *ratelimiter)),
+        }
+    }
+}
+
+impl OpenApiFromRequest<'_> for Ratelimiter {
+    fn from_request_input(
+        _gen: &mut OpenApiGenerator,
+        _name: String,
+        _required: bool,
+    ) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
+        Ok(RequestHeaderInput::None)
+    }
+}
+
+/// Attach ratelimiter to the Rocket application
+pub struct RatelimitFairing;
+
+#[async_trait]
+impl Fairing for RatelimitFairing {
+    fn info(&self) -> Info {
+        Info {
+            name: "Ratelimiter",
+            kind: Kind::Request | Kind::Response,
+        }
+    }
+
+    async fn on_request(&self, request: &mut Request<'_>, _: &mut Data<'_>) {
+        use rocket::outcome::Outcome;
+        if let Outcome::Error(_) = request.guard::<Ratelimiter>().await {
+            info!(
+                "User rate-limited on route {}! (IP = {:?})",
+                request.uri(),
+                to_real_ip(request).await
+            );
+
+            request.set_method(Method::Get);
+            request.set_uri(Origin::parse("/ratelimit").unwrap())
+        }
+    }
+
+    async fn on_response<'r>(&self, request: &'r Request<'_>, response: &mut Response<'r>) {
+        let guard = request.guard::<Ratelimiter>().await;
+        let (Outcome::Success(ratelimiter) | Outcome::Error((_, ratelimiter))) = guard else {
+            unreachable!()
+        };
+        let Ratelimiter {
+            key,
+            limit,
+            remaining,
+            reset,
+        } = ratelimiter;
+
+        response.set_raw_header("X-RateLimit-Limit", limit.to_string());
+        response.set_raw_header("X-RateLimit-Bucket", key.to_string());
+        response.set_raw_header("X-RateLimit-Remaining", remaining.to_string());
+        response.set_raw_header("X-RateLimit-Reset-After", reset.to_string());
+
+        if guard.is_error() {
+            response.set_status(Status::TooManyRequests);
+        }
+    }
+}
+
+#[async_trait]
+impl<'r> FromRequest<'r> for RatelimitInformation {
+    type Error = u128;
+
+    async fn from_request(request: &'r rocket::Request<'_>) -> Outcome<Self, Self::Error> {
+        let info = match request.guard::<Ratelimiter>().await {
+            Outcome::Success(ratelimiter) => RatelimitInformation::Success(ratelimiter),
+            Outcome::Error((_, ratelimiter)) => RatelimitInformation::Failure {
+                retry_after: ratelimiter.reset,
+            },
+            _ => unreachable!(),
+        };
+        Outcome::Success(info)
+    }
+}
+
+#[rocket::get("/ratelimit")]
+fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
+    Json(info)
+}
+
+pub fn routes() -> Vec<rocket::Route> {
+    rocket::routes![ratelimit_info]
+}

crates/daemons/crond/src/main.rs

@@ -1,7 +1,7 @@
 use revolt_config::configure;
 use revolt_database::DatabaseInfo;
 use revolt_result::Result;
-use tasks::{file_deletion, prune_dangling_files};
+use tasks::{file_deletion, prune_dangling_files, prune_members};
 use tokio::try_join;
 
 pub mod tasks;
@@ -13,7 +13,8 @@ async fn main() -> Result<()> {
     let db = DatabaseInfo::Auto.connect().await.expect("database");
     try_join!(
         file_deletion::task(db.clone()),
-        prune_dangling_files::task(db)
+        prune_dangling_files::task(db.clone()),
+        prune_members::task(db.clone())
     )
     .map(|_| ())
 }

crates/daemons/crond/src/tasks/mod.rs

@@ -1,2 +1,3 @@
 pub mod file_deletion;
 pub mod prune_dangling_files;
+pub mod prune_members;

crates/daemons/crond/src/tasks/prune_members.rs

@@ -0,0 +1,18 @@
+use std::time::Duration;
+
+use log::warn;
+use revolt_database::Database;
+use revolt_result::Result;
+use tokio::time::sleep;
+
+pub async fn task(db: Database) -> Result<()> {
+    loop {
+        let success = db.remove_dangling_members().await;
+        if let Err(s) = success {
+            revolt_config::capture_error(&s);
+            warn!("Failed to prune dangling members: {:?}", &s);
+        }
+
+        sleep(Duration::from_secs(90)).await;
+    }
+}

crates/delta/Cargo.toml

@@ -81,6 +81,7 @@ revolt-models = { path = "../core/models", features = [
 revolt-presence = { path = "../core/presence" }
 revolt-result = { path = "../core/result", features = ["rocket", "okapi"] }
 revolt-permissions = { path = "../core/permissions", features = ["schemas"] }
+revolt-ratelimits = { path = "../core/ratelimits", features = ["rocket"] }
 
 [build-dependencies]
 vergen = "7.5.0"

crates/delta/src/main.rs

@@ -11,6 +11,7 @@ pub mod util;
 use revolt_config::config;
 use revolt_database::events::client::EventV1;
 use revolt_database::AMQP;
+use revolt_ratelimits::rocket as ratelimiter;
 use rocket::{Build, Rocket};
 use rocket_cors::{AllowedOrigins, CorsOptions};
 use rocket_prometheus::PrometheusMetrics;
@@ -122,18 +123,22 @@ pub async fn web() -> Rocket<Build> {
     let rocket = rocket::build();
     let prometheus = PrometheusMetrics::new();
 
+    // Ratelimits
+    let ratelimits = ratelimiter::RatelimitStorage::new(util::ratelimits::DeltaRatelimits);
+
     routes::mount(config, rocket)
         .attach(prometheus.clone())
         .mount("/metrics", prometheus)
         .mount("/", rocket_cors::catch_all_options_routes())
-        .mount("/", util::ratelimiter::routes())
+        .mount("/", ratelimiter::routes())
         .mount("/swagger/", swagger)
         .mount("/0.8/swagger/", swagger_0_8)
         .manage(authifier)
         .manage(db)
         .manage(amqp)
         .manage(cors.clone())
-        .attach(util::ratelimiter::RatelimitFairing)
+        .manage(ratelimits)
+        .attach(ratelimiter::RatelimitFairing)
         .attach(cors)
         .configure(rocket::Config {
             limits: rocket::data::Limits::default().limit("string", 5.megabytes()),

crates/delta/src/routes/bots/invite.rs

@@ -168,8 +168,8 @@ mod test {
             .await;
 
         match event {
-            EventV1::ServerMemberJoin { user, .. } => {
-                assert_eq!(bot.id, user);
+            EventV1::ServerMemberJoin { member, .. } => {
+                assert_eq!(bot.id, member.id.user);
             }
             _ => unreachable!(),
         }

crates/delta/src/routes/users/find_mutual.rs

@@ -8,9 +8,9 @@ use revolt_result::{create_error, Result};
 use rocket::serde::json::Json;
 use rocket::State;
 
-/// # Fetch Mutual Friends And Servers
+/// # Fetch Mutual Friends, Servers, Groups and DMs
 ///
-/// Retrieve a list of mutual friends and servers with another user.
+/// Retrieve a list of mutual friends, servers, groups and DMs with another user.
 #[openapi(tag = "Relationships")]
 #[get("/<target>/mutual")]
 pub async fn mutual(
@@ -32,5 +32,6 @@ pub async fn mutual(
     Ok(Json(v0::MutualResponse {
         users: db.fetch_mutual_user_ids(&user.id, &target.id).await?,
         servers: db.fetch_mutual_server_ids(&user.id, &target.id).await?,
+        channels: db.fetch_mutual_channel_ids(&user.id, &target.id).await?,
     }))
 }

crates/delta/src/util/mod.rs

@@ -1,2 +1,2 @@
-pub mod ratelimiter;
+pub mod ratelimits;
 pub mod test;

crates/delta/src/util/ratelimiter.rs

@@ -1,344 +0,0 @@
-use std::collections::hash_map::DefaultHasher;
-use std::hash::Hasher;
-use std::ops::Add;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
-
-use authifier::models::Session;
-use rocket::fairing::{Fairing, Info, Kind};
-use rocket::http::uri::Origin;
-use rocket::http::{Method, Status};
-use rocket::request::{FromRequest, Outcome};
-use rocket::serde::json::Json;
-use rocket::{Data, Request, Response};
-
-use revolt_rocket_okapi::gen::OpenApiGenerator;
-use revolt_rocket_okapi::request::{OpenApiFromRequest, RequestHeaderInput};
-
-use serde::Serialize;
-
-use dashmap::DashMap;
-use once_cell::sync::Lazy;
-
-/// Ratelimit Bucket
-#[derive(Clone, Copy, Debug)]
-struct Entry {
-    used: u8,
-    reset: u128,
-}
-
-static MAP: Lazy<DashMap<u64, Entry>> = Lazy::new(DashMap::new);
-
-/// Get the current time from Unix Epoch as a Duration
-fn now() -> Duration {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .expect("Time went backwards...")
-}
-
-impl Entry {
-    /// Find bucket by its key
-    pub fn from(key: u64) -> Entry {
-        MAP.get(&key).map(|x| *x).unwrap_or_else(|| Entry {
-            used: 0,
-            reset: now().add(Duration::from_secs(10)).as_millis(),
-        })
-    }
-
-    /// Deduct one unit from the bucket and save
-    pub fn deduct(&mut self) {
-        let current_time = now().as_millis();
-        if current_time > self.reset {
-            self.used = 1;
-            self.reset = now().add(Duration::from_secs(10)).as_millis();
-        } else {
-            self.used += 1;
-        }
-    }
-
-    /// Save information
-    pub fn save(self, key: u64) {
-        MAP.insert(key, self);
-    }
-
-    /// Get remaining units in the bucket
-    pub fn get_remaining(&self, limit: u8) -> u8 {
-        if now().as_millis() > self.reset {
-            limit
-        } else {
-            limit - self.used
-        }
-    }
-
-    /// Get how long bucket has until reset
-    pub fn left_until_reset(&self) -> u128 {
-        let current_time = now().as_millis();
-        if current_time > self.reset {
-            0
-        } else {
-            self.reset - current_time
-        }
-    }
-}
-
-/// Ratelimit Guard
-#[derive(Serialize, Clone, Copy, Debug)]
-#[allow(dead_code)]
-pub struct Ratelimiter {
-    key: u64,
-    limit: u8,
-    remaining: u8,
-    reset: u128,
-}
-
-/// Find bucket from given request
-///
-/// Optionally, include a resource id to hash against.
-fn resolve_bucket<'r>(request: &'r rocket::Request<'_>) -> (&'r str, Option<&'r str>) {
-    let (segment, resource, extra) = if matches!(request.routed_segment(0), Some("0.8")) {
-        (
-            request.routed_segment(1),
-            request.routed_segment(2),
-            request.routed_segment(3),
-        )
-    } else {
-        (
-            request.routed_segment(0),
-            request.routed_segment(1),
-            request.routed_segment(2),
-        )
-    };
-
-    if let Some(segment) = segment {
-        let method = request.method();
-        match (segment, resource, method) {
-            ("users", target, Method::Patch) => ("user_edit", target),
-            ("users", _, _) => {
-                if let Some("default_avatar") = extra {
-                    return ("default_avatar", None);
-                }
-
-                ("users", None)
-            }
-            ("bots", _, _) => ("bots", None),
-            ("channels", Some(id), _) => {
-                if request.method() == Method::Post {
-                    if let Some("messages") = extra {
-                        return ("messaging", Some(id));
-                    }
-                }
-
-                ("channels", Some(id))
-            }
-            ("servers", Some(id), _) => ("servers", Some(id)),
-            ("auth", _, _) => {
-                if request.method() == Method::Delete {
-                    ("auth_delete", None)
-                } else {
-                    ("auth", None)
-                }
-            }
-            ("swagger", _, _) => ("swagger", None),
-            ("safety", Some("report"), _) => ("safety_report", Some("report")),
-            ("safety", _, _) => ("safety", None),
-            _ => ("any", None),
-        }
-    } else {
-        ("any", None)
-    }
-}
-
-/// Resolve per-bucket limits
-fn resolve_bucket_limit(bucket: &str) -> u8 {
-    match bucket {
-        "user_edit" => 2,
-        "users" => 20,
-        "bots" => 10,
-        "messaging" => 10,
-        "channels" => 15,
-        "servers" => 5,
-        "auth" => 15,
-        "auth_delete" => 255,
-        "default_avatar" => 255,
-        "swagger" => 100,
-        "safety" => 15,
-        "safety_report" => 3,
-        _ => 20,
-    }
-}
-
-/// Find the remote IP of the client
-fn to_ip(request: &'_ rocket::Request<'_>) -> String {
-    request
-        .remote()
-        .map(|x| x.ip().to_string())
-        .unwrap_or_default()
-}
-
-/// Find the actual IP of the client
-fn to_real_ip(request: &'_ rocket::Request<'_>) -> String {
-    if let Ok(true) = std::env::var("TRUST_CLOUDFLARE").map(|x| x == "1") {
-        request
-            .headers()
-            .get_one("CF-Connecting-IP")
-            .map(|x| x.to_string())
-            .unwrap_or_else(|| to_ip(request))
-    } else {
-        to_ip(request)
-    }
-}
-
-impl Ratelimiter {
-    /// Generate guard from identifier and target bucket
-    pub fn from(
-        identifier: &str,
-        (bucket, resource): (&str, Option<&str>),
-    ) -> Result<Ratelimiter, Ratelimiter> {
-        let mut key = DefaultHasher::new();
-        key.write(identifier.as_bytes());
-        key.write(bucket.as_bytes());
-
-        if let Some(id) = resource {
-            key.write(id.as_bytes());
-        }
-
-        let key = key.finish();
-        let limit = resolve_bucket_limit(bucket);
-        let mut entry = Entry::from(key);
-
-        let remaining = entry.get_remaining(limit);
-        let reset = entry.left_until_reset();
-        let mut ratelimiter = Ratelimiter {
-            key,
-            limit,
-            remaining,
-            reset,
-        };
-        if remaining == 0 {
-            return Err(ratelimiter);
-        }
-
-        entry.deduct();
-        entry.save(key);
-        ratelimiter.remaining -= 1;
-        ratelimiter.reset = entry.left_until_reset();
-
-        Ok(ratelimiter)
-    }
-}
-
-#[async_trait]
-impl<'r> FromRequest<'r> for Ratelimiter {
-    type Error = Ratelimiter;
-
-    async fn from_request(request: &'r rocket::Request<'_>) -> Outcome<Self, Self::Error> {
-        let ratelimiter = request
-            .local_cache_async(async {
-                use rocket::outcome::Outcome;
-                let identifier = if let Outcome::Success(session) = request.guard::<Session>().await
-                {
-                    session.id
-                } else {
-                    to_real_ip(request)
-                };
-
-                Ratelimiter::from(&identifier, resolve_bucket(request))
-            })
-            .await;
-
-        match ratelimiter {
-            Ok(ratelimiter) => Outcome::Success(*ratelimiter),
-            Err(ratelimiter) => Outcome::Error((Status::TooManyRequests, *ratelimiter)),
-        }
-    }
-}
-
-impl OpenApiFromRequest<'_> for Ratelimiter {
-    fn from_request_input(
-        _gen: &mut OpenApiGenerator,
-        _name: String,
-        _required: bool,
-    ) -> revolt_rocket_okapi::Result<RequestHeaderInput> {
-        Ok(RequestHeaderInput::None)
-    }
-}
-
-/// Attach ratelimiter to the Rocket application
-pub struct RatelimitFairing;
-
-#[rocket::async_trait]
-impl Fairing for RatelimitFairing {
-    fn info(&self) -> Info {
-        Info {
-            name: "Ratelimiter",
-            kind: Kind::Request | Kind::Response,
-        }
-    }
-
-    async fn on_request(&self, request: &mut Request<'_>, _: &mut Data<'_>) {
-        use rocket::outcome::Outcome;
-        if let Outcome::Error(_) = request.guard::<Ratelimiter>().await {
-            info!(
-                "User rate-limited on route {}! (IP = {:?})",
-                request.uri(),
-                to_real_ip(request)
-            );
-
-            request.set_method(Method::Get);
-            request.set_uri(Origin::parse("/ratelimit").unwrap())
-        }
-    }
-
-    async fn on_response<'r>(&self, request: &'r Request<'_>, response: &mut Response<'r>) {
-        let guard = request.guard::<Ratelimiter>().await;
-        let (Outcome::Success(ratelimiter) | Outcome::Error((_, ratelimiter))) = guard else {
-            unreachable!()
-        };
-        let Ratelimiter {
-            key,
-            limit,
-            remaining,
-            reset,
-        } = ratelimiter;
-
-        response.set_raw_header("X-RateLimit-Limit", limit.to_string());
-        response.set_raw_header("X-RateLimit-Bucket", key.to_string());
-        response.set_raw_header("X-RateLimit-Remaining", remaining.to_string());
-        response.set_raw_header("X-RateLimit-Reset-After", reset.to_string());
-
-        if guard.is_error() {
-            response.set_status(Status::TooManyRequests);
-        }
-    }
-}
-
-#[derive(Serialize)]
-#[serde(untagged)]
-pub enum RatelimitInformation {
-    Success(Ratelimiter),
-    Failure { retry_after: u128 },
-}
-
-#[async_trait]
-impl<'r> FromRequest<'r> for RatelimitInformation {
-    type Error = u128;
-
-    async fn from_request(request: &'r rocket::Request<'_>) -> Outcome<Self, Self::Error> {
-        let info = match request.guard::<Ratelimiter>().await {
-            Outcome::Success(ratelimiter) => RatelimitInformation::Success(ratelimiter),
-            Outcome::Error((_, ratelimiter)) => RatelimitInformation::Failure {
-                retry_after: ratelimiter.reset,
-            },
-            _ => unreachable!(),
-        };
-        Outcome::Success(info)
-    }
-}
-
-#[rocket::get("/ratelimit")]
-fn ratelimit_info(info: RatelimitInformation) -> Json<RatelimitInformation> {
-    Json(info)
-}
-
-pub fn routes() -> Vec<rocket::Route> {
-    rocket::routes![ratelimit_info]
-}

crates/delta/src/util/ratelimits.rs

@@ -0,0 +1,81 @@
+use revolt_ratelimits::ratelimiter::RatelimitResolver;
+use rocket::{http::Method, Request};
+
+pub struct DeltaRatelimits;
+
+impl<'a> RatelimitResolver<Request<'a>> for DeltaRatelimits {
+    fn resolve_bucket<'r>(&self, request: &'r Request<'_>) -> (&'r str, Option<&'r str>) {
+        let (segment, resource, extra) = if request.routed_segment(0) == Some("0.8") {
+            (
+                request.routed_segment(1),
+                request.routed_segment(2),
+                request.routed_segment(3),
+            )
+        } else {
+            (
+                request.routed_segment(0),
+                request.routed_segment(1),
+                request.routed_segment(2),
+            )
+        };
+
+        if let Some(segment) = segment {
+            #[allow(clippy::redundant_locals)]
+            let resource = resource;
+
+            let method = request.method();
+            match (segment, resource, method) {
+                ("users", target, Method::Patch) => ("user_edit", target),
+                ("users", _, _) => {
+                    if let Some("default_avatar") = extra {
+                        return ("default_avatar", None);
+                    }
+
+                    ("users", None)
+                }
+                ("bots", _, _) => ("bots", None),
+                ("channels", Some(id), _) => {
+                    if request.method() == Method::Post {
+                        if let Some("messages") = extra {
+                            return ("messaging", Some(id));
+                        }
+                    }
+
+                    ("channels", Some(id))
+                }
+                ("servers", Some(id), _) => ("servers", Some(id)),
+                ("auth", _, _) => {
+                    if request.method() == Method::Delete {
+                        ("auth_delete", None)
+                    } else {
+                        ("auth", None)
+                    }
+                }
+                ("swagger", _, _) => ("swagger", None),
+                ("safety", Some("report"), _) => ("safety_report", Some("report")),
+                ("safety", _, _) => ("safety", None),
+                _ => ("any", None),
+            }
+        } else {
+            ("any", None)
+        }
+    }
+
+    fn resolve_bucket_limit(&self, bucket: &str) -> u32 {
+        match bucket {
+            "user_edit" => 2,
+            "users" => 20,
+            "bots" => 10,
+            "messaging" => 10,
+            "channels" => 15,
+            "servers" => 5,
+            "auth" => 15,
+            "auth_delete" => 255,
+            "default_avatar" => 255,
+            "swagger" => 100,
+            "safety" => 15,
+            "safety_report" => 3,
+            _ => 20,
+        }
+    }
+}

crates/services/autumn/Cargo.toml

@@ -52,6 +52,7 @@ revolt-result = { version = "0.8.8", path = "../../core/result", features = [
     "utoipa",
     "axum",
 ] }
+revolt-ratelimits = { version = "0.8.8", path = "../../core/ratelimits", features = ["axum"] }
 
 # Axum / web server
 tempfile = "3.12.0"

crates/services/autumn/Dockerfile

@@ -1,11 +1,13 @@
 # Build Stage
 FROM ghcr.io/revoltchat/base:latest AS builder
+FROM debian:12 AS debian
 
 # Bundle Stage
 FROM gcr.io/distroless/cc-debian12:nonroot
 COPY --from=builder /home/rust/src/target/release/revolt-autumn ./
 COPY --from=mwader/static-ffmpeg:7.0.2 /ffmpeg /usr/local/bin/
 COPY --from=mwader/static-ffmpeg:7.0.2 /ffprobe /usr/local/bin/
+COPY --from=debian /usr/bin/uname /usr/bin/uname
 
 EXPOSE 14704
 USER nonroot

crates/services/autumn/src/api.rs

@@ -25,10 +25,10 @@ use tokio::time::Instant;
 use tower_http::cors::{AllowHeaders, Any, CorsLayer};
 use utoipa::ToSchema;
 
-use crate::{exif::strip_metadata, metadata::generate_metadata, mime_type::determine_mime_type};
+use crate::{exif::strip_metadata, metadata::generate_metadata, mime_type::determine_mime_type, AppState};
 
 /// Build the API router
-pub async fn router() -> Router<Database> {
+pub async fn router() -> Router<AppState> {
     let config = config().await;
 
     let cors = CorsLayer::new()

crates/services/autumn/src/main.rs

@@ -1,8 +1,10 @@
 use std::net::{Ipv4Addr, SocketAddr};
 
-use axum::Router;
+use axum::{middleware::from_fn_with_state, Router};
 
-use revolt_database::DatabaseInfo;
+use axum_macros::FromRef;
+use revolt_database::{Database, DatabaseInfo};
+use revolt_ratelimits::axum as ratelimiter;
 use tokio::net::TcpListener;
 use utoipa::{
     openapi::security::{ApiKey, ApiKeyValue, SecurityScheme},
@@ -15,6 +17,13 @@ pub mod clamav;
 pub mod exif;
 pub mod metadata;
 pub mod mime_type;
+mod ratelimits;
+
+#[derive(FromRef, Clone)]
+struct AppState {
+    database: Database,
+    ratelimit_storage: ratelimiter::RatelimitStorage,
+}
 
 #[tokio::main]
 async fn main() -> Result<(), std::io::Error> {
@@ -69,12 +78,23 @@ async fn main() -> Result<(), std::io::Error> {
 
     // Connect to the database
     let db = DatabaseInfo::Auto.connect().await.unwrap();
+    let ratelimits = ratelimiter::RatelimitStorage::new(ratelimits::AutumnRatelimits);
+
+    let state = AppState {
+        database: db,
+        ratelimit_storage: ratelimits,
+    };
 
     // Configure Axum and router
     let app = Router::new()
         .merge(Scalar::with_url("/scalar", ApiDoc::openapi()))
         .nest("/", api::router().await)
-        .with_state(db);
+        .nest("/", ratelimiter::routes())
+        .layer(from_fn_with_state(
+            state.clone(),
+            ratelimiter::ratelimit_middleware,
+        ))
+        .with_state(state);
 
     // Configure TCP listener and bind
     let address = SocketAddr::from((Ipv4Addr::UNSPECIFIED, 14704));

crates/services/autumn/src/ratelimits.rs

@@ -0,0 +1,28 @@
+use axum::http::{request::Parts, Method};
+use revolt_ratelimits::ratelimiter::RatelimitResolver;
+
+pub struct AutumnRatelimits;
+
+impl RatelimitResolver<Parts> for AutumnRatelimits {
+    fn resolve_bucket<'a>(&self, parts: &'a Parts) -> (&'a str, Option<&'a str>) {
+        let path = parts
+            .uri
+            .path()
+            .trim_matches('/')
+            .split_terminator("/")
+            .collect::<Vec<&str>>();
+
+        match (&parts.method, path.as_slice()) {
+            (&Method::POST, &[tag]) => ("upload", Some(tag)),
+            _ => ("any", None),
+        }
+    }
+
+    fn resolve_bucket_limit(&self, bucket: &str) -> u32 {
+        match bucket {
+            "upload" => 10,
+            "any" => u32::MAX,
+            _ => unreachable!("Bucket defined but no limit set"),
+        }
+    }
+}

crates/services/january/Dockerfile

@@ -1,11 +1,13 @@
 # Build Stage
 FROM ghcr.io/revoltchat/base:latest AS builder
+FROM debian:12 AS debian
 
 # Bundle Stage
 FROM gcr.io/distroless/cc-debian12:nonroot
 COPY --from=builder /home/rust/src/target/release/revolt-january ./
 COPY --from=mwader/static-ffmpeg:7.0.2 /ffmpeg /usr/local/bin/
 COPY --from=mwader/static-ffmpeg:7.0.2 /ffprobe /usr/local/bin/
+COPY --from=debian /usr/bin/uname /usr/bin/uname
 
 EXPOSE 14705
 USER nonroot