warpdrive_proxy/cache/

redis.rs

//! Redis-based distributed cache implementation
//!
//! This module provides a high-performance Redis cache backend with:
//! - Connection pooling via ConnectionManager
//! - Async/await with tokio
//! - Key prefixing for namespace isolation
//! - Batch operations for efficiency
//! - Cluster support (when enabled via Cargo features)
//!
//! # Performance Characteristics
//!
//! - Connection reuse: ConnectionManager handles pooling internally
//! - Zero-copy when possible: Uses bytes for efficient data transfer
//! - Pipelined batch operations: Reduces round trips
//! - Minimal allocations in hot paths
//!
//! # Example
//!
//! ```no_run
//! use warpdrive::cache::redis::RedisCache;
//!
//! # async fn example() -> anyhow::Result<()> {
//! let cache = RedisCache::from_url(
//!     "redis://localhost:6379",
//!     "warpdrive:cache:".to_string()
//! ).await?;
//!
//! // Single operations
//! cache.set("user:123", b"user_data", 60).await?;
//! let data = cache.get("user:123").await?;
//! cache.delete("user:123").await?;
//!
//! // Batch operations
//! let values = cache.batch_get(&[
//!     "user:123".to_string(),
//!     "user:456".to_string()
//! ]).await?;
//!
//! // Clear by pattern
//! let deleted = cache.clear_prefix("user:*").await?;
//! println!("Deleted {} keys", deleted);
//! # Ok(())
//! # }
//! ```

use async_trait::async_trait;
use redis::AsyncCommands;
use redis::aio::ConnectionManager;

use crate::cache::Cache;

/// Redis cache implementation with connection pooling
///
/// This struct wraps a Redis ConnectionManager and provides cache operations
/// with automatic key prefixing for namespace isolation.
///
/// # Connection Management
///
/// Uses `ConnectionManager` which provides:
/// - Automatic reconnection on connection loss
/// - Connection health checks
/// - Async operation support
/// - Internal connection pooling
///
/// # Key Prefixing
///
/// All keys are automatically prefixed to avoid collisions:
/// - User provides: `"user:123"`
/// - Stored as: `"warpdrive:cache:user:123"` (if prefix is `"warpdrive:cache:"`)
#[derive(Clone)]
pub struct RedisCache {
    /// Connection manager for async Redis operations
    client: ConnectionManager,

    /// Key prefix for namespace isolation
    prefix: String,
}

impl RedisCache {
    /// Create a new Redis cache from connection URL
    ///
    /// # Parameters
    ///
    /// - `url`: Redis connection URL (e.g., `"redis://localhost:6379"`)
    /// - `prefix`: Key prefix for namespace isolation (e.g., `"warpdrive:cache:"`)
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - URL is invalid
    /// - Cannot connect to Redis server
    /// - Authentication fails
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use warpdrive::cache::redis::RedisCache;
    /// # async fn example() -> anyhow::Result<()> {
    /// let cache = RedisCache::from_url(
    ///     "redis://localhost:6379/0",
    ///     "myapp:".to_string()
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn from_url(url: &str, prefix: String) -> anyhow::Result<Self> {
        let client = redis::Client::open(url)?;
        let connection_manager = ConnectionManager::new(client).await?;

        Ok(RedisCache {
            client: connection_manager,
            prefix,
        })
    }

    /// Build the full key with prefix
    ///
    /// Internal helper to construct namespaced keys.
    #[inline]
    fn prefixed_key(&self, key: &str) -> String {
        format!("{}{}", self.prefix, key)
    }

    /// Batch get multiple keys
    ///
    /// Retrieves multiple values in a single round trip using Redis pipelining.
    /// This is significantly more efficient than multiple individual GET operations.
    ///
    /// # Returns
    ///
    /// A vector of `Option<Vec<u8>>` corresponding to the input keys.
    /// `None` indicates the key was not found or expired.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use warpdrive::cache::redis::RedisCache;
    /// # async fn example(cache: &RedisCache) -> anyhow::Result<()> {
    /// let keys = vec!["user:1".to_string(), "user:2".to_string()];
    /// let values = cache.batch_get(&keys).await?;
    ///
    /// for (key, value) in keys.iter().zip(values.iter()) {
    ///     match value {
    ///         Some(data) => println!("{}: {} bytes", key, data.len()),
    ///         None => println!("{}: not found", key),
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn batch_get(&self, keys: &[String]) -> anyhow::Result<Vec<Option<Vec<u8>>>> {
        if keys.is_empty() {
            return Ok(vec![]);
        }

        let mut conn = self.client.clone();

        // Build prefixed keys
        let prefixed_keys: Vec<String> = keys.iter().map(|k| self.prefixed_key(k)).collect();

        // Use MGET for batch retrieval - need to build the command manually
        let mut cmd = redis::cmd("MGET");
        for key in &prefixed_keys {
            cmd.arg(key);
        }

        let values: Vec<Option<Vec<u8>>> = cmd.query_async(&mut conn).await?;

        Ok(values)
    }

    /// Batch delete multiple keys
    ///
    /// Deletes multiple keys in a single round trip using Redis DEL.
    /// Returns `Ok(())` regardless of whether keys existed.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use warpdrive::cache::redis::RedisCache;
    /// # async fn example(cache: &RedisCache) -> anyhow::Result<()> {
    /// let keys = vec!["user:1".to_string(), "user:2".to_string()];
    /// cache.batch_delete(&keys).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn batch_delete(&self, keys: &[String]) -> anyhow::Result<()> {
        if keys.is_empty() {
            return Ok(());
        }

        let mut conn = self.client.clone();

        // Build prefixed keys
        let prefixed_keys: Vec<String> = keys.iter().map(|k| self.prefixed_key(k)).collect();

        // Use DEL for batch deletion
        let _: () = conn.del(&prefixed_keys).await?;

        Ok(())
    }

    /// Clear all keys matching a pattern
    ///
    /// Deletes all keys matching the given pattern (after prefixing).
    /// Uses SCAN + DEL for safe iteration without blocking Redis.
    ///
    /// # Warning
    ///
    /// This operation scans the entire keyspace and can be slow on large datasets.
    /// Use sparingly and consider the impact on Redis performance.
    ///
    /// # Parameters
    ///
    /// - `pattern`: Glob-style pattern (e.g., `"user:*"`, `"session:*"`)
    ///
    /// # Returns
    ///
    /// The number of keys deleted.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use warpdrive::cache::redis::RedisCache;
    /// # async fn example(cache: &RedisCache) -> anyhow::Result<()> {
    /// // Delete all user cache entries
    /// let count = cache.clear_prefix("user:*").await?;
    /// println!("Cleared {} user entries", count);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn clear_prefix(&self, pattern: &str) -> anyhow::Result<usize> {
        let mut conn = self.client.clone();
        let full_pattern = self.prefixed_key(pattern);

        // Use SCAN to iterate over matching keys without blocking
        let mut cursor = 0;
        let mut total_deleted = 0;

        loop {
            // SCAN returns (cursor, keys)
            let (new_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
                .arg(cursor)
                .arg("MATCH")
                .arg(&full_pattern)
                .arg("COUNT")
                .arg(100) // Scan 100 keys per iteration
                .query_async(&mut conn)
                .await?;

            if !keys.is_empty() {
                let deleted: usize = conn.del(&keys).await?;
                total_deleted += deleted;
            }

            cursor = new_cursor;

            // Cursor of 0 means we've completed the scan
            if cursor == 0 {
                break;
            }
        }

        Ok(total_deleted)
    }

    /// Get connection manager for advanced operations
    ///
    /// Provides access to the underlying ConnectionManager for operations
    /// not covered by the Cache trait (e.g., INCR, ZADD, etc.).
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use warpdrive::cache::redis::RedisCache;
    /// # use redis::AsyncCommands;
    /// # async fn example(cache: &RedisCache) -> anyhow::Result<()> {
    /// let mut conn = cache.connection();
    /// let count: i64 = conn.incr("page_views", 1).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn connection(&self) -> ConnectionManager {
        self.client.clone()
    }
}

#[async_trait]
impl Cache for RedisCache {
    async fn get(&self, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
        let mut conn = self.client.clone();
        let full_key = self.prefixed_key(key);

        // GET returns Option<Vec<u8>>
        let value: Option<Vec<u8>> = conn.get(&full_key).await?;

        Ok(value)
    }

    async fn set(&self, key: &str, value: &[u8], ttl_seconds: u64) -> anyhow::Result<()> {
        let mut conn = self.client.clone();
        let full_key = self.prefixed_key(key);

        if ttl_seconds > 0 {
            // SETEX: Set with expiration
            let _: () = conn.set_ex(&full_key, value, ttl_seconds).await?;
        } else {
            // SET: No expiration
            let _: () = conn.set(&full_key, value).await?;
        }

        Ok(())
    }

    async fn delete(&self, key: &str) -> anyhow::Result<()> {
        let mut conn = self.client.clone();
        let full_key = self.prefixed_key(key);

        // DEL returns number of keys deleted, we ignore it
        let _: () = conn.del(&full_key).await?;

        Ok(())
    }

    async fn exists(&self, key: &str) -> anyhow::Result<bool> {
        let mut conn = self.client.clone();
        let full_key = self.prefixed_key(key);

        // EXISTS returns 1 if key exists, 0 otherwise
        let exists: bool = conn.exists(&full_key).await?;

        Ok(exists)
    }
}

#[cfg(test)]
mod tests {
    // Helper to test key prefixing without real Redis connection
    fn mock_redis_cache(prefix: String) -> impl Fn(&str) -> String {
        move |key: &str| format!("{}{}", prefix, key)
    }

    #[test]
    fn test_prefixed_key() {
        let prefix = "warpdrive:cache:".to_string();
        let prefixer = mock_redis_cache(prefix);

        assert_eq!(prefixer("user:123"), "warpdrive:cache:user:123");
        assert_eq!(prefixer("session:abc"), "warpdrive:cache:session:abc");
    }

    #[test]
    fn test_prefixed_key_empty_prefix() {
        let prefix = "".to_string();
        let prefixer = mock_redis_cache(prefix);

        assert_eq!(prefixer("user:123"), "user:123");
    }
}