Consumer: Rewards Commission Enrichment (pin-catalog-rewards-commission)

Kafka-consumer агентских комиссий из контекста rewards: денормализует действующий набор правил вознаграждения застройщика в owned VO PropertyObjectAgentCommission витринных карточек объектов недвижимости. Этот документ — дом алгоритма матчинга (раздел «Матчинг»): на него ссылаются chessboard-projection-sync.md (шаг D4) и invest-analytics-enrichment.md.

Назначение

Consumer обрабатывает событие CommissionRulesChanged (pin.rewards.commission-rules.v1) и пересчитывает AgentCommission всех затронутых объектов недвижимости застройщика (по DeveloperCompanyId + матчинг). Событие несёт per-developer снапшот всего действующего набора правил комиссий одного застройщика; для каждого объекта недвижимости правила сопоставляются по DeveloperCompanyId и ключам матчинга (PropertyType/ProjectId/LayoutId/LayoutGroupId/RoomsCount/ AreaTotal — условия правил) и вычисляется PropertyObjectAgentCommission (PercentMin/PercentMax/BaseKind/RewardRuleId/RuleVersion). PercentMin = PercentMax при одном подошедшем правиле; диапазон — при нескольких (точный процент зависит от программы сделки); null (весь VO) — ни одно активное правило не подошло, блок комиссии на фронте скрыт (поле отдаётся только по permission catalog.commissions.view).

Границы consumer'а

Собственных интеграционных событий consumer не публикует (витрина — конечная точка). Комиссия не влияет на видимость (INV-1) и агрегаты проекта (MinPrice/PropertyObjectsAvailable/Payback).

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

Поле	Значение
Статус / owner	draft / PIN Marketplace
Сервис/контекст	catalog
Событие	CommissionRulesChanged
Topic / key	pin.rewards.commission-rules.v1, key = DeveloperCompanyId (per-developer порядок)
ContractVersion	= 1; при несовпадении версии события — warn-and-proceed (catalog_commission_contract_mismatch_total)
Consumer group	pin-catalog-rewards-commission
Producer	rewards — owner-контракт commission-rules.md
Feature flag	catalog.commissions.enabled — гейт обработки и выдачи поля API

Событие → эффект

Событие	Эффект в витрине каталога
CommissionRulesChanged (набор правил застройщика)	Для каждого объекта недвижимости застройщика пересчитывается AgentCommission матчингом (см. «Матчинг»); пустой набор правил снимает комиссию со всех объектов недвижимости

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

• Из метаданных доступны: идентификатор сообщения (дедупликация), DeveloperCompanyId (AggregateId = Kafka key), тип агрегата, время события (OccurredAt). Корреляция трейса — из метаданных.
• Актор — системный; карточки не tenanted, DeveloperCompanyId события маппится на колонку-фильтр DeveloperId карточки (не tenant).
• Идемпотентность: ключ catalog-rewards:commission-rules:{MessageId}; бизнес-идемпотентность — per-property-object guard INV-5 (шаг 4).

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

Контракт — event-contract: pin.rewards.commission-rules.v1 (owner-документ rewards). Семантика — полный per-developer снапшот действующего набора правил комиссий застройщика на OccurredAt. Дом полей правила — reward-rule.md.

Поле payload	Тип	Обяз.	Как используется (маппинг)
DeveloperCompanyId	Guid	Да	= CompanyId застройщика; маппится на колонку-фильтр DeveloperId — ключ выборки объектов недвижимости застройщика
RulesetVersion	long (>= 1)	Да	Монотонная версия набора правил застройщика; per-property-object guard INV-5 → AgentCommission.RuleVersion
Rules	IReadOnlyList<CommissionRuleSnapshot>	Да (может быть пуст)	Действующие правила; пустой = комиссия снята со всех объектов недвижимости; вход матчинга
OccurredAt	DateTimeOffset	Да	Снапшот-момент; проверка периода правила; курсор RewardsSyncedAt для объектов недвижимости с AgentCommission = null

CommissionRuleSnapshot (дом — reward-rule.md): RuleId (Guid), RuleVersion (long), Priority (int), CalculationBase (CommissionCalculationBase: BasePrice/FinalPrice), CommissionPercent (decimal(5,2)), EffectiveFrom/EffectiveTo (DateTimeOffset?), Conditions (RewardRuleConditionsModel: PropertyTypes/ProjectIds/LayoutGroupIds/ LayoutIds/RoomsRange (NullableRange<int>)/AreaRangeM2 (NullableRange<decimal>); «пусто = any, И между измерениями»).

Алгоритм

1. Контекст события. При catalog.commissions.enabled = false consumer не подписан (топик копит события, replay безопасен). Восстановить метаданные; при несовпадении ContractVersion — warn-and-proceed; проверить идемпотентность по ключу catalog-rewards:commission-rules:{MessageId} — дубль → пропуск (ack).
2. Проверка входа. DeveloperCompanyId != Guid.Empty; RulesetVersion >= 1; OccurredAt не в будущем > 5 мин (clock-skew → warn + принять); каждый CommissionRuleSnapshot: RuleId != Guid.Empty, RuleVersion >= 1, 0 < CommissionPercent <= 100 (нарушение элемента → правило исключается + warn, событие не теряется), CalculationBase известен. Существование по id не проверяется (батч по DeveloperCompanyId); пустой набор объектов недвижимости допустим (правила заданы до импорта объектов) → ack no-op + метрика.
3. Что читаем. Объекты недвижимости застройщика страницами (по 500, keyset по Id): DeveloperId == payload.DeveloperCompanyId && Status != Delisted. Атрибуты матчинга уже на карточке — запросов в чужой контекст нет. Оверрайды не читаются: AgentCommission не оверрайдится (источник — только rewards).
4. Бизнес-правила и переходы (для каждого объекта недвижимости страницы).
   • Матчинг: правило сопоставляется по атрибутам объекта, набору Rules и OccurredAt (раздел «Матчинг»); результат — PropertyObjectAgentCommission?.
   • Guard INV-5 (монотонность): применять только если AgentCommission == null ? OccurredAt > RewardsSyncedAt : RulesetVersion >= AgentCommission.RuleVersion. Иначе (устаревший снапшот, обычно replay) — catalog_commission_stale_total, объект недвижимости пропускается.
   • Маппинг: новая комиссия перезаписывает VO (в т.ч. в null при пустом матче); RewardsSyncedAt = время события; при непустой комиссии → RuleVersion = RulesetVersion. Идентичный результат → объект недвижимости не помечается изменённым (экономия), курсор RewardsSyncedAt обновляется. Статус/видимость объекта недвижимости не меняются.
5. Что меняется. Для изменившихся объектов недвижимости — обновление полей commission_* + rewards_synced_at. Порядок внутри страницы не важен.
6. Согласованность. Граница — страница (не мега-транзакция); отметка идемпотентности — в финальном изменении после последней страницы, offset подтверждается после неё. Конфликт оптимистической блокировки с chessboard-projection-sync → перечитать объект недвижимости и повторить (до 3 раз).
7. Результат. Успех → ack. Retriable инфраструктурный сбой → retry; non-retriable (контракт) → DLQ + алерт. SLO лага < 15 мин — витрина показывает последний применённый снапшот. Доменных/ интеграционных событий consumer не порождает.

Диаграмма

Матчинг

Правило сопоставления правил rewards с объектом недвижимости — детерминированная функция без внешних вызовов: по атрибутам объекта, списку правил и моменту asOf возвращает PropertyObjectAgentCommission?. Пригодна для двух вызывающих (ниже), покрыта unit-тестами без Kafka/БД.

Вход PropertyObjectMatchingAttributes (проекция полей карточки)

Поле	Тип	Источник в property_object_cards
DeveloperId	Guid	DeveloperId (маппинг из payload DeveloperCompanyId)
ProjectId	Guid	ProjectId (ЖК объекта)
PropertyType	PropertyType (Apartment/Parking/Commercial)	маппинг из TypeName; unknown → fallback по Kind проекта + warn (DSC-CAT-2)
LayoutId	Guid?	LayoutId
LayoutGroupId	Guid?	LayoutGroupId
RoomsCount	short?	RoomsCount (null для паркинга/коммерции)
AreaTotalM2	decimal	AreaTotal

Предикат соответствия (правило подходит ⇔ все измерения истинны)

Семантика Conditions — дом reward-rule.md: пусто/null = «любое»; непустое — значение объекта недвижимости обязано попасть в множество/диапазон; между измерениями — логическое И.

Измерение	Условие соответствия
Период	(EffectiveFrom == null ‖ EffectiveFrom <= asOf) ∧ (EffectiveTo == null ‖ EffectiveTo > asOf)
PropertyTypes	пусто → any; иначе PropertyType ∈ PropertyTypes (unknown-тип при непустом списке → не подошло, fail-closed)
ProjectIds	пусто → any; иначе ProjectId ∈ ProjectIds
LayoutIds	пусто → any; иначе LayoutId != null ∧ LayoutId ∈ LayoutIds
LayoutGroupIds	пусто → any; иначе LayoutGroupId != null ∧ LayoutGroupId ∈ LayoutGroupIds
RoomsRange (NullableRange<int>)	пусто → any; иначе RoomsCount != null ∧ RoomsRange.Contains(RoomsCount) (открытая граница = сторона не ограничена)
AreaRangeM2 (NullableRange<decimal>)	пусто → any; иначе AreaRangeM2.Contains(AreaTotalM2) (открытая граница = сторона не ограничена)

Выход PropertyObjectAgentCommission?

Пусть matched = правила из rules, прошедшие предикат:

• matched пуст → null (комиссия не задана; фронт скрывает блок).
• иначе representative = правило с максимальным Priority (детерминированно — INV-3 rewards, приоритет уникален среди активных), VO: PercentMin = min(matched.CommissionPercent), PercentMax = max(matched.CommissionPercent) (Min == Max при одном правиле); BaseKind = map(representative.CalculationBase) (CommissionCalculationBase → CommissionBaseKind, 1:1 {BasePrice, FinalPrice}) — база начисляемого правила; RewardRuleId = representative.RuleId (без FK, чужой контекст); RuleVersion = payload.RulesetVersion (курсор монотонности INV-5, не версия одного правила).

Вызывающие и деградация

• Этот consumer — rules из payload (без внешних вызовов); батч по всем объектам недвижимости застройщика.
• chessboard-projection-sync.md шаг D4 — при создании объекта недвижимости/изменении полей матчинга rules берутся из internal API rewards (calculate-property-object-commission); недоступность/ошибка API не валит событие: AgentCommission удерживает прежнее/null + метрика catalog_commission_recompute_deferred_total. Internal API резолвит одно правило по приоритету (PercentMin == PercentMax), тогда как снапшот-событие даёт весь набор и потому полный диапазон PercentMin..PercentMax; при следующем CommissionRulesChanged объект дозаполняется диапазоном.

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

• Дедупликация: ключ catalog-rewards:commission-rules:{MessageId}.
• Бизнес-guard (INV-5): per-property-object — RulesetVersion против AgentCommission.RuleVersion (комиссия задана) либо OccurredAt против RewardsSyncedAt (null). Key = DeveloperCompanyId → per-developer порядок; guard закрывает replay и дубли.
• Replay: да (снапшот-семантика + детерминизм матчинга + guard). Нужен для первичного наполнения после включения producer-а/флага и после POST /admin/projections/rebuild (сброс AgentCommission/RewardsSyncedAt).

Retry и DLQ

DLQ-топик: pin.catalog.rewards-commission-dlq.v1 (событие + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset).

Сбой	Код	Retry	DLQ	Действие
Инфраструктурный (PostgreSQL/Kafka, timeout)	—	Бесконечный backoff 1s→60s с паузой партиции	Нет	Runbook catalog-overview
Конфликт оптимистической блокировки — гонка с chessboard-sync	—	3 немедленных повтора с перечиткой	После исчерпания	Расследование гонки
Неизвестный ContractVersion/структура payload	CATALOG_COMMISSION_CONTRACT_INVALID	Нет	Да + алерт	Согласовать контракт с owner rewards (commission-rules.md) → replay
Некорректный элемент Rules	—	Нет — правило исключается + warn	Нет	Правка правила в rewards → новое событие

«Не найдена» — невозможно

Карточка «не найдена» — невозможно: адресация по DeveloperCompanyId (не по id); отсутствие объектов недвижимости → штатный no-op.

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

• Logs: consumed/skipped-duplicate/skipped-stale/recomputed/deferred/dead-lettered (DeveloperCompanyId, RulesetVersion, propertyObjectsAffected/Set/Cleared); суммы/проценты комиссий internal, PII нет.
• Metrics: catalog_commission_lag_seconds, catalog_commission_consume_duration_ms, catalog_commission_property_objects_recomputed_total, catalog_commission_stale_total, catalog_commission_recompute_deferred_total (общая с D4), catalog_commission_retries_total, catalog_commission_dlq_total, catalog_commission_contract_mismatch_total.
• Traces: span приёма события → span обработки → per-page spans хранилища; correlation из метаданных.
• Alerts: лаг > 15 мин (SLO README каталога); DLQ > 0; рост ..._deferred_total (недоступность internal API rewards). Dashboard/Runbook: catalog-overview; runbooks «replay DLQ», «rebuild проекций».
• Безопасность: системный актор; trusted-граница (внутренний Kafka, m2m ACL); карточки не tenanted, DeveloperCompanyId события — только фильтр (cross-tenant-запись невозможна). AgentCommission отдаётся в ответ только по permission catalog.commissions.view; fail-closed на нарушении контракта (DLQ).

Acceptance Criteria

• Пересчёт комиссий застройщика. Дано CommissionRulesChanged (непустой Rules, RulesetVersion свежее). Когда обрабатывается. Тогда по каждому объекту недвижимости застройщика (Status != Delisted) матчинг даёт AgentCommission, изменившиеся объекты обновлены, RewardsSyncedAt = now, при непустой комиссии RuleVersion = RulesetVersion; ack.
• Диапазон vs единственное правило. Дано подошло несколько правил. Когда собирается VO. Тогда PercentMin = min(CommissionPercent), PercentMax = max, BaseKind — по правилу с максимальным Priority; при одном подошедшем правиле PercentMin == PercentMax.
• Пустой набор правил. Дано Rules = []. Когда обрабатывается. Тогда AgentCommission снимается со всех объектов недвижимости застройщика (VO → null).
• Ни одно правило не подошло. Дано объект не проходит предикат матчинга. Когда матчинг выполнен. Тогда AgentCommission = null (блок комиссии на фронте скрыт).
• Stale (INV-5). Дано RulesetVersion < AgentCommission.RuleVersion (или OccurredAt <= RewardsSyncedAt при AgentCommission = null). Когда обрабатывается. Тогда skipped-stale, объект недвижимости пропускается.
• Некорректный элемент Rules. Дано правило с CommissionPercent = 0. Когда валидируется. Тогда правило исключается + warn, остальные применяются (событие не падает, не DLQ).
• Нет объектов застройщика. Дано правила заданы до импорта объектов недвижимости. Когда обрабатывается. Тогда штатный no-op ack + метрика (карточка «не найдена» невозможна — адресация по DeveloperCompanyId).
• Гонка с chessboard-sync. Дано параллельная запись той же карточки. Когда обновляется комиссия. Тогда конфликт оптимистической блокировки разрешается перечиткой объекта и повтором (до 3 раз).
• Флаг выключен. Дано catalog.commissions.enabled = false. Тогда consumer не подписан и поле agentCommission в API не отдаётся.
• Неизвестный контракт. Дано неизвестный ContractVersion/структура payload. Когда обрабатывается. Тогда CATALOG_COMMISSION_CONTRACT_INVALID → DLQ + алерт (fail closed).
• Replay. Дано re-read топика после POST /admin/projections/rebuild. Когда переигрывается. Тогда комиссии восстановлены детерминированно (снапшот + матчинг + guard).

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

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

• Домен (CQRS) (1): Entity / Aggregate: RewardRuleEntity
• События (2): Event Contract: pin.rewards.commission-rules (CommissionRulesChangedKafkaEvent), Event Contract: pin.rewards.reward-rule-changed (RewardRuleChangedKafkaEvent)
• Консюмеры (3): Consumer: Chessboard Projection Sync (pin-catalog-chessboard-sync), Consumer: Invest-Analytics Enrichment (pin-catalog-analytics-enrichment), Consumer: User-Activity Popularity (pin-catalog-popularity)
• index.md (1): Catalog Service (Pin.Catalog)

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

• Целевой VO и инвариант: catalog-property-object-card.md (PropertyObjectAgentCommission, RewardsSyncedAt, INV-5 CommissionConsistency); карточка проекта комиссию не денормализует — catalog-project-card.md.
• Контракт события (owner): commission-rules.md (pin.rewards.commission-rules.v1, per-developer снапшот набора, key = DeveloperCompanyId).
• Источник правил: reward-rule.md (Conditions, Priority/INV-3), rewards/README.md; internal API (D4, одно правило) — calculate-property-object-commission.
• Соседние консюмеры: chessboard-projection-sync.md (шаг D4), invest-analytics-enrichment.md.
• ADR-0052, ADR-0056.