pleme_redis/

job_queue.rs

//! Redis-based FIFO job queue using LPUSH/BRPOP pattern
//!
//! This module provides a Redis-backed job queue with:
//! - FIFO ordering (jobs processed in submission order)
//! - Blocking pop with timeout (BRPOP)
//! - Atomic operations
//! - Multiple worker support
//! - Distributed locking for job de-duplication

use redis::aio::ConnectionManager;
use redis::{AsyncCommands, RedisError};
use serde::{Deserialize, Serialize};
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::time::Duration;
use thiserror::Error;
use tokio::time::timeout;
use uuid::Uuid;

use crate::error::RedisError as PlemeRedisError;

/// Calculate hash of job parameters for unique lock keys
///
/// This creates a unique identifier for each parameter combination,
/// allowing concurrent execution of jobs with different parameters.
fn calculate_hash(value: &serde_json::Value) -> String {
    let mut hasher = DefaultHasher::new();
    // Serialize to string for consistent hashing
    value.to_string().hash(&mut hasher);
    format!("{:x}", hasher.finish())
}

/// Redis-based FIFO job queue using LPUSH/BRPOP pattern
///
/// This implementation provides:
/// - FIFO ordering (jobs processed in submission order)
/// - Blocking pop with timeout (BRPOP)
/// - Atomic operations
/// - Multiple worker support
pub struct JobQueue {
    conn: ConnectionManager,
    queue_key: String,
}

/// Job message in Redis queue
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueuedJob {
    pub job_id: Uuid,
    pub job_type: String,
    pub enqueued_at: chrono::DateTime<chrono::Utc>,
}

impl JobQueue {
    /// Create new Redis queue
    ///
    /// # Arguments
    /// * `conn` - Redis multiplexed connection
    /// * `queue_key` - Redis key for the queue (e.g., "sync_jobs:queue")
    pub fn new(conn: ConnectionManager, queue_key: String) -> Self {
        Self { conn, queue_key }
    }

    /// Enqueue a job (LPUSH)
    ///
    /// Adds job to the left side of the list (head). Workers pop from the right (tail).
    /// This creates FIFO behavior.
    ///
    /// # Arguments
    /// * `job_id` - Job ID to enqueue
    /// * `job_type` - Job type string
    ///
    /// # Returns
    /// * `Ok(())` - Job enqueued successfully
    /// * `Err(QueueError)` - Failed to enqueue
    pub async fn enqueue(&mut self, job_id: Uuid, job_type: String) -> Result<(), QueueError> {
        tracing::info!(
            "📮 [REDIS_QUEUE] Enqueueing job - id: {}, type: {}, queue: {}",
            job_id,
            job_type,
            self.queue_key
        );

        let message = QueuedJob {
            job_id,
            job_type: job_type.clone(),
            enqueued_at: chrono::Utc::now(),
        };

        let serialized = serde_json::to_string(&message)
            .map_err(|e| {
                tracing::error!("❌ [REDIS_QUEUE] Serialization error for job {}: {}", job_id, e);
                QueueError::SerializationError(e.to_string())
            })?;

        tracing::debug!("💾 [REDIS_QUEUE] Pushing job {} to Redis list '{}'", job_id, self.queue_key);

        self.conn
            .lpush::<_, _, ()>(&self.queue_key, serialized)
            .await
            .map_err(|e| {
                tracing::error!("❌ [REDIS_QUEUE] Redis LPUSH error for job {}: {}", job_id, e);
                QueueError::RedisError(e.to_string())
            })?;

        tracing::info!("✅ [REDIS_QUEUE] Job {} enqueued successfully (type: {})", job_id, job_type);

        Ok(())
    }

    /// Dequeue a job with blocking wait (BRPOP)
    ///
    /// Blocks until a job is available or timeout is reached.
    ///
    /// # Arguments
    /// * `timeout_seconds` - Maximum time to wait for a job (0 = wait forever)
    ///
    /// # Returns
    /// * `Ok(Some(QueuedJob))` - Job dequeued successfully
    /// * `Ok(None)` - Timeout reached, no job available
    /// * `Err(QueueError)` - Failed to dequeue
    pub async fn dequeue(&mut self, timeout_seconds: u64) -> Result<Option<QueuedJob>, QueueError> {
        tracing::debug!(
            "🔍 [REDIS_QUEUE] Waiting for job - queue: {}, timeout: {}s",
            self.queue_key,
            timeout_seconds
        );

        // BRPOP returns (key, value) tuple or nil
        let result: Option<(String, String)> = self
            .conn
            .brpop(&self.queue_key, timeout_seconds as f64)
            .await
            .map_err(|e| {
                tracing::error!("❌ [REDIS_QUEUE] Redis BRPOP error: {}", e);
                QueueError::RedisError(e.to_string())
            })?;

        match result {
            Some((_key, value)) => {
                let job: QueuedJob = serde_json::from_str(&value)
                    .map_err(|e| {
                        tracing::error!("❌ [REDIS_QUEUE] Deserialization error: {}", e);
                        QueueError::DeserializationError(e.to_string())
                    })?;

                tracing::info!(
                    "📥 [REDIS_QUEUE] Job dequeued - id: {}, type: {}, queue_age: {}s",
                    job.job_id,
                    job.job_type,
                    (chrono::Utc::now() - job.enqueued_at).num_seconds()
                );

                Ok(Some(job))
            }
            None => {
                tracing::debug!("⏱️  [REDIS_QUEUE] Dequeue timeout - no jobs available");
                Ok(None) // Timeout, no job available
            }
        }
    }

    /// Dequeue with Tokio timeout wrapper
    ///
    /// Prevents indefinite blocking by adding application-level timeout.
    ///
    /// # Arguments
    /// * `timeout_duration` - Maximum wait time
    ///
    /// # Returns
    /// * `Ok(Some(QueuedJob))` - Job dequeued
    /// * `Ok(None)` - Timeout reached
    /// * `Err(QueueError)` - Error occurred
    pub async fn dequeue_with_timeout(
        &mut self,
        timeout_duration: Duration,
    ) -> Result<Option<QueuedJob>, QueueError> {
        let timeout_seconds = timeout_duration.as_secs();

        match timeout(timeout_duration, self.dequeue(timeout_seconds)).await {
            Ok(Ok(job)) => Ok(job),
            Ok(Err(e)) => Err(e),
            Err(_) => Ok(None), // Tokio timeout
        }
    }

    /// Get queue length (LLEN)
    ///
    /// Returns the number of jobs currently in the queue.
    pub async fn length(&mut self) -> Result<u64, QueueError> {
        self.conn
            .llen(&self.queue_key)
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))
    }

    /// Peek at next job without removing it (LINDEX)
    ///
    /// Returns the job that would be dequeued next, without removing it.
    pub async fn peek(&mut self) -> Result<Option<QueuedJob>, QueueError> {
        let result: Option<String> = self
            .conn
            .lindex(&self.queue_key, -1) // Last element (right side)
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))?;

        match result {
            Some(value) => {
                let job: QueuedJob = serde_json::from_str(&value)
                    .map_err(|e| QueueError::DeserializationError(e.to_string()))?;
                Ok(Some(job))
            }
            None => Ok(None),
        }
    }

    /// Clear all jobs from queue (LTRIM 1 0)
    ///
    /// WARNING: This removes all jobs. Use with caution.
    pub async fn clear(&mut self) -> Result<(), QueueError> {
        self.conn
            .ltrim::<_, ()>(&self.queue_key, 1, 0) // Invalid range = clear all
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))?;

        Ok(())
    }

    /// Get queue statistics
    pub async fn stats(&mut self) -> Result<QueueStats, QueueError> {
        let length = self.length().await?;
        let oldest_job = self.peek().await?;

        let oldest_age = oldest_job.map(|job| {
            let now = chrono::Utc::now();
            let age = now - job.enqueued_at;
            age.num_seconds()
        });

        Ok(QueueStats {
            queue_length: length,
            oldest_job_age_seconds: oldest_age,
        })
    }

    /// Check if a job exists in the queue (LRANGE scan)
    ///
    /// Scans the queue to see if a job with the given ID exists.
    /// This is an O(N) operation, use sparingly.
    ///
    /// # Arguments
    /// * `job_id` - Job ID to search for
    ///
    /// # Returns
    /// * `Ok(true)` - Job exists in queue
    /// * `Ok(false)` - Job not found in queue
    /// * `Err(QueueError)` - Failed to check
    pub async fn contains(&mut self, job_id: Uuid) -> Result<bool, QueueError> {
        // Get all jobs in the queue (LRANGE 0 -1)
        let queue_items: Vec<String> = self
            .conn
            .lrange(&self.queue_key, 0, -1)
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))?;

        // Check if any item contains the job_id
        for item in queue_items {
            if let Ok(job) = serde_json::from_str::<QueuedJob>(&item) {
                if job.job_id == job_id {
                    return Ok(true);
                }
            }
        }

        Ok(false)
    }

    /// Attempt to acquire a distributed lock for a job type + parameters combination
    ///
    /// Uses Redis SET NX EX to atomically set a key with TTL only if it doesn't exist.
    /// This prevents multiple jobs of the same type with the same parameters from running
    /// within the lock window. Jobs with different parameters can run concurrently.
    ///
    /// # Arguments
    /// * `job_type` - Job type to lock (e.g., "FETCH_PROVIDER_PRODUCTS")
    /// * `parameters` - Job parameters to include in lock key (enables concurrent jobs with different params)
    /// * `ttl_seconds` - Lock duration in seconds (e.g., 1800 for 30 minutes, 172800 for 48 hours)
    ///
    /// # Returns
    /// * `Ok(true)` - Lock acquired successfully
    /// * `Ok(false)` - Lock already held by another job
    /// * `Err(QueueError)` - Redis error occurred
    pub async fn acquire_job_type_lock(
        &mut self,
        job_type: &str,
        parameters: &serde_json::Value,
        ttl_seconds: u64,
    ) -> Result<bool, QueueError> {
        // Hash parameters to create unique lock key per job variant
        let params_hash = calculate_hash(parameters);
        let lock_key = format!("job_lock:{}:{}", job_type, params_hash);

        tracing::debug!(
            "🔒 [REDIS_LOCK] Attempting to acquire lock - job_type: '{}', params_hash: '{}', TTL: {}s",
            job_type,
            params_hash,
            ttl_seconds
        );

        // SET NX EX: Set if Not eXists with EXpiration
        // Returns true if key was set, false if key already existed
        let acquired: bool = redis::cmd("SET")
            .arg(&lock_key)
            .arg("locked")
            .arg("NX")  // Only set if key doesn't exist
            .arg("EX")  // Set expiration in seconds
            .arg(ttl_seconds)
            .query_async(&mut self.conn)
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))?;

        if acquired {
            tracing::info!(
                "✅ [REDIS_LOCK] Lock acquired - job_type: '{}', params_hash: '{}', TTL: {}s ({}h)",
                job_type,
                params_hash,
                ttl_seconds,
                ttl_seconds / 3600
            );
        } else {
            tracing::warn!(
                "⏸️  [REDIS_LOCK] Lock already held - job_type: '{}', params_hash: '{}' - job will be skipped",
                job_type,
                params_hash
            );
        }

        Ok(acquired)
    }

    /// Check if a job type lock is currently held
    ///
    /// # Arguments
    /// * `job_type` - Job type to check
    /// * `parameters` - Job parameters
    ///
    /// # Returns
    /// * `Ok(true)` - Lock is held
    /// * `Ok(false)` - Lock is not held
    /// * `Err(QueueError)` - Redis error occurred
    pub async fn is_job_type_locked(
        &mut self,
        job_type: &str,
        parameters: &serde_json::Value,
    ) -> Result<bool, QueueError> {
        let params_hash = calculate_hash(parameters);
        let lock_key = format!("job_lock:{}:{}", job_type, params_hash);

        let exists: bool = self
            .conn
            .exists(&lock_key)
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))?;

        Ok(exists)
    }

    /// Get remaining TTL for a job type lock
    ///
    /// # Arguments
    /// * `job_type` - Job type to check
    /// * `parameters` - Job parameters
    ///
    /// # Returns
    /// * `Ok(Some(seconds))` - Lock exists with TTL
    /// * `Ok(None)` - Lock doesn't exist or has no TTL
    /// * `Err(QueueError)` - Redis error occurred
    pub async fn get_lock_ttl(
        &mut self,
        job_type: &str,
        parameters: &serde_json::Value,
    ) -> Result<Option<i64>, QueueError> {
        let params_hash = calculate_hash(parameters);
        let lock_key = format!("job_lock:{}:{}", job_type, params_hash);

        let ttl: i64 = self
            .conn
            .ttl(&lock_key)
            .await
            .map_err(|e| QueueError::RedisError(e.to_string()))?;

        // TTL returns -2 if key doesn't exist, -1 if key exists but has no expire
        match ttl {
            -2 => Ok(None),  // Key doesn't exist
            -1 => Ok(None),  // Key exists but no TTL (shouldn't happen with our locks)
            seconds => Ok(Some(seconds)),
        }
    }
}

/// Queue statistics
#[derive(Debug, Clone)]
pub struct QueueStats {
    pub queue_length: u64,
    pub oldest_job_age_seconds: Option<i64>,
}

/// Queue errors
#[derive(Error, Debug)]
pub enum QueueError {
    #[error("Redis error: {0}")]
    RedisError(String),

    #[error("Serialization error: {0}")]
    SerializationError(String),

    #[error("Deserialization error: {0}")]
    DeserializationError(String),

    #[error("Timeout: no job available")]
    Timeout,
}

impl From<RedisError> for QueueError {
    fn from(err: RedisError) -> Self {
        QueueError::RedisError(err.to_string())
    }
}

impl From<PlemeRedisError> for QueueError {
    fn from(err: PlemeRedisError) -> Self {
        QueueError::RedisError(err.to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_queued_job_serialization() {
        let job = QueuedJob {
            job_id: Uuid::new_v4(),
            job_type: "FETCH_PROVIDER_PRODUCTS".to_string(),
            enqueued_at: chrono::Utc::now(),
        };

        let serialized = serde_json::to_string(&job).unwrap();
        let deserialized: QueuedJob = serde_json::from_str(&serialized).unwrap();

        assert_eq!(job.job_id, deserialized.job_id);
        assert_eq!(job.job_type, deserialized.job_type);
    }

    #[test]
    fn test_queue_stats() {
        let stats = QueueStats {
            queue_length: 10,
            oldest_job_age_seconds: Some(120),
        };

        assert_eq!(stats.queue_length, 10);
        assert_eq!(stats.oldest_job_age_seconds, Some(120));
    }

    #[test]
    fn test_calculate_hash() {
        let params1 = serde_json::json!({"provider_id": "dropi", "category": "electronics"});
        let params2 = serde_json::json!({"provider_id": "dropi", "category": "electronics"});
        let params3 = serde_json::json!({"provider_id": "dropi", "category": "fashion"});

        let hash1 = calculate_hash(&params1);
        let hash2 = calculate_hash(&params2);
        let hash3 = calculate_hash(&params3);

        // Same parameters should produce same hash
        assert_eq!(hash1, hash2);

        // Different parameters should produce different hash
        assert_ne!(hash1, hash3);
    }
}