Implementing Event Sourcing + CQRS in Practice with Kafka and PostgreSQL

When working on a financial domain project, I once received a request: "What was the exact balance of this account three months ago?" At the time, the system only stored the current balance in the DB, leaving change history in a separate audit table — which had been partially lost during a migration. We ended up spending two weeks doing manual log archaeology. Event Sourcing is the architecture that solves that situation structurally.

This article covers how to implement Event Sourcing + CQRS in a production-like manner using TypeScript (NestJS). The event store is implemented as a PostgreSQL events table, and Kafka serves as the streaming layer for Projection consumption. The example domain is a bank account.

After reading this article, you will have a clear picture of three things:

The flow by which an Aggregate produces and reassembles events
The structure in which PostgreSQL serves as the event log and Kafka serves as the streaming backbone
How Projections consume events to build read models, and how time-travel queries work

Core Concepts

Why Store Events Instead of Current State

Traditional CRUD only keeps the current state in the DB. One line — UPDATE accounts SET balance = 5000 WHERE id = 1 — and the previous value is gone. Event Sourcing says: instead of storing "balance is now 5000," store "a deposit of 2000 event occurred."

yaml

Traditional approach: [Current State] balance = 5000
ES approach:          [Event Log]
                        AccountOpened  { initialBalance: 3000 }  // version=1
                        DepositMade    { amount: 2500 }           // version=2
                        WithdrawalMade { amount: 500 }            // version=3

When you need the current balance, replay events from the beginning. When you need "the balance as of before version=2," replay only events with version ≤ 2. This is a time-travel query.

CQRS — Why Separate Reads from Writes

Repeating event replay to derive the current state on every query is far more expensive than a simple DB lookup. This becomes even more of a problem in reality, where read requests vastly outnumber writes. CQRS solves this problem as follows:

Command side: Execute domain logic → store events in the PostgreSQL events table → publish to Kafka. The goal is correctness.
Query side: A Projection consuming Kafka reads from a pre-built read-only view. The goal is speed.

Diagram 1

Aggregate — The Boundary of Domain Consistency

An Aggregate is a collection of domain logic. It receives an external Command, validates it, and on success produces events. An Aggregate does not write directly to the DB. The Command Handler pulls uncommitted events from the Aggregate and saves them to the EventStore.

Architectural Positioning: Separating the Roles of Kafka and PostgreSQL

You often hear "just use Kafka as the EventStore," but there are practical limitations to this. Kafka is not optimized for random access — selectively reading only the events for a specific aggregateId. Spinning up a Consumer to scan the entire topic is too costly for Aggregate rehydration.

This article separates the roles.

Layer	Role
PostgreSQL `events` table	Persistent store for events. Supports fast rehydration and time-travel queries via `aggregate_id + version` index. The enforcement point for optimistic locking.
Kafka `account-events` topic	Append-only streaming backbone for Projection consumption. Projections consume events with their own Consumer Groups to build read models.

The Command Handler INSERTs events into PostgreSQL and then publishes to Kafka — in that order. In this dual-write structure, if the Kafka publish fails after the PostgreSQL commit, the Projection misses the event. In production, the Transactional Outbox pattern — writing the event and an outbox record to PostgreSQL in the same transaction, then having a CDC tool like Debezium deliver it to Kafka — eliminates this risk. This article uses direct publishing to focus on the core flow; please read with that trade-off in mind.

Type Definitions

In a TypeScript production implementation, without type definitions it's hard to follow the code flow. First, define domain events and commands.

typescript

// types/domain-events.ts
import { IEvent } from '@nestjs/cqrs';
 
// Having BaseEvent extend IEvent lets us pass it to NestJS AggregateRoot.apply() without `as any`.
export interface BaseEvent extends IEvent {
  aggregateId: string;
  timestamp: Date;
  version: number;
}
 
export interface AccountOpenedEvent extends BaseEvent {
  type: 'AccountOpened';
  payload: { ownerId: string; initialBalance: number };
}
 
export interface DepositMadeEvent extends BaseEvent {
  type: 'DepositMade';
  payload: { amount: number; description: string };
}
 
export interface WithdrawalMadeEvent extends BaseEvent {
  type: 'WithdrawalMade';
  payload: { amount: number; description: string };
}
 
// Discriminated union — TypeScript narrows the type inside each case of the switch in applyEvent.
export type DomainEvent = AccountOpenedEvent | DepositMadeEvent | WithdrawalMadeEvent;

typescript

// types/commands.ts
export interface OpenAccountCommand {
  accountId: string;
  ownerId: string;
  initialBalance: number;
}
 
export interface DepositCommand {
  accountId: string;
  amount: number;
  description: string;
}
 
export interface WithdrawCommand {
  accountId: string;
  amount: number;
  description: string;
}

Aggregate Implementation

typescript

// account.aggregate.ts
import { AggregateRoot } from '@nestjs/cqrs';
import { DomainEvent } from './types/domain-events';
import { OpenAccountCommand, DepositCommand, WithdrawCommand } from './types/commands';
 
export type AccountStatus = 'ACTIVE' | 'FROZEN' | 'CLOSED';
 
export interface AccountState {
  id: string;
  balance: number;
  status: AccountStatus;
  version: number;
}
 
export class AccountAggregate extends AggregateRoot<DomainEvent> {
  private state: AccountState = { id: '', balance: 0, status: 'ACTIVE', version: 0 };
 
  // NestJS's default loadFromHistory implementation only manages the uncommitted list.
  // We wire in the reduceState call that actually changes domain state here.
  loadFromHistory(events: DomainEvent[]): void {
    events.forEach(event => this.reduceState(event));
  }
 
  static fromSnapshot(state: AccountState): AccountAggregate {
    const agg = new AccountAggregate();
    agg.state = { ...state };
    return agg;
  }
 
  openAccount(command: OpenAccountCommand): void {
    if (this.state.id) {
      throw new Error('Account already exists');
    }
    this.applyAndRecord({
      type: 'AccountOpened',
      aggregateId: command.accountId,
      payload: { ownerId: command.ownerId, initialBalance: command.initialBalance },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  deposit(command: DepositCommand): void {
    if (this.state.status !== 'ACTIVE') {
      throw new Error('Deposits can only be made to active accounts');
    }
    if (command.amount <= 0) {
      throw new Error('Deposit amount must be greater than 0');
    }
    this.applyAndRecord({
      type: 'DepositMade',
      aggregateId: this.state.id,
      payload: { amount: command.amount, description: command.description },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  withdraw(command: WithdrawCommand): void {
    if (this.state.status !== 'ACTIVE') {
      throw new Error('Withdrawals can only be made from active accounts');
    }
    if (command.amount <= 0) {
      throw new Error('Withdrawal amount must be greater than 0');
    }
    if (this.state.balance < command.amount) {
      throw new Error('Insufficient balance');
    }
    this.applyAndRecord({
      type: 'WithdrawalMade',
      aggregateId: this.state.id,
      payload: { amount: command.amount, description: command.description },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  private applyAndRecord(event: DomainEvent): void {
    this.reduceState(event);
    this.apply(event); // NestJS AggregateRoot — adds to the uncommitted list
  }
 
  private reduceState(event: DomainEvent): void {
    switch (event.type) {
      case 'AccountOpened':
        this.state.id = event.aggregateId;
        this.state.balance = event.payload.initialBalance;
        this.state.status = 'ACTIVE';
        break;
      case 'DepositMade':
        this.state.balance += event.payload.amount;
        break;
      case 'WithdrawalMade':
        this.state.balance -= event.payload.amount;
        break;
    }
    this.state.version = event.version;
  }
 
  getState(): AccountState {
    return { ...this.state };
  }
}

Let's clarify why we override loadFromHistory. NestJS AggregateRoot's default loadFromHistory only marks events as "not uncommitted." We have to wire in the reduceState call that changes domain state ourselves. When producing new events, this.apply(event) handles uncommitted list management, and they can be retrieved with getUncommittedEvents() — both of these methods are built into NestJS.

EventStore Implementation

typescript

// event-store/event-store.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { InjectDataSource } from '@nestjs/typeorm';
import { DataSource } from 'typeorm';
import { Kafka, Producer, CompressionTypes } from 'kafkajs';
import { DomainEvent } from '../types/domain-events';
 
export class OptimisticConcurrencyException extends Error {
  constructor(aggregateId: string, expectedVersion: number) {
    super(
      `[${aggregateId}] Version conflict: expected version ${expectedVersion} was already used by another command`,
    );
  }
}
 
@Injectable()
export class EventStoreService implements OnModuleInit {
  private producer: Producer;
 
  constructor(@InjectDataSource() private readonly dataSource: DataSource) {
    const kafka = new Kafka({
      clientId: 'account-service',
      brokers: [process.env.KAFKA_BROKER ?? 'localhost:9092'],
    });
    this.producer = kafka.producer();
  }
 
  async onModuleInit(): Promise<void> {
    await this.producer.connect();
    await this.ensureEventsTable();
  }
 
  async appendEvents(
    aggregateId: string,
    events: DomainEvent[],
    expectedVersion: number,
  ): Promise<void> {
    await this.dataSource.transaction(async (manager) => {
      for (const event of events) {
        try {
          // The UNIQUE (aggregate_id, version) constraint enforces optimistic locking.
          // If two commands try to write the same version concurrently, the second INSERT fails.
          await manager.query(
            `INSERT INTO events (aggregate_id, version, type, payload, timestamp)
             VALUES ($1, $2, $3, $4, $5)`,
            [
              event.aggregateId,
              event.version,
              event.type,
              JSON.stringify(event.payload),
              event.timestamp,
            ],
          );
        } catch (err: any) {
          if (err.code === '23505') {
            throw new OptimisticConcurrencyException(aggregateId, expectedVersion);
          }
          throw err;
        }
      }
    });
 
    // Publish to Kafka after the PostgreSQL transaction commits.
    // In production, use Transactional Outbox + CDC to guarantee atomicity.
    await this.producer.send({
      topic: 'account-events',
      compression: CompressionTypes.GZIP,
      messages: events.map(event => ({
        key: event.aggregateId, // Same Aggregate → same partition → guaranteed ordering
        value: JSON.stringify(event),
        headers: {
          eventType: event.type,
          version: String(event.version),
        },
      })),
    });
  }
 
  async readEvents(
    aggregateId: string,
    options?: { maxTimestamp?: Date; fromVersion?: number },
  ): Promise<DomainEvent[]> {
    const params: unknown[] = [aggregateId];
    let query = `
      SELECT type, payload, aggregate_id AS "aggregateId", version, timestamp
      FROM events
      WHERE aggregate_id = $1
    `;
 
    if (options?.fromVersion !== undefined) {
      params.push(options.fromVersion);
      query += ` AND version > $${params.length}`;
    }
    if (options?.maxTimestamp) {
      params.push(options.maxTimestamp.toISOString());
      query += ` AND timestamp <= $${params.length}`;
    }
 
    query += ' ORDER BY version ASC';
 
    const rows: Array<{
      type: string;
      payload: Record<string, unknown>;
      aggregateId: string;
      version: number;
      timestamp: string;
    }> = await this.dataSource.query(query, params);
 
    return rows.map(row => ({
      type: row.type,
      aggregateId: row.aggregateId,
      version: row.version,
      timestamp: new Date(row.timestamp),
      payload: row.payload,
    } as DomainEvent));
  }
 
  private async ensureEventsTable(): Promise<void> {
    await this.dataSource.query(`
      CREATE TABLE IF NOT EXISTS events (
        id            BIGSERIAL    PRIMARY KEY,
        aggregate_id  TEXT         NOT NULL,
        version       INT          NOT NULL,
        type          TEXT         NOT NULL,
        payload       JSONB        NOT NULL,
        timestamp     TIMESTAMPTZ  NOT NULL,
        CONSTRAINT uq_aggregate_version UNIQUE (aggregate_id, version)
      );
      CREATE INDEX IF NOT EXISTS idx_events_aggregate_id
        ON events (aggregate_id, version);
    `);
  }
}

The UNIQUE (aggregate_id, version) constraint is the heart of optimistic concurrency control. No separate version table or Redis is needed. Since the version auto-increments as the Aggregate is rehydrated and events are produced, when two transactions try to INSERT the same version, the second is rejected by a PostgreSQL constraint violation (23505).

Command Handler

typescript

// commands/deposit.command-handler.ts
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { DepositCommand } from '../types/commands';
import { AccountAggregate } from '../account.aggregate';
import { EventStoreService } from '../event-store/event-store.service';
import { DomainEvent } from '../types/domain-events';
 
@CommandHandler(DepositCommand)
export class DepositCommandHandler implements ICommandHandler<DepositCommand> {
  constructor(private readonly eventStore: EventStoreService) {}
 
  async execute(command: DepositCommand): Promise<void> {
    // 1. Fetch all events for this Aggregate from PostgreSQL
    const events = await this.eventStore.readEvents(command.accountId);
 
    // 2. Reassemble the Aggregate by replaying events
    const account = new AccountAggregate();
    account.loadFromHistory(events);
    const expectedVersion = account.getState().version;
 
    // 3. Execute domain logic (including validation)
    account.deposit(command);
 
    // 4. Pull out uncommitted events and save them — getUncommittedEvents is built into NestJS
    const newEvents = account.getUncommittedEvents() as DomainEvent[];
    await this.eventStore.appendEvents(command.accountId, newEvents, expectedVersion);
 
    account.uncommit();
  }
}

Projection — Transforming Events into Read Models

A Projection is a Kafka Consumer. It consumes events and maintains a query-optimized view in PostgreSQL.

Because Kafka guarantees at-least-once delivery, the same event may be consumed twice. Without a guard, DepositMade could be processed twice, doubling the balance. Two rules prevent this:

WHERE projectionVersion < :version condition: For a version that has already been applied, the UPDATE does not execute because the WHERE condition is not met.
Single UPDATE query: The balance increment and lastUpdatedAt update are bundled into one query to guarantee atomicity. Splitting into two queries risks state inconsistency if the process dies after the first query.

typescript

// projections/account-balance.projection.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { Kafka, Consumer, KafkaMessage } from 'kafkajs';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { AccountBalanceView } from './account-balance-view.entity';
import { DomainEvent } from '../types/domain-events';
 
@Injectable()
export class AccountBalanceProjection implements OnModuleInit {
  private consumer: Consumer;
 
  constructor(
    @InjectRepository(AccountBalanceView)
    private readonly viewRepo: Repository<AccountBalanceView>,
  ) {
    const kafka = new Kafka({
      clientId: 'balance-projection',
      brokers: [process.env.KAFKA_BROKER ?? 'localhost:9092'],
    });
    this.consumer = kafka.consumer({ groupId: 'account-balance-projection' });
  }
 
  async onModuleInit(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'account-events', fromBeginning: true });
    await this.consumer.run({
      eachMessage: async ({ message }) => this.handle(message),
    });
  }
 
  private async handle(message: KafkaMessage): Promise<void> {
    const event = JSON.parse(message.value!.toString()) as DomainEvent;
 
    try {
      switch (event.type) {
        case 'AccountOpened': {
          await this.viewRepo
            .createQueryBuilder()
            .insert()
            .values({
              accountId: event.aggregateId,
              balance: event.payload.initialBalance,
              ownerId: event.payload.ownerId,
              status: 'ACTIVE',
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .orIgnore() // Idempotency: ignore if already exists
            .execute();
          break;
        }
 
        case 'DepositMade': {
          // WHERE projectionVersion < version — skipped on duplicate consumption due to WHERE mismatch
          await this.viewRepo
            .createQueryBuilder()
            .update()
            .set({
              balance: () => `balance + ${event.payload.amount}`,
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .where('accountId = :id AND projectionVersion < :version', {
              id: event.aggregateId,
              version: event.version,
            })
            .execute();
          break;
        }
 
        case 'WithdrawalMade': {
          await this.viewRepo
            .createQueryBuilder()
            .update()
            .set({
              balance: () => `balance - ${event.payload.amount}`,
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .where('accountId = :id AND projectionVersion < :version', {
              id: event.aggregateId,
              version: event.version,
            })
            .execute();
          break;
        }
      }
    } catch (error) {
      if (error instanceof KnownBusinessError) {
        await this.sendToDeadLetterQueue(message, error);
      } else {
        throw error; // Transient errors re-throw → Kafka retries
      }
    }
  }
 
  private async sendToDeadLetterQueue(message: KafkaMessage, error: Error): Promise<void> {
    // Move to 'account-events-dlq' topic for manual review and reprocessing
  }
}

Rebuilding a Projection

When you've fixed a bug or need a new read model, you have to replay from the beginning. The procedure is as follows:

Reset the existing read model table

Reset the Consumer Group offset to the beginning:

bash

kafka-consumer-groups.sh \
  --reset-offsets --to-earliest \
  --group account-balance-projection \
  --topic account-events \
  --execute

Restart the Projection

The reason this is possible is one of the core strengths of Event Sourcing. Because events are never deleted, you can discard a read model at any time and rebuild it from scratch. This is the background behind "being able to create a new view just by adding a Projection" being cited as an advantage.

Time-Travel Queries

typescript

// time-travel/time-travel.service.ts
import { Injectable, NotFoundException, BadRequestException } from '@nestjs/common';
import { EventStoreService } from '../event-store/event-store.service';
import { AccountAggregate, AccountState } from '../account.aggregate';
import { DomainEvent } from '../types/domain-events';
 
@Injectable()
export class TimeTravelService {
  constructor(private readonly eventStore: EventStoreService) {}
 
  async getAccountStateAt(accountId: string, timestamp: Date): Promise<AccountState> {
    const events = await this.eventStore.readEvents(accountId, {
      maxTimestamp: timestamp,
    });
 
    if (events.length === 0) {
      throw new NotFoundException(
        `Account ${accountId} did not exist at ${timestamp.toISOString()}`,
      );
    }
 
    const account = new AccountAggregate();
    account.loadFromHistory(events);
    return account.getState();
  }
 
  async getEventsBetween(
    accountId: string,
    from: Date,
    to: Date,
  ): Promise<DomainEvent[]> {
    const allEvents = await this.eventStore.readEvents(accountId);
    return allEvents.filter(e => {
      const t = new Date(e.timestamp);
      return t >= from && t <= to;
    });
  }
}

typescript

// time-travel/time-travel.controller.ts
import { Controller, Get, Param, Query, BadRequestException } from '@nestjs/common';
import { TimeTravelService } from './time-travel.service';
 
@Controller('accounts')
export class TimeTravelController {
  constructor(private readonly timeTravelService: TimeTravelService) {}
 
  @Get(':id/state')
  async getStateAt(
    @Param('id') accountId: string,
    @Query('at') at: string,
  ) {
    if (!at || isNaN(Date.parse(at))) {
      throw new BadRequestException('The at parameter must be in ISO 8601 format');
    }
    return this.timeTravelService.getAccountStateAt(accountId, new Date(at));
  }
}
 
// GET /accounts/acc-123/state?at=2025-03-15T09:00:00Z

The key to time-travel queries lies in the PostgreSQL timestamp <= $n filter. This query cannot be implemented efficiently by doing a full scan of Kafka. This is why the EventStore was designed around PostgreSQL index-based access.

Diagram 2

Snapshots — When Events Accumulate

When events number in the thousands or tens of thousands, rehydration costs spike. A Snapshot saves the state at a specific version point, shortcutting to replaying only the events after that point.

typescript

// snapshot/snapshot.service.ts
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { SnapshotEntity } from './snapshot.entity';
import { AccountAggregate, AccountState } from '../account.aggregate';
import { EventStoreService } from '../event-store/event-store.service';
 
@Injectable()
export class SnapshotService {
  private readonly SNAPSHOT_INTERVAL = 100;
 
  constructor(
    @InjectRepository(SnapshotEntity)
    private readonly snapshotRepo: Repository<SnapshotEntity>,
  ) {}
 
  async loadAggregate(
    aggregateId: string,
    eventStore: EventStoreService,
  ): Promise<AccountAggregate> {
    const snapshot = await this.snapshotRepo.findOne({
      where: { aggregateId },
      order: { version: 'DESC' },
    });
 
    let account: AccountAggregate;
    let fromVersion: number | undefined;
 
    if (snapshot) {
      // fromSnapshot is a static method defined as AccountAggregate.fromSnapshot(state)
      account = AccountAggregate.fromSnapshot(snapshot.state as AccountState);
      fromVersion = snapshot.version;
    } else {
      account = new AccountAggregate();
    }
 
    const remainingEvents = await eventStore.readEvents(aggregateId, { fromVersion });
    account.loadFromHistory(remainingEvents);
 
    return account;
  }
 
  async maybeSaveSnapshot(account: AccountAggregate): Promise<void> {
    const state = account.getState();
    if (state.version > 0 && state.version % this.SNAPSHOT_INTERVAL === 0) {
      await this.snapshotRepo.save({
        aggregateId: state.id,
        version: state.version,
        state: state,
        createdAt: new Date(),
      });
    }
  }
}

When to Choose Event Sourcing + CQRS

This architecture carries significant complexity. Applying it to every service will actually slow down development velocity.

Diagram 3

There is one core criterion: Is this a domain where "when, who, what, and why something changed" has business value in itself? Financial transactions, medical records, and regulatory compliance fall into this category. Simple content management or storing configuration values likely do not.

Pros and Cons Summary

Category	Description	Practical Impact
Pro	Complete audit trail — no separate audit table needed	★★★★★
Pro	Time-travel queries — accurate reproduction of past state	★★★★★
Pro	Read model flexibility — new views creatable just by adding a Projection	★★★★☆
Pro	Independent scaling of Command/Query	★★★★☆
Pro	Bug reproduction — accurate state restoration via event replay	★★★☆☆
Con	Eventual consistency — delay in Projection updates	★★★★☆
Con	Event replay performance — increased latency on large Aggregates without Snapshots	★★★★☆
Con	Schema evolution difficulty — Upcasting strategy required when event structure changes	★★★☆☆
Con	Increased operational complexity — more moving parts	★★★☆☆
Con	Unsuitable where strong consistency is required	★★★☆☆

Common Mistakes in Practice

Storing Derived Data in Events

Events should only contain "what happened."

typescript

// Bad: derived data is mixed in
interface BadDepositEvent {
  type: 'DepositMade';
  amount: number;
  balanceAfter: number;          // ❌ A derived value the Aggregate can calculate
  allTransactions: unknown[];    // ❌ Too much to put in an event
}
 
// Good: only the fact of what occurred
interface GoodDepositEvent extends BaseEvent {
  type: 'DepositMade';
  payload: { amount: number; description: string };
}

Naming Event Types Using Verb Infinitives

Events are facts that occurred in the past.

typescript

// Bad
'DepositAccount', 'CreateOrder', 'UpdateUser'
 
// Good — the more granularly you split them, the higher the value of the audit log
'DepositMade', 'OrderPlaced', 'UserEmailChanged', 'UserPhoneNumberChanged'

Splitting UserUpdated into specific events like UserEmailChanged and UserPhoneNumberChanged lets anyone reading the audit log immediately know what changed, and makes it easier to subscribe to only specific changes in a Projection.

Infinitely Retrying Failed Projections

Repeatedly retrying the same malformed event will halt the entire Projection.

typescript

await this.consumer.run({
  eachMessage: async ({ message }) => {
    try {
      await this.handle(message);
    } catch (error) {
      if (error instanceof KnownBusinessError) {
        await this.sendToDeadLetterQueue(message, error); // Isolate and reprocess manually
      } else {
        throw error; // Transient error → Kafka retries
      }
    }
  },
});

Introducing It Where Strong Consistency Is Required

If you need strong consistency between reads and writes — such as "inventory must be decremented exactly at the moment of placing an order" — this architecture, which presupposes a Projection update delay, may not be appropriate. Applying it selectively per Bounded Context is the pragmatic approach.

Closing Thoughts

Here is a summary of the structure implemented in this article.

PostgreSQL events table: Persistent store for events. The UNIQUE (aggregate_id, version) constraint handles optimistic locking, and the timestamp filter supports time-travel queries.
Kafka: The streaming layer consumed by Projections. Uses aggregateId as the partition key to guarantee ordering.
Aggregate: The loadFromHistory + getUncommittedEvents pattern clearly separates rehydration from event production.
Projection: The WHERE projectionVersion < :version condition combined with a single UPDATE simultaneously ensures idempotency and atomicity.

Here is the recommended sequence for introducing this architecture for the first time:

Start by defining domain events — List the important state changes in your current system as past-tense events. The more granularly you split them, the higher the value of the audit log.
Start with a single small Aggregate — You don't need to migrate the entire system at once. Pick one independent Bounded Context and apply the EventStore pattern to it.
Add a Projection and a time-travel endpoint — Build a read model with a single Projection, and add a single GET /accounts/:id/state?at=<timestamp> endpoint. Once you have the event history, the rest follows naturally.

References

#EventSourcing#CQRS#Kafka#PostgreSQL#NestJS#TypeScript

Architecture

Implementing Event Sourcing + CQRS in Practice with Kafka and PostgreSQL

After reading this article, you will have a clear picture of three things:

The flow by which an Aggregate produces and reassembles events
The structure in which PostgreSQL serves as the event log and Kafka serves as the streaming backbone
How Projections consume events to build read models, and how time-travel queries work

Core Concepts

Why Store Events Instead of Current State

yaml

Traditional approach: [Current State] balance = 5000
ES approach:          [Event Log]
                        AccountOpened  { initialBalance: 3000 }  // version=1
                        DepositMade    { amount: 2500 }           // version=2
                        WithdrawalMade { amount: 500 }            // version=3

When you need the current balance, replay events from the beginning. When you need "the balance as of before version=2," replay only events with version ≤ 2. This is a time-travel query.

CQRS — Why Separate Reads from Writes

Command side: Execute domain logic → store events in the PostgreSQL events table → publish to Kafka. The goal is correctness.
Query side: A Projection consuming Kafka reads from a pre-built read-only view. The goal is speed.

Diagram 1

Aggregate — The Boundary of Domain Consistency

Architectural Positioning: Separating the Roles of Kafka and PostgreSQL

This article separates the roles.

Layer	Role
PostgreSQL `events` table	Persistent store for events. Supports fast rehydration and time-travel queries via `aggregate_id + version` index. The enforcement point for optimistic locking.
Kafka `account-events` topic	Append-only streaming backbone for Projection consumption. Projections consume events with their own Consumer Groups to build read models.

Type Definitions

In a TypeScript production implementation, without type definitions it's hard to follow the code flow. First, define domain events and commands.

typescript

// types/domain-events.ts
import { IEvent } from '@nestjs/cqrs';
 
// Having BaseEvent extend IEvent lets us pass it to NestJS AggregateRoot.apply() without `as any`.
export interface BaseEvent extends IEvent {
  aggregateId: string;
  timestamp: Date;
  version: number;
}
 
export interface AccountOpenedEvent extends BaseEvent {
  type: 'AccountOpened';
  payload: { ownerId: string; initialBalance: number };
}
 
export interface DepositMadeEvent extends BaseEvent {
  type: 'DepositMade';
  payload: { amount: number; description: string };
}
 
export interface WithdrawalMadeEvent extends BaseEvent {
  type: 'WithdrawalMade';
  payload: { amount: number; description: string };
}
 
// Discriminated union — TypeScript narrows the type inside each case of the switch in applyEvent.
export type DomainEvent = AccountOpenedEvent | DepositMadeEvent | WithdrawalMadeEvent;

typescript

// types/commands.ts
export interface OpenAccountCommand {
  accountId: string;
  ownerId: string;
  initialBalance: number;
}
 
export interface DepositCommand {
  accountId: string;
  amount: number;
  description: string;
}
 
export interface WithdrawCommand {
  accountId: string;
  amount: number;
  description: string;
}

Aggregate Implementation

typescript

// account.aggregate.ts
import { AggregateRoot } from '@nestjs/cqrs';
import { DomainEvent } from './types/domain-events';
import { OpenAccountCommand, DepositCommand, WithdrawCommand } from './types/commands';
 
export type AccountStatus = 'ACTIVE' | 'FROZEN' | 'CLOSED';
 
export interface AccountState {
  id: string;
  balance: number;
  status: AccountStatus;
  version: number;
}
 
export class AccountAggregate extends AggregateRoot<DomainEvent> {
  private state: AccountState = { id: '', balance: 0, status: 'ACTIVE', version: 0 };
 
  // NestJS's default loadFromHistory implementation only manages the uncommitted list.
  // We wire in the reduceState call that actually changes domain state here.
  loadFromHistory(events: DomainEvent[]): void {
    events.forEach(event => this.reduceState(event));
  }
 
  static fromSnapshot(state: AccountState): AccountAggregate {
    const agg = new AccountAggregate();
    agg.state = { ...state };
    return agg;
  }
 
  openAccount(command: OpenAccountCommand): void {
    if (this.state.id) {
      throw new Error('Account already exists');
    }
    this.applyAndRecord({
      type: 'AccountOpened',
      aggregateId: command.accountId,
      payload: { ownerId: command.ownerId, initialBalance: command.initialBalance },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  deposit(command: DepositCommand): void {
    if (this.state.status !== 'ACTIVE') {
      throw new Error('Deposits can only be made to active accounts');
    }
    if (command.amount <= 0) {
      throw new Error('Deposit amount must be greater than 0');
    }
    this.applyAndRecord({
      type: 'DepositMade',
      aggregateId: this.state.id,
      payload: { amount: command.amount, description: command.description },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  withdraw(command: WithdrawCommand): void {
    if (this.state.status !== 'ACTIVE') {
      throw new Error('Withdrawals can only be made from active accounts');
    }
    if (command.amount <= 0) {
      throw new Error('Withdrawal amount must be greater than 0');
    }
    if (this.state.balance < command.amount) {
      throw new Error('Insufficient balance');
    }
    this.applyAndRecord({
      type: 'WithdrawalMade',
      aggregateId: this.state.id,
      payload: { amount: command.amount, description: command.description },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  private applyAndRecord(event: DomainEvent): void {
    this.reduceState(event);
    this.apply(event); // NestJS AggregateRoot — adds to the uncommitted list
  }
 
  private reduceState(event: DomainEvent): void {
    switch (event.type) {
      case 'AccountOpened':
        this.state.id = event.aggregateId;
        this.state.balance = event.payload.initialBalance;
        this.state.status = 'ACTIVE';
        break;
      case 'DepositMade':
        this.state.balance += event.payload.amount;
        break;
      case 'WithdrawalMade':
        this.state.balance -= event.payload.amount;
        break;
    }
    this.state.version = event.version;
  }
 
  getState(): AccountState {
    return { ...this.state };
  }
}

EventStore Implementation

typescript

// event-store/event-store.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { InjectDataSource } from '@nestjs/typeorm';
import { DataSource } from 'typeorm';
import { Kafka, Producer, CompressionTypes } from 'kafkajs';
import { DomainEvent } from '../types/domain-events';
 
export class OptimisticConcurrencyException extends Error {
  constructor(aggregateId: string, expectedVersion: number) {
    super(
      `[${aggregateId}] Version conflict: expected version ${expectedVersion} was already used by another command`,
    );
  }
}
 
@Injectable()
export class EventStoreService implements OnModuleInit {
  private producer: Producer;
 
  constructor(@InjectDataSource() private readonly dataSource: DataSource) {
    const kafka = new Kafka({
      clientId: 'account-service',
      brokers: [process.env.KAFKA_BROKER ?? 'localhost:9092'],
    });
    this.producer = kafka.producer();
  }
 
  async onModuleInit(): Promise<void> {
    await this.producer.connect();
    await this.ensureEventsTable();
  }
 
  async appendEvents(
    aggregateId: string,
    events: DomainEvent[],
    expectedVersion: number,
  ): Promise<void> {
    await this.dataSource.transaction(async (manager) => {
      for (const event of events) {
        try {
          // The UNIQUE (aggregate_id, version) constraint enforces optimistic locking.
          // If two commands try to write the same version concurrently, the second INSERT fails.
          await manager.query(
            `INSERT INTO events (aggregate_id, version, type, payload, timestamp)
             VALUES ($1, $2, $3, $4, $5)`,
            [
              event.aggregateId,
              event.version,
              event.type,
              JSON.stringify(event.payload),
              event.timestamp,
            ],
          );
        } catch (err: any) {
          if (err.code === '23505') {
            throw new OptimisticConcurrencyException(aggregateId, expectedVersion);
          }
          throw err;
        }
      }
    });
 
    // Publish to Kafka after the PostgreSQL transaction commits.
    // In production, use Transactional Outbox + CDC to guarantee atomicity.
    await this.producer.send({
      topic: 'account-events',
      compression: CompressionTypes.GZIP,
      messages: events.map(event => ({
        key: event.aggregateId, // Same Aggregate → same partition → guaranteed ordering
        value: JSON.stringify(event),
        headers: {
          eventType: event.type,
          version: String(event.version),
        },
      })),
    });
  }
 
  async readEvents(
    aggregateId: string,
    options?: { maxTimestamp?: Date; fromVersion?: number },
  ): Promise<DomainEvent[]> {
    const params: unknown[] = [aggregateId];
    let query = `
      SELECT type, payload, aggregate_id AS "aggregateId", version, timestamp
      FROM events
      WHERE aggregate_id = $1
    `;
 
    if (options?.fromVersion !== undefined) {
      params.push(options.fromVersion);
      query += ` AND version > $${params.length}`;
    }
    if (options?.maxTimestamp) {
      params.push(options.maxTimestamp.toISOString());
      query += ` AND timestamp <= $${params.length}`;
    }
 
    query += ' ORDER BY version ASC';
 
    const rows: Array<{
      type: string;
      payload: Record<string, unknown>;
      aggregateId: string;
      version: number;
      timestamp: string;
    }> = await this.dataSource.query(query, params);
 
    return rows.map(row => ({
      type: row.type,
      aggregateId: row.aggregateId,
      version: row.version,
      timestamp: new Date(row.timestamp),
      payload: row.payload,
    } as DomainEvent));
  }
 
  private async ensureEventsTable(): Promise<void> {
    await this.dataSource.query(`
      CREATE TABLE IF NOT EXISTS events (
        id            BIGSERIAL    PRIMARY KEY,
        aggregate_id  TEXT         NOT NULL,
        version       INT          NOT NULL,
        type          TEXT         NOT NULL,
        payload       JSONB        NOT NULL,
        timestamp     TIMESTAMPTZ  NOT NULL,
        CONSTRAINT uq_aggregate_version UNIQUE (aggregate_id, version)
      );
      CREATE INDEX IF NOT EXISTS idx_events_aggregate_id
        ON events (aggregate_id, version);
    `);
  }
}

Command Handler

typescript

// commands/deposit.command-handler.ts
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { DepositCommand } from '../types/commands';
import { AccountAggregate } from '../account.aggregate';
import { EventStoreService } from '../event-store/event-store.service';
import { DomainEvent } from '../types/domain-events';
 
@CommandHandler(DepositCommand)
export class DepositCommandHandler implements ICommandHandler<DepositCommand> {
  constructor(private readonly eventStore: EventStoreService) {}
 
  async execute(command: DepositCommand): Promise<void> {
    // 1. Fetch all events for this Aggregate from PostgreSQL
    const events = await this.eventStore.readEvents(command.accountId);
 
    // 2. Reassemble the Aggregate by replaying events
    const account = new AccountAggregate();
    account.loadFromHistory(events);
    const expectedVersion = account.getState().version;
 
    // 3. Execute domain logic (including validation)
    account.deposit(command);
 
    // 4. Pull out uncommitted events and save them — getUncommittedEvents is built into NestJS
    const newEvents = account.getUncommittedEvents() as DomainEvent[];
    await this.eventStore.appendEvents(command.accountId, newEvents, expectedVersion);
 
    account.uncommit();
  }
}

Projection — Transforming Events into Read Models

A Projection is a Kafka Consumer. It consumes events and maintains a query-optimized view in PostgreSQL.

Because Kafka guarantees at-least-once delivery, the same event may be consumed twice. Without a guard, DepositMade could be processed twice, doubling the balance. Two rules prevent this:

WHERE projectionVersion < :version condition: For a version that has already been applied, the UPDATE does not execute because the WHERE condition is not met.
Single UPDATE query: The balance increment and lastUpdatedAt update are bundled into one query to guarantee atomicity. Splitting into two queries risks state inconsistency if the process dies after the first query.

typescript

// projections/account-balance.projection.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { Kafka, Consumer, KafkaMessage } from 'kafkajs';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { AccountBalanceView } from './account-balance-view.entity';
import { DomainEvent } from '../types/domain-events';
 
@Injectable()
export class AccountBalanceProjection implements OnModuleInit {
  private consumer: Consumer;
 
  constructor(
    @InjectRepository(AccountBalanceView)
    private readonly viewRepo: Repository<AccountBalanceView>,
  ) {
    const kafka = new Kafka({
      clientId: 'balance-projection',
      brokers: [process.env.KAFKA_BROKER ?? 'localhost:9092'],
    });
    this.consumer = kafka.consumer({ groupId: 'account-balance-projection' });
  }
 
  async onModuleInit(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'account-events', fromBeginning: true });
    await this.consumer.run({
      eachMessage: async ({ message }) => this.handle(message),
    });
  }
 
  private async handle(message: KafkaMessage): Promise<void> {
    const event = JSON.parse(message.value!.toString()) as DomainEvent;
 
    try {
      switch (event.type) {
        case 'AccountOpened': {
          await this.viewRepo
            .createQueryBuilder()
            .insert()
            .values({
              accountId: event.aggregateId,
              balance: event.payload.initialBalance,
              ownerId: event.payload.ownerId,
              status: 'ACTIVE',
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .orIgnore() // Idempotency: ignore if already exists
            .execute();
          break;
        }
 
        case 'DepositMade': {
          // WHERE projectionVersion < version — skipped on duplicate consumption due to WHERE mismatch
          await this.viewRepo
            .createQueryBuilder()
            .update()
            .set({
              balance: () => `balance + ${event.payload.amount}`,
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .where('accountId = :id AND projectionVersion < :version', {
              id: event.aggregateId,
              version: event.version,
            })
            .execute();
          break;
        }
 
        case 'WithdrawalMade': {
          await this.viewRepo
            .createQueryBuilder()
            .update()
            .set({
              balance: () => `balance - ${event.payload.amount}`,
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .where('accountId = :id AND projectionVersion < :version', {
              id: event.aggregateId,
              version: event.version,
            })
            .execute();
          break;
        }
      }
    } catch (error) {
      if (error instanceof KnownBusinessError) {
        await this.sendToDeadLetterQueue(message, error);
      } else {
        throw error; // Transient errors re-throw → Kafka retries
      }
    }
  }
 
  private async sendToDeadLetterQueue(message: KafkaMessage, error: Error): Promise<void> {
    // Move to 'account-events-dlq' topic for manual review and reprocessing
  }
}

Rebuilding a Projection

When you've fixed a bug or need a new read model, you have to replay from the beginning. The procedure is as follows:

Reset the existing read model table

Reset the Consumer Group offset to the beginning:

bash

kafka-consumer-groups.sh \
  --reset-offsets --to-earliest \
  --group account-balance-projection \
  --topic account-events \
  --execute

Restart the Projection

Time-Travel Queries

typescript

// time-travel/time-travel.service.ts
import { Injectable, NotFoundException, BadRequestException } from '@nestjs/common';
import { EventStoreService } from '../event-store/event-store.service';
import { AccountAggregate, AccountState } from '../account.aggregate';
import { DomainEvent } from '../types/domain-events';
 
@Injectable()
export class TimeTravelService {
  constructor(private readonly eventStore: EventStoreService) {}
 
  async getAccountStateAt(accountId: string, timestamp: Date): Promise<AccountState> {
    const events = await this.eventStore.readEvents(accountId, {
      maxTimestamp: timestamp,
    });
 
    if (events.length === 0) {
      throw new NotFoundException(
        `Account ${accountId} did not exist at ${timestamp.toISOString()}`,
      );
    }
 
    const account = new AccountAggregate();
    account.loadFromHistory(events);
    return account.getState();
  }
 
  async getEventsBetween(
    accountId: string,
    from: Date,
    to: Date,
  ): Promise<DomainEvent[]> {
    const allEvents = await this.eventStore.readEvents(accountId);
    return allEvents.filter(e => {
      const t = new Date(e.timestamp);
      return t >= from && t <= to;
    });
  }
}

typescript

// time-travel/time-travel.controller.ts
import { Controller, Get, Param, Query, BadRequestException } from '@nestjs/common';
import { TimeTravelService } from './time-travel.service';
 
@Controller('accounts')
export class TimeTravelController {
  constructor(private readonly timeTravelService: TimeTravelService) {}
 
  @Get(':id/state')
  async getStateAt(
    @Param('id') accountId: string,
    @Query('at') at: string,
  ) {
    if (!at || isNaN(Date.parse(at))) {
      throw new BadRequestException('The at parameter must be in ISO 8601 format');
    }
    return this.timeTravelService.getAccountStateAt(accountId, new Date(at));
  }
}
 
// GET /accounts/acc-123/state?at=2025-03-15T09:00:00Z

Diagram 2

Snapshots — When Events Accumulate

When events number in the thousands or tens of thousands, rehydration costs spike. A Snapshot saves the state at a specific version point, shortcutting to replaying only the events after that point.

typescript

// snapshot/snapshot.service.ts
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { SnapshotEntity } from './snapshot.entity';
import { AccountAggregate, AccountState } from '../account.aggregate';
import { EventStoreService } from '../event-store/event-store.service';
 
@Injectable()
export class SnapshotService {
  private readonly SNAPSHOT_INTERVAL = 100;
 
  constructor(
    @InjectRepository(SnapshotEntity)
    private readonly snapshotRepo: Repository<SnapshotEntity>,
  ) {}
 
  async loadAggregate(
    aggregateId: string,
    eventStore: EventStoreService,
  ): Promise<AccountAggregate> {
    const snapshot = await this.snapshotRepo.findOne({
      where: { aggregateId },
      order: { version: 'DESC' },
    });
 
    let account: AccountAggregate;
    let fromVersion: number | undefined;
 
    if (snapshot) {
      // fromSnapshot is a static method defined as AccountAggregate.fromSnapshot(state)
      account = AccountAggregate.fromSnapshot(snapshot.state as AccountState);
      fromVersion = snapshot.version;
    } else {
      account = new AccountAggregate();
    }
 
    const remainingEvents = await eventStore.readEvents(aggregateId, { fromVersion });
    account.loadFromHistory(remainingEvents);
 
    return account;
  }
 
  async maybeSaveSnapshot(account: AccountAggregate): Promise<void> {
    const state = account.getState();
    if (state.version > 0 && state.version % this.SNAPSHOT_INTERVAL === 0) {
      await this.snapshotRepo.save({
        aggregateId: state.id,
        version: state.version,
        state: state,
        createdAt: new Date(),
      });
    }
  }
}

When to Choose Event Sourcing + CQRS

This architecture carries significant complexity. Applying it to every service will actually slow down development velocity.

Diagram 3

Pros and Cons Summary

Category	Description	Practical Impact
Pro	Complete audit trail — no separate audit table needed	★★★★★
Pro	Time-travel queries — accurate reproduction of past state	★★★★★
Pro	Read model flexibility — new views creatable just by adding a Projection	★★★★☆
Pro	Independent scaling of Command/Query	★★★★☆
Pro	Bug reproduction — accurate state restoration via event replay	★★★☆☆
Con	Eventual consistency — delay in Projection updates	★★★★☆
Con	Event replay performance — increased latency on large Aggregates without Snapshots	★★★★☆
Con	Schema evolution difficulty — Upcasting strategy required when event structure changes	★★★☆☆
Con	Increased operational complexity — more moving parts	★★★☆☆
Con	Unsuitable where strong consistency is required	★★★☆☆

Common Mistakes in Practice

Storing Derived Data in Events

Events should only contain "what happened."

typescript

// Bad: derived data is mixed in
interface BadDepositEvent {
  type: 'DepositMade';
  amount: number;
  balanceAfter: number;          // ❌ A derived value the Aggregate can calculate
  allTransactions: unknown[];    // ❌ Too much to put in an event
}
 
// Good: only the fact of what occurred
interface GoodDepositEvent extends BaseEvent {
  type: 'DepositMade';
  payload: { amount: number; description: string };
}

Naming Event Types Using Verb Infinitives

Events are facts that occurred in the past.

typescript

// Bad
'DepositAccount', 'CreateOrder', 'UpdateUser'
 
// Good — the more granularly you split them, the higher the value of the audit log
'DepositMade', 'OrderPlaced', 'UserEmailChanged', 'UserPhoneNumberChanged'

Infinitely Retrying Failed Projections

Repeatedly retrying the same malformed event will halt the entire Projection.

typescript

await this.consumer.run({
  eachMessage: async ({ message }) => {
    try {
      await this.handle(message);
    } catch (error) {
      if (error instanceof KnownBusinessError) {
        await this.sendToDeadLetterQueue(message, error); // Isolate and reprocess manually
      } else {
        throw error; // Transient error → Kafka retries
      }
    }
  },
});

Introducing It Where Strong Consistency Is Required

Closing Thoughts

Here is a summary of the structure implemented in this article.

PostgreSQL events table: Persistent store for events. The UNIQUE (aggregate_id, version) constraint handles optimistic locking, and the timestamp filter supports time-travel queries.
Kafka: The streaming layer consumed by Projections. Uses aggregateId as the partition key to guarantee ordering.
Aggregate: The loadFromHistory + getUncommittedEvents pattern clearly separates rehydration from event production.
Projection: The WHERE projectionVersion < :version condition combined with a single UPDATE simultaneously ensures idempotency and atomicity.

Here is the recommended sequence for introducing this architecture for the first time:

Start by defining domain events — List the important state changes in your current system as past-tense events. The more granularly you split them, the higher the value of the audit log.
Start with a single small Aggregate — You don't need to migrate the entire system at once. Pick one independent Bounded Context and apply the EventStore pattern to it.
Add a Projection and a time-travel endpoint — Build a read model with a single Projection, and add a single GET /accounts/:id/state?at=<timestamp> endpoint. Once you have the event history, the rest follows naturally.

References

#EventSourcing#CQRS#Kafka#PostgreSQL#NestJS#TypeScript

Core Concepts

Why Store Events Instead of Current State

CQRS — Why Separate Reads from Writes

Aggregate — The Boundary of Domain Consistency

Architectural Positioning: Separating the Roles of Kafka and PostgreSQL

Type Definitions

Aggregate Implementation

EventStore Implementation

Command Handler

Projection — Transforming Events into Read Models

Rebuilding a Projection

Time-Travel Queries

Snapshots — When Events Accumulate

When to Choose Event Sourcing + CQRS

Pros and Cons Summary

Common Mistakes in Practice

Storing Derived Data in Events

Naming Event Types Using Verb Infinitives

Infinitely Retrying Failed Projections

Introducing It Where Strong Consistency Is Required

Closing Thoughts

References

Core Concepts

Why Store Events Instead of Current State

CQRS — Why Separate Reads from Writes

Aggregate — The Boundary of Domain Consistency

Architectural Positioning: Separating the Roles of Kafka and PostgreSQL

Type Definitions

Aggregate Implementation

EventStore Implementation

Command Handler

Projection — Transforming Events into Read Models

Rebuilding a Projection

Time-Travel Queries

Snapshots — When Events Accumulate

When to Choose Event Sourcing + CQRS

Pros and Cons Summary

Common Mistakes in Practice

Storing Derived Data in Events

Naming Event Types Using Verb Infinitives

Infinitely Retrying Failed Projections

Introducing It Where Strong Consistency Is Required

Closing Thoughts

References

Recommended Posts

Implementing Zero-Downtime PostgreSQL Schema Migrations with the Expand-Contract Pattern

CQRS and On-Demand Read Model Reconstruction Implemented with a PostgreSQL Event Store

Real-time PostgreSQL Sync to the Browser with ElectricSQL Shapes API — A sync-engine Pattern for Connecting Offline Writes and Server DB in Local-First Web Apps

TypeScript로 CQRS + Event Sourcing 직접 구현하기 — EventStore, 이벤트 리플레이, Projection까지

Building an Order and Payment State Machine with TypeScript and KurrentDB Using Event Sourcing and Point-in-Time Reconstruction

Event Sourcing + CQRS TypeScript Implementation Guide — A practical architecture that separates commands and queries while preserving all state changes as an event log