ri/queue/backends/
redis_backend.rs

1//! Copyright © 2025-2026 Wenze Wei. All Rights Reserved.
2//!
3//! This file is part of Ri.
4//! The Ri project belongs to the Dunimd Team.
5//!
6//! Licensed under the Apache License, Version 2.0 (the "License");
7//! You may not use this file except in compliance with the License.
8//! You may obtain a copy of the License at
9//!
10//!     http://www.apache.org/licenses/LICENSE-2.0
11//!
12//! Unless required by applicable law or agreed to in writing, software
13//! distributed under the License is distributed on an "AS IS" BASIS,
14//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15//! See the License for the specific language governing permissions and
16//! limitations under the License.
17
18#![allow(non_snake_case)]
19
20//! # Redis Queue Backend
21//! 
22//! This module provides a Redis implementation for the Ri queue system. It allows
23//! sending and receiving messages using Redis lists as the underlying message broker.
24//! 
25//! ## Key Components
26//! 
27//! - **RiRedisQueue**: Main Redis queue implementation
28//! - **RedisQueueProducer**: Redis producer implementation
29//! - **RedisQueueConsumer**: Redis consumer implementation
30//! 
31//! ## Design Principles
32//! 
33//! 1. **Async Trait Implementation**: Implements the RiQueue, RiQueueProducer, and RiQueueConsumer traits
34//! 2. **Redis Integration**: Uses the redis crate for Redis connectivity
35//! 3. **Thread Safety**: Uses Arc for safe sharing of connections and consumers
36//! 4. **Future-based API**: Leverages async/await for non-blocking operations
37//! 5. **Blocking Operations**: Uses BLPOP for efficient blocking message consumption
38//! 6. **Error Handling**: Comprehensive error handling with RiResult
39//! 7. **List-based Queue**: Uses Redis lists for simple FIFO queue functionality
40//! 8. **Batch Support**: Provides batch sending functionality
41//! 9. **Implicit Acknowledgment**: Acknowledgment is implicit when messages are popped from the list
42//! 10. **Stats Support**: Provides queue length statistics using Redis LLEN command
43//! 
44//! ## Usage
45//! 
46//! ```rust
47//! use ri::prelude::*;
48//! 
49//! async fn example() -> RiResult<()> {
50//!     // Create a new Redis queue
51//!     let queue = RiRedisQueue::new("test-queue", "redis://localhost:6379").await?;
52//!     
53//!     // Create a producer
54//!     let producer = queue.create_producer().await?;
55//!     
56//!     // Create a message
57//!     let message = RiQueueMessage {
58//!         id: "12345".to_string(),
59//!         payload: b"Hello, Redis!".to_vec(),
60//!         headers: vec![("key1".to_string(), "value1".to_string())],
61//!         timestamp: chrono::Utc::now().timestamp_millis() as u64,
62//!         priority: 0,
63//!     };
64//!     
65//!     // Send the message
66//!     producer.send(message).await?;
67//!     
68//!     // Create a consumer
69//!     let consumer = queue.create_consumer("test-consumer-group").await?;
70//!     
71//!     // Receive messages
72//!     if let Some(received_message) = consumer.receive().await? {
73//!         println!("Received message: {:?}", received_message);
74//!         consumer.ack(&received_message.id).await?;
75//!     }
76//!     
77//!     Ok(())
78//! }
79//! ```
80
81use async_trait::async_trait;
82use redis::{AsyncCommands, Client};
83use std::sync::Arc;
84use std::time::Duration;
85use tokio::sync::Mutex;
86use crate::core::RiResult;
87use crate::queue::{RiQueue, RiQueueMessage, RiQueueProducer, RiQueueConsumer, RiQueueStats};
88
89/// Redis queue implementation for the Ri queue system.
90///
91/// This struct provides a Redis implementation of the RiQueue trait, allowing
92/// sending and receiving messages using Redis lists as the underlying message broker.
93pub struct RiRedisQueue {
94    /// Queue name (Redis key)
95    name: String,
96    /// Redis client for connecting to Redis
97    client: Arc<Client>,
98}
99
100impl RiRedisQueue {
101    /// Creates a new Redis queue instance.
102    ///
103    /// # Parameters
104    ///
105    /// - `name`: The name of the queue (Redis key)
106    /// - `connection_string`: The Redis connection string
107    ///
108    /// # Returns
109    ///
110    /// A new RiRedisQueue instance wrapped in RiResult
111    pub async fn new(name: &str, connection_string: &str) -> RiResult<Self> {
112        let client = Client::open(connection_string)?;
113        Self::new_with_client(name, client)
114    }
115
116    /// Creates a new Redis queue instance with an existing client.
117    ///
118    /// # Parameters
119    ///
120    /// - `name`: The name of the queue (Redis key)
121    /// - `client`: The existing Redis client
122    ///
123    /// # Returns
124    ///
125    /// A new RiRedisQueue instance wrapped in RiResult
126    pub fn new_with_client(name: &str, client: Client) -> RiResult<Self> {
127        Ok(Self {
128            name: name.to_string(),
129            client: Arc::new(client),
130        })
131    }
132
133    /// Creates a new Redis queue instance with an existing connection.
134    ///
135    /// # Parameters
136    ///
137    /// - `name`: The name of the queue (Redis key)
138    /// - `connection_string`: The Redis connection string
139    ///
140    /// # Returns
141    ///
142    /// A new RiRedisQueue instance wrapped in RiResult
143    pub async fn new_with_connection(name: &str, connection_string: &str) -> RiResult<Self> {
144        let client = Client::open(connection_string)?;
145        Ok(Self {
146            name: name.to_string(),
147            client: Arc::new(client),
148        })
149    }
150}
151
152#[async_trait]
153impl RiQueue for RiRedisQueue {
154    /// Creates a new producer for the Redis queue.
155    ///
156    /// # Returns
157    ///
158    /// A new RiQueueProducer instance wrapped in RiResult
159    async fn create_producer(&self) -> RiResult<Box<dyn RiQueueProducer>> {
160        let conn = self.client.get_async_connection().await?;
161        
162        Ok(Box::new(RedisQueueProducer {
163            connection: Arc::new(Mutex::new(conn)),
164            queue_name: self.name.clone(),
165        }))
166    }
167
168    /// Creates a new consumer for the Redis queue.
169    ///
170    /// # Parameters
171    ///
172    /// - `_consumer_group`: The consumer group name (ignored in this implementation)
173    ///
174    /// # Returns
175    ///
176    /// A new RiQueueConsumer instance wrapped in RiResult
177    async fn create_consumer(&self, _consumer_group: &str) -> RiResult<Box<dyn RiQueueConsumer>> {
178        let conn = self.client.get_async_connection().await?;
179        
180        Ok(Box::new(RedisQueueConsumer {
181            connection: Arc::new(Mutex::new(conn)),
182            queue_name: self.name.clone(),
183            paused: Arc::new(Mutex::new(false)),
184        }))
185    }
186
187    /// Gets statistics for the Redis queue.
188    ///
189    /// # Returns
190    ///
191    /// RiQueueStats containing queue statistics wrapped in RiResult
192    async fn get_stats(&self) -> RiResult<RiQueueStats> {
193        let mut conn = self.client.get_async_connection().await?;
194        let len: i64 = conn.llen(&self.name).await?;
195        
196        Ok(RiQueueStats {
197            queue_name: self.name.clone(),
198            message_count: len as u64,
199            consumer_count: 0,
200            producer_count: 0,
201            processed_messages: 0,
202            failed_messages: 0,
203            avg_processing_time_ms: 0.0,
204            total_bytes_sent: 0,
205            total_bytes_received: 0,
206            last_message_time: 0,
207        })
208    }
209
210    /// Purges all messages from the Redis queue.
211    ///
212    /// # Returns
213    ///
214    /// RiResult indicating success or failure
215    async fn purge(&self) -> RiResult<()> {
216        let mut conn = self.client.get_async_connection().await?;
217        conn.del::<_, ()>(&self.name).await?;
218        Ok(())
219    }
220
221    /// Deletes the Redis queue.
222    ///
223    /// Note: This implementation simply calls purge since deleting a Redis key
224    /// is the same as purging all messages from the queue.
225    ///
226    /// # Returns
227    ///
228    /// RiResult indicating success or failure
229    async fn delete(&self) -> RiResult<()> {
230        self.purge().await
231    }
232}
233
234/// Redis queue producer implementation.
235///
236/// This struct provides a Redis implementation of the RiQueueProducer trait,
237/// allowing sending messages to Redis queues.
238struct RedisQueueProducer {
239    /// Redis async connection
240    connection: Arc<Mutex<redis::aio::Connection>>,
241    /// Queue name (Redis key) to send messages to
242    queue_name: String,
243}
244
245#[async_trait]
246impl RiQueueProducer for RedisQueueProducer {
247    /// Sends a single message to the Redis queue.
248    ///
249    /// # Parameters
250    ///
251    /// - `message`: The message to send
252    ///
253    /// # Returns
254    ///
255    /// RiResult indicating success or failure
256    async fn send(&self, message: RiQueueMessage) -> RiResult<()> {
257        let mut conn = self.connection.lock().await;
258        let payload = serde_json::to_vec(&message)?;
259        
260        conn.rpush::<_, _, ()>(&self.queue_name, payload).await?;
261        Ok(())
262    }
263
264    /// Sends multiple messages to the Redis queue.
265    ///
266    /// # Parameters
267    ///
268    /// - `messages`: A vector of messages to send
269    ///
270    /// # Returns
271    ///
272    /// RiResult indicating success or failure
273    async fn send_batch(&self, messages: Vec<RiQueueMessage>) -> RiResult<()> {
274        let mut conn = self.connection.lock().await;
275        
276        for message in messages {
277            let payload = serde_json::to_vec(&message)?;
278            conn.rpush::<_, _, ()>(&self.queue_name, payload).await?;
279        }
280        Ok(())
281    }
282}
283
284/// Redis queue consumer implementation.
285///
286/// This struct provides a Redis implementation of the RiQueueConsumer trait,
287/// allowing receiving messages from Redis queues.
288struct RedisQueueConsumer {
289    /// Redis async connection
290    connection: Arc<Mutex<redis::aio::Connection>>,
291    /// Queue name (Redis key) to receive messages from
292    queue_name: String,
293    /// Flag indicating if the consumer is paused
294    paused: Arc<Mutex<bool>>,
295}
296
297#[async_trait]
298impl RiQueueConsumer for RedisQueueConsumer {
299    /// Receives a message from the Redis queue.
300    ///
301    /// # Returns
302    ///
303    /// An Option containing the received message, or None if the consumer is paused
304    /// or the BLPOP operation timed out
305    async fn receive(&self) -> RiResult<Option<RiQueueMessage>> {
306        let paused = *self.paused.lock().await;
307        if paused {
308            return Ok(None);
309        }
310
311        let mut conn = self.connection.lock().await;
312        
313        // Use BLPOP for blocking pop with timeout
314        let result: Option<(String, Vec<u8>)> = conn.blpop(&self.queue_name, 5.0).await?;
315        
316        if let Some((_, payload)) = result {
317            // Security: Check message size before deserialization to prevent DoS
318            const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024; // 10 MB
319            if payload.len() > MAX_MESSAGE_SIZE {
320                log::warn!(
321                    "[Ri.Redis.Queue] Message too large: {} bytes (max {} bytes), rejecting",
322                    payload.len(), MAX_MESSAGE_SIZE
323                );
324                return Err(crate::core::RiError::Other(format!(
325                    "Message too large: {} bytes (max {} bytes)",
326                    payload.len(), MAX_MESSAGE_SIZE
327                )));
328            }
329            
330            // Security: Use bounded deserialization to prevent stack overflow
331            let message: RiQueueMessage = serde_json::from_slice(&payload)
332                .map_err(|e| {
333                    log::warn!("[Ri.Redis.Queue] Failed to deserialize message: {}", e);
334                    crate::core::RiError::Other(format!("Failed to deserialize message: {}", e))
335                })?;
336            Ok(Some(message))
337        } else {
338            Ok(None)
339        }
340    }
341
342    /// Acknowledges a message.
343    ///
344    /// Note: In Redis list-based queues, acknowledgment is implicit when messages
345    /// are popped from the list. This method is a no-op in this implementation.
346    ///
347    /// # Parameters
348    ///
349    /// - `_message_id`: The message ID to acknowledge (ignored in this implementation)
350    ///
351    /// # Returns
352    ///
353    /// RiResult indicating success or failure
354    async fn ack(&self, _message_id: &str) -> RiResult<()> {
355        // In Redis list-based queue, acknowledgment is implicit when message is popped
356        Ok(())
357    }
358
359    /// Negatively acknowledges a message.
360    ///
361    /// This implementation handles message retry by pushing the message back to the queue
362    /// with appropriate retry logic and delay mechanisms. It tracks retry counts and
363    /// implements exponential backoff for retry delays.
364    ///
365    /// # Parameters
366    ///
367    /// - `message_id`: The message ID to negatively acknowledge
368    ///
369    /// # Returns
370    ///
371    /// RiResult indicating success or failure
372    async fn nack(&self, message_id: &str) -> RiResult<()> {
373        log::info!("Message negatively acknowledged: {message_id}");
374
375        let mut conn = self.connection.lock().await;
376
377        let (original_data, retry_count): (Option<Vec<u8>>, u32) = conn.hgetall::<_, (Option<Vec<u8>>, u32)>(&format!("{}_meta", self.queue_name)).await
378            .map(|(data, count)| (data, count))
379            .unwrap_or((None, 0));
380
381        let max_retries = 3u32;
382        let base_delay_ms = 1000u64;
383
384        if retry_count >= max_retries {
385            log::warn!("Message {message_id} exceeded max retries ({max_retries}), moving to dead letter queue");
386            conn.rpush::<_, _, ()>(&format!("{}_dlq", self.queue_name), message_id.as_bytes()).await?;
387            conn.hdel::<_, &str, ()>(&format!("{}_meta", self.queue_name), "retry_count").await?;
388            return Ok(());
389        }
390
391        let new_retry_count = retry_count + 1;
392        let retry_delay_ms = base_delay_ms * (2u64.pow(new_retry_count - 1));
393
394        log::info!("Message {message_id} scheduled for retry {new_retry_count}/{max_retries} after {retry_delay_ms}ms delay");
395
396        conn.hset::<_, &str, u32, ()>(&format!("{}_meta", self.queue_name), "retry_count", new_retry_count).await?;
397
398        tokio::time::sleep(Duration::from_millis(retry_delay_ms)).await;
399
400        if let Some(data) = original_data {
401            conn.rpush::<_, _, ()>(&self.queue_name, &data).await?;
402        }
403
404        Ok(())
405    }
406
407    /// Pauses the consumer.
408    ///
409    /// # Returns
410    ///
411    /// RiResult indicating success or failure
412    async fn pause(&self) -> RiResult<()> {
413        let mut paused = self.paused.lock().await;
414        *paused = true;
415        Ok(())
416    }
417
418    /// Resumes the consumer.
419    ///
420    /// # Returns
421    ///
422    /// RiResult indicating success or failure
423    async fn resume(&self) -> RiResult<()> {
424        let mut paused = self.paused.lock().await;
425        *paused = false;
426        Ok(())
427    }
428}
ri/queue/backends/redis_backend.rs

ri/queue/backends/
redis_backend.rs
