Consumer: контроль смены телефона связанного инвестора (pin-investor-owner-identity-users)

Consumer контекста investor-owner на события пользователей identity: мониторинг расхождения телефона у уже связанного инвестора. Входной контракт — чужой дом identity (user-upserted.md); здесь описывается только то, что consumer читает и делает. Дом сравниваемого снапшот-поля OwnerMapping.PhoneNumber (зафиксирован на момент линковки, INV-7) — owner-mapping.md; правило «смена телефона НЕ разрывает связь» — assumption README (../README.md, раздел «Известные ограничения и assumptions»).

Назначение

Смена телефона инвестора в PIN после установки связи не разрывает связь: сопоставление Собственник ↔ Инвестор выполнено по телефону только на момент линковки. Этот consumer наблюдает поток pin.identity.users.v1 и, когда актуальный телефон связанного инвестора расходится со снапшотом связи, поднимает метрику расхождения и structured-лог для ручной сверки Службой заботы — без авторазрыва и без переустановки связи. Consumer не мутирует состояние связи (опционально — диагностическая пометка) и не производит событий (минимальность ADR-0056): расхождение — это операционный сигнал, а не доменный факт.

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

Поле	Значение
Сервис/контекст	investor-owner
Обрабатываемое событие	UserUpsertedKafkaEvent (event type pin.identity.user-upserted)
Topic	pin.identity.users.v1 (общий с событием блокировки пользователя; key = UserId)
ContractVersion	1; при несовпадении версии — warn-and-proceed (метрика investor_owner_identity_users_contract_mismatch_total)
Consumer group	pin-investor-owner-identity-users
Producer	identity (регистрация / обновление инвестора / подтверждение смены телефона / admin-API / синхронизация из Talent Identity) — контракт identity, upstream source of truth
Эффект (событие → что меняется)	Метрика/лог расхождения телефона (owned-состояние связи не меняется; опционально — диагностическая пометка)

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

• Из метаданных события доступны: MessageId (Guid v7), время события CreatedAt, версия контракта, AggregateId = UserId, AggregateType = User.
• Actor — system-actor consumer-а (не пользовательский контекст); tenant не применяется (ADR-0052 — данные инвестора вне tenant): чтение/запись связей ведётся с bypass record-level фильтра по InvestorUserId.
• Корреляция сквозного трейса восстанавливается из метаданных события (в исходящие события не пробрасывается — consumer событий не производит).
• Идемпотентность: на уровне сообщения — по MessageId (ключ investor-owner-identity-users:user-upserted:{MessageId}); на бизнес-уровне — сравнение значения телефона (шаг 4).

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

Контракт события — user-upserted.md (дом полей; FullNameSnapshot, CompanyId, Status, RegisteredVia, ReferrerUrl этим consumer-ом не читаются).

Поле payload	Тип	Обязательность	Как используется (маппинг)
UserId	Guid	Да	Ключ поиска связи по InvestorUserId; = AggregateId события
PhoneNumber	string (E.164)	Да	Нормализация E.164 → сравнение со снапшотом OwnerMapping.PhoneNumber
Type	string (enum UserType)	Да	Short-circuit: со связью Linked может совпасть только Investor; иное → not-linked no-op
Version	long	Да	Опциональный version-guard (tie-break при неизменном телефоне); structured-лог

Алгоритм

1. Контекст события и идемпотентность. Восстановить метаданные события (MessageId, CreatedAt, AggregateId = UserId); сверить версию контракта с 1 — при несовпадении warn-and-proceed (метрика investor_owner_identity_users_contract_mismatch_total); actor = system, tenant не применяется. Если сообщение с таким MessageId уже обработано → дубль: лог skipped-duplicate, ack (no-op).
2. Проверка входа. UserId заполнен; PhoneNumber непустой и нормализуем в E.164 — ненормализуемый номер → отказ INVESTOR_OWNER_IDENTITY_INVALID_PAYLOAD (non-retriable → DLQ + алерт). Существование связи входом не проверяется — её отсутствие есть допустимый no-op (шаги 3–4), а не ошибка.
3. Поиск связи. Найти активную связь инвестора: по InvestorUserId = UserId, Status = Linked, не архивированную. Единственность гарантирована ux_owner_mappings_investor_user_id (INV-5). Иных загрузок нет.
4. Сравнение телефона (перехода состояний нет):
   • not-linked (связь не найдена или Type != Investor): инвестор не связан с собственником (нет связи / PendingLink / Conflict / архив) → идемпотентный no-op, лог not-linked, фиксируется только отметка идемпотентности.
   • Сравнение: нормализованный телефон из payload сопоставляется со снапшотом OwnerMapping.PhoneNumber (зафиксирован на момент линковки — дом owner-mapping.md).
     • совпадение: лог skipped-no-change, no-op — телефон инвестора не расходится со снапшотом связи (бизнес-идемпотентность).
     • расхождение: НЕ разрывать и НЕ переустанавливать связь (assumption README; INV-7 ImmutableLinkKind — LinkKind/InvestorUserId иммутабельны) → поднять метрику investor_owner_link_phone_divergence_total, structured-лог phone-divergence с маскированными телефонами (+7***1234). Опционально — диагностическая пометка в ConflictDetails (маскированный телефон + MessageId/Version) без смены Status/LinkKind/ConflictReason (связь остаётся Linked — это диагностика, не конфликт связи).
   • Маппинга поле-в-поле в owned state нет: снапшот связи владеет Talent Owners, телефон инвестора владеет identity — consumer только сравнивает.
5. Согласованность. Базовый путь — фиксируется только отметка идемпотентности (owned state связи не меняется). Опциональный диагностический путь — обновление ConflictDetails/UpdatedAt связи; эта правка и отметка идемпотентности фиксируются в одном согласованном изменении. Offset подтверждается только после успешной фиксации.
6. Результат. Успех → ack (skipped-duplicate / not-linked / skipped-no-change / phone-divergence). Отказ INVESTOR_OWNER_IDENTITY_INVALID_PAYLOAD → non-retriable → DLQ + алерт. Инфраструктурный сбой (PostgreSQL/Kafka) → бесконечный backoff (событие не теряется).

Диаграмма

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

• На уровне сообщения: по MessageId события (ключ investor-owner-identity-users:user-upserted:{MessageId}) — повторная доставка после сбоя не дублирует alert/пометку.
• На бизнес-уровне: сравнение значения телефона — при неизменном телефоне skipped-no-change no-op; опциональный version-guard по payload.Version (применять только более новое событие) актуален лишь при добавлении watermark-поля версии identity в связь (вне текущего объёма — identity-версия в связи не денормализуется).
• Replay: да. Повторный проход по топику безопасен (чистое сравнение без разрушающих мутаций; dedup по MessageId подавляет повторный подсчёт расхождений). Runbook — «разбор расхождения телефона связи» (investor-owner-overview).

Retry и DLQ

DLQ-топик: pin.investor-owner.identity-users-dlq.v1 (исходное событие + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset).

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)	—	Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)	Нет	Runbook: восстановить зависимость
Невалидный payload (UserId пуст / телефон не нормализуется в E.164)	INVESTOR_OWNER_IDENTITY_INVALID_PAYLOAD	Нет (non-retriable)	Да + алерт	Исправление данных в identity → повторное событие

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

• Logs (structured, по событию): consumed / skipped-duplicate / not-linked / skipped-no-change / phone-divergence / invalid-payload c полями MessageId, InvestorUserId, payload.Version; телефоны маскируются (+7***1234), ФИО не логируется.
• Metrics: investor_owner_identity_users_lag_seconds (now − время события), investor_owner_identity_users_consume_duration_ms, investor_owner_link_phone_divergence_total, investor_owner_identity_users_dlq_total, investor_owner_identity_users_contract_mismatch_total.
• Traces: span приёма события → span обработки → span обращения к реестру связей; correlation из метаданных события.
• Alerts: рост investor_owner_link_phone_divergence_total — мониторинг Службой заботы (ручная сверка: смена номера vs передача номера другому владельцу); DLQ > 0; лаг > 15 мин (SLO README).
• Dashboard / Runbook: investor-owner-overview; runbook «разбор расхождения телефона связи».
• Безопасность: consumer под system-actor (не пользовательский контекст); tenant не применяется (ADR-0052) — чтение/запись связей с bypass record-level фильтра по InvestorUserId; PhoneNumber/ФИО — PII, маскирование в логах; телефон не публикуется (consumer не производит событий). Доступ к топику pin.identity.users.v1 — m2m ACL Kafka: investor-owner требуется добавить в перечень доверенных consumer-ов топика (в контракте user-upserted.md ещё не значится — расширение ACL для чтения PhoneNumber, assumption).

Acceptance Criteria

• Given связанный инвестор (Status = Linked) и user-upserted с телефоном, совпадающим со снапшотом связи, When consume, Then skipped-no-change no-op (бизнес-идемпотентность), связь не мутируется, ack.
• Given связанный инвестор и user-upserted с телефоном, расходящимся со снапшотом (INV-7 — снапшот иммутабелен), When consume, Then связь НЕ разрывается и НЕ переустанавливается, растёт investor_owner_link_phone_divergence_total, лог phone-divergence с маскированными телефонами; опционально диагностическая пометка в ConflictDetails без смены Status/LinkKind/ConflictReason.
• Given пользователь без связи, в PendingLink/Conflict/архиве, или Type != Investor, When consume, Then not-linked no-op (отсутствие связи — допустимо, не ошибка), ack.
• Given повторная доставка того же MessageId, When consume, Then skipped-duplicate no-op — расхождение не подсчитывается повторно.
• Given UserId пуст или телефон не нормализуется в E.164, When consume, Then INVESTOR_OWNER_IDENTITY_INVALID_PAYLOAD — non-retriable → DLQ + алерт.
• Given недоступность PostgreSQL/Kafka, When consume, Then бесконечный backoff (событие не теряется); повторный проход по топику безопасен (чистое сравнение без разрушающих мутаций).
• Given любой исход, When consume, Then consumer НЕ публикует событий и НЕ меняет Status/InvestorUserId связи (расхождение — операционный сигнал, не доменный факт).

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

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

• События (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)
• Консюмеры (1): Consumer: синхронизация из Talent Owners (pin-investor-owner-talent-owners-sync)
• index.md (1): Сервис Инвестор-Собственник (investor-owner)

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

• Дом снапшот-поля PhoneNumber связи и INV-7 (телефон зафиксирован на момент линковки): owner-mapping.md.
• Контракт входного события (чужой дом identity): user-upserted.md.
• Assumption «смена телефона не разрывает связь»: ../README.md.
• Смежный consumer (первичный источник связи и снапшота PhoneNumber): talent-owners-sync.md.
• Решения: ADR-0052 (инвестор вне tenant), ADR-0056 (simplicity-first: consumer не плодит событий).