Event Sourcing in PHP: Grundlagen ohne Framework

1. Zustand als Sequenz von Events statt aktuellem Snapshot

Der klassische CRUD-Ansatz speichert für ein Bankkonto schlicht den aktuellen Kontostand in einer Spalte und überschreibt diesen Wert bei jeder Ein- oder Auszahlung per UPDATE. Nach dem UPDATE ist nicht mehr nachvollziehbar, wie der Kontostand zustande kam, wie viele einzelne Buchungen dahinterstecken oder wann genau welche Änderung passierte, sofern das nicht mühsam in einer separaten Tabelle mitprotokolliert wird.

Event Sourcing löst genau dieses Problem, indem statt des aktuellen Kontostands eine Sequenz einzelner Events gespeichert wird: KontoEröffnet, GeldEingezahlt, GeldAbgehoben. Der aktuelle Kontostand ergibt sich dann jederzeit daraus, alle diese Events der Reihe nach durchzugehen und ihre jeweilige Wirkung aufzusummieren, statt ihn als eigenständigen Wert zu pflegen. Der große Vorteil ist ein von Natur aus vorhandener, lückenloser Audit-Trail sowie die Möglichkeit, den historischen Zustand zu jedem beliebigen Zeitpunkt der Vergangenheit exakt zu rekonstruieren, was beim Debugging eines real aufgetretenen Produktionsfehlers oft den entscheidenden Unterschied macht.

2. Einen minimalen Event Store selbst implementieren

Im Kern besteht ein Event Store aus einer Append-only-Tabelle mit den Spalten Aggregat-ID, Versionsnummer, Event-Typ, Nutzdaten als serialisiertes JSON und Zeitstempel. Events werden ausschließlich angehängt, niemals verändert oder gelöscht, was die Grundvoraussetzung für einen verlässlichen Audit-Trail ist.

Die zentrale Methode append() nimmt eine erwartete Versionsnummer entgegen und prüft implizit über einen eindeutigen Datenbank-Constraint auf Aggregat-ID und Version, ob zwischenzeitlich bereits ein anderer Prozess Events für dasselbe Aggregat gespeichert hat. Die Methode load() liefert alle gespeicherten Events zu einer Aggregat-ID in ihrer ursprünglichen Reihenfolge zurück.

<?php

declare(strict_types=1);

namespace App\EventSourcing;

interface StoredEvent
{
    public function eventType(): string;
    public function payload(): array;
}

final class ConcurrencyException extends \RuntimeException
{
}

final class EventStore
{
    public function __construct(private readonly \PDO $pdo)
    {
    }

    /**
     * @param StoredEvent[] $events
     * @throws ConcurrencyException if $expectedVersion no longer matches the stored version
     */
    public function append(string $aggregateId, int $expectedVersion, array $events): void
    {
        $stmt = $this->pdo->prepare(
            'INSERT INTO event_store (aggregate_id, version, event_type, payload, occurred_at)
             VALUES (:aggregate_id, :version, :event_type, :payload, :occurred_at)'
        );

        $version = $expectedVersion;
        foreach ($events as $event) {
            $version++;
            try {
                $stmt->execute([
                    'aggregate_id' => $aggregateId,
                    'version' => $version,
                    'event_type' => $event->eventType(),
                    'payload' => json_encode($event->payload()),
                    'occurred_at' => (new \DateTimeImmutable())->format(DATE_ATOM),
                ]);
            } catch (\PDOException $e) {
                // Relies on a UNIQUE constraint on (aggregate_id, version) to
                // detect a concurrent writer that already used this version.
                throw new ConcurrencyException(
                    "Aggregate '{$aggregateId}' was modified concurrently", previous: $e
                );
            }
        }
    }

    /**
     * @return array{version:int,event_type:string,payload:array}[]
     */
    public function load(string $aggregateId): array
    {
        $stmt = $this->pdo->prepare(
            'SELECT version, event_type, payload FROM event_store
             WHERE aggregate_id = :aggregate_id ORDER BY version ASC'
        );
        $stmt->execute(['aggregate_id' => $aggregateId]);

        return array_map(
            static fn (array $row) => [
                'version' => (int) $row['version'],
                'event_type' => $row['event_type'],
                'payload' => json_decode($row['payload'], true),
            ],
            $stmt->fetchAll(\PDO::FETCH_ASSOC)
        );
    }
}

3. Aggregate und Domain Events als unveränderliche Objekte modellieren

Ein Domain Event wird als unveränderliches, meist readonly modelliertes Objekt abgebildet, etwa OrderPlaced oder OrderShipped, das genau die zum Zeitpunkt des Eintretens relevanten Nutzdaten trägt. Ein Aggregat hält keinen direkt von außen gesetzten Zustand, sondern rekonstruiert ihn ausschließlich über eine apply()-Methode pro Event-Typ, die den internen Zustand anhand des jeweiligen Events aktualisiert.

Neue Zustandsänderungen entstehen ausschließlich über Befehlsmethoden auf dem Aggregat, etwa ship(), die zunächst die geltenden Geschäftsregeln prüfen und erst danach ein neues Event erzeugen. Dieses neue Event wird sofort per apply() auf den In-Memory-Zustand angewendet und zusätzlich einer Liste noch ungespeicherter Events hinzugefügt, die am Ende des Requests gebündelt im Event Store landen.

4. Aggregate-Rekonstruktion durch Replay aller Events

Ein Aggregat zu laden bedeutet in einem Event-Sourcing-System nicht, eine Zeile aus einer Tabelle zu lesen, sondern alle gespeicherten Events zur jeweiligen Aggregat-ID in ihrer ursprünglichen Reihenfolge zu laden und der Reihe nach auf ein frisches, leeres Aggregat anzuwenden, bis der aktuelle Zustand erreicht ist.

Dieses Vorgehen entkoppelt die Rekonstruktion vollständig von den ursprünglichen Erzeugungs- und Änderungsmethoden: Ein Aggregat, das über place() erzeugt und über ship() verändert wurde, lässt sich exakt genauso über reines Event-Replay wiederherstellen, ganz ohne diese Methoden erneut aufzurufen oder ihre Geschäftsregeln erneut zu prüfen.

<?php

declare(strict_types=1);

namespace App\Order;

final class Order
{
    private string $status = 'pending';
    private int $version = 0;

    /** @var object[] */
    private array $uncommittedEvents = [];

    private function __construct(private readonly string $orderId)
    {
    }

    public static function place(string $orderId, int $totalCents): self
    {
        $order = new self($orderId);
        $order->recordThat(new OrderPlaced($orderId, $totalCents));
        return $order;
    }

    /**
     * Reconstitutes an aggregate purely by replaying its historical events,
     * without touching the "place" factory or any business rules again.
     *
     * @param array{version:int,event_type:string,payload:array}[] $storedEvents
     */
    public static function reconstituteFromEvents(string $orderId, array $storedEvents): self
    {
        $order = new self($orderId);
        foreach ($storedEvents as $stored) {
            $order->apply(self::denormalize($stored['event_type'], $stored['payload']));
            $order->version = $stored['version'];
        }
        return $order;
    }

    public function ship(): void
    {
        if ($this->status !== 'pending') {
            throw new \DomainException("Order '{$this->orderId}' cannot be shipped from status '{$this->status}'");
        }
        $this->recordThat(new OrderShipped($this->orderId));
    }

    private function recordThat(object $event): void
    {
        $this->apply($event);
        $this->uncommittedEvents[] = $event;
    }

    private function apply(object $event): void
    {
        $this->status = match ($event::class) {
            OrderPlaced::class => 'pending',
            OrderShipped::class => 'shipped',
        };
    }

    private static function denormalize(string $type, array $payload): object
    {
        return match ($type) {
            'order_placed' => new OrderPlaced($payload['order_id'], $payload['total_cents']),
            'order_shipped' => new OrderShipped($payload['order_id']),
        };
    }

    public function pullUncommittedEvents(): array
    {
        $events = $this->uncommittedEvents;
        $this->uncommittedEvents = [];
        return $events;
    }
}

5. Optimistic Concurrency Control beim Anhängen neuer Events

Wenn zwei parallele Requests dasselbe Aggregat laden und beide darauf basierend neue Events erzeugen, entsteht ohne zusätzlichen Schutz ein Konflikt: Der zweite Schreibvorgang würde den ersten stillschweigend überschreiben oder zu einem inkonsistenten Endzustand führen, der so nie beabsichtigt war.

Die übliche Lösung ist Optimistic Concurrency Control: Jedes neue Event trägt die erwartete, zum Ladezeitpunkt gültige Versionsnummer des Aggregats. Der Event Store prüft beim Anhängen atomar, meist über einen eindeutigen Datenbank-Constraint auf Aggregat-ID und Version, ob diese erwartete Version noch der tatsächlich gespeicherten entspricht. Stimmt sie nicht mehr überein, wirft der Store eine ConcurrencyException, die der aufrufende Code behandeln muss, üblicherweise durch erneutes Laden des Aggregats und einen erneuten Versuch der ursprünglichen Operation.

6. Snapshotting: Performance bei langer Event-Historie

Bei einem Aggregat mit mehreren zehntausend Events wird das Laden bei jedem Zugriff spürbar langsamer, weil jedes Mal die komplette Historie von Anfang an neu abgespielt werden muss, nur um denselben aktuellen Zustand erneut zu berechnen, der schon beim letzten Laden bekannt war.

Snapshotting löst dieses Problem: In regelmäßigen Abständen, etwa alle hundert Events, wird der bereits berechnete Zustand eines Aggregats zusammen mit seiner Versionsnummer serialisiert und separat gespeichert. Beim nächsten Laden wird zunächst der neueste Snapshot geladen und anschließend nur noch die seither hinzugekommenen Events darauf angewendet, statt der gesamten Historie seit der Aggregat-Erzeugung.

<?php

declare(strict_types=1);

namespace App\EventSourcing;

final class SnapshotStore
{
    public function __construct(private readonly \PDO $pdo)
    {
    }

    public function save(string $aggregateId, int $version, array $state): void
    {
        $this->pdo->prepare(
            'REPLACE INTO aggregate_snapshots (aggregate_id, version, state)
             VALUES (:aggregate_id, :version, :state)'
        )->execute([
            'aggregate_id' => $aggregateId,
            'version' => $version,
            'state' => json_encode($state),
        ]);
    }

    public function loadLatest(string $aggregateId): ?array
    {
        $stmt = $this->pdo->prepare(
            'SELECT version, state FROM aggregate_snapshots WHERE aggregate_id = :aggregate_id'
        );
        $stmt->execute(['aggregate_id' => $aggregateId]);
        $row = $stmt->fetch(\PDO::FETCH_ASSOC);

        return $row === false
            ? null
            : ['version' => (int) $row['version'], 'state' => json_decode($row['state'], true)];
    }
}

// Loading now becomes: load the latest snapshot, then only replay events
// with a version greater than the snapshot's version, instead of every
// single event since the aggregate was first created.
function loadOrder(EventStore $store, SnapshotStore $snapshots, string $orderId): array
{
    $snapshot = $snapshots->loadLatest($orderId);
    $fromVersion = $snapshot['version'] ?? 0;

    return array_filter(
        $store->load($orderId),
        static fn (array $event) => $event['version'] > $fromVersion
    );
}

7. Projektionen: Lesemodelle aus Events aufbauen

Event Sourcing allein beantwortet Fragen wie alle offenen Bestellungen eines bestimmten Kunden nicht effizient, da dafür jedes Mal potenziell viele Aggregate vollständig rekonstruiert werden müssten. Deshalb werden aus demselben Event-Strom zusätzlich denormalisierte Projektionen aufgebaut, meist asynchron über einen Event-Listener, der jedes neue Event in eine für Abfragen optimierte Tabelle schreibt.

Diese Trennung von Schreib- und Lesemodell überschneidet sich bewusst mit dem CQRS-Pattern, das als eigenständiges Thema separat behandelt wird. Event Sourcing liefert dabei die zugrunde liegende Datenquelle, aus der sich beliebig viele unterschiedliche Projektionen für unterschiedliche Lesezugriffe ableiten lassen, ohne dass das Schreibmodell selbst davon etwas wissen muss.

8. Event-Versionierung: wenn sich das Schema ändert

Ein zentrales Problem in der Praxis ist, dass die Struktur eines vor zwei Jahren gespeicherten Events oft nicht mehr exakt zur aktuellen Erwartung des Codes passt, weil sich fachliche Anforderungen geändert haben, etwa durch ein neu hinzugekommenes Pflichtfeld. Da bereits gespeicherte Events aber unveränderlich bleiben müssen, kann das historische Schema nicht einfach nachträglich angepasst werden.

Die übliche Lösung ist Upcasting: eine kleine Transformationsschicht, die beim Laden ein altes Event-Format automatisch in das aktuell erwartete Format überführt, etwa indem ein fehlendes Feld mit einem sinnvollen Standardwert ergänzt wird. Für wirklich inkompatible Änderungen wird stattdessen ein neuer, versionierter Event-Typ eingeführt, während der alte Typ für historische Daten weiterhin gelesen werden kann.

9. Wann sich Event Sourcing lohnt, und wann nicht

Event Sourcing lohnt sich besonders bei Fachbereichen mit hohem Bedarf an Nachvollziehbarkeit, etwa Finanztransaktionen, Bestellprozesse oder Vertragsänderungen, sowie bei komplexer Geschäftslogik, deren Verlauf über die Zeit fachlich relevant ist und nicht nur der jeweilige Endzustand zählt.

Für einfache, CRUD-lastige Bereiche ohne fachlichen Bedarf an Historie ist Event Sourcing dagegen oft reiner Mehraufwand: eine höhere Einstiegshürde für neue Teammitglieder, aufwändigere Abfragen selbst für einfache Auflistungen und zusätzliche Infrastruktur für Projektionen. Die Entscheidung sollte deshalb pro Bounded Context getroffen werden, nicht als pauschale Architekturentscheidung für eine gesamte Anwendung.

10. Zusammenfassung

11. FAQ: Event Sourcing in PHP: Das Wichtigste auf einen Blick