CQRS (Command Query Responsibility Segregation) separates write models from read models. Event Sourcing treats the immutable event log as the source of truth — current state is a projection of replayed events. Both are architectural patterns, not Mongo features — but MongoDB fits them especially well: change streams as the projection bus (Post 12), $merge for materialised views (Post 06), aggregation for replaying events, transactions for atomic event-store appends (Post 09). This post covers when each pays off, the canonical models, the trade-offs vs CRUD, and a worked order-management example end-to-end.

Series map: 19 Beanie+FastAPI → 20 Outbox → 21 CQRS + Event Sourcing (this post) → 22 RAG → ... → 29 K8s + mongosync + cost.

Learning Outcome

  • Define CQRS and recognise when it earns its complexity.
  • Define Event Sourcing and the event-store schema in MongoDB.
  • Project events into read models via change streams.
  • Replay history for new projections + bug fixes.
  • Handle snapshots + projection rebuilds.
  • Combine outbox + CQRS + ES into a clean architecture.

Chapter 1 — CQRS in one paragraph

A normal CRUD app uses one model for reads and writes — same schema, same indexes, same database. CQRS splits them: command side handles writes (often normalised, validated, transactional), query side holds one or more read models optimised per use case (denormalised, fast). The read models are updated by events the command side publishes — usually asynchronously.

Benefit:

  • Read models match exactly the queries they serve → simple + fast.
  • Read + write scale independently.
  • Multiple read models per command model (one for search, one for analytics, one for a mobile API).

Cost:

  • More moving parts.
  • Read-after-write delay (eventual consistency).
  • Operational complexity.

Use CQRS when one of: read volume >> write volume; queries diverge wildly from write shape; you already need an outbox + projections.

Chapter 2 — Event Sourcing in one paragraph

Instead of storing current state, store every change as an immutable event. Current state is computed by replaying events. Events are append-only — the most reliable storage primitive in databases.

Benefits:

  • Full audit trail for free.
  • New projections retroactively built from history.
  • Bug fixes can replay events through corrected code.
  • Time-travel debugging.

Costs:

  • Reads must aggregate events (use snapshots + projections).
  • Schema evolution of events is forever — versioning matters.
  • Higher mental + ops load than CRUD.

Use ES when one of: regulatory audit trail mandatory; the domain is genuinely event-shaped (orders, payments, IoT, ledger systems); you need to replay history for new analytics or bug correction.

ES + CQRS combine naturally — ES is the command-side event log, CQRS is the read-side projections.

Chapter 3 — When NOT to use CQRS / ES

  • A CRUD app with simple reads + writes — KISS.
  • Small team / small product — overhead > benefit.
  • Tight read-after-write requirements with no tolerance for eventual consistency.
  • "Because it sounds modern" — biggest reason teams regret it.

Default = CRUD. Add CQRS only when reads + writes genuinely diverge. Add ES only when the audit-or-replay property is mandatory.

Chapter 4 — Event-store schema

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// events collection — append-only
{
  _id:           ObjectId(),
  aggregate:     "order",                  // entity type
  aggregate_id:  "o123",                   // entity id
  version:       7,                         // sequential per aggregate
  event_type:    "order.shipped",
  payload:       { tracking: "ABC123", carrier: "ups", shipped_at: ISODate(...) },
  metadata:      { user_id: "u42", request_id: "req-..." },
  occurred_at:   ISODate(),
}
 
db.events.createIndex({ aggregate: 1, aggregate_id: 1, version: 1 }, { unique: true })
db.events.createIndex({ occurred_at: -1 })

Constraints:

  • (aggregate, aggregate_id, version) unique — prevents concurrent appends with the same version. Acts as optimistic concurrency control.
  • Append-only — never update or delete events.

Use TTL only on very long-term retention if regulation allows. Most ES systems keep events forever (or in cold storage after N years).

Chapter 5 — Command handlers (write side)

A command handler:

  1. Loads the aggregate by replaying its events.
  2. Validates the command against current state.
  3. Generates new events.
  4. Appends them with the next sequential version.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
async function shipOrder(orderId, cmd) {
  // load events
  const events = await db.collection('events')
    .find({ aggregate: 'order', aggregate_id: orderId })
    .sort({ version: 1 })
    .toArray();
  if (!events.length) throw new Error('order not found');
 
  // fold to current state
  const state = events.reduce(apply, {});
 
  // validate
  if (state.status !== 'paid') throw new Error('not paid');
 
  // produce new events
  const newEvent = {
    aggregate: 'order',
    aggregate_id: orderId,
    version: events[events.length - 1].version + 1,
    event_type: 'order.shipped',
    payload: { tracking: cmd.tracking, carrier: cmd.carrier, shipped_at: new Date() },
    metadata: cmd.metadata,
    occurred_at: new Date(),
  };
 
  // append (unique index on (aggregate, aggregate_id, version) blocks conflicting writes)
  await db.collection('events').insertOne(newEvent);
}
 
function apply(state, event) {
  switch (event.event_type) {
    case 'order.created':  return { ...state, status: 'pending', items: event.payload.items, total: event.payload.total };
    case 'order.paid':     return { ...state, status: 'paid' };
    case 'order.shipped':  return { ...state, status: 'shipped', tracking: event.payload.tracking };
    case 'order.cancelled': return { ...state, status: 'cancelled' };
    default: return state;
  }
}

Concurrency: two handlers trying to append version: N for the same aggregate → one wins, the other gets E11000 duplicate key. The losing handler retries by re-loading the aggregate (now at version N+1) + reapplying validation.

Chapter 6 — Projections (read side)

A projection is a read model derived from events. One projection per query pattern.

Projection: current order view

1
2
3
// projections.orders_view collection
{ _id: "o123", user_id: 1, status: "shipped", total: 99, tracking: "ABC123", updated_at: ISODate(...) }

A projection process change-streams events (Post 12) and updates the view:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
const cs = db.collection('events').watch(
  [{ $match: { operationType: 'insert' } }],
  { fullDocument: 'required' }
);
 
for await (const change of cs) {
  const e = change.fullDocument;
  if (e.aggregate !== 'order') continue;
 
  switch (e.event_type) {
    case 'order.created':
      await db.collection('orders_view').insertOne({
        _id: e.aggregate_id,
        user_id: e.payload.user_id,
        status: 'pending',
        total: e.payload.total,
        items: e.payload.items,
        updated_at: e.occurred_at,
      });
      break;
    case 'order.paid':
      await db.collection('orders_view').updateOne(
        { _id: e.aggregate_id },
        { $set: { status: 'paid', updated_at: e.occurred_at } }
      );
      break;
    case 'order.shipped':
      await db.collection('orders_view').updateOne(
        { _id: e.aggregate_id },
        { $set: { status: 'shipped', tracking: e.payload.tracking, updated_at: e.occurred_at } }
      );
      break;
    case 'order.cancelled':
      await db.collection('orders_view').updateOne(
        { _id: e.aggregate_id },
        { $set: { status: 'cancelled', updated_at: e.occurred_at } }
      );
      break;
  }
 
  await persistResumeToken(change._id);
}

The read endpoint:

1
2
3
4
5
6
app.get('/orders/:id', async (req, res) => {
  const o = await db.collection('orders_view').findOne({ _id: req.params.id });
  if (!o) return res.status(404).end();
  res.json(o);
});

Read side never touches the event log directly. Fast. Indexed. Schema designed for the read pattern.

Multiple projections

Add a customer_orders_view keyed by user_id, an analytics_daily_revenue view, a search_index Atlas Search index — each is a different consumer of the event stream.

To add a new view: write the projection code, replay events from the beginning (no resume token), let it catch up, switch reads over.

Chapter 7 — Snapshots

Replaying thousands of events per command load is slow. Periodically write a snapshot of the aggregate state, then load snapshot + events_after_snapshot_version.

1
2
3
4
5
6
7
8
9
{
  _id:           "order:o123",
  aggregate:     "order",
  aggregate_id:  "o123",
  version:       100,
  state:         { status: "shipped", items: [...], total: 99, ... },
  taken_at:      ISODate(...),
}

Take snapshots every N events (e.g., every 50) or on a schedule. Command handler loads snapshot + events with version > snapshot.version.

1
2
3
4
5
6
7
8
9
async function loadAggregate(orderId) {
  const snap = await db.collection('snapshots').findOne({ _id: `order:${orderId}` });
  const fromVersion = snap?.version ?? 0;
  const events = await db.collection('events').find({
    aggregate: 'order', aggregate_id: orderId, version: { $gt: fromVersion }
  }).sort({ version: 1 }).toArray();
  return events.reduce(apply, snap?.state ?? {});
}

Snapshots are derived data — disposable. Wipe + rebuild from events if your apply logic changes.

Chapter 8 — Rebuilding projections

Bug in projection code? Wrote bad state? Rebuild from scratch:

1
2
3
4
5
6
7
8
9
// drop projection
await db.collection('orders_view').drop();
 
// replay all events from beginning
const cs = db.collection('events').watch([{ $match: { operationType: 'insert' } }], {
  startAtOperationTime: Timestamp(0, 0)    // from the start of the cluster
});
// ... feed every event through the projection logic

Or use find + sort by _id (events sorted naturally by ObjectId time) instead of change stream for a one-shot rebuild.

This retroactive correctness is one of the headline benefits of ES — no other persistence model lets you "just rerun history with the new code".

Chapter 9 — Event versioning

Events live forever. Their schema must evolve gracefully.

Patterns:

  1. Add a version field per event type + handle each version in the projection.
  2. Upcast old events to the new shape at read time.
  3. Migration events — emit a new event ("schema migrated to v2") rather than mutating history.

Never mutate old events. The event log is sacred — modifying it breaks every projection's correctness.

1
2
3
4
5
6
7
case 'order.created':
  if (e.version === 1) state.items = e.payload.items;
  else if (e.version === 2) {
    state.items = e.payload.line_items.map(li => ({ p: li.product_id, qty: li.quantity }));
  }
  break;

Chapter 10 — Worked end-to-end architecture

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
┌─────────────────────────────────────────────────────────────────────┐
│   Client (UI / API caller)                                          │
└─────────┬────────────────────────────────────────┬──────────────────┘
          │ POST /commands                          │ GET /queries
          ▼                                          ▼
  ┌─────────────────┐                       ┌────────────────────────┐
  │ Command Service │                       │  Query Service          │
  │  - loads agg    │                       │  - reads from projection│
  │  - validates    │                       │    collections          │
  │  - emits event  │                       └────────────┬───────────┘
  └────────┬────────┘                                    │
           │ insert into `events`                        │
           │   (unique index → optimistic concurrency)    │
           ▼                                             │
   ┌──────────────────────┐                              │
   │  MongoDB              │                              │
   │   ├ events            │ <─── change stream ─────────┘
   │   ├ snapshots         │
   │   ├ orders_view       │
   │   ├ user_orders_view  │
   │   └ daily_revenue     │
   └──────────────────────┘
           │
           │ change stream
           ▼
   ┌──────────────────────────┐
   │ Projection Workers       │
   │  - one per projection    │
   │  - updates view collection│
   └──────────────────────────┘

Three deployables:

  1. Command service — accepts commands, validates against loaded aggregate, appends events.
  2. Query service — reads from projection collections.
  3. Projection workers — one or many, each tail change stream + update one view.

Optional: outbox alongside events for publishing to Kafka (Post 20) — events going to external systems.

CQRS with event sourcing on MongoDB: commands append versioned events, projection workers update read views, queries read the views

Chapter 11 — Trade-offs vs CRUD

CRUD CQRS CQRS + ES
Code complexity low medium high
Read speed depends on index optimal (purpose-built views) optimal
Write speed direct mutation direct mutation append-only (very fast)
Audit trail manual manual free
Replay history impossible impossible core feature
Read-after-write strong (single DB) eventual (projection lag) eventual
Schema evolution migrations migrations event versioning
Best for most apps read-heavy / divergent regulated / event-shaped

The simplest pattern that meets your requirements wins. CQRS adds complexity; only pay for it when it earns its keep.

Chapter 12 — Common pitfalls

Pitfall Cause Fix
Eventual consistency surprises in UI read immediately after write optimistic UI; or read-from-command-side fallback for the actor's own writes
Projection stuck on a bug one bad event halts the worker log + skip the bad one + alert; never panic-mutate the event log
Aggregate load slow thousands of events snapshots every N events
Lost data on schema migration upcasting bug or missed event version exhaustive tests on every event type version
Sharded events cross-shard transaction aggregate scattered across shards shard key starts with aggregate_id
Projection worker DLQ overflowing downstream issue back-pressure; alert; never silently drop
Wrong projection idempotency duplicates inflate counts dedupe by event _id
ES adopted on a CRUD-friendly domain overengineering rollback to CRUD; lessons learned

Practice Exercises

Exercise 1 — Event store

Build the events collection from Chapter 4 with the unique index. Append 3 events for one aggregate. Try to append a duplicate version — confirm it fails.

Exercise 2 — Command handler

Implement payOrder and shipOrder. Load the aggregate, validate, emit the event. Confirm a payOrder against an already-paid order is rejected.

Exercise 3 — Projection

Build the orders_view projection (Chapter 6). Start the projection worker; run a few commands; confirm the view stays in sync.

Exercise 4 — Snapshot

After 100 simulated events for one aggregate, write a snapshot. Time the load-aggregate function before + after snapshot — confirm speedup.

Exercise 5 — Rebuild

Drop orders_view. Replay from the start (single-pass aggregate). Confirm final state matches what was there before.

Exercise 6 — Schema versioning

Introduce a v2 of order.created that renames items → line_items. Update the apply function to handle both versions. Replay history; confirm same final state.

Cheat-sheet

What How
CQRS separate write + read models
ES events as source of truth
Event store append-only collection, unique (agg, agg_id, version)
Concurrency unique index acts as optimistic CC
Command handler load → validate → emit event
Projection change-stream-driven update of a read view
Multiple views one worker per view per query pattern
Snapshot periodic state cache; load snapshot + events_after
Rebuild projection drop view + replay from start
Event versioning never mutate; handle each version in apply
Outbox + ES combine for cross-service publication
When CRUD most apps; stick with it
When CQRS reads heavy and divergent
When ES regulated audit + replay are mandatory
Pitfall over-engineering when CRUD suffices

What's next

Post 22 builds a production RAG (Retrieval-Augmented Generation) application end-to-end using Atlas Vector Search + LangChain + an embedder + an LLM.

References

Note: this post is in progress. Planned additions: complete TypeScript sample app; comparison with EventStore DB; saga orchestrator example; aggregate boundary heuristics.