Порождение событий, CQRS и Laravel

Перевод статьи подготовлен для студентов профессионального курса «Framework Laravel»

---

Введение

Эта статья посвящена основам создания событийных CQRS-систем на языке PHP и в фреймворке Laravel. Предполагается, что вы знакомы со схемой разработки с использованием командной шины и имеете представление о событиях (в частности, о публикации событий для массива слушателей). Чтобы освежить эти знания, вы можете воспользоваться сервисом Laracasts. Кроме того, предполагается, что вы имеете определенное представление о принципе CQRS. Если же нет, я настоятельно рекомендую прослушать две лекции: «Практикум по порождению событий» Матиаса Верраса (Mathias Verraes) и «CQRS и порождение событий» Грега Янга (Greg Young).

Не используйте приведенный здесь код в своих проектах! Он представляет собой обучающую платформу, позволяющую понять идеи, лежащие в основе CQRS. Этот код нельзя назвать надежным, он плохо протестирован, и кроме того, я редко программирую интерфейсы, поэтому будет гораздо труднее менять отдельные части кода. Гораздо лучший пример CQRS-пакета, которым вы можете воспользоваться — Broadway, разработанный Qandidate Lab. Это чистый, слабосвязанный код, правда, некоторые абстракции делают его не совсем понятным, если вы никогда не сталкивались с событийными системами.

И последнее — мой код связан с событиями и командной шиной Laravel. Я хотел посмотреть, как будет выглядеть код в Laravel (я обычно использую этот фреймворк для проектов небольших агентств), однако, оглядываясь назад, я думаю, что мне следовало бы создать собственные реализации. Я надеюсь, мой код будет понятен даже тем, кто не использует фреймворки.

На github код расположен по адресу https://github.com/scazz/cqrs-tutorial.git, и в нашем руководстве мы рассмотрим его составляющие по нарастанию логики.

Мы создадим начальный вариант системы регистрации для школы серфинга. С ее помощью клиенты школы могут записываться на занятия. Для процесса записи мы формулируем следующие правила:

• На каждом занятии должен быть хотя бы один клиент...
• … но не более трех.

Одна из самых впечатляющих возможностей событийных CQRS-систем — это создание моделей чтения, специфичных для каждого показателя, требуемого от системы. Вы найдете примеры проекции моделей чтения в ElasticSearch, а Грег Янг (Greg Young) внедрил в свое хранилище событий предметно-ориентированный язык для обработки сложных событий. Однако для простоты наша проекция чтения будет представлять собой стандартную базу данных SQL для использования с Eloquent. В итоге у нас будет одна таблица для занятий и одна для клиентов.

Концепция порождения событий также позволяет обрабатывать события в офлайн-режиме. Но в данной статье я буду по максимуму придерживаться «традиционных» моделей разработки (опять же для упрощения), и наши проекции чтения будут обновляться в режиме реального времени, сразу после сохранения событий в хранилище.

Настройка проекта и первый тест

git clone https://github.com/scazz/cqrs-tutorial

Создайте новый проект Laravel 5

$> laravel new cqrs-tutorial

И для начала нам нужен тест. Воспользуемся интеграционным тестом, который позволит убедиться, что запись клиента на занятия приводит к тому, что занятие создается в нашей модели Eloquent.

Листинг tests/CQRSTest.php:

use Illuminate\Foundation\Bus\DispatchesCommands;
class CQRSTest extends TestCase {
use DispatchesCommands;

/**
 * Проверка того, что команда BookLesson создает занятие в нашей проекции чтения
 * @return void
 */
public function testFiringEventUpdatesReadModel()
{
$testLessonId = '123e4567-e89b-12d3-a456-426655440000';
$clientName = "George";
$lessonId = new LessonId($testLessonId);
	$command = new BookLesson($lessonId, $clientName);
	$this->dispatch($command);

	$this->assertNotNull(Lesson::find($testLessonId));
	$this->assertEquals( Lesson::find($testLessonId)->clientName, $clientName );
}
}

Мы заранее присваиваем новому занятию ID, создаем команду для записи на новое занятие и указываем Laravel на необходимость ее отправки. В таблице занятий нам нужно создать новую запись, которую мы сможем прочитать с использованием модели Eloquent. Нам потребуется база данных, поэтому заполните свой файл .env должным образом.

Каждое событие, регистрируемое в нашем хранилище событий, присоединяется к корню агрегата, который мы будем называть просто сущностью (Entity) — абстракция в учебных целях только добавляет путаницы. ID — это универсальный уникальный идентификатор (UUID). Хранилищу событий все равно, применяется ли событие к занятию (Lesson) или клиенту (Client). Ему только известно, что оно связано с ID.

На основании ошибок, выявленных в ходе тестирования, мы можем создать отсутствующие классы. Сначала мы создадим класс LessonId, затем команду BookLesson (не беспокойтесь пока о методе обработчика, а просто продолжайте выполнять тест). Класс Lesson представляет собой модель чтения за пределами пространства имен Lesson. Исключительно модель чтения — здесь никогда не будет храниться логика предметной области. В завершение нам потребуется создать миграцию для таблицы занятий.

Для сохранения наглядности кода я воспользуюсь библиотекой проверки утверждений. Ее можно добавить следующей командой:

$> composer require beberlei/assert

Рассмотрим процесс, который должен инициироваться этой командой:

1. Валидация: императивные команды могут завершиться неудачей, а события уже произошли и поэтому завершаться неудачей не должны.
2. Создайте новое событие LessonWasBooked (на занятие записались).
3. Обновите состояние занятия. (Модель записи должна быть осведомлена о состоянии модели, чтобы она могла выполнить валидацию.)
4. Добавьте это событие в поток незафиксированных событий, хранящихся в модели записи занятия.
5. Сохраните поток незафиксированных событий в хранилище.
6. Инициируйте событие LessonWasBooked глобально для информирования всех проекторов чтения о необходимости обновить таблицу занятий.

Сначала необходимо создать модель записи для занятия. Воспользуемся статическим фабричным методом Lesson::bookClientOntoNewLesson(). Он генерирует новое событие LessonWasOpened (занятие открыто), применяет это событие к самому себе (просто устанавливает свой ID), добавляет новое событие в список незафиксированных событий в виде DomainEventMessage (событие плюс некоторые метаданные, которые мы используем при сохранении в хранилище событий).

Процесс повторяется для добавления клиента к событию. При применении события ClientWasBookedOntoLesson (клиент был записан на занятие) модель записи не отслеживает имена клиентов, а только лишь количество записавшихся клиентов. Модели записи не нужно знать имен клиентов для обеспечения неизменности.

Методы applyLessonWasOpened и applyClientWasBookedOntoLesson на данный момент могут показаться немного странными. Мы воспользуемся ими позднее, когда нам понадобится воспроизвести старые события, чтобы сформировать состояние модели записи. Объяснить это непросто, поэтому приведу код, который поможет разобраться в этом процессе. Позднее мы извлечем код, обрабатывающий незафиксированные события (uncommittedEvents) и создающий сообщения о событиях предметной области.

app/School/Lesson/Lesson.php
public function openLesson( LessonId $lessonId ) {
   	 /* здесь следовало бы проверить любые инварианты, но нам не нужно их защищать, поэтому мы можем просто генерировать события */ 
   	 $this->apply(
   		 new LessonWasOpened( $lessonId)
   	 );
    }
    protected function applyLessonWasOpened( LessonWasOpened $event ) {
   	 $this->lessonId = $event->getLessonId();
   	 $this->numberOfClients = 0;
    }
    public function bookClient( $clientName ) {
   	 if ($this->numberOfClients >= 3) {
   		 throw new TooManyClientsAddedToLesson();
   	 }
   	 $this->apply(
   		 new ClientBookedOntoLesson( $this->lessonId, $clientName)
   	 );
    }
/** 
*   Здесь мы просто отслеживаем количество клиентов — 
*   это единственное, что имеет значение для модели записи 
*   Если бы правило предметной области гласило, что клиенты не могут иметь одинаковые имена, 
* нам бы пришлось отслеживать имена клиентов. 
*/
    protected function applyClientBookedOntoLesson( ClientBookedOntoLesson $event ) {
   	 $this->numberOfClients++;
    }

Мы можем извлечь компоненты CQRS из нашей модели записи — фрагменты класса, участвующие в обработке незафиксированных событий. Мы также можем очистить API для сущности, порожденной на основании событий, создав защищенную функцию apply(), которая принимает событие, вызывает соответствующий метод applyEventName() и добавляет новое событие DomainEventMessage в список незафиксированных событий. Извлеченный класс является деталью реализации CQRS и не содержит логики предметной области, поэтому мы можем создать новое пространство имен: App\CQRS:

Обратите внимание на код app/CQRS/EventSourcedEntity.php
Чтобы код заработал, нам нужно добавить класс DomainEventMessage, который является простым DTO — его можно найти в app/CQRS/DomainEventMessage.php

Таким образом, мы получили систему, которая генерирует события для каждой попытки записи и использует события для записи изменений, необходимых для предотвращения инвариантов. Следующим шагом является сохранение этих событий в хранилище (EventStore). Прежде всего это хранилище событий надо создать. Для упрощения будем использовать модель Eloquent, простую таблицу SQL со следующими полями: *UUID (чтобы знать, к какой сущности применять событие) *event_payload (сериализованное сообщение, содержащее все необходимое для воссоздания события) *recordedAt — метка времени, чтобы знать, когда событие произошло. Если вы внимательно просмотрите код, то увидите, что я создал две команды — для создания и уничтожения нашей таблицы хранилища событий:

• php artisan eloquenteventstore:create (App\CQRS\EloquentEventStore\CreateEloquentEventStore)
• php artisan eloquenteventstore:drop (App\CQRS\EloquentEventStore\DropEloquentEventStore) (не забудьте добавить их в App\Console\Kernel.php, чтобы они загрузились).

Есть две очень веские причины не использовать SQL в качестве хранилища событий: она не реализует модель append-only (только добавление данных, события должны быть неизменяемыми), а также потому, что SQL не является идеальным языком запросов к темпоральным базам. Мы программируем интерфейс, чтобы облегчить замену хранилища событий в последующих публикациях.

Для сохранения событий воспользуемся репозиторием. Всякий раз, когда вызывается save() для модели записи, мы сохраняем список uncommittedEvents в хранилище событий. Для хранения событий нам потребуется механизм их сериализации и десериализации. Создадим для этого сериализатор (Serializer). Нам потребуются метаданные, такие как класс события (например, App\School\Lesson\Events\LessonWasOpened) и полезная нагрузка события (данные, необходимые для реконструкции события).

Все это будет закодировано в формате JSON, а затем записано в нашу базу данных вместе с UUID сущности и меткой времени. Мы хотим обновлять наши модели чтения после фиксации событий, поэтому репозиторий будет инициировать каждое событие после сохранения. Serializer будет отвечать за запись класса события, в то время как событие — за сериализацию своей полезной нагрузки. Полностью сериализованное событие будет выглядеть примерно следующим образом:

 { class: "App\\School\\Lesson\\Events\\", event: $event->serialize() }  

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

app/School/Lesson/Events/LessonWasOpened.php
 class LessonWasOpened implements SerializableEvent { 
public function serialize() 
{ 
    return array( 'lessonId'=> (string) $this->getLessonId() ); 
} 
}  

Создайте репозиторий LessonRepository. Мы можем выполнить рефакторинг и извлечь базовые компоненты CQRS позже.

app/School/Lesson/LessonRepository.php

eventStoreRepository = new EloquentEventStoreRepository( new EventSerializer() ); } public function save(Lesson $lesson) { /** @var DomainEventMessage $domainEventMessage */ foreach( $lesson->getUncommittedDomainEvents() as $domainEventMessage ) { $this->eventStoreRepository->append( $domainEventMessage->getId(), $domainEventMessage->getEvent(), $domainEventMessage->getRecordedAt() ); Event::fire($domainEventMessage->getEvent()); } } } 

Если вы снова запустите интеграционный тест, а затем проверите SQL-таблицу domain_events, вы должны увидеть в базе данных два события. 

Наш последний шаг к успешному прохождению теста — прослушивание транслируемых событий и обновление проекции модели чтения Lesson. Транслируемые события Lesson будут перехватываться проектором LessonProjector, который применит необходимые изменения к LessonProjection (Eloquent-модели таблицы занятий):

Листинг app/School/Lesson/Projections/LessonProjector.php
class LessonProjector {
    public function applyLessonWasOpened( LessonWasOpened $event ) {
   	 $lessonProjection = new LessonProjection();
   	 $lessonProjection->id = $event->getLessonId();
   	 $lessonProjection->save();
    }
    public function subscribe(Dispatcher $events) {
   	 $fullClassName = self::class;
   	 $events->listen( LessonWasOpened::class, $fullClassName.'@applyLessonWasOpened');
    }
}
Листинг app/School/Lesson/Projections/LessonProjection.php
class LessonProjection extends Model {
    public $timestamps = false;
    protected $table = "lessons";
}

Если вы запустите тест, то увидите, что произошла ошибка SQL:

Unknown column 'clientName' in 'field list'

Как только мы создадим миграцию для добавления clientName в таблицу занятий, мы успешно пройдем тест. Мы реализовали базовую функциональность CQRS: команды создают события, которые используются для генерации моделей чтения. 

Усовершенствование модели чтения с помощью связей

Мы достигли знаменательной вехи, но это еще не все! Пока модель чтения поддерживает только одного клиента (мы же указали в наших правилах предметной области трех). Изменения, которые мы вносим в модель чтения, довольно просты: мы просто создаем модель проекции Client и проектор ClientProjector, который отлавливает событие ClientBookedOntoLesson. Сначала обновим наш тест, чтобы отразить изменения, которые мы хотим видеть в нашей модели чтения:

tests/CQRSTest.php
public function testFiringEventUpdatesReadModel()
    {
   	 $lessonId = new LessonId( (string) \Rhumsaa\Uuid\Uuid::uuid1() );
   	 $clientName = "George";
   	 $command = new BookLesson($lessonId, $clientName);
   	 $this->dispatch($command);
   	 $lesson =  Lesson::find( (string) $lessonId);
   	 $this->assertEquals( $lesson->id, (string) $lessonId );
   	 $client = $lesson->clients()->first();
   	 $this->assertEquals($client->name, $clientName);
    } 

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

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

В нашей модели записи в данный момент отсутствует возможность загрузки текущего состояния. Если мы захотим добавить второго клиента к занятию, то можем просто создать второе событие ClientWasAddedToLesson, однако не сможем обеспечить защиту от инвариантов. Для большей наглядности предлагаю написать второй тест, имитирующий запись двух клиентов на занятие.

tests/CQRSTest.php
	public function testLoadingWriteModel()
    {
   	 $lessonId = new LessonId( (string) \Rhumsaa\Uuid\Uuid::uuid1() );
   	 $clientName_1 = "George";
   	 $clientName_2 = "Fred";
   	 $command = new BookLesson($lessonId, $clientName_1);
   	 $this->dispatch($command);
   	 $command = new BookClientOntoLesson($lessonId, $clientName_2);
   	 $this->dispatch($command);
   	 $lesson =  Lesson::find( (string) $lessonId );
   	 $this->assertClientCollectionContains($lesson->clients, $clientName_1);
   	 $this->assertClientCollectionContains($lesson->clients, $clientName_2);
    }

Для нашей модели записи требуется реализовать способ «загрузки» сущности, для которой в хранилище событий уже имеются применимые к ней события. Мы можем добиться этого путем воспроизведения каждого события, которое ссылается на UUID сущности. В общих чертах процесс выглядит следующим образом:

1. Получаем все релевантные сообщения о событиях от хранилища событий.
2. Для каждого сообщения воссоздаем соответствующее событие.
3. Создаем новую модель записи сущности и воспроизводим каждое событие.

В данный момент наши тесты генерируют исключения, поэтому мы начнем с создания необходимой команды BookClientOntoLesson (записать клиента на занятие), используя команду BookLesson в качестве шаблона. Метод обработчика будет выглядеть следующим образом:

Листинг app/School/Lesson/Commands/BookClientOntoLesson.php
public function handle(LessonRepository $repository) {
   	 /** @var Lesson $lesson */
   	 $lesson = $repository->load($this->lessonId);
   	 $lesson->bookClient($this->clientName);
   	 $repository->save($lesson);
    }

Добавим событие загрузки в репозиторий занятий:
app/School/Lesson/LessonRepository.php
public function load(LessonId $id) {
   	 $events = $this->eventStoreRepository->load($id);
   	 $lesson = new Lesson();
   	 $lesson->initializeState($events);
   	 return $lesson;
    }

Функция загрузки репозитория возвращает массив воссозданных событий. Для этого она сначала находит сообщения о событиях в хранилище, а затем передает их в Serializer для преобразования каждого сообщения в событие. Serializer создает сообщения из событий, поэтому нам нужно добавить метод deserialize() для выполнения обратного преобразования. Вспомним, что Serializer передается каждому событию для выполнения сериализации данных события (например, имени клиента). То же самое мы сделаем и для выполнения обратного преобразования, при этом наш интерфейс SerializableEvent должен быть обновлен при помощи метода deserialize(). Давайте рассмотрим код, чтобы все встало на свои места. Сначала функцию загрузки EventStoreRepository:

app/CQRS/EloquentEventStore/EloquentEventStoreRepository.php
public function load($uuid) {
   	 $eventMessages = EloquentEventStoreModel::where('uuid', $uuid)->get();
   	 $events = [];
   	 foreach($eventMessages as $eventMessage) {
   		 /* Мы сериализовали наше событие в сообщение event_payload, поэтому перед возвращением данных нам нужно выполнить десериализацию. */
   		 $events[] = $this->eventSerializer->deserialize( json_decode($eventMessage->event_payload));
   	 }
   	 return $events;
    }

С помощью соответствующей функции десериализации в eventSerializer:

app/CQRS/Serializer/EventSerializer.php
public function serialize( SerializableEvent $event ) {
   	 return array(
   		 'class'   => get_class($event),
   		 'payload' => $event->serialize()
   	 );
    }
    public function deserialize( $serializedEvent ) {
   	 $eventClass = $serializedEvent->class;
   	 $eventPayload = $serializedEvent->payload;
   	 return $eventClass::deserialize($eventPayload);
    }

В заключение воспользуемся статическим фабричным методом deserialize() в LessonWasOpened (нам нужно добавлять этот метод к каждому событию)

app/School/Lesson/Events/LessonWasOpened.php
public static function deserialize($data) {
   	 $lessonId = new LessonId($data->lessonId);
   	 return new self($lessonId);
    } 

Теперь у нас есть массив всех произошедших событий, которые мы просто воспроизводим относительно нашей модели записи Entity для инициализации состояния в методе initializeState в app/CQRS/EventSouredEntity.php

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

tests/CQRSTest.php
	public function testMoreThan3ClientsCannotBeAddedToALesson() {
   	 $lessonId = new LessonId( (string) \Rhumsaa\Uuid\Uuid::uuid1() );
   	 $this->dispatch( new BookLesson($lessonId, "bob") );
   	 $this->dispatch( new BookClientOntoLesson($lessonId, "george") );
   	 $this->dispatch( new BookClientOntoLesson($lessonId, "fred") );
   	 $this->setExpectedException( TooManyClientsAddedToLesson::class );
   	 $this->dispatch( new BookClientOntoLesson($lessonId, "emma") );
    }

Обратите внимание, что нам требуется только lessonId — этот тест повторно инициализирует состояние урока при выполнении каждой команды. 

На данный момент мы просто передаем созданные вручную UUID, тогда как на самом деле мы хотим генерировать их автоматически. Я собираюсь использовать пакет Ramsy\UUID, поэтому давайте установим его при помощи composer:

$> composer require ramsey/uuid

А теперь обновим наши тесты, чтобы использовать новый пакет:

tests/CQRSTest.php
	public function testEntityCreationWithUUIDGenerator() {
   	 $lessonId = new LessonId( (string) \Rhumsaa\Uuid\Uuid::uuid1() );
   	 $this->dispatch( new BookLesson($lessonId, "bob") );
   	 $this->assertInstanceOf( Lesson::class, Lesson::find( (string) $lessonId)  );
    }

Теперь новый разработчик проекта может посмотреть на код, увидеть App\School\ReadModels, где содержится набор моделей Eloquent, и использовать эти модели для записи изменений в таблицу занятий. Мы можем помешать этому, создав класс ImmutableModel, который расширяет класс Eloquent Model и переопределяет метод сохранения в app/CQRS/ReadModelImmutableModel.php.