← Back to DevBytes
Developer Tutorial July 08, 2026 14 min read

Event-Driven Architecture: Event Sourcing and CQRS

Understanding Event Sourcing and CQRS

Event-Driven Architecture (EDA) has become a cornerstone of modern distributed systems. Two patterns that sit at the heart of EDA are Event Sourcing and Command Query Responsibility Segregation (CQRS). When used together, they enable systems that are auditable, scalable, and capable of handling complex business logic with remarkable clarity. This tutorial takes you from first principles through to a complete, production-ready implementation.

What Is Event Sourcing?

Event Sourcing is a persistence pattern where you store state changes as a sequence of immutable events rather than storing the current state itself. Instead of saving a row in a database representing an entity's current values, you append an event to an event store describing what happened.

Think of it like a version control system for your data. Git doesn't store the latest snapshot of your codebase alone — it stores every commit (event) that ever happened. You can replay those commits to reconstruct any historical state. Event Sourcing works the same way:

What Is CQRS?

CQRS stands for Command Query Responsibility Segregation. It separates read operations (queries) from write operations (commands). In a traditional CRUD architecture, the same data model and database handle both reads and writes. With CQRS, you have distinct models:

The separation allows you to optimize each side independently — the write side for consistency and business logic, the read side for query performance and specific use-case needs.

Why Event Sourcing and CQRS Matter

Together, these patterns unlock powerful capabilities that are difficult to achieve with traditional architectures:

How to Implement Event Sourcing — Step by Step

1. Define Your Domain Events

Events are immutable facts about what happened. They should be named in the past tense and carry all relevant data. Here's an example for a bank account domain:

// domain-events.ts
export interface DomainEvent {
  readonly eventType: string;
  readonly aggregateId: string;
  readonly timestamp: number;
  readonly version: number;
}

export class AccountOpened implements DomainEvent {
  readonly eventType = 'AccountOpened';
  readonly version: number;

  constructor(
    readonly aggregateId: string,
    readonly accountHolder: string,
    readonly initialDeposit: number,
    readonly timestamp: number = Date.now(),
    version?: number
  ) {
    this.version = version ?? 1;
  }
}

export class MoneyDeposited implements DomainEvent {
  readonly eventType = 'MoneyDeposited';
  readonly version: number;

  constructor(
    readonly aggregateId: string,
    readonly amount: number,
    readonly description: string,
    readonly timestamp: number = Date.now(),
    version?: number
  ) {
    this.version = version ?? 1;
  }
}

export class MoneyWithdrawn implements DomainEvent {
  readonly eventType = 'MoneyWithdrawn';
  readonly version: number;

  constructor(
    readonly aggregateId: string,
    readonly amount: number,
    readonly description: string,
    readonly timestamp: number = Date.now(),
    version?: number
  ) {
    this.version = version ?? 1;
  }
}

export class AccountClosed implements DomainEvent {
  readonly eventType = 'AccountClosed';
  readonly version: number;

  constructor(
    readonly aggregateId: string,
    readonly reason: string,
    readonly timestamp: number = Date.now(),
    version?: number
  ) {
    this.version = version ?? 1;
  }
}

// Union type for all account events
export type AccountEvent =
  | AccountOpened
  | MoneyDeposited
  | MoneyWithdrawn
  | AccountClosed;

2. Build the Aggregate

An aggregate is a cluster of domain objects that are treated as a single unit for data changes. It enforces business rules (invariants) and produces events. The aggregate's state is rebuilt by replaying events:

// account-aggregate.ts
import { AccountEvent, AccountOpened, MoneyDeposited, MoneyWithdrawn, AccountClosed } from './domain-events';

export class AccountAggregate {
  private _id: string = '';
  private _holder: string = '';
  private _balance: number = 0;
  private _isClosed: boolean = false;
  private _version: number = 0;

  // Rehydrate the aggregate from an event history
  static fromHistory(events: AccountEvent[]): AccountAggregate {
    const account = new AccountAggregate();
    for (const event of events) {
      account.applyEvent(event);
    }
    return account;
  }

  // Expose current state for queries (read model building)
  get state() {
    return {
      id: this._id,
      holder: this._holder,
      balance: this._balance,
      isClosed: this._isClosed,
      version: this._version,
    };
  }

  // Command: Open a new account
  openAccount(id: string, holder: string, initialDeposit: number): AccountEvent[] {
    if (this._id) {
      throw new Error('Account already exists. Cannot open again.');
    }
    if (initialDeposit < 0) {
      throw new Error('Initial deposit cannot be negative.');
    }
    return [new AccountOpened(id, holder, initialDeposit, Date.now(), this._version + 1)];
  }

  // Command: Deposit money
  deposit(amount: number, description: string): AccountEvent[] {
    this.assertAccountActive();
    if (amount <= 0) {
      throw new Error('Deposit amount must be positive.');
    }
    return [new MoneyDeposited(this._id, amount, description, Date.now(), this._version + 1)];
  }

  // Command: Withdraw money
  withdraw(amount: number, description: string): AccountEvent[] {
    this.assertAccountActive();
    if (amount <= 0) {
      throw new Error('Withdrawal amount must be positive.');
    }
    if (this._balance - amount < 0) {
      throw new Error('Insufficient funds.');
    }
    return [new MoneyWithdrawn(this._id, amount, description, Date.now(), this._version + 1)];
  }

  // Command: Close account
  close(reason: string): AccountEvent[] {
    this.assertAccountActive();
    return [new AccountClosed(this._id, reason, Date.now(), this._version + 1)];
  }

  // Apply an event to mutate internal state (used during replay and for new events)
  private applyEvent(event: AccountEvent): void {
    switch (event.eventType) {
      case 'AccountOpened': {
        const e = event as AccountOpened;
        this._id = e.aggregateId;
        this._holder = e.accountHolder;
        this._balance = e.initialDeposit;
        this._isClosed = false;
        this._version = e.version;
        break;
      }
      case 'MoneyDeposited': {
        const e = event as MoneyDeposited;
        this._balance += e.amount;
        this._version = e.version;
        break;
      }
      case 'MoneyWithdrawn': {
        const e = event as MoneyWithdrawn;
        this._balance -= e.amount;
        this._version = e.version;
        break;
      }
      case 'AccountClosed': {
        const e = event as AccountClosed;
        this._isClosed = true;
        this._version = e.version;
        break;
      }
      default:
        throw new Error(`Unknown event type: ${(event as any).eventType}`);
    }
  }

  private assertAccountActive(): void {
    if (!this._id) {
      throw new Error('Account does not exist.');
    }
    if (this._isClosed) {
      throw new Error('Account is closed. No operations allowed.');
    }
  }
}

3. Implement the Event Store

The event store is responsible for persisting events and providing them back in order. A production event store often uses a dedicated database like EventStoreDB, or a pattern built on top of PostgreSQL or MongoDB. Here's a simplified in-memory version that mimics the essential contract:

// event-store.ts
import { AccountEvent } from './domain-events';

export interface IEventStore {
  append(aggregateId: string, events: AccountEvent[], expectedVersion: number): Promise;
  getEvents(aggregateId: string): Promise;
  getAllEvents(): AsyncIterable;
}

export class InMemoryEventStore implements IEventStore {
  private streams: Map = new Map();

  async append(aggregateId: string, events: AccountEvent[], expectedVersion: number): Promise {
    const existingEvents = this.streams.get(aggregateId) || [];
    const currentVersion = existingEvents.length > 0
      ? existingEvents[existingEvents.length - 1].version
      : 0;

    // Optimistic concurrency check
    if (expectedVersion !== currentVersion) {
      throw new Error(
        `Concurrency conflict for aggregate ${aggregateId}: ` +
        `expected version ${expectedVersion}, actual version ${currentVersion}`
      );
    }

    // Assign correct version numbers
    let nextVersion = currentVersion + 1;
    for (const event of events) {
      // Ensure version continuity
      const versionedEvent = { ...event, version: nextVersion } as AccountEvent;
      existingEvents.push(versionedEvent);
      nextVersion++;
    }

    this.streams.set(aggregateId, existingEvents);
  }

  async getEvents(aggregateId: string): Promise {
    return this.streams.get(aggregateId) || [];
  }

  async *getAllEvents(): AsyncIterable {
    // Return events in insertion order across all streams
    for (const [, events] of this.streams) {
      for (const event of events) {
        yield event;
      }
    }
  }
}

4. Wire Up the Command Handler

The command handler orchestrates the process: load the aggregate from its event history, execute the command, and persist the resulting new events:

// command-handler.ts
import { IEventStore } from './event-store';
import { AccountAggregate } from './account-aggregate';
import { AccountEvent } from './domain-events';

export class AccountCommandHandler {
  constructor(private eventStore: IEventStore) {}

  async handleOpenAccount(
    id: string,
    holder: string,
    initialDeposit: number
  ): Promise {
    // Create a fresh aggregate (no history for new accounts)
    const account = new AccountAggregate();
    const events = account.openAccount(id, holder, initialDeposit);

    // Persist with expected version 0 (new stream)
    await this.eventStore.append(id, events, 0);

    // Apply events to aggregate so it reflects new state
    const updatedAccount = AccountAggregate.fromHistory(events);
    console.log('Account opened:', updatedAccount.state);
  }

  async handleDeposit(
    accountId: string,
    amount: number,
    description: string
  ): Promise {
    // Load existing events
    const history = await this.eventStore.getEvents(accountId);
    const account = AccountAggregate.fromHistory(history);

    // Execute command
    const newEvents = account.deposit(amount, description);

    // Persist with optimistic concurrency check
    await this.eventStore.append(accountId, newEvents, account.state.version);

    console.log('Deposit processed. New balance:', account.state.balance + amount);
  }

  async handleWithdraw(
    accountId: string,
    amount: number,
    description: string
  ): Promise {
    const history = await this.eventStore.getEvents(accountId);
    const account = AccountAggregate.fromHistory(history);
    const newEvents = account.withdraw(amount, description);
    await this.eventStore.append(accountId, newEvents, account.state.version);
    console.log('Withdrawal processed. New balance:', account.state.balance - amount);
  }

  async handleCloseAccount(accountId: string, reason: string): Promise {
    const history = await this.eventStore.getEvents(accountId);
    const account = AccountAggregate.fromHistory(history);
    const newEvents = account.close(reason);
    await this.eventStore.append(accountId, newEvents, account.state.version);
    console.log('Account closed:', accountId);
  }
}

How to Implement CQRS — The Read Side

With Event Sourcing handling the write side, CQRS adds the read side through projections. A projection listens to events and builds a read-optimized view. This decouples reads from the write model entirely.

1. Build a Projection (Read Model)

// account-projection.ts
import { AccountEvent, AccountOpened, MoneyDeposited, MoneyWithdrawn, AccountClosed } from './domain-events';

export interface AccountReadModel {
  id: string;
  holder: string;
  balance: number;
  status: 'active' | 'closed';
  transactionCount: number;
  lastActivity: Date;
}

export class AccountProjection {
  private accounts: Map = new Map();

  // Process a single event to update the read model
  processEvent(event: AccountEvent): void {
    switch (event.eventType) {
      case 'AccountOpened': {
        const e = event as AccountOpened;
        this.accounts.set(e.aggregateId, {
          id: e.aggregateId,
          holder: e.accountHolder,
          balance: e.initialDeposit,
          status: 'active',
          transactionCount: 0,
          lastActivity: new Date(e.timestamp),
        });
        break;
      }
      case 'MoneyDeposited': {
        const e = event as MoneyDeposited;
        const account = this.accounts.get(e.aggregateId);
        if (account) {
          account.balance += e.amount;
          account.transactionCount++;
          account.lastActivity = new Date(e.timestamp);
        }
        break;
      }
      case 'MoneyWithdrawn': {
        const e = event as MoneyWithdrawn;
        const account = this.accounts.get(e.aggregateId);
        if (account) {
          account.balance -= e.amount;
          account.transactionCount++;
          account.lastActivity = new Date(e.timestamp);
        }
        break;
      }
      case 'AccountClosed': {
        const e = event as AccountClosed;
        const account = this.accounts.get(e.aggregateId);
        if (account) {
          account.status = 'closed';
          account.lastActivity = new Date(e.timestamp);
        }
        break;
      }
    }
  }

  // Rebuild entire projection from an event stream
  async rebuildFromStream(eventStream: AsyncIterable): Promise {
    this.accounts.clear();
    for await (const event of eventStream) {
      this.processEvent(event);
    }
  }

  // Query methods for the read side
  getAccount(id: string): AccountReadModel | undefined {
    return this.accounts.get(id);
  }

  getAllAccounts(): AccountReadModel[] {
    return Array.from(this.accounts.values());
  }

  getAccountsByStatus(status: 'active' | 'closed'): AccountReadModel[] {
    return this.getAllAccounts().filter(a => a.status === status);
  }

  getHighBalanceAccounts(minBalance: number): AccountReadModel[] {
    return this.getAllAccounts().filter(a => a.balance >= minBalance);
  }
}

2. Create a Query Service (Thin Read-Side API)

// query-service.ts
import { AccountProjection, AccountReadModel } from './account-projection';

export class AccountQueryService {
  constructor(private projection: AccountProjection) {}

  getAccount(id: string): AccountReadModel | null {
    const account = this.projection.getAccount(id);
    return account || null;
  }

  listActiveAccounts(): AccountReadModel[] {
    return this.projection.getAccountsByStatus('active');
  }

  listClosedAccounts(): AccountReadModel[] {
    return this.projection.getAccountsByStatus('closed');
  }

  getVipCandidates(minBalance: number = 10000): AccountReadModel[] {
    return this.projection.getHighBalanceAccounts(minBalance);
  }

  getAccountSummary(id: string): object | null {
    const account = this.projection.getAccount(id);
    if (!account) return null;
    return {
      holder: account.holder,
      balance: account.balance,
      status: account.status,
      transactionCount: account.transactionCount,
      lastActivity: account.lastActivity.toISOString(),
    };
  }
}

Putting It All Together: A Complete Working Example

Here's how the command and query sides work together in a single application entry point:

// main.ts
import { InMemoryEventStore } from './event-store';
import { AccountCommandHandler } from './command-handler';
import { AccountProjection } from './account-projection';
import { AccountQueryService } from './query-service';

async function main() {
  // ---- Infrastructure Setup ----
  const eventStore = new InMemoryEventStore();
  const projection = new AccountProjection();
  const commandHandler = new AccountCommandHandler(eventStore);
  const queryService = new AccountQueryService(projection);

  // ---- Subscribe projection to events (simulated polling here) ----
  // In production, you'd use a message broker or streaming platform (Kafka, RabbitMQ)
  // Here we manually feed events to the projection after each command for simplicity.

  // ---- Execute Commands (Write Side) ----
  console.log('=== COMMAND: Open Account ===');
  await commandHandler.handleOpenAccount('acc-001', 'Alice Johnson', 500);
  // Feed event to projection
  await rebuildProjection(eventStore, projection);

  console.log('\n=== COMMAND: Deposit Money ===');
  await commandHandler.handleDeposit('acc-001', 200, 'Salary deposit');
  await rebuildProjection(eventStore, projection);

  console.log('\n=== COMMAND: Withdraw Money ===');
  await commandHandler.handleWithdraw('acc-001', 100, 'ATM withdrawal');
  await rebuildProjection(eventStore, projection);

  console.log('\n=== COMMAND: Open Second Account ===');
  await commandHandler.handleOpenAccount('acc-002', 'Bob Smith', 15000);
  await rebuildProjection(eventStore, projection);

  // ---- Execute Queries (Read Side) ----
  console.log('\n=== QUERY: Get Account Summary ===');
  const summary = queryService.getAccountSummary('acc-001');
  console.log(JSON.stringify(summary, null, 2));

  console.log('\n=== QUERY: List Active Accounts ===');
  const activeAccounts = queryService.listActiveAccounts();
  console.table(activeAccounts.map(a => ({
    id: a.id,
    holder: a.holder,
    balance: a.balance,
    transactions: a.transactionCount,
  })));

  console.log('\n=== QUERY: VIP Candidates (balance >= 10000) ===');
  const vipCandidates = queryService.getVipCandidates(10000);
  console.table(vipCandidates.map(a => ({
    id: a.id,
    holder: a.holder,
    balance: a.balance,
  })));

  // ---- Demonstrate Temporal Query (Replay up to a point) ----
  console.log('\n=== TEMPORAL QUERY: State after first 3 events ===');
  const allEvents = await collectEvents(eventStore);
  const firstThreeEvents = allEvents.slice(0, 3);
  const tempProjection = new AccountProjection();
  for (const event of firstThreeEvents) {
    tempProjection.processEvent(event);
  }
  console.log('Account acc-001 after 3 events:',
    JSON.stringify(tempProjection.getAccount('acc-001'), null, 2));
}

async function rebuildProjection(eventStore: InMemoryEventStore, projection: AccountProjection) {
  // In a real system, this would be incremental via a subscription
  // For simplicity, we rebuild from the full stream each time
  await projection.rebuildFromStream(eventStore.getAllEvents());
}

async function collectEvents(eventStore: InMemoryEventStore): Promise {
  const events: import('./domain-events').AccountEvent[] = [];
  for await (const event of eventStore.getAllEvents()) {
    events.push(event);
  }
  return events;
}

// Run the example
main().catch(console.error);

When you run this, you'll see the write side producing events, the projection updating the read model, and queries returning denormalized data instantly — all backed by the immutable event stream.

Best Practices

Common Pitfalls and How to Avoid Them

Conclusion

Event Sourcing and CQRS together form a powerful architectural foundation for systems that demand auditability, scalability, and rich domain modeling. By storing state transitions as immutable events and separating reads from writes, you gain a complete history of every change, the ability to rebuild any view from scratch, and the flexibility to optimize each side of your system independently. The patterns require careful engineering — optimistic concurrency, schema evolution, eventual consistency handling — but the payoff in system resilience and business insight is substantial. Start small: identify one bounded context where audit trails or temporal queries would add real value, implement a clean event store and projection, and expand from there. The event-driven mindset, once adopted, tends to reveal benefits far beyond what you initially anticipated.

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles