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 crate::traits::StreamingBackend;
use anyhow::Result;
use async_trait::async_trait;
use redis::aio::ConnectionManager;
use tracing::debug;

/// 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(())
    /// # }
    /// ```
    /// # Errors
    ///
    /// Returns an error if Redis connection fails.
    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(())
    /// # }
    /// ```
    /// # Errors
    ///
    /// Returns an error if the XADD command fails.
    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(())
    /// # }
    /// ```
    /// # Errors
    ///
    /// Returns an error if the XREVRANGE command fails.
    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(())
    /// # }
    /// ```
    /// # Errors
    ///
    /// Returns an error if the XREAD command fails.
    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
                            if let Some(id_val) = entry_parts.first() {
                                let id = Self::value_to_string(id_val);

                                // Second element is the field-value array
                                if let Some(redis::Value::Array(field_values)) = entry_parts.get(1)
                                {
                                    let mut fields = Vec::new();

                                    // Process pairs of field-value
                                    for chunk in field_values.chunks(2) {
                                        if chunk.len() == 2 {
                                            if let (Some(field_val), Some(value_val)) =
                                                (chunk.first(), chunk.get(1))
                                            {
                                                let field = Self::value_to_string(field_val);
                                                let value = Self::value_to_string(value_val);
                                                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 Some(redis::Value::Array(entries)) = stream_parts.get(1) {
                                for entry in entries {
                                    if let redis::Value::Array(entry_parts) = entry {
                                        if entry_parts.len() >= 2 {
                                            if let Some(id_val) = entry_parts.first() {
                                                let id = Self::value_to_string(id_val);

                                                if let Some(redis::Value::Array(field_values)) =
                                                    entry_parts.get(1)
                                                {
                                                    let mut fields = Vec::new();

                                                    for chunk in field_values.chunks(2) {
                                                        if chunk.len() == 2 {
                                                            if let (
                                                                Some(field_val),
                                                                Some(value_val),
                                                            ) = (chunk.first(), chunk.get(1))
                                                            {
                                                                let field = Self::value_to_string(
                                                                    field_val,
                                                                );
                                                                let value = Self::value_to_string(
                                                                    value_val,
                                                                );
                                                                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(),
            _ => 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)
    }
}