Consumer: Referral Bond Sync (pin-rewards-referral-bonds)

Синхронизация реферальных связей из referral в контекст rewards. Одна consumer group обрабатывает оба события топика pin.referral.bonds.v1 — создание связи (bond-created) и закрытие (bond-closed); оба описаны в этом документе, чтобы политика retry/DLQ/replay группы была единой. Ссылки из referral-accrual.md и ../README.md на consumer реплики связей разрешаются в этот документ.

Назначение

Consumer group pin-rewards-referral-bonds потребляет события реферальных связей referral и поддерживает в rewards локальную реплику связей (rewards.referral_bond_replicas). Реплика — основание реферальных начислений: при завершении сделки consumer crm-deal-completed.md ищет по ней активную связь приведённого инвестора и создаёт ReferralAccrualEntity — без синхронного запроса в referral (решение № 7 referral: referral владеет фактом связи, rewards — деньгами).

Закрытие связи (bond-closed, Replaced/Revoked) помечает реплику неактивной — новые начисления по связи прекращаются; уже созданные начисления остаются (регламент referral-accrual.md: отзыв денег по фроду — только явной ручной отменой BondRevoked оператором ПИН, автоматической отмены по закрытию связи нет). Этот регламент — та самая «судьба ранее начисленного», на которую ссылается контракт ../../referral/events/bond-closed.md.

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

Поле	Значение
Сервис/контекст	rewards (фоновый обработчик)
Consumer group	pin-rewards-referral-bonds
Producer	referral (агрегат referral-bond.md; правила привязки R1/R2/R6, admin-rebind/revoke)
Топик	pin.referral.bonds.v1 — общий для ReferralBondCreatedKafkaEvent (bond-created) и ReferralBondClosedKafkaEvent (bond-closed)
Partitions / key	6 partitions / key = ReferredUserId — строгий порядок created → closed per-приглашённый (при rebind closed старой связи приходит раньше created новой)
ContractVersion	1 у обоих событий; несовпадение версии — warn-and-proceed (метрика rewards_contract_mismatch_total)

Событие	Эффект (событие → что меняется)
pin.referral.bond-created	создание строки реплики rewards.referral_bond_replicas (Status = Active) — связь становится основанием будущих начислений
pin.referral.bond-closed	пометка строки закрытой (Status = Replaced/Revoked) — новые начисления по BondId прекращаются; созданные начисления не мутируются

Целевая реплика (owned projection; дом полей — этот документ)

ReferralBondReplicaEntity (rewards.referral_bond_replicas) — плоская строка (не агрегат): без доменных событий и аудита (изменения вносит только system-actor синка). Физическое удаление не выполняется — история оснований начислений остаётся проверяемой.

Поле	Тип	Обязательность	Ограничения (PK/FK/index/constraint)	Источник (payload)	Описание
BondId	Guid	Да	PK pk_referral_bond_replicas	BondId события (= referral.referral_bonds.id)	Идентификатор связи — тот же, что в referral; на него ссылается ReferralAccrualEntity.BondId.
ReferredUserId	Guid	Да	index ix_referral_bond_replicas_referred (referred_user_id); partial-unique ux_referral_bond_replicas_referred_active (referred_user_id) where status = 'Active' — не более одной активной связи на приведённого (зеркало INV-REF-001 источника)	ReferredUserId	Приведённый пользователь; ключ поиска активной связи по InvestorUserId сделки.
ReferrerUserId	Guid	Да	—	ReferrerUserId	Реферер-партнёр — будущий получатель вознаграждения (ReferralAccrualEntity.PartnerUserId); участник guard-а самореферала (INV-RA-4).
ReferrerCompanyId	Guid?	Нет	—	ReferrerCompanyId	Компания реферера на момент привязки; null — реферер-инвестор (ADR-0052). Справочно (аналитика/отчёты).
Status	ReferralBondStatus	Да	хранится как text, varchar(32): Active / Replaced / Revoked (дом enum — referral-bond.md)	bond-created → Active; bond-closed → Status payload	Резолюция основания начисления смотрит только Status = Active — любое иное значение (в т.ч. неизвестное новое) fail-closed останавливает новые начисления.
BoundAt	DateTimeOffset	Да	—	BoundAt	Момент привязки (справочно; разбор споров «связь была активна на момент сделки»).
ReplicaVersion	long	Да	>= 1	Version	Anti-out-of-order guard per-BondId (монотонный, INV-BND-1).
SyncedAt	DateTimeOffset	Да	—	время события	Момент последнего применённого синка (метрика лага).

Поля ReferralLinkId, Channel, AppliedRule событий consumer-ом не читаются (канал/правило привязки на деньги не влияют); ReplacedByBondId/RevokeReason из bond-closed не сохраняются — цепочка коррекций живёт в referral, в логи пишется только BondId замены.

Инвариант	Условие	Где проверяется	Поведение
INV-BND-1 MonotonicReplica	Событие применяется только при payload.Version > row.ReplicaVersion	handler (шаг 4)	Устаревшее/дубль — идемпотентный skip (skipped-stale)
INV-BND-2 NoHardDelete	Строки не удаляются; закрытие — только смена Status	handler	История оснований начислений проверяема
INV-BND-3 SingleActiveReferred	Не более одной строки Active на ReferredUserId — гарантировано producer-ом (INV-REF-001) + порядком per-key (closed старой связи раньше created новой при rebind); конфликт partial-unique индекса — аномалия данных	unique index при записи	REWARDS_BOND_REPLICA_ACTIVE_CONFLICT → DLQ + алерт (fail closed)
INV-BND-4 ClosedIsTerminal	Строка в Replaced/Revoked не возвращается в Active (терминальность INV-REF-B6 источника); «восстановление» связи в referral — это новая связь с новым BondId (отдельное bond-created)	handler (guard перехода)	Событие, нарушающее терминальность, → DLQ REWARDS_BOND_EVENT_INVALID

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

• Из метаданных события доступны: идентификатор сообщения, время события, Version, агрегат-источник (AggregateId = BondId, AggregateType = "ReferralBond").
• Обработка — под system-actor синка вне tenant-контекста (platform-scope): реплика — платформенное состояние rewards (реферальный процесс ведёт ПИН, ADR-0052).
• Корреляция сквозного трейса восстанавливается из метаданных события; исходящих событий нет.
• Ключи идемпотентности сообщений — rewards:bond-created:{MessageId} / rewards:bond-closed:{MessageId}; бизнес-идемпотентность — version-guard INV-BND-1 per-BondId.

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

Контракты — bond-created.md, bond-closed.md (дом — контекст referral). Ниже — только поля, которые consumer реально читает.

ReferralBondCreatedKafkaEvent

Поле payload	Тип	Обязательность	Как используется (маппинг)
BondId	Guid	Да	PK строки → ReferralBondReplicaEntity.BondId
ReferredUserId	Guid	Да	→ ReferredUserId (ключ поиска активной связи)
ReferrerUserId	Guid	Да	→ ReferrerUserId (получатель будущих начислений)
ReferrerCompanyId	Guid?	Нет	→ ReferrerCompanyId
BoundAt	DateTimeOffset	Да	→ BoundAt
Version	long	Да	Монотонный guard INV-BND-1 → ReplicaVersion

Гарантия producer-а: ReferredUserId != ReferrerUserId (INV-REF-B2) — отдельная проверка самопривязки consumer-у не нужна; guard самореферала «реферер = агент сделки» (INV-RA-4) выполняется позже, при создании начисления (crm-deal-completed.md).

ReferralBondClosedKafkaEvent

Поле payload	Тип	Обязательность	Как используется (маппинг)
BondId	Guid	Да	Ключ строки реплики
ReferredUserId	Guid	Да	Сверка со строкой (fail-closed при несоответствии)
Status	string (enum ReferralBondStatus: Replaced / Revoked)	Да	→ Status; unknown-значение — fail-closed (см. шаг 4)
ReplacedByBondId	Guid?	При Replaced	Только лог (цепочка коррекций — состояние referral)
Version	long	Да	Монотонный guard INV-BND-1 → ReplicaVersion

Алгоритм

1. Контекст события. Восстановить метаданные события; сверить ContractVersion — при несовпадении warn-and-proceed. Проверить идемпотентность по ключу rewards:bond-created:{MessageId} (для закрытия — rewards:bond-closed:{MessageId}) — дубль → skipped-duplicate, пропуск (ack).
2. Проверка входа. BondId != Guid.Empty; ReferredUserId != Guid.Empty; для bond-created — ReferrerUserId != Guid.Empty, BoundAt задан; Version >= 1. Нарушение — REWARDS_BOND_EVENT_INVALID (non-retriable → DLQ: контракт producer-а нарушен).
3. Что читаем. Существующую строку реплики по BondId; отсутствие: для bond-created допустимо (первое событие связи), для bond-closed — аномалия порядка (см. шаг 4).
4. Бизнес-правила и переходы.
   • Version-guard (INV-BND-1): для существующей строки payload.Version <= ReplicaVersion — устаревшее/дубль: метрика rewards_bond_replica_stale_events_total, лог skipped-stale, только отметка идемпотентности, пропуск.
   • bond-created: создать строку (Status = Active, маппинг по таблице payload, ReplicaVersion = Version, SyncedAt = время события). Повтор с тем же BondId и большей версией — upsert-перезапись полей (immutable-поля источника не меняются — расхождение только логируется warn). Конфликт ux_referral_bond_replicas_referred_active (у приглашённого уже есть другая активная связь) — аномалия producer-а → REWARDS_BOND_REPLICA_ACTIVE_CONFLICT (non-retriable → DLQ + алерт, fail closed: INV-BND-3).
   • bond-closed, строка есть: guard INV-BND-4 — строка уже терминальна (Replaced/Revoked): идемпотентный skip skipped-already-closed (не более одного closed на BondId — гарантия контракта; повтор = redelivery). Иначе Status = payload.Status, ReplicaVersion = Version, SyncedAt = время события. Unknown Status — warning unknown-bond-status + значение сохраняется как есть (хранение text): резолюция основания смотрит только Status = Active, поэтому неизвестный статус закрытия fail-closed останавливает новые начисления (правило совместимости контракта bond-closed).
   • bond-closed, строки нет: порядок per-key (created → closed) нарушен быть не может — отсутствие строки означает пропуск created (потерянная запись/сбой до фиксации). 3 немедленных повтора (перечитать строку — вдруг параллельная партиция первоначального наполнения ещё пишет), затем REWARDS_BOND_EVENT_ORPHAN_CLOSED → DLQ + алерт; ручное действие — replay топика с offset 0 (восстановит и created, и closed).
   • Созданные начисления не трогаются: закрытие связи не мутирует ReferralAccrualEntity — ни статусы, ни суммы (отмена по фроду — ручная, CancelReason = BondRevoked, platform-API rewards).
5. Что меняется. Строка реплики создаётся или обновляется. Конкурентность разрешается version-guard INV-BND-1 (per-key порядок Kafka исключает гонку внутри партиции).
6. Согласованность. Строка реплики и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается после фиксации. Исходящих событий нет: реплика — приватное состояние rewards.
7. Результат. Успех → ack; non-retriable коды → DLQ + алерт; инфраструктурные сбои → retry (таблица ниже).

Окно согласованности с начислениями

Между закрытием связи в referral и применением bond-closed в реплике (SLA < 5 мин, bond-closed.md) consumer завершения сделки может успеть создать начисление по ещё-активной строке. Это ожидаемое поведение (окно зафиксировано контрактом bond-closed): созданное начисление остаётся и при необходимости отзывается вручную (BondRevoked). Обратного окна нет: пока created не применён, начисление по связи не создаётся (сделка без активной связи в реплике — штатный skip consumer-а завершения), восстановление — replay deal-completed после догона синка не требуется, т.к. отсутствие рефералки диагностируется сверкой (см. Наблюдаемость).

Диаграмма

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

• Consumer-level: отметка по идентификатору сообщения.
• Business-level: version-guard INV-BND-1 per-BondId — событие с Version <= ReplicaVersion пропускается (защита от redelivery и replay); терминальность закрытия (INV-BND-4) делает повтор bond-closed no-op.
• Replay: да. Полный re-read топика с offset 0 безопасен (upsert + version-guard) и используется как первоначальное наполнение реплики при внедрении rewards (отдельного import-job нет, ADR-0056) и как восстановление после REWARDS_BOND_EVENT_ORPHAN_CLOSED. Начисления при replay не создаются и не меняются — реплика от них отвязана.

Retry и DLQ

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

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)	— (exception)	Бесконечный exponential backoff 1s→60s с паузой партиции	Нет	Runbook rewards-overview: восстановить зависимость
Нарушение контракта payload (пустые BondId/ReferredUserId/ReferrerUserId, Version < 1, нарушение терминальности INV-BND-4)	REWARDS_BOND_EVENT_INVALID	Нет (non-retriable)	Да + алерт	Разбор с командой referral; после фикса producer-а — replay из DLQ
Вторая активная связь одного приглашённого (конфликт partial-unique, INV-BND-3)	REWARDS_BOND_REPLICA_ACTIVE_CONFLICT	Нет	Да + алерт	Разбор данных referral (нарушение INV-REF-001 / порядка публикации)
bond-closed без строки реплики	REWARDS_BOND_EVENT_ORPHAN_CLOSED	3 немедленных повтора	После исчерпания + алерт	Replay топика с offset 0
Устаревшее событие (Version <= ReplicaVersion) / повтор closed терминальной строки	— (не ошибка)	—	Нет	Не требуется (идемпотентный skip)
Unknown Status в bond-closed	— (warning)	—	Нет	Мониторинг rewards_bond_unknown_status_total; значение fail-closed (не Active)

Acceptance-критерии

• Создание реплики (happy path). Дано: первое bond-created связи. Когда: обработка. Тогда: создана строка rewards.referral_bond_replicas (Status = Active, ReplicaVersion = Version, SyncedAt = время события); ack. Связь доступна резолюции активного бонда в crm-deal-completed.md.
• Закрытие связи. Дано: строка Active; приходит bond-closed (Status = Revoked, Version > ReplicaVersion). Когда: обработка. Тогда: Status = Revoked; новые начисления по BondId не создаются; уже созданные ReferralAccrualEntity не изменены (ни статус, ни сумма).
• Rebind (пара closed + created). Дано: admin-rebind в referral — closed старой связи и created новой делят key ReferredUserId. Когда: обработка по порядку. Тогда: старая строка → Replaced, новая создана Active; в любой момент не более одной Active-строки приглашённого (INV-BND-3).
• Устаревшее/дубль (version-guard). Дано: Version <= ReplicaVersion. Когда: обработка. Тогда: skipped-stale, состояние не меняется, ack (INV-BND-1).
• Повтор закрытия. Дано: строка уже Replaced/Revoked, redelivery bond-closed. Когда: обработка. Тогда: skipped-already-closed, ack (INV-BND-4).
• Закрытие без строки. Дано: bond-closed по BondId, которого нет в реплике. Когда: обработка. Тогда: 3 немедленных повтора → REWARDS_BOND_EVENT_ORPHAN_CLOSED → DLQ + алерт; после replay топика реплика сходится.
• Unknown Status. Дано: bond-closed с неизвестным значением Status. Когда: обработка. Тогда: warning unknown-bond-status, значение сохранено, строка не Active — новые начисления остановлены (fail-closed), ack.
• Нарушение контракта. Дано: пустой BondId/ReferredUserId. Когда: обработка. Тогда: REWARDS_BOND_EVENT_INVALID → DLQ + алерт.
• Replay/наполнение. Дано: re-read топика с offset 0. Когда: обработка. Тогда: реплика сходится к последнему состоянию каждой связи (upsert + version-guard); начисления не задваиваются (реплика их не создаёт).

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

• Logs (structured): consumed / skipped-duplicate / skipped-stale / skipped-already-closed / created / closed / dead-lettered с полями MessageId, BondId, ReferredUserId, payload.Version, row.ReplicaVersion, Status. ФИО/телефонов в событии нет (PII-минимизация контракта); RevokeReason в реплике не хранится и в лог не пишется.
• Metrics: rewards_referral_bonds_lag_seconds (now − max(SyncedAt)), rewards_referral_bonds_consume_duration_ms, rewards_bond_replica_stale_events_total, rewards_bond_unknown_status_total, rewards_referral_bonds_dlq_total, rewards_contract_mismatch_total.
• Traces: consume-span → handler-span → DB-span; корреляция из метаданных события.
• Alerts: lag > 5 мин (SLA реплики — контракт bond-closed); DLQ > 0.
• Сверка: периодическое сравнение числа активных связей в реплике с referral (расхождение → replay); сделки приведённых инвесторов без реферального начисления при активной связи диагностируются метрикой rewards_referral_accrual_no_rate_total и сверкой consumer-а завершения сделки.
• Безопасность: system-actor, platform-scope; события — доверенный внутренний топик referral (m2m ACL Kafka); untrusted-ввода нет; чтение реплики — только внутренними операциями rewards (резолюция основания начисления).

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

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

• Домен (CQRS) (1): Entity / Aggregate: ReferralBondEntity
• Use Cases (2): UC-RWD-003. Сделка завершена — начисление комиссии (system), UC-RWD-009. Жизненный цикл реферального начисления (связь → начисление → гейт → выплата)
• События (2): Event Contract: pin.referral.bond-closed (ReferralBondClosedKafkaEvent), Event Contract: pin.referral.bond-created (ReferralBondCreatedKafkaEvent)
• Консюмеры (1): Consumer: CRM Deal Completed
• Саги (1): Saga «Commission Settlement» (финансовый контур вознаграждений)
• index.md (2): Referral Service, Rewards Service

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

• Потребители реплики: referral-accrual.md (поиск активной связи, создание начисления), crm-deal-completed.md (ветка реферального начисления).
• Контракты событий: bond-created.md, bond-closed.md; агрегат-источник — referral-bond.md (enum ReferralBondStatus, INV-REF-001/B2/B6).
• Регламент «созданные начисления остаются»: referral-accrual.md (CancelReason = BondRevoked — ручная отмена оператором ПИН).
• Шаблон: consumer-template.md; ADR-0052 (tenancy), ADR-0056 (simplicity-first — реплика в PostgreSQL), ADR-0066 (referral — факт связи, rewards — деньги).