@application-kit/transaction-outbox

Transactional outbox на TypeORM и NestJS: события пишутся в таблицу transaction_outboxes в одной транзакции с изменением и тут же удаляются — Debezium читает вставку из WAL и по полю queue_name раскладывает событие по топикам Kafka (EventRouter, route.by.field: queue_name).

Установка

yarn add @application-kit/transaction-outbox

Peer-зависимости: @application-kit/kafka (вход /event-contract), @application-kit/typeorm, @libs-for-dev/nestjs-ddd-library, @nestjs/common, @nestjs/core, @nestjs/typeorm, nestjs-pino, oxide.ts, typeorm. kafkajs и @nestjs/microservices не нужны: пакет берёт из kafka только правила версии контракта.

Подключение

Приложение приносит два токена — глобально (например, из корневого модуля): TransactionOutboxModule импортируется статически в каждый модуль-потребитель и опций не принимает.

import {
  TRANSACTION_OUTBOX_LOGGER,
  TRANSACTION_OUTBOX_REQUEST_FACTS,
  TransactionOutboxLogger,
} from '@application-kit/transaction-outbox'

@Global()
@Module({
  exports: [TRANSACTION_OUTBOX_LOGGER, TRANSACTION_OUTBOX_REQUEST_FACTS],
  providers: [
    { provide: TRANSACTION_OUTBOX_LOGGER, useClass: TransactionOutboxLogger },
    { provide: TRANSACTION_OUTBOX_REQUEST_FACTS, useValue: () => ({ ip: currentIp() }) },
  ],
})
class CommonModule {}

• TRANSACTION_OUTBOX_LOGGER — любой класс с error(context, message); TransactionOutboxLogger — реализация на nestjs-pino (нужен глобальный LoggerModule).
• TRANSACTION_OUTBOX_REQUEST_FACTS — хук обогащения: функция без аргументов, результат которой ложится в payload.facts событий с carriesRequestFacts: true. Пакет не знает, что такое факты.

Запись событий

const result = await outbox.save([new UserWasCreatedEvent({ id })], async () => {
  return repository.save(user)
})

Колбэк может ничего не возвращать: это штатный успех, save отвечает Ok(true). Result нужен, когда у работы есть ожидаемый отказ.

События пишутся до колбэка в той же транзакции. Err из колбэка откатывает её наравне со сбоем — иначе Debezium разослал бы потребителям факт о невыполненной работе; сам отказ возвращается значением. save внутри колбэка другого save присоединяется к открытой транзакции. Репозитории, которые участвуют в транзакции, регистрируются через TransactionAwareTypeOrmModule.forFeature([...]) вместо TypeOrmModule.forFeature.

Версия контракта

Имя очереди собирается versionedQueueName(queueName, eventVersion) из @application-kit/kafka/event-contract: у первой версии имя прежнее, у остальных — <queue>.v<N>. Базовый класс AbstractSingleIdOutboxEvent фиксирует версию 1.

Миграция

Пакет экспортирует orm — сущность TransactionOutboxEntity и миграцию TransactionOutbox1734780489810 классами: передайте его в packages фабрики CLI DataSource. Имя класса миграции записано в таблице migrations уже мигрированных баз — переименование заставит TypeORM прогнать миграцию повторно.

import { orm as outboxOrm } from '@application-kit/transaction-outbox'

createCliDataSource({ ..., packages: [outboxOrm] }) // @application-kit/typeorm/nest

Тесты потребителя

import { TransactionOutboxFake } from '@application-kit/transaction-outbox/testing'

const outbox = new TransactionOutboxFake(logger)
await service.run()
expect(outbox.events.size).toBe(1)

Фейк копит события в events и откатывает их на Err и исключении колбэка, как настоящая транзакция.