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

Consumer: CRM Identity User Sync (CrmUserUpsertedConsumer)

Назначение​

Консюмер CRM Identity User Sync поддерживает в контексте crm display-реплику ФИО и телефона зарегистрированного пользователя на карточках клиентской базы (ClientEntity). Идентити — source of truth ФИО/телефона зарегистрированного клиента (client.md); CRM хранит их копию, чтобы списки клиентов, карточка клиента, полнотекстовый поиск клиентской базы (проекция ClientSearchDocument по ФИО + телефон + email) и дубль-детект фиксаций работали без синхронного запроса в identity. Один identity-пользователь может быть клиентом в нескольких агентствах — консюмер обновляет все карточки, где InvestorUserId равен идентификатору пользователя из события (fan-out кросс-tenant, см. Безопасность). Консюмер интеграционных событий не публикует (клиентская база — внутренние данные агентства).

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

ПолеЗначение
Сервис/контекстcrm
ConsumerCRM Identity User Sync
Обрабатываемое событиеpin.identity.user-upserted
Topicpin.identity.users.v1 (общий с событием блокировки пользователя; key = UserId, 6 партиций — дом user-upserted.md)
ContractVersionожидается 1; при несовпадении версии события — warn-and-proceed (метрика crm_identity_sync_contract_mismatch_total)
Consumer groupcrm-identity-user-sync
Produceridentity (регистрация / обновление профиля инвестора / подтверждение смены телефона / синк Talent / admin API)
Эффект (событие → что меняется)Перезаписывает display-поля (ФИО, телефон) всех карточек ClientEntity, привязанных к пользователю
Границы scope

Топик общий с событием блокировки пользователя, но CRM-клиент хранит только ФИО/телефон — статус/блокировка пользователя на display клиента не влияют (client.md не моделирует UserStatus на клиенте), поэтому событие блокировки этот консюмер игнорирует (обрабатывается только user-upserted). Привязка вручную добавленной карточки (InvestorUserId = null) к новой учётке по телефону — не ответственность этого консюмера: InvestorUserId пишет консюмер лид-заявки/referral (client.md, поле InvestorUserId).

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

  • Из метаданных события доступны: идентификатор сообщения MessageId (для дедупликации), CreatedAt, версия контракта, идентификатор пользователя (UserId).
  • Актор — системный контекст консюмера (не пользовательский); запись выполняется в platform-scope (адресация по InvestorUserId через границы агентств — единственная санкционированная кросс-tenant запись, ограниченная display-полями; ADR-0052).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Идемпотентность: на уровне консюмера — по ключу crm-identity-user-upserted:{MessageId}; на бизнес-уровне — version-guard версии события против сохранённой версии реплики на карточке (см. дельту ниже).

Требуемая дельта к дому client.md​

Version-guard требует хранить последнюю применённую доменную версию identity на карточке. Дом поля — client.md; этот консюмер требует дельту: ClientEntity.IdentityUserVersion : long? (nullable — для клиентов без InvestorUserId не заполняется; монотонный guard применённых snapshot-ов). Поле объявляется в доме client.md, здесь только используется.

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

Контракт — дом event-contract: pin.identity.user-upserted (payload UserUpsertedKafkaEvent + вложенная FullNameSnapshot). Ниже — только поля, которые consumer реально читает, и их маппинг на ClientEntity.

Поле payloadТипОбязательностьКак используется (маппинг)
UserIdGuidДаКлюч поиска всех живых карточек с этим InvestorUserId (fan-out кросс-tenant)
NameFullNameSnapshotДаПерезапись ClientEntity.Name (FullName): FirstName/LastName/MiddleName поэлементно (LastName допускает пустую строку)
PhoneNumberstringДаНормализация E.164 → перезапись ClientEntity.PhoneNumber; конфликт уникальности в tenant — см. шаг 4
VersionlongДаМонотонный guard → ClientEntity.IdentityUserVersion (business-идемпотентность)
Поля вне обработки

Поля Type, CompanyId, Status, RegisteredVia, ReferrerUrl consumer не читает — на display клиента не влияют; неизвестные значения enum-ов не парсятся (unknown-enum DLQ здесь не применяется).

Алгоритм​

  1. Контекст события. Восстановить метаданные события (идентификатор сообщения, идентификатор пользователя, время, версия контракта); при несовпадении версии — warn-and-proceed. Проверить идемпотентность по ключу crm-identity-user-upserted:{MessageId}: если событие уже обработано → пропустить как дубль (ack).
  2. Проверка входа. UserId задан; Name.FirstName непустое (≤ 128, при превышении truncate + warn — событие не теряется); PhoneNumber непустой; версия ≥ 1. Телефон нормализуется в E.164 — невалидный номер (producer гарантирует E.164, потому это аномалия) → CRM_IDENTITY_SYNC_INVALID_PHONE (non-retriable → DLQ + алерт). Существование карточек заранее не проверяется — операция upsert-проекции, пустой набор допустим (шаг 3).
  3. Поиск карточек (fan-out). Найти все живые (DeletedAt is null) карточки с этим InvestorUserId — platform-scope, кросс-tenant; набор мал (≤ одна карточка на агентство), допустимо пусто. Пустой набор → пользователь ещё не в чьей-либо клиентской базе → лог no-match, ставится только отметка идемпотентности, ack. Для этого поиска нужен индекс по InvestorUserId (дельта к client.md, раздел «Индексы»).
  4. Правила и маппинг (по каждой карточке).
    • Version-guard: версия события ≤ IdentityUserVersion карточки → устаревшее/дубль: метрика crm_identity_sync_stale_total, лог skipped-stale, карточка пропускается.
    • Маппинг поле-в-поле: Name ← ФИО из события; PhoneNumber ← нормализованный E.164; IdentityUserVersion ← версия события. Snapshot-семантика: перезаписываются только display-поля; локальные поля агента (Comment, Status, Source, ActiveLeadId) не трогаются. Проекция поиска ClientSearchDocument обновляется в том же согласованном изменении → поиск свежий.
    • Конфликт телефона (INV-CLI-1): новый телефон совпал с другой живой карточкой того же TenantId → неблокирующе: применить только Name и IdentityUserVersion, телефон не перезаписывать, лог phone-conflict (CRM_IDENTITY_SYNC_PHONE_CONFLICT), метрика; событие не теряется и не идёт в DLQ (расхождение разбирается заботой). Карточка без изменений исключается из записи.
  5. Что меняется. Обновляются изменённые карточки (правка не должна затирать параллельное ручное редактирование карточки агентом — при конфликте перечитать и повторить, см. Retry). Обновление карточек, проекции поиска и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается только после успешной фиксации.
  6. Результат. Успех → ack (в т.ч. no-match/skipped-*). Retriable-код → retry по политике; non-retriable → DLQ + алерт. Freshness: display-реплика eventually consistent, SLA < 1 мин (дом user-upserted.md, таблица потребителей).

Диаграмма​

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

  • На уровне консюмера: ключ crm-identity-user-upserted:{MessageId} — повторная доставка после сбоя не меняет состояние.
  • На бизнес-уровне: version-guard версии события против ClientEntity.IdentityUserVersion по каждой карточке — защита от out-of-order (после retry) и от replay целого топика (last-write-wins по key=UserId).
  • Replay: да. Полный повторный проход топика с offset 0 безопасен: per-key (UserId) порядок + version-guard дают сходимость к последнему snapshot-у. Runbook — «пересинхронизация display-реплики клиентов» (docs/06-services/crm/runbooks/).

Retry и DLQ​

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)— (exception)Бесконечный exponential backoff 1s→60s с паузой партицииНетRunbook: восстановить зависимость
Конкурентная правка карточки (гонка с ручным редактированием агентом)—3 попытки немедленно (перечитать карточку)После исчерпанияРасследование гонки
Невалидный телефон (аномалия producer-а)CRM_IDENTITY_SYNC_INVALID_PHONEНетДа + алертИсправление данных в identity → повторное событие
Конфликт телефона в агентстве (INV-CLI-1)CRM_IDENTITY_SYNC_PHONE_CONFLICTНетНет (телефон не перезаписан, Name применён)Забота: разбор дублирующихся карточек агентства
ContractVersion mismatch—warn-and-proceedНетПроверить версию контракта producer-а

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

  • Logs (structured): consumed / skipped-duplicate / skipped-stale / no-match / updated(count) / phone-conflict / failed c полями MessageId, UserId, payload.Version, затронутых ClientId. PII — маскирование: телефон +7***1234, ФИО И***.
  • Metrics: crm_identity_sync_consume_duration_ms, crm_identity_sync_updated_clients_total, crm_identity_sync_stale_total, crm_identity_sync_phone_conflicts_total, crm_identity_sync_dlq_total, crm_identity_sync_contract_mismatch_total, crm_identity_sync_lag_seconds (now − envelope.CreatedAt).
  • Traces: span приёма события → span обработки → spans обращений к хранилищу; корреляция из метаданных события.
  • Alerts: лаг консюмера > 5 мин (README crm, «Alerts»); DLQ > 0; всплеск phone_conflicts.
  • Dashboard / Runbook: Grafana pin-crm; runbooks docs/06-services/crm/runbooks/ (будущий путь).
  • Безопасность: system-actor; platform-scope запись (санкционированный кросс-tenant fan-out по InvestorUserId, ограничен display-полями — ADR-0052); PhoneNumber/Name — PII, топик доступен по m2m ACL Kafka (доверенный consumer), в логах маскируются; tenant mismatch по построению невозможен (адресация по InvestorUserId, каждая карточка несёт собственный TenantId).

Acceptance Criteria​

  • AC-1 (fan-out обновление). Дано: пользователь — клиент в двух агентствах, событие с Version = 5. Когда: user-upserted. Тогда: обе карточки получают новые ФИО/телефон, IdentityUserVersion = 5, проекция поиска ClientSearchDocument обновлена, ack.
  • AC-2 (устаревшая версия). Дано: карточка с IdentityUserVersion = 5. Когда: событие с Version ≤ 5. Тогда: skipped-stale, карточка без изменений, ack.
  • AC-3 (нет карточек). Дано: пользователь ни в чьей клиентской базе. Когда: событие. Тогда: no-match, ставится только отметка идемпотентности, ack.
  • AC-4 (конфликт телефона в агентстве). Дано: новый телефон совпал с другой живой карточкой того же TenantId (INV-CLI-1). Когда: событие. Тогда: применяются только Name и IdentityUserVersion, телефон не перезаписан, phone-conflict (CRM_IDENTITY_SYNC_PHONE_CONFLICT), без DLQ, ack.
  • AC-5 (невалидный телефон). Дано: PhoneNumber не приводится к E.164 (аномалия producer). Когда: событие. Тогда: CRM_IDENTITY_SYNC_INVALID_PHONE (non-retriable) → DLQ + алерт.
  • AC-6 (дубль). Дано: уже обработанный MessageId. Когда: повторная доставка. Тогда: skipped-duplicate, без изменений, ack.
  • AC-7 (игнор события блокировки). Дано: в общем топике приходит событие блокировки пользователя. Когда: доставка. Тогда: консюмер его игнорирует (обрабатывается только user-upserted), display клиента не меняется.

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

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

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