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

Consumer: DevStatsSessionEndedConsumer

Пишущий процесс факта сессий dev-stats: принимает завершённую сессию из user-activity и добавляет строку session_facts в ClickHouse pin_dev_stats батчами. Даёт сессионные метрики контекста (длительность сессий, разрез гость/пользователь); привязка к застройщику — не денормализацией в сессию, а через связь SessionId с просмотрами его объектов недвижимости в property_object_view_facts (ADR-0057, ADR-0056 п.3: длительность считает сервер при закрытии по polling-delta).

Назначение​

Consumer DevStatsSessionEndedConsumer слушает событие session-ended (топик pin.user-activity.session-ended.v1) и добавляет (никогда не обновляет и не удаляет) строку факта завершённой сессии. Событие терминальное — ровно одно на сессию (INV-SES-006). Консюмер классифицирует ViewerKind ∈ Guest/User исключительно по наличию UserId (роль тут не нужна — в отличие от property_object_view_facts, где Agent/Investor), переносит DurationSeconds/StartedAt/LastActivityAt и пишет TenantId = платформенный sentinel (00000000-0000-0000-0000-000000000000): сессия кросс-застройщик, её tenant-скоуп задаётся на чтении через пересечение с property_object_view_facts (см. session-facts.md).

Обогащения иерархией нет (сессия не привязана к объекту недвижимости/застройщику) — это самый простой из двух допустимых по канону вариантов (ADR-0056, simplicity-first): единая узкая таблица без реплики измерений. Owned-состояния в PostgreSQL консюмер не меняет: пишет только в ClickHouse.

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

ПолеЗначение
Сервис/контекстdev-stats (проект DevStats.Background)
ConsumerDevStatsSessionEndedConsumer
Обрабатываемое событиеsession-ended
Topicpin.user-activity.session-ended.v1
ContractVersion1; при несовпадении — warn-and-proceed
Consumer groupdev-stats-session-ended
Produceruser-activity — worker activity-closer (закрытие по неактивности) и link-handler (форс-закрытие при входе другого пользователя, INV-SES-002)
Эффект (событие → что меняется)append строки в pin_dev_stats.session_facts (ReplacingMergeTree)

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

  • Из метаданных события доступны идентификатор источника (AggregateId = SessionId), тип агрегата (Session), время события (CreatedAt), версия snapshot-а (Version).
  • Сквозной трейс восстанавливается из метаданных события; идентификатор сообщения (MessageId) → SourceEventId факта (сквозная трассировка).
  • Бизнес-идемпотентность — декларативная, на стороне ClickHouse: FactId = SourceEventId — последний компонент ORDER BY ReplacingMergeTree; повтор доставки схлопывается при merge, чтение дубль-устойчиво (argMax(DurationSeconds, IngestionTime) по SessionId; FINAL запрещён — правило SDK). Дополнительно — in-batch dedup по SourceEventId. Отдельная таблица идемпотентности не используется: нет PG-транзакции.

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

Контракт события: event-contract: session-ended. Дом колонок факта — session-facts.md («Columns»); ниже — правила маппинга. Payload плоский, вложенных моделей нет.

Поле payloadТипОбязательностьКак используется (маппинг)
SessionIdGuidДа→ SessionId (ключ связи с property_object_view_facts); Guid.Empty → poison (DLQ)
AnonymousIdGuidДа→ AnonymousId (идентификатор гостя)
UserIdGuid?Нет→ UserId (NULL для гостя); единственный вход классификации ViewerKind
CreatedAtDateTimeOffsetДа→ StartedAt/EventTime; партиция toYYYYMM(EventTime); нижняя граница Duration
LastActivityAtDateTimeOffsetДа→ LastActivityAt; момент закрытия (polling-delta)
DurationSecondsintДа→ DurationSeconds; < 0 → poison (chk_sessions_duration_nonneg: >= 0)

Колонки, которые консюмер заполняет сам: FactId = SourceEventId (идентификатор сообщения); IngestionTime = момент вставки; TenantId = платформенный sentinel (Guid.Empty как строка); BrandId = NULL; DeveloperId = NULL (в v1 не заполняется — задел под будущую денормализацию single-developer-сессий); ViewerKind = UserId is null ? "Guest" : "User".

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

Ожидает от user-activity (входной контракт session-ended): плоский snapshot завершённой сессии (SessionId, AnonymousId, UserId?, CreatedAt, LastActivityAt, DurationSeconds); MessageId → SourceEventId. Гарантии продюсера: событие терминальное и единичное на SessionId (INV-SES-006), поэтому Version-guard не нужен (достаточно first-seen по SessionId); DurationSeconds посчитан сервером (LastActivityAt − CreatedAt, ≥ 0); key = SessionId (6 партиций); доставка at-least-once; порядок доставки не значим. Событие не несёт tenant/застройщика (сессия кросс-застройщик, ADR-0052) и не несёт Device.*.

Предоставляет downstream: append-only строку session_facts с TenantId = sentinel, DeveloperId = NULL. Привязка к застройщику не денормализуется в сессию — она задаётся на чтении пересечением SessionId с property_object_view_facts WHERE DeveloperId = актор (см. session-facts.md). Дубль-устойчивое чтение — argMax(DurationSeconds, IngestionTime) по SessionId без FINAL; свежесть ≤ 5 мин.

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

Предусловия: таблица session_facts создана миграцией; consumer group dev-stats-session-ended зарегистрирована. Обогащение измерениями не требуется — payload самодостаточен. Постусловие: для валидного сообщения в session_facts есть строка завершённой сессии с DurationSeconds/StartedAt/LastActivityAt, ViewerKind по наличию UserId, FactId = SourceEventId; повтор схлопывается по FactId; offset сдвинут.

Алгоритм​

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

  1. Проверить контекст события: сверить ContractVersion (= 1; при несовпадении — warn-and-proceed). Дубль по SourceEventId на входе не отсеивается (dedup декларативен), но повтор в текущем батче отбрасывается.
  2. Проверить payload: ошибка десериализации → DLQ (poison); SessionId == Guid.Empty → DLQ + error-лог; DurationSeconds < 0 (нарушение chk_sessions_duration_nonneg: >= 0), CreatedAt == default или CreatedAt > now() + 1 ч → DLQ.
  3. Маппинг поле-в-поле: построить строку факта session_facts — скалярные поля из payload; ViewerKind = UserId is null ? Guest : User; TenantId = sentinel; DeveloperId = NULL; FactId = SourceEventId; IngestionTime — момент вставки. Обогащения нет (сессия не привязана к объекту недвижимости/застройщику); ни измерения, ни PG не читаются — payload самодостаточен. Переходов состояний нет. Добавить строку в буфер батча.
  4. Пакетно вставить (append-only) буфер строк в pin_dev_stats.session_facts (батч ≤ 1000 строк / 5 с); точечное обновление/удаление/upsert запрещены — append-only; в CH пишет только этот background-консюмер. Смещение (offset) фиксируется строго после подтверждённой вставки; сбой между вставкой и commit → повторная доставка → дубли схлопывает ReplacingMergeTree (at-least-once без потерь).
  5. Записать метрики/логи; успех → ack; ошибка вставки → retry всего батча (идемпотентно), offset не двигается.

Диаграмма​

Чтение сессионных метрик застройщика (пример без FINAL): множество SessionId берётся из property_object_view_facts WHERE DeveloperId = актор за период и джойнится к session_facts по SessionId — так «сессии, в которых смотрели мои объекты недвижимости» и их длительности получаются без привязки застройщика к самой сессии (session-facts.md, «Связь с застройщиком»).

Acceptance criteria​

  • Happy. Given валидное session-ended, When обработано, Then в session_facts есть строка с DurationSeconds/StartedAt(= CreatedAt)/LastActivityAt, TenantId = sentinel, DeveloperId = NULL, FactId = SourceEventId.
  • ViewerKind. Given UserId == null, Then ViewerKind = Guest; Given UserId != null, Then ViewerKind = User (роль не нужна, в отличие от факта просмотров).
  • Идемпотентный повтор. Given терминальное событие доставлено дважды (at-least-once), When оба обработаны, Then дубль по FactId схлопывается ReplacingMergeTree, in-batch dedup не даёт повтора; событие единично на SessionId, поэтому логически одна строка.
  • Граничные значения. Given DurationSeconds < 0 (нарушение chk_sessions_duration_nonneg) или CreatedAt == default или CreatedAt > now() + 1 ч или SessionId == Guid.Empty или ошибка десериализации, When обработано, Then сообщение в DLQ (poison) + error-лог, партиция продолжается.
  • Привязка к застройщику на чтении. Given сессия и просмотры объектов актора связаны по SessionId, When застройщик читает сессионные метрики, Then возвращаются только сессии из пересечения с property_object_view_facts WHERE DeveloperId = актор (кросс-застройщик-сессии наружу не отдаются).

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

  • Business key: FactId = SourceEventId (идентификатор сообщения, уникален на событие; событие терминальное — на SessionId не более одной публикации). Дубль: строка вставляется повторно и схлопывается ReplacingMergeTree при merge; до merge читатели дубль-устойчивы (argMax(DurationSeconds, IngestionTime) по SessionId).
  • Replay: да — переигрывание топика группой dev-stats-session-ended с произвольного offset безопасно.
  • Пересечение с backfill сессий исключено схемой ключей: у backfill-строк SourceEventId — детерминированный backfill:{SessionId}, у консюмера — идентификатор сообщения события.

Retry и DLQ​

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

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

  • Logs: consumed (offset-диапазон, размер батча), skipped-poison (причина + SourceEventId), flushed (строк, длительность вставки), failed (ошибка CH). Число дублей не логируется — dedup на стороне CH.
  • Metrics: dev_stats_session_facts_ingested_total, dev_stats_ingest_lag_seconds (EventTime → IngestionTime), dev_stats_ingest_batch_size, счётчики retry/DLQ.
  • Traces: consume span + insert span; корреляция из метаданных события; SourceEventId в строке связывает CH-данные с событием.
  • PII/masking: факт несёт только псевдонимные UserId/AnonymousId/SessionId (Sensitive на уровне строк); ФИО/контакты не хранятся; Device.IpAddress/UserAgent в событие не входят (правило session.md). Наружу API отдаёт агрегаты (тип зрителя, счётчики, длительности) и только в пересечении с объектами недвижимости актора.
  • Actor/permissions: пишет только system-actor консюмера; прямой записи в CH из API/других сервисов нет. Tenant: сессия вне tenant (ADR-0052 п.2) → TenantId = sentinel; scope застройщика применяется на чтении через связь с property_object_view_facts (fail-closed).

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

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

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