Event-Sourcing-Patterns für stark audit-relevante Anwendungen
Event Sourcing für audit-kritische Anwendungen: append-only Event Stores, Projections und Snapshot-Strategien zwischen Compliance und Performance.
Traditionelle CRUD-Anwendungen überschreiben den Zustand. Wenn jemand einen Datensatz aktualisiert, ist der vorherige Wert weg, es sei denn, du hast Audit-Logging nachträglich hinzugefügt. In Domänen, in denen jede Änderung nachvollziehbar sein muss—Finanzdienstleistungen, Gesundheitswesen, Legal Tech—entsteht dadurch ein grundsätzlicher Konflikt zwischen der Funktionsweise der Anwendung und den Anforderungen der Regulierenden.
Event Sourcing dreht das Datenmodell um: Anstatt den aktuellen Zustand zu speichern, speicherst du die Sequenz der Events, die diesen Zustand erzeugt hat. Der aktuelle Zustand wird zu einer Ableitung, und die komplette Historie existiert standardmäßig. Für stark audit-relevante Anwendungen ist das kein fancy Architekturpattern—es ist der natürliche Pass.
Events als Source of Truth
Bei Event Sourcing ist dein Datenspeicher ein append-only Log aus Domain Events. Jedes Event erfasst, was passiert ist, wann und die relevanten Details. Der aktuelle Zustand wird durch Replay dieser Events berechnet.
// Domain events for an account management system
interface DomainEvent {
eventId: string;
aggregateId: string;
eventType: string;
timestamp: Date;
version: number;
data: Record<string, unknown>;
metadata: {
userId: string;
correlationId: string;
causationId: string;
ipAddress: string;
};
}
// ❌ CRUD approach: audit is an afterthought
class AccountRepository {
async updateBalance(accountId: string, newBalance: number) {
// Previous balance is gone forever
await db.query(
'UPDATE accounts SET balance = $1 WHERE id = $2',
[newBalance, accountId]
);
// Audit log added later, often incomplete
await db.query(
'INSERT INTO audit_log (entity, action, timestamp) VALUES ($1, $2, NOW())',
[accountId, 'balance_updated']
);
}
}
// ✅ Event sourcing: audit is inherent
class AccountEventStore {
async append(event: DomainEvent): Promise<void> {
await db.query(
`INSERT INTO events (event_id, aggregate_id, event_type,
timestamp, version, data, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
[
event.eventId,
event.aggregateId,
event.eventType,
event.timestamp,
event.version,
JSON.stringify(event.data),
JSON.stringify(event.metadata),
]
);
}
async getEvents(aggregateId: string): Promise<DomainEvent[]> {
const result = await db.query(
`SELECT * FROM events
WHERE aggregate_id = $1
ORDER BY version ASC`,
[aggregateId]
);
return result.rows;
}
}Das Feld metadata ist kritisch für Audits. Jedes Event erfasst, wer es ausgelöst hat, die Correlation Chain, die zusammengehörige Operationen verknüpft, und Kontextinformationen wie IP-Adressen, die Auditor:innen brauchen.
Read Models mit Projections aufbauen
Das Speichern von Events löst das Audit-Problem, erzeugt aber ein Abfrageproblem: Du kannst nicht effizient „alle Konten mit einem Saldo über 10.000 $“ abfragen, indem du jedes Mal Millionen von Events replayst. Projections lösen das, indem sie denormalisierte Read Models pflegen.
// Projection that builds a read model from events
class AccountBalanceProjection {
async handle(event: DomainEvent): Promise<void> {
switch (event.eventType) {
case 'AccountOpened':
await db.query(
`INSERT INTO account_balances
(account_id, balance, owner_name, opened_at)
VALUES ($1, $2, $3, $4)`,
[
event.aggregateId,
event.data.initialDeposit,
event.data.ownerName,
event.timestamp,
]
);
break;
case 'FundsDeposited':
await db.query(
`UPDATE account_balances
SET balance = balance + $1, updated_at = $2
WHERE account_id = $3`,
[event.data.amount, event.timestamp, event.aggregateId]
);
break;
case 'FundsWithdrawn':
await db.query(
`UPDATE account_balances
SET balance = balance - $1, updated_at = $2
WHERE account_id = $3`,
[event.data.amount, event.timestamp, event.aggregateId]
);
break;
}
}
}
// Separate audit-specific projection
class AuditTrailProjection {
async handle(event: DomainEvent): Promise<void> {
await db.query(
`INSERT INTO audit_trail
(event_id, aggregate_id, event_type, actor_id,
ip_address, correlation_id, timestamp, details)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
[
event.eventId,
event.aggregateId,
event.eventType,
event.metadata.userId,
event.metadata.ipAddress,
event.metadata.correlationId,
event.timestamp,
JSON.stringify(event.data),
]
);
}
}Projections sind wegwerfbar. Wenn ein Auditor ein neues Berichtsformat braucht, baust du eine neue Projection und replayst Events durch sie. Die Quelldaten ändern sich nie.
Snapshot-Strategien für Performance
Tausende von Events für den Wiederaufbau eines Aggregats zu replayen, wird langsam. Snapshots erfassen periodisch den berechneten Zustand, sodass du nur Events nach dem Snapshot replayst.
class SnapshotStore {
private readonly SNAPSHOT_INTERVAL = 100;
async getAggregate(aggregateId: string): Promise<Account> {
const snapshot = await this.getLatestSnapshot(aggregateId);
const startVersion = snapshot ? snapshot.version + 1 : 0;
const events = await this.eventStore.getEventsAfterVersion(
aggregateId,
startVersion
);
let account = snapshot
? Account.fromSnapshot(snapshot.state)
: Account.empty(aggregateId);
for (const event of events) {
account = account.apply(event);
}
// Create snapshot if enough events have accumulated
if (events.length >= this.SNAPSHOT_INTERVAL) {
await this.saveSnapshot({
aggregateId,
version: account.version,
state: account.toSnapshot(),
createdAt: new Date(),
});
}
return account;
}
private async getLatestSnapshot(
aggregateId: string
): Promise<Snapshot | null> {
const result = await db.query(
`SELECT * FROM snapshots
WHERE aggregate_id = $1
ORDER BY version DESC LIMIT 1`,
[aggregateId]
);
return result.rows[0] || null;
}
private async saveSnapshot(snapshot: Snapshot): Promise<void> {
await db.query(
`INSERT INTO snapshots
(aggregate_id, version, state, created_at)
VALUES ($1, $2, $3, $4)`,
[
snapshot.aggregateId,
snapshot.version,
JSON.stringify(snapshot.state),
snapshot.createdAt,
]
);
}
}Snapshots sind Optimierung, keine Quelldaten. Du kannst jederzeit alle Snapshots löschen und aus Events neu aufbauen—das System bleibt korrekt, nur vorübergehend langsamer.
Umgang mit Event-Schema-Evolution
Events sind immutable, aber dein Domänenverständnis entwickelt sich weiter. Du musst mit Events umgehen, die mit einem älteren Schema geschrieben wurden, als dein aktueller Code erwartet.
// Event upcasting: transform old event shapes to new ones
class EventUpcaster {
private upcasters = new Map<string, UpcastFunction[]>();
register(eventType: string, fromVersion: number, upcaster: UpcastFunction) {
const key = `${eventType}:${fromVersion}`;
const chain = this.upcasters.get(key) || [];
chain.push(upcaster);
this.upcasters.set(key, chain);
}
upcast(event: StoredEvent): DomainEvent {
let current = event;
while (current.schemaVersion < this.getCurrentVersion(current.eventType)) {
const key = `${current.eventType}:${current.schemaVersion}`;
const upcaster = this.upcasters.get(key);
if (!upcaster) {
throw new Error(
`No upcaster for ${current.eventType} v${current.schemaVersion}`
);
}
current = upcaster[0](current);
}
return current as DomainEvent;
}
}
// Example: FundsDeposited event evolved over time
// v1: { amount: number }
// v2: { amount: number, currency: string }
// v3: { amount: number, currency: string, channel: string }
const upcaster = new EventUpcaster();
upcaster.register('FundsDeposited', 1, (event) => ({
...event,
data: { ...event.data, currency: 'USD' },
schemaVersion: 2,
}));
upcaster.register('FundsDeposited', 2, (event) => ({
...event,
data: { ...event.data, channel: 'unknown' },
schemaVersion: 3,
}));Compliance-Queries: der Audit-Vorteil
Mit Event Sourcing werden Compliance-Queries, die in einem CRUD-System komplexe Audit-Log-Joins erfordern würden, zu straightforwarden Event-Stream-Queries.
class ComplianceQueryService {
// "Show me every change to this account, by whom, and when"
async getAccountHistory(accountId: string): Promise<AuditEntry[]> {
const events = await this.eventStore.getEvents(accountId);
return events.map((event) => ({
timestamp: event.timestamp,
action: event.eventType,
actor: event.metadata.userId,
ipAddress: event.metadata.ipAddress,
details: event.data,
correlationId: event.metadata.correlationId,
}));
}
// "What did the account look like at this specific point in time?"
async getStateAtTime(
accountId: string,
pointInTime: Date
): Promise<Account> {
const events = await this.eventStore.getEventsUntil(
accountId,
pointInTime
);
let account = Account.empty(accountId);
for (const event of events) {
account = account.apply(event);
}
return account;
}
// "Who accessed accounts with balances over $50k last quarter?"
async getHighValueAccessPatterns(
threshold: number,
startDate: Date,
endDate: Date
): Promise<AccessPattern[]> {
// This query runs against a projection built
// specifically for compliance reporting
return db.query(
`SELECT actor_id, account_id, event_type, timestamp
FROM compliance_access_log
WHERE balance_at_time > $1
AND timestamp BETWEEN $2 AND $3
ORDER BY timestamp DESC`,
[threshold, startDate, endDate]
);
}
}Die „Zustand zu einem bestimmten Zeitpunkt“-Query ist das Killer-Feature für Audits. In einem CRUD-System erfordert der Wiederaufbau historischer Zustände umfangreiche Audit-Log-Korrelation. Bei Event Sourcing ist es das Replay von Events bis zu einem timestamp.
Wichtige Erkenntnisse
Event Sourcing macht Audit Trails zur natürlichen Konsequenz des Datenmodells und nicht zu einer nachträglich angeflanschten Idee: Jeder Zustandswechsel ist ein immutable Event, das nicht nur erfasst, was sich geändert hat, sondern auch wer es ausgelöst hat, wann und warum. Projections lösen das Abfrageproblem, indem sie wegwerfbare Read Models pflegen: Baue neue Projections, wann immer Auditor:innen eine andere Sicht brauchen, replay die Event-Historie durch sie und mache dir nie Sorgen um Datenverlust, weil die Quell-Events immutable sind. Snapshots sind eine Performance-Optimierung, mit der du das Replay alter Events überspringen kannst—sie sind wegwerfbare Zusammenfassungen, keine Quelldaten, und du solltest sie erst einführen, wenn die Replay-Performance tatsächlich ein Problem wird, nicht präventiv. Schema-Evolution durch Upcasting lässt dein Domänenmodell wachsen, ohne gespeicherte Events neu schreiben zu müssen: Transformiere alte Event-Shapes zur Lesezeit in aktuelle Schemas und behalte so die Abwärtskompatibilität mit deiner kompletten Event-Historie.
