Предпочтительный способ обработки источника событий в рецепте NestJS CQRS

Поэтому я делаю нечто подобное и использую postgres, который действительно поддерживает json ('simple-json') в языке TypeORM (ссылка). К лучшему или худшему, моя сущность события выглядит так:

@Entity()
export class MyEvent {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column()
  name: string;

  @Column('simple-json')
  data: object;

  @CreateDateColumn({type: 'timestamp'})
  created_at: Date;
}

Важно отметить, что я использую свои постоянные события только для контрольного следа и гибкости потенциальных прогнозов, которые я еще не строю. Вы можете абсолютно запросить JSON в postgres, используя TypeORM, например. .where('my_event.data ::jsonb @> :data', {data: {someDataField: 2}}), но я понимаю, что запрос ваших событий для получения текущего состояния как бы упускает суть CQRS. Лучше создавать агрегаты в новых таблицах прогнозов или обновлять один огромный прогноз.

Я в порядке с тем, как я сейчас сохраняю свои события, но это определенно не СУХОЙ. Я бы подумал, что расширение базового класса с помощью общего метода saveEvent или использование класса EventHandlerFactory, который будет использовать репозиторий в своем конструкторе, будет немного чище, чем внедрение репозитория в каждый обработчик.

Может у кого-то есть хорошие мысли?

Во-первых, ваша первоначальная догадка была верной: модуль NestJS CQRS не имеет мнения о том, как вы реализуете Event Sourcing. Причина в том, что CQRS отличается от ES. Хотя вы можете комбинировать их, это совершенно необязательно. Затем, если вы решите использовать ES, снова есть масса способов реализовать.

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

Что касается вашей модели SQL, лучше всего иметь одну таблицу для хранения ваших событий. Очень хорошая статья, демонстрирующая это, - это Хранилище событий в Postgres от Кейси. Speakman.

Выбранный здесь Events вид таблицы выглядит следующим образом:

CREATE TABLE IF NOT EXISTS Event
(
    SequenceNum bigserial NOT NULL,
    StreamId uuid NOT NULL,
    Version int NOT NULL,
    Data jsonb NOT NULL,
    Type text NOT NULL,
    Meta jsonb NOT NULL,
    LogDate timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (SequenceNum),
    UNIQUE (StreamId, Version),
    FOREIGN KEY (StreamId)
        REFERENCES Stream (StreamId)
);

В статье дается четкое описание обоснования для каждого из столбцов, но вы можете построить свои агрегаты, используя запрос, основанный на StreamId + Version. Столбец Meta может содержать метаданные, например userId и correlationId (здесь подробнее о корреляции) и т. д. В статье также упоминается, как можно создавать снимки, что в некоторых случаях может быть удобно (но избегайте, пока не понадобится).

Обратите внимание на столбец Type, в котором хранится тип события и который может использоваться для целей десериализации (поэтому нет необходимости создавать собственный ORM;)

Другие проекты, которые показывают, как реализовать хранилище событий, - это PostgreSQL Event Sourcing и более полный решение message-db.

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

Сами события представляют собой простые классы, например:

class FooHappened {
  constructor(
    readonly root: string;
    readonly bar: string;
  ) {}
}

Я использовал свойство root для совокупного корня ObjectId для построения моделей чтения, и до сих пор это работало хорошо.