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

crates/core/coalesced/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "revolt-coalesced"
+version = "0.8.8"
+edition = "2021"
+license = "MIT"
+authors = ["Paul Makles <me@insrt.uk>", "Zomatree <me@zomatree.live>"]
+description = "Revolt Backend: Coalescion service"
+
+[features]
+tokio = ["dep:tokio"]
+queue = ["dep:indexmap"]
+cache = ["dep:lru"]
+
+default = ["tokio"]
+
+[dependencies]
+tokio = { version = "1.47.0", features = ["sync"], optional = true }
+indexmap = { version = "*", optional = true }
+lru = { version = "*", optional = true }
+
+[dev-dependencies]
+tokio = { version = "1.47.0", features = ["rt", "rt-multi-thread", "macros", "time"] }

crates/core/coalesced/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/coalesced/src/config.rs

@@ -0,0 +1,24 @@
+#[derive(Clone, PartialEq, Eq, Debug)]
+/// Config values for [`CoalescionService`].
+pub struct CoalescionServiceConfig {
+    /// How many tasks are running at once
+    pub max_concurrent: Option<usize>,
+    /// Whether to queue tasks once `max_concurrent` is reached
+    #[cfg(feature = "queue")]
+    pub queue_requests: bool,
+    /// Max amount of tasks in the buffer queue
+    #[cfg(feature = "queue")]
+    pub max_queue: Option<usize>,
+}
+
+impl Default for CoalescionServiceConfig {
+    fn default() -> Self {
+        Self {
+            max_concurrent: Some(100),
+            #[cfg(feature = "queue")]
+            queue_requests: true,
+            #[cfg(feature = "queue")]
+            max_queue: Some(100)
+        }
+    }
+}

crates/core/coalesced/src/error.rs

@@ -0,0 +1,27 @@
+use std::fmt;
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
+/// Coalescion service error.
+pub enum Error {
+    /// Failed to receive the actions return from the channel for unknown reason
+    RecvError,
+    /// Reached the `max_concurrent` amount of actions running at once and could not queue the action
+    MaxConcurrent,
+    /// Reached the `max_queue` amount of actions in the queue
+    MaxQueue,
+    /// Failed to downcast the type to the current type being returned, this will be most likely an ID collision
+    DowncastError,
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Error::RecvError => write!(f, "Unable to receive data from the channel"),
+            Error::MaxConcurrent => write!(f, "Max number of tasks running at once"),
+            Error::MaxQueue => write!(f, "Max number of tasks in queue"),
+            Error::DowncastError => write!(f, "Failed to downcast type, possible key collision with different types")
+        }
+    }
+}
+
+impl std::error::Error for Error {}

crates/core/coalesced/src/lib.rs

@@ -0,0 +1,39 @@
+//! # Coalesced
+//!
+//! Coalescion service to group, caching and queue duplicate actions.
+//! useful for deduplicating web requests, database lookups and other similar resource
+//! intensive or rate-limited actions.
+//!
+//! ## Features
+//! - `tokio`: Uses tokio for the async backend, this is currently the only backend.
+//! - `queue`: Whether to support queueing requests to only allow X amount of actions running at once.
+//! - `cache`: Whether to cache the actions results for future actions with the same id, uses an LRU cache internally.
+//!
+//! [`CoalescionService`] uses both [`Arc`] and [`RwLock`] internally and can be cheaply cloned to
+//! use in your codebase.
+//!
+//! It is common practice to wrap the service and in your own which delegates the executions to ensure all ids are tracked in one location across your codebase.
+//!
+//! All values are stored using [`Any`] and must be [`'static`] + [`Send`] + [`Sync`], if there is an id mismatch
+//! and a type is wrong the library will return an error, values returned from the service are also
+//! wrapped in an [`Arc`] as they are shared to each duplicate action.
+//!
+//! ## Example:
+//! ```rs
+//! use revolt_coalesced::CoalescionService;
+//!
+//! let service = CoalescionService::new();
+//!
+//! let user_id = "my_user_id";
+//! let user = service.execute(user_id, || async move {
+//!     database.fetch_user(user_id).await.unwrap()
+//! }).await;
+//! ```
+
+mod config;
+mod error;
+mod service;
+
+pub use config::CoalescionServiceConfig;
+pub use error::Error;
+pub use service::CoalescionService;

crates/core/coalesced/src/service.rs

@@ -0,0 +1,208 @@
+use std::{any::Any, collections::HashMap, fmt::Debug, future::Future, hash::Hash, sync::Arc};
+
+use tokio::sync::{
+    watch::{channel as watch_channel, Receiver},
+    RwLock,
+};
+
+#[cfg(feature = "cache")]
+use lru::LruCache;
+
+#[cfg(feature = "queue")]
+use indexmap::IndexMap;
+
+use crate::{CoalescionServiceConfig, Error};
+
+#[derive(Debug, Clone)]
+#[allow(clippy::type_complexity)]
+/// # Coalescion service
+///
+/// See module description for example usage.
+pub struct CoalescionService<Id: Hash + Clone + Eq> {
+    config: Arc<CoalescionServiceConfig>,
+    watchers: Arc<RwLock<HashMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
+    #[cfg(feature = "queue")]
+    queue: Arc<RwLock<IndexMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
+    #[cfg(feature = "cache")]
+    cache: Option<Arc<tokio::sync::Mutex<LruCache<Id, Arc<dyn Any + Send + Sync>>>>>,
+}
+
+impl<Id: Hash + Clone + Eq> CoalescionService<Id> {
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    pub fn from_config(config: CoalescionServiceConfig) -> Self {
+        Self {
+            config: Arc::new(config),
+            watchers: Arc::new(RwLock::new(HashMap::new())),
+            #[cfg(feature = "queue")]
+            queue: Arc::new(RwLock::new(IndexMap::new())),
+            #[cfg(feature = "cache")]
+            cache: None,
+        }
+    }
+
+    #[cfg(feature = "cache")]
+    pub fn from_cache(
+        config: CoalescionServiceConfig,
+        cache: LruCache<Id, Arc<dyn Any + Send + Sync>>,
+    ) -> Self {
+        Self {
+            cache: Some(Arc::new(Mutex::new(cache))),
+            ..Self::from_config(config)
+        }
+    }
+
+    async fn wait_for<Value: Any + Send + Sync>(
+        &self,
+        mut receiver: Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>,
+    ) -> Result<Arc<Value>, Error> {
+        receiver
+            .wait_for(|v| v.is_some())
+            .await
+            .map_err(|_| Error::RecvError)
+            .and_then(|r| r.clone().unwrap())
+            .and_then(|arc| Arc::downcast(arc).map_err(|_| Error::DowncastError))
+    }
+
+    async fn insert_and_execute<
+        Value: Send + Sync + 'static,
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Value>,
+    >(
+        &self,
+        id: Id,
+        func: F,
+    ) -> Result<Arc<Value>, Error> {
+        let (send, recv) = watch_channel(None);
+
+        self.watchers.write().await.insert(id.clone(), recv);
+
+        let value = Ok(Arc::new(func().await));
+
+        send.send_modify(|opt| {
+            opt.replace(value.clone().map(|v| v as Arc<dyn Any + Send + Sync>));
+        });
+
+        #[cfg(feature = "cache")]
+        if let Some(cache) = self.cache.as_ref() {
+            if let Ok(value) = &value {
+                cache.lock().await.push(id.clone(), value.clone());
+            }
+        };
+
+        self.watchers.write().await.remove(&id);
+
+        value
+    }
+
+    /// Coalesces an function, the actual function may not run if one with the same id is already running,
+    /// queued to be ran, or cached, the id should be globally unique for this specific action.
+    pub async fn execute<
+        Value: Send + Sync + 'static,
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Value>,
+    >(
+        &self,
+        id: Id,
+        func: F,
+    ) -> Result<Arc<Value>, Error> {
+        #[cfg(feature = "cache")]
+        if let Some(cache) = self.cache.as_ref() {
+            if let Some(value) = cache.lock().await.get(&id) {
+                return Arc::downcast::<Value>(value.clone()).map_err(|_| Error::DowncastError);
+            }
+        };
+
+        let (receiver, length) = {
+            let watchers = self.watchers.read().await;
+            let length = watchers.len();
+
+            (watchers.get(&id).cloned(), length)
+        };
+
+        if let Some(receiver) = receiver {
+            self.wait_for(receiver).await
+        } else {
+            match self.config.max_concurrent {
+                Some(max_concurrent) if length >= max_concurrent => {
+                    #[cfg(feature = "queue")]
+                    if self.config.queue_requests {
+                        let (receiver, length) = {
+                            let queue = self.queue.read().await;
+
+                            (queue.get(&id).cloned(), queue.len())
+                        };
+
+                        if let Some(receiver) = receiver {
+                            return self.wait_for(receiver).await;
+                        } else {
+                            if self
+                                .config
+                                .max_queue
+                                .is_some_and(|max_queue| max_queue >= length)
+                            {
+                                return Err(Error::MaxQueue);
+                            };
+
+                            let (send, recv) = watch_channel(None);
+
+                            self.queue.write().await.insert(id.clone(), recv);
+
+                            loop {
+                                let length = self.watchers.read().await.len();
+
+                                if length < max_concurrent {
+                                    let first_key = {
+                                        let queue = self.queue.read().await;
+                                        queue.first().map(|v| v.0).cloned()
+                                    };
+
+                                    if first_key == Some(id.clone()) {
+                                        self.queue.write().await.shift_remove(&id);
+
+                                        let response = self.insert_and_execute(id, func).await;
+
+                                        send.send_modify(|opt| {
+                                            opt.replace(
+                                                response
+                                                    .clone()
+                                                    .map(|v| v as Arc<dyn Any + Send + Sync>),
+                                            );
+                                        });
+
+                                        return response;
+                                    }
+                                }
+                            }
+                        }
+                    } else {
+                        Err(Error::MaxConcurrent)
+                    }
+
+                    #[cfg(not(feature = "queue"))]
+                    Err(Error::MaxConcurrent)
+                }
+                _ => self.insert_and_execute(id, func).await,
+            }
+        }
+    }
+
+    /// Fetches the amount of currently running tasks
+    pub async fn current_task_count(&self) -> usize {
+        self.watchers.read().await.len()
+    }
+
+    #[cfg(feature = "queue")]
+    /// Fetches the current length of the queue
+    pub async fn current_queue_len(&self) -> usize {
+        self.queue.read().await.len()
+    }
+}
+
+impl<Id: Hash + Clone + Eq> Default for CoalescionService<Id> {
+    fn default() -> Self {
+        Self::from_config(CoalescionServiceConfig::default())
+    }
+}

crates/core/config/Revolt.toml

@@ -58,6 +58,8 @@ voso_legacy_token = ""
 trust_cloudflare = false
 # easypwned endpoint
 easypwned = ""
+# Tenor API Key
+tenor_key = ""
 
 [api.security.captcha]
 # hCaptcha configuration
@@ -310,3 +312,4 @@ files = ""
 proxy = ""
 pushd = ""
 crond = ""
+gifbox = ""

crates/core/config/src/lib.rs

@@ -189,6 +189,7 @@ pub struct ApiSecurity {
     pub captcha: ApiSecurityCaptcha,
     pub trust_cloudflare: bool,
     pub easypwned: String,
+    pub tenor_key: String,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -394,6 +395,7 @@ pub struct Sentry {
     pub proxy: String,
     pub pushd: String,
     pub crond: String,
+    pub gifbox: String,
 }
 
 #[derive(Deserialize, Debug, Clone)]

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

@@ -17,17 +17,33 @@ pub enum Ping {
 }
 
 /// Fields provided in Ready payload
-#[derive(PartialEq)]
-pub enum ReadyPayloadFields {
-    Users,
-    Servers,
-    Channels,
-    Members,
-    Emoji,
-    VoiceStates,
+#[derive(PartialEq, Debug, Clone, Deserialize)]
+pub struct ReadyPayloadFields {
+    pub users: bool,
+    pub servers: bool,
+    pub channels: bool,
+    pub members: bool,
+    pub emojis: bool,
+    pub voice_states: bool,
+    pub user_settings: Vec<String>,
+    pub channel_unreads: bool,
+    pub policy_changes: bool,
+}
 
-    UserSettings(Vec<String>),
-    ChannelUnreads,
+impl Default for ReadyPayloadFields {
+    fn default() -> Self {
+        Self {
+            users: true,
+            servers: true,
+            channels: true,
+            members: true,
+            emojis: true,
+            voice_states: true,
+            user_settings: Vec::new(),
+            channel_unreads: false,
+            policy_changes: true,
+        }
+    }
 }
 
 /// Protocol Events
@@ -63,7 +79,8 @@ pub enum EventV1 {
         #[serde(skip_serializing_if = "Option::is_none")]
         channel_unreads: Option<Vec<ChannelUnread>>,
 
-        policy_changes: Vec<PolicyChange>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        policy_changes: Option<Vec<PolicyChange>>,
     },
 
     /// Ping response

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

@@ -1,14 +1,20 @@
-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> FromRequestParts<S> for User
+where
+    Database: FromRef<S>,
+    S: Send + Sync
+{
     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/voice/mod.rs

@@ -328,6 +328,9 @@ pub async fn get_channel_voice_state(channel: &Channel) -> Result<Option<v0::Cha
             }
         }
 
+        // In case a user voice state failed to be fetched, the vec's capacity will be larger than the length, shrink it
+        participants.shrink_to_fit();
+
         Ok(Some(v0::ChannelVoiceState {
             id: channel.id().to_string(),
             participants,

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/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]
+}