Event Sourcing in der Praxis: Muster und Fallstricke
Praktischer Leitfaden zu Event Sourcing: wann einsetzen, Event Stores implementieren, Projektionen bauen, Schema-Evolution und typische Fehler.
Event Sourcing speichert den Zustand als Sequenz von Ereignissen, nicht als veränderbaren aktuellen Zustand. Statt eine Zeile in der Datenbank zu aktualisieren, hängst du ein Ereignis an, das beschreibt, was passiert ist. Der aktuelle Zustand wird durch Wiedergabe der Ereignisse abgeleitet. Das liefert eine lückenlose Audit-Trail, die Möglichkeit, den Zustand zu jedem Zeitpunkt wiederherzustellen, und zeitliche Abfragen, die traditionelles CRUD nicht beantworten kann.
Es führt aber auch erhebliche Komplexität ein. Event Sourcing ist nicht für jede Domain geeignet. Dieser Leitfaden behandelt, wann es funktioniert, wie man es implementiert und die Fallstricke, die Teams überraschen.
Wann Event Sourcing Sinn ergibt
Event Sourcing glänzt in Domains, in denen die Historie der Änderungen so wertvoll ist wie der aktuelle Zustand.
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
Der Test: Wenn deine Stakeholder Wert darin sehen, die Fragen »Wie war der Zustand von X zum Zeitpunkt T?« oder »Welche Abfolge von Aktionen hat zu diesem Zustand geführt?« zu beantworten, lohnt sich Event Sourcing.
Der Event Store
Ein Event Store ist ein append-only Log, bei dem jeder Stream die Historie einer Entität repräsentiert. Ereignisse sind unveränderlich — einmal geschrieben, ändern sie sich nie.
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,
},
];Jedes Ereignis erfasst, was passiert ist, wann und von wem es ausgelöst wurde. Das Feld version ermöglicht optimistische Nebenläufigkeit: Zwei gleichzeitige Schreibzugriffe auf denselben Stream werden erkannt und einer wird abgelehnt.
Zustand aus Ereignissen wiederherstellen
Ein Aggregate lädt seinen Event-Stream und wendet jedes Ereignis an, um seinen aktuellen Zustand wiederherzustellen.
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...' }Die Funktion applyEvent ist eine reine Funktion: Bei denselben Ereignissen produziert sie immer denselben Zustand. Das macht sie einfach zu testen und nachzuvollziehen.
Projektionen: Lesemodelle
Ereignisse bei jedem Lesevorgang wiederzugeben, ist teuer. Projektionen bauen optimierte Lesemodelle auf, indem sie Ereignisse abonnieren und denormalisierte Views pflegen.
// 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 dashboardProjektionen können von Grund auf neu durch Wiedergabe aller Ereignisse aufgebaut werden. Das bedeutet, du kannst neue Lesemodelle nachträglich hinzufügen: Erstelle eine Projektion, die Bestellungen nach Region zählt, spiele die Ereignisse vom Anfang an durch und die neue View ist sofort mit historischen Daten gefüllt.
Schema-Evolution
Ereignisse sind unveränderlich, aber dein Domain-Modell entwickelt sich weiter. Neue Ereignisversionen müssen neben alten koexistieren.
// ❌ 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);
}Upcasting transformiert alte Ereignisformen zur aktuell erwarteten Form zum Zeitpunkt des Lesens. Die gespeicherten Ereignisse bleiben unverändert — die Transformation wird beim Laden des Streams im Arbeitsspeicher angewendet.
Snapshots für lange Streams
Streams mit Tausenden von Ereignissen werden teuer wiederzugeben. Snapshots speichern einen Checkpoint des aktuellen Zustands zu einer bestimmten Version.
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;
}Snapshots sind eine Optimierung, kein Muss. Das System muss auch ohne sie korrekt funktionieren — sie reduzieren nur die Anzahl der Ereignisse, die pro Lesevorgang wiedergegeben werden.
Wichtige Erkenntnisse
- Event Sourcing speichert, was passiert ist, nicht, was der aktuelle Zustand ist — der Zustand wird durch Wiedergabe der Ereignisse abgeleitet
- Nutze es für audit-relevante Domains, in denen Historie, zeitliche Abfragen und Audit-Trails Geschäftswert haben
- Projektionen bauen optimierte Lesemodelle aus Event-Streams auf — neue Projektionen kannst du nachträglich hinzufügen
- Handle Schema-Evolution durch Upcasting — transformiere alte Ereignisse zum aktuellen Shape zum Zeitpunkt des Lesens
- Snapshots optimieren lange Streams — speichere den Zustand regelmäßig als Checkpoint, um Tausende von Ereignissen wiederzugeben zu vermeiden
- Defaulte nicht auf Event Sourcing — die Komplexitätskosten sind real, nutze es nur, wo die Vorteile sie rechtfertigen
