multi_tier_cache/

l2_cache.rs

//! L2 Cache - Redis Cache
//!
//! Redis-based distributed cache for warm data storage with persistence.

use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use redis::{Client, AsyncCommands};
use redis::aio::ConnectionManager;
use serde_json;
use std::sync::atomic::{AtomicU64, Ordering};

/// L2 Cache using Redis with ConnectionManager for automatic reconnection
pub struct L2Cache {
    /// Redis connection manager - handles reconnection automatically
    conn_manager: ConnectionManager,
    /// Hit counter
    hits: Arc<AtomicU64>,
    /// Miss counter
    misses: Arc<AtomicU64>,
    /// Set counter
    sets: Arc<AtomicU64>,
}

impl L2Cache {
    /// Create new L2 cache with ConnectionManager for automatic reconnection
    pub async fn new() -> Result<Self> {
        println!("  🔴 Initializing L2 Cache (Redis with ConnectionManager)...");

        // Try to connect to Redis
        let redis_url = std::env::var("REDIS_URL")
            .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());

        let client = Client::open(redis_url.as_str())?;

        // Create ConnectionManager - handles reconnection automatically
        let conn_manager = ConnectionManager::new(client).await?;

        // Test connection
        let mut conn = conn_manager.clone();
        let _: String = redis::cmd("PING").query_async(&mut conn).await?;

        println!("  ✅ L2 Cache connected to Redis at {} (ConnectionManager enabled)", redis_url);

        Ok(Self {
            conn_manager,
            hits: Arc::new(AtomicU64::new(0)),
            misses: Arc::new(AtomicU64::new(0)),
            sets: Arc::new(AtomicU64::new(0)),
        })
    }
    
    /// Get value from L2 cache using persistent ConnectionManager
    pub async fn get(&self, key: &str) -> Option<serde_json::Value> {
        let mut conn = self.conn_manager.clone();

        match conn.get::<_, String>(key).await {
            Ok(json_str) => {
                match serde_json::from_str(&json_str) {
                    Ok(value) => {
                        self.hits.fetch_add(1, Ordering::Relaxed);
                        Some(value)
                    }
                    Err(_) => {
                        self.misses.fetch_add(1, Ordering::Relaxed);
                        None
                    }
                }
            }
            Err(_) => {
                self.misses.fetch_add(1, Ordering::Relaxed);
                None
            }
        }
    }

    /// Get value with its remaining TTL from L2 cache
    ///
    /// Returns tuple of (value, ttl_seconds) if key exists
    /// TTL is in seconds, None if key doesn't exist or has no expiration
    pub async fn get_with_ttl(&self, key: &str) -> Option<(serde_json::Value, Option<Duration>)> {
        let mut conn = self.conn_manager.clone();

        // Get value
        let json_str: String = match conn.get(key).await {
            Ok(s) => s,
            Err(_) => {
                self.misses.fetch_add(1, Ordering::Relaxed);
                return None;
            }
        };

        // Parse JSON
        let value: serde_json::Value = match serde_json::from_str(&json_str) {
            Ok(v) => v,
            Err(_) => {
                self.misses.fetch_add(1, Ordering::Relaxed);
                return None;
            }
        };

        // Get TTL (in seconds, -1 = no expiry, -2 = key doesn't exist)
        let ttl_secs: i64 = match redis::cmd("TTL").arg(key).query_async(&mut conn).await {
            Ok(ttl) => ttl,
            Err(_) => -1, // Fallback: treat as no expiry
        };

        self.hits.fetch_add(1, Ordering::Relaxed);

        let ttl = if ttl_secs > 0 {
            Some(Duration::from_secs(ttl_secs as u64))
        } else {
            None // No expiry or error
        };

        Some((value, ttl))
    }
    
    /// Set value with custom TTL using persistent ConnectionManager
    pub async fn set_with_ttl(&self, key: &str, value: serde_json::Value, ttl: Duration) -> Result<()> {
        let json_str = serde_json::to_string(&value)?;
        let mut conn = self.conn_manager.clone();

        let _: () = conn.set_ex(key, json_str, ttl.as_secs()).await?;
        self.sets.fetch_add(1, Ordering::Relaxed);
        println!("💾 [L2] Cached '{}' with TTL {:?}", key, ttl);
        Ok(())
    }
    
    /// Remove value from cache using persistent ConnectionManager
    pub async fn remove(&self, key: &str) -> Result<()> {
        let mut conn = self.conn_manager.clone();
        let _: () = conn.del(key).await?;
        Ok(())
    }

    /// Scan keys matching a pattern (glob-style: *, ?, [])
    ///
    /// Uses Redis SCAN command (non-blocking, cursor-based iteration)
    /// This is safe for production use, unlike KEYS command.
    ///
    /// # Arguments
    /// * `pattern` - Glob-style pattern (e.g., "user:*", "product:123:*")
    ///
    /// # Returns
    /// Vector of matching key names
    ///
    /// # Examples
    /// ```
    /// // Find all user cache keys
    /// let keys = cache.scan_keys("user:*").await?;
    ///
    /// // Find specific user's cache keys
    /// let keys = cache.scan_keys("user:123:*").await?;
    /// ```
    pub async fn scan_keys(&self, pattern: &str) -> Result<Vec<String>> {
        let mut conn = self.conn_manager.clone();
        let mut keys = Vec::new();
        let mut cursor: u64 = 0;

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

            cursor = result.0;
            keys.extend(result.1);

            // Cursor 0 means iteration is complete
            if cursor == 0 {
                break;
            }
        }

        println!("🔍 [L2] Scanned keys matching '{}': {} found", pattern, keys.len());
        Ok(keys)
    }

    /// Remove multiple keys at once (bulk delete)
    ///
    /// More efficient than calling remove() multiple times
    pub async fn remove_bulk(&self, keys: &[String]) -> Result<usize> {
        if keys.is_empty() {
            return Ok(0);
        }

        let mut conn = self.conn_manager.clone();
        let count: usize = conn.del(keys).await?;
        println!("🗑️  [L2] Removed {} keys", count);
        Ok(count)
    }
    
    /// Health check
    pub async fn health_check(&self) -> bool {
        let test_key = "health_check_l2";
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or(Duration::from_secs(0))
            .as_secs();
        let test_value = serde_json::json!({"test": true, "timestamp": timestamp});

        match self.set_with_ttl(test_key, test_value.clone(), Duration::from_secs(10)).await {
            Ok(_) => {
                match self.get(test_key).await {
                    Some(retrieved) => {
                        let _ = self.remove(test_key).await;
                        retrieved["test"].as_bool().unwrap_or(false)
                    }
                    None => false
                }
            }
            Err(_) => false
        }
    }
    
    /// Publish data to Redis Stream using XADD
    ///
    /// # Arguments
    /// * `stream_key` - Name of the Redis Stream (e.g., "events_stream")
    /// * `fields` - Vec of field-value pairs to add to the stream
    /// * `maxlen` - Optional maximum length for stream trimming (uses MAXLEN ~ for approximate trimming)
    ///
    /// # Returns
    /// The stream entry ID generated by Redis (e.g., "1234567890123-0")
    pub async fn stream_add(
        &self,
        stream_key: &str,
        fields: Vec<(String, String)>,
        maxlen: Option<usize>
    ) -> Result<String> {
        let mut conn = self.conn_manager.clone();

        // Build XADD command with MAXLEN trimming if specified
        let mut cmd = redis::cmd("XADD");
        cmd.arg(stream_key);

        if let Some(max) = maxlen {
            // Use MAXLEN ~ for approximate trimming (more efficient)
            cmd.arg("MAXLEN").arg("~").arg(max);
        }

        cmd.arg("*");  // Auto-generate ID

        // Add all field-value pairs
        for (field, value) in &fields {
            cmd.arg(field).arg(value);
        }

        let entry_id: String = cmd.query_async(&mut conn).await?;

        println!("📤 [L2 Stream] Published to '{}' (ID: {}, fields: {})", stream_key, entry_id, fields.len());
        Ok(entry_id)
    }
    
    /// Read latest entries from Redis Stream using XREVRANGE
    /// 
    /// # Arguments
    /// * `stream_key` - Name of the Redis Stream
    /// * `count` - Number of latest entries to retrieve
    /// 
    /// # Returns
    /// Vector of (entry_id, fields) tuples, ordered from newest to oldest
    pub async fn stream_read_latest(
        &self,
        stream_key: &str,
        count: usize
    ) -> Result<Vec<(String, Vec<(String, String)>)>> {
        let mut conn = self.conn_manager.clone();

        // XREVRANGE returns a raw Value that we need to parse manually
        let raw_result: redis::Value = redis::cmd("XREVRANGE")
            .arg(stream_key)
            .arg("+")  // Start from the newest
            .arg("-")  // To the oldest
            .arg("COUNT")
            .arg(count)
            .query_async(&mut conn)
            .await?;

        let entries = Self::parse_stream_response(raw_result)?;

        println!("📥 [L2 Stream] Read {} entries from '{}'", entries.len(), stream_key);
        Ok(entries)
    }
    
    /// Read entries from Redis Stream using XREAD (blocking or non-blocking)
    /// 
    /// # Arguments
    /// * `stream_key` - Name of the Redis Stream
    /// * `last_id` - Last entry ID seen (use "0" to read from beginning, "$" for only new entries)
    /// * `count` - Maximum number of entries to retrieve
    /// * `block_ms` - Optional blocking timeout in milliseconds (None for non-blocking)
    /// 
    /// # Returns
    /// Vector of (entry_id, fields) tuples
    pub async fn stream_read(
        &self,
        stream_key: &str,
        last_id: &str,
        count: usize,
        block_ms: Option<usize>
    ) -> Result<Vec<(String, Vec<(String, String)>)>> {
        let mut conn = self.conn_manager.clone();

        let mut cmd = redis::cmd("XREAD");

        // Add BLOCK if specified
        if let Some(block) = block_ms {
            cmd.arg("BLOCK").arg(block);
        }

        cmd.arg("COUNT").arg(count);
        cmd.arg("STREAMS").arg(stream_key).arg(last_id);

        let raw_result: redis::Value = cmd.query_async(&mut conn).await?;

        // XREAD returns [[stream_name, [[id, [field, value, ...]], ...]]]
        let entries = Self::parse_xread_response(raw_result)?;

        println!("📥 [L2 Stream] XREAD retrieved {} entries from '{}'", entries.len(), stream_key);
        Ok(entries)
    }
    
    /// Helper function to parse Redis stream response from XREVRANGE/XRANGE
    /// Format: [[id, [field, value, field, value, ...]], ...]
    fn parse_stream_response(value: redis::Value) -> Result<Vec<(String, Vec<(String, String)>)>> {
        match value {
            redis::Value::Array(entries) => {
                let mut result = Vec::new();
                
                for entry in entries {
                    if let redis::Value::Array(entry_parts) = entry {
                        if entry_parts.len() >= 2 {
                            // First element is the ID
                            let id = Self::value_to_string(&entry_parts[0]);
                            
                            // Second element is the field-value array
                            if let redis::Value::Array(field_values) = &entry_parts[1] {
                                let mut fields = Vec::new();
                                
                                // Process pairs of field-value
                                for chunk in field_values.chunks(2) {
                                    if chunk.len() == 2 {
                                        let field = Self::value_to_string(&chunk[0]);
                                        let value = Self::value_to_string(&chunk[1]);
                                        fields.push((field, value));
                                    }
                                }
                                
                                result.push((id, fields));
                            }
                        }
                    }
                }
                
                Ok(result)
            }
            redis::Value::Nil => Ok(Vec::new()),
            _ => Err(anyhow::anyhow!("Unexpected Redis stream response format"))
        }
    }
    
    /// Helper function to parse XREAD response
    /// Format: [[stream_name, [[id, [field, value, ...]], ...]], ...]
    fn parse_xread_response(value: redis::Value) -> Result<Vec<(String, Vec<(String, String)>)>> {
        match value {
            redis::Value::Array(streams) => {
                let mut all_entries = Vec::new();
                
                for stream in streams {
                    if let redis::Value::Array(stream_parts) = stream {
                        if stream_parts.len() >= 2 {
                            // Second element contains the entries
                            if let redis::Value::Array(entries) = &stream_parts[1] {
                                for entry in entries {
                                    if let redis::Value::Array(entry_parts) = entry {
                                        if entry_parts.len() >= 2 {
                                            let id = Self::value_to_string(&entry_parts[0]);
                                            
                                            if let redis::Value::Array(field_values) = &entry_parts[1] {
                                                let mut fields = Vec::new();
                                                
                                                for chunk in field_values.chunks(2) {
                                                    if chunk.len() == 2 {
                                                        let field = Self::value_to_string(&chunk[0]);
                                                        let value = Self::value_to_string(&chunk[1]);
                                                        fields.push((field, value));
                                                    }
                                                }
                                                
                                                all_entries.push((id, fields));
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
                
                Ok(all_entries)
            }
            redis::Value::Nil => Ok(Vec::new()),
            _ => Err(anyhow::anyhow!("Unexpected XREAD response format"))
        }
    }
    
    /// Helper function to convert Redis Value to String
    fn value_to_string(value: &redis::Value) -> String {
        match value {
            redis::Value::BulkString(bytes) => String::from_utf8_lossy(bytes).to_string(),
            redis::Value::SimpleString(s) => s.clone(),
            redis::Value::Int(i) => i.to_string(),
            redis::Value::Okay => "OK".to_string(),
            redis::Value::Nil => String::new(),
            _ => String::new(),
        }
    }
}

// ===== Trait Implementations =====

use crate::traits::{CacheBackend, L2CacheBackend, StreamingBackend};
use async_trait::async_trait;

/// Implement CacheBackend trait for L2Cache
///
/// This allows L2Cache to be used as a pluggable backend in the multi-tier cache system.
#[async_trait]
impl CacheBackend for L2Cache {
    async fn get(&self, key: &str) -> Option<serde_json::Value> {
        L2Cache::get(self, key).await
    }

    async fn set_with_ttl(
        &self,
        key: &str,
        value: serde_json::Value,
        ttl: Duration,
    ) -> Result<()> {
        L2Cache::set_with_ttl(self, key, value, ttl).await
    }

    async fn remove(&self, key: &str) -> Result<()> {
        L2Cache::remove(self, key).await
    }

    async fn health_check(&self) -> bool {
        L2Cache::health_check(self).await
    }

    fn name(&self) -> &str {
        "Redis (L2)"
    }
}

/// Implement L2CacheBackend trait for L2Cache
///
/// This extends CacheBackend with TTL introspection capabilities needed for L2->L1 promotion.
#[async_trait]
impl L2CacheBackend for L2Cache {
    async fn get_with_ttl(
        &self,
        key: &str,
    ) -> Option<(serde_json::Value, Option<Duration>)> {
        L2Cache::get_with_ttl(self, key).await
    }
}

/// Implement StreamingBackend trait for L2Cache
///
/// This enables Redis Streams functionality for event-driven architectures.
#[async_trait]
impl StreamingBackend for L2Cache {
    async fn stream_add(
        &self,
        stream_key: &str,
        fields: Vec<(String, String)>,
        maxlen: Option<usize>,
    ) -> Result<String> {
        L2Cache::stream_add(self, stream_key, fields, maxlen).await
    }

    async fn stream_read_latest(
        &self,
        stream_key: &str,
        count: usize,
    ) -> Result<Vec<(String, Vec<(String, String)>)>> {
        L2Cache::stream_read_latest(self, stream_key, count).await
    }

    async fn stream_read(
        &self,
        stream_key: &str,
        last_id: &str,
        count: usize,
        block_ms: Option<usize>,
    ) -> Result<Vec<(String, Vec<(String, String)>)>> {
        L2Cache::stream_read(self, stream_key, last_id, count, block_ms).await
    }
}