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

Consumer: CRM Events → Notifications (notifications-crm-events)

Назначение​

Группа notifications-crm-events слушает интеграционные события crm (лиды, фиксации, сделки, бронирования, документы, изменения состава участников сделки/лида) и для каждого создаёт набор уведомлений (NotificationEntity) — fan-out получатель × канал — с учётом настроек получателя (user-notification-settings.md) и шаблонов (notification-template.md). Собственных доменных фактов consumer не меняет (кросс-контекстной записи в crm нет); действует по схеме persist-then-send: сначала уведомление в статусе Pending, затем отправку выполняет dispatcher.

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

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

Топики и маппинг событие → Kind → получатели​

Топик (pin.crm.*.v1)NotificationKindПолучатели (из payload)Urgency
lead-createdLeadCreatedагент-владелец лида (+ руководитель группы)Normal
lead-expiredLeadExpiredагент-владелец лидаNormal
fixation-approvedFixationApprovedагент-инициатор фиксацииNormal
fixation-rejectedFixationRejectedагент-инициатор фиксацииNormal
deal-createdDealCreatedполный активный состав сделки (Members — все UserId снапшота)Normal
deal-stage-changedDealStageChangedучастники сделкиNormal
deal-completedDealCompletedучастники сделкиNormal
deal-cancelledDealCancelledучастники сделкиNormal
booking-createdBookingCreatedучастники сделкиNormal
booking-confirmedBookingConfirmedучастники сделкиNormal
booking-paidBookingPaidучастники сделкиCritical
booking-cancelledBookingCancelledучастники сделкиNormal
booking-expiredBookingExpiredучастники сделкиCritical
document-requestedDocumentRequestedадресат запроса документаNormal
document-status-changedDocumentStatusChangedинициатор/участники сделкиNormal
deal-members-changedDealMemberAdded / DealMemberRemoved / DealExecutorChanged / DealResponsibleChangedпо дельте (см. ниже)Normal
lead-participants-changedLeadParticipantAdded / LeadParticipantRemoved / LeadExecutorChanged / LeadResponsibleChangedпо дельте (см. ниже)Normal

Дом полей payload и получателей — соответствующий event-контракт crm (напр. deal-created.md, booking-paid.md, deal-members-changed.md, lead-participants-changed.md).

Маппинг дельты состава участников (deal-members-changed / lead-participants-changed)​

События несут и полный активный состав (ActiveMembers/ActiveParticipants), и дельту (Members/Participants с ChangeType). Уведомления адресуются по характеру изменения:

ChangeType дельтыNotificationKind (deal / lead)ПолучателиСмысл
AddedDealMemberAdded / LeadParticipantAddedдобавленный участник (UserId записи дельты)«Вас добавили в сделку/лид» (с ролью из Role)
RemovedDealMemberRemoved / LeadParticipantRemovedудалённый участник (UserId записи дельты)«Вас удалили из сделки/лида» (в т.ч. каскад увольнения агента)
ExecutorChangedDealExecutorChanged / LeadExecutorChangedполный активный состав (ActiveMembers/ActiveParticipants)«Сменился исполнитель» — новому исполнителю передано ведение, состав информируется
ResponsibleChangedDealResponsibleChanged / LeadResponsibleChangedполный активный состав (ActiveMembers/ActiveParticipants)«Сменился ответственный (куратор)»
RoleChanged——смена роли того же участника уведомлением не сопровождается (пропуск)

Одно событие может нести несколько записей дельты (напр. передача ведения — Removed прежнего + ExecutorChanged нового): каждая обрабатывается независимо; итоговое множество (получатель, Kind) дедуплицируется (INV-NTF-1). Payload уведомления — DealMembershipNotificationPayload / LeadMembershipNotificationPayload (notification.md); ActorName (кто изменил состав) резолвится из контактной реплики по ChangedByUserId (null — системное изменение).

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

  • Из метаданных события доступны: MessageId (Guid v7, дедупликация), CreatedAt, Version (= ContractVersion), AggregateId, AggregateType.
  • Actor — system-actor консюмера; tenant уведомления берётся из получателя (INullableTenantedEntity: компания или null), не из tenant события.
  • Correlation сквозного трейса восстанавливается из метаданных события.
  • Идемпотентность: consumer-level — ключ notifications-crm:{EventType}:{MessageId}; business-level — unique (SourceMessageId, RecipientUserId, Channel, Kind) (INV-NTF-1) защищает от дублей fan-out при replay.

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

Consumer читает из каждого события: идентификаторы получателей (участники/владелец), маршрутные и display-поля для payload уведомления (notification.md) — маппинг поле-в-поле по таблице ниже и дому event-контракта. ФИО/контакты получателя в payload не приходят — адрес и локаль берутся снапшотом из контактной реплики recipient-contact.md на enqueue (шаг 3 алгоритма), а не синхронным lookup в identity/profile.

Поле payload (типовое)ТипОбяз.Как используется
DealId/BookingId/LeadId/FixationIdGuidДа→ id в payload уведомления + dedup-контекст + DeepLink
Number/Stage/Status/ExpiresAt/Amountскаляр/enum/MoneyЗав. от события→ типизированные поля payload (ref-модели для id+name)
состав участников (Members/ActiveMembers/ActiveParticipants — снапшот { UserId, Role })ListДля событий сделки/лидаполный активный состав → множество получателей fan-out (по всем UserId снапшота, после дедупликации), а не скалярные AgentUserId/InvestorUserId
дельта состава (Members/Participants c ChangeType)ListДля deal-members-changed/lead-participants-changedадресация точечных Kind по ChangeType (added/removed/executor/responsible — таблица выше)
адресат-скаляр (RecipientUserId — адресат документа)GuidДля document-requestedполучатель точечного уведомления, не несущего состав
TenantId/CompanyIdGuidДаконтекст; tenant уведомления берётся из получателя

Алгоритм​

  1. Контекст события. Восстановить метаданные события (MessageId, агрегат-источник, время); сверить Version с ContractVersion = 1 (несовпадение — warn-and-proceed); actor = system. Проверить consumer-идемпотентность по ключу notifications-crm:{EventType}:{MessageId}: событие уже обработано → лог skipped-duplicate, подтвердить (ack, no-op). Определить Kind по типу события и снять снапшот payload (SourceMessageId = MessageId, SourceTopic).
  2. Проверка входа. Обязательные идентификаторы непусты (Guid.Empty запрещён); обязательные для Kind поля payload присутствуют. Нарушение контракта → NTF_INVALID_PAYLOAD (non-retriable → DLQ + алерт).
  3. Резолв получателей и доставки. Собрать множество RecipientUserId из полного активного состава участников события (Members/ActiveMembers/ActiveParticipants — все UserId снапшота), а не из скалярных AgentUserId/InvestorUserId; для точечных Kind по дельте состава (deal-members-changed/lead-participants-changed, таблица «Маппинг дельты состава») получатель берётся из записи дельты (Added/Removed — конкретный участник) либо из полного ActiveMembers/ ActiveParticipants (ExecutorChanged/ResponsibleChanged); для точечных событий без состава (document-requested) — адресат-скаляр. Множество дедуплицировать, исключить инициатора там, где самоуведомление не нужно (напр. смена статуса собственным действием; для Removed — сам удалённый участник получает уведомление); пустое множество → лог no-recipients, зафиксировать отметку идемпотентности, ack. По каждому получателю:
    • определить каналы для Kind/Urgency по настройкам получателя (INV-UNS-3/4: дефолты Kind + явные opt-in/out; GlobalMute и правило Critical);
    • подобрать шаблон (Kind, Locale, Channel) (INV-TPL-5; fallback локали ru; если тела канала нет — канал пропускается); если активного шаблона под Kind/каналы нет — NTF_TEMPLATE_NOT_RESOLVED (non-retriable → DLQ + алерт);
    • адрес доставки и локаль — снапшот из контактной реплики RecipientContactEntity (recipient-contact.md) по RecipientUserId (INV-RC-4/5); отсутствие реплики/точки канала не ошибка: внешний адрес остаётся пустым и догоняется адаптером/следующим апдейтом реплики (graceful); in-app адрес не требуется;
    • тихие часы: если получатель в окне тишины, канал interruptive и Urgency != Critical — доставка откладывается до открытия окна (INV-NTF-4).
  4. Материализация. Для каждой валидной пары (получатель, канал) создаётся NotificationEntity в статусе Pending с отрисованными Title/Body/DeepLink; дубль по ключу (SourceMessageId, RecipientUserId, Channel, Kind) (INV-NTF-1) пропускается (идемпотентно). In-app уведомление после фиксации доставляется realtime.
  5. Согласованность. Все созданные Notification(Pending) и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается только после успешной фиксации. Отправку по каналам выполняет dispatcher (persist-then-send); интеграционный факт dispatched (../events/notification-dispatched.md) эмитит dispatcher при переходе Sent/терминальный Failed, а не этот consumer.
  6. Результат. Успех (в т.ч. no-recipients/skipped-duplicate) → ack. Retriable-код → retry; non-retriable → DLQ + алерт.

Диаграмма​

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

  • Consumer-level: ключ notifications-crm:{EventType}:{MessageId} — повторная доставка того же события пропускается.
  • Business-level: unique (SourceMessageId, RecipientUserId, Channel, Kind) (INV-NTF-1) — replay топика не создаёт дублирующих уведомлений.
  • Replay: да. Повторный re-read безопасен (dedup-ключ). Runbook — notifications/runbooks/ (будущий путь).

Retry и DLQ​

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны)—Exponential backoff с паузой партицииНетRunbook: восстановить зависимость
Нарушение контракта payloadNTF_INVALID_PAYLOADНетДа + алертСогласовать контракт с crm, replay
Шаблон (Kind, Locale) не резолвится (нет Active)NTF_TEMPLATE_NOT_RESOLVEDНетДа + алертЗавести/опубликовать шаблон (SEED/admin), replay
Конкурентная гонка при dedup—Оптимистичный ретрай по версииНетАвтоматически

Отсутствие контактной реплики/точки канала — не failure: уведомление создаётся, внешний адрес email/sms догоняется после апдейта реплики или адаптером (in-app доставляется всегда).

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

  • Ожидает от crm: интеграционные события на 17 топиках pin.crm.*.v1, key = id агрегата (DealId/BookingId/LeadId/FixationId/документ), ContractVersion = 1, доставка at-least-once (публикация через outbox источника). Порядок гарантирован только внутри одного агрегата (по ключу партиции); между разными агрегатами/типами — нет. Дом полей payload — event-контракты crm (../../crm/events/); ФИО/контакты в payload не приходят (только id).
  • Предоставляет: набор NotificationEntity (Pending) на пару получатель × канал; в crm ничего не пишет (кросс-контекстной записи нет). Финальный исход доставки эмитит dispatcher как pin.notifications.dispatched.v1, а не consumer.
  • При устаревшей реплике/несовпадении версии: ContractVersion != 1 — warn-and-proceed (обрабатывает по известным полям); отсутствующее обязательное поле — NTF_INVALID_PAYLOAD (DLQ), контракт согласуется с crm.

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

  • Предусловие. Событие получено; ContractVersion совместима; для целевого Kind/Locale существует Active-шаблон с включённым каналом.
  • Постусловие (успех). Для каждой разрешённой пары (получатель, канал) существует ровно один NotificationEntity (Pending, INV-NTF-1); отметка идемпотентности зафиксирована в том же согласованном изменении; offset подтверждён после фиксации; in-app доставлено realtime. Для BookingPaid/ BookingExpired (Urgency = Critical) доставка идёт без учёта тихих часов и GlobalMute (INV-NTF-4, INV-UNS-4). Инициатор действия исключён там, где самоуведомление не нужно.

Acceptance criteria​

  • Given booking-paid с 3 участниками и Active-шаблоном, when consumer обрабатывает, then для каждого разрешённого канала создан Pending, Urgency = Critical игнорирует тихие часы, offset подтверждён после фиксации.
  • Given повторная доставка того же события (тот же MessageId), when обработка, then skipped-duplicate, новых уведомлений нет, ack.
  • Given replay всего топика, when повторное чтение, then dedup (SourceMessageId, RecipientUserId, Channel, Kind) (INV-NTF-1) не создаёт дублей.
  • Given событие, из которого множество получателей пусто (после дедупа/исключения инициатора), when обработка, then no-recipients, отметка идемпотентности зафиксирована, ack (без DLQ).
  • Given deal-created с составом Members из 4 участников, when обработка, then получатели DealCreated резолвятся по всем UserId снапшота Members (полный активный состав), а не по скалярным ExecutorAgentUserId/InvestorUserId.
  • Given deal-members-changed c дельтой ChangeType = Added, when обработка, then создаётся DealMemberAdded только для добавленного участника (UserId записи дельты) с ролью из Role.
  • Given lead-participants-changed c дельтой ChangeType = Removed (в т.ч. каскад увольнения агента), when обработка, then создаётся LeadParticipantRemoved для удалённого участника (сам удалённый — получатель, из состава не исключается).
  • Given deal-members-changed с передачей ведения (Removed прежнего + ExecutorChanged нового), when обработка, then DealExecutorChanged адресуется полному активному составу (ActiveMembers), DealMemberRemoved — прежнему исполнителю; множество (получатель, Kind) дедуплицируется (INV-NTF-1).
  • Given дельта ChangeType = RoleChanged, when обработка, then уведомление не создаётся (пропуск), остальные записи дельты обрабатываются.
  • Given для Kind нет Active-шаблона, when обработка, then NTF_TEMPLATE_NOT_RESOLVED → DLQ + алерт.
  • Given payload без обязательного идентификатора, when обработка, then NTF_INVALID_PAYLOAD → DLQ + алерт.
  • Given отсутствие контактной реплики/точки канала для получателя, when обработка, then уведомление создано, in-app доставлено, внешний адрес email/sms догоняется после апдейта реплики (не failure).
  • Given недоступность PostgreSQL/Kafka, when обработка, then backoff с паузой партиции, offset не сдвигается, после восстановления — повтор без дублей.

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

  • Logs: consumed / skipped-duplicate / no-recipients / enqueued (count по каналам) / failed с полями MessageId, EventType, Kind, затронутые RecipientUserId. PII (адрес/тело) — маскируется.
  • Metrics: notifications_enqueue_duration_ms, notifications_enqueued_total{kind,channel}, notifications_skipped_total{reason}, notifications_dlq_total, notifications_consumer_lag_seconds.
  • Traces: span приёма события → span обработки → spans обращений к хранилищу (уведомления + контактная реплика); correlation из метаданных события.
  • Alerts: лаг консюмера > 5 мин (README); DLQ > 0; всплеск NTF_TEMPLATE_NOT_RESOLVED.
  • Безопасность: system-actor; уведомление несёт TenantId получателя (null — инвестор); адрес/тело — PII; топики crm доступны по m2m ACL Kafka.

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

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

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