PolyWeather/web/services/system_api.py

"""System and observability API service functions."""

from __future__ import annotations

import time
import os
from collections import Counter
from datetime import datetime, timezone
from typing import Any, Dict, Optional

from fastapi import BackgroundTasks, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import PlainTextResponse
from loguru import logger

from src.database.db_manager import DBManager
from src.database.sqlite_connection import connect_sqlite
from src.utils.metrics import export_prometheus_metrics, gauge_set
from web.core import build_health_payload, build_system_status_payload
import web.routes as legacy_routes


def get_health_payload() -> Dict[str, Any]:
    return build_health_payload()


async def get_system_status_payload(request: Request) -> Dict[str, Any]:
    legacy_routes._require_ops_admin(request)
    payload = await run_in_threadpool(build_system_status_payload)
    payload["realtime"] = await run_in_threadpool(_realtime_status_payload)
    return payload


def _realtime_status_payload() -> Dict[str, Any]:
    try:
        from web.routers import sse_router

        store = sse_router.event_store
        status_fn = getattr(store, "status", None)
        if callable(status_fn):
            status = dict(status_fn())
        else:
            store_name = "degraded_sqlite" if getattr(store, "degraded_from", None) == "redis" else "sqlite"
            status = {
                "store": store_name,
                "latest_revision": int(store.latest_revision()),
            }
        connection_count = getattr(sse_router.sse_manager, "connection_count", None)
        status["sse_connections"] = int(connection_count()) if callable(connection_count) else 0
        return status
    except Exception as exc:
        return {
            "store": "unknown",
            "latest_revision": 0,
            "sse_connections": 0,
            "error": str(exc),
        }


def get_system_cache_status(request: Request, cities: Optional[str] = None) -> Dict[str, Any]:
    legacy_routes._require_ops_admin(request)
    selected = legacy_routes._normalize_city_list(cities)
    if not selected:
        selected = legacy_routes._normalize_city_list(None)
    kinds = {
        "summary": legacy_routes.CITY_SUMMARY_CACHE_TTL_SEC,
        "panel": legacy_routes.CITY_PANEL_CACHE_TTL_SEC,
        "nearby": legacy_routes.CITY_NEARBY_CACHE_TTL_SEC,
        "market": legacy_routes.CITY_MARKET_CACHE_TTL_SEC,
        "full": legacy_routes.CITY_FULL_CACHE_TTL_SEC,
    }
    items = []
    for city in selected:
        row = {"city": city}
        for kind, ttl_sec in kinds.items():
            entry = legacy_routes._CACHE_DB.get_city_cache(kind, city)
            row[kind] = {
                "exists": bool(entry),
                "fresh": legacy_routes._city_cache_is_fresh(entry, ttl_sec),
                "updated_at": entry.get("updated_at") if entry else None,
                "age_sec": round(max(0.0, time.time() - float(entry.get("updated_at_ts") or 0.0)), 1)
                if entry
                else None,
                "ttl_sec": ttl_sec,
            }
        items.append(row)
    return {"cities": items}


def run_system_priority_warm(
    request: Request,
    background_tasks: BackgroundTasks,
    timezone: Optional[str] = None,
) -> Dict[str, Any]:
    legacy_routes._assert_entitlement(request)
    batches = legacy_routes._select_priority_city_batches(timezone)
    primary = list(batches.get("primary") or [])
    secondary = list(batches.get("secondary") or [])

    def _queue_city_refresh(city: str, *, priority: str) -> None:
        enqueue = getattr(legacy_routes._CACHE_DB, "enqueue_observation_refresh_request", None)
        if not callable(enqueue):
            logger.warning("priority warm queue unavailable city={} timezone={}", city, timezone)
            return
        enqueue(
            city=city,
            kind="panel",
            priority=priority,
            reason="system_priority_warm",
        )

    def _runner() -> None:
        for city in primary:
            try:
                _queue_city_refresh(city, priority="high")
            except Exception as exc:
                logger.warning("priority warm primary failed city={} timezone={}: {}", city, timezone, exc)
        for city in secondary:
            try:
                _queue_city_refresh(city, priority="normal")
            except Exception as exc:
                logger.warning("priority warm secondary failed city={} timezone={}: {}", city, timezone, exc)

    background_tasks.add_task(_runner)
    return {
        "ok": True,
        "region": batches.get("region"),
        "timezone": batches.get("timezone"),
        "primary": primary,
        "secondary": secondary,
    }


def get_prometheus_metrics_response(request: Request) -> PlainTextResponse:
    legacy_routes._require_ops_admin(request)
    _refresh_operational_metrics()
    return PlainTextResponse(
        export_prometheus_metrics(),
        media_type="text/plain; version=0.0.4; charset=utf-8",
    )


def _payment_event_reason(row: Dict[str, Any]) -> str:
    payload = row.get("payload") if isinstance(row, dict) else {}
    payload = payload if isinstance(payload, dict) else {}
    confirm_failure = (
        payload.get("confirm_failure")
        if isinstance(payload.get("confirm_failure"), dict)
        else {}
    )
    return str(
        payload.get("reason")
        or confirm_failure.get("reason")
        or payload.get("error")
        or "unknown"
    ).strip().lower() or "unknown"


def _payment_event_is_resolved(row: Dict[str, Any]) -> bool:
    payload = row.get("payload") if isinstance(row, dict) else {}
    payload = payload if isinstance(payload, dict) else {}
    return bool(str(payload.get("resolved_at") or "").strip())


def _refresh_operational_metrics() -> None:
    try:
        db = DBManager()
    except Exception:
        return

    payment_events = []
    for event_type in ("payment_intent_failed", "payment_refund_required"):
        try:
            payment_events.extend(
                db.list_payment_audit_events(limit=500, event_type=event_type)
            )
        except Exception:
            continue

    open_events = [row for row in payment_events if not _payment_event_is_resolved(row)]
    gauge_set("polyweather_payment_incidents_open", len(open_events))
    for reason, count in Counter(_payment_event_reason(row) for row in open_events).items():
        gauge_set("polyweather_payment_incidents_by_reason", count, reason=reason)

    try:
        refund_cases = db.list_refund_cases(limit=500)
    except Exception:
        refund_cases = []
    terminal_statuses = {"refunded", "rejected", "closed"}
    open_refunds = [
        case
        for case in refund_cases
        if str(case.get("status") or "").strip().lower() not in terminal_statuses
    ]
    gauge_set("polyweather_refund_cases_open", len(open_refunds))

    realtime = _realtime_status_payload()
    gauge_set("polyweather_sse_connections", int(realtime.get("sse_connections") or 0))
    gauge_set(
        "polyweather_realtime_latest_revision",
        int(realtime.get("latest_revision") or 0),
    )
    degraded_from = str(realtime.get("degraded_from") or "").strip().lower()
    store = str(realtime.get("store") or "").strip().lower()
    gauge_set(
        "polyweather_realtime_redis_fallback",
        1 if degraded_from == "redis" or store == "degraded_sqlite" else 0,
    )

    db_path = str(getattr(db, "db_path", "") or "").strip()
    if db_path:
        try:
            gauge_set("polyweather_sqlite_db_size_bytes", os.path.getsize(db_path))
        except OSError:
            pass
        _refresh_training_data_metrics(db_path)


def _target_date_stale_days(target_date: Optional[str]) -> Optional[int]:
    raw = str(target_date or "").strip()
    if not raw:
        return None
    try:
        parsed = datetime.strptime(raw, "%Y-%m-%d").date()
    except Exception:
        return None
    return max(0, (datetime.now(timezone.utc).date() - parsed).days)


def _max_target_date(db_path: str, table_name: str) -> Optional[str]:
    with connect_sqlite(db_path) as conn:
        row = conn.execute(f"SELECT MAX(target_date) AS max_date FROM {table_name}").fetchone()
    if not row:
        return None
    return row[0]


def _refresh_training_data_metrics(db_path: str) -> None:
    stale_threshold_days = max(
        0,
        int(os.getenv("POLYWEATHER_TRAINING_DATA_STALE_WARN_DAYS", "2") or "2"),
    )
    table_metrics = {
        "daily_records_store": "polyweather_daily_records_stale_days",
        "truth_records_store": "polyweather_truth_records_stale_days",
        "training_feature_records_store": "polyweather_training_features_stale_days",
    }
    max_stale_days: Optional[int] = None
    for table_name, metric_name in table_metrics.items():
        try:
            stale_days = _target_date_stale_days(_max_target_date(db_path, table_name))
        except Exception:
            stale_days = None
        if stale_days is None:
            gauge_set(metric_name, -1)
            max_stale_days = max(max_stale_days or 0, stale_threshold_days + 1)
            continue
        gauge_set(metric_name, stale_days)
        max_stale_days = max(max_stale_days or 0, stale_days)
    gauge_set(
        "polyweather_training_data_stale",
        1 if max_stale_days is None or max_stale_days > stale_threshold_days else 0,
    )