Crate eventcore

Expand description

§EventCore

Type-safe, multi-stream event sourcing for Rust with dynamic consistency boundaries.

EventCore lets a single command read from and atomically write to multiple event streams in one optimistic-concurrency-controlled transaction. You describe what a command does — which streams it touches, how it folds past events into state, and what new events it produces — and EventCore handles the infrastructure: loading state, detecting concurrent writes, retrying on conflict, and committing atomically.

APIs exposed only through feature flags whose names start with experimental- are disabled by default and are not covered by EventCore’s stable compatibility guarantee while they retain that prefix.

§Core concepts

Stream — an ordered, append-only sequence of events identified by a StreamId. Each stream has a version that increments with every append.
Command — a unit of business logic implementing CommandLogic. Its apply method folds events into command-local state (the write model); its handle method validates business rules and returns the new events to append. The streams a command may touch are declared with #[derive(Command)] and the #[stream] attribute.
execute — the canonical entry point. It loads the declared streams, folds them into state, calls handle, and atomically appends the resulting events with optimistic concurrency control, retrying per the supplied RetryPolicy.
Projection — a read model built by replaying events. The legacy, backend-neutral API implements Projector and uses run_projection. With the postgres feature, eventcore::postgres::projections adds a separate effect-plus-progress transactional runner without changing the legacy API. Read models and write models remain on separate code paths.

§Quick start: your first command

This example defines a Deposit command for a bank account, executes it against the in-memory store, and is fully runnable. Add eventcore and eventcore-memory to your Cargo.toml, then:

use eventcore::{
    execute, Command, CommandError, CommandLogic, Event, NewEvents, RetryPolicy, StreamId,
};
use eventcore_memory::InMemoryEventStore;
use serde::{Deserialize, Serialize};

// 1. Define your domain events. Each event knows which stream it belongs to.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
enum BankAccountEvent {
    MoneyDeposited { account_id: StreamId, amount: u32 },
}

impl Event for BankAccountEvent {
    fn stream_id(&self) -> &StreamId {
        match self {
            BankAccountEvent::MoneyDeposited { account_id, .. } => account_id,
        }
    }
    fn event_type_name() -> &'static str {
        "BankAccountEvent"
    }
}

// 2. Define a command. `#[derive(Command)]` generates the stream
//    declarations from the `#[stream]`-tagged fields.
#[derive(Command)]
struct Deposit {
    #[stream]
    account_id: StreamId,
    amount: u32,
}

// 3. Implement the business logic: how events fold into state (`apply`)
//    and what events the command produces (`handle`).
impl CommandLogic for Deposit {
    type Event = BankAccountEvent;
    type State = ();

    fn apply(&self, state: Self::State, _event: &Self::Event) -> Self::State {
        state
    }

    fn handle(&self, _state: Self::State) -> Result<NewEvents<Self::Event>, CommandError> {
        Ok(vec![BankAccountEvent::MoneyDeposited {
            account_id: self.account_id.clone(),
            amount: self.amount,
        }]
        .into())
    }
}

// 4. Execute the command against a store.
let rt = tokio::runtime::Runtime::new().expect("runtime");
rt.block_on(async {
    let store = InMemoryEventStore::new();
    let account_id = StreamId::try_new("account-42").expect("valid stream id");

    let command = Deposit { account_id, amount: 100 };
    execute(&store, command, RetryPolicy::new())
        .await
        .expect("deposit to succeed");
});

From here, see the user manual for projections, multi-stream atomicity, and backend selection, or the eventcore-demo crate for a complete bank application backed by PostgreSQL.

§Reading events directly

Most applications never read events by hand — execute does it for you. When you do need a stream’s raw history (for tooling or a projection), EventStore::read_stream returns a lazy EventStream, an async Stream of events. To materialize the whole history into a Vec, pass it to the collect_events helper.

§Backends

EventCore works with several EventStore implementations:

eventcore-memory — a separate zero-dependency crate added directly to your Cargo.toml (used in the quick start above) for tests and examples.
postgres feature — PostgreSQL backend with ACID transactions, re-exported as eventcore::postgres.
sqlite feature — SQLite backend with optional SQLCipher encryption, re-exported as eventcore::sqlite.

§Error handling

execute returns Ok(ExecutionResponse) on success; the response exposes ExecutionResponse::attempts so callers can observe how many retries occurred.
CommandError is returned by execute on failure — business-rule violations, concurrency conflicts (after retries are exhausted), and store failures.
EventStoreError (in eventcore-types) is returned by backend operations, including version conflicts and event deserialization failures.

Macros§

require: Validates a business rule condition and returns early with a CommandError when the condition is false.

Structs§

AttemptNumber: Attempt number for retry operations (1-based).
CommandStateReplayCheckpoint: A command-state projection checkpoint after one stream was fully replayed.
CommandStateSnapshot: A durable command-state read-model projection.
CommandStateSnapshotId: Stable identifier for a command-state projection.
DelayMilliseconds: Delay in milliseconds for retry or backoff operations.
EventStream: An async stream of events read from a single stream, generic over the consumer’s event payload type.
ExecutionResponse: The result of a successful execute call.
FailureContext: Context provided to error handler when event processing fails.
NewEvents: Collection of new events produced by a command.
ProjectionConfig: Configuration for running projections via run_projection.
RetryContext: Context information passed to metrics hooks during retry lifecycle.
RetryPolicy: Configuration for automatic retry behavior on concurrency conflicts.
StreamDeclarations
StreamId: Stream identifier domain type.
StreamPosition: Opaque cursor representing a location in an EventReader delivery sequence.

Enums§

BackoffStrategy: Defines the delay strategy between retry attempts.
CommandError: Error type for command execution failures.
FailureStrategy: Strategy for handling event processing failures.
ProjectionError: Error type for projection operations.

Constants§

COMMAND_STATE_SNAPSHOT_REFRESH_THRESHOLD: Number of reconstructed events at which an opt-in command-state projection is first persisted.

Traits§

CommandLogic: Trait defining the business logic of a command.
CommandStreams: Infrastructure trait describing the streams required to execute a command.
Event: Event trait for domain-first event sourcing.
MetricsHook: Callback trait for integrating with metrics systems during retry lifecycle.
Projector: Trait for transforming events into read model updates.
StreamResolver: Trait for runtime stream discovery when static declarations are insufficient.

Functions§

collect_events: Collect every event from a stream into a Vec, in stream-version order.
execute: Execute a command against the event store with a custom retry policy.
run_projection: Runs a projector against a backend that provides events, checkpoints, and coordination.

Derive Macros§

Command: Entry point for #[derive(Command)].

Crate eventcore

Crate eventcore Copy item path

§EventCore

§Core concepts

§Quick start: your first command

§Reading events directly

§Backends

§Error handling

Macros§

Structs§

Enums§

Constants§

Traits§

Functions§

Derive Macros§

Crate eventcore