CQRS + Read-model projections

เนื้อหานี้ยังไม่ได้แปลเป็นภาษาไทย แสดงเป็นภาษาอังกฤษแทน

This tutorial walks through CQRS (Command Query Responsibility Segregation) in a CephalonEngine app: the write model owns commands + state, the read model owns optimized projections, and Wolverine events bridge the two.

You’ll build:

A Products write model on Postgres (transactional, normalized).
A Products read model on ClickHouse (denormalized, analytics-friendly).
A Wolverine event bus that publishes ProductCreated / ProductPriceChanged / ProductDiscontinued from the write side and projects them into the read side.
Two transport behaviors: POST /products writes via the write model, GET /products/search?... reads via the read model.
Reasoning about eventual consistency between the two sides + how to measure the projection lag.

Time: ~60 minutes. Prerequisite: Build your first app — this tutorial picks up where step 3 (data) ends.

What is CQRS, and when do you actually want it?

CQRS = different models for reads vs writes. It’s not about different code paths — every web app already has those. It’s about different storage shapes:

Side	Optimizes for	Typical shape
Write	Transactional consistency, invariants, FK integrity	Normalized relational (3NF)
Read	Query latency, aggregation, full-text, faceting	Denormalized — wide rows, materialized views, search indexes

When CQRS pays off

Read traffic > 10× write traffic and reads are doing expensive joins/aggregations.
The same data is queried in many shapes (per-customer summary, search, dashboards, exports).
You need independent scaling — read replicas / read-only nodes that can be added without touching the write side.
Eventual consistency is acceptable (the read side lags writes by milliseconds-to-seconds).

When CQRS is wasted complexity

Low traffic — a single Postgres can serve both sides for years.
Strong-consistency reads required — e.g. checkout flows that depend on the latest write within the same request.
Single query shape — if every read is the same shape as the write entity, you’re just adding lag for nothing.

CephalonEngine doesn’t force CQRS — it just makes it cheap when you need it.

Two flavors

1. Lightweight CQRS (this tutorial)

State-based projections. The write model stores entity state. When state changes, a behavior publishes a *Changed event. A handler on the read side updates a denormalized table.

Command  ──→  Write model (Postgres)  ──→  Event  ──→  Projection handler  ──→  Read model (ClickHouse)
                                              │
                                              └──→  (also: external integrations, audit, search index)

Pros: simple mental model, easy to debug, current state is in one place. Cons: read-side rebuild requires re-emitting all current state (no historical replay).

2. Full event-sourced CQRS

Events are the source of truth. No “current state” table on the write side — just an append-only event stream. Read projections rebuild by replaying events.

Command  ──→  Validate against event-replayed state  ──→  Append events  ──→  Read projection rebuilds from events

Pros: full audit trail, time-travel queries, rebuild any projection from history. Cons: bigger learning curve, event schema evolution is a real problem, debugging needs event-store tooling.

This tutorial does lightweight CQRS. We cover full event-sourcing at the end with pointers to the Cephalon.EventSourcing family.

Setup

Pick up from first-app step 3 — you should have an Acme.Store.Modules.Products module with Product and Category EF Core entities on Postgres.

# At the repo root
docker run -d --name acme-store-clickhouse -p 9000:9000 -p 8123:8123 \
  -e CLICKHOUSE_DB=acme_read \
  -e CLICKHOUSE_USER=acme \
  -e CLICKHOUSE_PASSWORD=acme \
  clickhouse/clickhouse-server:24.3

docker run -d --name acme-store-rabbit -p 5672:5672 -p 15672:15672 \
  rabbitmq:3.13-management

You should already have Postgres running from step 3. After this, you have 3 containers: Postgres (write), ClickHouse (read), RabbitMQ (eventing).

Step 1 — Configure read/write split

Add ClickHouse + Wolverine packages:

cd src/Acme.Store.Host
dotnet add package Cephalon.Data.ClickHouse --prerelease
dotnet add package Cephalon.Eventing --prerelease
dotnet add package Cephalon.Eventing.Wolverine --prerelease

Then configure appsettings.json:

1
{
2
  "Engine": {
3
    "Data": {
4
      "IdStrategy": "Sfid",
5
      "Provider": null,
6
      "WriteModel": {
7
        "Provider": "Postgres",
8
        "Migrations": { "ApplyOn": "Startup" }
9
      },
10
      "ReadModel": {
11
        "Provider": "ClickHouse",
12
        "Migrations": { "ApplyOn": "Startup" }
13
      }
14
    },
15
    "Messaging": {
16
      "Enabled": true,
17
      "Provider": "Wolverine",
18
      "Wolverine": {
19
        "Transport": "RabbitMQ",
20
        "ConnectionString": "amqp://guest:guest@localhost:5672",
21
        "Concurrency": 4,
22
        "Outbox": { "Enabled": true }
23
      }
24
    }
25
  },
26
  "ConnectionStrings": {
27
    "ProductsWrite":  "Host=localhost;Port=5432;Database=acme_store;Username=acme;Password=acme",
28
    "ProductsRead":   "Host=localhost;Port=8123;Database=acme_read;Username=acme;Password=acme"
29
  }
30
}

What’s happening:

Engine:Data:WriteModel:Provider="Postgres" + ReadModel:Provider="ClickHouse" tells the data capability that this app has split storage.
ConnectionStrings:ProductsWrite / ProductsRead follow the {Module}{Side} convention so the engine routes each DbContext to the right backend.
Messaging:Wolverine:Outbox:Enabled=true enables the transactional outbox — events publish in the same transaction as the write, guaranteeing read-side projections see every committed write (even if the broker is briefly down).

Full schema: Reference → Configuration → Data + Messaging.

Step 2 — Define the write model

The write model is the source of truth for commands and invariants. Keep it normalized and small.

1
using Cephalon.Data.EntityFramework;
2
using Microsoft.EntityFrameworkCore;
3

4
namespace Acme.Store.Modules.Products.Write;
5

6
public sealed class ProductsWriteDbContext(DbContextOptions<ProductsWriteDbContext> options)
7
    : CephalonDbContext(options)
8
{
9
    public DbSet<Product> Products => Set<Product>();
10

11
    protected override void OnModelCreating(ModelBuilder b)
12
    {
13
        base.OnModelCreating(b);
14

15
        b.Entity<Product>(e =>
16
        {
17
            e.UseSfidPrimaryKey();
18
            e.Property(p => p.Sku).HasMaxLength(64).IsRequired();
19
            e.HasIndex(p => p.Sku).IsUnique();
20
            e.Property(p => p.Name).HasMaxLength(200).IsRequired();
21
            e.Property(p => p.Price).HasPrecision(18, 2);
22
            e.Property(p => p.IsDiscontinued).HasDefaultValue(false);
23
            e.Property(p => p.UpdatedAt).IsRowVersion();
24
        });
25
    }
26
}
27

28
public sealed class Product
29
{
30
    public Sfid Id { get; init; }
31
    public required string Sku { get; init; }
32
    public required string Name { get; set; }
33
    public required string Description { get; set; }
34
    public required decimal Price { get; set; }
35
    public required Sfid CategoryId { get; init; }
36
    public bool IsDiscontinued { get; set; }
37
    public byte[]? UpdatedAt { get; init; }    // optimistic concurrency
38
}

1
var builder = WebApplication.CreateBuilder(args);
2

3
builder.Services
4
    .AddCephalonAspNetCore()
5
    .AddModulesFromAssemblies(typeof(Program).Assembly)
6
    .AddCephalonEntityFramework<ProductsWriteDbContext>((sp, opts) =>
7
    {
8
        var conn = sp.GetRequiredService<IConfiguration>().GetConnectionString("ProductsWrite");
9
        opts.UseNpgsql(conn);
10
    })
11
    .Build(builder);

Step 3 — Define the read model

The read model is denormalized for queries. Cram the per-product details + category name + computed aggregations into wide rows.

1
using Cephalon.Data.EntityFramework;
2
using Microsoft.EntityFrameworkCore;
3

4
namespace Acme.Store.Modules.Products.Read;
5

6
public sealed class ProductsReadDbContext(DbContextOptions<ProductsReadDbContext> options)
7
    : CephalonDbContext(options)
8
{
9
    public DbSet<ProductView> Products => Set<ProductView>();
10

11
    protected override void OnModelCreating(ModelBuilder b)
12
    {
13
        base.OnModelCreating(b);
14

15
        b.Entity<ProductView>(e =>
16
        {
17
            e.HasKey(v => v.Id);
18
            e.Property(v => v.Sku).HasMaxLength(64);
19
            e.Property(v => v.Name).HasMaxLength(200);
20
            e.Property(v => v.CategoryName).HasMaxLength(200);
21
            // ClickHouse-specific MergeTree config via annotation
22
            e.HasAnnotation("ClickHouse:Engine", "MergeTree");
23
            e.HasAnnotation("ClickHouse:OrderBy", "(CategoryId, Sku)");
24
        });
25
    }
26
}
27

28
public sealed class ProductView
29
{
30
    public Sfid Id { get; set; }
31
    public string Sku { get; set; } = "";
32
    public string Name { get; set; } = "";
33
    public string Description { get; set; } = "";
34
    public decimal Price { get; set; }
35
    public Sfid CategoryId { get; set; }
36
    public string CategoryName { get; set; } = "";   // denormalized — no JOIN needed
37
    public bool IsDiscontinued { get; set; }
38
    public int ViewCount { get; set; }                // denormalized aggregation
39
    public DateTimeOffset CreatedAt { get; set; }
40
    public DateTimeOffset LastUpdatedAt { get; set; }
41
    public DateTimeOffset ProjectedAt { get; set; }   // when this row was last projected
42
}

1
builder.Services
2
    .AddCephalonAspNetCore()
3
    .AddModulesFromAssemblies(typeof(Program).Assembly)
4
    .AddCephalonEntityFramework<ProductsWriteDbContext>((sp, opts) =>
5
    {
6
        opts.UseNpgsql(sp.GetRequiredService<IConfiguration>().GetConnectionString("ProductsWrite"));
7
    })
8
    .AddCephalonEntityFramework<ProductsReadDbContext>((sp, opts) =>
9
    {
10
        opts.UseConnection(sp.GetRequiredService<IConfiguration>().GetConnectionString("ProductsRead"));
11
        // (Cephalon.Data.ClickHouse registers UseConnection on top of EF Core)
12
    })
13
    .Build(builder);

Denormalize aggressively in the read model. Every JOIN you avoid on the read side is a query you don’t have to optimize later. CategoryName being in the read row means category filtering and display are zero-join. ViewCount being denormalized means “popular products” queries are one column scan, not a left-join + count.

Step 4 — Define the events

Events are the public contract between write and read sides. Keep them small and stable.

1
namespace Acme.Store.Modules.Products;
2

3
[Event]
4
public sealed record ProductCreated(
5
    Sfid Id, string Sku, string Name, string Description, decimal Price,
6
    Sfid CategoryId, string CategoryName, DateTimeOffset CreatedAt);
7

8
[Event]
9
public sealed record ProductPriceChanged(
10
    Sfid Id, decimal OldPrice, decimal NewPrice, DateTimeOffset ChangedAt);
11

12
[Event]
13
public sealed record ProductRenamed(
14
    Sfid Id, string OldName, string NewName, DateTimeOffset ChangedAt);
15

16
[Event]
17
public sealed record ProductDiscontinued(
18
    Sfid Id, DateTimeOffset DiscontinuedAt);

Conventions:

Past-tense names. Events describe what happened, not commands.
Self-contained payload. The read side should not need to fetch additional data — include everything needed for the projection.
Id first. Convention so handlers can dispatch by aggregate.
[Event] attribute registers the type with Cephalon.Eventing for typed routing.

Step 5 — Write-side command behavior with outbox publishing

The command handler does the transactional write and publishes the event in the same transaction. The outbox ensures broker outages don’t cause lost events.

1
public sealed record CreateProductCommand(
2
    string Sku, string Name, string Description, decimal Price, Sfid CategoryId);
3

4
public sealed class CreateProductBehavior(
5
    ProductsWriteDbContext db,
6
    IMessagePublisher events,
7
    IIdGenerator<Sfid> ids)
8
    : IRestBehavior<CreateProductCommand, Sfid>
9
{
10
    public RestRoute Route => RestRoute.Post("/products");
11

12
    [BehaviorPipeline(typeof(WithRequireScope), Scopes = "products:write")]
13
    [BehaviorPipeline(typeof(WithOutbox))]   // ← outbox decorator
14
    public async Task<Sfid> Handle(CreateProductCommand cmd, CancellationToken ct)
15
    {
16
        var category = await db.Categories.FirstAsync(c => c.Id == cmd.CategoryId, ct);
17

18
        var product = new Product
19
        {
20
            Id = ids.NewId(),
21
            Sku = cmd.Sku,
22
            Name = cmd.Name,
23
            Description = cmd.Description,
24
            Price = cmd.Price,
25
            CategoryId = cmd.CategoryId,
26
        };
27
        db.Products.Add(product);
28

29
        // Publish event in the SAME transaction as the DB write.
30
        // The outbox holds it until SaveChanges commits.
31
        await events.PublishAsync(new ProductCreated(
32
            product.Id, product.Sku, product.Name, product.Description, product.Price,
33
            product.CategoryId, category.Name, DateTimeOffset.UtcNow), ct);
34

35
        await db.SaveChangesAsync(ct);   // commits write + flushes outbox to broker
36
        return product.Id;
37
    }
38
}

Key points:

IMessagePublisher is the Cephalon.Eventing interface — broker-neutral.
WithOutbox decorator wraps the handler so the Publish enqueues into the outbox table instead of going directly to the broker. The outbox flushes when SaveChanges commits — atomically with the DB write.
The handler returns only the Sfid — the read model isn’t consulted here.

Step 6 — Read-side projection handler

The projection handler subscribes to events and updates the read model. Idempotent + at-least-once-safe.

1
public sealed class ProductProjection(ProductsReadDbContext read, ILogger<ProductProjection> log) :
2
    IMessageHandler<ProductCreated>,
3
    IMessageHandler<ProductPriceChanged>,
4
    IMessageHandler<ProductRenamed>,
5
    IMessageHandler<ProductDiscontinued>
6
{
7
    [BehaviorPipeline(typeof(WithInbox))]   // dedupe by event id
8
    [BehaviorPipeline(typeof(WithRetry), MaxAttempts = 5, Backoff = "00:00:30")]
9
    public async Task Handle(ProductCreated e, MessageContext ctx)
10
    {
11
        var view = new ProductView
12
        {
13
            Id = e.Id,
14
            Sku = e.Sku,
15
            Name = e.Name,
16
            Description = e.Description,
17
            Price = e.Price,
18
            CategoryId = e.CategoryId,
19
            CategoryName = e.CategoryName,
20
            IsDiscontinued = false,
21
            CreatedAt = e.CreatedAt,
22
            LastUpdatedAt = e.CreatedAt,
23
            ProjectedAt = DateTimeOffset.UtcNow,
24
        };
25
        await read.Products.AddAsync(view, ctx.CancellationToken);
26
        await read.SaveChangesAsync(ctx.CancellationToken);
27
        log.LogInformation("Projected ProductCreated for {ProductId} in {Lag} ms",
28
            e.Id, (DateTimeOffset.UtcNow - e.CreatedAt).TotalMilliseconds);
29
    }
30

31
    [BehaviorPipeline(typeof(WithInbox))]
32
    public async Task Handle(ProductPriceChanged e, MessageContext ctx)
33
    {
34
        var view = await read.Products.FindAsync([e.Id], ctx.CancellationToken)
35
            ?? throw new InvalidOperationException($"ProductView {e.Id} not found");
36
        view.Price = e.NewPrice;
37
        view.LastUpdatedAt = e.ChangedAt;
38
        view.ProjectedAt = DateTimeOffset.UtcNow;
39
        await read.SaveChangesAsync(ctx.CancellationToken);
40
    }
41

42
    // Handle(ProductRenamed) and Handle(ProductDiscontinued) follow the same shape.
43
}

Key points:

One handler class per aggregate (Product). It implements IMessageHandler<T> once per event type.
WithInbox decorator deduplicates by event id — so if the broker redelivers (network blip, ack lost), the projection runs at most once per logical event.
WithRetry handles transient ClickHouse failures.
The handler logs the projection lag — useful for measuring eventual-consistency window.

Step 7 — Read-side query behavior

The read behavior queries the read model directly. No write-side dependency, no JOIN to category.

1
public sealed record ProductSearchInput(
2
    string? Query, Sfid? CategoryId, decimal? MinPrice, decimal? MaxPrice,
3
    int Limit = 20, int Offset = 0);
4

5
public sealed record ProductSearchResult(
6
    IReadOnlyList<ProductView> Items, int Total);
7

8
public sealed class SearchProductsBehavior(ProductsReadDbContext read)
9
    : IRestBehavior<ProductSearchInput, ProductSearchResult>
10
{
11
    public RestRoute Route => RestRoute.Get("/products/search");
12

13
    public async Task<ProductSearchResult> Handle(ProductSearchInput q, CancellationToken ct)
14
    {
15
        var query = read.Products.Where(p => !p.IsDiscontinued);
16

17
        if (!string.IsNullOrWhiteSpace(q.Query))
18
            query = query.Where(p => p.Name.Contains(q.Query) || p.Sku.Contains(q.Query));
19
        if (q.CategoryId is { } cat)
20
            query = query.Where(p => p.CategoryId == cat);
21
        if (q.MinPrice is { } min) query = query.Where(p => p.Price >= min);
22
        if (q.MaxPrice is { } max) query = query.Where(p => p.Price <= max);
23

24
        var total = await query.CountAsync(ct);
25
        var items = await query
26
            .OrderByDescending(p => p.ViewCount)   // pre-denormalized aggregation
27
            .Skip(q.Offset).Take(q.Limit)
28
            .ToListAsync(ct);
29

30
        return new ProductSearchResult(items, total);
31
    }
32
}

ClickHouse runs this in single-digit milliseconds across millions of rows because:

No JOIN — CategoryName is in the row.
OrderBy matches the table’s MergeTree sort key.
ViewCount is denormalized — no aggregation at query time.

Step 8 — Verify

Run the host + place a request:

dotnet run --project src/Acme.Store.Host

# Place a write
curl -X POST http://localhost:5000/products \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <token-with-products:write>" \
  -d '{"sku":"WIDGET-001","name":"Blue widget","description":"...","price":12.50,"categoryId":"01HQ..."}'
# → {"id": "01HQ8RVKSXM4Y8RBVPDC0E9YHZ"}

# Wait ~50-200ms for projection...
sleep 0.5

# Read back
curl 'http://localhost:5000/products/search?query=widget'
# → {"items": [{ "id": "01HQ...", "name": "Blue widget", "categoryName": "Tools", ... }], "total": 1}

You can also verify via the engine manifest:

curl http://localhost:5000/engine/manifest | jq '.capabilities'
# → ["Data", "Eventing", "Identity", "Observability"]

Eventual consistency window

Reads see writes after the projection runs. Typical lag:

Scenario	Lag p50	Lag p99
In-memory transport (dev)	`<1 ms`	`<5 ms`
RabbitMQ, same data center	10-30 ms	50-150 ms
RabbitMQ across regions	50-100 ms	300-500 ms
Kafka, single partition	5-15 ms	30-80 ms

Acceptable for: catalog browsing, search, dashboards, reporting. Not acceptable for: checkout reading the cart you just wrote (use the write model directly for that), idempotency checks against the same request.

Measuring the lag

The projection handler logs ProjectionLag = ProjectedAt - EventCreatedAt. Surface this in observability:

1
log.LogInformation("Projected {Event} in {Lag} ms", event.GetType().Name,
2
    (DateTimeOffset.UtcNow - event.CreatedAt).TotalMilliseconds);

Then in Grafana / Tempo, alert when p99 > 500ms.

Going full event-sourced

The lightweight CQRS above stores current state on the write side. To go full event-sourced — events ARE the state, replay rebuilds projections:

1
{
2
  "Engine": {
3
    "EventSourcing": {
4
      "Enabled": true,
5
      "Provider": "EntityFramework",      // or "Cassandra", "ClickHouse", etc.
6
      "Snapshots": { "Every": 100 },
7
      "Projections": { "RebuildOnSchemaChange": true }
8
    }
9
  }
10
}

Then refactor the write side from Product (state) to ProductAggregate (event-replayer):

1
public sealed class ProductAggregate : EventSourcedAggregate<Sfid>
2
{
3
    public string Sku { get; private set; } = "";
4
    public decimal Price { get; private set; }
5
    public bool IsDiscontinued { get; private set; }
6

7
    public void Apply(ProductCreated e) { /* update state from event */ }
8
    public void Apply(ProductPriceChanged e) { Price = e.NewPrice; }
9
    public void Apply(ProductDiscontinued e) { IsDiscontinued = true; }
10

11
    public void ChangePrice(decimal newPrice)
12
    {
13
        if (IsDiscontinued) throw new InvariantException("Cannot price a discontinued product");
14
        RaiseEvent(new ProductPriceChanged(Id, Price, newPrice, DateTimeOffset.UtcNow));
15
    }
16
}

Read projections subscribe to the event stream and rebuild from history (not just from new events). Full coverage in Reference → Components → Event Sourcing.

Anti-patterns

Don’t	Do
Query the read model from inside a command handler	Read-after-write inside a command = consistency bug. Use the write DbContext if you need read-after-write.
Use the write model for queries that don’t touch the writes	If the query doesn’t need invariants, it shouldn’t pay the write-model cost (JOINs, locks). Project to the read side.
Skip the outbox because “in-memory transport doesn’t drop messages”	Production transport WILL drop messages without an outbox. Always enable outbox.
Denormalize the read model from the write model directly via DB triggers	DB triggers tie write and read storage tightly. Use events — they’re decoupled.
Make every event a `*Changed` summary of the full entity	Event payloads should be what changed, not the full entity. Smaller events = cleaner audit trail + cheaper to publish.
Run write and read on the same Postgres “for now”	If you’re going to deploy read replicas later, set up the read model on day 1 with the projection pipeline. Switching later is expensive.
Allow handlers to read external services (other APIs, files, etc.)	Projections should be deterministic. Bring the data into the event payload OR call the external service from a separate behavior.

Tips & tricks

Rebuild projections

When the read schema changes (or a bug is fixed), rebuild from scratch:

1
public sealed class ProductsReadModelRebuilder(
2
    ProductsWriteDbContext write,
3
    IMessagePublisher events) : IHostedService
4
{
5
    public async Task StartAsync(CancellationToken ct)
6
    {
7
        // Re-emit "synthetic" events from the current state of the write model.
8
        var products = await write.Products.Include(p => p.Category).ToListAsync(ct);
9
        foreach (var p in products)
10
        {
11
            await events.PublishAsync(new ProductCreated(
12
                p.Id, p.Sku, p.Name, p.Description, p.Price,
13
                p.CategoryId, p.Category.Name, p.CreatedAt), ct);
14
        }
15
    }
16
}

Run once, then disable. Better: keep an event store (full event-sourcing) and replay from there.

Multiple read models per aggregate

Nothing forces 1:1 between aggregate and read model. Common pattern: 3 read models per aggregate, one per consumer:

ProductAggregate (write)
   ├── ProductSearchView (ClickHouse — for full-text search)
   ├── ProductDashboardView (Postgres — for admin UI)
   └── ProductPricingView (Redis — for fast cart calculations)

Each projection handler subscribes to the same events but writes to its own optimized shape.

Versioning events

Events are the public contract. Breaking them breaks consumers. Two patterns:

Additive only. Add optional fields, never remove. Consumers ignore unknown fields.
Versioned envelope. ProductCreatedV2 with a new schema; handlers explicitly opt into the new version.

Cephalon.Eventing supports both — pick one and stick to it.

Snapshots (event sourcing)

If you replay events at startup, snapshots avoid the long warmup:

1
{ "Engine": { "EventSourcing": { "Snapshots": { "Every": 100 } } } }

Stores a snapshot every 100 events. Recovery: load latest snapshot + replay events after it.

Where to go next

Reference → Configuration → Data — full read/write split schema.
Reference → Configuration → Messaging — outbox/inbox, transport choice, DLQ.
Reference → Components → Event Sourcing — when you graduate from lightweight CQRS.
Technology → Eventing — patterns deep-dive.
Tutorial → Event-driven pipeline — saga / process-manager patterns that pair well with CQRS.
Tutorial → Observability stack — measure projection lag in production.