CQRS and On-Demand Read Model Reconstruction Implemented with a PostgreSQL Event Store
Systems in audit-required domains that cannot explain "how this state came to be" carry a structural flaw. Storing a single balance column in an accounts table tells you the current balance, but who changed it, when, and why is buried forever. CQRS + Event Sourcing is the structural answer to this problem. Instead of current state, it uses the history of changes as the source of truth.
This article covers three core topics: an event store schema design that is self-contained within a single PostgreSQL instance, optimistic concurrency control that restores Aggregate state via event replay, and a checkpoint strategy that prevents data loss when rebuilding a Read Model on-demand from an existing event stream. It is structured so you can get started with only PostgreSQL — something your team already knows — without a separate message broker or dedicated event database.
Core Concepts
CQRS: Completely Separate the Write Model from the Read Model
CQRS (Command Query Responsibility Segregation) is a pattern that separates the responsibilities of Commands and Queries. It is a far stronger separation than splitting layers, because the storage itself can differ.
The diagram makes it explicit that the entity sending NOTIFY is the PostgreSQL DB trigger, not the event store service. Write requests are handled by the Command Handler, and read requests are handled by the Query Handler. Each can use an independently optimized data model, so if read load spikes, only the Read Model server needs to scale out.
Event Sourcing: A Sequence of Events — Not Current State — Is the Truth
Let's look at the familiar approach first.
-- Traditional approach: overwrites state
UPDATE accounts SET balance = 950 WHERE id = 'acc-001';
-- Why did 50 get deducted? Who did it? When? → UnknownEvent Sourcing changes this to:
-- Event Sourcing: append events (Append-Only)
INSERT INTO events (aggregate_id, event_type, event_data, sequence_number)
VALUES ('acc-001', 'MoneyWithdrawn', '{"amount": 50, "reason": "ATM"}', 5);
-- Current balance = sum of all deposit/withdrawal events
-- Who, when, and why it was deducted → it's all in the eventsThe current balance is calculated by applying DepositMade and WithdrawalMade events in order from the beginning. Events become the Source of Truth, and current state is a derived value.
Key point: Once an event is saved, it is never modified. If an incorrect withdrawal occurred, you don't delete the withdrawal event — you append a
WithdrawalReversedevent. You record a new fact rather than altering the past.
How PostgreSQL Works as an Event Store
Dedicated event databases (EventStoreDB, EventSourcingDB, etc.) are fine, but you can build a production-grade event store with PostgreSQL as well. The biggest advantage is that you can accommodate heterogeneous event payloads with JSONB while still using plain SQL queries.
CREATE TABLE events (
event_id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
aggregate_id UUID NOT NULL,
aggregate_type TEXT NOT NULL, -- 'Account', 'Order', etc.
event_type TEXT NOT NULL, -- 'MoneyWithdrawn', etc.
event_data JSONB NOT NULL,
sequence_number INTEGER NOT NULL,
global_position BIGSERIAL NOT NULL, -- global order: used as checkpoint for polling recovery and rebuilds
occurred_at TIMESTAMPTZ DEFAULT now()
);
-- Core of optimistic concurrency control
CREATE UNIQUE INDEX idx_events_aggregate_seq
ON events (aggregate_id, sequence_number);
-- Query optimization
CREATE INDEX idx_events_aggregate_id ON events (aggregate_id, occurred_at);
CREATE INDEX idx_events_event_type ON events (event_type);
CREATE INDEX idx_events_global_pos ON events (global_position);The (aggregate_id, sequence_number) unique index is the heart of concurrency control. When two processes try to write an event to the same Aggregate simultaneously, one will always fail with a unique constraint violation. The failing side retries and the conflict is resolved.
global_position is a globally ordered sequence number that the DB fills automatically via BIGSERIAL. It is used as a checkpoint when rebuilding a Read Model, and for polling missed events when a worker reconnects.
Aggregates and Event Streams
Each Aggregate — Account, Order, etc. — has its own event stream keyed by its unique aggregate_id. State restoration works by loading all events for that Aggregate in ascending sequence_number order and applying them in sequence in memory.
An Aggregate is not a separate service — it is an in-memory object inside the application process. As the number of events grows, the cost of replaying all of them each time increases, which is when you introduce Snapshots. You serialize and store the state at a specific version, then replay only the events after that point.
Practical Application
We use a financial account as an example here. It is a domain that simultaneously requires complex state transitions and regulatory audit requirements, so the advantages of this pattern stand out most clearly.
Defining and Storing Event Types
type AccountEvent =
| { type: 'AccountOpened'; ownerId: string; initialBalance: number }
| { type: 'MoneyDeposited'; amount: number; description: string }
| { type: 'MoneyWithdrawn'; amount: number; description: string }
| { type: 'AccountClosed'; reason: string };
async function appendEvents(
aggregateId: string,
events: AccountEvent[],
expectedVersion: number, // optimistic concurrency control
db: Pool
): Promise<void> {
const client = await db.connect();
try {
await client.query('BEGIN');
// In production, prefer a multi-row INSERT over repeated single INSERTs
// (a single-row loop becomes a bottleneck as the number of events grows)
for (let i = 0; i < events.length; i++) {
const event = events[i];
const nextSeq = expectedVersion + i + 1;
await client.query(
`INSERT INTO events
(aggregate_id, aggregate_type, event_type, event_data, sequence_number)
VALUES ($1, 'Account', $2, $3, $4)`,
[aggregateId, event.type, JSON.stringify(event), nextSeq]
);
}
await client.query('COMMIT');
} catch (err) {
await client.query('ROLLBACK');
// PostgreSQL unique constraint violation = concurrency conflict
if ((err as any).code === '23505') {
throw new ConcurrencyError('Concurrent modification conflict. Retry required.');
}
throw err;
} finally {
client.release();
}
}Command Handler: The Core of the Write Flow
In CQRS + Event Sourcing, the Command Handler is the entry point that connects business rule validation with event storage. Let's walk through the entire flow using a withdrawal command as an example.
async function withdrawMoney(
aggregateId: string,
amount: number,
description: string,
db: Pool
): Promise<void> {
// ① Restore current state by replaying events
const state = await loadAccount(aggregateId, db);
// ② Validate business rules (performed in the Write model)
if (state.isClosed) {
throw new Error('Cannot withdraw from a closed account.');
}
if (state.balance < amount) {
throw new Error(`Insufficient balance: current ${state.balance}, requested ${amount}`);
}
// ③ Save event. state.version = last sequence_number → detects concurrency conflicts
await appendEvents(
aggregateId,
[{ type: 'MoneyWithdrawn', amount, description }],
state.version, // expectedVersion
db
);
}The part in ③ where state.version is passed as expectedVersion is where optimistic concurrency control actually operates. If two requests read the same account simultaneously and obtain the same version, the first to write succeeds and the rest receive a ConcurrencyError and retry.
Restoring Aggregate State via Event Replay
interface AccountState {
id: string;
ownerId: string; // extracted from AccountOpened event
balance: number;
isClosed: boolean;
version: number; // last processed sequence_number
}
function applyEvent(state: AccountState, event: AccountEvent): AccountState {
switch (event.type) {
case 'AccountOpened':
return { ...state, ownerId: event.ownerId, balance: event.initialBalance };
case 'MoneyDeposited':
return { ...state, balance: state.balance + event.amount };
case 'MoneyWithdrawn':
return { ...state, balance: state.balance - event.amount };
case 'AccountClosed':
return { ...state, isClosed: true };
}
}
async function loadAccount(aggregateId: string, db: Pool): Promise<AccountState> {
const result = await db.query(
`SELECT event_type, event_data, sequence_number
FROM events
WHERE aggregate_id = $1 AND aggregate_type = 'Account'
ORDER BY sequence_number ASC`,
[aggregateId]
);
const initial: AccountState = {
id: aggregateId,
ownerId: '',
balance: 0,
isClosed: false,
version: 0,
};
return result.rows.reduce((state, row) => ({
...applyEvent(state, row.event_data as AccountEvent),
version: row.sequence_number,
}), initial);
}In the AccountOpened case of applyEvent, event.ownerId must be reflected in the state. If this value is missing, subsequent Read Model INSERTs will fail with an owner_id NOT NULL constraint violation.
Triggering Async Projections with LISTEN/NOTIFY
When an event is saved, PostgreSQL's LISTEN/NOTIFY can be used to notify the projection worker.
CREATE OR REPLACE FUNCTION notify_event_inserted()
RETURNS TRIGGER AS $$
BEGIN
PERFORM pg_notify(
'new_event',
json_build_object(
'event_id', NEW.event_id,
'aggregate_id', NEW.aggregate_id,
'event_type', NEW.event_type,
'global_position', NEW.global_position
)::text -- NOTIFY payload is limited to 8,000 bytes. Send only metadata; fetch data separately
);
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER event_inserted_trigger
AFTER INSERT ON events
FOR EACH ROW EXECUTE FUNCTION notify_event_inserted();// Projection worker: LISTEN + polling fallback combination
async function startProjectionWorker(db: Pool): Promise<void> {
const workerClient = await db.connect();
let lastProcessedPosition = await getLastProcessedPosition(db);
await workerClient.query('LISTEN new_event');
workerClient.on('notification', async (msg) => {
const payload = JSON.parse(msg.payload!);
await updateAccountSummaryReadModel(payload.aggregate_id, db);
lastProcessedPosition = payload.global_position;
await saveLastProcessedPosition(lastProcessedPosition, db);
});
// Poll for events missed during disconnection after reconnecting
// NOTIFY is only delivered while the worker is connected; events during a disconnect are handled here
workerClient.on('connect', async () => {
await catchUpFromPosition(lastProcessedPosition, db);
});
}
async function catchUpFromPosition(fromPosition: number, db: Pool): Promise<void> {
const missed = await db.query(
`SELECT DISTINCT aggregate_id
FROM events
WHERE aggregate_type = 'Account' AND global_position > $1
ORDER BY MIN(global_position) ASC`,
[fromPosition]
);
for (const row of missed.rows) {
await updateAccountSummaryReadModel(row.aggregate_id, db);
}
}Because the NOTIFY payload is limited to 8,000 bytes, you cannot include the full event data in it. The correct approach is to pass only aggregate_id and global_position, as shown, and re-query the DB for the actual state update. When the worker disconnects and reconnects, the connect handler polls for missed events using global_position as a checkpoint.
Read Model Projection: Account Summary View
-- Read Model table: denormalized view optimized for queries
CREATE TABLE account_summary (
account_id UUID PRIMARY KEY,
owner_id TEXT NOT NULL,
balance NUMERIC(15, 2) NOT NULL DEFAULT 0,
is_closed BOOLEAN NOT NULL DEFAULT false,
last_updated TIMESTAMPTZ,
version INTEGER NOT NULL DEFAULT 0
);async function updateAccountSummaryReadModel(
aggregateId: string,
db: Pool
): Promise<void> {
const state = await loadAccount(aggregateId, db);
await db.query(
`INSERT INTO account_summary (account_id, owner_id, balance, is_closed, last_updated, version)
VALUES ($1, $2, $3, $4, now(), $5)
ON CONFLICT (account_id) DO UPDATE
SET owner_id = EXCLUDED.owner_id,
balance = EXCLUDED.balance,
is_closed = EXCLUDED.is_closed,
last_updated = EXCLUDED.last_updated,
version = EXCLUDED.version`,
[state.id, state.ownerId, state.balance, state.isClosed, state.version]
);
}The INSERT ... ON CONFLICT DO UPDATE pattern guarantees idempotency. Even if the same event is processed twice, the result is the same.
On-Demand Rebuild: Preventing Data Loss with Checkpoints
There are two main situations that call for an on-demand rebuild. First, when a new business requirement arises — you can immediately build a new Read Model from existing events alone. Second, when there is a bug in the Read Model calculation logic — after fixing the logic, a rebuild recovers everything without touching the source-of-truth events.
The core challenge of a rebuild is not missing any new events that arrive while the rebuild is in progress. While the rebuild loop runs, new events continue to be applied to the existing account_summary, and swapping the table would discard those changes. A global_position-based checkpoint solves this problem.
async function rebuildReadModel(db: Pool): Promise<void> {
const client = await db.connect();
try {
// pg_advisory_lock is a session-level lock. It is released automatically when the connection drops,
// so we explicitly unlock in the finally block so the next rebuild request can enter immediately
await client.query('SELECT pg_advisory_lock(12345)');
// CREATE TABLE IF NOT EXISTS: prevents errors on retry after partial failure
await client.query(
'CREATE TABLE IF NOT EXISTS account_summary_new (LIKE account_summary INCLUDING ALL)'
);
await client.query('TRUNCATE account_summary_new');
// Checkpoint: record the global position at the time the rebuild starts
const { rows: [{ checkpoint }] } = await client.query(
'SELECT COALESCE(MAX(global_position), 0) AS checkpoint FROM events'
);
// Build new Read Model from events up to the checkpoint
const aggregates = await client.query(
`SELECT DISTINCT aggregate_id FROM events
WHERE aggregate_type = 'Account' AND global_position <= $1`,
[checkpoint]
);
for (const row of aggregates.rows) {
const state = await loadAccountUpTo(row.aggregate_id, checkpoint, db);
await client.query(
`INSERT INTO account_summary_new
(account_id, owner_id, balance, is_closed, last_updated, version)
VALUES ($1, $2, $3, $4, now(), $5)`,
[state.id, state.ownerId, state.balance, state.isClosed, state.version]
);
}
// Catch up events that arrived after the checkpoint (missed during rebuild)
const lateAggregates = await client.query(
`SELECT DISTINCT aggregate_id FROM events
WHERE aggregate_type = 'Account' AND global_position > $1`,
[checkpoint]
);
for (const row of lateAggregates.rows) {
const state = await loadAccount(row.aggregate_id, db);
await client.query(
`INSERT INTO account_summary_new
(account_id, owner_id, balance, is_closed, last_updated, version)
VALUES ($1, $2, $3, $4, now(), $5)
ON CONFLICT (account_id) DO UPDATE
SET owner_id = EXCLUDED.owner_id,
balance = EXCLUDED.balance,
is_closed = EXCLUDED.is_closed,
last_updated = EXCLUDED.last_updated,
version = EXCLUDED.version`,
[state.id, state.ownerId, state.balance, state.isClosed, state.version]
);
}
// Atomic table swap: no downtime for the read service
await client.query('BEGIN');
await client.query('ALTER TABLE account_summary RENAME TO account_summary_old');
await client.query('ALTER TABLE account_summary_new RENAME TO account_summary');
await client.query('DROP TABLE account_summary_old');
await client.query('COMMIT');
} finally {
await client.query('SELECT pg_advisory_unlock(12345)');
client.release();
}
}
// Replay up to a specific checkpoint by specifying an upper bound on global_position
async function loadAccountUpTo(
aggregateId: string,
maxPosition: number,
db: Pool
): Promise<AccountState> {
const result = await db.query(
`SELECT event_type, event_data, sequence_number
FROM events
WHERE aggregate_id = $1 AND aggregate_type = 'Account'
AND global_position <= $2
ORDER BY sequence_number ASC`,
[aggregateId, maxPosition]
);
const initial: AccountState = {
id: aggregateId, ownerId: '', balance: 0, isClosed: false, version: 0,
};
return result.rows.reduce((state, row) => ({
...applyEvent(state, row.event_data as AccountEvent),
version: row.sequence_number,
}), initial);
}Optimizing Replay Performance with Snapshots
When events grow into the millions, replaying all of them every time is not practical.
CREATE TABLE snapshots (
aggregate_id UUID NOT NULL,
aggregate_type TEXT NOT NULL,
state_data JSONB NOT NULL,
version INTEGER NOT NULL, -- last sequence_number reflected
created_at TIMESTAMPTZ DEFAULT now(),
PRIMARY KEY (aggregate_id, version)
);const SNAPSHOT_THRESHOLD = 100; // save a snapshot every 100 events
async function saveSnapshot(state: AccountState, db: Pool): Promise<void> {
await db.query(
`INSERT INTO snapshots (aggregate_id, aggregate_type, state_data, version)
VALUES ($1, 'Account', $2, $3)
ON CONFLICT (aggregate_id, version) DO NOTHING`,
[state.id, JSON.stringify(state), state.version]
);
}
async function loadAccountWithSnapshot(aggregateId: string, db: Pool): Promise<AccountState> {
// 1. Fetch the latest snapshot
const snapResult = await db.query(
`SELECT state_data, version FROM snapshots
WHERE aggregate_id = $1 AND aggregate_type = 'Account'
ORDER BY version DESC LIMIT 1`,
[aggregateId]
);
let state: AccountState;
let fromVersion: number;
if (snapResult.rows.length > 0) {
state = snapResult.rows[0].state_data as AccountState;
fromVersion = snapResult.rows[0].version;
} else {
state = { id: aggregateId, ownerId: '', balance: 0, isClosed: false, version: 0 };
fromVersion = 0;
}
// 2. Replay only events after the snapshot
const events = await db.query(
`SELECT event_type, event_data, sequence_number FROM events
WHERE aggregate_id = $1 AND sequence_number > $2
ORDER BY sequence_number ASC`,
[aggregateId, fromVersion]
);
const finalState = events.rows.reduce((s, row) => ({
...applyEvent(s, row.event_data as AccountEvent),
version: row.sequence_number,
}), state);
// 3. Decide whether to save a snapshot: save if events since last snapshot exceed the threshold
if (events.rows.length >= SNAPSHOT_THRESHOLD) {
await saveSnapshot(finalState, db);
}
return finalState;
}"Every N events" is the simplest and most predictable snapshot timing strategy. There is also a strategy of saving after a specific event type (e.g., AccountClosed), in which case the state no longer changes after that event, making the reconstruction cost permanently near zero. Combining both strategies is also common.
Pros and Cons Analysis
Advantages
| Advantage | Description |
|---|---|
| Automated audit trail | Append-Only event storage automatically preserves change history. No separate audit log system is needed, and financial/medical regulatory requirements are naturally satisfied. |
| Time travel queries | Replaying events only up to a specific point lets you reconstruct the state at any past moment. You can answer precisely "what was the balance just before a fraudulent transaction." |
| Flexible Read Model evolution | When new business requirements arise, replay existing events to generate a new projection. New query patterns are supported without data migration. |
| Independent scalability | Read and write loads are separated and can each be scaled independently. |
| Bug recovery | If a bug is found in Read Model logic, fix the logic and recover by replaying events to rebuild. |
| Event-driven integration | The event stream becomes a natural medium for inter-microservice communication. |
Disadvantages
| Disadvantage | Description |
|---|---|
| Structural complexity | The number of components increases: Command Handler, Event Store, Projection Worker, Read Model, Query Handler, and more. For simple CRUD, this is severe over-engineering. |
| Eventual consistency | Async projections are not reflected in the Read Model immediately after an event is saved. This must be considered from a UX perspective during the design phase. |
| Difficulty evolving event schemas | Past event structures cannot be modified, so a versioning (Upcasting) strategy is required. |
| Rebuild time | Rebuilding large event streams takes considerable time. Checkpoint-based catch-up and Advisory Lock coordination are essential. |
| Snapshot requirement | When events grow into the millions, state restoration performance degrades sharply without snapshots. |
| Operational complexity | Ensuring idempotency, polling catch-up on reconnect, and retry logic all raise operational difficulty. |
Common Mistakes in Practice
1. Applying it to the wrong domain
Introducing Event Sourcing to auxiliary domains like simple post CRUD or configuration management only increases maintenance burden. It is recommended to apply it only to Core Domains that have complex state transitions and audit requirements.
2. Including derived computed values in events
Events record only "what happened." Including the current balance (balanceAfter) in a MoneyWithdrawn event complicates schema evolution. The cleaner approach is to compute the current balance in the projection and store only the value that caused the change in the event.
3. Missing concurrency control
Saving events without expectedVersion causes data inconsistency when two requests modify the same Aggregate simultaneously. The (aggregate_id, sequence_number) unique index and the expectedVersion check must always be implemented together. Passing state.version as expectedVersion in the Command Handler completes this flow.
4. Not guaranteeing projection idempotency
The result must be the same even if an event is processed twice. The INSERT ... ON CONFLICT DO UPDATE pattern is the simplest solution.
5. Applying eventual consistency unconditionally to the entire flow
If the Read Model is queried in the same request before async projection processing is complete, the previous state is returned. For flows where the user must immediately see what they just changed, design that specific part to use on-demand recalculation or mix in a synchronous projection.
6. Trusting LISTEN/NOTIFY alone
NOTIFY is only delivered while the worker is connected. Events during a disconnection are lost. A strategy of using global_position as a checkpoint and polling for missed events on reconnect must always be implemented alongside it.
Closing Thoughts
CQRS + Event Sourcing is a pattern that uses "history of changes" — not "current state" — as the source of truth. This means audit trails come for free, Read Models that match new business requirements can be rebuilt at any time, and reads and writes can each be optimized independently. On the other hand, structural complexity and eventual consistency are trade-offs that must be accepted.
Here are 3 steps you can start with right now.
-
The first step is creating the event store schema in PostgreSQL. Pick one core domain from your existing service, define the event types, and actually create the
eventstable with aglobal_positioncolumn. -
The second step is building one
loadAggregateand one Command Handler. Implementing the flow of reading events, accumulating state withapplyEvent, validating business rules, and saving withappendEventswill give you a fast, hands-on grasp of what Event Sourcing really is. -
The third step is attaching a small projection. Adding a worker that receives event insertions via PostgreSQL LISTEN/NOTIFY and updates a denormalized Read Model table will put the entire CQRS flow in your hands.
Starting with one core domain that has strong audit requirements will naturally reveal when and how far to extend it to the rest of your domains.
References
- Building a Production-Ready Event Store in PostgreSQL: Schema Design, Projections, and Replay - DEV Community
- GitHub - eugene-khyst/postgresql-event-sourcing (Spring Boot reference implementation)
- Rebuilding Event-Driven Read Models in a safe and resilient way - Event-Driven.io
- Projections - Marten .NET Document DB & Event Store
- Designing Read Models - EventSourcingDB official documentation
- CQRS Pattern - Azure Architecture Center
- Event Sourcing and CQRS Pattern - Microservices.io
- Event Sourcing and CQRS with Databases: EventStoreDB, Axon, Polecat
- CQRS and Event Sourcing: Practical Implementation Patterns 2026 - Calmops
- Event Sourcing, CQRS and Micro Services: Real FinTech Example - Medium
- Go EventSourcing and CQRS with PostgreSQL, Kafka, MongoDB and ElasticSearch - DEV Community
- GitHub - fraktalio/fstore-sql
- GitHub - oskardudycz/EventSourcing.NetCore
- How I Built an Aggregateless Event Store with TypeScript and PostgreSQL
- Event Sourcing and Event Storage with Apache Kafka - Confluent
