Building Projections — Event Sourcing & CQRS

A projection consumes events (from the global log or a subscription) and updates a read model. Make projections idempotent (upserts keyed by entity, track processed positions), store the checkpoint (last processed position) in the same transaction as the read-model update, handle events in order per stream, and design read models per query (denormalize freely). Build several read models from the same events: SQL tables, search indexes, caches, analytics.

Many read models, one event log

Read model	Built for	Store
Order summaries	Customer order history page	SQL table
Order search	Support agents searching orders	Elasticsearch/OpenSearch
Daily revenue	Finance dashboard	Analytics DB / warehouse
Inventory available	Product pages	Redis / SQL

async def run_order_summary_projection(db, store):
    async with db.transaction():
        checkpoint = await db.fetchval("SELECT position FROM checkpoints WHERE name = 'order_summaries'") or 0
    async for batch in store.read_all(after=checkpoint, batch_size=500):
        async with db.transaction():
            for e in batch:
                if e.type == "OrderPlaced":
                    await db.execute('''INSERT INTO order_summaries (id, customer_id, total, status, placed_at)
                                        VALUES ($1,$2,$3,'placed',$4) ON CONFLICT (id) DO NOTHING''',
                                     e.data["order_id"], e.data["customer_id"], e.data["total"], e.recorded_at)
                elif e.type == "OrderShipped":
                    await db.execute("UPDATE order_summaries SET status='shipped' WHERE id=$1", e.data["order_id"])
            await db.execute('''INSERT INTO checkpoints (name, position) VALUES ('order_summaries', $1)
                                ON CONFLICT (name) DO UPDATE SET position = EXCLUDED.position''', batch[-1].position)

export async function project(events: StoredEvent[]) {
  await db.$transaction(async (tx) => {
    for (const e of events) {
      if (e.type === "OrderPlaced") {
        await tx.orderSummary.upsert({ where: { id: e.data.orderId }, update: {},
          create: { id: e.data.orderId, customerId: e.data.customerId, total: e.data.total, status: "placed" } });
      } else if (e.type === "OrderShipped") {
        await tx.orderSummary.update({ where: { id: e.data.orderId }, data: { status: "shipped" } });
      }
    }
    await tx.checkpoint.upsert({ where: { name: "order_summaries" }, update: { position: events.at(-1)!.position },
      create: { name: "order_summaries", position: events.at(-1)!.position } });
  });
}

Common pitfalls

Checkpoint saved separately from the read-model write.
Non-idempotent projection logic.
Projections that call external services (slow, non-replayable).

Key points

Idempotent updates
Checkpoint in the same transaction
Read model per query

Further reading: Azure Architecture: Materialized View pattern

Practice questions

Why store the projection checkpoint in the same transaction as the read-model update?

Answer

So a crash can't leave the read model updated without advancing the checkpoint (duplicates) or the checkpoint advanced without the update (lost events).

Why should projections avoid calling external services?

Answer

Projections must be fast, deterministic and replayable; external calls make rebuilds slow, non-reproducible and side-effecting.