Consumer: DevStatsProjectUpsertedConsumer

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

Назначение

Consumer DevStatsProjectUpsertedConsumer слушает событие project-upserted (топик pin.chessboard.project-upserted.v1) и пишет срез проекта (Name/Status/Type/Class/ ConstructionStatus/RegionId+RegionName + DeveloperId-родитель + версия) в dim_project. Измерение даёт подпись графика (Name) и атрибуты разбивки/фильтра просмотров по проекту без JOIN-ов; факты уже несут денормализованный ProjectId/DeveloperId. Owned-состояния в PostgreSQL нет (ClickHouse-only, ADR-0057); денормализованные цены/счётчики (PriceMin/PropertyObjectsFreeCount), jsonb-вкладки, Stars/HotelierId не реплицируются (метрики считаются по фактам просмотров, а не по атрибутам продаж).

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

Поле	Значение
Сервис/контекст	dev-stats (проект DevStats.Background)
Consumer	DevStatsProjectUpsertedConsumer
Обрабатываемое событие	project-upserted
Topic	pin.chessboard.project-upserted.v1
ContractVersion	1; при несовпадении — warn-and-proceed
Consumer group	dev-stats-dim-project
Producer	chessboard (Developers API create/update/publish/hide, синк Talent, job project-aggregates-recalc)
Эффект (событие → что меняется)	upsert строки в pin_dev_stats.dim_project (ReplacingMergeTree)

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

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

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

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

Поле payload	Тип	Обязательность	Как используется (маппинг)
Id	Guid	Да	→ ProjectId (UUID, ключ ORDER BY); Guid.Empty → poison (DLQ)
DeveloperId	Guid	Да	→ DeveloperId (родитель вверх по иерархии, = TenantId); Guid.Empty → poison
Name	string	Да	→ Name (подпись графика)
Status	string (enum ProjectStatus)	Да	→ Status (LowCardinality): Draft/Published/Hidden/Archived; unknown → сохранить как есть + warning + метрика (на чтении — как Hidden, fail closed)
Type	string (enum ProjectType)	Да	→ Type; unknown → сохранить как есть + warning (fallback чтения ResidentialComplex, reference-data.md)
Class	string (enum ProjectClass)	Да	→ Class; unknown → сохранить как есть + warning (fallback Comfort)
ConstructionStatus	string (enum)	Да	→ ConstructionStatus; unknown → сохранить как есть + warning (fallback UnderConstruction)
RegionId	long?	Нет	→ RegionId (Int64, ключ разбивки/фильтра); null до резолва гео в источнике (у Published заполнен, INV-PRJ-3 chessboard)
RegionSnapshot	snapshot (RegionSnapshot?)	Нет (⇔ RegionId != null)	→ RegionName (денормализованное имя региона из RegionSnapshot.Name, snapshots.md); пусто при RegionId = null
Version	long	Да	→ SourceVersion (guard/диагностика)
TenantId	Guid	Да	сверяется с DeveloperId (tenant boundary, ADR-0052); mismatch → warning
Stars,HotelierId,Price,CurrencyId/CurrencySnapshot,CountryId/CountrySnapshot,CityId/CitySnapshot,PropertyObjectsTotalCount,PropertyObjectsFreeCount,PaybackYears,CoverImageFileId	—	—	не реплицируются (страна/город — вне минимального набора dev-stats; разбивку по гео несёт только регион)

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

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

Ожидает от chessboard (входной контракт project-upserted):

• Поля snapshot-а: Id, DeveloperId, Name, Status, Type, Class, ConstructionStatus, RegionId, RegionSnapshot (гео-ссылка + display-снапшот региона), Version, TenantId (перечень и типы — таблица payload выше). Страна/город (CountryId/CityId + снапшоты) и валюта (CurrencyId/CurrencySnapshot) событием несутся, но в измерение не реплицируются.
• Гарантии продюсера: key = Id → все события одного проекта попадают в одну партицию и приходят по порядку публикации; Version монотонно растёт per-project; событие публикуется атомарно со снапшотом проекта (нет «полу-обновлённых» строк); доставка at-least-once (возможны дубликаты); совместимость additive-only для minor (новые optional-поля без bump ContractVersion), breaking → новый топик .v2.
• Порядок между агрегатами разных уровней (проект vs объект недвижимости vs застройщик) не гарантирован — измерение самодостаточно и не зависит от прихода родителя.

Предоставляет downstream (в пределах dev-stats, без публикации событий — leaf):

• JOIN-free строку dim_project для подписи (Name) и разбивки/фильтра метрик по уровню Project; ключ адресации — ProjectId; окно свежести ≤ 5 мин (событие → видимость в чтении).
• Измерение не лежит на критическом пути обогащения факта: иерархию просмотра консюмер факта резолвит из dim_property_object (несёт денормализованный ProjectId), а dim_project нужен только для человекочитаемой подписи/фильтра при чтении метрик.

Поведение при отсутствии/устаревании строки:

• Строки dim_project ещё нет на момент чтения метрики (родитель не долетел) → метрика всё равно резолвится по ProjectId из факта, подпись деградирует до ProjectId (плейсхолдер), ошибка наружу не отдаётся.
• Устаревшая доставка (повтор/переупорядочивание) → precedence по UpsertedAt: при чтении побеждает строка с максимальным UpsertedAt; SourceVersion служит диагностикой рассинхронизации, не блокирует запись.

Алгоритм

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

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

Диаграмма

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

Предусловия (до обработки батча): таблица dim_project создана миграцией; consumer group dev-stats-dim-project зарегистрирована на топике; при первом вводе offset стартует с earliest (полное наполнение измерения). Существование родителя (dim_developer) для обработки не требуется.

Постусловие (после успешного батча): для каждого валидного Id в dim_project присутствует строка с UpsertedAt = время события и SourceVersion = Version; при последующем merge остаётся строка с максимальным UpsertedAt на ProjectId; offset сдвинут за обработанный диапазон. Poison-сообщения ушли в DLQ, батч их не блокирует. Интеграционные события не публикуются, PostgreSQL-строки не меняются (ClickHouse-only).

Acceptance criteria

• Happy upsert. Given валидное project-upserted (непустые Id/DeveloperId), When консюмер обработал батч, Then в dim_project есть строка проекта с его Name/Status/Type/Class/ConstructionStatus/ RegionId/RegionName и UpsertedAt = время события; чтение argMax(…, UpsertedAt) GROUP BY ProjectId (без FINAL) отдаёт эти значения.
• Идемпотентный повтор. Given то же событие доставлено дважды (at-least-once), When оба обработаны, Then после merge логически одна строка на ProjectId; до merge дубль-устойчивое чтение не задваивает подпись.
• Устаревшая доставка. Given повтор старого снапшота с меньшим UpsertedAt при уже применённом свежем, When обработан, Then при чтении побеждает строка с максимальным UpsertedAt (старая не понижает).
• Unknown enum. Given Status/Type/Class/ConstructionStatus вне известного множества, When обработан, Then значение сохранено как есть + warning + метрика chessboard_sync_unknown_enum_total; на чтении применяется fail-closed/ fallback (Status→Hidden, Type→ResidentialComplex, Class→Comfort, ConstructionStatus→UnderConstruction).
• Poison. Given Id == Guid.Empty или DeveloperId == Guid.Empty или ошибка десериализации, When обработан, Then сообщение уходит в DLQ + error-лог, батч продолжается, offset остальных сдвигается.
• Tenant-mismatch. Given TenantId ≠ DeveloperId, When обработан, Then warning tenant-mismatch, строка всё равно пишется с DeveloperId из payload (диагностика, не отказ).
• Отсутствие downstream-строки. Given метрика читается по проекту, чья строка dim_project ещё не долетела, Then метрика резолвится по ProjectId, подпись деградирует до id без ошибки.

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

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

Retry и DLQ

Сбой	Retry	DLQ	Ручное действие
ClickHouse недоступен / timeout вставки	Да: retry батча, backoff (1с → 60с), offset не commit-ится	Нет (инфраструктура)	Алерт ingest-lag > 15 мин; восстановить CH
Ошибка десериализации payload	Нет	Да (poison)	Разбор из DLQ; issue producer-у chessboard
Id/DeveloperId == 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/Type/Class/ConstructionStatus), tenant-mismatch (TenantId ≠ DeveloperId).
• Metrics: dev_stats_dim_project_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 (DeveloperId = актор).

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

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

• События (2): События dev-stats, Event Contract: pin.chessboard.project-upserted (ProjectUpsertedKafkaEvent)
• Консюмеры (5): Consumer: DevStatsBuildingUpsertedConsumer, Consumer: DevStatsDeveloperUpsertedConsumer, Consumer: DevStatsLayoutGroupUpsertedConsumer, Consumer: DevStatsLayoutUpsertedConsumer, Consumer: DevStatsPropertyObjectUpsertedConsumer

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

• Дом колонок и dedup: dimensions/project.md; контракт события: chessboard/events/project-upserted.md; snapshot-модель региона (RegionSnapshot): handbook/domain/snapshots.md.
• Соседние ingest-консюмеры: developer-upserted.md, building-upserted.md, layout-group-upserted.md, layout-upserted.md, property-object-upserted.md.
• Обзор: ../README.md; backfill: ../background/dimension-backfill.md.
• ADR-0052, ADR-0056, ADR-0057.