Event Sourcing & CQRS: Building Scalable, Auditable Node.js Microservices

Beyond CRUD: The Imperative for Event Sourcing & CQRS

In traditional software systems, the ubiquitous CRUD (Create, Read, Update, Delete) paradigm dominates: when an entity changes, we issue an UPDATE statement that overwrites previous database records. While intuitive for basic applications, this destructive state mutation introduces severe limitations in enterprise fintech, logistics, and healthcare systems:

The Audit Blindspot: Overwriting an account balance from $1,200 to $850 destroys historical truth. You see the current state, but you have erased the chain of events that created that state.
Read/Write Bottlenecks: A single relational schema must compromise between write performance (normalized tables, foreign keys) and read performance (complex JOINs, aggregations).
Impedance Mismatch: Business stakeholders reason in terms of domain occurrences—"Customer Ordered Goods", "Invoice Disputed"—while developers translate these into generic database rows.

Event Sourcing and Command Query Responsibility Segregation (CQRS) solve these architectural bottlenecks by decoupling writes from reads and treating state transitions as an immutable, append-only ledger of historical facts.

SQL

+-------------------------------------------------------------------------------+
|                        Traditional CRUD vs. CQRS + ES                         |
+-------------------------------------------------------------------------------+
| Traditional CRUD:                                                             |
| Client ---> [API Controller] ---> UPDATE users SET balance = 500 WHERE id = 1 |
| (Destructive overwrite, zero historical causality, lock contention)          |
|                                                                               |
| Event Sourcing & CQRS:                                                        |
| Command ---> [Aggregate Invariants] ---> Append [MoneyWithdrawnEvent] (Log)   |
|                                                  │                            |
|                                                  ▼ (Async Event Bus)          |
| Query   <--- [Read Model Projection] <─── Update Denormalized View (Redis/SQL)|
+-------------------------------------------------------------------------------+

MERMAID

graph TD
    User([Client Application]) -->|Send Command: WithdrawFunds| CmdHandler[Command Handler]
    CmdHandler -->|Load History & Reconstitute| Agg[Account Aggregate]
    Agg -->|Enforce Invariants: Balance >= Amount| Event[Emit MoneyWithdrawnEvent]
    Event -->|Append with Version Check| EStore[(Immutable Event Store)]
    EStore -->|Publish Event| Bus[Kafka / EventBus]
    Bus -->|Consume Event| Projector[Read Model Projector]
    Projector -->|Upsert Read View| ReadDB[(PostgreSQL / Redis Read Model)]
    User -->|Query Account Balance| QueryHandler[Query Handler]
    QueryHandler -->|Sub-millisecond Read| ReadDB

Core Principles of Event Sourcing

In an Event-Sourced architecture:

Events are Immutable Facts: Named in the past tense (AccountOpened, FundsDeposited, PaymentDeclined). Once written to the event store, an event is never modified or deleted.
State is a Left-Fold Projection: An entity's current state is derived by replaying all historical events from origin: $$\text{Current State} = \sum_{t=0}^{N} \text{Apply}(\text{Event}_t)$$
Optimistic Concurrency Control (OCC): Every event is appended with a monotonically increasing sequence version number. If two concurrent requests attempt to append version 5 simultaneously, the event store rejects the second request with a concurrency conflict.

Complete TypeScript Implementation

Let us construct an auditable, enterprise-grade banking ledger using pure TypeScript and modern Node.js patterns.

1. Defining Immutable Domain Events

TYPESCRIPT

// src/domain/events.ts
export interface DomainEvent<T = Record<string, unknown>> {
  readonly eventId: string;
  readonly aggregateId: string;
  readonly eventType: string;
  readonly version: number;
  readonly timestamp: Date;
  readonly payload: T;
}

export type AccountOpenedEvent = DomainEvent<{
  ownerName: string;
  currency: string;
  initialDeposit: number;
}>;

export type MoneyDepositedEvent = DomainEvent<{
  amount: number;
  reference: string;
}>;

export type MoneyWithdrawnEvent = DomainEvent<{
  amount: number;
  reason: string;
}>;

2. The Aggregate Root (`AccountAggregate`)

The aggregate root enforces business rules and encapsulates state transitions:

TYPESCRIPT

// src/domain/account.aggregate.ts
import { DomainEvent, AccountOpenedEvent, MoneyDepositedEvent, MoneyWithdrawnEvent } from './events';
import crypto from 'node:crypto';

export class AccountAggregate {
  public id: string = '';
  public ownerName: string = '';
  public balance: number = 0;
  public currency: string = 'USD';
  public version: number = 0;
  public isClosed: boolean = false;

  private uncommittedEvents: DomainEvent[] = [];

  // Reconstitute aggregate state from historical events
  public static fromHistory(events: DomainEvent[]): AccountAggregate {
    const aggregate = new AccountAggregate();
    for (const event of events) {
      aggregate.applyEvent(event, false);
    }
    return aggregate;
  }

  // --- Business Commands ---

  public openAccount(id: string, ownerName: string, initialDeposit: number, currency: string = 'USD'): void {
    if (this.version > 0) {
      throw new Error('Account is already initialized.');
    }
    if (initialDeposit < 0) {
      throw new Error('Initial deposit cannot be negative.');
    }

    const event: AccountOpenedEvent = {
      eventId: crypto.randomUUID(),
      aggregateId: id,
      eventType: 'ACCOUNT_OPENED',
      version: this.version + 1,
      timestamp: new Date(),
      payload: { ownerName, currency, initialDeposit },
    };

    this.applyEvent(event, true);
  }

  public deposit(amount: number, reference: string): void {
    if (this.isClosed) throw new Error('Cannot deposit into a closed account.');
    if (amount <= 0) throw new Error('Deposit amount must be positive.');

    const event: MoneyDepositedEvent = {
      eventId: crypto.randomUUID(),
      aggregateId: this.id,
      eventType: 'MONEY_DEPOSITED',
      version: this.version + 1,
      timestamp: new Date(),
      payload: { amount, reference },
    };

    this.applyEvent(event, true);
  }

  public withdraw(amount: number, reason: string): void {
    if (this.isClosed) throw new Error('Cannot withdraw from a closed account.');
    if (amount <= 0) throw new Error('Withdrawal amount must be positive.');
    
    // Business Invariant Enforcement
    if (this.balance - amount < 0) {
      throw new Error(`Insufficient funds. Current balance: ${this.balance}, requested: ${amount}`);
    }

    const event: MoneyWithdrawnEvent = {
      eventId: crypto.randomUUID(),
      aggregateId: this.id,
      eventType: 'MONEY_WITHDRAWN',
      version: this.version + 1,
      timestamp: new Date(),
      payload: { amount, reason },
    };

    this.applyEvent(event, true);
  }

  // State Transition Mutation
  private applyEvent(event: DomainEvent, isNew: boolean): void {
    switch (event.eventType) {
      case 'ACCOUNT_OPENED': {
        const p = (event as AccountOpenedEvent).payload;
        this.id = event.aggregateId;
        this.ownerName = p.ownerName;
        this.currency = p.currency;
        this.balance = p.initialDeposit;
        break;
      }
      case 'MONEY_DEPOSITED': {
        const p = (event as MoneyDepositedEvent).payload;
        this.balance += p.amount;
        break;
      }
      case 'MONEY_WITHDRAWN': {
        const p = (event as MoneyWithdrawnEvent).payload;
        this.balance -= p.amount;
        break;
      }
      default:
        throw new Error(`Unknown event type: ${event.eventType}`);
    }

    this.version = event.version;
    if (isNew) {
      this.uncommittedEvents.push(event);
    }
  }

  public getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents];
  }

  public clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }
}

3. Append-Only Event Store with Concurrency Control

TYPESCRIPT

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

export class ConcurrencyError extends Error {
  constructor(expected: number, actual: number) {
    super(`Optimistic Concurrency Failure: Expected version ${expected}, but found ${actual}`);
    this.name = 'ConcurrencyError';
  }
}

export class InMemoryEventStore {
  // In production, backed by PostgreSQL append-only table or EventStoreDB
  private storage: Map<string, DomainEvent[]> = new Map();

  public async getEvents(aggregateId: string): Promise<DomainEvent[]> {
    return this.storage.get(aggregateId) || [];
  }

  public async appendEvents(
    aggregateId: string,
    expectedVersion: number,
    events: DomainEvent[]
  ): Promise<void> {
    const existing = this.storage.get(aggregateId) || [];
    const currentVersion = existing.length > 0 ? existing[existing.length - 1].version : 0;

    if (currentVersion !== expectedVersion) {
      throw new ConcurrencyError(expectedVersion, currentVersion);
    }

    this.storage.set(aggregateId, [...existing, ...events]);
  }
}

4. The CQRS Read Model Projector

The read side listens to committed events and builds pre-aggregated, denormalized views designed for instantaneous querying without runtime computations:

TYPESCRIPT

// src/read-model/account-projector.ts
import { DomainEvent, AccountOpenedEvent, MoneyDepositedEvent, MoneyWithdrawnEvent } from '../domain/events';

export interface AccountView {
  accountId: string;
  ownerName: string;
  currentBalance: number;
  totalDeposited: number;
  totalWithdrawn: number;
  transactionCount: number;
  lastUpdated: Date;
}

export class AccountReadModelProjector {
  private readDatabase: Map<string, AccountView> = new Map();

  public project(event: DomainEvent): void {
    const accountId = event.aggregateId;

    switch (event.eventType) {
      case 'ACCOUNT_OPENED': {
        const payload = (event as AccountOpenedEvent).payload;
        this.readDatabase.set(accountId, {
          accountId,
          ownerName: payload.ownerName,
          currentBalance: payload.initialDeposit,
          totalDeposited: payload.initialDeposit,
          totalWithdrawn: 0,
          transactionCount: 1,
          lastUpdated: event.timestamp,
        });
        break;
      }

      case 'MONEY_DEPOSITED': {
        const payload = (event as MoneyDepositedEvent).payload;
        const view = this.readDatabase.get(accountId);
        if (view) {
          view.currentBalance += payload.amount;
          view.totalDeposited += payload.amount;
          view.transactionCount += 1;
          view.lastUpdated = event.timestamp;
        }
        break;
      }

      case 'MONEY_WITHDRAWN': {
        const payload = (event as MoneyWithdrawnEvent).payload;
        const view = this.readDatabase.get(accountId);
        if (view) {
          view.currentBalance -= payload.amount;
          view.totalWithdrawn += payload.amount;
          view.transactionCount += 1;
          view.lastUpdated = event.timestamp;
        }
        break;
      }
    }
  }

  public getAccountView(accountId: string): AccountView | undefined {
    return this.readDatabase.get(accountId);
  }
}

End-to-End Orchestration & Verification

TYPESCRIPT

// src/index.ts
import crypto from 'node:crypto';
import { AccountAggregate } from './domain/account.aggregate';
import { InMemoryEventStore } from './store/event-store';
import { AccountReadModelProjector } from './read-model/account-projector';

async function runDemo() {
  const eventStore = new InMemoryEventStore();
  const projector = new AccountReadModelProjector();
  const accountId = crypto.randomUUID();

  console.log('--- 1. Executing Command: Open Account ---');
  const account = new AccountAggregate();
  account.openAccount(accountId, 'Tahir Idrees', 1000);

  const events1 = account.getUncommittedEvents();
  await eventStore.appendEvents(accountId, 0, events1);
  events1.forEach((e) => projector.project(e));
  account.clearUncommittedEvents();

  console.log('--- 2. Executing Command: Deposit & Withdrawal ---');
  account.deposit(500, 'Salary Payment Ref #9921');
  account.withdraw(350, 'Cloud Infrastructure Hosting Fee');

  const events2 = account.getUncommittedEvents();
  await eventStore.appendEvents(accountId, 1, events2);
  events2.forEach((e) => projector.project(e));
  account.clearUncommittedEvents();

  console.log('--- 3. Querying CQRS Denormalized Read View ---');
  const view = projector.getAccountView(accountId);
  console.log('Denormalized Read Model:', JSON.stringify(view, null, 2));

  console.log('\n--- 4. Auditing Complete Event History ---');
  const fullAuditTrail = await eventStore.getEvents(accountId);
  console.log(`Total Stored Events: ${fullAuditTrail.length}`);
  fullAuditTrail.forEach((evt) => {
    console.log(`[v${evt.version}] ${evt.eventType} at ${evt.timestamp.toISOString()}`);
  });
}

runDemo().catch(console.error);

Architectural Comparison: Event Sourcing vs Traditional CRUD

Metric	Traditional CRUD	Event Sourcing + CQRS
Auditability	Poor (Requires manual triggers / log tables)	100% Native (Source of truth is the audit log)
Data Loss on Mutation	High (Overwrites destructive state)	Zero (Append-only immutability)
Read Performance	Degrades with table normalization and JOINs	Sub-millisecond (Denormalized read projections)
Time-Travel Debugging	Impossible	Replay events up to any arbitrary timestamp
Implementation Complexity	Minimal	Higher (Requires event buses, projectors, OCC)

Production Verification Checklist

Append-Only Immutability: Event store database credentials have only INSERT and SELECT grants; UPDATE and DELETE are revoked at the database level.
Optimistic Concurrency Control: Every append transaction asserts that the expected version equals the latest version in storage.
Snapshotting Strategy: Aggregates with more than 500 events periodically persist snapshot states to avoid replaying thousands of events on startup.
Idempotent Projections: Projectors check eventId or version before modifying read views to tolerate duplicate message delivery.
Schema Evolution Plan: Backward-compatible upcasters are established for evolving domain event schemas over time.

Event Sourcing & CQRS: Building Scalable, Auditable Node.js Microservices

Beyond CRUD: The Imperative for Event Sourcing & CQRS

Core Principles of Event Sourcing

Complete TypeScript Implementation

1. Defining Immutable Domain Events

2. The Aggregate Root (AccountAggregate)

3. Append-Only Event Store with Concurrency Control

4. The CQRS Read Model Projector

End-to-End Orchestration & Verification

Architectural Comparison: Event Sourcing vs Traditional CRUD

Production Verification Checklist

Related Posts

2. The Aggregate Root (`AccountAggregate`)