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

Consumer: Talent Identity Sync (pin-identity-talent-sync)

Консолидированный документ Kafka-синка из talent.microservices.identity в контекст identity PIN — один consumer group обрабатывает четыре потока событий (tenant-upsert, tenant-status, user-upsert, user-status).

Один документ на все четыре потока

Правила маппинга полей и владения полями — дом talent-replica.md; целевые сущности — company.md, user.md. Retry/DLQ/replay-политика группы описана здесь единожды на все четыре потока (см. разделы «Идемпотентность и replay» и «Retry и DLQ» ниже).

Назначение​

Consumer group pin-identity-talent-sync потребляет события Talent Identity о tenant-ах (компаниях-застройщиках) и их пользователях (менеджерах застройщиков) и поддерживает в PIN реплики: CompanyEntity (SyncSource = TalentIdentity) и UserEntity типа DeveloperManager. Каждое применённое изменение re-publish-ится собственными событиями PIN (pin.identity.companies.v1, pin.identity.users.v1, pin.identity.access.v1), чтобы downstream-контексты PIN не подписывались на топики Talent напрямую (правило границ — service-boundaries.md).

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

ПолеЗначение
Сервис/контекстidentity
Consumer grouppin-identity-talent-sync
Producertalent.microservices.identity (внешняя система; контракты — upstream source of truth, schema registry Talent; имена топиков — зафиксированное assumption README identity)
Топик 1talent.identity.tenants.v1 — события TenantUpsertedEvent, TenantBlockedEvent, TenantUnblockedEvent, TenantDeletedEvent
Топик 2talent.identity.users.v1 — события UserUpsertedEvent, UserBlockedEvent, UserUnblockedEvent, UserDeletedEvent
ContractVersionожидается 1 для всех событий Talent; при несовпадении версии — warn-and-proceed (метрика talent_sync_contract_mismatch_total)
Порядокkey события = TenantId/UserId Talent → все события одного агрегата в одной партиции, строгий порядок per-aggregate; кросс-агрегатный порядок tenant→user НЕ гарантирован (обрабатывается правилом R4 talent-replica.md)

Маппинг «событие → эффект» (бизнес-термины):

Событие(я)Эффект
TenantUpsertedEventСоздание/обновление CompanyEntity-реплики
TenantBlockedEvent / TenantUnblockedEvent / TenantDeletedEventCompany.Suspend(TalentBlocked) / Resume() / Archive()
UserUpsertedEventСоздание/обновление UserEntity (DeveloperManager) + дефолтный RoleAssignment при создании
UserBlockedEvent / UserUnblockedEvent / UserDeletedEventUser.Block()/Unblock()/Archive() + отзыв активных сессий/назначений

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

  • Из метаданных события доступны: MessageId (Guid v7, для дедупликации), CreatedAt (= OccurredAt события Talent), Version (= ContractVersion producer-а), идентификатор и тип агрегата-источника (TenantId/UserId Talent, Tenant/User).
  • Актор — система синка (не пользовательский контекст): создатель/изменивший мутируемых сущностей штампуется служебным идентификатором; синк пишет в platform-таблицу companies и кросс-tenant в users (tenant-фильтр не применяется).
  • Корреляция сквозного трейса восстанавливается из метаданных и пробрасывается в исходящие pin.identity.*.
  • Ключ идемпотентности: talent-sync:{EventType}:{MessageId}; бизнес-идемпотентность — version-guard R2 (payload.Version > entity.ReplicaVersion, INV-6 company / INV-10 user).

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

Контракты внешние (schema registry Talent — upstream source of truth); ниже — поля, которые consumers реально читают, и их маппинг. Полная таблица правил маппинга — дом talent-replica.md (разделы «Маппинг Talent tenant → CompanyEntity» и «Маппинг Talent user → UserEntity»); здесь — только связка «поле payload → шаг алгоритма».

TenantUpsertedEvent (snapshot tenant-а)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДаКлюч поиска компании по внешнему id Talent; при создании → CompanyEntity.Id и ExternalTalentTenantId (Id PIN = Id Talent)
Typestring (enum Talent)ДаМаппинг → CompanyType (Developer — основной поток); только при создании (INV-4 ImmutableType); неизвестное значение → DLQ
NamestringДаПерезапись CompanyEntity.Name (синк-владеемое поле, INV-5)
LogoFileIdGuid?НетПерезапись CompanyEntity.LogoFileId
Requisites.LegalNamestring?НетПерезапись Requisites.LegalName (owned VO CompanyRequisites)
Requisites.Innstring?НетПерезапись Requisites.Inn
Requisites.Kppstring?НетПерезапись Requisites.Kpp
Requisites.Ogrnstring?НетПерезапись Requisites.Ogrn
Requisites.LegalAddressstring?НетПерезапись Requisites.LegalAddress
Statusstring: Active/Blocked/DeletedДаМаппинг статуса: Active → Active, Blocked → Suspended, Deleted → Archived — методами агрегата с guard-ами переходов company.md
VersionlongДаМонотонный guard R2 → ReplicaVersion; денормализуется в TalentSyncState.LastAppliedVersion

TenantBlockedEvent / TenantUnblockedEvent / TenantDeletedEvent​

Поле payloadТипОбязательностьКак используется
IdGuidДапоиск компании по внешнему id Talent
VersionlongДаМонотонный guard R2

UserUpsertedEvent (snapshot пользователя Talent)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДапоиск пользователя по внешнему id Talent; при создании → UserEntity.Id и ExternalTalentUserId
TenantIdGuidДарезолв UserEntity.CompanyId по внешнему id tenant-а; компании нет → retriable-ошибка R4
FullName.FirstNamestringДаПерезапись Name.FirstName (owned FullName)
FullName.LastNamestringДаПерезапись Name.LastName
FullName.MiddleNamestring?НетПерезапись Name.MiddleName
PhoneNumberstringДаНормализация E.164 → перезапись PhoneNumber; конфликт уникальности → R6 (DLQ + Conflict)
Emailstring?НетПерезапись Email; при непустом — EmailConfirmedAt = момент события (Talent считается подтвердившим)
Statusstring: Active/Blocked/DeletedДаПри создании — Invited (в PIN ещё не входил); при обновлении: Active → Active, Blocked → Blocked, Deleted → Archived
VersionlongДаМонотонный guard R2 → ReplicaVersion

UserBlockedEvent / UserUnblockedEvent / UserDeletedEvent​

Поле payloadТипОбязательностьКак используется
IdGuidДапоиск пользователя по внешнему id Talent
VersionlongДаМонотонный guard R2

Алгоритм​

Отдельный проход для каждого из четырёх потоков событий.

A. Tenant upsert (TenantUpsertedEvent)​

  1. Контекст события. Восстановить метаданные (MessageId, CreatedAt, Id); сверить Version с ожидаемым ContractVersion = 1 — при несовпадении warn-and-proceed; идемпотентность по ключу talent-sync:tenant-upserted:{MessageId} — дубль → лог skipped-duplicate, ack.
  2. Проверка входа. Id задан; Name — непустая, ≤ 256 (при превышении — truncate + warn, событие не теряется); Version >= 0; маппинг Type → CompanyType — неизвестное значение → отказ IDENTITY_SYNC_UNKNOWN_COMPANY_TYPE (non-retriable → DLQ + алерт); маппинг Status → CompanyStatus — неизвестное значение → IDENTITY_SYNC_UNKNOWN_STATUS (та же DLQ-ветка). Существование сущности не проверяется — операция upsert, отсутствие допустимо.
  3. Что читаем. Компания по внешнему id Talent (может отсутствовать — первое событие tenant-а); запись синк-статуса TalentSyncState этого tenant-а (может отсутствовать).
  4. Правила и переходы.
    • Version-guard (R2/INV-6): если компания уже есть и payload.Version <= ReplicaVersion — событие устаревшее/дубль: метрика talent_sync_stale_events_total, лог skipped-stale, к шагу 7 (только отметка идемпотентности), ack.
    • Type-guard (INV-4): если компания есть и map(payload.Type) ≠ Type — отказ IDENTITY_SYNC_TYPE_CHANGED (non-retriable → DLQ, TalentSyncState.Status = DeadLettered): смена типа компании в Talent — аномалия, требует ручного разбора.
    • Создание (компании нет): создать CompanyEntity из снапшота Talent — Id = payload.Id, ExternalTalentTenantId = payload.Id, SyncSource = TalentIdentity, Status = Pending; сгенерировать Slug из Name (транслитерация + lower-kebab + разрешение коллизии суффиксом -2, -3… с проверкой уникальности, INV-1); дефолтный WorkspaceSettings по типу компании; затем активировать (Pending → Active) — либо сразу Suspend(TalentBlocked)/Archive(), если payload.Status = Blocked/Deleted. Поднимаются доменные CompanyCreatedEvent и CompanyStatusChangedEvent.
    • Обновление (компания есть): применить снапшот (R3 snapshot-семантика) — перезаписать ВСЕ синк-владеемые поля: Name, LogoFileId, Requisites поэлементно; ReplicaVersion = payload.Version; LastSyncedAt = момент события; локальные поля PIN (Slug, Description, WorkspaceSettings) не трогаются (R5, INV-5). Сверка статуса: map(payload.Status) ≠ Status → переход методом агрегата (Suspend(TalentBlocked) / Resume() / Archive()); совпадает — no-op.
    • Маппинг поле-в-поле — по таблице «Маппинг Talent tenant → CompanyEntity» talent-replica.md.
  5. Что меняется. Компания (создание/обновление; concurrency синка — version-guard R2, а не версия строки). Запись TalentSyncState: при отсутствии создаётся (AggregateType = Tenant, ExternalId = payload.Id, LocalId = company.Id), затем LastAppliedVersion = payload.Version, LastEventId = MessageId, LastEventAt = момент события, Status = Synced, LastError = null.
  6. События. CompanyUpsertedKafkaEvent → топик pin.identity.companies.v1, key = company.Id, AggregateType = Company, ContractVersion = 1 — полный snapshot (контракт); при переходе Active → Suspended дополнительно CompanySuspendedKafkaEvent (Reason = TalentBlocked, контракт).
  7. Согласованность. Компания + TalentSyncState + отметка идемпотентности + исходящие события — атомарно (R8). Offset подтверждается только после успешной фиксации.
  8. Результат. Успех → ack. Retriable-код → retry по политике (таблица ниже); non-retriable → DLQ + TalentSyncState.Status = DeadLettered + алерт.

B. Tenant status (TenantBlocked/TenantUnblocked/TenantDeleted)​

  1. Контекст. Метаданные соответствующего события; ключ идемпотентности talent-sync:tenant-{blocked|unblocked|deleted}:{MessageId}; дубль → skip.
  2. Проверка. Id задан, Version >= 0. Существование: компания по внешнему id Talent; отсутствует → отказ IDENTITY_SYNC_COMPANY_NOT_FOUND — retriable (out-of-order: TenantUpsertedEvent ещё не применён; key-партиционирование Talent делает это маловероятным, но политика R4 действует). Допустимость состояния — guard-ы переходов в методах агрегата (шаг 4).
  3. Что читаем. Компания (шаг 2) + её TalentSyncState.
  4. Правила и переходы. Version-guard R2 (устаревшее → skip). По типу события: TenantBlockedEvent → Suspend(TalentBlocked) (guard INV-8: только из Active; уже Suspended — идемпотентный no-op с обновлением ReplicaVersion); TenantUnblockedEvent → Resume() (Suspended → Active; из Archived — no-op + warn: Talent разблокировал архивированную в PIN компанию — метрика talent_sync_conflicts_total); TenantDeletedEvent → Archive() (R7: soft-delete, Status = Archived + DeletedAt, INV-7). Каждый переход поднимает CompanyStatusChangedEvent.
  5. Что меняется. Компания (обновление); TalentSyncState (Version/EventId/EventAt/Synced — как в A5).
  6. События. TenantBlockedEvent → CompanySuspendedKafkaEvent (Reason = TalentBlocked); TenantUnblockedEvent/TenantDeletedEvent → CompanyUpsertedKafkaEvent с актуальным snapshot (потребители видят Status = Active/Archived).
  7. Согласованность. Всё — атомарно.
  8. Результат. Как A8.

C. User upsert (UserUpsertedEvent)​

  1. Контекст. Метаданные события; ключ идемпотентности talent-sync:user-upserted:{MessageId}; дубль → skip.
  2. Проверка входа.
    • Id задан; TenantId задан; FullName.FirstName непустое (≤ 128, truncate + warn); PhoneNumber непустой; Version >= 0; Email при непустом — формат email (невалидный → warn + Email = null, событие не теряется);
    • нормализация PhoneNumber в E.164 — невалидный номер → отказ IDENTITY_SYNC_INVALID_PHONE (non-retriable → DLQ: без телефона менеджер не сможет войти СМС-flow);
    • существование компании (R4): компания по внешнему id tenant-а; отсутствует → отказ IDENTITY_SYNC_COMPANY_NOT_FOUND — retriable (exponential backoff → DLQ); компания появится своим TenantUpsertedEvent, событие пользователя не теряется;
    • допустимость состояния компании: Status = Archived → warn + продолжение (реплика пользователя создаётся; вход всё равно заблокирован INV-9 user.md).
  3. Что читаем. Пользователь по внешнему id Talent (может отсутствовать); его TalentSyncState; при создании дополнительно — владелец того же телефона среди активных пользователей PIN (проверка конфликта R6); для дефолтного назначения — системная SEED-роль developer-manager (CompanyId = null; отсутствие — фатальная ошибка конфигурации → retriable fail + алерт).
  4. Правила и переходы.
    • Version-guard R2/INV-10: payload.Version <= ReplicaVersion → skip-stale (метрика, отметка идемпотентности, ack).
    • Конфликт телефона (R6): при создании, если телефон занят другим PIN-пользователем (обычно инвестором) — отказ IDENTITY_SYNC_PHONE_CONFLICT (non-retriable): TalentSyncState.Status = Conflict, LastError (без PII), событие в DLQ, алерт → тикет заботы, ручной merge учёток; авто-merge запрещён (риск отдать чужой аккаунт). При обновлении с изменившимся номером — та же проверка перед перезаписью.
    • Создание (пользователя нет): создать UserEntity из снапшота — Type = DeveloperManager, Status = Invited (в Active переведёт первый успешный вход), ContactsManagedBy = TalentSync (self-service запрещён), RegisteredVia = TalentSync, SyncSource = TalentIdentity, ExternalTalentUserId = payload.Id; секреты (PasswordHash/PinCodeHash/2FA) НЕ реплицируются — вход в PIN тем же СМС-flow по реплицированному номеру. Если payload.Status = Blocked — сразу Block(); Deleted — Archive(). Поднимается UserRegisteredEvent. Дефолтное назначение роли: RoleAssignment(role = developer-manager, scope = Organization(CompanyId), grantedBy = система) — единственная роль, которую назначает синк (сужение до Project — вручную руководителем); инварианты INV-R2/R4/R6 проверяются при создании. Поднимается RoleAssignmentGrantedEvent.
    • Обновление (пользователь есть): применить снапшот — перезапись Name (owned FullName), PhoneNumber (E.164), Email (+EmailConfirmedAt при непустом); ReplicaVersion, LastSyncedAt; локальные поля (секреты, роли, LastSignInAt) не трогаются (R5). Сверка статуса: Blocked → Block() + отзыв активных RefreshSession (INV-S7, шаг 5); Active при Status = Blocked → Unblock(); Deleted → Archive() (+ отзыв назначений, шаг 5). Invited/Active PIN-статусы при payload.Status = Active не меняются (переход Invited → Active — только первым входом).
  5. Что меняется. Пользователь (создание/обновление) + при создании — дефолтный RoleAssignment. При блокировке — отзыв всех активных RefreshSession пользователя (Status = Revoked, причина AdminBlock); при архиве — тот же отзыв сессий + отзыв всех активных RoleAssignment. TalentSyncState — как A5.
  6. События. Всегда — UserUpsertedKafkaEvent → pin.identity.users.v1, key = user.Id (контракт); при создании назначения — UserAccessChangedKafkaEvent → pin.identity.access.v1 (контракт); при Block() — UserBlockedKafkaEvent (Reason = TalentBlocked, контракт); при Archive() — UserUpsertedKafkaEvent со Status = Archived + UserAccessChangedKafkaEvent (назначения отозваны).
  7. Согласованность. Пользователь + назначение + сессии + TalentSyncState + отметка идемпотентности + исходящие события — атомарно.
  8. Результат. Как A8.

D. User status (UserBlocked/UserUnblocked/UserDeleted)​

  1. Контекст. Ключ идемпотентности talent-sync:user-{blocked|unblocked|deleted}:{MessageId}.
  2. Проверка. Id задан, Version >= 0; пользователь по внешнему id Talent; отсутствует → отказ IDENTITY_SYNC_USER_NOT_FOUND — retriable (out-of-order против UserUpsertedEvent).
  3. Что читаем. Пользователь (шаг 2) + его TalentSyncState.
  4. Правила и переходы. Version-guard R2. UserBlockedEvent → Block() (guard user.md: Active|Invited → Blocked; уже Blocked — no-op + version bump); UserUnblockedEvent → Unblock() (Blocked → Active; Archived — no-op + warn + talent_sync_conflicts_total); UserDeletedEvent → Archive() (R7 soft-delete).
  5. Что меняется. Пользователь (обновление); при Block/Archive — отзыв активных RefreshSession (INV-S7); при Archive — отзыв активных RoleAssignment; TalentSyncState.
  6. События. Block → UserBlockedKafkaEvent (pin.identity.users.v1); Unblock → UserUpsertedKafkaEvent (snapshot Status = Active); Archive → UserUpsertedKafkaEvent (Status = Archived) + UserAccessChangedKafkaEvent (pin.identity.access.v1, отзыв назначений). Доменные — SessionRevokedEvent на каждый отзыв, RoleAssignmentRevokedEvent.
  7. Согласованность. Всё — атомарно.
  8. Результат. Как A8.

Диаграмма​

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

  • По событию: дедупликация по MessageId (ключи talent-sync:{eventType}:{MessageId}) — повторная доставка after-commit-crash не меняет состояние.
  • По бизнес-правилу: version-guard R2 — событие с Version <= ReplicaVersion пропускается (защита от out-of-order и от replay целого топика).
  • Replay: да. Полный re-read топиков с offset 0 безопасен (upsert + version-guard); runbook — «полная пересинхронизация» talent-replica.md (R9, job talent-initial-import). Точечный replay DLQ — после ручного resolution (merge учёток при R6) через операцию resolution конфликтов (перевод TalentSyncState.Status → Synced
    • повторная подача события).

Retry и DLQ​

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

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/шина недоступны, timeout)—Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)НетRunbook identity-overview: восстановить зависимость
Компания не найдена (out-of-order tenant→user, R4)IDENTITY_SYNC_COMPANY_NOT_FOUND6 попыток, backoff 30s → 1m → 2m → 4m → 8m → 16mДа, после исчерпанияПроверить, пришёл ли TenantUpsertedEvent; replay из DLQ
Пользователь не найден (status-событие раньше upsert)IDENTITY_SYNC_USER_NOT_FOUNDта же политикаДаТо же
Неизвестный enum (Type/Status Talent)IDENTITY_SYNC_UNKNOWN_COMPANY_TYPE / IDENTITY_SYNC_UNKNOWN_STATUSНет (non-retriable)Да + алертОбновить маппинг (релиз) → replay
Смена типа компании (INV-4)IDENTITY_SYNC_TYPE_CHANGEDНетДа + алертРучной разбор с Talent; решение по данным
Конфликт телефона (R6)IDENTITY_SYNC_PHONE_CONFLICTНетДа + TalentSyncState.Status = Conflict + тикет заботыРучной merge учёток → resolution → replay
Невалидный телефонIDENTITY_SYNC_INVALID_PHONEНетДа + алертИсправление данных в Talent → повторное событие
Конфликт версий при гонке с admin-операцией—3 попытки немедленно (перечитать агрегат)После исчерпанияРасследование гонки

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

  • Logs (structured, по событию): consumed / skipped-duplicate / skipped-stale / created / updated / status-changed / conflict / dead-lettered c полями EventId, ExternalId, payload.Version, entity.ReplicaVersion; PII (телефон/ФИО/email) — маскируется (+7***1234, d***@mail).
  • Metrics: talent_sync_lag_seconds (now − TalentSyncState.LastEventAt, по AggregateType), talent_sync_consume_duration_ms, talent_sync_stale_events_total, talent_sync_retries_total, talent_sync_dlq_total, talent_sync_conflicts_total, talent_sync_contract_mismatch_total.
  • Traces: consume-span → span обработки → spans хранилища; correlation из метаданных пробрасывается в исходящие pin.identity.*.
  • Alerts: лаг > 5 мин (SLO README identity); DLQ > 0; Conflict > 0; ошибки конфигурации (SEED-роль developer-manager не найдена).
  • Dashboard / Runbook: identity-overview; runbooks «replay DLQ синка», «merge учёток (R6)», «полная пересинхронизация (R9)».
  • Безопасность: consumers работают под system-actor (не пользовательский контекст); секреты входа не реплицируются; доступ к топикам Talent — m2m ACL Kafka; tenant mismatch невозможен по построению (адресация по ExternalTalent*Id), несоответствие payload.TenantId компании существующего пользователя → fail closed (IDENTITY_SYNC_TENANT_MISMATCH, DLQ).

Пред/постусловия​

  • До обработки события. Реплика может содержать или не содержать целевую компанию/пользователя (операция upsert). Порядок per-aggregate гарантирован ключом Talent (TenantId/UserId); кросс-агрегатный порядок tenant→user источником не гарантируется — компенсируется R4 (retry + backoff для user до появления компании).
  • После обработки применённого события. Целевая сущность отражает последний snapshot (ReplicaVersion = payload.Version, синк-владеемые поля перезаписаны, локальные — нетронуты, R5); TalentSyncState = Synced (либо Conflict/DeadLettered при R6/аномалии); ушли соответствующие pin.identity.* (companies/users [+ access при изменении прав/блокировке/архиве]); offset ack только после фиксации. Устаревшее (Version <= ReplicaVersion) или дубль (MessageId) — состояние не изменено (skip), ack. Downstream-контексты PIN зависят только от pin.identity.*, на топики Talent не подписаны (service-boundaries.md).

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

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

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