Author: Chengyi Xu Date: 2026-05-09 Target: LogRocket / SitePoint / Smashing Magazine Status: ready
Picture this: your payment service processed a refund twice. The customer is angry, the finance team is confused, and nobody can tell you exactly what happened because the state says "refunded" and your audit log was a best-effort afterthought that someone forgot to update in the second code path. If this scenario keeps you up at night, you need event sourcing.
Event sourcing is a fundamentally different way to model application state — every mutation is captured as an immutable event, and the current state is a projection of all events that occurred up to this moment. When implemented correctly, it gives you complete auditability, temporal queries, and the ability to rebuild state at any point in time.
Before diving into code, let's nail down the semantics because this pattern inverts some deeply ingrained assumptions. In traditional CRUD applications, when a user changes their email address, you find the user row and update the email column. The old value is gone forever unless you built a separate audit trail — which usually lives in a different table, uses a different code path, and drifts from actual state changes as features are added and someone forgets to wire them into the audit logger.
Event sourcing inverts this. Instead of storing the current state and mutating it, you store the sequence of events that led to the current state — a user registered, an email changed, a payment was authorized. These events are immutable and append-only. The current state of any entity is derived by replaying its event stream from the beginning. This sounds computationally expensive, and for very long streams it can be. We will cover snapshots to make it fast, but the important shift is philosophical: you stop thinking of your database as a snapshot of now and start thinking of it as a journal of what happened.
Here is the contrast in concrete terms. In a traditional system, changing a user's email looks like this:
UPDATE users SET email = 'new@example.com' WHERE id = 123;In an event-sourced system, you instead insert a new event:
// Event 1: User was registered
{ type: 'UserRegistered', userId: '123', email: 'old@example.com', occurredAt: '2026-01-15T10:00:00Z' }
// Event 2: Email was changed
{ type: 'UserEmailChanged', userId: '123', oldEmail: 'old@example.com', newEmail: 'new@example.com', occurredAt: '2026-05-09T14:30:00Z' }The current state — user 123 has email new@example.com — is not stored in a row anywhere. It lives only as the result of applying these two events in order. If you need to know what the user's email was on March 1st, you replay only the events that occurred before that date and stop. No separate audit table, no logging code, no drift. The write path IS the audit trail.
An event sourcing system needs four core types: events, aggregates, an event store, and projections. The StoredEvent is the canonical representation of something that happened. Every event gets a globally unique ID, belongs to a specific stream identified by streamId, and carries a monotonically increasing streamVersion within that stream. The metadata object carries tracing information — correlation IDs, causation IDs for event chains, and timestamps — this is what makes the system audit-ready without additional effort.
// --- Event Definition ---
type EventType =
| 'BankAccountOpened'
| 'MoneyDeposited'
| 'MoneyWithdrawn'
| 'AccountClosed';
interface StoredEvent {
id: string;
streamId: string;
streamVersion: number;
type: EventType;
data: Record<string, unknown>;
metadata: {
correlationId: string;
causationId: string | null;
userId: string | null;
occurredAt: Date;
ipAddress: string | null;
};
}
// --- Aggregate Definition ---
interface Aggregate {
id: string;
version: number;
applyEvent(event: StoredEvent): void;
getUncommittedEvents(): StoredEvent[];
clearUncommittedEvents(): void;
}The Aggregate interface defines the contract for domain objects. An aggregate tracks its version and maintains a buffer of uncommitted events — events applied in memory but not yet persisted. This buffer lets multiple commands run against the same aggregate instance within a single operation, with each command seeing the state changes from the previous one.
Event sourcing shines brightest in domains where the history of changes is as important as the current state. Banking is the canonical example — regulators do not just care about your current balance; they need to know every deposit and withdrawal that led to it, going back seven years in most jurisdictions. Let's model a bank account as a concrete example.
The key design decision in any aggregate is where validation lives. In an event-sourced system, validation happens before events are recorded, not after. A deposit command checks that the account is open, that the amount is positive, and that it does not exceed a single-transaction limit. Only if all guards pass does it record a MoneyDeposited event. This means events represent facts that have already been validated — they do not need to be re-validated during replay.
The recordEvent method does two things atomically in memory: it applies the event to the aggregate's state immediately, and it buffers the event for later persistence. The immediate application is critical for correctness. If a command deposits $100 and then a second command attempts to withdraw $150 within the same operation (before events hit the database), the second command will correctly throw "Insufficient funds" because the first deposit was already reflected in the in-memory balance.
import { v4 as uuidv4 } from 'uuid';
class BankAccount implements Aggregate {
id: string;
version: number;
balance: number;
ownerId: string;
isClosed: boolean;
private uncommittedEvents: StoredEvent[];
constructor(id: string, ownerId: string) {
this.id = id;
this.version = 0;
this.balance = 0;
this.ownerId = ownerId;
this.isClosed = false;
this.uncommittedEvents = [];
}
deposit(amount: number, description: string): void {
if (this.isClosed) {
throw new Error('Cannot deposit to a closed account');
}
if (amount <= 0) {
throw new Error('Deposit amount must be positive');
}
if (amount > 1_000_000) {
throw new Error('Deposit exceeds single transaction limit');
}
this.recordEvent({
type: 'MoneyDeposited',
streamId: this.id,
data: { amount, description },
});
}
withdraw(amount: number, description: 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');
}
if (this.balance < amount) {
throw new Error('Insufficient funds');
}
this.recordEvent({
type: 'MoneyWithdrawn',
streamId: this.id,
data: { amount, description },
});
}
applyEvent(event: StoredEvent): void {
switch (event.type) {
case 'MoneyDeposited': {
const data = event.data as { amount: number };
this.balance += data.amount;
break;
}
case 'MoneyWithdrawn': {
const data = event.data as { amount: number };
this.balance -= data.amount;
break;
}
case 'AccountClosed':
this.isClosed = true;
break;
default:
throw new Error(`Unknown event type: ${event.type}`);
}
this.version = event.streamVersion;
}
getUncommittedEvents(): StoredEvent[] {
return [...this.uncommittedEvents];
}
clearUncommittedEvents(): void {
this.uncommittedEvents = [];
}
private recordEvent(
partial: Pick<StoredEvent, 'type' | 'streamId' | 'data'>
): void {
const event: StoredEvent = {
id: uuidv4(),
streamId: partial.streamId,
streamVersion: this.version + this.uncommittedEvents.length + 1,
type: partial.type,
data: partial.data,
metadata: {
correlationId: uuidv4(),
causationId: null,
userId: null,
occurredAt: new Date(),
ipAddress: null,
},
};
this.applyEvent(event);
this.uncommittedEvents.push(event);
}
}A subtle but important detail: the streamVersion calculation uses this.version + this.uncommittedEvents.length + 1. When loading from the event store, this.version reflects the last persisted event's version. As we buffer uncommitted events, each gets the next sequential version number. When we persist, the event store will verify that no other process wrote to this stream in the meantime — this is optimistic concurrency control, and it is how event sourcing systems avoid distributed locks while still guaranteeing consistency.
The event store is the persistence layer. Its job is simple — append events, retrieve events by stream, never modify or delete — but the constraints are strict: serializable isolation for writes, efficient retrieval by stream and version, and cross-stream queries by correlation ID.
PostgreSQL is an excellent choice. Its JSONB columns store event data without a rigid schema, its unique constraints and SELECT FOR UPDATE give you optimistic concurrency without external coordination, and its index support handles the read patterns you need. Here is a complete implementation using the pg driver:
import { Pool } from 'pg';
class PostgresEventStore {
private pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async initialize(): Promise<void> {
await this.pool.query(`
CREATE TABLE IF NOT EXISTS events (
id UUID PRIMARY KEY,
stream_id VARCHAR(255) NOT NULL,
stream_version INT NOT NULL,
type VARCHAR(255) NOT NULL,
data JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (stream_id, stream_version)
);
CREATE INDEX IF NOT EXISTS idx_events_stream_id_version
ON events (stream_id, stream_version);
CREATE INDEX IF NOT EXISTS idx_events_type_occurred
ON events (type, occurred_at);
CREATE INDEX IF NOT EXISTS idx_events_correlation
ON events ((metadata->>'correlationId'));
`);
}
async appendEvents(
streamId: string,
expectedVersion: number,
events: StoredEvent[]
): Promise<void> {
const client = await this.pool.connect();
try {
await client.query('BEGIN');
const result = await client.query(
`SELECT COALESCE(MAX(stream_version), 0) as current_version
FROM events WHERE stream_id = $1 FOR UPDATE`,
[streamId]
);
const currentVersion = parseInt(result.rows[0].current_version, 10);
if (currentVersion !== expectedVersion) {
throw new Error(
`Concurrency conflict: expected ${expectedVersion}, got ${currentVersion}`
);
}
for (const event of events) {
await client.query(
`INSERT INTO events
(id, stream_id, stream_version, type, data, metadata, occurred_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
[
event.id, event.streamId, event.streamVersion,
event.type, JSON.stringify(event.data),
JSON.stringify(event.metadata), event.metadata.occurredAt,
]
);
}
await client.query('COMMIT');
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
async getEvents(
streamId: string,
fromVersion?: number
): Promise<StoredEvent[]> {
const result = await this.pool.query(
`SELECT * FROM events
WHERE stream_id = $1
AND ($2::int IS NULL OR stream_version > $2)
ORDER BY stream_version ASC`,
[streamId, fromVersion ?? null]
);
return result.rows.map((row) => ({
id: row.id,
streamId: row.stream_id,
streamVersion: row.stream_version,
type: row.type,
data: row.data,
metadata: row.metadata,
}));
}
async getEventsByCorrelationId(
correlationId: string
): Promise<StoredEvent[]> {
const result = await this.pool.query(
`SELECT * FROM events
WHERE metadata->>'correlationId' = $1
ORDER BY occurred_at ASC`,
[correlationId]
);
return result.rows.map((row) => ({
id: row.id,
streamId: row.stream_id,
streamVersion: row.stream_version,
type: row.type,
data: row.data,
metadata: row.metadata,
}));
}
}The UNIQUE (stream_id, stream_version) constraint is the single most important line in this entire system. It provides a database-level guarantee that no two processes can write the same version number to the same stream. Combined with SELECT ... FOR UPDATE in the append transaction, this gives us serializable isolation for event writes without any distributed locks. If two requests try to append events simultaneously to the same stream, one will succeed and the other will receive a constraint violation — forcing the loser to reload the stream and retry.
The getEventsByCorrelationId method is what makes command idempotency possible. Every incoming command carries a correlation ID. Before processing, the command handler checks whether any events with that correlation ID already exist. If they do, the command has already been processed and the handler returns immediately. This is how you handle duplicate webhooks, retried HTTP requests, and at-least-once message delivery without creating duplicate state transitions.
The repository bridges the aggregate and the event store: loading hydrates an aggregate by replaying its event stream, and saving persists new events with optimistic concurrency checks.
The expectedVersion calculation in the save method is worth studying: When you load an account from the store, replaying N events puts the aggregate at version N. You then run a command that records M new events, bumping the in-memory version to N + M. When you save, you tell the event store: "I expect this stream to still be at version N. If another process wrote to it while I was processing, reject my write." The event store compares N against the current actual version — if they match, the write proceeds. If not, a concurrency conflict error is thrown, and the client retries the entire operation.
class BankAccountRepository {
private store: PostgresEventStore;
constructor(store: PostgresEventStore) {
this.store = store;
}
async getById(accountId: string): Promise<BankAccount | null> {
const events = await this.store.getEvents(accountId);
if (events.length === 0) {
return null;
}
const account = new BankAccount(accountId, 'placeholder');
for (const event of events) {
account.applyEvent(event);
}
account.clearUncommittedEvents();
return account;
}
async save(account: BankAccount): Promise<void> {
const uncommitted = account.getUncommittedEvents();
if (uncommitted.length === 0) {
return;
}
const expectedVersion = account.version - uncommitted.length;
await this.store.appendEvents(account.id, expectedVersion, uncommitted);
account.clearUncommittedEvents();
}
}This repository is intentionally simple. In production, you would add retry logic around the save method — if a concurrency conflict occurs, you reload the aggregate, re-execute the command against the updated state, and try again. For commands that are not idempotent by nature (like generating a unique confirmation number), you would also add a deduplication check using the correlation ID before re-executing.
If you have ever tried to query "show me all accounts with a balance over $10,000" by replaying every account's event stream, you know why projections exist. Event streams are optimized for writes but terrible for reads that filter, sort, or aggregate across streams.
A projection is a materialized view of the event stream stored in a query-friendly format — typically a relational table. The projection handler consumes events as they are written and updates the read table. The critical property is idempotence: applying the same event twice must produce the same result as applying it once.
Here is a bank account projection that maintains a bank_account_read table:
class BankAccountProjection {
private pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async initialize(): Promise<void> {
await this.pool.query(`
CREATE TABLE IF NOT EXISTS bank_account_read (
account_id VARCHAR(255) PRIMARY KEY,
owner_id VARCHAR(255) NOT NULL,
balance DECIMAL(18, 2) NOT NULL DEFAULT 0,
is_closed BOOLEAN NOT NULL DEFAULT false,
version INT NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
`);
}
async project(event: StoredEvent): Promise<void> {
switch (event.type) {
case 'BankAccountOpened': {
const data = event.data as { ownerId: string };
await this.pool.query(
`INSERT INTO bank_account_read (account_id, owner_id, balance, version)
VALUES ($1, $2, 0, $3)
ON CONFLICT (account_id) DO NOTHING`,
[event.streamId, data.ownerId, event.streamVersion]
);
break;
}
case 'MoneyDeposited': {
const data = event.data as { amount: number };
await this.pool.query(
`UPDATE bank_account_read
SET balance = balance + $1, version = $2, updated_at = NOW()
WHERE account_id = $3 AND version < $2`,
[data.amount, event.streamVersion, event.streamId]
);
break;
}
case 'MoneyWithdrawn': {
const data = event.data as { amount: number };
await this.pool.query(
`UPDATE bank_account_read
SET balance = balance - $1, version = $2, updated_at = NOW()
WHERE account_id = $3 AND version < $2`,
[data.amount, event.streamVersion, event.streamId]
);
break;
}
case 'AccountClosed':
await this.pool.query(
`UPDATE bank_account_read
SET is_closed = true, version = $1, updated_at = NOW()
WHERE account_id = $2 AND version < $1`,
[event.streamVersion, event.streamId]
);
break;
}
}
}The WHERE version < $2 clause in every UPDATE is the idempotency guard. It ensures that if a projection handler processes the same event twice — due to a retry, a re-delivery, or a replay from an earlier point in the stream — the second application is a no-op because the projection's version is already at or beyond the event's version. This pattern is universal across all event-sourced projections and is one of the first things to verify when reviewing projection code.
A note on architecture: in production, projections should not be updated synchronously within the command handler that writes events. Instead, events should be published to a message broker or polled from the event store by dedicated projection workers. This decoupling prevents slow projections from blocking command processing and allows different projections to update at different speeds. PostgreSQL's LISTEN/NOTIFY provides a simple pub-sub mechanism for this within a single database, while Kafka or RabbitMQ is appropriate for multi-service architectures.
The command handler is the entry point for business operations. It receives a command, loads the aggregate, executes the command, persists events, and updates the read side. In a well-structured system, the command handler is thin — it coordinates but contains no business logic.
class BankAccountCommandHandler {
constructor(
private repo: BankAccountRepository,
private projection: BankAccountProjection
) {}
async handleDeposit(
accountId: string,
amount: number,
description: string,
correlationId: string
): Promise<void> {
// Idempotency check
const existing = await this.store.getEventsByCorrelationId(correlationId);
if (existing.length > 0) {
return;
}
const account = await this.repo.getById(accountId);
if (!account) {
throw new Error(`Account ${accountId} not found`);
}
account.deposit(amount, description);
await this.repo.save(account);
for (const event of account.getUncommittedEvents()) {
await this.projection.project(event);
}
}
async handleWithdraw(
accountId: string,
amount: number,
description: string,
correlationId: string
): Promise<void> {
const existing = await this.store.getEventsByCorrelationId(correlationId);
if (existing.length > 0) {
return;
}
const account = await this.repo.getById(accountId);
if (!account) {
throw new Error(`Account ${accountId} not found`);
}
account.withdraw(amount, description);
await this.repo.save(account);
for (const event of account.getUncommittedEvents()) {
await this.projection.project(event);
}
}
}The idempotency check at the top of each handler is the guard against double-processing. In distributed systems, network timeouts and retries mean that every command handler will eventually receive the same logical request more than once. Checking for an existing event with the same correlation ID before processing ensures that the retry is a no-op rather than a duplicate transaction. This pattern is load-bearing in production — without it, every network blip becomes a potential double-charge.
Loading an aggregate by replaying every event works until it doesn't. A bank account opened in 2018 with daily transactions might have 3,000 events by 2026. Replaying all of them on every read becomes a bottleneck.
Snapshots solve this by periodically capturing the aggregate's state at a specific version. When loading, you find the most recent snapshot and replay only events that occurred after it. If you snapshot every 100 events, loading a stream with 3,000 events goes from 3,000 event applications to at most 100.
The snapshot store is a separate table that maps stream ID and version to a JSON blob of the aggregate's state. A snapshot is a performance optimization, not a source of truth — the event store remains the authoritative record, and snapshots can be deleted and rebuilt from events at any time.
class SnapshotStore {
private pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async initialize(): Promise<void> {
await this.pool.query(`
CREATE TABLE IF NOT EXISTS snapshots (
stream_id VARCHAR(255) NOT NULL,
version INT NOT NULL,
state JSONB NOT NULL,
snapshot_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (stream_id, version)
);
`);
}
async saveSnapshot(
streamId: string,
version: number,
state: Record<string, unknown>
): Promise<void> {
await this.pool.query(
`INSERT INTO snapshots (stream_id, version, state)
VALUES ($1, $2, $3)
ON CONFLICT (stream_id, version) DO UPDATE SET state = $3`,
[streamId, version, JSON.stringify(state)]
);
}
async getLatestSnapshot(streamId: string): Promise<{
version: number;
state: Record<string, unknown>;
} | null> {
const result = await this.pool.query(
`SELECT version, state FROM snapshots
WHERE stream_id = $1
ORDER BY version DESC LIMIT 1`,
[streamId]
);
if (result.rows.length === 0) return null;
return { version: result.rows[0].version, state: result.rows[0].state };
}
}The repository with snapshot support loads the snapshot, hydrates the aggregate's fields directly from it, then replays only the events that happened after the snapshot version. If no snapshot exists, it falls back to a full replay. The snapshot frequency — "snapshot every 50 events" — is a tuning parameter. For high-throughput streams, snapshot less frequently (every 200-500 events) because the snapshot write cost is non-trivial. For low-throughput streams with bursty reads, snapshot more aggressively because read latency matters more than snapshot overhead.
class BankAccountRepositoryWithSnapshots extends BankAccountRepository {
private snapshots: SnapshotStore;
constructor(store: PostgresEventStore, snapshots: SnapshotStore) {
super(store);
this.snapshots = snapshots;
}
async getById(accountId: string): Promise<BankAccount | null> {
const snapshot = await this.snapshots.getLatestSnapshot(accountId);
let account: BankAccount;
if (snapshot) {
account = new BankAccount(accountId, snapshot.state.ownerId as string);
account.balance = snapshot.state.balance as number;
account.isClosed = snapshot.state.isClosed as boolean;
account.version = snapshot.version;
account.clearUncommittedEvents();
const subsequentEvents = await this.store.getEvents(
accountId,
snapshot.version
);
for (const event of subsequentEvents) {
account.applyEvent(event);
}
} else {
const events = await this.store.getEvents(accountId);
if (events.length === 0) return null;
account = new BankAccount(accountId, 'placeholder');
for (const event of events) {
account.applyEvent(event);
}
}
account.clearUncommittedEvents();
return account;
}
async saveAndSnapshot(account: BankAccount): Promise<void> {
await this.save(account);
if (account.version % 50 === 0) {
await this.snapshots.saveSnapshot(account.id, account.version, {
balance: account.balance,
ownerId: account.ownerId,
isClosed: account.isClosed,
});
}
}
}The true payoff of event sourcing is not in architecture diagrams — it is in queries that would be impossible in a traditional system. Because every state change is an immutable event with a timestamp, you can query the system at any point in time, reconstruct the exact state of any aggregate on any date, and trace transaction lifecycles across services by following correlation IDs.
When a customer disputes a charge from three months ago, you query the event store with a timestamp filter and get the exact state as it existed at that moment. No "the logs might have rotated," no "that was before we added audit logging to that endpoint." The event store is the authoritative record of everything that happened.
class TemporalQueryService {
private pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async getBalanceAt(accountId: string, at: Date): Promise<number> {
const result = await this.pool.query(
`SELECT type, data FROM events
WHERE stream_id = $1 AND occurred_at <= $2
ORDER BY stream_version ASC`,
[accountId, at]
);
let balance = 0;
for (const event of result.rows) {
if (event.type === 'MoneyDeposited') {
balance += event.data.amount;
} else if (event.type === 'MoneyWithdrawn') {
balance -= event.data.amount;
}
}
return balance;
}
async getTransactionHistory(
accountId: string,
from: Date,
to: Date
): Promise<Array<{ type: string; amount: number; at: Date }>> {
const result = await this.pool.query(
`SELECT type, data, occurred_at FROM events
WHERE stream_id = $1
AND occurred_at BETWEEN $2 AND $3
AND type IN ('MoneyDeposited', 'MoneyWithdrawn')
ORDER BY occurred_at ASC`,
[accountId, from, to]
);
return result.rows.map((e) => ({
type: e.type,
amount: e.data.amount,
at: e.occurred_at,
}));
}
async reconstructStateAt(
accountId: string,
at: Date
): Promise<{ balance: number; isClosed: boolean; version: number }> {
const result = await this.pool.query(
`SELECT type, data, stream_version FROM events
WHERE stream_id = $1 AND occurred_at <= $2
ORDER BY stream_version ASC`,
[accountId, at]
);
let balance = 0;
let isClosed = false;
let version = 0;
for (const event of result.rows) {
switch (event.type) {
case 'MoneyDeposited':
balance += event.data.amount;
break;
case 'MoneyWithdrawn':
balance -= event.data.amount;
break;
case 'AccountClosed':
isClosed = true;
break;
}
version = event.stream_version;
}
return { balance, isClosed, version };
}
}This temporal query capability exposes something subtle: event sourcing is a different relationship with data. In a CRUD system, you ask "what is the current state?" In an event-sourced system, you can ask "what was the state at any point in time, and how did it get there?" The second question is almost always more useful for debugging.
One of the hardest problems in event sourcing is that events are immutable but your understanding of the domain evolves. An event written two years ago might be missing fields today's code expects. For example, you might have started without currency support:
// Version 1 of MoneyDeposited data
{ amount: 100, description: "Paycheck" }
// Version 2 of MoneyDeposited data (added currency field)
{ amount: 100, currency: "USD", description: "Paycheck" }When your event handler encounters a v1 event, the currency field is missing. Without a migration strategy, you get undefined where you expect a string, and the replay either crashes or silently corrupts state.
The solution is upcasters — pure functions that transform old event data into the current schema when events are loaded. Upcasters run during replay, between the event store and the aggregate, and they do not modify the stored events. The stored event remains in its original form forever. The upcaster produces a transformed version for the current code to consume.
interface StoredEvent {
id: string;
streamId: string;
streamVersion: number;
type: EventType;
dataVersion: number; // <-- add this to every event
data: Record<string, unknown>;
metadata: StoredEventMetadata;
}
const upcasters: Record<
string,
(data: Record<string, unknown>) => Record<string, unknown>
> = {
'MoneyDeposited_v1_to_v2': (data) => ({
amount: data.amount,
currency: (data as any).currency || 'USD',
description: data.description,
}),
};
function upcast(event: StoredEvent): StoredEvent {
let data = event.data;
let version = event.dataVersion;
while (true) {
const key = `${event.type}_v${version}_to_v${version + 1}`;
if (!upcasters[key]) break;
data = upcasters[key](data);
version += 1;
}
return { ...event, data, dataVersion: version };
}The upcaster chain is applied during event retrieval, before events reach the aggregate. The upcasters themselves are pure functions with no side effects and no database access — they transform data shapes and nothing else. This constraint is important because upcasters run on every event replay, potentially thousands of times per stream load, and any IO in an upcaster would destroy read performance.
Event-sourced systems are remarkably testable. Every test follows the same three-step pattern: given a history of events that establishes the aggregate's state, when I execute a command, I expect specific events to be produced. There are no mocks, no database fixtures, and no shared state. Every test is self-contained because the aggregate's entire state is derived from the events you feed it.
This is a radical improvement over testing CRUD systems, where tests typically need a database with pre-inserted rows or elaborate mocking of repository methods. In event sourcing, the test setup is literally a list of events. The test assertion is literally a list of expected events. The code under test is a pure function from (events, command) to (new events).
function makeEvent(
type: EventType,
data: Record<string, unknown>,
version: number = 1
): StoredEvent {
return {
id: `evt-${version}`,
streamId: 'acc-1',
streamVersion: version,
type,
data,
metadata: {
correlationId: `corr-${version}`,
causationId: null,
userId: null,
occurredAt: new Date(),
ipAddress: null,
},
};
}
describe('BankAccount', () => {
test('withdraw reduces balance and emits correct event', () => {
const account = new BankAccount('acc-1', 'user-1');
account.applyEvent(makeEvent('BankAccountOpened', { ownerId: 'user-1' }));
account.applyEvent(makeEvent('MoneyDeposited', { amount: 100 }, 2));
account.clearUncommittedEvents();
account.withdraw(40, 'ATM withdrawal');
const events = account.getUncommittedEvents();
expect(events).toHaveLength(1);
expect(events[0].type).toBe('MoneyWithdrawn');
expect(events[0].data).toEqual({
amount: 40,
description: 'ATM withdrawal',
});
expect(account.balance).toBe(60);
});
test('prevents overdraft', () => {
const account = new BankAccount('acc-1', 'user-1');
account.applyEvent(makeEvent('BankAccountOpened', { ownerId: 'user-1' }));
account.applyEvent(makeEvent('MoneyDeposited', { amount: 25 }, 2));
account.clearUncommittedEvents();
expect(() =>
account.withdraw(100, 'Overdraft attempt')
).toThrow('Insufficient funds');
expect(account.getUncommittedEvents()).toHaveLength(0);
expect(account.balance).toBe(25);
});
test('prevents operations on closed account', () => {
const account = new BankAccount('acc-1', 'user-1');
account.applyEvent(makeEvent('BankAccountOpened', { ownerId: 'user-1' }));
account.applyEvent(makeEvent('MoneyDeposited', { amount: 100 }, 2));
account.applyEvent(makeEvent('AccountClosed', {}, 3));
account.clearUncommittedEvents();
expect(() =>
account.withdraw(10, 'After close')
).toThrow('Cannot withdraw from a closed account');
expect(() =>
account.deposit(10, 'After close')
).toThrow('Cannot deposit to a closed account');
});
});These tests verify business rules without touching a database or a network. The entire test suite runs in milliseconds. More importantly, the tests document the business rules. Reading the test "prevents overdraft" tells you exactly what the system does when someone tries to withdraw more than their balance. The events are the documentation.
Event sourcing is a specialized tool with real costs. Applying it where it does not fit will slow your team down without proportional benefit.
Your domain requires a complete, regulatory-grade audit trail. Banking, healthcare, insurance, and legal applications must be able to explain every state change for years. Event sourcing provides this as a natural consequence of the architecture rather than a bolt-on logging system that drifts over time.
You need temporal queries as a first-class feature. If your users routinely ask "what did this look like on March 15?" or your compliance team needs end-of-day snapshots for every business day of the year, event sourcing gives you this with a parameterized query instead of a data pipeline.
Your system must support compensating transactions. In financial systems, you do not delete a mistaken transaction — you post a reversal. Event sourcing models this naturally: the original event stays, a correction event adjusts the state, and the full history is preserved.
You are building collaborative editing or multi-user sync. Event sourcing and CRDTs are natural partners — the event stream serves as the sync protocol, and each participant replays events to converge on the same state.
Your domain has no auditing requirements and you are building a standard CRUD application. A blog, a content management system, or an internal dashboard does not need the complexity of event sourcing. A simple updated_at timestamp on your rows is sufficient.
Your team is small and still learning the domain. Event sourcing doubles the number of concepts in your codebase: you must understand events, aggregates, projections, snapshots, and replay. If your team is figuring out what to build, event sourcing will slow that discovery process.
Your reads vastly outnumber your writes and you cannot afford the projection infrastructure. If your application does 10,000 reads per second and 10 writes per second, maintaining projections for every read pattern becomes expensive. Consider a traditional approach with a lightweight audit log instead.
You just need an audit log, not full event sourcing. A separate audit_events table appended to in the same database transaction as your state update gives you roughly 80% of the audit benefit with 10% of the complexity. Event sourcing is worth it when you need temporal queries, replay capability, and event-driven projections — not just when you want to know who changed what.
The litmus test: If someone asked you "tell me exactly what happened to record X on February 14 at 3:42 PM" and you would need to consult logs, backups, or a disorganized audit table, event sourcing is a strong candidate. If you can answer that question with a simple SELECT on your existing audit_log table, you probably do not need to re-architect.
Before shipping an event-sourced system, address these five concerns. Each one has caused production incidents in real event-sourced systems, and each one has a straightforward mitigation.
Add a dataVersion field to every event from day one. Write upcasters as pure functions. Test that every historical event shape can be upcast to the current schema without errors. Do this even if you think your schema is final — it never is.
Event stores grow monotonically. A system processing 100 events per second generates 8.6 million events per day and over 3 billion per year. Mitigations include partitioning the events table by month (PostgreSQL supports declarative partitioning), archiving events older than your retention period to cold storage, and being aggressive about snapshot frequency to reduce replay depth. Most importantly, decide on a retention policy before launch — regulatory requirements, not convenience, should drive it.
Every command handler must survive being called twice with the same inputs. The correlation ID pattern — checking for an existing event with the same correlation ID before processing — is the standard approach. For commands that generate unique identifiers (order numbers, confirmation codes), generate them from deterministic inputs so that the same command always produces the same identifier.
Projections are eventually consistent with the event store. Between the moment an event is written and the moment the projection updates, the read model shows stale data. Decide what consistency guarantees your application needs. If reads must see the user's own writes, maintain a read-through cache for the writing user's aggregates. For everyone else, eventual consistency is usually acceptable.
Monitor the lag between event writes and projection updates. Monitor the event store's append rate and disk usage. Monitor snapshot freshness — if snapshots stop being created, replay times will grow until reads time out. Monitor concurrency conflict rates — a rising conflict rate indicates contention on specific aggregates that may need redesign.
Event sourcing is a paradigm shift from "store the current state" to "store everything that happened." In TypeScript, with a PostgreSQL event store, the repository pattern, projections, and snapshots, you can build a system that gives you complete auditability, temporal query capabilities, and an architecture that aligns naturally with how businesses think about data: as a sequence of events, not a point-in-time snapshot.
The code in this article is production-grade. The patterns — optimistic concurrency control, idempotent projections, snapshot-based hydration, correlation-based deduplication, upcasters for schema evolution — are proven in systems handling millions of events per day across banking, e-commerce, and collaboration platforms.
Start small. Event-source a single aggregate in your application — the one where auditability matters most. See if the benefits of temporal queries and immutable history outweigh the costs of projection maintenance and schema evolution. If they do, expand the pattern to other aggregates. If they do not, you will still walk away with a deeper understanding of how your application's state flows, and a reusable event store that you can apply when the right problem comes along.
Not every system needs event sourcing. But for the ones that do, nothing else comes close.