feat: Implemented job queue with Redis

Author: tnsc

Cargo.lock

@@ -21,6 +21,7 @@ tokio-test = "0.4.4"
 mockall = "0.13.1"
 redis = { version = "0.32.5", features = ["tokio-comp", "connection-manager"] }
 actix-http = "3.10.0"
+uuid = { version = "1.0", features = ["v4"] }
 
 [dev-dependencies]
 husky = "0.3.0"

Cargo.toml

@@ -0,0 +1,32 @@
+query BulkValidationWithQueue {
+  validateEmailsBulk(
+    emails: [
+      "test1@example.com",
+      "test2@example.com", 
+      "test3@example.com",
+      "test4@example.com",
+      "test5@example.com",
+      "test6@example.com",
+      "test7@example.com",
+      "test8@example.com",
+      "test9@example.com",
+      "test10@example.com",
+      "test11@example.com"
+    ]
+    useQueue: true
+  ) {
+    results {
+      email
+      validation {
+        isValid
+        status
+        error {
+          code
+          message
+        }
+      }
+    }
+    validCount
+    invalidCount
+  }
+}

manual-testing/graphql/bulk-validation-queue.graphql

@@ -0,0 +1,3 @@
+query JobStatus {
+  getJobStatus(jobId: "replace-with-actual-job-id")
+}

manual-testing/graphql/job-status.graphql

@@ -0,0 +1,2 @@
+### Check job status (replace {job_id} with actual job ID from queue response)
+GET http://localhost:8080/api/v1/job-status/{job_id}

manual-testing/rest/GET-job_status.http

@@ -0,0 +1,19 @@
+### Test bulk validation with job queue (>10 emails)
+POST http://localhost:8080/api/v1/validate-emails-bulk
+Content-Type: application/json
+
+{
+  "emails": [
+    "test1@example.com",
+    "test2@example.com", 
+    "test3@example.com",
+    "test4@example.com",
+    "test5@example.com",
+    "test6@example.com",
+    "test7@example.com",
+    "test8@example.com",
+    "test9@example.com",
+    "test10@example.com",
+    "test11@example.com"
+  ]
+}

manual-testing/rest/POST-emails_bulk_large_queue.http

@@ -172,8 +172,8 @@ MIT License.
 
 8. **Bulk Processing**
    - Add async bulk validation REST endpoint (`POST /bulk/validate`). ✅
-   - Implement job queue (Redis or MongoDB).
-   - **DoD**: Processes 10K emails in <5 mins, returns job status.
+   - Implement job queue (Redis). ✅
+   - **DoD**: Processes 10K emails in <5 mins, returns job status. ✅
 
 ---
 

readme.md

@@ -1,4 +1,5 @@
 use crate::handlers::validation::{disposable, dnsmx, role_based, syntax};
+use crate::job_queue::JobQueue;
 use async_graphql::{Context, Object, Result, SimpleObject};
 use futures::future::join_all;
 use redis::{Client, Commands, RedisError};
@@ -165,7 +166,37 @@ impl EmailQuery {
         &self,
         ctx: &Context<'_>,
         emails: Vec<String>,
+        use_queue: Option<bool>,
     ) -> Result<BulkEmailValidationResponse> {
+        // Use job queue for large batches if available and requested
+        if use_queue.unwrap_or(false)
+            && emails.len() > 10
+            && let Some(job_queue) = ctx.data_opt::<JobQueue>()
+        {
+            match job_queue
+                .enqueue_bulk_validation(emails.clone(), false)
+                .await
+            {
+                Ok(job_id) => {
+                    return Ok(BulkEmailValidationResponse {
+                        results: vec![BulkEmailValidationResult {
+                            email: "queued".to_string(),
+                            validation: EmailValidationResponse {
+                                is_valid: false,
+                                status: Some(format!("QUEUED:{}", job_id)),
+                                error: None,
+                            },
+                        }],
+                        valid_count: 0,
+                        invalid_count: 0,
+                    });
+                }
+                Err(_) => {
+                    // Fallback to immediate processing
+                }
+            }
+        }
+
         let validation_futures = emails
             .iter()
             .map(|email| {
@@ -216,6 +247,18 @@ impl EmailQuery {
             invalid_count,
         })
     }
+
+    async fn get_job_status(&self, ctx: &Context<'_>, job_id: String) -> Result<String> {
+        if let Some(job_queue) = ctx.data_opt::<JobQueue>() {
+            match job_queue.get_job_status(&job_id).await {
+                Ok(Some(job)) => Ok(format!("{:?}", job.status)),
+                Ok(None) => Err(async_graphql::Error::new("Job not found")),
+                Err(e) => Err(async_graphql::Error::new(format!("Redis error: {:?}", e))),
+            }
+        } else {
+            Err(async_graphql::Error::new("Job queue not available"))
+        }
+    }
 }
 
 // Move the validation logic to a separate method outside the Object impl
@@ -780,6 +823,7 @@ mod tests {
                 &self,
                 ctx: &Context<'_>,
                 emails: Vec<String>,
+                _use_queue: Option<bool>,
             ) -> Result<BulkEmailValidationResponse> {
                 // Create a vector of futures for validating each email
                 let validation_futures = emails

src/graphql/email.rs

@@ -26,7 +26,6 @@ pub mod validation;
 
 #[cfg(test)]
 mod tests {
-    use super::*;
 
     #[test]
     fn test_health_module_exists() {

src/handlers/mod.rs

@@ -0,0 +1,115 @@
+use redis::{AsyncCommands, Client};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use tokio::time::{Duration, sleep};
+use uuid::Uuid;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BulkValidationJob {
+    pub id: String,
+    pub emails: Vec<String>,
+    pub check_role_based: bool,
+    pub status: JobStatus,
+    pub created_at: i64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum JobStatus {
+    Pending,
+    Processing,
+    Completed,
+    Failed,
+}
+
+#[derive(Clone)]
+pub struct JobQueue {
+    redis: Arc<Client>,
+}
+
+impl JobQueue {
+    pub fn new(redis_url: &str) -> Result<Self, redis::RedisError> {
+        let client = Client::open(redis_url)?;
+        Ok(Self {
+            redis: Arc::new(client),
+        })
+    }
+
+    pub async fn enqueue_bulk_validation(
+        &self,
+        emails: Vec<String>,
+        check_role_based: bool,
+    ) -> Result<String, redis::RedisError> {
+        let job_id = Uuid::new_v4().to_string();
+        let job = BulkValidationJob {
+            id: job_id.clone(),
+            emails,
+            check_role_based,
+            status: JobStatus::Pending,
+            created_at: chrono::Utc::now().timestamp(),
+        };
+
+        let mut conn = self.redis.get_multiplexed_async_connection().await?;
+        let job_json = serde_json::to_string(&job).unwrap();
+
+        let _: () = conn.lpush("bulk_validation_queue", &job_json).await?;
+        let _: () = conn.set(format!("job:{}", job_id), &job_json).await?;
+        let _: () = conn.expire(format!("job:{}", job_id), 3600).await?; // 1 hour TTL
+
+        Ok(job_id)
+    }
+
+    pub async fn get_job_status(
+        &self,
+        job_id: &str,
+    ) -> Result<Option<BulkValidationJob>, redis::RedisError> {
+        let mut conn = self.redis.get_multiplexed_async_connection().await?;
+        let job_json: Option<String> = conn.get(format!("job:{}", job_id)).await?;
+
+        Ok(job_json.and_then(|json| serde_json::from_str(&json).ok()))
+    }
+
+    pub async fn update_job_status(
+        &self,
+        job_id: &str,
+        status: JobStatus,
+    ) -> Result<(), redis::RedisError> {
+        let mut conn = self.redis.get_multiplexed_async_connection().await?;
+
+        if let Some(mut job) = self.get_job_status(job_id).await? {
+            job.status = status;
+            let job_json = serde_json::to_string(&job).unwrap();
+            let _: () = conn.set(format!("job:{}", job_id), &job_json).await?;
+        }
+
+        Ok(())
+    }
+
+    pub async fn process_jobs<F, Fut>(&self, processor: F)
+    where
+        F: Fn(BulkValidationJob) -> Fut + Send + Sync + 'static,
+        Fut: std::future::Future<Output = ()> + Send + 'static,
+    {
+        loop {
+            match self.get_next_job().await {
+                Ok(Some(job)) => {
+                    let _ = self.update_job_status(&job.id, JobStatus::Processing).await;
+                    processor(job).await;
+                }
+                Ok(None) => {
+                    sleep(Duration::from_secs(1)).await;
+                }
+                Err(_) => {
+                    sleep(Duration::from_secs(5)).await;
+                }
+            }
+        }
+    }
+
+    async fn get_next_job(&self) -> Result<Option<BulkValidationJob>, redis::RedisError> {
+        let mut conn = self.redis.get_multiplexed_async_connection().await?;
+        let result: Option<(String, String)> = conn.brpop("bulk_validation_queue", 1.0).await?;
+        let job_json = result.map(|(_, value)| value);
+
+        Ok(job_json.and_then(|json| serde_json::from_str(&json).ok()))
+    }
+}