huiali/rust-skills/rust-cache

use redis::{aio::ConnectionManager, AsyncCommands};
use serde::{de::DeserializeOwned, Serialize};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
 
pub struct CacheManager {
    config: CacheConfig,
    redis: Option<ConnectionManager>,
    stats: Arc<RwLock<CacheStats>>,
}
 
impl CacheManager {
    pub async fn new(config: CacheConfig) -> Result<Self, CacheError> {
        let redis = if config.enabled && config.redis.enabled {
            let client = redis::Client::open(config.redis.url.as_str())
                .map_err(|e| CacheError::Connection(format!("Redis connection failed: {}", e)))?;
 
            // Timeout control for remote Redis
            let timeout = Duration::from_secs(30);
 
            match tokio::time::timeout(timeout, ConnectionManager::new(client)).await {
                Ok(Ok(conn)) => Some(conn),
                Ok(Err(e)) => return Err(CacheError::Connection(format!("Redis failed: {}", e))),
                Err(_) => return Err(CacheError::Timeout(format!("Redis timeout ({}s)", timeout.as_secs()))),
            }
        } else {
            None
        };
 
        Ok(Self {
            config,
            redis,
            stats: Arc::new(RwLock::new(CacheStats::new())),
        })
    }
 
    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CacheError> {
        if !self.config.enabled {
            return Ok(None);
        }
 
        // Update stats
        {
            let mut stats = self.stats.write().await;
            stats.total_requests += 1;
        }
 
        if let Some(mut redis) = self.redis.clone() {
            match redis.get::<&str, Vec<u8>>(key).await {
                Ok(bytes) if !bytes.is_empty() => {
                    {
                        let mut stats = self.stats.write().await;
                        stats.redis_hits += 1;
                    }
                    self.deserialize(&bytes)
                }
                Ok(_) => {
                    let mut stats = self.stats.write().await;
                    stats.redis_misses += 1;
                    Ok(None)
                }
                Err(e) => {
                    log::warn!("Redis read failed: key={}, error={}", key, e);
                    let mut stats = self.stats.write().await;
                    stats.redis_misses += 1;
                    Ok(None)  // Cache failures shouldn't block business logic
                }
            }
        } else {
            Ok(None)
        }
    }
 
    pub async fn set<T: Serialize>(
        &self,
        key: &str,
        value: &T,
        ttl: Option<u64>,
    ) -> Result<(), CacheError> {
        if !self.config.enabled {
            return Ok(());
        }
 
        let bytes = self.serialize(value)?;
 
        if let Some(mut redis) = self.redis.clone() {
            let ttl_seconds = ttl.unwrap_or(self.config.default_ttl);
            match redis.set_ex::<&str, Vec<u8>, ()>(key, bytes, ttl_seconds).await {
                Ok(_) => log::debug!("Redis write: key={}, ttl={}s", key, ttl_seconds),
                Err(e) => log::warn!("Redis write failed: key={}, error={}", key, e),
            }
        }
 
        Ok(())
    }
 
    fn serialize<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, CacheError> {
        serde_json::to_vec(value).map_err(|e| {
            CacheError::Serialization(format!("Serialization failed: {}", e))
        })
    }
 
    fn deserialize<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<Option<T>, CacheError> {
        if bytes.is_empty() {
            return Ok(None);
        }
 
        match serde_json::from_slice(bytes) {
            Ok(value) => Ok(Some(value)),
            Err(e) => {
                log::warn!("Deserialization failed: {}", e);
                Ok(None)  // Corrupted data should be skipped, not error
            }
        }
    }
}
 
#[derive(Debug, Clone, Default)]
pub struct CacheStats {
    pub total_requests: u64,
    pub redis_hits: u64,
    pub redis_misses: u64,
}
 
impl CacheStats {
    pub fn hit_rate(&self) -> f64 {
        if self.total_requests == 0 {
            0.0
        } else {
            self.redis_hits as f64 / self.total_requests as f64 * 100.0
        }
    }
}

use sha2::{Digest, Sha256};
 
pub struct CacheKeyBuilder;
 
impl CacheKeyBuilder {
    /// Build namespaced key
    /// Format: {namespace}:{entity}:{id}
    pub fn build(namespace: &str, entity: &str, id: impl std::fmt::Display) -> String {
        format!("{}:{}:{}", namespace, entity, id)
    }
 
    /// Build list cache key with query hash
    /// Format: {namespace}:{entity}:list:{query_hash}
    pub fn list_key(namespace: &str, entity: &str, query: &str) -> String {
        let mut hasher = Sha256::new();
        hasher.update(query.as_bytes());
        let hash = format!("{:x}", hasher.finalize());
        format!("{}:{}:list:{}", namespace, entity, &hash[..8])
    }
 
    /// Build pattern matching key
    /// Format: {namespace}:{entity}:*
    pub fn pattern(namespace: &str, entity: &str) -> String {
        format!("{}:{}:*", namespace, entity)
    }
 
    /// Build versioned key
    /// Format: {namespace}:{entity}:{id}:v{version}
    pub fn versioned(namespace: &str, entity: &str, id: impl std::fmt::Display, version: u64) -> String {
        format!("{}:{}:{}:v{}", namespace, entity, id, version)
    }
}
 
// Usage examples
fn example_key_usage() {
    // Single entity
    let key = CacheKeyBuilder::build("myapp", "user", 123);
    // myapp:user:123
 
    // List query
    let key = CacheKeyBuilder::list_key("myapp", "posts", "tag=rust&sort=date");
    // myapp:posts:list:a3f2d8e1
 
    // Pattern for deletion
    let pattern = CacheKeyBuilder::pattern("myapp", "user");
    // myapp:user:*
}

pub struct CacheBreaker<K, T> {
    manager: Arc<CacheManager>,
    _phantom: std::marker::PhantomData<(K, T)>,
}
 
impl<K, T> CacheBreaker<K, T>
where
    K: std::fmt::Display + Clone + Send + Sync,
    T: DeserializeOwned + Serialize + Clone,
{
    pub fn new(manager: Arc<CacheManager>) -> Self {
        Self {
            manager,
            _phantom: std::marker::PhantomData
        }
    }
 
    /// Get data with cache protection
    pub async fn get_or_load<F, Fut>(&self, key: &str, loader: F) -> Result<Option<T>, CacheError>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<Option<T>, CacheError>>,
    {
        // 1. Try cache first
        if let Some(cached) = self.manager.get::<T>(key).await? {
            return Ok(Some(cached));
        }
 
        // 2. Cache miss, load from data source
        let result = loader().await?;
 
        // 3. Write back to cache
        if let Some(ref value) = result {
            self.manager.set(key, value, None).await?;
        }
 
        Ok(result)
    }
}
 
// Usage
async fn get_user(cache: &CacheBreaker<UserId, User>, id: UserId) -> Result<Option<User>> {
    let key = format!("user:{}", id);
 
    cache.get_or_load(&key, || async move {
        database.fetch_user(id).await
    }).await
}

impl CacheManager {
    /// Batch delete with pattern matching using SCAN
    /// Avoids blocking with DEL command
    pub async fn delete_pattern(&self, pattern: &str) -> Result<usize, CacheError> {
        let mut deleted_count = 0;
 
        if let Some(redis) = &self.redis {
            let mut cursor: u64 = 0;
 
            loop {
                let result: std::result::Result<(u64, Vec<String>), redis::RedisError> =
                    redis::cmd("SCAN")
                        .arg(cursor)
                        .arg("MATCH")
                        .arg(pattern)
                        .arg("COUNT")
                        .arg(100)
                        .query_async(&mut redis.clone())
                        .await;
 
                match result {
                    Ok((new_cursor, keys)) => {
                        if !keys.is_empty() {
                            let del_result: std::result::Result<(), redis::RedisError> =
                                redis::cmd("DEL")
                                    .arg(&keys)
                                    .query_async(&mut redis.clone())
                                    .await;
 
                            if del_result.is_ok() {
                                deleted_count += keys.len();
                            }
                        }
                        cursor = new_cursor;
                        if cursor == 0 {
                            break;
                        }
                    }
                    Err(e) => {
                        log::warn!("Redis SCAN failed: {}", e);
                        break;
                    }
                }
            }
 
            log::info!("Batch delete completed: pattern={}, deleted={}", pattern, deleted_count);
        }
 
        Ok(deleted_count)
    }
}
 
// Usage
async fn invalidate_user_cache(cache: &CacheManager, user_id: u64) -> Result<()> {
    // Delete all user-related cache entries
    let pattern = format!("myapp:user:{}:*", user_id);
    cache.delete_pattern(&pattern).await?;
    Ok(())
}

use rand::Rng;
 
/// Add random jitter to TTL to prevent cache avalanche
fn calculate_jitter_ttl(base_ttl: u64) -> u64 {
    let jitter_range = (base_ttl as f64 * 0.1)..(base_ttl as f64 * 0.2);
    let jitter_seconds = rand::thread_rng().gen_range(jitter_range);
    (base_ttl as f64 + jitter_seconds) as u64
}
 
pub async fn set_with_jitter(
    cache: &CacheManager,
    key: &str,
    value: &impl Serialize,
    base_ttl: u64,
) -> Result<(), CacheError> {
    let ttl = calculate_jitter_ttl(base_ttl);
    cache.set(key, value, Some(ttl)).await
}
 
// Usage
async fn cache_with_protection(cache: &CacheManager) -> Result<()> {
    // Cache 1000 items with 1 hour base TTL
    // Actual TTLs will range from 3600s to 4320s (10-20% jitter)
    for i in 0..1000 {
        let key = format!("item:{}", i);
        set_with_jitter(&cache, &key, &i, 3600).await?;
    }
    Ok(())
}

Consider:
  → Read/write ratio? Read-heavy = Cache-Aside
  → Consistency requirements? Strong = Write-Through
  → Write performance critical? High throughput = Write-Behind
  → Predictable access? Refresh-Ahead

TTL tiers:
  → Hot data (high frequency): 5-15 minutes
  → Medium data (moderate frequency): 1 hour
  → Cold data (low frequency): 24 hours
  → Static data (rarely changes): 7 days
  → Always add jitter (10-20%) to prevent avalanche

Options:
  → Time-based: Let TTL expire (simplest)
  → Event-based: Invalidate on writes (accurate)
  → Pattern-based: Delete by pattern (bulk invalidation)
  → Version-based: Include version in key (no deletion needed)