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
182impl cursor::Bind for Event {
183 type T = Self;
184
185 fn sort_by(data: &mut Vec<Self::T>, is_order_desc: bool) {
186 if !is_order_desc {
187 data.sort_by(|a, b| {
188 if a.timestamp_subsec != b.timestamp_subsec {
189 return a.timestamp_subsec.cmp(&b.timestamp_subsec);
190 }
191
192 if a.timestamp != b.timestamp {
193 return a.timestamp.cmp(&b.timestamp);
194 }
195
196 if a.version != b.version {
197 return a.version.cmp(&b.version);
198 }
199
200 a.id.cmp(&b.id)
201 });
202 } else {
203 data.sort_by(|a, b| {
204 if a.timestamp_subsec != b.timestamp_subsec {
205 return b.timestamp_subsec.cmp(&a.timestamp_subsec);
206 }
207
208 if a.timestamp != b.timestamp {
209 return b.timestamp.cmp(&a.timestamp);
210 }
211
212 if a.version != b.version {
213 return b.version.cmp(&a.version);
214 }
215
216 b.id.cmp(&a.id)
217 });
218 }
219 }
220
221 fn retain(
222 data: &mut Vec<Self::T>,
223 cursor: <<Self as cursor::Bind>::T as Cursor>::T,
224 is_order_desc: bool,
225 ) {
226 data.retain(|event| {
227 if is_order_desc {
228 event.timestamp < cursor.t
229 || (event.timestamp == cursor.t
230 && (event.timestamp_subsec < cursor.s
231 || (event.timestamp_subsec == cursor.s
232 && (event.version < cursor.v
233 || (event.version == cursor.v
234 && event.id.to_string() < cursor.i)))))
235 } else {
236 event.timestamp > cursor.t
237 || (event.timestamp == cursor.t
238 && (event.timestamp_subsec > cursor.s
239 || (event.timestamp_subsec == cursor.s
240 && (event.version > cursor.v
241 || (event.version == cursor.v
242 && event.id.to_string() > cursor.i)))))
243 }
244 });
245 }
246}
247
248#[cfg(any(feature = "sqlite", feature = "mysql", feature = "postgres"))]
249impl<R: sqlx::Row> sqlx::FromRow<'_, R> for Event
250where
251 i32: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
252 Vec<u8>: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
253 String: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
254 i64: sqlx::Type<R::Database> + for<'r> sqlx::Decode<'r, R::Database>,
255 for<'r> &'r str: sqlx::Type<R::Database> + sqlx::Decode<'r, R::Database>,
256 for<'r> &'r str: sqlx::ColumnIndex<R>,
257{
258 fn from_row(row: &R) -> Result<Self, sqlx::Error> {
259 let timestamp: i64 = sqlx::Row::try_get(row, "timestamp")?;
260 let timestamp_subsec: i64 = sqlx::Row::try_get(row, "timestamp_subsec")?;
261 let version: i32 = sqlx::Row::try_get(row, "version")?;
262
263 Ok(Event {
264 id: Ulid::from_string(sqlx::Row::try_get(row, "id")?)
265 .map_err(|err| sqlx::Error::InvalidArgument(err.to_string()))?,
266 aggregator_id: sqlx::Row::try_get(row, "aggregator_id")?,
267 aggregator_type: sqlx::Row::try_get(row, "aggregator_type")?,
268 version: version as u16,
269 name: sqlx::Row::try_get(row, "name")?,
270 routing_key: sqlx::Row::try_get(row, "routing_key")?,
271 data: sqlx::Row::try_get(row, "data")?,
272 metadata: sqlx::Row::try_get(row, "metadata")?,
273 timestamp: timestamp as u64,
274 timestamp_subsec: timestamp_subsec as u32,
275 })
276 }
277}