Практика: публикация событий после сохранения (Publishing domain events) ​

Подписка меняет тариф, биллинг и почта узнают об этом из событий, но только если изменение действительно записано в базу: событие о том, чего нет в базе, хуже, чем никакого.

Базовый пример: Доменное событие. Там агрегат копит события и отдаёт их releaseEvents, но не видно, кто их забирает, когда публикует и что будет, если сохранение упадёт. Здесь это делает сервис приложения: загрузить, изменить, сохранить, опубликовать, и именно в таком порядке.

Задача ​

Сервис подписок: клиент меняет тариф или отменяет подписку, а биллинг пересчитывает счёт, почта шлёт письмо, аналитика пишет в воронку. Первая версия публиковала событие прямо из агрегата в момент изменения. Однажды база отказала на записи: изменение откатилось, а письмо «вы перешли на тариф pro» уже ушло и биллинг выставил новый счёт. Клиент платит за тариф, которого у него нет.

Вторая беда — порядок. Клиент дважды за минуту поменял тариф, обработчик биллинга получил события в обратном порядке и оставил старый тариф. Нужен способ понять, какое событие новее, не полагаясь на часы разных машин.

Что урезано ​

• База данных — InMemorySubscriptions со словарём «id -> тариф, статус, версия». Отказ базы имитирует обёртка в тестах, которая падает на каждом save.
• Нет транзакций и оптимистической блокировки: версия хранится, но не сверяется при записи.
• Диспетчер синхронный и без обработки ошибок подписчиков: это подробно разобрано в практике наблюдателя.
• Подписка уже существует в базе (тариф basic, версия 1); оформление новой подписки опущено.
• Часы — функция или интерфейс с одним методом; в тестах они шагают на минуту при каждом вызове.

Решение ​

События PlanChanged и SubscriptionCancelled несут три общих поля: id подписки, версию агрегата после изменения и время. По версии получатель понимает порядок и замечает пропуск, по id находит, чего касается факт.

Агрегат Subscription сам увеличивает версию при каждом записанном событии и отказывает, если операция запрещена: отменённая подписка не меняет тариф и не отменяется второй раз. Отказ — исключение или ошибка до записи события, поэтому отказанная операция не оставляет следов.

Диспетчер EventDispatcher подписывает обработчики на конкретный тип события и вызывает их в порядке событий.

Сервис приложения SubscriptionService связывает всё в одном методе handle: загрузить агрегат из хранилища Subscriptions, вызвать доменный метод с временем из часов Clock, сохранить и только потом опубликовать то, что вернул releaseEvents.

Структура ​

Агрегат хранит версию рядом с состоянием, а события — отдельно, только до releaseEvents. Хранилище сохраняет тариф, статус и версию, но не события: после get агрегат собирается заново (Subscription::restore в PHP, RestoreSubscription в Go, конструктор в Python и TS), и его список событий пуст. Версия увеличивается в момент записи факта (nextMeta в Go и TS, ++$this->version в PHP, self.version += 1 в Python), поэтому версии событий идут подряд: подписка версии 1 после двух изменений даёт события версий 2 и 3.

Публикация после сохранения ​

Сервис держит порядок шагов в одном месте, handle:

1. get — загрузить агрегат; если подписки нет, дальше не идём.
2. Доменный метод с clock.now(); если агрегат отказал, исключение выходит из handle до сохранения, и публиковать нечего.
3. save; если база упала, исключение выходит из handle, и события остаются в агрегате, который никто больше не увидит.
4. publish(releaseEvents()) — только сюда доходят события, чьё изменение уже в базе.

Время берётся из внедряемых часов, а не из системных: в тестах события сравниваются целиком, с точным occurredAt. Подмена хранилища в тестах (обёртка, которая читает как обычно и падает на save) возможна потому, что сервис зависит от интерфейса Subscriptions, а не от InMemorySubscriptions.

Как это собирается ​

При старте приложения собираются хранилище, часы и EventDispatcher, и на него подписываются три слушателя: биллинг на PlanChanged, почта на SubscriptionCancelled, журнал аудита на оба. В приложении это делает контейнер, в примере - несколько строк в начале. Все они получают один SubscriptionService, и каждый запрос - это один вызов его метода: обработчик ничего не знает ни о слушателях, ни о версиях.

Запросы к sub-1 проходят: после каждого слушатели получают ровно те события, что сохранились, с версиями 2 и 3 подряд. sub-2 уже отменена, агрегат отказывает до записи события, обработчик отвечает 409, и никто из слушателей не вызван. Подписки sub-9 нет: get падает раньше доменного метода, и это 404. Часы в примере фиксированные, в приложении на их месте системное время.

Сценарий целиком с выводом: пример на PHP, он же на Go, Python и TypeScript в панели кода.

Реализации ​

Диспетчер знает только типы событий: подписка — это пара «тип -> обработчик», публикация проходит события по порядку и для каждого вызывает обработчики его типа. Обработчик биллинга, подписанный на PlanChanged, не получает SubscriptionCancelled.

Запуск: make test T=patterns/enterprise/domain-event/practice

Что проверяют тесты ​

Тесты одинаковы во всех языках. В хранилище подписка sub-1 на тарифе basic версии 1, часы начинают с 10:00 и шагают на минуту. Пять случаев:

• Событие несёт id, версию и время: смена тарифа публикует ровно PlanChanged(sub-1, 2, 10:00, basic -> pro), в базе версия 2.
• Изменения подряд: pro, team, отмена дают события версий 2, 3, 4 со временем 10:00, 10:01, 10:02, именно в этом порядке.
• Сохранение упало: сервис бросает database is down, опубликовано ничего, в базе по-прежнему basic.
• Агрегат отказал: после отмены смена тарифа падает с subscription sub-1 is already cancelled, опубликовано только событие отмены, версия в базе осталась 2.
• Подписка по типу: обработчик биллинга на PlanChanged получает basic -> pro и не видит отмену.

Где граница ​

• Публикация после сохранения закрывает «событие без изменения», но оставляет обратную дыру: база записала, а процесс упал до publish, и событие потеряно навсегда. Если событие обязано дойти (счёт, письмо об оплате), его сохраняют в базу в той же транзакции, что и агрегат, а доставляет отдельный процесс. Это Outbox: доменное событие отвечает за «что случилось», outbox — за «не потерять».
• Наблюдатель из практики про доменные события заказа решает другую задачу: кому и как разослать уже готовое событие, что делать с упавшим подписчиком и каскадом. Здесь вопрос — когда событие имеет право уйти. В проекте нужны оба: сервис решает «когда», диспетчер — «кому».
• Версия здесь только помечает события. Если два запроса загрузили подписку версии 1 одновременно, оба сохранят версию 2 и оба опубликуют. Следующий шаг — оптимистическая блокировка: save сверяет ожидаемую версию и отказывает при расхождении.
• Если реакций нет или она одна и синхронная, сервис может вызвать её напрямую после save: событие и диспетчер окупаются, когда реакций несколько и они принадлежат разным командам.

Отличия между языками ​

• PHP — readonly класс и свойства в интерфейсе (PHP 8.4) дают неизменяемое событие без геттеров; агрегат открывает состояние через public private(set). Шаг сервиса — Closure, его тип описан для PHPStan в @param Closure(Subscription, DateTimeImmutable): void. Обработчики хранятся как Closure без параметра типа: Closure(PlanChanged) не подтип Closure(SubscriptionEvent), а тип связывает @template в subscribe.
• Go — исключений нет, поэтому каждый шаг handle проверяет error, и порядок «отказ, сохранение, публикация» виден в коде буквально. Ошибка базы обёрнута fmt.Errorf("save subscription %s: %w", ...), тест находит её через errors.Is. Подмена хранилища в тесте — структура со встроенным *InMemorySubscriptions и своим Save.
• Python — хранилище описано Protocol, и BrokenSubscriptions подходит без наследования. Часы — любой вызываемый объект; в тестах это next по бесконечному генератору. Поля агрегата открыты на запись по соглашению языка: менять их снаружи не принято, но запрета нет.
• TypeScript — состояние агрегата закрыто приватными полями и открыто геттерами, иначе subscription.plan = "pro" меняло бы тариф без события. Обработчик хранится как (event: SubscriptionEvent) => void через приведение типа: безопасно, потому что диспетчер вызывает его только для событий своего type.

Связанные темы ​

• Доменное событие (базовый пример)
• Практика Наблюдателя: доменные события заказа — доставка уже готовых событий: ошибки подписчиков, отписка во время рассылки, каскад
• Практика Состояния: жизненный цикл заказа — правила переходов; запрещённый переход так же отказывает, не меняя состояния