Consumer: identity-users-sync (ProfileUserUpsertedConsumer, ProfileUserBlockedConsumer)

Consumer-документ контекста profile для топика pin.identity.users.v1: два consumer-а одной consumer group обрабатывают события учётки identity и ведут реплику профиля ProfileEntity (profile.md) с автосозданием персоны Self инвестора (person.md). Контракты входных событий — user-upserted и user-blocked (payload здесь не переописывается).

Назначение

profile не владеет контактами/ФИО/статусом учётки — владелец identity (user.md). Consumer-ы создают профиль ЛК при регистрации пользователя (первое UserUpserted, Version = 1 — покрывает сценарий «user-registered» исходного ТЗ), обновляют read-only реплику полей учётки при каждом изменении и блокируют/архивируют профиль вслед за identity. Для инвестора при создании профиля автосоздаётся персона Relation = Self (базовый набор персональных данных) — ФИО из события шифруется перед записью.

Поле	Значение
Owning team	PIN Platform Core
Статус	draft

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

Поле	Значение
Сервис/контекст	profile
Consumer A	ProfileUserUpsertedConsumer (событие UserUpsertedKafkaEvent)
Consumer B	ProfileUserBlockedConsumer (событие UserBlockedKafkaEvent)
Обрабатываемые события	pin.identity.user-upserted : UserUpsertedKafkaEvent; pin.identity.user-blocked : UserBlockedKafkaEvent
Topic	pin.identity.users.v1 (общий; key = UserId → строгий порядок upserted/blocked per user)
ContractVersion	UserUpsertedKafkaEvent.ContractVersion = 1; UserBlockedKafkaEvent.ContractVersion = 1; при несовпадении — warn-and-proceed
Consumer group	profile-identity-users-sync (оба consumer-а — одна group, один offset-поток на партицию)
Producer	identity (регистрация, updateInvestor, confirm-flow контактов, admin-API, talent-identity-sync)
Эффект (событие → что меняется)	UserUpsertedKafkaEvent → создание/обновление реплики ProfileEntity (+ автосоздание персоны Self инвестора при регистрации); UserBlockedKafkaEvent → перевод реплики профиля в Blocked

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

• из метаданных события доступны AggregateId = UserId, AggregateType = "User", время события; TenantId не используется (инвестор вне tenant, ADR-0052 — изоляция record-level по Id = UserId);
• корреляция сквозного трейса восстанавливается из метаданных события;
• бизнес-идемпотентность двухуровневая: MessageId (дубль доставки) + монотонный guard event.Version > Profile.ReplicaVersion (INV-PRF-1 MonotonicReplica — replay и out-of-order события применяются не более одного раза; ключи бизнес-идемпотентности: pin.identity.user-upserted:{UserId}:{Version} / pin.identity.user-blocked:{UserId}:{Version}).

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

Ожидается от identity (входные гарантии):

• Порядок. Топик pin.identity.users.v1 партиционирован по UserId → строгий per-user порядок UserUpserted/UserBlocked; между разными пользователями порядок не гарантирован.
• Версионирование. Version монотонно растёт на пользователя; ContractVersion = 1 (несовпадение — warn-and-proceed); неизвестные enum трактуются fail-closed (шаг 2 алгоритма A).
• Первое событие пользователя — всегда UserUpserted (Version = 1) со снапшотом полей учётки; UserBlocked снапшота для создания не несёт → при отсутствии профиля ретраится до догона upserted.
• Payload — trusted (внутренний контур, без секретов/пароля); email в контракте v1 не публикуется (реплика ContactEmail остаётся null до additive-расширения).

Предоставляется наружу (исходящие гарантии):

• Реплика ProfileEntity сходится к последнему применённому Version (eventual consistency, SLO лага консюмера < 1 мин); собственные поля ЛК (Locale/Timezone/VerificationStatus/ DefaultRequisiteTemplateId) реплика не перетирает (INV-PRF-1).
• При first-seen инвестора создаётся ровно одна персона Self (INV-PRS-1) и публикуется PersonUpsertedKafkaEvent (pin.profile.persons.v1, key = Id, ContractVersion = 1, без PII) для нижестоящих; при replay/гонке Self повторно не создаётся.
• Профиль собственных Kafka-событий наружу не транслирует; статус учётки потребителям рассылает identity, статус верификации — события заявки (pin.profile.verification-requests.v1).

Постусловие успешной обработки: для UserId существует профиль с ReplicaVersion = max(applied Version) и LastSyncedAt = now; для инвестора-first-seen — активная персона Self; offset подтверждён после фиксации.

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

Контракты: event-contract: pin.identity.user-upserted, event-contract: pin.identity.user-blocked. Ниже — только то, что consumer-ы реально читают, и маппинг в поля ProfileEntity / PersonEntity.

UserUpsertedKafkaEvent → ProfileEntity (+ PersonEntity при first-seen инвестора)

Поле payload	Тип	Обязательность	Как используется (маппинг)
UserId	Guid	Да	Profile.Id (PK, 1:1 c учёткой); Person.OwnerUserId при автосоздании Self.
Type	string (enum UserType)	Да	Profile.UserType (значения идентичны ProfileUserType); unknown-значение → DLQ, профиль не создаётся (правило profile.md). Type = Investor при first-seen → автосоздание персоны Self.
CompanyId	Guid?	Нет	Profile.CompanyId (информационная реплика).
Status	string (enum UserStatus)	Да	Profile.Status: Invited схлопывается в Active (для ЛК различие не нужно); Archived дополнительно ставит Profile.DeletedAt = envelope.CreatedAt, DeletedByUserId = null (system).
Name	FullNameSnapshot	Да	Profile.Name (owned FullName — открытая реплика отображаемого имени); при first-seen инвестора — plaintext-вход для EncryptedPersonName персоны Self (шифруется, см. шаг 4).
PhoneNumber	string	Да	Profile.ContactPhone (E.164, read-only реплика).
RegisteredVia	string (enum RegistrationSource)	Да	Производное Profile.ContactsManagedBy: TalentSync → TalentSync, иначе SelfService (иммутабельно с регистрации; в payload v1 отдельного поля ContactsManagedBy нет).
Version	long	Да	Profile.ReplicaVersion (монотонный guard INV-PRF-1); Version = 1 ⇔ регистрация (first-seen).
ReferrerUrl	string?	Нет	Не используется (правила привязки — контекст referral).

Поле Profile.ContactEmail: в payload user-upserted v1 email отсутствует (правило контракта identity — секреты/email не публикуются). До additive-расширения контракта (optional-поле Email — зафиксированный открытый вопрос между profile и identity) реплика остаётся null; ЛК отображает email из ответа identity getCurrentInvestor напрямую, инвариантов profile это не затрагивает.

UserBlockedKafkaEvent → ProfileEntity

Поле payload	Тип	Обязательность	Как используется (маппинг)
UserId	Guid	Да	Ключ загрузки Profile.Id.
BlockedAt	DateTimeOffset	Да	Profile.UpdatedAt мутации (фиксация момента блокировки).
Version	long	Да	Profile.ReplicaVersion (монотонный guard).
CompanyId, Reason	Guid?, string	Нет/Да	Не используются (Reason — для gateway/notifications; profile достаточно факта Blocked).

Алгоритм

Алгоритм A — обработка UserUpsertedKafkaEvent

1. Контекст события. Восстановить метаданные события (идентификатор сообщения, AggregateId = UserId, trace); сверить ContractVersion = 1 — при несовпадении warn-and-proceed. Проверить бизнес-идемпотентность по MessageId: дубль доставки → пропустить (ack, no-op).
2. Проверка входа. Разобрать enum Type/Status/RegisteredVia: unknown Type → PROFILE_UNKNOWN_USER_TYPE (обычно non-retriable → DLQ; профиль не создаётся до расширения enum — правило profile.md); unknown Status → трактовать как Blocked (fail closed); unknown RegisteredVia → SelfService + warning. Существование связанных сущностей не проверяется: событие identity — trusted (внутренний контур), UserId и есть создаваемый ключ.
3. Что читаем. Найти профиль по Id = UserId (может отсутствовать при первом событии). В ветке first-seen дополнительно проверить, нет ли уже активной персоны Self владельца (страховка от гонки replay; обычно нет).
4. Бизнес-правила и маппинг.
   • Монотонный guard (INV-PRF-1): если профиль есть и event.Version <= ReplicaVersion — идемпотентный пропуск (лог skipped-stale, без мутаций).
   • Создание (профиля нет): создать реплику профиля из полей события (маппинг по таблице выше); собственные поля по умолчанию: Locale = "ru-RU", Timezone = null, VerificationStatus = NotVerified, DefaultRequisiteTemplateId = null, Version = 1, ReplicaVersion = event.Version, LastSyncedAt = now, создатель — system. Профиль поднимает ProfileCreatedEvent. Если Type = Investor и активной Self ещё нет — автосоздаётся персона Self: ФИО из события шифруется текущим ключом (пустая фамилия при lazy-регистрации шифруется как есть; ограничение «1–128» действует для self-service ввода, инвестор дополнит фамилию при редактировании), Relation = Self, Status = Active, BirthDateCipher = null, CitizenshipCountryId = null, Gender = null, DocumentsCount = 0, Version = 1. Инвариант INV-PRS-1 SingleSelf защищён unique partial index ux_persons_owner_self (гонка → шаг 8). Персона поднимает PersonCreatedEvent.
   • Обновление (профиль есть): обновить только реплика-поля (UserType, CompanyId, Status, Name, ContactPhone, ContactsManagedBy) + ReplicaVersion = event.Version, LastSyncedAt = now; собственные поля (Locale/Timezone/VerificationStatus/DefaultRequisiteTemplateId) не трогаются (INV-PRF-1: реплика и self-операции меняют непересекающиеся наборы полей). Status = Archived → soft-delete вслед за identity; Status = Active при ранее Blocked — разблокировка. Смена ФИО учётки персону Self не перезаписывает: после создания персональные данные персоны — собственность инвестора (source of truth — person.md); PII персон стирается отдельным admin-регламентом crypto-erase (ADR-0053).
5. Что меняется. Создание: добавляются профиль и (для инвестора) персона Self в порядке FK-зависимости (сначала профиль-владелец). Обновление: обновляется профиль (правка не должна затирать более свежее состояние).
6. События. Доменные ProfileCreatedEvent, PersonCreatedEvent (в процессе при коммите). Интеграционное: при создании персоны Self — PersonUpsertedKafkaEvent (Id, OwnerUserId, Relation = "Self", Status = "Active", DocumentsCount = 0, Version = 1; topic pin.profile.persons.v1, ContractVersion = 1, key = Id, AggregateType = "Person" — контракт person-upserted.md). Ветка обновления собственных Kafka-событий не публикует (профиль наружу не транслируется — profile.md).
7. Согласованность. Профиль, персона, отметка идемпотентности и outbox-запись фиксируются в одном согласованном изменении; offset подтверждается после успешной фиксации.
8. Результат. Успех → ack. Конфликт unique-индекса ux_persons_owner_self или конкурентная self-операция → повторная обработка перечитывает состояние и сходится идемпотентно.

Алгоритм B — обработка UserBlockedKafkaEvent

1. Контекст события. Аналогично A (событие UserBlockedKafkaEvent, идемпотентность по MessageId, version warn-and-proceed).
2. Проверка входа. Reason не используется. Payload-валидация не требуется — trusted.
3. Что читаем. Найти профиль по Id = UserId. Если профиля нет — событие пришло раньше создания профиля (blocked не несёт снапшота для create): PROFILE_NOT_FOUND_RETRY → retry с backoff до догона upserted того же ключа (key = UserId упорядочен; ситуация возможна только при replay из середины топика).
4. Бизнес-правила. Guard event.Version > ReplicaVersion, иначе skip. Переход реплики Status: Active → Blocked (stateDiagram profile.md); ReplicaVersion = event.Version, LastSyncedAt = now, UpdatedAt = event.BlockedAt. Активные заявки верификации не отзываются: self-операции заблокированного отклоняет guard INV-PRF-3 ActiveOnly, ревьюер видит статус профиля в admin API.
5. Что меняется. Обновляется профиль (перевод в Blocked).
6. События. Не публикуются — profile наружу статус учётки не транслирует; блокировку потребителям уже разослал identity.
7. Согласованность. Профиль и отметка идемпотентности фиксируются в одном согласованном изменении.
8. Результат. Успех → ack; контролируемая ошибка → retry/DLQ по политике ниже.

Диаграмма

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

• Business key: pin.identity.user-upserted:{UserId}:{Version} (и аналогично для blocked); плюс MessageId на дубль доставки. Поведение дубля/устаревшего события — skip без мутаций (INV-PRF-1 MonotonicReplica).
• Replay всего топика — да, безопасен: реплика сходится к последнему Version; автосоздание Self повторно не выполняется (проверка активной Self + unique index). Runbook — «replay DLQ identity-sync» (README, раздел Runtime).

Retry и DLQ

Сбой	Retry	DLQ	Ручное действие
Transient БД / недоступность PostgreSQL	Да, экспоненциальный backoff	Нет	—
Недоступность ключей PersonalDataCrypto (шифрование ФИО Self)	Да, backoff; fail-closed — событие не ack-ается	Нет (алерт «ошибки PersonalDataCrypto > 0»)	Восстановить доступ к Vault/k8s secrets
Unknown enum UserType	Ограниченный retry (не лечится повтором)	Да	Расширить ProfileUserType + replay DLQ
UserBlocked раньше профиля (replay из середины)	Да, backoff до появления профиля	Да, после исчерпания	Replay события после догона upserted
Конфликт ux_persons_owner_self / оптимистической блокировки (гонка с self-операцией)	Да — повторная обработка сходится идемпотентно	Нет	—
Poison message (десериализация envelope)	Нет	Да	Разбор вручную, инцидент producer-у identity

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

• Logs: consumed / skipped-duplicate / skipped-stale (version guard) / profile-created / self-person-created / profile-blocked / failed. Без значений PII: ФИО/телефон в логах consumer-а маскируются по правилам identity (+7***1234), значения персоны не логируются вообще — только UserId/PersonId.
• Metrics: consume duration, lag consumer group profile-identity-users-sync (SLO < 1 мин — README), retries, DLQ depth, duplicates/stale-skips, счётчик self-person-created.
• Traces: span приёма события → span обработки → spans обращений к хранилищу/шифрованию; correlation из метаданных события.
• Dashboard: profile-overview (лаг consumer-а, DLQ); alerts: DLQ identity-sync > 0, ошибки PersonalDataCrypto > 0.
• Actor/permissions: system-actor (consumer), permission-проверок нет — trusted внутренний контур; tenant isolation не применяется (ADR-0052, record-level по Id = UserId); untrusted-вход отсутствует (payload identity — trusted; тем не менее enum-ы парсятся с fail-closed fallback).

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

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

• Домен (CQRS) (1): Entity / Aggregate: ProfileEntity
• API (5): GET /api/v1/internal/profile/verification-status/{userId} — Статус верификации для m2m (getVerificationStatus), GET /api/v1/profile/me — Получение своего профиля ЛК (getMyProfile), GET /api/v1/profile/verification — Моя заявка на верификацию (getMyVerification), POST /api/v1/profile/persons — Создание персоны (createPerson), POST /api/v1/profile/verification/{requestId}/revoke — Отзыв заявки KYC на доработку (revokeVerification)
• Use Cases (3): UC-PRF-001. Заполнение профиля и персоны «Сам» (первый вход инвестора в ЛК), UC-PRF-005. Запрет редактирования профиля для менеджера застройщика (TalentSync), UC-PRF-006. Смена email/phone из ЛК (делегирование в identity + обновление реплики)
• События (7): Event Contract: pin.crm.lead-created (LeadCreatedEvent), Event Contract: pin.identity.user-blocked (UserBlockedKafkaEvent), Event Contract: pin.identity.user-upserted (UserUpsertedKafkaEvent), Event Contract: pin.profile.person-upserted (PersonUpsertedKafkaEvent), Event Contract: pin.support.ticket-created (TicketCreatedKafkaEvent), Event Contract: pin.support.ticket-message-posted (TicketMessagePostedKafkaEvent), Event Contract: pin.support.ticket-status-changed (TicketStatusChangedKafkaEvent)
• Фоновые задачи (1): Background Job: ReEncryptRotationJob
• index.md (1): Profile Service (Pin.Profile)

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

• Входные контракты: user-upserted.md, user-blocked.md; дом события — user.md.
• Owned state: profile.md (INV-PRF-1..3, stateDiagram реплики), person.md (INV-PRS-1 SingleSelf, EncryptedPersonName).
• Производимое событие: person-upserted.md.
• Решения: ADR-0052 (инвестор вне tenant, record-level по UserId), ADR-0053 (crypto-erase PII), ADR-0056 (simplicity-first).