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

Consumer: DevStatsLayoutUpsertedConsumer

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

Назначение​

Consumer DevStatsLayoutUpsertedConsumer слушает событие layout-upserted (топик pin.chessboard.layout-upserted.v1) и пишет срез планировки (Name/Rooms/TotalArea/IsArchived + LayoutGroupId/ProjectId/DeveloperId-родители + версия) в dim_layout. Измерение даёт подпись графика (Name — «1К-А 32.5»), атрибуты разбивки (Rooms, TotalArea) и родительские id (в т.ч. группу) для восхождения по иерархии/фильтрации без JOIN-ов. У планировки нет статусного цикла — вместо Status реплицируется флаг IsArchived. Owned-состояния в PostgreSQL нет; жилая площадь (LivingArea), схемы (Schemas), счётчик объектов недвижимости (PropertyObjectsCount) не реплицируются.

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

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

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

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

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

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

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДа→ LayoutId (UUID, ключ ORDER BY); Guid.Empty → poison (DLQ)
LayoutGroupIdGuid?Нет→ LayoutGroupId (Nullable(UUID), родитель; null — планировка вне группы)
ProjectIdGuidДа→ ProjectId (планировки типовые в проекте); Guid.Empty → poison
TenantIdGuidДа→ DeveloperId (денормализация вверх, ADR-0052); Guid.Empty → poison
NamestringДа→ Name (подпись графика)
RoomsshortДа→ Rooms (Int16, 0 = студия; атрибут разбивки)
TotalAreadecimalДа→ TotalArea (Decimal(8,2), типовая общая площадь, м²)
IsArchivedboolДа→ IsArchived (UInt8): true → 1, false → 0 (архив — пометка, строка не удаляется)
VersionlongДа→ SourceVersion (guard/диагностика)
LivingArea,PropertyObjectsCount,Schemas——не реплицируются

Колонки, заполняемые консюмером: 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, LayoutGroupId записывается как есть (nullable), резолв иерархии факта идёт из dim_property_object.
  3. Маппинг поле-в-поле: построить строку dim_layout (DeveloperId = TenantId, IsArchived = payload.IsArchived ? 1 : 0, UpsertedAt = время события, SourceVersion = Version). Добавить в буфер; внутрибатчевая свёртка по LayoutId (max UpsertedAt).
  4. Пакетно вставить (upsert) буфер строк в pin_dev_stats.dim_layout. Upsert обеспечивает ReplacingMergeTree(UpsertedAt) (merge → max UpsertedAt per LayoutId); точечное обновление/удаление не применяются (архив — upsert с IsArchived = 1). Смещение (offset) фиксируется строго после подтверждённой вставки; при ошибке — retry всего батча.

Диаграмма​

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

  • Business key: LayoutId; версия-precedence — UpsertedAt (ReplacingMergeTree). Дубль перезаписывается идемпотентно; до merge чтение дубль-устойчиво — argMax(Name, UpsertedAt) … GROUP BY LayoutId (FINAL запрещён).
  • Version guard (dimensions/layout.md) — декларативный: per-key-ordering
    • ReplacingMergeTree(UpsertedAt) + внутрибатчевая свёртка; version-lookup нет (ADR-0056).
  • Replay/backfill: да — повтор топика layout-upserted группой dev-stats-dim-layout безопасен; первичное наполнение — 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_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-upserted (pin.chessboard.layout-upserted.v1, ContractVersion = 1, at-least-once). Гарантии: ключ сообщения = Id (per-layout ordering), монотонная Version, TenantId = DeveloperId и ProjectId-родитель; LayoutGroupId — nullable (планировка вне группы). Статусного цикла нет — состояние выражено bool IsArchived.
ПредоставляетСтроку dim_layout (ReplacingMergeTree) с Name/Rooms/TotalArea/IsArchived + LayoutGroupId/ProjectId/DeveloperId. Потребители реплики: обогащение фактов уровнем Layout, атрибуты разбивки (Rooms, TotalArea) и подпись графика («1К-А 32.5»).
Окно eventual-consistency≤ 5 мин. Пока реплика отстаёт, факт берёт последнюю известную версию dim_layout; подпись/атрибуты планировки могут отставать.

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

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

Acceptance Criteria​

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

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

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

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