Consumer: Invest-Analytics Enrichment (pin-catalog-analytics-enrichment)

Kafka-consumer обогащения витринных карточек метриками доходности из контекста invest-analytics: окупаемость (Payback), прогнозный рост цены (GrowthPercent), арендный доход (RentalIncomePerYear), рейтинг (Rating), индекс надёжности (Reliability). Дом полей целевой проекции — catalog-property-object-card.md (поля Payback, Reliability, GrowthPercent, RentalIncomePerYear, Rating, AnalyticsSyncedAt, SourceSnapshot.Payback*/Reliability*/Rating/AnalyticsAsOf) и catalog-project-card.md (поля Payback, GrowthPercent, Rating, AnalyticsSyncedAt).

Назначение

Consumer обрабатывает событие PropertyObjectMetricsUpdated (pin.invest-analytics.property-object-metrics.v1) и обновляет денормализованные метрики доходности в catalog.property_object_cards (сырые значения — в SourceSnapshot, эффективные — по правилу Effective = Override ?? SourceSnapshot, INV-3 property-object-card) и в catalog.project_cards (проектные GrowthPercent/Rating из payload; Payback проекта — агрегат «лучшая окупаемость по объектам недвижимости», INV-2 project-card). Благодаря денормализации лента /lots сортируется «по окупаемости» и показывает метрики без запросов в invest-analytics.

Публикуемые события

Собственных интеграционных событий consumer не публикует (витрина — конечная точка). Внутренний пересчёт агрегатов проекта происходит в той же транзакции.

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

Поле	Значение
Статус	draft (контракт события опубликован владельцем, DSC-CAT-4 закрыт)
Owning team	PIN Marketplace (контекст catalog)
Обрабатываемое событие	PropertyObjectMetricsUpdated
Topic	pin.invest-analytics.property-object-metrics.v1, key = PropertyObjectId
ContractVersion	= 1; при несовпадении версии события — warn-and-proceed (catalog_analytics_enrichment_contract_mismatch_total)
Consumer group	pin-catalog-analytics-enrichment
Producer	invest-analytics — контракт опубликован владельцем (property-object-metrics.md, DSC-CAT-4 закрыт)

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

Событие	Эффект в витрине каталога
PropertyObjectMetricsUpdated	Метрики доходности объекта недвижимости (Payback/Reliability/GrowthPercent/RentalIncomePerYear/Rating) перезаписываются сырым снапшотом; эффективные значения пересчитываются с учётом оверрайда; при наличии проектной части обновляются GrowthPercent/Rating проекта; агрегат Payback проекта пересчитывается по опубликованным объектам недвижимости

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

• Из метаданных доступны: идентификатор сообщения (дедупликация), PropertyObjectId (AggregateId), тип агрегата, время события. Корреляция трейса — из метаданных.
• Актор — системный; карточки не tenanted (платформенная витрина).
• Идемпотентность: ключ catalog-analytics:property-object-metrics-updated:{MessageId}; бизнес-идемпотентность — монотонный guard по AsOf (payload.AsOf > SourceSnapshot.AnalyticsAsOf).

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

Контракт события: event-contract: pin.invest-analytics.property-object-metrics-updated — опубликован владельцем (DSC-CAT-4 закрыт).

Семантика payload

Полный snapshot метрик объекта недвижимости на момент AsOf: null в поле означает «метрика недоступна/отозвана» и очищает сырое значение (эффективное может удержать оверрайд).

Поле payload	Тип	Обязательность	Как используется (маппинг)
PropertyObjectId	Guid	Да	Карточка объекта недвижимости; нет карточки → retriable CATALOG_SYNC_PROPERTY_OBJECT_NOT_FOUND (out-of-order против pin.chessboard.property-object-upserted.v1)
ProjectId	Guid	Да	Сверка с card.ProjectId (расхождение → DLQ, fail closed) + адресация проектной части
PaybackValue	decimal?	Нет	→ SourceSnapshot.PaybackValue; эффективный VO Payback = Override(Payback*) ?? Source (INV-3; пара с PaybackPeriod — оба заданы или оба null, иначе warn + пара отбрасывается)
PaybackPeriod	string? (enum PaybackPeriod: Years/Months)	Нет	→ SourceSnapshot.PaybackPeriod; unknown → пара отбрасывается + warn
GrowthPercent	decimal?	Нет	→ GrowthPercent объекта недвижимости напрямую (диапазон -99.99..999.99, вне — clamp + warn); оверрайда для поля нет
RentalIncomePerYear	decimal?	Нет	→ RentalIncomePerYear (>= 0, иначе null + warn)
Rating	decimal?	Нет	→ SourceSnapshot.Rating; эффективный Rating = Override.Rating ?? Source (0.0..10.0, вне — clamp + warn)
ReliabilityValue	short?	Нет	→ SourceSnapshot.ReliabilityValue; эффективный VO Reliability = Override(Reliability*) ?? Source (пара с ReliabilityMax; 0 <= Value <= Max, иначе пара отбрасывается + warn)
ReliabilityMax	short?	Нет	→ SourceSnapshot.ReliabilityMax (1..100)
ProjectGrowthPercent	decimal?	Нет	→ project_cards.GrowthPercent (метрика уровня проекта; null = проектная часть не пересчитывалась — поле проекта не трогается)
ProjectRating	decimal?	Нет	→ project_cards.Rating (та же семантика null)
AsOf	DateTimeOffset	Да	Монотонный guard: применяется только при AsOf > SourceSnapshot.AnalyticsAsOf; → SourceSnapshot.AnalyticsAsOf, AnalyticsSyncedAt = now (объект недвижимости и, при наличии проектной части, проект)

Алгоритм

1. Контекст события. Восстановить метаданные; при несовпадении ContractVersion — warn-and-proceed; проверить идемпотентность по ключу catalog-analytics:property-object-metrics-updated:{MessageId} — дубль → пропуск (ack).
2. Проверка входа. PropertyObjectId != Guid.Empty; ProjectId != Guid.Empty; AsOf задан и не в будущем более чем на 5 минут (clock-skew guard, иначе warn + принять); диапазоны метрик — по таблице payload (нарушение отдельного поля не валит событие: clamp/null + warn, unknown enum PaybackPeriod — пара отбрасывается). Существование: карточка объекта недвижимости; нет карточки → CATALOG_SYNC_PROPERTY_OBJECT_NOT_FOUND — retriable (объект появится событием chessboard). Допустимость состояния: Status == Delisted → идемпотентный skip + warn (INV-7 DelistedFrozen — метрики удалённого объекта не применяются). Связь: card.ProjectId != payload.ProjectId → CATALOG_SYNC_PROJECT_MISMATCH (non-retriable → DLQ, fail closed).
3. Что читаем. Карточка объекта недвижимости (единственная по id); живой оверрайд объекта (нужен для эффективных Payback/Reliability/Rating); при наличии проектной части (ProjectGrowthPercent != null || ProjectRating != null) — карточка проекта (её отсутствие — аномалия целостности проекций → retriable, см. Retry).
4. Бизнес-правила и переходы.
   • AsOf-guard (монотонность): payload.AsOf <= SourceSnapshot.AnalyticsAsOf (при заполненном AnalyticsAsOf) → метрика catalog_analytics_enrichment_stale_total, лог skipped-stale, к согласованной фиксации (только отметка идемпотентности), ack.
   • Snapshot-семантика: перезаписать сырые значения SourceSnapshot.PaybackValue/PaybackPeriod/ReliabilityValue/ReliabilityMax/Rating (в т.ч. null-ами), SourceSnapshot.AnalyticsAsOf = payload.AsOf.
   • Эффективные поля (INV-3 property-object-card), маппинг поле-в-поле: Payback = Override(PaybackValue, PaybackPeriod) ?? Source(...) (VO целиком: оверрайд-пара приоритетнее, обе стороны — «оба или ничего»); Reliability — аналогично; Rating = Override.Rating ?? Source.Rating; GrowthPercent = payload.GrowthPercent; RentalIncomePerYear = payload.RentalIncomePerYear (оверрайдов у этих двух полей нет — источник единственный).
   • AnalyticsSyncedAt = now. Статус карточки не меняется (метрики на INV-1 не влияют).
   • Проектная часть (если загружена карточка проекта): GrowthPercent = payload.ProjectGrowthPercent ?? current; Rating = payload.ProjectRating ?? current; SourceSnapshot.AnalyticsAsOf = payload.AsOf (при AsOf-монотонности проекта; иначе проектная часть пропускается + warn); AnalyticsSyncedAt = now.
5. Что меняется. Обновление карточки объекта недвижимости; при проектной части — обновление карточки проекта. Порядок: сначала объект недвижимости, затем проект (агрегатный пересчёт шага 6 читает уже обновлённый объект).
6. Внутренний пересчёт агрегата. Изменение Payback объекта недвижимости запускает пересчёт Payback проекта: минимум нормализованной окупаемости в годах (Value при Period = Years, Value / 12 при Months) по опубликованным объектам недвижимости проекта с непустой окупаемостью → projectCard.Payback, AggregatesRecomputedAt = now (INV-2 project-card).
7. Согласованность. Объект недвижимости + проект + отметка идемпотентности — в одном согласованном изменении; offset подтверждается после фиксации. Конфликт оптимистической блокировки с конкурирующими консюмерами (chessboard-sync пишет ту же строку) → перечитать и повторить (до 3 раз).
8. Результат. Успех → ack; retriable fail → retry; non-retriable → DLQ + алерт. Freshness: SLO лага метрик < 15 мин — метрики допускают eventual-обновление, сортировка «по окупаемости» использует последние применённые значения.

Диаграмма

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

• Дедупликация: ключ catalog-analytics:property-object-metrics-updated:{MessageId}.
• Бизнес-guard: монотонный AsOf-guard (payload.AsOf > SourceSnapshot.AnalyticsAsOf) — защита от out-of-order и replay; повтор события с тем же AsOf — идемпотентный skip.
• Replay: да. Полный re-read топика безопасен (snapshot-семантика + guard); используется для первичного наполнения метрик после включения producer-а (DSC-CAT-4) и после ручного rebuild проекций (POST /api/admin/v1/projections/rebuild с ResetAnalyticsAsOf=true сбрасывает AnalyticsAsOf пересобранных карточек — метрики доедут replay-ем/следующими событиями).

Retry и DLQ

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

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka, timeout)	—	Бесконечный exponential backoff 1s→60s с паузой партиции	Нет	Runbook catalog-overview
Карточка объекта недвижимости не найдена (out-of-order против chessboard-sync)	CATALOG_SYNC_PROPERTY_OBJECT_NOT_FOUND	6 попыток, backoff 30s→1m→2m→4m→8m→16m	Да, после исчерпания	Проверить pin.chessboard.property-object-upserted.v1; replay из DLQ
Карточка проекта не найдена (аномалия целостности проекций)	—	та же политика	Да + алерт	Rebuild карточки проекта; replay
ProjectId события ≠ ProjectId карточки	CATALOG_SYNC_PROJECT_MISMATCH	Нет (non-retriable)	Да + алерт	Ручной разбор с invest-analytics (fail closed)
Конфликт оптимистической блокировки	—	3 немедленных повтора с перечиткой	После исчерпания	Расследование гонки

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

• Logs: consumed / skipped-duplicate / skipped-stale / enriched / project-part-applied / dead-lettered с полями EventId, PropertyObjectId, AsOf, SourceSnapshot.AnalyticsAsOf. PII нет.
• Metrics: catalog_analytics_enrichment_lag_seconds, ..._consume_duration_ms, ..._stale_total, ..._retries_total, ..._dlq_total, ..._contract_mismatch_total, ..._clamped_fields_total (поля вне диапазона).
• Traces: span приёма события → span обработки → spans хранилища; correlation из метаданных.
• Alerts: лаг топика > 15 мин (SLO метрик README каталога); DLQ > 0.
• Dashboard / Runbook: catalog-overview; runbooks «replay DLQ», «rebuild проекций».
• Безопасность: системный актор; trusted-граница (внутренний Kafka, m2m ACL); метрики — public display-данные, маскирование не требуется; fail closed на несоответствии связей.

Acceptance Criteria

• Обогащение метрик. Дано PropertyObjectMetricsUpdated (AsOf свежее AnalyticsAsOf) для существующей карточки. Когда обрабатывается. Тогда SourceSnapshot метрик перезаписан, эффективные Payback/Reliability/Rating = Override ?? Source, GrowthPercent/RentalIncomePerYear = payload, AnalyticsSyncedAt = now; ack.
• Проектная часть. Дано ProjectGrowthPercent/ProjectRating заданы. Когда обрабатывается. Тогда обновляются GrowthPercent/Rating проекта; агрегат Payback проекта = минимум нормализованной окупаемости (в годах) по опубликованным объектам недвижимости (INV-2 project-card).
• Отзыв метрики (null). Дано поле пришло null (метрика отозвана). Когда применяется snapshot. Тогда сырое значение очищается; эффективное может удержать оверрайд.
• Stale (AsOf-guard). Дано AsOf <= SourceSnapshot.AnalyticsAsOf. Когда обрабатывается. Тогда skipped-stale, без мутаций, ack.
• Out-of-order раньше chessboard. Дано карточки объекта ещё нет. Когда приходит метрика. Тогда CATALOG_SYNC_PROPERTY_OBJECT_NOT_FOUND (retriable) → backoff → DLQ.
• Project mismatch. Дано payload.ProjectId != card.ProjectId. Когда обрабатывается. Тогда CATALOG_SYNC_PROJECT_MISMATCH → DLQ + алерт (fail closed).
• Delisted-объект. Дано Status = Delisted. Когда приходит метрика. Тогда идемпотентный skip + warn (INV-7 DelistedFrozen — метрики удалённого объекта не применяются).
• Значение вне диапазона. Дано Rating = 42. Когда применяется. Тогда clamp в 0.0..10.0 + warn + catalog_analytics_enrichment_clamped_fields_total; событие не падает.
• Неполная пара окупаемости. Дано задан только PaybackValue без PaybackPeriod. Когда собирается VO. Тогда пара отбрасывается + warn (сырое Payback не устанавливается).
• Гонка с chessboard-sync. Дано параллельная запись той же строки консюмером chessboard-sync. Когда сохраняется обогащение. Тогда конфликт оптимистической блокировки разрешается перечиткой и повтором (до 3 раз).
• Replay. Дано re-read топика (в т.ч. первичное наполнение после включения producer-а). Когда переигрывается. Тогда итог идентичен (snapshot-семантика + AsOf-guard).

Расхождения контрактов (реестр)

#	Расхождение	Статус	Действие
DSC-CAT-4	README invest-analytics ранее декларировал «Produced events: нет», тогда как домен каталога (catalog-property-object-card.md, README каталога) требует событие pin.invest-analytics.property-object-metrics.v1 для денормализации метрик витрины (синхронное чтение API в ленте противоречит «одна строка без джойнов», ADR-0056)	закрыто	Владелец опубликовал контракт PropertyObjectMetricsUpdated (property-object-metrics.md) и внёс событие в «Produced» README invest-analytics. Payload совпадает с таблицей выше. До включения producer-а метрики карточек остаются null — витрина работает, блоки метрик скрыты

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

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

• API (1): POST /api/admin/v1/projections/rebuild — Принудительный пересчёт карточек по id (admin)
• События (3): События invest-analytics: карта контрактов, Event Contract: pin.invest-analytics.model-changed (EconomicModelChangedKafkaEvent), Event Contract: pin.invest-analytics.property-object-metrics-updated (PropertyObjectMetricsUpdatedKafkaEvent)
• Консюмеры (3): Consumer: Chessboard Projection Sync (pin-catalog-chessboard-sync), Consumer: Rewards Commission Enrichment (pin-catalog-rewards-commission), Consumer: User-Activity Popularity (pin-catalog-popularity)
• index.md (1): Catalog Service (Pin.Catalog)

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

• Целевые проекции: catalog-property-object-card.md (VO PropertyObjectPayback/PropertyObjectReliability, INV-3, нормализация окупаемости), catalog-project-card.md (агрегат Payback, INV-2); оверрайды — display-override.md.
• Producer: invest-analytics/README.md; контракт события — ../../invest-analytics/events/property-object-metrics.md.
• Соседние консюмеры: chessboard-projection-sync.md, rewards-commission.md.
• ADR-0056.