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

Consumer: Selections Events (UserActivitySelectionItemAddedConsumer)

Назначение​

Когда инвестор добавляет объект недвижимости в подборку, selections публикует pin.selections.selection-item-added (поэлементно). Consumer записывает в user_activity.action_events запись AddToSelection — так действие попадает в историю действий инвестора (/profile/events), а через собственное action-tracked — в ленту карточки лида CRM (агенту видно намерение клиента собрать подборку, поток «ACT → CRM» service-boundaries.md). user-activity не владеет подборками — пишет производную запись действия (source of truth — selections).

Событие selection-item-added.v1 эмитируется по событию на каждый добавленный объект недвижимости из диффа PUT /selections/{id}/items и из POST /selections/from-favorites (контракт); payload несёт готовые PropertyObjectId + SelectionId, маппинг в AddToSelectionActionEvent прямой.

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

ПолеЗначение
Сервис/контекстuser-activity (консюмеры и worker-ы)
ConsumerUserActivitySelectionItemAddedConsumer
Consumer grouppin-user-activity-selections
Producerselections (операция добавления элемента подборки)
Обрабатываемое событиеpin.selections.selection-item-added (контракт)
Topicpin.selections.selection-item-added.v1 (key = SelectionId)
ContractVersion1; при несовпадении версии — warn-and-proceed
Эффект (событие → что меняется)append-only запись AddToSelectionActionEvent (user_activity.action_events) + исходящее action-tracked

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

  • Из метаданных события доступны: MessageId (ключ дедупа), CreatedAt (= OccurredAt), Version, AggregateId = SelectionId, AggregateType = "Selection".
  • Actor — system-actor консюмера. Контекст вне tenant (ADR-0052 п.2): изоляция record-level по UserId/AnonymousId/SessionId.
  • Корреляция сквозного трейса восстанавливается из метаданных события и пробрасывается в исходящее action-tracked.
  • Идемпотентность: consumer-level — ключ user-activity-selections:{MessageId}; business-level — INV-ACT-004 (SourceMessageId).

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

Дом контракта — ../../selections/events/selection-item-added.md. Consumer читает нижеперечисленное и маппит в AddToSelectionActionEvent (дом полей — action-event.md); остальные поля payload читаются как игнорируемые additive.

Поле payloadТипОбязательностьКак используется (маппинг)
SelectionIdGuidДа→ AddToSelectionActionEvent.SelectionId (логическая ссылка на подборку, без FK); = AggregateId, Kafka key
PropertyObjectIdGuidДа→ AddToSelectionActionEvent.PropertyObjectId (логическая ссылка на объект недвижимости, INV-LV-006)
OwnerUserIdGuidДаКлюч резолва Active-сессии владельца; снимок UserId берётся из найденной сессии
OccurredAtDateTimeOffsetДа→ AddToSelectionActionEvent.OccurredAt (доменное время; server-clock rule INV-ACT-005)
SelectionItemId, SelectionItemsCount, Version—Нетне маппятся (дедуп — по MessageId; лента фиксирует факт, не счётчик/порядок)

OwnerUserId всегда заполнен: подборка принадлежит авторизованному владельцу — гость подборки не создаёт (INV-SEL-002; в отличие от гостевого избранного). Пустой OwnerUserId → error-лог invalid-owner

  • skip (ack, не DLQ). Гостевой ветки (AnonymousId) у этого события нет.

Алгоритм​

  1. Приём входа и контекста. Восстановить контекст события (MessageId, CreatedAt, Version, payload); actor = system; сверить версию с ContractVersion = 1 — warn-and-proceed. Consumer- идемпотентность по ключу user-activity-selections:{MessageId} — дубль → skipped-duplicate, ack.
  2. Проверка входа. SelectionId, PropertyObjectId, OwnerUserId, SourceMessageId непустые — пустой OwnerUserId → invalid-owner, ack-skip. Существование подборки/объекта недвижимости синхронно не проверяется (кросс-контекст, логические ссылки).
  3. Business-дедуп и резолв сессии. Если запись с этим SourceMessageId уже есть → ack (INV-ACT-004). Резолв Active-сессии владельца (INV-ACT-003) по OwnerUserId; фоллбэк на последнюю сессию владельца по LastActivityAt (consumer не синтезирует сессию — Device в payload нет). OwnerUserId всегда авторизован — гостевой ветки (AnonymousId) нет. Сессии нет → skipped-no-session, ack.
  4. Бизнес-правила + маппинг. Собрать AddToSelectionActionEvent (INV-ACT-002 required-поля): PropertyObjectId, SelectionId, SessionId = session.Id, AnonymousId = session.AnonymousId, UserId = session.UserId (снимок), OccurredAt = OccurredAt, SourceMessageId = MessageId. Запись immutable (INV-ACT-001) — переходов статуса нет.
  5. Что меняется. Появляется append-only запись action_events + отметка idempotency-ключа.
  6. Какие события фиксируются. ActionTrackedKafkaEvent → pin.user-activity.action-tracked.v1 (AggregateId = ActionEventId, ContractVersion = 1; payload зеркалит TPH: ActionType = AddToSelection, SessionId, AnonymousId, UserId?, OccurredAt, PropertyObjectId, SelectionId). Downstream crm пока не мапит AddToSelection в ленту лида напрямую (no-op; см. crm-консюмеры) — событие всё равно эмитится для единства ленты действий (news-рекомендации — будущий потребитель, ADR-0054).
  7. Согласованность. Вставка action_events + отметка идемпотентности + исходящее событие фиксируются в одном согласованном изменении; offset коммитится после фиксации.
  8. Результат. Успех → ack. Контролируемые skip-ветки (дубль, invalid-owner, skipped-no-session) — ack. Инфраструктурная ошибка → retry по политике консюмера.

Диаграмма​

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

  • Consumer-level: дедуп по MessageId (ключ user-activity-selections:{MessageId}).
  • Business-level (INV-ACT-004): partial-unique ux_action_events_source_message на SourceMessageId = MessageId — повтор доставки ловится проверкой существования/constraint-ом → ack (одна запись на сообщение).
  • Событие — факт (append-only): version-guard не нужен; достаточно дедупа по MessageId.
  • Replay: да — повторный re-read безопасен (idempotency-ключ + unique SourceMessageId).

Retry и DLQ​

DLQ-топик группы: pin.user-activity.selections-dlq.v1 (событие + x-dlq-reason, x-source-topic).

СбойRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)Бесконечный exponential backoff 1s→60s с паузой партицииНетRunbook: восстановить зависимость
Пустой OwnerUserIdНет — контролируемый invalid-owner, ackНетИсправить selections → replay
Нет сессии владельца (skipped-no-session)Нет — контролируемый ackНетОжидаемо вне трекнутой сессии; не ошибка
Contract mismatch (несовпадение версии)Нет — warn-and-proceedНетОбновить маппинг при breaking-bump (.v2)
Гонка SourceMessageId (unique violation)НетНетТрактуется как дубль → ack
Непредвиденная ошибка обработки3 попытки немедленноПосле исчерпанияРасследование гонки/бага

Acceptance Criteria​

#Given / When / Then
AC1Given selection-item-added с валидными SelectionId/PropertyObjectId/OwnerUserId и Active-сессией владельца When обработано Then создаётся AddToSelectionActionEvent со снимком UserId/SessionId, публикуется action-tracked (ActionType = AddToSelection)
AC2Given повторная доставка события с тем же MessageId When обработано Then запись не дублируется (дедуп по MessageId/SourceMessageId, INV-ACT-004), ack без изменений
AC3Given событие с пустым OwnerUserId (гость подборки не создаёт, INV-SEL-002) When обработано Then invalid-owner, ack-skip, без записи и без DLQ
AC4Given владелец без трекнутой сессии When резолв сессии (по OwnerUserId) пуст Then skipped-no-session, ack (не ошибка), запись не создаётся
AC5Given запись создана When downstream crm получает action-tracked Then лента лида не меняется (no-op для AddToSelection); событие всё равно эмитится для единства ленты действий (будущий потребитель news)
AC6Given PostgreSQL/Kafka недоступны When обработка падает Then бесконечный backoff-ретрай с паузой партиции; событие не теряется

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

  • Logs: consumed / skipped-duplicate / invalid-owner / skipped-no-session / recorded / failed с полями MessageId, SelectionId, PropertyObjectId, SessionId. PII нет.
  • Metrics: user_activity_selections_consume_duration_ms, user_activity_selections_recorded_total, user_activity_selections_skipped_total{reason}, user_activity_selections_retries_total, user_activity_selections_dlq_total (0), user_activity_selections_lag_seconds.
  • Traces: consume-span → span обработки → spans обращений к хранилищу; correlation из метаданных события → исходящее action-tracked.
  • Alerts: лаг > 5 мин; DLQ > 0; всплеск invalid-owner (регресс контракта selections).
  • Безопасность: system-actor; вне tenant (ADR-0052 п.2), record-level-изоляция; доступ к топику pin.selections.* — m2m ACL Kafka. Секретов/PII в payload нет.

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

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

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