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

Consumer: Identity Events → Notifications (notifications-identity-events)

Назначение​

Группа notifications-identity-events слушает события identity (регистрация, доступы, входы, инвайты компаний, членство агента) и решает две задачи: (1) создаёт уведомления безопасности/доступа/приглашений (NotificationEntity); (2) ведёт локальную реплику контактов получателя (RecipientContactEntity) из pin.identity.users.v1 — ФИО, телефон, статус учётки, tenant. Часть событий общих топиков в уведомления не отображается (upsert без релевантного пользователю факта), но users-upsert при этом не выбрасывается: его ФИО/телефон/статус применяются к контактной реплике, чтобы при enqueue не делать синхронный lookup в identity. Собственных доменных фактов в identity consumer не меняет (кросс-контекстной записи нет; реплика контактов — внутренняя проекция самого notifications); persist-then-send. Общий алгоритм enqueue — как в crm-events.md; здесь — дельта источника и ведение реплики.

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

ПолеЗначение
Сервис/контекстnotifications (проект Pin.Notifications.Background)
Consumer groupnotifications-identity-events
Обрабатываемые событиятопики pin.identity.*.v1 (таблица ниже)
Produceridentity
ContractVersionожидается 1 по каждому топику; несовпадение — warn-and-proceed
Эффект (событие → что меняется)создаются NotificationEntity (Pending) для затронутых пользователей; апсертится RecipientContactEntity (реплика контактов) из users-топика

Топики и маппинг​

ТопикСобытие → NotificationKindПолучательUrgency
pin.identity.users.v1first-registration → WelcomeRegistered; любой upsert → апдейт контактной реплики (ФИО/телефон/статус/tenant); блокировка → реплика RecipientStatus = Blocked; уведомления по прочему upsert нетсам пользователь (уведомление) / реплика (все)Low
pin.identity.companies.v1suspend → CompanySuspended; прочий upsert — игнорадмины/менеджеры компанииNormal
pin.identity.access.v1смена ролей/permissions → AccessChangedзатронутый пользовательNormal
pin.identity.sign-ins.v1новый вход/устройство → SecurityNewSignInвошедший пользовательCritical
pin.identity.company-invitation-changed.v1 (forward, ADR-0061)приглашение создано → CompanyInvitationReceived; принято/отклонено → CompanyInvitationResolvedприглашаемый + инициаторNormal
pin.identity.agent-membership-changed.v1 (forward, ADR-0061)вступление/выход/смена агентства → AgentMembershipChangedагент + руководитель агентстваNormal

Дом полей payload — event-контракты identity (напр. user-upserted.md, user-signed-in.md, user-access-changed.md); инвайты/членство — forward-контракты (создаются работой ADR-0061), поля описаны в notification.md (SecurityNotificationPayload/AccessNotificationPayload/InvitationNotificationPayload). Контактные поля users-топика (Name/PhoneNumber/Status/CompanyId/Version) применяются к реплике RecipientContactEntity (а не к уведомлению).

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

  • Из метаданных события доступны MessageId, CreatedAt, Version, AggregateId; actor = system; tenant уведомления — по получателю (INullableTenantedEntity).
  • Идемпотентность: consumer-level ключ notifications-identity:{EventType}:{MessageId}; business-level — unique (SourceMessageId, RecipientUserId, Channel, Kind) (INV-NTF-1).

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

Поле payload (типовое)ТипОбяз.Как используется
UserId/CompanyId/InvitationIdGuidДаполучатель fan-out + id для payload уведомления
SignedInAt/Ip/Device/Locationскалярыsign-in→ SecurityNotificationPayload (Ip маскируется)
Role/Scope/Changeenum/строкаaccess→ AccessNotificationPayload (ref-модель роли)
CompanyName/Role/Stateстрока/enuminvite/membership→ InvitationNotificationPayload (ref-модель компании)
Name/PhoneNumber/Status/CompanyId/VersionFullNameSnapshot/строка/enum/Guid?/longusers-upsert→ апдейт реплики RecipientContactEntity (guard по Version, INV-RC-2); в уведомление не идут

Алгоритм​

Общая логика enqueue (резолв каналов/шаблона/адреса, материализация, согласованность) — как в crm-events.md; дельта источника — фильтр не-уведомляемых событий (шаг 1) и ведение контактной реплики из users-топика (шаг 1a).

  1. Контекст события. Восстановить метаданные и сверить ContractVersion (несовпадение — warn-and-proceed); проверить consumer-идемпотентность по notifications-identity:{EventType}:{MessageId} (дубль → skipped-duplicate, ack). Определить Kind по типу события; для users-топика (upsert/blocked) Kind может отсутствовать (не уведомляемый факт) — событие всё равно не выбрасывается, а идёт в ведение реплики (шаг 1a). 1a. Ведение реплики контактов (users-топик). Для pin.identity.users.v1 применить контактные поля к RecipientContactEntity по UserId (first-seen — создать, иначе — обновить): Name/PhoneNumber/RecipientStatus/TenantId, guard монотонности по Version (устаревшее событие — no-op, INV-RC-2); user-blocked → RecipientStatus = Blocked; анонимизация (Archived, наибольшая Version) очищает PII (INV-RC-3). Если событие не уведомляемое (upsert без Kind, кроме first-registration → WelcomeRegistered) — после апдейта реплики лог replica-updated, ack (уведомление не создаётся). Апдейт реплики фиксируется в том же согласованном изменении, что и отметка идемпотентности (шаг 5).
  2. Проверка входа. Обязательные идентификаторы непусты; иначе → NTF_INVALID_PAYLOAD (DLQ).
  3. Резолв получателя. Пользователь / админы компании / агент + руководитель; дедуп; пустое множество → no-recipients (ack).
  4. Резолв доставки и материализация. По каждому получателю — каналы для Kind/Urgency (INV-UNS-3/4; SecurityNewSignIn — Critical, opt-out игнорируется); локаль и адрес — снапшот из контактной реплики RecipientContactEntity (INV-RC-4/5; отсутствие точки канала не блокирует — graceful, in-app доступен всегда); шаблон (Kind, Locale, Channel); тихие часы; создаётся Notification(Pending) с рендером (dedup INV-NTF-1). Нет активного шаблона → NTF_TEMPLATE_NOT_RESOLVED (DLQ). Отправку выполняет dispatcher; собственных интеграционных событий consumer не публикует.
  5. Согласованность и результат. Уведомления, апдейт реплики и отметка идемпотентности фиксируются в одном согласованном изменении, offset — после фиксации. Успех/replica-updated/no-recipients → ack; retriable → retry; non-retriable → DLQ.

Диаграмма​

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

  • Consumer-level по MessageId; business-level — dedup-unique (INV-NTF-1). Replay: да (dedup-ключ).

Retry и DLQ​

СбойКодRetryDLQ
Инфраструктурный—backoff + пауза партицииНет
Нарушение payloadNTF_INVALID_PAYLOADНетДа + алерт
Шаблон не резолвитсяNTF_TEMPLATE_NOT_RESOLVEDНетДа + алерт

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

  • Ожидает от identity: события на топиках pin.identity.*.v1, key = id агрегата (UserId/CompanyId/ InvitationId), ContractVersion = 1, доставка at-least-once. Топик users несёт и не-уведомляемые upsert-факты, но они не выбрасываются: их контактные поля (Name/PhoneNumber/Status/CompanyId, guard по Version) применяются к контактной реплике RecipientContactEntity; уведомление создаётся только по уведомляемым фактам (маппинг «Событие → Kind»). Топик companies фильтруется как прежде (уведомляемый факт → Kind, прочее — ignored-event). Дом полей — event-контракты identity (../../identity/events/).
  • Предоставляет (реплика): RecipientContactEntity — долгоживущая контактная проекция получателя, из которой enqueue снимает RecipientAddress/Locale без синхронного lookup в identity; source of truth контактов остаётся за identity/profile, notifications держит только реплику.
  • Forward-топики company-invitation-changed/agent-membership-changed создаются работой ADR-0061: до их появления consumer подписан, но сообщений нет (пустой поток — не ошибка); при появлении контракта маппинг уже задан.
  • Предоставляет: NotificationEntity (Pending) для затронутых пользователей; в identity не пишет. Собственных интеграционных событий не публикует; финальный исход — за dispatcher-ом.

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

  • Предусловие. Событие получено; ContractVersion совместима. Для уведомляемого факта — есть Active-шаблон (Kind, Locale) с включённым каналом; для users-upsert достаточно валидного UserId (даже без уведомляемого факта — ведётся реплика).
  • Постусловие (успех). Для уведомляемого факта — для каждой разрешённой пары (получатель, канал) создан Pending (dedup INV-NTF-1) с адресом/локалью из контактной реплики; для users-upsert — реплика RecipientContactEntity отражает актуальные ФИО/телефон/статус (guard по Version). Отметка идемпотентности и апдейт реплики зафиксированы; offset подтверждён после фиксации. SecurityNewSignIn (Urgency = Critical) доставляется без учёта тихих часов и GlobalMute; Ip/устройство маскируются.

Acceptance criteria​

  • Given pin.identity.sign-ins.v1 о новом входе, when обработка, then SecurityNewSignIn создаётся, Critical игнорирует тихие часы/opt-out, Ip замаскирован, ack.
  • Given users-upsert без уведомляемого факта (обновление ФИО/телефона), when обработка, then контактная реплика RecipientContactEntity обновлена (guard по Version), уведомлений нет, replica-updated, ack (без DLQ); given companies-upsert без релевантного факта — ignored-event, ack.
  • Given для получателя есть контактная реплика, when enqueue, then RecipientAddress/Locale берутся из реплики без синхронного lookup; given реплики ещё нет (событие опередило users-upsert), then in-app создаётся, внешний адрес пуст и догоняется после апдейта реплики (graceful, не DLQ).
  • Given user-blocked/анонимизация в users-топике, when обработка, then реплика получает RecipientStatus = Blocked/Archived (PII очищена при Archived), внешние каналы подавляются, in-app сохраняется.
  • Given повтор того же события (тот же MessageId), when обработка, then skipped-duplicate, дублей нет.
  • Given пустое множество получателей после дедупа, when обработка, then no-recipients, отметка идемпотентности зафиксирована, ack.
  • Given для Kind нет Active-шаблона, when обработка, then NTF_TEMPLATE_NOT_RESOLVED → DLQ + алерт; given payload без обязательного id — NTF_INVALID_PAYLOAD → DLQ + алерт.
  • Given forward-топик инвайтов ещё не существует, when цикл потребления, then пустой поток без ошибок; при появлении события — маппится в CompanyInvitationReceived/CompanyInvitationResolved.

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

  • Logs: consumed/replica-updated/ignored-event/skipped-duplicate/no-recipients/enqueued/failed (PII/Ip/ФИО/телефон маскируются).
  • Metrics: notifications_enqueued_total{kind,channel}, notifications_consumer_lag_seconds{source=identity}, notifications_dlq_total.
  • Безопасность: system-actor; SecurityNewSignIn содержит маскированный Ip/устройство; топики по m2m ACL Kafka.

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

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

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