Уровень и эволюция CQRS — Node/NestJS: от маркеров к event-driven read-model

Опирается на правила: R-CQRS-TIER-1 … R-CQRS-TIER-5 и R-CQRS-TIER-X1 … R-CQRS-TIER-X2 из CQRS Style Guide → раздел 6. Уровень и эволюция.

Важно знать

• Уровень 1 (плоский Service): CQRS не применяется — маркеры Command<R> / Query<R> не нужны.
• Уровень 2 (Use Case Pattern): маркеры Command<R> / Query<R> обязательны. Один <X>Repository, query-handler работает без транзакции.
• Уровень 3 split (DDD + Hexagonal): появляется <X>ViewRepository с raw select → read-DTO; write — через <X>Repository с агрегатом и pessimistic lock.
• Уровень 3 event-driven: read-model в отдельной таблице / Redis / ES, sync через outbox + Kafka.
• Эволюция строго снизу вверх: 1 → 2 → 3-split → 3-event-driven. Возврат назад — сигнал ошибки при старте.
• Маркеры без enforcement (query-handler с транзакцией на запись, общий repository для R+W при event-driven инфра) — карго-культ.
• Если read-инфра отдельная — интерфейс тоже отдельный. <X>ViewRepository и <X>Repository не смешиваются.

CQRS — не «есть или нет», а шкала зрелости. На каждом уровне берётся ровно столько, сколько даёт пользу при текущем размере и нагрузке. Избыточная event-driven инфра на молодом сервисе стоит дороже, чем lightweight на старте и эволюция по метрикам.

Уровень 1 — CQRS не применяется

R-CQRS-TIER-1: на Уровне 1 (плоский NestJS @Injectable() Service) CQRS не используется. Никаких маркеров, никакого Handler, общий метод на чтение и запись.

@Injectable()
export class OrderService {
  constructor(private readonly dataSource: DataSource) {}

  async createOrder(dto: CreateOrderDto): Promise<string> {
    const order = Order.create(dto.customerId, dto.items);
    await this.dataSource.getRepository(OrderEntity).save(order.toEntity());
    return order.id;
  }

  async getOrder(id: string): Promise<OrderDto> {
    const row = await this.dataSource.getRepository(OrderEntity).findOneBy({ id });
    if (!row) throw new OrderNotFoundError(id);
    return toOrderDto(row);
  }
}

Уровень 1 — тонкие CRUD-сервисы, внутренние утилиты, proxy. Вводить маркеры здесь нет смысла: одна модель данных, простая структура, TypeORM Entity покрывает всё.

Уровень 2 — lightweight CQRS обязателен

R-CQRS-TIER-2: на Уровне 2 (Use Case Pattern) маркеры Command<R> / Query<R> обязательны на всех use-case-классах. Read и write идут через один <X>Repository, но query-handler работает без транзакции.

// core/order/port/in/create-order.command.ts
export class CreateOrderCommand implements Command<OrderId> {
  constructor(
    readonly customerId: CustomerId,
    readonly items: ReadonlyArray<OrderItemInput>,
  ) {}
}

// core/order/port/in/get-order.query.ts
export class GetOrderQuery implements Query<OrderSummary> {
  constructor(readonly orderId: OrderId) {}
}

// application/order/create-order.handler.ts
@Injectable()
export class CreateOrderHandler implements Handler<CreateOrderCommand, OrderId> {
  constructor(
    @Inject(ORDER_REPOSITORY) private readonly orders: OrderRepository,
    @Inject(TX_RUNNER) private readonly tx: TransactionRunner,
  ) {}

  async execute(cmd: CreateOrderCommand): Promise<OrderId> {
    return this.tx.run(async () => {
      const order = Order.create(cmd.customerId, cmd.items);
      await this.orders.save(order);
      return order.id;
    });
  }
}

// application/order/get-order.handler.ts — без транзакции (enforcement маркера)
@Injectable()
export class GetOrderHandler implements Handler<GetOrderQuery, OrderSummary> {
  constructor(
    @Inject(ORDER_REPOSITORY) private readonly orders: OrderRepository,
  ) {}

  async execute(query: GetOrderQuery): Promise<OrderSummary> {
    const order = await this.orders.byId(query.orderId);   // read без TX
    if (!order) throw new OrderNotFoundError(query.orderId);
    return toOrderSummary(order);
  }
}

Enforcement маркеров на Уровне 2:

• Query-handler работает без tx.run() — read без транзакции и без lock. Это и есть enforcement Query<R>.
• Command-handler обёрнут в tx.run() — один агрегат, один commit на границе Handler (R-CQRS-CMD-3).
• Command возвращает минимум — OrderId, не OrderSummary. Контроллер дёрнет GetOrderHandler отдельно если UI нужны данные после создания (R-CQRS-CMD-4).

Read и write используют один OrderRepository — отдельный OrderViewRepository здесь избыточен.

Уровень 3 split — отдельный ViewRepository

R-CQRS-TIER-3: на Уровне 3 (DDD + Hexagonal) появляются два отдельных интерфейса. OrderRepository для записи (агрегат + pessimistic lock), OrderViewRepository для чтения (raw select → read-DTO).

// core/order/port/out/order.repository.ts
export interface OrderRepository {
  byId(id: OrderId): Promise<Order | null>;
  save(order: Order): Promise<void>;
}

// core/order/port/out/order-view.repository.ts
export interface OrderViewRepository {
  summary(orderId: OrderId): Promise<OrderSummary | null>;
  search(customerId: CustomerId, status: OrderStatus, page: PageRequest): Promise<Page<OrderSummary>>;
}

// adapters/out/persistence/typeorm-order-view.repository.ts
@Injectable()
export class TypeOrmOrderViewRepository implements OrderViewRepository {
  constructor(private readonly dataSource: DataSource) {}

  async summary(orderId: OrderId): Promise<OrderSummary | null> {
    const rows = await this.dataSource.query(
      `SELECT o.id, o.status, o.customer_name, o.total_amount, o.created_at
         FROM orders o
        WHERE o.id = $1`,
      [orderId],
    );
    return rows[0] ? toOrderSummary(rows[0]) : null;
  }

  async search(customerId: CustomerId, status: OrderStatus, page: PageRequest): Promise<Page<OrderSummary>> {
    const rows = await this.dataSource.query(
      `SELECT o.id, o.status, o.customer_name, o.total_amount, o.created_at
         FROM orders o
        WHERE o.customer_id = $1
          AND ($2::text IS NULL OR o.status = $2)
        ORDER BY o.created_at DESC
        LIMIT $3 OFFSET $4`,
      [customerId, status ?? null, page.size, page.offset()],
    );
    return toPage(rows, page);
  }
}

Что меняется:

• Два интерфейса в core/. OrderRepository возвращает агрегат, OrderViewRepository возвращает read-DTO. Без пересечений.
• Raw select в <X>ViewRepository (R-TYPEORM-QRY-4). Не find() с relations, а явный SQL с bind-параметрами — структура диктуется API/UI, не TypeORM-маппингом агрегата.
• Read-DTO — frozen plain object или readonly-класс в core/<bc>/port/ (или core/<bc>/port/out/view/). Наружу — только данные, без доменных методов.
• Запрос через OrderViewRepository в query-handler. OrderRepository query-handler не видит.

@Injectable()
export class GetOrderHandler implements Handler<GetOrderQuery, OrderSummary> {
  constructor(
    @Inject(ORDER_VIEW_REPOSITORY) private readonly view: OrderViewRepository,
  ) {}

  async execute(query: GetOrderQuery): Promise<OrderSummary> {
    const summary = await this.view.summary(query.orderId);
    if (!summary) throw new OrderNotFoundError(query.orderId);
    return summary;
  }
}

Read и write по-прежнему используют одно физическое хранилище — PostgreSQL. Разделение пока на уровне интерфейсов и запросов, не инфраструктуры.

Уровень 3 event-driven — отдельное хранилище

R-CQRS-TIER-4: эволюция от split-варианта, когда нагрузки или паттерны чтения требуют отдельной инфраструктуры. Read-model переезжает в денормализованную PG-таблицу, Redis, ElasticSearch — то, что оптимально под конкретный паттерн чтения.

write-side:                        read-side:
  PostgreSQL                         order_summary (PG-таблица)
  ├── orders (агрегат)               ├── customer_name (денормализовано)
  └── outbox_events                  └── индексы под query-сторону
        ↓
  outbox-relay (SKIP LOCKED)
        ↓
  Kafka (order.events)
        ↓
  read-side consumer
        ↓
  UPSERT order_summary

Что добавляется к split-варианту:

• Таблица outbox_events в write-БД — строка пишется в той же транзакции, что и агрегат.
• Outbox-relay — scheduled NestJS job или отдельный сервис с SELECT ... FOR UPDATE SKIP LOCKED.
• Kafka топик — order.events (или sber.acquiring.events, product.catalog.events — по домену).
• Read-side consumer — в этом же NestJS-приложении или в отдельном; обновляет order_summary через UPSERT.
• Idempotency-защита (R-CQRS-SYNC-2) — таблица processed_event или version-check в UPSERT.
• Bootstrap-rebuilder — отдельный скрипт или NestJS command (nest command), который перебирает агрегаты и пересобирает order_summary.

// adapters/in/events/order-event.consumer.ts
@Injectable()
export class OrderEventConsumer {
  constructor(private readonly dataSource: DataSource) {}

  @EventPattern('order.events')
  async handle(payload: OrderEventPayload): Promise<void> {
    const already = await this.dataSource.query(
      `SELECT 1 FROM processed_event WHERE event_id = $1`, [payload.eventId],
    );
    if (already.length) return;

    await this.dataSource.transaction(async (em) => {
      await em.query(
        `INSERT INTO order_summary (id, status, customer_name, total_amount, updated_at)
         VALUES ($1, $2, $3, $4, NOW())
         ON CONFLICT (id) DO UPDATE
           SET status = EXCLUDED.status,
               total_amount = EXCLUDED.total_amount,
               updated_at = EXCLUDED.updated_at`,
        [payload.orderId, payload.status, payload.customerName, payload.totalAmount],
      );
      await em.query(
        `INSERT INTO processed_event (event_id, processed_at) VALUES ($1, NOW())`,
        [payload.eventId],
      );
    });
  }
}

Стоимость, которую принимаем:

• Eventual consistency (100ms–1s в норме, больше при деградации consumer-а или relay).
• Новые failure modes: отставание consumer, зависший outbox, рассинхронизация проекции.
• Дополнительный мониторинг: lag consumer, возраст необработанных outbox-строк.

Окупается, когда write-сторона страдает от read-нагрузки или read-проекция фундаментально другая (full-text, аналитические сводки). До этого порога read-replica + кеш (R-CQRS-WHEN-X2) решают дешевле.

Эволюция строго снизу вверх

R-CQRS-TIER-5: движение по уровням — строго 1 → 2 → 3-split → 3-event-driven. Возврат назад — редкость и обычно сигнал, что начали слишком высоко.

Типичный путь жизни NestJS-сервиса:

1. Уровень 1 — стартовали как утилитный микросервис без явной бизнес-домены. Плоский ProductService, TypeORM Entity наружу.
2. Появился реальный домен — ввели Command<R> / Query<R>, Handler, TransactionRunner. Уровень 2.
3. Read-сторона стала отличаться от write: Customer-портал хочет проекций с customer_name inline, без join. Завели ProductViewRepository с raw select. Уровень 3 split.
4. p95 latency query пробил SLA при пике каталога Sber Marketplace — переехали на Redis-проекцию с outbox+Kafka. Уровень 3 event-driven.

Каждый переход бизнес-обоснован: метриками, новыми требованиями к чтению, фактом боли. Не «потому что event-driven — это современно».

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

Куда дальше

• Command side — write-handler на Уровне 2 / 3 в NestJS.
• Query side — query-handler с <X>ViewRepository и raw select.
• Read-model — где хранить отдельную проекцию на Уровне 3 event-driven.
• Sync через события — outbox + Kafka для event-driven в NestJS.
• Когда CQRS оправдан — пороги перехода между уровнями.