Consumer: DevStatsFunnelSelectionItemAddedConsumer

Наполняет этап Selections воронки funnel_daily: добавление объекта недвижимости в подборку → uniqState(SelectionItemId) по срезу (иерархия × дата). Обогащение — из dim_property_object по PropertyObjectId. Часть views → favorites → selections → leads → bookings → deals (ADR-0057). Источник — контракт pin.selections.selection-item-added.v1 (selections эмитит поэлементно: одно событие на добавленный объект недвижимости через PUT /selections/{id}/items и на каждый скопированный через POST /selections/from-favorites).

Назначение

Consumer DevStatsFunnelSelectionItemAddedConsumer слушает событие selection-item-added (топик pin.selections.selection-item-added.v1), по PropertyObjectId резолвит иерархию объекта недвижимости из dim_property_object и добавляет частичную строку funnel_daily с заполненным состоянием этапа Selections; прочие этапы — пустые *State(). Запись — append-only в аналитическое хранилище pin_dev_stats, не source of truth (данные подборок — в selections).

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

Поле	Значение
Сервис/контекст	dev-stats (DevStats.Background)
Consumer	DevStatsFunnelSelectionItemAddedConsumer
Обрабатываемое событие	selection-item-added (контракт)
Topic	pin.selections.selection-item-added.v1
ContractVersion	1 (при несовпадении — warn-and-proceed)
Consumer group	dev-stats-funnel-selections
Producer	selections (поэлементное событие подборки)
Эффект (событие → что меняется)	pin_dev_stats.funnel_daily, этап Selections = uniqState(SelectionItemId)
Feature flag	dev-stats.funnel.enabled (этап Selections активен)

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

• Из метаданных события доступны идентификатор сообщения (MessageId), время события (CreatedAt) и идентификатор источника (AggregateId).
• Сквозной трейс восстанавливается из метаданных события.
• Ключ идемпотентности — uniqState(SelectionItemId) + отсев повторного MessageId в батче.

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

Дом контракта — selections/events/selection-item-added.md. Consumer читает три поля:

Поле payload	Тип	Обяз.	Как используется
SelectionItemId	Guid	Да	бизнес-ключ этапа → uniqState(SelectionItemId)
PropertyObjectId	Guid	Да	lookup dim_property_object → иерархия
OccurredAt	DateTimeOffset	Да	EventDate = toDate(OccurredAt)

SelectionId/OwnerUserId/SelectionItemsCount/Version в воронке не используются (агрегат по объекту недвижимости, не по подборке/пользователю) — читаются как игнорируемые additive-поля.

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

Ожидает от selections (входной контракт selection-item-added): поэлементное событие (одно на добавленный объект недвижимости) с SelectionItemId (уникален на добавление — эмитится только для нового элемента, survived/reorder не re-эмитятся), PropertyObjectId, OccurredAt; key = SelectionId (3 партиции, порядок добавлений в одну подборку сохранён); публикуется атомарно с изменением состава; доставка at-least-once. Иерархию событие не несёт.

Ожидает от dim_property_object (внутри dev-stats): строку иерархии по PropertyObjectId (argMax(…, UpsertedAt) без FINAL); отсутствие → deferred re-enrich.

Предоставляет downstream: частичную строку funnel_daily с заполненным этапом Selections = uniqState(SelectionItemId) (прочие этапы — пустые *State()); AggregatingMergeTree сольёт по ключу (DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId, PropertyObjectId, EventDate); окно свежести ≤ 5 мин. Этап активен под флагом dev-stats.funnel.enabled; нижние этапы воронки могут отставать от Views (разные источники) — это ожидаемое eventual-consistency-окно, не ошибка.

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

Предусловия: таблица funnel_daily и MV созданы; флаг dev-stats.funnel.enabled включён для этапа Selections; dim_property_object наполнено. Постусловие: для валидного, обогащённого события в funnel_daily есть частичная строка среза с uniqState(SelectionItemId); повтор SelectionItemId счётчик не растит; offset сдвинут после подтверждённой вставки.

Алгоритм

1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warn-and-proceed); идемпотентность по SelectionItemId/MessageId — повтор пропускается (no-op).
2. Проверить payload (SelectionItemId/PropertyObjectId/OccurredAt); при невалидном — пропуск с warning (skip + warning).
3. Обогащение иерархией: по PropertyObjectId взять из dim_property_object последнюю версию пути (argMax(…иерархия…, UpsertedAt), без FINAL); строки нет → отложенное повторное обогащение (deferred re-enrich).
4. Собрать частичную строку funnel_daily: ключ (DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId, PropertyObjectId, EventDate), Selections = uniqState(SelectionItemId), прочие — пустые *State().
5. Пакетно вставить строку в хранилище funnel_daily (батч ≤ 1000 строк / 5 с); AggregatingMergeTree сольёт состояния по ключу. Смещение (offset) фиксируется только после подтверждённой вставки.
6. Записать метрики/логи; успех → ack.

Диаграмма

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

• Business key SelectionItemId (uniqState) — повтор не растит счётчик; + MessageId в батче. Replay с offset безопасен. Late-dimension → deferred re-enrich (как property_object_view_facts).

Retry и DLQ

Сбой	Retry	DLQ	Ручное действие
Транзиент ClickHouse	Да (offset не закоммичен)	Нет	авто-догон
Нет строки dim_property_object	deferred re-enrich	Нет	backfill dim_property_object
Битый payload	Нет	Нет	skip + warning

Acceptance criteria

• Happy. Given валидное selection-item-added и существующая строка dim_property_object, When обработано, Then в funnel_daily растёт этап Selections среза объекта на дату toDate(OccurredAt); прочие этапы не затронуты.
• Идемпотентный повтор. Given то же SelectionItemId/MessageId доставлено повторно, When обработано, Then uniqState(SelectionItemId) не задваивается (повтор — no-op).
• Late-dimension. Given строки dim_property_object нет, When обрабатывается, Then deferred re-enrich; после backfill измерения событие обогащается; строка воронки не теряется.
• Битый payload. Given отсутствует SelectionItemId/PropertyObjectId/OccurredAt, When обработано, Then skip + warning (без DLQ — воронка не критична к единичной потере, source of truth в selections).
• Флаг выключен. Given dev-stats.funnel.enabled = false, Then этап Selections не наполняется/скрыт на чтении, апстрим-события не ломают ingestion.

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

• Logs/metrics/traces — как у прочих funnel-консюмеров (consume/enrich/ingest): consumed, enriched/deferred (late-dimension), ingested (строк/длительность), skipped (битый payload). PII нет; DeveloperId клеймится из dim_property_object (fail-closed). Воронка агрегатна; наружу — только счётчики этапов.

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

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

• События (3): События контекста selections: карта топиков и отклонённые события, События dev-stats, Event Contract: pin.selections.selection-item-added (SelectionItemAddedKafkaEvent)
• Консюмеры (1): Consumer: DevStatsFunnelFavoriteAddedConsumer
• index.md (2): DevStats Service (Pin.DevStats), Selections Service (Pin.Selections)
• facts (1): funnel_daily (ClickHouse rollup / воронка)

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

• Контракт события: ../../selections/events/selection-item-added.md; producer-операции: set-selection-items.md, create-selection-from-favorites.md.
• Роллап: funnel-daily-rollup.md; измерение: dim_property_object.
• Соседи-воронки: favorite-added.md, lead-created.md.
• ADR-0057, ADR-0056.