use crate::db::get_redis_conn;
use crate::types::DynError;
use redis::AsyncCommands;
use serde::Deserialize;
use utoipa::ToSchema;
#[derive(Clone, Deserialize, Debug, ToSchema, Default)]
#[serde(rename_all = "snake_case")]
pub enum SortOrder {
Ascending,
#[default]
Descending,
}
pub enum ScoreAction {
Increment(f64),
Decrement(f64),
}
pub const SORTED_PREFIX: &str = "Sorted";
pub async fn check_member(
prefix: &str,
key: &str,
member: &str,
) -> Result<Option<isize>, DynError> {
let index_key = format!("{prefix}:{key}");
let mut redis_conn = get_redis_conn().await?;
let rank = redis_conn.zscore(index_key, member).await?;
Ok(rank)
}
pub async fn put(
prefix: &str,
key: &str,
items: &[(f64, &str)],
expiration: Option<i64>,
) -> Result<(), DynError> {
if items.is_empty() {
return Ok(());
}
let index_key = format!("{prefix}:{key}");
let mut redis_conn = get_redis_conn().await?;
let mut pipe = redis::pipe();
pipe.zadd_multiple(&index_key, items);
if let Some(ttl) = expiration {
pipe.expire(&index_key, ttl);
}
let _: () = pipe.query_async(&mut redis_conn).await?;
Ok(())
}
pub async fn put_score(
prefix: &str,
key: &str,
member: &str,
score_mutation: ScoreAction,
) -> Result<(), DynError> {
let index_key = format!("{prefix}:{key}");
let mut redis_conn = get_redis_conn().await?;
let value = match score_mutation {
ScoreAction::Increment(val) => val,
ScoreAction::Decrement(val) => -val,
};
let _: () = redis_conn.zincr(&index_key, member, value).await?;
Ok(())
}
pub async fn get_range(
prefix: &str,
key: &str,
min_score: Option<f64>,
max_score: Option<f64>,
skip: Option<usize>,
limit: Option<usize>,
sorting: SortOrder,
) -> Result<Option<Vec<(String, f64)>>, DynError> {
let mut redis_conn = get_redis_conn().await?;
let index_key = format!("{prefix}:{key}");
if !redis_conn.exists(&index_key).await? {
return Ok(None);
}
let min_score = min_score.unwrap_or(f64::MIN);
let max_score = max_score.unwrap_or(f64::MAX);
let skip = skip.unwrap_or(0) as isize;
let limit = limit.unwrap_or(1000) as isize;
let elements: Vec<(String, f64)> = match sorting {
SortOrder::Ascending => {
redis_conn
.zrangebyscore_limit_withscores(index_key, min_score, max_score, skip, limit)
.await?
}
SortOrder::Descending => {
redis_conn
.zrevrangebyscore_limit_withscores(index_key, max_score, min_score, skip, limit)
.await?
}
};
Ok(Some(elements))
}
pub async fn get_lex_range(
prefix: &str,
key: &str,
min: &str,
max: &str,
skip: Option<usize>,
limit: Option<usize>,
) -> Result<Option<Vec<String>>, DynError> {
let mut redis_conn = get_redis_conn().await?;
let index_key = format!("{prefix}:{key}");
let skip = skip.unwrap_or(0) as isize;
let limit = limit.unwrap_or(1000) as isize;
let elements: Vec<String> = redis_conn
.zrangebylex_limit(index_key, min, max, skip, limit)
.await?;
match elements.len() {
0 => Ok(None),
_ => Ok(Some(elements)),
}
}
pub async fn _remove(prefix: &str, key: &str, items: &[&str]) -> Result<(), DynError> {
if items.is_empty() {
return Ok(());
}
let index_key = format!("{prefix}:{key}");
let mut redis_conn = get_redis_conn().await?;
let _: () = redis_conn.zrem(&index_key, items).await?;
Ok(())
}
pub async fn del(prefix: &str, key: &str, values: &[&str]) -> Result<(), DynError> {
if values.is_empty() {
return Ok(());
}
let index_key = format!("{prefix}:{key}");
let mut redis_conn = get_redis_conn().await?;
let _: () = redis_conn.zrem(index_key, values).await?;
Ok(())
}