CQRS and Event Sourcing

Command Query Responsibility Segregation (CQRS) gives writes and reads separate models. Event sourcing stores every change as an event instead of only the latest state. The patterns are independent and often combined. This lesson builds an event-sourced Storefront order service with a Postgres event store, a projection, and query endpoints.

Project setup

Start PostgreSQL:

Terminal

docker run --name events-pg -e POSTGRES_PASSWORD=pw -e POSTGRES_DB=events -p 5432:5432 -d postgres:16

Cargo.toml

TOML

[package]
name = "order-events"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
tokio = { version = "1", features = ["full"] }
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "json", "uuid"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["serde"] }
thiserror = "2"

migrations/0001_events.sql

SQL

CREATE TABLE IF NOT EXISTS events (
    seq          BIGSERIAL UNIQUE,
    aggregate_id UUID        NOT NULL,
    version      INT         NOT NULL,
    event_type   TEXT        NOT NULL,
    payload      JSONB       NOT NULL,
    recorded_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (aggregate_id, version)
);

CREATE TABLE IF NOT EXISTS snapshots (
    aggregate_id UUID PRIMARY KEY,
    version      INT   NOT NULL,
    state        JSONB NOT NULL
);

CREATE TABLE IF NOT EXISTS order_summaries (
    id          UUID PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    status      TEXT   NOT NULL,
    line_count  INT    NOT NULL,
    total_cents BIGINT NOT NULL,
    version     INT    NOT NULL
);

CREATE TABLE IF NOT EXISTS projection_checkpoints (
    name     TEXT PRIMARY KEY,
    last_seq BIGINT NOT NULL
);

INSERT INTO projection_checkpoints (name, last_seq)
VALUES ('order_summaries', 0)
ON CONFLICT DO NOTHING;

events is the append-only store. The primary key (aggregate_id, version) implements optimistic concurrency. seq is a global order across all aggregates.
snapshots caches an aggregate state at a version.
order_summaries is the read model. projection_checkpoints records how far the projector has read.

CQRS in one sentence

A command is a request to change state and can be rejected.
A query reads state and changes nothing.
An event is a fact that happened, named in the past tense.
An aggregate is a group of objects changed as one unit that protects its own rules (here, one order).
A projection is a read model built by applying events in order.
A snapshot is a saved aggregate state at a given version.

CQRS can use one database (separate handlers over the same tables) or a separate read store. A separate read store gives eventual consistency: after a command succeeds, the read model updates a moment later.

Example 1: The aggregate as a fold over events

The state of an order is the result of applying its events in order. apply records facts and never checks rules. Rules live in the decide_* methods, which return new events or an error. A repeated command returns no events, so a retry is harmless.

src/domain.rs

Rust

use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum OrderEvent {
    Placed { customer_id: i64 },
    LineAdded { sku: String, qty: u32, unit_cents: i64 },
    Paid,
    Cancelled,
}

impl OrderEvent {
    pub fn name(&self) -> &'static str {
        match self {
            OrderEvent::Placed { .. } => "Placed",
            OrderEvent::LineAdded { .. } => "LineAdded",
            OrderEvent::Paid => "Paid",
            OrderEvent::Cancelled => "Cancelled",
        }
    }
}

#[derive(Debug, PartialEq, thiserror::Error)]
pub enum OrderError {
    #[error("order has not been placed")]
    NotPlaced,
    #[error("order is already placed")]
    AlreadyPlaced,
    #[error("order is already paid")]
    AlreadyPaid,
    #[error("order is cancelled")]
    Cancelled,
    #[error("order has no lines")]
    EmptyOrder,
    #[error("quantity must be greater than zero")]
    InvalidQuantity,
}

#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct Order {
    pub placed: bool,
    pub customer_id: i64,
    pub lines: Vec<(String, u32, i64)>,
    pub paid: bool,
    pub cancelled: bool,
}

impl Order {
    /// Records a fact. It never fails and never checks rules.
    pub fn apply(&mut self, event: &OrderEvent) {
        match event {
            OrderEvent::Placed { customer_id } => {
                self.placed = true;
                self.customer_id = *customer_id;
            }
            OrderEvent::LineAdded { sku, qty, unit_cents } => {
                self.lines.push((sku.clone(), *qty, *unit_cents));
            }
            OrderEvent::Paid => self.paid = true,
            OrderEvent::Cancelled => self.cancelled = true,
        }
    }

    pub fn total_cents(&self) -> i64 {
        self.lines.iter().map(|(_, qty, cents)| *qty as i64 * cents).sum()
    }

    pub fn decide_place(&self, customer_id: i64) -> Result<Vec<OrderEvent>, OrderError> {
        if self.placed {
            return Err(OrderError::AlreadyPlaced);
        }
        Ok(vec![OrderEvent::Placed { customer_id }])
    }

    pub fn decide_add_line(
        &self,
        sku: String,
        qty: u32,
        unit_cents: i64,
    ) -> Result<Vec<OrderEvent>, OrderError> {
        self.require_open()?;
        if qty == 0 {
            return Err(OrderError::InvalidQuantity);
        }
        Ok(vec![OrderEvent::LineAdded { sku, qty, unit_cents }])
    }

    pub fn decide_pay(&self) -> Result<Vec<OrderEvent>, OrderError> {
        if !self.placed {
            return Err(OrderError::NotPlaced);
        }
        if self.cancelled {
            return Err(OrderError::Cancelled);
        }
        if self.paid {
            return Ok(vec![]); // repeating the command changes nothing
        }
        if self.lines.is_empty() {
            return Err(OrderError::EmptyOrder);
        }
        Ok(vec![OrderEvent::Paid])
    }

    pub fn decide_cancel(&self) -> Result<Vec<OrderEvent>, OrderError> {
        if !self.placed {
            return Err(OrderError::NotPlaced);
        }
        if self.cancelled {
            return Ok(vec![]);
        }
        if self.paid {
            return Err(OrderError::AlreadyPaid);
        }
        Ok(vec![OrderEvent::Cancelled])
    }

    fn require_open(&self) -> Result<(), OrderError> {
        if !self.placed {
            Err(OrderError::NotPlaced)
        } else if self.cancelled {
            Err(OrderError::Cancelled)
        } else if self.paid {
            Err(OrderError::AlreadyPaid)
        } else {
            Ok(())
        }
    }
}

src/main.rs (demo for this step)

Rust

mod domain;

use domain::{Order, OrderEvent};

fn apply_all(order: &mut Order, events: &[OrderEvent]) {
    for event in events {
        order.apply(event);
    }
}

fn main() {
    let mut order = Order::default();

    let events = order.decide_place(7).unwrap();
    apply_all(&mut order, &events);
    let events = order.decide_add_line("notebook".to_string(), 2, 450).unwrap();
    apply_all(&mut order, &events);
    let events = order.decide_add_line("pen".to_string(), 1, 120).unwrap();
    apply_all(&mut order, &events);
    let events = order.decide_pay().unwrap();
    apply_all(&mut order, &events);

    println!("customer {} total {} paid {}", order.customer_id, order.total_cents(), order.paid);
    println!("{:?}", order.decide_cancel());
    println!("{}", order.decide_pay().unwrap().len());
}

Output

customer 7 total 1020 paid true
Err(AlreadyPaid)
0

Example 2: The event store and snapshots

The store appends events with expected_version. Two writers that loaded the same version both try to insert the next one, and the primary key lets only one succeed. The loser gets VersionConflict and reloads and retries, or returns 409. Events are never updated or deleted. A mistake is fixed by a new event.

decode upcasts old events. Version 1 of LineAdded stored price in whole currency units. Version 2 stores unit_cents. Domain code sees only the newest shape.
Other schema changes: an optional field gets a serde default, and a renamed field accepts both names with #[serde(alias = "...")]. A removed event type is still read and no longer written.
Personal data in immutable events conflicts with deletion duties. One approach encrypts personal fields with a per-person key and deletes the key.

src/store.rs (full file in Complete example)

append inserts each event with the next version in one transaction. A unique violation on (aggregate_id, version) becomes StoreError::VersionConflict.
load_from reads the events after a given version and applies them to an Order.
decode renames a version 1 LineAdded price field to unit_cents before deserializing.

Replaying many events on every command costs time. A snapshot stores the aggregate at some version, and loading reads the snapshot and only the later events. Snapshots are a cache: the events remain the truth, and a snapshot can be deleted and rebuilt. Short-lived aggregates need none.

src/snapshot.rs (full file in Complete example)

load reads the snapshot row, if any, and passes it with its version to store::load_from, so only later events are read.
maybe_save upserts the snapshot when the version crosses a multiple of 50.

Example 3: The command side

Each handler loads the aggregate, calls a decide_* method, appends the events, and returns the new version. Order ids are chosen by the client (PUT /orders/{id}), so creation can be retried.

src/error.rs (full file in Complete example)

AppError(StatusCode, String) renders as {"error": message}.
From<OrderError>: NotPlaced becomes 404, InvalidQuantity becomes 422, every other rule violation becomes 409.
From<StoreError>: VersionConflict becomes 409. Other errors are logged and become 500.

src/commands.rs (full file in Complete example)

execute loads the aggregate with snapshot::load, runs the decide_* closure, appends the events when there are any, applies them, calls snapshot::maybe_save, and returns {"order_id", "version"}.
place (PUT /orders/{id}, 201), add_line, pay, and cancel (POST) each call execute with one decide_* method.
Rule violations map to 409, 404, or 422 through From<OrderError>.
The command response carries the new version. The client uses it to wait on the read side.

Example 4: Projections and the query side

The projector reads events after its checkpoint, updates order_summaries, and moves the checkpoint in the same transaction. A crash cannot apply an event twice or skip one. FOR UPDATE on the checkpoint row keeps a second projector instance from running the same batch.

New read needs mean new projections built from history: revenue per day, a search index.
seq values can be assigned to a transaction that commits after a later one, so a projector that reads by seq can skip a late-committing row. Production systems read only rows older than the oldest open transaction, or assign seq in commit order.

src/projector.rs (full file in Complete example)

project_batch opens a transaction, locks the checkpoint row with FOR UPDATE, reads up to 500 events after it, and applies each through decode.
Placed inserts a summary row. LineAdded adds one line and qty * unit_cents to the total. Paid and Cancelled set the status. Every branch also stores the event version.
The checkpoint moves to the last applied seq before commit.
rebuild rewinds the checkpoint to 0 and deletes all summaries in one transaction.
run loops: it sleeps 200 ms when a batch is empty and 1 second after an error.

The query handlers read only the read model. GET /orders/{id}/summary?min_version=N waits up to two seconds for the projection to reach version N, which handles eventual consistency for a client that just wrote.

src/queries.rs (full file in Complete example)

Summary maps one order_summaries row (sqlx::FromRow) and serializes as the response body.
get_summary polls the read model every 100 ms, at most 20 times. It returns 404 when the row is missing and min_version is 0, and 503 when the version is not reached in time.
list_summaries returns all rows ordered by id.
rebuild calls projector::rebuild.

Example 5: Replay and lag

Replay rebuilds a read model by running the projector from sequence zero. It fixes a projection bug and gives a new projection its history. POST /admin/rebuild calls projector::rebuild, which empties the table and rewinds the checkpoint, and the running projector reprocesses every event. The read model is incomplete while replay runs. A production system replays into a new table and switches reads when it catches up.

Responses to eventual consistency:

Return the changed resource, or its version, from the command.
Accept min_version on queries, as get_summary does.
Show a pending state until the projection catches up.
Expose lag as a metric: the newest seq in events minus last_seq in the checkpoint.

When to use these patterns

Use when	Avoid when
History is a requirement (audit, finance, undo)	The service is create, read, update, delete on simple records
Several read shapes serve one write model	One model serves reads and writes well
Read and write loads scale differently	The team cannot maintain projections, replay, and versioning
Business rules are complex and stateful	Immediate read-your-writes is required everywhere

CQRS without event sourcing is the smaller change. In the Storefront, orders and payments have audit needs, and the product catalog stays plain CRUD. Events reach other services through the outbox from Sagas and Distributed Transactions.

Complete example

File tree:

Text

order-events/
  Cargo.toml
  migrations/0001_events.sql
  src/main.rs
  src/domain.rs
  src/store.rs
  src/snapshot.rs
  src/error.rs
  src/commands.rs
  src/projector.rs
  src/queries.rs

Cargo.toml

TOML

[package]
name = "order-events"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
tokio = { version = "1", features = ["full"] }
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "json", "uuid"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["serde"] }
thiserror = "2"

migrations/0001_events.sql

SQL

CREATE TABLE IF NOT EXISTS events (
    seq          BIGSERIAL UNIQUE,
    aggregate_id UUID        NOT NULL,
    version      INT         NOT NULL,
    event_type   TEXT        NOT NULL,
    payload      JSONB       NOT NULL,
    recorded_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (aggregate_id, version)
);

CREATE TABLE IF NOT EXISTS snapshots (
    aggregate_id UUID PRIMARY KEY,
    version      INT   NOT NULL,
    state        JSONB NOT NULL
);

CREATE TABLE IF NOT EXISTS order_summaries (
    id          UUID PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    status      TEXT   NOT NULL,
    line_count  INT    NOT NULL,
    total_cents BIGINT NOT NULL,
    version     INT    NOT NULL
);

CREATE TABLE IF NOT EXISTS projection_checkpoints (
    name     TEXT PRIMARY KEY,
    last_seq BIGINT NOT NULL
);

INSERT INTO projection_checkpoints (name, last_seq)
VALUES ('order_summaries', 0)
ON CONFLICT DO NOTHING;

src/main.rs

Rust

mod commands;
mod domain;
mod error;
mod projector;
mod queries;
mod snapshot;
mod store;

use axum::{
    routing::{get, post, put},
    Router,
};
use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://postgres:pw@localhost:5432/events")
        .await?;
    sqlx::raw_sql(include_str!("../migrations/0001_events.sql")).execute(&pool).await?;

    tokio::spawn(projector::run(pool.clone()));

    let app = Router::new()
        .route("/orders/{id}", put(commands::place))
        .route("/orders/{id}/lines", post(commands::add_line))
        .route("/orders/{id}/pay", post(commands::pay))
        .route("/orders/{id}/cancel", post(commands::cancel))
        .route("/orders/{id}/summary", get(queries::get_summary))
        .route("/orders", get(queries::list_summaries))
        .route("/admin/rebuild", post(queries::rebuild))
        .with_state(pool);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

src/domain.rs

Rust

use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum OrderEvent {
    Placed { customer_id: i64 },
    LineAdded { sku: String, qty: u32, unit_cents: i64 },
    Paid,
    Cancelled,
}

impl OrderEvent {
    pub fn name(&self) -> &'static str {
        match self {
            OrderEvent::Placed { .. } => "Placed",
            OrderEvent::LineAdded { .. } => "LineAdded",
            OrderEvent::Paid => "Paid",
            OrderEvent::Cancelled => "Cancelled",
        }
    }
}

#[derive(Debug, PartialEq, thiserror::Error)]
pub enum OrderError {
    #[error("order has not been placed")]
    NotPlaced,
    #[error("order is already placed")]
    AlreadyPlaced,
    #[error("order is already paid")]
    AlreadyPaid,
    #[error("order is cancelled")]
    Cancelled,
    #[error("order has no lines")]
    EmptyOrder,
    #[error("quantity must be greater than zero")]
    InvalidQuantity,
}

#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct Order {
    pub placed: bool,
    pub customer_id: i64,
    pub lines: Vec<(String, u32, i64)>,
    pub paid: bool,
    pub cancelled: bool,
}

impl Order {
    /// Records a fact. It never fails and never checks rules.
    pub fn apply(&mut self, event: &OrderEvent) {
        match event {
            OrderEvent::Placed { customer_id } => {
                self.placed = true;
                self.customer_id = *customer_id;
            }
            OrderEvent::LineAdded { sku, qty, unit_cents } => {
                self.lines.push((sku.clone(), *qty, *unit_cents));
            }
            OrderEvent::Paid => self.paid = true,
            OrderEvent::Cancelled => self.cancelled = true,
        }
    }

    pub fn total_cents(&self) -> i64 {
        self.lines.iter().map(|(_, qty, cents)| *qty as i64 * cents).sum()
    }

    pub fn decide_place(&self, customer_id: i64) -> Result<Vec<OrderEvent>, OrderError> {
        if self.placed {
            return Err(OrderError::AlreadyPlaced);
        }
        Ok(vec![OrderEvent::Placed { customer_id }])
    }

    pub fn decide_add_line(
        &self,
        sku: String,
        qty: u32,
        unit_cents: i64,
    ) -> Result<Vec<OrderEvent>, OrderError> {
        self.require_open()?;
        if qty == 0 {
            return Err(OrderError::InvalidQuantity);
        }
        Ok(vec![OrderEvent::LineAdded { sku, qty, unit_cents }])
    }

    pub fn decide_pay(&self) -> Result<Vec<OrderEvent>, OrderError> {
        if !self.placed {
            return Err(OrderError::NotPlaced);
        }
        if self.cancelled {
            return Err(OrderError::Cancelled);
        }
        if self.paid {
            return Ok(vec![]); // repeating the command changes nothing
        }
        if self.lines.is_empty() {
            return Err(OrderError::EmptyOrder);
        }
        Ok(vec![OrderEvent::Paid])
    }

    pub fn decide_cancel(&self) -> Result<Vec<OrderEvent>, OrderError> {
        if !self.placed {
            return Err(OrderError::NotPlaced);
        }
        if self.cancelled {
            return Ok(vec![]);
        }
        if self.paid {
            return Err(OrderError::AlreadyPaid);
        }
        Ok(vec![OrderEvent::Cancelled])
    }

    fn require_open(&self) -> Result<(), OrderError> {
        if !self.placed {
            Err(OrderError::NotPlaced)
        } else if self.cancelled {
            Err(OrderError::Cancelled)
        } else if self.paid {
            Err(OrderError::AlreadyPaid)
        } else {
            Ok(())
        }
    }
}

src/store.rs

Rust

use serde_json::Value;
use sqlx::PgPool;
use uuid::Uuid;

use crate::domain::{Order, OrderEvent};

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    #[error("version conflict")]
    VersionConflict,
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),
    #[error("bad event payload: {0}")]
    Decode(#[from] serde_json::Error),
}

/// Turns a stored payload into the current event shape.
/// Version 1 of `LineAdded` stored `price` in whole currency units.
pub fn decode(mut payload: Value) -> Result<OrderEvent, serde_json::Error> {
    if let Some(obj) = payload.as_object_mut() {
        if obj.get("type").and_then(|t| t.as_str()) == Some("LineAdded") {
            if let Some(price) = obj.remove("price").and_then(|p| p.as_i64()) {
                obj.insert("unit_cents".to_string(), serde_json::json!(price * 100));
            }
        }
    }
    serde_json::from_value(payload)
}

/// Appends events after `expected_version`. Returns the new version.
pub async fn append(
    pool: &PgPool,
    aggregate_id: Uuid,
    expected_version: i32,
    events: &[OrderEvent],
) -> Result<i32, StoreError> {
    let mut tx = pool.begin().await?;
    let mut version = expected_version;
    for event in events {
        version += 1;
        let result = sqlx::query(
            "INSERT INTO events (aggregate_id, version, event_type, payload)
             VALUES ($1, $2, $3, $4)",
        )
        .bind(aggregate_id)
        .bind(version)
        .bind(event.name())
        .bind(serde_json::to_value(event)?)
        .execute(&mut *tx)
        .await;

        match result {
            Err(sqlx::Error::Database(db)) if db.is_unique_violation() => {
                return Err(StoreError::VersionConflict);
            }
            Err(e) => return Err(e.into()),
            Ok(_) => {}
        }
    }
    tx.commit().await?;
    Ok(version)
}

/// Applies the events after `from_version` to `order`.
pub async fn load_from(
    pool: &PgPool,
    aggregate_id: Uuid,
    mut order: Order,
    from_version: i32,
) -> Result<(Order, i32), StoreError> {
    let rows: Vec<(i32, Value)> = sqlx::query_as(
        "SELECT version, payload FROM events
         WHERE aggregate_id = $1 AND version > $2 ORDER BY version",
    )
    .bind(aggregate_id)
    .bind(from_version)
    .fetch_all(pool)
    .await?;

    let mut version = from_version;
    for (v, payload) in rows {
        order.apply(&decode(payload)?);
        version = v;
    }
    Ok((order, version))
}

src/snapshot.rs

Rust

use sqlx::PgPool;
use uuid::Uuid;

use crate::domain::Order;
use crate::store::{self, StoreError};

const SNAPSHOT_EVERY: i32 = 50;

/// Loads the latest snapshot, then only the events after it.
pub async fn load(pool: &PgPool, id: Uuid) -> Result<(Order, i32), StoreError> {
    let snapshot: Option<(i32, serde_json::Value)> =
        sqlx::query_as("SELECT version, state FROM snapshots WHERE aggregate_id = $1")
            .bind(id)
            .fetch_optional(pool)
            .await?;

    match snapshot {
        Some((version, state)) => {
            let order: Order = serde_json::from_value(state)?;
            store::load_from(pool, id, order, version).await
        }
        None => store::load_from(pool, id, Order::default(), 0).await,
    }
}

/// Saves a snapshot when the version crosses a multiple of SNAPSHOT_EVERY.
pub async fn maybe_save(
    pool: &PgPool,
    id: Uuid,
    old_version: i32,
    new_version: i32,
    order: &Order,
) -> Result<(), StoreError> {
    if new_version / SNAPSHOT_EVERY == old_version / SNAPSHOT_EVERY {
        return Ok(());
    }
    sqlx::query(
        "INSERT INTO snapshots (aggregate_id, version, state) VALUES ($1, $2, $3)
         ON CONFLICT (aggregate_id)
         DO UPDATE SET version = EXCLUDED.version, state = EXCLUDED.state",
    )
    .bind(id)
    .bind(new_version)
    .bind(serde_json::to_value(order)?)
    .execute(pool)
    .await?;
    Ok(())
}

src/error.rs

Rust

use axum::{http::StatusCode, response::{IntoResponse, Response}, Json};
use serde_json::json;

use crate::domain::OrderError;
use crate::store::StoreError;

pub struct AppError(pub StatusCode, pub String);

impl From<OrderError> for AppError {
    fn from(e: OrderError) -> Self {
        let status = match e {
            OrderError::NotPlaced => StatusCode::NOT_FOUND,
            OrderError::InvalidQuantity => StatusCode::UNPROCESSABLE_ENTITY,
            _ => StatusCode::CONFLICT,
        };
        AppError(status, e.to_string())
    }
}

impl From<StoreError> for AppError {
    fn from(e: StoreError) -> Self {
        match e {
            StoreError::VersionConflict => {
                AppError(StatusCode::CONFLICT, "version conflict, retry the command".to_string())
            }
            other => {
                eprintln!("store error: {other}");
                AppError(StatusCode::INTERNAL_SERVER_ERROR, "internal error".to_string())
            }
        }
    }
}

impl From<sqlx::Error> for AppError {
    fn from(e: sqlx::Error) -> Self {
        eprintln!("database error: {e}");
        AppError(StatusCode::INTERNAL_SERVER_ERROR, "internal error".to_string())
    }
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        (self.0, Json(json!({ "error": self.1 }))).into_response()
    }
}

src/commands.rs

Rust

use axum::{
    extract::{Path, State},
    http::StatusCode,
    Json,
};
use serde::Deserialize;
use serde_json::{json, Value};
use sqlx::PgPool;
use uuid::Uuid;

use crate::domain::{Order, OrderError, OrderEvent};
use crate::error::AppError;
use crate::{snapshot, store};

#[derive(Deserialize)]
pub struct PlaceOrder {
    pub customer_id: i64,
}

#[derive(Deserialize)]
pub struct AddLine {
    pub sku: String,
    pub qty: u32,
    pub unit_cents: i64,
}

/// Load the aggregate, decide, append, and snapshot when due.
async fn execute(
    pool: &PgPool,
    id: Uuid,
    decide: impl FnOnce(&Order) -> Result<Vec<OrderEvent>, OrderError>,
) -> Result<Value, AppError> {
    let (mut order, version) = snapshot::load(pool, id).await?;
    let events = decide(&order)?;
    let new_version = if events.is_empty() {
        version
    } else {
        store::append(pool, id, version, &events).await?
    };
    for event in &events {
        order.apply(event);
    }
    snapshot::maybe_save(pool, id, version, new_version, &order).await?;
    Ok(json!({ "order_id": id, "version": new_version }))
}

pub async fn place(
    State(pool): State<PgPool>,
    Path(id): Path<Uuid>,
    Json(body): Json<PlaceOrder>,
) -> Result<(StatusCode, Json<Value>), AppError> {
    let result = execute(&pool, id, |o| o.decide_place(body.customer_id)).await?;
    Ok((StatusCode::CREATED, Json(result)))
}

pub async fn add_line(
    State(pool): State<PgPool>,
    Path(id): Path<Uuid>,
    Json(body): Json<AddLine>,
) -> Result<Json<Value>, AppError> {
    let result = execute(&pool, id, |o| o.decide_add_line(body.sku, body.qty, body.unit_cents)).await?;
    Ok(Json(result))
}

pub async fn pay(
    State(pool): State<PgPool>,
    Path(id): Path<Uuid>,
) -> Result<Json<Value>, AppError> {
    Ok(Json(execute(&pool, id, |o| o.decide_pay()).await?))
}

pub async fn cancel(
    State(pool): State<PgPool>,
    Path(id): Path<Uuid>,
) -> Result<Json<Value>, AppError> {
    Ok(Json(execute(&pool, id, |o| o.decide_cancel()).await?))
}

src/projector.rs

Rust

use std::time::Duration;

use serde_json::Value;
use sqlx::PgPool;
use uuid::Uuid;

use crate::domain::OrderEvent;
use crate::store::{decode, StoreError};

/// Applies up to 500 new events to `order_summaries`.
/// The checkpoint moves in the same transaction as the read-model rows.
pub async fn project_batch(pool: &PgPool) -> Result<usize, StoreError> {
    let mut tx = pool.begin().await?;

    let checkpoint: i64 = sqlx::query_scalar(
        "SELECT last_seq FROM projection_checkpoints WHERE name = 'order_summaries' FOR UPDATE",
    )
    .fetch_one(&mut *tx)
    .await?;

    let rows: Vec<(i64, Uuid, i32, Value)> = sqlx::query_as(
        "SELECT seq, aggregate_id, version, payload FROM events
         WHERE seq > $1 ORDER BY seq LIMIT 500",
    )
    .bind(checkpoint)
    .fetch_all(&mut *tx)
    .await?;

    let mut last = checkpoint;
    for (seq, id, version, payload) in &rows {
        match decode(payload.clone())? {
            OrderEvent::Placed { customer_id } => {
                sqlx::query(
                    "INSERT INTO order_summaries (id, customer_id, status, line_count, total_cents, version)
                     VALUES ($1, $2, 'placed', 0, 0, $3)
                     ON CONFLICT (id) DO NOTHING",
                )
                .bind(id)
                .bind(customer_id)
                .bind(version)
                .execute(&mut *tx)
                .await?;
            }
            OrderEvent::LineAdded { qty, unit_cents, .. } => {
                sqlx::query(
                    "UPDATE order_summaries
                     SET line_count = line_count + 1, total_cents = total_cents + $2, version = $3
                     WHERE id = $1",
                )
                .bind(id)
                .bind(qty as i64 * unit_cents)
                .bind(version)
                .execute(&mut *tx)
                .await?;
            }
            OrderEvent::Paid => set_status(&mut tx, *id, "paid", *version).await?,
            OrderEvent::Cancelled => set_status(&mut tx, *id, "cancelled", *version).await?,
        }
        last = *seq;
    }

    if last != checkpoint {
        sqlx::query("UPDATE projection_checkpoints SET last_seq = $1 WHERE name = 'order_summaries'")
            .bind(last)
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    Ok(rows.len())
}

async fn set_status(
    tx: &mut sqlx::PgConnection,
    id: Uuid,
    status: &str,
    version: i32,
) -> Result<(), sqlx::Error> {
    sqlx::query("UPDATE order_summaries SET status = $2, version = $3 WHERE id = $1")
        .bind(id)
        .bind(status)
        .bind(version)
        .execute(tx)
        .await?;
    Ok(())
}

/// Empties the read model and rewinds the checkpoint. The projector then replays history.
pub async fn rebuild(pool: &PgPool) -> Result<(), sqlx::Error> {
    let mut tx = pool.begin().await?;
    sqlx::query("UPDATE projection_checkpoints SET last_seq = 0 WHERE name = 'order_summaries'")
        .execute(&mut *tx)
        .await?;
    sqlx::query("DELETE FROM order_summaries").execute(&mut *tx).await?;
    tx.commit().await
}

pub async fn run(pool: PgPool) {
    loop {
        match project_batch(&pool).await {
            Ok(0) => tokio::time::sleep(Duration::from_millis(200)).await,
            Ok(_) => {}
            Err(e) => {
                eprintln!("projector error: {e}");
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}

src/queries.rs

Rust

use std::time::Duration;

use axum::{
    extract::{Path, Query, State},
    http::StatusCode,
    Json,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sqlx::PgPool;
use uuid::Uuid;

use crate::error::AppError;
use crate::projector;

#[derive(Serialize, sqlx::FromRow)]
pub struct Summary {
    pub id: Uuid,
    pub customer_id: i64,
    pub status: String,
    pub line_count: i32,
    pub total_cents: i64,
    pub version: i32,
}

#[derive(Deserialize)]
pub struct SummaryQuery {
    pub min_version: Option<i32>,
}

const COLUMNS: &str = "id, customer_id, status, line_count, total_cents, version";

/// Waits up to two seconds for the projection to reach `min_version`.
pub async fn get_summary(
    State(pool): State<PgPool>,
    Path(id): Path<Uuid>,
    Query(q): Query<SummaryQuery>,
) -> Result<Json<Summary>, AppError> {
    let min_version = q.min_version.unwrap_or(0);
    for _ in 0..20 {
        let row: Option<Summary> =
            sqlx::query_as(&format!("SELECT {COLUMNS} FROM order_summaries WHERE id = $1"))
                .bind(id)
                .fetch_optional(&pool)
                .await?;
        match row {
            Some(s) if s.version >= min_version => return Ok(Json(s)),
            None if min_version == 0 => {
                return Err(AppError(StatusCode::NOT_FOUND, "order not found".to_string()));
            }
            _ => tokio::time::sleep(Duration::from_millis(100)).await,
        }
    }
    Err(AppError(
        StatusCode::SERVICE_UNAVAILABLE,
        "read model has not reached that version".to_string(),
    ))
}

pub async fn list_summaries(State(pool): State<PgPool>) -> Result<Json<Vec<Summary>>, AppError> {
    let rows: Vec<Summary> =
        sqlx::query_as(&format!("SELECT {COLUMNS} FROM order_summaries ORDER BY id"))
            .fetch_all(&pool)
            .await?;
    Ok(Json(rows))
}

pub async fn rebuild(State(pool): State<PgPool>) -> Result<Json<Value>, AppError> {
    projector::rebuild(&pool).await?;
    Ok(Json(json!({ "rebuild": "started" })))
}

Start PostgreSQL with the docker run line from Project setup, then cargo run. The examples use ID=11111111-1111-1111-1111-111111111111.

Terminal

ID=11111111-1111-1111-1111-111111111111
curl -s -X PUT localhost:3000/orders/$ID -H 'content-type: application/json' -d '{"customer_id":7}'

Output

{"order_id":"11111111-1111-1111-1111-111111111111","version":1}

Terminal

curl -s -X POST localhost:3000/orders/$ID/lines -H 'content-type: application/json' \
  -d '{"sku":"notebook","qty":2,"unit_cents":450}'
curl -s -X POST localhost:3000/orders/$ID/lines -H 'content-type: application/json' \
  -d '{"sku":"pen","qty":1,"unit_cents":120}'

Output

{"order_id":"11111111-1111-1111-1111-111111111111","version":2}
{"order_id":"11111111-1111-1111-1111-111111111111","version":3}

Terminal

curl -s "localhost:3000/orders/$ID/summary?min_version=3"

Output

{"id":"11111111-1111-1111-1111-111111111111","customer_id":7,"status":"placed","line_count":2,"total_cents":1020,"version":3}

Terminal

curl -s -X POST localhost:3000/orders/$ID/pay
curl -s -X POST localhost:3000/orders/$ID/pay
curl -s -X POST localhost:3000/orders/$ID/cancel
curl -s -X PUT localhost:3000/orders/$ID -H 'content-type: application/json' -d '{"customer_id":7}'

Output

{"order_id":"11111111-1111-1111-1111-111111111111","version":4}
{"order_id":"11111111-1111-1111-1111-111111111111","version":4}
{"error":"order is already paid"}
{"error":"order is already placed"}

The second pay appends no event, so the version stays 4. Replay the history into a fresh read model:

Terminal

curl -s -X POST localhost:3000/admin/rebuild
curl -s "localhost:3000/orders/$ID/summary?min_version=4"

Output

{"rebuild":"started"}
{"id":"11111111-1111-1111-1111-111111111111","customer_id":7,"status":"paid","line_count":2,"total_cents":1020,"version":4}

Summary

CQRS gives commands and queries separate models. Event sourcing stores immutable events and derives state by applying them.
apply records facts. decide_* checks rules and returns events. A repeated command returns no events.
A primary key on (aggregate_id, version) gives optimistic concurrency. A conflict means reload and retry.
A projection updates its read model and its checkpoint in one transaction.
Snapshots shorten loading. Replay rebuilds a projection from history.
Event shapes change through serde defaults, aliases, or upcasting in decode. Events are never edited.
Read models lag behind writes. Clients use the returned version, a min_version query, or a pending state.
Apply the patterns to aggregates with history or complex read needs. Keep simple records as plain tables.