By completing this guide, you will design and build a fully functional, zero-dependency Event Sourcing and CQRS engine in modern TypeScript. You will master append-only event streams, aggregate root state hydration, optimistic concurrency control, and asynchronous read model projections from raw code.
- The mechanical differences between traditional ORM state mutation and immutable event streams
- How to build a thread-safe, append-only Event Store with concurrency guarantees in TypeScript
- How to construct an Aggregate Root pattern that rehydrates domain state on demand
- How to build asynchronous Projection Engines that construct optimized SQL/NoSQL read models
Introduction
Traditional CRUD architectures discard business domain history every single time an SQL UPDATE statement runs against your database. When an account balance shifts from $500 to $300, the previous state vanishes forever unless you manually maintain clunky audit tables.
Implementing cqrs event sourcing typescript architectures solves this structural loss by recording every state change as an immutable record of historical truth. As engineering teams migrate complex domain logic away from traditional monolithic ORMs to audit-proof microservices, light-weight event sourcing patterns are rapidly replacing overly heavy enterprise frameworks.
In this hands-on guide, we will cut through academic framework abstraction and build a production-ready event sourcing engine from scratch. You will construct an append-only event store, implement aggregate state rehydration, enforce concurrency guards, and project optimized read models using pure, strongly-typed TypeScript.
Deconstructing the Write Model vs Read Model
Traditional systems force a single data model to handle both high-velocity writes and complex query requests. This unified model forces compromises: your relational database normalization optimizes data integrity during mutations, but slows multi-table join reads down to a crawl.
The CQRS (Command Query Responsibility Segregation) pattern splits your data architecture into two distinct pipelines. The write side accepts commands, enforces domain invariants, and appends immutable events. The read side listens to these events and projects them into denormalized, query-optimized views.
Think of an event-sourced aggregate like a ledger in accounting. You never erase a mistake or directly edit a line item; you write a correcting entry. The ledger represents your write model, while your monthly bank statement is a read projection computed from that underlying history.
By decoupling the write model vs read model, you can scale your database reading infrastructure independently from your domain validation logic. Writes require ultra-fast, single-row append locks, while reads pull directly from pre-computed indexes or flat caches.
Before writing engine code, we must establish strict, immutably typed events that capture domain execution intent. The code block below defines our core primitives for commands, domain events, and event envelopes.
// Step 1: Define core event sourcing type definitions
export interface DomainEvent<Payload = Record<string, unknown>> {
readonly eventType: string;
readonly aggregateId: string;
readonly payload: Payload;
readonly timestamp: number;
}
export interface EventEnvelope<Payload = Record<string, unknown>> {
readonly eventId: string;
readonly streamId: string;
readonly version: number;
readonly event: DomainEvent<Payload>;
}
export interface Command<Payload = Record<string, unknown>> {
readonly commandType: string;
readonly targetAggregateId: string;
readonly payload: Payload;
}This initial structural layout uses TypeScript's readonly modifier to enforce runtime immutability across event structures. The EventEnvelope wraps the inner domain payload with critical infrastructure meta-data: sequential versions and unique stream identifiers necessary for tracking stream boundaries.
Step 1: Build an Append-Only Event Store in Node.js
An event store is not a traditional relational table or a key-value store; it is an append-only log tailored specifically for sequential event payloads. The primary responsibility of the event store is guaranteeing that event streams append sequentially without silent data corruption or race conditions.
When multiple microservice instances attempt to process commands on the same aggregate simultaneously, our event store must reject conflicting writes using optimistic concurrency control. We achieve this by validating the aggregate's expected version before accepting new events into the log.
Never perform an SQL UPDATE or DELETE operation on an event store table. The audit log must remain strictly append-only. Correct state changes must always be recorded by appending new compensating events.
Let's build an event store to examine how to build event store nodejs components with robust stream concurrency guarantees. This implementation handles stream generation, version verification, and synchronous projection broadcasting.
// Step 2: Implement the Event Store with Concurrency Guards
export class ConcurrencyError extends Error {
constructor(
public readonly streamId: string,
public readonly expectedVersion: number,
public readonly actualVersion: number
) {
super(`Concurrency conflict on aggregate stream "${streamId}". Expected: ${expectedVersion}, Actual: ${actualVersion}`);
this.name = 'ConcurrencyError';
}
}
export type EventSubscriber = (envelope: EventEnvelope) => Promise<void>;
export class InMemoryEventStore {
private readonly streams: Map<string, EventEnvelope[]> = new Map();
private readonly subscribers: Set<EventSubscriber> = new Set();
public subscribe(subscriber: EventSubscriber): () => void {
this.subscribers.add(subscriber);
return () => this.subscribers.delete(subscriber);
}
public async getStream(streamId: string): Promise<EventEnvelope[]> {
return this.streams.get(streamId) || [];
}
public async append(
streamId: string,
events: DomainEvent[],
expectedVersion: number
): Promise<EventEnvelope[]> {
const history = this.streams.get(streamId) || [];
const currentVersion = history.length;
if (currentVersion !== expectedVersion) {
throw new ConcurrencyError(streamId, expectedVersion, currentVersion);
}
const newEnvelopes: EventEnvelope[] = events.map((event, index) => ({
eventId: crypto.randomUUID(),
streamId,
version: currentVersion + index + 1,
event,
}));
const updatedStream = [...history, ...newEnvelopes];
this.streams.set(streamId, updatedStream);
// Publish to asynchronous subscribers
for (const envelope of newEnvelopes) {
for (const subscriber of this.subscribers) {
await subscriber(envelope);
}
}
return newEnvelopes;
}
}The InMemoryEventStore maintains a collection of event arrays scoped by stream keys. When append() is invoked, it evaluates expectedVersion against the existing stream length. If another command worker mutated the aggregate in the fraction of a second between execution and commit, the operation fails instantly with a ConcurrencyError.
Step 2: Construct the Aggregate Root with State Replay
An Aggregate Root in DDD (Domain-Driven Design) acts as the primary gatekeeper for domain invariants. In traditional systems, an aggregate maps directly to a row in a database table. In an event-sourced engine, an aggregate root has no stored persistent state; its state is entirely computed by replaying historical events from zero.
This process of building state from scratch is known as event sourcing from scratch through event replay. The aggregate processes incoming commands, checks current business rules against its rehydrated state, and emits new domain events if validation passes.
Keep your domain aggregate methods pure. An aggregate should never make external HTTP calls, query databases, or access system clocks directly. All decision context must come from past events and command arguments.
The code sample below implements an abstract base class for aggregate roots along with a concrete BankAccountAggregate domain model.
// Step 3: Define Aggregate Root with state rehydration patterns
export abstract class AggregateRoot<TState> {
private uncommittedEvents: DomainEvent[] = [];
private currentVersion: number = 0;
constructor(public readonly id: string) {}
public abstract getInitialState(): TState;
public abstract applyEvent(state: TState, event: DomainEvent): TState;
public hydrate(history: EventEnvelope[]): TState {
let state = this.getInitialState();
for (const envelope of history) {
state = this.applyEvent(state, envelope.event);
this.currentVersion = envelope.version;
}
return state;
}
protected raiseEvent(event: DomainEvent): void {
this.uncommittedEvents.push(event);
}
public getUncommittedEvents(): DomainEvent[] {
return [...this.uncommittedEvents];
}
public clearUncommittedEvents(): void {
this.uncommittedEvents = [];
}
public getVersion(): number {
return this.currentVersion;
}
}
// Domain Implementation: Bank Account Aggregate
export interface BankAccountState {
balance: number;
isClosed: boolean;
}
export class BankAccountAggregate extends AggregateRoot<BankAccountState> {
public getInitialState(): BankAccountState {
return { balance: 0, isClosed: false };
}
public applyEvent(state: BankAccountState, event: DomainEvent): BankAccountState {
switch (event.eventType) {
case 'AccountOpened':
return { ...state, balance: (event.payload.initialDeposit as number) || 0 };
case 'MoneyDeposited':
return { ...state, balance: state.balance + (event.payload.amount as number) };
case 'MoneyWithdrawn':
return { ...state, balance: state.balance - (event.payload.amount as number) };
case 'AccountClosed':
return { ...state, isClosed: true };
default:
return state;
}
}
// Domain Business Rules
public deposit(amount: number): void {
if (amount <= 0) throw new Error("Deposit amount must be positive.");
this.raiseEvent({
eventType: 'MoneyDeposited',
aggregateId: this.id,
payload: { amount },
timestamp: Date.now(),
});
}
public withdraw(amount: number, currentState: BankAccountState): void {
if (currentState.isClosed) throw new Error("Cannot withdraw from closed account.");
if (currentState.balance < amount) throw new Error("Insufficient funds.");
this.raiseEvent({
eventType: 'MoneyWithdrawn',
aggregateId: this.id,
payload: { amount },
timestamp: Date.now(),
});
}
}Notice how state modification occurs cleanly inside applyEvent. The aggregate business methods like withdraw() contain no direct balance adjustments. Instead, they assert invariants, generate a new event payload via raiseEvent(), and defer actual state updates to the standard replay handler.
Step 3: Create Asynchronous Read Model Projections
Now that the write side is safely persisting events, we must construct the query side. Read models are specialized data representations tailored to user queries, dashboards, and search indexes.
Using a read model projection typescript handler, we listen to stream pushes from the event store and update external database structures. This operation can happen synchronously or in the background as an eventual consistency worker.
// Step 4: Construct Read Model Projection handlers
export interface CustomerBalanceReadModel {
accountId: string;
currentBalance: number;
totalTransactions: number;
lastUpdated: number;
}
export class AccountBalanceProjection {
private readonly readDatabase: Map<string, CustomerBalanceReadModel> = new Map();
public async project(envelope: EventEnvelope): Promise<void> {
const { event } = envelope;
const existing = this.readDatabase.get(event.aggregateId) || {
accountId: event.aggregateId,
currentBalance: 0,
totalTransactions: 0,
lastUpdated: event.timestamp,
};
switch (event.eventType) {
case 'AccountOpened':
this.readDatabase.set(event.aggregateId, {
...existing,
currentBalance: (event.payload.initialDeposit as number) || 0,
totalTransactions: 1,
lastUpdated: event.timestamp,
});
break;
case 'MoneyDeposited':
this.readDatabase.set(event.aggregateId, {
...existing,
currentBalance: existing.currentBalance + (event.payload.amount as number),
totalTransactions: existing.totalTransactions + 1,
lastUpdated: event.timestamp,
});
break;
case 'MoneyWithdrawn':
this.readDatabase.set(event.aggregateId, {
...existing,
currentBalance: existing.currentBalance - (event.payload.amount as number),
totalTransactions: existing.totalTransactions + 1,
lastUpdated: event.timestamp,
});
break;
}
}
public getAccountView(accountId: string): CustomerBalanceReadModel | undefined {
return this.readDatabase.get(accountId);
}
}This projection decouples queries completely from the write model logic. A complex web UI query does not need to compute aggregates or replay 10,000 events to show a balance. It simply queries the fast AccountBalanceProjection key-value lookup directly.
Step 4: Wire up Command and Query Dispatchers
To execute system commands cleanly inside microservices without coupling handlers directly to database stores, we need a Command Bus orchestrator. The dispatcher loads the aggregate history, rehydrates state, runs domain actions, and saves emitted events back into the log.
The code example below completes our cqrs implementation example by unifying command handling, persistence pipelines, and read model projections into a working execution system.
// Step 5: Command Handler and Dispatch Orchestration Engine
export class CommandDispatcher {
constructor(private readonly store: InMemoryEventStore) {}
public async handleWithdrawal(accountId: string, amount: number): Promise<void> {
// 1. Fetch historical stream from store
const history = await this.store.getStream(accountId);
// 2. Instantiate and rehydrate domain aggregate
const aggregate = new BankAccountAggregate(accountId);
const currentState = aggregate.hydrate(history);
// 3. Execute domain method (validates invariants & records events)
aggregate.withdraw(amount, currentState);
// 4. Commit new uncommitted events back to event store
const newEvents = aggregate.getUncommittedEvents();
const expectedVersion = aggregate.getVersion();
await this.store.append(accountId, newEvents, expectedVersion);
aggregate.clearUncommittedEvents();
}
}This four-step command execution pipeline encapsulates aggregate state recovery safely. It guarantees that write operations always operate against fully hydrated up-to-date state while isolating the underlying persistence mechanism.
Step 5: Putting It All Together (End-to-End Execution)
Now let's execute a real-world script demonstrating a full command-to-projection pipeline. We will set up the store, subscribe our projection listener, execute commands via our command dispatcher, and inspect the resulting read model state.
// Step 6: Complete End-to-End Application Workflow Setup
async function runEngineDemo() {
const eventStore = new InMemoryEventStore();
const projection = new AccountBalanceProjection();
const dispatcher = new CommandDispatcher(eventStore);
// Wire up the async read model projection subscriber
eventStore.subscribe(async (envelope) => {
await projection.project(envelope);
});
const accountId = "acc-9923-xyz";
// Bootstrap account with initial event payload
await eventStore.append(accountId, [
{
eventType: 'AccountOpened',
aggregateId: accountId,
payload: { initialDeposit: 500 },
timestamp: Date.now(),
},
{
eventType: 'MoneyDeposited',
aggregateId: accountId,
payload: { amount: 250 },
timestamp: Date.now(),
}
], 0);
// Execute command via dispatcher pipeline
console.log("Executing $150 withdrawal command...");
await dispatcher.handleWithdrawal(accountId, 150);
// Fetch updated Read Model projection
const readView = projection.getAccountView(accountId);
console.log("Current Read Model State:", JSON.stringify(readView, null, 2));
// Retrieve raw event log to verify historical transparency
const fullStream = await eventStore.getStream(accountId);
console.log(`Total persisted stream events: ${fullStream.length}`);
}
runEngineDemo().catch(console.error);Executing this workflow populates the stream and updates our read view without explicit relational state updates. If you run this script, you will see the full log output detailing how events seamlessly flow from commands down to projection updates.
{
"accountId": "acc-9923-xyz",
"currentBalance": 600,
"totalTransactions": 3,
"lastUpdated": 1793836800000
}The system computes state derived strictly from the underlying stream log events: an initial deposit of $500, a credit of $250, and a withdrawal command of $150 resulting in a final projected balance of $600 across 3 distinct transaction records.
Best Practices and Common Pitfalls
Implement Snapshotting for Long-Lived Streams
As an aggregate processes thousands of events over months or years, replaying its complete stream history from version zero during every command execution becomes a major performance bottleneck. Implementing snapshot patterns solves this scaling issue.
A snapshot periodically serializes the aggregate's rehydrated state every 100 or 1,000 events. When loading the aggregate, fetch the newest snapshot image first, then only apply incoming events recorded after the snapshot's creation version.
Version your event schemas systematically! Include a version property inside your events (e.g., schemaVersion: 2). When business rules evolve, write Upcaster functions that dynamically translate legacy event models into modern structural interfaces before sending them to aggregates.
Avoid Dual Writing (The Outbox Pattern)
A critical bug pattern developers face when constructing event engines is publishing messages to external queue systems like Kafka or RabbitMQ inside the main database transaction. If the database transaction aborts after publishing the message, downstream read models will become permanently corrupted.
Always write your domain events to an atomic persistent Event Store first. Use a dedicated background tailing worker or transactional Outbox processing pattern to pick up committed store records and asynchronously broadcast them across external messaging networks.
Real-World Example
Consider a modern fintech payment processor processing millions of cross-border wallet settlements daily under strict audit compliance mandates. Standard ORM structures fail in these environments because overwritten balance rows make regulatory verification impossible without heavy secondary audit logging.
Using our custom TypeScript event sourcing engine, the platform records every incoming transaction, currency conversion, and fee deduction as immutable stream entries. If an auditing team queries a account's balance as it existed on March 14th at 10:14 AM, the system simply replays events up to that precise timestamp.
Furthermore, read-model projections run on isolated Redis or Elasticsearch infrastructure. While high-volume batch payments continuously update write stores, client dashboards execute ultra-fast read queries without locking aggregate records or impacting core checkout latency.
Future Outlook and What's Coming Next
As microservice architectures evolve through 2026 and into 2027, manual framework-heavy event sourcing setups are giving way to high-performance, lightweight runtime native implementations. Native WebAssembly (WASM) stream engines and zero-copy JSON parsing strategies are dramatically lowering memory overhead for Event Store engines running on Node.js and Bun.
Additionally, TypeScript's expanding capabilities with compile-time type-level validation are enabling developer teams to catch invalid state machine transitions directly inside the IDE before commands ever touch database layers. Combining strongly typed event schemas with lightweight native runtimes makes building event-sourced systems simpler than ever before.
Conclusion
Event Sourcing and CQRS shift your architectural perspective from managing mutable state tables to preserving absolute chronological history. By decoupling write validation models from consumer read views, you unlock auditable domain boundaries and massive read-side scalability.
We built a functional, strongly typed event sourcing engine from scratch using plain TypeScript — including an append-only event store, optimistic concurrency checks, aggregate hydration pipelines, and projection handlers. You now possess the foundation needed to shed heavy enterprise frameworks and write clean, auditable systems.
Take this codebase, extend the InMemoryEventStore to push records into PostgreSQL or DynamoDB, build an Upcaster pipeline for versioning event payloads, and start building audit-proof microservices in your systems today!
- Event stores must be strictly append-only; state is derived through event rehydration rather than in-place database mutations
- Optimistic concurrency guards prevent race conditions by asserting stream version numbers before persisting new event arrays
- CQRS separates write-side domain enforcement from read-side query denormalization, allowing independent optimization and scaling
- Always keep aggregate root models pure and free of side effects like HTTP calls or unhandled database operations
