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

Consumer: DevStatsFunnelBookingCreatedConsumer

Наполняет этап Bookings воронки funnel_daily: создание брони (Резерв) → uniqState(BookingId) по срезу (иерархия × дата). Обогащение иерархией — из dim_property_object по PropertyObjectId. Часть views → favorites → selections → leads → bookings → deals (ADR-0057).

Назначение​

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

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

ПолеЗначение
Сервис/контекстdev-stats (DevStats.Background)
ConsumerDevStatsFunnelBookingCreatedConsumer
Обрабатываемое событиеbooking-created
Topicpin.crm.booking-created.v1
ContractVersion1 (warn-and-proceed)
Consumer groupdev-stats-funnel-crm-booking
Producercrm (операция create-booking → Reserved)
Эффект (событие → что меняется)pin_dev_stats.funnel_daily, этап Bookings = uniqState(BookingId)
Feature flagdev-stats.funnel.enabled

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

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

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

Контракт: event: booking-created. Consumer читает:

Поле payloadТипОбяз.Как используется
BookingIdGuidДабизнес-ключ этапа → uniqState(BookingId)
PropertyObjectIdGuidДаlookup dim_property_object → иерархия
TenantIdGuidДаtenant-guard: должен = dim_property_object.DeveloperId (иначе fail-closed)
CreatedAtDateTimeOffsetДаEventDate = toDate(CreatedAt)

InitiatedBy/ProgramTerms/ReserveExpiresAt в воронке не используются.

Алгоритм​

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

  1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warn-and-proceed); идемпотентность по BookingId/MessageId — повтор пропускается (no-op).
  2. Проверить payload (BookingId/PropertyObjectId/TenantId/CreatedAt); при невалидном — пропуск с warning (skip + warning).
  3. Обогащение иерархией: по PropertyObjectId взять из dim_property_object последнюю версию пути DeveloperId → … → LayoutId (argMax(…иерархия…, UpsertedAt), без FINAL); строки нет → отложенное повторное обогащение (deferred re-enrich, backoff, ≤ N попыток; при устойчивом промахе → DLQ + алерт «dimension gap»). Tenant-guard: dim_property_object.DeveloperId должен совпадать с TenantId события; иначе — пропуск + alert (нарушение tenant), строка не пишется (fail-closed, без кросс-tenant утечки).
  4. Собрать частичную строку funnel_daily: ключ (DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId, PropertyObjectId, EventDate), Bookings = uniqState(BookingId), прочие этапы — пустые *State().
  5. Пакетно вставить строку в хранилище funnel_daily (батч ≤ 1000 строк / 5 с); AggregatingMergeTree сольёт частичные состояния по ключу. Смещение (offset) фиксируется только после подтверждённой вставки.
  6. Записать метрики/логи; успех → ack.

Диаграмма​

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

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

Retry и DLQ​

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

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

  • Logs/metrics/traces — consume/enrich/ingest, счётчик tenant-mismatch. PII нет; DeveloperId клеймится из dim_property_object и сверяется с TenantId (fail-closed). Воронка агрегатна.

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

НаправлениеКонтракт
Ожидает от crmСобытие booking-created (pin.crm.booking-created.v1, ContractVersion = 1, at-least-once) при создании брони (переход объекта в Reserved). Обязательные для воронки поля: BookingId (уникален и стабилен на бронь → идемпотентность этапа), PropertyObjectId, TenantId = DeveloperId, CreatedAt. Отмена/протухание брони этап Bookings не уменьшает (воронка кумулятивна по факту резерва; следующий этап — booking-paid).
Ожидает от chessboard (через dim-ingest)Строку dim_property_object с иерархией и DeveloperId; при отставании — late-dimension (park).
ПредоставляетЧастичную строку funnel_daily: Bookings = uniqState(BookingId) по срезу (иерархия × EventDate). Потребитель — read API get-conversion-funnel (uniqMerge, без FINAL). Порядок прихода этапов роли не играет — AggregatingMergeTree сливает состояния по ключу.
Окно eventual-consistency≤ 5 мин. В промежутке конверсия bookings/leads по объекту занижена — догоняет после batch-вставки; при parked-событии — после backfill измерения.

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

  • Предусловия: включён флаг dev-stats.funnel.enabled; для объекта наполнена строка dim_property_object; событие проходит валидацию payload и tenant-guard.
  • Постусловия: в funnel_daily по срезу (иерархия объекта, день CreatedAt) состояние Bookings учитывает BookingId ровно один раз; offset группы dev-stats-funnel-crm-booking продвинут строго после подтверждённой вставки; операционные данные crm не изменены.

Acceptance Criteria​

  • Happy path: Дано — флаг включён, dim_property_object для PO1 есть, TenantId = dim.DeveloperId; Когда — приходит booking-created (B1, PO1, день D); Тогда — uniqMerge(Bookings) по срезу PO1 × D увеличивается на 1.
  • Идемпотентность (повтор доставки): Дано — B1 учтён; Когда — повторная доставка того же события; Тогда — Bookings не меняется, offset продвинут.
  • Tenant-guard: Дано — TenantId события ≠ dim_property_object.DeveloperId; Когда — обработка; Тогда — строка не пишется, skip + alert (нарушение tenant), кросс-tenant утечки нет.
  • Late-dimension: Дано — строки dim_property_object нет; Когда — событие обработано; Тогда — сообщение паркуется; после появления измерения — дозаполнение; при устойчивом промахе — DLQ + алерт.
  • Флаг выключен / битый payload: Дано — funnel.enabled = off или отсутствует PropertyObjectId; Когда — обработка; Тогда — этап не наполняется, событие ack/skip без искажения агрегатов.
  • Отказ хранилища: Дано — ClickHouse недоступен; Когда — вставка падает; Тогда — offset не коммитится, retry с backoff, догон без дублей.

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

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

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