CQRS and Event Sourcing with TypeScript
CQRS (Command Query Responsibility Segregation) and Event Sourcing are two architectural patterns that, when combined, offer a radically different way of designing software systems. Instead of reading and writing against the same data model, CQRS separates write operations (commands) from read operations (queries). Event Sourcing, in turn, replaces traditional state persistence with an ordered sequence of immutable events that describe everything that has happened in the system.
In this article, we will explore both patterns in depth with concrete TypeScript implementations. We will use an order management system as an example to illustrate commands, queries, event stores, projections, and snapshots.
What You Will Learn
- The CQRS pattern: separating commands and queries
- Command Bus and Query Bus in TypeScript
- Event Sourcing: event-based persistence
- Implementing an Event Store
- Projections and Read Models
- Snapshots for performance optimization
- When to use (and when to avoid) CQRS and Event Sourcing
CQRS: Separating Commands and Queries
The principle behind CQRS is straightforward: the model used to write data should not be the same model used to read it. In a traditional architecture, a single model (and often a single table) handles both write and read operations. CQRS introduces two distinct models, each optimized for its own purpose.
Commands
A Command represents the intent to change the system's state. It is an immutable object containing all the data needed to perform the operation. Commands do not return data: their sole effect is state modification.
// cqrs/command.ts
export interface Command {
readonly type: string;
}
export interface CommandHandler<T extends Command> {
execute(command: T): Promise<void>;
}
// commands/create-order.command.ts
export class CreateOrderCommand implements Command {
readonly type = 'CreateOrder';
constructor(
public readonly orderId: string,
public readonly customerId: string,
public readonly items: {
productId: string;
productName: string;
quantity: number;
unitPrice: number;
}[]
) {}
}
// commands/handlers/create-order.handler.ts
export class CreateOrderHandler
implements CommandHandler<CreateOrderCommand> {
constructor(private readonly eventStore: EventStore) {}
async execute(command: CreateOrderCommand): Promise<void> {
// Check that the order does not already exist
const existing = await this.eventStore.getEvents(command.orderId);
if (existing.length > 0) {
throw new Error('Order already exists');
}
// Create the aggregate and generate events
const order = Order.create(
command.orderId,
command.customerId,
command.items
);
// Persist the generated events
const events = order.pullUncommittedEvents();
await this.eventStore.appendEvents(command.orderId, events, 0);
}
}
Queries
A Query represents a request for data. It does not modify the system state and returns a result optimized for the specific use case. The Read Model can be denormalized, precomputed, and structured exactly as the UI needs it.
// cqrs/query.ts
export interface Query {
readonly type: string;
}
export interface QueryHandler<TQuery extends Query, TResult> {
execute(query: TQuery): Promise<TResult>;
}
// queries/get-order-summary.query.ts
export class GetOrderSummaryQuery implements Query {
readonly type = 'GetOrderSummary';
constructor(public readonly orderId: string) {}
}
export interface OrderSummaryReadModel {
orderId: string;
customerName: string;
status: string;
itemCount: number;
totalAmount: number;
createdAt: string;
lastUpdatedAt: string;
}
// queries/handlers/get-order-summary.handler.ts
export class GetOrderSummaryHandler
implements QueryHandler<GetOrderSummaryQuery, OrderSummaryReadModel> {
constructor(private readonly readDb: ReadDatabase) {}
async execute(
query: GetOrderSummaryQuery
): Promise<OrderSummaryReadModel> {
const summary = await this.readDb.findOne<OrderSummaryReadModel>(
'order_summaries',
{ orderId: query.orderId }
);
if (!summary) {
throw new Error('Order not found');
}
return summary;
}
}
Command Bus and Query Bus
The Bus is the mechanism that connects commands/queries to their respective handlers. It acts as a centralized dispatcher, decoupling the sender from the handler.
// cqrs/command-bus.ts
export class CommandBus {
private handlers = new Map<string, CommandHandler<any>>();
register<T extends Command>(
commandType: string,
handler: CommandHandler<T>
): void {
if (this.handlers.has(commandType)) {
throw new Error(`Handler already registered for: ${commandType}`);
}
this.handlers.set(commandType, handler);
}
async dispatch(command: Command): Promise<void> {
const handler = this.handlers.get(command.type);
if (!handler) {
throw new Error(`No handler found for command: ${command.type}`);
}
await handler.execute(command);
}
}
// cqrs/query-bus.ts
export class QueryBus {
private handlers = new Map<string, QueryHandler<any, any>>();
register<TQuery extends Query, TResult>(
queryType: string,
handler: QueryHandler<TQuery, TResult>
): void {
this.handlers.set(queryType, handler);
}
async dispatch<TResult>(query: Query): Promise<TResult> {
const handler = this.handlers.get(query.type);
if (!handler) {
throw new Error(`No handler found for query: ${query.type}`);
}
return handler.execute(query);
}
}
Event Sourcing: State as a Sequence of Events
Event Sourcing turns the traditional approach to persistence upside down. Instead of saving the current state of an entity, we save every event that led to that state. The current state is obtained by replaying all events in chronological order.
Benefits of Event Sourcing
| Benefit | Description |
|---|---|
| Complete audit trail | Every change is recorded; the state can be reconstructed at any point in time |
| Advanced debugging | Replaying events reveals exactly how the current state was reached |
| Multiple projections | Different read models can be built from the same events for different needs |
| Temporal evolution | New projections can be created retroactively by reprocessing historical events |
| No data loss | Events are immutable and append-only; nothing is ever deleted |
Domain Events for Event Sourcing
// events/base-event.ts
export interface DomainEvent {
readonly eventType: string;
readonly aggregateId: string;
readonly occurredOn: Date;
readonly version: number;
readonly payload: Record<string, unknown>;
}
// events/order-events.ts
export class OrderCreatedEvent implements DomainEvent {
readonly eventType = 'OrderCreated';
readonly occurredOn = new Date();
constructor(
public readonly aggregateId: string,
public readonly version: number,
public readonly payload: {
customerId: string;
items: {
productId: string;
productName: string;
quantity: number;
unitPrice: number;
}[];
}
) {}
}
export class OrderConfirmedEvent implements DomainEvent {
readonly eventType = 'OrderConfirmed';
readonly occurredOn = new Date();
constructor(
public readonly aggregateId: string,
public readonly version: number,
public readonly payload: {
confirmedAt: string;
totalAmount: number;
}
) {}
}
export class OrderItemAddedEvent implements DomainEvent {
readonly eventType = 'OrderItemAdded';
readonly occurredOn = new Date();
constructor(
public readonly aggregateId: string,
public readonly version: number,
public readonly payload: {
productId: string;
productName: string;
quantity: number;
unitPrice: number;
}
) {}
}
export class OrderCancelledEvent implements DomainEvent {
readonly eventType = 'OrderCancelled';
readonly occurredOn = new Date();
constructor(
public readonly aggregateId: string,
public readonly version: number,
public readonly payload: {
reason: string;
cancelledAt: string;
}
) {}
}
Event-Sourced Aggregate
In an Event Sourced system, the aggregate does not maintain state directly but reconstructs it by applying events. Each business operation generates new events that are accumulated and then persisted to the Event Store.
// aggregates/order.aggregate.ts
export class Order {
private _id: string = '';
private _customerId: string = '';
private _status: OrderStatus = OrderStatus.CREATED;
private _items: OrderItem[] = [];
private _version: number = 0;
private _uncommittedEvents: DomainEvent[] = [];
// Factory method: creates a new order
static create(
orderId: string,
customerId: string,
items: { productId: string; productName: string;
quantity: number; unitPrice: number }[]
): Order {
const order = new Order();
order.apply(new OrderCreatedEvent(orderId, 1, {
customerId,
items,
}));
return order;
}
// Reconstruction from historical events
static fromHistory(events: DomainEvent[]): Order {
const order = new Order();
for (const event of events) {
order.applyFromHistory(event);
}
return order;
}
confirm(): void {
if (this._status !== OrderStatus.CREATED) {
throw new Error('Only orders with CREATED status can be confirmed');
}
if (this._items.length === 0) {
throw new Error('Cannot confirm an empty order');
}
const total = this._items.reduce(
(sum, item) => sum + item.quantity * item.unitPrice, 0
);
this.apply(new OrderConfirmedEvent(this._id, this._version + 1, {
confirmedAt: new Date().toISOString(),
totalAmount: total,
}));
}
cancel(reason: string): void {
if (this._status === OrderStatus.SHIPPED) {
throw new Error('Cannot cancel an order that has already shipped');
}
this.apply(new OrderCancelledEvent(this._id, this._version + 1, {
reason,
cancelledAt: new Date().toISOString(),
}));
}
// Applies a new event (generates uncommitted event)
private apply(event: DomainEvent): void {
this.mutate(event);
this._uncommittedEvents.push(event);
}
// Applies a historical event (without generating uncommitted)
private applyFromHistory(event: DomainEvent): void {
this.mutate(event);
}
// Modifies internal state based on the event
private mutate(event: DomainEvent): void {
switch (event.eventType) {
case 'OrderCreated':
this._id = event.aggregateId;
this._customerId = (event.payload as any).customerId;
this._items = (event.payload as any).items.map((i: any) => ({
productId: i.productId,
productName: i.productName,
quantity: i.quantity,
unitPrice: i.unitPrice,
}));
this._status = OrderStatus.CREATED;
break;
case 'OrderConfirmed':
this._status = OrderStatus.CONFIRMED;
break;
case 'OrderItemAdded':
this._items.push(event.payload as any);
break;
case 'OrderCancelled':
this._status = OrderStatus.CANCELLED;
break;
}
this._version = event.version;
}
pullUncommittedEvents(): DomainEvent[] {
const events = [...this._uncommittedEvents];
this._uncommittedEvents = [];
return events;
}
get id(): string { return this._id; }
get version(): number { return this._version; }
get status(): OrderStatus { return this._status; }
}
Event Store
The Event Store is the event database. Each event stream corresponds to an aggregate. The two fundamental operations are appending new events and reading events from a stream. The Event Store guarantees ordering and supports optimistic concurrency through versioning.
// infrastructure/event-store.ts
export interface EventStore {
appendEvents(
streamId: string,
events: DomainEvent[],
expectedVersion: number
): Promise<void>;
getEvents(streamId: string): Promise<DomainEvent[]>;
getEventsFromVersion(
streamId: string,
fromVersion: number
): Promise<DomainEvent[]>;
getAllEvents(): Promise<DomainEvent[]>;
}
export class InMemoryEventStore implements EventStore {
private streams = new Map<string, DomainEvent[]>();
private allEvents: DomainEvent[] = [];
private subscribers: ((event: DomainEvent) => void)[] = [];
async appendEvents(
streamId: string,
events: DomainEvent[],
expectedVersion: number
): Promise<void> {
const currentEvents = this.streams.get(streamId) || [];
const currentVersion = currentEvents.length > 0
? currentEvents[currentEvents.length - 1].version
: 0;
// Optimistic concurrency check
if (currentVersion !== expectedVersion) {
throw new ConcurrencyError(
`Concurrency conflict: expected version ${expectedVersion}, ` +
`current version ${currentVersion}`
);
}
const updatedStream = [...currentEvents, ...events];
this.streams.set(streamId, updatedStream);
this.allEvents.push(...events);
// Notify subscribers (for projections)
for (const event of events) {
this.subscribers.forEach(sub => sub(event));
}
}
async getEvents(streamId: string): Promise<DomainEvent[]> {
return this.streams.get(streamId) || [];
}
async getEventsFromVersion(
streamId: string,
fromVersion: number
): Promise<DomainEvent[]> {
const events = this.streams.get(streamId) || [];
return events.filter(e => e.version > fromVersion);
}
async getAllEvents(): Promise<DomainEvent[]> {
return [...this.allEvents];
}
subscribe(handler: (event: DomainEvent) => void): void {
this.subscribers.push(handler);
}
}
Projections and Read Models
Projections are the mechanism that transforms events into Read Models optimized for reading. Each projection listens to specific event types and updates a denormalized read model. This model is designed to answer queries as efficiently as possible.
// projections/order-summary.projection.ts
export interface OrderSummaryReadModel {
orderId: string;
customerId: string;
status: string;
itemCount: number;
totalAmount: number;
createdAt: string;
updatedAt: string;
}
export class OrderSummaryProjection {
private summaries = new Map<string, OrderSummaryReadModel>();
constructor(eventStore: EventStore) {
// Subscribe to relevant events
eventStore.subscribe((event) => this.handleEvent(event));
}
private handleEvent(event: DomainEvent): void {
switch (event.eventType) {
case 'OrderCreated':
this.onOrderCreated(event);
break;
case 'OrderConfirmed':
this.onOrderConfirmed(event);
break;
case 'OrderItemAdded':
this.onOrderItemAdded(event);
break;
case 'OrderCancelled':
this.onOrderCancelled(event);
break;
}
}
private onOrderCreated(event: DomainEvent): void {
const payload = event.payload as any;
const items = payload.items || [];
const totalAmount = items.reduce(
(sum: number, i: any) => sum + i.quantity * i.unitPrice, 0
);
this.summaries.set(event.aggregateId, {
orderId: event.aggregateId,
customerId: payload.customerId,
status: 'created',
itemCount: items.length,
totalAmount,
createdAt: event.occurredOn.toISOString(),
updatedAt: event.occurredOn.toISOString(),
});
}
private onOrderConfirmed(event: DomainEvent): void {
const summary = this.summaries.get(event.aggregateId);
if (summary) {
summary.status = 'confirmed';
summary.totalAmount = (event.payload as any).totalAmount;
summary.updatedAt = event.occurredOn.toISOString();
}
}
private onOrderItemAdded(event: DomainEvent): void {
const summary = this.summaries.get(event.aggregateId);
if (summary) {
const payload = event.payload as any;
summary.itemCount += 1;
summary.totalAmount += payload.quantity * payload.unitPrice;
summary.updatedAt = event.occurredOn.toISOString();
}
}
private onOrderCancelled(event: DomainEvent): void {
const summary = this.summaries.get(event.aggregateId);
if (summary) {
summary.status = 'cancelled';
summary.updatedAt = event.occurredOn.toISOString();
}
}
// Query methods
getById(orderId: string): OrderSummaryReadModel | undefined {
return this.summaries.get(orderId);
}
getByCustomer(customerId: string): OrderSummaryReadModel[] {
return Array.from(this.summaries.values())
.filter(s => s.customerId === customerId);
}
getByStatus(status: string): OrderSummaryReadModel[] {
return Array.from(this.summaries.values())
.filter(s => s.status === status);
}
}
Snapshots: Optimizing Performance
Over time, the number of events for an aggregate can grow significantly. Reconstructing state from thousands of events becomes expensive. Snapshots solve this problem by periodically saving a photograph of the aggregate's state. Reconstruction starts from the most recent snapshot and applies only the subsequent events.
// infrastructure/snapshot-store.ts
export interface Snapshot {
aggregateId: string;
version: number;
state: Record<string, unknown>;
createdAt: Date;
}
export interface SnapshotStore {
save(snapshot: Snapshot): Promise<void>;
getLatest(aggregateId: string): Promise<Snapshot | null>;
}
export class InMemorySnapshotStore implements SnapshotStore {
private snapshots = new Map<string, Snapshot>();
async save(snapshot: Snapshot): Promise<void> {
this.snapshots.set(snapshot.aggregateId, snapshot);
}
async getLatest(aggregateId: string): Promise<Snapshot | null> {
return this.snapshots.get(aggregateId) || null;
}
}
// repository/order.repository.ts
export class EventSourcedOrderRepository {
private static readonly SNAPSHOT_INTERVAL = 50;
constructor(
private readonly eventStore: EventStore,
private readonly snapshotStore: SnapshotStore
) {}
async getById(orderId: string): Promise<Order> {
// 1. Look for the most recent snapshot
const snapshot = await this.snapshotStore.getLatest(orderId);
let events: DomainEvent[];
let order: Order;
if (snapshot) {
// 2a. Reconstruct from snapshot + subsequent events
order = Order.fromSnapshot(snapshot.state);
events = await this.eventStore.getEventsFromVersion(
orderId,
snapshot.version
);
} else {
// 2b. Reconstruct from all events
events = await this.eventStore.getEvents(orderId);
order = Order.fromHistory(events);
return order;
}
// 3. Apply missing events
for (const event of events) {
order.applyFromHistory(event);
}
return order;
}
async save(order: Order): Promise<void> {
const uncommitted = order.pullUncommittedEvents();
const expectedVersion = order.version - uncommitted.length;
await this.eventStore.appendEvents(
order.id,
uncommitted,
expectedVersion
);
// Create snapshot if needed
if (order.version % EventSourcedOrderRepository.SNAPSHOT_INTERVAL === 0) {
await this.snapshotStore.save({
aggregateId: order.id,
version: order.version,
state: order.toSnapshot(),
createdAt: new Date(),
});
}
}
}
Complete Flow: From Command to Query
Let us trace the complete flow of an operation in the CQRS + Event Sourcing system, from receiving the command to querying the updated read model.
// app/order-service.ts
export class OrderService {
constructor(
private readonly commandBus: CommandBus,
private readonly queryBus: QueryBus
) {}
// Write side: dispatch a command
async createOrder(
customerId: string,
items: { productId: string; productName: string;
quantity: number; unitPrice: number }[]
): Promise<string> {
const orderId = crypto.randomUUID();
await this.commandBus.dispatch(
new CreateOrderCommand(orderId, customerId, items)
);
return orderId;
}
async confirmOrder(orderId: string): Promise<void> {
await this.commandBus.dispatch(
new ConfirmOrderCommand(orderId)
);
}
// Read side: execute a query
async getOrderSummary(orderId: string): Promise<OrderSummaryReadModel> {
return this.queryBus.dispatch<OrderSummaryReadModel>(
new GetOrderSummaryQuery(orderId)
);
}
async getCustomerOrders(
customerId: string
): Promise<OrderSummaryReadModel[]> {
return this.queryBus.dispatch<OrderSummaryReadModel[]>(
new GetCustomerOrdersQuery(customerId)
);
}
}
// Initial configuration
function bootstrap(): OrderService {
// Event Store
const eventStore = new InMemoryEventStore();
const snapshotStore = new InMemorySnapshotStore();
const orderRepo = new EventSourcedOrderRepository(eventStore, snapshotStore);
// Projections
const orderSummaryProjection = new OrderSummaryProjection(eventStore);
const readDb = new InMemoryReadDatabase(orderSummaryProjection);
// Command Bus
const commandBus = new CommandBus();
commandBus.register('CreateOrder', new CreateOrderHandler(eventStore));
commandBus.register('ConfirmOrder', new ConfirmOrderHandler(orderRepo));
// Query Bus
const queryBus = new QueryBus();
queryBus.register('GetOrderSummary', new GetOrderSummaryHandler(readDb));
queryBus.register('GetCustomerOrders', new GetCustomerOrdersHandler(readDb));
return new OrderService(commandBus, queryBus);
}
CQRS Without Event Sourcing
It is important to emphasize that CQRS and Event Sourcing are independent patterns. You can adopt CQRS without Event Sourcing by maintaining a traditional database for writes and creating separate read models through synchronous or asynchronous projections.
| Approach | Complexity | Benefits | Limitations |
|---|---|---|---|
| Simple CQRS | Medium | Optimized read models, clear separation | No native audit trail, eventual consistency |
| CQRS + Event Sourcing | High | Audit trail, state reconstruction, multiple projections | Management complexity, learning curve |
| Traditional architecture | Low | Simple, familiar, mature tooling | Read and write on the same model, less flexibility |
When to Use CQRS and Event Sourcing
Ideal Scenarios
- Audit trail requirements: regulated industries (finance, healthcare, compliance)
- Asymmetric read/write: far more reads than writes (or vice versa)
- Multiple projections: data that needs to be read in very different formats
- Collaborative systems: multiple users modifying the same data concurrently
- Event-driven architecture: integration with systems that communicate through events
When to Avoid
- Simple CRUD applications: the added complexity does not justify the benefits
- Small teams without experience: the learning curve is significant
- Strong consistency requirements: CQRS introduces eventual consistency between read and write
- Prototypes and MVPs: initial development speed matters more than scalability
Challenges and Considerations
| Challenge | Description | Mitigation Strategy |
|---|---|---|
| Eventual Consistency | The read model may not immediately reflect the latest write | Optimistic UI, polling, real-time notifications |
| Event evolution | Event structure changes over time | Event versioning, upcasting, schema registry |
| Projection rebuilding | Reprocessing millions of events can take time | Snapshots, parallel processing, incremental projections |
| Operational complexity | More components to monitor and maintain | Observability, structured logging, health checks |
Conclusion
CQRS and Event Sourcing are powerful tools that solve specific problems of scale, auditing, and flexibility. CQRS separates read and write models, allowing each to be optimized independently. Event Sourcing transforms persistence into an immutable record of facts, unlocking capabilities such as time-travel debugging, retroactive projections, and native audit trails.
Like any advanced pattern, they require careful evaluation of the context. They are not the solution to every architectural problem, but in the right scenarios -- complex systems, event-driven domains, compliance requirements -- they offer advantages that traditional architectures cannot match. The key is to start with CQRS in its simple form and introduce Event Sourcing only where the added value is clear and measurable.
Related deep dive: 2 - Domain-Driven Design per Applicazioni Web.
Related deep dive: 4 - Architettura Esagonale (Ports and Adapters).
Frequently asked questions
What is the main difference between a Command and a Query in CQRS?
A Command represents an intent to change the system's state, while a Query represents a request for data without modifying the system state.
How does Event Sourcing differ from traditional persistence approaches?
Event Sourcing saves every event that led to the current state of an entity instead of saving the current state itself.
What is the purpose of a Projection in CQRS + Event Sourcing?
A Projection transforms events into Read Models optimized for reading, designed to answer queries as efficiently as possible.
Can CQRS be used without Event Sourcing?
Yes, CQRS can be adopted independently by maintaining a traditional database for writes and creating separate read models through synchronous or asynchronous projections.
Commenti
Caricamento commenti...
Sign in to comment