cqrs-rust-lib

A pragmatic CQRS / Event Sourcing library for Rust with pluggable storage backends, structured domain errors, and REST integration.

Features

• Split Aggregate / CommandHandler traits (Single Responsibility)
• Structured domain errors — CqrsError + define_domain_errors! macro
• Pluggable storage backends: InMemory, MongoDB, PostgreSQL, SurrealDB
• Unified Query trait — auto-derives filter from struct fields (RSQL under the hood)
• HTTP Codex convention — CqrsHttpQuery<Q> extracts _q, skip/limit, page/page_size, sort from HTTP params
• RFC 9457 application/problem+json error responses (feature: problem-json)
• Backend prelude pattern — swap the entire backend with one use line
• REST routers with Axum and auto-generated OpenAPI/Swagger (feature: rest)
• Audit log router for event history
• Snapshot support
• WASM-compatible core (no Tokio in production deps)

Installation

[dependencies]
cqrs-rust-lib = { version = "0.7", features = ["postgres"] }

Feature flags

Quick Start

1. Define your domain

use cqrs_rust_lib::{Aggregate, CommandHandler, CqrsContext, CqrsError, Event};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum AccountEvent {
    Opened { owner: String },
    Deposited { amount: i64 },
    Withdrawn { amount: i64 },
}

impl Event for AccountEvent {
    fn event_type(&self) -> String {
        match self {
            Self::Opened { .. } => "opened".into(),
            Self::Deposited { .. } => "deposited".into(),
            Self::Withdrawn { .. } => "withdrawn".into(),
        }
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Account {
    pub id: String,
    pub balance: i64,
}

impl Aggregate for Account {
    const TYPE: &'static str = "account";
    type Event = AccountEvent;
    type Error = CqrsError;

    fn aggregate_id(&self) -> String { self.id.clone() }
    fn with_aggregate_id(mut self, id: String) -> Self { self.id = id; self }

    fn apply(&mut self, event: Self::Event) -> Result<(), Self::Error> {
        match event {
            AccountEvent::Opened { .. } => {}
            AccountEvent::Deposited { amount } => self.balance += amount,
            AccountEvent::Withdrawn { amount } => self.balance -= amount,
        }
        Ok(())
    }
}

impl CommandHandler for Account {
    type CreateCommand = CreateCommand;
    type UpdateCommand = UpdateCommand;
    type Services = ();

    async fn handle_create(&self, cmd: CreateCommand, _: &(), _: &CqrsContext)
        -> Result<Vec<AccountEvent>, CqrsError>
    {
        match cmd {
            CreateCommand::Open { owner } => Ok(vec![AccountEvent::Opened { owner }]),
        }
    }

    async fn handle_update(&self, cmd: UpdateCommand, _: &(), _: &CqrsContext)
        -> Result<Vec<AccountEvent>, CqrsError>
    {
        match cmd {
            UpdateCommand::Deposit { amount } => Ok(vec![AccountEvent::Deposited { amount }]),
            UpdateCommand::Withdraw { amount } => Ok(vec![AccountEvent::Withdrawn { amount }]),
        }
    }
}

2. Execute commands

use cqrs_rust_lib::es::{inmemory::InMemoryPersist, EventStoreImpl};
use cqrs_rust_lib::{CqrsCommandEngine, CqrsContext};

let store = EventStoreImpl::new(InMemoryPersist::<Account>::new());
let engine = CqrsCommandEngine::new(store, vec![], (), Box::new(|_e| {}));

let ctx = CqrsContext::default();
let id = engine.execute_create(CreateCommand::Open { owner: "Alice".into() }, &ctx).await?;
engine.execute_update(&id, UpdateCommand::Deposit { amount: 100 }, &ctx).await?;

Domain Error Codes

use cqrs_rust_lib::{define_domain_errors, CqrsError, CqrsErrorCode};
use http::StatusCode;

define_domain_errors! {
    domain: "account",
    prefix: 10,
    errors: {
        InsufficientFunds => (1, StatusCode::BAD_REQUEST, "INSUFFICIENT_FUNDS"),
        AccountClosed     => (3, StatusCode::GONE,        "ACCOUNT_CLOSED"),
    }
}

impl From<ErrorCode> for CqrsError {
    fn from(e: ErrorCode) -> Self { e.error(e.to_string()) }
}

Response shape (default):

{
  "domain": "account",
  "code": "ACCOUNT_INSUFFICIENT_FUNDS",
  "internalCode": 10001,
  "message": "Cannot withdraw 500, balance is 200",
  "requestId": "req-123"
}

CqrsError::from_status never degrades a status: any code without a dedicated GenericErrorCode variant keeps its value through GenericErrorCode::Other (GENERIC_HTTP_418, internal code 1418). 402, 405, 406, 408, 412, 413, 415, 422, 423, 428, 429, 501, 503 and 504 have dedicated variants whose internal code is 1000 + status.

RFC 9457 problem details (feature: problem-json)

With the problem-json feature the REST layer serves application/problem+json documents instead:

{
  "type": "urn:cqrs-error:account:ACCOUNT_INSUFFICIENT_FUNDS",
  "title": "ACCOUNT_INSUFFICIENT_FUNDS",
  "status": 400,
  "detail": "Cannot withdraw 500, balance is 200",
  "instance": "urn:cqrs-request:req-123",
  "domain": "account",
  "code": "ACCOUNT_INSUFFICIENT_FUNDS",
  "internalCode": 10001,
  "requestId": "req-123"
}

The type member defaults to urn:cqrs-error:{domain}:{code}. Point it at your own documentation with a base URI, or override it per error:

use cqrs_rust_lib::problem::set_problem_type_base_uri;

set_problem_type_base_uri("https://api.example.com/errors").unwrap();
// -> "type": "https://api.example.com/errors/ACCOUNT_INSUFFICIENT_FUNDS"

CqrsError::conflict("slug taken").with_type_uri("https://api.example.com/errors/slug");

CqrsError::to_problem() is available without the feature, for hand-rolled routes. See docs/migration_guide/problem_json.md.

Backend Preludes

Each backend exposes canonical type aliases under cqrs_rust_lib::prelude::<backend>. Swapping the backend requires changing a single import line — the rest of the wiring is identical.

// Change only this line to swap backends:
use cqrs_rust_lib::prelude::postgres as db;
// use cqrs_rust_lib::prelude::mongodb as db;
// use cqrs_rust_lib::prelude::surrealdb as db;

// Everything below stays the same:
let es = db::EventStorePersist::<MyAggregate>::new(connection.clone());
let repo = Arc::new(db::ReadStorage::<MyView, MyQuery>::new(connection.clone(), "my_view", ...));
// Reads the event store's own snapshot table, so it takes the table, not a view storage.
let snap = Arc::new(db::FromSnapshotStorage::<MyAggregate, MyQuery>::new(
    connection.clone(),
    es.snapshot_table_name(),
));

The connection setup (client, pool, URI) is necessarily backend-specific and stays outside the prelude.

FromSnapshotStorage reads the event store's snapshot table directly — its layout differs from a view table on every backend — and defaults to a mapper naming where the aggregate actually sits: data->>'field' on Postgres, data.field on SurrealDB, state.field on MongoDB. See docs/migration_guide/snapshot_read_storage.md.

Query Trait (Read Side)

Query is the unified read-side filter/pagination/sort interface. It requires Serialize (supertrait) so that equality filters are auto-derived from struct fields — no boilerplate needed in most cases.

use cqrs_rust_lib::read::Query;
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GameQuery {
    pub category: Option<String>,   // non-None → category == "value"
    pub available: Option<bool>,    // non-None → available == true/false
}

// Empty impl: filter auto-derived, no pagination override, no sort
impl Query for GameQuery {}

Override only what you need:

use cqrs_rust_lib::read::{Query, Sorter, SortDirection};
use cqrs_rust_lib::rsql::{Ast, Constraint, Operator, RestSql, Value};

impl Query for ProductQuery {
    // Custom filter: min_price uses >= instead of ==
    fn filter(&self) -> Option<RestSql> {
        self.min_price.and_then(|p| {
            RestSql::from_ast(Ast::Constraint(Constraint {
                field: "price".into(),
                operator: Operator::Gte,
                value: Value::Float(p),
            })).ok()
        })
    }

    // Static default sort — applied when no HTTP sort param is given
    fn default_sort() -> Option<Vec<Sorter>> {
        Some(vec![Sorter { field: "name".into(), direction: SortDirection::Asc }])
    }
}

HTTP Codex convention (feature: rest)

CqrsHttpQuery<Q> is an Axum extractor that adds _q (RSQL), pagination and sort on top of any typed Q. Use it directly with CQRSCodexReadRouter:

use cqrs_rust_lib::rest::{CQRSCodexReadRouter, CqrsHttpQuery};

// GET /games?_q=available==true&skip=20&limit=20&sort=-title
CQRSCodexReadRouter::<Game, GameView, GameQuery>::routes(storage, "games")

Filter priority: _q (RSQL) AND Q::filter() — combined. Sort priority: HTTP sort → Q::sort() → Q::default_sort().

The typed params of Q and the RSQL _q string are one set of filterable fields in two syntaxes — RSQL exists because a flat ?field=value cannot express >=, =in=, or or a range. So _q may only name fields of the query struct: a field not reachable as a query param has no reason to be reachable from _q. The set is derived from Q's Deserialize impl, so there is no second list to keep in step and every filterable field is a typed OpenAPI parameter by construction. A field the struct does not declare is rejected with 422 naming it; a query struct with no fields offers no filter at all.

#[derive(Debug, Clone, Serialize, Deserialize, IntoParams)]
pub struct GameQuery {
    pub category: Option<String>,
    pub title: Option<String>,   // filterable, therefore a field
}

impl Query for GameQuery {
    // Sorting gets its own list: ordering by a column of the view is reasonable
    // where filtering on it is not, so there is nothing to derive it from. Empty
    // (the default) means the view offers no sort at all.
    fn sortable_fields(&self) -> Vec<&str> {
        vec!["id", "title", "category"]
    }
}

Both constrain the caller, not Query::default_sort() — a view can order its own results while offering the caller no say. Note what this does and does not do: the fields are still returned in the response body, so it stops a listing being used as a lookup by an unoffered field — it does not hide it. See ADR-0002, ADR-0003 and docs/migration_guide/queryable_fields.md.

A query parameter the extractor cannot read is rejected with 422 Unprocessable Entity, not silently dropped: a _q that fails to parse (the response carries rest-sql's positioned error, caret included), and a skip/limit/page/page_size that is not a non-negative integer. An empty value — ?_q=&limit=10 — means the parameter is unset, not unreadable, and is accepted. See docs/migration_guide/codex_query_rejection.md.

A sort field name must be one or more .-separated segments matching [A-Za-z_][A-Za-z0-9_]* — the dot addresses a nested path on MongoDB and SurrealDB. The name is interpolated into the generated ORDER BY, never bound as a parameter, so anything else (a space, a quote, a hyphen, a non-ASCII letter) is rejected with 400 Validation failed naming the field. The check runs in the storage layer, so it applies to a Sorter built in Rust and handed to Storage::filter just as much as to the HTTP sort param. A view whose stored keys do not fit that grammar needs a FieldMapper translating a legal logical name to it.

Pagination accepts both vocabularies; skip/limit wins when both are present:

Paged<T> reports both forms, so skip/limit stay exact even when skip is not a multiple of limit:

{ "items": [], "total": 137, "skip": 25, "limit": 10, "page": 2, "pageSize": 10 }

Storage Backends

PostgreSQL

use cqrs_rust_lib::prelude::postgres as db;
use tokio_postgres::NoTls;

let (client, conn) = tokio_postgres::connect("postgres://user:pass@localhost/db", NoTls).await?;
tokio::spawn(async move { let _ = conn.await; });
let client = Arc::new(client);

client.batch_execute(&db::EventStorePersist::<Account>::schema()).await?;
let es = db::EventStorePersist::<Account>::from_client(client.clone());
let views = db::ReadStorage::<AccountView, AccountQuery>::new(client, "account", "account_view");

Connection pooling

Both the event store and the read storage acquire connections through the same PgPool trait. The default SharedClient wraps a single Arc<Client>; plug a real pool (deadpool-postgres, bb8, …) by implementing the two traits — no extra dependency is pulled into the library:

use cqrs_rust_lib::prelude::postgres::{PgConn, PgPool};
use cqrs_rust_lib::{cqrs_async_trait, CqrsError};

#[derive(Debug, Clone)]
struct DeadPool(deadpool_postgres::Pool);
struct DeadConn(deadpool_postgres::Object);

impl PgConn for DeadConn {
    fn client(&self) -> &tokio_postgres::Client { &self.0 }
}

cqrs_async_trait! {
impl PgPool for DeadPool {
    type Connection = DeadConn;
    async fn acquire(&self) -> Result<Self::Connection, CqrsError> {
        self.0.get().await.map(DeadConn).map_err(CqrsError::database_error)
    }
}
}

let es = db::EventStorePersist::<Account>::with_pool(pool.clone());
let views = db::ReadStorage::<AccountView, AccountQuery>::with_pool(pool, "account", "account_view");

MongoDB

use cqrs_rust_lib::prelude::mongodb as db;

let options = ClientOptions::parse(uri).await?;
let db_client = mongodb::Client::with_options(options.clone())?;
let database = db_client.database(&options.default_database.unwrap());

let es = db::EventStorePersist::<Account>::new(database.clone());

SurrealDB

use cqrs_rust_lib::prelude::surrealdb as db;
use surrealdb::engine::any::connect;

let surreal = connect(uri).await?;
surreal.use_ns("myns").use_db("mydb").await?;
surreal.query(db::EventStorePersist::<Game>::schema()).await?.check()?;

let es = db::EventStorePersist::<Game>::new(surreal.clone());

REST Routers (feature: rest)

use cqrs_rust_lib::rest::{CQRSWriteRouter, CQRSReadRouter, CQRSAuditLogRouter, CQRSCodexReadRouter};

// Standard router — typed query params only
CQRSReadRouter::routes(repository, Aggregate::TYPE)

// Codex router — adds _q, page, page_size, sort HTTP params
CQRSCodexReadRouter::<A, V, Q>::routes(storage, tag)

// Write + audit
CQRSWriteRouter::routes(engine)
CQRSAuditLogRouter::routes(event_store, tag)

See example/todolist/src/api.rs for complete wiring with Swagger UI.

Architecture

Aggregate (state + events)     CommandHandler (commands → events)
         \                       /
          CqrsCommandEngine ────── EventStore (persist)
                │                         │
           Dispatchers              Storage backends
          (projections)          (InMemory / PG / Mongo / Surreal)
                │
           ReadStorage ← Query (filter + sort + pagination)

Key Types

Examples

cargo run -p todolist    -- start --pg-uri="postgres://..." --http-port=8081
cargo run -p ludotheque  -- start --surreal-uri="ws://..." --http-port=8082

cargo test               # lib unit tests
cargo test -p todolist   # todolist integration tests
cargo test -p ludotheque # ludotheque integration tests

Migration Guides

• Aggregate / CommandHandler Split
• Domain Error Codes
• WASM Compatibility
• Query Trait (0.6 → 0.7)

License

MIT — see LICENSE.