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

Consumer: Messenger User Replica Sync (MessengerUserUpsertedConsumer)

Назначение​

Consumer слушает событие pin.identity.users.v1 и поддерживает реплику MessengerUserEntity (ФИО + нормализованный телефон) в контексте messenger, чтобы мессенджер работал без синхронного запроса в identity: глобальный поиск контакта по телефону (маска фамилии до Accepted), валидация адресата заявки/сообщения (IsActive), display участника чата и карточки контакта. Обновление идёт с монотонным version-guard (INV-MU-1, last-write-wins по key = UserId); после сохранения реплики тот же обработчик реиндексирует search-документ контактов MessengerContactSearchDocument (Elasticsearch, search/contacts.md, ADR-0059). Интеграционных Kafka-событий consumer не публикует (внутренняя реплика).

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

ПолеЗначение
Сервис/контекстmessenger
ConsumerMessengerUserUpsertedConsumer
Обрабатываемое событиеUserUpsertedKafkaEvent
Topicpin.identity.users.v1 (общий с UserBlockedKafkaEvent; key = UserId, 6 партиций)
ContractVersionожидается 1; при несовпадении версии — warn-and-proceed (messenger_user_sync_contract_mismatch_total)
Consumer groupmessenger-identity-user-sync
Produceridentity: регистрация / обновление инвестора / подтверждение смены телефона / синк Talent / admin API
Эффект (событие → что меняется)Обновляется реплика пользователя и реиндексируется search-документ контактов

Границы scope. Топик общий с UserBlockedKafkaEvent; реплика хранит IsActive (адресуемость), а не enum-статус. Изменение статуса на Blocked/Archived приходит в UserUpsertedKafkaEvent.Status (fail closed: unknown-статус трактуется как Blocked) → IsActive = false; отдельный UserBlockedKafkaEvent этот consumer не обрабатывает.

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

  • Из метаданных события доступны: идентификатор сообщения (для дедупликации), время события, версия контракта, идентификатор и тип агрегата-источника (UserId / User).
  • Актор — системный; реплика person-level, не tenant-фильтруется (глобальный поиск сквозной, ADR-0052).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Ключ бизнес-идемпотентности: messenger-identity-user-upserted:{MessageId}; плюс монотонный payload.Version-guard против MessengerUserEntity.IdentityUserVersion (INV-MU-1).

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

Контракт — дом user-upserted.md (payload + FullNameSnapshot). Ниже — только читаемые поля и их маппинг на MessengerUserEntity.

Поле payloadТипОбязательностьКак используется (маппинг)
UserIdGuidДаКлюч реплики (Id = UserId)
NameFullNameSnapshotДаПлоские колонки FirstName/MiddleName/LastName (LastName допускает пустую строку)
PhoneNumberstringДаНормализация E.164 → PhoneNumber (уникален, INV-MU-2)
Statusstring (UserStatus)ДаIsActive = Status ∈ {Invited, Active}; {Blocked, Archived}/unknown → false (fail closed)
VersionlongДаМонотонный guard → IdentityUserVersion (INV-MU-1)

Поля Type, CompanyId, RegisteredVia, ReferrerUrl consumer не читает (на реплику контактов не влияют). ФИО/телефон/IsActive дополнительно маппятся в search-документ MessengerContactSearchDocument (поля — search/contacts.md); индексная запись выполняется после сохранения реплики.

Алгоритм​

  1. Контекст события. Восстановить метаданные события; сверить версию контракта с 1 (при несовпадении — warn-and-proceed). Проверить бизнес-идемпотентность по messenger-identity-user-upserted:{MessageId}: событие уже обработано → пропустить как дубль (skipped-duplicate, ack).
  2. Проверка входа. UserId задан; Name.FirstName непустое (≤ 128, при превышении — усечь + warning); PhoneNumber непустой; Version >= 1. Телефон нормализуется в E.164; невалидный номер (аномалия producer, который гарантирует E.164) → MSG_USER_SYNC_INVALID_PHONE (DLQ + алерт, без ретрая).
  3. Что читаем. Найти реплику по UserId (прямая адресация: Id = UserId). Отсутствие — первое появление реплики (создание на шаге изменений).
  4. Бизнес-правила и маппинг.
    • Version-guard (INV-MU-1): если реплика есть и payload.Version <= existing.IdentityUserVersion — доставка устаревшая/дубль/out-of-order: метрика messenger_identity_stale_events_total, лог skipped-stale, ack без изменений.
    • Маппинг поле-в-поле: FirstName/MiddleName/LastName из Name; PhoneNumber — нормализованный E.164; IsActive — по Status (fail closed для unknown/Blocked/Archived); IdentityUserVersion = payload.Version.
    • Конфликт телефона (INV-MU-2): новый PhoneNumber совпал с другой живой строкой реплики → неблокирующе: применить ФИО/IsActive/IdentityUserVersion, телефон не перезаписывать, лог phone-conflict (MSG_USER_PHONE_CONFLICT) + тикет заботы; событие не теряется и не идёт в DLQ.
  5. Что меняется. Реплика создаётся (первое появление) или обновляется по snapshot-у. Конкуренции нет: порядок обеспечивают монотонный IdentityUserVersion + key = UserId.
  6. Что фиксируется. Интеграционных Kafka-событий и realtime-push нет (INV-MU-4, downstream реагирует на события identity). После сохранения реплики обработчик реиндексирует search-документ контактов MessengerContactSearchDocument (Elasticsearch, search/contacts.md) — это индексная запись, не Kafka. Elasticsearch не участвует в транзакции PostgreSQL: индексация идёт после сохранения (read-your-writes eventual); сбой индексации ретраится, устойчивый рассинхрон добирает MessengerContactSearchRebuildJob.
  7. Согласованность. Реплика и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается только после фиксации. Индексная запись Elasticsearch — вне транзакции PostgreSQL (best-effort + rebuild-reconciliation, шаг 6).
  8. Результат. Успех → ack (в т.ч. skipped-*), SLA < 5 мин. Инфраструктурный сбой → ретрай (backoff); MSG_USER_SYNC_INVALID_PHONE → DLQ + алерт.

Диаграмма​

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

До обработки (истинно на входе):

  • Событие опубликовано identity при регистрации/обновлении публичных полей пользователя (ФИО/телефон/статус); messenger — только проекция, source of truth в identity.
  • Топик pin.identity.users.v1 общий с UserBlockedKafkaEvent; этот consumer обрабатывает только UserUpsertedKafkaEvent, статусные изменения приходят полем Status.
  • Порядок per-user сохранён ключом UserId; монотонность применения — payload.Version.

После успешной обработки (гарантируется):

  • Реплика MessengerUserEntity создана (первое появление) или обновлена snapshot-ом при payload.Version > IdentityUserVersion; IdentityUserVersion = payload.Version.
  • FirstName/MiddleName/LastName, нормализованный E.164 PhoneNumber (при отсутствии конфликта), IsActive (fail-closed для unknown/Blocked/Archived) отражают последний применённый snapshot.
  • Search-документ контактов реиндексирован (eventual, после сохранения реплики; при сбое — ретрай/rebuild-джоб).
  • Наружу Kafka не ушло (INV-MU-4); offset подтверждён после фиксации.
  • При устаревшей доставке (payload.Version <= IdentityUserVersion) реплика не изменена (skipped-stale), учтена метрика.

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

Ожидается от identity (вход pin.identity.users.v1): at-least-once доставка; ключ партиционирования UserId (порядок per-user); монотонный payload.Version (last-write-wins, INV-MU-1); PhoneNumber в формате E.164 (гарантия продьюсера — нарушение считается аномалией → DLQ); обязательные UserId, Name (FullNameSnapshot), PhoneNumber, Status, Version, MessageId, ContractVersion. Неизвестный Status трактуется fail-closed как Blocked (IsActive = false). Поля Type/CompanyId/RegisteredVia/ ReferrerUrl игнорируются (на реплику контактов не влияют).

Предоставляет: реплику downstream-операциям messenger (поиск по телефону, валидация адресата, display) — не через Kafka, а как локальное read-only состояние. Search-проекция MessengerContactSearchDocument (Elasticsearch) — вне транзакции PostgreSQL: read-your-writes eventual, устойчивый рассинхрон добирает MessengerContactSearchRebuildJob.

Окно консистентности: только что зарегистрированный пользователь становится находимым по телефону и адресуемым в заявке/сообщении лишь после применения события (обычно секунды; при лаге консюмера — до нескольких минут). Пока реплики нет — операции контакта видят его как несуществующего (MSG_CONTACT_USER_NOT_FOUND, contact.md INV-CT-6); это допустимая гонка, сходится после доставки.

Критерии приёмки (Given/When/Then)​

  • Happy (создание): Given нет реплики по UserId и событие с Version = N; When обработка; Then реплика создана, IdentityUserVersion = N, search-документ проиндексирован, ack.
  • Happy (обновление): Given реплика с IdentityUserVersion = N и событие Version = N+1; When обработка; Then поля обновлены, IdentityUserVersion = N+1, реиндексация.
  • Устаревшее событие: Given реплика Version = N и событие Version ≤ N; When обработка; Then реплика не изменена (skipped-stale), метрика messenger_identity_stale_events_total, ack.
  • Дубль: Given уже обработанный MessageId; When повторная доставка; Then no-op (skipped-duplicate), ack.
  • Смена статуса: Given событие Status ∈ {Blocked, Archived} или unknown; When обработка; Then IsActive = false (fail-closed), адресат недоступен для новых заявок/сообщений.
  • Конфликт телефона: Given PhoneNumber совпал с другой живой строкой реплики; When обработка; Then ФИО/IsActive/Version применены, телефон не перезаписан, phone-conflict (MSG_USER_PHONE_CONFLICT) + тикет заботы, событие не в DLQ.
  • Невалидный телефон: Given номер не приводится к E.164; When обработка; Then MSG_USER_SYNC_INVALID_PHONE, DLQ + алерт, без ретрая.
  • Сбой реиндексации: Given реплика сохранена, индексация Elasticsearch упала; When обработка; Then реплика зафиксирована и ack, реиндексация ретраится/добирается rebuild-джобом (метрика messenger_contact_search_index_failures_total).
  • Иная версия контракта: Given ContractVersion ≠ 1; When обработка; Then warn-and-proceed.

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

  • По сообщению: ключ messenger-identity-user-upserted:{MessageId} — повтор после сбоя не мутирует.
  • По версии: монотонный payload.Version-guard против IdentityUserVersion (INV-MU-1) — защита от out-of-order (после retry) и replay целого топика (last-write-wins по key = UserId).
  • Replay: да. Полный re-read с offset 0 безопасен: per-key порядок + version-guard дают сходимость к последнему snapshot-у. Runbook — «пересинхронизация реплики пользователей messenger» (runbooks/, будущий путь).

Retry и DLQ​

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka)—Бесконечный backoff 1s→60sНетRunbook: восстановить зависимость
Невалидный телефон (аномалия producer)MSG_USER_SYNC_INVALID_PHONEНетДа + алертИсправить данные в identity → повторное событие
Конфликт телефона (INV-MU-2)MSG_USER_PHONE_CONFLICTНетНет (ФИО/IsActive применены, телефон не перезаписан)Забота: разбор дублей телефона
ContractVersion mismatch—warn-and-proceedНетПроверить версию контракта producer

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

  • Logs (structured): consumed / skipped-duplicate / skipped-stale / upserted / phone-conflict / failed c MessageId, UserId, payload.Version. PII — маскирование: телефон +7***1234, ФИО И***.
  • Metrics: messenger_user_sync_consume_duration_ms, messenger_users_upserted_total, messenger_identity_stale_events_total, messenger_user_phone_conflicts_total, messenger_user_sync_dlq_total, messenger_user_sync_lag_seconds, messenger_contact_search_index_failures_total (сбои реиндексации search-документа — добор rebuild-джобом).
  • Traces: приём события → обработка → обращения к хранилищу; корреляция из метаданных события.
  • Alerts: лаг > 5 мин; DLQ > 0; всплеск phone_conflicts.
  • Безопасность: системный актор, person-level (не tenant-фильтруется); PhoneNumber/Name — PII, топик доступен по m2m ACL Kafka (доверенный consumer), в логах маскируются; поиск отдаёт маску фамилии до Accepted (INV-MU-3, messenger-user.md).

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

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

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