oxidite_queue/
redis.rs

1use async_trait::async_trait;
2use redis::{Client, AsyncCommands};
3use crate::{QueueBackend, job::JobWrapper, Result, QueueError};
4
5/// Redis queue backend
6pub struct RedisBackend {
7    client: Client,
8    queue_key: String,
9    dlq_key: String,
10}
11
12impl RedisBackend {
13    /// Create a new Redis queue backend
14    pub fn new(url: &str, queue_key: &str) -> Result<Self> {
15        let client = Client::open(url)
16            .map_err(|e| QueueError::BackendError(e.to_string()))?;
17        
18        let dlq_key = format!("{}_dlq", queue_key);
19        
20        Ok(Self {
21            client,
22            queue_key: queue_key.to_string(),
23            dlq_key,
24        })
25    }
26}
27
28#[async_trait]
29impl QueueBackend for RedisBackend {
30    async fn enqueue(&self, job: JobWrapper) -> Result<()> {
31        let mut conn = self.client.get_multiplexed_async_connection()
32            .await
33            .map_err(|e| QueueError::BackendError(e.to_string()))?;
34            
35        let payload = serde_json::to_string(&job)?;
36        
37        // Use LPUSH to add to the head of the list (or tail, depending on how we want to process)
38        // Standard queue is usually LPUSH (enqueue) and RPOP (dequeue)
39        let _: () = conn.lpush(&self.queue_key, payload)
40            .await
41            .map_err(|e| QueueError::BackendError(e.to_string()))?;
42            
43        Ok(())
44    }
45
46    async fn dequeue(&self) -> Result<Option<JobWrapper>> {
47        let mut conn = self.client.get_multiplexed_async_connection()
48            .await
49            .map_err(|e| QueueError::BackendError(e.to_string()))?;
50            
51        // RPOP removes and returns the last element of the list
52        let result: Option<String> = conn.rpop(&self.queue_key, None)
53            .await
54            .map_err(|e| QueueError::BackendError(e.to_string()))?;
55            
56        if let Some(payload) = result {
57            let job: JobWrapper = serde_json::from_str(&payload)?;
58            Ok(Some(job))
59        } else {
60            Ok(None)
61        }
62    }
63
64    async fn complete(&self, _job_id: &str) -> Result<()> {
65        // In a simple RPOP implementation, the job is already removed from the queue.
66        // For more reliability, we'd use RPOPLPUSH to a processing queue and then remove from there.
67        // For this v1 implementation, we'll keep it simple.
68        Ok(())
69    }
70
71    async fn fail(&self, job_id: &str, error: String) -> Result<()> {
72        // Move failed jobs to a separate failed queue for inspection
73        let failed_key = format!("{}_failed", self.queue_key);
74        
75        let mut conn = self.client.get_multiplexed_async_connection()
76            .await
77            .map_err(|e| QueueError::BackendError(e.to_string()))?;
78        
79        // Store the failed job with error information
80        let failed_job = serde_json::json!({
81            "job_id": job_id,
82            "error": error,
83            "failed_at": chrono::Utc::now().to_rfc3339()
84        });
85        
86        let payload = serde_json::to_string(&failed_job)
87            .map_err(|e| QueueError::SerializationError(e))?;
88        
89        let _: () = conn.lpush(&failed_key, payload)
90            .await
91            .map_err(|e| QueueError::BackendError(e.to_string()))?;
92        
93        Ok(())
94    }
95
96    async fn retry(&self, job: JobWrapper) -> Result<()> {
97        self.enqueue(job).await
98    }
99
100    async fn move_to_dead_letter(&self, job: JobWrapper) -> Result<()> {
101        let mut conn = self.client.get_multiplexed_async_connection()
102            .await
103            .map_err(|e| QueueError::BackendError(e.to_string()))?;
104            
105        let payload = serde_json::to_string(&job)?;
106        let _: () = conn.lpush(&self.dlq_key, payload)
107            .await
108            .map_err(|e| QueueError::BackendError(e.to_string()))?;
109            
110        Ok(())
111    }
112
113    async fn list_dead_letter(&self) -> Result<Vec<JobWrapper>> {
114        let mut conn = self.client.get_multiplexed_async_connection()
115            .await
116            .map_err(|e| QueueError::BackendError(e.to_string()))?;
117            
118        let results: Vec<String> = conn.lrange(&self.dlq_key, 0, -1)
119            .await
120            .map_err(|e| QueueError::BackendError(e.to_string()))?;
121            
122        let mut jobs = Vec::new();
123        for payload in results {
124            if let Ok(job) = serde_json::from_str::<JobWrapper>(&payload) {
125                jobs.push(job);
126            }
127        }
128        
129        Ok(jobs)
130    }
131
132    async fn retry_from_dead_letter(&self, job_id: &str) -> Result<()> {
133        let jobs = self.list_dead_letter().await?;
134        
135        // Find the job by ID
136        if let Some(job) = jobs.iter().find(|j| j.id == job_id) {
137            // Remove from DLQ
138            let mut conn = self.client.get_multiplexed_async_connection()
139                .await
140                .map_err(|e| QueueError::BackendError(e.to_string()))?;
141                
142            let payload = serde_json::to_string(&job)?;
143            let _: () = conn.lrem(&self.dlq_key, 1, payload)
144                .await
145                .map_err(|e| QueueError::BackendError(e.to_string()))?;
146            
147            // Clone and reset before re-enqueue
148            let mut job_clone = job.clone();
149            job_clone.status = crate::job::JobStatus::Pending;
150            job_clone.attempts = 0;
151            job_clone.error = None;
152            self.enqueue(job_clone).await?;
153        }
154        
155        Ok(())
156    }
157
158    async fn clear(&self) -> Result<()> {
159        let mut conn = self.client.get_multiplexed_async_connection()
160            .await
161            .map_err(|e| QueueError::BackendError(e.to_string()))?;
162
163        let _: () = conn.del(&self.queue_key)
164            .await
165            .map_err(|e| QueueError::BackendError(e.to_string()))?;
166        Ok(())
167    }
168}
oxidite_queue/redis.rs

oxidite_queue/
redis.rs
