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

Consumer: Favorites Events (UserActivityFavoriteAddedConsumer / ...RemovedConsumer)

Назначение​

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

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

ПолеЗначение
Сервис/контекстuser-activity (хост Pin.UserActivity.Background — консюмеры и worker-ы)
Consumer grouppin-user-activity-favorites
Producerfavorites: POST /favorites, POST /favorites/import, DELETE /favorites/{propertyObjectId}
Топик 1pin.favorites.favorite-added.v1 — FavoriteAddedKafkaEvent (6 партиций, key = FavoriteId)
Топик 2pin.favorites.favorite-removed.v1 — FavoriteRemovedKafkaEvent (6 партиций, key = FavoriteId)
ContractVersion1; при несовпадении версии события — warn-and-proceed (апкаста нет)
Эффект (событие → что меняется)новая производная запись ActionEvent в ленте действий + исходящее action-tracked (source of truth избранного — favorites)

Группа обрабатывает два события одним общим правилом; тип записи ленты определяется событием (паттерн как в ../../crm/consumers/user-activity-actions.md):

СобытиеЭффект — запись ленты (ActionEventType)
favorite-added (pin.favorites.favorite-added.v1)AddToFavoritesActionEvent (ActionType = AddToFavorites)
favorite-removed (pin.favorites.favorite-removed.v1)RemoveFromFavoritesActionEvent (ActionType = RemoveFromFavorites)

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

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

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

Оба события имеют идентичную форму (дом полей — контракты favorites). Ниже — только поля, которые consumer реально читает, и их маппинг в подтип ActionEvent (дом полей — action-event.md).

Поле payloadТипОбязательностьКак используется (маппинг)
FavoriteIdGuidДа→ *ActionEvent.FavoriteId (логическая ссылка на favorites, без FK)
PropertyObjectIdGuidДа→ *ActionEvent.PropertyObjectId (логическая ссылка на объект недвижимости, существование не проверяется, INV-LV-006)
UserIdGuid?Условно (XOR с AnonymousId, INV-FAV-001)Ключ резолва сессии владельца (при AnonymousId == null); снимок UserId берётся из найденной сессии
AnonymousIdGuid?Условно (XOR с UserId)Ключ резолва Active-сессии устройства по AnonymousId
OccurredAtDateTimeOffsetДа→ *ActionEvent.OccurredAt (доменное время действия; сервер клиентский timestamp не подменяет, INV-ACT-005)

Инвариант payload: заполнен ровно один из UserId / AnonymousId; оба null/оба заданы — контрактная ошибка producer-а → error-лог + skip (ack, не DLQ — ретрай не восстановит данные, правило favorites).

Алгоритм​

Оба события обрабатываются одним правилом; различие — только тип создаваемой записи (ActionType по событию).

  1. Контекст события. Восстановить метаданные (MessageId, CreatedAt/OccurredAt, версия, payload); actor = system; сверить версию с ContractVersion = 1 — при несовпадении warn-and-proceed. Определить ActionType по событию (AddToFavorites/RemoveFromFavorites). Consumer-идемпотентность по ключу user-activity-favorites:{ActionType}:{MessageId} — дубль → лог skipped-duplicate, ack (no-op).
  2. Проверка входа. FavoriteId != Guid.Empty, PropertyObjectId != Guid.Empty, SourceMessageId != Guid.Empty; заполнен ровно один из UserId/AnonymousId (XOR, INV-FAV-001) — нарушение → лог invalid-owner, ack (skip, не DLQ). Существование favorites.favorites(FavoriteId)/объекта недвижимости синхронно не проверяется (кросс-контекст, логические ссылки).
  3. Что читаем. Business-дедуп: запись с этим SourceMessageId уже есть → ack (no-op, INV-ACT-004). Резолв сессии-владельца (INV-ACT-003): по AnonymousId (или UserId) — Active-сессия; если её нет — последняя сессия владельца по LastActivityAt. Действие якорится к существующей сессии; consumer не синтезирует сессию (в payload нет Device, ср. identity-signed-in.md ветка A). Сессии нет вовсе → лог skipped-no-session, ack.
  4. Бизнес-правила и маппинг. По ActionType создаётся подтип (INV-ACT-002 required-поля): AddToFavoritesActionEvent / RemoveFromFavoritesActionEvent с PropertyObjectId, FavoriteId, SessionId, AnonymousId, UserId (снимок сессии на момент действия), OccurredAt = OccurredAt события, SourceMessageId = MessageId. Запись immutable (INV-ACT-001) — переходов статуса нет.
  5. Что меняется. Новая строка ActionEvent в ленте действий + отметка идемпотентности.
  6. Какие события фиксируются. ActionTrackedKafkaEvent → pin.user-activity.action-tracked.v1 (AggregateType = "ActionEvent", ContractVersion = 1; payload зеркалит поля: ActionType, SessionId, AnonymousId, UserId?, OccurredAt, PropertyObjectId, FavoriteId) через outbox. Потребитель CRM переводит AddToFavorites → FavoriteAdded; RemoveFromFavorites в ленте лида пока не мапится (no-op downstream, ../../crm/consumers/user-activity-actions.md).
  7. Согласованность. Вставка action_events, отметка идемпотентности и запись outbox фиксируются в одном согласованном изменении; offset подтверждается только после успешной фиксации.
  8. Результат. Успех → ack. Контролируемые skip-ветки (дубль, invalid-owner, skipped-no-session) — тоже ack. Инфраструктурная ошибка → retry по политике консюмера (см. «Retry и DLQ»).

Диаграмма​

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

  • Consumer-level: дедуп по MessageId (ключ user-activity-favorites:{ActionType}:{MessageId}).
  • Business-level (INV-ACT-004): partial-unique ux_action_events_source_message на SourceMessageId = MessageId — повтор at-least-once после after-commit-crash ловится проверкой/constraint-ом → трактуется как дубль → ack. Одна запись действия на одно исходное сообщение.
  • Событие — факт (append-only): version-guard состояния не нужен (правило favorite-*.md), достаточно дедупа.
  • Replay: да — повторный re-read топиков безопасен (idempotency-ключ + unique SourceMessageId).

Retry и DLQ​

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

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

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

  • Logs (structured): consumed / skipped-duplicate / invalid-owner / skipped-no-session / recorded / failed с полями MessageId, FavoriteId, PropertyObjectId, SessionId, ActionType. PII нет — Favorite.Note в payload отсутствует и не логируется.
  • Metrics: user_activity_favorites_consume_duration_ms, user_activity_favorites_recorded_total{action_type}, user_activity_favorites_skipped_total{reason}, user_activity_favorites_retries_total, user_activity_favorites_dlq_total (ожидается 0), user_activity_favorites_lag_seconds (now − envelope.CreatedAt).
  • Traces: consume-span → handler-span → DB-spans; correlation из envelope пробрасывается в outbox action-tracked.
  • Alerts: лаг > 5 мин (SLO README); DLQ > 0; всплеск invalid-owner (регресс контракта favorites).
  • Безопасность: system-actor (не внешний); контекст вне tenant (ADR-0052 п.2) — изоляция record-level; доступ к топикам pin.favorites.* — m2m ACL Kafka. Событие не содержит secrets/токенов/заметок.

Предусловия, окно согласованности и постусловия​

  • Предусловие: объект недвижимости добавлен/удалён из избранного в favorites (source of truth), событие favorite-added/favorite-removed опубликовано; у устройства/пользователя есть трекнутая сессия (иначе действие не привяжется — skipped-no-session).
  • Постусловие: в ленте действий владельца появилась ровно одна запись AddToFavorites/ RemoveFromFavorites на исходное сообщение (INV-ACT-004); опубликовано action-tracked.
  • Окно согласованности: цепочка асинхронна — favorites → (Kafka) → этот consumer → (Kafka action-tracked) → crm. Инвестор видит лайк в своей истории /profile/events после обработки consumer-ом (лаг обычно в пределах SLO README); агент видит FavoriteAdded в ленте лида после дополнительного лага CRM-консюмера. RemoveFromFavorites в ленту лида не попадает (downstream no-op), но в истории инвестора присутствует. В промежуточном состоянии (событие принято, но ещё не обработано) действие в истории не видно — это ожидаемая eventual-consistency, не потеря.

Acceptance Criteria​

#Given / When / Then
AC1Given favorite-added с валидными FavoriteId/PropertyObjectId и Active-сессией владельца When consumer обрабатывает событие Then создаётся AddToFavoritesActionEvent со снимком UserId/SessionId, публикуется action-tracked (ActionType = AddToFavorites), offset подтверждается после фиксации
AC2Given favorite-removed When обработано Then создаётся RemoveFromFavoritesActionEvent, action-tracked эмитится; лента лида CRM не меняется (no-op downstream)
AC3Given повторная доставка события с тем же MessageId When consumer её видит Then запись не дублируется (дедуп по MessageId и SourceMessageId, INV-ACT-004), ack без изменений
AC4Given событие с обоими или ни одним из UserId/AnonymousId (нарушение XOR INV-FAV-001) When обработано Then invalid-owner, ack-skip, без записи и без DLQ
AC5Given событие владельца без трекнутой сессии When резолв сессии пуст Then skipped-no-session, ack (не ошибка), запись не создаётся
AC6Given гостевой лайк (AnonymousId задан, UserId = null) с Active-сессией устройства When обработано Then запись создаётся со снимком AnonymousId и UserId = null; после связывания сессии действие читается в истории пользователя через связанную сессию (без backfill)
AC7Given PostgreSQL/Kafka недоступны When обработка падает Then бесконечный backoff-ретрай с паузой партиции; событие не теряется и в DLQ не уходит

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

  • Имя consumer-дока. README user-activity и action-event.md фиксируют канон consumers/favorites-events.md (этот файл); контракты favorites ссылаются на устаревший путь consumers/favorites-actions.md — расхождение имён, требует reconciliation в доках favorites перед реализацией.
  • action-tracked vs action-recorded. README (owning-service) именует produced-событие ActionTrackedKafkaEvent/pin.user-activity.action-tracked.v1; action-event.md — ActionRecordedKafkaEvent/ action-recorded.v1 (README «Принятые решения» #4). Здесь принято имя README; расхождение — assumption.
  • Якорение к сессии. INV-ACT-003 допускает создание сессии «оживлением устройства», но payload favorites не несёт Device — consumer только резолвит существующую сессию, иначе skipped-no-session (ADR-0056, консистентно с identity-signed-in.md).

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

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

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