Реалізація CQRS для веб-застосунку

Реалізація CQRS (Command Query Responsibility Segregation) для веб-додатку

CQRS розділяє моделі запису (Commands) та читання (Queries) в додатку. Замість однієї моделі, яка одночасно приймає зміни та відповідає на запити, створюються дві незалежні сторони: Command Side обробляє бізнес-операції та змінює стан, Query Side повертає денормалізовані представлення для UI.

Мотивація

Стандартна CRUD-архітектура ламається при зростанні навантаження, коли:

• Читають на порядки частіше, ніж пишуть — потрібно масштабувати незалежно
• Форма даних для запису відрізняється від форми для читання (нормалізація vs. денормалізація)
• Складні бізнес-правила при записі конфліктують з продуктивністю запитів
• Потрібна історія змін або аудит

CQRS вирішує ці завдання ціною збільшення складності системи. Для простої CRUD-програми він надмірний.

Структура Command Side

Command — намір змінити стан. Містить все необхідне для виконання операції:

// Commands — невіддільні об'єкти намірів
interface CreateOrderCommand {
  readonly type: 'CreateOrder';
  readonly customerId: string;
  readonly items: Array<{ productId: string; quantity: number; price: number }>;
  readonly shippingAddress: Address;
}

interface CancelOrderCommand {
  readonly type: 'CancelOrder';
  readonly orderId: string;
  readonly reason: string;
  readonly requestedBy: string;
}

Command Handler — обробляє один тип команди, містить бізнес-логіку:

class CreateOrderCommandHandler {
  constructor(
    private orderRepo: OrderRepository,
    private productRepo: ProductRepository,
    private eventBus: EventBus
  ) {}

  async handle(command: CreateOrderCommand): Promise<string> {
    // 1. Завантажити агрегат або створити новий
    const order = Order.create(command.customerId);

    // 2. Застосувати бізнес-правила
    for (const item of command.items) {
      const product = await this.productRepo.findById(item.productId);
      if (!product.isAvailable(item.quantity)) {
        throw new InsufficientStockError(item.productId);
      }
      order.addItem(item);
    }

    order.setShippingAddress(command.shippingAddress);
    order.submit();

    // 3. Зберегти агрегат
    await this.orderRepo.save(order);

    // 4. Опублікувати доменні події
    await this.eventBus.publishAll(order.pullDomainEvents());

    return order.id;
  }
}

Command Bus — маршрутизує команди до відповідних хендлерів:

class CommandBus {
  private handlers = new Map<string, CommandHandler>();

  register<T extends Command>(type: string, handler: CommandHandler<T>) {
    this.handlers.set(type, handler);
  }

  async dispatch<T extends Command>(command: T): Promise<unknown> {
    const handler = this.handlers.get(command.type);
    if (!handler) throw new Error(`No handler for ${command.type}`);

    // middleware: validation, logging, retry
    return handler.handle(command);
  }
}

Структура Query Side

Query — запит даних без побічних ефектів:

interface GetOrderDetailsQuery {
  readonly type: 'GetOrderDetails';
  readonly orderId: string;
}

interface GetCustomerOrdersQuery {
  readonly type: 'GetCustomerOrders';
  readonly customerId: string;
  readonly status?: OrderStatus;
  readonly page: number;
  readonly perPage: number;
}

Read Model — денормалізоване представлення, оптимізоване під конкретний UI:

interface OrderDetailsReadModel {
  id: string;
  status: string;
  customer: { id: string; name: string; email: string };
  items: Array<{
    productId: string;
    productName: string;
    quantity: number;
    unitPrice: number;
    subtotal: number;
  }>;
  shippingAddress: Address;
  totals: { subtotal: number; shipping: number; tax: number; total: number };
  timeline: Array<{ event: string; occurredAt: Date; actor: string }>;
  createdAt: Date;
  updatedAt: Date;
}

Query Handler — читає безпосередньо з Read Model без завантаження агрегату:

class GetOrderDetailsQueryHandler {
  constructor(private db: Database) {}

  async handle(query: GetOrderDetailsQuery): Promise<OrderDetailsReadModel> {
    return this.db.queryOne(`
      SELECT
        o.id, o.status, o.created_at, o.updated_at,
        c.id as customer_id, c.name as customer_name, c.email,
        json_agg(json_build_object(
          'productId', oi.product_id,
          'productName', p.name,
          'quantity', oi.quantity,
          'unitPrice', oi.unit_price,
          'subtotal', oi.quantity * oi.unit_price
        )) as items,
        o.shipping_address,
        o.total_amount
      FROM orders_view o
      JOIN customers c ON c.id = o.customer_id
      JOIN order_items_view oi ON oi.order_id = o.id
      JOIN products p ON p.id = oi.product_id
      WHERE o.id = $1
      GROUP BY o.id, c.id
    `, [query.orderId]);
  }
}

Синхронізація Read Model

Read Model оновлюється асинхронно через доменні події. Eventual consistency — Read Model може кратковременно відставати від Write Model.

class OrderReadModelUpdater {
  // Підписаний на події з EventBus або Kafka

  async on(event: DomainEvent) {
    switch (event.eventType) {
      case 'OrderCreated':
        await this.db.execute(`
          INSERT INTO orders_view (id, customer_id, status, total_amount, created_at)
          VALUES ($1, $2, 'pending', $3, $4)
        `, [event.aggregateId, event.payload.customerId,
            event.payload.total, event.occurredAt]);
        break;

      case 'OrderStatusChanged':
        await this.db.execute(`
          UPDATE orders_view
          SET status = $2, updated_at = $3
          WHERE id = $1
        `, [event.aggregateId, event.payload.newStatus, event.occurredAt]);
        break;

      case 'OrderItemAdded':
        await this.db.execute(`
          INSERT INTO order_items_view (order_id, product_id, quantity, unit_price)
          VALUES ($1, $2, $3, $4)
          ON CONFLICT (order_id, product_id) DO UPDATE
            SET quantity = EXCLUDED.quantity
        `, [event.aggregateId, event.payload.productId,
            event.payload.quantity, event.payload.price]);
        break;
    }
  }
}

Масштабування

Write Side — вертикальне масштабування основної БД, шардування за aggregate_id.

Read Side — горизонтальне масштабування read replicas PostgreSQL, Redis-кеш для hot data, ElasticSearch для повнотекстового пошуку, окрема таблиця або схема під кожен Read Model.

Рівні реалізації CQRS

Починати варто з логічного розділення. Переходити до різних БД тільки при доведеній необхідності.

Терміни реалізації

• Рефакторинг існуючого додатку до Command/Query розділення — 1–2 тижні
• Новий додаток з CQRS від старту — 2–3 тижні
• Повний CQRS + Event Sourcing + async Read Models — 4–8 тижнів залежно від складності домену