Kapitola 13 · Vzory · Event Sourcing v DDD a Symfony

Event Sourcing v DDD a Symfony

Q: Co je Event Sourcing?

Event Sourcing je přístup k persistenci stavu, při kterém se neukládá aktuální snímek dat, ale append-only sekvence neměnných událostí, které k aktuálnímu stavu vedly. Aktuální stav agregátu vzniká přehráním těchto událostí od počátku, což poskytuje úplný audit trail a možnost zpětně rekonstruovat jakýkoli stav v čase. Platí princip „current state is derived from the history of events“: event log se pouze rozšiřuje o nové záznamy. Viz úvodní sekci .

Q: Jaký je vztah mezi Event Sourcingem a CQRS?

Event Sourcing a CQRS jsou dva nezávislé vzory, které se často kombinují. Každý z nich lze zavést samostatně: CQRS funguje i s klasickou ORM persistencí, ES lze implementovat i bez rozdělení na write a read modely. V praxi se však hodí dohromady, protože ES přirozeně vede k oddělení zápisu (event store) a čtení (projekce do read modelů) – což je přesně myšlenka CQRS. Více v sekci Vztah k CQRS .

Q: Co je Event Store a k čemu slouží?

Event Store je specializované append-only úložiště, které persistuje doménové události jednotlivých agregátů chronologicky seřazené. Typicky poskytuje dotazy na event stream konkrétního agregátu pro jeho rekonstrukci a globální dotaz pro čtení událostí všemi projekcemi. Základní metody jsou append(streamId, events) a readStream(streamId) ; pokročilejší řešení zahrnují optimistické zamykání verzí a publikování událostí do event busu. Implementačně může jít o specializovaný produkt (KurrentDB, dříve EventStoreDB), o PHP knihovnu nad relační databází (EventSauce, Broadway, prooph), nebo o vlastní minimalistickou nadstavbu, jakou staví tato kapitola. Detailní rozbor v sekci Implementace Event Store .

Q: Co jsou projekce v Event Sourcingu?

Projekce je proces, který naslouchá událostem z event store a buduje z nich read modely – denormalizované datové struktury určené pro rychlé dotazy. Projekce bývá jednoúčelová: každý read model má obvykle vlastní projekci, která ho od začátku nebo od posledního zpracovaného offsetu udržuje aktuální. Projekce lze kdykoli přebudovat (rebuild) přehráním událostí od počátku, čímž se bezpečně opravují chyby v read modelech. Praktický příklad v sekci Projekce .

Q: K čemu slouží snapshotting v Event Sourcingu?

Snapshotting je technika, při které se periodicky ukládá serializovaný stav agregátu, aby se při jeho rekonstrukci nemuselo přehrávat celé event history od začátku. Při načtení se vezme poslední snapshot a aplikují se pouze události, které nastaly po něm. Snapshoty řeší výkonnostní problém dlouhých streamů, typicky u agregátů s řádově tisíci událostí – pro krátké streamy jsou zbytečné a přidávají operační komplexitu. Podrobný rozbor v sekci Snapshotting .

Q: Kdy se vyplatí Event Sourcing nasadit?

Event Sourcing se vyplatí tam, kde je historie změn sama o sobě doménově cenná – finanční systémy, sklady, auditované procesy, regulovaná odvětví – nebo kde je třeba rekonstruovat stav v libovolném bodě minulosti. Nevhodný je pro prototypy, MVP a prosté CRUD aplikace. Nasazuje se zpravidla selektivně na jeden bounded context, nikoli plošně na celou aplikaci. Rozhodovací kritéria v sekci Kdy použít Event Sourcing .

Event Sourcing v kontextu Domain-Driven Design a Symfony – implementace Event Store, event-sourcovaných agregátů, projekcí, Outbox patternu, snapshottingu a verzování událostí. Včetně praktických problémů: idempotence projektorů, rebuild projekcí a eventual consistency.

Autor M. Katuščák

Doba čtení ≈ 45 min

Náročnost náročná

Publikováno · Aktualizováno 24. 4. 2025 · 8. 7. 2026

Obsah kapitoly

13.01 Co je Event Sourcing?#

Tradiční CRUD persistence má slepou skvrnu: při každé změně přepíše předchozí stav a veškerá historie se nenávratně ztrácí. Event Sourcing (ES) ukládá stav systému jako sekvenci neměnných událostí, jež k němu vedly [1]. Každá změna stavu domény je zaznamenána jako samostatná, pojmenovaná událost se svými daty. Aktuální stav agregátu pak vzniká přehráním (replay) těchto událostí od počátku.

Princip lze vyjádřit větou: current state is derived from the history of events. Namísto jediného řádku v databázové tabulce, který je při každé změně přepisován, existuje append-only log všech událostí, jež kdy na agregátu nastaly.

Porovnání s tradiční CRUD persistencí

V klasickém CRUD přístupu drží tabulka pouze aktuální stav entity – jakmile se hodnota změní, předchozí je pryč. Event Sourcing zapisuje každou změnu jako nový řádek event logu, takže žádná informace se nikdy nepřepisuje ani nemaže.

◢ Vzor

CRUD vs. Event Sourcing – přehled

Vlastnost	CRUD (tradiční)	Event Sourcing
Co se ukládá	Aktuální stav entity	Sekvence událostí (změn)
Auditní log	Vyžaduje další implementaci	Zabudován ve struktuře
Obnova stavu	Přímé čtení z tabulky	Replay event streamu
Temporální dotazy	Obtížné / nemožné	Přirozené (replay do libovolného bodu)
Složitost implementace	Nízká až střední	Vysoká
Výkon čtení	Závisí na schématu (rychlé pro jednoduché tabulky, pomalé pro mnoho JOINů)	Závisí na strategii (rychlé přes denormalizované projekce, pomalé bez snapshotů)

§ Poznámka

Pojmy Event Sourcingu:

Event (Událost) – Neměnný záznam o tom, co se v doméně přihodilo, vyjádřený v minulém čase (např. OrderPlaced, PaymentReceived). Obsahuje všechna data potřebná k rekonstrukci změny stavu.
Event Store – Specializované append-only úložiště pro události. Události se do něj pouze přidávají. Každá událost patří do event streamu konkrétního agregátu.
Aggregate (Agregát) – V kontextu ES je agregát rekonstruován přehráním všech událostí ze svého event streamu. Každá mutace stavu agregátu produkuje novou událost místo přímé modifikace atributů.
Projection (Projekce) – Read model sestavený z událostí. Projekce transformují event stream do podoby vhodné pro konkrétní dotazy (query) – například denormalizovaná tabulka pro přehled objednávek.
Snapshot – Periodicky ukládaný snímek aktuálního stavu agregátu, který slouží jako zkratka při replay. Umožňuje přehrát pouze události novější než poslední snapshot místo celého event streamu od počátku.

13.02 Vztah k CQRS#

Event Sourcing a CQRS jsou dva samostatné vzory [2]. Nejsou totéž – lze aplikovat CQRS bez Event Sourcingu a naopak ES bez CQRS. V praxi DDD aplikací se ale obvykle objevují společně.

Důvod je technický: Event Sourcing produkuje události jako základní artefakt persistence a CQRS potřebuje způsob, jak aktualizovat read modely při každé změně write strany. Události tuto propagaci pokrývají bez další infrastruktury – write side uloží událost do Event Store, read side ji přečte a aktualizuje projekci.

13.03 Kdy použít Event Sourcing#

Event Sourcing přidává konkrétní možnosti – auditní log, replay, temporální dotazy – výměnou za vyšší složitost infrastruktury i kódu. Před zavedením stojí za úvahu, zda v daném kontextu přínosy převažují nad náklady na implementaci a provoz. Tato sekce dává rozhodovací rámec; zbytek kapitoly rozebírá implementaci.

Vhodné případy užití

Auditní log jako doménový požadavek – Finanční systémy, zdravotnické záznamy nebo jakákoli doména, kde je zákonná povinnost uchovávat kompletní historii změn. Auditní log v ES vychází přímo z formátu úložiště – nepotřebuje samostatnou implementaci.
Komplexní doménová logika s bohatými stavovými přechody – Agregáty procházejí mnoha stavy, každý přechod má svou sémantiku a musí být rekonstruovatelný. Typicky: objednávkové systémy, workflow enginy, bankovní transakce.
Temporální dotazy – Potřeba „přehrát“ stav systému k libovolnému bodu v minulosti (debugging, analýza, „what-if“ scénáře). U ES stačí replay eventů do daného timestampu.
Event-driven integrace – Systém produkuje události, které konzumují jiné bounded contexts nebo externí systémy. ES zajišťuje, že žádná událost nebude ztracena – Event Store je zdrojem pravdy pro integraci.
CQRS s vysokou čtecí zátěží – ES umožňuje vybudovat libovolný počet optimalizovaných read modelů z jednoho event streamu, aniž by bylo nutné měnit write model.

Nevhodné případy užití

Jednoduché CRUD aplikace – Pokud doménová logika spočívá v základních operacích Create/Read/Update/Delete bez složitých stavových přechodů, ES přináší jen zbytečnou složitost.
Systémy orientované převážně na reporting – Pokud je primárním požadavkem rychlé čtení a agregace dat (BI, analytics), jsou vhodnější klasická DW řešení nebo OLAP databáze.
Prototypy a MVP – Rychlá validace produktového nápadu nepotřebuje složitou infrastrukturu. ES lze přidat do zralého systému inkrementálně, pokud se ukáže potřeba – viz Migrace z CRUD.
Týmy bez zkušeností s ES – Implementace Event Sourcingu bez předchozí zkušenosti přináší vysoké riziko chyb v kritické infrastruktuře (Event Store, serializace, versioning). Začíná se typicky menším bounded contextem jako experimentem.

! Pozor

Varování: Event Sourcing výrazně zvyšuje složitost systému

Event Sourcing CRUD nenahrazuje. Cenu zaplatíte na všech úrovních: infrastruktura (Event Store, event bus, snapshot store), doménový model (apply metody, immutabilita událostí, verzování schémat), testování (given/when/then scénáře s event streamy) a provoz (migrace schémat událostí, rebuildy projekcí, monitoring lag asynchronních projektorů). Podrobněji o výkonnostních dopadech pojednává kapitola Výkonnostní aspekty.

Nepoužívejte Event Sourcing paušálně pro celou aplikaci. V DDD se ES nasazuje selektivně na bounded contexts, kde se vrátí investice – typicky Core Domain s komplexní doménovou logikou. Ostatní kontexty mohou nadále používat klasickou CRUD persistenci. Časté chyby při zavádění ES shrnuje kapitola Anti-vzory.

Hotové knihovny, nebo vlastní store?

Padne-li rozhodnutí pro Event Sourcing, zbývá volba implementace. PHP ekosystém nabízí několik knihoven s odlišnou váhou i filozofií:

EventSauce – doporučený výchozí bod. Malé jádro, srozumitelná dokumentace a žádná vazba na konkrétní framework; integraci do Symfony si napíšete sami v řádu desítek řádků.
Broadway – starší knihovna s historicky silnou vazbou na Symfony. Funguje, ale tempo vývoje za poslední roky zvolnilo.
prooph – dlouho platil za standard PHP Event Sourcingu, dnes jej udržuje komunita. Pro nové projekty bývá první volbou spíš EventSauce.
Ecotone – ne knihovna, ale celý framework s ES, CQRS a Sagami. Má výraznou filozofii postavenou na message-driven architektuře; přijímáte ji vcelku, ne po částech.

Vlastní minimalistický store má smysl ve třech situacích: při učení, kdy chcete vidět principy bez vrstvy cizích abstrakcí, při požadavku na plnou kontrolu nad schématem a serializací, a u malé domény s několika typy událostí, kde by knihovna byla větší než problém. Tato kniha staví vlastní store z prvního důvodu – ukázky slouží k výuce principů, nikoli jako náhrada prověřené knihovny v produkci.

13.04 Doménové události jako základ Event Sourcingu#

V Event Sourcingu jsou doménové události (Domain Events) zdrojem pravdy o stavu systému – nejen notifikací o vedlejších efektech, jako je tomu u událostí v Doctrine ORM aplikaci. Tomu odpovídají i přísnější požadavky na jejich tvar:

První dva požadavky se týkají tvaru samotné třídy. Událost je po vytvoření neměnná – veškeré properties jsou read-only, nastavené v konstruktoru. A musí jít serializovat do trvalého formátu (JSON, MessagePack…) a deserializovat zpět bez ztráty informace.

Zbylé tři míří na obsah a životní cyklus:

schéma události se v čase vyvíjí – stará data v Event Store je třeba udržet čitelná, typicky upcastingem (transformací starých verzí na aktuální),
název vyjadřuje fakt v minulém čase: UserRegistered, OrderPlaced, PaymentFailed,
data musí být dost granulární na to, aby z události šel rekonstruovat stav bez přístupu k externím zdrojům.

◢ Vzor

PHP: Bázová třída DomainEvent a konkrétní třída UserRegistered

php src/Shared/Domain/Event/DomainEvent.php

1<?php2 3declare(strict_types=1);4 5namespace App\Shared\Domain\Event;6 7use DateTimeImmutable;8 9/**10 * Společná bázová třída pro doménové události. Identita a čas jsou11 * public readonly vlastnosti (přímý přístup `$event->eventId`,12 * `$event->occurredAt`); serializaci do Event Store a zpět řeší metody.13 * Všechny potomky jsou immutabilní value objekty.14 */15abstract class DomainEvent16{17    public function __construct(18        /** Unikátní identifikátor události (UUID v7). */19        public readonly string $eventId,20        /** Čas vzniku události - vždy UTC. */21        public readonly DateTimeImmutable $occurredAt,22    ) {}23 24    /**25     * Název události sloužící k jejímu uložení a vyhledání v Event Store.26     * Konvence: FQCN nebo krátký slug ve tvaru "user.registered".27     */28    abstract public function eventType(): string;29 30    /**31     * Verze schématu payloadu - klíčová pro upcasting starých událostí.32     * Nové verze události inkrementují toto číslo.33     */34    abstract public function schemaVersion(): int;35 36    /**37     * Serializace do pole pro uložení do Event Store.38     * Musí obsahovat všechna data potřebná k rekonstrukci stavu39     * VČETNĚ eventId a occurredAt - identita události je součástí payloadu.40     *41     * @return array<string, mixed>42     */43    abstract public function toPayload(): array;44 45    /**46     * Rekonstrukce události z payloadu načteného z Event Store.47     * Nesmí generovat nové UUID ani čas - obojí přebírá z payloadu.48     *49     * @param array<string, mixed> $payload50     */51    abstract public static function fromPayload(array $payload): static;52}

src/Shared/Domain/Event/DomainEvent.php

php src/Identity/Domain/Event/UserRegistered.php

1<?php2 3declare(strict_types=1);4 5namespace App\Identity\Domain\Event;6 7use App\Shared\Domain\Event\DomainEvent;8use DateTimeImmutable;9use Symfony\Component\Uid\Uuid;10 11/**12 * Událost emitovaná po úspěšné registraci uživatele.13 * Immutabilní - všechny properties jsou readonly.14 */15final class UserRegistered extends DomainEvent16{17    private function __construct(18        string $eventId,19        DateTimeImmutable $occurredAt,20        public readonly string $userId,21        public readonly string $email,22        public readonly string $fullName,23    ) {24        parent::__construct($eventId, $occurredAt);25    }26 27    /**28     * Named constructor pro PRVNÍ vytvoření události v doménovém kódu.29     * Pouze zde se generuje nové UUID a aktuální čas.30     */31    public static function create(string $userId, string $email, string $fullName): self32    {33        return new self(34            eventId:    (string) Uuid::v7(),35            occurredAt: new DateTimeImmutable('now', new \DateTimeZone('UTC')),36            userId:     $userId,37            email:      $email,38            fullName:   $fullName,39        );40    }41 42    /**43     * Rekonstrukce z Event Store: identita a čas se PŘEBÍRAJÍ z payloadu.44     * Negenerovat zde nové UUID - rozbilo by to idempotenci přes event_id.45     *46     * @param array<string, mixed> $payload47     */48    public static function fromPayload(array $payload): static49    {50        return new self(51            eventId:    $payload['eventId'],52            // Payload nenese offset, proto UTC uvádíme explicitně -53            // jinak by čas převzal default timezone serveru.54            occurredAt: new DateTimeImmutable($payload['occurredAt'], new \DateTimeZone('UTC')),55            userId:     $payload['userId'],56            email:      $payload['email'],57            fullName:   $payload['fullName'],58        );59    }60 61    public function eventType(): string62    {63        return 'identity.user_registered';64    }65 66    public function schemaVersion(): int67    {68        return 1; // při změně schématu inkrementujeme a vytvoříme upcaster69    }70 71    /** @return array<string, mixed> */72    public function toPayload(): array73    {74        return [75            'eventId'    => $this->eventId,76            'occurredAt' => $this->occurredAt->format('Y-m-d H:i:s.u'), // vždy UTC (viz create())77            'userId'     => $this->userId,78            'email'      => $this->email,79            'fullName'   => $this->fullName,80        ];81    }82}

src/Identity/Domain/Event/UserRegistered.php

Na rozdělení create() / fromPayload() stojí celá idempotence systému. Kdyby konstruktor generoval eventId a occurredAt při každém vytvoření instance, dostala by tatáž uložená událost po deserializaci nové UUID a nový čas. Tracking tabulka zpracovaných událostí by duplicitní doručení nepoznala a rebuild projekcí by pracoval s jinými časy, než jaké platily při zápisu. Identita události proto vzniká právě jednou – v create() – a payload ji nese s sebou; sloupce event_id a occurred_on v tabulce slouží už jen jako indexovaná metadata. Čas události se ukládá i parsuje v UTC explicitně. Formát payloadu offset nenese, takže by jinak deserializace na serveru s odlišnou default timezone časy posunula.

Pro eventType() se osvědčil formát <bounded_context>.<past_tense_verb_noun>, například ordering.order_placed nebo payment.payment_received. Tato konvence usnadňuje routing událostí v Symfony Messenger a jejich filtrování v Event Store.

13.05 Implementace Event Store#

Event Store je append-only databázové úložiště pro všechny doménové události. Každý záznam nese jednu událost s jejím kontextem – ke kterému agregátu patří, v jaké verzi streamu a kdy nastala. Záznamy se nikdy nepřepisují ani nemažou.

Struktura tabulky Event Store

◢ Vzor

SQL: Migrace tabulky `event_store` (MySQL/MariaDB)

sql migrations/snippet.sql

1CREATE TABLE event_store (2    id            BIGINT UNSIGNED  NOT NULL AUTO_INCREMENT,3    event_id      CHAR(36)         NOT NULL COMMENT 'UUID v7 události - globálně unikátní',4    aggregate_id  CHAR(36)         NOT NULL COMMENT 'UUID agregátu (vlastníka streamu)',5    aggregate_type VARCHAR(255)    NOT NULL COMMENT 'FQCN nebo slug agregátu, napr. ordering.order',6    event_type    VARCHAR(255)     NOT NULL COMMENT 'Typ události, napr. ordering.order_placed',7    payload       JSON             NOT NULL COMMENT 'Serializovaná data události',8    -- DEFAULT jako výraz v závorkách vyžaduje MySQL 8.0.13+.9    -- MariaDB mapuje JSON na LONGTEXT s kontrolou JSON_VALID; default zvládá od 10.2.10    metadata      JSON             NOT NULL DEFAULT ('{}') COMMENT 'Korelační ID, causation ID, user ID…',11    schema_version SMALLINT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'Verze schématu payloadu - pro upcasting',12    version       INT UNSIGNED     NOT NULL COMMENT 'Pořadové číslo ve streamu agregátu (od 1)',13    occurred_on   DATETIME(6)      NOT NULL COMMENT 'UTC čas vzniku události',14 15    PRIMARY KEY (id),16    UNIQUE KEY uq_event_id (event_id),17    -- Optimistic locking: dvojice (aggregate_id, version) musí být unikátní18    UNIQUE KEY uq_aggregate_version (aggregate_id, version),19    KEY idx_aggregate_id (aggregate_id),20    KEY idx_aggregate_type (aggregate_type),21    KEY idx_event_type (event_type),22    KEY idx_occurred_on (occurred_on)23) ENGINE=InnoDB24  DEFAULT CHARSET=utf8mb425  COLLATE=utf8mb4_unicode_ci26  COMMENT='Append-only store všech doménových událostí';

migrations/snippet.sql

Sloupec version nese optimistic locking. Před zápisem nové události command handler přečte poslední verzi streamu agregátu. Pokud mezitím jiný proces zapsal událost se stejnou verzí, databáze při insertu vyvolá výjimku z porušení unikátního indexu uq_aggregate_version. Souběžné zápisy se tak detekují bez pesimistického zamykání řádků.

◢ Vzor

PHP: Interface EventStore a Doctrine implementace

php src/Infrastructure/EventSourcing/EventStore.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\EventSourcing;6 7use App\Shared\Domain\Event\DomainEvent;8 9interface EventStore10{11    /**12     * Uloží nové události do event streamu agregátu.13     *14     * @param DomainEvent[] $events15     * @param int           $expectedVersion Verze posledního uloženého eventu - slouží16     *                                        pro optimistic locking. Použijte 0 pro nový agregát.17     *18     * @throws ConcurrencyException Pokud $expectedVersion neodpovídá skutečné verzi streamu.19     */20    public function append(21        string $aggregateId,22        string $aggregateType,23        array $events,24        int $expectedVersion,25    ): void;26 27    /**28     * Načte celý event stream agregátu (nebo od dané verze pro snapshot support).29     *30     * @return EventEnvelope[]31     */32    public function loadStream(33        string $aggregateId,34        int $fromVersion = 1,35    ): array;36 37    /**38     * Načte všechny události z celého Event Store (pro rebuild projekcí).39     * Vrací generátor pro paměťově efektivní iteraci nad miliony záznamů.40     *41     * @return \Generator<EventEnvelope>42     */43    public function loadAll(int $batchSize = 500): \Generator;44}

src/Infrastructure/EventSourcing/EventStore.php

php src/Infrastructure/EventSourcing/DoctrineEventStore.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\EventSourcing;6 7use App\Shared\Domain\Event\DomainEvent;8use Doctrine\DBAL\Connection;9use Doctrine\DBAL\Exception\UniqueConstraintViolationException;10use Doctrine\DBAL\ParameterType;11 12final class DoctrineEventStore implements EventStore13{14    public function __construct(15        private readonly Connection $connection,16        private readonly EventSerializer $serializer,17    ) {}18 19    /**20     * @param DomainEvent[] $events21     */22    public function append(23        string $aggregateId,24        string $aggregateType,25        array $events,26        int $expectedVersion,27    ): void {28        $version = $expectedVersion;29 30        $this->connection->beginTransaction();31 32        try {33            foreach ($events as $event) {34                $version++;35 36                $this->connection->insert('event_store', [37                    'event_id'       => $event->eventId,38                    'aggregate_id'   => $aggregateId,39                    'aggregate_type' => $aggregateType,40                    'event_type'     => $event->eventType(),41                    'payload'        => json_encode($event->toPayload(), JSON_THROW_ON_ERROR),42                    'metadata'       => '{}',43                    'schema_version' => $event->schemaVersion(),44                    'version'        => $version,45                    'occurred_on'    => $event->occurredAt->format('Y-m-d H:i:s.u'),46                ]);47            }48 49            $this->connection->commit();50        } catch (UniqueConstraintViolationException $e) {51            $this->connection->rollBack();52            throw new ConcurrencyException(53                "Concurrency conflict for aggregate {$aggregateId} at version {$version}.",54                previous: $e,55            );56        } catch (\Throwable $e) {57            $this->connection->rollBack();58            throw $e;59        }60    }61 62    /**63     * @return EventEnvelope[]64     */65    public function loadStream(string $aggregateId, int $fromVersion = 1): array66    {67        $rows = $this->connection->fetchAllAssociative(68            'SELECT event_type, payload, schema_version, version, occurred_on69               FROM event_store70              WHERE aggregate_id = :aggregateId71                AND version >= :fromVersion72           ORDER BY version ASC',73            ['aggregateId' => $aggregateId, 'fromVersion' => $fromVersion],74        );75 76        return array_map(77            fn(array $row) => $this->serializer->deserialize($row),78            $rows,79        );80    }81 82    /**83     * Iteruje přes celý Event Store v dávkách - paměťově efektivní pro rebuild projekcí.84     *85     * @return \Generator<EventEnvelope>86     */87    public function loadAll(int $batchSize = 500): \Generator88    {89        $lastId = 0;90 91        do {92            $rows = $this->connection->fetchAllAssociative(93                'SELECT id, event_type, payload, schema_version, version, occurred_on94                   FROM event_store95                  WHERE id > :lastId96               ORDER BY id ASC97                  LIMIT :limit',98                ['lastId' => $lastId, 'limit' => $batchSize],99                ['limit' => ParameterType::INTEGER],100            );101 102            foreach ($rows as $row) {103                $lastId = (int) $row['id'];104                yield $this->serializer->deserialize($row);105            }106        } while (count($rows) === $batchSize);107    }108}

src/Infrastructure/EventSourcing/DoctrineEventStore.php

Parametr :limit v metodě loadAll() má explicitní typ ParameterType::INTEGER. Bez něj DBAL hodnotu naváže jako řetězec a MySQL výraz LIMIT '500' odmítne jako syntaktickou chybu.

13.06 Agregát s Event Sourcingem#

V klasickém DDD agregát mění svůj stav přímou modifikací vlastních atributů. V Event Sourcingu každá změna stavu prochází přes doménovou událost. Metody agregátu nemodifikují atributy přímo – nahrají událost a teprve její aplikace na stav vyvolá změnu.

Výsledkem je, že agregát obsahuje dvě sady metod:

Mutační metody (veřejné rozhraní agregátu) – validují invarianty, rozhodují, která událost nastane, a volají interní metodu pro nahrání události (typicky recordEvent()). Jméno se záměrně liší od record() ve stavově ukládaném AggregateRoot – zde metoda událost navíc aplikuje na stav a inkrementuje verzi streamu.
apply*() metody (private/protected) – přijmou konkrétní typ události a aplikují změnu na interní stav. Tyto metody jsou volány jak při nahrávání nové události, tak při replay z Event Store.

Pro testování to znamená vzor given/when/then – given jsou historické události, when je volání metody na agregátu, then jsou nově emitované události. Podrobně v kapitole Testování DDD kódu.

◢ Vzor

PHP: Base class EventSourcedAggregate

php src/Shared/Domain/EventSourcedAggregate.php

1<?php2 3declare(strict_types=1);4 5namespace App\Shared\Domain;6 7use App\Shared\Domain\Event\DomainEvent;8 9abstract class EventSourcedAggregate10{11    /** @var DomainEvent[] Události nahrané v aktuální transakci - čekají na uložení. */12    private array $recordedEvents = [];13 14    private int $version = 0;15 16    /**17     * Nahrajeme novou událost: aplikujeme ji na stav, zapamatujeme ji pro persistenci18     * a inkrementujeme verzi streamu - nezbytné pro optimistic locking.19     */20    protected function recordEvent(DomainEvent $event): void21    {22        $this->applyEvent($event);23        $this->recordedEvents[] = $event;24        $this->version++;25    }26 27    /**28     * Přehrajeme historické události z Event Store (bez přidávání do $recordedEvents).29     *30     * @param DomainEvent[] $events31     */32    public static function reconstituteFromEvents(array $events): static33    {34        $aggregate = new static();35 36        foreach ($events as $event) {37            $aggregate->applyEvent($event);38            $aggregate->version++;39        }40 41        return $aggregate;42    }43 44    /**45     * Přehraje dodatečné události na existující instanci (pro snapshot support).46     *47     * @param DomainEvent[] $events48     */49    public function replayEvents(array $events): void50    {51        foreach ($events as $event) {52            $this->applyEvent($event);53            $this->version++;54        }55    }56 57    /**58     * Obnoví verzi streamu při rekonstrukci ze snapshotu.59     * Bez obnovení by verze začínala na nule a optimistic locking60     * by při prvním uložení selhal na konfliktu.61     */62    protected function restoreVersion(int $version): void63    {64        $this->version = $version;65    }66 67    /**68     * Dynamické dispatchování na apply*() metody podle třídy události.69     * Konvence: apply + ShortClassName, napr. applyOrderCreated().70     * apply*() metody v podtřídách MUSÍ být protected (ne private),71     * jinak je PHP nemůže volat z kontextu této nadtřídy.72     */73    private function applyEvent(DomainEvent $event): void74    {75        $method = 'apply' . (new \ReflectionClass($event))->getShortName();76 77        if (!method_exists($this, $method)) {78            throw new \LogicException(79                sprintf('Aggregate %s must implement %s().', static::class, $method)80            );81        }82 83        $this->$method($event);84    }85 86    /** @return DomainEvent[] */87    public function releaseEvents(): array88    {89        $events = $this->recordedEvents;90        $this->recordedEvents = [];91        return $events;92    }93 94    public function version(): int95    {96        return $this->version;97    }98}

src/Shared/Domain/EventSourcedAggregate.php

U event-sourced agregátu se stav rekonstruuje replayem a vnitřní properties zůstávají privátní – metody apply* je opakovaně přepisují. Gettery na konci třídy proto nahrazují public private(set) z kapitoly Návrh agregátu; jde o záměrnou odchylku od konvence zbytku knihy.

◢ Vzor

PHP: Order agregát s Event Sourcingem

php src/Ordering/Domain/Order.php

1<?php2 3declare(strict_types=1);4 5namespace App\Ordering\Domain;6 7use App\Ordering\Domain\Event\OrderConfirmed;8use App\Ordering\Domain\Event\OrderCreated;9use App\Ordering\Domain\Event\OrderItemAdded;10use App\Ordering\Domain\Event\OrderShipped;11use App\Shared\Domain\EventSourcedAggregate;12 13final class Order extends EventSourcedAggregate14{15    private string $orderId;16    private string $customerId;17    private OrderStatus $status;18 19    /** @var OrderItem[] */20    private array $items = [];21 22    private ?string $trackingNumber = null;23 24    // Statická továrna - vytvoří objednávku ve stavu Draft25    public static function create(string $orderId, string $customerId): self26    {27        $order = new self();28        $order->recordEvent(OrderCreated::create($orderId, $customerId));29 30        return $order;31    }32 33    public function addItem(OrderItem $item): void34    {35        if ($this->status !== OrderStatus::Draft) {36            throw new \DomainException('Items can only be added to draft orders.');37        }38 39        $this->recordEvent(OrderItemAdded::create($this->orderId, $item));40    }41 42    public function confirm(): void43    {44        if ($this->status !== OrderStatus::Draft) {45            throw new \DomainException('Only draft orders can be confirmed.');46        }47        if (empty($this->items)) {48            throw new \DomainException('Cannot confirm an empty order.');49        }50 51        $this->recordEvent(OrderConfirmed::create($this->orderId));52    }53 54    public function ship(string $trackingNumber): void55    {56        if ($this->status !== OrderStatus::Confirmed) {57            throw new \DomainException('Only confirmed orders can be shipped.');58        }59 60        $this->recordEvent(OrderShipped::create($this->orderId, $trackingNumber));61    }62 63    // --- apply* metody - MUSÍ být protected (ne private), aby je base class mohla volat dynamicky ---64    // --- Obsahují POUZE změnu interního stavu, žádnou doménovou logiku ---65 66    protected function applyOrderCreated(OrderCreated $event): void67    {68        $this->orderId    = $event->orderId;69        $this->customerId = $event->customerId;70        $this->items      = [];71        $this->status     = OrderStatus::Draft;72    }73 74    protected function applyOrderItemAdded(OrderItemAdded $event): void75    {76        $this->items[] = $event->item;77    }78 79    protected function applyOrderConfirmed(OrderConfirmed $event): void80    {81        $this->status = OrderStatus::Confirmed;82    }83 84    protected function applyOrderShipped(OrderShipped $event): void85    {86        $this->status         = OrderStatus::Shipped;87        $this->trackingNumber = $event->trackingNumber;88    }89 90    // Gettery pro aplikační vrstvu91    public function orderId(): string         { return $this->orderId; }92    public function status(): OrderStatus     { return $this->status; }93    public function trackingNumber(): ?string { return $this->trackingNumber; }94}

src/Ordering/Domain/Order.php

Načítání agregátu z event streamu (replay)

Repozitář pro event-sourcovaný agregát neprovádí SELECT do tabulky entit. Místo toho načte event stream z Event Store a předá jej statické tovární metodě reconstituteFromEvents(). Výsledný agregát má přesně takový stav, jaký odpovídá historii jeho událostí.

◢ Vzor

PHP: EventSourced repozitář pro Order agregát

php src/Infrastructure/Ordering/EventSourcedOrderRepository.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\Ordering;6 7use App\Ordering\Domain\Order;8use App\Infrastructure\EventSourcing\EventStore;9use App\Infrastructure\EventSourcing\EventSerializer;10 11final class EventSourcedOrderRepository12{13    private const AGGREGATE_TYPE = 'ordering.order';14 15    public function __construct(16        private readonly EventStore $eventStore,17        private readonly EventSerializer $serializer,18    ) {}19 20    public function load(string $orderId): Order21    {22        $envelopes = $this->eventStore->loadStream($orderId);23 24        if (empty($envelopes)) {25            throw new \DomainException("Order {$orderId} not found.");26        }27 28        $events = array_map(29            fn($envelope) => $this->serializer->toEvent($envelope),30            $envelopes,31        );32 33        return Order::reconstituteFromEvents($events);34    }35 36    public function save(Order $order): void37    {38        $newEvents = $order->releaseEvents();39 40        if (empty($newEvents)) {41            return;42        }43 44        // expectedVersion = aktuální verze PŘED novými událostmi45        $expectedVersion = $order->version() - count($newEvents);46 47        $this->eventStore->append(48            $order->orderId(),49            self::AGGREGATE_TYPE,50            $newEvents,51            $expectedVersion,52        );53    }54}

src/Infrastructure/Ordering/EventSourcedOrderRepository.php

13.07 Projekce (Projections)#

Event Store je append-only a neumí ad-hoc dotazy typu „všechny objednávky zákazníka X s celkovou hodnotou nad 1000 Kč“. Pro takové dotazy vznikají vedle něj projekce – denormalizované read modely budované z event streamu specificky pro tvar dotazů aplikace.

Tok eventu z agregátu do read modelu přes projektor

Synchronní vs. asynchronní projekce

Synchronní projekce – Projekce se aktualizuje přímo v téže transakci jako zápis události. Garantuje konzistenci dat v okamžiku odpovědi na command, ale zvyšuje latenci zápisu a zavádí těsnou vazbu mezi write a read stranou.
Asynchronní projekce – Události jsou po uložení do Event Store zařazeny do fronty (Symfony Messenger + transport jako RabbitMQ nebo Redis). Projektor je konzument, který zpracovává zprávy nezávisle. Read model je v krátkém časovém okně nekonzistentní (eventual consistency), ale write side je rychlejší a oddělená.

◢ Vzor

PHP: OrderSummaryProjection a asynchronní projektor

php src/Infrastructure/Ordering/Projection/OrderSummaryProjector.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\Ordering\Projection;6 7use App\Ordering\Domain\Event\OrderConfirmed;8use App\Ordering\Domain\Event\OrderCreated;9use App\Ordering\Domain\Event\OrderItemAdded;10use App\Ordering\Domain\Event\OrderShipped;11use Doctrine\DBAL\Connection;12use Symfony\Component\Messenger\Attribute\AsMessageHandler;13 14/**15 * Projector budující tabulku order_summary z doménových událostí.16 *17 * Každá metoda handle*() odpovídá jednomu typu události a je registrována18 * jako samostatný Messenger handler atributem #[AsMessageHandler] na úrovni metody.19 */20final class OrderSummaryProjector21{22    public function __construct(23        private readonly Connection $connection,24    ) {}25 26    #[AsMessageHandler]27    public function handleOrderCreated(OrderCreated $event): void28    {29        $this->connection->insert('order_summary', [30            'order_id'      => $event->orderId,31            'customer_id'   => $event->customerId,32            'status'        => 'draft',33            'item_count'    => 0,34            'total_amount'  => 0,35            'placed_at'     => $event->occurredAt->format('Y-m-d H:i:s'),36            'shipped_at'    => null,37            'tracking_no'   => null,38        ]);39    }40 41    #[AsMessageHandler]42    public function handleOrderItemAdded(OrderItemAdded $event): void43    {44        $this->connection->executeStatement(45            'UPDATE order_summary46                SET item_count   = item_count + 1,47                    total_amount = total_amount + :price48              WHERE order_id = :orderId',49            ['price' => $event->item->unitPrice(), 'orderId' => $event->orderId],50        );51    }52 53    #[AsMessageHandler]54    public function handleOrderConfirmed(OrderConfirmed $event): void55    {56        $this->connection->executeStatement(57            'UPDATE order_summary SET status = :status WHERE order_id = :orderId',58            ['status' => 'confirmed', 'orderId' => $event->orderId],59        );60    }61 62    #[AsMessageHandler]63    public function handleOrderShipped(OrderShipped $event): void64    {65        $this->connection->executeStatement(66            'UPDATE order_summary67                SET status      = :status,68                    shipped_at  = :shippedAt,69                    tracking_no = :trackingNo70              WHERE order_id = :orderId',71            [72                'status'     => 'shipped',73                'shippedAt'  => $event->occurredAt->format('Y-m-d H:i:s'),74                'trackingNo' => $event->trackingNumber,75                'orderId'    => $event->orderId,76            ],77        );78    }79}

src/Infrastructure/Ordering/Projection/OrderSummaryProjector.php

Asynchronní doručování událostí projektorům přes Symfony Messenger vyžaduje nastavený transport a routing v config/packages/messenger.yaml:

◢ Vzor

YAML: Konfigurace Symfony Messenger pro asynchronní projekce

yaml config/packages/messenger.yaml

1framework:2    messenger:3        transports:4            async:5                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'6                options:7                    auto_setup: true8 9        routing:10            # Všechny doménové události routujeme na async transport11            'App\Ordering\Domain\Event\OrderCreated':    async12            'App\Ordering\Domain\Event\OrderItemAdded': async13            'App\Ordering\Domain\Event\OrderConfirmed': async14            'App\Ordering\Domain\Event\OrderShipped':   async

config/packages/messenger.yaml

Projekce lze přebudovat (rebuild) přehráním celého Event Store od začátku. Při změně doménových požadavků stačí vytvořit novou projekci a přehrát historii. CRUD systémy tuto možnost nemají – historická data v nich už nejsou k dispozici.

13.08 Event Store jako outbox#

Předchozí sekce ukazovala projektory jako Messenger handlery, které dostávají doménové události z asynchronní fronty. Implicitně jsme předpokládali, že se událost po zápisu do Event Store spolehlivě dostane do message brokeru. V produkci to bez další infrastruktury neplatí. Zápis do databáze a publikace do brokeru jsou dvě nezávislé operace; spadne-li proces mezi nimi, událost je uložená, ale ke konzumentům nikdy nedorazí. Tento dual-write problem, jeho varianty i obecné řešení s kompletním kódem rozebírá kapitola Outbox Pattern. Zde jen to, co je na Event Sourcingu specifické: druhá tabulka není potřeba.

Relay čte přímo z event_store

Tabulka event_store splňuje vlastnosti outbox tabulky sama o sobě. Je to append-only log a každý záznam vzniká ve stejné transakci jako odpovídající doménová změna. Přidává se relay worker, který čte nové řádky podle id a posílá je do Messengeru. Pozici posledního publikovaného řádku si ukládá do checkpoint tabulky, takže po restartu pokračuje tam, kde skončil. Samotný checkpoint ovšem nestačí: kvůli gap problému popsanému v následujícím calloutu se musí kombinovat s některou z mitigací – překryvem s deduplikací, výběrem přes FOR UPDATE SKIP LOCKED, nebo CDC. Implementace relay – polling worker pod supervisord, nebo varianta s CDC – je shodná s běžným outboxem, viz Relay process – dvě varianty.

Záruky doručení a jejich důsledky

Outbox dává at-least-once doručení uvnitř jednoho kanálu. Konkrétně:

At-least-once: pokud relay spadne mezi dispatchem a updatem checkpointu, stejná událost se po restartu publikuje znovu. Konzumenti musí být idempotentní – přesně tak, jak ukazuje následující sekce u projektorů.
Pořadí: relay publikuje vzestupně podle id. Uvnitř streamu jednoho agregátu to odpovídá pořadí verzí, takže projektor uvidí OrderCreated před OrderShipped. Napříč agregáty pořadí zajištěno není a kvůli gap problému popsanému výše nejde o spolehlivé globální pořadí commitů.
Latence: mezi commitem události a jejím doručením k projektoru vzniká okno odpovídající polling intervalu relay. V praxi 100 ms až 1 s; nižší latenci dává výstupní transport, který umí push (např. PostgreSQL LISTEN/NOTIFY nebo Debezium).

13.09 Praktické problémy projekcí#

Předchozí sekce ukázaly, jak projekci vybudovat a jak události spolehlivě doručit. V praxi se ale objevují problémy, které z jednoduchých ukázek nejsou patrné. Tato sekce pokrývá nejčastější z nich – idempotenci, chybové stavy, rebuild a eventual consistency z pohledu uživatelského rozhraní.

Idempotence projektorů

Asynchronní transport (RabbitMQ, Redis Streams, Amazon SQS) garantuje doručení zprávy alespoň jednou (at-least-once delivery). Zpráva se proto může doručit opakovaně – po timeoutu, restartu workeru nebo síťovém výpadku. Pokud projektor není idempotentní, opakované zpracování způsobí poškozená data: duplicitní řádky, zdvojené částky, nekonzistentní počty.

Idempotenci lze zajistit dvěma způsoby: upsert (INSERT … ON DUPLICATE KEY UPDATE) místo prostého INSERT, nebo tracking tabulka již zpracovaných událostí. U tracking tabulky musí záznam checkpointu a zápis projekce proběhnout v jedné databázové transakci – atomicita obou zápisů je podstatou idempotence. Pád workeru mezi nimi by jinak událost tiše ztratil.

◢ Vzor

PHP: Idempotentní projektor s tracking tabulkou

php src/Infrastructure/Ordering/Projection/IdempotentOrderProjector.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\Ordering\Projection;6 7use App\Ordering\Domain\Event\OrderCreated;8use Doctrine\DBAL\Connection;9use Symfony\Component\Messenger\Attribute\AsMessageHandler;10 11/**12 * Idempotentní projektor: před zpracováním ověří, zda událost13 * již nebyla zpracována, pomocí tabulky projection_checkpoint.14 *15 * Pozn.: Atribut #[AsMessageHandler] na třídě registruje __invoke() jako handler.16 * Pro projektory zpracovávající více typů událostí použijte atribut17 * na jednotlivých metodách - viz OrderSummaryProjector výše.18 */19#[AsMessageHandler]20final class IdempotentOrderProjector21{22    public function __construct(23        private readonly Connection $connection,24    ) {}25 26    public function __invoke(OrderCreated $event): void27    {28        // Checkpoint i projekce běží v jedné transakci. Pád workeru mezi29        // oběma zápisy by jinak zanechal checkpoint bez projekce30        // a opakované doručení by událost tiše přeskočilo.31        $this->connection->transactional(function () use ($event): void {32            // Atomická kontrola + záznam: INSERT IGNORE vrátí 0 affected rows33            // pokud eventId již existuje → událost byla již zpracována.34            $affected = $this->connection->executeStatement(35                'INSERT IGNORE INTO projection_checkpoint (event_id, projection_name, processed_at)36                 VALUES (:eventId, :projection, NOW(6))',37                ['eventId' => $event->eventId, 'projection' => 'order_summary'],38            );39 40            if ($affected === 0) {41                return; // Duplicitní doručení - přeskočíme42            }43 44            // Vlastní projekční logika45            $this->connection->insert('order_summary', [46                'order_id'     => $event->orderId,47                'customer_id'  => $event->customerId,48                'status'       => 'draft',49                'item_count'   => 0,50                'total_amount' => 0,51                'placed_at'    => $event->occurredAt->format('Y-m-d H:i:s'),52            ]);53        });54    }55}

src/Infrastructure/Ordering/Projection/IdempotentOrderProjector.php

◢ Vzor

SQL: Tabulka `projection_checkpoint` pro tracking zpracovaných událostí

sql migrations/snippet.sql

1CREATE TABLE projection_checkpoint (2    event_id        CHAR(36)     NOT NULL COMMENT 'UUID události - odkaz na event_store.event_id',3    projection_name VARCHAR(100) NOT NULL COMMENT 'Název projekce, napr. order_summary',4    processed_at    DATETIME(6)  NOT NULL COMMENT 'Čas zpracování',5 6    PRIMARY KEY (event_id, projection_name)7) ENGINE=InnoDB8  DEFAULT CHARSET=utf8mb49  COLLATE=utf8mb4_unicode_ci10  COMMENT='Tracking tabulka pro idempotentní projektory - zabraňuje duplicitnímu zpracování';

migrations/snippet.sql

Chybové stavy a retry strategie

Projektor může selhat z mnoha důvodů: dočasná nedostupnost databáze, neplatný payload u staré události bez upcasteru, nebo bug v projekční logice. Symfony Messenger nabízí dvě hlavní mechaniky pro řešení:

Retry transport – zpráva se po selhání automaticky vrátí do fronty s exponenciálním backoffem (výchozí: 3 pokusy s násobičem 2).
Failed transport (dead letter queue) – po vyčerpání retry pokusů se zpráva přesune do samostatné fronty, kde čeká na manuální zásah. Nedojde ke ztrátě události ani k zablokování zbytku fronty.

◢ Vzor

YAML: Kompletní konfigurace Messenger s retry a dead letter queue

Následující konfigurace rozšiřuje základní nastavení z předchozí sekce o retry strategii a failed transport:

yaml config/packages/messenger.yaml

1framework:2    messenger:3        # Failed transport - sem padají zprávy po vyčerpání retry pokusů4        failure_transport: failed5 6        transports:7            async:8                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'9                retry_strategy:10                    max_retries: 311                    delay: 1000        # 1 sekunda12                    multiplier: 2      # exponenciální backoff: 1s, 2s, 4s13                    max_delay: 60000   # max 60 sekund mezi pokusy14 15            failed:16                dsn: 'doctrine://default?queue_name=failed'17 18        routing:19            'App\Ordering\Domain\Event\OrderCreated':    async20            'App\Ordering\Domain\Event\OrderItemAdded': async21            'App\Ordering\Domain\Event\OrderConfirmed': async22            'App\Ordering\Domain\Event\OrderShipped':   async

config/packages/messenger.yaml

Pro diagnostiku a opětovné zpracování selhalých zpráv slouží příkazy Symfony Messenger:

bin/console messenger:failed:show – zobrazí zprávy v dead letter queue
bin/console messenger:failed:retry – pokusí se zprávy znovu zpracovat
bin/console messenger:failed:remove {id} – odstraní neplatnou zprávu

Rebuild projekcí

Možnost přebudovat projekci od začátku je v Event Sourcingu praktická obrana proti chybám v projekční logice. V provozu jde ale o netriviální operaci. Rebuild musí běžet odděleně od normálního provozu projektoru, stará data se musí korektně odstranit a po dokončení musí projekce odpovídat aktuálnímu stavu Event Store.

◢ Vzor

PHP: Symfony konzolový příkaz pro rebuild projekce

php src/Infrastructure/EventSourcing/Console/RebuildProjectionCommand.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\EventSourcing\Console;6 7use App\Infrastructure\EventSourcing\EventStore;8use App\Infrastructure\EventSourcing\EventSerializer;9use Doctrine\DBAL\Connection;10use Symfony\Component\Console\Attribute\AsCommand;11use Symfony\Component\Console\Command\Command;12use Symfony\Component\Console\Input\InputArgument;13use Symfony\Component\Console\Input\InputInterface;14use Symfony\Component\Console\Output\OutputInterface;15use Symfony\Component\Console\Style\SymfonyStyle;16 17#[AsCommand(18    name: 'app:projection:rebuild',19    description: 'Přebuduje projekci přehráním všech událostí z Event Store.',20)]21final class RebuildProjectionCommand extends Command22{23    /** @var array<string, array{projector: callable, table: string}> Registr projektorů dle názvu */24    private array $projectors;25 26    /**27     * @param iterable<string, object> $projectors Symfony tagged_iterator28     * @param array<string, string>    $projectionTables Mapa: název projekce → název tabulky29     */30    public function __construct(31        private readonly Connection $connection,32        private readonly EventStore $eventStore,33        private readonly EventSerializer $serializer,34        iterable $projectors,35        array $projectionTables,36    ) {37        parent::__construct();38        foreach ($projectors as $name => $projector) {39            $this->projectors[$name] = [40                'projector' => $projector,41                'table'     => $projectionTables[$name] ?? throw new \InvalidArgumentException(42                    "Projekce '{$name}' nemá definovanou tabulku v \$projectionTables.",43                ),44            ];45        }46    }47 48    protected function configure(): void49    {50        $this->addArgument('projection', InputArgument::REQUIRED, 'Název projekce k přebudování');51    }52 53    protected function execute(InputInterface $input, OutputInterface $output): int54    {55        $io = new SymfonyStyle($input, $output);56        $name = $input->getArgument('projection');57 58        if (!isset($this->projectors[$name])) {59            $io->error("Projekce '{$name}' neexistuje. Dostupné: " . implode(', ', array_keys($this->projectors)));60            return Command::FAILURE;61        }62 63        $config = $this->projectors[$name];64        $table  = $config['table'];65 66        $io->warning("Rebuild smaže data projekce '{$name}' (tabulka '{$table}') a přehraje celý Event Store.");67 68        // 1. Smazat stávající data projekce - název tabulky pochází z whitelistu,69        //    nikoli z uživatelského vstupu, takže nehrozí SQL injection.70        $this->connection->executeStatement("TRUNCATE TABLE {$table}");71 72        // 2. Vymazat checkpoint záznamy pro tuto projekci73        $this->connection->executeStatement(74            'DELETE FROM projection_checkpoint WHERE projection_name = :name',75            ['name' => $name],76        );77 78        // 3. Přehrát všechny události z Event Store. Dispatch podle konvence79        //    handle{NázevUdálosti}() - události, pro které projektor80        //    handler nemá, se přeskočí.81        $projector = $config['projector'];82        $count = 0;83        $batchSize = 500;84 85        foreach ($this->eventStore->loadAll($batchSize) as $envelope) {86            $event  = $this->serializer->toEvent($envelope);87            $method = 'handle' . (new \ReflectionClass($event))->getShortName();88 89            if (!method_exists($projector, $method)) {90                continue;91            }92 93            $projector->$method($event);94            $count++;95 96            if ($count % $batchSize === 0) {97                $io->text("Zpracováno {$count} událostí…");98            }99        }100 101        $io->success("Projekce '{$name}' přebudována. Celkem {$count} událostí.");102        return Command::SUCCESS;103    }104}

src/Infrastructure/EventSourcing/Console/RebuildProjectionCommand.php

Eventual consistency a uživatelské rozhraní

Asynchronní projekce vytváří časové okno, typicky milisekundy až jednotky sekund, kdy uživatel akci provedl, ale read model ji ještě nezobrazuje. Po kliknutí na „Potvrdit“ svítí na výpisu stále „Draft“. Nejde o bug, nýbrž o vlastnost architektury. Strategie pro UI – optimistickou aktualizaci, potvrzovací stránku, polling či SSE – rozebírá sekce Eventual Consistency v praxi v kapitole CQRS.

13.10 Snapshotting#

Se stárnutím systému rostou event streamy agregátů. Agregát s tisíci událostmi vyžaduje při každém zpracování commandu načíst a přehrát celý ten objem řádků z databáze. Výkonnostní problém se v provozu objeví dřív, než tým očekává.

Vzor snapshotting uchová aktuální stav agregátu v pravidelných intervalech – po každých N událostech nebo časově. Při příštím načtení repozitář vyhledá poslední snapshot a z Event Store dotáhne jen události novější než tento snapshot.

Snapshot strategie: zhuštěný stav místo plného replay

Kdy vytvářet snapshots

Poté, co replay agregátu začne měřitelně zpomalovat – práh závisí na doméně, typicky se pohybuje od stovek po tisíce událostí.
Periodicky (např. jednou denně) pro agregáty s vysokou frekvencí událostí.
Na vyžádání – jako optimalizační krok po migraci nebo importu dat.

◢ Vzor

PHP: Snapshot třída a repozitář se snapshot podporou

php src/Infrastructure/EventSourcing/Snapshot.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\EventSourcing;6 7use DateTimeImmutable;8 9/**10 * Snapshot uchovává serializovaný stav agregátu v konkrétní verzi event streamu.11 */12final class Snapshot13{14    public function __construct(15        public readonly string $aggregateId,16        public readonly string $aggregateType,17        public readonly int $version,18        public readonly array $state,       // serializovaný stav agregátu19        public readonly DateTimeImmutable $takenAt,20    ) {}21}

src/Infrastructure/EventSourcing/Snapshot.php

php src/Infrastructure/Ordering/SnapshottingOrderRepository.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\Ordering;6 7use App\Ordering\Domain\Order;8use App\Infrastructure\EventSourcing\EventStore;9use App\Infrastructure\EventSourcing\Snapshot;10use App\Infrastructure\EventSourcing\SnapshotStore;11use App\Infrastructure\EventSourcing\EventSerializer;12 13final class SnapshottingOrderRepository14{15    private const AGGREGATE_TYPE    = 'ordering.order';16    private const SNAPSHOT_INTERVAL = 50; // snapshot každých 50 událostí17 18    public function __construct(19        private readonly EventStore    $eventStore,20        private readonly SnapshotStore $snapshotStore,21        private readonly EventSerializer $serializer,22    ) {}23 24    public function load(string $orderId): Order25    {26        // 1. Pokusíme se načíst snapshot27        $snapshot = $this->snapshotStore->findLatest($orderId, self::AGGREGATE_TYPE);28 29        if ($snapshot !== null) {30            // 2a. Máme snapshot - přehrajeme pouze události novější než snapshot31            $fromVersion = $snapshot->version + 1;32            $aggregate   = Order::reconstituteFromSnapshot($snapshot->state, $snapshot->version);33        } else {34            // 2b. Nemáme snapshot - přehrajeme celý event stream od začátku35            $fromVersion = 1;36            $aggregate   = null;37        }38 39        $envelopes = $this->eventStore->loadStream($orderId, $fromVersion);40 41        if (empty($envelopes) && $aggregate === null) {42            throw new \DomainException("Order {$orderId} not found.");43        }44 45        if (!empty($envelopes)) {46            $events = array_map(47                fn($e) => $this->serializer->toEvent($e),48                $envelopes,49            );50 51            if ($aggregate !== null) {52                $aggregate->replayEvents($events);53            } else {54                $aggregate = Order::reconstituteFromEvents($events);55            }56        }57 58        return $aggregate;59    }60 61    public function save(Order $order): void62    {63        $newEvents = $order->releaseEvents();64 65        if (empty($newEvents)) {66            return;67        }68 69        $expectedVersion = $order->version() - count($newEvents);70 71        $this->eventStore->append(72            $order->orderId(),73            self::AGGREGATE_TYPE,74            $newEvents,75            $expectedVersion,76        );77 78        // Automatické snapshotování79        if ($order->version() % self::SNAPSHOT_INTERVAL === 0) {80            $this->snapshotStore->save(new Snapshot(81                aggregateId:   $order->orderId(),82                aggregateType: self::AGGREGATE_TYPE,83                version:       $order->version(),84                state:         $order->toSnapshot(),85                takenAt:       new \DateTimeImmutable('now', new \DateTimeZone('UTC')),86            ));87        }88    }89}

src/Infrastructure/Ordering/SnapshottingOrderRepository.php

Aby byl snapshotting funkční, musí agregát implementovat metody toSnapshot(): array (serializace aktuálního stavu) a statickou reconstituteFromSnapshot(array $state, int $version): static (deserializace). Na rozdíl od reconstituteFromEvents() tato metoda nevytváří apply*() volání – přímo nastaví properties z uloženého snímku a přes restoreVersion() z base class obnoví verzi streamu. Bez obnovené verze by optimistic locking při prvním uložení selhal. Formát snapshotu se proto musí vyvíjet spolu s doménovým modelem.

13.11 Verzování událostí (Event Versioning)#

Události v Event Store jsou permanentní – jednou uložené zůstávají ve své podobě natrvalo. Doménový model se přitom v čase vyvíjí: přibývají atributy, mění se struktura dat, původní pole se rozdělují nebo slučují. Otázka tedy zní: jak přečíst starou událost novým kódem?

Odpověď je event versioning – strategie, která zachovává zpětnou čitelnost starých událostí i po změně jejich schématu. Nejrozšířenějším vzorem je upcasting: při deserializaci se starší verze payloadu transformuje na aktuální formát, takže doménový model pracuje pouze s nejnovější verzí.

Proč je verzování nezbytné

Append-only princip – Události v Event Store nelze měnit. Pokud změníte schéma události, stará data zůstávají v původním formátu navždy.
Replay a projekce – Při přebudování projekcí nebo replay agregátu se přehrávají všechny historické události, včetně těch z prvních verzí systému.
Dlouhověkost systému – Event-sourcovaný systém může běžet roky. Za tu dobu se doménové požadavky změní mnohokrát a schémata událostí se musejí vyvíjet spolu s nimi.

Vzor Upcaster

Upcaster je objekt, který transformuje payload události z jedné verze do následující. Upcasters se řetězí: pokud existuje událost ve verzi 1 a aktuální verze je 3, proběhne transformace v1 → v2 → v3. Upcasting se provádí při čtení (deserializaci), nikoli při zápisu – původní data v Event Store zůstávají nedotčena.

◢ Vzor

PHP: Interface EventUpcaster

php src/Infrastructure/EventSourcing/Versioning/EventUpcaster.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\EventSourcing\Versioning;6 7/**8 * Upcaster transformuje payload události ze starší verze na novější.9 * Každý upcaster je zodpovědný za přesně jeden přechod verze (např. v1 → v2).10 */11interface EventUpcaster12{13    /**14     * Typ události, na který se upcaster vztahuje (např. "identity.user_registered").15     */16    public function eventType(): string;17 18    /**19     * Zdrojová verze payloadu, kterou tento upcaster transformuje.20     */21    public function fromVersion(): int;22 23    /**24     * Cílová verze payloadu po transformaci.25     */26    public function toVersion(): int;27 28    /**29     * Transformuje payload ze zdrojové verze na cílovou.30     *31     * @param array<string, mixed> $payload Data události ve zdrojové verzi.32     * @return array<string, mixed> Data události v cílové verzi.33     */34    public function upcast(array $payload): array;35}

src/Infrastructure/EventSourcing/Versioning/EventUpcaster.php

Konkrétní příklad: rozdělení pole `fullName`

Představme si reálnou situaci: při spuštění systému událost UserRegistered obsahovala pole fullName (celé jméno jako jeden řetězec). Později se objevil požadavek rozlišit křestní jméno a příjmení – vznikla verze 2 se dvěma poli firstName a lastName. V Event Store ale stále existují tisíce událostí v1 s polem fullName.

◢ Vzor

PHP: Upcaster pro UserRegistered v1 → v2

php src/Infrastructure/Identity/Versioning/UserRegisteredV1ToV2Upcaster.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\Identity\Versioning;6 7use App\Infrastructure\EventSourcing\Versioning\EventUpcaster;8 9/**10 * Transformuje UserRegistered v1 (fullName) na v2 (firstName + lastName).11 *12 * Strategie rozdělení: první slovo je firstName, zbytek lastName.13 * Pokud jméno obsahuje pouze jedno slovo, lastName se nastaví na prázdný řetězec.14 */15final readonly class UserRegisteredV1ToV2Upcaster implements EventUpcaster16{17    public function eventType(): string18    {19        return 'identity.user_registered';20    }21 22    public function fromVersion(): int23    {24        return 1;25    }26 27    public function toVersion(): int28    {29        return 2;30    }31 32    /**33     * @param array<string, mixed> $payload34     * @return array<string, mixed>35     */36    public function upcast(array $payload): array37    {38        $fullName = $payload['fullName'] ?? '';39        $parts    = explode(' ', trim($fullName), 2);40 41        $payload['firstName'] = $parts[0];42        $payload['lastName']  = $parts[1] ?? '';43 44        // Odstraníme původní pole - v2 schéma jej již nepoužívá45        unset($payload['fullName']);46 47        return $payload;48    }49}

src/Infrastructure/Identity/Versioning/UserRegisteredV1ToV2Upcaster.php

◢ Vzor

PHP: UpcasterChain – řetězení upcasterů při deserializaci

php src/Infrastructure/EventSourcing/Versioning/UpcasterChain.php

1<?php2 3declare(strict_types=1);4 5namespace App\Infrastructure\EventSourcing\Versioning;6 7/**8 * Řetězí upcasters a transformuje payload z libovolné historické verze9 * na aktuální verzi. Upcasters se aplikují postupně: v1 → v2 → v3 → …10 */11final readonly class UpcasterChain12{13    /** @var array<string, array<int, EventUpcaster>> Klíč = eventType, vnitřní klíč = fromVersion */14    private array $upcasters;15 16    /**17     * @param EventUpcaster[] $upcasters18     */19    public function __construct(array $upcasters)20    {21        $map = [];22 23        foreach ($upcasters as $upcaster) {24            $map[$upcaster->eventType()][$upcaster->fromVersion()] = $upcaster;25        }26 27        $this->upcasters = $map;28    }29 30    /**31     * Aplikuje všechny relevantní upcasters na payload.32     *33     * @param string              $eventType      Typ události (např. "identity.user_registered").34     * @param int                 $schemaVersion  Verze payloadu uloženého v Event Store.35     * @param array<string, mixed> $payload        Původní payload z Event Store.36     * @return array<string, mixed> Transformovaný payload v aktuální verzi.37     */38    public function upcast(string $eventType, int $schemaVersion, array $payload): array39    {40        if (!isset($this->upcasters[$eventType])) {41            return $payload;42        }43 44        $version = $schemaVersion;45 46        while (isset($this->upcasters[$eventType][$version])) {47            $upcaster = $this->upcasters[$eventType][$version];48            $payload  = $upcaster->upcast($payload);49            $version  = $upcaster->toVersion();50        }51 52        return $payload;53    }54}

src/Infrastructure/EventSourcing/Versioning/UpcasterChain.php

V praxi se UpcasterChain integruje do EventSerializer: při deserializaci se z uloženého záznamu přečte event_type a schema_version, payload projde řetězem upcasterů a teprve výsledná transformovaná data se předají konstruktoru aktuální třídy události.

§ Poznámka

Weak vs. strong schema strategie

Pro tvar payloadu existují dva přístupy, které se v praxi míchají:

Weak schema (slabé schéma) – Payload je uložen jako volný JSON bez formální definice. Upcasters transformují data ad-hoc. Výhodou je flexibilita a rychlost vývoje; nevýhodou je, že chyby v transformaci se projeví až za běhu a je obtížné ověřit konzistenci napříč verzemi.
Strong schema (silné schéma) – Každá verze události má explicitně definované schéma (např. pomocí JSON Schema nebo PHP třídy s validací). Upcaster pak transformuje mezi dvěma dobře definovanými strukturami. Výhodou je vyšší bezpečnost a možnost automatického testování kompatibility; nevýhodou je vyšší režie při každé změně schématu.

Pro většinu projektů je rozumným kompromisem kombinace obou přístupů: silné schéma pro kritické události v Core Doméně (finanční transakce, stavy objednávek) a slabé schéma pro méně kritické události v podpůrných kontextech (notifikace, logy aktivit).

Změny, které upcasting neřeší

Upcasting předpokládá, že stará data lze deterministicky přeložit na nový formát. Některé změny tuto vlastnost nemají:

Sémantická změna pole. Order.shippingPrice původně zahrnoval DPH, od v3 ho neobsahuje. Stará data nelze správně přeložit – DPH sazba v okamžiku vystavení objednávky není v eventu uložená. Upcaster může jen předpokládat (např. konstantní 21 %), což je nepřesné a generuje reporty s chybnými čísly.
Event splitting. Původní OrderPlaced obsahoval customerData inline. V nové verzi se rozděluje na OrderPlaced + CustomerSnapshotted (samostatný event). Upcaster by musel vytvořit druhý event z prvního, což porušuje princip „1 fyzický event v Event Store = 1 logický fakt“.
Event merging. Dva eventy ItemAdded + ItemQuantityChanged se v nové doméně spojí do jednoho ItemUpserted. Upcasting jdoucí jednou cestou nestačí – potřebujete agregátní transformaci napříč streamem.
Sémantický bug v doménové logice. Stará data byla validní podle starého modelu, ale ten model byl chybný. Replay přes opravený kód vyhodí výjimky.

Tři možné cesty, podle závažnosti:

◢ Vzor

Strategie 1: Copy-and-replace stream

Spustí se one-time migrace, která čte starý stream, transformuje events v PHP kódu (žádný upcaster, plnohodnotná migrace) a zapíše do nového streamu (order_v2). Starý stream zůstává jako audit trail, ale doménový kód ho ignoruje.

order_v1 (frozen, audit only)
    │
    ▼ migration script
order_v2 (active)

Cena: doba běhu migrace (může to být hodiny u velkých streamů), nutnost double-write během přechodného období (aplikace zapisuje do obou streamů, dokud migrace neskončí).

◢ Vzor

Strategie 3: Compensating event

Místo přepisování historie se vloží nová událost, která starý fakt opravuje:

v1: OrderPlaced(price=100, includedVAT=true)
v2: PriceCorrectedDueToVATBug(orderId, originalPrice=100, correctedNet=82.64)
v3: ... (další eventy pracují s opravenou hodnotou)

Doménový kód při replay aplikuje obě události a stav konverguje na správnou hodnotu. Audit trail je explicitní – stará data jsou zachována, oprava je samostatný fakt.

Cena: doménový model získává „šum“ event typů, které řeší minulé bugy. Po pár letech provozu tvoří historické opravy znatelnou část event typů.

Stream archivation a storage tiering

Agregáty s dlouhým životním cyklem (UserAccount, Subscription, LedgerAccount) nasbírají za roky provozu desítky až stovky tisíc událostí. Aktivní Event Store tabulka roste, dotazy se zpomalují a snapshoty musí vznikat častěji.

Standardní řešení: storage tiering podle stáří streamu.

Hot tier (PostgreSQL master) – události za posledních 90 dní, dotazy < 10 ms.
Warm tier (Postgres replika nebo tatáž databáze na pomalejším disku) – události 90 dní – 2 roky. Hydration sahá sem jen pro forenzní dotazy nebo plný replay projekce.
Cold tier (S3, Glacier, on-prem object storage) – události starší než 2 roky. Pouze pro čtení; přístup k němu vyžadují jen auditní reporty a compliance.

Implementace: každou noc se spustí job, který přesune event_store řádky starší než N dní do event_store_archive tabulky (nebo přímo do S3 jako Parquet). Repozitář při hydration ve výchozím nastavení cold tier nečte – pokud agregát potřebuje plný replay, operátor jej explicitně obnoví ze snapshotu novějšího, než je hranice cold tieru. Pro audit dotazy funguje zvlášť query service, který umí číst všechny tři tiers.

! Pozor

Event Store je z definice append-only, ale GDPR požaduje právo na výmaz (čl. 17). Konflikt řeší crypto shredding: osobní údaje v eventech jsou zašifrovány klíčem per-subject. Po žádosti o výmaz se zničí klíč, události zůstávají, ale jsou nečitelné.

Implementace v PHP: každý subject (uživatel) má v separátní tabulce subject_keys symetrický klíč. Doménová událost při serializaci zašifruje PII pole (email, name, address) tímto klíčem; zbytek payloadu zůstává čitelný (audit trail funguje). Smazání klíče = právo na zapomnění, audit zachycen na úrovni „uživatel #42 učinil akci v čase T“, ale identifikace uživatele není možná.

Alternativou je referenční přístup: event nese jen ID subjektu a osobní údaje leží v separátní tabulce s možností DELETE.

Časté otázky

Co je Event Sourcing?

Event Sourcing je přístup k persistenci stavu, při kterém se neukládá aktuální snímek dat, ale append-only sekvence neměnných událostí, které k aktuálnímu stavu vedly. Aktuální stav agregátu vzniká přehráním těchto událostí od počátku, což poskytuje úplný audit trail a možnost zpětně rekonstruovat jakýkoli stav v čase. Platí princip „current state is derived from the history of events“: event log se pouze rozšiřuje o nové záznamy. Viz úvodní sekci.

Jaký je vztah mezi Event Sourcingem a CQRS?

Event Sourcing a CQRS jsou dva nezávislé vzory, které se často kombinují. Každý z nich lze zavést samostatně: CQRS funguje i s klasickou ORM persistencí, ES lze implementovat i bez rozdělení na write a read modely. V praxi se však hodí dohromady, protože ES přirozeně vede k oddělení zápisu (event store) a čtení (projekce do read modelů) – což je přesně myšlenka CQRS. Více v sekci Vztah k CQRS.

Co je Event Store a k čemu slouží?

Event Store je specializované append-only úložiště, které persistuje doménové události jednotlivých agregátů chronologicky seřazené. Typicky poskytuje dotazy na event stream konkrétního agregátu pro jeho rekonstrukci a globální dotaz pro čtení událostí všemi projekcemi. Základní metody jsou append(streamId, events) a readStream(streamId); pokročilejší řešení zahrnují optimistické zamykání verzí a publikování událostí do event busu. Implementačně může jít o specializovaný produkt (KurrentDB, dříve EventStoreDB), o PHP knihovnu nad relační databází (EventSauce, Broadway, prooph), nebo o vlastní minimalistickou nadstavbu, jakou staví tato kapitola. Detailní rozbor v sekci Implementace Event Store.

Co jsou projekce v Event Sourcingu?

Projekce je proces, který naslouchá událostem z event store a buduje z nich read modely – denormalizované datové struktury určené pro rychlé dotazy. Projekce bývá jednoúčelová: každý read model má obvykle vlastní projekci, která ho od začátku nebo od posledního zpracovaného offsetu udržuje aktuální. Projekce lze kdykoli přebudovat (rebuild) přehráním událostí od počátku, čímž se bezpečně opravují chyby v read modelech. Praktický příklad v sekci Projekce.

K čemu slouží snapshotting v Event Sourcingu?

Snapshotting je technika, při které se periodicky ukládá serializovaný stav agregátu, aby se při jeho rekonstrukci nemuselo přehrávat celé event history od začátku. Při načtení se vezme poslední snapshot a aplikují se pouze události, které nastaly po něm. Snapshoty řeší výkonnostní problém dlouhých streamů, typicky u agregátů s řádově tisíci událostí – pro krátké streamy jsou zbytečné a přidávají operační komplexitu. Podrobný rozbor v sekci Snapshotting.

Kdy se vyplatí Event Sourcing nasadit?

Event Sourcing se vyplatí tam, kde je historie změn sama o sobě doménově cenná – finanční systémy, sklady, auditované procesy, regulovaná odvětví – nebo kde je třeba rekonstruovat stav v libovolném bodě minulosti. Nevhodný je pro prototypy, MVP a prosté CRUD aplikace. Nasazuje se zpravidla selektivně na jeden bounded context, nikoli plošně na celou aplikaci. Rozhodovací kritéria v sekci Kdy použít Event Sourcing.

13.01 Co je Event Sourcing?#

Porovnání s tradiční CRUD persistencí

13.02 Vztah k CQRS#

13.03 Kdy použít Event Sourcing#

Vhodné případy užití

Nevhodné případy užití

Hotové knihovny, nebo vlastní store?

13.04 Doménové události jako základ Event Sourcingu#

13.05 Implementace Event Store#

Struktura tabulky Event Store

13.06 Agregát s Event Sourcingem#

Načítání agregátu z event streamu (replay)

13.07 Projekce (Projections)#

Synchronní vs. asynchronní projekce

13.08 Event Store jako outbox#

Relay čte přímo z event_store

Záruky doručení a jejich důsledky

13.09 Praktické problémy projekcí#

Idempotence projektorů

Chybové stavy a retry strategie

Rebuild projekcí

Eventual consistency a uživatelské rozhraní

13.10 Snapshotting#

Kdy vytvářet snapshots

13.11 Verzování událostí (Event Versioning)#

Proč je verzování nezbytné

Vzor Upcaster

Konkrétní příklad: rozdělení pole fullName

Změny, které upcasting neřeší

Stream archivation a storage tiering

Časté otázky

Konkrétní příklad: rozdělení pole `fullName`