Event Sourcing Patterns for Audit-Heavy Applications
Use event sourcing to keep a complete, immutable audit trail of every state change, with patterns for projections, snapshotting and event replay.
Why State Alone Is Not Enough
Most applications store the current state of entities. A user's balance is $500. An order's status is "shipped." But when an auditor asks "how did this balance reach $500?" or "who changed the order status and when?", you are guessing from incomplete logs. Event sourcing solves this by storing every state change as an immutable event, making the full history the source of truth.
The Event Store Foundation
An event store is an append-only log. Events are never updated or deleted. To know the current state of an entity, you replay its events from the beginning.
interface DomainEvent {
eventId: string;
aggregateId: string;
aggregateType: string;
eventType: string;
payload: Record<string, unknown>;
metadata: {
timestamp: string;
userId: string;
correlationId: string;
causationId: string;
};
version: number;
}
class EventStore {
constructor(private readonly db: Database) {}
async append(
aggregateId: string,
events: DomainEvent[],
expectedVersion: number
): Promise<void> {
await this.db.transaction(async (tx) => {
// Optimistic concurrency check
const current = await tx.query<{ max_version: number }>(
"SELECT COALESCE(MAX(version), 0) as max_version FROM events WHERE aggregate_id = $1",
[aggregateId]
);
if (current.rows[0].max_version !== expectedVersion) {
throw new ConcurrencyError(
`Expected version ${expectedVersion}, ` +
`but found ${current.rows[0].max_version}`
);
}
for (const event of events) {
await tx.query(
`INSERT INTO events (event_id, aggregate_id, aggregate_type,
event_type, payload, metadata, version)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
[
event.eventId,
event.aggregateId,
event.aggregateType,
event.eventType,
JSON.stringify(event.payload),
JSON.stringify(event.metadata),
event.version,
]
);
}
});
}
async getEvents(
aggregateId: string,
fromVersion?: number
): Promise<DomainEvent[]> {
const result = await this.db.query<DomainEvent>(
`SELECT * FROM events
WHERE aggregate_id = $1 AND version > $2
ORDER BY version ASC`,
[aggregateId, fromVersion ?? 0]
);
return result.rows;
}
}Aggregates That Produce Events
Aggregates are the write model in event sourcing. Commands come in, business rules validate them, and the aggregate emits events describing what happened—not what was requested.
// ❌ Mutable state with no history
class Account {
balance: number = 0;
withdraw(amount: number): void {
this.balance -= amount; // History is lost
}
}
// ✅ Event-sourced aggregate with full audit trail
class AccountAggregate {
private balance: number = 0;
private status: "active" | "frozen" = "active";
private uncommittedEvents: DomainEvent[] = [];
private version: number = 0;
static fromEvents(events: DomainEvent[]): AccountAggregate {
const account = new AccountAggregate();
for (const event of events) {
account.apply(event);
account.version = event.version;
}
return account;
}
withdraw(amount: number, userId: string, correlationId: string): void {
if (this.status === "frozen") {
throw new BusinessRuleError("Cannot withdraw from a frozen account");
}
if (amount <= 0) {
throw new BusinessRuleError("Withdrawal amount must be positive");
}
if (this.balance < amount) {
throw new BusinessRuleError("Insufficient funds");
}
this.emit({
eventType: "MoneyWithdrawn",
payload: { amount, previousBalance: this.balance },
userId,
correlationId,
});
}
private apply(event: DomainEvent): void {
switch (event.eventType) {
case "AccountOpened":
this.balance = event.payload.initialDeposit as number;
this.status = "active";
break;
case "MoneyDeposited":
this.balance += event.payload.amount as number;
break;
case "MoneyWithdrawn":
this.balance -= event.payload.amount as number;
break;
case "AccountFrozen":
this.status = "frozen";
break;
}
}
getUncommittedEvents(): DomainEvent[] {
return [...this.uncommittedEvents];
}
}Projections for Read Models
Event sourcing separates writes from reads. The event store handles writes. Projections subscribe to events and build optimized read models for queries. This is the CQRS pattern applied naturally.
interface Projection {
name: string;
handle(event: DomainEvent): Promise<void>;
}
class AccountBalanceProjection implements Projection {
name = "account-balance";
constructor(private readonly readDb: Database) {}
async handle(event: DomainEvent): Promise<void> {
switch (event.eventType) {
case "AccountOpened":
await this.readDb.query(
`INSERT INTO account_balances (account_id, balance, owner_name, updated_at)
VALUES ($1, $2, $3, $4)`,
[
event.aggregateId,
event.payload.initialDeposit,
event.payload.ownerName,
event.metadata.timestamp,
]
);
break;
case "MoneyDeposited":
case "MoneyWithdrawn": {
const delta =
event.eventType === "MoneyDeposited"
? (event.payload.amount as number)
: -(event.payload.amount as number);
await this.readDb.query(
`UPDATE account_balances
SET balance = balance + $1, updated_at = $2
WHERE account_id = $3`,
[delta, event.metadata.timestamp, event.aggregateId]
);
break;
}
}
}
}
class AuditLogProjection implements Projection {
name = "audit-log";
constructor(private readonly readDb: Database) {}
async handle(event: DomainEvent): Promise<void> {
await this.readDb.query(
`INSERT INTO audit_log (event_id, aggregate_id, event_type,
user_id, timestamp, correlation_id, details)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
[
event.eventId,
event.aggregateId,
event.eventType,
event.metadata.userId,
event.metadata.timestamp,
event.metadata.correlationId,
JSON.stringify(event.payload),
]
);
}
}Snapshots for Performance
Replaying thousands of events for every command is expensive. Snapshots periodically capture the aggregate state so replay only needs events since the last snapshot.
class SnapshotStore {
constructor(private readonly db: Database) {}
async save(
aggregateId: string,
state: Record<string, unknown>,
version: number
): Promise<void> {
await this.db.query(
`INSERT INTO snapshots (aggregate_id, state, version, created_at)
VALUES ($1, $2, $3, NOW())
ON CONFLICT (aggregate_id) DO UPDATE
SET state = $2, version = $3, created_at = NOW()`,
[aggregateId, JSON.stringify(state), version]
);
}
async load(
aggregateId: string
): Promise<{ state: Record<string, unknown>; version: number } | null> {
const result = await this.db.query(
"SELECT state, version FROM snapshots WHERE aggregate_id = $1",
[aggregateId]
);
return result.rows[0] ?? null;
}
}
class AccountRepository {
private static SNAPSHOT_INTERVAL = 100;
constructor(
private readonly eventStore: EventStore,
private readonly snapshotStore: SnapshotStore
) {}
async load(aggregateId: string): Promise<AccountAggregate> {
const snapshot = await this.snapshotStore.load(aggregateId);
const fromVersion = snapshot?.version ?? 0;
const events = await this.eventStore.getEvents(aggregateId, fromVersion);
let aggregate: AccountAggregate;
if (snapshot) {
aggregate = AccountAggregate.fromSnapshot(snapshot.state, snapshot.version);
aggregate.replayEvents(events);
} else {
aggregate = AccountAggregate.fromEvents(events);
}
return aggregate;
}
async save(aggregate: AccountAggregate): Promise<void> {
const events = aggregate.getUncommittedEvents();
await this.eventStore.append(
aggregate.id,
events,
aggregate.version - events.length
);
if (aggregate.version % AccountRepository.SNAPSHOT_INTERVAL === 0) {
await this.snapshotStore.save(
aggregate.id,
aggregate.toSnapshot(),
aggregate.version
);
}
}
}Event Versioning and Migration
Events are immutable, but your understanding of the domain evolves. When event schemas change, use upcasters to transform old events into the current format during replay.
type Upcaster = (event: DomainEvent) => DomainEvent;
const upcasters: Map<string, Upcaster[]> = new Map([
[
"MoneyWithdrawn",
[
// v1 → v2: Added 'channel' field
(event) => {
if (!event.payload.channel) {
return {
...event,
payload: { ...event.payload, channel: "unknown" },
};
}
return event;
},
// v2 → v3: Renamed 'previousBalance' to 'balanceBefore'
(event) => {
if ("previousBalance" in event.payload) {
const { previousBalance, ...rest } = event.payload;
return {
...event,
payload: {
...rest,
balanceBefore: previousBalance,
},
};
}
return event;
},
],
],
]);
function upcastEvent(event: DomainEvent): DomainEvent {
const eventUpcasters = upcasters.get(event.eventType) ?? [];
return eventUpcasters.reduce((e, upcaster) => upcaster(e), event);
}Key Takeaways
Event sourcing turns every state change into a permanent, queryable record. For audit-heavy domains—finance, healthcare, compliance—this is not optional, it is the correct architecture. The event store is append-only and immutable: events record facts that already happened.
Aggregates enforce business rules and emit events. Projections consume events to build optimized read models for any query pattern. Snapshots prevent performance degradation as event streams grow. Upcasters handle schema evolution without corrupting the historical record.
Start with a simple event store, one aggregate, and one projection. Add complexity—snapshots, upcasters, multiple projections—only when the domain demands it. The audit trail you build today is the compliance answer you give tomorrow.
