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

Consumer: DevStatsPropertyObjectViewEndedConsumer

Пишущий процесс главного факта dev-stats: принимает завершённый просмотр объекта недвижимости из user-activity, обогащает его иерархией застройщика из dim_property_object и добавляет closed-строку факта property_object_view_facts в ClickHouse pin_dev_stats батчами. Реализует ADR-0057: трёхлетний массив просмотров с произвольными агрегациями по иерархии проект→здание→группа планировок→планировка→объект недвижимости и по типу зрителя живёт в ClickHouse, а не в операционном PostgreSQL user-activity (ADR-0056 п.2–3).

Назначение​

Consumer DevStatsPropertyObjectViewEndedConsumer слушает событие property-object-view-ended (топик pin.user-activity.property-object-view-ended.v1) и добавляет (никогда не обновляет и не удаляет) строку closed-факта просмотра. Событие несёт PropertyObjectId, SessionId, AnonymousId, UserId?, ViewSource, ViewedAt, LastActivityAt, DurationSeconds, но не несёт вышестоящую иерархию застройщика — консюмер резолвит DeveloperId → ProjectId → BuildingId → LayoutGroupId → LayoutId из dim_property_object перед вставкой, чтобы факт лёг в property_object_view_facts уже денормализованным (JOIN-free аналитика по любому уровню). Он же классифицирует ViewerKind (Guest/Agent/Investor) и переносит ViewSource.

Closed-строка (реальная DurationSeconds, IngestionTime позже open-факта из action-tracked) склеивается с open-строкой того же просмотра по PropertyObjectViewId при чтении (argMax(*, IngestionTime)) — см. «Deduplication» в property-object-view-facts.md. Owned-состояния в PostgreSQL консюмер не меняет: пишет только в ClickHouse.

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

ПолеЗначение
Сервис/контекстdev-stats (проект DevStats.Background)
ConsumerDevStatsPropertyObjectViewEndedConsumer
Обрабатываемое событиеproperty-object-view-ended
Topicpin.user-activity.property-object-view-ended.v1
ContractVersion1; при несовпадении — warn-and-proceed
Consumer groupdev-stats-property-object-view-ended
Produceruser-activity — worker activity-closer (закрытие просмотра по неактивности / форс-закрытие при закрытии родительской сессии)
Эффект (событие → что меняется)append closed-строки в pin_dev_stats.property_object_view_facts (ReplacingMergeTree)

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

  • Из метаданных события доступны идентификатор источника (AggregateId = PropertyObjectViewId), тип агрегата (PropertyObjectView), время события (CreatedAt), версия snapshot-а (Version).
  • Сквозной трейс восстанавливается из метаданных события; идентификатор сообщения (MessageId) прописывается в факт как SourceEventId (сквозная трассировка событие → строка CH).
  • Бизнес-идемпотентность — декларативная, на стороне ClickHouse: FactId = SourceEventId — последний компонент ORDER BY ReplacingMergeTree, повторная доставка (at-least-once) схлопывается при merge, а читающие запросы дубль-устойчивы (argMax(*, IngestionTime) по PropertyObjectViewId; FINAL запрещён — правило SDK). Дополнительно консюмер отсекает дубль того же SourceEventId в батче, чтобы rollup property_object_view_daily/funnel_views_mv не получил повтор. Отдельная таблица идемпотентности не используется: у консюмера нет PG-транзакции.

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

Контракт события: event-contract: property-object-view-ended. Дом колонок факта — property-object-view-facts.md («Columns»); ниже — только правила маппинга.

Поле payloadТипОбязательностьКак используется (маппинг)
PropertyObjectViewIdGuidДа→ PropertyObjectViewId (бизнес-ключ склейки open/closed); Guid.Empty → poison (DLQ)
PropertyObjectIdGuidДа→ PropertyObjectId; ключ обогащения из dim_property_object; Guid.Empty → poison
SessionIdGuidДа→ SessionId (связь с session_facts)
AnonymousIdGuidДа→ AnonymousId (идентификатор гостя; вход uniqViewers = coalesce(UserId, AnonymousId))
UserIdGuid?Нет→ UserId (NULL для гостя); вход классификации ViewerKind
ViewSourcestring (enum ViewSource)Да→ ViewSource; unknown-значение → Unknown + warning (fallback контракта)
ViewedAtDateTimeOffsetДа→ EventTime/ViewedAt; партиция toYYYYMM; default/> now + 1 ч → poison
LastActivityAtDateTimeOffsetДа→ LastActivityAt
DurationSecondsintДа→ DurationSeconds; < 0 → poison (INV-LV: >= 0)
FilterSnapshotAppliedFilterModel?Нетв факт не маппится (воронку каталога строит action-tracked); в логах не разворачивается

Колонки, которые консюмер заполняет сам: FactId = SourceEventId (идентификатор сообщения); IngestionTime = момент вставки; TenantId = DeveloperId (после обогащения); BrandId = NULL; иерархия (DeveloperId/ProjectId/BuildingId/LayoutGroupId/LayoutId) — из dim_property_object; ViewerKind — из UserId + роль (см. ниже).

Обогащение иерархией из dim_property_object​

Lookup WHERE PropertyObjectId = ? в dim_property_object, последняя версия — argMax(col, UpsertedAt) с GROUP BY PropertyObjectId (без FINAL); из строки берётся весь путь вверх (DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId). Lookup'ы батчатся: для батча сообщений — один WHERE PropertyObjectId IN (…) вместо запроса на строку.

Классификация ViewerKind (зависимость от контракта / assumption)​

ViewerKind ∈ Guest/Agent/Investor. UserId == null → Guest (полностью выводится из payload). Для аутентифицированного зрителя разделение Agent/Investor требует роли зрителя, которой в v1 нет ни в payload, ни в измерениях dev-stats (реплики dim_user нет по ADR-0056). Простой faithful-вариант v1: producer добавляет в событие optional-поле роли актора (ViewerRole, additive-only — bump ContractVersion не нужен) либо заводится реплика dim_user из identity. До этого аутентифицированный зритель пишется как Investor (доминирующий тип аудитории каталога), а Agent детектируется по мере появления сигнала роли. Требование зафиксировано в «Известных ограничениях» README.

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

Ожидает от user-activity (входной контракт property-object-view-ended):

  • Поля payload: PropertyObjectViewId, PropertyObjectId, SessionId, AnonymousId, UserId?, ViewSource, ViewedAt, LastActivityAt, DurationSeconds, FilterSnapshot?; MessageId в метаданных → SourceEventId.
  • Гарантии продюсера: событие публикуется worker-ом activity-closer атомарно с закрытием просмотра; DurationSeconds уже посчитан сервером по polling-delta (INV-LV ≥ 0); доставка at-least-once (дубли снимаются dedup по FactId); один PropertyObjectViewId разделяется с open-фактом action-tracked (ViewPropertyObject) для склейки open/closed по PropertyObjectViewId при чтении.
  • Событие не несёт иерархию застройщика — её обязан долить сам консюмер из dim_property_object.

Ожидает от dim_property_object (внутри dev-stats): строку с полным путём вверх по PropertyObjectId (гарантия консюмера property-object-upserted); lookup argMax(…, UpsertedAt) GROUP BY PropertyObjectId без FINAL. Отсутствие строки → late-dimension (park/deferred re-enrich/DLQ).

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

  • Append-only closed-строку property_object_view_facts уже денормализованную по иерархии — источник всех метрик внимания и вход rollup-ов property_object_view_daily/funnel_views_mv (стадия Views) и этапа воронки. Гарантия дубль-устойчивости чтения: argMax(*, IngestionTime) по PropertyObjectViewId, uniq* по PropertyObjectViewId; окно свежести факта ≤ 5 мин.

Алгоритм​

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

  1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warn-and-proceed). Дубль по SourceEventId на входе не отсеивается (dedup декларативен), но повтор в текущем батче отбрасывается.
  2. Проверить payload (по колонке «Обязательность»): ошибка десериализации → DLQ (poison), партиция продолжается; PropertyObjectViewId/PropertyObjectId == Guid.Empty → DLQ + error-лог; DurationSeconds < 0 (нарушение INV-LV: >= 0), ViewedAt == default или ViewedAt > now() + 1 ч → DLQ; unknown-enum ViewSource → Unknown + warning (не DLQ).
  3. Обогащение иерархией: batch-lookup dim_property_object по PropertyObjectId (argMax(…, UpsertedAt), без FINAL). Late-dimension (гонка: факт раньше property-object-upserted): строки нет → сообщение паркуется (deferred re-enrich): отложенный ретрай с backoff (напр. 30 c → 5 мин, ≤ N попыток); при устойчивом промахе → DLQ + алерт «dimension gap». Просмотр не теряется (см. dim_property_object.md).
  4. Маппинг поле-в-поле: построить строку факта property_object_view_facts — скалярные поля из payload; иерархия из dim_property_object; ViewerKind из UserId + роль; FactId = SourceEventId; TenantId = DeveloperId; IngestionTime — момент вставки. Переходов состояний нет. Добавить строку в буфер батча.
  5. Пакетно вставить (append-only) буфер строк в pin_dev_stats.property_object_view_facts (батч ≤ 1000 строк / 5 с); точечное обновление/удаление/upsert запрещены — append-only факт; в CH пишет только этот background-консюмер. Смещение (offset) фиксируется строго после подтверждённой вставки; сбой между вставкой и commit → повторная доставка → дубли схлопывает ReplacingMergeTree (at-least-once без потерь).
  6. Записать метрики/логи; успех → ack; ошибка вставки → retry всего батча (идемпотентно), offset не двигается.

Диаграмма​

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

Предусловия (до обработки батча): таблица property_object_view_facts и её rollup-MV созданы миграцией; consumer group dev-stats-property-object-view-ended зарегистрирована; измерение dim_property_object наполнено (backfill при вводе) — иначе просмотры массово паркуются на late-dimension.

Постусловие (после успешного батча): для каждого валидного, обогащённого сообщения в property_object_view_facts появилась closed-строка с реальной DurationSeconds, денормализованной иерархией (DeveloperId = TenantId), FactId = SourceEventId, IngestionTime = момент вставки; open/closed того же PropertyObjectViewId склеиваются при чтении (побеждает closed по IngestionTime); rollup-ы получили строку без задвоения (in-batch dedup по SourceEventId); offset сдвинут. Спарканные сообщения не потеряны; poison — в DLQ. PostgreSQL не меняется.

Acceptance criteria​

  • Happy closed-факт. Given валидное property-object-view-ended и существующая строка dim_property_object, When консюмер обработал батч, Then в property_object_view_facts есть closed-строка с денормализованной иерархией и реальной DurationSeconds; чтение (argMax/uniq без FINAL) отдаёт один просмотр.
  • Склейка open/closed. Given ранее записан open-факт (action-tracked, DurationSeconds = 0) с тем же PropertyObjectViewId, When пришёл closed, Then при чтении склейка по PropertyObjectViewId выбирает closed-строку (argMax(*, IngestionTime)); просмотр не задваивается.
  • Late-dimension. Given строки dim_property_object ещё нет, When обрабатывается просмотр, Then сообщение паркуется (deferred re-enrich с backoff), при устойчивом промахе → DLQ + алерт; после backfill измерения replay из DLQ дописывает факт; dev_stats_dim_property_object_miss_total растёт.
  • Идемпотентный повтор. Given то же событие доставлено дважды, When оба обработаны, Then дубль по FactId = SourceEventId схлопывается ReplacingMergeTree, in-batch dedup не даёт rollup-повтора.
  • Граничные значения. Given DurationSeconds < 0 (нарушение INV-LV) или ViewedAt == default или ViewedAt > now() + 1 ч или пустой PropertyObjectViewId/PropertyObjectId, When обработан, Then сообщение в DLQ (poison) + error-лог, партиция продолжается.
  • Unknown ViewSource. Given ViewSource вне множества контракта, When обработан, Then Unknown + warning (не DLQ), строка пишется.
  • Гость vs аутентифицированный. Given UserId == null, Then ViewerKind = Guest; Given UserId != null без сигнала роли, Then ViewerKind = Investor (v1-допущение), Agent — по мере появления сигнала роли.

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

  • Business key: FactId = SourceEventId (идентификатор сообщения, уникален на событие). Дубль: строка вставляется повторно и схлопывается ReplacingMergeTree при merge; до merge читатели дубль-устойчивы (argMax(*, IngestionTime) по PropertyObjectViewId; open/closed одной сессии просмотра склеиваются по PropertyObjectViewId).
  • Replay: да — переигрывание топика группой dev-stats-property-object-view-ended с произвольного offset безопасно; «дырки» после инцидента закрываются повтором диапазона без ручной очистки CH.
  • Пересечение с backfill фактов исключено схемой ключей: у backfill-строк SourceEventId — детерминированный backfill:{PropertyObjectViewId} (см. facts-backfill), у консюмера — идентификатор сообщения события.

Retry и DLQ​

СбойRetryDLQРучное действие
ClickHouse недоступен / timeout вставкиДа: батч с backoff (1с → 60с), offset не commit-итсяНет (сбой инфраструктуры)Алерт по росту ingest-lag > 15 мин; восстановить CH — консюмер догонит
dim_property_object не найден (late-dimension)Да: park + deferred re-enrich (backoff, ≤ N попыток)Да при устойчивом промахе + алерт «dimension gap»Backfill dim_property_object, затем replay сообщения из DLQ
Ошибка десериализации / невалидный payload (шаг 2)НетДа (poison, per-message) + error-логПроверить контракт producer-а user-activity; replay после исправления
Версия события ≠ ContractVersionНе сбой: warning-лог, продолжитьНетСверить график миграции контракта
Сбой между вставкой батча и commit offsetПовторная доставка (at-least-once)НетНе требуется — дубли схлопывает ReplacingMergeTree

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

  • Logs: consumed (offset-диапазон, размер батча), enriched/parked (late-dimension), skipped-poison (причина
    • SourceEventId), flushed (строк, длительность вставки), failed (ошибка CH). Число дублей не логируется — dedup на стороне CH.
  • Metrics: dev_stats_property_object_view_facts_ingested_total, dev_stats_ingest_lag_seconds (ViewedAt → IngestionTime), dev_stats_ingest_batch_size, dev_stats_dim_property_object_miss_total, счётчики retry/DLQ.
  • Traces: consume span + dim-lookup span + insert span; корреляция из метаданных события; SourceEventId в строке связывает CH-данные с событием.
  • PII/masking: факт несёт только псевдонимные UserId/AnonymousId/SessionId (Sensitive на уровне строк); ФИО/контакты не хранятся; наружу API отдаёт агрегаты (типы зрителей, счётчики). FilterSnapshot в логах не разворачивается.
  • Actor/permissions: пишет только system-actor консюмера; прямой записи в CH из API/других сервисов нет. Tenant: TenantId = DeveloperId берётся из dim_property_object (обогащение замыкает tenant-scope на источнике данных); scope-фильтрация — на чтении (DeveloperId = актор, fail-closed).

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

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

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