feat(tradein): per-source listing price history — schema + daily snapshot (#570) (#646)

tradein-mvp/backend/app/services/scheduler.py

@@ -12,6 +12,8 @@ Sources:
   - yandex_city_sweep       → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT)
   - cian_history_backfill   → run_cian_backfill (this module, #560)
   - rosreestr_dkp_import    → import_rosreestr_dkp (this module, #563)
+  - listing_source_snapshot → snapshot_listing_sources (tasks/listing_source_snapshot.py, #570;
+                              pure internal DB snapshot — no external calls, enabled by default)
 """
 from __future__ import annotations
 
@@ -368,6 +370,41 @@ async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -
     return run_id
 
 
+async def trigger_listing_source_snapshot_run(
+    db: Session, schedule_row: dict[str, Any]
+) -> int | None:
+    """Создать scrape_runs + launch snapshot_listing_sources в executor (sync DB-only task).
+
+    Per-source price-history daily snapshot (#570). Mirror trigger_rosreestr_dkp_run:
+    задача синхронная (чистый internal DB-snapshot, БЕЗ внешних HTTP-вызовов / анти-бота),
+    поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (079),
+    в отличие от avito/yandex sweep (dormant).
+
+    Returns run_id (или None если skip — есть running run).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            # snapshot_listing_sources is synchronous (DB-only, no async HTTP)
+            from app.tasks.listing_source_snapshot import snapshot_listing_sources
+
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, snapshot_listing_sources, run_db, run_id)
+        except Exception:
+            logger.exception("scheduler: snapshot_listing_sources crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered listing_source_snapshot run_id=%d", run_id)
+    return run_id
+
+
 def import_rosreestr_dkp(
     db: Session,
     run_id: int,
@@ -637,6 +674,8 @@ async def scheduler_loop() -> None:
                         await trigger_cian_backfill_run(db, sch)
                     elif source == "rosreestr_dkp_import":
                         await trigger_rosreestr_dkp_run(db, sch)
+                    elif source == "listing_source_snapshot":
+                        await trigger_listing_source_snapshot_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

tradein-mvp/backend/app/tasks/listing_source_snapshot.py

@@ -0,0 +1,141 @@
+"""Daily per-source snapshot writer (#570).
+
+Берёт текущее состояние listing_sources (последний снимок на canonical listing × source)
+и пишет ежедневный ряд в listing_source_snapshots + change-log price_change в
+listing_source_events. Так история per-source цены копится СРАЗУ, независимо от
+(сейчас DORMANT) скраперов — см. шапку data/sql/079_listing_source_history.sql.
+
+Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app
+scheduler'ом через trigger_listing_source_snapshot_run() (scheduler.py),
+по образцу import_rosreestr_dkp (sync task в run_in_executor).
+
+Вся работа — два set-based SQL statement'а (snapshot upsert + event-diff CTE),
+никакого row-by-row Python: 18 355 строк обслуживаются одним INSERT … SELECT каждый.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+# Окно свежести: источник считается активным, если last_seen_at не старше N дней.
+FRESHNESS_WINDOW_DAYS = 7
+
+# ── Daily snapshot upsert ─────────────────────────────────────────────────────
+# Снимок на (listing_source_id, CURRENT_DATE). ON CONFLICT → last-write-wins за день
+# (повторный прогон в те же сутки перезаписывает снимок свежими значениями).
+# is_active derived: last_seen_at в пределах окна свежести на момент снимка.
+# payload_hash = md5(raw_payload::text) — ::text на колонке допустим (это не bind-param).
+# run_id через CAST(:run_id AS bigint) — psycopg v3 (никогда :run_id::bigint).
+_SNAPSHOT_SQL = text(
+    """
+    INSERT INTO listing_source_snapshots (
+        listing_source_id, snapshot_date, price_rub, is_active,
+        last_seen_at, payload_hash, observed_at, run_id
+    )
+    SELECT
+        id,
+        CURRENT_DATE,
+        price_rub,
+        (last_seen_at > now() - make_interval(days => :freshness_days)) AS is_active,
+        last_seen_at,
+        md5(raw_payload::text),
+        now(),
+        CAST(:run_id AS bigint)
+    FROM listing_sources
+    ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE SET
+        price_rub    = EXCLUDED.price_rub,
+        is_active    = EXCLUDED.is_active,
+        last_seen_at = EXCLUDED.last_seen_at,
+        payload_hash = EXCLUDED.payload_hash,
+        observed_at  = EXCLUDED.observed_at,
+        run_id       = EXCLUDED.run_id
+    """
+)
+
+# ── Event diff: price_change ──────────────────────────────────────────────────
+# Для каждого источника сравниваем сегодняшнюю цену (snapshot_date = CURRENT_DATE) с
+# самым свежим ПРЕДЫДУЩИМ снимком (snapshot_date < CURRENT_DATE). Если цена изменилась
+# (обе NOT NULL, old <> 0) — пишем price_change.
+#   today  — снимок за сегодня (только что записан _SNAPSHOT_SQL).
+#   prior  — последний снимок строго ДО сегодня (DISTINCT ON … ORDER BY date DESC).
+# Полностью set-based: один INSERT … SELECT по всем источникам, без Python-цикла.
+# change_time = now() детерминирует UNIQUE(listing_source_id, change_time, event_type)
+# в пределах прогона → ON CONFLICT DO NOTHING делает писатель идемпотентным.
+_EVENT_DIFF_SQL = text(
+    """
+    WITH today AS (
+        SELECT listing_source_id, price_rub
+        FROM listing_source_snapshots
+        WHERE snapshot_date = CURRENT_DATE
+    ),
+    prior AS (
+        SELECT DISTINCT ON (listing_source_id)
+            listing_source_id, price_rub
+        FROM listing_source_snapshots
+        WHERE snapshot_date < CURRENT_DATE
+        ORDER BY listing_source_id, snapshot_date DESC
+    )
+    INSERT INTO listing_source_events (
+        listing_source_id, change_time, event_type, price_rub, diff_percent
+    )
+    SELECT
+        t.listing_source_id,
+        now(),
+        'price_change',
+        t.price_rub,
+        round((t.price_rub - p.price_rub)::numeric / p.price_rub * 100, 4)
+    FROM today t
+    JOIN prior p ON p.listing_source_id = t.listing_source_id
+    WHERE t.price_rub IS NOT NULL
+      AND p.price_rub IS NOT NULL
+      AND p.price_rub <> 0
+      AND t.price_rub <> p.price_rub
+    ON CONFLICT (listing_source_id, change_time, event_type) DO NOTHING
+    """
+)
+
+
+def snapshot_listing_sources(db: Session, run_id: int) -> dict[str, int]:
+    """Записать дневной снимок listing_sources + price_change-события.
+
+    Sync (вызывается scheduler-триггером в executor, как import_rosreestr_dkp).
+    Два set-based statement'а в одной транзакции:
+      1. upsert снимка на (listing_source_id, CURRENT_DATE) — last-write-wins.
+      2. diff сегодняшней цены против последнего предыдущего снимка → price_change-события.
+
+    Финализирует scrape_runs (mark_done / mark_failed) и пишет counters.
+
+    Returns {"snapshotted": N, "price_change_events": M}.
+    """
+    counters: dict[str, int] = {"snapshotted": 0, "price_change_events": 0}
+    try:
+        snap_result = db.execute(
+            _SNAPSHOT_SQL,
+            {"freshness_days": FRESHNESS_WINDOW_DAYS, "run_id": run_id},
+        )
+        counters["snapshotted"] = snap_result.rowcount or 0
+
+        event_result = db.execute(_EVENT_DIFF_SQL)
+        counters["price_change_events"] = event_result.rowcount or 0
+
+        db.commit()
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "snapshot_listing_sources run_id=%d done: snapshotted=%d price_change_events=%d",
+            run_id,
+            counters["snapshotted"],
+            counters["price_change_events"],
+        )
+        return counters
+    except Exception as exc:
+        logger.exception("snapshot_listing_sources run_id=%d failed", run_id)
+        db.rollback()
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/079_listing_source_history.sql

@@ -0,0 +1,140 @@
+-- 079_listing_source_history.sql
+-- #570 — per-source listing price history.
+--
+-- ПРОБЛЕМА: listing_sources хранит ТОЛЬКО последний снимок на (canonical listing × source)
+--   (price_rub / last_seen_at / raw_payload перезаписываются при каждом скрейпе) — истории нет.
+--   Скраперы сейчас DORMANT (avito/yandex sweep enabled=false), так что ждать накопления
+--   истории «органически» нельзя. Решение: ежедневный internal snapshot существующих 18 355
+--   строк listing_sources (4 источника) — история начинает копиться СРАЗУ, независимо от
+--   скраперов. Снимок пишет snapshot_listing_sources() (app/tasks/listing_source_snapshot.py),
+--   запускаемый in-app scheduler'ом (source='listing_source_snapshot', seed ниже, enabled=true).
+--
+-- СОЗДАЁТ:
+--   listing_source_snapshots — daily snapshot на источник (PK (listing_source_id, snapshot_date)).
+--   listing_source_events    — change-log (price_change/delisted/relisted/edited/first_seen).
+--   offer_price_history.listing_source_id — backlink на конкретный источник (был только source text).
+--   v_listing_source_price_on_date — view «цена источника на дату» (snapshots ⋈ listing_sources).
+--
+-- ПАРТИЦИОНИРОВАНИЕ — НАМЕРЕННО НЕТ (отложено).
+--   ADR предлагал партиционировать listing_source_snapshots по snapshot_date (RANGE),
+--   но на текущем объёме это преждевременно: ~18 355 строк/день × ~30 ≈ 540k строк/мес —
+--   plain-таблица с индексом (snapshot_date DESC) обслуживает это без проблем. Партиционирование
+--   вводим позже отдельной миграцией, когда retention/объём этого реально потребуют.
+--
+-- ЗАВИСИМОСТИ: listing_sources (existing), scrape_runs (015_scrape_runs.sql),
+--   offer_price_history (existing), scrape_schedules (052_scrape_schedules.sql, UNIQUE(source)).
+-- Idempotent: CREATE TABLE/INDEX IF NOT EXISTS, ADD COLUMN IF NOT EXISTS, CREATE OR REPLACE VIEW,
+--   ON CONFLICT (source) DO NOTHING — безопасно запускать повторно.
+-- Apply after: 078_scrape_schedules_seed_yandex_sweep.sql
+
+BEGIN;
+
+-- ── listing_source_snapshots — ежедневный снимок на источник ──────────────────
+-- is_active — derived: last_seen_at в пределах окна свежести (7 дней) на момент снимка.
+-- payload_hash — md5(raw_payload::text), дешёвый детектор «изменился ли payload» между днями.
+-- run_id — провенанс: какой scrape_runs создал снимок (ON DELETE SET NULL — run могут чистить).
+CREATE TABLE IF NOT EXISTS listing_source_snapshots (
+    listing_source_id bigint      NOT NULL REFERENCES listing_sources(id) ON DELETE CASCADE,
+    snapshot_date     date        NOT NULL DEFAULT CURRENT_DATE,
+    price_rub         bigint,
+    is_active         boolean     NOT NULL,
+    last_seen_at      timestamptz,
+    payload_hash      text,
+    observed_at       timestamptz NOT NULL DEFAULT now(),
+    run_id            bigint      REFERENCES scrape_runs(id) ON DELETE SET NULL,
+    PRIMARY KEY (listing_source_id, snapshot_date)
+);
+
+COMMENT ON TABLE listing_source_snapshots IS
+    'Ежедневный снимок listing_sources на источник (#570). Plain-таблица (не партиционирована — '
+    'см. шапку 079). PK (listing_source_id, snapshot_date), last-write-wins за день.';
+
+CREATE INDEX IF NOT EXISTS idx_lss_snapshot_date
+    ON listing_source_snapshots (snapshot_date DESC);
+CREATE INDEX IF NOT EXISTS idx_lss_source_date
+    ON listing_source_snapshots (listing_source_id, snapshot_date DESC);
+
+-- ── listing_source_events — change-log по источнику ───────────────────────────
+-- diff_percent — для price_change: (new-old)/old*100. payload_diff — опциональный JSON-дифф.
+-- UNIQUE (listing_source_id, change_time, event_type) — идемпотентность писателя (ON CONFLICT).
+CREATE TABLE IF NOT EXISTS listing_source_events (
+    id                bigserial   PRIMARY KEY,
+    listing_source_id bigint      NOT NULL REFERENCES listing_sources(id) ON DELETE CASCADE,
+    change_time       timestamptz NOT NULL,
+    event_type        text        NOT NULL CHECK (
+        event_type IN ('price_change', 'delisted', 'relisted', 'edited', 'first_seen')
+    ),
+    price_rub         bigint,
+    diff_percent      numeric,
+    payload_diff      jsonb,
+    recorded_at       timestamptz NOT NULL DEFAULT now(),
+    UNIQUE (listing_source_id, change_time, event_type)
+);
+
+COMMENT ON TABLE listing_source_events IS
+    'Change-log на источник (#570): price_change/delisted/relisted/edited/first_seen. '
+    'Писатель — snapshot_listing_sources() (set-based, ON CONFLICT DO NOTHING).';
+
+CREATE INDEX IF NOT EXISTS idx_lse_source_change_time
+    ON listing_source_events (listing_source_id, change_time DESC);
+CREATE INDEX IF NOT EXISTS idx_lse_type_change_time
+    ON listing_source_events (event_type, change_time DESC);
+
+-- ── offer_price_history — backlink на конкретный источник ─────────────────────
+-- Раньше была только колонка source (text) — без ссылки на listing_sources.id.
+-- ON DELETE SET NULL: история переживает удаление источника (canonical price-history сохраняется).
+ALTER TABLE offer_price_history
+    ADD COLUMN IF NOT EXISTS listing_source_id bigint
+        REFERENCES listing_sources(id) ON DELETE SET NULL;
+
+CREATE INDEX IF NOT EXISTS idx_oph_source_change_time
+    ON offer_price_history (listing_source_id, change_time DESC);
+
+-- ── v_listing_source_price_on_date — «цена источника на дату» ─────────────────
+-- Удобный join snapshots ⋈ listing_sources: (listing_id, ext_source, дата, цена, активность).
+CREATE OR REPLACE VIEW v_listing_source_price_on_date AS
+SELECT
+    ls.listing_id,
+    ls.ext_source,
+    lss.snapshot_date,
+    lss.price_rub,
+    lss.is_active
+FROM listing_source_snapshots lss
+JOIN listing_sources ls ON ls.id = lss.listing_source_id;
+
+COMMENT ON VIEW v_listing_source_price_on_date IS
+    'Цена/активность источника на дату снимка (#570): listing_source_snapshots ⋈ listing_sources.';
+
+-- ── Seed scrape_schedules row для daily snapshot ──────────────────────────────
+-- enabled = true — БЕЗОПАСНО включать сразу: pure-internal DB snapshot, никаких внешних
+--   HTTP-вызовов / анти-бота (в отличие от avito/yandex sweep, которые seed'ятся DORMANT).
+-- Окно 01:00-02:00 UTC — РАНЬШЕ остальных job'ов (cian/rosreestr 02:00+), чтобы снять
+--   снимок до того, как backfill-задачи начнут менять listing_sources в тот же день.
+-- next_run_at = завтрашнее наступление окна (tomorrow + 01:00 UTC), иначе при NULL
+--   get_due_schedules() выстрелит сразу после деплоя. AT TIME ZONE 'UTC' фиксирует
+--   wall-clock час в UTC независимо от session TimeZone GUC.
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'listing_source_snapshot',
+    true,                                        -- SAFE: pure internal DB, no external calls
+    1,
+    2,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 1)) AT TIME ZONE 'UTC',
+    '{}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет cron-script setup). '
+    'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
+    'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570).';
+
+COMMIT;

tradein-mvp/backend/tests/test_listing_source_snapshot.py

@@ -0,0 +1,281 @@
+"""Tests for the per-source listing price-history snapshot writer (#570).
+
+Writer (snapshot_listing_sources) is SQL-heavy, so most assertions are static: we read
+the emitted SQL via inspect/text-attr and check the upsert + event-diff CTE shape, table/
+column names, and the psycopg-v3 cast discipline (no :param::type). We also assert the
+scheduler wiring (trigger fn + dispatch branch) and the migration 079 contents.
+
+Plus one cheap behavioural test: a fake db (monkeypatched .execute) drives the counter
+logic without a real Postgres.
+
+Static style mirrors tests/test_rosreestr_dedup_key.py / test_yandex_city_sweep.py.
+"""
+
+import inspect
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+# Importing app.services.scheduler / app.tasks pulls app.core.config.Settings → needs
+# DATABASE_URL. Stub it BEFORE app imports (as in test_scheduler.py) — these tests are
+# static / fake-db; no live database is touched.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import scheduler
+from app.tasks import listing_source_snapshot as snap_mod
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_079 = _SQL_DIR / "079_listing_source_history.sql"
+
+# Emitted SQL text (SQLAlchemy text() clause → .text gives the raw string).
+_SNAPSHOT_SQL = str(snap_mod._SNAPSHOT_SQL.text)
+_EVENT_DIFF_SQL = str(snap_mod._EVENT_DIFF_SQL.text)
+_ALL_SQL = _SNAPSHOT_SQL + "\n" + _EVENT_DIFF_SQL
+_WRITER_SRC = inspect.getsource(snap_mod.snapshot_listing_sources)
+
+
+# ── Snapshot upsert SQL ───────────────────────────────────────────────────────
+
+
+def test_snapshot_targets_correct_table_and_columns() -> None:
+    assert "INSERT INTO listing_source_snapshots" in _SNAPSHOT_SQL
+    for col in (
+        "listing_source_id",
+        "snapshot_date",
+        "price_rub",
+        "is_active",
+        "last_seen_at",
+        "payload_hash",
+        "observed_at",
+        "run_id",
+    ):
+        assert col in _SNAPSHOT_SQL, f"snapshot SQL missing column {col!r}"
+    assert "FROM listing_sources" in _SNAPSHOT_SQL
+    assert "CURRENT_DATE" in _SNAPSHOT_SQL
+
+
+def test_snapshot_is_upsert_last_write_wins() -> None:
+    """ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE — last-write-wins for the day."""
+    assert "ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE" in _SNAPSHOT_SQL
+    # Collapse runs of whitespace so column-alignment spacing doesn't matter.
+    flat = re.sub(r"\s+", " ", _SNAPSHOT_SQL)
+    for col in ("price_rub", "is_active", "last_seen_at", "payload_hash", "observed_at", "run_id"):
+        assert f"{col} = EXCLUDED.{col}" in flat, f"upsert SET missing {col}"
+
+
+def test_snapshot_derives_is_active_and_payload_hash() -> None:
+    # is_active derived from freshness window; payload_hash = md5(raw_payload::text).
+    assert "make_interval(days => :freshness_days)" in _SNAPSHOT_SQL
+    assert "AS is_active" in _SNAPSHOT_SQL
+    assert "md5(raw_payload::text)" in _SNAPSHOT_SQL
+
+
+# ── Event-diff CTE SQL ────────────────────────────────────────────────────────
+
+
+def test_event_diff_is_set_based_cte_not_python_loop() -> None:
+    """Event diff is one set-based INSERT … SELECT over a CTE — never a row-by-row loop."""
+    assert "WITH today AS" in _EVENT_DIFF_SQL
+    assert "prior AS" in _EVENT_DIFF_SQL
+    assert "DISTINCT ON (listing_source_id)" in _EVENT_DIFF_SQL
+    assert "INSERT INTO listing_source_events" in _EVENT_DIFF_SQL
+    # Prior = most-recent snapshot strictly before today.
+    assert "snapshot_date < CURRENT_DATE" in _EVENT_DIFF_SQL
+    assert "snapshot_date = CURRENT_DATE" in _EVENT_DIFF_SQL
+    assert "ORDER BY listing_source_id, snapshot_date DESC" in _EVENT_DIFF_SQL
+    # No Python iteration over rows in the writer body (set-based only).
+    body = _WRITER_SRC.split('"""', 2)[-1]
+    assert "for " not in body, "writer must be set-based — no Python row loop"
+
+
+def test_event_diff_emits_price_change_with_diff_percent() -> None:
+    assert "'price_change'" in _EVENT_DIFF_SQL
+    # diff_percent = (new-old)/old*100.
+    assert "(t.price_rub - p.price_rub)" in _EVENT_DIFF_SQL
+    assert "/ p.price_rub * 100" in _EVENT_DIFF_SQL
+    # Only when the price actually changed and old is a usable denominator.
+    assert "t.price_rub <> p.price_rub" in _EVENT_DIFF_SQL
+    assert "p.price_rub <> 0" in _EVENT_DIFF_SQL
+    assert "p.price_rub IS NOT NULL" in _EVENT_DIFF_SQL
+
+
+def test_event_diff_is_idempotent_on_conflict() -> None:
+    assert "ON CONFLICT (listing_source_id, change_time, event_type) DO NOTHING" in _EVENT_DIFF_SQL
+
+
+# ── psycopg v3 cast discipline ────────────────────────────────────────────────
+
+
+def test_sql_uses_psycopg_v3_casts_not_double_colon() -> None:
+    """psycopg v3: bind params via CAST(:x AS type), never :x::type.
+
+    Literal ::-casts on columns (raw_payload::text, ::numeric) are fine — the ban is
+    only on :param::type, which psycopg v3 breaks.
+    """
+    # No bind-param (:name) followed by ::type anywhere in the emitted SQL.
+    assert not re.search(r":\w+::", _ALL_SQL)
+    # run_id bound via CAST(:run_id AS bigint).
+    assert "CAST(:run_id AS bigint)" in _SNAPSHOT_SQL
+    # freshness window bound via named param inside make_interval (no cast needed).
+    assert ":freshness_days" in _SNAPSHOT_SQL
+
+
+# ── Writer return / finalisation contract ─────────────────────────────────────
+
+
+def test_writer_returns_counters_and_finalises_run() -> None:
+    assert "snapshotted" in _WRITER_SRC
+    assert "price_change_events" in _WRITER_SRC
+    assert "mark_done" in _WRITER_SRC
+    assert "mark_failed" in _WRITER_SRC
+
+
+# ── Scheduler wiring ──────────────────────────────────────────────────────────
+
+
+def test_trigger_fn_exists_and_is_async() -> None:
+    fn = getattr(scheduler, "trigger_listing_source_snapshot_run", None)
+    assert fn is not None, "trigger_listing_source_snapshot_run missing from scheduler"
+    assert inspect.iscoroutinefunction(fn)
+
+
+def test_dispatch_branch_wired() -> None:
+    """scheduler_loop dispatches source='listing_source_snapshot' to the trigger."""
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source == "listing_source_snapshot"' in loop_src
+    assert "trigger_listing_source_snapshot_run(db, sch)" in loop_src
+
+
+def test_trigger_runs_sync_task_in_executor() -> None:
+    """Sync DB-only task → run_in_executor (mirror rosreestr), not a bare await."""
+    trig_src = inspect.getsource(scheduler.trigger_listing_source_snapshot_run)
+    assert "run_in_executor(None, snapshot_listing_sources" in trig_src
+    assert "_claim_run(db, schedule_row)" in trig_src
+
+
+# ── Migration 079 ─────────────────────────────────────────────────────────────
+
+
+def test_migration_079_exists() -> None:
+    assert _MIGRATION_079.is_file(), f"missing migration: {_MIGRATION_079}"
+
+
+def test_migration_079_creates_three_schema_objects_and_view() -> None:
+    sql = _MIGRATION_079.read_text("utf-8")
+    # 1. snapshots table (plain — explicitly NOT partitioned, noted in header).
+    assert "CREATE TABLE IF NOT EXISTS listing_source_snapshots" in sql
+    assert "PRIMARY KEY (listing_source_id, snapshot_date)" in sql
+    assert "REFERENCES listing_sources(id) ON DELETE CASCADE" in sql
+    assert "REFERENCES scrape_runs(id) ON DELETE SET NULL" in sql
+    assert "CREATE INDEX IF NOT EXISTS idx_lss_snapshot_date" in sql
+    assert "CREATE INDEX IF NOT EXISTS idx_lss_source_date" in sql
+    # 2. events table + CHECK + UNIQUE.
+    assert "CREATE TABLE IF NOT EXISTS listing_source_events" in sql
+    assert "id                bigserial   PRIMARY KEY" in sql.replace("\t", " ") or (
+        "id" in sql and "bigserial" in sql and "PRIMARY KEY" in sql
+    )
+    assert "'price_change', 'delisted', 'relisted', 'edited', 'first_seen'" in sql
+    assert "UNIQUE (listing_source_id, change_time, event_type)" in sql
+    # 3. offer_price_history backlink column + index.
+    assert "ALTER TABLE offer_price_history" in sql
+    assert "ADD COLUMN IF NOT EXISTS listing_source_id bigint" in sql
+    assert "idx_oph_source_change_time" in sql
+    # 4. view.
+    assert "CREATE OR REPLACE VIEW v_listing_source_price_on_date" in sql
+    assert "JOIN listing_sources ls ON ls.id = lss.listing_source_id" in sql
+    # Transactional + idempotent + partitioning note.
+    assert "BEGIN;" in sql and "COMMIT;" in sql
+    assert "НЕТ" in sql and "артицион" in sql  # partitioning explicitly deferred
+
+
+def test_migration_079_seeds_schedule_enabled_true() -> None:
+    sql = _MIGRATION_079.read_text("utf-8")
+    assert "'listing_source_snapshot'" in sql
+    assert "'{}'::jsonb" in sql
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+    # enabled=true (SAFE — pure internal DB). Assert the seeded VALUES row has true,
+    # not false (unlike the dormant yandex/avito sweeps).
+    assert "true,                                        -- SAFE" in sql
+    assert "false" not in sql.split("INSERT INTO scrape_schedules")[1]
+
+
+def test_migration_079_uses_psycopg_safe_sql() -> None:
+    """Migration is plain DDL (no bind params), but guard against accidental :x::type."""
+    sql = _MIGRATION_079.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+# ── Cheap behavioural test: counter logic via fake db ─────────────────────────
+
+
+class _FakeResult:
+    def __init__(self, rowcount: int) -> None:
+        self.rowcount = rowcount
+
+
+class _FakeDB:
+    """Minimal stand-in for a SQLAlchemy Session — records execute() calls, returns rowcounts."""
+
+    def __init__(self, rowcounts: list[int]) -> None:
+        self._rowcounts = list(rowcounts)
+        self.executed: list[Any] = []
+        self.committed = False
+        self.rolled_back = False
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        self.executed.append((stmt, params))
+        return _FakeResult(self._rowcounts.pop(0))
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+
+def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None:
+    """snapshot_listing_sources maps the two execute() rowcounts to its counters + marks done."""
+    marked: dict[str, Any] = {}
+    monkeypatch.setattr(
+        snap_mod.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(snap_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+
+    db = _FakeDB(rowcounts=[18355, 42])  # snapshot rowcount, then event rowcount
+    out = snap_mod.snapshot_listing_sources(db, run_id=99)  # type: ignore[arg-type]
+
+    assert out == {"snapshotted": 18355, "price_change_events": 42}
+    assert db.committed is True
+    assert len(db.executed) == 2
+    # run_id threaded into the snapshot statement's bind params.
+    _stmt, params = db.executed[0]
+    assert params is not None and params["run_id"] == 99
+    # Run finalised via mark_done with the same counters.
+    assert marked["run_id"] == 99
+    assert marked["counters"] == {"snapshotted": 18355, "price_change_events": 42}
+
+
+def test_counter_logic_failure_path_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    """On execute error: rollback + mark_failed + re-raise (no silent swallow)."""
+    failed: dict[str, Any] = {}
+    monkeypatch.setattr(snap_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(
+        snap_mod.runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
+    )
+
+    class _BoomDB(_FakeDB):
+        def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+            raise RuntimeError("boom")
+
+    db = _BoomDB(rowcounts=[])
+    with pytest.raises(RuntimeError, match="boom"):
+        snap_mod.snapshot_listing_sources(db, run_id=7)  # type: ignore[arg-type]
+    assert db.rolled_back is True
+    assert failed["run_id"] == 7