Expand description
§Aggregate Module
Core abstractions for domain aggregates and commands in event sourcing.
§Overview
This module provides the traits needed to implement aggregates following the Event Sourcing and CQRS patterns. Aggregates are the fundamental building blocks that encapsulate business logic, enforce invariants, and produce events.
§Design Philosophy
Complexity is Opt-In: You can use aggregates in two ways:
- Simple Path: Define enums for commands/events, implement the trait
- Complex Path: Add rich domain logic, validation, and business rules
Both approaches use the same infrastructure and are first-class citizens.
§Core Concepts
§Commands
Commands represent intent to change state. They are imperative (e.g., CreateUser,
UpdateProfile) and can be rejected if business rules aren’t satisfied.
- Commands are validated before producing events
- Commands may produce zero events (validation failure)
- Commands may produce multiple events (complex operations)
- Commands from one aggregate must be atomic
§Events
Events represent facts about things that have happened. They are past tense
(e.g., UserCreated, ProfileUpdated) and cannot be rejected once produced.
- Events are immutable once written
- Events are the source of truth
- Events are used to reconstruct aggregate state
- Events are published to event bus for subscribers
§Aggregates
Aggregates are consistency boundaries that:
- Encapsulate domain logic and business rules
- Validate commands and produce events
- Apply events to update internal state
- Can be reconstructed from their event stream
- Enforce invariants within their boundary
§Quick Start
§1. Define Your Domain Events
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UserEvent {
UserCreated {
id: String,
name: String,
email: String,
},
ProfileUpdated {
name: String,
},
EmailChanged {
email: String,
},
}§2. Define Your Commands
use arc_core::aggregate::Command;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UserCommand {
CreateUser {
id: String,
name: String,
email: String,
},
UpdateProfile {
id: String,
name: String,
},
ChangeEmail {
id: String,
email: String,
},
}
impl Command for UserCommand {
fn aggregate_id(&self) -> &str {
match self {
UserCommand::CreateUser { id, .. } => id,
UserCommand::UpdateProfile { id, .. } => id,
UserCommand::ChangeEmail { id, .. } => id,
}
}
}§3. Define Your Aggregate
use arc_core::aggregate::Aggregate;
use arc_core::event::Event;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum UserError {
#[error("User already exists")]
AlreadyExists,
#[error("User not found")]
NotFound,
#[error("Invalid email format")]
InvalidEmail,
}
#[derive(Default)]
pub struct UserAggregate {
id: Option<String>,
name: Option<String>,
email: Option<String>,
version: i64,
created: bool,
}
#[async_trait::async_trait]
impl Aggregate for UserAggregate {
type Command = UserCommand;
type Event = UserEvent;
type Error = UserError;
fn aggregate_type() -> &'static str {
"User"
}
fn version(&self) -> i64 {
self.version
}
async fn handle(&self, command: Self::Command) -> Result<Vec<Event>, Self::Error> {
match command {
UserCommand::CreateUser { id, name, email } => {
// Enforce invariant: user cannot be created twice
if self.created {
return Err(UserError::AlreadyExists);
}
// Validate email
if !email.contains('@') {
return Err(UserError::InvalidEmail);
}
// Produce event
Ok(vec![Event::new(
"User",
&id,
self.version + 1,
"UserCreated",
serde_json::json!({
"id": id,
"name": name,
"email": email,
}),
)])
}
UserCommand::UpdateProfile { id, name } => {
if !self.created {
return Err(UserError::NotFound);
}
Ok(vec![Event::new(
"User",
&id,
self.version + 1,
"ProfileUpdated",
serde_json::json!({ "name": name }),
)])
}
UserCommand::ChangeEmail { id, email } => {
if !self.created {
return Err(UserError::NotFound);
}
if !email.contains('@') {
return Err(UserError::InvalidEmail);
}
Ok(vec![Event::new(
"User",
&id,
self.version + 1,
"EmailChanged",
serde_json::json!({ "email": email }),
)])
}
}
}
fn apply(&mut self, event: &Event) {
self.version = event.sequence;
match event.event_type.as_str() {
"UserCreated" => {
self.id = Some(event.payload["id"].as_str().unwrap().to_string());
self.name = Some(event.payload["name"].as_str().unwrap().to_string());
self.email = Some(event.payload["email"].as_str().unwrap().to_string());
self.created = true;
}
"ProfileUpdated" => {
self.name = Some(event.payload["name"].as_str().unwrap().to_string());
}
"EmailChanged" => {
self.email = Some(event.payload["email"].as_str().unwrap().to_string());
}
_ => {}
}
}
}§4. Use Your Aggregate
// Create a command
let command = UserCommand::CreateUser {
id: "user-123".to_string(),
name: "Alice".to_string(),
email: "alice@example.com".to_string(),
};
// Create aggregate (typically loaded from event store)
let aggregate = UserAggregate::default();
// Handle command
let events = aggregate.handle(command).await.unwrap();
// Events would be persisted to event store and published to event bus
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_type, "UserCreated");§Testing Your Aggregates
Aggregates are easy to test because they’re pure functions (commands → events → state).
#[tokio::test]
async fn test_user_creation() {
// Given: A new user aggregate
let aggregate = UserAggregate::default();
// When: Creating a user
let command = UserCommand::CreateUser {
id: "user-123".to_string(),
name: "Alice".to_string(),
email: "alice@example.com".to_string(),
};
let events = aggregate.handle(command).await.unwrap();
// Then: UserCreated event is produced
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_type, "UserCreated");
assert_eq!(events[0].aggregate_id, "user-123");
}
#[tokio::test]
async fn test_invalid_email() {
// Given: A new user aggregate
let aggregate = UserAggregate::default();
// When: Creating a user with invalid email
let command = UserCommand::CreateUser {
id: "user-456".to_string(),
name: "Bob".to_string(),
email: "not-an-email".to_string(), // Invalid!
};
let result = aggregate.handle(command).await;
// Then: Error is returned
assert!(result.is_err());
}
#[tokio::test]
async fn test_cannot_create_twice() {
// Given: An existing user (reconstructed from events)
let mut aggregate = UserAggregate::default();
let event = Event::new(
"User",
"user-789",
1,
"UserCreated",
serde_json::json!({
"id": "user-789",
"name": "Charlie",
"email": "charlie@example.com"
}),
);
aggregate.apply(&event);
// When: Trying to create the user again
let command = UserCommand::CreateUser {
id: "user-789".to_string(),
name: "Charlie".to_string(),
email: "charlie@example.com".to_string(),
};
let result = aggregate.handle(command).await;
// Then: Error is returned
assert!(result.is_err());
}§Advanced Patterns
§Multiple Events from One Command
Sometimes a command should produce multiple events atomically:
ⓘ
async fn handle(&self, command: Self::Command) -> Result<Vec<Event>, Self::Error> {
match command {
OrderCommand::PlaceOrder { order_id, items } => {
// Validate stock
// ...
// Produce multiple events
Ok(vec![
Event::new("Order", &order_id, self.version + 1, "OrderPlaced", ...),
Event::new("Order", &order_id, self.version + 2, "InventoryReserved", ...),
Event::new("Order", &order_id, self.version + 3, "PaymentRequested", ...),
])
}
}
}§Conditional Events
Commands may produce zero events if preconditions aren’t met:
ⓘ
async fn handle(&self, command: Self::Command) -> Result<Vec<Event>, Self::Error> {
match command {
UserCommand::MarkAsActive { id } => {
// If already active, no event needed
if self.is_active {
return Ok(vec![]);
}
Ok(vec![Event::new("User", &id, self.version + 1, "UserActivated", ...)])
}
}
}§Complex Validation
Use the aggregate state to enforce complex business rules:
ⓘ
async fn handle(&self, command: Self::Command) -> Result<Vec<Event>, Self::Error> {
match command {
AccountCommand::Withdraw { amount } => {
// Check balance
if self.balance < amount {
return Err(AccountError::InsufficientFunds);
}
// Check withdrawal limit
if self.daily_withdrawals + amount > self.daily_limit {
return Err(AccountError::DailyLimitExceeded);
}
Ok(vec![Event::new("Account", &self.id, self.version + 1, "Withdrawn", ...)])
}
}
}