offline_protocol_reliability/

deduplicator.rs

//! Message deduplication using rotating bloom filters for O(1) memory.
//!
//! This module implements a scalable deduplication system using bloom filters
//! that can handle millions of messages with constant memory usage.
//!
//! The rotating bloom filter approach uses multiple time-windowed filters:
//! - Current filter: accepts new message IDs
//! - Previous filters: used for lookups, rotated out after retention period

use chrono::{DateTime, Utc};
use offline_protocol_core::MessageId;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Configuration for deduplication.
#[derive(Debug, Clone)]
pub struct DeduplicatorConfig {
    /// Maximum number of message IDs to track (used for HashMap fallback).
    pub max_tracked_messages: usize,

    /// Time to keep message IDs in memory (seconds).
    pub retention_time_secs: u64,

    /// Enable bloom filter mode for scalable deduplication.
    pub use_bloom_filter: bool,

    /// Number of bits in each bloom filter (default: 2^20 = ~1MB per filter).
    pub bloom_filter_bits: usize,

    /// Number of hash functions for bloom filter (default: 7 for ~1% false positive).
    pub bloom_hash_count: usize,

    /// Number of rotating bloom filters (default: 4 for hourly windows).
    pub bloom_filter_count: usize,

    /// Rotation interval in seconds (default: 900 = 15 minutes).
    pub bloom_rotation_secs: u64,
}

impl Default for DeduplicatorConfig {
    fn default() -> Self {
        Self {
            // Exact-match mode by default: no false positives (Bloom can drop ~1% of legitimate messages).
            //
            // Sized for the seen set to be persisted across restarts
            // (`export_seen`/`import_seen`): a message can reach a receiver
            // twice on two paths — a push-injected copy and the relay socket
            // copy after a reconnect, hours apart if the app was closed in
            // between — and the second copy is only recognisable as a
            // duplicate while its id is still tracked. An hour did not cover
            // "reopened the next day", and a day did not cover a sender that
            // keeps re-sending an unacknowledged id for up to
            // `REDELIVERY_WINDOW_MS`: that replay reached the ratchet, failed
            // to decrypt and was never ACKed, so it came back on every
            // reconnect. Retention is that window; a duplicate is re-ACKed,
            // which is what finally settles the sender.
            //
            // The cap has to hold a week of a busy device's inbound ids,
            // which 2000 did not.
            // Every flush rewrites the whole persisted set (one sealed record,
            // ~73 bytes of JSON per id), so it is also a write-size ceiling:
            // 5000 ids is ~365 KB per write, and a write needs 32 new ids
            // or a change 5 s old, so a quiet device writes nothing.
            max_tracked_messages: 5000,
            retention_time_secs: crate::constants::REDELIVERY_WINDOW_MS / 1000, // 7 days
            use_bloom_filter: false,
            bloom_filter_bits: 1 << 20, // ~1MB per filter (1,048,576 bits)
            bloom_hash_count: 7,        // ~1% false positive rate when Bloom is enabled
            bloom_filter_count: 4,      // 4 windows for 1-hour retention
            bloom_rotation_secs: 900,   // 15 minutes per window
        }
    }
}

/// A simple bloom filter implementation.
#[derive(Clone)]
struct BloomFilter {
    bits: Vec<u64>,
    bit_count: usize,
    hash_count: usize,
    items_added: usize,
}

impl BloomFilter {
    fn new(bit_count: usize, hash_count: usize) -> Self {
        let word_count = bit_count.div_ceil(64);
        Self {
            bits: vec![0u64; word_count],
            bit_count,
            hash_count,
            items_added: 0,
        }
    }

    /// Computes multiple hash values for an item using double hashing.
    fn hashes(&self, item: &str) -> Vec<usize> {
        let mut hashes = Vec::with_capacity(self.hash_count);

        // First hash using FNV-1a
        let h1 = fnv1a_hash(item.as_bytes());
        // Second hash using a different seed, forced odd: with a
        // power-of-two bit_count an even h2 collapses the probe indices
        // onto few bits (all onto one when h2 ≡ 0 mod bit_count), inflating
        // the false-positive rate; an odd step keeps all probes distinct.
        let h2 = fnv1a_hash_seeded(item.as_bytes(), 0x5f356495) | 1;

        for i in 0..self.hash_count {
            // Kirsch-Mitzenmacher optimization: h(i) = h1 + i * h2
            let combined = h1.wrapping_add((i as u64).wrapping_mul(h2));
            hashes.push((combined as usize) % self.bit_count);
        }

        hashes
    }

    fn insert(&mut self, item: &str) {
        for bit_index in self.hashes(item) {
            let word_index = bit_index / 64;
            let bit_offset = bit_index % 64;
            self.bits[word_index] |= 1u64 << bit_offset;
        }
        self.items_added += 1;
    }

    fn contains(&self, item: &str) -> bool {
        for bit_index in self.hashes(item) {
            let word_index = bit_index / 64;
            let bit_offset = bit_index % 64;
            if (self.bits[word_index] & (1u64 << bit_offset)) == 0 {
                return false;
            }
        }
        true
    }

    fn clear(&mut self) {
        self.bits.fill(0);
        self.items_added = 0;
    }

    fn items_count(&self) -> usize {
        self.items_added
    }

    /// Estimates the current false positive probability.
    fn estimated_false_positive_rate(&self) -> f64 {
        if self.items_added == 0 {
            return 0.0;
        }
        let m = self.bit_count as f64;
        let n = self.items_added as f64;
        let k = self.hash_count as f64;
        (1.0 - (-k * n / m).exp()).powf(k)
    }
}

/// FNV-1a hash function.
fn fnv1a_hash(data: &[u8]) -> u64 {
    fnv1a_hash_seeded(data, 0xcbf29ce484222325)
}

/// FNV-1a hash with custom seed.
fn fnv1a_hash_seeded(data: &[u8], seed: u64) -> u64 {
    const FNV_PRIME: u64 = 0x100000001b3;
    let mut hash = seed;
    for byte in data {
        hash ^= *byte as u64;
        hash = hash.wrapping_mul(FNV_PRIME);
    }
    hash
}

/// A rotating bloom filter with time windows.
struct RotatingBloomFilter {
    filters: Vec<BloomFilter>,
    current_index: usize,
    last_rotation: DateTime<Utc>,
    rotation_interval_secs: u64,
}

impl RotatingBloomFilter {
    fn new(filter_count: usize, bit_count: usize, hash_count: usize, rotation_secs: u64) -> Self {
        let filters = (0..filter_count)
            .map(|_| BloomFilter::new(bit_count, hash_count))
            .collect();
        Self {
            filters,
            current_index: 0,
            last_rotation: Utc::now(),
            rotation_interval_secs: rotation_secs,
        }
    }

    fn maybe_rotate(&mut self, now: DateTime<Utc>) {
        let elapsed = now.signed_duration_since(self.last_rotation);
        if elapsed.num_seconds() >= self.rotation_interval_secs as i64 {
            // Move to next filter and clear it
            self.current_index = (self.current_index + 1) % self.filters.len();
            self.filters[self.current_index].clear();
            self.last_rotation = now;
        }
    }

    fn insert(&mut self, item: &str) {
        self.maybe_rotate(Utc::now());
        self.filters[self.current_index].insert(item);
    }

    fn contains(&mut self, item: &str) -> bool {
        self.maybe_rotate(Utc::now());
        // Check all filters (current and previous windows)
        self.filters.iter().any(|f| f.contains(item))
    }

    fn total_items(&self) -> usize {
        self.filters.iter().map(|f| f.items_count()).sum()
    }

    fn clear(&mut self) {
        for filter in &mut self.filters {
            filter.clear();
        }
        self.current_index = 0;
        self.last_rotation = Utc::now();
    }

    fn average_false_positive_rate(&self) -> f64 {
        let rates: Vec<f64> = self
            .filters
            .iter()
            .map(|f| f.estimated_false_positive_rate())
            .collect();
        if rates.is_empty() {
            return 0.0;
        }
        rates.iter().sum::<f64>() / rates.len() as f64
    }
}

/// One id of the exact-mode seen set, as exported for persistence and
/// imported on the next launch. `seen_at_ms` is wall-clock milliseconds so
/// the retention window can be re-applied on import.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SeenId {
    /// The message id.
    pub id: String,
    /// When the id was first seen, unix milliseconds.
    pub seen_at_ms: i64,
}

/// Entry for a seen message (used in HashMap mode).
#[derive(Debug, Clone)]
struct SeenEntry {
    /// When this message was first seen.
    seen_at: DateTime<Utc>,
    /// When this entry was last accessed (for LRU eviction).
    last_accessed: DateTime<Utc>,
    /// Whether [`Deduplicator::export_seen`] includes this id. `false` for
    /// an id tracked with [`Deduplicator::mark_seen_local`]: a sender's own
    /// outgoing frame, which dedup only has to recognise while this process
    /// lives.
    exportable: bool,
}

/// Deduplicator for tracking seen messages and preventing duplicates.
///
/// Supports two modes:
/// - HashMap mode: exact tracking with FIFO eviction (original behavior)
/// - Bloom filter mode: O(1) memory with probabilistic duplicate detection
pub struct Deduplicator {
    config: DeduplicatorConfig,
    /// HashMap for exact tracking (fallback mode).
    seen_messages: HashMap<String, SeenEntry>,
    /// Rotating bloom filter for scalable tracking.
    bloom_filter: Option<RotatingBloomFilter>,
}

impl Deduplicator {
    /// Creates a new deduplicator with default configuration.
    pub fn new() -> Self {
        Self::with_config(DeduplicatorConfig::default())
    }

    /// Creates a new deduplicator with custom configuration.
    pub fn with_config(config: DeduplicatorConfig) -> Self {
        // Keep the low-level constructor panic-free for direct users of this
        // crate. The top-level ProtocolConfig rejects these invalid values;
        // here we fail safe to exact HashMap mode instead of constructing a
        // Bloom filter that would divide by zero or rotate an empty list.
        let valid_bloom_config = config.bloom_filter_bits > 0
            && config.bloom_hash_count > 0
            && config.bloom_filter_count > 0
            && config.bloom_rotation_secs > 0;
        let bloom_filter = if config.use_bloom_filter && valid_bloom_config {
            Some(RotatingBloomFilter::new(
                config.bloom_filter_count,
                config.bloom_filter_bits,
                config.bloom_hash_count,
                config.bloom_rotation_secs,
            ))
        } else {
            None
        };

        Self {
            config,
            seen_messages: HashMap::new(),
            bloom_filter,
        }
    }

    /// Applies a new configuration in place, carrying the tracked set across.
    ///
    /// Rebuilding from scratch on a runtime configuration change forgets every
    /// id held, which was equivalent to a restart while the set lived in memory
    /// only and is a loss now that it is persisted: the ids restored at launch
    /// would be dropped, and the next export would overwrite the record with a
    /// near-empty set — so the restart-time duplicate the record exists to
    /// recognise would be processed after all. The exact-mode entries therefore
    /// survive, re-bounded by the new configuration: ids past the new retention
    /// window are dropped, and past the new cap the newest are kept, with each
    /// entry's exportable flag intact so a carried-over
    /// [`Self::mark_seen_local`] id still stays out of [`Self::export_seen`].
    ///
    /// A change of mode carries nothing: a bloom filter has no ids to hand
    /// over and cannot take ids in.
    pub fn reconfigure(&mut self, config: DeduplicatorConfig) {
        let previous = std::mem::take(&mut self.seen_messages);
        let was_exact = self.bloom_filter.is_none();
        *self = Self::with_config(config);
        if !was_exact || self.bloom_filter.is_some() {
            return;
        }
        let retention = chrono::Duration::seconds(self.config.retention_time_secs as i64);
        let cutoff = Utc::now() - retention;
        let mut entries: Vec<(String, SeenEntry)> = previous
            .into_iter()
            .filter(|(_, entry)| entry.seen_at > cutoff)
            .collect();
        entries.sort_by(|left, right| {
            right
                .1
                .seen_at
                .cmp(&left.1.seen_at)
                .then_with(|| left.0.cmp(&right.0))
        });
        entries.truncate(self.config.max_tracked_messages);
        self.seen_messages = entries.into_iter().collect();
    }

    /// Returns `true` when deduplication is exact (HashMap mode): a positive
    /// [`Self::is_duplicate`] answer is then authoritative. In bloom mode
    /// (`false`) a positive answer may be a false positive (~1% with default
    /// settings), so callers must not treat a hit as proof of a prior send.
    pub fn is_exact(&self) -> bool {
        self.bloom_filter.is_none()
    }

    /// Checks if a message has been seen before.
    ///
    /// # Arguments
    ///
    /// * `message_id` - The message ID to check
    ///
    /// # Returns
    ///
    /// Returns `true` if the message has been seen, `false` otherwise.
    ///
    /// Note: In bloom filter mode, this may return false positives (claiming a
    /// message is a duplicate when it isn't). The false positive rate is typically
    /// below 1% with default settings.
    pub fn is_duplicate(&self, message_id: &MessageId) -> bool {
        let msg_id_str = message_id.as_str();

        if let Some(ref bloom) = self.bloom_filter {
            // Note: We need mutable access for rotation check, but this is safe
            // because we're only reading. We'll handle rotation in mark_seen.
            // For now, just check all filters.
            bloom.filters.iter().any(|f| f.contains(&msg_id_str))
        } else {
            self.seen_messages.contains_key(&msg_id_str)
        }
    }

    /// Checks if a message has been seen before (mutable version for bloom rotation
    /// and LRU access tracking).
    pub fn is_duplicate_mut(&mut self, message_id: &MessageId) -> bool {
        let msg_id_str = message_id.as_str();

        if let Some(ref mut bloom) = self.bloom_filter {
            bloom.contains(&msg_id_str)
        } else if let Some(entry) = self.seen_messages.get_mut(&msg_id_str) {
            entry.last_accessed = Utc::now();
            true
        } else {
            false
        }
    }

    /// Marks a message as seen.
    ///
    /// # Arguments
    ///
    /// * `message_id` - The message ID to mark as seen
    ///
    /// # Returns
    ///
    /// Returns `true` if this is a new message, `false` if it was already seen.
    pub fn mark_seen(&mut self, message_id: MessageId) -> bool {
        self.mark_seen_with(message_id, true)
    }

    /// Marks a message as seen for the life of this process only.
    ///
    /// Identical to [`Self::mark_seen`] for [`Self::is_duplicate`], eviction
    /// and retention, but the id is left out of [`Self::export_seen`]. For a
    /// sender marking its own outgoing frame: the mark exists to stop a
    /// relayed echo of that frame re-entering the receive path, which a
    /// restart does not change, and exporting it would spend the persisted
    /// record's cap on ids no peer will ever push at this node. A later
    /// [`Self::mark_seen`] of an id already tracked here is a no-op and does
    /// not make it exportable.
    ///
    /// # Returns
    ///
    /// Returns `true` if the message was newly marked, `false` if it was
    /// already tracked.
    pub fn mark_seen_local(&mut self, message_id: MessageId) -> bool {
        self.mark_seen_with(message_id, false)
    }

    fn mark_seen_with(&mut self, message_id: MessageId, exportable: bool) -> bool {
        let msg_id_str = message_id.as_str();

        if let Some(ref mut bloom) = self.bloom_filter {
            // Check if already seen
            if bloom.contains(&msg_id_str) {
                return false;
            }
            // Add to bloom filter
            bloom.insert(&msg_id_str);
            true
        } else {
            // HashMap mode (original implementation)
            if self.seen_messages.contains_key(&msg_id_str) {
                return false;
            }

            // Check if we've hit the limit
            if self.seen_messages.len() >= self.config.max_tracked_messages {
                self.cleanup_expired();

                if self.seen_messages.len() >= self.config.max_tracked_messages {
                    // LRU eviction, local marks first: they only stop our own
                    // frames echoing back through the mesh, which is a matter
                    // of minutes, so at the cap a local mark may last only
                    // until the next insert. An inbound id has to last the
                    // whole retention, because a sender may still replay it
                    // days later, and a replay this set forgot reaches
                    // decryption instead of a re-ACK. `false < true`, so
                    // local marks sort first.
                    if let Some((lru_id, _)) = self
                        .seen_messages
                        .iter()
                        .min_by_key(|(_, entry)| (entry.exportable, entry.last_accessed))
                        .map(|(id, entry)| (id.clone(), entry.clone()))
                    {
                        self.seen_messages.remove(&lru_id);
                    }
                }
            }

            let now = Utc::now();
            self.seen_messages.insert(
                msg_id_str,
                SeenEntry {
                    seen_at: now,
                    last_accessed: now,
                    exportable,
                },
            );

            true
        }
    }

    /// Removes a message from the seen set so a future copy is processed
    /// fresh instead of being treated as a duplicate.
    ///
    /// Intended for receivers that reject a message *after* marking it seen
    /// (e.g. an encryption-policy or security-gate rejection): forgetting the
    /// id makes a replay re-enter the same gate — and be rejected again
    /// without a delivery ACK — instead of hitting the duplicate re-ACK path,
    /// which would confirm to an injector that the target is online.
    ///
    /// # Returns
    ///
    /// Returns `true` if the message was tracked and has been removed.
    /// Bloom filters do not support removal, so in bloom filter mode this is
    /// a no-op returning `false`: a replayed rejected message is treated as a
    /// duplicate until its filter window rotates out.
    pub fn unmark_seen(&mut self, message_id: &MessageId) -> bool {
        if self.bloom_filter.is_some() {
            return false;
        }
        self.seen_messages.remove(&message_id.as_str()).is_some()
    }

    /// Exports up to `max` tracked ids, newest first, for persistence.
    ///
    /// Exact (HashMap) mode only: a bloom filter has no ids to export, so
    /// bloom mode returns an empty vector and the persisted set simply stays
    /// empty. Newest first so that a cap smaller than the tracked set keeps
    /// the ids most likely to be replayed — the recent ones. Ids tracked with
    /// [`Self::mark_seen_local`] are left out.
    pub fn export_seen(&self, max: usize) -> Vec<SeenId> {
        if self.bloom_filter.is_some() {
            return Vec::new();
        }
        let mut entries: Vec<SeenId> = self
            .seen_messages
            .iter()
            .filter(|(_, entry)| entry.exportable)
            .map(|(id, entry)| SeenId {
                id: id.clone(),
                seen_at_ms: entry.seen_at.timestamp_millis(),
            })
            .collect();
        entries.sort_by(|left, right| {
            right
                .seen_at_ms
                .cmp(&left.seen_at_ms)
                .then_with(|| left.id.cmp(&right.id))
        });
        entries.truncate(max);
        entries
    }

    /// Re-admits ids exported by [`Self::export_seen`] on a previous run.
    ///
    /// Skips ids already tracked (the live set wins), ids whose retention
    /// window has closed as of `now`, and stops at `max_tracked_messages` so
    /// a restore can never push the set over its cap or evict live entries.
    /// Entries are admitted in the order given, so a caller that wants the
    /// newest to survive a tight cap passes them newest first — which is the
    /// order `export_seen` produces. Returns how many were admitted. A no-op
    /// in bloom mode, which has nothing to import into.
    pub fn import_seen(&mut self, entries: Vec<SeenId>, now: DateTime<Utc>) -> usize {
        if self.bloom_filter.is_some() {
            return 0;
        }
        let retention = chrono::Duration::seconds(self.config.retention_time_secs as i64);
        let cutoff = now - retention;
        let mut imported = 0;
        for entry in entries {
            if self.seen_messages.len() >= self.config.max_tracked_messages {
                break;
            }
            let Some(seen_at) = DateTime::<Utc>::from_timestamp_millis(entry.seen_at_ms) else {
                continue;
            };
            if seen_at <= cutoff || self.seen_messages.contains_key(&entry.id) {
                continue;
            }
            self.seen_messages.insert(
                entry.id,
                SeenEntry {
                    seen_at,
                    last_accessed: seen_at,
                    exportable: true,
                },
            );
            imported += 1;
        }
        imported
    }

    /// Removes expired entries that exceed the retention time.
    ///
    /// # Returns
    ///
    /// Returns the number of entries removed.
    ///
    /// Note: In bloom filter mode, expiration is handled automatically through
    /// filter rotation, so this returns 0.
    pub fn cleanup_expired(&mut self) -> usize {
        if self.bloom_filter.is_some() {
            // Bloom filter handles expiration through rotation
            return 0;
        }

        let now = Utc::now();
        let retention = chrono::Duration::seconds(self.config.retention_time_secs as i64);
        let cutoff = now - retention;

        let before_count = self.seen_messages.len();
        self.seen_messages.retain(|_, entry| entry.seen_at > cutoff);
        before_count - self.seen_messages.len()
    }

    /// Gets the number of messages currently being tracked.
    ///
    /// Note: In bloom filter mode, this returns an estimate based on insertion count.
    pub fn tracked_count(&self) -> usize {
        if let Some(ref bloom) = self.bloom_filter {
            bloom.total_items()
        } else {
            self.seen_messages.len()
        }
    }

    /// Clears all tracked messages.
    pub fn clear(&mut self) {
        if let Some(ref mut bloom) = self.bloom_filter {
            bloom.clear();
        }
        self.seen_messages.clear();
    }

    /// Checks if a message ID is being tracked (seen).
    pub fn is_tracked(&self, message_id: &MessageId) -> bool {
        self.is_duplicate(message_id)
    }

    /// Returns whether bloom filter mode is active.
    pub fn is_bloom_filter_mode(&self) -> bool {
        self.bloom_filter.is_some()
    }

    /// Gets statistics about the deduplicator.
    pub fn stats(&self) -> DeduplicatorStats {
        if let Some(ref bloom) = self.bloom_filter {
            let total = bloom.total_items();
            // Estimate "capacity" based on expected items before FP rate gets too high
            let estimated_capacity = self.config.bloom_filter_bits / self.config.bloom_hash_count;
            let capacity_percent =
                ((total as f32 / estimated_capacity as f32) * 100.0).min(100.0) as u8;

            DeduplicatorStats {
                total_tracked: total,
                recent_tracked: total, // In bloom mode, we can't distinguish recent
                capacity_used_percent: capacity_percent,
                false_positive_rate: Some(bloom.average_false_positive_rate()),
                mode: DeduplicatorMode::BloomFilter,
            }
        } else {
            let now = Utc::now();
            let recent_cutoff = now - chrono::Duration::seconds(60);

            let recent_count = self
                .seen_messages
                .values()
                .filter(|entry| entry.seen_at > recent_cutoff)
                .count();

            DeduplicatorStats {
                total_tracked: self.seen_messages.len(),
                recent_tracked: recent_count,
                capacity_used_percent: (self.seen_messages.len() as f32
                    / self.config.max_tracked_messages as f32
                    * 100.0) as u8,
                false_positive_rate: None,
                mode: DeduplicatorMode::HashMap,
            }
        }
    }

    /// Gets the estimated memory usage in bytes.
    pub fn estimated_memory_bytes(&self) -> usize {
        if self.bloom_filter.is_some() {
            // Each filter uses bit_count/8 bytes, times number of filters
            let filter_bytes = self.config.bloom_filter_bits.div_ceil(8);
            filter_bytes * self.config.bloom_filter_count
                + std::mem::size_of::<RotatingBloomFilter>()
        } else {
            // Rough estimate: each entry is ~50 bytes (String + DateTime)
            self.seen_messages.len() * 50 + std::mem::size_of::<HashMap<String, SeenEntry>>()
        }
    }
}

impl Default for Deduplicator {
    fn default() -> Self {
        Self::new()
    }
}

/// The mode of operation for the deduplicator.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeduplicatorMode {
    /// Using HashMap for exact tracking (original behavior).
    HashMap,
    /// Using bloom filters for O(1) memory.
    BloomFilter,
}

/// Statistics about the deduplicator.
#[derive(Debug, Clone)]
pub struct DeduplicatorStats {
    /// Total number of messages being tracked.
    pub total_tracked: usize,
    /// Number of messages seen in the last minute.
    pub recent_tracked: usize,
    /// Percentage of capacity used (0-100).
    pub capacity_used_percent: u8,
    /// Estimated false positive rate (only in bloom filter mode).
    pub false_positive_rate: Option<f64>,
    /// Current operating mode.
    pub mode: DeduplicatorMode,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::thread;
    use std::time::Duration;

    fn hashmap_config() -> DeduplicatorConfig {
        DeduplicatorConfig {
            use_bloom_filter: false,
            ..Default::default()
        }
    }

    #[test]
    fn test_default_config_is_hashmap_and_capacity_5000() {
        let config = DeduplicatorConfig::default();
        assert!(
            !config.use_bloom_filter,
            "default should be exact-match HashMap to avoid false positives"
        );
        assert_eq!(config.max_tracked_messages, 5000);
        // 7 days: the sender's redelivery window, not a day.
        assert_eq!(config.retention_time_secs, 604_800);

        let dedup = Deduplicator::new();
        assert!(!dedup.is_bloom_filter_mode());
        let stats = dedup.stats();
        assert_eq!(stats.mode, DeduplicatorMode::HashMap);
    }

    #[test]
    fn test_invalid_bloom_config_falls_back_without_panicking() {
        let config = DeduplicatorConfig {
            use_bloom_filter: true,
            bloom_filter_bits: 0,
            ..Default::default()
        };

        let mut dedup = Deduplicator::with_config(config);
        let message_id = MessageId::new();

        assert!(!dedup.is_bloom_filter_mode());
        assert!(dedup.mark_seen(message_id.clone()));
        assert!(dedup.is_duplicate(&message_id));
    }

    #[test]
    fn test_default_hashmap_no_false_positives() {
        let mut dedup = Deduplicator::new();
        let seen = MessageId::new();
        dedup.mark_seen(seen.clone());

        // Brand-new ID must never be reported as duplicate (HashMap is exact match)
        for _ in 0..20 {
            let fresh = MessageId::new();
            assert!(
                !dedup.is_duplicate(&fresh),
                "fresh message ID must not be false positive"
            );
        }
        assert!(dedup.is_duplicate(&seen));
    }

    #[test]
    fn test_basic_deduplication() {
        let mut dedup = Deduplicator::new();
        let msg_id = MessageId::new();

        // First time should not be duplicate
        assert!(!dedup.is_duplicate(&msg_id));
        assert!(dedup.mark_seen(msg_id.clone()));

        // Second time should be duplicate
        assert!(dedup.is_duplicate(&msg_id));
        assert!(!dedup.mark_seen(msg_id.clone()));
    }

    #[test]
    fn test_unmark_seen_hashmap_mode() {
        let mut dedup = Deduplicator::with_config(hashmap_config());
        let msg_id = MessageId::new();

        dedup.mark_seen(msg_id.clone());
        assert!(dedup.is_duplicate(&msg_id));

        assert!(dedup.unmark_seen(&msg_id), "tracked id must be removed");
        assert!(
            !dedup.is_duplicate(&msg_id),
            "unmarked message must be processed fresh on replay"
        );
        assert!(
            !dedup.unmark_seen(&msg_id),
            "second unmark of the same id is a no-op"
        );
    }

    #[test]
    fn test_unmark_seen_noop_in_bloom_mode() {
        let config = DeduplicatorConfig {
            use_bloom_filter: true,
            bloom_filter_bits: 1 << 16,
            bloom_hash_count: 5,
            bloom_filter_count: 2,
            bloom_rotation_secs: 60,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);
        let msg_id = MessageId::new();

        dedup.mark_seen(msg_id.clone());
        assert!(
            !dedup.unmark_seen(&msg_id),
            "bloom filters cannot remove entries"
        );
        assert!(
            dedup.is_duplicate(&msg_id),
            "bloom mode keeps the id until its window rotates"
        );
    }

    #[test]
    fn test_basic_deduplication_hashmap_mode() {
        let mut dedup = Deduplicator::with_config(hashmap_config());
        let msg_id = MessageId::new();

        assert!(!dedup.is_duplicate(&msg_id));
        assert!(dedup.mark_seen(msg_id.clone()));

        assert!(dedup.is_duplicate(&msg_id));
        assert!(!dedup.mark_seen(msg_id.clone()));
    }

    #[test]
    fn test_multiple_messages() {
        let mut dedup = Deduplicator::new();

        let msg1 = MessageId::new();
        let msg2 = MessageId::new();
        let msg3 = MessageId::new();

        assert!(dedup.mark_seen(msg1.clone()));
        assert!(dedup.mark_seen(msg2.clone()));
        assert!(dedup.mark_seen(msg3.clone()));

        assert_eq!(dedup.tracked_count(), 3);

        assert!(dedup.is_duplicate(&msg1));
        assert!(dedup.is_duplicate(&msg2));
        assert!(dedup.is_duplicate(&msg3));
    }

    #[test]
    fn test_capacity_limit() {
        // HashMap mode required for FIFO eviction
        let config = DeduplicatorConfig {
            max_tracked_messages: 3,
            retention_time_secs: 3600,
            use_bloom_filter: false,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        let msg1 = MessageId::new();
        let msg2 = MessageId::new();
        let msg3 = MessageId::new();
        let msg4 = MessageId::new();

        dedup.mark_seen(msg1.clone());
        dedup.mark_seen(msg2.clone());
        dedup.mark_seen(msg3.clone());

        assert_eq!(dedup.tracked_count(), 3);

        // Adding fourth message should evict oldest
        dedup.mark_seen(msg4.clone());
        assert_eq!(dedup.tracked_count(), 3);

        // msg1 should have been evicted
        assert!(!dedup.is_duplicate(&msg1));
        assert!(dedup.is_duplicate(&msg4));
    }

    #[test]
    fn test_cleanup_expired() {
        // HashMap mode required for explicit cleanup
        let config = DeduplicatorConfig {
            max_tracked_messages: 100,
            retention_time_secs: 1,
            use_bloom_filter: false,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        let msg1 = MessageId::new();
        let msg2 = MessageId::new();

        dedup.mark_seen(msg1.clone());

        thread::sleep(Duration::from_millis(500));

        dedup.mark_seen(msg2.clone());

        thread::sleep(Duration::from_millis(600));

        let removed = dedup.cleanup_expired();
        assert_eq!(removed, 1);
        assert_eq!(dedup.tracked_count(), 1);

        assert!(!dedup.is_duplicate(&msg1));
        assert!(dedup.is_duplicate(&msg2));
    }

    #[test]
    fn test_clear() {
        let mut dedup = Deduplicator::new();

        dedup.mark_seen(MessageId::new());
        dedup.mark_seen(MessageId::new());
        dedup.mark_seen(MessageId::new());

        assert_eq!(dedup.tracked_count(), 3);

        dedup.clear();
        assert_eq!(dedup.tracked_count(), 0);
    }

    #[test]
    fn test_is_tracked() {
        let mut dedup = Deduplicator::new();
        let msg_id = MessageId::new();

        assert!(!dedup.is_tracked(&msg_id));

        dedup.mark_seen(msg_id.clone());
        assert!(dedup.is_tracked(&msg_id));
    }

    #[test]
    fn test_stats() {
        let mut dedup = Deduplicator::new();

        for _ in 0..5 {
            dedup.mark_seen(MessageId::new());
        }

        let stats = dedup.stats();
        assert_eq!(stats.total_tracked, 5);
    }

    #[test]
    fn test_stats_hashmap_mode() {
        let mut dedup = Deduplicator::with_config(hashmap_config());

        for _ in 0..5 {
            dedup.mark_seen(MessageId::new());
        }

        let stats = dedup.stats();
        assert_eq!(stats.total_tracked, 5);
        assert_eq!(stats.recent_tracked, 5);
        assert!(stats.capacity_used_percent < 1);
        assert_eq!(stats.mode, DeduplicatorMode::HashMap);
    }

    #[test]
    fn test_capacity_percent() {
        let config = DeduplicatorConfig {
            max_tracked_messages: 10,
            retention_time_secs: 3600,
            use_bloom_filter: false,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        for _ in 0..5 {
            dedup.mark_seen(MessageId::new());
        }

        let stats = dedup.stats();
        assert_eq!(stats.capacity_used_percent, 50);
    }

    #[test]
    fn test_bloom_filter_mode() {
        let config = DeduplicatorConfig {
            use_bloom_filter: true,
            bloom_filter_bits: 1 << 16, // Smaller for testing
            bloom_hash_count: 5,
            bloom_filter_count: 2,
            bloom_rotation_secs: 60,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        assert!(dedup.is_bloom_filter_mode());

        let msg1 = MessageId::new();
        let msg2 = MessageId::new();

        assert!(!dedup.is_duplicate(&msg1));
        dedup.mark_seen(msg1.clone());
        assert!(dedup.is_duplicate(&msg1));

        assert!(!dedup.is_duplicate(&msg2));
        dedup.mark_seen(msg2.clone());
        assert!(dedup.is_duplicate(&msg2));

        let stats = dedup.stats();
        assert_eq!(stats.mode, DeduplicatorMode::BloomFilter);
        assert!(stats.false_positive_rate.is_some());
    }

    #[test]
    fn test_bloom_filter_constant_memory() {
        let config = DeduplicatorConfig {
            use_bloom_filter: true,
            bloom_filter_bits: 1 << 16,
            bloom_hash_count: 5,
            bloom_filter_count: 2,
            bloom_rotation_secs: 3600,
            ..Default::default()
        };
        let dedup = Deduplicator::with_config(config);

        let initial_memory = dedup.estimated_memory_bytes();

        // Memory should be roughly constant regardless of items added
        // (Unlike HashMap which grows with items)
        let expected_filter_bytes = (1 << 16) / 8 * 2; // 2 filters, 64KB each
        assert!(initial_memory >= expected_filter_bytes);
        assert!(initial_memory < expected_filter_bytes * 2); // Some overhead
    }

    #[test]
    fn test_bloom_filter_high_volume() {
        let config = DeduplicatorConfig {
            use_bloom_filter: true,
            bloom_filter_bits: 1 << 20, // 1MB
            bloom_hash_count: 7,
            bloom_filter_count: 4,
            bloom_rotation_secs: 3600,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        // Insert 10,000 messages - should work with O(1) memory. The filter
        // is probabilistic: a fresh id can (rarely) be reported as already
        // seen, so tolerate a handful of false positives rather than
        // asserting on every insert.
        let mut inserted = 0usize;
        for _ in 0..10_000 {
            let msg = MessageId::new();
            if dedup.mark_seen(msg) {
                inserted += 1;
            }
        }

        assert!(
            inserted >= 9_990,
            "too many bloom false positives: {}",
            10_000 - inserted
        );
        assert_eq!(dedup.tracked_count(), inserted);

        let stats = dedup.stats();
        // False positive rate should still be low
        assert!(stats.false_positive_rate.unwrap() < 0.02);
    }

    #[test]
    fn test_eviction_takes_local_marks_before_inbound_ids() {
        let config = DeduplicatorConfig {
            max_tracked_messages: 3,
            retention_time_secs: 3600,
            use_bloom_filter: false,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        // The inbound id is the oldest, so plain LRU would take it first.
        let inbound = MessageId::new();
        dedup.mark_seen(inbound.clone());
        thread::sleep(Duration::from_millis(10));
        let local_a = MessageId::new();
        dedup.mark_seen_local(local_a.clone());
        thread::sleep(Duration::from_millis(10));
        let local_b = MessageId::new();
        dedup.mark_seen_local(local_b.clone());

        dedup.mark_seen(MessageId::new());

        assert!(
            dedup.is_duplicate(&inbound),
            "a replay of the inbound id must still be recognised"
        );
        assert!(!dedup.is_duplicate(&local_a), "the oldest local mark goes");
        assert!(dedup.is_duplicate(&local_b));
    }

    #[test]
    fn test_lru_eviction_preserves_recently_accessed() {
        let config = DeduplicatorConfig {
            max_tracked_messages: 3,
            retention_time_secs: 3600,
            use_bloom_filter: false,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);

        let msg_a = MessageId::new();
        let msg_b = MessageId::new();
        let msg_c = MessageId::new();

        dedup.mark_seen(msg_a.clone());
        thread::sleep(Duration::from_millis(10));
        dedup.mark_seen(msg_b.clone());
        thread::sleep(Duration::from_millis(10));
        dedup.mark_seen(msg_c.clone());

        // Re-access A so it becomes the most recently accessed
        assert!(dedup.is_duplicate_mut(&msg_a));

        // Insert D — should evict B (least recently accessed), not A
        let msg_d = MessageId::new();
        dedup.mark_seen(msg_d.clone());

        assert_eq!(dedup.tracked_count(), 3);
        assert!(
            dedup.is_duplicate(&msg_a),
            "A was recently accessed, should survive"
        );
        assert!(
            !dedup.is_duplicate(&msg_b),
            "B should have been evicted (LRU)"
        );
        assert!(dedup.is_duplicate(&msg_c), "C should survive");
        assert!(dedup.is_duplicate(&msg_d), "D was just inserted");
    }
    #[test]
    fn test_export_import_round_trip_newest_first() {
        let mut dedup = Deduplicator::new();
        let ids: Vec<MessageId> = (0..5).map(|_| MessageId::new()).collect();
        for id in &ids {
            dedup.mark_seen(id.clone());
        }
        let exported = dedup.export_seen(usize::MAX);
        assert_eq!(exported.len(), 5);
        assert!(
            exported
                .windows(2)
                .all(|pair| pair[0].seen_at_ms >= pair[1].seen_at_ms),
            "export is newest first"
        );
        // A cap keeps the newest, not an arbitrary subset.
        let capped = dedup.export_seen(2);
        assert_eq!(capped.len(), 2);
        assert_eq!(capped, exported[..2].to_vec());

        let mut restored = Deduplicator::new();
        assert_eq!(restored.import_seen(exported.clone(), Utc::now()), 5);
        for id in &ids {
            assert!(restored.is_duplicate(id));
        }
        // The seen_at survives the trip, so a later export agrees.
        let mut again = restored.export_seen(usize::MAX);
        let mut original = exported;
        again.sort_by(|a, b| a.id.cmp(&b.id));
        original.sort_by(|a, b| a.id.cmp(&b.id));
        assert_eq!(again, original);
    }

    #[test]
    fn test_import_ignores_expired_existing_and_respects_cap() {
        let config = DeduplicatorConfig {
            max_tracked_messages: 3,
            retention_time_secs: 3600,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);
        let now = Utc::now();
        let live = MessageId::new();
        let expired = MessageId::new();
        let fresh: Vec<MessageId> = (0..3).map(|_| MessageId::new()).collect();
        dedup.mark_seen(live.clone());

        let seen = |id: &MessageId, secs_ago: i64| SeenId {
            id: id.as_str(),
            seen_at_ms: (now - chrono::Duration::seconds(secs_ago)).timestamp_millis(),
        };
        let entries = vec![
            // Expired: past the retention window as of `now`.
            seen(&expired, 3601),
            // Already tracked: the live entry wins and this does not count.
            seen(&live, 10),
            seen(&fresh[0], 5),
            seen(&fresh[1], 4),
            // Over the cap once the two above are in: refused.
            seen(&fresh[2], 3),
        ];
        assert_eq!(dedup.import_seen(entries, now), 2);
        assert_eq!(dedup.tracked_count(), 3);
        assert!(!dedup.is_duplicate(&expired));
        assert!(dedup.is_duplicate(&live));
        assert!(dedup.is_duplicate(&fresh[0]));
        assert!(dedup.is_duplicate(&fresh[1]));
        assert!(!dedup.is_duplicate(&fresh[2]));
    }

    #[test]
    fn test_local_marks_dedup_but_are_not_exported() {
        let mut dedup = Deduplicator::new();
        let inbound = MessageId::new();
        let own = MessageId::new();
        assert!(dedup.mark_seen(inbound.clone()));
        assert!(dedup.mark_seen_local(own.clone()));
        // Both dedup in memory alike.
        assert!(dedup.is_duplicate(&inbound));
        assert!(dedup.is_duplicate(&own));
        assert!(!dedup.mark_seen_local(own.clone()), "already tracked");
        // A later plain mark of a local id is a no-op and does not promote it.
        assert!(!dedup.mark_seen(own.clone()));
        // Only the inbound id reaches the export.
        let exported = dedup.export_seen(usize::MAX);
        assert_eq!(exported.len(), 1);
        assert_eq!(exported[0].id, inbound.as_str());
        // A local id can still be released like any other.
        assert!(dedup.unmark_seen(&own));
        assert!(!dedup.is_duplicate(&own));
    }

    #[test]
    fn test_export_import_are_noops_in_bloom_mode() {
        let config = DeduplicatorConfig {
            use_bloom_filter: true,
            ..Default::default()
        };
        let mut dedup = Deduplicator::with_config(config);
        dedup.mark_seen(MessageId::new());
        assert!(dedup.export_seen(usize::MAX).is_empty());
        let imported = MessageId::new();
        assert_eq!(
            dedup.import_seen(
                vec![SeenId {
                    id: imported.as_str(),
                    seen_at_ms: Utc::now().timestamp_millis(),
                }],
                Utc::now()
            ),
            0
        );
        assert!(!dedup.is_duplicate(&imported));
    }

    #[test]
    fn test_reconfigure_carries_tracked_ids_within_the_new_bounds() {
        let mut dedup = Deduplicator::new();
        let now = Utc::now();
        let expired = MessageId::new();
        let oldest_live = MessageId::new();
        let inbound = MessageId::new();
        let own = MessageId::new();
        // Plant two ids with explicit ages, then two fresh ones.
        assert_eq!(
            dedup.import_seen(
                vec![
                    SeenId {
                        id: expired.as_str(),
                        seen_at_ms: (now - chrono::Duration::seconds(7200)).timestamp_millis(),
                    },
                    SeenId {
                        id: oldest_live.as_str(),
                        seen_at_ms: (now - chrono::Duration::seconds(60)).timestamp_millis(),
                    },
                ],
                now
            ),
            2
        );
        assert!(dedup.mark_seen(inbound.clone()));
        assert!(dedup.mark_seen_local(own.clone()));

        // Tighter retention drops the expired id; a cap of 2 keeps the two
        // newest of what is left, whichever way they were marked.
        dedup.reconfigure(DeduplicatorConfig {
            max_tracked_messages: 2,
            retention_time_secs: 3600,
            ..Default::default()
        });
        assert_eq!(dedup.tracked_count(), 2);
        assert!(
            !dedup.is_duplicate(&expired),
            "past the new retention window"
        );
        assert!(!dedup.is_duplicate(&oldest_live), "past the new cap");
        assert!(dedup.is_duplicate(&inbound));
        assert!(dedup.is_duplicate(&own));
        // The exportable flag rides along: the local mark stays local.
        let exported = dedup.export_seen(usize::MAX);
        assert_eq!(exported.len(), 1);
        assert_eq!(exported[0].id, inbound.as_str());

        // A mode change carries nothing either way.
        dedup.reconfigure(DeduplicatorConfig {
            use_bloom_filter: true,
            ..Default::default()
        });
        assert!(dedup.is_bloom_filter_mode());
        assert!(!dedup.is_duplicate(&inbound));
        dedup.mark_seen(MessageId::new());
        dedup.reconfigure(DeduplicatorConfig::default());
        assert!(!dedup.is_bloom_filter_mode());
        assert_eq!(dedup.tracked_count(), 0);
    }
}