evento_core/lib.rs
1//! Core types and traits for the Evento event sourcing library.
2//!
3//! This crate provides the foundational abstractions for building event-sourced applications
4//! with Evento. It defines the core traits, types, and builders used throughout the framework.
5//!
6//! # Features
7//!
8//! - **`macro`** (default) - Procedural macros from `evento-macro`
9//! - **`group`** - Multi-executor support via `EventoGroup`
10//! - **`rw`** - Read-write split executor pattern via `Rw`
11//! - **`sqlite`**, **`mysql`**, **`postgres`** - Database support via sqlx
12//! - **`fjall`** - Embedded key-value storage with Fjall
13//!
14//! # Core Concepts
15//!
16//! ## Events
17//!
18//! Events are immutable facts that represent something that happened in your domain.
19//! The [`Event`] struct stores serialized event data with metadata:
20//!
21//! ```rust,ignore
22//! // Define events using the aggregate macro
23//! #[evento::aggregate]
24//! pub enum BankAccount {
25//! AccountOpened { owner_id: String, initial_balance: i64 },
26//! MoneyDeposited { amount: i64 },
27//! }
28//! ```
29//!
30//! ## Executor
31//!
32//! The [`Executor`] trait abstracts event storage and retrieval. Implementations
33//! handle persisting events, querying, and managing subscriptions.
34//!
35//! ## Aggregate Builder
36//!
37//! Use [`create()`] or [`append()`] to build and commit events:
38//!
39//! ```rust,ignore
40//! use evento::metadata::Metadata;
41//!
42//! let id = evento::create()
43//! .event(&AccountOpened { owner_id: "user1".into(), initial_balance: 1000 })
44//! .metadata(&Metadata::default())
45//! .commit(&executor)
46//! .await?;
47//! ```
48//!
49//! ## Projections
50//!
51//! Build read models by replaying events. Use the [`projection`](mod@projection) module for loading
52//! aggregate state:
53//!
54//! ```rust,ignore
55//! use evento::projection::Projection;
56//!
57//! #[evento::projection]
58//! #[derive(Debug)]
59//! pub struct AccountView {
60//! pub balance: i64,
61//! }
62//!
63//! #[evento::handler]
64//! async fn on_deposited(
65//! event: Event<MoneyDeposited>,
66//! projection: &mut AccountView,
67//! ) -> anyhow::Result<()> {
68//! projection.balance += event.data.amount;
69//! Ok(())
70//! }
71//!
72//! let result = Projection::<_, AccountView>::new::<BankAccount>()
73//! .handler(on_deposited())
74//! .load("account-123")
75//! .execute(&executor)
76//! .await?;
77//! ```
78//!
79//! ## Subscriptions
80//!
81//! Process events continuously in real-time. See the [`subscription`](mod@subscription) module:
82//!
83//! ```rust,ignore
84//! use evento::subscription::SubscriptionBuilder;
85//!
86//! #[evento::subscription]
87//! async fn on_deposited<E: Executor>(
88//! context: &Context<'_, E>,
89//! event: Event<MoneyDeposited>,
90//! ) -> anyhow::Result<()> {
91//! // Perform side effects
92//! Ok(())
93//! }
94//!
95//! let subscription = SubscriptionBuilder::<Sqlite>::new("deposit-processor")
96//! .handler(on_deposited())
97//! .routing_key("accounts")
98//! .start(&executor)
99//! .await?;
100//! ```
101//!
102//! ## Cursor-based Pagination
103//!
104//! GraphQL-style pagination for querying events. See the [`cursor`] module.
105//!
106//! # Modules
107//!
108//! - [`context`] - Type-safe request context for storing arbitrary data
109//! - [`cursor`] - Cursor-based pagination types and traits
110//! - [`metadata`] - Standard event metadata types
111//! - [`projection`](mod@projection) - Projections for loading aggregate state
112//! - [`subscription`](mod@subscription) - Continuous event processing with subscriptions
113//!
114//! # Example
115//!
116//! ```rust,ignore
117//! use evento::{Executor, metadata::Metadata, cursor::Args, EventFilter};
118//!
119//! // Create and persist an event
120//! let id = evento::create()
121//! .event(&AccountOpened { owner_id: "user1".into(), initial_balance: 1000 })
122//! .metadata(&Metadata::default())
123//! .commit(&executor)
124//! .await?;
125//!
126//! // Query events with pagination
127//! let events = executor.read(
128//! Some(vec![EventFilter::by_id("myapp/Account", &id)]),
129//! None,
130//! Args::forward(10, None),
131//! ).await?;
132//! ```
133
134mod aggregator;
135pub mod context;
136pub mod cursor;
137mod executor;
138pub mod metadata;
139pub mod projection;
140pub mod subscription;
141
142#[cfg(feature = "macro")]
143pub use evento_macro::*;
144
145pub use aggregator::*;
146pub use executor::*;
147pub use subscription::RoutingKey;
148
149use std::fmt::Debug;
150use ulid::Ulid;
151
152use crate::{cursor::Cursor, metadata::Metadata};
153
154/// Cursor data for event pagination.
155///
156/// Used internally for base64-encoded cursor values in paginated queries.
157/// Contains the essential fields needed to uniquely identify an event's position.
158#[derive(Debug, bitcode::Encode, bitcode::Decode)]
159pub struct EventCursor {
160 /// Event ID (ULID string)
161 pub i: String,
162 /// Event version
163 pub v: u16,
164 /// Event timestamp (Unix timestamp in seconds)
165 pub t: u64,
166 /// Sub-second precision (milliseconds)
167 pub s: u32,
168}
169
170/// A stored event in the event store.
171///
172/// Events are immutable records of facts that occurred in your domain.
173/// They contain serialized data and metadata, along with positioning
174/// information for the aggregate they belong to.
175///
176/// # Fields
177///
178/// - `id` - Unique event identifier (ULID format for time-ordering)
179/// - `aggregate_id` - The aggregate instance this event belongs to
180/// - `aggregate_type` - Type name like `"myapp/BankAccount"`
181/// - `version` - Sequence number within the aggregate (for optimistic concurrency)
182/// - `name` - Event type name like `"AccountOpened"`
183/// - `routing_key` - Optional key for event distribution/partitioning
184/// - `data` - Serialized event payload (bitcode format)
185/// - `metadata` - Event metadata (see [`metadata::Metadata`])
186/// - `timestamp` - When the event occurred (Unix seconds)
187/// - `timestamp_subsec` - Sub-second precision (milliseconds)
188///
189/// # Serialization
190///
191/// Event data is serialized using [bitcode](https://crates.io/crates/bitcode)
192/// for compact binary representation. Use [`metadata::Event`] to deserialize typed events.
193#[derive(Debug, Clone, PartialEq, Default)]
194pub struct Event {
195 /// Unique event identifier (ULID)
196 pub id: Ulid,
197 /// ID of the aggregate this event belongs to
198 pub aggregate_id: String,
199 /// Type name of the aggregate (e.g., "myapp/User")
200 pub aggregate_type: String,
201 /// Version number of the aggregate after this event
202 pub version: u16,
203 /// Event type name
204 pub name: String,
205 /// Optional routing key for event distribution
206 pub routing_key: Option<String>,
207 /// Serialized event data (bitcode format)
208 pub data: Vec<u8>,
209 /// Event metadata
210 pub metadata: Metadata,
211 /// Unix timestamp when the event occurred (seconds)
212 pub timestamp: u64,
213 /// Sub-second precision (milliseconds)
214 pub timestamp_subsec: u32,
215}
216
217impl Cursor for Event {
218 type T = EventCursor;
219
220 fn serialize(&self) -> Self::T {
221 EventCursor {
222 i: self.id.to_string(),
223 v: self.version,
224 t: self.timestamp,
225 s: self.timestamp_subsec,
226 }
227 }
228}
229
230impl cursor::Bind for Event {
231 type T = Self;
232
233 fn sort_by(data: &mut Vec<Self::T>, is_order_desc: bool) {
234 if !is_order_desc {
235 data.sort_by(|a, b| {
236 if a.timestamp != b.timestamp {
237 return a.timestamp.cmp(&b.timestamp);
238 }
239
240 if a.timestamp_subsec != b.timestamp_subsec {
241 return a.timestamp_subsec.cmp(&b.timestamp_subsec);
242 }
243
244 if a.version != b.version {
245 return a.version.cmp(&b.version);
246 }
247
248 a.id.cmp(&b.id)
249 });
250 } else {
251 data.sort_by(|a, b| {
252 if a.timestamp != b.timestamp {
253 return b.timestamp.cmp(&a.timestamp);
254 }
255
256 if a.timestamp_subsec != b.timestamp_subsec {
257 return b.timestamp_subsec.cmp(&a.timestamp_subsec);
258 }
259
260 if a.version != b.version {
261 return b.version.cmp(&a.version);
262 }
263
264 b.id.cmp(&a.id)
265 });
266 }
267 }
268
269 fn retain(
270 data: &mut Vec<Self::T>,
271 cursor: <<Self as cursor::Bind>::T as Cursor>::T,
272 is_order_desc: bool,
273 ) {
274 data.retain(|event| {
275 if is_order_desc {
276 event.timestamp < cursor.t
277 || (event.timestamp == cursor.t
278 && (event.timestamp_subsec < cursor.s
279 || (event.timestamp_subsec == cursor.s
280 && (event.version < cursor.v
281 || (event.version == cursor.v
282 && event.id.to_string() < cursor.i)))))
283 } else {
284 event.timestamp > cursor.t
285 || (event.timestamp == cursor.t
286 && (event.timestamp_subsec > cursor.s
287 || (event.timestamp_subsec == cursor.s
288 && (event.version > cursor.v
289 || (event.version == cursor.v
290 && event.id.to_string() > cursor.i)))))
291 }
292 });
293 }
294}
295
296#[cfg(test)]
297mod tests {
298 use super::*;
299 use crate::cursor::Bind;
300
301 fn event_at(timestamp: u64, timestamp_subsec: u32) -> Event {
302 Event {
303 id: Ulid::new(),
304 timestamp,
305 timestamp_subsec,
306 ..Default::default()
307 }
308 }
309
310 /// Events must be ordered by whole seconds first, then sub-seconds — matching
311 /// the SQL `ORDER BY timestamp, timestamp_subsec, version, id`. A regression for
312 /// the bug where `timestamp_subsec` was (incorrectly) the major sort key, which
313 /// reordered events whose larger second carried a smaller sub-second.
314 #[test]
315 fn sort_orders_by_timestamp_before_subsec() {
316 // (t=1000, s=500) must come before (t=1001, s=100) ascending.
317 let earlier = event_at(1000, 500);
318 let later = event_at(1001, 100);
319
320 let mut asc = vec![later.clone(), earlier.clone()];
321 Event::sort_by(&mut asc, false);
322 assert_eq!(
323 (asc[0].timestamp, asc[1].timestamp),
324 (1000, 1001),
325 "ascending order must place the smaller whole-second first"
326 );
327
328 let mut desc = vec![earlier, later];
329 Event::sort_by(&mut desc, true);
330 assert_eq!(
331 (desc[0].timestamp, desc[1].timestamp),
332 (1001, 1000),
333 "descending order must place the larger whole-second first"
334 );
335 }
336
337 /// `retain` (cursor keyset filter) must agree with `sort_by`: forward pagination
338 /// from a cursor at (t=1000, s=500) keeps the strictly-later (t=1001, s=100).
339 #[test]
340 fn retain_agrees_with_sort_order() {
341 let cursor = EventCursor {
342 i: Ulid::nil().to_string(),
343 v: 0,
344 t: 1000,
345 s: 500,
346 };
347
348 let mut forward = vec![event_at(1001, 100), event_at(1000, 400)];
349 Event::retain(&mut forward, cursor, false);
350 assert_eq!(forward.len(), 1);
351 assert_eq!(forward[0].timestamp, 1001);
352 }
353}