jmug/stoatchat
1
0
Fork 1
mirror of https://github.com/stoatchat/stoatchat.git synced 2026-07-06 03:06:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs: document revolt-coalesced

Browse Source
This commit is contained in:
Zomatree
2025-09-15 01:44:21 +01:00
committed by Angelo Kontaxis
parent 5885e067a6
commit db55998546
12 changed files with 154 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/core/coalesced/Cargo.toml
View File

@@ -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 }

1
crates/core/coalesced/src/config.rs
View File

@@ -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>,

15
crates/core/coalesced/src/error.rs
View File

@@ -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")
}
}
}

32
crates/core/coalesced/src/lib.rs
View File

@@ -1,3 +1,35 @@
//! # 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;

88
crates/core/coalesced/src/service.rs
View File

@@ -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,22 +44,37 @@ 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);
@@ -61,7 +82,7 @@ impl<Id: Hash + PartialEq + Eq + Clone + Ord, Value> CoalescionService<Id, Value
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())
}

9
crates/services/gifbox/Cargo.toml
View File

@@ -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"] }

1
crates/services/gifbox/src/routes/categories.rs
View File

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

5
crates/services/gifbox/src/routes/search.rs
View File

@@ -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>,
}

20
crates/services/gifbox/src/routes/trending.rs
View File

@@ -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())

48
crates/services/gifbox/src/tenor/mod.rs
View File

@@ -1,3 +1,5 @@
//! Interal Tenor API wrapper
use std::{sync::Arc, time::Duration};
use lru_time_cache::LruCache;
@@ -20,18 +22,11 @@ pub enum TenorError {
pub struct Tenor {
pub key: Arc<str>,
pub client: Client,
pub coalescion: CoalescionService<String>,
pub cache: Arc<RwLock<LruCache<String, Arc<types::PaginatedMediaResponse>>>>,
pub cache_coalescion:
CoalescionService<String, Result<Arc<types::PaginatedMediaResponse>, TenorError>>,
pub categories: Arc<RwLock<LruCache<String, Arc<types::CategoriesResponse>>>>,
pub categories_coalescion:
CoalescionService<String, Result<Arc<types::CategoriesResponse>, TenorError>>,
pub featured: Arc<RwLock<LruCache<String, Arc<types::PaginatedMediaResponse>>>>,
pub featured_coalescion:
CoalescionService<String, Result<Arc<types::PaginatedMediaResponse>, TenorError>>,
}
impl Tenor {
@@ -39,39 +34,29 @@ impl Tenor {
Self {
key: Arc::from(key),
client: Client::new(),
coalescion: CoalescionService::from_config(CoalescionServiceConfig {
max_concurrent: Some(100),
queue_requests: true,
max_queue: None,
}),
// 1 hour, 1k requests
cache: Arc::new(RwLock::new(LruCache::with_expiry_duration_and_capacity(
Duration::from_secs(60 * 60),
1000,
))),
cache_coalescion: CoalescionService::from_config(CoalescionServiceConfig {
max_concurrent: Some(100),
queue_requests: true,
max_queue: None,
}),
// 1 day, 1k requests
categories: Arc::new(RwLock::new(LruCache::with_expiry_duration_and_capacity(
Duration::from_secs(60 * 60 * 24),
1000,
))),
categories_coalescion: CoalescionService::from_config(CoalescionServiceConfig {
max_concurrent: Some(100),
queue_requests: true,
max_queue: None,
}),
// 1 day, 1k requests
featured: Arc::new(RwLock::new(LruCache::with_expiry_duration_and_capacity(
Duration::from_secs(60 * 60 * 24),
1000,
))),
featured_coalescion: CoalescionService::from_config(CoalescionServiceConfig {
max_concurrent: Some(100),
queue_requests: true,
max_queue: None,
}),
}
}
@@ -110,7 +95,7 @@ impl Tenor {
}
}
let res = self.cache_coalescion.execute(unique_key.clone(), || {
let res = self.coalescion.execute(unique_key.clone(), || {
let mut path = format!(
"/search?key={}&q={}&client_key=Gifbox&media_filter=webm,tinywebm&locale={}&contentfilter=high&limit={limit}",
&self.key,
@@ -127,7 +112,7 @@ impl Tenor {
path.push_str("&component=categories");
}
self.request(path)
self.request::<types::PaginatedMediaResponse>(path)
})
.await
.unwrap();
@@ -152,7 +137,7 @@ impl Tenor {
}
let res = self
.categories_coalescion
.coalescion
.execute(unique_key.clone(), || {
let path = format!(
"/categories?key={}&client_key=Gifbox&locale={}&contentfilter=high",
@@ -160,7 +145,7 @@ impl Tenor {
url_encode(locale),
);
self.request(path)
self.request::<types::CategoriesResponse>(path)
})
.await
.unwrap();
@@ -189,7 +174,7 @@ impl Tenor {
}
}
let res = self.featured_coalescion.execute(unique_key.clone(), || {
let res = self.coalescion.execute(unique_key.clone(), || {
let mut path = format!(
"/featured?key={}&client_key=Gifbox&media_filter=webm,tinywebm&locale={}&contentfilter=high&limit={limit}",
&self.key,
@@ -201,16 +186,13 @@ impl Tenor {
path.push_str(position);
};
self.request(path)
self.request::<types::PaginatedMediaResponse>(path)
})
.await
.unwrap();
if let Ok(resp) = &*res {
self.featured
.write()
.await
.insert(unique_key, resp.clone());
self.featured.write().await.insert(unique_key, resp.clone());
}
(*res).clone()

2
crates/services/gifbox/src/tenor/types.rs
View File

@@ -1,3 +1,5 @@
//! Tenor API models
use std::collections::HashMap;
use serde::{Deserialize, Serialize};