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;
pub struct EventStore {
pub(crate) inner: Arc<EventStoreInner>,
#[cfg(feature = "tokio")]
pub(crate) auto_persist_handle: Option<tokio::task::JoinHandle<()>>,
}
impl EventStore {
pub fn new() -> Self {
Self::from_parts(
SystemClock::new(),
None,
None,
EventCounterConfig::default(),
)
}
pub fn clock_now(&self) -> DateTime<Utc> {
self.inner.clock_now()
}
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,
}
}
pub fn record(&self, event_id: impl EventId) {
self.record_count(event_id, 1);
}
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();
}
pub fn record_at(&self, event_id: impl EventId, time: DateTime<Utc>) -> Result<()> {
self.record_count_at(event_id, 1, time)
}
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(())
}
pub fn record_ago(&self, event_id: impl EventId, duration: Duration) {
self.record_count_ago(event_id, 1, duration);
}
pub fn record_count_ago(&self, event_id: impl EventId, count: u32, duration: Duration) {
let now = self.inner.clock.now();
let time = now - duration;
let _ = self.record_count_at(event_id, count, time);
}
pub fn query(&self, event_id: impl EventId) -> Query {
Query::new(self.inner.clone(), event_id.as_ref().to_string())
}
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)
}
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(),
)
}
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(),
)
}
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
}
pub fn limit(&self) -> Limiter {
Limiter::new(self.inner.clone())
}
pub fn balance_delta(&self, positive: impl EventId, negative: impl EventId) -> Result<()> {
let positive_str = positive.as_ref();
let negative_str = negative.as_ref();
let delta = self.query_delta(positive_str, negative_str).ever().sum();
if delta > 0 {
self.record_count(negative_str, delta.min(u32::MAX as i64) as u32);
} else if delta < 0 {
self.record_count(positive_str, (-delta).min(u32::MAX as i64) as u32);
}
Ok(())
}
pub fn persist(&self) -> Result<()> {
self.persist_if_dirty(false)
}
pub fn close(self) -> Result<()> {
self.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()))?;
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(())
})();
match persist_result {
Ok(()) => {
storage.commit_transaction()?;
Ok(())
}
Err(e) => {
let _ = storage.rollback_transaction();
Err(e)
}
}
}
pub fn persist_all(&self) -> Result<()> {
self.persist_if_dirty(true)
}
pub fn reset_dirty(&self) {
for entry in self.inner.events.iter() {
let mut counter = entry.value().lock().unwrap();
counter.reset_dirty();
}
}
pub fn compact(&mut self) -> Result<()> {
if self.inner.storage.is_none() {
return Err(Error::Storage("No storage configured".to_string()));
}
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();
for key in keys {
if let Some(counter) = self.inner.get_counter_for_query(&key) {
let mut counter_guard = counter.lock().unwrap();
counter_guard.advance_if_needed(now);
}
}
self.persist_all()?;
self.inner.events.clear();
Ok(())
}
pub fn memory_usage(&self) -> usize {
let mut total: usize = 0;
for entry in self.inner.events.iter() {
let counter = entry.value().lock().unwrap();
total = total.saturating_add(counter.memory_usage())
}
total
}
pub fn tracked_intervals(&self) -> Vec<(TimeUnit, usize)> {
self.inner
.config
.as_vec()
.iter()
.map(|config| (config.time_unit(), config.bucket_count()))
.collect()
}
pub fn export_all(&self) -> Result<HashMap<String, SingleEventCounter>> {
let mut result = HashMap::new();
let mut all_event_ids: HashSet<String> = HashSet::new();
for entry in self.inner.events.iter() {
all_event_ids.insert(entry.key().clone());
}
if let Some(storage) = &self.inner.storage {
let storage_guard = storage.lock().unwrap();
let storage_keys = storage_guard.list_keys()?;
drop(storage_guard); for key in storage_keys {
all_event_ids.insert(key);
}
}
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)
}
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)
}
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(())
}
pub fn import_all(&self, events: HashMap<String, SingleEventCounter>) -> Result<()> {
for (event_id, counter) in events {
self.import_event(event_id, counter)?;
}
Ok(())
}
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(())
}
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)
}
#[cfg(feature = "tokio")]
#[allow(dead_code)] pub(crate) fn spawn_auto_persist(
inner: Arc<EventStoreInner>,
interval: chrono::Duration,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let std_interval = interval
.to_std()
.expect("auto_persist interval must be positive");
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) = {
let mut counter = counter.lock().unwrap();
if force_dirty || counter.is_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(());
}
} {
storage.save(event_id, data)
} else {
storage.delete(event_id)
}
.map_err(|_e| {
let mut counter = counter.lock().unwrap();
counter.mark_dirty();
_e
})
}
}
impl Default for EventStore {
fn default() -> Self {
Self::new()
}
}
impl Drop for EventStore {
fn drop(&mut self) {
#[cfg(feature = "tokio")]
if let Some(handle) = &self.auto_persist_handle {
handle.abort();
}
if self.is_dirty() {
let _ = self.persist();
}
}
}
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());
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");
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();
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));
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();
store.record_count("login", 5);
store.record_count("logout", 3);
store.record_count("error", 1);
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");
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());
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");
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));
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();
let past = now - Duration::days(2);
let result = store.record_at("test_event", past);
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));
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();
store.record_ago("ancient_event", Duration::days(365 * 10));
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();
store.record_count_ago("ancient_event", 100, Duration::days(365 * 10));
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);
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_balance_delta_negative() {
let store = EventStore::new();
store.record_count("credits", 3);
store.record_count("debits", 10);
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_balance_delta_zero() {
let store = EventStore::new();
store.record_count("credits", 10);
store.record_count("debits", 10);
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");
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());
let result = store.close();
assert!(result.is_ok());
}
#[cfg(feature = "serde-bincode")]
#[test]
fn test_close_returns_error_on_persist_failure() {
use crate::EventStoreBuilder;
let store = EventStoreBuilder::new().build().unwrap();
store.record("event1");
assert!(store.is_dirty());
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();
store.record("event1");
store.persist().unwrap();
assert!(!store.is_dirty());
store.record("event2");
assert!(store.is_dirty());
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();
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();
store.record("event1");
store.record("event2");
store.persist_all().unwrap();
assert!(!store.is_dirty());
store.compact().unwrap();
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;
let base_store = EventStore::from_parts(
SystemClock::new(),
Some(Box::new(MemoryStorage::new())),
Some(Arc::new(crate::formatter::BincodeFormat)),
EventCounterConfig::default(),
);
let store = Arc::new(Mutex::new(base_store));
let store_writer = Arc::clone(&store);
let store_compactor = Arc::clone(&store);
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));
}
});
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();
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();
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();
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();
store.record_count("event1", 10);
store.persist().unwrap();
store.record_count("event2", 20);
let storage2 = {
let storage_arc = store.inner.storage.as_ref().unwrap().clone();
let storage_guard = storage_arc.lock().unwrap();
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();
store2.record_count("event3", 30);
let exported = store2.export_all().unwrap();
assert_eq!(exported.len(), 2);
assert!(exported.contains_key("event1"));
assert!(exported.contains_key("event3"));
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();
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();
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)); assert_eq!(store1.query("event2").last_days(1).sum(), Some(20)); assert_eq!(store1.query("event3").last_days(1).sum(), Some(30));
}
#[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);
let b1_export = store_b1.export_all().unwrap();
store_a1.merge_all(b1_export).unwrap();
let a2_export = store_a2.export_all().unwrap();
store_b2.merge_all(a2_export).unwrap();
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);
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();
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();
assert_eq!(store_a1.query("event1").last_days(1).sum(), Some(60));
assert_eq!(store_a2.query("event1").last_days(1).sum(), Some(60));
}
#[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;
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(),
);
assert!(store.auto_persist_handle.is_none());
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());
}
#[cfg(all(feature = "tokio", feature = "serde"))]
#[tokio::test]
async fn test_spawn_auto_persist_creates_background_task() {
use crate::storage::MemoryStorage;
use chrono::Duration;
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(),
);
store.record("test");
assert!(store.is_dirty());
let handle =
EventStore::spawn_auto_persist(store.inner.clone(), Duration::milliseconds(50));
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(!store.is_dirty());
handle.abort();
}
#[cfg(all(feature = "tokio", feature = "serde"))]
#[tokio::test]
async fn test_drop_aborts_handle_and_persists() {
use crate::storage::MemoryStorage;
use std::time::Duration;
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(),
);
let handle = tokio::spawn(async {
loop {
tokio::time::sleep(Duration::from_secs(1)).await;
}
});
store.auto_persist_handle = Some(handle);
store.record("test");
assert!(store.is_dirty());
drop(store);
}
#[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");
let store = EventStore::builder()
.with_storage(FilePerEvent::new(&storage_path, ".dat").unwrap())
.auto_persist(Duration::milliseconds(100))
.build()
.unwrap();
store.record("event1");
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
{
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"
);
}
store.record("event2");
drop(store);
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![];
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);
}
for handle in handles {
handle.join().unwrap();
}
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);
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));
}
});
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();
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;
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);
let writer = thread::spawn(move || {
for _ in 0..100 {
store_writer.record("persist_event");
thread::sleep(std::time::Duration::from_micros(10));
}
});
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();
store.persist().unwrap();
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![];
store.record_count("event1", 100);
store.record_count("event2", 200);
store.record_count("event3", 300);
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();
assert_eq!(sum, Some(600));
}
});
handles.push(handle);
}
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![];
store.record_count("numerator", 100);
store.record_count("denominator", 50);
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);
assert_eq!(ratio, Some(2.0));
}
});
handles.push(handle);
}
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![];
store.record_count("positive", 150);
store.record_count("negative", 50);
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();
assert_eq!(delta, 100);
}
});
handles.push(handle);
}
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);
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));
}
});
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();
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();
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);
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));
}
});
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();
let ratio = store.query_ratio("numerator", "denominator").last_days(1);
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);
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));
}
});
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();
let delta = store.query_delta("positive", "negative").last_days(1).sum();
assert_eq!(delta, 4950);
}
}