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

Consumer: синхронизация из Talent Owners (pin-investor-owner-talent-owners-sync)

Консолидированный документ Kafka-синхронизации из внешней системы Talent Owners в контекст investor-owner PIN. Одна consumer group обрабатывает три события Talent Owners. Дома полей целевых сущностей — owner-mapping.md и investor-property.md; здесь описываются только маппинг payload → поля и алгоритмы.

Назначение​

Consumer group pin-investor-owner-talent-owners-sync потребляет события Talent Owners о собственниках (talent.owners.owners.v1) и их владениях (talent.owners.ownerships.v1) и поддерживает в PIN:

  1. Реестр связей Собственник ↔ Инвестор (OwnerMappingEntity): сопоставление выполняется напрямую по PhoneNumber (E.164); если инвестора с таким телефоном ещё нет — consumer инициирует его создание в identity через internal-API provision-investor с заданным Id = OwnerId (id собственника = id пользователя-инвестора, правило INV-OWN-001);
  2. Портфель объектов инвестора (InvestorPropertyEntity, Source = TalentOwnersSync): владения собственника из Talent Owners становятся записями «Мои объекты недвижимости» связанного инвестора, включая уже проданные объекты («за сколько купили/продали, сколько и когда заработали»).

Каждое применённое изменение re-publish-ится собственными событиями PIN (pin.investor-owner.owner-linked.v1, pin.investor-owner.property-added.v1, pin.investor-owner.property-sold.v1), чтобы downstream-контексты PIN (notifications, profile, user-activity) не подписывались на топики Talent напрямую (правило границ — service-boundaries.md).

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

ПолеЗначение
Сервис/контекстinvestor-owner
Consumer grouppin-investor-owner-talent-owners-sync
ProducerTalent Owners (внешняя система; контракты — upstream source of truth, schema registry Talent; имена топиков и состав payload — зафиксированное assumption README investor-owner до подтверждения)
Топик 1talent.owners.owners.v1 — события OwnerUpsertedEvent, OwnerDeletedEvent
Топик 2talent.owners.ownerships.v1 — событие OwnershipUpsertedEvent (снапшот владения; удаление владения Talent не транслирует — компенсация только admin-операцией, см. ограничения)
ContractVersionожидается 1 для всех событий Talent; при несовпадении версии — warn-and-proceed (метрика investor_owner_sync_contract_mismatch_total)
Порядокkey Kafka: OwnerId (топик 1) / OwnershipId (топик 2) → строгий порядок per-aggregate внутри топика; кросс-топиковый порядок owner→ownership НЕ гарантирован — обрабатывается retriable-правилом INV-8 (владение ждёт связь)
DLQpin.investor-owner.talent-owners-sync-dlq.v1 (исходное событие + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset)

Три события группы и их эффект в owned-состоянии контекста:

СобытиеЭффект (событие → что меняется)
OwnerUpsertedEventUpsert связи OwnerMappingEntity; provision инвестора в identity (INV-OWN-001); переходы PendingLink → Linked / Conflict
OwnerDeletedEventSoft-delete связи (DeletedAt); записи портфеля НЕ трогаются (история живёт)
OwnershipUpsertedEventСоздание/upsert InvestorPropertyEntity (Source = TalentOwnersSync); фиксация продажи при проданном владении

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

  • Из метаданных события доступны: MessageId (Guid v7), время события CreatedAt (= OccurredAt события Talent), версия контракта, AggregateId (OwnerId/OwnershipId Talent), AggregateType (Owner/Ownership).
  • Actor — system-actor синхронизации (не пользовательский контекст): мутируемые сущности без пользователя-создателя (CreatedByUserId/UpdatedByUserId = null, правило домов). Сущности контекста не tenanted (ADR-0052); record-level фильтр по InvestorUserId для system-actor выключен (bypass, запись любых инвесторов).
  • Корреляция сквозного трейса восстанавливается из метаданных события и пробрасывается в исходящие pin.investor-owner.* события и в вызовы internal-API identity/chessboard.
  • Идемпотентность: на уровне сообщения — по MessageId (ключ investor-owner-talent-sync:{EventType}:{MessageId}); на бизнес-уровне — version-guard INV-6 (payload.Version > mapping.ReplicaVersion) для связей и unique ExternalOwnershipId (INV-7) для объектов портфеля.

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

Контракты внешние (schema registry Talent — upstream source of truth); ниже — поля, которые consumers реально читают, и маппинг в поля домов owner-mapping.md / investor-property.md.

OwnerUpsertedEvent (snapshot собственника, топик talent.owners.owners.v1)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДаКлюч поиска связи; при создании → OwnerMappingEntity.Id (= OwnerId, INV-OWN-001) и требуемый Id инвестора в запросе provision-investor
PhoneNumberstringДаНормализация E.164 → OwnerMappingEntity.PhoneNumber (ключ сопоставления); ненормализуемый → Conflict(PhoneInvalid) (INV-8 PhoneFormat)
FullName.FirstNamestringДа→ OwnerFullName.FirstName (снапшот; ≤ 128, при превышении truncate + warn)
FullName.LastNamestring?Нет→ OwnerFullName.LastName (null → пустая строка — допустимо домом полей)
FullName.MiddleNamestring?Нет→ OwnerFullName.MiddleName
VersionlongДаМонотонный guard INV-6 → ReplicaVersion; время события → LastSyncedAt

OwnerDeletedEvent (топик talent.owners.owners.v1)​

Поле payloadТипОбязательностьКак используется
IdGuidДаКлюч поиска связи
VersionlongДаМонотонный guard INV-6

OwnershipUpsertedEvent (snapshot владения, топик talent.owners.ownerships.v1)​

Поле payloadТипОбязательностьКак используется (маппинг)
IdGuidДа→ InvestorPropertyEntity.ExternalOwnershipId — ключ идемпотентности INV-7
OwnerIdGuidДаКлюч поиска связи; InvestorUserId берётся из OwnerMapping.InvestorUserId, OwnerMappingId = OwnerId
ObjectIdGuidДаСтабильный id объекта недвижимости в Talent Owners; → PropertyObjectId, если PinPropertyObjectId = null (см. правило PropertyObjectId ниже)
PinPropertyObjectIdGuid?НетId объекта недвижимости PIN (chessboard), если объект Talent сопоставлен с шахматкой; → PropertyObjectId при наличии
ObjectTitlestringДа→ PropertyObjectSnapshot.Title (≤ 256, truncate + warn)
Numberstring?Нет→ PropertyObjectSnapshot.Number
ProjectNamestring?Нет→ PropertyObjectSnapshot.Project (ref {ProjectId, ProjectName}; ProjectId — из обогащения chessboard; null, если объект недвижимости не сопоставлен с шахматкой PIN)
BuildingNamestring?Нет→ PropertyObjectSnapshot.Building (ref {BuildingId, BuildingName}; BuildingId — из обогащения chessboard; null при отсутствии сопоставления)
AreaTotaldecimal?Нет→ PropertyObjectSnapshot.AreaTotal (> 0, иначе null + warn)
Floorint?Нет→ PropertyObjectSnapshot.Floor (>= -5, <= 200, иначе null + warn)
Statusstring: Active / SoldДаActive → создание записи Active; Sold → создание сразу Sold либо фиксация продажи существующей записи; неизвестное значение → DLQ
PurchasePrice.AmountdecimalДа→ PurchasePrice.Amount (> 0, INV-4)
PurchasePrice.CurrencystringДаКод ISO-4217 Talent → резолв в PurchasePrice.CurrencyId по справочнику валют handbook (правило резолва валют, шаг C2); Currency.IsoCode резолвленной валюты → денормализованный CurrencyIsoCode записи
PurchasedAtDateTimeOffsetДа→ PurchasedAt (<= now() + 5 мин clock-skew allowance)
SalePrice.Amount / SalePrice.Currencydecimal / stringНет (обязательны при Status = Sold)→ цена продажи; код валюты резолвится тем же правилом → SalePrice.CurrencyId; RealizedProfit вычисляет агрегат (INV-5), извне не принимается
SoldAtDateTimeOffset?Нет (обязателен при Status = Sold)→ дата продажи; SoldAt >= PurchasedAt (INV-3)
VersionlongДаЛогирование/диагностика out-of-order; version-guard на записи портфеля не ведётся — порядок обеспечен Kafka key = Id + иммутабельностью покупки (INV-9) и невозвратностью Sold

Правило PropertyObjectId: PropertyObjectId = payload.PinPropertyObjectId ?? payload.ObjectId. При PinPropertyObjectId = null объект Talent не сопоставлен с шахматкой PIN: ProjectId/BuildingId/ DeveloperId остаются null, история цен invest-analytics и карточка каталога недоступны — карточка портфеля показывает только PropertyObjectSnapshot (ограничение README investor-owner). Уникальность INV-1 (InvestorUserId, PropertyObjectId, Active) при этом работает по ObjectId.

Internal-API identity provision-investor​

Контракт — будущий путь ../../identity/api/provision-investor.md (конвенция); здесь — модели, которые consumer реально использует. Операция на стороне identity идемпотентна (поиск по телефону/Id до создания), поэтому повторный вызов при retry безопасен — распределённая транзакция не нужна (simplicity-first, ADR-0056).

Запрос ProvisionInvestorRequest:

ПолеТипОбязательностьИсточникОписание
RequiredUserIdGuidДаpayload.Id (OwnerId)Требуемый id пользователя при создании (INV-OWN-001).
PhoneNumberstringДанормализованный payload.PhoneNumberE.164; ключ поиска существующего инвестора.
FullNameFullNameSnapshot (FirstName, LastName, MiddleName?)Даpayload.FullNameФИО для создаваемого инвестора (RegisteredVia = LazyRegistration); у существующего пользователя имя НЕ перезаписывается (правило дома owner-mapping.md).

Ответ ProvisionInvestorResponse:

ПолеТипОбязательностьОписание
InvestorUserIdGuidДаId найденного или созданного инвестора.
CreatedboolДаtrue — пользователь создан (InvestorUserId = RequiredUserId); false — найден существующий инвестор по телефону (его Id может отличаться от RequiredUserId).

Коды ошибок ответа → реакция consumer-а:

Код identityСмыслРеакция
IDENTITY_PROVISION_USER_NOT_INVESTORПользователь с телефоном существует, но Type != InvestorСвязь → Conflict(MatchedUserNotInvestor) — non-retriable, в admin-очередь
IDENTITY_PROVISION_ID_COLLISIONВ identity уже есть пользователь с Id = RequiredUserId, но с другим телефономСвязь → Conflict(IdentityIdCollision) — non-retriable, в admin-очередь
HTTP 5xx / timeout / недоступностьИнфраструктурный сбойСвязь сохраняется в PendingLink, отказ INVESTOR_OWNER_SYNC_IDENTITY_UNAVAILABLE — retriable

Алгоритм​

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

A. Собственник (OwnerUpsertedEvent) → связь + provision инвестора​

  1. Контекст события и идемпотентность. Восстановить метаданные (MessageId, CreatedAt, AggregateId = payload.Id); сверить версию контракта с 1 — при несовпадении warn-and-proceed. Если сообщение с таким MessageId уже обработано → дубль: лог skipped-duplicate, ack.
  2. Проверка входа. Id заполнен; FullName.FirstName непустое (≤ 128, truncate + warn); PhoneNumber непустой. Нормализация PhoneNumber в E.164 — ненормализуемый номер НЕ теряет событие: связь создаётся/обновляется снапшотом и переводится в Conflict(PhoneInvalid) (шаг 4), исходный номер — в ConflictDetails маскированно. Событие — upsert, отсутствие связи допустимо.
  3. Поиск связи и проверка телефона. Найти связь по Id = OwnerId (включая архивированную — повторное появление удалённого собственника снимает архив, восстановление реплики). Отдельно проверить, не занят ли нормализованный телефон другой не-конфликтной связью (INV-4 UniquePhone; unique ux_owner_mappings_phone_number — вторая линия защиты при гонке).
  4. Бизнес-правила и переходы.
    • Version-guard (INV-6, с поправкой на PendingLink): если связь существует и версия события <= ReplicaVersion и статус связи != PendingLink — событие устаревшее/дубль: метрика investor_owner_sync_stale_events_total, лог skipped-stale, фиксируется только отметка идемпотентности, ack. Записи в PendingLink guard не пропускает: это ретрай линковки после недоступности identity — снапшот уже применён, нужен только шаг provision (ветка линковки ниже).
    • Невалидный телефон: связь создаётся (если её нет) либо обновляется снапшотом и переводится в Conflict(PhoneInvalid) (INV-3 ConflictCompleteness: причина обязательна). Фиксируется доменный факт OwnerMappingConflictDetected. Provision не вызывается.
    • Конфликт телефона (INV-4): телефон уже занят другой связью → та же ветка с Conflict(PhoneAlreadyMappedToAnotherOwner) (в ConflictDetails — id конкурирующей связи, маскированный телефон). Уже установленная (Linked) связь при этом НЕ разрушается — конфликтует только новый/несвязанный собственник; разбор — admin-API resolve-owner-mapping-conflict.
    • Создание (связи нет): создать OwnerMappingEntity из снапшота Talent — Status = PendingLink, InvestorUserId = null, Id = OwnerId, ReplicaVersion = версия события.
    • Обновление (связь есть, версия события новее): перезаписать снапшот-поля OwnerFullName, PhoneNumber; обновить ReplicaVersion, LastSyncedAt; снять архив, если был. Поля состояния связи (InvestorUserId, LinkKind, LinkedAt) НЕ трогаются (INV-7 ImmutableLinkKind). Особый случай: связь Linked, телефон изменился, новый номер занят другой не-конфликтной связью — телефон НЕ перезаписывается (нарушил бы ux_owner_mappings_phone_number): warn + метрика investor_owner_sync_conflicts_total, связь и старый номер сохраняются, разбор — Служба заботы.
    • Линковка (только при Status = PendingLink после применения снапшота): вызвать internal-API identity provision-investor (RequiredUserId = OwnerId, нормализованный телефон, ФИО) — внешний вызов ДО локальной фиксации (идемпотентен на стороне identity):
      • успех, Created = true → связь PendingLink → Linked, LinkKind = InvestorCreated, InvestorUserId = OwnerId (guard INV-OWN-001 InvestorUserId == Id; при нарушении — INVESTOR_OWNER_ID_MISMATCH, DLQ + алерт: identity нарушил контракт), LinkedAt = now() (INV-2 LinkedCompleteness); фиксируется OwnerMappingLinked;
      • успех, Created = false → связь Linked, LinkKind = ExistingInvestorMatched (расхождение InvestorUserId != Id допустимо и фиксируется этим LinkKind); нарушение INV-5 UniqueInvestor (инвестор уже связан с другим собственником) ловится unique ux_owner_mappings_investor_user_id при фиксации → Conflict(PhoneAlreadyMappedToAnotherOwner) на повторной попытке;
      • IDENTITY_PROVISION_USER_NOT_INVESTOR → связь Conflict(MatchedUserNotInvestor);
      • IDENTITY_PROVISION_ID_COLLISION → связь Conflict(IdentityIdCollision);
      • недоступность/5xx/timeout → связь остаётся PendingLink (снапшот и ReplicaVersion сохраняются), отказ INVESTOR_OWNER_SYNC_IDENTITY_UNAVAILABLE — retriable; отметка идемпотентности при retriable-ошибке НЕ пишется (повторная доставка обработает линковку заново; version-guard пропустит PendingLink).
  5. Согласованность. Создание/обновление связи, отметка идемпотентности и (при переходе в Linked) исходящее событие фиксируются в одном согласованном изменении; offset — после фиксации. Гонка с admin-операциями (resolve-conflict) разрешается оптимистической блокировкой: конфликт → 3 немедленные попытки с перечиткой связи.
  6. Событие о связи. При переходе в Linked публикуется InvestorOwnerLinkedKafkaEvent → топик pin.investor-owner.owner-linked.v1, key = id связи; payload: OwnerId (= Id), InvestorUserId, LinkKind, LinkedAt, Version — телефон в payload НЕ передаётся (PII, правило дома owner-mapping.md). При Conflict интеграционное событие не публикуется (конфликт — внутренняя admin-очередь), фиксируется только доменный OwnerMappingConflictDetected.
  7. Результат. Успех → ack (включая ветки Conflict — конфликт зафиксирован в модели, событие обработано). Отказ retriable → retry по политике (таблица ниже); non-retriable → DLQ + алерт.

B. Собственник удалён (OwnerDeletedEvent) → архив связи​

  1. Контекст и идемпотентность. Метаданные из OwnerDeletedEvent (ключ investor-owner-talent-sync:owner-deleted:{MessageId}); дубль → skip.
  2. Проверка входа и поиск связи. Id заполнен. Найти связь по Id (включая архивированную). Если связи нет → лог warn delete-for-unknown-owner + метрика, отметка идемпотентности, ack (удаление никогда не виденного собственника — replay-безопасный no-op; retry не нужен: upsert после delete невозможен по version-guard).
  3. Version-guard и архив. INV-6 (версия события <= ReplicaVersion → skip-stale). Иначе — архивировать связь (DeletedAt = время события, DeletedByUserId = null, обновить ReplicaVersion, LastSyncedAt). Записи портфеля (по OwnerMappingId) не трогаются — delete behavior restrict дома owner-mapping.md: история покупок/заработка остаётся у InvestorUserId. Повторный delete → идемпотентный no-op. Downstream-события при архиве не публикуются: архив связи не меняет портфель (минимальность ADR-0056).
  4. Согласованность. Архив связи и отметка идемпотентности фиксируются в одном согласованном изменении; offset — после фиксации.
  5. Результат. Как A7.

C. Владение (OwnershipUpsertedEvent) → объект портфеля​

  1. Контекст и идемпотентность. Метаданные из OwnershipUpsertedEvent (ключ investor-owner-talent-sync:ownership-upserted:{MessageId}); дубль → skip.
  2. Проверка входа и резолв валют. Id, OwnerId, ObjectId заполнены; ObjectTitle непустой (≤ 256, truncate + warn); PurchasePrice.Amount > 0 (INV-4), PurchasedAt <= now() + 5 мин — нарушение → отказ INVESTOR_OWNER_SYNC_INVALID_PAYLOAD (non-retriable → DLQ: данные Talent требуют исправления). Маппинг Status: неизвестное значение → INVESTOR_OWNER_SYNC_UNKNOWN_STATUS (non-retriable → DLQ, unknown enum handling). Резолв валют: строковый код PurchasePrice.Currency (и SalePrice.Currency при продаже) сопоставляется по IsoCode со справочником валют handbook (../../handbook/domain/currency.md; микросправочник, кэшируется на стороне consumer-а; актуализация — через internal resolve ../../handbook/api/internal/resolve-references.md): найден → CurrencyId резолвленной валюты; пустой или неизвестный код → валюта по умолчанию RUB + warn + метрика investor_owner_sync_currency_fallback_total (данные Talent де-факто рублёвые, RUB-only ADR-0048 — событие не теряется). Недоступность справочника при пустом кэше → INVESTOR_OWNER_SYNC_HANDBOOK_UNAVAILABLE — retriable. При Status = Sold: SalePrice.Amount > 0, SoldAt != null, SoldAt >= PurchasedAt (INV-3), валюта продажи = валюте покупки — после резолва по CurrencyId (INV-5) — нарушение → та же ветка INVESTOR_OWNER_SYNC_INVALID_PAYLOAD.
  3. Существование и состояние связи (INV-8 LinkedOwnerOnly). Найти связь по OwnerId (не архивированную) — родитель проверяется первым. Если связи нет → INVESTOR_OWNER_MAPPING_NOT_FOUND — retriable (out-of-order ownership→owner: связь придёт своим OwnerUpsertedEvent). Если связь не Linked → INVESTOR_OWNER_MAPPING_NOT_LINKED — retriable (PendingLink доработает ветка линковки A; Conflict разрешит Служба заботы — после исчерпания retry событие ждёт в DLQ и переигрывается runbook-ом после resolution).
  4. Поиск записи портфеля и обогащение.
    • Найти запись портфеля по ExternalOwnershipId = payload.Id (включая архивированную): архивированная заботой запись НЕ пересоздаётся (компенсирующие правки не затираются синхронизацией) — лог warn skipped-archived, отметка идемпотентности, ack.
    • При создании определить объект недвижимости по правилу PropertyObjectId (PinPropertyObjectId ?? ObjectId) и проверить дубль по объекту (INV-1, шаг 5).
    • При создании и заданном PinPropertyObjectId — best-effort обогащение из chessboard (internal-API get-property-objects-by-ids, будущий путь): ProjectId, BuildingId, DeveloperId + дозаполнение пустых полей снапшота. Недоступность chessboard НЕ блокирует появление объекта: warn + метрика, снапшот создаётся из payload, ProjectId/BuildingId/DeveloperId = null, дозаполнение — компенсирующая admin-операция.
  5. Бизнес-правила и переходы.
    • Дубль между источниками (INV-1 SingleActivePerLot): если по паре (InvestorUserId связи, PropertyObjectId) уже есть активная запись (обычно объект уже создан consumer-ом crm-deal-completed по сделке PIN) → INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVE — non-retriable → DLQ + алерт: ручная сверка заботой (admin archive-investor-property + replay). Автоматический merge запрещён — финансовые данные. (unique ux_investor_properties_investor_property_object_active — вторая линия при гонке.)
    • Создание (записи нет): создать InvestorPropertyEntity из владения — Source = TalentOwnersSync, DealId = null (INV-2 SourceConsistency), INV-4; InvestorUserId — из связи, OwnerMappingId = OwnerId, ExternalOwnershipId = payload.Id; снапшот и цены — по таблице payload (CurrencyId — из резолва валют шага 2, денормализованный CurrencyIsoCode — код резолвленной валюты). Фиксируется InvestorPropertyAdded. Если Status = Sold — сразу фиксируется продажа (переход [*] → Sold; агрегат вычисляет RealizedProfit = SalePrice.Amount − PurchasePrice.Amount, INV-5); фиксируется InvestorPropertySold.
    • Upsert (запись есть): поля покупки иммутабельны (INV-9) — расхождения PurchasePrice/PurchasedAt/PropertyObjectId с payload логируются warn + метрика investor_owner_sync_conflicts_total, но НЕ перезаписываются (исправление — admin-операция). Единственная применяемая дельта: Status = Sold в payload при Active-записи → фиксация продажи (guard-ы INV-3/INV-4/INV-5). Status = Active при проданной записи — no-op + warn (продажа не откатывается синхронизацией; разбор вручную). Полное совпадение — идемпотентный no-op (лог skipped-no-change).
  6. Согласованность и события. Запись портфеля, отметка идемпотентности и исходящие события фиксируются в одном согласованном изменении; offset — после фиксации. При создании — InvestorPropertyAddedKafkaEvent → топик pin.investor-owner.property-added.v1, key = id записи (контракт property-added.md); при фиксации продажи (в т.ч. при создании сразу в Sold) — дополнительно InvestorPropertySoldKafkaEvent → pin.investor-owner.property-sold.v1, key = id записи; payload обоих — по дому investor-property.md: денежные суммы — {Amount, CurrencyId} + CurrencySnapshot резолвленной валюты (шаг 2), снятый на момент публикации. Гонка двух инстансов по одному владению гасится unique ux_investor_properties_external_ownership_id (проигравший — перечитка + no-op).
  7. Результат. Как A7.

Диаграмма​

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

  • На уровне сообщения: по MessageId события (ключи investor-owner-talent-sync:{eventType}:{MessageId}) — повторная доставка после сбоя не мутирует состояние; отметка пишется только при успешном исходе (retriable-ошибка оставляет событие «необработанным»).
  • На бизнес-уровне: связи — version-guard INV-6 (payload.Version <= ReplicaVersion при Status != PendingLink → skip-stale); объекты портфеля — unique ExternalOwnershipId (INV-7, upsert) + иммутабельность покупки (INV-9) + невозвратность Sold.
  • Provision: вызов identity идемпотентен на его стороне — повторный retry вернёт уже созданного/найденного инвестора, дубль учётки невозможен.
  • Replay: да. Полный re-read обоих топиков с offset 0 безопасен (upsert + guard-ы); события Sold поверх Sold — no-op. Runbook — «replay DLQ синхронизации» (README investor-owner). Точечный replay DLQ — после ручного resolution (разбор конфликта связи, admin-архив дубля портфеля).

Retry и DLQ​

DLQ-топик группы: pin.investor-owner.talent-owners-sync-dlq.v1.

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)—Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)НетRunbook investor-owner-overview: восстановить зависимость
identity provision недоступен (5xx/timeout)INVESTOR_OWNER_SYNC_IDENTITY_UNAVAILABLE6 попыток, backoff 30s → 1m → 2m → 4m → 8m → 16m; связь сохранена в PendingLinkДа, после исчерпанияПроверить identity; replay из DLQ (линковка доработает PendingLink)
Связь не найдена (out-of-order ownership→owner)INVESTOR_OWNER_MAPPING_NOT_FOUNDта же политикаДаПроверить, пришёл ли OwnerUpsertedEvent; replay из DLQ
Связь не Linked (INV-8: PendingLink/Conflict)INVESTOR_OWNER_MAPPING_NOT_LINKEDта же политикаДаРазрешить конфликт (admin resolve-owner-mapping-conflict) → replay из DLQ
Справочник валют handbook недоступен при пустом кэше (резолв кода валюты)INVESTOR_OWNER_SYNC_HANDBOOK_UNAVAILABLEта же политикаДа, после исчерпанияПроверить handbook; replay из DLQ
Невалидный payload владения (цены/даты/валюты, INV-3/4/5)INVESTOR_OWNER_SYNC_INVALID_PAYLOADНет (non-retriable)Да + алертИсправление данных в Talent → повторное событие
Неизвестный enum Status владенияINVESTOR_OWNER_SYNC_UNKNOWN_STATUSНетДа + алертОбновить маппинг (релиз) → replay
Активный дубль по объекту недвижимости (INV-1, объект уже из сделки PIN)INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVEНетДа + алертСверка заботой: admin archive-investor-property лишней записи → replay
identity вернул InvestorUserId != OwnerId при Created = trueINVESTOR_OWNER_ID_MISMATCHНетДа + алертРазбор контракта provision с командой identity
Конфликты связи (дубль телефона / не-инвестор / коллизия id / невалидный номер)— (Status = Conflict в модели)Нет (не ошибка consumer-а)НетAdmin-очередь list-owner-mapping-conflicts → resolve-owner-mapping-conflict
Оптимистическая блокировка при гонке с admin-операцией—3 попытки немедленно (перечитать связь/запись)После исчерпанияРасследование гонки

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

  • Logs (structured, по событию): consumed / skipped-duplicate / skipped-stale / skipped-archived / skipped-no-change / mapping-created / mapping-linked / mapping-conflict / property-created / property-sold / dead-lettered с полями EventId, OwnerId/OwnershipId, payload.Version, mapping.ReplicaVersion, ConflictReason. PII маскируется: телефон +7***1234, ФИО И*** П***; суммы цен/дохода в логи открытым текстом не пишутся.
  • Metrics: investor_owner_sync_lag_seconds (now − max(LastSyncedAt), по топику), investor_owner_sync_consume_duration_ms, investor_owner_sync_stale_events_total, investor_owner_sync_retries_total, investor_owner_sync_dlq_total, investor_owner_sync_conflicts_total (+label reason), investor_owner_sync_contract_mismatch_total, investor_owner_sync_currency_fallback_total (пустой/неизвестный код валюты Talent → RUB), investor_owner_investors_provisioned_total (+label created|matched), investor_owner_properties_synced_total (+label created|sold|noop).
  • Traces: span приёма события → span обработки → span вызовов identity/chessboard → spans хранилища; correlation из метаданных события пробрасывается в исходящие pin.investor-owner.* события.
  • Alerts (README investor-owner): лаг синхронизации > 15 мин; DLQ > 0; рост Status = Conflict > 10/день; ошибки provision-API identity.
  • Dashboard / Runbooks: investor-owner-overview; runbooks «replay DLQ синхронизации», «разбор конфликтов связи», «компенсирующее исправление записи портфеля».
  • Безопасность: consumers работают под system-actor; вызовы identity/chessboard — m2m-аутентификация; доступ к топикам Talent — m2m ACL Kafka; секреты входа НЕ реплицируются (создание учётки — целиком ответственность identity); tenant isolation не применяется (данные инвестора вне tenant, ADR-0052) — изоляция record-level по InvestorUserId для user-чтений, consumer пишет с bypass; телефон/ФИО не попадают в исходящие pin.investor-owner.* события.

Известные ограничения и assumptions​

  • Имена топиков и состав payload Talent Owners — assumption до подтверждения schema registry Talent; расхождения правятся в этом документе без изменения домов сущностей.
  • Удаление владения Talent не транслируется событием (ownership-deleted в контракте нет) — ошибочные записи портфеля исправляются admin-операцией archive-investor-property.
  • Смена телефона собственника в Talent после установки связи обновляет только снапшот PhoneNumber и не пересматривает связь (INV-7); расхождения телефона инвестора PIN мониторит consumer identity-users.md.

Acceptance Criteria​

  • Given OwnerUpsertedEvent по новому собственнику с валидным телефоном, инвестора с которым в identity ещё нет, When consume, Then создаётся связь (PendingLink), вызывается provision-investor (RequiredUserId = OwnerId), при Created = true связь → Linked, LinkKind = InvestorCreated, InvestorUserId = OwnerId (INV-OWN-001), публикуется owner-linked без телефона в payload.
  • Given тот же поток, но инвестор с телефоном уже существует, When provision, Then Created = false, связь → Linked, LinkKind = ExistingInvestorMatched (расхождение InvestorUserId != OwnerId допустимо).
  • Given identity вернул не-инвестора / коллизию id / телефон невалиден / телефон занят другой связью, When consume, Then связь → Conflict(MatchedUserNotInvestor|IdentityIdCollision|PhoneInvalid| PhoneAlreadyMappedToAnotherOwner), доменный OwnerMappingConflictDetected, интеграционного owner-linked нет; событие ack (конфликт зафиксирован), разбор — admin-очередь; уже Linked-связь при этом не рушится.
  • Given identity недоступен (5xx/timeout), When линковка, Then связь остаётся PendingLink, INVESTOR_OWNER_SYNC_IDENTITY_UNAVAILABLE — retriable (6 попыток → DLQ), отметка идемпотентности НЕ пишется, version-guard не пропускает PendingLink — повтор дорабатывает линковку.
  • Given событие <= ReplicaVersion при Status != PendingLink, When consume, Then skipped-stale no-op (INV-6 version-guard), инкремент investor_owner_sync_stale_events_total.
  • Given OwnershipUpsertedEvent пришёл раньше своего OwnerUpsertedEvent (кросс-топиковый out-of-order), When consume, Then INVESTOR_OWNER_MAPPING_NOT_FOUND — retriable; после прихода/линковки owner replay из DLQ проходит.
  • Given связь есть, но не Linked (PendingLink/Conflict), When событие владения, Then INVESTOR_OWNER_MAPPING_NOT_LINKED (INV-8) — retriable; после resolution события переигрываются из DLQ.
  • Given валидное OwnershipUpsertedEvent при Linked-связи и отсутствии записи по ExternalOwnershipId, When consume, Then создаётся InvestorPropertyEntity (Source = TalentOwnersSync, OwnerMappingId = OwnerId), при Status = Sold — сразу продажа с RealizedProfit = SalePrice − PurchasePrice (INV-5); публикуются property-added (+property-sold).
  • Given PinPropertyObjectId = null, When создание объекта, Then PropertyObjectId = ObjectId, ProjectId/BuildingId/DeveloperId = null, карточка показывает только снапшот; Given chessboard недоступен при обогащении, Then объект всё равно создаётся из payload (best-effort, warn).
  • Given владение с пустым или неизвестным справочнику handbook кодом валюты, When consume, Then запись создаётся с валютой по умолчанию RUB (fallback), warn + инкремент investor_owner_sync_currency_fallback_total; событие не теряется и в DLQ не попадает. Given справочник валют недоступен при пустом кэше, Then INVESTOR_OWNER_SYNC_HANDBOOK_UNAVAILABLE — retriable (6 попыток → DLQ).
  • Given запись по ExternalOwnershipId уже есть, When повторное/обновляющее событие, Then идемпотентный upsert (INV-7): поля покупки иммутабельны (INV-9, расхождения только warn+метрика), применяется лишь дельта Active → Sold; полное совпадение — skipped-no-change.
  • Given объект уже присутствует активной записью из сделки PIN (дубль INV-1), When событие владения, Then INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVE — non-retriable → DLQ + алерт; авто-merge запрещён.
  • Given OwnerDeletedEvent, When consume, Then связь soft-delete (DeletedAt), записи портфеля НЕ трогаются (delete restrict, история живёт), downstream-событий нет; delete для неизвестного собственника — replay-безопасный no-op.
  • Given невалидный payload владения (цены/даты/валюты, INV-3/4/5) или неизвестный Status, When consume, Then INVESTOR_OWNER_SYNC_INVALID_PAYLOAD / INVESTOR_OWNER_SYNC_UNKNOWN_STATUS — non-retriable → DLQ + алерт.

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

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

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