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

Consumer: DevStatsBuildingUpsertedConsumer

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

Назначение​

Consumer DevStatsBuildingUpsertedConsumer слушает событие building-upserted (топик pin.chessboard.building-upserted.v1) и пишет срез здания (Name/Status/ConstructionStatus/FloorsTotal

  • ProjectId/DeveloperId-родители + версия) в dim_building. Измерение даёт подпись графика (Name — «Литер 3») и денормализованный DeveloperId (из TenantId), чтобы восхождение по иерархии и проверка scope шли без обращения к dim_project. Owned-состояния в PostgreSQL нет (ClickHouse-only); сроки сдачи (Delivery*), SectionsCount, ReadinessPercent, счётчики объектов недвижимости не реплицируются.

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

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

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

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

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

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

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДа→ BuildingId (UUID, ключ ORDER BY); Guid.Empty → poison (DLQ)
ProjectIdGuidДа→ ProjectId (родитель); Guid.Empty → poison
TenantIdGuidДа→ DeveloperId (денормализация вверх по иерархии, ADR-0052); Guid.Empty → poison
NamestringДа→ Name («Литер 3», подпись графика)
Statusstring (enum BuildingStatus)Да→ Status (LowCardinality): Draft/Published/Hidden; unknown → сохранить как есть + warning + метрика (на чтении — как Hidden)
ConstructionStatusstring (enum)Да→ ConstructionStatus; unknown → сохранить как есть + warning (fallback чтения UnderConstruction)
FloorsTotalshortДа→ FloorsTotal (Int16, подпись корпуса)
VersionlongДа→ SourceVersion (guard/диагностика)
SectionsCount,DeliveryYear,DeliveryQuarter,ReadinessPercent,PropertyObjects*Count——не реплицируются

Колонки, заполняемые консюмером: UpsertedAt = время события.

Алгоритм​

Консюмер батчевый: шаги 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_building (DeveloperId = TenantId, UpsertedAt = время события, SourceVersion = Version); переходов состояний нет. Добавить в буфер; внутрибатчевая свёртка по BuildingId (max UpsertedAt).
  4. Пакетно вставить (upsert) буфер строк в pin_dev_stats.dim_building. Upsert обеспечивает ReplacingMergeTree(UpsertedAt) (merge → max UpsertedAt per BuildingId); точечное обновление/удаление не применяются (Status = Hidden — обычная строка snapshot-а). Смещение (offset) фиксируется строго после подтверждённой вставки; при ошибке — retry всего батча.

Диаграмма​

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

  • Business key: BuildingId; версия-precedence — UpsertedAt (ReplacingMergeTree). Дубль перезаписывается идемпотентно; до merge чтение дубль-устойчиво — argMax(Name, UpsertedAt) … GROUP BY BuildingId (FINAL запрещён).
  • Version guard (dimensions/building.md) — декларативный: per-key-ordering
    • ReplacingMergeTree(UpsertedAt) + внутрибатчевая свёртка; version-lookup нет (ADR-0056).
  • Replay/backfill: да — повтор топика building-upserted группой dev-stats-dim-building безопасен; первичное наполнение — 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), unknown-enum (Status/ConstructionStatus).
  • Metrics: dev_stats_dim_building_upserted_total, dev_stats_dim_ingest_lag_seconds, dev_stats_dim_ingest_batch_size, chessboard_sync_unknown_enum_total, retry/DLQ.
  • Traces: consume span + insert span; корреляция из метаданных события.
  • PII/masking: PII отсутствует (здание — объект). Tenant: DeveloperId = TenantId (ADR-0052); scope-фильтрация — на чтении в API.

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

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

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

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

Acceptance Criteria​

  • Happy path (создание): Дано — миграции применены; Когда — приходит building-upserted (B1, ProjectId = P1, Name = "Литер 3"); Тогда — argMax(Name, UpsertedAt) … GROUP BY BuildingId для B1 = "Литер 3", DeveloperId = TenantId.
  • Обновление (новая версия): Дано — строка B1 есть; Когда — событие с большим UpsertedAt и новым Status; Тогда — чтение отдаёт новый статус (старая версия схлопывается при merge).
  • Идемпотентность (повтор доставки): Дано — событие B1 применено; Когда — повторная доставка; Тогда — логических дублей нет, offset продвинут.
  • Poison: Дано — Id/ProjectId/TenantId == Guid.Empty или битый payload; Тогда — DLQ (poison) + error-лог, партиция продолжается.
  • Unknown-enum: Дано — Status/ConstructionStatus вне набора; Тогда — значение сохраняется как есть + warning + метрика; на чтении fallback (Hidden/UnderConstruction).
  • Отказ хранилища: Дано — ClickHouse недоступен; Тогда — offset не коммитится, retry с backoff, догон без дублей.

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

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

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