fesature: add Redis queue deduplication

CLAUDE.md

···
 - `QUEUE_BUFFER_SIZE`: Buffer size for MPSC queue (default: 1000)
 - `QUEUE_SQLITE_MAX_SIZE`: Max queue size for SQLite work shedding (default: 10000)
 - `QUEUE_REDIS_TIMEOUT`: Redis blocking timeout in seconds (default: 5)
+- `QUEUE_REDIS_DEDUP_ENABLED`: Enable queue deduplication to prevent duplicate handles (default: false)
+- `QUEUE_REDIS_DEDUP_TTL`: TTL for deduplication keys in seconds (default: 60)
 
 ### Optional - Rate Limiting
 - `RESOLVER_MAX_CONCURRENT`: Maximum concurrent handle resolutions (default: 0 = disabled)

README.md

···
 - `QUEUE_BUFFER_SIZE`: MPSC queue buffer size (default: 1000)
 - `QUEUE_REDIS_PREFIX`: Redis key prefix for queues (default: queue:handleresolver:)
 - `QUEUE_REDIS_TIMEOUT`: Redis blocking timeout in seconds (default: 5)
+- `QUEUE_REDIS_DEDUP_ENABLED`: Enable queue deduplication (default: false)
+- `QUEUE_REDIS_DEDUP_TTL`: TTL for deduplication keys in seconds (default: 60)
 - `QUEUE_SQLITE_MAX_SIZE`: Max SQLite queue size for work shedding (default: 10000)
 
 #### Rate Limiting

src/bin/quickdid.rs

···
     metrics::create_metrics_publisher,
     queue::{
         HandleResolutionWork, QueueAdapter, create_mpsc_queue_from_channel, create_noop_queue,
-        create_redis_queue, create_sqlite_queue, create_sqlite_queue_with_max_size,
+        create_redis_queue, create_redis_queue_with_dedup, create_sqlite_queue, 
+        create_sqlite_queue_with_max_size,
     },
     sqlite_schema::create_sqlite_pool,
     task_manager::spawn_cancellable_task,
···
                     "    QUEUE_REDIS_PREFIX  Redis key prefix for queues (default: queue:handleresolver:)"
                 );
                 println!("    QUEUE_REDIS_TIMEOUT Queue blocking timeout in seconds (default: 5)");
+                println!("    QUEUE_REDIS_DEDUP_ENABLED Enable queue deduplication (default: false)");
+                println!("    QUEUE_REDIS_DEDUP_TTL TTL for dedup keys in seconds (default: 60)");
                 println!("    QUEUE_WORKER_ID     Worker ID for Redis queue (default: worker1)");
                 println!("    QUEUE_BUFFER_SIZE   Buffer size for MPSC queue (default: 1000)");
                 println!("    QUEUE_SQLITE_MAX_SIZE Maximum SQLite queue size (default: 10000)");
···
                 if let Some(url) = queue_redis_url {
                     if let Some(pool) = try_create_redis_pool(url, "queue adapter") {
                         tracing::info!(
-                            "Creating Redis queue adapter with prefix: {}",
-                            config.queue_redis_prefix
+                            "Creating Redis queue adapter with prefix: {}, dedup: {}, dedup_ttl: {}s",
+                            config.queue_redis_prefix,
+                            config.queue_redis_dedup_enabled,
+                            config.queue_redis_dedup_ttl
                         );
-                        create_redis_queue::<HandleResolutionWork>(
-                            pool,
-                            config.queue_worker_id.clone(),
-                            config.queue_redis_prefix.clone(),
-                            config.queue_redis_timeout,
-                        )
+                        if config.queue_redis_dedup_enabled {
+                            create_redis_queue_with_dedup::<HandleResolutionWork>(
+                                pool,
+                                config.queue_worker_id.clone(),
+                                config.queue_redis_prefix.clone(),
+                                config.queue_redis_timeout,
+                                true,
+                                config.queue_redis_dedup_ttl,
+                            )
+                        } else {
+                            create_redis_queue::<HandleResolutionWork>(
+                                pool,
+                                config.queue_worker_id.clone(),
+                                config.queue_redis_prefix.clone(),
+                                config.queue_redis_timeout,
+                            )
+                        }
                     } else {
                         tracing::warn!("Falling back to MPSC queue adapter");
                         // Fall back to MPSC if Redis fails

src/config.rs

···
     /// Redis blocking timeout for queue operations in seconds (e.g., 5)
     pub queue_redis_timeout: u64,
 
+    /// Enable deduplication for Redis queue to prevent duplicate handles
+    /// Default: false
+    pub queue_redis_dedup_enabled: bool,
+
+    /// TTL for Redis queue deduplication keys in seconds
+    /// Default: 60 (1 minute)
+    pub queue_redis_dedup_ttl: u64,
+
     /// Maximum queue size for SQLite adapter work shedding (e.g., 10000)
     /// When exceeded, oldest entries are deleted to maintain this limit.
     /// Set to 0 to disable work shedding (unlimited queue size).
···
             cache_ttl_redis: parse_env("CACHE_TTL_REDIS", 7776000)?,
             cache_ttl_sqlite: parse_env("CACHE_TTL_SQLITE", 7776000)?,
             queue_redis_timeout: parse_env("QUEUE_REDIS_TIMEOUT", 5)?,
+            queue_redis_dedup_enabled: parse_env("QUEUE_REDIS_DEDUP_ENABLED", false)?,
+            queue_redis_dedup_ttl: parse_env("QUEUE_REDIS_DEDUP_TTL", 60)?,
             queue_sqlite_max_size: parse_env("QUEUE_SQLITE_MAX_SIZE", 10000)?,
             resolver_max_concurrent: parse_env("RESOLVER_MAX_CONCURRENT", 0)?,
             resolver_max_concurrent_timeout_ms: parse_env("RESOLVER_MAX_CONCURRENT_TIMEOUT_MS", 0)?,
···
         if self.queue_redis_timeout == 0 {
             return Err(ConfigError::InvalidTimeout(
                 "QUEUE_REDIS_TIMEOUT must be > 0".to_string(),
+            ));
+        }
+        if self.queue_redis_dedup_enabled && self.queue_redis_dedup_ttl == 0 {
+            return Err(ConfigError::InvalidTtl(
+                "QUEUE_REDIS_DEDUP_TTL must be > 0 when deduplication is enabled".to_string(),
             ));
         }
         match self.queue_adapter.as_str() {

src/queue/factory.rs

···
 
 use super::{
     adapter::QueueAdapter, mpsc::MpscQueueAdapter, noop::NoopQueueAdapter,
-    redis::RedisQueueAdapter, sqlite::SqliteQueueAdapter,
+    redis::RedisQueueAdapter, sqlite::SqliteQueueAdapter, work::DedupKey,
 };
 
 // ========= MPSC Queue Factories =========
···
 /// # Examples
 ///
 /// ```no_run
-/// use quickdid::queue::create_redis_queue;
+/// use quickdid::queue::{create_redis_queue, HandleResolutionWork};
 /// use deadpool_redis::Config;
 ///
 /// # async fn example() -> anyhow::Result<()> {
 /// let cfg = Config::from_url("redis://localhost:6379");
 /// let pool = cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1))?;
 ///
-/// let queue = create_redis_queue::<String>(
+/// let queue = create_redis_queue::<HandleResolutionWork>(
 ///     pool,
 ///     "worker-1".to_string(),
 ///     "queue:myapp:".to_string(),
···
     timeout_seconds: u64,
 ) -> Arc<dyn QueueAdapter<T>>
 where
-    T: Send + Sync + Serialize + for<'de> Deserialize<'de> + 'static,
+    T: Send + Sync + Serialize + for<'de> Deserialize<'de> + DedupKey + 'static,
 {
     Arc::new(RedisQueueAdapter::new(
         pool,
         worker_id,
         key_prefix,
         timeout_seconds,
+    ))
+}
+
+/// Create a new Redis-backed queue adapter with deduplication.
+///
+/// This creates a distributed queue with deduplication to prevent duplicate items
+/// from being queued within the specified TTL window.
+///
+/// # Arguments
+///
+/// * `pool` - Redis connection pool
+/// * `worker_id` - Worker identifier for this queue instance
+/// * `key_prefix` - Redis key prefix for queue operations
+/// * `timeout_seconds` - Timeout for blocking operations
+/// * `dedup_enabled` - Whether to enable deduplication
+/// * `dedup_ttl` - TTL for deduplication keys in seconds
+///
+/// # Examples
+///
+/// ```no_run
+/// use quickdid::queue::{create_redis_queue_with_dedup, HandleResolutionWork};
+/// use deadpool_redis::Config;
+///
+/// # async fn example() -> anyhow::Result<()> {
+/// let cfg = Config::from_url("redis://localhost:6379");
+/// let pool = cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1))?;
+///
+/// let queue = create_redis_queue_with_dedup::<HandleResolutionWork>(
+///     pool,
+///     "worker-1".to_string(),
+///     "queue:myapp:".to_string(),
+///     5,
+///     true,  // Enable deduplication
+///     60,    // 60 second dedup window
+/// );
+/// # Ok(())
+/// # }
+/// ```
+pub fn create_redis_queue_with_dedup<T>(
+    pool: RedisPool,
+    worker_id: String,
+    key_prefix: String,
+    timeout_seconds: u64,
+    dedup_enabled: bool,
+    dedup_ttl: u64,
+) -> Arc<dyn QueueAdapter<T>>
+where
+    T: Send + Sync + Serialize + for<'de> Deserialize<'de> + DedupKey + 'static,
+{
+    Arc::new(RedisQueueAdapter::with_dedup(
+        pool,
+        worker_id,
+        key_prefix,
+        timeout_seconds,
+        dedup_enabled,
+        dedup_ttl,
     ))
 }
 

src/queue/mod.rs

···
 // Re-export core types
 pub use adapter::QueueAdapter;
 pub use error::{QueueError, Result};
-pub use work::HandleResolutionWork;
+pub use work::{DedupKey, HandleResolutionWork};
 
 // Re-export implementations (with limited visibility)
 pub use mpsc::MpscQueueAdapter;
···
 // Re-export factory functions
 pub use factory::{
     create_mpsc_queue, create_mpsc_queue_from_channel, create_noop_queue, create_redis_queue,
-    create_sqlite_queue, create_sqlite_queue_with_max_size,
+    create_redis_queue_with_dedup, create_sqlite_queue, create_sqlite_queue_with_max_size,
 };

src/queue/redis.rs

···
 
 use super::adapter::QueueAdapter;
 use super::error::{QueueError, Result};
+use super::work::DedupKey;
 
 /// Redis-backed queue adapter implementation.
 ///
···
 /// # Examples
 ///
 /// ```no_run
-/// use quickdid::queue::RedisQueueAdapter;
-/// use quickdid::queue::QueueAdapter;
+/// use quickdid::queue::{RedisQueueAdapter, QueueAdapter, HandleResolutionWork};
 /// use deadpool_redis::Config;
 ///
 /// # async fn example() -> anyhow::Result<()> {
···
 /// let pool = cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1))?;
 ///
 /// // Create queue adapter
-/// let queue = RedisQueueAdapter::<String>::new(
+/// let queue = RedisQueueAdapter::<HandleResolutionWork>::new(
 ///     pool,
 ///     "worker-1".to_string(),
 ///     "queue:myapp:".to_string(),
···
 /// );
 ///
 /// // Use the queue
-/// queue.push("work-item".to_string()).await?;
+/// let work = HandleResolutionWork::new("alice.bsky.social".to_string());
+/// queue.push(work.clone()).await?;
 /// if let Some(item) = queue.pull().await {
 ///     // Process item
 ///     queue.ack(&item).await?;
···
     key_prefix: String,
     /// Timeout for blocking RPOPLPUSH operations (in seconds)
     timeout_seconds: u64,
+    /// Enable deduplication to prevent duplicate items in queue
+    dedup_enabled: bool,
+    /// TTL for deduplication keys in seconds
+    dedup_ttl: u64,
     /// Type marker for generic parameter
     _phantom: std::marker::PhantomData<T>,
 }
···
         key_prefix: String,
         timeout_seconds: u64,
     ) -> Self {
+        Self::with_dedup(
+            pool,
+            worker_id,
+            key_prefix,
+            timeout_seconds,
+            false,
+            60, // Default TTL of 60 seconds
+        )
+    }
+
+    /// Create a new Redis queue adapter with deduplication settings.
+    ///
+    /// # Arguments
+    ///
+    /// * `pool` - Redis connection pool
+    /// * `worker_id` - Unique identifier for this worker instance
+    /// * `key_prefix` - Redis key prefix for queue operations
+    /// * `timeout_seconds` - Timeout for blocking pull operations
+    /// * `dedup_enabled` - Whether to enable deduplication
+    /// * `dedup_ttl` - TTL for deduplication keys in seconds
+    pub fn with_dedup(
+        pool: RedisPool,
+        worker_id: String,
+        key_prefix: String,
+        timeout_seconds: u64,
+        dedup_enabled: bool,
+        dedup_ttl: u64,
+    ) -> Self {
         Self {
             pool,
             worker_id,
             key_prefix,
             timeout_seconds,
+            dedup_enabled,
+            dedup_ttl,
             _phantom: std::marker::PhantomData,
         }
     }
···
     fn worker_queue_key(&self) -> String {
         format!("{}{}", self.key_prefix, self.worker_id)
     }
+
+    /// Get the deduplication key for an item.
+    /// This key is used to track if an item is already queued.
+    fn dedup_key(&self, item_id: &str) -> String {
+        format!("{}dedup:{}", self.key_prefix, item_id)
+    }
+
+    /// Check and mark an item for deduplication.
+    /// Returns true if the item was successfully marked (not duplicate),
+    /// false if it was already in the deduplication set (duplicate).
+    async fn check_and_mark_dedup(
+        &self,
+        conn: &mut deadpool_redis::Connection,
+        item_id: &str,
+    ) -> Result<bool> {
+        if !self.dedup_enabled {
+            return Ok(true); // Always allow if dedup is disabled
+        }
+
+        let dedup_key = self.dedup_key(&item_id.to_string());
+        
+        // Use SET NX EX to atomically set if not exists with expiry
+        // Returns OK if the key was set, Nil if it already existed
+        let result: Option<String> = deadpool_redis::redis::cmd("SET")
+            .arg(&dedup_key)
+            .arg("1")
+            .arg("NX") // Only set if not exists
+            .arg("EX") // Set expiry
+            .arg(self.dedup_ttl)
+            .query_async(conn)
+            .await
+            .map_err(|e| QueueError::RedisOperationFailed {
+                operation: "SET NX EX".to_string(),
+                details: e.to_string(),
+            })?;
+
+        // If result is Some("OK"), the key was set (not duplicate)
+        // If result is None, the key already existed (duplicate)
+        Ok(result.is_some())
+    }
 }
 
 #[async_trait]
 impl<T> QueueAdapter<T> for RedisQueueAdapter<T>
 where
-    T: Send + Sync + Serialize + for<'de> Deserialize<'de> + 'static,
+    T: Send + Sync + Serialize + for<'de> Deserialize<'de> + DedupKey + 'static,
 {
     async fn pull(&self) -> Option<T> {
         match self.pool.get().await {
···
             .get()
             .await
             .map_err(|e| QueueError::RedisConnectionFailed(e.to_string()))?;
+
+        // Check for deduplication if enabled
+        if self.dedup_enabled {
+            let dedup_id = work.dedup_key();
+            let is_new = self.check_and_mark_dedup(&mut conn, &dedup_id).await?;
+            
+            if !is_new {
+                debug!(
+                    dedup_key = %dedup_id,
+                    "Item already queued, skipping duplicate"
+                );
+                return Ok(()); // Successfully deduplicated
+            }
+        }
 
         let data = serde_json::to_vec(&work)
             .map_err(|e| QueueError::SerializationFailed(e.to_string()))?;
···
 
         // Should be healthy if Redis is running
         assert!(adapter.is_healthy().await);
+    }
+
+    #[tokio::test]
+    async fn test_redis_queue_deduplication() {
+        use crate::queue::HandleResolutionWork;
+
+        let pool = match crate::test_helpers::get_test_redis_pool() {
+            Some(p) => p,
+            None => {
+                eprintln!("Skipping Redis test - no Redis connection available");
+                return;
+            }
+        };
+
+        let test_prefix = format!(
+            "test:queue:dedup:{}:",
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        );
+        
+        // Create adapter with deduplication enabled
+        let adapter = RedisQueueAdapter::<HandleResolutionWork>::with_dedup(
+            pool.clone(),
+            "test-worker-dedup".to_string(),
+            test_prefix.clone(),
+            1,
+            true,  // Enable deduplication
+            2,     // 2 second TTL for quick testing
+        );
+
+        let work = HandleResolutionWork::new("alice.example.com".to_string());
+
+        // First push should succeed
+        adapter.push(work.clone()).await.expect("First push should succeed");
+        
+        // Second push of same item should be deduplicated (but still return Ok)
+        adapter.push(work.clone()).await.expect("Second push should succeed (deduplicated)");
+        
+        // Queue should only have one item
+        let depth = adapter.depth().await;
+        assert_eq!(depth, Some(1), "Queue should only have one item after deduplication");
+        
+        // Pull the item
+        let pulled = adapter.pull().await;
+        assert_eq!(pulled, Some(work.clone()));
+        
+        // Queue should now be empty
+        let depth = adapter.depth().await;
+        assert_eq!(depth, Some(0), "Queue should be empty after pulling");
+        
+        // Wait for dedup TTL to expire
+        tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+        
+        // Should be able to push again after TTL expires
+        adapter.push(work.clone()).await.expect("Push after TTL expiry should succeed");
+        
+        let depth = adapter.depth().await;
+        assert_eq!(depth, Some(1), "Queue should have one item after TTL expiry");
+    }
+
+    #[tokio::test]
+    async fn test_redis_queue_deduplication_disabled() {
+        use crate::queue::HandleResolutionWork;
+
+        let pool = match crate::test_helpers::get_test_redis_pool() {
+            Some(p) => p,
+            None => {
+                eprintln!("Skipping Redis test - no Redis connection available");
+                return;
+            }
+        };
+
+        let test_prefix = format!(
+            "test:queue:nodedup:{}:",
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        );
+        
+        // Create adapter with deduplication disabled
+        let adapter = RedisQueueAdapter::<HandleResolutionWork>::with_dedup(
+            pool.clone(),
+            "test-worker-nodedup".to_string(),
+            test_prefix.clone(),
+            1,
+            false,  // Disable deduplication
+            60,
+        );
+
+        let work = HandleResolutionWork::new("bob.example.com".to_string());
+
+        // Push same item twice
+        adapter.push(work.clone()).await.expect("First push should succeed");
+        adapter.push(work.clone()).await.expect("Second push should succeed");
+        
+        // Queue should have two items (no deduplication)
+        let depth = adapter.depth().await;
+        assert_eq!(depth, Some(2), "Queue should have two items when deduplication is disabled");
+        
+        // Pull both items
+        let pulled1 = adapter.pull().await;
+        assert_eq!(pulled1, Some(work.clone()));
+        
+        let pulled2 = adapter.pull().await;
+        assert_eq!(pulled2, Some(work.clone()));
+        
+        // Queue should now be empty
+        let depth = adapter.depth().await;
+        assert_eq!(depth, Some(0), "Queue should be empty after pulling all items");
     }
 
     #[tokio::test]

src/queue/work.rs

···
     }
 }
 
+/// Trait for getting a unique deduplication key from a work item.
+/// This is used by the Redis queue adapter to prevent duplicate items.
+pub trait DedupKey {
+    /// Get a unique key for deduplication purposes.
+    /// This should return a consistent identifier for equivalent work items.
+    fn dedup_key(&self) -> String;
+}
+
+impl DedupKey for HandleResolutionWork {
+    fn dedup_key(&self) -> String {
+        // Use the handle itself as the dedup key
+        self.handle.clone()
+    }
+}
+
+// For testing purposes, implement DedupKey for String
+#[cfg(test)]
+impl DedupKey for String {
+    fn dedup_key(&self) -> String {
+        self.clone()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
···
 
         assert_eq!(work1, work2);
         assert_ne!(work1, work3);
+    }
+
+    #[test]
+    fn test_handle_resolution_work_dedup_key() {
+        let work1 = HandleResolutionWork::new("alice.example.com".to_string());
+        let work2 = HandleResolutionWork::new("alice.example.com".to_string());
+        let work3 = HandleResolutionWork::new("bob.example.com".to_string());
+
+        // Same handle should have same dedup key
+        assert_eq!(work1.dedup_key(), work2.dedup_key());
+        assert_eq!(work1.dedup_key(), "alice.example.com");
+        
+        // Different handle should have different dedup key
+        assert_ne!(work1.dedup_key(), work3.dedup_key());
+        assert_eq!(work3.dedup_key(), "bob.example.com");
     }
 }