Event Sourcing en la práctica: patrones y trampas
Guía práctica de event sourcing: cuándo usarlo, cómo implementar event stores, construir proyecciones, manejar esquemas y evitar errores.
El event sourcing almacena el estado como una secuencia de eventos en lugar de un estado actual mutable. En vez de actualizar una fila en la base de datos, añades un evento que describe lo que ocurrió. El estado actual se deriva reproduciendo los eventos. Esto te da una auditoría completa, la capacidad de reconstruir el estado en cualquier momento y consultas temporales que el CRUD tradicional no puede responder.
También introduce una complejidad considerable. El event sourcing no es adecuado para todos los dominios. Esta guía cubre cuándo funciona, cómo implementarlo y los errores que suelen sorprender a los equipos.
Cuándo tiene sentido el event sourcing
El event sourcing brilla en dominios donde el historial de cambios es tan valioso como el estado actual.
Good fits:
- Financial systems — every transaction must be auditable
- Inventory management — track every stock movement
- Booking/reservation systems — cancellations and modifications matter
- Collaborative editing — track every change by every user
- Compliance-heavy domains — regulatory audit trails required
Poor fits:
- Simple CRUD applications — blog posts, user profiles
- High-throughput writes with no audit needs — telemetry data
- Systems where current state is all that matters
La prueba: si tus interesados de negocio verían valor en responder «¿cuál era el estado de X en el momento T?» o «¿qué secuencia de acciones llevó a este estado?», vale la pena considerar el event sourcing.
El event store
Un event store es un registro de solo anexado donde cada stream representa el historial de una entidad. Los eventos son inmutables: una vez escritos, no cambian.
interface DomainEvent {
eventId: string;
streamId: string;
eventType: string;
data: Record<string, unknown>;
metadata: {
timestamp: string;
userId: string;
correlationId: string;
};
version: number;
}
// ❌ Mutable state — no history, no audit trail
await db.orders.update(
{ _id: orderId },
{ $set: { status: 'shipped', shippedAt: new Date() } }
);
// Previous state is gone forever// ✅ Event sourcing — append events, derive state
const events: DomainEvent[] = [
{
eventId: 'evt-001',
streamId: 'order-123',
eventType: 'OrderPlaced',
data: {
customerId: 'cust-456',
items: [{ productId: 'prod-1', quantity: 2, price: 29.99 }],
total: 59.98,
},
metadata: { timestamp: '2021-08-01T10:00:00Z', userId: 'cust-456', correlationId: 'req-789' },
version: 1,
},
{
eventId: 'evt-002',
streamId: 'order-123',
eventType: 'PaymentProcessed',
data: { paymentId: 'pay-001', amount: 59.98, method: 'card' },
metadata: { timestamp: '2021-08-01T10:00:05Z', userId: 'system', correlationId: 'req-789' },
version: 2,
},
{
eventId: 'evt-003',
streamId: 'order-123',
eventType: 'OrderShipped',
data: { trackingNumber: '1Z999AA10123456784', carrier: 'UPS' },
metadata: { timestamp: '2021-08-01T14:30:00Z', userId: 'staff-001', correlationId: 'req-912' },
version: 3,
},
];Cada evento captura qué ocurrió, cuándo y quién lo desencadenó. El campo version permite la concurrencia optimista: se detectan dos escrituras simultáneas en el mismo stream y se rechaza una.
Reconstruyendo el estado a partir de eventos
Un agregado carga su stream de eventos y aplica cada evento para reconstruir su estado actual.
interface OrderState {
id: string;
status: 'placed' | 'paid' | 'shipped' | 'delivered' | 'cancelled';
items: Array<{ productId: string; quantity: number; price: number }>;
total: number;
trackingNumber?: string;
}
function applyEvent(state: OrderState, event: DomainEvent): OrderState {
switch (event.eventType) {
case 'OrderPlaced':
return {
id: event.streamId,
status: 'placed',
items: event.data.items as OrderState['items'],
total: event.data.total as number,
};
case 'PaymentProcessed':
return { ...state, status: 'paid' };
case 'OrderShipped':
return {
...state,
status: 'shipped',
trackingNumber: event.data.trackingNumber as string,
};
case 'OrderDelivered':
return { ...state, status: 'delivered' };
case 'OrderCancelled':
return { ...state, status: 'cancelled' };
default:
return state;
}
}
function rehydrate(events: DomainEvent[]): OrderState {
return events.reduce(
(state, event) => applyEvent(state, event),
{} as OrderState
);
}
// Usage:
const orderEvents = await eventStore.getStream('order-123');
const currentState = rehydrate(orderEvents);
// { id: 'order-123', status: 'shipped', total: 59.98, trackingNumber: '1Z...' }La función applyEvent es una función pura: dados los mismos eventos, siempre produce el mismo estado. Esto facilita probarla y razonar sobre ella.
Proyecciones: modelos de lectura
Reproducir eventos en cada lectura es costoso. Las proyecciones construyen modelos de lectura optimizados suscribiéndose a eventos y manteniendo vistas desnormalizadas.
// Projection: Order summary view (optimized for listing orders)
interface OrderSummary {
orderId: string;
customerName: string;
total: number;
status: string;
itemCount: number;
lastUpdated: string;
}
class OrderSummaryProjection {
constructor(private db: Database) {}
async handle(event: DomainEvent): Promise<void> {
switch (event.eventType) {
case 'OrderPlaced':
await this.db.orderSummaries.insert({
orderId: event.streamId,
customerName: event.data.customerName,
total: event.data.total,
status: 'placed',
itemCount: (event.data.items as unknown[]).length,
lastUpdated: event.metadata.timestamp,
});
break;
case 'OrderShipped':
await this.db.orderSummaries.update(
{ orderId: event.streamId },
{
$set: {
status: 'shipped',
lastUpdated: event.metadata.timestamp,
},
}
);
break;
case 'OrderCancelled':
await this.db.orderSummaries.update(
{ orderId: event.streamId },
{
$set: {
status: 'cancelled',
lastUpdated: event.metadata.timestamp,
},
}
);
break;
}
}
}
// Multiple projections from the same events:
// - OrderSummaryProjection → order list page
// - CustomerOrderHistoryProjection → customer profile page
// - RevenueReportProjection → analytics dashboardLas proyecciones pueden reconstruirse desde cero reproduciendo todos los eventos. Esto significa que puedes añadir nuevos modelos de lectura de forma retroactiva: creas una proyección que cuente pedidos por región, reproduces los eventos desde el principio y la nueva vista se llena inmediatamente con datos históricos.
Evolución de esquemas
Los eventos son inmutables, pero tu modelo de dominio evoluciona. Las nuevas versiones de eventos deben coexistir con las antiguas.
// ❌ Breaking change — old events in the store can't be read
interface OrderPlacedV2 {
eventType: 'OrderPlaced';
data: {
customerId: string;
shippingAddress: Address; // New required field
items: OrderItem[];
total: number;
};
}
// ✅ Upcasting — transform old events to the current schema
function upcast(event: DomainEvent): DomainEvent {
if (event.eventType === 'OrderPlaced' && !event.data.shippingAddress) {
return {
...event,
data: {
...event.data,
shippingAddress: {
street: 'Unknown',
city: 'Unknown',
country: 'Unknown',
},
},
};
}
return event;
}
// Apply upcasting when reading events
async function getStream(streamId: string): Promise<DomainEvent[]> {
const rawEvents = await eventStore.read(streamId);
return rawEvents.map(upcast);
}El upcasting transforma las formas antiguas de los eventos en la forma esperada actual en el momento de lectura. Los eventos almacenados permanecen inmutables: la transformación se aplica en memoria al cargar el stream.
Snapshots para streams largos
Los streams con miles de eventos se vuelven costosos de reproducir. Los snapshots almacenan un punto de control del estado actual en una versión específica.
interface Snapshot<T> {
streamId: string;
state: T;
version: number; // The event version this snapshot reflects
createdAt: string;
}
async function loadAggregate(streamId: string): Promise<OrderState> {
// Try loading from snapshot first
const snapshot = await snapshotStore.get<OrderState>(streamId);
let state: OrderState;
let fromVersion: number;
if (snapshot) {
state = snapshot.state;
fromVersion = snapshot.version + 1;
} else {
state = {} as OrderState;
fromVersion = 1;
}
// Load only events after the snapshot
const events = await eventStore.getStream(streamId, { fromVersion });
const currentState = events.reduce(
(s, event) => applyEvent(s, event),
state
);
// Save new snapshot if enough events accumulated
const totalEvents = (snapshot?.version ?? 0) + events.length;
if (events.length > 100) {
await snapshotStore.save({
streamId,
state: currentState,
version: totalEvents,
createdAt: new Date().toISOString(),
});
}
return currentState;
}Los snapshots son una optimización, no un requisito. El sistema debe funcionar correctamente sin ellos: solo reducen el número de eventos reproducidos por lectura.
Puntos clave
- El event sourcing almacena lo que ocurrió, no lo que es el estado actual: el estado se deriva reproduciendo eventos
- Úsalo en dominios con mucha auditoría donde el historial, las consultas temporales y los registros de auditoría tienen valor de negocio
- Las proyecciones construyen modelos de lectura optimizados a partir de streams de eventos; puedes añadir nuevas proyecciones de forma retroactiva
- Maneja la evolución de esquemas mediante upcasting: transforma los eventos antiguos en la forma actual en el momento de lectura
- Los snapshots optimizan streams largos: guarda el estado periódicamente para evitar reproducir miles de eventos
- No uses event sourcing por defecto: el coste de complejidad es real; úsalo donde los beneficios lo justifiquen
