feat(tradein): systemd timer scheduler — trigger script + units + SCHEDULER_ENABLE flag (#581)

tradein-mvp/backend/app/core/config.py

@@ -12,6 +12,12 @@ class Settings(BaseSettings):
     # required — задаётся через env DATABASE_URL. Нет дефолта: fail-fast при старте
     # если переменная не задана (C-3 security audit).
     database_url: str
+
+    # In-app asyncio scheduler enable flag (#581).
+    # Set SCHEDULER_ENABLE=false when using the systemd timer trigger instead.
+    # Default true preserves existing behaviour.
+    scheduler_enable: bool = Field(default=True, validation_alias="SCHEDULER_ENABLE")
+
     cors_origins: list[str] = ["http://localhost", "http://localhost:3000", "http://localhost:8080"]
     environment: str = "dev"
 

tradein-mvp/backend/app/main.py

@@ -74,17 +74,23 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     except Exception:
         logger.exception("FDW user mapping bootstrap failed — cadastral queries may fail")
 
-    # Startup: launch scheduler background task
-    scheduler_task = asyncio.create_task(scheduler_loop())
-    logger.info("FastAPI lifespan: scheduler task spawned")
+    # Startup: launch scheduler background task (disabled when SCHEDULER_ENABLE=false,
+    # e.g. when using the systemd timer trigger — see tradein-mvp/ops/systemd/ #581).
+    scheduler_task = None
+    if settings.scheduler_enable:
+        scheduler_task = asyncio.create_task(scheduler_loop())
+        logger.info("FastAPI lifespan: scheduler task spawned")
+    else:
+        logger.info("FastAPI lifespan: in-app scheduler disabled (SCHEDULER_ENABLE=false)")
     yield
-    # Shutdown: cancel scheduler
-    scheduler_task.cancel()
-    try:
-        await scheduler_task
-    except asyncio.CancelledError:
-        pass
-    logger.info("FastAPI lifespan: scheduler cancelled cleanly")
+    # Shutdown: cancel scheduler if it was started
+    if scheduler_task is not None:
+        scheduler_task.cancel()
+        try:
+            await scheduler_task
+        except asyncio.CancelledError:
+            pass
+        logger.info("FastAPI lifespan: scheduler cancelled cleanly")
 
 
 app = FastAPI(

tradein-mvp/ops/systemd/README.md

@@ -0,0 +1,57 @@
+# TradeIn systemd scrape trigger
+
+Replaces the in-app asyncio scheduler with a systemd timer that calls the admin API every 60s.
+
+## Install
+
+Run as root on the VPS after deploy:
+
+```bash
+cd /opt/gendesign/tradein-mvp/ops/systemd
+bash install.sh
+```
+
+`install.sh` copies the unit files to `/etc/systemd/system/`, reloads systemd, and
+enables + starts the timer.
+
+## Environment variables
+
+Create `/etc/tradein/trigger.env` before enabling the timer:
+
+```ini
+# psycopg v3 connection string (postgresql://... or postgresql+psycopg://...)
+TRADEIN_DATABASE_URL=postgresql://tradein:<password>@localhost:5432/tradein
+
+# Admin username from roles.yaml with role=admin — passed as X-Authenticated-User header
+TRADEIN_ADMIN_TOKEN=admin
+
+# Backend base URL (no trailing slash). Default: http://localhost:8000
+TRADEIN_API_BASE=http://localhost:8000
+```
+
+## Diagnostic commands
+
+```bash
+# Follow live timer logs
+journalctl -u tradein-scrape-trigger -f
+
+# List all active timers and next trigger times
+systemctl list-timers
+
+# Check timer status
+systemctl status tradein-scrape-trigger.timer
+
+# Check last service run
+systemctl status tradein-scrape-trigger.service
+
+# Manual one-shot trigger (for testing)
+systemctl start tradein-scrape-trigger.service
+```
+
+## Notes
+
+- The script uses `flock` on `/var/run/tradein-trigger.lock` to prevent overlapping
+  runs if the previous invocation is still running.
+- Sources without a dedicated admin endpoint (e.g. `rosreestr_dkp_import`,
+  `listing_source_snapshot`) are logged as warnings and skipped. Set
+  `SCHEDULER_ENABLE=true` in the backend env to handle them via the in-app scheduler.

tradein-mvp/ops/systemd/install.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+# Run as root on VPS after deploy.
+# Installs (or updates) the systemd timer that triggers scrape schedules every 60s.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+cp "$SCRIPT_DIR/tradein-scrape-trigger.service" /etc/systemd/system/
+cp "$SCRIPT_DIR/tradein-scrape-trigger.timer" /etc/systemd/system/
+systemctl daemon-reload
+systemctl enable --now tradein-scrape-trigger.timer
+systemctl status tradein-scrape-trigger.timer

tradein-mvp/ops/systemd/tradein-scrape-trigger.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=TradeIn Scrape Schedule Trigger
+After=network-online.target docker.service
+
+[Service]
+Type=oneshot
+User=root
+EnvironmentFile=/etc/tradein/trigger.env
+ExecStart=/usr/bin/python3 /opt/gendesign/tradein-mvp/scripts/trigger-schedules.py
+StandardOutput=journal
+StandardError=journal
+TimeoutStartSec=55
+
+[Install]
+WantedBy=multi-user.target

tradein-mvp/ops/systemd/tradein-scrape-trigger.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=TradeIn Scrape Trigger — every 60s
+Requires=tradein-scrape-trigger.service
+
+[Timer]
+OnBootSec=30s
+OnUnitActiveSec=60s
+AccuracySec=5s
+
+[Install]
+WantedBy=timers.target

tradein-mvp/scripts/trigger-schedules.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""Trigger due scrape schedules via admin API. Called by systemd timer every 60s."""
+
+from __future__ import annotations
+
+import fcntl
+import logging
+import os
+import random
+import sys
+import time
+
+import httpx
+import psycopg
+
+LOCK_FILE = "/var/run/tradein-trigger.lock"
+DB_URL = os.environ["TRADEIN_DATABASE_URL"]
+API_BASE = os.environ.get("TRADEIN_API_BASE", "http://localhost:8000")
+# Username that maps to role=admin in roles.yaml — passed as X-Authenticated-User.
+ADMIN_TOKEN = os.environ["TRADEIN_ADMIN_TOKEN"]
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger(__name__)
+
+# Mapping from scrape_schedules.source → admin API path.
+# Sources handled entirely inside the scheduler loop (no separate admin endpoint)
+# are listed in _NO_ENDPOINT_SOURCES: they are skipped when the systemd trigger
+# runs, and require SCHEDULER_ENABLE=true to fire.
+_SOURCE_TO_PATH: dict[str, str] = {
+    "avito_city_sweep": "/api/v1/admin/scrape/avito-city-sweep",
+    "yandex_city_sweep": "/api/v1/admin/scrape/yandex-city-sweep",
+    "n1_city_sweep": "/api/v1/admin/scrape/n1",
+    "cian_history_backfill": "/api/v1/admin/scrape/cian-backfill-history",
+    "yandex_address_backfill": "/api/v1/admin/scrape/yandex-address-backfill",
+}
+
+# Sources with no dedicated admin trigger endpoint.
+_NO_ENDPOINT_SOURCES = frozenset(
+    {
+        "rosreestr_dkp_import",
+        "listing_source_snapshot",
+        "asking_to_sold_ratio_refresh",
+        "refresh_search_matview",
+        "deactivate_stale_avito",
+        "sber_index_pull",
+        "rosreestr_quarter_poll",
+    }
+)
+
+_MAX_RETRIES = 3
+_RETRY_BASE_S = 2.0
+
+
+def _normalize_db_url(url: str) -> str:
+    """Strip SQLAlchemy driver prefix so psycopg v3 accepts the connection string."""
+    if url.startswith("postgresql+psycopg://"):
+        return "postgresql://" + url[len("postgresql+psycopg://"):]
+    return url
+
+
+def get_due_schedules() -> list[dict[str, str]]:
+    """Fetch scrape_schedules rows that are due for execution."""
+    conn_url = _normalize_db_url(DB_URL)
+    with psycopg.connect(conn_url) as conn:
+        rows = conn.execute(
+            """
+            SELECT id, source, schedule_type
+            FROM scrape_schedules
+            WHERE enabled = true
+              AND (next_run_at IS NULL OR next_run_at <= NOW())
+            """,
+        ).fetchall()
+        description = conn.execute(
+            "SELECT id, source, schedule_type FROM scrape_schedules LIMIT 0"
+        ).description or []
+    col_names = [desc.name for desc in description]
+    if not col_names:
+        col_names = ["id", "source", "schedule_type"]
+    return [
+        {col_names[i]: (str(row[i]) if row[i] is not None else "") for i in range(len(col_names))}
+        for row in rows
+    ]
+
+
+def trigger_source(client: httpx.Client, source: str) -> None:
+    """POST to the admin trigger endpoint for a given source. Retries on connection error."""
+    path = _SOURCE_TO_PATH.get(source)
+    if path is None:
+        if source in _NO_ENDPOINT_SOURCES:
+            log.warning(
+                "trigger: source=%s has no admin API endpoint — "
+                "ensure SCHEDULER_ENABLE=true or add a dedicated admin endpoint",
+                source,
+            )
+        else:
+            log.warning("trigger: unknown source=%s, skipping", source)
+        return
+
+    url = f"{API_BASE}{path}"
+    last_exc: Exception | None = None
+
+    for attempt in range(1, _MAX_RETRIES + 1):
+        try:
+            resp = client.post(
+                url,
+                headers={"X-Authenticated-User": ADMIN_TOKEN},
+                timeout=50.0,
+            )
+            log.info(
+                "trigger: source=%s url=%s status=%d attempt=%d",
+                source,
+                url,
+                resp.status_code,
+                attempt,
+            )
+            if resp.status_code >= 500:
+                log.warning(
+                    "trigger: source=%s got HTTP %d — body: %s",
+                    source,
+                    resp.status_code,
+                    resp.text[:200],
+                )
+            return
+        except httpx.ConnectError as exc:
+            last_exc = exc
+            wait = _RETRY_BASE_S * (2 ** (attempt - 1))
+            log.warning(
+                "trigger: source=%s connection error (attempt %d/%d), "
+                "retrying in %.1fs: %s",
+                source,
+                attempt,
+                _MAX_RETRIES,
+                wait,
+                exc,
+            )
+            if attempt < _MAX_RETRIES:
+                time.sleep(wait)
+
+    log.error(
+        "trigger: source=%s failed after %d attempts: %s",
+        source,
+        _MAX_RETRIES,
+        last_exc,
+    )
+
+
+def main() -> int:
+    # Exclusive non-blocking flock: if a previous run is still active, exit immediately
+    # with code 0 so systemd does not log a failure.
+    lock_fd = open(LOCK_FILE, "w")
+    try:
+        fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        log.info("trigger: lock busy — previous run still active, exiting")
+        lock_fd.close()
+        return 0
+
+    try:
+        due = get_due_schedules()
+        if not due:
+            log.info("trigger: no schedules due")
+            return 0
+
+        log.info("trigger: %d schedule(s) due", len(due))
+
+        with httpx.Client() as client:
+            for i, row in enumerate(due):
+                source = row.get("source", "")
+                trigger_source(client, source)
+                if i < len(due) - 1:
+                    # Stagger requests to avoid thundering herd on the API
+                    time.sleep(random.uniform(1, 4))
+
+        return 0
+
+    except Exception:
+        log.exception("trigger: unexpected error")
+        return 1
+    finally:
+        try:
+            fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN)
+        except Exception:
+            pass
+        lock_fd.close()
+
+
+if __name__ == "__main__":
+    sys.exit(main())