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

Consumer: User-Activity Actions (crm-user-activity-actions)

Один документ на два входных топика user-activity, наполняющих ленту событий лида (LeadEventEntity). Контракты входных событий — дом property-object-view.md и action-event.md; целевой агрегат — lead.md.

Назначение​

Consumer group crm-user-activity-actions потребляет активность инвестора на витрине и, если у инвестора есть активный лид, дописывает в его ленту типизированные события: PropertyObjectViewed (из property-object-view-ended), FilterApplied / FavoriteAdded (из action-tracked). Так агент видит «что делает и чего хочет клиент» (UC-1..3), не запрашивая user-activity синхронно. Если активного лида у инвестора нет — событие не относится ни к какому лиду: no-op ack (лента не создаётся вне лида). Консюмер собственных Kafka-событий не публикует (лента — внутреннее состояние crm).

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

ПолеЗначение
Сервис/контекстcrm
Consumer groupcrm-user-activity-actions
Produceruser-activity (источник истины входных контрактов)
Топик 1pin.user-activity.property-object-view-ended.v1 — PropertyObjectViewEndedKafkaEvent
Топик 2pin.user-activity.action-tracked.v1 — ActionRecordedKafkaEvent (см. Change context)
ContractVersionожидается 1; при несовпадении версии события — warn-and-proceed
Эффект (событие → что меняется)Дописывает в ленту активного лида LeadEventEntity (PropertyObjectViewed/FilterApplied/FavoriteAdded)

Маппинг «событие → эффект» (тип события в ленте лида, LeadEventType):

Входное событиеУсловиеТип события лида (LeadEventType)
PropertyObjectViewEndedKafkaEvent—PropertyObjectViewed
ActionRecordedKafkaEventActionType = ApplyFilterFilterApplied
ActionRecordedKafkaEventActionType = AddToFavoritesFavoriteAdded
ActionRecordedKafkaEventActionType ∈ {Search, ViewPropertyObject, RemoveFromFavorites, AddToSelection}нет отображения → no-op ack
Change context

README контекста и lead.md фиксируют топик как pin.user-activity.action-tracked.v1; дом события action-event.md публикует ActionRecordedKafkaEvent на канонический pin.user-activity.action-recorded.v1. Здесь принято имя из README crm (owning-сервис консюмера); расхождение имён топика/типа — assumption, требует reconciliation с user-activity перед реализацией. ViewPropertyObject намеренно не мапится — просмотры приходят отдельным топиком property-object-view-ended; AddToSelection — консюмером selection-shared.

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

  • Из метаданных события доступны: идентификатор сообщения MessageId (для дедупликации), CreatedAt, версия контракта, идентификатор агрегата-источника (PropertyObjectViewId / ActionEventId).
  • Актор — системный контекст консюмера (не пользовательский); запись в ленту идёт в tenant найденного лида (TenantId = AgencyCompanyId).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Бизнес-идемпотентность — по ключу crm-user-activity-actions:{LeadEventType}:{MessageId}; страховка — unique ux_lead_events_source_message_id на LeadEventEntity.SourceMessageId = MessageId (INV ниже).

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

Ниже — только поля, которые consumer реально читает, и их маппинг в LeadEventEntity (дом типов/ограничений payload лид-событий — lead.md, раздел «Enum LeadEventType и typed payload»).

PropertyObjectViewEndedKafkaEvent → LeadPropertyObjectViewedPayload (Type = PropertyObjectViewed)​

Поле payload событияТипОбяз.Как используется (маппинг)
UserIdGuid?НетКлюч резолва активного лида (PrimaryInvestorUserId); null (гостевой просмотр) → no-op ack
PropertyObjectIdGuidДа→ LeadPropertyObjectViewedPayload.PropertyObjectId
FilterSnapshot.Project.ProjectIdGuid?Нет→ LeadPropertyObjectViewedPayload.ProjectId (null, если фильтра/проекта не было)
DurationSecondsint (>=0)Да→ LeadPropertyObjectViewedPayload.DurationSeconds
SessionIdGuidДа→ LeadPropertyObjectViewedPayload.SessionId
ViewedAtDateTimeOffsetДа→ LeadEventEntity.OccurredAt

ActionRecordedKafkaEvent → LeadFilterAppliedPayload (ApplyFilter → Type = FilterApplied)​

Поле payload событияТипОбяз.Как используется (маппинг)
UserIdGuid?НетРезолв активного лида (PrimaryInvestorUserId); null → no-op ack
OccurredAtDateTimeOffsetДа→ LeadEventEntity.OccurredAt
FilterSnapshot.City.CityIdlong?Нет→ LeadFilterAppliedPayload.CityId (ref-модель CityRefModel; id справочника городов pin.handbook)
FilterSnapshot.City.CityNamestring?Нет→ LeadFilterAppliedPayload.CityName (display-имя из ref-модели как есть, без резолва в справочнике — мягкая семантика активности)
FilterSnapshot.Developer.DeveloperIdGuid?Нет→ LeadFilterAppliedPayload.DeveloperCompanyId (ref-модель DeveloperRefModel)
FilterSnapshot.Project.ProjectIdGuid?Нет→ LeadFilterAppliedPayload.ProjectId (ref-модель ProjectRefModel)
FilterSnapshot.PriceNullableRange<decimal>Нетграницы Price.Min / Price.Max → payload Price (NullableRange<decimal>)
FilterSnapshot.RoomsList<int>?Нет→ RoomsCount = единственное значение при Count == 1, иначе null (лента показывает конкретную комнатность; мультивыбор не детализируется)
FilterSnapshot (display)AppliedFilterModelДа→ RawQuery — денормализованное резюме фильтра (город/застройщик/проект/цена/комнаты) из *Name-полей ref-моделей, чтобы лента читалась без lookup в catalog
к сведению
Частота FilterApplied снижена дебаунсом

ApplyFilter приходит из user-activity редко: на витрине каталога кнопки «Найти» нет, фронт трекает применение фильтра отдельным запросом с trailing-дебаунсом 5000–7000 мс (track-action.md, action-event.md), поэтому в топик попадает только устоявшийся фильтр, а не каждое изменение контрола. Как следствие лента лида получает осмысленные FilterApplied без флуда — дополнительного троттлинга на стороне crm не требуется.

ActionRecordedKafkaEvent → LeadFavoriteAddedPayload (AddToFavorites → Type = FavoriteAdded)​

Поле payload событияТипОбяз.Как используется (маппинг)
UserIdGuid?НетРезолв активного лида; null → no-op ack
OccurredAtDateTimeOffsetДа→ LeadEventEntity.OccurredAt
PropertyObjectIdGuid?Да (для AddToFavorites)→ LeadFavoriteAddedPayload.PropertyObjectId

Алгоритм​

Оба входных события различаются только шагами 1 и 4 (приведение к нормализованному событию ленты); шаги 2–3, 5–6 — общая обработка.

  1. Контекст события. Восстановить метаданные события (идентификатор сообщения, время, версия контракта); при несовпадении версии — warn-and-proceed. Проверить идемпотентность по ключу crm-user-activity-actions:{LeadEventType}:{MessageId}: дубль → пропустить (ack). Для action-tracked — диспетчеризация по ActionType; неотображаемый тип → skipped-unmapped-action, ack.
  2. Проверка входа. SourceMessageId задан; для PropertyObjectViewed — DurationSeconds >= 0; для FilterApplied — FilterSnapshot содержит ≥ 1 заполненное поле; для FavoriteAdded — PropertyObjectId задан. Нарушение контракта payload (обяз. поле отсутствует) → CRM_ACTIVITY_INVALID_PAYLOAD (non-retriable → DLQ). Если UserId == null — резолв активного лида невозможен → no-op ack (см. шаг 3).
  3. Что читаем. Найти активный лид инвестора (PrimaryInvestorUserId, Status = Active; резолв по событийному UserId). Если активного лида нет → no-op ack (skipped-no-active-lead) — активность вне лида в ленту не пишется. При нескольких активных лидах (мульти-агентство) выбор детерминирован по LastEventAt desc (см. Assumptions).
  4. Правила и маппинг поле-в-поле. Построить typed payload лид-события (таблицы выше) и дописать в ленту через метод агрегата: создаётся LeadEventEntity (Type, Payload, OccurredAt из события, SourceMessageId = MessageId, TenantId = activeLead.TenantId, CreatedAt). Guard INV-LEAD-2 (Status = Active, иначе CRM_LEAD_NOT_ACTIVE); метод также ставит activeLead.LastEventAt = OccurredAt и поднимает доменное событие LeadEventAdded.
  5. Что меняется. Появляется строка ленты и обновляется LastEventAt лида. Запись события ленты и отметка идемпотентности фиксируются в одном согласованном изменении; интеграционных событий консюмер не публикует (лента — внутреннее состояние crm, LeadEventAdded остаётся доменным). Offset подтверждается только после успешной фиксации.
  6. Результат. Успех → ack. Retriable-код → retry по политике; non-retriable → DLQ + алерт. No-op-ветки (нет UserId, нет активного лида, неотображаемый ActionType, дубль) — тоже ack.

Диаграмма​

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

  • На уровне консюмера: ключ crm-user-activity-actions:{Type}:{MessageId}.
  • На уровне хранилища: unique partial ux_lead_events_source_message_id на SourceMessageId — повтор доставки после сбоя ловится ограничением: конфликт уникальности → трактуется как дубль → ack (no-op). Глобальная уникальность SourceMessageId означает одну строку ленты на одно исходное сообщение — событие адресуется одному лиду (детерминированный выбор при мульти-агентстве, см. Assumptions).
  • Replay: да. Повторный проход топиков безопасен (ключ идемпотентности + unique SourceMessageId); runbook — crm/runbooks/.

Retry и DLQ​

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)— (exception)Бесконечный exponential backoff 1s→60s с паузой партицииНетRunbook: восстановить зависимость
Нарушение контракта payload (обяз. поле отсутствует)CRM_ACTIVITY_INVALID_PAYLOADНет (non-retriable)Да + алертИсправить продьюсер user-activity → replay
Неизвестный ActionType (additive у продьюсера)—НетНетWarn + no-op ack (forward-compatible); при необходимости добавить маппинг (релиз)
Гонка SourceMessageId (unique violation)—НетНетТрактуется как дубль → ack
Лид ушёл в Expired между шагами 3 и 5CRM_LEAD_NOT_ACTIVEНетНетNo-op ack (активность истёкшему лиду не пишется)

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

  • Logs: consumed / skipped-duplicate / skipped-unmapped-action / skipped-no-investor / skipped-no-active-lead / appended / dead-lettered с полями MessageId, AggregateId, LeadEventType, LeadId (при наличии). PII в payload нет (id и агрегированные величины; CityName/DeveloperName/ProjectName — display-справочники, не PII).
  • Metrics: crm_activity_consume_duration_ms, crm_activity_appended_total{type}, crm_activity_skipped_total{reason}, crm_activity_retries_total, crm_activity_dlq_total, crm_activity_lag_seconds (now − envelope.CreatedAt).
  • Traces: span приёма события → span обработки → spans обращений к хранилищу; корреляция из метаданных события.
  • Alerts: лаг > 5 мин (SLO README); DLQ > 0; всплеск skipped-no-active-lead (возможен разрыв привязки лидов).
  • Безопасность: consumer под system-actor; запись в tenant найденного лида (AgencyCompanyId), кросс-tenant-запись невозможна по построению (адресация через сам лид). Топики user-activity — платформенные (ADR-0052 п.2), доступ — m2m ACL Kafka. Инвестор не видит ленты; лента — данные агента/лида (permissions crm.leads.view на чтение, см. lead.md).

Assumptions и ограничения​

  • Топик action-tracked vs action-recorded — расхождение имён (см. Change context), reconciliation обязателен перед реализацией.
  • Мульти-агентство: INV-CRM-001 допускает несколько активных лидов у одного инвестора (разные агенты); из-за глобально-уникального SourceMessageId строка ленты пишется одному лиду — выбор детерминирован по LastEventAt desc. Fan-out одной активности в несколько лент — отдельное решение (вне scope).
  • Авто-продление лида по активности клиента (Lead.LastExtendedAt, lead.md) в этом консюмере не реализуется — отдельная политика.

Acceptance Criteria​

  • AC-1 (просмотр объекта). Дано: у инвестора есть активный лид. Когда: PropertyObjectViewEndedKafkaEvent с UserId. Тогда: в ленту дописывается PropertyObjectViewed (LeadPropertyObjectViewedPayload), LastEventAt = OccurredAt, ack.
  • AC-2 (применён фильтр). Дано: активный лид есть. Когда: ActionRecordedKafkaEvent с ActionType = ApplyFilter. Тогда: дописывается FilterApplied с денормализованным RawQuery, ack.
  • AC-3 (добавлено в избранное). Дано: активный лид есть. Когда: ActionType = AddToFavorites с PropertyObjectId. Тогда: дописывается FavoriteAdded, ack.
  • AC-4 (неотображаемый ActionType). Дано: ActionType ∈ {Search, ViewPropertyObject, RemoveFromFavorites, AddToSelection}. Когда: событие. Тогда: skipped-unmapped-action, лента не меняется, ack.
  • AC-5 (гостевая активность). Дано: UserId == null. Когда: событие. Тогда: skipped-no-investor, ack.
  • AC-6 (нет активного лида). Дано: у инвестора нет активного лида. Когда: событие. Тогда: skipped-no-active-lead, лента не создаётся вне лида, ack.
  • AC-7 (дубль по сообщению). Дано: уже виденный MessageId/SourceMessageId. Когда: повторная доставка/replay. Тогда: дубль (ux_lead_events_source_message_id) → no-op, ack.
  • AC-8 (нарушен контракт payload). Дано: отсутствует обязательное поле payload. Когда: событие. Тогда: CRM_ACTIVITY_INVALID_PAYLOAD (non-retriable) → DLQ + алерт.

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

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

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