feat(db): quarter_price_index FDW foreign table + monthly refresh (Refs #762) (#797)

Co-authored-by: bot-backend <bot-backend@gendsgn.local>
Co-committed-by: bot-backend <bot-backend@gendsgn.local>

backend/app/services/site_finder/quarter_price_index_refresh.py

@@ -0,0 +1,85 @@
+"""Refresh helper for mv_quarter_price_index (Issue #762).
+
+mv_quarter_price_index depends on mv_quarter_price_per_m2, so both must be
+refreshed in order:
+  1. mv_quarter_price_per_m2  (source MV — deals aggregation)
+  2. mv_quarter_price_index   (derived MV — price index normalised to city median)
+
+Scheduled via Celery beat hardcoded entry in workers/beat_schedule.py.
+Cadence: monthly on the 5th at 05:00 MSK (after refresh_ekb_districts_medians
+which runs at 04:00 MSK on the same day — deals data is already settled).
+
+Usage example (manual, via psql-connected shell or admin endpoint):
+    from sqlalchemy.orm import Session
+    from app.services.site_finder.quarter_price_index_refresh import refresh_quarter_price_index
+
+    count = refresh_quarter_price_index(db)
+    # logs: "mv_quarter_price_index refreshed: 1972 rows"
+"""
+
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def _refresh_mv(db: Session, mv_name: str, *, concurrently: bool) -> None:
+    """Run REFRESH MATERIALIZED VIEW [CONCURRENTLY] <mv_name>.
+
+    Falls back to non-concurrent on the known "cannot refresh concurrently"
+    error (MV empty or no UNIQUE index — should not happen in prod, but
+    provides a safe recovery path for first-run edge cases).
+    """
+    try:
+        if concurrently:
+            db.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {mv_name}"))
+        else:
+            db.execute(text(f"REFRESH MATERIALIZED VIEW {mv_name}"))
+        db.commit()
+    except OperationalError as e:
+        if concurrently and "cannot refresh materialized view" in str(e).lower():
+            logger.warning(
+                "%s: CONCURRENTLY failed (MV likely not populated), "
+                "falling back to non-concurrent refresh",
+                mv_name,
+            )
+            db.rollback()
+            db.execute(text(f"REFRESH MATERIALIZED VIEW {mv_name}"))
+            db.commit()
+        else:
+            raise
+
+
+def refresh_quarter_price_index(db: Session, *, concurrently: bool = True) -> int:
+    """Refresh mv_quarter_price_per_m2 then mv_quarter_price_index (in order).
+
+    mv_quarter_price_index reads from mv_quarter_price_per_m2, so the source MV
+    must be refreshed first. Both refreshes happen in the same call.
+
+    Args:
+        db: SQLAlchemy Session (sync).
+        concurrently: When True, uses REFRESH CONCURRENTLY for both MVs —
+            non-blocking (readers continue). Requires the respective UNIQUE
+            indexes (mv_quarter_price_pk on source, mv_quarter_price_index_uq
+            on derived) and both MVs to be already populated.
+            Pass False only for first populate or after MV recreation.
+
+    Returns:
+        Row count of mv_quarter_price_index after refresh (for observability).
+    """
+    # Step 1: refresh source MV (deals aggregation layer)
+    _refresh_mv(db, "mv_quarter_price_per_m2", concurrently=concurrently)
+    logger.info("mv_quarter_price_per_m2 refreshed (chain step 1/2)")
+
+    # Step 2: refresh derived MV (price index, reads from source)
+    _refresh_mv(db, "mv_quarter_price_index", concurrently=concurrently)
+
+    row = db.execute(text("SELECT COUNT(*) FROM mv_quarter_price_index")).first()
+    count = int(row[0]) if row else 0
+    logger.info("mv_quarter_price_index refreshed: %d rows (chain step 2/2)", count)
+    return count

backend/app/workers/beat_schedule.py

@@ -264,6 +264,16 @@ def build_beat_schedule() -> dict:
     #     "options": {"queue": "celery"},
     # }
 
+    # Quarter price index refresh (#762) — monthly on the 5th at 05:00 MSK (02:00 UTC).
+    # Chains: mv_quarter_price_per_m2 → mv_quarter_price_index (in order, both CONCURRENTLY).
+    # Runs 1 h after refresh-ekb-districts-medians (04:00 MSK) so deals data is settled.
+    # Not in job_settings (technical infra task, no UI config needed).
+    schedule["refresh-quarter-price-index"] = {
+        "task": "tasks.refresh_quarter_price_index.refresh_quarter_price_index_chain",
+        "schedule": _parse_cron("0 2 5 * *"),  # 02:00 UTC = 05:00 MSK, 5th of month
+        "options": {"queue": "celery"},
+    }
+
     # NSPD quarter dump refresh — re-enabled 2026-05-17 после Sub-PR B (#260)
     # переключения search_by_quarter на grid-walk. Foundation (#247) + integration
     # (#260) теперь возвращают полноценные dumps (territorial_zones, ЗОУИТ, risk

backend/app/workers/tasks/refresh_quarter_price_index.py

@@ -0,0 +1,45 @@
+"""Celery task: refresh mv_quarter_price_per_m2 → mv_quarter_price_index chain.
+
+Scheduled via hardcoded beat entry in workers/beat_schedule.py:
+  'refresh-quarter-price-index' — monthly on the 5th at 05:00 MSK (02:00 UTC).
+  Runs after 'refresh-ekb-districts-medians' (04:00 MSK same day) so that
+  deals data is settled before the price index is recomputed.
+
+Issue: #762.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.core.db import SessionLocal
+from app.services.site_finder.quarter_price_index_refresh import refresh_quarter_price_index
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(
+    bind=True,
+    name="tasks.refresh_quarter_price_index.refresh_quarter_price_index_chain",
+    max_retries=2,
+)
+def refresh_quarter_price_index_chain(self: Any) -> dict[str, Any]:
+    """Refresh mv_quarter_price_per_m2 then mv_quarter_price_index in sequence.
+
+    Both MVs are refreshed CONCURRENTLY (non-blocking). Falls back to
+    non-concurrent if the MV is found unpopulated (edge case on first run).
+
+    Returns result dict for Celery task result store / logging.
+    """
+    db = SessionLocal()
+    try:
+        count = refresh_quarter_price_index(db, concurrently=True)
+        logger.info("refresh_quarter_price_index_chain: completed, index rows=%d", count)
+        return {"status": "ok", "mv_quarter_price_index_rows": count}
+    except Exception as e:
+        logger.exception("refresh_quarter_price_index_chain failed: %s", e)
+        raise
+    finally:
+        db.close()

data/sql/99b_grant_quarter_price_index_fdw.sql

@@ -0,0 +1,16 @@
+-- 99b_grant_quarter_price_index_fdw.sql
+-- Issue #762 (#647-2) — grant SELECT on mv_quarter_price_index to the tradein
+-- FDW reader role so the tradein foreign table (085_quarter_price_index_fdw.sql)
+-- can read it. Mirrors 122_grant_rosreestr_to_fdw_reader.sql.
+--
+-- Ordering: '99b_' sorts AFTER '99a_quarter_price_index.sql' (which CREATEs the MV)
+-- because 'a' < 'b'. The MV must exist before the GRANT. Files 100_–122_ sort
+-- BEFORE 99_ lexicographically ('1' < '9'), so they cannot grant on this object.
+-- Idempotent: GRANT is repeatable; role tradein_fdw_reader created in
+-- 100_tradein_fdw_role.sql (REVOKE ALL baseline there).
+
+BEGIN;
+
+GRANT SELECT ON public.mv_quarter_price_index TO tradein_fdw_reader;
+
+COMMIT;

tradein-mvp/backend/data/sql/085_quarter_price_index_fdw.sql

@@ -0,0 +1,58 @@
+-- 085_quarter_price_index_fdw.sql
+-- Issue #762 — Foreign table over MAIN DB's mv_quarter_price_index MV.
+--
+-- Context:
+--   #760 (PR #792) created public.mv_quarter_price_index in the MAIN gendesign DB.
+--   The estimator (#647-3) needs O(1) lookup by quarter_cad_number without a
+--   cross-service HTTP round-trip. FDW foreign table lets tradein read the MV
+--   directly via the already-established gendesign_remote server.
+--
+-- Reused infra:
+--   SERVER gendesign_remote — created by 060_postgres_fdw_extension.sql.
+--   Pattern: per-table FOREIGN TABLE (not IMPORT FOREIGN SCHEMA) — matches repo convention.
+--   USER MAPPING is managed at backend startup from env (see 060_postgres_fdw_extension.sql).
+--
+-- Column mapping (verified against live MV via pg_attribute, 2026-05-30):
+--   quarter_cad_number  character varying(30)  → varchar(30) / text (FDW accepts text for varchar)
+--   price_index         double precision        → double precision
+--   n_deals             bigint                  → bigint
+--   basis               text                    → text
+--   computed_at         timestamp with time zone → timestamptz
+--
+-- Index note:
+--   The remote MV has UNIQUE index mv_quarter_price_index_uq on quarter_cad_number.
+--   FDW pushes equality predicates (WHERE quarter_cad_number = $1) down to MAIN,
+--   so remote index gives O(1) lookup. No local index on the foreign table is needed
+--   or possible (postgres_fdw does not support local indexes on foreign tables).
+--
+-- Idempotency:
+--   DROP FOREIGN TABLE IF EXISTS → CREATE is safe to re-apply.
+--   BEGIN/COMMIT per sql.md conventions.
+--
+-- Apply order:
+--   After 060_postgres_fdw_extension.sql (server must exist).
+--   After 99a_quarter_price_index.sql is applied in MAIN (MV must exist remotely).
+--   Before estimator code referencing this table is deployed (#647-3).
+
+BEGIN;
+
+DROP FOREIGN TABLE IF EXISTS quarter_price_index;
+
+CREATE FOREIGN TABLE quarter_price_index (
+    quarter_cad_number  varchar(30)                  NOT NULL,
+    price_index         double precision             NOT NULL,
+    n_deals             bigint                       NOT NULL,
+    basis               text                         NOT NULL,
+    computed_at         timestamp with time zone     NOT NULL
+)
+SERVER gendesign_remote
+OPTIONS (schema_name 'public', table_name 'mv_quarter_price_index');
+
+COMMENT ON FOREIGN TABLE quarter_price_index IS
+    'FDW read-only view of gendesign.public.mv_quarter_price_index. '
+    'Per-cadastral-quarter price index normalised to EKB city median (=1.0). '
+    'Lookup by quarter_cad_number is O(1) via remote UNIQUE index on MAIN. '
+    'Refresh cadence: MAIN refreshes MV via Celery beat (see quarter_price_index_refresh task). '
+    'Issue: #762. Source MV: #760.';
+
+COMMIT;