use std::sync::Arc;
use redis::aio::ConnectionManager;
use redis::{AsyncCommands, Client};
use serde::{de::DeserializeOwned, Serialize};
use uuid::Uuid;
use crate::error::{DbError, Result};
pub mod keys {
pub const USER_SESSION: &str = "session:";
pub const USER_PROFILE: &str = "user:";
pub const TOKEN_PRICE: &str = "price:";
pub const TOKEN_META: &str = "token:";
pub const USER_BALANCE: &str = "balance:";
pub const RATE_LIMIT: &str = "ratelimit:";
pub const LOCK: &str = "lock:";
}
#[derive(Debug, Clone)]
pub struct CacheConfig {
pub redis_url: String,
pub default_ttl_secs: u64,
pub session_ttl_secs: u64,
pub price_ttl_secs: u64,
pub profile_ttl_secs: u64,
pub balance_ttl_secs: u64,
pub key_prefix: String,
}
impl Default for CacheConfig {
fn default() -> Self {
Self {
redis_url: "redis://127.0.0.1:6379".to_string(),
default_ttl_secs: 3600, session_ttl_secs: 86400, price_ttl_secs: 5, profile_ttl_secs: 300, balance_ttl_secs: 30, key_prefix: "kaccy:".to_string(),
}
}
}
impl CacheConfig {
pub fn from_env() -> Self {
let redis_url =
std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
Self {
redis_url,
..Default::default()
}
}
pub fn with_ttls(
mut self,
default: u64,
session: u64,
price: u64,
profile: u64,
balance: u64,
) -> Self {
self.default_ttl_secs = default;
self.session_ttl_secs = session;
self.price_ttl_secs = price;
self.profile_ttl_secs = profile;
self.balance_ttl_secs = balance;
self
}
}
#[derive(Clone)]
pub struct RedisCache {
conn: ConnectionManager,
config: CacheConfig,
}
impl RedisCache {
pub async fn new(config: CacheConfig) -> Result<Self> {
let client = Client::open(config.redis_url.as_str())
.map_err(|e| DbError::Connection(format!("Redis client error: {}", e)))?;
let conn = ConnectionManager::new(client)
.await
.map_err(|e| DbError::Connection(format!("Redis connection error: {}", e)))?;
tracing::info!("Redis cache connected to {}", config.redis_url);
Ok(Self { conn, config })
}
fn full_key(&self, key: &str) -> String {
format!("{}{}", self.config.key_prefix, key)
}
pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let value: Option<String> = conn
.get(&full_key)
.await
.map_err(|e| DbError::Cache(format!("Redis GET error: {}", e)))?;
match value {
Some(json) => {
let parsed: T = serde_json::from_str(&json)
.map_err(|e| DbError::Cache(format!("Deserialization error: {}", e)))?;
Ok(Some(parsed))
}
None => Ok(None),
}
}
pub async fn set<T: Serialize>(&self, key: &str, value: &T, ttl_secs: u64) -> Result<()> {
let full_key = self.full_key(key);
let json = serde_json::to_string(value)
.map_err(|e| DbError::Cache(format!("Serialization error: {}", e)))?;
let mut conn = self.conn.clone();
let _: () = conn
.set_ex(&full_key, json, ttl_secs)
.await
.map_err(|e| DbError::Cache(format!("Redis SET error: {}", e)))?;
Ok(())
}
pub async fn set_default<T: Serialize>(&self, key: &str, value: &T) -> Result<()> {
self.set(key, value, self.config.default_ttl_secs).await
}
pub async fn delete(&self, key: &str) -> Result<bool> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let deleted: i64 = conn
.del(&full_key)
.await
.map_err(|e| DbError::Cache(format!("Redis DEL error: {}", e)))?;
Ok(deleted > 0)
}
pub async fn delete_pattern(&self, pattern: &str) -> Result<u64> {
let full_pattern = self.full_key(pattern);
let mut conn = self.conn.clone();
let keys: Vec<String> = conn
.keys(&full_pattern)
.await
.map_err(|e| DbError::Cache(format!("Redis KEYS error: {}", e)))?;
if keys.is_empty() {
return Ok(0);
}
let deleted: i64 = conn
.del(&keys)
.await
.map_err(|e| DbError::Cache(format!("Redis DEL error: {}", e)))?;
Ok(deleted as u64)
}
pub async fn exists(&self, key: &str) -> Result<bool> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let exists: bool = conn
.exists(&full_key)
.await
.map_err(|e| DbError::Cache(format!("Redis EXISTS error: {}", e)))?;
Ok(exists)
}
pub async fn incr(&self, key: &str) -> Result<i64> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let value: i64 = conn
.incr(&full_key, 1)
.await
.map_err(|e| DbError::Cache(format!("Redis INCR error: {}", e)))?;
Ok(value)
}
pub async fn incr_with_expiry(&self, key: &str, ttl_secs: u64) -> Result<i64> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let (value,): (i64,) = redis::pipe()
.atomic()
.incr(&full_key, 1)
.expire(&full_key, ttl_secs as i64)
.ignore()
.query_async(&mut conn)
.await
.map_err(|e| DbError::Cache(format!("Redis INCR/EXPIRE error: {}", e)))?;
Ok(value)
}
pub async fn expire(&self, key: &str, ttl_secs: u64) -> Result<bool> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let set: bool = conn
.expire(&full_key, ttl_secs as i64)
.await
.map_err(|e| DbError::Cache(format!("Redis EXPIRE error: {}", e)))?;
Ok(set)
}
pub async fn ttl(&self, key: &str) -> Result<i64> {
let full_key = self.full_key(key);
let mut conn = self.conn.clone();
let ttl: i64 = conn
.ttl(&full_key)
.await
.map_err(|e| DbError::Cache(format!("Redis TTL error: {}", e)))?;
Ok(ttl)
}
pub async fn try_lock(&self, resource: &str, ttl_secs: u64) -> Result<Option<String>> {
let key = format!("{}{}:{}", keys::LOCK, resource, Uuid::new_v4());
let full_key = self.full_key(&key);
let lock_id = Uuid::new_v4().to_string();
let mut conn = self.conn.clone();
let set: bool = conn
.set_nx(&full_key, &lock_id)
.await
.map_err(|e| DbError::Cache(format!("Redis SETNX error: {}", e)))?;
if set {
let _: () = conn
.expire(&full_key, ttl_secs as i64)
.await
.map_err(|e| DbError::Cache(format!("Redis EXPIRE error: {}", e)))?;
Ok(Some(lock_id))
} else {
Ok(None)
}
}
pub async fn release_lock(&self, resource: &str, lock_id: &str) -> Result<bool> {
let key = format!("{}{}:{}", keys::LOCK, resource, lock_id);
self.delete(&key).await
}
pub async fn health_check(&self) -> Result<bool> {
let mut conn = self.conn.clone();
let pong: String = redis::cmd("PING")
.query_async(&mut conn)
.await
.map_err(|e| DbError::Cache(format!("Redis PING error: {}", e)))?;
Ok(pong == "PONG")
}
}
impl RedisCache {
pub async fn set_session(&self, session_id: &str, user_id: Uuid) -> Result<()> {
let key = format!("{}{}", keys::USER_SESSION, session_id);
self.set(&key, &user_id, self.config.session_ttl_secs).await
}
pub async fn get_session(&self, session_id: &str) -> Result<Option<Uuid>> {
let key = format!("{}{}", keys::USER_SESSION, session_id);
self.get(&key).await
}
pub async fn invalidate_session(&self, session_id: &str) -> Result<bool> {
let key = format!("{}{}", keys::USER_SESSION, session_id);
self.delete(&key).await
}
#[allow(dead_code)]
pub async fn invalidate_user_sessions(&self, user_id: Uuid) -> Result<u64> {
tracing::warn!(
"invalidate_user_sessions: scanning all sessions for user {}",
user_id
);
Ok(0) }
}
impl RedisCache {
pub async fn set_user_profile<T: Serialize>(&self, user_id: Uuid, profile: &T) -> Result<()> {
let key = format!("{}{}", keys::USER_PROFILE, user_id);
self.set(&key, profile, self.config.profile_ttl_secs).await
}
pub async fn get_user_profile<T: DeserializeOwned>(&self, user_id: Uuid) -> Result<Option<T>> {
let key = format!("{}{}", keys::USER_PROFILE, user_id);
self.get(&key).await
}
pub async fn invalidate_user_profile(&self, user_id: Uuid) -> Result<bool> {
let key = format!("{}{}", keys::USER_PROFILE, user_id);
self.delete(&key).await
}
}
impl RedisCache {
pub async fn set_token_price(&self, token_id: Uuid, price_btc: f64) -> Result<()> {
let key = format!("{}{}", keys::TOKEN_PRICE, token_id);
self.set(&key, &price_btc, self.config.price_ttl_secs).await
}
pub async fn get_token_price(&self, token_id: Uuid) -> Result<Option<f64>> {
let key = format!("{}{}", keys::TOKEN_PRICE, token_id);
self.get(&key).await
}
pub async fn set_token_prices(&self, prices: &[(Uuid, f64)]) -> Result<()> {
for (token_id, price) in prices {
self.set_token_price(*token_id, *price).await?;
}
Ok(())
}
}
impl RedisCache {
pub async fn set_token_meta<T: Serialize>(&self, token_id: Uuid, meta: &T) -> Result<()> {
let key = format!("{}{}", keys::TOKEN_META, token_id);
self.set(&key, meta, self.config.default_ttl_secs).await
}
pub async fn get_token_meta<T: DeserializeOwned>(&self, token_id: Uuid) -> Result<Option<T>> {
let key = format!("{}{}", keys::TOKEN_META, token_id);
self.get(&key).await
}
pub async fn invalidate_token_meta(&self, token_id: Uuid) -> Result<bool> {
let key = format!("{}{}", keys::TOKEN_META, token_id);
self.delete(&key).await
}
}
impl RedisCache {
pub async fn set_balance(&self, user_id: Uuid, token_id: Uuid, balance: f64) -> Result<()> {
let key = format!("{}{}:{}", keys::USER_BALANCE, user_id, token_id);
self.set(&key, &balance, self.config.balance_ttl_secs).await
}
pub async fn get_balance(&self, user_id: Uuid, token_id: Uuid) -> Result<Option<f64>> {
let key = format!("{}{}:{}", keys::USER_BALANCE, user_id, token_id);
self.get(&key).await
}
pub async fn invalidate_user_balances(&self, user_id: Uuid) -> Result<u64> {
let pattern = format!("{}{}:*", keys::USER_BALANCE, user_id);
self.delete_pattern(&pattern).await
}
pub async fn invalidate_token_balances(&self, token_id: Uuid) -> Result<u64> {
let pattern = format!("{}*:{}", keys::USER_BALANCE, token_id);
self.delete_pattern(&pattern).await
}
}
impl RedisCache {
pub async fn check_rate_limit(
&self,
identifier: &str,
limit: u64,
window_secs: u64,
) -> Result<(u64, bool)> {
let key = format!("{}{}", keys::RATE_LIMIT, identifier);
let count = self.incr_with_expiry(&key, window_secs).await? as u64;
Ok((count, count <= limit))
}
pub async fn get_rate_limit_count(&self, identifier: &str) -> Result<u64> {
let key = format!("{}{}", keys::RATE_LIMIT, identifier);
let count: Option<u64> = self.get(&key).await?;
Ok(count.unwrap_or(0))
}
pub async fn reset_rate_limit(&self, identifier: &str) -> Result<bool> {
let key = format!("{}{}", keys::RATE_LIMIT, identifier);
self.delete(&key).await
}
}
pub struct CachedRepository<R> {
cache: Arc<RedisCache>,
repo: R,
}
impl<R> CachedRepository<R> {
pub fn new(cache: Arc<RedisCache>, repo: R) -> Self {
Self { cache, repo }
}
pub fn repo(&self) -> &R {
&self.repo
}
pub fn cache(&self) -> &RedisCache {
&self.cache
}
}
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct CacheStats {
pub hits: u64,
pub misses: u64,
pub sets: u64,
pub deletes: u64,
}
impl CacheStats {
pub fn hit_rate(&self) -> f64 {
let total = self.hits + self.misses;
if total == 0 {
0.0
} else {
self.hits as f64 / total as f64
}
}
}