Deduplication Strategies

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

/**
 * Multi-layer deduplication architecture
 * 
 * Each layer catches different types of duplicates:
 * - Producer: Network retry duplicates
 * - Broker: Producer restart duplicates  
 * - Consumer: Broker redelivery duplicates
 */
 
// Layer 1: Idempotent Producer
class IdempotentProducer {
    private sequenceNumber: number = 0;
    private pendingMessages: Map<number, Message> = new Map();
 
    async send(message: Message): Promise<void> {
        const seqNum = ++this.sequenceNumber;
        
        // Broker tracks (producerId, sequenceNumber) tuples
        // If we retry and send same sequence, broker ignores
        await this.broker.send({
            ...message,
            producerId: this.producerId,
            sequenceNumber: seqNum,
        });
    }
}
 
// Layer 2: Broker Deduplication
class BrokerDeduplication {
    private recentMessageIds: Map<string, number> = new Map(); // messageId → timestamp
    private windowMs: number = 5 * 60 * 1000; // 5-minute window
 
    acceptMessage(message: Message): boolean {
        const now = Date.now();
        
        // Check if we've seen this message ID recently
        const lastSeen = this.recentMessageIds.get(message.id);
        if (lastSeen && (now - lastSeen) < this.windowMs) {
            console.log(`Broker: Duplicate message ${message.id} rejected`);
            return false; // Duplicate within window
        }
 
        // Accept and record
        this.recentMessageIds.set(message.id, now);
        this.cleanupOldEntries(now);
        return true;
    }
 
    private cleanupOldEntries(now: number): void {
        for (const [id, timestamp] of this.recentMessageIds) {
            if (now - timestamp > this.windowMs) {
                this.recentMessageIds.delete(id);
            }
        }
    }
}
 
// Layer 3: Consumer Deduplication
class DedupConsumer {
    async processMessage(message: Message): Promise<void> {
        // Final layer: Database-backed deduplication
        const processed = await this.db.processedMessages.findUnique({
            where: { messageId: message.id },
        });
 
        if (processed) {
            console.log(`Consumer: Duplicate ${message.id}, skipping`);
            await message.ack();
            return;
        }
 
        // Process and record atomically
        await this.db.$transaction([
            this.businessLogic(message),
            this.db.processedMessages.create({
                data: { messageId: message.id },
            }),
        ]);
 
        await message.ack();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

import { v4 as uuidv4 } from 'uuid';
 
/**
 * Identifier-based deduplication service
 */
class IdentifierDeduplication {
    private storage: DeduplicationStorage;
    private ttl: number;
 
    constructor(storage: DeduplicationStorage, ttlSeconds: number = 86400) {
        this.storage = storage;
        this.ttl = ttlSeconds;
    }
 
    /**
     * Check if a message is a duplicate and record it if not.
     * 
     * @returns true if duplicate (should skip), false if new (should process)
     */
    async isDuplicate(messageId: string): Promise<boolean> {
        // Atomic check-and-set operation
        const wasSet = await this.storage.setIfNotExists(
            `dedup:${messageId}`,
            Date.now().toString(),
            this.ttl
        );
 
        // setIfNotExists returns true if we set it (new message)
        // Returns false if it already existed (duplicate)
        return !wasSet;
    }
 
    /**
     * Generate a dedup-friendly message ID
     */
    static generateMessageId(producerId: string): string {
        // Format: {producerId}-{timestamp}-{uuid}
        // Includes producer for debugging, timestamp for ordering,
        // UUID for guaranteed uniqueness
        return `${producerId}-${Date.now()}-${uuidv4()}`;
    }
}
 
// Storage implementations
class RedisDeduplicationStorage implements DeduplicationStorage {
    private redis: Redis;
 
    async setIfNotExists(key: string, value: string, ttlSeconds: number): Promise<boolean> {
        // Redis SETNX with expiry - atomic check-and-set
        const result = await this.redis.set(key, value, 'EX', ttlSeconds, 'NX');
        return result === 'OK';
    }
}
 
class PostgresDeduplicationStorage implements DeduplicationStorage {
    private db: Database;
 
    async setIfNotExists(key: string, value: string, ttlSeconds: number): Promise<boolean> {
        try {
            await this.db.deduplicationRecords.create({
                data: {
                    messageId: key,
                    processedAt: new Date(),
                    expiresAt: new Date(Date.now() + ttlSeconds * 1000),
                },
            });
            return true; // Successfully inserted (new message)
        } catch (error: any) {
            if (error.code === 'P2002') {
                return false; // Unique constraint violation (duplicate)
            }
            throw error;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({
    clientId: 'order-service',
    brokers: ['kafka:9092'],
});
 
/**
 * Kafka's idempotent producer provides automatic deduplication
 * at the broker level.
 * 
 * How it works:
 * 1. Producer gets a unique ProducerId (PID) from the broker
 * 2. Each message includes (PID, sequenceNumber)
 * 3. Broker tracks the last sequenceNumber per PID per partition
 * 4. Duplicate (PID, seq) pairs are silently ignored
 */
const producer = kafka.producer({
    idempotent: true,  // Enable broker-level deduplication
    
    // With idempotency enabled, these settings are required:
    maxInFlightRequests: 5,  // Max 5 (Kafka limitation)
    acks: -1,  // Wait for all replicas (required for safety)
});
 
await producer.connect();
 
// Even if this is retried multiple times due to network issues,
// the broker ensures the message appears exactly once in the log
await producer.send({
    topic: 'orders',
    messages: [{
        key: orderId,
        value: JSON.stringify(order),
    }],
});
 
/**
 * Kafka idempotent producer dedup scope:
 * 
 * ✓ Catches: Network retry duplicates (same producer session)
 * ✗ Misses: Producer restart duplicates (new PID, new sequence)
 * ✗ Misses: Consumer-side duplicates (redelivery after crash)
 * 
 * For complete coverage, also need consumer-side deduplication
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

import { createHash } from 'crypto';
 
class ContentBasedDeduplication {
    private storage: DeduplicationStorage;
    private ttl: number;
 
    /**
     * Generate a content hash for a message.
     * 
     * Includes only the fields that define message identity,
     * excluding metadata like timestamps or message IDs.
     */
    generateContentHash(message: any): string {
        // Extract identity-defining fields only
        const contentToHash = {
            type: message.type,
            entityId: message.entityId,
            operation: message.operation,
            payload: message.payload,
            // Explicitly exclude: message.id, message.timestamp, message.metadata
        };
 
        const serialized = this.canonicalSerialize(contentToHash);
        return createHash('sha256').update(serialized).digest('hex');
    }
 
    /**
     * Canonical serialization ensures identical objects produce identical strings
     * regardless of property order.
     */
    private canonicalSerialize(obj: any): string {
        if (obj === null || obj === undefined) {
            return 'null';
        }
 
        if (typeof obj !== 'object') {
            return JSON.stringify(obj);
        }
 
        if (Array.isArray(obj)) {
            return '[' + obj.map(item => this.canonicalSerialize(item)).join(',') + ']';
        }
 
        // Sort keys for consistent ordering
        const sortedKeys = Object.keys(obj).sort();
        const pairs = sortedKeys.map(key => 
            `"${key}":${this.canonicalSerialize(obj[key])}`
        );
        return '{' + pairs.join(',') + '}';
    }
 
    async isDuplicate(message: any): Promise<boolean> {
        const contentHash = this.generateContentHash(message);
        return await this.storage.isDuplicate(`content:${contentHash}`);
    }
}
 
// Usage
const dedup = new ContentBasedDeduplication(storage, 3600);
 
// These two messages have different IDs but identical content
// Content-based dedup will catch this as a duplicate
const message1 = {
    id: 'msg-001',
    timestamp: 1704700000000,
    type: 'ORDER_CREATED',
    entityId: 'order-123',
    payload: { items: ['A', 'B'], total: 99.99 }
};
 
const message2 = {
    id: 'msg-002',  // Different ID!
    timestamp: 1704700001000,  // Different timestamp!
    type: 'ORDER_CREATED',
    entityId: 'order-123',
    payload: { items: ['A', 'B'], total: 99.99 }
};
 
await dedup.isDuplicate(message1);  // false (new)
await dedup.isDuplicate(message2);  // true (duplicate by content)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

/**
 * Semantic deduplication considers business meaning,
 * not just byte-level content equality.
 */
 
class SemanticDeduplication {
    /**
     * Extract the semantic identity of a message.
     * Two messages are duplicates if they have the same semantic identity.
     */
    getSemanticId(message: any): string {
        switch (message.type) {
            case 'ORDER_CREATED':
                // For orders, the orderId defines identity
                // Multiple ORDER_CREATED for same orderId = duplicate
                return `order-created:${message.payload.orderId}`;
 
            case 'PAYMENT_COMPLETED':
                // For payments, orderId + amount defines identity
                // (same order could have multiple partial payments)
                return `payment:${message.payload.orderId}:${message.payload.amount}`;
 
            case 'USER_PROFILE_UPDATED':
                // For profile updates, user + version defines identity
                // Only care about latest version
                return `profile:${message.payload.userId}:${message.payload.version}`;
 
            case 'INVENTORY_CHANGED':
                // For inventory, sku + final quantity defines identity
                // Multiple messages with same final state = duplicate
                return `inventory:${message.payload.sku}:${message.payload.finalQuantity}`;
 
            default:
                // Fallback to content hash
                return `unknown:${this.hashContent(message)}`;
        }
    }
 
    async isDuplicate(message: any): Promise<boolean> {
        const semanticId = this.getSemanticId(message);
        return await this.storage.isDuplicate(semanticId);
    }
}
 
/**
 * Example: These messages are semantically equivalent
 * (same order creation, different message structure)
 */
const msg1 = {
    type: 'ORDER_CREATED',
    payload: { orderId: '12345', items: ['A', 'B'], createdBy: 'user-1' }
};
 
const msg2 = {
    type: 'ORDER_CREATED',
    payload: { orderId: '12345', items: ['A', 'B'], createdBy: 'user-1', notes: 'Added later' }
};
 
// Both produce semanticId "order-created:12345"
// Second message is treated as duplicate despite different content

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

class TimeWindowedDeduplication {
    /**
     * Redis-based time-windowed deduplication.
     * Messages are tracked for a fixed time window,
     * then automatically expire.
     */
    private redis: Redis;
    private windowSeconds: number;
    private keyPrefix: string;
 
    constructor(redis: Redis, windowMinutes: number, keyPrefix: string = 'dedup') {
        this.redis = redis;
        this.windowSeconds = windowMinutes * 60;
        this.keyPrefix = keyPrefix;
    }
 
    async isDuplicate(messageId: string): Promise<boolean> {
        const key = `${this.keyPrefix}:${messageId}`;
        
        // SETNX with expiry - atomic check and set
        // Returns null if key already exists (duplicate)
        // Returns 'OK' if key was set (new message)
        const result = await this.redis.set(
            key,
            '1',
            'EX', this.windowSeconds,
            'NX'  // Only set if not exists
        );
 
        // If result is null, the key existed → duplicate
        return result === null;
    }
 
    /**
     * Check multiple message IDs at once (for batch processing)
     */
    async checkBatch(messageIds: string[]): Promise<Map<string, boolean>> {
        const pipeline = this.redis.pipeline();
        
        for (const id of messageIds) {
            pipeline.set(
                `${this.keyPrefix}:${id}`,
                '1',
                'EX', this.windowSeconds,
                'NX'
            );
        }
 
        const results = await pipeline.exec();
        const duplicates = new Map<string, boolean>();
 
        results?.forEach((result, index) => {
            // result[1] is the value returned by SET NX
            // null means key existed (duplicate)
            duplicates.set(messageIds[index], result[1] === null);
        });
 
        return duplicates;
    }
}
 
// Usage with different windows for different message types
const realTimeDedup = new TimeWindowedDeduplication(redis, 15, 'rt'); // 15 min
const batchDedup = new TimeWindowedDeduplication(redis, 1440, 'batch'); // 24 hours
const auditDedup = new TimeWindowedDeduplication(redis, 10080, 'audit'); // 7 days

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

/**
 * Sliding window deduplication with first-seen timestamp tracking.
 * This allows us to know when a message was first seen,
 * useful for debugging and analytics.
 */
 
class SlidingWindowDeduplication {
    private redis: Redis;
    private windowMs: number;
 
    async checkAndRecord(messageId: string): Promise<{
        isDuplicate: boolean;
        firstSeenAt?: Date;
    }> {
        const key = `dedup:${messageId}`;
        const now = Date.now();
 
        // Use HSETNX to store first-seen timestamp atomically
        const wasSet = await this.redis.hsetnx(key, 'firstSeen', now.toString());
 
        if (wasSet) {
            // New message - set expiry
            await this.redis.pexpire(key, this.windowMs);
            return { isDuplicate: false };
        }
 
        // Duplicate - retrieve first-seen time
        const firstSeen = await this.redis.hget(key, 'firstSeen');
        return {
            isDuplicate: true,
            firstSeenAt: firstSeen ? new Date(parseInt(firstSeen)) : undefined,
        };
    }
}
 
// Usage
const result = await dedup.checkAndRecord('msg-12345');
if (result.isDuplicate) {
    console.log(`Duplicate! First seen: ${result.firstSeenAt}`);
    // Useful for monitoring duplicate rates and debugging
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

/**
 * Partitioned Deduplication
 * 
 * Distribute dedup checks across multiple shards based on message ID.
 * Each shard handles a subset of the ID space.
 */
 
class PartitionedDeduplication {
    private shards: Redis[];
    private numShards: number;
 
    constructor(shardConnections: Redis[]) {
        this.shards = shardConnections;
        this.numShards = shardConnections.length;
    }
 
    /**
     * Determine which shard owns a given message ID.
     * Uses consistent hashing to distribute evenly.
     */
    private getShardIndex(messageId: string): number {
        // Simple hash-based partitioning
        // For production, use consistent hashing (e.g., jump hash)
        let hash = 0;
        for (let i = 0; i < messageId.length; i++) {
            hash = ((hash << 5) - hash) + messageId.charCodeAt(i);
            hash = hash & hash; // Convert to 32-bit integer
        }
        return Math.abs(hash) % this.numShards;
    }
 
    async isDuplicate(messageId: string): Promise<boolean> {
        const shardIndex = this.getShardIndex(messageId);
        const shard = this.shards[shardIndex];
 
        const result = await shard.set(
            `dedup:${messageId}`,
            '1',
            'EX', 3600,
            'NX'
        );
 
        return result === null;
    }
 
    /**
     * Batch check across shards.
     * Groups IDs by shard, executes in parallel.
     */
    async checkBatch(messageIds: string[]): Promise<Map<string, boolean>> {
        // Group IDs by shard
        const shardGroups: Map<number, string[]> = new Map();
        for (const id of messageIds) {
            const shardIndex = this.getShardIndex(id);
            if (!shardGroups.has(shardIndex)) {
                shardGroups.set(shardIndex, []);
            }
            shardGroups.get(shardIndex)!.push(id);
        }
 
        // Execute checks in parallel across shards
        const results = new Map<string, boolean>();
        const promises: Promise<void>[] = [];
 
        for (const [shardIndex, ids] of shardGroups) {
            const promise = this.checkShardBatch(shardIndex, ids, results);
            promises.push(promise);
        }
 
        await Promise.all(promises);
        return results;
    }
 
    private async checkShardBatch(
        shardIndex: number,
        ids: string[],
        results: Map<string, boolean>
    ): Promise<void> {
        const shard = this.shards[shardIndex];
        const pipeline = shard.pipeline();
 
        for (const id of ids) {
            pipeline.set(`dedup:${id}`, '1', 'EX', 3600, 'NX');
        }
 
        const pipelineResults = await pipeline.exec();
        pipelineResults?.forEach((result, index) => {
            results.set(ids[index], result[1] === null);
        });
    }
}
 
// Deploy: 16 Redis shards across 4 servers
// Each shard handles 1/16th of the ID space
// Scales to ~160K dedup checks/second (10K per shard)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

/**
 * Two-tier deduplication: Local cache + distributed store.
 * 
 * Local cache catches recent duplicates instantly.
 * Distributed store provides durability and cross-instance dedup.
 */
 
import { LRUCache } from 'lru-cache';
 
class HybridDeduplication {
    private localCache: LRUCache<string, boolean>;
    private distributedStore: PartitionedDeduplication;
 
    constructor(distributedStore: PartitionedDeduplication) {
        this.localCache = new LRUCache({
            max: 100_000,            // 100K entries
            ttl: 1000 * 60 * 5,      // 5 minute local TTL
        });
        this.distributedStore = distributedStore;
    }
 
    async isDuplicate(messageId: string): Promise<boolean> {
        // Layer 1: Check local cache (fastest)
        if (this.localCache.has(messageId)) {
            return true; // Definitely duplicate (cached locally)
        }
 
        // Layer 2: Check distributed store
        const isDup = await this.distributedStore.isDuplicate(messageId);
 
        if (!isDup) {
            // New message - add to local cache
            this.localCache.set(messageId, true);
        }
 
        return isDup;
    }
}
 
// Performance characteristics:
// - Local cache hit: ~0.001ms (microseconds)
// - Redis hit: ~0.5-2ms (network round-trip)
// - With 80% local cache hit rate, average latency drops from 1ms to 0.2ms
// - 5x improvement in dedup check latency

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

/**
 * Cross-region deduplication strategies.
 * 
 * Challenge: We want duplicates caught even when processing
 * moves between regions, but cross-region latency is expensive.
 */
 
// Strategy 1: Synchronous Cross-Region Check (Latency-heavy)
class SyncCrossRegionDedup {
    async isDuplicate(messageId: string): Promise<boolean> {
        // Check all regions - high latency (50-200ms per region)
        const results = await Promise.all([
            this.usEastStore.isDuplicate(messageId),
            this.euWestStore.isDuplicate(messageId),
            this.apSouthStore.isDuplicate(messageId),
        ]);
 
        return results.some(isDup => isDup);
    }
}
 
// Strategy 2: Async Replication (Eventual consistency)
class AsyncReplicatedDedup {
    private localStore: DeduplicationStore;
    private replicationQueue: MessageQueue;
 
    async isDuplicate(messageId: string): Promise<boolean> {
        // Check only local store (fast)
        const isDup = await this.localStore.isDuplicate(messageId);
 
        if (!isDup) {
            // New message - replicate to other regions asynchronously
            await this.replicationQueue.send({
                type: 'DEDUP_RECORD',
                messageId,
                region: process.env.REGION,
                timestamp: Date.now(),
            });
        }
 
        return isDup;
    }
}
 
// Strategy 3: Region-Affine Message IDs (Recommended)
class RegionAffineDedup {
    /**
     * Include region in message ID or routing key.
     * All duplicates of a message go to the same region.
     * No cross-region dedup needed.
     */
    private localStore: DeduplicationStore;
    private currentRegion: string;
 
    async isDuplicate(messageId: string): Promise<boolean> {
        // Extract region from message routing
        const messageRegion = this.extractRegion(messageId);
 
        if (messageRegion !== this.currentRegion) {
            // Message belongs to different region - not our responsibility
            // This shouldn't happen if routing is correct
            throw new Error(`Message ${messageId} routed to wrong region`);
        }
 
        // Check only local store
        return await this.localStore.isDuplicate(messageId);
    }
 
    extractRegion(messageId: string): string {
        // Example: "us-east:msg-12345" → "us-east"
        return messageId.split(':')[0];
    }
}
 
/**
 * Recommendation:
 * 
 * Prefer region-affine routing where possible.
 * - Route messages to fixed regions based on partition key
 * - No cross-region dedup needed
 * - No cross-region latency
 * - Simplest to operate
 * 
 * Use async replication only when region affinity is impossible.
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

import { BloomFilter } from 'bloom-filters';
 
class BloomFilterDeduplication {
    private filter: BloomFilter;
    private backupStore: DeduplicationStore;
    private falsePositiveRate: number;
 
    constructor(
        expectedItems: number,
        falsePositiveRate: number = 0.001 // 0.1%
    ) {
        // Calculate optimal size for target false positive rate
        // 1 billion items at 0.1% FP rate ≈ 1.4 GB memory
        this.filter = BloomFilter.create(expectedItems, falsePositiveRate);
        this.falsePositiveRate = falsePositiveRate;
    }
 
    async isDuplicate(messageId: string): Promise<boolean> {
        // Fast check: Bloom filter
        if (!this.filter.has(messageId)) {
            // DEFINITELY new - bloom filters have no false negatives
            this.filter.add(messageId);
            return false;
        }
 
        // MAYBE duplicate - could be false positive
        // Fall back to authoritative store
        const confirmed = await this.backupStore.isDuplicate(messageId);
 
        if (!confirmed) {
            // False positive! Actually new message.
            // Already in bloom filter, record in backup store
            await this.backupStore.record(messageId);
        }
 
        return confirmed;
    }
 
    /**
     * Memory usage calculation:
     * 
     * For n items with false positive rate p:
     * bits = -n * ln(p) / (ln(2))^2
     * 
     * Examples:
     * - 1M items, 0.1% FP: ~1.4 MB
     * - 100M items, 0.1% FP: ~143 MB
     * - 1B items, 0.1% FP: ~1.4 GB
     * 
     * Compare to storing 1B 36-char UUIDs: ~36 GB
     */
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

import { CuckooFilter } from 'bloom-filters';
 
class TimeWindowedCuckooDedup {
    private currentWindow: CuckooFilter;
    private previousWindow: CuckooFilter;
    private windowDurationMs: number;
    private lastRotation: number;
 
    constructor(
        expectedItems: number,
        windowMinutes: number = 60
    ) {
        // Two windows for time-windowed dedup with rotation
        this.currentWindow = CuckooFilter.create(expectedItems);
        this.previousWindow = CuckooFilter.create(expectedItems);
        this.windowDurationMs = windowMinutes * 60 * 1000;
        this.lastRotation = Date.now();
    }
 
    async isDuplicate(messageId: string): Promise<boolean> {
        await this.maybeRotate();
 
        // Check both windows
        if (this.currentWindow.has(messageId) || 
            this.previousWindow.has(messageId)) {
            return true;
        }
 
        // New message - add to current window
        this.currentWindow.add(messageId);
        return false;
    }
 
    private async maybeRotate(): Promise<void> {
        const now = Date.now();
        if (now - this.lastRotation > this.windowDurationMs) {
            // Rotate windows
            this.previousWindow = this.currentWindow;
            this.currentWindow = CuckooFilter.create(this.currentWindow.size);
            this.lastRotation = now;
        }
    }
}
 
/**
 * Cuckoo filter advantages:
 * - Supports deletion (can remove expired entries)
 * - Better space efficiency than Bloom for false positive rates < 3%
 * - Faster lookups than Bloom filter
 * 
 * Trade-offs:
 * - More complex implementation
 * - Insertion can fail if filter is too full
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

class MonitoredDeduplication {
    private metrics: MetricsClient;
    private underlying: DeduplicationStore;
 
    async isDuplicate(messageId: string): Promise<boolean> {
        const startTime = Date.now();
        
        try {
            const result = await this.underlying.isDuplicate(messageId);
            
            // Record latency
            const latency = Date.now() - startTime;
            this.metrics.histogram('dedup.check_latency_ms', latency);
            
            // Record duplicate rate
            this.metrics.increment('dedup.checks_total');
            if (result) {
                this.metrics.increment('dedup.duplicates_found');
            }
            
            return result;
        } catch (error) {
            this.metrics.increment('dedup.check_errors');
            throw error;
        }
    }
 
    /**
     * Periodic health check and reporting
     */
    async reportHealth(): Promise<void> {
        // Report store size
        const storeSize = await this.underlying.getSize();
        this.metrics.gauge('dedup.store_size', storeSize);
 
        // Report memory usage
        const memoryUsage = await this.underlying.getMemoryUsage();
        this.metrics.gauge('dedup.memory_bytes', memoryUsage);
 
        // Check for anomalies
        const duplicateRate = await this.calculateDuplicateRate();
        if (duplicateRate > 0.05) { // >5% duplicate rate
            console.warn(`High duplicate rate: ${(duplicateRate * 100).toFixed(2)}%`);
            this.metrics.increment('dedup.high_duplicate_rate_alert');
        }
 
        if (duplicateRate < 0.0001) { // <0.01% duplicate rate
            console.warn(`Suspiciously low duplicate rate: ${(duplicateRate * 100).toFixed(4)}%`);
            // This might indicate dedup is not working correctly
            this.metrics.increment('dedup.low_duplicate_rate_alert');
        }
    }
 
    private async calculateDuplicateRate(): Promise<number> {
        const total = await this.metrics.getCounter('dedup.checks_total');
        const duplicates = await this.metrics.getCounter('dedup.duplicates_found');
        return total > 0 ? duplicates / total : 0;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

class DeduplicationMaintenance {
    private db: Database;
    private redis: Redis;
    private metrics: MetricsClient;
 
    /**
     * Cleanup expired deduplication records.
     * Should run periodically (e.g., hourly).
     */
    async cleanupExpiredRecords(): Promise<void> {
        console.log('Starting dedup cleanup...');
        const startTime = Date.now();
 
        // Database cleanup
        const dbDeleted = await this.db.dedupRecords.deleteMany({
            where: {
                expiresAt: { lt: new Date() },
            },
        });
 
        console.log(`Deleted ${dbDeleted.count} expired DB records`);
        this.metrics.gauge('dedup.cleanup.db_deleted', dbDeleted.count);
 
        // Redis cleanup happens automatically via TTL
        // But we can force cleanup of keys without TTL
        const redisCleanup = await this.cleanupRedisOrphans();
 
        const duration = Date.now() - startTime;
        console.log(`Cleanup completed in ${duration}ms`);
        this.metrics.histogram('dedup.cleanup.duration_ms', duration);
    }
 
    /**
     * Verify deduplication is working correctly.
     * Run periodically as a health check.
     */
    async verifyDeduplication(): Promise<boolean> {
        const testId = `test-${Date.now()}-${Math.random()}`;
 
        // First check should return false (new)
        const firstCheck = await this.dedupStore.isDuplicate(testId);
        if (firstCheck) {
            console.error('Dedup verification FAILED: New ID flagged as duplicate');
            this.metrics.increment('dedup.verification.failed');
            return false;
        }
 
        // Second check should return true (duplicate)
        const secondCheck = await this.dedupStore.isDuplicate(testId);
        if (!secondCheck) {
            console.error('Dedup verification FAILED: Duplicate not detected');
            this.metrics.increment('dedup.verification.failed');
            return false;
        }
 
        console.log('Dedup verification PASSED');
        this.metrics.increment('dedup.verification.passed');
        return true;
    }
 
    /**
     * Handle dedup store failure gracefully.
     * When dedup fails, we must decide: skip or process?
     */
    async handleDedupFailure(
        messageId: string,
        error: Error,
        message: Message
    ): Promise<'process' | 'skip'> {
        console.error(`Dedup check failed for ${messageId}: ${error.message}`);
        this.metrics.increment('dedup.failures');
 
        // Decision depends on business requirements:
 
        // Option 1: Fail-open (process anyway, risk duplicates)
        // Use when: Duplicates are tolerable, message loss is not
        // return 'process';
 
        // Option 2: Fail-closed (skip, rely on redelivery)
        // Use when: Duplicates are dangerous
        // return 'skip';
 
        // Option 3: Circuit breaker (fail fast if too many failures)
        if (await this.isDedupCircuitOpen()) {
            throw new Error('Dedup circuit breaker open');
        }
 
        // Default: process, log for investigation
        return 'process';
    }
}