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

Consumer: ChessboardPropertyObjectPriceChangedConsumer

Назначение​

Единственный пишущий процесс ленты истории цен invest-analytics. Слушает событие изменения цены объекта недвижимости PropertyObjectPriceChangedKafkaEvent из контекста chessboard и добавляет (никогда не обновляет и не удаляет) append-only факт PropertyObjectPriceHistoryFact в ленту ClickHouse pin_analytics.property_object_price_history_facts батчами. Из фактов public API строит график роста цены (day/month/year + growthPercent, ../api/public/get-property-object-price-history.md), а расчёт серий доходности берёт историческую часть серии price-per-meter (../domain/investor-series-models.md). Owned-состояние в PostgreSQL consumer не меняет: снапшот измерений (Developer/Project/Building/LayoutGroup/Layout/AreaTotal) приходит в payload события — реплика измерений не заводится.

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

ПолеЗначение
Сервис/контекстinvest-analytics
ConsumerChessboardPropertyObjectPriceChangedConsumer
Обрабатываемое событиеPropertyObjectPriceChangedKafkaEvent
Topicpin.chessboard.property-object-price-changed.v1
ContractVersion1; при несовпадении версии — warn-and-proceed
Consumer groupinvest-analytics-price-history
Producerchessboard (агрегат PropertyObject, каждое изменение Price, в т.ч. синком из Talent — docs/06-services/chessboard/domain/property-object.md)
Эффект (событие → что меняется)добавляется строка факта цены в ленту ClickHouse pin_analytics.property_object_price_history_facts (append-only); PG-состояние не меняется

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

  • Из метаданных события доступны: TenantId, AggregateId (= PropertyObjectId), AggregateType (= PropertyObject), время события (CreatedAt); идентификатор сообщения (EventId) прописывается в факт как SourceEventId (сквозная трассировка событие → строка ленты).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Бизнес-идемпотентность — декларативная, на стороне ленты ClickHouse: FactId = SourceEventId — последний компонент ключа сортировки ReplacingMergeTree, повторная доставка схлопывается при merge, а читающие запросы дубль-устойчивы (argMax(..., IngestionTime); FINAL запрещён). Отдельная таблица идемпотентности не ведётся: у консюмера нет PG-транзакции, а отметка в PG отдельно от вставки в ленту была бы вторым несогласованным хранилищем вместо одного встроенного механизма dedup.

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

Контракт входного события — docs/06-services/chessboard/events/property-object-price-changed.md (будущий путь, владелец — chessboard). Ниже — поля payload, которые consumer реально читает, и маппинг в колонки факта. Дом полей и типы колонок — ../domain/price-history-fact.md, раздел «Columns».

Поле payloadТипОбязательностьКак используется (маппинг)
PropertyObjectIdGuidДа→ колонка PropertyObjectId (UUID, 1-й компонент ORDER BY); Guid.Empty → poison (DLQ)
DeveloperIdGuidДа→ колонки DeveloperId (UUID) и TenantId (Guid как строка, ADR-0052: tenant = компания застройщика)
ProjectIdGuidДа→ колонка ProjectId
BuildingIdlongДа→ колонка BuildingId (Int64)
LayoutGroupIdlongДа→ колонка LayoutGroupId; связывает факт с property_object_type_config.GroupId
LayoutIdlongДа→ колонка LayoutId
AreaTotaldecimalДа→ колонка AreaTotal (Decimal(10,2)); знаменатель fallback-вычисления цены за метр; <= 0 → poison (DLQ)
OldPricedecimal?Нет→ колонка OldPrice; null для первой зафиксированной цены объекта недвижимости
NewPricedecimalДа→ колонка NewPrice (Decimal(18,2), полная цена объекта недвижимости, руб); < 0 → poison (DLQ)
OldPricePerMeterdecimal?Нет→ колонка OldPricePerMeter; если отсутствует, а OldPrice задан — вычисляется round(OldPrice / AreaTotal, 2) (симметрично NewPricePerMeter)
NewPricePerMeterdecimal?Нет→ колонка NewPricePerMeter; если отсутствует — вычисляется round(NewPrice / AreaTotal, 2)
CurrencyIdshortДа→ колонка CurrencyId (Int16, типизированная ссылка справочника валют handbook); RUB-only (ADR-0048)
CurrencySnapshotCurrencySnapshot ({ CurrencyId, IsoCode, Name, Sign? })ДаCurrencySnapshot.IsoCode → колонка CurrencyIsoCode (денорм ISO-4217-кода для display); RUB-выравнивание по IsoCode: IsoCode ≠ "RUB" → warning-лог (мультивалютность вне v1, дом факта), но факт пишется с фактическими CurrencyId/IsoCode. Строкового кода валюты payload отдельным полем не несёт — код живёт только внутри snapshot
ChangedAtDateTimeOffsetДа→ колонка EventTime (DateTime, UTC) — бизнес-время изменения цены в Talent; определяет партицию toYYYYMM; default/будущее время > now + 1 ч → poison (DLQ)
SourcePropertyObjectStatusSourceНетне маппится в колонку: ChangeSource факта проставляет consumer (talent-sync), а не producer; поле используется только в логе
VersionuintНетне маппится; используется только в логе (диагностика гонок в chessboard)

Колонки, которые consumer заполняет сам (не из payload): FactId = SourceEventId = EventId события; IngestionTime = now() момента вставки; BrandId = NULL; ChangeSource = talent-sync; TransformVersion = 1 (версия маппинга этого consumer-а; bump при изменении правил маппинга).

Зависимость от контракта producer-а​

Текущий набросок payload в docs/06-services/chessboard/domain/property-object.md (PropertyObjectId, TenantId, ProjectId, OldPrice, NewPrice, CurrencyId + CurrencySnapshot, PricePerMeter, ChangedAt, Source, Version) не содержит снапшота измерений DeveloperId, BuildingId, LayoutGroupId, LayoutId, AreaTotal, OldPricePerMeter. Требование к детальному ТЗ chessboard (зафиксировано и в «Известных ограничениях» ../README.md): событие обязано нести снапшот измерений — иначе invest-analytics пришлось бы держать реплику измерений или ходить в chessboard синхронно из consumer-а. Добавление полей — additive-only, bump ContractVersion не требуется.

Алгоритм​

Consumer батчевый: шаги 1–4 выполняются на сообщение, шаги 5, 7–8 — на батч (flush по порогу: ≤ 1000 строк или 5 секунд — freshness SLA факта ≤ 5 минут выполняется с запасом).

  1. Контекст события. Восстановить трейс и контекст из метаданных события (идентификатор сообщения EventId → SourceEventId, PropertyObjectId, тип агрегата, время); сверить ContractVersion (= 1) — при несовпадении warning-лог и продолжить (warn-and-proceed). Бизнес-идемпотентность — без lookup: dedup декларативен (FactId = SourceEventId), поэтому дубль не отсеивается на входе, а схлопывается хранилищем.
  2. Проверка входа (по-полевой чеклист — колонка «Обязательность» таблицы payload):
    • десериализация payload не удалась → сообщение в DLQ (poison), обработка партиции продолжается;
    • PropertyObjectId == Guid.Empty / DeveloperId == Guid.Empty / ProjectId == Guid.Empty → DLQ + error-лог (нарушение контракта producer-а);
    • BuildingId <= 0 / LayoutGroupId <= 0 / LayoutId <= 0 → DLQ + error-лог (см. «Зависимость от контракта producer-а» — алерт, пока chessboard не публикует снапшот измерений);
    • NewPrice < 0, AreaTotal <= 0, ChangedAt == default или ChangedAt > now() + 1 час → DLQ;
    • существование сущностей по id не проверяется: consumer не проверяет FK в PG — измерения приходят снапшотом, факт самодостаточен;
    • допустимость состояния не проверяется: у append-only факта нет статусов.
  3. Что читаем — ничего: consumer не читает ни PG, ни ClickHouse, ни chessboard; вход события достаточен для построения строки факта.
  4. Бизнес-правила и маппинг поле-в-поле. Построить PropertyObjectPriceHistoryFact по таблице маппинга выше; вычислимые правила: NewPricePerMeter = payload.NewPricePerMeter ?? round(NewPrice / AreaTotal, 2); OldPricePerMeter = payload.OldPricePerMeter ?? (OldPrice is null ? null : round(OldPrice / AreaTotal, 2)); EventTime = ChangedAt (UTC); TenantId = DeveloperId; CurrencyId = payload.CurrencyId, CurrencyIsoCode = payload.CurrencySnapshot.IsoCode (RUB-выравнивание: IsoCode ≠ "RUB" → warning-лог, не сбой); ChangeSource = "talent-sync"; TransformVersion = 1. Переходов состояний нет. Добавить строку в буфер батча.
  5. Что меняется. По достижении порога (1000 строк или 5 секунд с первой строки буфера) — один batch INSERT буфера в ленту pin_analytics.property_object_price_history_facts. Update/Delete/Upsert запрещены — append-only; пишет только этот consumer.
  6. Какие факты/события фиксируются. Доменных событий нет (агрегата нет); сам ingestion интеграционных событий не публикует (append-only ClickHouse, без PG/outbox). Пересчёт витринных метрик доходности и публикация pin.invest-analytics.property-object-metrics.v1 — отдельный recompute-компонент, реагирующий на это же изменение цены (../events/property-object-metrics.md); карта событий — ../events/README.md.
  7. Согласованность. Границей атомарности служит один batch INSERT ClickHouse; commit Kafka offset выполняется строго после подтверждённой вставки батча. Сбой между вставкой и commit → повторная доставка батча → дубли схлопываются ReplacingMergeTree (at-least-once без потерь и без видимых дублей).
  8. Результат. Успех батча → commit offset; метрики/логи (см. «Наблюдаемость»); контролируемая ошибка вставки → retry всего батча (идемпотентно), offset не двигается.

Диаграмма​

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

  • Business key: FactId = SourceEventId (= EventId события, уникален на событие). Поведение дубля: строка вставляется повторно и схлопывается ReplacingMergeTree при merge; до merge дубли невидимы читателям — запросы API дубль-устойчивы (argMax(NewPricePerMeter, IngestionTime) внутри периода).
  • Replay: да, штатный runbook (см. ../README.md, «Runtime И Эксплуатация») — переигрывание топика с произвольного offset consumer-group-ой invest-analytics-price-history безопасно; «дырки» истории после инцидента закрываются повтором диапазона offset без ручной очистки CH.
  • Пересечение с backfill-ом исключено схемой ключей: у backfill-строк SourceEventId = backfill:{PropertyObjectId}:{ChangedAt:O}, у consumer-строк — Guid EventId события (см. talent-price-sync.md).

Retry и DLQ​

СбойRetryDLQРучное действие
ClickHouse недоступен / timeout вставкиДа: retry батча с экспоненциальным backoff (1с → 60с), без ограничения попыток; offset не commit-итсяНет (батч не «яд» — сбой инфраструктуры)Алерт по росту invest_analytics_price_ingest_lag_seconds > 15 мин; восстановить CH; consumer догонит сам
Ошибка десериализации payloadНетДа (poison, per-message)Разбор сообщения из DLQ; при системной причине — issue producer-у chessboard
Невалидный payload (шаг 2: пустые id, AreaTotal <= 0, NewPrice < 0, битое ChangedAt)НетДа (poison, per-message) + error-логПроверить контракт producer-а (см. «Зависимость от контракта producer-а»); после исправления — replay сообщения из DLQ
ContractVersion события не совпалНе сбой: warning-лог, обработка продолжаетсяНетСверить график миграции контракта с chessboard
Сбой между вставкой батча и commit offsetПовторная доставка батча (at-least-once)НетНе требуется — дубли схлопывает ReplacingMergeTree

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

  • Logs: consumed (offset-диапазон, размер батча), skipped-poison (причина + SourceEventId), flushed (строк, длительность вставки), failed (ошибка CH). Число дублей не логируется — dedup на стороне CH.
  • Metrics: invest_analytics_price_facts_ingested_total, invest_analytics_price_ingest_lag_seconds (EventTime → IngestionTime), invest_analytics_price_ingest_batch_size, счётчики retry и DLQ.
  • Traces: span приёма события + span вставки; correlation из метаданных события; SourceEventId в строке факта связывает CH-данные с событием.
  • PII/masking: PII в payload и факте отсутствует (факт описывает объект, не пользователя).
  • Actor/permissions: пишет только consumer; прямой записи в CH из API/других сервисов нет. Tenant: TenantId факта берётся из payload (DeveloperId), scope-фильтрация на запись не применяется — consumer платформенный; tenant mismatch метаданных vs payload → warning-лог (fail closed не требуется: чтение цен публично).

Acceptance Criteria​

  • Given валидное событие изменения цены, When consumer обработал сообщение и батч сброшен, Then в ленте ClickHouse появляется ровно одна строка факта (FactId = SourceEventId); Kafka offset commit-ится строго после подтверждённой вставки батча.
  • Given то же событие доставлено повторно (at-least-once), When обработано, Then дубль строки схлопывается ReplacingMergeTree при merge; до merge читающие запросы дубль-устойчивы (argMax(..., IngestionTime)); lookup идемпотентности не выполняется.
  • Given событие с более ранним EventTime пришло с задержкой (out-of-order), When обработано, Then факт добавляется в партицию своего EventTime и не затирает более свежие; порядок по объекту восстанавливается сортировкой API по EventTime.
  • Given payload не десериализуется или пустой PropertyObjectId/DeveloperId/ProjectId или AreaTotal <= 0/NewPrice < 0 или ChangedAt = default/> now + 1 ч, When обработано, Then сообщение → DLQ (poison, per-message) + error-лог; обработка партиции продолжается.
  • Given ClickHouse недоступен/timeout вставки, When flush батча, Then батч ретраится с backoff (1с → 60с) без ограничения попыток, offset не двигается; в DLQ ничего не уходит (сбой инфраструктуры).
  • Given ContractVersion события ≠ 1, When обработано, Then warning-лог и продолжение (warn-and-proceed), не сбой.
  • Given событие без NewPricePerMeter, When маппинг, Then значение вычисляется как round(NewPrice / AreaTotal, 2); CurrencyId = payload.CurrencyId, CurrencyIsoCode = payload.CurrencySnapshot.IsoCode, ChangeSource = talent-sync.
  • Given событие с CurrencySnapshot.IsoCode ≠ "RUB" (мультивалютность вне v1), When маппинг, Then warning-лог RUB-выравнивания, факт всё равно пишется с фактическими CurrencyId/CurrencyIsoCode (не DLQ, не сбой).

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

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

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

  • Контракт события: docs/06-services/chessboard/events/property-object-price-changed.md (будущий путь); набросок payload — docs/06-services/chessboard/domain/property-object.md.
  • Дом факта: ../domain/price-history-fact.md; канал синка Talent: talent-price-sync.md; первичная загрузка: ../jobs/price-history-backfill.md (будущий док).
  • Produced-события контекста: ../events/README.md (property-object-metrics.md — переиздаётся recompute-компонентом при изменении цены).
  • ADR-0052 (tenancy), ADR-0056 (simplicity-first, границы ClickHouse), ADR-0057 (паттерн CH-facts).