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

Consumer: DevStatsFunnelFavoriteAddedConsumer

Наполняет этап Favorites роллапа-воронки funnel_daily: каждое добавление объекта недвижимости в избранное → uniqState(FavoriteId) по срезу (иерархия × дата). Обогащение иерархией (DeveloperId → … → LayoutId) — из dim_property_object по PropertyObjectId (событие несёт PropertyObjectId, но не иерархию). Часть воронки views → favorites → selections → leads → bookings → deals (ADR-0057).

Назначение​

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

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

ПолеЗначение
Сервис/контекстdev-stats (DevStats.Background)
ConsumerDevStatsFunnelFavoriteAddedConsumer
Обрабатываемое событиеfavorite-added
Topicpin.favorites.favorite-added.v1
ContractVersion1 (при несовпадении — warn-and-proceed)
Consumer groupdev-stats-funnel-favorites
Producerfavorites (добавление в избранное / импорт избранного)
Эффект (событие → что меняется)pin_dev_stats.funnel_daily, этап Favorites = uniqState(FavoriteId)
Feature flagdev-stats.funnel.enabled

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

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

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

Контракт: event: favorite-added. Consumer читает только:

Поле payloadТипОбяз.Как используется
FavoriteIdGuidДабизнес-ключ этапа → uniqState(FavoriteId) (идемпотентность)
PropertyObjectIdGuidДаlookup dim_property_object → иерархия (DeveloperId … LayoutId)
OccurredAtDateTimeOffsetДаEventDate = toDate(OccurredAt) — ключ дня

UserId/AnonymousId/Note в воронке не используются (воронка про конверсию объекта недвижимости, не сегмент/PII).

Алгоритм​

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

  1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warn-and-proceed); идемпотентность по FavoriteId/MessageId — повтор пропускается (no-op).
  2. Проверить payload: заполнены FavoriteId/PropertyObjectId/OccurredAt; OccurredAt == default или > now() + 1 ч — расценивается как битый payload (skip + warning); иначе — пропуск с warning (skip + warning).
  3. Обогащение иерархией: по PropertyObjectId взять из dim_property_object последнюю версию пути DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId (argMax(…, UpsertedAt), без FINAL); строки нет → late-dimension: отложенное повторное обогащение (park/deferred re-enrich, backoff, ≤ N попыток; при устойчивом промахе → DLQ + алерт «dimension gap», см. funnel-daily-rollup). Просмотр/этап не теряется — восстанавливается после backfill измерения и replay.
  4. Собрать частичную строку funnel_daily: ключ (DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId, PropertyObjectId, EventDate), Favorites = uniqState(FavoriteId), прочие этапы — пустые *State().
  5. Пакетно вставить строку в хранилище pin_dev_stats.funnel_daily (батч ≤ 1000 строк / 5 с); AggregatingMergeTree сольёт частичные состояния по ключу. Смещение (offset) фиксируется только после подтверждённой вставки.
  6. Записать метрики/логи; успех → ack, контролируемая ошибка → retry/DLQ.

Диаграмма​

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

  • Business key: FavoriteId (uniqState) — повтор события не увеличивает счётчик этапа; дополнительно отсев MessageId в батче. Replay топика/переигровка с offset безопасны (вставка идемпотентна).
  • Late-dimension (dim_property_object ещё не пришёл): стратегия park/deferred re-enrich — как у property_object_view_facts.

Retry и DLQ​

СбойRetryDLQРучное действие
Транзиент ClickHouse / пакетной вставкиДа (offset не закоммичен)Нетавто-догон по offset
Нет строки dim_property_object (late-dimension)deferred re-enrich (backoff, ≤ N попыток)Да при устойчивом промахе + алерт «dimension gap»backfill dim_property_object, затем replay из DLQ
Битый payload (нет PropertyObjectId / невалидный OccurredAt)НетНетskip + warning (данные невосстановимы)
Версия события ≠ ContractVersionНе сбой: warning-логНетсверить миграцию контракта favorites
Сбой между вставкой и commit offsetПовторная доставка (at-least-once)Нетне требуется — uniqState(FavoriteId) идемпотентен

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

  • Logs: consumed / skipped-duplicate / enriched / deferred / failed. Metrics: consume duration, batch size, ingest lag, retries, duplicates. Traces: consume + enrich + ingest spans (корреляция из метаданных события).
  • PII: нет (UserId/AnonymousId в воронку не пишутся); tenant DeveloperId клеймится из dim_property_object (fail-closed при отсутствии). Воронка агрегатна — персональные данные не раскрываются.

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

НаправлениеКонтракт
Ожидает от favoritesСобытие favorite-added (pin.favorites.favorite-added.v1, ContractVersion = 1, at-least-once). Гарантии: FavoriteId уникален и стабилен на факт добавления (повтор доставки — тот же id → идемпотентность этапа); PropertyObjectId указывает на существующий (в перспективе) объект недвижимости; OccurredAt — бизнес-время добавления. Удаление из избранного (RemoveFromFavorites) этап воронки не уменьшает (воронка кумулятивна по факту интереса).
Ожидает от chessboard (через собственный dim-ingest)Строку dim_property_object с полной иерархией DeveloperId → … → LayoutId и DeveloperId = TenantId. При отставании реплики — late-dimension (park).
ПредоставляетЧастичную строку funnel_daily: этап Favorites = uniqState(FavoriteId) по срезу (иерархия × EventDate); прочие этапы — пустые state. Потребитель — read API get-conversion-funnel (uniqMerge, без FINAL). Порядок прихода этапов воронки роли не играет — AggregatingMergeTree сливает состояния по ключу.
Окно eventual-consistency≤ 5 мин (event → видимость в API). В промежутке застройщик видит конверсию favorites/views заниженной по этому объекту — она догоняет после batch-вставки/enrich; при parked-событии — после backfill измерения.

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

  • Предусловия: включён флаг dev-stats.funnel.enabled; для объекта наполнена строка dim_property_object (иначе событие паркуется, а не теряется); событие проходит валидацию payload.
  • Постусловия: в funnel_daily по срезу (иерархия объекта, день OccurredAt) состояние Favorites учитывает FavoriteId ровно один раз (независимо от числа доставок); offset группы dev-stats-funnel-favorites продвинут строго после подтверждённой вставки; операционные данные favorites не изменены (запись только в аналитическое хранилище).

Acceptance Criteria​

  • Happy path: Дано — флаг включён, dim_property_object для PO1 есть; Когда — приходит favorite-added (F1, PO1, день D); Тогда — uniqMerge(Favorites) по срезу PO1 × D увеличивается на 1, иерархия соответствует dim_property_object.
  • Идемпотентность (повтор доставки): Дано — F1 уже учтён; Когда — то же событие доставлено повторно; Тогда — Favorites не меняется, offset продвинут (нет двойного счёта).
  • Late-dimension: Дано — строки dim_property_object для PO1 ещё нет; Когда — событие обработано; Тогда — сообщение паркуется и не влияет на счётчик; после появления измерения и replay — этап дозаполняется; при устойчивом промахе (> N попыток) — DLQ + алерт «dimension gap».
  • Флаг выключен: Дано — dev-stats.funnel.enabled = off; Когда — приходит событие; Тогда — этап не наполняется, событие ack/skip без записи.
  • Битый payload: Дано — событие без PropertyObjectId или с OccurredAt в будущем; Когда — обработка; Тогда — skip + warning, offset продвинут, агрегаты не искажены.
  • Отказ хранилища: Дано — ClickHouse недоступен; Когда — batch-вставка падает; Тогда — offset не коммитится, retry с backoff; после восстановления — догон без дублей.

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

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

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