prax_query/data_cache/

redis.rs

//! Redis cache backend for distributed caching.
//!
//! This module provides a Redis-based cache implementation with:
//!
//! - **Connection pooling** using bb8 or deadpool
//! - **Cluster support** for horizontal scaling
//! - **Pipelining** for batch operations
//! - **Lua scripting** for atomic operations
//! - **Pub/Sub** for cache invalidation across instances
//!
//! # Example
//!
//! ```rust,ignore
//! use prax_query::data_cache::redis::{RedisCache, RedisCacheConfig};
//!
//! let cache = RedisCache::new(RedisCacheConfig {
//!     url: "redis://localhost:6379".to_string(),
//!     pool_size: 10,
//!     ..Default::default()
//! }).await?;
//! ```

use std::time::Duration;

use super::backend::{BackendStats, CacheBackend, CacheError, CacheResult};
use super::invalidation::EntityTag;
use super::key::{CacheKey, KeyPattern};

/// Configuration for Redis cache.
#[derive(Debug, Clone)]
pub struct RedisCacheConfig {
    /// Redis connection URL.
    pub url: String,
    /// Connection pool size.
    pub pool_size: u32,
    /// Connection timeout.
    pub connection_timeout: Duration,
    /// Command timeout.
    pub command_timeout: Duration,
    /// Key prefix for all entries.
    pub key_prefix: String,
    /// Default TTL.
    pub default_ttl: Option<Duration>,
    /// Enable cluster mode.
    pub cluster_mode: bool,
    /// Database number (0-15).
    pub database: u8,
    /// Enable TLS.
    pub tls: bool,
    /// Username for AUTH.
    pub username: Option<String>,
    /// Password for AUTH.
    pub password: Option<String>,
}

impl Default for RedisCacheConfig {
    fn default() -> Self {
        Self {
            url: "redis://localhost:6379".to_string(),
            pool_size: 10,
            connection_timeout: Duration::from_secs(5),
            command_timeout: Duration::from_secs(2),
            key_prefix: "prax:cache".to_string(),
            default_ttl: Some(Duration::from_secs(300)),
            cluster_mode: false,
            database: 0,
            tls: false,
            username: None,
            password: None,
        }
    }
}

impl RedisCacheConfig {
    /// Create a new config with the given URL.
    pub fn new(url: impl Into<String>) -> Self {
        Self {
            url: url.into(),
            ..Default::default()
        }
    }

    /// Set pool size.
    pub fn with_pool_size(mut self, size: u32) -> Self {
        self.pool_size = size;
        self
    }

    /// Set key prefix.
    pub fn with_prefix(mut self, prefix: impl Into<String>) -> Self {
        self.key_prefix = prefix.into();
        self
    }

    /// Set default TTL.
    pub fn with_ttl(mut self, ttl: Duration) -> Self {
        self.default_ttl = Some(ttl);
        self
    }

    /// Enable cluster mode.
    pub fn cluster(mut self) -> Self {
        self.cluster_mode = true;
        self
    }

    /// Set database number.
    pub fn database(mut self, db: u8) -> Self {
        self.database = db;
        self
    }

    /// Set authentication.
    pub fn auth(mut self, username: Option<String>, password: impl Into<String>) -> Self {
        self.username = username;
        self.password = Some(password.into());
        self
    }

    /// Build the full key with prefix.
    fn full_key(&self, key: &CacheKey) -> String {
        format!("{}:{}", self.key_prefix, key.as_str())
    }
}

/// Represents a Redis connection (placeholder for actual implementation).
///
/// In a real implementation, this would use `redis-rs` or `fred` crate.
#[derive(Clone)]
pub struct RedisConnection {
    config: RedisCacheConfig,
    // In real impl: pool: Pool<RedisConnectionManager>
}

impl RedisConnection {
    /// Create a new connection.
    pub async fn new(config: RedisCacheConfig) -> CacheResult<Self> {
        // In real implementation:
        // - Create connection pool using bb8 or deadpool
        // - Establish initial connections
        // - Verify connectivity

        Ok(Self { config })
    }

    /// Get the config.
    pub fn config(&self) -> &RedisCacheConfig {
        &self.config
    }

    /// Execute a Redis command (placeholder).
    async fn execute<T>(&self, _cmd: &str, _args: &[&str]) -> CacheResult<T>
    where
        T: Default,
    {
        // Placeholder - real impl would use redis-rs
        // Example with redis-rs:
        // let mut conn = self.pool.get().await?;
        // redis::cmd(cmd).arg(args).query_async(&mut *conn).await
        Ok(T::default())
    }

    /// GET command.
    pub async fn get(&self, key: &str) -> CacheResult<Option<Vec<u8>>> {
        // Placeholder
        let _ = key;
        Ok(None)
    }

    /// SET command with optional TTL.
    pub async fn set(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> CacheResult<()> {
        // Placeholder
        let _ = (key, value, ttl);
        Ok(())
    }

    /// DEL command.
    pub async fn del(&self, key: &str) -> CacheResult<bool> {
        // Placeholder
        let _ = key;
        Ok(false)
    }

    /// EXISTS command.
    pub async fn exists(&self, key: &str) -> CacheResult<bool> {
        // Placeholder
        let _ = key;
        Ok(false)
    }

    /// KEYS command (use SCAN in production).
    pub async fn keys(&self, pattern: &str) -> CacheResult<Vec<String>> {
        // Placeholder - use SCAN in production for large datasets
        let _ = pattern;
        Ok(Vec::new())
    }

    /// MGET command.
    pub async fn mget(&self, keys: &[String]) -> CacheResult<Vec<Option<Vec<u8>>>> {
        // Placeholder
        Ok(vec![None; keys.len()])
    }

    /// MSET command.
    pub async fn mset(&self, pairs: &[(String, Vec<u8>)]) -> CacheResult<()> {
        // Placeholder
        let _ = pairs;
        Ok(())
    }

    /// FLUSHDB command.
    pub async fn flush(&self) -> CacheResult<()> {
        // Placeholder
        Ok(())
    }

    /// DBSIZE command.
    pub async fn dbsize(&self) -> CacheResult<usize> {
        // Placeholder
        Ok(0)
    }

    /// INFO command.
    pub async fn info(&self) -> CacheResult<String> {
        // Placeholder
        Ok(String::new())
    }

    /// SCAN for pattern matching.
    pub async fn scan(&self, pattern: &str, count: usize) -> CacheResult<Vec<String>> {
        // Placeholder - in real impl, iterate through all matches
        let _ = (pattern, count);
        Ok(Vec::new())
    }

    /// Pipeline multiple commands.
    pub fn pipeline(&self) -> RedisPipeline {
        RedisPipeline::new(self.clone())
    }
}

/// A Redis pipeline for batching commands.
pub struct RedisPipeline {
    conn: RedisConnection,
    commands: Vec<PipelineCommand>,
}

enum PipelineCommand {
    Get(String),
    Set(String, Vec<u8>, Option<Duration>),
    Del(String),
}

impl RedisPipeline {
    fn new(conn: RedisConnection) -> Self {
        Self {
            conn,
            commands: Vec::new(),
        }
    }

    /// Add a GET command.
    pub fn get(mut self, key: impl Into<String>) -> Self {
        self.commands.push(PipelineCommand::Get(key.into()));
        self
    }

    /// Add a SET command.
    pub fn set(mut self, key: impl Into<String>, value: Vec<u8>, ttl: Option<Duration>) -> Self {
        self.commands.push(PipelineCommand::Set(key.into(), value, ttl));
        self
    }

    /// Add a DEL command.
    pub fn del(mut self, key: impl Into<String>) -> Self {
        self.commands.push(PipelineCommand::Del(key.into()));
        self
    }

    /// Execute the pipeline.
    pub async fn execute(self) -> CacheResult<Vec<PipelineResult>> {
        // Placeholder - real impl would batch execute
        Ok(vec![PipelineResult::Ok; self.commands.len()])
    }
}

/// Result of a pipeline command.
#[derive(Debug, Clone)]
pub enum PipelineResult {
    Ok,
    Value(Option<Vec<u8>>),
    Error(String),
}

/// Redis cache backend.
#[derive(Clone)]
pub struct RedisCache {
    conn: RedisConnection,
    config: RedisCacheConfig,
}

impl RedisCache {
    /// Create a new Redis cache.
    pub async fn new(config: RedisCacheConfig) -> CacheResult<Self> {
        let conn = RedisConnection::new(config.clone()).await?;
        Ok(Self { conn, config })
    }

    /// Create from a URL.
    pub async fn from_url(url: &str) -> CacheResult<Self> {
        Self::new(RedisCacheConfig::new(url)).await
    }

    /// Get the connection.
    pub fn connection(&self) -> &RedisConnection {
        &self.conn
    }

    /// Get the config.
    pub fn config(&self) -> &RedisCacheConfig {
        &self.config
    }

    /// Build the full key with prefix.
    fn full_key(&self, key: &CacheKey) -> String {
        self.config.full_key(key)
    }
}

impl CacheBackend for RedisCache {
    async fn get<T>(&self, key: &CacheKey) -> CacheResult<Option<T>>
    where
        T: serde::de::DeserializeOwned,
    {
        let full_key = self.full_key(key);

        match self.conn.get(&full_key).await? {
            Some(data) => {
                let value: T = serde_json::from_slice(&data)
                    .map_err(|e| CacheError::Deserialization(e.to_string()))?;
                Ok(Some(value))
            }
            None => Ok(None),
        }
    }

    async fn set<T>(
        &self,
        key: &CacheKey,
        value: &T,
        ttl: Option<Duration>,
    ) -> CacheResult<()>
    where
        T: serde::Serialize + Sync,
    {
        let full_key = self.full_key(key);
        let data = serde_json::to_vec(value)
            .map_err(|e| CacheError::Serialization(e.to_string()))?;

        let effective_ttl = ttl.or(self.config.default_ttl);
        self.conn.set(&full_key, &data, effective_ttl).await
    }

    async fn delete(&self, key: &CacheKey) -> CacheResult<bool> {
        let full_key = self.full_key(key);
        self.conn.del(&full_key).await
    }

    async fn exists(&self, key: &CacheKey) -> CacheResult<bool> {
        let full_key = self.full_key(key);
        self.conn.exists(&full_key).await
    }

    async fn get_many<T>(&self, keys: &[CacheKey]) -> CacheResult<Vec<Option<T>>>
    where
        T: serde::de::DeserializeOwned,
    {
        let full_keys: Vec<String> = keys.iter().map(|k| self.full_key(k)).collect();
        let results = self.conn.mget(&full_keys).await?;

        results
            .into_iter()
            .map(|opt| {
                opt.map(|data| {
                    serde_json::from_slice(&data)
                        .map_err(|e| CacheError::Deserialization(e.to_string()))
                })
                .transpose()
            })
            .collect()
    }

    async fn invalidate_pattern(&self, pattern: &KeyPattern) -> CacheResult<u64> {
        let full_pattern = format!("{}:{}", self.config.key_prefix, pattern.to_redis_pattern());

        // Use SCAN to find matching keys
        let keys = self.conn.scan(&full_pattern, 1000).await?;

        if keys.is_empty() {
            return Ok(0);
        }

        // Delete in batches
        let mut deleted = 0u64;
        for key in keys {
            if self.conn.del(&key).await? {
                deleted += 1;
            }
        }

        Ok(deleted)
    }

    async fn invalidate_tags(&self, tags: &[EntityTag]) -> CacheResult<u64> {
        // Tags stored as sets: tag:<tag_value> -> [key1, key2, ...]
        let mut total = 0u64;

        for tag in tags {
            let tag_key = format!("{}:tag:{}", self.config.key_prefix, tag.value());
            // In real impl: SMEMBERS to get keys, then DEL
            let _ = tag_key;
            total += 0; // Placeholder
        }

        Ok(total)
    }

    async fn clear(&self) -> CacheResult<()> {
        // In production, use SCAN + DEL with prefix
        // FLUSHDB would clear everything
        self.conn.flush().await
    }

    async fn len(&self) -> CacheResult<usize> {
        self.conn.dbsize().await
    }

    async fn stats(&self) -> CacheResult<BackendStats> {
        let info = self.conn.info().await?;
        let entries = self.conn.dbsize().await?;

        Ok(BackendStats {
            entries,
            memory_bytes: None, // Parse from INFO
            connections: Some(self.config.pool_size as usize),
            info: Some(info),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_redis_config() {
        let config = RedisCacheConfig::new("redis://localhost:6379")
            .with_pool_size(20)
            .with_prefix("myapp")
            .with_ttl(Duration::from_secs(600));

        assert_eq!(config.pool_size, 20);
        assert_eq!(config.key_prefix, "myapp");
        assert_eq!(config.default_ttl, Some(Duration::from_secs(600)));
    }

    #[test]
    fn test_full_key() {
        let config = RedisCacheConfig::new("redis://localhost")
            .with_prefix("app:cache");

        let key = CacheKey::new("User", "id:123");
        let full = config.full_key(&key);

        assert_eq!(full, "app:cache:prax:User:id:123");
    }

    #[tokio::test]
    async fn test_redis_cache_creation() {
        // This test just verifies the API works
        // Real tests would need a Redis instance
        let config = RedisCacheConfig::default();
        let cache = RedisCache::new(config).await.unwrap();

        assert_eq!(cache.config().pool_size, 10);
    }
}