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

Consumer: Support Identity User Sync

Назначение​

Consumer «Support Identity User Sync» слушает событие pin.identity.user-upserted и поддерживает в контексте support display-реплику ФИО и телефона заявителя на карточках обращений (TicketEntity) и арбитражных дел (ArbitrageCaseEntity), чтобы списки/карточки и поиск тикетов (Elasticsearch-проекция support-tickets по номеру/теме/ФИО/телефону — ../search/tickets.md) работали без синхронного запроса в identity. Один identity-пользователь может быть заявителем нескольких тикетов и/или дел — консюмер обновляет все карточки, где RequesterUserId = payload.UserId (тикеты) и ComplainantUserId = payload.UserId (дела), fan-out в едином tenant платформы. Интеграционных событий консюмер не публикует (внутренние данные support, ADR-0056).

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

ПолеЗначение
Сервис/контекстsupport
ConsumerSupport Identity User Sync
Обрабатываемое событиеpin.identity.user-upserted (UserUpsertedKafkaEvent)
Topicpin.identity.users.v1 (общий с UserBlockedKafkaEvent; key = UserId, 6 партиций — дом user-upserted.md)
ContractVersionожидается 1; при несовпадении версии — warn-and-proceed (метрика support_identity_sync_contract_mismatch_total)
Consumer groupsupport-identity-user-sync
Produceridentity (регистрация / updateInvestor / confirmPhoneChange / синк Talent / admin-API)
Эффект (событие → что меняется)обновляется display-реплика ФИО/телефона заявителя: RequesterName/RequesterPhone на тикетах и ComplainantName/ComplainantPhone на арбитражных делах (+ переиндексация тикетов в support-tickets)

Границы scope. Топик общий с UserBlockedKafkaEvent, но display заявителя хранит только ФИО/телефон — статус/блокировка на карточку тикета не влияют (ticket.md/arbitrage-case.md не моделируют UserStatus заявителя), поэтому UserBlockedKafkaEvent этот consumer игнорирует. Email не синхронизируется: pin.identity.users.v1 email в payload не несёт (user-upserted.md — «Email не входит»); RequesterEmail/RequesterName.email фиксируются при создании тикета из ввода заявителя и этим консюмером не обновляются (задокументированное расхождение дома ticket.md, поле RequesterEmail).

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

  • Из метаданных события доступны: MessageId (Guid v7), CreatedAt, Version (= ContractVersion producer-а), AggregateId = UserId, AggregateType = "User".
  • Актор — system (не пользовательский контекст); scope — единый tenant платформы (support single-tenant, TenantId = PIN CompanyId, ADR-0052), fan-out идёт в пределах одного tenant (кросс-tenant записи нет).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Идемпотентность: consumer-level — ключ support-identity-user-upserted:{MessageId}; business-level — version-guard payload.Version против сохранённой версии реплики на каждой карточке (IdentityRequesterVersion / IdentityComplainantVersion, дома ticket.md/arbitrage-case.md).

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

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

Поле payloadТипОбязательностьКак используется (маппинг)
UserIdGuidДаКлюч fan-out: карточки заявителя — тикеты по RequesterUserId + дела по ComplainantUserId
NameFullNameSnapshotДаПерезапись display-ФИО RequesterName/ComplainantName: FirstName/LastName/MiddleName поэлементно (LastName допускает пустую строку)
PhoneNumberstringДаНормализация E.164 → перезапись RequesterPhone/ComplainantPhone
VersionlongДаМонотонный guard → IdentityRequesterVersion/IdentityComplainantVersion (business-идемпотентность)

Поля Type, CompanyId, Status, RegisteredVia, ReferrerUrl consumer не читает (на display заявителя не влияют); Email в payload отсутствует (см. «Границы scope»).

Алгоритм​

  1. Контекст события. Восстановить метаданные (MessageId, AggregateId = UserId, время, версия); сверить версию с ContractVersion = 1 — при несовпадении warn-and-proceed (метрика support_identity_sync_contract_mismatch_total). Актор — system, единый tenant платформы. Проверить consumer-идемпотентность по ключу support-identity-user-upserted:{MessageId} — событие уже обработано → пропустить как дубль (ack, no-op), лог skipped-duplicate.
  2. Проверка входа. UserId задан; Name.FirstName непустое (≤ 128, при превышении truncate + warn — событие не теряется); PhoneNumber непустой; Version >= 1. Нормализация PhoneNumber в E.164 — невалидный номер (аномалия producer-а, гарантирующего E.164) → SUPPORT_IDENTITY_SYNC_INVALID_PHONE (non-retriable → DLQ + алерт). Проверка существования by-id не выполняется — это upsert-проекция, пустой набор карточек допустим (шаг 3).
  3. Что читаем (fan-out по двум агрегатам). В едином tenant платформы найти все карточки заявителя: тикеты, где RequesterUserId = payload.UserId и не скрыты (индекс ix_tickets_requester_user_id), и арбитражные дела, где ComplainantUserId = payload.UserId и не скрыты (индекс ix_arbitrage_complainant_user_id). Оба набора малы (десятки строк). Оба пусты → пользователь ещё не заявитель ни обращения, ни дела → лог no-match, только отметка идемпотентности, ack.
  4. Бизнес-правила и переходы (по каждой карточке из обоих наборов).
    • Version-guard: если payload.Version <= card.IdentityRequesterVersion (или card.IdentityComplainantVersion) → устаревшее/дубль: метрика support_identity_sync_stale_total, лог skipped-stale, карточка пропускается.
    • Маппинг display-реплики: ФИО из payload.Name (FirstName/LastName/MiddleName), телефон — нормализованный E.164 из payload.PhoneNumber, версия → guard-поле. Snapshot-семантика: перезаписываются только display-поля (RequesterName/RequesterPhone для тикета, ComplainantName/ComplainantPhone для дела); статусы/очередь/assignee/решение не трогаются. Для тикетов обновление реплики переиндексирует документ support-tickets (../search/tickets.md); арбитраж Elasticsearch-индекса не имеет (ILIKE по Number/Title). Карточка без изменений (guard/no-op) исключается из записи.
  5. Что меняется. Обновлённые тикеты и дела + отметка идемпотентности фиксируются в одном согласованном изменении; смещение (offset) подтверждается только после успешной фиксации. Гонка с командой оператора/арбитра разводится оптимистической блокировкой → повтор с перечитыванием (см. «Retry и DLQ»).
  6. Результат. Успех → ack (в т.ч. no-match/skipped-*). Retriable-код → retry по политике; non-retriable → DLQ + алерт. Freshness: display-реплика eventually consistent, SLA < 5 мин (README контекста, «Alerts»: лаг > 5 мин).

Диаграмма​

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

  • Consumer-level: по MessageId (support-identity-user-upserted:{MessageId}) — повторная доставка after-commit-crash не мутирует состояние.
  • Business-level: version-guard payload.Version против IdentityRequesterVersion / IdentityComplainantVersion по каждой карточке — защита от out-of-order и от replay целого топика (last-write-wins по key=UserId).
  • Replay: да. Полный re-read топика с offset 0 безопасен: per-key (UserId) порядок + version-guard дают сходимость к последнему snapshot-у. Runbook — «пересинхронизация display-реплики заявителей» (docs/06-services/support/runbooks/, будущий путь).

Retry и DLQ​

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

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

  • Logs (structured): consumed / skipped-duplicate / skipped-stale / no-match / updated(tickets+cases count) / failed c полями MessageId, UserId, payload.Version, затронутых TicketId/CaseId. PII — маскирование: телефон +7***1234, ФИО И***.
  • Metrics: support_identity_sync_consume_duration_ms, support_identity_sync_updated_tickets_total, support_identity_sync_updated_cases_total, support_identity_sync_stale_total, support_identity_sync_dlq_total, support_identity_sync_contract_mismatch_total, support_identity_sync_lag_seconds (now − время события).
  • Traces: span приёма события → span обработки → spans обращений к хранилищу; correlation из метаданных события.
  • Alerts: лаг консюмера > 5 мин (README контекста, «Alerts»); DLQ > 0.
  • Dashboard / Runbook: Grafana pin-support; runbooks docs/06-services/support/runbooks/ (будущий путь).
  • Безопасность: system-актор; единый tenant платформы (кросс-tenant записи нет, ADR-0052); PhoneNumber/Name — PII, топик доступен по m2m ACL Kafka (доверенный consumer), в логах маскируются; каждая карточка несёт собственный TenantId = PIN CompanyId.

Acceptance Criteria (Given / When / Then)​

#GivenWhenThen
AC1Пользователь — заявитель N тикетов и M арбитражных дел; payload.Version больше сохранённого на карточкахдоставлен UserUpsertedKafkaEventВсе N тикетов и M дел получают новые RequesterName/RequesterPhone и ComplainantName/ComplainantPhone; guard-версии обновлены; тикеты переиндексируются в support-tickets; ack после фиксации
AC2На карточке уже стоит версия >= payload.Versionдоставка событияКарточка пропускается (skipped-stale, метрика support_identity_sync_stale_total); last-write-wins по UserId
AC3UserId ещё не заявитель ни тикета, ни деладоставка событияno-match: только отметка идемпотентности, ack; мутаций нет
AC4Событие с тем же MessageId уже обработано (after-commit-crash / повтор)повторная доставкаskipped-duplicate (consumer-level ключ), состояние не мутируется, ack
AC5PhoneNumber не приводится к E.164 (аномалия producer-а)доставка событияSUPPORT_IDENTITY_SYNC_INVALID_PHONE → non-retriable → DLQ + алерт; исправление данных в identity → повторное событие
AC6ContractVersion события != 1доставка событияwarn-and-proceed (метрика support_identity_sync_contract_mismatch_total), обработка продолжается
AC7Гонка с командой оператора/арбитра (карточка изменена между выборкой и записью)конкурентная фиксацияКонфликт версии → 3 немедленных ретрая с перечиткой; после исчерпания — DLQ + расследование; ложной перезаписи чужих полей нет
AC8Полный re-read топика с offset 0 (replay)пересинхронизацияСходимость к последнему snapshot-у на каждый UserId (per-key порядок + version-guard); повторных мутаций уже актуальных карточек нет

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

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

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