Уровень и эволюция CQRS в Go

← назад к разделу

CQRS — не переключатель, который включают или выключают целиком. Это шкала: на каждом уровне берёшь ровно столько разделения, сколько даёт реальную пользу здесь и сейчас. Запускаться с event-driven read-model на сервисе без доказанной проблемы с нагрузкой — значит платить инфраструктурную цену за несуществующую боль.

Разберём четыре уровня на примере сервиса заказов.

Уровень 1 — обычный сервис, CQRS не нужен

Это точка отсчёта. Обычный сервис-обработчик с одним репозиторием, без разделения операций чтения и записи. Никаких маркеров, одинаковые транзакции для всего.

// internal/order/service.go
type OrderService struct {
    db *pgxpool.Pool
}

func (s *OrderService) CreateOrder(ctx context.Context, customerID string, items []string) (string, error) {
    // ...запись...
}

func (s *OrderService) GetOrder(ctx context.Context, id string) (OrderDTO, error) {
    // ...чтение, та же транзакционная стратегия...
}

Этот уровень подходит для внутренних утилит, тонких прокси, простых CRUD-сервисов без выраженной бизнес-логики. Вводить маркеры Command/Query здесь не нужно — они ничего не добавят.

Уровень 2 — маркеры Command и Query

Когда у сервиса появляется настоящий домен с инвариантами, имеет смысл явно разделить команды и запросы. Это даёт видимость в коде: сразу понятно, меняет ли операция состояние или только читает.

Маркеры — пустые интерфейсы с неэкспортируемым методом. Неэкспортируемый метод работает как замок: реализовать его из другого пакета не получится вовсе, поэтому команды и запросы получают его встраиванием cqrs.CommandMarker и cqrs.QueryMarker.

// core/cqrs/cqrs.go
package cqrs

type Command interface{ isCommand() }
type Query   interface{ isQuery()   }

// Маркеры для встраивания: неэкспортируемый метод из чужого пакета
// не реализовать, команда получает его через встроенную структуру
type CommandMarker struct{}
type QueryMarker struct{}

func (CommandMarker) isCommand() {}
func (QueryMarker) isQuery()     {}

В коде:

// core/order/command/create_order.go
type CreateOrder struct {
    cqrs.CommandMarker
    CustomerID string
    Items      []string
}

// core/order/query/get_order_summary.go
type GetOrderSummary struct {
    cqrs.QueryMarker
    OrderID string
}

Репозиторий на этом уровне остаётся один, но обработчики команд и запросов получают разные транзакционные стратегии.

Обработчик команды — обычная транзакция на запись:

func (h *CreateOrderHandler) Handle(ctx context.Context, cmd command.CreateOrder) (string, error) {
    var id string
    err := h.uow.Within(ctx, func(ctx context.Context) error {
        order := NewOrder(cmd.CustomerID, cmd.Items)
        if err := h.orders.Save(ctx, order); err != nil {
            return fmt.Errorf("save order: %w", err)
        }
        id = order.ID
        return nil
    })
    return id, err
}

Обработчик запроса — транзакция только на чтение (pgx.ReadOnly). Без этой опции маркер Query ничего не гарантирует: код выглядит красиво, но база никак не защищена от случайной записи.

func (h *GetOrderSummaryHandler) Handle(ctx context.Context, q query.GetOrderSummary) (OrderSummaryDTO, error) {
    tx, err := h.db.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly})
    if err != nil {
        return OrderSummaryDTO{}, fmt.Errorf("begin read tx: %w", err)
    }
    defer tx.Rollback(ctx)
    ctx = withTx(ctx, tx) // репозиторий берёт транзакцию из контекста, как и на write-стороне

    summary, err := h.orders.SummaryByID(ctx, q.OrderID)
    if err != nil {
        return OrderSummaryDTO{}, fmt.Errorf("read summary %s: %w", q.OrderID, err)
    }
    return summary, nil
}

Единый OrderRepository на этом уровне — это осознанное упрощение, а не недостаток. Разделять интерфейсы раньше, чем это нужно, — преждевременная сложность.

// core/order/port/out/order_repository.go
type OrderRepository interface {
    ByID(ctx context.Context, id string) (*Order, error)
    SummaryByID(ctx context.Context, id string) (OrderSummaryDTO, error)
    Save(ctx context.Context, o *Order) error
}

PostgreSQL сам отловит попытку записи в ReadOnly-транзакции и вернёт ошибку — никакого дополнительного кода не нужно. Одно условие: репозиторий должен выполнять запрос внутри этой транзакции, поэтому её кладут в контекст (withTx), как делает UnitOfWork на write-стороне.

Уровень 3 — раздельные репозитории

Когда появляются сложные проекции — история транзакций, сводки по клиенту, таблицы с фильтрами — становится неудобно держать методы чтения и записи в одном интерфейсе. Читающей части нужны отдельные запросы, отдельные DTO, своя форма данных.

Решение: завести два интерфейса.

// core/order/port/out/order_repository.go — только запись
type OrderRepository interface {
    ByID(ctx context.Context, id string) (*Order, error)
    Save(ctx context.Context, o *Order) error
}

// core/order/port/out/order_view_repository.go — только чтение
type OrderViewRepository interface {
    SummaryByID(ctx context.Context, id string) (view.OrderSummaryDTO, error)
    ListByCustomer(ctx context.Context, customerID string, p Pagination) ([]view.OrderSummaryDTO, error)
}

Read-DTO — самостоятельные структуры, форма которых продиктована тем, что нужно API, а не тем, как устроен агрегат:

// core/order/dto/view/order_summary.go
type OrderSummaryDTO struct {
    OrderID      string
    CustomerName string
    TotalAmount  int64
    Status       string
    ItemCount    int
    CreatedAt    time.Time
}

Обработчик запроса теперь работает с OrderViewRepository, а не с основным репозиторием:

type GetOrderSummaryHandler struct {
    views OrderViewRepository
    db    *pgxpool.Pool
}

func (h *GetOrderSummaryHandler) Handle(ctx context.Context, q query.GetOrderSummary) (view.OrderSummaryDTO, error) {
    tx, err := h.db.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly})
    if err != nil {
        return view.OrderSummaryDTO{}, fmt.Errorf("begin read tx: %w", err)
    }
    defer tx.Rollback(ctx)
    ctx = withTx(ctx, tx)

    return h.views.SummaryByID(ctx, q.OrderID)
}

На этом уровне база данных по-прежнему одна — PostgreSQL. Разделение пока только в типах и интерфейсах, не в инфраструктуре.

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

Переход нужен, когда read-нагрузка начинает мешать write-стороне — или когда форма данных для чтения фундаментально другая: full-text поиск, аналитические сводки, данные из нескольких источников.

Read-модель переезжает в отдельную денормализованную таблицу (order_summary) или в Redis и синхронизируется через outbox + Kafka. Если таблица живёт в той же базе и обновляется дёшево, её можно вести и той же транзакцией — событийная синхронизация нужна ради независимости хранилищ и нагрузки, а не сама по себе.

Интерфейс OrderViewRepository не меняется — меняется только реализация адаптера: теперь он читает из order_summary, а не из orders.

// adapters/out/persistence/order_summary_repository.go
func (r *OrderSummaryRepository) SummaryByID(ctx context.Context, id string) (view.OrderSummaryDTO, error) {
    row, err := r.queries.GetOrderSummary(ctx, id)   // sqlc → order_summary
    if err != nil {
        if errors.Is(err, pgx.ErrNoRows) {
            return view.OrderSummaryDTO{}, fmt.Errorf("order summary not found: %w", pgx.ErrNoRows)
        }
        return view.OrderSummaryDTO{}, fmt.Errorf("get order summary: %w", err)
    }
    return toOrderSummaryDTO(row), nil
}

Этот уровень несёт реальную стоимость:

• Задержка синхронизации — в норме 100 мс–2 с. Read-сторона видит не самые свежие данные.
• Новые точки отказа — consumer может отстать, outbox — зависнуть, данные read и write — разъехаться.
• Больше компонентов — outbox-relay, Kafka consumer, мониторинг отставания.

До этого порога read-реплика + кеш решают дешевле и без этих рисков.

Как выглядит эволюция на практике

Сервис заказов проходит эти уровни постепенно, и каждый шаг диктует реальная боль, а не план «сделать правильно сразу»:

1. Уровень 1 — стартовали как внутренний CRUD-сервис.
2. Уровень 2 — появился домен «Заказ» с инвариантами. Добавили маркеры Command/Query и pgx.ReadOnly на обработчиках чтения.
3. Уровень 3 split — понадобились сложные проекции (история, сводки по клиенту). Выделили OrderViewRepository с отдельными запросами.
4. Уровень 3 event-driven — p95 latency списочных запросов пробил SLA под нагрузкой. Перенесли read-модель в отдельную таблицу, синхронизация через outbox + Kafka.

Возврат назад (упрощение структуры) происходит при слиянии сервисов или сокращении продукта — это редкость, не обычный рефакторинг.

Частые ошибки

Маркеры без ReadOnly-транзакции. Маркер Query на структуре ничего не гарантирует, если обработчик открывает обычную rw-транзакцию. Либо добавляйте pgx.ReadOnly, либо не используйте маркеры — половинчатое решение хуже обоих крайних.

Event-driven read-model с единым репозиторием. Если read-модель живёт в отдельной таблице, но OrderRepository один и обслуживает и чтение, и запись — разделение потеряло смысл. Нужен отдельный OrderViewRepository.

Прыжок через уровни. Переход с уровня 1 сразу на event-driven без промежуточных шагов — это максимальная сложность без проверенной необходимости. Каждый шаг должен быть обоснован наблюдаемой проблемой.

Read-методы в основном репозитории на уровне 3. Если SummaryByID и ListByCustomer остались в OrderRepository после выделения OrderViewRepository — разделение интерфейсов формальное, а не реальное.

Коротко

• CQRS — шкала из четырёх уровней; стартовать с верхнего без обоснования — преждевременная сложность.
• Уровень 1: обычный сервис, маркеры не нужны.
• Уровень 2: маркеры Command/Query обязательно с pgx.ReadOnly на обработчиках запросов — одно без другого бессмысленно.
• Уровень 3 split: два интерфейса — OrderRepository (запись, агрегат) и OrderViewRepository (чтение, read-DTO). База всё ещё одна.
• Уровень 3 event-driven: read-модель в отдельном хранилище, синхронизация через outbox + Kafka. Несёт задержку синхронизации и новые точки отказа.
• Переходить вверх — по метрикам и реальной боли, не потому что «так правильно».

Что почитать дальше

• Когда CQRS оправдан — пороги перехода между уровнями.
• Command side в Go — write-handler: UnitOfWork, pgx.Tx через context.
• Query side в Go — read-handler с OrderViewRepository и pgx.ReadOnly.
• Read-model в Go — денормализованная таблица, Redis, rebuild.
• Синхронизация через события — outbox + segmentio/kafka-go, идемпотентный consumer.