Event Sourcing v DDD a Symfony

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ždou změnu stavu domény zaznamenává samostatná pojmenovaná událost s vlastními daty. Aktuální stav agregátu pak vzniká přehráním (replay) těchto událostí od počátku.

Fowler princip formuluje jako „Event Sourcing ensures that all changes to application state are stored as a sequence of events“ [1]. Místo jednoho přepisovaného řádku v tabulce vzniká append-only log všech událostí, které kdy na agregátu nastaly.

Fowlerova formulace z roku 2005 je široká. Sedne na ni auditní log i stream processing. Verraes ji o čtrnáct let později zúžil [3]. Systém je podle něj event-sourced tehdy, když jediným zdrojem pravdy je uložená historie událostí. Přidává k tomu podmínku, kterou Fowler nemá: z téže historie se musí vynucovat pravidla pro nové události. Tím se Event Sourcing odděluje od pouhého logování změn – log, ze kterého se nikdo nerozhoduje, je archiv, ne zdroj pravdy.

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

V CRUD přístupu drží tabulka jen aktuální stav entity a změněná hodnota je pryč. Event Sourcing zapisuje každou změnu jako nový řádek event logu, takže se nic nepřepisuje ani nemaže.

13.02 Vztah k CQRS#

Event Sourcing a CQRS jsou dva samostatné vzory [2]. Nejsou totéž: CQRS jde nasadit bez Event Sourcingu a ES bez CQRS. V doménově orientovaných aplikacích 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.

Symetrie mezi oběma vzory ale neplatí. CQRS nad klasickou ORM persistencí je běžná konfigurace bez zvláštních nároků. Event Sourcing bez CQRS možný je, jenže write model se pak nedá dotazovat ničím jiným než čtením streamu podle ID. Dotaz „vypiš všechny uživatele se jménem Greg“ nad event streamem nesestavíte, jak upozorňuje už Young [2]. Read stranu proto postavíte tak jako tak – a tím se dostáváte ke CQRS oklikou.

13.03 Kdy použít Event Sourcing#

Event Sourcing přidává auditní log, replay a temporální dotazy. Platíte za ně vyšší složitostí infrastruktury i kódu, takže se vyplatí jen tam, kde přínosy převáží náklady na implementaci a provoz.

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ě a 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. Událost je v ES zároveň zápisem stavu, takže se z úložiště neztratí. Doručení konzumentům ale zajišťuje až relay, viz Event Store jako outbox. Ven se navíc publikuje jen vybraná podmnožina událostí, ne interní stream agregátu; viz Interní a publikované události.
• 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 na DDD.
• 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.

Broker není Event Store

Kafka, RabbitMQ ani Redis Streams roli Event Store nezastanou, i když se v nich události také objevují v pořadí. Dudycz uvádí dva technické důvody [4]. Brokery neumí optimistic concurrency na úrovni streamu, takže nemají čím ochránit invariant agregátu proti souběžnému zápisu. Stream v nich jde odebírat, ne přečíst. Poskládat z něj agregát je proto buď nespolehlivé, nebo drahé. Brokery jsou podle něj stavěné na přepravu událostí, ne na trvalé uložení. Ukazuje to i retence: data po nastavené době mizí, a to je u zdroje pravdy nepřijatelné.

Rozdělení rolí je tedy jednoznačné: Event Store drží historii a vynucuje nad ní pravidla, broker ji rozváží konzumentům. Zbytek kapitoly proto používá Symfony Messenger a RabbitMQ výhradně jako přepravu, nikdy jako úložiště.

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í. Následující přehled popisuje stav k září 2026 a stárne rychleji než zbytek kapitoly – aktuální kondici balíčku si před volbou ověřte na Packagistu.

• EventSauce – malé jádro, srozumitelná dokumentace, žádná vazba na framework. Doctrine repozitář zpráv i outbox dodává v samostatných balíčcích, takže vlastní kód píšete hlavně kolem DI a Messengeru.
• patchlevel/event-sourcing – bundle deklaruje podporu Symfony 8, stejně jako Symfony bundle Ecotone. Přináší hotové subscriptions, snapshoty, upcasting i crypto-shredding. Při preferenci hotového řešení pro Symfony je to dnes první, na co se podívat.
• Ecotone – ne knihovna, ale celý framework s ES, CQRS a ságami. Má výraznou filozofii postavenou na message-driven architektuře; přijímáte ji vcelku, ne po částech. Stabilní řada 1.x Symfony 8 podporuje, řada 2.0 je zatím v beta verzi.
• prooph – řada 7.x je udržovaná a používaná, doprovodný Symfony bundle se ale od roku 2024 nehnul a končí u Symfony 7. Integraci pro Symfony 8 si napíšete sami.

Broadway do tohoto seznamu už nepatří. V srpnu 2026 vyšla verze 3.0.1 označená jako abandoned a repozitář skončil v archivu. Pro nové projekty odpadá; v existujících kódových základnách zůstává jako zátěž k odstranění.

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. V aplikaci nad Doctrine ORM jsou jen notifikací o vedlejších efektech. Tomu odpovídají i přísnější požadavky na jejich tvar.

První dva se týkají samotné třídy. Událost je po vytvoření neměnná – všechny vlastnosti jsou jen pro čtení a nastavuje je konstruktor. 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 proto musí zůstat č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.

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>.<podstatné_jméno>_<sloveso_v_minulém_čase>, například ordering.order_placed nebo payment.payment_succeeded. Usnadňuje routing událostí v Symfony Messenger i jejich filtrování v Event Store.

Interní a publikované události

Události v Event Store slouží dvěma různým účelům a smíchat je stojí draho. Interní událost je jednotka stavu agregátu. Její tvar se mění spolu s doménovým modelem a nikdo mimo Bounded Context o něm nemá vědět. Publikovaná (integrační) událost je naopak veřejné rozhraní kontextu – smlouva, na které staví cizí týmy.

Verraes k tomu formuluje pravidlo: veřejná je malá vybraná podmnožina událostí, všechno ostatní zůstává ve výchozím stavu privátní [7]. Bez tohoto rozdělení se vnitřní struktura kontextu stává jeho API. Každé přejmenování pole v agregátu je pak breaking change pro konzumenty a refaktoring interního modelu přestává být možný. Většina bolesti, kterou řeší sekce Verzování událostí, má původ právě zde.

Prakticky to znamená dvě sady tříd. Interní události zůstávají v Domain\Event a čtou je pouze agregát a vlastní projekce. Publikované události žijí v samostatném namespace (typicky Application\IntegrationEvent), mají stabilní eventType, vlastní verzování a překlad z interní události do publikované obstará explicitní mapper. Cena je jedna vrstva navíc; protihodnotou je možnost měnit doménový model bez koordinace s okolím.

13.05 Implementace Event Store#

Každý záznam v Event Store 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

SQL v této kapitole cílí na MySQL 8: AUTO_INCREMENT, INSERT IGNORE, NOW(6), TRUNCATE TABLE a ENGINE=InnoDB. Skeleton Symfony přitom ve výchozím stavu míří na PostgreSQL a lokální zkoušky často běží na SQLite. Na obojím se ukázky bez překladu neobejdou: INSERT IGNORE nezná ani jedna z těch databází, TRUNCATE nezná SQLite.

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ů.

Existuje i druhá varianta. Young pracuje vedle tabulky událostí ještě s tabulkou agregátů, která nese aktuální verzi streamu denormalizovaně [2]. Konflikt se pak kontroluje předem, proti uložené hodnotě, a chybová hláška vzniká v aplikačním kódu. Cenou je jeden zápis navíc při každém commitu a nutnost udržet obě tabulky v jedné transakci. Unikátní index žádný další zápis nepotřebuje, zato konflikt zjistí až z výjimky, kterou kód přeloží na ConcurrencyException. Přesně to dělá následující implementace.

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#

Stavově ukládaný agregát mění své atributy přímo. V Event Sourcingu každá změna stavu prochází přes doménovou událost: metoda agregátu událost nahraje a změnu stavu vyvolá teprve její aplikace.

Agregát proto 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 (protected, viz komentář v applyEvent()) – přijmou konkrétní typ události a aplikují změnu na interní stav. Volají se 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 volání metody na agregátu, then nově emitované události. Podrobně v kapitole Testování DDD.

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 Order níže proto nahrazují public private(set) z kapitoly Návrh agregátu; jde o záměrnou odchylku od konvence zbytku knihy.

PHP: události event-sourced agregátu

Události dědí z DomainEvent, takže nesou eventId i occurredAt a umí se přeložit do payloadu a zpět. Vlastní data drží v primitivních polích – právě proto, že cestují do Event Store a zpět přes serializer. Vytváří je pojmenovaný konstruktor create(), identitu a čas tedy dostanou právě jednou:

1<?php2 3declare(strict_types=1);4 5namespace App\Ordering\EventSourced\Event;6 7use App\Ordering\EventSourced\OrderItem;8use App\SharedKernel\Domain\Event\DomainEvent;9use DateTimeImmutable;10use Symfony\Component\Uid\Uuid;11 12final class OrderPlaced extends DomainEvent13{14    private function __construct(15        string $eventId,16        DateTimeImmutable $occurredAt,17        public readonly string $orderId,18        public readonly string $customerId,19    ) {20        parent::__construct($eventId, $occurredAt);21    }22 23    public static function create(string $orderId, string $customerId): self24    {25        return new self(26            eventId:    (string) Uuid::v7(),27            occurredAt: new DateTimeImmutable('now', new \DateTimeZone('UTC')),28            orderId:    $orderId,29            customerId: $customerId,30        );31    }32 33    /** @param array<string, mixed> $payload */34    public static function fromPayload(array $payload): static35    {36        return new self(37            eventId:    $payload['eventId'],38            occurredAt: new DateTimeImmutable($payload['occurredAt'], new \DateTimeZone('UTC')),39            orderId:    $payload['orderId'],40            customerId: $payload['customerId'],41        );42    }43 44    public function eventType(): string { return 'ordering.order_placed'; }45 46    public function schemaVersion(): int { return 1; }47 48    /** @return array<string, mixed> */49    public function toPayload(): array50    {51        return [52            'eventId'    => $this->eventId,53            'occurredAt' => $this->occurredAt->format('Y-m-d H:i:s.u'),54            'orderId'    => $this->orderId,55            'customerId' => $this->customerId,56        ];57    }58}59 60final class OrderItemAdded extends DomainEvent61{62    private function __construct(63        string $eventId,64        DateTimeImmutable $occurredAt,65        public readonly string $orderId,66        public readonly OrderItem $item,67    ) {68        parent::__construct($eventId, $occurredAt);69    }70 71    public static function create(string $orderId, OrderItem $item): self72    {73        return new self(74            eventId:    (string) Uuid::v7(),75            occurredAt: new DateTimeImmutable('now', new \DateTimeZone('UTC')),76            orderId:    $orderId,77            item:       $item,78        );79    }80 81    /** @param array<string, mixed> $payload */82    public static function fromPayload(array $payload): static83    {84        return new self(85            eventId:    $payload['eventId'],86            occurredAt: new DateTimeImmutable($payload['occurredAt'], new \DateTimeZone('UTC')),87            orderId:    $payload['orderId'],88            // Hodnotový objekt uvnitř události se skládá zpět ručně;89            // serializer zná jen skalární payload.90            item:       new OrderItem(91                $payload['item']['productId'],92                $payload['item']['quantity'],93                $payload['item']['unitPriceInCents'],94            ),95        );96    }97 98    public function eventType(): string { return 'ordering.order_item_added'; }99 100    public function schemaVersion(): int { return 1; }101 102    /** @return array<string, mixed> */103    public function toPayload(): array104    {105        return [106            'eventId'    => $this->eventId,107            'occurredAt' => $this->occurredAt->format('Y-m-d H:i:s.u'),108            'orderId'    => $this->orderId,109            'item'       => [110                'productId'        => $this->item->productId,111                'quantity'         => $this->item->quantity,112                'unitPriceInCents' => $this->item->unitPriceInCents,113            ],114        ];115    }116}117 118final class OrderConfirmed extends DomainEvent119{120    private function __construct(121        string $eventId,122        DateTimeImmutable $occurredAt,123        public readonly string $orderId,124    ) {125        parent::__construct($eventId, $occurredAt);126    }127 128    public static function create(string $orderId): self129    {130        return new self(131            eventId:    (string) Uuid::v7(),132            occurredAt: new DateTimeImmutable('now', new \DateTimeZone('UTC')),133            orderId:    $orderId,134        );135    }136 137    /** @param array<string, mixed> $payload */138    public static function fromPayload(array $payload): static139    {140        return new self(141            eventId:    $payload['eventId'],142            occurredAt: new DateTimeImmutable($payload['occurredAt'], new \DateTimeZone('UTC')),143            orderId:    $payload['orderId'],144        );145    }146 147    public function eventType(): string { return 'ordering.order_confirmed'; }148 149    public function schemaVersion(): int { return 1; }150 151    /** @return array<string, mixed> */152    public function toPayload(): array153    {154        return [155            'eventId'    => $this->eventId,156            'occurredAt' => $this->occurredAt->format('Y-m-d H:i:s.u'),157            'orderId'    => $this->orderId,158        ];159    }160}161 162final class OrderShipped extends DomainEvent163{164    private function __construct(165        string $eventId,166        DateTimeImmutable $occurredAt,167        public readonly string $orderId,168        public readonly string $trackingNumber,169    ) {170        parent::__construct($eventId, $occurredAt);171    }172 173    public static function create(string $orderId, string $trackingNumber): self174    {175        return new self(176            eventId:        (string) Uuid::v7(),177            occurredAt:     new DateTimeImmutable('now', new \DateTimeZone('UTC')),178            orderId:        $orderId,179            trackingNumber: $trackingNumber,180        );181    }182 183    /** @param array<string, mixed> $payload */184    public static function fromPayload(array $payload): static185    {186        return new self(187            eventId:        $payload['eventId'],188            occurredAt:     new DateTimeImmutable($payload['occurredAt'], new \DateTimeZone('UTC')),189            orderId:        $payload['orderId'],190            trackingNumber: $payload['trackingNumber'],191        );192    }193 194    public function eventType(): string { return 'ordering.order_shipped'; }195 196    public function schemaVersion(): int { return 1; }197 198    /** @return array<string, mixed> */199    public function toPayload(): array200    {201        return [202            'eventId'        => $this->eventId,203            'occurredAt'     => $this->occurredAt->format('Y-m-d H:i:s.u'),204            'orderId'        => $this->orderId,205            'trackingNumber' => $this->trackingNumber,206        ];207    }208}

Položku agregát drží jako neměnný záznam. Musí být serializovatelná, protože cestuje uvnitř události do Event Store a zpátky:

1<?php2 3declare(strict_types=1);4 5namespace App\Ordering\EventSourced;6 7final readonly class OrderItem8{9    public function __construct(10        public string $productId,11        public int $quantity,12        public int $unitPriceInCents,13    ) {}14}

Model má vlastní namespace App\Ordering\EventSourced, ne App\Ordering\Domain\Model. Důvod je věcný: event-sourced Order je jiný model téhož konceptu než stavově ukládaný agregát z ostatních kapitol. Sdílet jméno i namespace by znamenalo dvě neslučitelné třídy na jednom místě.

Identifikátory jsou zde primitivní řetězce, ne hodnotové objekty jako ve zbytku knihy. Je to druhá záměrná odchylka této kapitoly: událost se serializuje do Event Store a zpět, takže primitivní tvar drží ukázku čitelnou bez vrstvy převodních typů. Ze stejného důvodu nese OrderItem cenu jako int v haléřích místo Money a addItem() přijímá celou položku, ne trojici ProductId, množství a Money jako kanonický agregát. V produkci hodnotové objekty zůstávají, převod obstará serializer.

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

Repozitář event-sourcovaného agregátu nečte tabulku entit. Načte event stream z Event Store a předá ho statické tovární metodě reconstituteFromEvents(). Výsledný stav agregátu přesně odpovídá historii jeho událostí.

FIG. 13.6-A Replay agregátu z event streamu

+ − ⤢ ⛶

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.

FIG. 13.7-A 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 se po uložení do Event Store zařadí 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á.

Projekce jako pozice ve streamu

Asynchronní projekci lépe než „handler, kterému chodí zprávy“ popisuje model catch-up subscription: projekce si drží pozici (checkpoint) ve streamu a od ní čte dál. Odsud plynou tři provozní vlastnosti. Restart projektoru nic nerozbije; pokračuje se od uložené pozice. Rebuild znamená nastavit pozici na nulu a nechat projekci dojet historii. A rozdíl mezi poslední zapsanou událostí a pozicí projekce je lag, jediné číslo, které o zdraví read strany opravdu vypovídá; patří do monitoringu.

Sekce Praktické problémy projekcí používá checkpoint tabulku k idempotenci, tedy k odpovědi na otázku „zpracoval jsem už tuto událost?“. Jde o odvozenou roli téhož záznamu. Primární je pozice. Stejný model pozice ve streamu používají i ságy a Process Managery. Nad event streamem ale neudržují read model, nýbrž rozpracovaný proces (kapitola Ságy a Process Managery).

Read model má vlastní tabulku. Vzniká migrací jako každá jiná, jen se z ní dá kdykoli smazat a postavit znovu z Event Store:

1CREATE TABLE order_summary (2    order_id     VARCHAR(36)  NOT NULL PRIMARY KEY,3    customer_id  VARCHAR(36)  NOT NULL,4    status       VARCHAR(20)  NOT NULL,5    item_count   INT UNSIGNED NOT NULL DEFAULT 0,6    -- Peníze v haléřích; sloupec drží součet za celou objednávku.7    total_amount BIGINT       NOT NULL DEFAULT 0,8    placed_at    DATETIME     NOT NULL,9    shipped_at   DATETIME     NULL,10    tracking_no  VARCHAR(100) NULL,11 12    KEY idx_customer (customer_id, placed_at)13) ENGINE=InnoDB;

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

Výčet tříd v routing je u Event Sourcingu položka, na kterou se zapomíná. Typů událostí rychle přibývá a chybějící řádek se projeví až tím, že projekce mlčí. Od Symfony 7.2 lze routing připsat rovnou k události atributem #[AsMessage('async_events')] a YAML seznam zrušit. Konfigurace se pak nemůže rozejít s doménovým modelem, protože žije v téže třídě.

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.

Odkud se berou samotné typy událostí, je otázka pro doménu, ne pro infrastrukturu. Nejrychleji je odhalí workshop popsaný v kapitole Event Storming a Domain Storytelling: oranžové lístky s doménovými událostmi jsou přímými kandidáty na obsah event streamu.

13.08 Event Store jako outbox#

Projektory z předchozí sekce dostávají události z asynchronní fronty. Mlčky přitom předpokládají, že se událost po zápisu do Event Store spolehlivě dostane do message brokeru. Bez další infrastruktury to 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 je nutné ho doplnit o některou z mitigací: překryv s deduplikací, výběr přes FOR UPDATE SKIP LOCKED, nebo CDC. Implementace relay je shodná s běžným outboxem, ať jde o polling worker pod supervisord, nebo o variantu s CDC; 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í OrderPlaced 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).

Push variantu nad PostgreSQL nemusíte psát sami. Doctrine transport Symfony Messengeru používá na PostgreSQL LISTEN/NOTIFY už od Symfony 5.1; ovládá se volbou use_notify v konfiguraci transportu a ve výchozím stavu je zapnutý. Worker, který z transportu konzumuje, pak nečeká na tik pollingu, ale probudí ho notifikace z databáze. Relay nad event_store tím ale zrychlený není: tam, kde čte tabulku přímo a Messenger v této roli nefiguruje, je potřeba vlastní implementace.

13.09 Praktické problémy projekcí#

Jednoduché ukázky projekcí zamlčují čtyři provozní problémy: 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. Když projektor není idempotentní, opakované zpracování poškodí data: duplicitní řádky, zdvojené částky, nekonzistentní počty.

Projektor níže je varianta OrderSummaryProjector z předchozí sekce, ne přídavek k němu. Kdo opíše obě třídy, dostane na OrderPlaced dva registrované handlery a druhý z nich skončí na UniqueConstraintViolationException. Do aplikace patří jeden z nich.

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.

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 na ně nabízí dva mechanismy:

• Retry transport – zpráva se po selhání automaticky vrátí do fronty s exponenciálním backoffem (výchozí: 3 opakování s násobičem 2, max_delay 0, tedy bez stropu, a jitter 0,1, který opakování rozprostře v čase).
• 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.

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

Při běžném provozu se hodí přepínače, které z těchto příkazů udělají nástroj pro triáž. messenger:failed:show --stats vypíše počty podle tříd zpráv, takže jedna vadná projekce je vidět na první pohled; --class-filter pak omezí výpis (a od Symfony 7.3 i mazání) na jediný typ zprávy. Hromadný úklid po opravené chybě zvládne messenger:failed:remove --all. Pro projektory s dlouho běžícími dávkami existuje od Symfony 7.2 messenger:consume --keepalive, který transportu průběžně hlásí, že se na zprávě pracuje, a zabrání předčasnému redelivery. Při vysokém průtoku událostí snižuje režii --fetch-size (Symfony 8.1), který si z transportu bere zprávy po dávkách místo po jedné.

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.

Příkaz si projektory nenajde sám. AutowireIterator sbírá služby podle tagu a mapu tabulek bere z parametru, takže obojí musí být v konfiguraci:

1parameters:2    app.projection_tables:3        order_summary: 'order_summary'4 5services:6    App\Infrastructure\Ordering\Projection\OrderSummaryProjector:7        tags:8            # 'key' se stane jménem projekce, které příkaz čeká v argumentu9            - { name: 'app.projection', key: 'order_summary' }

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“. Architektura s tím počítá, chybou to není. Strategie pro UI rozebírá sekce Eventual Consistency v praxi v kapitole CQRS: optimistickou aktualizaci, Post-Redirect-Get, polling i synchronní projekce pro kritické cesty.

13.10 Snapshotting#

Event streamy agregátů s věkem systému rostou. Agregát s tisíci událostmi musí při každém commandu načíst a přehrát všechny tyto řádky. Výkonnostní problém se v provozu objeví dřív, než tým čeká.

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. Young pro tuto konstrukci používá název Rolling Snapshot a zdůrazňuje její povahu [2]: jde o denormalizaci stavu agregátu k danému okamžiku, tedy o výkonnostní heuristiku, nikoli o zdroj pravdy. Zdrojem pravdy zůstává event stream a snapshot lze kdykoli zahodit a postavit znovu.

FIG. 13.10-A 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í. Interval v ukázce níže (50) je laditelný parametr, ne doporučení.
• 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.

Snapshotting stojí na dvou metodách agregátu, které ukazuje sekce Order agregát s Event Sourcingem: toSnapshot(): array serializuje aktuální stav, statická reconstituteFromSnapshot(array $state, int $version) ho načte zpět. Na rozdíl od reconstituteFromEvents() nevolá apply*(): properties nastaví přímo ze 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. Protože snímek kopíruje vnitřní stav agregátu, musí se jeho formát vyvíjet spolu s doménovým modelem.

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

Uložená událost zůstává ve své podobě natrvalo. Doménový model se přitom vyvíjí: přibývají atributy, mění se struktura dat, pole se rozdělují nebo slučují. Starou událost pak musí přečíst nový kód.

Tuto zpětnou čitelnost zajišťuje event versioning. 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í. Nejpodrobněji téma zpracovává Young v knize Versioning in an Event Sourced System [8], odkud pochází i většina dnes používané terminologie – weak schema, copy and replace, double write.

O tom, kolik práce verzování obnáší, rozhoduje z velké části rozdělení popsané v sekci Interní a publikované události. U interní události jste jediným konzumentem a stačí upcaster. U publikované měníte smlouvu s cizími týmy a bez přechodného období se neobejdete.

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 transformuje payload události z jedné verze do následující. Upcastery 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.

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

Při spuštění systému obsahovala událost UserRegistered pole fullName (celé jméno jako jeden řetězec). Později se objevil požadavek rozlišit křestní jméno a příjmení, a tak 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.

UpcasterChain se integruje do EventSerializer. Při deserializaci se z uloženého záznamu přečte event_type a schema_version a payload projde řetězem upcasterů. Metoda fromPayload() aktuální třídy události pak dostane už jen transformovaná data.

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 %). To je nepřesné a reporty pak nesou chybná čísla.
• Event splitting. Původní OrderPlaced obsahoval customerData inline. V nové verzi se rozděluje na OrderPlaced + CustomerSnapshotted (samostatný event). Upcaster by musel z prvního eventu vytvořit druhý, a to 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čí, je potřeba transformace nad celým 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.

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

Archivace streamů 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 (primární databáze, ať už MySQL nebo PostgreSQL) – události za posledních 90 dní, dotazy < 10 ms.
• Warm tier (čtecí replika nebo tatáž databáze na pomalejším disku) – události od 90 dní do 2 let. 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 ze všech tří vrstev.