Consumer: Wallet Identity Display Sync (wallet-identity-users)

Назначение

Консюмер поддерживает в контексте wallet display-реплику холдеров (../domain/user-display-replica.md): ФИО агента/партнёра для подписи счёта, текущее агентство агента (id + название) для резолва ростера агентства и подписи счёта агентства. Цель — рисовать балансы человекочитаемо и определять состав агентства без синхронного запроса в identity на read-path. Идентити — source of truth ФИО и названий компаний; кошелёк хранит копию только отображаемых полей, обновляемую last-write-wins по версии. Собственных интеграционных событий консюмер не публикует.

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

Поле	Значение
Сервис/контекст	wallet
Consumer	Wallet Identity Display Sync
Обрабатываемые события	pin.identity.user-upserted (ФИО + текущее агентство), pin.identity.company-upserted (название агентства)
Topics	pin.identity.users.v1 (key = UserId; дом — user-upserted.md), pin.identity.companies.v1 (key = CompanyId; дом — company-upserted.md)
ContractVersion	1 по обоим; при несовпадении — warn-and-proceed (метрика wallet_identity_sync_contract_mismatch_total)
Consumer group	wallet-identity-users
Producer	identity (регистрация/обновление пользователя, смена агентства; создание/переименование компании)
Эффект (событие → что меняется)	Upsert строки UserDisplayReplicaEntity (по user-upserted) / обновление CompanyName строк агентов компании (по company-upserted)

Границы scope. Кошелёк хранит из identity только отображаемые поля: ФИО (DisplayName), текущее агентство агента (CompanyId + CompanyName). Телефон, email, Type, Status, RegisteredVia, ReferrerUrl, реквизиты компании, Slug, логотип не читаются и не реплицируются (data minimization: подписи счетов нужны только имя и название агентства). События блокировки пользователя и приостановки компании из общих топиков игнорируются: на подпись они не влияют, доступ отсекает identity (JWT), а учёт балансов ведёт журнал независимо от реплики. Permissions/права кошелёк не реплицирует — проверяет по токену.

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

Контракты — дом user-upserted.md и company-upserted.md. Ниже — только поля, которые консюмер реально читает.

pin.identity.user-upserted

Поле payload	Тип	Обязательность	Как используется (маппинг)
UserId	Guid	Да	Ключ upsert-а строки реплики (= Kafka key).
Name	FullNameSnapshot	Да	Схлопывается в DisplayName (INV-WLT-UDR-3).
CompanyId	Guid?	Нет	Текущее агентство агента → CompanyId; null — партнёр/инвестор/агент вне агентства.
Version	long	Да	Монотонный guard → IdentityUserVersion (INV-WLT-UDR-1).

pin.identity.company-upserted

Поле payload	Тип	Обязательность	Как используется (маппинг)
CompanyId	Guid	Да	Ключ выборки строк агентов этой компании (= Kafka key).
Name	string	Да	→ CompanyName строк с CompanyId = этой компании.
Version	long	Да	Монотонный guard → IdentityCompanyVersion (INV-WLT-UDR-2).

Type/Status/Requisites/Slug/LogoFileId не читаются: подписи нужно только название компании.

Алгоритм

1. Контекст события. Восстановить метаданные (идентификатор сообщения, ключ, время, версия контракта); несовпадение версии — warn-and-proceed. Идемпотентность по ключу wallet-identity-users:{MessageId}: уже обработано → пропустить как дубль (ack).
2. Проверка входа. Для user-upserted: UserId задан, Version ≥ 1; Name.FirstName при превышении 128 — truncate + warn (событие не теряется). Для company-upserted: CompanyId и Name заданы, Version ≥ 1. Нарушение → WALLET_IDENTITY_SYNC_INVALID_PAYLOAD (non-retriable → DLQ + алерт).
3. Version-guard.
   • user-upserted: строка существует и Version ≤ IdentityUserVersion → устаревшее/дубль (лог skipped-stale{source="user"}, метрика, ack).
   • company-upserted: применяется к каждой строке-агенту компании, у которой Version > IdentityCompanyVersion; строки с бо́льшим guard пропускаются (skipped-stale{source="company"}).
4. Применение.
   • user-upserted: first-seen UserId → строка создаётся; известный → DisplayName/CompanyId замещаются целиком (полный snapshot, не мёрж), IdentityUserVersion ← Version. При смене CompanyId CompanyName берётся из уже известной строки нового агентства (INV-WLT-UDR-4); если ни одной нет — CompanyName = null, IdentityCompanyVersion = null (дозаполнится ближайшим company-upserted). Анонимизированный snapshot (admin purge в identity) применяется тем же путём — PII-очистка доходит до реплики.
   • company-upserted: CompanyName ← Name, IdentityCompanyVersion ← Version для всех строк с CompanyId = этой компании (bulk). Строк нет (агентов ещё не реплицировали) — no-op, ack.
5. Согласованность. Применение и отметка идемпотентности — в одном согласованном изменении; offset подтверждается после фиксации.
6. Результат. Успех → ack (включая skipped-stale); retriable-ошибка → retry по политике; non-retriable → DLQ + алерт. Freshness: реплика eventually consistent, SLA < 1 мин (README, SLO лага consumer-ов).

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

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

Retry и DLQ

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

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

• Logs (structured): consumed / skipped-duplicate / skipped-stale / upserted / failed — с MessageId, ключом (UserId/CompanyId), Version, source (user/company). ФИО в логах маскируется (инициалы); телефон в payload user-upserted есть, но консюмером не читается и в логи не попадает. Названия компаний — Internal, не маскируются.
• Metrics: wallet_identity_sync_consume_duration_ms, wallet_identity_sync_upserted_total{source}, wallet_identity_sync_stale_total{source}, wallet_identity_sync_dlq_total, wallet_identity_sync_contract_mismatch_total, wallet_identity_sync_lag_seconds{source}.
• Traces: span приёма → span обработки → span записи; корреляция из метаданных события.
• Alerts: лаг консюмера > 5 мин; DLQ > 0.
• Безопасность: system-actor; реплика — platform-scope справочник без tenant-принадлежности (ФИО и название компании не являются tenant-данными; выборки кошелька применяют собственные record-level/scope правила поверх). Топики доступны по m2m ACL Kafka (доверенный consumer).

Acceptance Criteria

• AC-1 (first-seen пользователь). Дано: строки нет. Когда: user-upserted (Version = 1, агент с CompanyId = C). Тогда: строка создана с DisplayName, CompanyId = C, IdentityUserVersion = 1; ack.
• AC-2 (обновление ФИО). Дано: строка с IdentityUserVersion = 3. Когда: user-upserted Version = 5 с новым ФИО. Тогда: DisplayName замещён, IdentityUserVersion = 5.
• AC-3 (устаревшая версия пользователя). Дано: строка с IdentityUserVersion = 5. Когда: user-upserted Version ≤ 5. Тогда: skipped-stale{source="user"}, без изменений, ack.
• AC-4 (смена агентства). Дано: строка агента с CompanyId = A. Когда: user-upserted с CompanyId = B (агент сменил агентство). Тогда: CompanyId = B; CompanyName — из известной строки агентства B либо null до ближайшего company-upserted (INV-WLT-UDR-4). Ростер агентства A его больше не содержит.
• AC-5 (переименование компании). Дано: несколько строк агентов с CompanyId = C. Когда: company-upserted (C, новое Name, Version > текущего guard-а). Тогда: CompanyName обновлён у всех строк C, IdentityCompanyVersion поднят.
• AC-6 (устаревшая версия компании). Дано: строки с IdentityCompanyVersion = 4. Когда: company-upserted Version ≤ 4. Тогда: skipped-stale{source="company"}, названия без изменений, ack.
• AC-7 (анонимизация). Дано: admin purge в identity опубликовал анонимизированный snapshot с наибольшей версией. Когда: обработка. Тогда: DisplayName замещён анонимизированным значением — PII в кошельке не переживает удаление в identity.
• AC-8 (дубль). Дано: обработанный MessageId. Когда: повторная доставка. Тогда: skipped-duplicate, без изменений, ack.
• AC-9 (невалидный payload). Дано: company-upserted без Name. Когда: обработка. Тогда: WALLET_IDENTITY_SYNC_INVALID_PAYLOAD → DLQ + алерт, остальные холдеры не блокируются.

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

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

• Домен (CQRS) (1): Entity (Replica): UserDisplayReplicaEntity
• API (2): GET /api/v1/agency/wallet/agent-balances — Балансы агентов агентства, Wallet API — реестр операций и общие контрактные модели
• События (6): Event Contract: pin.crm.lead-created (LeadCreatedEvent), Event Contract: pin.identity.user-blocked (UserBlockedKafkaEvent), Event Contract: pin.identity.user-upserted (UserUpsertedKafkaEvent), Event Contract: pin.support.ticket-created (TicketCreatedKafkaEvent), Event Contract: pin.support.ticket-message-posted (TicketMessagePostedKafkaEvent), Event Contract: pin.support.ticket-status-changed (TicketStatusChangedKafkaEvent)
• index.md (1): Wallet Service

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

• Контракты входных событий (дом): user-upserted.md, company-upserted.md.
• Целевая реплика: ../domain/user-display-replica.md.
• Читатели реплики: ../api/platform/list-wallet-accounts.md, ../api/agency/list-agency-agent-balances.md, ../api/user/get-my-wallet.md.
• Эталон display-реплики: calendar/consumers/identity-user-upserted.md, messenger/domain/messenger-user.md.
• Шаблон consumer: consumer-template.md; README контекста wallet; ADR-0052, ADR-0056, ADR-0070.