Event Sourcing in der Praxis: Wann es hilft und wann es schmerzt

interface EventSourcingFitAssessment {
  domain: string;
  fit: "excellent" | "good" | "poor";
  reason: string;
  keyBenefit: string;
}
 
const fitAssessments: EventSourcingFitAssessment[] = [
  {
    domain: "Financial transactions",
    fit: "excellent",
    reason: "Regulatory requirements demand complete audit trail",
    keyBenefit: "Immutable ledger satisfies compliance out of the box",
  },
  {
    domain: "Order management",
    fit: "excellent",
    reason: "Order lifecycle has distinct, meaningful state transitions",
    keyBenefit: "Full order history enables dispute resolution",
  },
  {
    domain: "Collaborative editing",
    fit: "good",
    reason: "Changes must be mergeable and replayable",
    keyBenefit: "Conflict resolution through event replay",
  },
  {
    domain: "User profile CRUD",
    fit: "poor",
    reason: "Current state is all that matters, changes have no business value",
    keyBenefit: "None — adds complexity without value",
  },
  {
    domain: "Static content management",
    fit: "poor",
    reason: "Low write frequency, query-heavy, history rarely needed",
    keyBenefit: "None — a simple database is better",
  },
];

interface DomainEvent {
  eventId: string;
  streamId: string;
  eventType: string;
  data: Record<string, unknown>;
  metadata: {
    timestamp: Date;
    version: number;
    correlationId: string;
    causationId: string;
    userId?: string;
  };
}
 
interface EventStore {
  append(
    streamId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void>;
  readStream(
    streamId: string,
    fromVersion?: number
  ): Promise<DomainEvent[]>;
  readAll(
    fromPosition?: number,
    limit?: number
  ): Promise<DomainEvent[]>;
}
 
class PostgresEventStore implements EventStore {
  constructor(private readonly db: Database) {}
 
  async append(
    streamId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void> {
    await this.db.transaction(async (tx) => {
      // Optimistic concurrency check
      const currentVersion = await tx.queryOne<{ max_version: number }>(
        "SELECT COALESCE(MAX(version), 0) as max_version FROM events WHERE stream_id = $1",
        [streamId]
      );
 
      if (currentVersion.max_version !== expectedVersion) {
        throw new ConcurrencyError(
          `Expected version ${expectedVersion}, found ${currentVersion.max_version}`
        );
      }
 
      // Append events
      for (let i = 0; i < events.length; i++) {
        const event = events[i];
        const version = expectedVersion + i + 1;
 
        await tx.execute(
          `INSERT INTO events (event_id, stream_id, event_type, data, metadata, version)
           VALUES ($1, $2, $3, $4, $5, $6)`,
          [
            event.eventId,
            streamId,
            event.eventType,
            JSON.stringify(event.data),
            JSON.stringify({ ...event.metadata, version }),
            version,
          ]
        );
      }
    });
  }
 
  async readStream(
    streamId: string,
    fromVersion: number = 0
  ): Promise<DomainEvent[]> {
    const rows = await this.db.query(
      "SELECT * FROM events WHERE stream_id = $1 AND version > $2 ORDER BY version ASC",
      [streamId, fromVersion]
    );
 
    return rows.map(this.mapRowToEvent);
  }
 
  async readAll(fromPosition = 0, limit = 1000): Promise<DomainEvent[]> {
    const rows = await this.db.query(
      "SELECT * FROM events WHERE global_position > $1 ORDER BY global_position ASC LIMIT $2",
      [fromPosition, limit]
    );
 
    return rows.map(this.mapRowToEvent);
  }
 
  private mapRowToEvent(row: any): DomainEvent {
    return {
      eventId: row.event_id,
      streamId: row.stream_id,
      eventType: row.event_type,
      data: row.data,
      metadata: row.metadata,
    };
  }
}

// ❌ Mutable state without event history
class MutableBankAccount {
  balance: number = 0;
 
  deposit(amount: number): void {
    this.balance += amount;
    // How did we get here? Who deposited what and when? No idea.
  }
}
 
// ✅ Event-sourced aggregate
interface AccountEvent extends DomainEvent {
  eventType: "AccountOpened" | "MoneyDeposited" | "MoneyWithdrawn" | "AccountClosed";
}
 
class BankAccount {
  private balance = 0;
  private status: "active" | "closed" = "active";
  private version = 0;
  private uncommittedEvents: DomainEvent[] = [];
 
  static fromEvents(events: DomainEvent[]): BankAccount {
    const account = new BankAccount();
    for (const event of events) {
      account.apply(event, false);
    }
    return account;
  }
 
  deposit(amount: number, correlationId: string): void {
    if (this.status !== "active") {
      throw new Error("Cannot deposit to a closed account");
    }
    if (amount <= 0) {
      throw new Error("Deposit amount must be positive");
    }
 
    this.apply(
      this.createEvent("MoneyDeposited", { amount }),
      true
    );
  }
 
  withdraw(amount: number, correlationId: string): void {
    if (this.status !== "active") {
      throw new Error("Cannot withdraw from a closed account");
    }
    if (amount > this.balance) {
      throw new Error("Insufficient funds");
    }
 
    this.apply(
      this.createEvent("MoneyWithdrawn", { amount }),
      true
    );
  }
 
  private apply(event: DomainEvent, isNew: boolean): void {
    switch (event.eventType) {
      case "AccountOpened":
        this.status = "active";
        this.balance = 0;
        break;
      case "MoneyDeposited":
        this.balance += (event.data as { amount: number }).amount;
        break;
      case "MoneyWithdrawn":
        this.balance -= (event.data as { amount: number }).amount;
        break;
      case "AccountClosed":
        this.status = "closed";
        break;
    }
 
    this.version++;
    if (isNew) {
      this.uncommittedEvents.push(event);
    }
  }
 
  private createEvent(
    eventType: string,
    data: Record<string, unknown>
  ): DomainEvent {
    return {
      eventId: crypto.randomUUID(),
      streamId: "",
      eventType,
      data,
      metadata: {
        timestamp: new Date(),
        version: this.version + 1,
        correlationId: "",
        causationId: "",
      },
    };
  }
 
  getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents];
  }
 
  getBalance(): number {
    return this.balance;
  }
 
  getVersion(): number {
    return this.version;
  }
}

interface Projection {
  name: string;
  handle(event: DomainEvent): Promise<void>;
  rebuild(): Promise<void>;
}
 
class AccountBalanceProjection implements Projection {
  name = "account-balances";
 
  constructor(private readonly db: Database) {}
 
  async handle(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case "AccountOpened":
        await this.db.execute(
          "INSERT INTO account_balances (account_id, balance, last_updated) VALUES ($1, 0, $2)",
          [event.streamId, event.metadata.timestamp]
        );
        break;
      case "MoneyDeposited":
        await this.db.execute(
          "UPDATE account_balances SET balance = balance + $1, last_updated = $2 WHERE account_id = $3",
          [event.data.amount, event.metadata.timestamp, event.streamId]
        );
        break;
      case "MoneyWithdrawn":
        await this.db.execute(
          "UPDATE account_balances SET balance = balance - $1, last_updated = $2 WHERE account_id = $3",
          [event.data.amount, event.metadata.timestamp, event.streamId]
        );
        break;
    }
  }
 
  async rebuild(): Promise<void> {
    await this.db.execute("TRUNCATE account_balances");
    // Replay all events from the beginning
  }
}

interface EventSourcingPitfall {
  mistake: string;
  consequence: string;
  prevention: string;
}
 
const pitfalls: EventSourcingPitfall[] = [
  {
    mistake: "Not versioning event schemas",
    consequence: "Old events break when deserialized with new code",
    prevention: "Include schema version, use upcasters to transform old events",
  },
  {
    mistake: "Storing too much data in events",
    consequence: "Event store grows uncontrollably, replay becomes slow",
    prevention: "Store meaningful domain facts, not raw request payloads",
  },
  {
    mistake: "No snapshotting for long-lived aggregates",
    consequence: "Loading an aggregate with 10000 events takes seconds",
    prevention: "Snapshot every N events and replay only from the last snapshot",
  },
  {
    mistake: "Coupling projections to event structure",
    consequence: "Every event schema change requires updating all projections",
    prevention: "Use anti-corruption layers between events and projections",
  },
];