multi_tier_cache/

redis_streams.rs

//! Redis Streams Integration
//!
//! Provides high-level Redis Streams functionality for event-driven architectures.
//!
//! Redis Streams is a data structure that acts like an append-only log, perfect for:
//! - Event sourcing
//! - Real-time analytics
//! - Message queues
//! - Activity feeds
//! - Audit logs
//!
//! # Features
//!
//! - **XADD**: Publish events with automatic ID generation
//! - **XREVRANGE**: Read latest N entries (newest first)
//! - **XREAD**: Read new entries (with optional blocking)
//! - **MAXLEN**: Automatic stream trimming to prevent unbounded growth
//!
//! # Example
//!
//! ```no_run
//! use multi_tier_cache::RedisStreams;
//!
//! # async fn example() -> anyhow::Result<()> {
//! let streams = RedisStreams::new("redis://127.0.0.1:6379").await?;
//!
//! // Publish event
//! let event_id = streams.stream_add(
//!     "events",
//!     vec![
//!         ("user_id".to_string(), "123".to_string()),
//!         ("action".to_string(), "login".to_string()),
//!     ],
//!     Some(1000)  // Keep only latest 1000 events
//! ).await?;
//!
//! // Read latest 10 events
//! let latest = streams.stream_read_latest("events", 10).await?;
//! # Ok(())
//! # }
//! ```

use anyhow::Result;
use redis::aio::ConnectionManager;
use async_trait::async_trait;
use tracing::debug;
use crate::traits::StreamingBackend;

/// Type alias for Redis Stream entry: (ID, Field-Value Pairs)
pub type StreamEntry = (String, Vec<(String, String)>);

/// Redis Streams client for event-driven architectures
///
/// Wraps Redis ConnectionManager to provide high-level streaming operations.
#[derive(Clone)]
pub struct RedisStreams {
    conn_manager: ConnectionManager,
}

impl RedisStreams {
    /// Create a new RedisStreams instance
    ///
    /// # Arguments
    /// * `redis_url` - Redis connection URL (e.g., "redis://127.0.0.1:6379")
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use multi_tier_cache::RedisStreams;
    /// # async fn example() -> anyhow::Result<()> {
    /// let streams = RedisStreams::new("redis://127.0.0.1:6379").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn new(redis_url: &str) -> Result<Self> {
        let client = redis::Client::open(redis_url)?;
        let conn_manager = ConnectionManager::new(client).await?;

        debug!("Redis Streams initialized at {}", redis_url);

        Ok(Self { conn_manager })
    }

    /// 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")
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use multi_tier_cache::RedisStreams;
    /// # async fn example() -> anyhow::Result<()> {
    /// # let streams = RedisStreams::new("redis://127.0.0.1:6379").await?;
    /// let event_id = streams.stream_add(
    ///     "user_events",
    ///     vec![
    ///         ("user_id".to_string(), "42".to_string()),
    ///         ("action".to_string(), "purchase".to_string()),
    ///         ("amount".to_string(), "99.99".to_string()),
    ///     ],
    ///     Some(10000)  // Keep max 10k events
    /// ).await?;
    /// println!("Event ID: {}", event_id);
    /// # Ok(())
    /// # }
    /// ```
    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?;

        debug!("[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
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use multi_tier_cache::RedisStreams;
    /// # async fn example() -> anyhow::Result<()> {
    /// # let streams = RedisStreams::new("redis://127.0.0.1:6379").await?;
    /// let latest_10 = streams.stream_read_latest("events", 10).await?;
    /// for (id, fields) in latest_10 {
    ///     println!("Event {}: {:?}", id, fields);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn stream_read_latest(
        &self,
        stream_key: &str,
        count: usize
    ) -> Result<Vec<StreamEntry>> {
        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)?;

        debug!("[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
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use multi_tier_cache::RedisStreams;
    /// # async fn example() -> anyhow::Result<()> {
    /// # let streams = RedisStreams::new("redis://127.0.0.1:6379").await?;
    /// // Non-blocking: read new entries since last seen ID
    /// let entries = streams.stream_read("events", "1234567890-0", 100, None).await?;
    ///
    /// // Blocking: wait up to 5 seconds for new entries
    /// let new_entries = streams.stream_read("events", "$", 100, Some(5000)).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn stream_read(
        &self,
        stream_key: &str,
        last_id: &str,
        count: usize,
        block_ms: Option<usize>
    ) -> Result<Vec<StreamEntry>> {
        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)?;

        debug!("[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<StreamEntry>> {
        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<StreamEntry>> {
        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(),
        }
    }
}

/// Implement StreamingBackend trait for RedisStreams
///
/// This enables RedisStreams to be used as a pluggable streaming backend.
#[async_trait]
impl StreamingBackend for RedisStreams {
    async fn stream_add(
        &self,
        stream_key: &str,
        fields: Vec<(String, String)>,
        maxlen: Option<usize>,
    ) -> Result<String> {
        RedisStreams::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)>)>> {
        let entries = RedisStreams::stream_read_latest(self, stream_key, count).await?;
        // Convert StreamEntry to the trait's expected type if they match, or just return
        // Since StreamEntry IS (String, Vec<(String, String)>), this is fine.
        // However, the trait definition in traits.rs might not use the alias.
        // Let's check traits.rs first or just return the result since the types are identical.
        Ok(entries)
    }

    async fn stream_read(
        &self,
        stream_key: &str,
        last_id: &str,
        count: usize,
        block_ms: Option<usize>,
    ) -> Result<Vec<(String, Vec<(String, String)>)>> {
        let entries = RedisStreams::stream_read(self, stream_key, last_id, count, block_ms).await?;
        Ok(entries)
    }
}