Consumer: завершённая сделка CRM → объект в портфеле инвестора (pin-investor-owner-crm-deal-completed)

Consumer контекста investor-owner: после завершения сделки в CRM купленный объект недвижимости должен появиться у инвестора-покупателя в его портфеле. Дом полей целевой сущности — investor-property.md; здесь — маппинг payload → поля и алгоритм.

Назначение

Consumer обрабатывает событие pin.crm.deal-completed и создаёт запись портфеля InvestorPropertyEntity (Source = PinCrmDeal, Status = Active): купленный объект появляется на странице «Мои объекты недвижимости» (/profile/my-lots) у инвестора-покупателя. Идемпотентность — по DealId (INV-6 DealIdempotency): повторная доставка события той же сделки не создаёт дубль. О появлении объекта публикуется pin.investor-owner.property-added.v1 (контракт) — для notifications и user-activity.

Сделки продажи объекта инвестора в текущем объёме CRM не ведёт — переход Active → Sold наполняется синхронизацией Talent Owners (talent-owners-sync.md) или admin-операцией (ограничение README investor-owner).

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

Поле	Значение
Сервис/контекст	investor-owner
Обрабатываемое событие	DealCompletedKafkaEvent (event type pin.crm.deal-completed)
Topic	pin.crm.deal-completed.v1; key = DealId (строгий порядок per-deal)
ContractVersion	1; при несовпадении версии — warn-and-proceed (метрика investor_owner_deal_contract_mismatch_total)
Consumer group	pin-investor-owner-crm-deal-completed
Producer	crm (контракт — будущий путь ../../crm/events/deal-completed.md; требование к составу payload зафиксировано в README investor-owner: событие несёт снапшот объекта и итоговую цену сделки)
Эффект (событие → что меняется)	Создание записи портфеля InvestorPropertyEntity у инвестора-покупателя
SLA	появление объекта в портфеле < 1 мин после deal-completed (README investor-owner)
DLQ	pin.investor-owner.crm-deal-completed-dlq.v1 (исходное событие + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset)

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

• Из метаданных события доступны: идентификатор сообщения MessageId (Guid v7), время события CreatedAt, версия контракта, AggregateId = DealId, AggregateType = Deal.
• Actor — system-actor consumer-а: создаваемая запись без пользователя-создателя (CreatedByUserId = null, правило дома investor-property.md). Сущность не tenanted (ADR-0052); consumer пишет записи любых инвесторов (record-level фильтр по InvestorUserId для system-actor выключен).
• Корреляция сквозного трейса восстанавливается из метаданных события и пробрасывается в исходящее property-added.
• Идемпотентность двухуровневая: на уровне сообщения — по MessageId (ключ investor-owner-crm-deal:deal-completed:{MessageId}); бизнес-идемпотентность — по DealId (unique ux_investor_properties_deal_id, INV-6). MessageId ловит повторную доставку того же сообщения, DealId — повторную публикацию producer-ом.

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

Контракт события: pin.crm.deal-completed (будущий путь — контекст crm документируется отдельно; состав ниже — зафиксированное требование investor-owner к контракту, README investor-owner «Известные ограничения»). Ниже — поля payload, которые consumer реально читает, и маппинг в поля дома investor-property.md.

Поле payload	Тип	Обязательность	Как используется (маппинг)
DealId	Guid	Да	→ InvestorPropertyEntity.DealId; ключ бизнес-идемпотентности (INV-6)
InvestorUserId	Guid	Да	→ InvestorUserId (владелец записи; id пользователя-инвестора identity, зафиксирован CRM при сделке)
PropertyObjectId	Guid	Да	→ PropertyObjectId (объект недвижимости chessboard; ключ истории цен invest-analytics и проверки INV-1)
ProjectId	Guid?	Нет	→ ProjectId (денормализация для группировки портфеля)
BuildingId	Guid?	Нет	→ BuildingId
DeveloperId	Guid?	Нет	→ DeveloperId
FinalPrice.Amount	decimal	Да	→ PurchasePrice.Amount — итоговая цена сделки, «за сколько купили» (> 0, INV-4)
FinalPrice.CurrencyId	short	Да	→ PurchasePrice.CurrencyId — типизированная ссылка на справочник валют handbook; незаполненный (<= 0) → INVESTOR_OWNER_DEAL_INVALID_PAYLOAD
Currency	CurrencySnapshot	Да	Display-снапшот валюты события (snapshots.md): Currency.IsoCode → денормализованный CurrencyIsoCode записи; переиспользуется в исходящем property-added (снапшот на момент публикации совпадает — публикация атомарна с записью)
CompletedAt	DateTimeOffset	Да	→ PurchasedAt — момент завершения сделки (<= now() + 5 мин clock-skew allowance)
PropertyObject.Title	string	Да	→ PropertyObjectSnapshot.Title (≤ 256, при превышении truncate + warn)
PropertyObject.Number	string?	Нет	→ PropertyObjectSnapshot.Number
PropertyObject.ProjectName	string?	Нет	→ PropertyObjectSnapshot.Project (ref-модель {ProjectId, ProjectName}; ProjectId — из payload.ProjectId; null, если проект не задан)
PropertyObject.BuildingName	string?	Нет	→ PropertyObjectSnapshot.Building (ref-модель {BuildingId, BuildingName}; BuildingId — из payload.BuildingId; null, если корпус не задан)
PropertyObject.AreaTotal	decimal?	Нет	→ PropertyObjectSnapshot.AreaTotal (> 0, иначе null + warn)
PropertyObject.Floor	int?	Нет	→ PropertyObjectSnapshot.Floor (>= -5, <= 200, иначе null + warn)
Version	long	Да	Версия агрегата сделки — только логирование/диагностика (сделка завершается один раз; version-guard на записи портфеля не нужен)

PropertyObjectSnapshot.SnapshotAt = время события (момент фиксации «что купили»). Поля payload, не перечисленные выше (участники сделки, комиссии, этапы), consumer-ом не читаются — комиссии считает rewards, сделку ведёт crm (границы README investor-owner).

Алгоритм

1. Контекст события и идемпотентность. Восстановить метаданные события (MessageId, CreatedAt, AggregateId = DealId); сверить версию контракта с 1 — при несовпадении warn-and-proceed (метрика investor_owner_deal_contract_mismatch_total). Если сообщение с таким MessageId уже обработано → дубль: лог skipped-duplicate, ack (no-op).
2. Проверка входа. Убедиться, что заполнены DealId, InvestorUserId, PropertyObjectId; итоговая цена FinalPrice.Amount > 0 (INV-4 PositivePrices) и FinalPrice.CurrencyId > 0 при заполненном снапшоте Currency; CompletedAt <= now() + 5 мин; заголовок объекта непустой (при превышении 256 — truncate + warn, событие не теряется). При нарушении → отказ INVESTOR_OWNER_DEAL_INVALID_PAYLOAD (non-retriable → DLQ + алерт команде crm: contract violation не чинится ретраем). Существование InvestorUserId в identity и PropertyObjectId в chessboard синхронно не проверяется (simplicity-first, ADR-0056): producer crm гарантирует существование инвестора и объекта, снапшот объекта приходит в payload, ссылки — без FK (чужой контекст). FinalPrice.CurrencyId в справочнике handbook синхронно не резолвится по той же причине: crm валидировал валюту на своём write-path, событие несёт готовый CurrencySnapshot.
3. Бизнес-идемпотентность по сделке (INV-6). Проверить, есть ли уже запись портфеля с этим DealId (включая архивированную заботой — компенсирующая правка не пересоздаётся). Если есть → идемпотентный no-op: лог skipped-duplicate-deal, метрика investor_owner_deal_duplicates_total, фиксируется только отметка идемпотентности, ack.
4. Проверка дубля между источниками (INV-1 SingleActivePerLot). Проверить, нет ли уже активной записи по паре (InvestorUserId, PropertyObjectId) — объект мог прийти раньше синхронизацией Talent Owners с другим ключом идемпотентности. Если активный дубль есть → отказ INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVE (non-retriable → DLQ + алерт: ручная сверка Службой заботы — admin archive-investor-property лишней записи → replay из DLQ). Автоматический merge запрещён — финансовые данные двух источников могут расходиться.
5. Создание записи портфеля. Создать InvestorPropertyEntity из сделки: Source = PinCrmDeal, Status = Active (переход [*] → Active), INV-2 SourceConsistency (DealId != null, OwnerMappingId и ExternalOwnershipId = null), INV-4; SalePrice/SoldAt/RealizedProfit = null (объект куплен, не продан). Маппинг поле-в-поле — таблица payload выше; PurchasePrice = FinalPrice (Amount + CurrencyId), денормализованный CurrencyIsoCode = Currency.IsoCode (display-код, снятый при записи), PurchasedAt = CompletedAt, снапшот объекта — из PropertyObject.*; поля аудита: время создания — now(), создатель — system (null), Version = 1. Фиксируется доменный факт InvestorPropertyAdded.
6. Событие о появлении объекта. Публикуется InvestorPropertyAddedKafkaEvent → топик pin.investor-owner.property-added.v1, key = id записи; payload собирается из созданной записи (Id, InvestorUserId, PropertyObjectId, ProjectId, Source = PinCrmDeal, PurchasePrice {Amount, CurrencyId}, Currency — CurrencySnapshot из входного события, PurchasedAt, Version) — контракт property-added.md.
7. Согласованность. Запись портфеля, отметка идемпотентности и исходящее событие фиксируются в одном согласованном изменении; offset подтверждается только после успешной фиксации. Гонка двух инстансов consumer-а по одной сделке гасится unique ux_investor_properties_deal_id (проигравший перечитывает по DealId → идемпотентный no-op).
8. Результат. Успех (включая идемпотентные no-op ветки) → ack. Отказ non-retriable (INVESTOR_OWNER_DEAL_INVALID_PAYLOAD, INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVE) → DLQ + алерт; инфраструктурные сбои → retry (таблица ниже).

Диаграмма

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

• Business key: DealId — «одна завершённая сделка = максимум одна запись портфеля»; поведение дубля: skip (запись не обновляется — сделка завершается один раз, поля покупки иммутабельны, INV-9). Двухуровневая защита: MessageId (отметка идемпотентности) + DealId (проверка + unique-индекс).
• Архивированная заботой запись по DealId дубль тоже гасит: архив — компенсирующее решение, синхронизация его не пересоздаёт (пересоздание — только admin create-investor-property).
• Replay: да. Полный re-read топика с offset 0 безопасен (все сделки уже применены → skip). Точечный replay DLQ — после ручного разбора INV-1 (архив лишней записи) или исправления контракта; runbook — «replay DLQ» investor-owner-overview.

Retry и DLQ

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)	—	Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)	Нет	Runbook investor-owner-overview: восстановить зависимость
Невалидный payload (пустые id, цена ≤ 0, незаполненный FinalPrice.CurrencyId/Currency-снапшот, нет снапшота объекта)	INVESTOR_OWNER_DEAL_INVALID_PAYLOAD	Нет (non-retriable — contract violation)	Да + алерт команде crm	Исправить producer/данные сделки → повторная публикация или replay
Активный дубль по объекту недвижимости (INV-1 — объект уже из синхронизации Talent)	INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVE	Нет	Да + алерт	Сверка заботой: admin archive-investor-property → replay из DLQ
Гонка по unique ux_investor_properties_deal_id	—	1 немедленная попытка (перечитка по DealId → no-op)	Нет	Не требуется
Оптимистическая блокировка при гонке с admin-операцией	—	3 попытки немедленно (перечитать запись)	После исчерпания	Расследование гонки

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

• Logs (structured): consumed / skipped-duplicate / skipped-duplicate-deal / created / dead-lettered с полями EventId, DealId, InvestorUserId, PropertyObjectId, payload.Version. Суммы (FinalPrice) в логи открытым текстом не пишутся (финансовые данные — правило дома investor-property.md); PII в payload отсутствует.
• Metrics: investor_owner_deal_lag_seconds (now − время события на consume), investor_owner_deal_consume_duration_ms, investor_owner_deal_properties_created_total, investor_owner_deal_duplicates_total, investor_owner_deal_dlq_total, investor_owner_deal_contract_mismatch_total.
• Traces: span приёма события → span обработки → spans обращений к хранилищу; correlation из метаданных события пробрасывается в исходящее property-added.
• Alerts: лаг > 15 мин (SLA появления объекта — 1 мин, алерт с запасом на burst); DLQ > 0.
• Dashboard / Runbook: investor-owner-overview («созданные объекты в день», лаг, DLQ); runbooks «replay DLQ», «компенсирующее исправление записи портфеля».
• Безопасность: consumer работает под system-actor; доступ к топику pin.crm.deal-completed.v1 — m2m ACL Kafka (investor-owner, rewards, notifications — по контракту crm); tenant isolation не применяется (данные инвестора вне tenant, ADR-0052), запись — с bypass record-level фильтра; входной payload — trusted boundary внутри платформы PIN (producer crm аутентифицирован ACL), но валидируется полностью (шаг 2 — fail closed в DLQ).

Acceptance Criteria

• Given валидное deal-completed по новой сделке, When consumer обрабатывает событие, Then создаётся InvestorPropertyEntity (Source = PinCrmDeal, Status = Active, PurchasePrice = FinalPrice с CurrencyId события, CurrencyIsoCode = Currency.IsoCode снапшота, PurchasedAt = CompletedAt, снапшот из PropertyObject.*), публикуется property-added (payload — CurrencyId + CurrencySnapshot) в том же согласованном изменении, offset подтверждается после фиксации.
• Given повторная доставка того же сообщения (MessageId) или повторная публикация той же сделки (DealId), When consume, Then идемпотентный no-op (skipped-duplicate / skipped-duplicate-deal), дубль не создаётся (INV-6 + unique ux_investor_properties_deal_id), ack.
• Given архивированная заботой запись по этому DealId, When повторное событие, Then запись НЕ пересоздаётся (компенсирующая правка не затирается) — no-op, ack.
• Given два инстанса consumer-а обрабатывают одну сделку одновременно, When оба доходят до записи, Then один создаёт запись, второй проигрывает по unique-индексу и перечитывает по DealId → no-op.
• Given невалидный payload (пустые id, FinalPrice.Amount <= 0, незаполненный FinalPrice.CurrencyId, CompletedAt в будущем сверх допуска), When consume, Then INVESTOR_OWNER_DEAL_INVALID_PAYLOAD — non-retriable → DLQ + алерт команде crm, портфель не мутирует.
• Given объект уже присутствует активной записью из синхронизации Talent Owners (дубль INV-1), When событие сделки, Then INVESTOR_OWNER_PROPERTY_ALREADY_ACTIVE — non-retriable → DLQ + алерт; авто-merge запрещён (финансовые данные могут расходиться); разбор — admin archive-investor-property → replay.
• Given недоступность PostgreSQL/Kafka, When consume, Then бесконечный backoff (событие не теряется), запись создаётся после восстановления; полный re-read топика с offset 0 безопасен (все сделки уже применены → skip).

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

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

• Домен (CQRS) (1): Entity / Aggregate: InvestorPropertyEntity
• API (1): POST /api/v1/admin/portfolio/properties/{propertyId}/archive — Архивирование записи портфеля (admin, archiveInvestorProperty)
• Use Cases (1): UC-OWN-002. Завершение сделки в CRM → купленный объект в «Мои объекты недвижимости» (system)
• События (2): Event Contract: pin.crm.deal-completed (DealCompletedEvent), Event Contract: pin.investor-owner.property-added (InvestorPropertyAddedKafkaEvent)
• Консюмеры (1): Consumer: синхронизация из Talent Owners (pin-investor-owner-talent-owners-sync)
• Саги (1): Saga «Deal Completion» (fan-out завершения сделки)
• scenarios (1): UC-OVR-004. От реферальной ссылки до объекта у собственника
• index.md (1): Сервис Инвестор-Собственник (investor-owner)

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

• Дом сущности: investor-property.md (поля, INV-1/2/4/6/9, переходы состояний).
• Контракт входного события: ../../crm/events/deal-completed.md (будущий путь, контекст crm; требование к составу payload — README investor-owner).
• Производимый контракт: property-added.md; snapshot-модели справочника (CurrencySnapshot) — ../../handbook/domain/snapshots.md.
• Смежный consumer: talent-owners-sync.md (второй источник портфеля; правило разбора дублей INV-1 — общее).
• Решения: ADR-0052 (инвестор вне tenant), ADR-0056 (simplicity-first: без синхронных проверок identity/ chessboard — доверие контракту + fail closed валидация), prototype-mapping.md (/profile/my-lots).