Design event-sourced systems with CQRS — event stores, aggregate roots, projections, snapshots, and replay for auditable, scalable domain architectures
Design event-sourced systems with CQRS — event stores, aggregate roots, projections, snapshots, and replay for auditable, scalable domain architectures
["Domain and aggregate boundaries","Consistency requirements (strong vs. eventual)","Query patterns (read:write ratio, required projections)","Infrastructure (PostgreSQL, EventStoreDB, Kafka)","Scale expectations (events/sec, total event count)"]
outputs
["Event schema definitions and versioning strategy","Event store implementation (append-only log)","Aggregate root with command handling","Projection builders for read models","Snapshot strategy for performance","Replay and migration tooling"]
Event sourcing stores every state change as an immutable event in an append-only log, rather than storing only the current state. The current state is derived by replaying events. Combined with CQRS (Command Query Responsibility Segregation), this pattern enables complete audit trails, temporal queries ("what was the state at time T?"), and independent scaling of reads and writes. It is the correct architecture when auditability, traceability, or complex domain logic are primary requirements.
-- Append-only event store tableCREATE TABLE events (
event_id UUID PRIMARY KEYDEFAULT gen_random_uuid(),
aggregate_id UUID NOT NULL,
aggregate_type VARCHAR(100) NOT NULL,
event_type VARCHAR(100) NOT NULL,
version INTEGERNOT NULL,
payload JSONB NOT NULL,
metadata JSONB NOT NULLDEFAULT'{}',
created_at TIMESTAMPTZ NOT NULLDEFAULT NOW(),
-- Optimistic concurrency: no two events for the same aggregate at the same versionUNIQUE (aggregate_id, version)
);
CREATE INDEX idx_events_aggregate ON events (aggregate_id, version);
CREATE INDEX idx_events_type ON events (event_type, created_at);
-- Snapshot table for performanceCREATE TABLE snapshots (
aggregate_id UUID PRIMARY KEY,
aggregate_type VARCHAR(100) NOT NULL,
version INTEGERNOT NULL,
state JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULLDEFAULT NOW()
);
// event-store.tsimport { Pool } from'pg';
interfaceStoredEvent {
eventId: string;
aggregateId: string;
aggregateType: string;
eventType: string;
version: number;
payload: Record<string, unknown>;
metadata: Record<string, unknown>;
createdAt: Date;
}
exportclassPostgresEventStore {
constructor(privatepool: Pool) {}
asyncappendEvents(
aggregateId: string,
aggregateType: string,
events: Array<{ type: string; payload: Record<string, unknown> }>,
expectedVersion: number
): Promise<StoredEvent[]> {
const client = awaitthis.pool.connect();
try {
await client.query('BEGIN');
// Optimistic concurrency checkconst { rows } = await client.query(
'SELECT MAX(version) as max_version FROM events WHERE aggregate_id = $1',
[aggregateId]
);
const currentVersion = rows[0].max_version ?? 0;
if (currentVersion !== expectedVersion) {
thrownewConcurrencyError(
`Expected version ${expectedVersion}, but aggregate is at version ${currentVersion}`
);
}
conststored: StoredEvent[] = [];
for (let i = 0; i < events.length; i++) {
const event = events[i];
const version = expectedVersion + i + 1;
const result = await client.query(
`INSERT INTO events (aggregate_id, aggregate_type, event_type, version, payload, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING *`,
[aggregateId, aggregateType, event.type, version, event.payload, {}]
);
stored.push(this.mapRow(result.rows[0]));
}
await client.query('COMMIT');
return stored;
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
asyncgetEvents(aggregateId: string, afterVersion = 0): Promise<StoredEvent[]> {
const { rows } = awaitthis.pool.query(
'SELECT * FROM events WHERE aggregate_id = $1 AND version > $2 ORDER BY version',
[aggregateId, afterVersion]
);
return rows.map(this.mapRow);
}
asyncgetAllEvents(afterTimestamp?: Date, limit = 1000): Promise<StoredEvent[]> {
const { rows } = awaitthis.pool.query(
`SELECT * FROM events
WHERE ($1::timestamptz IS NULL OR created_at > $1)
ORDER BY created_at, version
LIMIT $2`,
[afterTimestamp ?? null, limit]
);
return rows.map(this.mapRow);
}
privatemapRow(row: Record<string, unknown>): StoredEvent {
return {
eventId: row.event_idasstring,
aggregateId: row.aggregate_idasstring,
aggregateType: row.aggregate_typeasstring,
eventType: row.event_typeasstring,
version: row.versionasnumber,
payload: row.payloadasRecord<string, unknown>,
metadata: row.metadataasRecord<string, unknown>,
createdAt: row.created_atasDate,
};
}
}
classConcurrencyErrorextendsError {
constructor(message: string) {
super(message);
this.name = 'ConcurrencyError';
}
}
Name events in past tense. Events represent facts that already happened: OrderCreated, not CreateOrder. Commands are imperative: CreateOrder.
Events are immutable. Never modify or delete events. Use compensating events (OrderCancelled) to reverse a previous event's effect.
Keep aggregates small. An aggregate with thousands of events is slow to rehydrate. Use snapshots every ~100 events or when replay exceeds 50ms.
Version your events from day one. Event schemas will change. Use upcasters to transform old events to the current format on read.
Projections are disposable. They can always be rebuilt from the event log. Do not treat projection tables as the source of truth.
Use optimistic concurrency on the event store. The (aggregate_id, version) unique constraint prevents two concurrent commands from producing conflicting events.
Common Pitfalls
Pitfall
Symptom
Fix
Events too granular
Thousands of tiny events per aggregate, slow replay
Combine related changes into meaningful domain events
Events too coarse
OrderUpdated with entire state diff — loses meaning
Each event should represent a single domain-meaningful action
Missing optimistic concurrency
Lost updates when two commands modify same aggregate
Add (aggregate_id, version) unique constraint; retry on conflict
Projection treated as source of truth
Data loss when projection DB fails
Always rebuild projections from event store; never write to event store from projection
No snapshot strategy
Aggregates with 10K+ events take seconds to load
Snapshot every N events or on a time schedule
Coupling projections to event store writes
Event store write fails if projection is down
Project asynchronously; use outbox pattern or CDC
No event versioning
Old events break new code after schema change
Implement upcasters from the start; include schema version in event metadata
Giant aggregate boundaries
Contention and serialization bottleneck
Split into smaller aggregates; use sagas for cross-aggregate coordination
