docs: document revolt-coalesced

Author: Zomatree

crates/core/coalesced/Cargo.toml

@@ -11,7 +11,7 @@ tokio = ["dep:tokio"]
 queue = ["dep:indexmap"]
 cache = ["dep:lru"]
 
-default = ["tokio", "queue", "cache"]
+default = ["tokio"]
 
 [dependencies]
 tokio = { version = "1.47.0", features = ["sync"], optional = true }

crates/core/coalesced/src/config.rs

@@ -1,4 +1,5 @@
 #[derive(Clone, PartialEq, Eq, Debug)]
+/// Config values for [`CoalescionService`].
 pub struct CoalescionServiceConfig {
     /// How many tasks are running at once
     pub max_concurrent: Option<usize>,

crates/core/coalesced/src/error.rs

@@ -1,18 +1,25 @@
-use std::fmt::Display;
+use std::fmt;
 
-#[derive(Clone, PartialEq, Eq, Debug)]
+#[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 Display for Error {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+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")
         }
     }
 }

crates/core/coalesced/src/lib.rs

@@ -1,7 +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;
+pub use service::CoalescionService;

crates/core/coalesced/src/service.rs

@@ -1,6 +1,9 @@
-use std::{collections::{HashMap}, fmt::Debug, hash::Hash, sync::Arc, future::Future};
+use std::{any::Any, collections::HashMap, fmt::Debug, future::Future, hash::Hash, sync::Arc};
 
-use tokio::{sync::{watch::{channel as watch_channel, Receiver}, RwLock, Mutex}};
+use tokio::sync::{
+    watch::{channel as watch_channel, Receiver},
+    RwLock,
+};
 
 #[cfg(feature = "cache")]
 use lru::LruCache;
@@ -10,20 +13,23 @@ use indexmap::IndexMap;
 
 use crate::{CoalescionServiceConfig, Error};
 
-#[derive(Clone, Debug)]
+#[derive(Debug, Clone)]
 #[allow(clippy::type_complexity)]
-pub struct CoalescionService<Id: Hash + Eq, Value> {
+/// # 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<Value>, Error>>>>>>,
+    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<Value>, Error>>>>>>,
+    queue: Arc<RwLock<IndexMap<Id, Receiver<Option<Result<Arc<dyn Any + Send + Sync>, Error>>>>>>,
     #[cfg(feature = "cache")]
-    cache: Option<Arc<Mutex<LruCache<Id, Arc<Value>>>>>,
+    cache: Option<Arc<tokio::sync::Mutex<LruCache<Id, Arc<dyn Any + Send + Sync>>>>>,
 }
 
-impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value> {
+impl<Id: Hash + Clone + Eq> CoalescionService<Id> {
     pub fn new() -> Self {
-        Self::default()
+        Default::default()
     }
 
     pub fn from_config(config: CoalescionServiceConfig) -> Self {
@@ -38,30 +44,45 @@ impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value
     }
 
     #[cfg(feature = "cache")]
-    pub fn from_cache(config: CoalescionServiceConfig, cache: LruCache<Id, Arc<Value>>) -> Self {
+    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(&self, mut receiver: Receiver<Option<Result<Arc<Value>, Error>>>) -> Result<Arc<Value>, Error> {
+    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<F: FnOnce() -> Fut, Fut: Future<Output = Value>>(&self, id: Id, func: F) -> Result<Arc<Value>, Error> {
+    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());
+            opt.replace(value.clone().map(|v| v as Arc<dyn Any + Send + Sync>));
         });
 
         #[cfg(feature = "cache")]
@@ -76,11 +97,21 @@ impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value
         value
     }
 
-    pub async fn execute<F: FnOnce() -> Fut, Fut: Future<Output = Value>>(&self, id: Id, func: F) -> Result<Arc<Value>, Error> {
+    /// 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 Ok(value.clone())
+                return Arc::downcast::<Value>(value.clone()).map_err(|_| Error::DowncastError);
             }
         };
 
@@ -105,10 +136,14 @@ impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value
                         };
 
                         if let Some(receiver) = receiver {
-                            return self.wait_for(receiver).await
+                            return self.wait_for(receiver).await;
                         } else {
-                            if self.config.max_queue.is_some_and(|max_queue| max_queue >= length) {
-                                return Err(Error::MaxQueue)
+                            if self
+                                .config
+                                .max_queue
+                                .is_some_and(|max_queue| max_queue >= length)
+                            {
+                                return Err(Error::MaxQueue);
                             };
 
                             let (send, recv) = watch_channel(None);
@@ -130,10 +165,14 @@ impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value
                                         let response = self.insert_and_execute(id, func).await;
 
                                         send.send_modify(|opt| {
-                                            opt.replace(response.clone());
+                                            opt.replace(
+                                                response
+                                                    .clone()
+                                                    .map(|v| v as Arc<dyn Any + Send + Sync>),
+                                            );
                                         });
 
-                                        return response
+                                        return response;
                                     }
                                 }
                             }
@@ -145,23 +184,24 @@ impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value
                     #[cfg(not(feature = "queue"))]
                     Err(Error::MaxConcurrent)
                 }
-                _ => {
-                    self.insert_and_execute(id, func).await
-                }
+                _ => 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 + PartialEq + Eq + Clone + Ord, Value> Default for CoalescionService<Id, Value> {
+impl<Id: Hash + Clone + Eq> Default for CoalescionService<Id> {
     fn default() -> Self {
         Self::from_config(CoalescionServiceConfig::default())
     }

crates/services/gifbox/Cargo.toml

@@ -22,8 +22,13 @@ revolt-result = { version = "0.8.8", path = "../../core/result", features = [
     "utoipa",
     "axum",
 ] }
-revolt-coalesced = { version = "0.8.8", path = "../../core/coalesced" }
-revolt-database = { version = "0.8.8", path = "../../core/database", features = ["axum-impl"] }
+revolt-coalesced = { version = "0.8.8", path = "../../core/coalesced", features = [
+    "queue",
+] }
+revolt-database = { version = "0.8.8", path = "../../core/database", features = [
+    "axum-impl",
+] }
+
 # Axum / web server
 axum = { version = "0.7.5" }
 axum-extra = { version = "0.9", features = ["typed-header"] }
@@ -37,4 +42,4 @@ tracing = "0.1"
 
 # Utils
 lru_time_cache = "0.11.11"
-urlencoding = "2.1.3"
+urlencoding = "2.1.3"

crates/services/gifbox/src/routes/categories.rs

@@ -11,6 +11,7 @@ use crate::{tenor, types};
 
 #[derive(Deserialize, IntoParams)]
 pub struct CategoriesQueryParams {
+    /// Users locale
     #[param(example = "en_US")]
     pub locale: String,
 }

crates/services/gifbox/src/routes/root.rs

@@ -19,4 +19,4 @@ pub async fn root() -> Json<types::RootResponse<'static>> {
         message: "Gifbox lives on!",
         version: CRATE_VERSION,
     })
-}
+}

crates/services/gifbox/src/routes/search.rs

@@ -11,12 +11,17 @@ use crate::{tenor, types};
 
 #[derive(Deserialize, IntoParams)]
 pub struct SearchQueryParams {
+    /// Search query
     #[param(example = "Wave")]
     pub query: String,
+    /// Users locale
     #[param(example = "en_US")]
     pub locale: String,
+    /// Amount of results to respond with
     pub limit: Option<u32>,
+    /// Flag for if searching in a gif category
     pub is_category: Option<bool>,
+    /// Value of `next` for getting the next page of results with the current search query
     pub position: Option<String>,
 }
 

crates/services/gifbox/src/routes/trending.rs

@@ -4,20 +4,20 @@ use axum::{
 };
 use revolt_database::User;
 use revolt_result::{create_error, Result};
-use serde::{Deserialize};
-use utoipa::{IntoParams};
+use serde::Deserialize;
+use utoipa::IntoParams;
 
-use crate::{
-    tenor,
-    types,
-};
+use crate::{tenor, types};
 
 #[derive(Deserialize, IntoParams)]
 pub struct TrendingQueryParams {
     #[param(example = "en_US")]
+    /// Users locale
     pub locale: String,
+    /// Amount of results to respond with
     pub limit: Option<u32>,
-    pub position: Option<String>
+    /// Value of `next` for getting the next page of results of featured gifs
+    pub position: Option<String>,
 }
 
 /// Trending GIFs
@@ -37,7 +37,11 @@ pub async fn trending(
     State(tenor): State<tenor::Tenor>,
 ) -> Result<Json<types::PaginatedMediaResponse>> {
     tenor
-        .featured(&params.locale, params.limit.unwrap_or(50), params.position.as_deref().unwrap_or_default())
+        .featured(
+            &params.locale,
+            params.limit.unwrap_or(50),
+            params.position.as_deref().unwrap_or_default(),
+        )
         .await
         .map_err(|_| create_error!(InternalError))
         .map(|results| results.as_ref().clone().into())