CQRS e Event Sourcing con TypeScript
CQRS (Command Query Responsibility Segregation) e Event Sourcing sono due pattern architetturali che, combinati, offrono un modo radicalmente diverso di progettare sistemi software. Invece di leggere e scrivere sullo stesso modello dati, CQRS separa le operazioni di scrittura (comandi) da quelle di lettura (query). Event Sourcing, a sua volta, sostituisce la persistenza tradizionale dello stato con una sequenza ordinata di eventi immutabili che descrivono tutto ciò che è accaduto nel sistema.
In questo articolo esploreremo entrambi i pattern in profondità, con implementazioni concrete in TypeScript. Useremo un sistema di gestione ordini come esempio per illustrare comandi, query, event store, proiezioni e snapshot.
Cosa Imparerai
- Il pattern CQRS: separazione tra comandi e query
- Command Bus e Query Bus in TypeScript
- Event Sourcing: persistenza basata su eventi
- Implementazione di un Event Store
- Proiezioni e Read Models
- Snapshot per ottimizzare le performance
- Quando usare (e quando evitare) CQRS e Event Sourcing
CQRS: Separazione tra Comandi e Query
Il principio alla base di CQRS è semplice: il modello usato per scrivere dati non deve essere lo stesso usato per leggerli. In un'architettura tradizionale, un unico modello (e spesso un'unica tabella) gestisce sia le operazioni di scrittura sia quelle di lettura. CQRS introduce due modelli distinti, ciascuno ottimizzato per il proprio scopo.
Comandi
Un Command rappresenta l'intenzione di modificare lo stato del sistema. È un oggetto immutabile che contiene tutti i dati necessari per eseguire l'operazione. I comandi non restituiscono dati: il loro unico effetto è la modifica dello stato.
// 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> {
// Verifica che l'ordine non esista gia
const existing = await this.eventStore.getEvents(command.orderId);
if (existing.length > 0) {
throw new Error('Ordine già esistente');
}
// Crea l'aggregate e genera gli eventi
const order = Order.create(
command.orderId,
command.customerId,
command.items
);
// Salva gli eventi generati
const events = order.pullUncommittedEvents();
await this.eventStore.appendEvents(command.orderId, events, 0);
}
}
Query
Una Query rappresenta una richiesta di dati. Non modifica lo stato del sistema e restituisce un risultato ottimizzato per il caso d'uso specifico. Il modello di lettura (Read Model) può essere denormalizzato, precompilato e strutturato esattamente come serve alla UI.
// 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('Ordine non trovato');
}
return summary;
}
}
Command Bus e Query Bus
Il Bus è il meccanismo che collega comandi/query ai rispettivi handler. Funge da dispatcher centralizzato, disaccoppiando chi invia il comando da chi lo gestisce.
// 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 già registrato per: ${commandType}`);
}
this.handlers.set(commandType, handler);
}
async dispatch(command: Command): Promise<void> {
const handler = this.handlers.get(command.type);
if (!handler) {
throw new Error(`Nessun handler per il comando: ${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(`Nessun handler per la query: ${query.type}`);
}
return handler.execute(query);
}
}
Event Sourcing: Lo Stato come Sequenza di Eventi
L'Event Sourcing capovolge il modo tradizionale di pensare alla persistenza. Invece di salvare lo stato corrente di un'entità, salviamo ogni evento che ha portato a quello stato. Lo stato attuale si ottiene riproducendo (replay) tutti gli eventi in ordine cronologico.
Vantaggi dell'Event Sourcing
| Vantaggio | Descrizione |
|---|---|
| Audit trail completo | Ogni cambiamento è registrato; si può ricostruire lo stato a qualsiasi punto nel tempo |
| Debugging avanzato | Riproducendo gli eventi si può capire esattamente come si è arrivati allo stato attuale |
| Proiezioni multiple | Dagli stessi eventi si possono costruire read model diversi per esigenze diverse |
| Evoluzione temporale | Nuove proiezioni possono essere create retroattivamente, riprocessando gli eventi storici |
| Nessuna perdita di dati | Gli eventi sono immutabili e append-only; non si cancella mai nulla |
Domain Events per 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;
}
) {}
}
Aggregate con Event Sourcing
In un sistema Event Sourced, l'aggregate non mantiene lo stato direttamente ma lo ricostruisce applicando gli eventi. Ogni operazione di business genera nuovi eventi che vengono accumulati e poi persistiti nell'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: crea un nuovo ordine
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;
}
// Ricostruzione da eventi storici
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('Solo ordini in stato CREATED possono essere confermati');
}
if (this._items.length === 0) {
throw new Error('Impossibile confermare un ordine vuoto');
}
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('Impossibile cancellare un ordine già spedito');
}
this.apply(new OrderCancelledEvent(this._id, this._version + 1, {
reason,
cancelledAt: new Date().toISOString(),
}));
}
// Applica un nuovo evento (genera uncommitted event)
private apply(event: DomainEvent): void {
this.mutate(event);
this._uncommittedEvents.push(event);
}
// Applica un evento storico (senza generare uncommitted)
private applyFromHistory(event: DomainEvent): void {
this.mutate(event);
}
// Modifica lo stato interno in base all'evento
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
L'Event Store è il database degli eventi. Ogni stream di eventi corrisponde a un aggregate. Le operazioni fondamentali sono due: appendere nuovi eventi e leggere gli eventi di uno stream. L'Event Store garantisce l'ordinamento e supporta l'optimistic concurrency tramite il 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(
`Conflitto di concorrenza: versione attesa ${expectedVersion}, ` +
`versione corrente ${currentVersion}`
);
}
const updatedStream = [...currentEvents, ...events];
this.streams.set(streamId, updatedStream);
this.allEvents.push(...events);
// Notifica i subscriber (per le proiezioni)
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);
}
}
Proiezioni e Read Models
Le proiezioni sono il meccanismo che trasforma gli eventi in Read Models ottimizzati per la lettura. Ogni proiezione ascolta specifici tipi di eventi e aggiorna un modello di lettura denormalizzato. Questo modello è pensato per rispondere alle query nel modo più efficiente possibile.
// 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) {
// Sottoscrizione agli eventi rilevanti
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);
}
}
Snapshot: Ottimizzare le Performance
Con il passare del tempo, il numero di eventi per un aggregate può crescere significativamente. Ricostruire lo stato da migliaia di eventi diventa costoso. Gli snapshot risolvono questo problema salvando periodicamente una fotografia dello stato dell'aggregate. La ricostruzione parte dallo snapshot più recente e applica solo gli eventi successivi.
// 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. Cerca lo snapshot più recente
const snapshot = await this.snapshotStore.getLatest(orderId);
let events: DomainEvent[];
let order: Order;
if (snapshot) {
// 2a. Ricostruisci dallo snapshot + eventi successivi
order = Order.fromSnapshot(snapshot.state);
events = await this.eventStore.getEventsFromVersion(
orderId,
snapshot.version
);
} else {
// 2b. Ricostruisci da tutti gli eventi
events = await this.eventStore.getEvents(orderId);
order = Order.fromHistory(events);
return order;
}
// 3. Applica gli eventi mancanti
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
);
// Crea snapshot se necessario
if (order.version % EventSourcedOrderRepository.SNAPSHOT_INTERVAL === 0) {
await this.snapshotStore.save({
aggregateId: order.id,
version: order.version,
state: order.toSnapshot(),
createdAt: new Date(),
});
}
}
}
Flusso Completo: Dal Comando alla Query
Vediamo il flusso completo di un'operazione nel sistema CQRS + Event Sourcing, dalla ricezione del comando fino alla query sul read model aggiornato.
// app/order-service.ts
export class OrderService {
constructor(
private readonly commandBus: CommandBus,
private readonly queryBus: QueryBus
) {}
// Lato scrittura: invia un comando
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)
);
}
// Lato lettura: esegui una 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)
);
}
}
// Configurazione iniziale
function bootstrap(): OrderService {
// Event Store
const eventStore = new InMemoryEventStore();
const snapshotStore = new InMemorySnapshotStore();
const orderRepo = new EventSourcedOrderRepository(eventStore, snapshotStore);
// Proiezioni
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 senza Event Sourcing
E' importante sottolineare che CQRS e Event Sourcing sono pattern indipendenti. Si può adottare CQRS senza Event Sourcing, mantenendo un database tradizionale per la scrittura e creando read model separati tramite proiezioni sincrone o asincrone.
| Approccio | Complessità | Vantaggi | Limitazioni |
|---|---|---|---|
| CQRS semplice | Media | Read model ottimizzati, separazione chiara | Nessun audit trail nativo, eventuale consistency |
| CQRS + Event Sourcing | Alta | Audit trail, ricostruzione stato, proiezioni multiple | Complessità gestionale, curva di apprendimento |
| Architettura tradizionale | Bassa | Semplice, familiare, tooling maturo | Read e write sullo stesso modello, meno flessibilità |
Quando Usare CQRS e Event Sourcing
Scenari Ideali
- Requisiti di audit trail: settori regolamentati (finanza, sanita, compliance)
- Read e write asimmetrici: molte più letture che scritture (o viceversa)
- Proiezioni multiple: dati che devono essere letti in formati molto diversi
- Sistemi collaborativi: più utenti modificano gli stessi dati concorrentemente
- Event-driven architecture: integrazione con sistemi che comunicano tramite eventi
Quando Evitare
- Applicazioni CRUD semplici: la complessità aggiunta non giustifica i benefici
- Team piccoli senza esperienza: la curva di apprendimento e significativa
- Requisiti di strong consistency: CQRS introduce eventual consistency tra read e write
- Prototipi e MVP: la velocità di sviluppo iniziale è più importante della scalabilità
Sfide e Considerazioni
| Sfida | Descrizione | Strategia di Mitigazione |
|---|---|---|
| Eventual Consistency | Il read model potrebbe non riflettere immediatamente l'ultima scrittura | UI ottimistica, polling, notifiche in tempo reale |
| Evoluzione degli eventi | La struttura degli eventi cambia nel tempo | Versioning degli eventi, upcasting, schema registry |
| Ricostruzione proiezioni | Riprocessare milioni di eventi può richiedere tempo | Snapshot, processing parallelo, proiezioni incrementali |
| Complessità operativa | Più componenti da monitorare e mantenere | Observability, logging strutturato, health check |
Conclusione
CQRS e Event Sourcing sono strumenti potenti che risolvono problemi specifici di scala, audit e flessibilità. CQRS separa i modelli di lettura e scrittura, consentendo di ottimizzare ciascuno indipendentemente. Event Sourcing trasforma la persistenza in un registro immutabile di fatti, aprendo possibilità come il time-travel debugging, le proiezioni retroattive e l'audit trail nativo.
Come ogni pattern avanzato, richiedono una valutazione attenta del contesto. Non sono la soluzione a ogni problema di architettura, ma nei scenari giusti - sistemi complessi, domini event-driven, requisiti di compliance - offrono vantaggi che le architetture tradizionali non possono eguagliare. La chiave è iniziare con CQRS in forma semplice e introdurre Event Sourcing solo dove il valore aggiunto è chiaro e misurabile.
Domande frequenti
Qual è la principale differenza tra un modello di lettura (Read Model) e un modello di scrittura (Write Model) in CQRS?
Il modello di lettura (Read Model) è ottimizzato per le query e può essere denormalizzato, precompilato e strutturato esattamente come serve alla UI, mentre il modello di scrittura (Write Model) rappresenta l'intenzione di modificare lo stato del sistema.
Come funziona l'Event Store in un sistema Event Sourced?
L'Event Store è il database degli eventi e ogni stream di eventi corrisponde a un aggregate. Le operazioni fondamentali sono due: appendere nuovi eventi e leggere gli eventi di uno stream.
Qual è lo scopo delle proiezioni in un sistema Event Sourced?
Lo scopo delle proiezioni è trasformare gli eventi in Read Models ottimizzati per la lettura, aggiornando un modello di lettura denormalizzato e pensato per rispondere alle query nel modo più efficiente possibile.
Perché è importante utilizzare snapshot in un sistema Event Sourced?
Gli snapshot sono importanti perché salvano periodicamente una fotografia dello stato dell'aggregate, permettendo di ricostruire lo stato da migliaia di eventi e risolvendo il problema della crescita significativa del numero di eventi.
Commenti
Caricamento commenti...
Accedi per commentare