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 aggregator macro
23//! #[evento::aggregator]
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//! ## Aggregator Builder
36//!
37//! Use [`create()`] or [`aggregator()`] 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 subscribing to events. See the [`projection`] module.
52//!
53//! ## Cursor-based Pagination
54//!
55//! GraphQL-style pagination for querying events. See the [`cursor`] module.
56//!
57//! # Modules
58//!
59//! - [`context`] - Type-safe request context for storing arbitrary data
60//! - [`cursor`] - Cursor-based pagination types and traits
61//! - [`metadata`] - Standard event metadata types
62//! - [`projection`] - Projections, subscriptions, and event handlers
63//!
64//! # Example
65//!
66//! ```rust,ignore
67//! use evento::{Executor, metadata::Metadata, cursor::Args, ReadAggregator};
68//!
69//! // Create and persist an event
70//! let id = evento::create()
71//! .event(&AccountOpened { owner_id: "user1".into(), initial_balance: 1000 })
72//! .metadata(&Metadata::default())
73//! .commit(&executor)
74//! .await?;
75//!
76//! // Query events with pagination
77//! let events = executor.read(
78//! Some(vec![ReadAggregator::id("myapp/Account", &id)]),
79//! None,
80//! Args::forward(10, None),
81//! ).await?;
82//! ```
83
84mod aggregator;
85pub mod context;
86pub mod cursor;
87mod executor;
88pub mod metadata;
89pub mod projection;
90
91#[cfg(feature = "macro")]
92pub use evento_macro::*;
93
94pub use aggregator::*;
95pub use executor::*;
96pub use projection::RoutingKey;
97
98use std::fmt::Debug;
99use ulid::Ulid;
100
101use crate::cursor::Cursor;
102
103/// Cursor data for event pagination.
104///
105/// Used internally for base64-encoded cursor values in paginated queries.
106/// Contains the essential fields needed to uniquely identify an event's position.
107#[derive(Debug, bitcode::Encode, bitcode::Decode)]
108pub struct EventCursor {
109 /// Event ID (ULID string)
110 pub i: String,
111 /// Event version
112 pub v: u16,
113 /// Event timestamp (Unix timestamp in seconds)
114 pub t: u64,
115 /// Sub-second precision (milliseconds)
116 pub s: u32,
117 /// Event routing_key
118 pub r: Option<String>,
119}
120
121/// A stored event in the event store.
122///
123/// Events are immutable records of facts that occurred in your domain.
124/// They contain serialized data and metadata, along with positioning
125/// information for the aggregate they belong to.
126///
127/// # Fields
128///
129/// - `id` - Unique event identifier (ULID format for time-ordering)
130/// - `aggregator_id` - The aggregate instance this event belongs to
131/// - `aggregator_type` - Type name like `"myapp/BankAccount"`
132/// - `version` - Sequence number within the aggregate (for optimistic concurrency)
133/// - `name` - Event type name like `"AccountOpened"`
134/// - `routing_key` - Optional key for event distribution/partitioning
135/// - `data` - Serialized event payload (bitcode format)
136/// - `metadata` - Serialized metadata (bitcode format)
137/// - `timestamp` - When the event occurred (Unix seconds)
138/// - `timestamp_subsec` - Sub-second precision (milliseconds)
139///
140/// # Serialization
141///
142/// Event data and metadata are serialized using [bitcode](https://crates.io/crates/bitcode)
143/// for compact binary representation. Use [`projection::EventData`] to deserialize.
144#[derive(Debug, Clone, PartialEq, Default)]
145pub struct Event {
146 /// Unique event identifier (ULID)
147 pub id: Ulid,
148 /// ID of the aggregate this event belongs to
149 pub aggregator_id: String,
150 /// Type name of the aggregate (e.g., "myapp/User")
151 pub aggregator_type: String,
152 /// Version number of the aggregate after this event
153 pub version: u16,
154 /// Event type name
155 pub name: String,
156 /// Optional routing key for event distribution
157 pub routing_key: Option<String>,
158 /// Serialized event data (bitcode format)
159 pub data: Vec<u8>,
160 /// Serialized event metadata (bitcode format)
161 pub metadata: Vec<u8>,
162 /// Unix timestamp when the event occurred (seconds)
163 pub timestamp: u64,
164 /// Sub-second precision (milliseconds)
165 pub timestamp_subsec: u32,
166}
167
168impl Cursor for Event {
169 type T = EventCursor;
170
171 fn serialize(&self) -> Self::T {
172 EventCursor {
173 i: self.id.to_string(),
174 v: self.version,
175 t: self.timestamp,
176 s: self.timestamp_subsec,
177 r: self.routing_key.to_owned(),
178 }
179 }
180
181 fn serialize_cursor(&self) -> Result<cursor::Value, cursor::CursorError> {
182 use base64::{alphabet, engine::general_purpose, engine::GeneralPurpose, Engine};
183
184 let cursor = self.serialize();
185 let encoded = bitcode::encode(&cursor);
186
187 let engine = GeneralPurpose::new(&alphabet::URL_SAFE, general_purpose::PAD);
188
189 Ok(cursor::Value(engine.encode(&encoded)))
190 }
191
192 fn deserialize_cursor(value: &cursor::Value) -> Result<Self::T, cursor::CursorError> {
193 use base64::{alphabet, engine::general_purpose, engine::GeneralPurpose, Engine};
194
195 let engine = GeneralPurpose::new(&alphabet::URL_SAFE, general_purpose::PAD);
196 let decoded = engine.decode(value)?;
197
198 let result = bitcode::decode::<Self::T>(&decoded)
199 .map_err(|e| cursor::CursorError::Bitcode(e.to_string()))?;
200
201 Ok(result)
202 }
203}
204
205impl cursor::Bind for Event {
206 type T = Self;
207
208 fn sort_by(data: &mut Vec<Self::T>, is_order_desc: bool) {
209 if !is_order_desc {
210 data.sort_by(|a, b| {
211 if a.timestamp_subsec != b.timestamp_subsec {
212 return a.timestamp_subsec.cmp(&b.timestamp_subsec);
213 }
214
215 if a.timestamp != b.timestamp {
216 return a.timestamp.cmp(&b.timestamp);
217 }
218
219 if a.version != b.version {
220 return a.version.cmp(&b.version);
221 }
222
223 a.id.cmp(&b.id)
224 });
225 } else {
226 data.sort_by(|a, b| {
227 if a.timestamp_subsec != b.timestamp_subsec {
228 return b.timestamp_subsec.cmp(&a.timestamp_subsec);
229 }
230
231 if a.timestamp != b.timestamp {
232 return b.timestamp.cmp(&a.timestamp);
233 }
234
235 if a.version != b.version {
236 return b.version.cmp(&a.version);
237 }
238
239 b.id.cmp(&a.id)
240 });
241 }
242 }
243
244 fn retain(
245 data: &mut Vec<Self::T>,
246 cursor: <<Self as cursor::Bind>::T as Cursor>::T,
247 is_order_desc: bool,
248 ) {
249 data.retain(|event| {
250 if is_order_desc {
251 event.timestamp < cursor.t
252 || (event.timestamp == cursor.t
253 && (event.timestamp_subsec < cursor.s
254 || (event.timestamp_subsec == cursor.s
255 && (event.version < cursor.v
256 || (event.version == cursor.v
257 && event.id.to_string() < cursor.i)))))
258 } else {
259 event.timestamp > cursor.t
260 || (event.timestamp == cursor.t
261 && (event.timestamp_subsec > cursor.s
262 || (event.timestamp_subsec == cursor.s
263 && (event.version > cursor.v
264 || (event.version == cursor.v
265 && event.id.to_string() > cursor.i)))))
266 }
267 });
268 }
269}
270
271#[cfg(any(feature = "sqlite", feature = "mysql", feature = "postgres"))]
272impl<R: sqlx::Row> sqlx::FromRow<'_, R> for Event
273where
274 i32: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
275 Vec<u8>: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
276 String: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
277 i64: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
278 for<'r> &'r str: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
279 for<'r> &'r str: sqlx::ColumnIndex<R>,
280{
281 fn from_row(row: &R) -> Result<Self, sqlx::Error> {
282 let timestamp: i64 = sqlx::Row::try_get(row, "timestamp")?;
283 let timestamp_subsec: i64 = sqlx::Row::try_get(row, "timestamp_subsec")?;
284 let version: i32 = sqlx::Row::try_get(row, "version")?;
285
286 Ok(Event {
287 id: Ulid::from_string(sqlx::Row::try_get(row, "id")?)
288 .map_err(|err| sqlx::Error::InvalidArgument(err.to_string()))?,
289 aggregator_id: sqlx::Row::try_get(row, "aggregator_id")?,
290 aggregator_type: sqlx::Row::try_get(row, "aggregator_type")?,
291 version: version as u16,
292 name: sqlx::Row::try_get(row, "name")?,
293 routing_key: sqlx::Row::try_get(row, "routing_key")?,
294 data: sqlx::Row::try_get(row, "data")?,
295 metadata: sqlx::Row::try_get(row, "metadata")?,
296 timestamp: timestamp as u64,
297 timestamp_subsec: timestamp_subsec as u32,
298 })
299 }
300}