Command side в CQRS (NestJS) — write-handler через агрегат и TransactionRunner

Опирается на правила: R-CQRS-CMD-1 … R-CQRS-CMD-5 и R-CQRS-CMD-X1 … R-CQRS-CMD-X3 из CQRS контракта → раздел 2. Command side.

Важно знать

• Command — это намерение изменить состояние. Класс с readonly-полями, реализует Command<R> из core/usecase.ts. Без вычислений в конструкторе.
• Один command — один агрегат. Если меняется два — это либо saga, либо границы агрегатов нарезаны неверно.
• Command-handler: запускает tx.run(async () => { ... }) через TransactionRunner; грузит агрегат; вызывает доменный метод; сохраняет.
• Возвращает минимум: OrderId, string, пустой результат. Никаких полных read-DTO.
• Read внутри command — только в рамках load-aggregate одним вызовом byId. Отдельный запрос «прочитать и решить» — антипаттерн.
• Валидация: контракт DTO через class-validator на request-объекте; бизнес-инварианты — внутри агрегата (бросает domain exception).
• Изменение нескольких агрегатов — через saga, не через один tx.run с двумя save.

Command — это намерение пользователя или системы изменить состояние. В CQRS это пишущая половина: всё, что может что-то поменять, едет через command-handler. Граница чёткая: если handler открывает транзакцию для записи — это command, и для него действуют другие правила, чем для query. Статья раскрывает раздел 2 CQRS-контракта в идиомах NestJS/TypeScript.

Command — класс с маркером Command<R>

R-CQRS-CMD-1: command — класс с readonly-полями, который реализует Command<R> из core/usecase.ts.

// core/order/command/confirm-order.command.ts
import { Command } from '@core/usecase';

export class ConfirmOrder implements Command<OrderId> {
  constructor(
    readonly orderId: OrderId,
    readonly idempotencyKey: string,
  ) {}
}

Что важно:

• Только readonly-поля, без методов. Command — данные, не логика. Маппинг из request-DTO происходит в контроллере (см. Use Case Pattern).
• Параметр типа <R> — возвращаемое значение handler-а. Здесь OrderId. Что именно возвращать — см. ниже (R-CQRS-CMD-4).
• idempotencyKey — обычное поле для денежных и других неидемпотентных операций. Контроллер берёт его из заголовка Idempotency-Key.
• Нет @IsNotEmpty и прочих декораторов. Command — внутренний объект core; контракт входа проверяется на request-DTO ещё до создания command.

// adapters/in/http/order.controller.ts
@Post(':id/confirm')
async confirm(
  @Param('id', ParseUUIDPipe) id: string,
  @Headers('idempotency-key') key: string,
): Promise<{ id: string }> {
  const result = await this.handler.execute(
    new ConfirmOrder(OrderId.of(id), key),
  );
  return { id: result.value };
}

Контроллер не содержит бизнес-логики — только маппинг параметров запроса в command и обратно.

Command меняет один агрегат

R-CQRS-CMD-2: одна команда — одно изменение одного агрегата. Если business-логика требует затронуть два — что-то не так:

• Либо это saga: распределённая последовательность из нескольких локальных команд с компенсациями.
• Либо границы агрегатов нарезаны неверно: два объекта, которые всегда меняются вместе, — это один агрегат.

// ПЛОХО — два агрегата в одной транзакции
async execute(cmd: CreateOrder): Promise<OrderId> {
  return this.tx.run(async () => {
    const customer = await this.customers.byId(cmd.customerId);
    customer.incrementOrderCount();          // мутация Customer
    await this.customers.save(customer);

    const order = Order.create(cmd.customerId, cmd.items);
    await this.orders.save(order);           // мутация Order
    return order.id;
  });
}

Что не так: транзакция удерживает локи на двух агрегатах, растут конкуренция и вероятность deadlock. Customer и Order живут разными жизнями с разным темпом изменений.

// ХОРОШО — один агрегат меняется, событие летит дальше
async execute(cmd: CreateOrder): Promise<OrderId> {
  return this.tx.run(async () => {
    const order = Order.create(cmd.customerId, cmd.items);
    await this.orders.save(order);
    // OrderCreated зарегистрировано внутри агрегата;
    // outbox-relay опубликует, Customer-сервис обновит счётчик асинхронно
    return order.id;
  });
}

Customer.orderCount обновляется через event-driven. Eventual consistency между агрегатами — норма DDD.

Структура command-handler-а

R-CQRS-CMD-3: классический command-handler в NestJS — четыре шага в строгом порядке.

// core/order/command/confirm-order.handler.ts
@Injectable()
export class ConfirmOrderHandler implements Handler<ConfirmOrder, OrderId> {
  constructor(
    @Inject(ORDER_REPOSITORY) private readonly orders: OrderRepository,
    @Inject(TX_RUNNER)        private readonly tx: TransactionRunner,
    @Inject(CLOCK)            private readonly clock: Clock,
  ) {}

  async execute(cmd: ConfirmOrder): Promise<OrderId> {
    return this.tx.run(async () => {
      // 1. Загрузить агрегат (pessimistic lock через SELECT FOR UPDATE внутри репозитория)
      const order = await this.orders.byId(cmd.orderId);
      if (!order) throw new OrderNotFoundError(cmd.orderId);

      // 2. Вызвать доменный метод — он же проверяет инварианты
      order.confirm(this.clock.now());

      // 3. Сохранить (outbox-событие OrderConfirmed зарегистрировано внутри order.confirm())
      await this.orders.save(order);

      // 4. Вернуть минимум
      return order.id;
    });
  }
}

Что важно в каждом шаге:

• tx.run(async () => { ... }) — транзакция открывается на границе handler-а, не внутри репозитория (R-TYPEORM-TX-1). Всё внутри коллбэка выполняется в одной connection, commit — при возврате без ошибки, rollback — при любом исключении.
• byId без отдельного lock-параметра — но реализация OrderRepository использует SELECT ... FOR UPDATE при открытой транзакции, исключая lost-update race.
• Доменный метод проверяет инварианты. order.confirm() сам бросит OrderAlreadyConfirmedError, если статус не подходит. Handler не делает if (order.status === ...) throw.
• registerEvent — внутри агрегата. order.confirm() регистрирует OrderConfirmed; repository при save записывает событие в outbox. Это критично для CQRS sync — см. Sync через события.

Command возвращает минимум

R-CQRS-CMD-4: возвращаемое значение command — id новой/изменённой сущности, статус-объект или void. Не полный read-DTO.

// OK — возвращает только id созданного заказа
export class CreateOrder implements Command<OrderId> {
  constructor(readonly customerId: CustomerId, readonly items: OrderItem[]) {}
}

// OK — подтверждение возвращает id
export class ConfirmOrder implements Command<OrderId> {
  constructor(readonly orderId: OrderId, readonly idempotencyKey: string) {}
}

// OK — для идемпотентных команд без смыслового ответа
export class CancelOrder implements Command<void> {
  constructor(readonly orderId: OrderId, readonly reason: string) {}
}

// ПЛОХО — возвращает полную read-проекцию
export class CreateOrder implements Command<OrderSummaryView> { ... }

Почему не полный read-DTO (R-CQRS-CMD-X2):

• Смешение ответственностей. Write-handler начинает собирать read-проекцию: join'ы, маппинги, кастомные view — это работа query-handler-а.
• При eventual consistency данные могут быть рассинхронизированы. Write видит order.status = CONFIRMED уже сейчас; query через 100 мс увидит то же. Если клиенту нужна полная проекция после write — два вызова с понятным контрактом надёжнее одного с двумя ответственностями.
• Контроллер делает второй вызов при необходимости. POST /orders → 201 с Location: /orders/{id}. Клиент, если нужно, запрашивает GET /orders/{id}/summary.

Валидация: контракт vs инвариант

R-CQRS-CMD-5: валидация в command-side происходит в двух местах, и их нельзя путать.

// 1. Контракт — на request-DTO через class-validator, до создания command
export class ConfirmOrderRequest {
  @IsUUID()
  orderId: string;

  @IsString()
  @MaxLength(64)
  idempotencyKey: string;
}

// 2. Бизнес-инвариант — внутри метода агрегата
export class Order {
  confirm(now: Date): void {
    if (this.status !== OrderStatus.NEW) {
      throw new OrderAlreadyConfirmedError(this.id, this.status);
    }
    if (this.items.length === 0) {
      throw new EmptyOrderError(this.id);
    }
    this.status = OrderStatus.CONFIRMED;
    this.confirmedAt = now;
    this.registerEvent(new OrderConfirmed(this.id, now));
  }
}

Первый слой останавливает невалидные данные на входе HTTP — ответ 400 клиенту. Второй слой охраняет domain-инварианты — бросает typed domain exception, который error-handler транслирует в 409 или 422. Подробно — в Validation → Где валидировать и Error Handling → Иерархия исключений.

Пример: CreateProduct

Domains разные, структура handler-а одна и та же:

// core/product/command/create-product.command.ts
export class CreateProduct implements Command<ProductId> {
  constructor(
    readonly name: string,
    readonly sku: string,
    readonly price: Money,
    readonly categoryId: CategoryId,
  ) {}
}

@Injectable()
export class CreateProductHandler implements Handler<CreateProduct, ProductId> {
  constructor(
    @Inject(PRODUCT_REPOSITORY) private readonly products: ProductRepository,
    @Inject(TX_RUNNER)          private readonly tx: TransactionRunner,
  ) {}

  async execute(cmd: CreateProduct): Promise<ProductId> {
    return this.tx.run(async () => {
      const existing = await this.products.bySku(cmd.sku);
      if (existing) throw new DuplicateSkuError(cmd.sku);

      const product = Product.create(cmd.name, cmd.sku, cmd.price, cmd.categoryId);
      await this.products.save(product);
      return product.id;
    });
  }
}

bySku — проверка уникальности как часть load-aggregate-логики (читается в той же транзакции, а не как отдельный «посмотреть и решить» read). Сравни с антипаттерном R-CQRS-CMD-X1 ниже.

Пример: SberPay — идемпотентная команда

// core/payment/command/charge-account.command.ts
export class ChargeAccount implements Command<void> {
  constructor(
    readonly accountId: AccountId,
    readonly amount: Money,
    readonly idempotencyKey: string,
  ) {}
}

@Injectable()
export class ChargeAccountHandler implements Handler<ChargeAccount, void> {
  constructor(
    @Inject(ACCOUNT_REPOSITORY) private readonly accounts: AccountRepository,
    @Inject(TX_RUNNER)          private readonly tx: TransactionRunner,
  ) {}

  async execute(cmd: ChargeAccount): Promise<void> {
    return this.tx.run(async () => {
      const account = await this.accounts.byId(cmd.accountId);
      if (!account) throw new AccountNotFoundError(cmd.accountId);

      const alreadyProcessed = await this.accounts.hasProcessed(cmd.idempotencyKey);
      if (alreadyProcessed) return;

      account.charge(cmd.amount, cmd.idempotencyKey);
      await this.accounts.save(account);
    });
  }
}

ChargeAccount возвращает void: UI не нуждается в ответных данных — достаточно HTTP 204. Идемпотентная проверка внутри транзакции: если ключ уже обработан, операция пропускается без ошибки.

Что запрещено

Отдельный SELECT «прочитать и решить» в command

R-CQRS-CMD-X1: внутри command-handler нет отдельного read-запроса «посмотреть и подумать». Всё, что нужно для решения — агрегат, загруженный одним byId.

// ПЛОХО — отдельный read в command
async execute(cmd: ConfirmOrder): Promise<OrderId> {
  return this.tx.run(async () => {
    const hasPayment = await this.payments.existsByOrderId(cmd.orderId); // отдельный read
    if (!hasPayment) throw new PaymentRequiredError(cmd.orderId);

    const order = await this.orders.byId(cmd.orderId);
    if (!order) throw new OrderNotFoundError(cmd.orderId);
    order.confirm(this.clock.now());
    await this.orders.save(order);
    return order.id;
  });
}

Что не так:

• Read-логика просочилась в write-handler; при добавлении новой проверки handler разрастается.
• Race condition: между existsByOrderId и byId состояние payment может измениться.

Корректно: либо paymentStatus — поле агрегата Order (тогда order.confirm() сам проверит), либо это разные bounded context-ы и проверка идёт через явный query, а не приклеена внутрь command.

Command возвращает полный read-DTO

R-CQRS-CMD-X2 — см. выше «Command возвращает минимум». Read-проекция — работа query-handler-а.

Несколько агрегатов в одном tx.run

R-CQRS-CMD-X3: если команда меняет два независимых агрегата — это saga, не один tx.run.

// ПЛОХО — два агрегата в одной транзакции
async execute(cmd: TransferMoney): Promise<void> {
  return this.tx.run(async () => {
    const from = await this.accounts.byId(cmd.fromId);
    const to   = await this.accounts.byId(cmd.toId);
    from.debit(cmd.amount);
    to.credit(cmd.amount);
    await this.accounts.save(from);
    await this.accounts.save(to);
  });
}

// ХОРОШО — orchestration saga
// 1. Command DebitAccount → сохраняет, публикует AccountDebited в outbox
// 2. Orchestrator получает событие → Command CreditAccount
// 3. На failure → компенсирующий CreditAccount (refund)

Pessimistic lock на два агрегата одновременно — прямой путь к deadlock при конкурирующих переводах. Saga даёт явную state-machine с компенсациями.

Что запрещено — таблица

Антипаттерн	Правило	Что взамен
Command-handler делает отдельный SELECT для решения	R-CQRS-CMD-X1	Загрузка агрегата one-shot; проверки внутри агрегата
Command возвращает полный read-DTO	R-CQRS-CMD-X2	OrderId / void + отдельный query при необходимости
Несколько агрегатов в одном tx.run command	R-CQRS-CMD-X3	Saga с локальными командами и компенсациями
tx.run не на границе handler-а, а внутри репозитория	R-CQRS-CMD-3	TransactionRunner в execute, save — без открытия TX
if (order.status !== X) throw в handler вместо метода агрегата	R-CQRS-CMD-5	Инвариант в методе агрегата + typed domain error

Куда дальше

• Query side — read-handler-ы с Query<R> и ViewRepository.
• Sync через события — как outbox-событие из command-handler доходит до read-model.
• Read-model — структура и восстановимость read-model.
• Уровень и эволюция — lightweight vs full CQRS, эволюция по уровням.
• Когда CQRS оправдан — когда вводить CQRS, а когда нет.