Перейти к основному содержимому

Consumer: DevStatsLayoutGroupUpsertedConsumer

Ingest-консюмер реплики-измерения «Группа планировок» в dev-stats: принимает snapshot группы из chessboard и upsert-ит строку в ClickHouse-измерение dim_layout_group (ReplacingMergeTree, pin_dev_stats). Уровень иерархии между проектом и планировкой (ADR-0057). Дом payload — event-contract: layout-group-upserted; дом колонок — dimensions/layout-group.md.

Назначение​

Consumer DevStatsLayoutGroupUpsertedConsumer слушает событие layout-group-upserted (топик pin.chessboard.layout-group-upserted.v1) и пишет срез группы (Name/SortOrder/IsArchived

  • ProjectId/DeveloperId-родители + версия) в dim_layout_group. Измерение даёт подпись графика (Name — «Студии», «Люкс с видом на море»), порядок отображения (SortOrder) и денормализованный DeveloperId для восхождения по иерархии/фильтрации без JOIN-ов. У группы нет статусного цикла — вместо Status реплицируется флаг IsArchived. Owned-состояния в PostgreSQL нет; счётчики (LayoutsCount) и площади (AreaMin/AreaMax) не реплицируются.

Owning-сервис и триггер​

ПолеЗначение
Сервис/контекстdev-stats (проект DevStats.Background)
ConsumerDevStatsLayoutGroupUpsertedConsumer
Обрабатываемое событиеlayout-group-upserted
Topicpin.chessboard.layout-group-upserted.v1
ContractVersion1; при несовпадении — warn-and-proceed
Consumer groupdev-stats-dim-layout-group
Producerchessboard (Developers API create/update/archive группы, синк Talent, job project-aggregates-recalc)
Эффект (событие → что меняется)upsert строки в pin_dev_stats.dim_layout_group (ReplacingMergeTree)

Контекст события​

  • Из метаданных события доступны идентификатор источника (AggregateId = Id группы), тип агрегата (LayoutGroup), время события (CreatedAt → UpsertedAt, версия ReplacingMergeTree(UpsertedAt)), версия snapshot-а (Version).
  • Сквозной трейс восстанавливается из метаданных события.
  • Идемпотентность — декларативная, на стороне ClickHouse: dedup-ключ = LayoutGroupId (единственный компонент ORDER BY), версионирование — ReplacingMergeTree(UpsertedAt). Порядок per-group гарантирован ключом сообщения (key = Id); отдельная таблица идемпотентности не используется (PG-транзакции нет).

Входные данные / payload события​

Ссылка на контракт: event-contract: layout-group-upserted. Дом и типы колонок — dimensions/layout-group.md, раздел «Columns»; ниже — поля, которые консюмер реально читает, и маппинг.

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДа→ LayoutGroupId (UUID, ключ ORDER BY); Guid.Empty → poison (DLQ)
ProjectIdGuidДа→ ProjectId (родитель); Guid.Empty → poison
TenantIdGuidДа→ DeveloperId (денормализация вверх, ADR-0052); Guid.Empty → poison
NamestringДа→ Name (подпись графика)
SortOrderintДа→ SortOrder (Int32, порядок отображения групп)
IsArchivedboolДа→ IsArchived (UInt8): true → 1, false → 0 (архив — пометка, строка не удаляется)
VersionlongДа→ SourceVersion (guard/диагностика)
LayoutsCount,AreaMin,AreaMax——не реплицируются

Колонки, заполняемые консюмером: UpsertedAt = время события. Enum-полей в payload нет — обработка unknown-enum неприменима (состояние группы — bool IsArchived).

Алгоритм​

Консюмер батчевый: шаги 1–3 — на сообщение, шаг 4 — на батч (порог ≤ 1000 строк или 5 секунд; freshness SLA ≤ 5 мин).

  1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warning-лог и продолжить). Дубль на входе не отсеивается (dedup декларативен).
  2. Проверить payload: ошибка десериализации / Id / ProjectId / TenantId == Guid.Empty → DLQ (poison) + error-лог. Проверки существования родителей нет: DeveloperId берётся из TenantId, резолв иерархии факта идёт из dim_property_object.
  3. Маппинг поле-в-поле: построить строку dim_layout_group (DeveloperId = TenantId, IsArchived = payload.IsArchived ? 1 : 0, UpsertedAt = время события, SourceVersion = Version). Добавить в буфер; внутрибатчевая свёртка по LayoutGroupId (max UpsertedAt).
  4. Пакетно вставить (upsert) буфер строк в pin_dev_stats.dim_layout_group. Upsert обеспечивает ReplacingMergeTree(UpsertedAt) (merge → max UpsertedAt per LayoutGroupId); точечное обновление/удаление не применяются (архив — upsert с IsArchived = 1). Смещение (offset) фиксируется строго после подтверждённой вставки; при ошибке — retry всего батча.

Диаграмма​

Идемпотентность и replay​

  • Business key: LayoutGroupId; версия-precedence — UpsertedAt (ReplacingMergeTree). Дубль перезаписывается идемпотентно; до merge чтение дубль-устойчиво — argMax(Name, UpsertedAt) … GROUP BY LayoutGroupId (FINAL запрещён).
  • Version guard (dimensions/layout-group.md) — декларативный: per-key-ordering + ReplacingMergeTree(UpsertedAt) + внутрибатчевая свёртка; version-lookup нет (ADR-0056).
  • Replay/backfill: да — повтор топика layout-group-upserted группой dev-stats-dim-layout-group безопасен; первичное наполнение — background/dimension-backfill.md.

Retry и DLQ​

СбойRetryDLQРучное действие
ClickHouse недоступен / timeout вставкиДа: retry батча, backoff (1с → 60с), offset не commit-итсяНет (инфраструктура)Алерт ingest-lag > 15 мин; восстановить CH
Ошибка десериализации payloadНетДа (poison)Разбор из DLQ; issue producer-у chessboard
Id/ProjectId/TenantId == Guid.Empty (шаг 2)НетДа (poison) + error-логПроверить контракт; replay из DLQ после фикса
Версия события ≠ ContractVersionНе сбой: warning-логНетСверить миграцию контракта
Сбой между вставкой и commit offsetПовторная доставка (at-least-once)НетНе требуется — dedup движком

Наблюдаемость и безопасность​

  • Logs: consumed, skipped-poison (Id), flushed (строк, длительность), failed (CH).
  • Metrics: dev_stats_dim_layout_group_upserted_total, dev_stats_dim_ingest_lag_seconds, dev_stats_dim_ingest_batch_size, retry/DLQ.
  • Traces: consume span + insert span; корреляция из метаданных события.
  • PII/masking: PII отсутствует (группа планировок — справочный объект). Tenant: DeveloperId = TenantId (ADR-0052); scope-фильтрация — на чтении в API.

Кросс-сервисные контракты​

НаправлениеКонтракт
Ожидает от chessboardСобытие layout-group-upserted (pin.chessboard.layout-group-upserted.v1, ContractVersion = 1, at-least-once). Гарантии: ключ сообщения = Id (per-group ordering), монотонная Version, TenantId = DeveloperId и ProjectId-родитель в snapshot-е. Статусного цикла нет — состояние выражено bool IsArchived.
ПредоставляетСтроку dim_layout_group (ReplacingMergeTree) с Name/SortOrder/IsArchived + ProjectId/DeveloperId. Потребители реплики: обогащение фактов уровнем LayoutGroup и подписи/порядок групп в ЛК («Студии», «Люкс с видом на море»).
Окно eventual-consistency≤ 5 мин. Пока реплика отстаёт, факт берёт последнюю известную версию dim_layout_group; подпись/порядок группы может отставать.

Предусловия и постусловия​

  • Предусловия: миграции DDL измерения применены; событие валидно (Id/ProjectId/TenantId != Guid.Empty).
  • Постусловия: в dim_layout_group для LayoutGroupId актуальна строка с максимальным UpsertedAt (архив — upsert с IsArchived = 1, строка не удаляется); offset группы dev-stats-dim-layout-group продвинут строго после подтверждённой вставки; операционные данные chessboard не изменены.

Acceptance Criteria​

  • Happy path (создание): Дано — миграции применены; Когда — приходит layout-group-upserted (G1, ProjectId = P1, Name = "Студии", IsArchived = false); Тогда — argMax(Name, UpsertedAt) … GROUP BY LayoutGroupId для G1 = "Студии", IsArchived = 0, DeveloperId = TenantId.
  • Архивация: Дано — строка G1 есть; Когда — событие с IsArchived = true и большим UpsertedAt; Тогда — чтение отдаёт IsArchived = 1 (строка сохраняется, не удаляется).
  • Идемпотентность (повтор доставки): Дано — событие G1 применено; Когда — повторная доставка; Тогда — логических дублей нет, offset продвинут.
  • Poison: Дано — Id/ProjectId/TenantId == Guid.Empty или битый payload; Тогда — DLQ (poison) + error-лог, партиция продолжается.
  • Отказ хранилища: Дано — ClickHouse недоступен; Тогда — offset не коммитится, retry с backoff, догон без дублей.

Обратные ссылки​

Автоссылки: где используется этот документ.

Связанные документы​