Background Job: PriceHistoryBackfillJob

Назначение

PriceHistoryBackfillJob — разовая (one-shot), перезапускаемая задача первичной загрузки: читает текущие цены существующих объектов недвижимости из Talent Chessboard API и материализует по одной стартовой строке PropertyObjectPriceHistoryFact на объект недвижимости в ClickHouse pin_analytics.property_object_price_history_facts (ChangeSource = backfill). Нужна потому, что топик pin.chessboard.property-object-price-changed.v1 на момент ввода сервиса не содержит истории «до начала публикации» — стартовые цены существующих объектов недвижимости из потока взять неоткуда (обоснование «job, не consumer» — ../consumers/talent-price-sync.md).

Постоянный поток изменений цен обеспечивает consumer chessboard-price-changed (ChangeSource = talent-sync, OldPrice из payload); backfill только дозаполняет старт (OldPrice = NULL) и разово. Прямого постоянного синка invest-analytics → Talent не существует (единственный пишущий поток в постоянном режиме — consumer).

Metadata	Значение
Owning team	Backend PIN (инвест-аналитика)
Статус документа	draft
Область действия	контекст invest-analytics; запись — ClickHouse pin_analytics.property_object_price_history_facts; состояние (checkpoint) — PostgreSQL invest_analytics.price_backfill_checkpoints
Assumptions/ограничения	backfill исполняется один раз при вводе сервиса; повторный прогон диапазона PropertyObjectId допустим и идемпотентен; freshness backfill не входит в SLA факта ≤ 5 мин (стартовая догрузка, не поток); ключ PropertyObjectId (UUID) упорядочивается по возрастанию — курсор постраничного чтения
Зависимости	Talent Chessboard API (источник цен, разовое чтение); ClickHouse (pin_analytics); PostgreSQL (invest_analytics — checkpoint); feature-flag invest-analytics.price-backfill.enabled
Кросс-ссылки	../README.md («Events И Consumers» → Background-джобы); ../consumers/talent-price-sync.md; ../domain/price-history-fact.md («Consumers И Backfill»); ../consumers/chessboard-price-changed.md; ../use-cases/UC-PRC-001-view-property-object-price-growth-chart.md

Классификация

Поле	Значение
Сервис	invest-analytics
Тип job	One-off (разовая, перезапускаемая конечная задача; не бесконечный consumer — классификация в ../consumers/talent-price-sync.md)
Trigger	вручную (admin/CLI) или по feature-flag invest-analytics.price-backfill.enabled при вводе сервиса в эксплуатацию; непрерывного расписания нет
Область tenant	платформенная (backfill вне tenant — как и запись consumer-а, ADR-0052)
Критичность	low (стартовая догрузка вне денежного и потокового пути; поток цен идёт независимо через consumer)

Расписание И Locking

Аспект	Правило
Schedule	нет — разовый запуск оператором/флагом; после Completed повторно не запускается без явного триггера
Distributed lock	Redis не используется (ADR-0056): единственность прогона даёт claim активной строки checkpoint блокировкой FOR UPDATE SKIP LOCKED — второй инстанс получает пустой результат и завершается no-op; даже при гонке идемпотентность (dedup FactId) исключает дубли
Batch size	BatchSize строк на страницу (default 1000, <= 1000); прогон повторяет чтение страниц, пока источник не исчерпан
Max duration	не ограничивается искусственно: задача конечна и возобновляема — рестарт продолжает с checkpoint
Cancellation	отмена учитывается между батчами — graceful stop, курсор закоммичен

Вход RunPriceHistoryBackfill

Actor — system (ручной триггер/флаг); публичного входа у job нет. Контекст вне tenant (платформенная maintenance-операция): скан идёт по всем объектам недвижимости Talent безотносительно DeveloperId. Проверки входа: From <= To при обоих заданных; BatchSize ∈ [1, 1000] — иначе ошибка валидации.

Поле	Тип	Обязательность	Источник	Описание
FromPropertyObjectId	Guid?	Нет	вход оператора	Нижняя граница диапазона PropertyObjectId для повторного прогона (null = с начала/с checkpoint)
ToPropertyObjectId	Guid?	Нет	вход оператора	Верхняя граница диапазона (null = до конца); при обоих заданных — From <= To
BatchSize	int	Нет	вход/конфиг	Размер страницы чтения и батча вставки (default 1000, 1..1000)

Состояние PriceBackfillCheckpoint (PostgreSQL invest_analytics.price_backfill_checkpoints)

Состояние прогона; дом полей — этот документ.

Поле	Тип	Описание
Id	Guid	PK, = RunId прогона
Status	PriceBackfillStatus (enum строкой: Running/Completed/Failed)	Состояние прогона
FromPropertyObjectId / ToPropertyObjectId	Guid?	Диапазон прогона (из входа)
LastProcessedPropertyObjectId	Guid?	Курсор: последний обработанный PropertyObjectId (точка возобновления)
ProcessedPropertyObjectCount	long	Обработано объектов недвижимости (для наблюдаемости)
IngestedFactCount	long	Вставлено строк факта
StartedAt / UpdatedAt / CompletedAt	DateTimeOffset / DateTimeOffset?	Тайминги; UpdatedAt — heartbeat
Version	int	Оптимистичная блокировка

Источник TalentPropertyObjectPriceSnapshot (Talent Chessboard API)

Строго типизованная модель страницы источника (сортировка по PropertyObjectId asc, курсор after).

Поле	Тип	Описание
PropertyObjectId	Guid	Идентификатор объекта недвижимости (property_object_id Talent)
DeveloperId / ProjectId	Guid	Измерения (застройщик = tenant; проект)
BuildingId / LayoutGroupId / LayoutId	long	Измерения (здание/группа планировок/планировка)
AreaTotal	decimal	Общая площадь, м²
CurrentPrice	decimal	Текущая цена объекта недвижимости, руб → NewPrice
PricePerMeter	decimal?	Цена за метр; если null — вычисляется
ChangedAt	DateTimeOffset	Бизнес-время текущей цены в Talent → EventTime

Валюты источник не несёт: Talent Chessboard API кода валюты не передаёт (как и поток изменений через Talent-синк). Backfill проставляет RUB-дефолт в CurrencyId/CurrencyIsoCode (RUB-only, ADR-0048) — задел под мультивалютность закрыт справочником handbook на стороне chessboard, а не первичной догрузкой.

Алгоритм

Шаги 1–2 — на прогон; 3–8 — цикл по батчам страниц источника до исчерпания.

1. Контекст. Actor = system, scope платформенный (не tenant), correlationId = RunId. Бизнес-идемпотентность — декларативная: dedup FactId в ClickHouse (шаг 6) + возобновление по checkpoint (шаг 3); отдельного ключа идемпотентности нет (PG-мутации факта нет, аналогично consumer-у).

2. Проверка входа. From <= To при обоих заданных; BatchSize ∈ [1,1000] — иначе ошибка валидации. Feature-flag invest-analytics.price-backfill.enabled выключен → отказ price-backfill-disabled.

3. Claim checkpoint + чтение источника.

   • Claim: найти активную строку checkpoint для того же диапазона (Status == Running, совпадающие From/To) под блокировкой FOR UPDATE SKIP LOCKED. Строка найдена и залочена → resume с cursor = LastProcessedPropertyObjectId. Пустой результат (лочит другой инстанс) → выход no-op (единственность прогона без Redis). Активной строки нет → создать новый checkpoint (RunId, From, To, Status = Running).
   • Источник: прочитать очередную страницу из Talent Chessboard API (after = cursor, to = ToPropertyObjectId, take = BatchSize), сортировка PropertyObjectId asc.

4. Маппинг поле-в-поле. На каждый TalentPropertyObjectPriceSnapshot построить одну PropertyObjectPriceHistoryFact (дом колонок/типов — ../domain/price-history-fact.md, «Columns»; здесь только backfill-дельта):

   Колонка факта	Значение для backfill
   FactId = SourceEventId	backfill:{PropertyObjectId}:{ChangedAt:O} (формат ключа ≠ идентификатор события consumer-а → коллизии со строками потока невозможны)
   ChangeSource	backfill
   OldPrice, OldPricePerMeter	NULL (нет предыстории у стартовой цены)
   EventTime	ChangedAt.UtcDateTime (партиция toYYYYMM)
   IngestionTime	now() в момент вставки
   NewPrice	snapshot.CurrentPrice
   NewPricePerMeter	snapshot.PricePerMeter ?? round(CurrentPrice / AreaTotal, 2)
   AreaTotal, PropertyObjectId, DeveloperId, ProjectId, BuildingId, LayoutGroupId, LayoutId	из snapshot
   TenantId	DeveloperId.ToString() (ADR-0052)
   CurrencyId / CurrencyIsoCode	RUB-запись справочника валют handbook / "RUB" — Talent Chessboard API кода валюты не несёт, backfill проставляет RUB-дефолт (RUB-only, ADR-0048; симметрично резолву IsoCode = 'RUB' в chessboard-синке)
   BrandId / TransformVersion	NULL / 1

   Переходов состояний у факта нет. Строки складываются в буфер батча (<= BatchSize).

5. Вставка факта. Один batch INSERT буфера в pin_analytics.property_object_price_history_facts. Update/Delete/Upsert в CH запрещены (append-only).

6. События. Интеграционных событий backfill не публикует (разовая append-only загрузка фактов в CH; produced-события контекста — ../events/README.md). При завершении прогона поднимается доменное PriceBackfillCompletedDomainEvent — in-process для метрик/логов, в Kafka не уходит.

7. Advance курсора (checkpoint). После подтверждённого batch INSERT в CH — обновить в checkpoint курсор (LastProcessedPropertyObjectId), счётчики (ProcessedPropertyObjectCount, IngestedFactCount) и UpdatedAt под той же блокировкой. CH-вставка и PG-checkpoint — два хранилища, не атомарны; порядок фиксирован: сперва INSERT в CH, затем advance курсора PG (как подтверждение offset после вставки у consumer-а). Последняя страница (источник исчерпан) → перевод Status: Running → Completed, CompletedAt = now.

8. Результат. Успех по завершении прогона (или no-op при skip-locked/пустом источнике); метрики/логи (см. «Наблюдаемость»). Сбой источника/вставки → отказ, курсор остаётся на последнем закоммиченном checkpoint, Status = Failed (или остаётся Running) — рестарт продолжает.

Граница Транзакции

Элемент	Правило
Unit of work	per-batch: один batch INSERT в CH + одно обновление checkpoint в PG на страницу
Сохраняемое состояние	ClickHouse property_object_price_history_facts (append); PG price_backfill_checkpoints (курсор/счётчики/статус)
External calls	Talent Chessboard API — до вставки (чтение страницы); внутри PG-транзакции внешних вызовов нет
Частичная ошибка	stop-at-checkpoint: сбой на батче не откатывает уже вставленные/закоммиченные батчи; рестарт продолжает с LastProcessedPropertyObjectId

Взаимодействие с потоком изменений цен

Backfill и постоянный consumer пишут в одну таблицу разными пространствами FactId (backfill:{PropertyObjectId}:{ChangedAt:O} vs идентификатор события потока) — коллизии ключей исключены, поэтому оба потока безопасно перекрываются во времени независимо от порядка запуска:

• Если цена объекта недвижимости изменилась между чтением backfill-снапшота и первым событием потока, в таблице окажутся обе строки (стартовая backfill + событийная) с близким EventTime; агрегация «график/ API» берёт «последнюю цену бакета» через argMax(NewPrice, (EventTime, IngestionTime)), поэтому две строки одного периода не искажают ряд.
• Backfill-строка несёт OldPrice = NULL (предыстории нет); первое событие потока после неё несёт свой OldPrice из payload — «разрыв» между стартовой ценой и первым переходом допустим и не является ошибкой (append-only лента изменений, а не сплошная цепочка Old→New).
• Порядок запуска backfill и consumer не важен: обе вставки идемпотентны (dedup FactId), конечное состояние ряда одинаково.

Идемпотентность И Retry

Сценарий	Правило
Job restart	resume с LastProcessedPropertyObjectId (курсор в PG); уже обработанные объекты недвижимости пропускаются, недообработанные переигрываются
Duplicate item / повторный прогон	dedup ReplacingMergeTree по FactId = SourceEventId = backfill:{PropertyObjectId}:{ChangedAt:O}: повторный backfill даёт строки с идентичным ключом → схлопываются при merge, дублей нет
Рестарт середины батча	безопасен: между CH INSERT и advance курсора PG рестарт переигрывает страницу → те же FactId схлопываются (at-least-once, без видимых дублей)
Параллельные инстансы	блокировка активного checkpoint (FOR UPDATE SKIP LOCKED): второй инстанс завершается no-op; при гонке дубли невозможны (dedup)
Повторяемая ошибка (Talent API/CH недоступны)	retry батча с backoff; курсор не двигается; без бесконечного retry внутри прогона — падение оставляет прогон на checkpoint
Чтение	FINAL в online-запросах запрещён; агрегации API дубль-устойчивы (argMax(..., IngestionTime))

Наблюдаемость

Тип	Правило
Logs	job started (RunId, From/To, cursor); batch processed (диапазон PropertyObjectId, размер батча, вставлено); job completed (ProcessedPropertyObjectCount, IngestedFactCount, durationMs); job failed (последний PropertyObjectId, exception). PII нет (факт описывает объект)
Metrics	invest_analytics_price_facts_ingested_total (общий с consumer-ом, лейбл change_source=backfill); invest_analytics_price_backfill_processed_lots_total; invest_analytics_price_backfill_batch_size; invest_analytics_price_backfill_running (gauge 0/1)
Traces	span на прогон с RunId; атрибуты from_property_object_id/to_property_object_id/batch_size; SourceEventId связывает CH-строку с backfill-прогоном
Alerts	backfill не участвует в freshness-SLA факта (стартовая догрузка); алерт только на Status = Failed без рестарта > 1 ч и на рост ошибок вставки CH (см. ../README.md, «Alerts»)

SLO

Показатель	Цель	Обоснование
Полнота	1 стартовая строка на каждый существующий объект недвижимости Talent после Completed	инвариант backfill; проверяется ProcessedPropertyObjectCount vs число объектов недвижимости
Идемпотентность	0 дублей строк при рестартах/повторных прогонах диапазона	dedup FactId (ReplacingMergeTree)
Возобновляемость	рестарт продолжает с последнего checkpoint без ручной очистки CH	курсор LastProcessedPropertyObjectId в PG
Freshness	вне SLA факта ≤ 5 мин	это разовая стартовая догрузка, а не поток (поток — consumer)

Связанный Runbook

Сценарий	Runbook
Падение источника Talent Chessboard API	job останавливается на последнем checkpoint; перезапуск продолжает с LastProcessedPropertyObjectId; вставка идемпотентна
Пропуски/дырки после инцидента	повторный backfill диапазона PropertyObjectId (From/ToPropertyObjectId) — dedup схлопнёт повторы (см. ../README.md, «Runbooks»)
Прогон «завис» (Running, нет heartbeat)	проверить процесс; UpdatedAt устарел → перезапустить (claim через FOR UPDATE SKIP LOCKED возьмёт строку)

Тесты

Тест	Ожидание
Пустой источник	нет объектов недвижимости — no-op success, строк факта нет, Status = Completed
Успешный прогон	на каждый объект недвижимости — 1 строка, ChangeSource = backfill, OldPrice = NULL, NewPricePerMeter вычислен при отсутствии в источнике
Рестарт середины	повторный запуск продолжает с LastProcessedPropertyObjectId, итог без дублей (dedup после merge)
Повторный прогон диапазона	те же FactId → 0 дополнительных строк после merge
Параллельные инстансы	второй инстанс завершается no-op (FOR UPDATE SKIP LOCKED)
Разграничение с потоком	backfill-строки (SourceEventId = backfill:...) не конфликтуют со строками потока (идентификатор события consumer-а)
Cancellation	отмена между батчами — graceful stop, курсор закоммичен, открытых транзакций не рвёт

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

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

• Домен (CQRS) (1): PropertyObjectPriceHistoryFact (ClickHouse)
• Use Cases (1): UC-PRC-001. Просмотр графика роста цены объекта недвижимости (гость / инвестор / реф-гость readOnly)
• Консюмеры (2): Consumer: ChessboardPropertyObjectPriceChangedConsumer, Talent Price Sync: сведён к consumer-у chessboard-price-changed (решение, не отдельный consumer)
• index.md (1): InvestAnalytics Service

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

• Дом факта (колонки/dedup/партиции): ../domain/price-history-fact.md, «Consumers И Backfill».
• Решение «job, не consumer» и разграничение потока/backfill: ../consumers/talent-price-sync.md; постоянный поток — ../consumers/chessboard-price-changed.md.
• Обзор/эксплуатация: ../README.md («Events И Consumers», «Runtime И Эксплуатация», «Alerts»); потребитель истории: ../use-cases/UC-PRC-001-view-property-object-price-growth-chart.md.
• Produced-события контекста: ../events/README.md (backfill сам не публикует).
• Решения: ADR-0052 (tenancy), ADR-0056 (simplicity-first, без Redis), ADR-0031 (background jobs).