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

@@ -69,6 +69,16 @@ Sources:
                               city_sweep; fetch via curl_cffi chrome120 + scraper_proxy_url,
                               parse via YandexDetailScraper.parse;
                               window 12:00-15:00 UTC — после avito_detail_backfill 09-12 UTC)
+  - cadastral_geo_match     → run_cadastral_geo_match
+                              (tasks/cadastral_geo_match.py; combined nightly job — (1) REFRESH
+                              cad_buildings_local from gendesign_cad_buildings FDW (one bulk
+                              scan, ~36.7k EKB rows, Point geom + GIST), (2) MATCH set-based
+                              LATERAL KNN UPDATE filling listings.building_cadastral_number from
+                              the nearest cadastral building within threshold_m (default 50m).
+                              GEO-NEAREST approximation (nearest building, not exact cadastral).
+                              Pure internal DB op — one FDW read + local UPDATE, no HTTP/anti-bot;
+                              window 09:00-10:00 UTC — после geocode_missing_listings 06-09 UTC
+                              чтобы listings имели свежие lat/lon/geom)
 """
 
 from __future__ import annotations
@@ -1354,6 +1364,48 @@ async def trigger_yandex_detail_backfill_run(
     return run_id
 
 
+async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch run_cadastral_geo_match в executor (sync DB-only task).
+
+    Combined refresh+match (#cadastral-geo-match): (1) REFRESH cad_buildings_local from the
+    gendesign_cad_buildings FDW (one bulk scan, builds Point geom), (2) MATCH set-based
+    LATERAL KNN UPDATE filling listings.building_cadastral_number from the nearest cadastral
+    building within threshold_m (default 50m). GEO-NEAREST approximation.
+
+    Sync task (one FDW read + local UPDATE, no async HTTP) — run in run_in_executor by the
+    same pattern as trigger_listing_source_snapshot_run / trigger_asking_to_sold_ratio_run.
+    run_cadastral_geo_match owns the scrape_runs lifecycle (mark_done/mark_failed). SAFE to
+    enable — seed 125 enabled=true, pure internal DB op.
+
+    Returns run_id или None (skip — already running).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.tasks.cadastral_geo_match import run_cadastral_geo_match
+
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(
+                None,
+                lambda: run_cadastral_geo_match(run_db, run_id=run_id, params=params),
+            )
+        except Exception:
+            logger.exception("scheduler: run_cadastral_geo_match 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 cadastral_geo_match run_id=%d", run_id)
+    return run_id
+
+
 def get_due_schedules(db: Session) -> list[dict[str, Any]]:
     """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
     rows = (
@@ -1427,6 +1479,8 @@ async def scheduler_loop() -> None:
                         await trigger_avito_detail_backfill_run(db, sch)
                     elif source == "yandex_detail_backfill":
                         await trigger_yandex_detail_backfill_run(db, sch)
+                    elif source == "cadastral_geo_match":
+                        await trigger_cadastral_geo_match_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

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

@@ -0,0 +1,323 @@
+"""Geo-nearest cadastral building matcher for listings (#cadastral-geo-match).
+
+Fills `listings.building_cadastral_number` (currently 0% / ~43k rows) from the nearest
+cadastral building, via a LOCAL materialized mirror of the gendesign_cad_buildings FDW.
+
+WHY a local mirror (perf fact, measured live):
+    gendesign_cad_buildings is a postgres_fdw foreign table (~36.7k EKB rows) with NO
+    PostGIS geom — only scalar lat/lon. A per-listing nearest-building query over the FDW
+    takes ~1.16s/row → ~13h for 43k listings. UNUSABLE per-row.
+    ⇒ We materialize the FDW once (single bulk scan) into cad_buildings_local with a real
+      Point geom + GIST, then run a fast LOCAL KNN nearest-neighbour join (geom <-> point).
+    The whole local backfill runs in well under a minute (see PR EXPLAIN evidence).
+
+APPROXIMATION (deliberate first increment):
+    This is a GEO-NEAREST match — a street-level-geocoded listing is matched to the nearest
+    cadastral building within `threshold_m`, NOT necessarily its exact cadastral building.
+    The threshold is always logged. Exact cadastral resolution + parcel-containment are
+    deferred (cad_parcels FDW not exposed). Tier-0 house matching in the estimator already
+    treats building_cadastral_number as a hint, not ground truth, so an approximate fill is
+    a net win over 0% coverage.
+
+Pipeline (one combined run, scheduler source='cadastral_geo_match'):
+    1. refresh_cad_buildings_local(db) — TRUNCATE + bulk INSERT from FDW (one scan).
+    2. match_listings_to_buildings(db, threshold_m) — set-based LATERAL KNN UPDATE.
+    Refresh-then-match in ONE run guarantees the match always reads a fresh mirror.
+
+psycopg v3: app SQL uses CAST(:x AS type), never :x::type.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_THRESHOLD_M = 50
+DEFAULT_BATCH_SIZE = 5000
+
+# EKB latitude — used to convert the metric threshold into a degree gate radius for the
+# GIST-indexable geometry-space pre-filter (the gate is a SUPERSET of the true circle; the
+# ST_DistanceSphere recheck decides acceptance, so a slightly-too-wide gate is harmless).
+_EKB_LAT_DEG = 56.84
+_M_PER_DEG_LAT = 111_320.0
+
+
+def _deg_gate_for(threshold_m: float) -> float:
+    """Degree radius that always encloses `threshold_m` metres at EKB latitude.
+
+    Longitude degrees shrink by cos(lat), so we size the gate on the tighter (longitude)
+    axis and apply a 1.5× safety factor. The gate only pre-filters candidates for the GIST
+    KNN scan; the exact ST_DistanceSphere(...) <= threshold_m recheck is authoritative, so
+    over-sizing the gate never produces a false match, only a few extra candidates to rank.
+    """
+    m_per_deg_lon = _M_PER_DEG_LAT * math.cos(math.radians(_EKB_LAT_DEG))
+    return 1.5 * threshold_m / m_per_deg_lon
+
+
+# ── Refresh: materialize FDW → cad_buildings_local (single bulk scan) ─────────
+def refresh_cad_buildings_local(db: Session) -> int:
+    """TRUNCATE cad_buildings_local; bulk INSERT from gendesign_cad_buildings FDW.
+
+    Single FDW scan (no per-row round-trips). Builds Point(4326) geom from lon/lat.
+    Idempotent: TRUNCATE+INSERT in one transaction — the table is fully replaced atomically.
+
+    Returns the number of rows loaded.
+    """
+    db.execute(text("TRUNCATE cad_buildings_local"))
+    db.execute(
+        text(
+            """
+            INSERT INTO cad_buildings_local (
+                cad_num, readable_address, year_built, floors, area_m2, purpose,
+                lat, lon, geom
+            )
+            SELECT
+                cad_num, readable_address, year_built, floors, area_m2, purpose,
+                lat, lon,
+                ST_SetSRID(ST_MakePoint(lon, lat), 4326)
+            FROM gendesign_cad_buildings
+            WHERE lat IS NOT NULL
+              AND lon IS NOT NULL
+              AND cad_num IS NOT NULL
+            ON CONFLICT (cad_num) DO NOTHING
+            """
+        )
+    )
+    count = db.execute(text("SELECT count(*) FROM cad_buildings_local")).scalar() or 0
+    db.commit()
+    logger.info("refresh_cad_buildings_local: loaded %d rows from FDW", count)
+    return int(count)
+
+
+# ── Match: LATERAL KNN nearest building → listings.building_cadastral_number ──
+# Set-based UPDATE. For each candidate listing (geo present, optionally only NULL bcn) the
+# LATERAL subquery picks the single nearest cad_buildings_local building and accepts it only
+# if its TRUE distance is within :threshold_m metres.
+#
+# PERF (measured live, the whole point of this design):
+#   The gate is a GEOMETRY-space ST_DWithin(cb.geom, l_geom, :deg_gate) — planar degrees,
+#   fully GIST-indexable, NO per-row geography cast. The `<-> ` KNN order then returns the
+#   single nearest candidate via the GIST index. ONLY that one nearest row is converted to a
+#   true metric distance via ST_DistanceSphere (spherical metres) and compared to :threshold_m.
+#   This is ~175× faster than a geography-cast ST_DWithin in the WHERE: a geography filter
+#   forced a per-candidate cast+recheck that the planner could not push into the index, so the
+#   full 41k-listing UPDATE ran 58s+ and never finished; the geometry-gate version does a
+#   5000-row chunk in ~330ms and the full ~41k backfill in a few seconds.
+#
+#   :deg_gate is a degree radius that ALWAYS encloses :threshold_m at EKB latitude (~56.8°):
+#   1° lat ≈ 111_320 m, and we widen by /cos(lat) for longitude + a safety factor, so the
+#   gate is a superset of the true threshold circle and the ST_DistanceSphere recheck is what
+#   actually decides acceptance (no false negatives from the gate).
+#
+# Batching: pick a chunk of candidate listing ids first (LIMIT :batch_size), then run the
+# LATERAL UPDATE only over that chunk — bounds lock/WAL footprint on the ~43k table and lets
+# the run loop checkpoint a heartbeat between chunks. only_missing=true makes each chunk drain
+# monotonically (filled rows drop out of the candidate set), so the loop terminates.
+# Single set-based LATERAL KNN UPDATE over ALL candidates in one statement.
+# NOT chunked by OFFSET: an OFFSET walk over the `building_cadastral_number IS NULL`
+# predicate is unsound because the UPDATE mutates that very predicate mid-walk —
+# matched rows drop out of the candidate set, so OFFSET (reset OR advance) skips
+# un-processed rows. The full UPDATE is ~5s for ~41k listings (GIST KNN, measured),
+# so a single statement is both correct and fast enough; no chunking needed.
+_MATCH_SQL = text(
+    """
+    WITH matched AS (
+        SELECT l.id AS id, m.cad_num AS cad_num
+        FROM listings l
+        JOIN LATERAL (
+            SELECT cb.cad_num,
+                   ST_DistanceSphere(cb.geom, l.geom) AS dist_m
+            FROM cad_buildings_local cb
+            WHERE ST_DWithin(cb.geom, l.geom, CAST(:deg_gate AS double precision))
+            ORDER BY cb.geom <-> l.geom
+            LIMIT 1
+        ) m ON true
+        WHERE l.geom IS NOT NULL
+          AND (
+              CAST(:only_missing AS boolean) = false
+              OR l.building_cadastral_number IS NULL
+          )
+          AND m.dist_m <= CAST(:threshold_m AS double precision)
+    )
+    UPDATE listings l
+    SET building_cadastral_number = matched.cad_num
+    FROM matched
+    WHERE l.id = matched.id
+    """
+)
+
+_CANDIDATES_SQL = text(
+    """
+    SELECT count(*) FROM listings l
+    WHERE l.geom IS NOT NULL
+      AND (CAST(:only_missing AS boolean) = false OR l.building_cadastral_number IS NULL)
+    """
+)
+
+
+@dataclass
+class CadMatchResult:
+    buildings_loaded: int = 0  # rows in cad_buildings_local after refresh
+    candidates_total: int = 0  # listings eligible for matching (geo present + filter)
+    listings_matched: int = 0  # listings updated with a building_cadastral_number
+    threshold_m: int = DEFAULT_THRESHOLD_M
+    duration_sec: float = field(default=0.0)
+
+    def to_counters(self) -> dict[str, int]:
+        return {
+            "buildings_loaded": self.buildings_loaded,
+            "candidates_total": self.candidates_total,
+            "listings_matched": self.listings_matched,
+            "threshold_m": self.threshold_m,
+            "duration_sec": int(self.duration_sec),
+        }
+
+
+def match_listings_to_buildings(
+    db: Session,
+    *,
+    threshold_m: int = DEFAULT_THRESHOLD_M,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+    only_missing: bool = True,
+) -> int:
+    """Set-based LATERAL KNN UPDATE: fill listings.building_cadastral_number.
+
+    GEO-NEAREST (approximate): each candidate listing is matched to the nearest
+    cad_buildings_local building within `threshold_m` metres. The threshold is logged.
+
+    Single set-based UPDATE over ALL candidates (no chunking). An OFFSET walk would be
+    unsound: the UPDATE mutates the `building_cadastral_number IS NULL` predicate it pages
+    over, so matched rows drop out and OFFSET (reset OR advance) skips un-processed rows.
+    The full UPDATE is ~5s for ~41k listings (GIST KNN), so one statement is correct and
+    fast. `batch_size` is accepted for schedule-param compatibility but no longer used.
+    Returns listings matched.
+    """
+    logger.info(
+        "match_listings_to_buildings: GEO-NEAREST match threshold_m=%d "
+        "only_missing=%s (approximate: nearest building, not exact cadastral)",
+        threshold_m,
+        only_missing,
+    )
+
+    deg_gate = _deg_gate_for(float(threshold_m))
+    result = db.execute(
+        _MATCH_SQL,
+        {
+            "threshold_m": float(threshold_m),
+            "deg_gate": deg_gate,
+            "only_missing": only_missing,
+        },
+    )
+    total_matched = result.rowcount
+    db.commit()
+
+    logger.info(
+        "match_listings_to_buildings: DONE total_matched=%d threshold_m=%d",
+        total_matched,
+        threshold_m,
+    )
+    return total_matched
+
+
+# ── Combined refresh + match (callable directly or via run wrapper) ───────────
+def refresh_and_match(
+    db: Session,
+    *,
+    threshold_m: int = DEFAULT_THRESHOLD_M,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+    only_missing: bool = True,
+) -> CadMatchResult:
+    """Refresh the local cad mirror, then geo-match listings. Returns a CadMatchResult."""
+    start = time.monotonic()
+    res = CadMatchResult(threshold_m=threshold_m)
+
+    res.buildings_loaded = refresh_cad_buildings_local(db)
+    res.candidates_total = int(
+        db.execute(_CANDIDATES_SQL, {"only_missing": only_missing}).scalar() or 0
+    )
+    res.listings_matched = match_listings_to_buildings(
+        db, threshold_m=threshold_m, batch_size=batch_size, only_missing=only_missing
+    )
+    res.duration_sec = time.monotonic() - start
+
+    coverage = 100.0 * res.listings_matched / res.candidates_total if res.candidates_total else 0.0
+    logger.info(
+        "refresh_and_match: DONE buildings=%d candidates=%d matched=%d "
+        "coverage=%.1f%% threshold_m=%d duration=%.1fs",
+        res.buildings_loaded,
+        res.candidates_total,
+        res.listings_matched,
+        coverage,
+        threshold_m,
+        res.duration_sec,
+    )
+    return res
+
+
+# ── Run lifecycle wrapper (scheduler entrypoint) ─────────────────────────────
+def run_cadastral_geo_match(db: Session, *, run_id: int, params: dict) -> CadMatchResult:
+    """Run-lifecycle wrapper for the combined refresh+match job (sync, DB-only).
+
+    Launched by the in-app scheduler (source='cadastral_geo_match') via
+    trigger_cadastral_geo_match_run, or manually. Refreshes cad_buildings_local from the
+    FDW (one bulk scan) then geo-matches listings.building_cadastral_number to the nearest
+    cadastral building within threshold_m.
+
+    Params (default_params jsonb):
+        threshold_m:  max nearest-building distance to accept (metres, default 50).
+        batch_size:   listings per LATERAL UPDATE chunk (default 5000).
+        only_missing: only fill WHERE building_cadastral_number IS NULL (default true).
+
+    Finalises scrape_runs (mark_done / mark_failed) with counters.
+    """
+    threshold_m = int(params.get("threshold_m", DEFAULT_THRESHOLD_M))
+    batch_size = int(params.get("batch_size", DEFAULT_BATCH_SIZE))
+    only_missing = bool(params.get("only_missing", True))
+
+    counters: dict[str, int] = {
+        "buildings_loaded": 0,
+        "candidates_total": 0,
+        "listings_matched": 0,
+        "threshold_m": threshold_m,
+    }
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        res = refresh_and_match(
+            db,
+            threshold_m=threshold_m,
+            batch_size=batch_size,
+            only_missing=only_missing,
+        )
+
+        counters = res.to_counters()
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "run_cadastral_geo_match: run_id=%d DONE buildings=%d candidates=%d "
+            "matched=%d threshold_m=%d duration=%.1fs",
+            run_id,
+            res.buildings_loaded,
+            res.candidates_total,
+            res.listings_matched,
+            threshold_m,
+            res.duration_sec,
+        )
+        return res
+    except Exception as exc:
+        logger.exception("run_cadastral_geo_match: run_id=%d FAILED", run_id)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/124_cad_buildings_local.sql

@@ -0,0 +1,47 @@
+-- 124_cad_buildings_local.sql
+-- Local materialized mirror of gendesign_cad_buildings (postgres_fdw foreign table)
+-- for fast geo-nearest building matching of listings.
+--
+-- WHY (perf fact, measured live):
+--   gendesign_cad_buildings is a postgres_fdw foreign table (~36.7k EKB rows) with NO
+--   PostGIS geom — only scalar lat/lon. A per-listing nearest-building query over the FDW
+--   takes ~1.16s/row (lat/lon bbox prefilter). With ~43k listings that is ~13h — UNUSABLE.
+--   Solution: materialize the FDW once into a LOCAL table with a real Point geom + GIST
+--   index, then run a fast local KNN nearest-neighbour join (geom <-> point, GIST-backed).
+--
+-- This migration creates the EMPTY table + indexes only. It does NOT read the FDW —
+--   so deploy never depends on the gendesign DB / FDW server being reachable.
+--   The table is populated by the refresh job (app/tasks/refresh_cad_buildings_local.py),
+--   driven by the in-app scheduler (source='cadastral_geo_match', combined refresh+match).
+--
+-- Schedule seed (cadastral_geo_match) lives in 125_scrape_schedules_seed_cadastral_geo_match.sql.
+--
+-- DEPENDENCIES: PostGIS (postgis extension, present since geom on listings/houses).
+-- Idempotent: CREATE TABLE/INDEX IF NOT EXISTS. Safe to re-run.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS cad_buildings_local (
+    cad_num          text PRIMARY KEY,
+    readable_address text,
+    year_built       int,
+    floors           int,
+    area_m2          numeric,
+    purpose          text,
+    lat              double precision,
+    lon              double precision,
+    geom             geometry(Point, 4326)
+);
+
+COMMENT ON TABLE cad_buildings_local IS
+    'Local mirror of gendesign_cad_buildings FDW (Rosreestr building registry, EKB ~36.7k). '
+    'Populated by app/tasks/refresh_cad_buildings_local.py (single bulk FDW scan, TRUNCATE+INSERT). '
+    'Point geom + GIST enables fast local KNN nearest-building matching '
+    '(run_cadastral_geo_match) for listings.building_cadastral_number. '
+    'Migration 124. Refresh+match driven by scheduler source=cadastral_geo_match.';
+
+-- GIST on geom — required for the KNN <-> operator and ST_DWithin predicate.
+CREATE INDEX IF NOT EXISTS cad_buildings_local_geom_idx
+    ON cad_buildings_local USING GIST (geom);
+
+COMMIT;

tradein-mvp/backend/data/sql/125_scrape_schedules_seed_cadastral_geo_match.sql

@@ -0,0 +1,58 @@
+-- 125_scrape_schedules_seed_cadastral_geo_match.sql
+-- Scheduler seed for the geo-nearest cadastral building matcher (combined refresh + match).
+--
+-- WHAT (source='cadastral_geo_match'):
+--   trigger_cadastral_geo_match_run (scheduler.py) → run_cadastral_geo_match
+--   (tasks/cadastral_geo_match.py). One combined nightly job:
+--     1. REFRESH: TRUNCATE cad_buildings_local; bulk INSERT from gendesign_cad_buildings FDW
+--        (single FDW scan, ~36.7k EKB rows, builds Point geom).
+--     2. MATCH:   set-based LATERAL KNN UPDATE — fills listings.building_cadastral_number
+--        from the nearest cad_buildings_local within threshold_m (default 50m).
+--   Refresh-then-match in ONE run guarantees the match always reads a fresh mirror.
+--
+--   The match is GEO-NEAREST (approximate): a street-level-geocoded listing matches the
+--   nearest building, not necessarily its exact cadastral building. Deliberate first
+--   increment; exact cadastral + parcel-containment deferred (cad_parcels FDW not exposed).
+--
+-- Pure internal DB op (one FDW read + local UPDATE) — no external HTTP, no anti-bot.
+--   SAFE to enable by default (enabled=true), like listing_source_snapshot / refresh_search_matview.
+--
+-- Window 09:00-10:00 UTC (12:00-13:00 MSK):
+--   - After geocode_missing_listings (06:00-09:00 UTC) → listings have fresh lat/lon/geom.
+--   - Before/independent of avito_detail_backfill proxy work; pure-DB so no contention.
+--
+-- default_params:
+--   threshold_m   -- max nearest-building distance to accept a match (metres). 50 = balance
+--                    of coverage vs precision for street-level geocodes (see PR EXPLAIN evidence).
+--   batch_size    -- listings updated per LATERAL UPDATE statement (chunked to bound lock/WAL).
+--   only_missing  -- if true (default) only fills WHERE building_cadastral_number IS NULL
+--                    (idempotent: a re-run is a no-op once filled). false = re-match all.
+--
+-- next_run_at = tomorrow 09:00 UTC (NOT NULL — avoid deploy-time fire, follow avito/cian seed
+--   convention; NULL would make get_due_schedules pick it up on the very next tick).
+--
+-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)), 124_cad_buildings_local.sql.
+-- Idempotent: ON CONFLICT (source) DO NOTHING.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'cadastral_geo_match',
+    true,
+    9,
+    10,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 9)) AT TIME ZONE 'UTC',
+    '{"threshold_m": 50, "batch_size": 5000, "only_missing": true}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMIT;

tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py

@@ -0,0 +1,384 @@
+"""Tests for the geo-nearest cadastral building matcher (#cadastral-geo-match).
+
+Convention mirrors test_listing_source_snapshot / test_asking_to_sold_ratio: the matcher
+is SQL-heavy and the gate has no live Postgres, so most assertions are STATIC — we read the
+emitted SQL via the text() clauses and inspect.getsource and check:
+  - the LATERAL KNN shape (ST_DWithin geography gate + `<->` KNN order + LIMIT 1),
+  - the threshold/only_missing params are bound (psycopg-v3 CAST discipline, no :p::type),
+  - the refresh does a single bulk FDW scan (TRUNCATE + INSERT ... FROM gendesign_cad_buildings),
+  - scheduler wiring (trigger fn + dispatch branch),
+  - migration 124 (table + GIST) and 125 (schedule seed) contents.
+
+Plus a fake-db behavioural test driving the chunk-loop + counter logic without Postgres.
+
+An OPTIONAL real-PostGIS behavioural test (test_real_knn_*) asserts the actual
+nearest-within-threshold / beyond-threshold→NULL semantics when a PostGIS database is
+reachable; it self-SKIPS otherwise so the hermetic gate never depends on it.
+"""
+
+from __future__ import annotations
+
+import inspect
+import math
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import scheduler
+from app.tasks import cadastral_geo_match as cgm
+
+_SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql"
+_MIGRATION_124 = _SQL_DIR / "124_cad_buildings_local.sql"
+_MIGRATION_125 = _SQL_DIR / "125_scrape_schedules_seed_cadastral_geo_match.sql"
+
+_MATCH_SQL = str(cgm._MATCH_SQL.text)
+_CANDIDATES_SQL = str(cgm._CANDIDATES_SQL.text)
+_REFRESH_SRC = inspect.getsource(cgm.refresh_cad_buildings_local)
+_MATCH_SRC = inspect.getsource(cgm.match_listings_to_buildings)
+
+
+# ── Refresh: single bulk FDW scan ─────────────────────────────────────────────
+
+
+def test_refresh_truncates_then_bulk_inserts_from_fdw() -> None:
+    assert "TRUNCATE cad_buildings_local" in _REFRESH_SRC
+    assert "INSERT INTO cad_buildings_local" in _REFRESH_SRC
+    assert "FROM gendesign_cad_buildings" in _REFRESH_SRC
+    # Builds a Point(4326) geom from lon/lat in the same scan.
+    assert "ST_SetSRID(ST_MakePoint(lon, lat), 4326)" in _REFRESH_SRC
+
+
+def test_refresh_filters_null_coords_and_cad_num() -> None:
+    assert "lat IS NOT NULL" in _REFRESH_SRC
+    assert "lon IS NOT NULL" in _REFRESH_SRC
+    assert "cad_num IS NOT NULL" in _REFRESH_SRC
+
+
+def test_refresh_is_single_scan_not_per_row() -> None:
+    """The FDW must be read once (one INSERT...SELECT), never per listing — the whole point."""
+    assert _REFRESH_SRC.count("FROM gendesign_cad_buildings") == 1
+
+
+# ── Match: LATERAL KNN shape ──────────────────────────────────────────────────
+
+
+def test_match_uses_lateral_knn_nearest_one() -> None:
+    flat = re.sub(r"\s+", " ", _MATCH_SQL)
+    assert "JOIN LATERAL" in flat
+    # GIST-backed KNN order + single nearest candidate.
+    assert "ORDER BY cb.geom <-> l.geom" in flat
+    assert "LIMIT 1" in flat
+
+
+def test_match_gates_with_geometry_dwithin_and_distancesphere_recheck() -> None:
+    """Perf-critical: geometry-space ST_DWithin (GIST, degrees) pre-gate + ST_DistanceSphere
+    metric recheck — NOT a geography-cast ST_DWithin in the WHERE (that ran 58s+ unfinished)."""
+    flat = re.sub(r"\s+", " ", _MATCH_SQL)
+    # GIST-indexable degree-space gate (no geography cast on the filtered table).
+    assert "ST_DWithin(cb.geom, l.geom, CAST(:deg_gate AS double precision))" in flat
+    # The geography cast must NOT appear in the gate (that defeats the index).
+    assert "CAST(cb.geom AS geography)" not in flat
+    # True metric distance computed only on the single nearest row, gated by threshold_m.
+    assert "ST_DistanceSphere(cb.geom, l.geom) AS dist_m" in flat
+    assert "m.dist_m <= CAST(:threshold_m AS double precision)" in flat
+
+
+def test_match_targets_building_cadastral_number() -> None:
+    flat = re.sub(r"\s+", " ", _MATCH_SQL)
+    assert "UPDATE listings" in flat
+    assert "SET building_cadastral_number = matched.cad_num" in flat
+
+
+def test_match_only_missing_param_is_bound_and_filters() -> None:
+    """only_missing=true → only fills WHERE building_cadastral_number IS NULL (idempotent)."""
+    flat = re.sub(r"\s+", " ", _MATCH_SQL)
+    assert "CAST(:only_missing AS boolean)" in flat
+    assert "l.building_cadastral_number IS NULL" in flat
+    # geo-present gate so non-geocoded listings never match.
+    assert "l.geom IS NOT NULL" in flat
+
+
+def test_match_is_single_statement_no_offset() -> None:
+    # OFFSET-paging over the mutated `building_cadastral_number IS NULL` predicate skips
+    # un-processed rows — the matcher must be ONE set-based UPDATE, no OFFSET/LIMIT paging.
+    flat = re.sub(r"\s+", " ", _MATCH_SQL)
+    assert "OFFSET" not in flat
+    assert "ST_DistanceSphere" in flat
+    assert "UPDATE listings l" in flat
+
+
+def test_deg_gate_encloses_threshold() -> None:
+    """The degree gate must be a SUPERSET of the metric threshold at EKB latitude.
+
+    A point exactly threshold_m metres away (in the tighter longitude axis) must fall
+    INSIDE the gate, else the GIST pre-filter would drop true-positive nearest buildings.
+    """
+    m_per_deg_lon = cgm._M_PER_DEG_LAT * math.cos(math.radians(cgm._EKB_LAT_DEG))
+    for threshold_m in (30, 50, 100):
+        deg_gate = cgm._deg_gate_for(float(threshold_m))
+        # Longitude-axis metres covered by the gate must exceed the threshold.
+        assert deg_gate * m_per_deg_lon >= threshold_m
+        # And by the latitude axis too (lat degrees are longer → even more margin).
+        assert deg_gate * cgm._M_PER_DEG_LAT >= threshold_m
+
+
+def test_no_psycopg_v3_colon_colon_cast() -> None:
+    """psycopg v3: never :param::type — must use CAST(:param AS type)."""
+    # No ':name::type' bound-param cast anywhere in the app SQL (the v3 trap).
+    assert not re.search(r":\w+::", _MATCH_SQL)
+    assert not re.search(r":\w+::", _CANDIDATES_SQL)
+    assert not re.search(r":\w+::", _REFRESH_SRC)
+
+
+def test_matcher_docstring_marks_approximation_and_threshold() -> None:
+    """Spec: the geo-nearest approximation must be explicit + threshold logged."""
+    doc = cgm.__doc__ or ""
+    assert "GEO-NEAREST" in doc
+    match_doc = cgm.match_listings_to_buildings.__doc__ or ""
+    assert "GEO-NEAREST" in match_doc
+    # threshold logged in match body
+    assert "threshold_m=%d" in _MATCH_SRC
+
+
+# ── Scheduler wiring ──────────────────────────────────────────────────────────
+
+
+def test_scheduler_has_trigger_and_dispatch() -> None:
+    assert hasattr(scheduler, "trigger_cadastral_geo_match_run")
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source == "cadastral_geo_match"' in loop_src
+    assert "trigger_cadastral_geo_match_run(db, sch)" in loop_src
+
+
+def test_trigger_claims_run_and_runs_in_executor() -> None:
+    src = inspect.getsource(scheduler.trigger_cadastral_geo_match_run)
+    assert "_claim_run(db, schedule_row)" in src
+    # sync DB-only task → run_in_executor (not a bare async call).
+    assert "run_in_executor" in src
+    assert "run_cadastral_geo_match" in src
+
+
+# ── Migration content ─────────────────────────────────────────────────────────
+
+
+def test_migration_124_creates_table_and_gist() -> None:
+    sql = _MIGRATION_124.read_text(encoding="utf-8")
+    assert "CREATE TABLE IF NOT EXISTS cad_buildings_local" in sql
+    assert "cad_num          text PRIMARY KEY" in sql
+    assert "geom             geometry(Point, 4326)" in sql
+    assert "USING GIST (geom)" in sql
+    assert "IF NOT EXISTS cad_buildings_local_geom_idx" in sql
+    assert "BEGIN;" in sql and "COMMIT;" in sql
+    # The migration must NOT read the FDW (deploy independence) — no FROM/INSERT against it.
+    # (The FDW name may appear in comments; only executable references are forbidden.)
+    code_lines = [ln for ln in sql.splitlines() if not ln.lstrip().startswith("--")]
+    code = "\n".join(code_lines)
+    assert "FROM gendesign_cad_buildings" not in code
+    assert "INSERT INTO cad_buildings_local" not in code  # populate is the refresh job, not DDL
+
+
+def test_migration_125_seeds_schedule_not_null_next_run() -> None:
+    sql = _MIGRATION_125.read_text(encoding="utf-8")
+    assert "INSERT INTO scrape_schedules" in sql
+    assert "'cadastral_geo_match'" in sql
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+    # next_run_at NOT NULL = tomorrow (avoid deploy-time fire).
+    assert "CURRENT_DATE + INTERVAL '1 day'" in sql
+    assert '"threshold_m": 50' in sql
+    assert "BEGIN;" in sql and "COMMIT;" in sql
+
+
+# ── Fake-db behavioural: chunk loop + counters ───────────────────────────────
+
+
+class _FakeResult:
+    def __init__(self, rowcount: int, scalar: Any = None) -> None:
+        self.rowcount = rowcount
+        self._scalar = scalar
+
+    def scalar(self) -> Any:
+        return self._scalar
+
+
+class _FakeDB:
+    """Minimal Session stand-in: scripts execute() returns queued results in order."""
+
+    def __init__(self, results: list[_FakeResult]) -> None:
+        self._results = list(results)
+        self.executed: list[str] = []
+        self.commits = 0
+
+    def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
+        self.executed.append(str(getattr(clause, "text", clause)))
+        return self._results.pop(0)
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:  # pragma: no cover
+        pass
+
+
+def test_match_runs_single_update_and_counts(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Single set-based UPDATE: one execute → rowcount is the match count, one commit."""
+    db = _FakeDB([_FakeResult(5)])
+    total = cgm.match_listings_to_buildings(db, threshold_m=50, batch_size=10, only_missing=True)
+    assert total == 5
+    assert db.commits == 1
+    assert len(db.executed) == 1  # exactly one UPDATE — no OFFSET paging
+
+
+def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None:
+    """run_cadastral_geo_match: refresh → candidates → match → mark_done(counters)."""
+    marked: dict[str, Any] = {}
+    monkeypatch.setattr(cgm.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(
+        cgm.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(cgm.runs_mod, "mark_failed", lambda *a, **k: None)
+
+    # Stub the heavy SQL fns; assert the wrapper plumbs counters correctly.
+    monkeypatch.setattr(cgm, "refresh_cad_buildings_local", lambda _db: 36732)
+    monkeypatch.setattr(
+        cgm,
+        "match_listings_to_buildings",
+        lambda _db, **k: 12000,
+    )
+
+    class _CandDB:
+        def execute(self, *a: Any, **k: Any) -> _FakeResult:
+            return _FakeResult(0, scalar=40000)
+
+    out = cgm.run_cadastral_geo_match(
+        _CandDB(),  # type: ignore[arg-type]
+        run_id=7,
+        params={"threshold_m": 50, "batch_size": 100, "only_missing": True},
+    )
+    assert out.buildings_loaded == 36732
+    assert out.candidates_total == 40000
+    assert out.listings_matched == 12000
+    assert out.threshold_m == 50
+    assert marked["run_id"] == 7
+    assert marked["counters"]["listings_matched"] == 12000
+    assert marked["counters"]["threshold_m"] == 50
+
+
+def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    failed: dict[str, Any] = {}
+    monkeypatch.setattr(cgm.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(cgm.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(
+        cgm.runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
+    )
+
+    def _boom(_db: Any) -> int:
+        raise RuntimeError("fdw down")
+
+    monkeypatch.setattr(cgm, "refresh_cad_buildings_local", _boom)
+
+    class _DB:
+        def rollback(self) -> None:
+            pass
+
+    with pytest.raises(RuntimeError):
+        cgm.run_cadastral_geo_match(_DB(), run_id=9, params={})  # type: ignore[arg-type]
+    assert failed["run_id"] == 9
+    assert "fdw down" in failed["err"]
+
+
+# ── Optional real-PostGIS behavioural test (self-skips without a DB) ──────────
+
+
+def _live_session() -> Any | None:
+    """Return a SQLAlchemy Session if a PostGIS DB is reachable, else None."""
+    try:
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+        if not dsn or "localhost:5432/test" in dsn:
+            return None
+        engine = create_engine(dsn, future=True)
+        conn = engine.connect()
+        conn.execute(__import__("sqlalchemy").text("SELECT PostGIS_Version()"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable PostGIS test DB")
+def test_real_knn_nearest_within_threshold_picked() -> None:
+    """With a real PostGIS DB: nearest building within threshold is picked; beyond → NULL."""
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    try:
+        db.execute(
+            _t(
+                "CREATE TEMP TABLE cad_buildings_local "
+                "(LIKE cad_buildings_local INCLUDING ALL) ON COMMIT DROP"
+            )
+        )
+    except Exception:
+        # Local table may not exist in this DB → build a minimal temp equivalent.
+        db.rollback()
+        db.execute(
+            _t(
+                "CREATE TEMP TABLE cad_buildings_local ("
+                "cad_num text PRIMARY KEY, geom geometry(Point,4326)) ON COMMIT DROP"
+            )
+        )
+    # Two buildings: one 10m from the listing, one 500m away.
+    db.execute(
+        _t(
+            "INSERT INTO cad_buildings_local (cad_num, geom) VALUES "
+            "('66:01:NEAR', ST_SetSRID(ST_MakePoint(60.6000, 56.8380),4326)),"
+            "('66:01:FAR',  ST_SetSRID(ST_MakePoint(60.6100, 56.8380),4326))"
+        )
+    )
+    # Same geometry-gate + ST_DistanceSphere recheck shape as production.
+    deg_gate = cgm._deg_gate_for(50.0)
+    # listing point ~ at NEAR; nearest within 50m must be 66:01:NEAR.
+    row = db.execute(
+        _t(
+            "SELECT cad_num FROM ("
+            "  SELECT cb.cad_num, "
+            "         ST_DistanceSphere(cb.geom, ST_SetSRID(ST_MakePoint(60.6001,56.8380),4326)) "
+            "             AS dist_m "
+            "  FROM cad_buildings_local cb "
+            "  WHERE ST_DWithin(cb.geom, ST_SetSRID(ST_MakePoint(60.6001,56.8380),4326), :g) "
+            "  ORDER BY cb.geom <-> ST_SetSRID(ST_MakePoint(60.6001,56.8380),4326) LIMIT 1"
+            ") m WHERE m.dist_m <= 50"
+        ),
+        {"g": deg_gate},
+    ).fetchone()
+    assert row is not None and row[0] == "66:01:NEAR"
+    # A point 500m+ from any building → no match within 50m.
+    none_row = db.execute(
+        _t(
+            "SELECT cad_num FROM ("
+            "  SELECT cb.cad_num, "
+            "         ST_DistanceSphere(cb.geom, ST_SetSRID(ST_MakePoint(60.7000,56.9000),4326)) "
+            "             AS dist_m "
+            "  FROM cad_buildings_local cb "
+            "  WHERE ST_DWithin(cb.geom, ST_SetSRID(ST_MakePoint(60.7000,56.9000),4326), :g) "
+            "  ORDER BY cb.geom <-> ST_SetSRID(ST_MakePoint(60.7000,56.9000),4326) LIMIT 1"
+            ") m WHERE m.dist_m <= 50"
+        ),
+        {"g": deg_gate},
+    ).fetchone()
+    assert none_row is None
+    db.rollback()
+    db.close()