tiny_counter/store/

mod.rs

pub mod builder;
pub(crate) mod inner;
pub mod limiter;
pub mod query;

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};

use chrono::{DateTime, Duration, Utc};
use dashmap::DashMap;

use crate::config_converter::convert_if_needed;
use crate::{
    Clock, DeltaQuery, Error, EventCounterConfig, Formatter, MultiQuery, Query, RatioQuery, Result,
    SingleEventCounter, Storage, SystemClock, TimeUnit,
};

use self::inner::EventStoreInner;
use self::limiter::Limiter;

/// Top-level store for tracking multiple events with automatic counter creation.
///
/// EventStore provides a high-level API for recording and querying events.
/// Each event is tracked independently with its own SingleEventCounter.
/// Most methods use &self and internal locking for thread safety.
///
/// # Thread Safety and Sharing
///
/// EventStore is NOT Clone. To share across threads:
/// - Use `Arc<EventStore>` for read-heavy workloads (record, query)
/// - Use `Arc<Mutex<EventStore>>` when calling `compact()` (&mut self methods)
///
/// Auto-persistence is configured via the builder and managed internally.
///
/// # Drop and Synchronous I/O Behavior
///
/// **IMPORTANT**: When an EventStore is dropped, it performs **synchronous I/O** by calling
/// [`persist()`](Self::persist) if any events are dirty. This ensures data is not lost, but has implications:
///
/// ## When Drop I/O Matters
///
/// - **Async contexts**: Drop runs synchronously and will block the executor. This can cause
///   performance issues or even deadlocks in async code.
/// - **Performance-sensitive code**: Drop may take time proportional to the number of dirty events
///   and the storage backend's performance.
/// - **Panic unwinding**: If a panic occurs, Drop still runs, but persistence errors are silently
///   ignored (they cannot be returned during unwinding).
///
/// ## Best Practices for Cleanup
///
/// **Option 1 - Use `close()`** (recommended for clarity):
///
/// ```rust
/// # #[cfg(feature = "serde")]
/// # {
/// use tiny_counter::{EventStore, storage::MemoryStorage};
///
/// # async fn example() {
/// let store = EventStore::builder()
///     .with_storage(MemoryStorage::new())
///     .build()
///     .unwrap();
///
/// store.record("user_action");
///
/// // Explicitly close with error handling
/// store.close().expect("Failed to close store");
/// // Store is now consumed
/// # }
/// # }
/// ```
///
/// **Option 2 - Persist then drop**:
///
/// ```rust
/// # #[cfg(feature = "serde")]
/// # {
/// use tiny_counter::{EventStore, storage::MemoryStorage};
///
/// # async fn example() {
/// let store = EventStore::builder()
///     .with_storage(MemoryStorage::new())
///     .build()
///     .unwrap();
///
/// store.record("user_action");
///
/// // Explicitly persist before drop in async context
/// store.persist().expect("Failed to persist events");
/// // Now drop is safe - either no dirty data, or we already handled errors
/// drop(store);
/// # }
/// # }
/// ```
///
/// **With auto-persist**: If you configured auto-persist via the builder, Drop still performs
/// a final persist to catch any events recorded after the last auto-persist cycle:
///
/// ```rust
/// # #[cfg(all(feature = "tokio", feature = "serde"))]
/// # {
/// use tiny_counter::{EventStore, storage::MemoryStorage};
/// use chrono::Duration;
///
/// # async fn example() {
/// let store = EventStore::builder()
///     .with_storage(MemoryStorage::new())
///     .auto_persist(Duration::seconds(60))  // Background task every 60s
///     .build()
///     .unwrap();
///
/// store.record("event1");
/// // Auto-persist will eventually save this
///
/// tokio::time::sleep(std::time::Duration::from_millis(100)).await;
///
/// store.record("event2");
/// // This might not be saved yet by auto-persist!
///
/// // Option 1: Explicit persist before drop (recommended for critical data)
/// store.persist().expect("Failed to persist");
/// drop(store);
///
/// // Option 2: Let Drop handle it (blocks executor, errors ignored)
/// // drop(store);  // Will call persist() synchronously
/// # }
/// # }
/// ```
///
/// ## Tradeoffs
///
/// - **Drop with persist**: Convenient, prevents data loss, but blocks and may fail silently
/// - **Explicit persist**: More verbose, but allows error handling and avoids blocking Drop
/// - **Auto-persist**: Reduces manual persist calls, but Drop still needed for final flush
///
/// ## Alternative: No-Op Drop
///
/// If you don't configure storage (no `with_storage()` in builder), Drop is a no-op and has
/// no I/O implications.
pub struct EventStore {
    pub(crate) inner: Arc<EventStoreInner>,
    #[cfg(feature = "tokio")]
    pub(crate) auto_persist_handle: Option<tokio::task::JoinHandle<()>>,
}

impl EventStore {
    /// Creates a new EventStore with default configuration.
    ///
    /// Default configuration includes 6 time units totaling 256 buckets:
    /// - 60 Minutes
    /// - 72 Hours
    /// - 56 Days
    /// - 52 Weeks
    /// - 12 Months
    /// - 4 Years
    pub fn new() -> Self {
        Self::from_parts(
            SystemClock::new(),
            None,
            None,
            EventCounterConfig::default(),
        )
    }

    /// Gets current time from clock.
    pub fn clock_now(&self) -> DateTime<Utc> {
        self.inner.clock_now()
    }

    /// Creates an EventStore from individual components.
    ///
    /// This is an internal constructor used by the builder.
    pub(crate) fn from_parts(
        clock: Arc<dyn Clock>,
        storage: Option<Box<dyn Storage>>,
        formatter: Option<Arc<dyn crate::Formatter>>,
        config: EventCounterConfig,
    ) -> Self {
        Self {
            inner: Arc::new(EventStoreInner {
                events: DashMap::new(),
                clock,
                storage: storage.map(|s| Arc::new(Mutex::new(s))),
                formatter,
                config,
            }),
            #[cfg(feature = "tokio")]
            auto_persist_handle: None,
        }
    }

    /// Records a single event at the current time.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record("app_launch");
    /// store.record("button_click");
    /// ```
    pub fn record(&self, event_id: impl EventId) {
        self.record_count(event_id, 1);
    }

    /// Records multiple events at the current time.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record_count("api_call", 5);
    /// store.record_count("page_view", 10);
    /// ```
    pub fn record_count(&self, event_id: impl EventId, count: u32) {
        let counter = self.inner.get_counter_for_record(event_id.as_ref());
        let now = self.inner.clock.now();
        let mut counter = counter.lock().unwrap();
        counter.advance_if_needed(now);
        counter.record(count);
        counter.mark_dirty();
    }

    /// Records a single event at a specific time.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    /// use chrono::{Duration, Utc};
    ///
    /// let store = EventStore::new();
    /// let two_days_ago = Utc::now() - Duration::days(2);
    /// store.record_at("feature_used", two_days_ago).unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// Returns `Error::FutureEvent` if the timestamp is in the future.
    ///
    /// **Note on old events**: Events beyond the tracking window are silently dropped with no error.
    /// This is intentional - the library uses fixed-size rotating buckets, and old data falls off
    /// as new data arrives. This represents a loss of granularity (e.g., "25 hours ago" becomes
    /// "1 day ago") rather than complete data loss. For production use, prefer `record()` for
    /// real-time events. Use `record_at()` and `record_ago()` primarily for testing and backfilling.
    pub fn record_at(&self, event_id: impl EventId, time: DateTime<Utc>) -> Result<()> {
        self.record_count_at(event_id, 1, time)
    }

    /// Records multiple events at a specific time.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    /// use chrono::{Duration, Utc};
    ///
    /// let store = EventStore::new();
    /// let yesterday = Utc::now() - Duration::days(1);
    /// store.record_count_at("sync_event", 3, yesterday).unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// Returns `Error::FutureEvent` if the timestamp is in the future.
    ///
    /// **Note on old events**: Events beyond the tracking window are silently dropped with no error.
    /// This is intentional - the library uses fixed-size rotating buckets, and old data falls off
    /// as new data arrives. This represents a loss of granularity (e.g., "25 hours ago" becomes
    /// "1 day ago") rather than complete data loss. For production use, prefer `record()` for
    /// real-time events. Use `record_count_at()` and `record_count_ago()` primarily for testing and backfilling.
    pub fn record_count_at(
        &self,
        event_id: impl EventId,
        count: u32,
        time: DateTime<Utc>,
    ) -> Result<()> {
        let counter = self.inner.get_counter_for_record(event_id.as_ref());
        let now = self.inner.clock.now();
        let mut counter = counter.lock().unwrap();
        counter.advance_if_needed(now);
        counter.record_at(count, time)?;
        counter.mark_dirty();
        Ok(())
    }

    /// Records a single event that occurred a duration ago.
    ///
    /// **Important**: Events outside the tracking window are silently dropped with no error or warning.
    /// The tracking window depends on your configuration. With default settings (256 buckets across
    /// 6 time units), events older than approximately 4 years are dropped.
    ///
    /// This represents a loss of granularity (e.g., "25 hours ago" becomes "1 day ago"), not complete
    /// data loss. The library uses fixed-size rotating buckets - old data falls off as new data arrives.
    ///
    /// **Recommendation**: Use `record()` for production real-time events. Use `record_ago()` primarily
    /// for testing queries or backfilling recent historical data within the tracking window.
    ///
    /// This method never returns an error. Use [`record_at`](Self::record_at) if you need
    /// to detect when events fall outside the tracking window.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    /// use chrono::Duration;
    ///
    /// let store = EventStore::new();
    /// store.record_ago("sync", Duration::hours(3));
    ///
    /// // Events too old are dropped silently
    /// store.record_ago("ancient", Duration::days(365 * 10));
    /// let sum = store.query("ancient").ever().sum();
    /// assert_eq!(sum, Some(0)); // Event was dropped
    /// ```
    pub fn record_ago(&self, event_id: impl EventId, duration: Duration) {
        self.record_count_ago(event_id, 1, duration);
    }

    /// Records multiple events that occurred a duration ago.
    ///
    /// **Important**: Events outside the tracking window are silently dropped with no error or warning.
    /// The tracking window depends on your configuration. With default settings (256 buckets across
    /// 6 time units), events older than approximately 4 years are dropped.
    ///
    /// This represents a loss of granularity (e.g., "25 hours ago" becomes "1 day ago"), not complete
    /// data loss. The library uses fixed-size rotating buckets - old data falls off as new data arrives.
    ///
    /// **Recommendation**: Use `record_count()` for production real-time events. Use `record_count_ago()`
    /// primarily for testing queries or backfilling recent historical data within the tracking window.
    ///
    /// This method never returns an error. Use [`record_count_at`](Self::record_count_at) if you need
    /// to detect when events fall outside the tracking window.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    /// use chrono::Duration;
    ///
    /// let store = EventStore::new();
    /// store.record_count_ago("batch_process", 5, Duration::days(1));
    ///
    /// // Events too old are dropped silently
    /// store.record_count_ago("ancient_batch", 100, Duration::days(365 * 10));
    /// let sum = store.query("ancient_batch").ever().sum();
    /// assert_eq!(sum, Some(0)); // Events were dropped
    /// ```
    pub fn record_count_ago(&self, event_id: impl EventId, count: u32, duration: Duration) {
        let now = self.inner.clock.now();
        let time = now - duration;
        // Events outside tracking window are silently dropped.
        // This matches the API design where record_ago methods are infallible.
        let _ = self.record_count_at(event_id, count, time);
    }

    /// Creates a query builder for a single event.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record("app_launch");
    ///
    /// let count = store.query("app_launch").last_days(7).sum();
    /// assert_eq!(count, Some(1));
    /// ```
    pub fn query(&self, event_id: impl EventId) -> Query {
        Query::new(self.inner.clone(), event_id.as_ref().to_string())
    }

    /// Creates a query builder for multiple events.
    ///
    /// Combines counts from multiple events into a single sum.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record("app_launch");
    /// store.record("app_resume");
    ///
    /// let total_opens = store
    ///     .query_many(&["app_launch", "app_resume"])
    ///     .last_days(7)
    ///     .sum();
    ///
    /// assert_eq!(total_opens, Some(2));
    /// ```
    pub fn query_many(&self, event_ids: &[impl EventId]) -> MultiQuery {
        let event_ids_owned: Vec<String> =
            event_ids.iter().map(|s| s.as_ref().to_string()).collect();
        MultiQuery::new(self.inner.clone(), event_ids_owned)
    }

    /// Creates a ratio query builder for two events.
    ///
    /// Calculates the ratio of numerator to denominator events.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record_count("conversions", 25);
    /// store.record_count("visits", 100);
    ///
    /// let conversion_rate = store
    ///     .query_ratio("conversions", "visits")
    ///     .last_days(7);
    ///
    /// assert_eq!(conversion_rate, Some(0.25));
    /// ```
    pub fn query_ratio(&self, numerator: impl EventId, denominator: impl EventId) -> RatioQuery {
        RatioQuery::new(
            self.inner.clone(),
            numerator.as_ref().to_string(),
            denominator.as_ref().to_string(),
        )
    }

    /// Creates a delta query builder for two events.
    ///
    /// Calculates the net change (positive - negative) between two events.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record_count("items_added", 10);
    /// store.record_count("items_removed", 3);
    ///
    /// let inventory_change = store
    ///     .query_delta("items_added", "items_removed")
    ///     .last_days(7)
    ///     .sum();
    ///
    /// assert_eq!(inventory_change, 7);
    /// ```
    pub fn query_delta(&self, positive: impl EventId, negative: impl EventId) -> DeltaQuery {
        DeltaQuery::new(
            self.inner.clone(),
            positive.as_ref().to_string(),
            negative.as_ref().to_string(),
        )
    }

    /// Returns whether any events have been modified since the last persist.
    pub fn is_dirty(&self) -> bool {
        for entry in self.inner.events.iter() {
            let counter = entry.value().lock().unwrap();
            if counter.is_dirty() {
                return true;
            }
        }
        false
    }

    /// Creates a rate limiter builder for checking constraints.
    ///
    /// Use this to create complex rate limiting rules with multiple constraints.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::{EventStore, TimeUnit};
    ///
    /// let store = EventStore::new();
    ///
    /// let result = store
    ///     .limit()
    ///     .at_most("api_call", 10, TimeUnit::Minutes)
    ///     .at_most("api_call", 100, TimeUnit::Hours)
    ///     .check_and_record("api_call");
    ///
    /// assert!(result.is_ok());
    /// ```
    pub fn limit(&self) -> Limiter {
        Limiter::new(self.inner.clone())
    }

    /// Reconciles the delta between two events and records to balance them.
    ///
    /// This method calculates the all-time delta (positive - negative) and:
    /// - If delta > 0: records delta to the negative event
    /// - If delta < 0: records |delta| to the positive event
    /// - If delta == 0: does nothing
    ///
    /// This is useful for tracking net changes like credits/debits or joins/leaves.
    ///
    /// # Examples
    ///
    /// ```
    /// use tiny_counter::EventStore;
    ///
    /// let store = EventStore::new();
    /// store.record_count("credits", 100);
    /// store.record_count("debits", 30);
    ///
    /// // Balance adds 70 to debits to equalize
    /// store.balance_delta("credits", "debits").unwrap();
    ///
    /// let delta = store.query_delta("credits", "debits").ever().sum();
    /// assert_eq!(delta, 0);
    /// ```
    pub fn balance_delta(&self, positive: impl EventId, negative: impl EventId) -> Result<()> {
        let positive_str = positive.as_ref();
        let negative_str = negative.as_ref();

        // Query all-time delta
        let delta = self.query_delta(positive_str, negative_str).ever().sum();

        if delta > 0 {
            // Positive delta: record to negative event to balance
            self.record_count(negative_str, delta.min(u32::MAX as i64) as u32);
        } else if delta < 0 {
            // Negative delta: record |delta| to positive event to balance
            self.record_count(positive_str, (-delta).min(u32::MAX as i64) as u32);
        }
        // delta == 0: no-op

        Ok(())
    }

    /// Persists only dirty (modified) events to storage.
    ///
    /// This method performs synchronous I/O to save modified events to the configured storage
    /// backend. It's automatically called by the Drop implementation when the EventStore is
    /// dropped, but **explicit calls are recommended** in async contexts and for error handling.
    ///
    /// Returns an error if no storage is configured or if serialization/storage fails.
    /// Requires a formatter - either a built-in formatter (serde-bincode, serde-json) or a custom implementation.
    ///
    /// # Best Practices
    ///
    /// - **Async contexts**: Call `persist()` explicitly before the store goes out of scope
    ///   to avoid blocking the executor during Drop
    /// - **Error handling**: Explicit calls allow you to handle persistence errors, while
    ///   Drop silently ignores errors
    /// - **With auto-persist**: Still call `persist()` before drop to ensure the final batch
    ///   of events is saved and to catch any errors
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "serde")]
    /// # {
    /// use tiny_counter::{EventStore, storage::MemoryStorage};
    ///
    /// let store = EventStore::builder()
    ///     .with_storage(MemoryStorage::new())
    ///     .build()
    ///     .unwrap();
    ///
    /// store.record("event");
    /// assert!(store.is_dirty());
    ///
    /// // Explicit persist with error handling
    /// store.persist().expect("Failed to persist");
    /// assert!(!store.is_dirty());
    /// # }
    /// ```
    pub fn persist(&self) -> Result<()> {
        self.persist_if_dirty(false)
    }

    /// Explicitly persist and close the event store.
    ///
    /// This is a convenience method that calls [`persist()`](Self::persist) and then consumes
    /// the EventStore. It's semantically clearer than calling `persist()` + `drop()` and makes
    /// the intent of cleanup explicit.
    ///
    /// **Preferred over relying on Drop** in production code, especially in:
    /// - Async contexts where Drop would block the executor
    /// - Code where error handling is important
    /// - Shutdown sequences where explicit cleanup is desired
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "serde")]
    /// # {
    /// use tiny_counter::{EventStore, storage::MemoryStorage};
    ///
    /// let store = EventStore::builder()
    ///     .with_storage(MemoryStorage::new())
    ///     .build()
    ///     .unwrap();
    ///
    /// store.record("event");
    ///
    /// // Explicit cleanup with error handling
    /// store.close().expect("Failed to close store");
    /// // Store is now consumed and dropped
    /// # }
    /// ```
    pub fn close(self) -> Result<()> {
        self.persist()
        // self is dropped here after successful persist
    }

    fn persist_if_dirty(&self, force_dirty: bool) -> Result<()> {
        let storage = self
            .inner
            .storage
            .as_ref()
            .ok_or_else(|| Error::Storage("No storage configured".to_string()))?;

        let mut storage = storage.lock().unwrap();

        let formatter = self
            .inner
            .formatter
            .as_ref()
            .ok_or_else(|| Error::Serialization("No formatter configured".to_string()))?;

        // Begin transaction for atomic multi-event persistence
        storage.begin_transaction()?;

        let longest_time_unit = self.inner.config.specified_time_unit(TimeUnit::Ever);
        let persist_result = (|| {
            for entry in self.inner.events.iter() {
                let counter = entry.value();
                let event_id = entry.key();
                self.persist_counter(
                    &mut **storage,
                    &**formatter,
                    event_id,
                    counter,
                    longest_time_unit,
                    force_dirty,
                )?;
            }
            Ok(())
        })();

        // Commit or rollback based on result
        match persist_result {
            Ok(()) => {
                storage.commit_transaction()?;
                Ok(())
            }
            Err(e) => {
                // Attempt rollback, but return original error
                let _ = storage.rollback_transaction();
                Err(e)
            }
        }
    }

    /// Persists all events to storage, regardless of dirty status.
    ///
    /// Returns an error if no storage is configured or if serialization/storage fails.
    pub fn persist_all(&self) -> Result<()> {
        self.persist_if_dirty(true)
    }

    /// Clears the dirty flag on all events without persisting.
    ///
    /// Use this when you want to mark all events as clean without saving.
    pub fn reset_dirty(&self) {
        for entry in self.inner.events.iter() {
            let mut counter = entry.value().lock().unwrap();
            counter.reset_dirty();
        }
    }

    /// Compacts storage by loading all events, advancing to current time,
    /// persisting back to storage, and clearing memory.
    ///
    /// This method:
    /// - Loads all events from storage into memory (triggers convert_if_needed)
    /// - Advances all counters to current time
    /// - Persists all counters (saves non-empty, deletes empty)
    /// - Clears in-memory cache (events will be lazy-loaded as needed)
    ///
    /// # Thread Safety
    ///
    /// Requires `&mut self` for exclusive access. When sharing EventStore across
    /// threads, wrap in `Arc<Mutex<EventStore>>` to safely call this method.
    pub fn compact(&mut self) -> Result<()> {
        // Verify storage is configured
        if self.inner.storage.is_none() {
            return Err(Error::Storage("No storage configured".to_string()));
        }

        // List all keys from storage
        let keys = {
            let storage = self.inner.storage.as_ref().unwrap();
            let storage_guard = storage.lock().unwrap();
            storage_guard.list_keys()?
        };

        let now = self.inner.clock.now();

        // Load all events into memory and advance them
        for key in keys {
            // get_counter_for_query loads from storage if not already in memory
            // and automatically triggers convert_if_needed
            if let Some(counter) = self.inner.get_counter_for_query(&key) {
                let mut counter_guard = counter.lock().unwrap();
                counter_guard.advance_if_needed(now);
            }
        }

        // Persist all in-memory counters (handles save/delete logic)
        self.persist_all()?;

        // Clear in-memory cache to free memory
        self.inner.events.clear();

        Ok(())
    }

    /// Returns the approximate memory usage in bytes for all tracked events.
    ///
    /// This includes the bucket storage for all intervals across all events.
    pub fn memory_usage(&self) -> usize {
        let mut total: usize = 0;
        for entry in self.inner.events.iter() {
            let counter = entry.value().lock().unwrap();
            // Calculate memory for each interval in the counter
            total = total.saturating_add(counter.memory_usage())
        }
        total
    }

    /// Returns a list of tracked time units and their bucket counts.
    ///
    /// This reflects the default configuration used for new events.
    pub fn tracked_intervals(&self) -> Vec<(TimeUnit, usize)> {
        self.inner
            .config
            .as_vec()
            .iter()
            .map(|config| (config.time_unit(), config.bucket_count()))
            .collect()
    }

    /// Exports all event counters.
    ///
    /// Returns a HashMap mapping event IDs to their SingleEventCounter instances.
    /// Useful for serialization, backup, or multi-device sync.
    ///
    /// This method queries storage to find all event_ids on disk and loads any
    /// counters that aren't already in memory, ensuring a complete snapshot.
    pub fn export_all(&self) -> Result<HashMap<String, SingleEventCounter>> {
        let mut result = HashMap::new();

        // Collect all event_ids from both memory and storage
        let mut all_event_ids: HashSet<String> = HashSet::new();

        // Add all event_ids from memory
        for entry in self.inner.events.iter() {
            all_event_ids.insert(entry.key().clone());
        }

        // Add all event_ids from storage (if storage is configured)
        if let Some(storage) = &self.inner.storage {
            let storage_guard = storage.lock().unwrap();
            let storage_keys = storage_guard.list_keys()?;
            drop(storage_guard); // Release lock before loading counters
            for key in storage_keys {
                all_event_ids.insert(key);
            }
        }

        // Load each counter (from memory or storage) and add to result
        for event_id in all_event_ids {
            if let Some(counter_arc) = self.inner.get_counter_for_query(&event_id) {
                let counter = counter_arc.lock().unwrap();
                result.insert(event_id, counter.clone());
            }
        }

        Ok(result)
    }

    /// Exports only dirty (modified) event counters.
    ///
    /// Returns a HashMap of event IDs to SingleEventCounter for events that have
    /// been modified since the last persist or reset_dirty.
    pub fn export_dirty(&self) -> Result<HashMap<String, SingleEventCounter>> {
        let mut result = HashMap::new();
        for entry in self.inner.events.iter() {
            let counter = entry.value().lock().unwrap();
            if counter.is_dirty() {
                result.insert(entry.key().clone(), counter.clone());
            }
        }
        Ok(result)
    }

    /// Imports a single event counter, creating or replacing it.
    ///
    /// If the event already exists, it is replaced entirely (not merged).
    /// Marks the event as dirty after import.
    pub fn import_event(&self, event_id: impl EventId, counter: SingleEventCounter) -> Result<()> {
        let event_id_str = event_id.as_ref();
        let mut counter = convert_if_needed(counter, &self.inner.config);
        counter.mark_dirty();
        self.inner
            .events
            .insert(event_id_str.to_string(), Arc::new(Mutex::new(counter)));
        Ok(())
    }

    /// Imports multiple event counters, creating or replacing them.
    ///
    /// This is a batch version of import_event.
    pub fn import_all(&self, events: HashMap<String, SingleEventCounter>) -> Result<()> {
        for (event_id, counter) in events {
            self.import_event(event_id, counter)?;
        }
        Ok(())
    }

    /// Merges a single event counter into the store.
    ///
    /// If the event doesn't exist, creates it with the merged counter.
    /// If it exists, merges the counts using SingleEventCounter::merge.
    /// Marks the event as dirty after merge.
    pub fn merge_event(&self, event_id: impl EventId, counter: SingleEventCounter) -> Result<()> {
        let event_id_str = event_id.as_ref();
        if let Some(existing_entry) = self.inner.get_counter_for_query(event_id_str) {
            let mut existing = existing_entry.lock().unwrap();
            existing.merge(counter)?;
            existing.mark_dirty();
        } else {
            let mut new_counter = counter;
            new_counter.mark_dirty();
            self.inner
                .events
                .insert(event_id_str.to_string(), Arc::new(Mutex::new(new_counter)));
        }

        Ok(())
    }

    /// Merges multiple event counters into the store.
    ///
    /// This is a batch version of merge_event.
    pub fn merge_all(&self, events: HashMap<String, SingleEventCounter>) -> Result<()> {
        for (event_id, counter) in events {
            self.merge_event(event_id, counter)?;
        }
        Ok(())
    }

    pub fn merge(&self, other: Self) -> Result<()> {
        let events = other.export_all()?;
        self.merge_all(events)
    }

    /// Spawns a background task for automatic persistence.
    ///
    /// This method creates a tokio task that periodically checks if the store
    /// is dirty and persists if needed. The task runs until aborted.
    ///
    /// Returns a JoinHandle that can be used to abort the task.
    #[cfg(feature = "tokio")]
    #[allow(dead_code)] // Will be used by builder in ev-5e4
    pub(crate) fn spawn_auto_persist(
        inner: Arc<EventStoreInner>,
        interval: chrono::Duration,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            // Convert to std::time::Duration at tokio boundary
            let std_interval = interval
                .to_std()
                .expect("auto_persist interval must be positive");

            // Create a temporary BaseEventStore to call methods
            let store = EventStore {
                inner: inner.clone(),
                auto_persist_handle: None,
            };

            loop {
                tokio::time::sleep(std_interval).await;

                if store.is_dirty() {
                    if let Err(e) = store.persist() {
                        eprintln!("Auto-persist failed: {}", e);
                    }
                }
            }
        })
    }

    fn persist_counter(
        &self,
        storage: &mut dyn Storage,
        formatter: &dyn Formatter,
        event_id: &str,
        counter: &Mutex<SingleEventCounter>,
        longest_time_unit: TimeUnit,
        force_dirty: bool,
    ) -> Result<()> {
        if let Some(data) = {
            // We get the lock, so we can get the data and serialize it.
            let mut counter = counter.lock().unwrap();
            if force_dirty || counter.is_dirty() {
                // We optimistically mark the event as not dirty
                if !counter.is_empty(longest_time_unit) {
                    let data = formatter.serialize(&counter)?;
                    counter.reset_dirty();
                    Some(data)
                } else {
                    counter.reset_dirty();
                    None
                }
            } else {
                return Ok(());
            }
            // We release the lock here, with the data serialized, and marked as not dirty.
        } {
            // We're now out of the lock, and can let the storage take as long as it wants to
            // store the event.
            // If another thread now records another event, then dirty will be set to true,
            // but it will have missed this save, and will be done next time. That's ok.
            storage.save(event_id, data)
        } else {
            storage.delete(event_id)
        }
        .map_err(|_e| {
            // If storage errors, then we should re-mark the event as dirty.
            // Dirty may already be set to true (if another thread has recorded an event),
            // but we won't lose that just because we serialized.
            // We'll just try again later.
            let mut counter = counter.lock().unwrap();
            counter.mark_dirty();
            _e
        })
    }
}

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

/// Drop implementation for EventStore.
///
/// **IMPORTANT**: This implementation performs **synchronous I/O** which can block the
/// current thread. See the struct-level documentation for [`EventStore`] for best practices.
///
/// When an EventStore is dropped:
/// 1. The background auto-persist task is aborted (if configured)
/// 2. A final `persist()` is attempted if the store has dirty (unsaved) events
///
/// # Behavior Details
///
/// - **Blocking I/O**: The persist operation blocks until complete. In async contexts,
///   this blocks the executor thread.
/// - **Silent errors**: Any errors during persist are ignored (logged to stderr in debug
///   builds). Errors cannot be returned from Drop.
/// - **Panic safety**: Drop runs during panic unwinding, so errors are swallowed to avoid
///   double-panics.
///
/// # Recommendations
///
/// Call [`persist()`](Self::persist) explicitly before dropping to:
/// - Avoid blocking in async code
/// - Handle errors properly
/// - Make cleanup behavior explicit
impl Drop for EventStore {
    fn drop(&mut self) {
        // Abort the background auto-persist task if present.
        // This ensures the task doesn't outlive the EventStore and prevents
        // potential use-after-free of the inner Arc<EventStoreInner>.
        #[cfg(feature = "tokio")]
        if let Some(handle) = &self.auto_persist_handle {
            handle.abort();
        }

        // Perform a final synchronous persist if there are unsaved changes.
        // This is a best-effort attempt - errors are silently ignored because:
        // 1. Drop cannot return an error
        // 2. Drop may be called during panic unwinding (can't panic in Drop)
        // 3. For critical data, users should call persist() explicitly before drop
        if self.is_dirty() {
            let _ = self.persist();
            // Errors are intentionally ignored. In production code, call persist()
            // explicitly before dropping the store to handle errors properly.
        }
    }
}

/// A generic event id trait for type-safe event identification.
///
/// This trait enables using custom types (especially enums) as event identifiers,
/// providing compile-time guarantees about which events exist in your system.
///
/// # Benefits of Type-Safe Event IDs
///
/// Using enums instead of raw strings provides several advantages:
///
/// 1. **Bounded event set**: Compiler enforces a fixed set of events, preventing typos
/// 2. **Memory safety**: Enum variants guarantee a bounded number of event counters
/// 3. **Refactoring safety**: Renaming events shows all usage sites at compile time
/// 4. **Documentation**: Enum definition serves as single source of truth for all events
///
/// # Memory Implications
///
/// Each unique event ID creates a new counter in memory (~2KB per event with default config).
/// Using patterns like `format!("user:{}:event", user_id)` creates unbounded event IDs
/// that grow with your user base, defeating the fixed-memory guarantee.
///
/// **Safe pattern** (bounded events):
/// ```rust
/// use tiny_counter::{EventStore, EventId};
///
/// #[derive(Debug)]
/// enum AppEvent {
///     UserLogin,
///     UserLogout,
///     ApiCall,
/// }
///
/// impl AsRef<str> for AppEvent {
///     fn as_ref(&self) -> &str {
///         match self {
///             AppEvent::UserLogin => "user_login",
///             AppEvent::UserLogout => "user_logout",
///             AppEvent::ApiCall => "api_call",
///         }
///     }
/// }
///
/// impl EventId for AppEvent {}
///
/// let store = EventStore::new();
/// store.record(AppEvent::UserLogin);  // Type-safe, bounded memory
/// ```
///
/// **Unsafe pattern** (unbounded events, avoid this):
/// ```rust,no_run
/// # use tiny_counter::EventStore;
/// let store = EventStore::new();
/// for user_id in 0..1_000_000 {
///     // WARNING: Creates 1M separate counters, ~2GB memory!
///     store.record(format!("user:{}:login", user_id));
/// }
/// ```
///
/// For per-user tracking, use a single event ID and separate EventStore instances per user,
/// or aggregate at the application level rather than in the event store.
pub trait EventId: AsRef<str> {}

impl EventId for str {}
impl EventId for String {}
impl EventId for &str {}
impl EventId for &String {}

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

    use chrono::TimeZone;

    #[test]
    fn test_new_creates_empty_store_with_default_intervals() {
        let store = EventStore::new();
        assert!(!store.is_dirty());

        // Verify default intervals work by querying each time unit
        let intervals = store.tracked_intervals();
        assert_eq!(intervals.len(), 6);
    }

    #[test]
    fn test_default_configs_have_256_buckets_total() {
        let config = EventCounterConfig::default();
        let total: usize = config.as_vec().iter().map(|c| c.bucket_count()).sum();
        assert_eq!(total, 256);
    }

    #[test]
    fn test_record_creates_counter_on_demand() {
        let store = EventStore::new();
        store.record("test_event");

        // Verify counter was created by querying the event
        let sum = store.query("test_event").last_days(1).sum();
        assert_eq!(sum, Some(1));
    }

    #[test]
    fn test_record_count_with_count_5() {
        let store = EventStore::new();
        store.record_count("test_event", 5);

        let sum = store.query("test_event").last_days(1).sum();
        assert_eq!(sum, Some(5));
    }

    #[test]
    fn test_record_at_with_past_time() {
        let store = EventStore::new();
        let now = Utc::now();
        let past = now - Duration::days(2);

        store.record_at("test_event", past).unwrap();

        // Verify event was recorded
        assert!(store.is_dirty());
        let sum = store.query("test_event").last_days(7).sum();
        assert_eq!(sum, Some(1));
    }

    #[test]
    fn test_record_ago_with_duration() {
        let store = EventStore::new();
        store.record_ago("test_event", Duration::hours(3));

        // Verify event was recorded
        assert!(store.is_dirty());
        let sum = store.query("test_event").last_hours(24).sum();
        assert_eq!(sum, Some(1));
    }

    #[test]
    fn test_query_returns_query_builder() {
        let store = EventStore::new();
        store.record("test_event");

        let query = store.query("test_event");
        let sum = query.last_days(1).sum();
        assert_eq!(sum, Some(1));
    }

    #[test]
    fn test_query_nonexistent_event_returns_none() {
        let store = EventStore::new();
        let query = store.query("nonexistent");
        let sum = query.last_days(1).sum();
        assert_eq!(sum, None);
    }

    #[test]
    fn test_query_many() {
        let store = EventStore::new();
        store.record_count("event1", 5);
        store.record_count("event2", 3);

        let event_ids = &["event1", "event2"];
        let query = store.query_many(event_ids);
        let sum = query.last_days(1).sum();
        assert_eq!(sum, Some(8));
    }

    #[test]
    fn test_query_ratio() {
        let store = EventStore::new();
        store.record_count("numerator", 10);
        store.record_count("denominator", 5);

        let ratio = store.query_ratio("numerator", "denominator").last_days(1);
        assert_eq!(ratio, Some(2.0));
    }

    #[test]
    fn test_query_delta() {
        let store = EventStore::new();
        store.record_count("positive", 10);
        store.record_count("negative", 3);

        let delta = store.query_delta("positive", "negative").last_days(1).sum();
        assert_eq!(delta, 7);
    }

    #[test]
    fn test_dirty_tracking_starts_clean() {
        let store = EventStore::new();
        assert!(!store.is_dirty());
    }

    #[test]
    fn test_dirty_tracking_becomes_dirty_after_record() {
        let store = EventStore::new();
        store.record("test_event");
        assert!(store.is_dirty());
    }

    #[test]
    fn test_query_returns_none_for_nonexistent_event() {
        let store = EventStore::new();
        let result = store.query("nonexistent").last_days(1).sum();
        assert_eq!(result, None);
    }

    #[test]
    fn test_integration_record_multiple_events_query_each() {
        let store = EventStore::new();

        // Record different events
        store.record_count("login", 5);
        store.record_count("logout", 3);
        store.record_count("error", 1);

        // Query each
        let login_sum = store.query("login").last_days(1).sum();
        let logout_sum = store.query("logout").last_days(1).sum();
        let error_sum = store.query("error").last_days(1).sum();

        assert_eq!(login_sum, Some(5));
        assert_eq!(logout_sum, Some(3));
        assert_eq!(error_sum, Some(1));
    }

    #[test]
    fn test_record_creates_counter_with_default_config() {
        let store = EventStore::new();
        store.record("test_event");

        // Verify counter has all 4 default intervals by querying each
        assert!(store.query("test_event").last_minutes(1).sum().is_some());
        assert!(store.query("test_event").last_hours(1).sum().is_some());
        assert!(store.query("test_event").last_days(1).sum().is_some());
        assert!(store.query("test_event").last_months(1).sum().is_some());
    }

    #[test]
    fn test_default_trait() {
        let store = EventStore::default();
        assert!(!store.is_dirty());

        // Verify default intervals are configured
        let intervals = store.tracked_intervals();
        assert_eq!(intervals.len(), 6);
    }

    #[test]
    fn test_default_config_includes_all_six_time_units() {
        let store = EventStore::new();
        store.record("test_event");

        // Verify that all 6 time units are in the default configuration
        let intervals = store.tracked_intervals();
        let time_units: Vec<TimeUnit> = intervals.iter().map(|(unit, _)| *unit).collect();

        assert!(time_units.contains(&TimeUnit::Minutes));
        assert!(time_units.contains(&TimeUnit::Hours));
        assert!(time_units.contains(&TimeUnit::Days));
        assert!(time_units.contains(&TimeUnit::Weeks));
        assert!(time_units.contains(&TimeUnit::Months));
        assert!(time_units.contains(&TimeUnit::Years));

        // Verify that all 6 default time units work
        assert!(store.query("test_event").last_minutes(1).sum().is_some());
        assert!(store.query("test_event").last_hours(1).sum().is_some());
        assert!(store.query("test_event").last_days(1).sum().is_some());
        assert!(store.query("test_event").last_weeks(1).sum().is_some());
        assert!(store.query("test_event").last_months(1).sum().is_some());
        assert!(store.query("test_event").last_years(1).sum().is_some());
    }

    #[test]
    fn test_record_at_with_time_before_creation() {
        let store = EventStore::new();
        let now = Utc.with_ymd_and_hms(2025, 1, 10, 12, 0, 0).unwrap();

        // Mock time by using TestClock would be better, but for this test
        // we'll just verify it doesn't panic
        let past = now - Duration::days(2);
        let result = store.record_at("test_event", past);

        // Should succeed (not a future event)
        assert!(result.is_ok());
    }

    #[test]
    fn test_multiple_records_to_same_event() {
        let store = EventStore::new();
        store.record("test_event");
        store.record("test_event");
        store.record("test_event");

        let sum = store.query("test_event").last_days(1).sum();
        assert_eq!(sum, Some(3));
    }

    #[test]
    fn test_record_count_ago() {
        let store = EventStore::new();
        store.record_count_ago("test_event", 5, Duration::hours(2));

        // Should have recorded in the past
        let sum = store.query("test_event").last_hours(24).sum();
        assert_eq!(sum, Some(5));
    }

    #[test]
    fn test_record_ago_outside_tracking_window_silently_drops() {
        let store = EventStore::new();

        // Default store tracks up to 4 years in the past
        // Record an event 10 years ago - should be silently dropped
        store.record_ago("ancient_event", Duration::days(365 * 10));

        // Query returns Some(0) because event counter exists but event was dropped
        let sum = store.query("ancient_event").ever().sum();
        assert_eq!(sum, Some(0));
    }

    #[test]
    fn test_record_count_ago_outside_tracking_window_silently_drops() {
        let store = EventStore::new();

        // Default store tracks up to 4 years in the past
        // Record events 10 years ago - should be silently dropped
        store.record_count_ago("ancient_event", 100, Duration::days(365 * 10));

        // Query returns Some(0) because event counter exists but events were dropped
        let sum = store.query("ancient_event").ever().sum();
        assert_eq!(sum, Some(0));
    }

    #[test]
    fn test_balance_delta_positive() {
        let store = EventStore::new();
        store.record_count("credits", 10);
        store.record_count("debits", 3);

        // Delta = 10 - 3 = 7 (positive), so record 7 to debits
        store.balance_delta("credits", "debits").unwrap();

        let credits_sum = store.query("credits").ever().sum();
        let debits_sum = store.query("debits").ever().sum();

        assert_eq!(credits_sum, Some(10));
        assert_eq!(debits_sum, Some(10)); // 3 + 7 = 10
    }

    #[test]
    fn test_balance_delta_negative() {
        let store = EventStore::new();
        store.record_count("credits", 3);
        store.record_count("debits", 10);

        // Delta = 3 - 10 = -7 (negative), so record 7 to credits
        store.balance_delta("credits", "debits").unwrap();

        let credits_sum = store.query("credits").ever().sum();
        let debits_sum = store.query("debits").ever().sum();

        assert_eq!(credits_sum, Some(10)); // 3 + 7 = 10
        assert_eq!(debits_sum, Some(10));
    }

    #[test]
    fn test_balance_delta_zero() {
        let store = EventStore::new();
        store.record_count("credits", 10);
        store.record_count("debits", 10);

        // Delta = 10 - 10 = 0, so no-op
        store.balance_delta("credits", "debits").unwrap();

        let credits_sum = store.query("credits").ever().sum();
        let debits_sum = store.query("debits").ever().sum();

        assert_eq!(credits_sum, Some(10));
        assert_eq!(debits_sum, Some(10));
    }

    #[test]
    fn test_reset_dirty() {
        let store = EventStore::new();
        store.record("test_event");
        assert!(store.is_dirty());

        store.reset_dirty();
        assert!(!store.is_dirty());
    }

    #[test]
    fn test_memory_usage_empty_store() {
        let store = EventStore::new();
        assert_eq!(store.memory_usage(), 0);
    }

    #[test]
    fn test_memory_usage_with_events() {
        let store = EventStore::new();
        store.record("test_event");

        // Memory usage should be greater than 0
        let usage = store.memory_usage();
        assert!(usage > 0);
    }

    #[test]
    fn test_tracked_intervals_returns_default_config() {
        let store = EventStore::new();
        let intervals = store.tracked_intervals();

        assert_eq!(intervals.len(), 6);
        assert!(intervals.contains(&(TimeUnit::Minutes, 60)));
        assert!(intervals.contains(&(TimeUnit::Hours, 72)));
        assert!(intervals.contains(&(TimeUnit::Days, 56)));
        assert!(intervals.contains(&(TimeUnit::Weeks, 52)));
        assert!(intervals.contains(&(TimeUnit::Months, 12)));
        assert!(intervals.contains(&(TimeUnit::Years, 4)));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_without_storage_returns_error() {
        let store = EventStore::new();
        store.record("test_event");

        let result = store.persist();
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::Storage(_) => (),
            _ => panic!("Expected Storage error"),
        }
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_all_without_storage_returns_error() {
        let store = EventStore::new();
        store.record("test_event");

        let result = store.persist_all();
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::Storage(_) => (),
            _ => panic!("Expected Storage error"),
        }
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_with_storage() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let store = EventStoreBuilder::new()
            .with_storage(MemoryStorage::new())
            .build()
            .unwrap();

        store.record("event1");
        store.record("event2");
        assert!(store.is_dirty());

        let result = store.persist();
        assert!(result.is_ok());
        assert!(!store.is_dirty());
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_close_persists_and_consumes() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let storage = MemoryStorage::new();
        let store = EventStoreBuilder::new()
            .with_storage(storage)
            .build()
            .unwrap();

        store.record("event1");
        store.record("event2");
        assert!(store.is_dirty());

        // close() should persist and consume the store
        let result = store.close();
        assert!(result.is_ok());
        // store is now consumed and cannot be used
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_close_returns_error_on_persist_failure() {
        use crate::EventStoreBuilder;

        // Create store without storage - persist will fail
        let store = EventStoreBuilder::new().build().unwrap();

        store.record("event1");
        assert!(store.is_dirty());

        // close() should return error when persist fails
        let result = store.close();
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("No storage configured"));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_only_dirty_events() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let store = EventStoreBuilder::new()
            .with_storage(MemoryStorage::new())
            .build()
            .unwrap();

        // Record and persist event1
        store.record("event1");
        store.persist().unwrap();
        assert!(!store.is_dirty());

        // Record event2
        store.record("event2");
        assert!(store.is_dirty());

        // Persist should only write event2
        store.persist().unwrap();
        assert!(!store.is_dirty());
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_all() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let store = EventStoreBuilder::new()
            .with_storage(MemoryStorage::new())
            .build()
            .unwrap();

        store.record("event1");
        store.record("event2");
        store.record("event3");

        let result = store.persist_all();
        assert!(result.is_ok());
        assert!(!store.is_dirty());
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_serialization_roundtrip() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let store = EventStoreBuilder::new()
            .with_storage(MemoryStorage::new())
            .build()
            .unwrap();

        store.record_count("test_event", 42);
        store.persist().unwrap();

        // Verify the counter has the right value
        let sum = store.query("test_event").last_days(1).sum();
        assert_eq!(sum, Some(42));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_clears_dirty_events() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let store = EventStoreBuilder::new()
            .with_storage(MemoryStorage::new())
            .build()
            .unwrap();

        store.record("event1");
        store.record("event2");
        assert!(store.is_dirty());

        store.persist().unwrap();
        assert!(!store.is_dirty());
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_compact_advances_and_saves() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let mut store = EventStoreBuilder::new()
            .track_days(7)
            .with_storage(MemoryStorage::new())
            .build()
            .unwrap();

        // Record events
        store.record("event1");
        store.record("event2");

        // Persist both
        store.persist_all().unwrap();
        assert!(!store.is_dirty());

        // Compact should re-save all events after advancing
        store.compact().unwrap();

        // Events should still be queryable
        assert_eq!(store.query("event1").last_days(7).sum(), Some(1));
        assert_eq!(store.query("event2").last_days(7).sum(), Some(1));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_compact_while_recording() {
        use crate::storage::MemoryStorage;
        use crate::{EventCounterConfig, SystemClock};
        use std::thread;

        // Create store with from_parts to avoid Clone issue
        let base_store = EventStore::from_parts(
            SystemClock::new(),
            Some(Box::new(MemoryStorage::new())),
            Some(Arc::new(crate::formatter::BincodeFormat)),
            EventCounterConfig::default(),
        );

        // Wrap in Arc+Mutex since compact needs &mut self
        let store = Arc::new(Mutex::new(base_store));
        let store_writer = Arc::clone(&store);
        let store_compactor = Arc::clone(&store);

        // Writer thread - continuously record
        let writer = thread::spawn(move || {
            for _ in 0..100 {
                let store = store_writer.lock().unwrap();
                store.record("compact_event");
                drop(store);
                thread::sleep(std::time::Duration::from_micros(10));
            }
        });

        // Compactor thread - compact multiple times
        let compactor = thread::spawn(move || {
            for _ in 0..10 {
                thread::sleep(std::time::Duration::from_micros(50));
                let mut store = store_compactor.lock().unwrap();
                let _ = store.compact();
            }
        });

        writer.join().unwrap();
        compactor.join().unwrap();

        // Verify all 100 events were recorded (no data loss)
        let store = store.lock().unwrap();
        let sum = store.query("compact_event").last_days(7).sum();
        assert_eq!(sum, Some(100));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_compact_without_storage_returns_error() {
        let mut store = EventStore::new();
        store.record("event1");

        let result = store.compact();
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("No storage configured"));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_memory_usage_calculation() {
        let store = EventStore::new();
        store.record("event1");
        store.record("event2");

        let usage = store.memory_usage();
        // Each event has 6 intervals (default config)
        // Each interval has at least 1 bucket (8 bytes)
        // So minimum is 2 events * 6 intervals * 1 bucket * 8 bytes = 96 bytes
        assert!(usage >= 96);
    }

    #[test]
    fn test_export_all_returns_all_counters() {
        let store = EventStore::new();
        store.record("event1");
        store.record("event2");
        store.record("event3");

        let exported = store.export_all().unwrap();
        assert_eq!(exported.len(), 3);
        assert!(exported.contains_key("event1"));
        assert!(exported.contains_key("event2"));
        assert!(exported.contains_key("event3"));
    }

    #[test]
    fn test_export_dirty_returns_only_dirty_counters() {
        let store = EventStore::new();
        store.record("event1");
        store.record("event2");
        store.reset_dirty();

        // Now record event3 (should be dirty)
        store.record("event3");

        let exported = store.export_dirty().unwrap();
        assert_eq!(exported.len(), 1);
        assert!(exported.contains_key("event3"));
        assert!(!exported.contains_key("event1"));
        assert!(!exported.contains_key("event2"));
    }

    #[test]
    fn test_export_all_returns_empty_map_for_empty_store() {
        let store = EventStore::new();
        let exported = store.export_all().unwrap();
        assert_eq!(exported.len(), 0);
    }

    #[test]
    fn test_export_dirty_returns_empty_map_when_nothing_dirty() {
        let store = EventStore::new();
        store.record("event1");
        store.reset_dirty();

        let exported = store.export_dirty().unwrap();
        assert_eq!(exported.len(), 0);
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_export_all_includes_disk_only_counters() {
        use crate::storage::MemoryStorage;
        use crate::EventStoreBuilder;

        let storage = MemoryStorage::new();
        let store = EventStoreBuilder::new()
            .with_storage(storage)
            .build()
            .unwrap();

        // Record and persist event1
        store.record_count("event1", 10);
        store.persist().unwrap();

        // Record event2 (stays in memory)
        store.record_count("event2", 20);

        // Clear memory by dropping store and creating new one with same storage
        let storage2 = {
            let storage_arc = store.inner.storage.as_ref().unwrap().clone();
            let storage_guard = storage_arc.lock().unwrap();
            // Create a new MemoryStorage and copy data from old one
            let mut new_storage = MemoryStorage::new();
            for key in storage_guard.list_keys().unwrap() {
                let data = storage_guard.load(&key).unwrap().unwrap();
                new_storage.save(&key, data).unwrap();
            }
            new_storage
        };

        let store2 = EventStoreBuilder::new()
            .with_storage(storage2)
            .build()
            .unwrap();

        // Record event3 (only in memory)
        store2.record_count("event3", 30);

        // export_all should include:
        // - event1 (from storage, not in memory)
        // - event3 (from memory, not in storage)
        let exported = store2.export_all().unwrap();
        assert_eq!(exported.len(), 2);
        assert!(exported.contains_key("event1"));
        assert!(exported.contains_key("event3"));

        // Verify the counts are correct
        assert_eq!(store2.query("event1").last_days(1).sum(), Some(10));
        assert_eq!(store2.query("event3").last_days(1).sum(), Some(30));
    }

    #[test]
    fn test_import_event_creates_new_counter() {
        let store1 = EventStore::new();
        store1.record_count("event1", 42);

        let exported = store1.export_all().unwrap();
        let counter = exported.get("event1").unwrap().clone();

        let store2 = EventStore::new();
        store2.import_event("event1", counter).unwrap();

        let sum = store2.query("event1").last_days(1).sum();
        assert_eq!(sum, Some(42));
    }

    #[test]
    fn test_import_event_overwrites_existing() {
        let store = EventStore::new();
        store.record_count("event1", 10);

        let store2 = EventStore::new();
        store2.record_count("event1", 42);

        let exported = store2.export_all().unwrap();
        let counter = exported.get("event1").unwrap().clone();

        store.import_event("event1", counter).unwrap();

        let sum = store.query("event1").last_days(1).sum();
        // Should have overwritten 10 with 42
        assert_eq!(sum, Some(42));
    }

    #[test]
    fn test_import_all_batch_imports() {
        let store1 = EventStore::new();
        store1.record_count("event1", 10);
        store1.record_count("event2", 20);
        store1.record_count("event3", 30);

        let exported = store1.export_all().unwrap();

        let store2 = EventStore::new();
        store2.import_all(exported).unwrap();

        assert_eq!(store2.query("event1").last_days(1).sum(), Some(10));
        assert_eq!(store2.query("event2").last_days(1).sum(), Some(20));
        assert_eq!(store2.query("event3").last_days(1).sum(), Some(30));
    }

    #[test]
    fn test_merge_event_combines_counts() {
        let store1 = EventStore::new();
        store1.record_count("event1", 10);

        let store2 = EventStore::new();
        store2.record_count("event1", 20);

        let exported = store2.export_all().unwrap();
        let counter = exported.get("event1").unwrap().clone();

        store1.merge_event("event1", counter).unwrap();

        let sum = store1.query("event1").last_days(1).sum();
        // Should have 10 + 20 = 30
        assert_eq!(sum, Some(30));
    }

    #[test]
    fn test_merge_event_creates_counter_if_not_exists() {
        let store1 = EventStore::new();

        let store2 = EventStore::new();
        store2.record_count("event1", 42);

        let exported = store2.export_all().unwrap();
        let counter = exported.get("event1").unwrap().clone();

        store1.merge_event("event1", counter).unwrap();

        let sum = store1.query("event1").last_days(1).sum();
        assert_eq!(sum, Some(42));
    }

    #[test]
    fn test_merge_all_combines_multiple_events() {
        let store1 = EventStore::new();
        store1.record_count("event1", 10);
        store1.record_count("event2", 20);

        let store2 = EventStore::new();
        store2.record_count("event1", 5);
        store2.record_count("event3", 30);

        let exported = store2.export_all().unwrap();
        store1.merge_all(exported).unwrap();

        assert_eq!(store1.query("event1").last_days(1).sum(), Some(15)); // 10 + 5
        assert_eq!(store1.query("event2").last_days(1).sum(), Some(20)); // unchanged
        assert_eq!(store1.query("event3").last_days(1).sum(), Some(30));
        // new
    }

    #[test]
    fn test_merge_is_commutative_at_store_level() {
        let store_a1 = EventStore::new();
        store_a1.record_count("event1", 10);

        let store_b1 = EventStore::new();
        store_b1.record_count("event1", 20);

        let store_a2 = EventStore::new();
        store_a2.record_count("event1", 10);

        let store_b2 = EventStore::new();
        store_b2.record_count("event1", 20);

        // a + b
        let b1_export = store_b1.export_all().unwrap();
        store_a1.merge_all(b1_export).unwrap();

        // b + a
        let a2_export = store_a2.export_all().unwrap();
        store_b2.merge_all(a2_export).unwrap();

        // Both should equal 30
        assert_eq!(store_a1.query("event1").last_days(1).sum(), Some(30));
        assert_eq!(store_b2.query("event1").last_days(1).sum(), Some(30));
    }

    #[test]
    fn test_merge_is_associative_at_store_level() {
        let store_a1 = EventStore::new();
        store_a1.record_count("event1", 10);

        let store_b1 = EventStore::new();
        store_b1.record_count("event1", 20);

        let store_c1 = EventStore::new();
        store_c1.record_count("event1", 30);

        let store_a2 = EventStore::new();
        store_a2.record_count("event1", 10);

        let store_b2 = EventStore::new();
        store_b2.record_count("event1", 20);

        let store_c2 = EventStore::new();
        store_c2.record_count("event1", 30);

        // (a + b) + c
        let b1_export = store_b1.export_all().unwrap();
        store_a1.merge_all(b1_export).unwrap();
        let c1_export = store_c1.export_all().unwrap();
        store_a1.merge_all(c1_export).unwrap();

        // a + (b + c)
        let c2_export = store_c2.export_all().unwrap();
        store_b2.merge_all(c2_export).unwrap();
        let b2_export = store_b2.export_all().unwrap();
        store_a2.merge_all(b2_export).unwrap();

        // Both should equal 60
        assert_eq!(store_a1.query("event1").last_days(1).sum(), Some(60));
        assert_eq!(store_a2.query("event1").last_days(1).sum(), Some(60));
    }

    // Test for ev-dcq: BaseEventStore with auto_persist_handle field
    #[cfg(all(feature = "tokio", feature = "serde"))]
    #[tokio::test]
    async fn test_base_event_store_has_auto_persist_handle_field() {
        use crate::storage::MemoryStorage;
        use std::time::Duration;

        // Create store with storage
        let mut store = EventStore::from_parts(
            SystemClock::new(),
            Some(Box::new(MemoryStorage::new())),
            #[cfg(feature = "serde-bincode")]
            Some(Arc::new(crate::formatter::BincodeFormat)),
            #[cfg(all(feature = "serde-json", not(feature = "serde-bincode")))]
            Some(Arc::new(crate::formatter::JsonFormat)),
            EventCounterConfig::default(),
        );

        // Verify field exists and is None initially
        assert!(store.auto_persist_handle.is_none());

        // Simulate setting the handle (this will fail until we add the field)
        let handle = tokio::spawn(async {
            tokio::time::sleep(Duration::from_millis(10)).await;
        });
        store.auto_persist_handle = Some(handle);
        assert!(store.auto_persist_handle.is_some());
    }

    // Test for ev-dcq: spawn_auto_persist creates background task
    #[cfg(all(feature = "tokio", feature = "serde"))]
    #[tokio::test]
    async fn test_spawn_auto_persist_creates_background_task() {
        use crate::storage::MemoryStorage;
        use chrono::Duration;

        // Create store with storage
        let store = EventStore::from_parts(
            SystemClock::new(),
            Some(Box::new(MemoryStorage::new())),
            #[cfg(feature = "serde-bincode")]
            Some(Arc::new(crate::formatter::BincodeFormat)),
            #[cfg(all(feature = "serde-json", not(feature = "serde-bincode")))]
            Some(Arc::new(crate::formatter::JsonFormat)),
            EventCounterConfig::default(),
        );

        // Record an event to make it dirty
        store.record("test");
        assert!(store.is_dirty());

        // Spawn auto-persist task
        let handle =
            EventStore::spawn_auto_persist(store.inner.clone(), Duration::milliseconds(50));

        // Wait for auto-persist to run
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Store should no longer be dirty
        assert!(!store.is_dirty());

        // Clean up
        handle.abort();
    }

    // Test for ev-dcq: Drop implementation aborts handle and does final persist
    #[cfg(all(feature = "tokio", feature = "serde"))]
    #[tokio::test]
    async fn test_drop_aborts_handle_and_persists() {
        use crate::storage::MemoryStorage;
        use std::time::Duration;

        // Create store with storage
        let mut store = EventStore::from_parts(
            SystemClock::new(),
            Some(Box::new(MemoryStorage::new())),
            #[cfg(feature = "serde-bincode")]
            Some(Arc::new(crate::formatter::BincodeFormat)),
            #[cfg(all(feature = "serde-json", not(feature = "serde-bincode")))]
            Some(Arc::new(crate::formatter::JsonFormat)),
            EventCounterConfig::default(),
        );

        // Spawn a long-running task
        let handle = tokio::spawn(async {
            loop {
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        });
        store.auto_persist_handle = Some(handle);

        // Record an event
        store.record("test");
        assert!(store.is_dirty());

        // Drop the store
        drop(store);

        // Task should be aborted (we can't directly test this, but Drop runs)
        // The test passes if Drop doesn't panic
    }

    // Test for ev-rh6: Drop persists dirty data even when auto-persist is enabled
    #[cfg(all(
        feature = "tokio",
        feature = "serde",
        feature = "serde",
        feature = "storage-fs"
    ))]
    #[tokio::test]
    async fn test_drop_persists_dirty_data_with_auto_persist() {
        use crate::storage::FilePerEvent;
        use chrono::Duration;
        use tempfile::tempdir;

        let temp_dir = tempdir().unwrap();
        let storage_path = temp_dir.path().join("events");

        // Create store with auto-persist enabled (short interval)
        let store = EventStore::builder()
            .with_storage(FilePerEvent::new(&storage_path, ".dat").unwrap())
            .auto_persist(Duration::milliseconds(100))
            .build()
            .unwrap();

        // Record first event and wait for auto-persist to run
        store.record("event1");
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;

        // Verify first event was persisted
        {
            let storage = FilePerEvent::new(&storage_path, ".dat").unwrap();
            let data = storage.load("event1").unwrap();
            assert!(
                data.is_some(),
                "First event should be persisted by auto-persist"
            );
        }

        // Record second event (makes store dirty again)
        store.record("event2");

        // Drop the store BEFORE next auto-persist cycle
        drop(store);

        // Load from storage and verify BOTH events are present
        let storage = FilePerEvent::new(&storage_path, ".dat").unwrap();
        let event1_data = storage.load("event1").unwrap();
        let event2_data = storage.load("event2").unwrap();

        assert!(
            event1_data.is_some(),
            "First event should still be in storage"
        );
        assert!(
            event2_data.is_some(),
            "Second event MUST be persisted on drop, not lost"
        );
    }
}

#[cfg(test)]
mod concurrency_tests {
    use std::thread;

    use super::*;

    #[test]
    fn test_concurrent_record_from_multiple_threads() {
        let store = Arc::new(EventStore::new());
        let mut handles = vec![];

        // Spawn 10 threads, each recording 100 events
        for _ in 0..10 {
            let store_clone = Arc::clone(&store);
            let handle = thread::spawn(move || {
                for _ in 0..100 {
                    store_clone.record("concurrent_event");
                }
            });
            handles.push(handle);
        }

        // Wait for all threads to complete
        for handle in handles {
            handle.join().unwrap();
        }

        // Should have 1000 total events
        let sum = store.query("concurrent_event").last_days(1).sum();
        assert_eq!(sum, Some(1000));
    }

    #[test]
    fn test_query_while_recording() {
        let store = Arc::new(EventStore::new());
        let store_writer = Arc::clone(&store);
        let store_reader = Arc::clone(&store);

        // Writer thread
        let writer = thread::spawn(move || {
            for i in 0..100 {
                store_writer.record_count("test_event", i);
                thread::sleep(std::time::Duration::from_micros(10));
            }
        });

        // Reader thread
        let reader = thread::spawn(move || {
            for _ in 0..50 {
                let _sum = store_reader.query("test_event").last_days(1).sum();
                thread::sleep(std::time::Duration::from_micros(20));
            }
        });

        writer.join().unwrap();
        reader.join().unwrap();

        // Should have sum of 0..100 = 4950
        let final_sum = store.query("test_event").last_days(1).sum();
        assert_eq!(final_sum, Some(4950));
    }

    #[cfg(feature = "serde-bincode")]
    #[test]
    fn test_persist_while_recording() {
        use crate::storage::MemoryStorage;

        // Create base store directly to avoid enum issues with Arc+Clone
        let base_store = EventStore::from_parts(
            SystemClock::new(),
            Some(Box::new(MemoryStorage::new())),
            #[cfg(feature = "serde")]
            {
                #[cfg(feature = "serde-bincode")]
                {
                    Some(Arc::new(crate::formatter::BincodeFormat))
                }
                #[cfg(not(feature = "serde-bincode"))]
                {
                    None
                }
            },
            EventCounterConfig::default(),
        );
        let store = Arc::new(base_store);

        let store_writer = Arc::clone(&store);
        let store_persister = Arc::clone(&store);

        // Writer thread
        let writer = thread::spawn(move || {
            for _ in 0..100 {
                store_writer.record("persist_event");
                thread::sleep(std::time::Duration::from_micros(10));
            }
        });

        // Persister thread
        let persister = thread::spawn(move || {
            for _ in 0..20 {
                let _ = store_persister.persist();
                thread::sleep(std::time::Duration::from_micros(50));
            }
        });

        writer.join().unwrap();
        persister.join().unwrap();

        // Final persist
        store.persist().unwrap();

        // Should have 100 events
        let sum = store.query("persist_event").last_days(1).sum();
        assert_eq!(sum, Some(100));
        assert!(!store.is_dirty());
    }

    #[test]
    fn test_concurrent_query_many_from_multiple_threads() {
        let store = Arc::new(EventStore::new());
        let mut handles = vec![];

        // Pre-populate events
        store.record_count("event1", 100);
        store.record_count("event2", 200);
        store.record_count("event3", 300);

        // Spawn 10 threads, each querying multiple events 50 times
        for _ in 0..10 {
            let store_clone = Arc::clone(&store);
            let handle = thread::spawn(move || {
                for _ in 0..50 {
                    let event_ids = &["event1", "event2", "event3"];
                    let sum = store_clone.query_many(event_ids).last_days(1).sum();
                    // Should always get the same sum
                    assert_eq!(sum, Some(600));
                }
            });
            handles.push(handle);
        }

        // Wait for all threads to complete
        for handle in handles {
            handle.join().unwrap();
        }
    }

    #[test]
    fn test_concurrent_query_ratio_from_multiple_threads() {
        let store = Arc::new(EventStore::new());
        let mut handles = vec![];

        // Pre-populate events
        store.record_count("numerator", 100);
        store.record_count("denominator", 50);

        // Spawn 10 threads, each querying ratio 50 times
        for _ in 0..10 {
            let store_clone = Arc::clone(&store);
            let handle = thread::spawn(move || {
                for _ in 0..50 {
                    let ratio = store_clone
                        .query_ratio("numerator", "denominator")
                        .last_days(1);
                    // Should always get the same ratio
                    assert_eq!(ratio, Some(2.0));
                }
            });
            handles.push(handle);
        }

        // Wait for all threads to complete
        for handle in handles {
            handle.join().unwrap();
        }
    }

    #[test]
    fn test_concurrent_query_delta_from_multiple_threads() {
        let store = Arc::new(EventStore::new());
        let mut handles = vec![];

        // Pre-populate events
        store.record_count("positive", 150);
        store.record_count("negative", 50);

        // Spawn 10 threads, each querying delta 50 times
        for _ in 0..10 {
            let store_clone = Arc::clone(&store);
            let handle = thread::spawn(move || {
                for _ in 0..50 {
                    let delta = store_clone
                        .query_delta("positive", "negative")
                        .last_days(1)
                        .sum();
                    // Should always get the same delta
                    assert_eq!(delta, 100);
                }
            });
            handles.push(handle);
        }

        // Wait for all threads to complete
        for handle in handles {
            handle.join().unwrap();
        }
    }

    #[test]
    fn test_concurrent_query_many_while_recording() {
        let store = Arc::new(EventStore::new());
        let store_writer = Arc::clone(&store);
        let store_reader = Arc::clone(&store);

        // Writer thread
        let writer = thread::spawn(move || {
            for i in 0..100 {
                store_writer.record_count("event1", i);
                store_writer.record_count("event2", i * 2);
                thread::sleep(std::time::Duration::from_micros(10));
            }
        });

        // Reader thread querying multiple events
        let reader = thread::spawn(move || {
            for _ in 0..50 {
                let event_ids = &["event1", "event2"];
                let _sum = store_reader.query_many(event_ids).last_days(1).sum();
                thread::sleep(std::time::Duration::from_micros(20));
            }
        });

        writer.join().unwrap();
        reader.join().unwrap();

        // Verify final sums
        let event1_sum = store.query("event1").last_days(1).sum().unwrap();
        let event2_sum = store.query("event2").last_days(1).sum().unwrap();
        let multi_sum = store
            .query_many(&["event1", "event2"])
            .last_days(1)
            .sum()
            .unwrap();

        // event1: sum of 0..100 = 4950
        // event2: sum of (0..100)*2 = 9900
        assert_eq!(event1_sum, 4950);
        assert_eq!(event2_sum, 9900);
        assert_eq!(multi_sum, 14850);
    }

    #[test]
    fn test_concurrent_query_ratio_while_recording() {
        let store = Arc::new(EventStore::new());
        let store_writer = Arc::clone(&store);
        let store_reader = Arc::clone(&store);

        // Writer thread
        let writer = thread::spawn(move || {
            for i in 1..=100 {
                store_writer.record_count("numerator", i * 2);
                store_writer.record_count("denominator", i);
                thread::sleep(std::time::Duration::from_micros(10));
            }
        });

        // Reader thread querying ratio
        let reader = thread::spawn(move || {
            for _ in 0..50 {
                let _ratio = store_reader
                    .query_ratio("numerator", "denominator")
                    .last_days(1);
                thread::sleep(std::time::Duration::from_micros(20));
            }
        });

        writer.join().unwrap();
        reader.join().unwrap();

        // Verify final ratio
        let ratio = store.query_ratio("numerator", "denominator").last_days(1);
        // numerator: sum of 2,4,6,...,200 = 10100
        // denominator: sum of 1,2,3,...,100 = 5050
        // ratio = 10100/5050 = 2.0
        assert_eq!(ratio, Some(2.0));
    }

    #[test]
    fn test_concurrent_query_delta_while_recording() {
        let store = Arc::new(EventStore::new());
        let store_writer = Arc::clone(&store);
        let store_reader = Arc::clone(&store);

        // Writer thread
        let writer = thread::spawn(move || {
            for i in 0..100 {
                store_writer.record_count("positive", i * 2);
                store_writer.record_count("negative", i);
                thread::sleep(std::time::Duration::from_micros(10));
            }
        });

        // Reader thread querying delta
        let reader = thread::spawn(move || {
            for _ in 0..50 {
                let _delta = store_reader
                    .query_delta("positive", "negative")
                    .last_days(1)
                    .sum();
                thread::sleep(std::time::Duration::from_micros(20));
            }
        });

        writer.join().unwrap();
        reader.join().unwrap();

        // Verify final delta
        let delta = store.query_delta("positive", "negative").last_days(1).sum();
        // positive: sum of 0,2,4,...,198 = 9900
        // negative: sum of 0,1,2,...,99 = 4950
        // delta = 9900 - 4950 = 4950
        assert_eq!(delta, 4950);
    }
}