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

Consumer: Identity Sign-Ins (UserActivitySignedInConsumer)

Назначение​

При успешном входе (confirmSignIn / verifySignInPassword / пин-код) identity публикует pin.identity.user-signed-in. Consumer применяет к активной сессии устройства правило INV-SES-002 «LinkOnce» (session.md):

  • гостевая Active-сессия (UserId == null) связывается с вошедшим пользователем (UserId: null → value, LinkedAt = now()) — гостевая активность «до входа» приписывается учётке (просмотры/действия читаются в истории пользователя, property-object-view.md «Правило чтения»);
  • вход другого пользователя на устройстве с уже связанной Active-сессией — текущая сессия форс-закрывается (Active → Inactive, Duration считается, session-ended), а новая рождается уже связанной (UserId установлен при создании);
  • повтор того же пользователя — no-op (идемпотентно).
MetadataЗначение
Owning teamPIN Engagement
Статус документаdraft
Область действияконтекст user-activity; таблицы user_activity.sessions, user_activity.property_object_views
Change contextgreenfield (ADR-0054 объединение; ADR-0056 simplicity-first, PostgreSQL-only)

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

ПолеЗначение
Сервис/контекстuser-activity (консюмеры и worker-ы)
ConsumerUserActivitySignedInConsumer
Обрабатываемое событиеpin.identity.user-signed-in
Topicpin.identity.sign-ins.v1 (6 партиций, key = UserId)
ContractVersion1; при несовпадении версии — warn-and-proceed
Consumer grouppin-user-activity-sign-ins
Produceridentity (confirm-операции входа; refresh-flow не публикует)
Эффект (событие → что меняется)связывание / форс-закрытие Active-сессии устройства (user_activity.sessions); закрытие открытых просмотров форс-закрываемой сессии (user_activity.property_object_views)

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

  • Из метаданных события доступны: MessageId (ключ дедупа), CreatedAt (= время входа), Version, AggregateId = UserId, AggregateType = "User".
  • Actor — system-actor консюмера (не пользовательский контекст). Контекст вне tenant (ADR-0052 п.2): изоляция record-level по AnonymousId/UserId.
  • Корреляция сквозного трейса восстанавливается из метаданных события и пробрасывается в исходящие события pin.user-activity.session-* / property-object-view-ended.
  • Идемпотентность: consumer-level — ключ user-activity-sign-in:{MessageId}; business-level — INV-SES-002 (см. «Идемпотентность и replay»).

Assumptions и зависимости (корреляция устройства)​

Событие входа (identity v1, PII-минимизировано) несёт UserId и identity-SessionId (RefreshSession.Id), но не device-AnonymousId этого контекста. Разрешение сессии устройства:

  • Deterministic-путь (фронт): POST /sessions/link (заголовок X-Anonymous-Id + JWT claim sub) — авторитетный линкер, фронт держит и AnonymousId (localStorage), и JWT; session.md перечисляет его источником UserId, session-linked.md — producer-ом session-linked.
  • Event-путь (этот consumer): идемпотентный комплемент, гарантирующий связывание/форс-закрытие без участия фронта. Требует device-AnonymousId; берёт его как additive-корреляцию на pin.identity.sign-ins.v1 (клиентский X-Anonymous-Id, пробрасываемый identity), консистентно с формулировкой user-signed-in.md «Связывание активной anonymous_id-сессии». Если identity ещё не эмитит поле — consumer инертен (warn + ack), связывание полностью обслуживает POST /sessions/link.

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

Контракт — event: pin.identity.user-signed-in; ниже — только поля, которые consumer реально читает, и их маппинг.

Поле payloadТипОбязательностьКак используется (маппинг)
AnonymousIdGuidДа (корреляция, см. Assumptions)Ключ поиска активной сессии устройства (AnonymousId + Status = Active)
UserIdGuidДаЦель связывания → Session.UserId; сравнение с Session.UserId (ветвление B/C/D)
OccurredAt (= время события)DateTimeOffsetДаМонотонный best-effort guard против out-of-order (ветка D); server-clock now() ставит сам consumer (INV-SES-004)
SignInMethodstring (enum Sms/Password/PinCode)ДаТолько лог/метрика; на семантику связывания не влияет; unknown → warn-and-proceed
SessionId (identity RefreshSession.Id)GuidДаНе ключ нашей Session; логируется как correlation-хэндл витрины сессий
IsFirstSignIn, DevicePlatform, DeviceDisplayNamebool/string?/string?—Не используются (Device фиксируется при startSession, не обновляется)

Алгоритм​

  1. Приём входа и контекста. Восстановить контекст события (MessageId, CreatedAt, AggregateId = UserId); actor = system; сверить версию с ContractVersion = 1 — при несовпадении warn-and-proceed; consumer- идемпотентность по ключу user-activity-sign-in:{MessageId} — дубль → лог skipped-duplicate, ack.
  2. Проверка входа. UserId непустой; AnonymousId непустой — при отсутствии/пустоте (корреляция не эмитится) лог skipped-no-correlation, ack (consumer инертен, см. Assumptions). SignInMethod вне enum → warn, продолжение. Существование пользователя identity.users(UserId) синхронно не проверяется (кросс-контекст, логическая ссылка session.md; валидность гарантирует producer).
  3. Резолв активной сессии устройства. Найти Active-сессию по AnonymousId под блокировкой строки (блокирующе, не с пропуском: consumer должен получить авторитетное состояние и сериализоваться против worker-а activity-closer и heartbeat, INV-SES-006). Результат единичный (INV-SES-001 SingleActivePerDevice), может отсутствовать.
  4. Бизнес-правила и переходы (INV-SES-002):
    • A — нет активной сессии (session == null): связывать/закрывать нечего (сессия закрыта по неактивности или не создана server-side). No-op, лог skipped-no-active-session; born-linked сессию создаст ближайший authenticated track/heartbeat/link (у consumer-а нет Device.UserAgent для создания).
    • B — гостевая сессия (session.UserId == null): связать сессию с UserId → UserId = UserId, LinkedAt = now(); Status остаётся Active. Поднимается событие SessionLinked.
    • C — уже связана с тем же (session.UserId == UserId): no-op (LinkOnce, повтор входа), лог skipped-already-linked.
    • D — связана с другим (session.UserId != null && != UserId), guard CreatedAt > session.LinkedAt (иначе — устаревший вход out-of-order → skipped-stale):
      • форс-закрытие текущей: Active → Inactive, EndedAt = now(), Duration = LastActivityAt − CreatedAt (INV-SES-005); открытые PropertyObjectView сессии закрываются тем же согласованным изменением (INV-LV-007, см. шаг 5). Поднимается событие SessionEnded.
      • создание новой born-linked: Status = Active, UserId установлен при создании (LinkedAt = CreatedAt = now()), Device переиспользуется (то же физическое устройство). Это не переход null → value, поэтому session-linked для новой сессии не публикуется (session-linked.md).
  5. Что меняется (порядок с учётом FK/INV-SES-001):
    • B: сессия обновляется (связана).
    • D: открытые Active-просмотры форс-закрываемой сессии закрываются (каждый — Duration, EndedAt, событие PropertyObjectViewEnded); затем старая сессия сохраняется как Inactive, новая — как Active в одном согласованном изменении (сохраняется partial-unique ux_sessions_active_anonymous, INV-SES-001).
    • Во всех ветках — отметка idempotency-ключа.
  6. Какие события фиксируются:
    • B: SessionLinkedKafkaEvent → pin.user-activity.session-linked.v1 (AggregateId = SessionId, ContractVersion = 1; payload — session-linked.md).
    • D: SessionEndedKafkaEvent → pin.user-activity.session-ended.v1 (закрытая сессия, UserId = прежний; payload — session-ended.md) + по одному PropertyObjectViewEndedKafkaEvent → pin.user-activity.property-object-view-ended.v1 на каждый закрытый PropertyObjectView (property-object-view-ended.md, чтобы PropertyObjectViewStats учла длительность). Новая born-linked сессия событий не даёт.
    • A/C: событий нет.
  7. Согласованность. Изменения sessions + property_object_views + отметка идемпотентности + исходящие события фиксируются в одном согласованном изменении; Kafka-offset коммитится только после успешной фиксации.
  8. Результат. Успех → ack. Инфраструктурная ошибка → retry по политике консюмера (см. «Retry и DLQ»). Метрики/логи — раздел «Наблюдаемость».

Диаграмма​

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

  • Consumer-level: дедуп по MessageId (ключ user-activity-sign-in:{MessageId}) — повторная at-least-once доставка after-commit-crash не мутирует состояние.
  • Business-level (INV-SES-002): связывание идемпотентно по построению — повторная обработка того же входа попадает в ветку C (сессия уже связана с UserId) → no-op. Форс-закрытие (D) при повторе не повторяется: после первой обработки активная сессия связана с UserId → снова ветка C.
  • Event — факт (append-only, не snapshot): version-guard не нужен (правило user-signed-in.md), достаточно дедупа по MessageId. Против out-of-order разных входов на одном устройстве — best-effort guard ветки D CreatedAt > session.LinkedAt (толерантен к межсерверному clock-skew).
  • Replay: безопасен (дедуп по MessageId + идемпотентность INV-SES-002). Полный re-read топика с offset 0 не нужен (identity — owner входов); точечный replay из DLQ — после ручного разбора.

Retry и DLQ​

DLQ-топик группы: pin.user-activity.sign-ins-dlq.v1 (событие + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset).

СбойRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)НетRunbook: восстановить зависимость
Нет корреляции AnonymousId (identity не эмитит поле)Нет — контролируемый skipped-no-correlation, ackНетНе ошибка: связывание обслуживает POST /sessions/link; при включении поля — replay не требуется
Устаревший вход out-of-order (guard ветки D)Нет — контролируемый skipped-stale, ackНетНет (штатная защита от out-of-order)
Contract mismatch (несовпадение версии) / unknown SignInMethodНет — warn-and-proceedНетОбновить маппинг при breaking-bump (.v2)
Непредвиденная ошибка обработки3 попытки немедленно (перечитать сессию)После исчерпанияРасследование гонки/бага

Acceptance Criteria​

#Given / When / Then
AC1 (ветка A)Given user-signed-in с AnonymousId, для которого нет Active-сессии When consumer обрабатывает Then no-op, skipped-no-active-session, ack; born-linked сессию создаст ближайший authenticated track/heartbeat/link (у consumer-а нет Device для создания)
AC2 (ветка B)Given гостевая Active-сессия устройства (UserId = null) When приходит вход UserId = U Then сессия связывается (UserId = U, LinkedAt = now(), Status остаётся Active), публикуется session-linked атомарно со сменой состояния
AC3 (ветка C)Given сессия уже связана с тем же UserId = U When приходит повторный вход U (или replay того же события) Then no-op, skipped-already-linked, событий нет (LinkOnce идемпотентен)
AC4 (ветка D)Given сессия связана с UserId = U1, CreatedAt входа > session.LinkedAt When входит другой UserId = U2 Then текущая сессия форс-закрывается (Active → Inactive, Duration считается), её открытые просмотры закрыты, публикуются session-ended + по одному property-object-view-ended; создаётся новая born-linked сессия (UserId = U2), для неё session-linked не публикуется
AC5 (out-of-order)Given сессия связана с U1, приходит устаревший вход (CreatedAt <= session.LinkedAt) When обработан Then skipped-stale, ack, состояние не меняется (guard ветки D)
AC6 (нет корреляции)Given identity ещё не эмитит AnonymousId When событие обработано Then consumer инертен: skipped-no-correlation, ack; связывание полностью обслуживает POST /sessions/link
AC7 (дубль)Given повторная at-least-once доставка с тем же MessageId When обработано Then skipped-duplicate, ack, без мутаций

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

  • Logs (structured, по событию): consumed / skipped-duplicate / skipped-no-correlation / skipped-no-active-session / linked / skipped-already-linked / user-switched / skipped-stale / failed с полями MessageId, AnonymousId, UserId, SessionId. PII нет; Device.* (UA/IP) в событии отсутствуют и не логируются.
  • Metrics: user_activity_signin_consume_duration_ms, user_activity_sessions_linked_total, user_activity_sessions_user_switched_total, user_activity_signin_stale_total, user_activity_signin_no_correlation_total, user_activity_signin_duplicates_total, user_activity_signin_retries_total, user_activity_signin_dlq_total (ожидается 0).
  • Traces: consume-span → span обработки → spans обращений к хранилищу; correlation из метаданных события пробрасывается в исходящие session-linked/session-ended/property-object-view-ended.
  • Alerts: lag > 1 мин (SLA user-signed-in.md); DLQ > 0; всплеск no_correlation (регресс контракта identity) или stale (нарушение порядка).
  • Безопасность: system-actor (не внешний); контекст вне tenant (ADR-0052 п.2) — изоляция record-level по AnonymousId/UserId, tenant-фильтр не применяется; доступ к топику pin.identity.sign-ins.v1 — m2m ACL Kafka. Событие не содержит secrets/токенов/кодов/телефона (user-signed-in.md).

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

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

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