Trait Aggregate 

pub trait Aggregate:
    Send
    + Sync
    + Default {
    type Command: Command;
    type Event;
    type Error: Error + Send + Sync + 'static;

    // Required methods
    fn aggregate_type() -> &'static str;
    fn version(&self) -> i64;
    fn handle<'life0, 'async_trait>(
        &'life0 self,
        command: Self::Command,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, Self::Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn apply(&mut self, event: &Event);

    // Provided methods
    fn from_events(events: Vec<Event>) -> Self { ... }
    fn to_snapshot(&self) -> Option<Value> { ... }
    fn from_snapshot(state: Value) -> Option<Self>
       where Self: Sized { ... }
}

Trait for domain aggregates in event sourcing.

Aggregates are consistency boundaries that encapsulate business logic, enforce invariants, and produce events in response to commands.

Lifecycle

1. Load: Reconstruct aggregate from event stream using from_events()
2. Command: Handle command with handle() to produce new events
3. Store: Persist events to event store (done by command bus)
4. Apply: Apply new events to update aggregate state using apply()
5. Publish: Publish events to event bus (done by command bus)

Design Principles

• State is private: Aggregate state should not be exposed outside
• Commands produce events: handle() validates and produces events
• Events update state: apply() updates internal state deterministically
• Pure functions: handle() has no side effects (no I/O, no mutations)
• Deterministic apply: Same events always produce same state
• Default implementation: Aggregate must implement Default for initial state

Type Parameters

• Command: The command type this aggregate handles (must implement Command)
• Event: The domain event type (typically an enum, must be serializable)
• Error: The error type for validation failures (must implement std::error::Error)

Example

See the module-level documentation for a complete example.

Required Associated Types

type Command: Command

The command type this aggregate handles

type Event

The domain event type (your custom enum)

type Error: Error + Send + Sync + 'static

The error type for validation failures

Required Methods

fn aggregate_type() -> &'static str

Get the aggregate type name (e.g., “User”, “Order”, “Account”).

This is used for event metadata and debugging. Should be a static string that uniquely identifies this aggregate type in your domain.

Example

fn aggregate_type() -> &'static str {
    "User"
}
assert_eq!(UserAggregate::aggregate_type(), "User");

fn version(&self) -> i64

Get the current version (sequence number) of this aggregate.

The version represents how many events have been applied to this aggregate. It starts at 0 for a new aggregate and increments with each event.

This is used for optimistic concurrency control - the event store checks that the version hasn’t changed since the aggregate was loaded.

Example

fn version(&self) -> i64 {
    self.version
}
let aggregate = UserAggregate::default();
assert_eq!(aggregate.version(), 0); // New aggregate

fn handle<'life0, 'async_trait>( &'life0 self, command: Self::Command, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>, Self::Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handle a command and produce events.

This is where your business logic lives. The method should:

1. Inspect current state (self) to make decisions
2. Validate the command against business rules
3. Return error if validation fails
4. Produce one or more events if validation succeeds
5. Return empty vec if command has no effect

Important: This method should have no side effects:

• Don’t write to databases
• Don’t call external APIs
• Don’t mutate state

Side effects happen in projections and event handlers.

Arguments

• command: The command to handle

Returns

• Ok(Vec<Event>): One or more events to persist and publish
• Err(Self::Error): Validation or business rule failure

Example

async fn handle(&self, command: Self::Command) -> Result<Vec<Event>, Self::Error> {
    match command {
        UserCommand::CreateUser { id, name, email } => {
            // Check invariant
            if self.created {
                return Err(UserError::AlreadyExists);
            }

            // Validate input
            if !email.contains('@') {
                return Err(UserError::InvalidEmail);
            }

            // Produce event
            Ok(vec![Event::new("User", &id, self.version + 1, "UserCreated", ...)])
        }
    }
}

fn apply(&mut self, event: &Event)

Apply an event to update aggregate state.

This method must be deterministic and side-effect free:

• Same events always produce same state
• No I/O operations
• No randomness
• No external dependencies

The event is already persisted when this is called. Your job is to update the aggregate’s internal state to reflect the event.

Arguments

• event: The event to apply (already persisted)

Example

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.created = true;
        }
        "ProfileUpdated" => {
            self.name = Some(event.payload["name"].as_str().unwrap().to_string());
        }
        _ => {} // Unknown event types are ignored
    }
}

Provided Methods

fn from_events(events: Vec<Event>) -> Self

Reconstruct aggregate from its event stream.

This method has a default implementation that:

1. Creates a new aggregate instance using Default::default()
2. Applies each event in order using apply()
3. Returns the reconstructed aggregate

You typically don’t need to override this unless you have special requirements.

Arguments

• events: The complete event stream for this aggregate

Returns

A fully reconstructed aggregate with all events applied

Example

// Load events from event store
let events = vec![
    Event::new("User", "user-123", 1, "UserCreated", serde_json::json!({"id": "user-123"})),
    Event::new("User", "user-123", 2, "ProfileUpdated", serde_json::json!({"name": "Alice"})),
];

// Reconstruct aggregate
let aggregate = UserAggregate::from_events(events);

assert_eq!(aggregate.version(), 2);
assert!(aggregate.created);

fn to_snapshot(&self) -> Option<Value>

Serialize current state for snapshotting.

Default returns None: an aggregate opts out and is always reconstructed by replaying its stream. Override to return Some(state) — typically via serde_json::to_value(self) — once the aggregate is Serialize.

fn from_snapshot(state: Value) -> Option<Self>

where Self: Sized,

Reconstruct an aggregate from a previously snapshotted state.

Default returns None so a loader falls back to stream replay. Override to deserialize the value produced by to_snapshot. Option (rather than Result) keeps the contract simple: a None — whether opted out or a failed decode — is always handled by replaying from sequence 0.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors