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

Consumer: DevStatsDeveloperUpsertedConsumer

Один из шести ingest-консюмеров реплик-измерений dev-stats: принимает snapshot застройщика из контекста chessboard и upsert-ит строку в ClickHouse-измерение dim_developer (ReplacingMergeTree, pin_dev_stats). Верхнее измерение иерархии Developer → Project → Building → LayoutGroup → Layout → PropertyObject, из которой обогащаются факты просмотров (ADR-0057). Дом payload — event-contract: developer-upserted; дом колонок — dimensions/developer.md.

Назначение​

Consumer DevStatsDeveloperUpsertedConsumer слушает событие developer-upserted (топик pin.chessboard.developer-upserted.v1) и пишет минимальный срез профиля застройщика (Name/LegalName/ Status/LogoFileId + версия) в dim_developer. Назначение измерения — JOIN-free-подпись и проверка scope: факты уже несут денормализованный DeveloperId, а измерение резолвит человекочитаемое имя/лого для подписей графиков ЛК и подтверждает существование застройщика для scope-контроля (DeveloperId = актор). Owned-состояния в PostgreSQL нет — контекст ClickHouse-only (ADR-0057); показатели карточки (Highlights), FoundedYear, контакты не реплицируются (в dim_developer не нужны).

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

ПолеЗначение
Сервис/контекстdev-stats (проект DevStats.Background)
ConsumerDevStatsDeveloperUpsertedConsumer
Обрабатываемое событиеdeveloper-upserted
Topicpin.chessboard.developer-upserted.v1
ContractVersion1; при несовпадении — warn-and-proceed
Consumer groupdev-stats-dim-developer
Producerchessboard (Developers API профиля, consumer компаний identity при Type = Developer, синк Talent)
Эффект (событие → что меняется)upsert строки в pin_dev_stats.dim_developer (ReplacingMergeTree)

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

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

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

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

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДа→ DeveloperId (UUID, единственный ключ ORDER BY); Guid.Empty → poison (DLQ)
NamestringДа→ Name (подпись графиков ЛК)
LegalNamestring?Нет→ LegalName (Nullable(String))
Statusstring (enum DeveloperStatus)Да→ Status (LowCardinality(String)): Draft/Published/Hidden; unknown — сохранить как есть + warning + метрика chessboard_sync_unknown_enum_total (на чтении трактуется как наиболее ограничивающий Hidden)
LogoFileIdGuid?Нет→ LogoFileId (Nullable(UUID))
VersionlongДа→ SourceVersion (монотонный guard, диагностика гонок)
FoundedYear, Highlights——не реплицируются (в dim_developer не нужны)

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

Алгоритм​

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

  1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warning-лог и продолжить, warn-and-proceed). Дубль на входе не отсеивается — дедуп декларативен (движок + порядок сообщений по ключу).
  2. Проверить payload (по колонке «Обязательность»): ошибка десериализации или Id == Guid.Empty → DLQ (poison) + error-лог. У верхнего измерения родителей нет — snapshot самодостаточен, проверок связей нет.
  3. Маппинг поле-в-поле: построить строку dim_developer по таблице выше (UpsertedAt = время события, SourceVersion = Version); переходов состояний нет. Добавить в буфер батча; внутрибатчевая свёртка по DeveloperId — оставить строку с максимальным UpsertedAt (страховка от дублей в одном батче до merge).
  4. Пакетно вставить (upsert) буфер строк в pin_dev_stats.dim_developer. Логический upsert обеспечивает движок ReplacingMergeTree(UpsertedAt) (merge схлопывает до строки с max UpsertedAt per DeveloperId); точечное обновление/удаление не применяются. Смещение (offset) фиксируется строго после подтверждённой вставки батча; при ошибке — retry всего батча (идемпотентно), offset не двигается.

Диаграмма​

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

  • Business key: DeveloperId (dedup-ключ ORDER BY); версия-precedence — UpsertedAt (ReplacingMergeTree). Поведение дубля: строка перезаписывается идемпотентно, до merge чтение дубль-устойчиво — argMax(Name, UpsertedAt) … GROUP BY DeveloperId (FINAL в online запрещён, правило SDK).
  • Version guard (по dimensions/developer.md) реализован декларативно: per-key-ordering + ReplacingMergeTree(UpsertedAt) + внутрибатчевая свёртка — синхронного event.Version > SourceVersion-lookup нет (ADR-0056, batch-ingestion).
  • Replay/backfill: да — повтор топика developer-upserted с произвольного offset группой dev-stats-dim-developer безопасен (upsert идемпотентен); первичное наполнение измерений — часть backfill при вводе сервиса (background/dimension-backfill.md).

Retry и DLQ​

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

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

  • Logs: consumed (offset-диапазон, размер батча), skipped-poison (причина + Id), flushed (строк, длительность вставки), failed (ошибка CH), unknown-enum (Status). Число дублей не логируется — dedup на стороне CH.
  • Metrics: dev_stats_dim_developer_upserted_total, dev_stats_dim_ingest_lag_seconds (CreatedAt → вставка), dev_stats_dim_ingest_batch_size, chessboard_sync_unknown_enum_total, счётчики retry/DLQ.
  • Traces: consume span + insert span; корреляция из метаданных события.
  • PII/masking: PII отсутствует (застройщик — организация; контакты/Highlights не реплицируются). Tenant: DeveloperId = TenantId контекста (ADR-0052); консюмер платформенный, scope-фильтрация — на чтении в API.

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

НаправлениеКонтракт
Ожидает от chessboardСобытие developer-upserted (pin.chessboard.developer-upserted.v1, ContractVersion = 1, at-least-once). Гарантии: ключ сообщения = Id (per-developer ordering — старая версия не приходит после новой), монотонная Version, самодостаточный snapshot (у верхнего измерения родителей нет). Id = AggregateId, CreatedAt — время snapshot-а.
ПредоставляетСтроку dim_developer (ReplacingMergeTree, версия по UpsertedAt) с Name/LegalName/Status/LogoFileId. Потребители реплики: read API dev-stats (подписи графиков ЛК, подтверждение существования застройщика для scope DeveloperId = актор). В обогащении фактов не участвует напрямую — DeveloperId уже денормализован в dim_property_object.
Окно eventual-consistency≤ 5 мин (event → строка в измерении). Пока реплика отстаёт, подпись графика может показать прежнее имя застройщика; на резолв иерархии факта это не влияет.

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

  • Предусловия: миграции DDL измерения применены; событие валидно (Id != Guid.Empty).
  • Постусловия: в dim_developer для DeveloperId актуальна строка с максимальным UpsertedAt (merge схлопывает дубли); offset группы dev-stats-dim-developer продвинут строго после подтверждённой вставки; никакие операционные данные chessboard не изменены (реплика только для чтения).

Acceptance Criteria​

  • Happy path (создание): Дано — миграции применены; Когда — приходит developer-upserted (D1, Name = "Стройтрест"); Тогда — argMax(Name, UpsertedAt) … GROUP BY DeveloperId для D1 = "Стройтрест".
  • Обновление (новая версия): Дано — строка D1 есть; Когда — приходит событие с большим UpsertedAt и новым Name; Тогда — чтение отдаёт новое имя (старая версия схлопывается при merge).
  • Идемпотентность (повтор доставки): Дано — событие D1 уже применено; Когда — повторная доставка; Тогда — логических дублей нет (ReplacingMergeTree + внутрибатчевая свёртка), offset продвинут.
  • Poison: Дано — Id == Guid.Empty или битый payload; Когда — обработка; Тогда — сообщение в DLQ (poison) + error-лог, партиция продолжается.
  • Unknown-enum Status: Дано — Status вне известного набора; Когда — обработка; Тогда — значение сохраняется как есть + warning + метрика chessboard_sync_unknown_enum_total; на чтении трактуется как наиболее ограничивающий Hidden.
  • Отказ хранилища: Дано — ClickHouse недоступен; Когда — batch-вставка падает; Тогда — offset не коммитится, retry с backoff, догон без дублей.

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

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

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