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

Consumer: DevStatsFunnelLeadCreatedConsumer

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

Зависимость по контракту (важно). Текущий pin.crm.lead-created.v1 не несёт PropertyObjectId (payload: LeadId, TenantId = AgencyCompanyId, AgentUserId, InvestorUserId, ClientId, Source, ReferralLinkId, даты) — лид это пара «инвестор↔агент», объект недвижимости у него появляется только по источнику (BookingRequested/ConsultationRequested, дом lead.md). Этап Leads воронки требует объект недвижимости для привязки к иерархии застройщика, поэтому нужен additive-only OriginPropertyObjectId: Guid? в контракте (без bump версии). Пока поле не выпущено — consumer выключен флагом dev-stats.funnel.enabled, лиды без резолвимого объекта недвижимости не учитываются в объект недвижимости-уровневой воронке (счётчик Leads пуст, конверсия пропускает этап). Ниже алгоритм описан для контракта с OriginPropertyObjectId.

Назначение​

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

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

ПолеЗначение
Сервис/контекстdev-stats (DevStats.Background)
ConsumerDevStatsFunnelLeadCreatedConsumer
Обрабатываемое событиеlead-created
Topicpin.crm.lead-created.v1
ContractVersion1 (warn-and-proceed при несовпадении)
Consumer groupdev-stats-funnel-crm-lead
Producercrm (лид-заявка / referral-attach / автосоздание при фиксации)
Эффект (событие → что меняется)pin_dev_stats.funnel_daily, этап Leads = uniqState(LeadId)
Feature flagdev-stats.funnel.enabled (ждёт OriginPropertyObjectId)

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

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

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

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

Поле payloadТипОбяз.Как используется
LeadIdGuidДабизнес-ключ этапа → uniqState(LeadId)
OriginPropertyObjectIdGuid?Условноlookup dim_property_object → иерархия; null → лид не учитывается в объект недвижимости-воронке (skip)
CreatedAtDateTimeOffsetДаEventDate = toDate(CreatedAt)

TenantId(=AgencyCompanyId), ФИО/PII — в воронке не используются (агрегат по объекту недвижимости застройщика).

Алгоритм​

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

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

Диаграмма​

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

  • Business key LeadId (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
OriginPropertyObjectId = null / отсутствует в контрактеНетНетskip + warning (см. зависимость выше)
Версия события ≠ ContractVersionНе сбой: warning-логНетсверить миграцию контракта crm
Сбой между вставкой и commit offsetПовторная доставка (at-least-once)Нетне требуется — uniqState(LeadId) идемпотентен

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

  • Logs/metrics/traces — consume/enrich/ingest, счётчик skip-по-OriginPropertyObjectId. PII (ФИО/телефон/ClientId) в воронку не пишется; DeveloperId клеймится из dim_property_object (fail-closed). Воронка агрегатна.

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

НаправлениеКонтракт
Ожидает от crmСобытие lead-created (pin.crm.lead-created.v1, ContractVersion = 1, at-least-once). Обязательные: LeadId (уникален и стабилен на лид → идемпотентность этапа), CreatedAt. Требуется additive-only OriginPropertyObjectId: Guid? — без него привязка к иерархии застройщика невозможна (см. блок «Зависимость по контракту»). TenantId = AgencyCompanyId — это агентство лида, не застройщик, и в воронку не переносится; застройщик резолвится из иерархии объекта. ФИО/телефон/ClientId — не переносятся.
Ожидает от chessboard (через dim-ingest)Строку dim_property_object с иерархией и DeveloperId для OriginPropertyObjectId; при отставании — late-dimension (park).
ПредоставляетЧастичную строку funnel_daily: Leads = uniqState(LeadId) по срезу (иерархия × EventDate). Потребитель — read API get-conversion-funnel (uniqMerge, без FINAL).
Окно eventual-consistency≤ 5 мин. Пока OriginPropertyObjectId не выпущен producer-ом — этап Leads объектной воронки пуст (конверсия пропускает этап; зафиксировано в блоке зависимости и «Известных ограничениях» README). После выпуска и включения флага — этап наполняется, историческая глубина восстанавливается replay/backfill.

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

  • Предусловия: включён флаг dev-stats.funnel.enabled; контракт lead-created несёт OriginPropertyObjectId; для объекта наполнена строка dim_property_object; событие валидно.
  • Постусловия: для лида с резолвимым объектом — в funnel_daily по срезу (иерархия объекта, день CreatedAt) состояние Leads учитывает LeadId ровно один раз; лид без объекта — не учитывается в объектной воронке (счётчик skip); offset группы dev-stats-funnel-crm-lead продвинут после подтверждённой вставки; операционные данные crm не изменены.

Acceptance Criteria​

  • Happy path: Дано — флаг включён, контракт с OriginPropertyObjectId, dim_property_object для PO1 есть; Когда — приходит lead-created (L1, OriginPropertyObjectId = PO1, день D); Тогда — uniqMerge(Leads) по срезу PO1 × D увеличивается на 1.
  • Идемпотентность (повтор доставки): Дано — L1 учтён; Когда — повторная доставка; Тогда — Leads не меняется.
  • Лид без объекта: Дано — OriginPropertyObjectId = null (или поле ещё не в контракте); Когда — обработка; Тогда — событие skip + инкремент счётчика skip-по-OriginPropertyObjectId, объектная воронка не изменяется.
  • Late-dimension: Дано — строки dim_property_object нет; Тогда — park → дозаполнение после измерения; при устойчивом промахе — DLQ + алерт.
  • Отказ хранилища: Дано — ClickHouse недоступен; Тогда — offset не коммитится, retry с backoff, догон без дублей.

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

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

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