Module deduplication 

Event Deduplication System

This module provides enterprise-grade event deduplication capabilities with Redis backend support, configurable event ID extraction, TTL-based cleanup, and comprehensive metrics.

Features

• Multiple ID Extraction Strategies: UUID, hash-based, custom extractors
• Redis-Backed Storage: Production-ready distributed deduplication
• TTL-Based Cleanup: Automatic memory management with configurable expiration
• Batch Operations: Efficient bulk deduplication checks
• Retry Logic: Exponential backoff for transient failures
• Comprehensive Metrics: Track duplicates, cache hits, and performance
• Thread-Safe: Lock-free concurrent operations with Arc/DashMap
• Zero Data Loss: Guaranteed at-most-once processing semantics

Architecture

The deduplication system uses a layered architecture:

┌─────────────────────────────────┐
│   EventDeduplicator (API)       │
└─────────────┬───────────────────┘
              │
┌─────────────┴───────────────────┐
│   EventIdExtractor (Strategy)   │
└─────────────┬───────────────────┘
              │
┌─────────────┴───────────────────┐
│ RedisDeduplicationBackend       │
│  - Connection pooling           │
│  - Retry logic                  │
│  - Metrics tracking             │
└─────────────────────────────────┘

Usage Example

use processor::deduplication::{
    EventDeduplicator, DeduplicationConfig, EventIdExtractor,
};
use serde::{Deserialize, Serialize};
use std::time::Duration;

#[derive(Debug, Clone, Serialize, Deserialize)]
struct MyEvent {
    id: String,
    data: String,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Configure deduplicator
    let config = DeduplicationConfig::builder()
        .redis_url("redis://localhost:6379")
        .key_prefix("dedup:")
        .default_ttl(Duration::from_secs(3600))
        .build()?;

    // Create custom ID extractor
    let extractor = EventIdExtractor::custom(|event: &MyEvent| {
        event.id.clone()
    });

    // Initialize deduplicator
    let dedup = EventDeduplicator::new(config, extractor).await?;

    // Check and process events
    let event = MyEvent {
        id: "event-123".to_string(),
        data: "payload".to_string(),
    };

    if !dedup.is_duplicate(&event).await? {
        // Process event
        println!("Processing new event: {}", event.id);

        // Mark as seen
        dedup.mark_seen(&event).await?;
    } else {
        println!("Duplicate event detected: {}", event.id);
    }

    // Get statistics
    let stats = dedup.stats().await;
    println!("Duplicate rate: {:.2}%", stats.duplicate_rate() * 100.0);

    Ok(())
}

Batch Processing

For high-throughput scenarios, use batch operations:

let events = vec![
    MyEvent { id: "event-1".to_string() },
    MyEvent { id: "event-2".to_string() },
    MyEvent { id: "event-3".to_string() },
];

// Check all events in one batch
let results = dedup.batch_check_duplicates(&events).await?;

for (event, is_duplicate) in events.iter().zip(results.iter()) {
    if !is_duplicate {
        println!("Processing: {}", event.id);
    }
}

// Mark all non-duplicates as seen
let new_events: Vec<_> = events.iter()
    .zip(results.iter())
    .filter_map(|(e, dup)| if !dup { Some(e) } else { None })
    .collect();

dedup.batch_mark_seen(&new_events).await?;

Structs

DeduplicationConfig

Configuration for event deduplication

DeduplicationConfigBuilder

Builder for DeduplicationConfig

DeduplicationStats

Statistics for deduplication operations

EventDeduplicator

High-level event deduplicator

Enums

EventIdExtractor

Strategy for extracting unique event IDs