Northern-Thailand-Ping-River-Monitor/src/web_api.py

#!/usr/bin/env python3
"""
FastAPI web interface for water monitoring system
"""

import asyncio
import os
import secrets
import time
from contextlib import asynccontextmanager
from datetime import date, datetime, timedelta
from decimal import Decimal
from threading import Lock
from typing import Any, Dict, List, Optional

import requests
from fastapi import (
    BackgroundTasks,
    Depends,
    FastAPI,
    Header,
    HTTPException,
    Query,
    Response,
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles

from .config import Config
from .health_check import (
    APIHealthCheck,
    DatabaseHealthCheck,
    HealthCheckManager,
    MemoryHealthCheck,
)
from .logging_config import get_logger, setup_logging
from .metrics import get_metrics_collector, increment_counter, set_gauge
from .postgres_history import PostgresHistory
from .schemas import (
    HealthResponse,
    MeasurementResponse,
    MetricsResponse,
    ScrapingStatusResponse,
    StationCreateModel,
    StationResponse,
    StationUpdateModel,
)
from .thaiwater import ThaiWaterClient
from .water_scraper_v3 import EnhancedWaterMonitorScraper

logger = get_logger(__name__)

# Simple thread-safe TTL cache for PostgreSQL history queries
HISTORY_CACHE: Dict[str, tuple] = {}
HISTORY_CACHE_LOCK = Lock()
HISTORY_TTL = 300  # 5 minutes

FORECAST_CACHE: Dict[str, tuple] = {}
FORECAST_CACHE_LOCK = Lock()
FORECAST_COMPUTE_LOCK = asyncio.Lock()  # single-flight for expensive inference
FORECAST_TTL = 900  # 15 minutes

DB_STATS_CACHE: Dict[str, tuple] = {}
DB_STATS_CACHE_LOCK = Lock()
_DB_STATS_COMPUTE_LOCK = Lock()
DB_STATS_TTL = 300  # 5 minutes

# Admin API protection. Read/dashboard endpoints stay public; anything that
# mutates state or leaks configuration requires the X-API-Key header matching
# ADMIN_API_KEY. Secure by default: with no key configured, those endpoints
# are disabled entirely rather than open.
ADMIN_API_KEY = os.getenv("ADMIN_API_KEY")


def require_admin_key(x_api_key: Optional[str] = Header(None, alias="X-API-Key")):
    if not ADMIN_API_KEY:
        raise HTTPException(
            status_code=503,
            detail="Admin API disabled: ADMIN_API_KEY is not configured on the server",
        )
    if not x_api_key or not secrets.compare_digest(x_api_key, ADMIN_API_KEY):
        raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")


# Dashboard HTML is loaded once at import from src/static/dashboard.html.
_DASHBOARD_HTML_PATH = os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "static", "dashboard.html"
)
try:
    with open(_DASHBOARD_HTML_PATH, encoding="utf-8") as _dashboard_file:
        DASHBOARD_HTML = _dashboard_file.read()
except OSError as _dashboard_error:  # pragma: no cover - defensive fallback
    logger.error(f"Could not load dashboard HTML: {_dashboard_error}")
    DASHBOARD_HTML = "<h1>Northern Thailand Ping River Monitor API</h1><p>See <code>/docs</code>.</p>"


# Global application state
app_state = {
    "scraper": None,
    "health_manager": None,
    "scraping_task": None,
    "is_scraping": False,
    "scraping_stats": {
        "total_runs": 0,
        "successful_runs": 0,
        "failed_runs": 0,
        "last_run": None,
        "next_run": None,
    },
}


def _acquire_collection_leadership(port: int):
    """Elect one background-collection leader per machine via a localhost bind.

    With multiple uvicorn workers every process runs this lifespan; only the
    worker holding the lock port runs the scraper/HII loops, so external APIs
    are polled once per cycle instead of once per worker. The socket is held
    for the process lifetime and releases automatically if the worker dies.
    """
    import socket

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.bind(("127.0.0.1", port))
        sock.listen(1)
        return sock
    except OSError:
        sock.close()
        return None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan manager"""
    # Startup
    logger.info("Starting Water Monitor API...")

    # Larger dedicated executor: the default (cpu+4 threads) was exhausted
    # under load by concurrent blocking work; waiting threads are cheap.
    from concurrent.futures import ThreadPoolExecutor

    app_state["executor"] = ThreadPoolExecutor(
        max_workers=Config.EXECUTOR_THREADS, thread_name_prefix="api"
    )
    asyncio.get_running_loop().set_default_executor(app_state["executor"])

    # Initialize configuration
    try:
        Config.validate_config()
        logger.info("Configuration validated successfully")
    except Exception as e:
        logger.error(f"Configuration validation failed: {e}")
        raise

    # Initialize scraper
    db_config = Config.get_database_config()
    app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)

    # Initialize HII/ThaiWater collector (rainfall + backup water level)
    try:
        from .hii_collector import create_collector_from_config

        app_state["hii_collector"] = create_collector_from_config()
        if app_state["hii_collector"]:
            logger.info("HII collection enabled (Ping-basin rainfall + water level)")
    except Exception as e:
        app_state["hii_collector"] = None
        logger.error(f"HII collector initialization failed: {e}")

    # Initialize health checks
    health_manager = HealthCheckManager()
    health_manager.add_check(DatabaseHealthCheck(app_state["scraper"].db_adapter))
    health_manager.add_check(
        APIHealthCheck(Config.API_URL, app_state["scraper"].session)
    )
    health_manager.add_check(MemoryHealthCheck(max_memory_mb=1000))
    app_state["health_manager"] = health_manager

    # Start background scraping in exactly one worker per machine
    app_state["leader_lock"] = _acquire_collection_leadership(
        Config.COLLECTION_LEADER_PORT
    )
    if app_state["leader_lock"]:
        app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
        logger.info("This worker is the background-collection leader")
    else:
        logger.info(
            "Another worker holds collection leadership; "
            "background scraping disabled in this process"
        )

    logger.info("Water Monitor API started successfully")

    yield

    # Shutdown
    logger.info("Shutting down Water Monitor API...")

    if app_state["scraping_task"]:
        app_state["scraping_task"].cancel()
        try:
            await app_state["scraping_task"]
        except asyncio.CancelledError:
            pass
    if app_state.get("leader_lock"):
        app_state["leader_lock"].close()
    if app_state.get("executor"):
        app_state["executor"].shutdown(wait=False)

    logger.info("Water Monitor API shutdown complete")


# Create FastAPI app
app = FastAPI(
    title="Northern Thailand Ping River Monitor API",
    description="Real-time water level monitoring system for Northern Thailand's Ping River Basin stations",
    version="3.1.3",
    lifespan=lifespan,
)
app.mount(
    "/static",
    StaticFiles(directory=os.path.dirname(_DASHBOARD_HTML_PATH)),
    name="static",
)

# Add CORS middleware.
# Origins come from CORS_ALLOW_ORIGINS (comma-separated). When none are configured
# we fall back to a wildcard WITHOUT credentials (a safe, spec-valid combination);
# credentials are only enabled when explicit origins are provided.
# Compress large responses end-to-end: the dashboard HTML and station JSON
# payloads shrink ~5-6x, which matters both on the internal VPN hop to the
# Caddy TLS terminator and on the public leg (Caddy passes Content-Encoding
# through). Load testing showed the deployment is bandwidth-bound, not
# compute-bound, once the response caches are hit.
app.add_middleware(GZipMiddleware, minimum_size=500)

_cors_origins = Config.CORS_ALLOW_ORIGINS or ["*"]
_cors_allow_credentials = bool(Config.CORS_ALLOW_ORIGINS)
app.add_middleware(
    CORSMiddleware,
    allow_origins=_cors_origins,
    allow_credentials=_cors_allow_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)


async def background_scraping_task():
    """Background task for periodic data scraping"""
    while True:
        try:
            if not app_state["is_scraping"]:
                app_state["is_scraping"] = True

                # Run scraping cycle
                scraper = app_state["scraper"]
                if scraper:
                    logger.info("Starting background scraping cycle")
                    start_time = datetime.now()

                    try:
                        # run_scraping_cycle() does blocking network/DB I/O and time.sleep
                        # retries; run it in a thread so it doesn't freeze the event loop.
                        result = await asyncio.get_event_loop().run_in_executor(
                            None, scraper.run_scraping_cycle
                        )

                        # Update stats
                        app_state["scraping_stats"]["total_runs"] += 1
                        app_state["scraping_stats"]["last_run"] = start_time

                        if result:
                            app_state["scraping_stats"]["successful_runs"] += 1
                            increment_counter("scraping_cycles_successful")
                            logger.info(
                                "Background scraping cycle completed successfully"
                            )
                        else:
                            app_state["scraping_stats"]["failed_runs"] += 1
                            increment_counter("scraping_cycles_failed")
                            logger.warning(
                                "Background scraping cycle completed with no new data"
                            )

                        # Update metrics
                        set_gauge("last_scraping_timestamp", start_time.timestamp())

                    except Exception as e:
                        app_state["scraping_stats"]["failed_runs"] += 1
                        increment_counter("scraping_cycles_failed")
                        logger.error(f"Background scraping cycle failed: {e}")

                    # HII rainfall/water-level snapshot (independent of RID cycle)
                    hii_collector = app_state.get("hii_collector")
                    if hii_collector:
                        try:
                            hii_counts = await asyncio.get_event_loop().run_in_executor(
                                None, hii_collector.run_cycle
                            )
                            set_gauge(
                                "hii_rainfall_rows_saved", hii_counts["rainfall"]
                            )
                            set_gauge(
                                "hii_waterlevel_rows_saved", hii_counts["waterlevel"]
                            )
                        except Exception as e:
                            logger.error(f"HII collection failed: {e}")

                app_state["is_scraping"] = False

            # Calculate next run time
            interval_seconds = Config.SCRAPING_INTERVAL_HOURS * 3600
            app_state["scraping_stats"]["next_run"] = datetime.now() + timedelta(
                seconds=interval_seconds
            )

            # Wait for next cycle
            await asyncio.sleep(interval_seconds)

        except asyncio.CancelledError:
            logger.info("Background scraping task cancelled")
            break
        except Exception as e:
            logger.error(f"Error in background scraping task: {e}")
            await asyncio.sleep(60)  # Wait a minute before retrying


# Umami server-side API tracking (fire-and-forget; never blocks a response)

_UMAMI_TRACK_PREFIXES = (
    "/api/",
    "/measurements",
    "/forecast",
    "/stations",
    "/sensors",
)


def _send_umami_event(path: str, method: str, status: int, host: str, user_agent: str):
    try:
        requests.post(
            Config.UMAMI_API_URL,
            json={
                "type": "event",
                "payload": {
                    "website": Config.UMAMI_WEBSITE_ID,
                    "url": path,
                    "hostname": host,
                    "name": "api-request",
                    "data": {"method": method, "status": status},
                },
            },
            headers={"User-Agent": user_agent or "api-client"},
            timeout=3,
        )
    except Exception:
        pass  # analytics must never affect API behavior


# Cache-Control by path: lets browsers and any edge cache (e.g. Cloudflare in
# front of Caddy) absorb repeat traffic — the public uplink is the scarce
# resource. Max-ages mirror the server-side cache TTLs / data cadence.
_CACHE_CONTROL_RULES = (
    ("/static/", "public, max-age=3600"),
    ("/measurements/latest", "public, max-age=30"),
    ("/api/hii/", "public, max-age=60"),
    ("/measurements/history", "public, max-age=300"),
    ("/forecast", "public, max-age=120"),
    ("/api/stats", "public, max-age=300"),
    ("/stations", "public, max-age=300"),
)


@app.middleware("http")
async def cache_control_headers(request, call_next):
    response = await call_next(request)
    if request.method == "GET" and response.status_code == 200:
        path = request.url.path
        if path == "/":
            response.headers.setdefault("Cache-Control", "public, max-age=120")
        else:
            for prefix, value in _CACHE_CONTROL_RULES:
                if path.startswith(prefix):
                    response.headers.setdefault("Cache-Control", value)
                    break
    return response


@app.middleware("http")
async def umami_api_tracking(request, call_next):
    response = await call_next(request)
    if (
        Config.UMAMI_TRACK_API
        and Config.UMAMI_WEBSITE_ID
        and request.url.path.startswith(_UMAMI_TRACK_PREFIXES)
    ):
        asyncio.get_event_loop().create_task(
            asyncio.to_thread(
                _send_umami_event,
                request.url.path,
                request.method,
                response.status_code,
                request.headers.get("host", "water.buildfor.life"),
                request.headers.get("user-agent", ""),
            )
        )
    return response


# API Routes


@app.get("/", response_class=HTMLResponse)
async def root():
    """Root endpoint with basic dashboard"""
    return HTMLResponse(content=DASHBOARD_HTML)


# Crawler/indexing files must live at the domain root (served from src/static)
_STATIC_DIR = os.path.dirname(_DASHBOARD_HTML_PATH)


@app.get("/robots.txt", include_in_schema=False)
async def robots_txt():
    return FileResponse(os.path.join(_STATIC_DIR, "robots.txt"), media_type="text/plain")


@app.get("/llms.txt", include_in_schema=False)
async def llms_txt():
    return FileResponse(os.path.join(_STATIC_DIR, "llms.txt"), media_type="text/plain")


@app.get("/sitemap.xml", include_in_schema=False)
async def sitemap_xml():
    return FileResponse(
        os.path.join(_STATIC_DIR, "sitemap.xml"), media_type="application/xml"
    )


@app.get("/health", response_model=HealthResponse)
async def get_health():
    """Get system health status"""
    increment_counter("api_requests", labels={"endpoint": "health"})

    health_manager = app_state["health_manager"]
    if not health_manager:
        raise HTTPException(status_code=503, detail="Health manager not initialized")

    # Health checks include a DB query and an external RID-API probe (seconds
    # of blocking I/O). Cached briefly so hammering /health cannot flood the
    # shared thread-pool executor; fresh hits answer inline.
    def compute():
        health_manager.run_all_checks()
        return health_manager.get_health_summary()

    summary, _ = await _cached_swr(
        HEALTH_CACHE,
        HEALTH_CACHE_LOCK,
        _HEALTH_COMPUTE_LOCK,
        "health",
        Config.HEALTH_CACHE_TTL_SECONDS,
        compute,
    )

    return HealthResponse(**summary)


@app.get("/metrics", response_model=MetricsResponse)
async def get_metrics():
    """Get application metrics"""
    increment_counter("api_requests", labels={"endpoint": "metrics"})

    metrics_collector = get_metrics_collector()
    metrics = metrics_collector.get_all_metrics()

    return MetricsResponse(**metrics)


@app.get("/stations", response_model=List[StationResponse])
async def get_stations():
    """Get list of all monitoring stations"""
    increment_counter("api_requests", labels={"endpoint": "stations"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    stations = []
    for station_id, station_info in scraper.station_mapping.items():
        stations.append(
            StationResponse(
                station_id=int(station_id),
                station_code=station_info["code"],
                thai_name=station_info["thai_name"],
                english_name=station_info["english_name"],
                latitude=station_info.get("latitude"),
                longitude=station_info.get("longitude"),
                status="active",
            )
        )

    return stations


@app.post(
    "/stations",
    response_model=StationResponse,
    dependencies=[Depends(require_admin_key)],
)
async def create_station(station: StationCreateModel):
    """Create a new monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "create_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    try:
        # Find next available station ID
        existing_ids = [int(sid) for sid in scraper.station_mapping.keys()]
        new_station_id = max(existing_ids) + 1 if existing_ids else 1

        # Add to station mapping and persist
        new_key = str(new_station_id)
        scraper.station_mapping[new_key] = {
            "code": station.station_code,
            "thai_name": station.thai_name,
            "english_name": station.english_name,
            "latitude": station.latitude,
            "longitude": station.longitude,
            "geohash": station.geohash,
        }
        if not scraper.save_stations():
            scraper.station_mapping.pop(new_key, None)
            raise HTTPException(status_code=500, detail="Failed to persist new station")

        logger.info(
            f"Created new station: {station.station_code} ({station.english_name})"
        )

        return StationResponse(
            station_id=new_station_id,
            station_code=station.station_code,
            thai_name=station.thai_name,
            english_name=station.english_name,
            latitude=station.latitude,
            longitude=station.longitude,
            geohash=station.geohash,
            status=station.status,
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error creating station: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.put(
    "/stations/{station_id}",
    response_model=StationResponse,
    dependencies=[Depends(require_admin_key)],
)
async def update_station(station_id: int, updates: StationUpdateModel):
    """Update an existing monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "update_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    station_key = str(station_id)
    if station_key not in scraper.station_mapping:
        raise HTTPException(status_code=404, detail="Station not found")

    try:
        station_info = scraper.station_mapping[station_key]
        original = dict(station_info)  # snapshot for rollback if persistence fails

        # Update fields if provided
        if updates.thai_name is not None:
            station_info["thai_name"] = updates.thai_name
        if updates.english_name is not None:
            station_info["english_name"] = updates.english_name
        if updates.latitude is not None:
            station_info["latitude"] = updates.latitude
        if updates.longitude is not None:
            station_info["longitude"] = updates.longitude
        if updates.geohash is not None:
            station_info["geohash"] = updates.geohash

        if not scraper.save_stations():
            scraper.station_mapping[station_key] = original
            raise HTTPException(
                status_code=500, detail="Failed to persist station update"
            )

        logger.info(f"Updated station {station_id}: {station_info['code']}")

        return StationResponse(
            station_id=station_id,
            station_code=station_info["code"],
            thai_name=station_info["thai_name"],
            english_name=station_info["english_name"],
            latitude=station_info.get("latitude"),
            longitude=station_info.get("longitude"),
            geohash=station_info.get("geohash"),
            status=updates.status or "active",
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error updating station {station_id}: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/stations/{station_id}", dependencies=[Depends(require_admin_key)])
async def delete_station(station_id: int):
    """Delete a monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "delete_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    station_key = str(station_id)
    if station_key not in scraper.station_mapping:
        raise HTTPException(status_code=404, detail="Station not found")

    try:
        station_info = scraper.station_mapping.pop(station_key)

        if not scraper.save_stations():
            scraper.station_mapping[station_key] = station_info  # restore
            raise HTTPException(
                status_code=500, detail="Failed to persist station deletion"
            )

        logger.info(f"Deleted station {station_id}: {station_info['code']}")

        return {"message": f"Station {station_info['code']} deleted successfully"}

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error deleting station {station_id}: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/stations/{station_id}", response_model=StationResponse)
async def get_station(station_id: int):
    """Get details of a specific monitoring station"""
    increment_counter("api_requests", labels={"endpoint": "get_station"})

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    station_key = str(station_id)
    if station_key not in scraper.station_mapping:
        raise HTTPException(status_code=404, detail="Station not found")

    station_info = scraper.station_mapping[station_key]

    return StationResponse(
        station_id=station_id,
        station_code=station_info["code"],
        thai_name=station_info["thai_name"],
        english_name=station_info["english_name"],
        latitude=station_info.get("latitude"),
        longitude=station_info.get("longitude"),
        geohash=station_info.get("geohash"),
        status="active",
    )


def _to_measurement_response(measurement: Dict[str, Any]) -> MeasurementResponse:
    """Map a raw measurement dict from a DB adapter to the API response model.

    ``discharge`` is optional in the data (some stations report only level), so
    it is read with ``.get`` rather than assumed present.
    """
    return MeasurementResponse(
        timestamp=measurement["timestamp"],
        station_code=measurement["station_code"],
        station_name_en=measurement["station_name_en"],
        station_name_th=measurement["station_name_th"],
        water_level=measurement["water_level"],
        discharge=measurement.get("discharge"),
        discharge_percent=measurement.get("discharge_percent"),
        status=measurement.get("status", "active"),
    )


@app.get("/sensors/thaiwater")
async def get_thaiwater_sensors():
    """Get current ThaiWater water-level sensors in the Ping basin."""
    increment_counter("api_requests", labels={"endpoint": "thaiwater_sensors"})
    try:
        client = ThaiWaterClient(
            api_key=Config.THAIWATER_API_KEY,
            timeout=Config.REQUEST_TIMEOUT,
        )
        return await asyncio.to_thread(client.fetch_ping_sensors)
    except RuntimeError as error:
        raise HTTPException(status_code=503, detail=str(error))
    except requests.RequestException as error:
        logger.error(f"Error fetching ThaiWater sensors: {error}")
        raise HTTPException(status_code=502, detail="ThaiWater API unavailable")


def _hii_engine():
    """Engine of the HII store, or None when collection is disabled."""
    collector = app_state.get("hii_collector")
    if not collector:
        return None
    store = collector.store
    if not store.engine and not store.connect():
        return None
    return store.engine


def _hii_rows(sql: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Run a read query against the hii_* tables, JSON-normalizing numerics."""
    engine = _hii_engine()
    if engine is None:
        return []
    from sqlalchemy import text

    with engine.connect() as conn:
        rows = [dict(row._mapping) for row in conn.execute(text(sql), params)]
    for row in rows:
        for key, value in row.items():
            if isinstance(value, Decimal):
                row[key] = float(value)
    return rows


# Short-TTL response caches with stale-on-error. Process-local by design
# (single-worker deployment); _ttl_cached_stale is the seam where a shared
# backend (e.g. Redis) would slot in if we ever run multiple workers.
# Thread-based single-flight keeps it event-loop-agnostic. Expired entries are
# kept as a fallback: if the recompute fails (typically the DB briefly
# unreachable), the last good response is served instead of a 5xx — during a
# flood, slightly stale readings with a visible timestamp beat an error page.
HII_CACHE: Dict[str, Any] = {}
HII_CACHE_LOCK = Lock()
_HII_COMPUTE_LOCKS = {"rain": Lock(), "waterlevel": Lock()}
LATEST_CACHE: Dict[str, Any] = {}
LATEST_CACHE_LOCK = Lock()
_LATEST_COMPUTE_LOCK = Lock()
HEALTH_CACHE: Dict[str, Any] = {}
HEALTH_CACHE_LOCK = Lock()
_HEALTH_COMPUTE_LOCK = Lock()


_REFRESH_IN_FLIGHT: set = set()
_REFRESH_FLAG_LOCK = Lock()


async def _cached_swr(cache, cache_lock, compute_lock, key, ttl, compute):
    """Stale-while-revalidate. Returns (value, served_stale).

    fresh -> return inline; expired-but-present -> return the stale value
    immediately and kick ONE background refresh (nobody waits, and no executor
    thread is parked on the single-flight lock — under stampede that parking
    exhausted the pool and timed out unrelated endpoints); absent -> cold
    single-flight compute (first request per key since startup).
    """
    fresh = _cache_fresh(cache, cache_lock, key, ttl)
    if fresh is not None:
        return fresh, False
    with cache_lock:
        entry = cache.get(key)
    if entry is not None:
        with _REFRESH_FLAG_LOCK:
            should_start = key not in _REFRESH_IN_FLIGHT
            if should_start:
                _REFRESH_IN_FLIGHT.add(key)
        if should_start:

            async def _refresh():
                try:
                    await asyncio.to_thread(
                        _ttl_cached_stale,
                        cache,
                        cache_lock,
                        compute_lock,
                        key,
                        ttl,
                        compute,
                    )
                except Exception as error:
                    logger.warning(f"{key}: background refresh failed: {error}")
                finally:
                    with _REFRESH_FLAG_LOCK:
                        _REFRESH_IN_FLIGHT.discard(key)

            asyncio.create_task(_refresh())
        return entry[1], True
    value, stale = await asyncio.to_thread(
        _ttl_cached_stale, cache, cache_lock, compute_lock, key, ttl, compute
    )
    return value, stale


def _cache_fresh(cache, cache_lock, key, ttl):
    """Non-blocking fresh-cache read for the async fast path. On-box load
    testing showed cache HITS queuing ~11s behind slow work in the shared
    thread-pool executor — so handlers must check the cache inline and only
    dispatch to a thread on a miss."""
    with cache_lock:
        entry = cache.get(key)
        if entry and time.monotonic() - entry[0] < ttl:
            return entry[1]
    return None


def _ttl_cached_stale(cache, cache_lock, compute_lock, key, ttl, compute):
    """Return (value, is_stale). Single-flight; keeps expired entries as an
    error fallback; never caches empty results so recovery is immediate."""
    with cache_lock:
        entry = cache.get(key)
        if entry and time.monotonic() - entry[0] < ttl:
            return entry[1], False
    with compute_lock:
        with cache_lock:
            entry = cache.get(key)
            if entry and time.monotonic() - entry[0] < ttl:
                return entry[1], False
        try:
            value = compute()
        except Exception as error:
            if entry is not None:
                age = int(time.monotonic() - entry[0])
                logger.warning(
                    f"{key}: recompute failed ({error}); serving {age}s-stale copy"
                )
                return entry[1], True
            raise
        if value:
            with cache_lock:
                cache[key] = (time.monotonic(), value)
        return value, False


async def _hii_swr(feed: str, hours: int, sql: str):
    def compute():
        cutoff = datetime.now() - timedelta(hours=hours)
        return _hii_rows(sql, {"cutoff": cutoff})

    return await _cached_swr(
        HII_CACHE,
        HII_CACHE_LOCK,
        _HII_COMPUTE_LOCKS[feed],
        f"{feed}:{hours}",
        Config.HII_CACHE_TTL_SECONDS,
        compute,
    )


@app.get("/api/hii/rainfall/latest")
async def get_hii_rainfall_latest(
    response: Response, hours: int = Query(26, ge=1, le=168)
):
    """Latest rainfall reading per HII station (Ping basin, collected hourly)."""
    increment_counter("api_requests", labels={"endpoint": "hii_rainfall"})
    sql = """
        SELECT s.id AS station_id, s.oldcode, s.name_en, s.name_th,
               s.latitude, s.longitude, s.agency,
               m.timestamp, m.rain_1h, m.rain_24h
        FROM hii_rainfall m
        JOIN hii_rain_stations s ON s.id = m.station_id
        JOIN (SELECT station_id, MAX(timestamp) AS latest_ts
              FROM hii_rainfall WHERE timestamp >= :cutoff
              GROUP BY station_id) latest
          ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp
    """
    rows, stale = await _hii_swr("rain", hours, sql)
    if stale:
        response.headers["X-Data-Stale"] = "true"
    return rows


@app.get("/api/hii/waterlevel/latest")
async def get_hii_waterlevel_latest(
    response: Response, hours: int = Query(26, ge=1, le=168)
):
    """Latest water-level reading per HII station (Ping basin, m MSL)."""
    increment_counter("api_requests", labels={"endpoint": "hii_waterlevel"})
    sql = """
        SELECT s.id AS station_id, s.oldcode, s.rid_code, s.name_en, s.name_th,
               s.latitude, s.longitude, s.agency, s.river_name,
               s.offset_msl, s.min_bank_msl, s.is_key_station,
               m.timestamp, m.wl_msl, m.discharge, m.storage_percent,
               m.situation_level, m.diff_wl_bank
        FROM hii_waterlevel m
        JOIN hii_wl_stations s ON s.id = m.station_id
        JOIN (SELECT station_id, MAX(timestamp) AS latest_ts
              FROM hii_waterlevel WHERE timestamp >= :cutoff
              GROUP BY station_id) latest
          ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp
    """
    rows, stale = await _hii_swr("waterlevel", hours, sql)
    if stale:
        response.headers["X-Data-Stale"] = "true"
    return rows


@app.get("/measurements/history/{station_code}")
async def get_postgres_history(
    station_code: str,
    hours: int = Query(168, ge=1),
    limit: int = Query(50000, ge=1, le=100000),
    start: Optional[date] = Query(None, description="First day (overrides hours)"),
    end: Optional[date] = Query(None, description="Last day, inclusive"),
):
    """Get historical measurements for a station from the configured database."""
    cache_key = f"{station_code}:{hours}:{limit}:{start}:{end}"
    now = time.monotonic()
    with HISTORY_CACHE_LOCK:
        cached = HISTORY_CACHE.get(cache_key)
        if cached and now - cached[0] < HISTORY_TTL:
            return cached[1]
    try:
        db_config = Config.get_database_config()
        end_time = (
            datetime.combine(end, datetime.max.time()) if end else datetime.now()
        )
        start_time = (
            datetime.combine(start, datetime.min.time())
            if start
            else end_time - timedelta(hours=hours)
        )
        if db_config["type"] == "postgresql":
            history = PostgresHistory(db_config["connection_string"])
            data = await asyncio.to_thread(
                history.station_history,
                station_code,
                start_time,
                end_time,
                limit,
            )
        else:
            scraper = app_state["scraper"]
            if not scraper or not scraper.db_adapter:
                raise RuntimeError("Database not available")
            rows = await asyncio.to_thread(
                scraper.db_adapter.get_measurements_by_timerange,
                start_time,
                end_time,
                [station_code],
            )
            # adapter returns newest-first; keep the newest `limit` rows, chart wants ascending
            data = list(reversed(rows[:limit]))
        with HISTORY_CACHE_LOCK:
            HISTORY_CACHE[cache_key] = (now, data)
        return data
    except RuntimeError as error:
        raise HTTPException(status_code=503, detail=str(error))
    except Exception as error:
        logger.error(f"Error fetching measurement history: {error}")
        raise HTTPException(status_code=502, detail="Measurement history unavailable")


@app.get("/forecast")
async def get_flood_forecasts():
    """Flood-risk forecasts per station for the 6/12/24 h horizons."""
    increment_counter("api_requests", labels={"endpoint": "forecast"})
    now = time.monotonic()
    with FORECAST_CACHE_LOCK:
        cached = FORECAST_CACHE.get("all")
        if cached and now - cached[0] < FORECAST_TTL:
            return cached[1]
    try:
        from .ml.predict import get_latest_forecasts
    except ImportError as error:
        raise HTTPException(status_code=503, detail=f"Forecasting unavailable: {error}")
    # Single-flight: inference takes seconds; without this, N concurrent cache
    # misses ran N full inferences and starved the thread pool (load test:
    # /forecast timeouts at 10 concurrent clients rippled into every endpoint).
    async with FORECAST_COMPUTE_LOCK:
        with FORECAST_CACHE_LOCK:
            cached = FORECAST_CACHE.get("all")
            if cached and time.monotonic() - cached[0] < FORECAST_TTL:
                return cached[1]
        return await _compute_forecasts(get_latest_forecasts)


async def _compute_forecasts(get_latest_forecasts):
    now = time.monotonic()
    try:
        data = await asyncio.to_thread(get_latest_forecasts)
    except FileNotFoundError:
        raise HTTPException(status_code=503, detail="No trained flood models found")
    except RuntimeError as error:
        raise HTTPException(status_code=503, detail=str(error))
    except Exception as error:
        logger.error(f"Error computing flood forecasts: {error}")
        raise HTTPException(status_code=502, detail="Flood forecast unavailable")
    with FORECAST_CACHE_LOCK:
        FORECAST_CACHE["all"] = (now, data)
    return data


@app.get("/measurements/latest", response_model=List[MeasurementResponse])
async def get_latest_measurements(response: Response, limit: int = 100):
    """Get latest measurements from all stations"""
    increment_counter("api_requests", labels={"endpoint": "measurements_latest"})

    scraper = app_state["scraper"]
    if not scraper or not scraper.db_adapter:
        raise HTTPException(status_code=503, detail="Database not available")

    def compute():
        return scraper.get_latest_data(limit=limit)

    try:
        # SWR-cached: the most frequently hit endpoint (every dashboard poll)
        measurements, stale = await _cached_swr(
            LATEST_CACHE,
            LATEST_CACHE_LOCK,
            _LATEST_COMPUTE_LOCK,
            f"latest:{limit}",
            Config.LATEST_CACHE_TTL_SECONDS,
            compute,
        )
        if stale:
            response.headers["X-Data-Stale"] = "true"
        return [_to_measurement_response(m) for m in measurements]

    except Exception as e:
        logger.error(f"Error fetching latest measurements: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get(
    "/measurements/station/{station_code}", response_model=List[MeasurementResponse]
)
async def get_station_measurements(
    station_code: str, hours: int = 24, limit: int = 1000
):
    """Get measurements for a specific station"""
    increment_counter("api_requests", labels={"endpoint": "measurements_station"})

    scraper = app_state["scraper"]
    if not scraper or not scraper.db_adapter:
        raise HTTPException(status_code=503, detail="Database not available")

    try:
        # Get measurements for the specified time range
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours)

        measurements = scraper.db_adapter.get_measurements_by_timerange(
            start_time, end_time, station_codes=[station_code]
        )

        # Limit results
        measurements = measurements[:limit]

        return [_to_measurement_response(m) for m in measurements]

    except Exception as e:
        logger.error(f"Error fetching station measurements: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/stats")
async def get_database_stats():
    """Get database coverage statistics (totals, date range, hourly coverage)"""
    increment_counter("api_requests", labels={"endpoint": "api_stats"})

    scraper = app_state["scraper"]
    if not scraper or not scraper.db_adapter:
        raise HTTPException(status_code=503, detail="Database not available")

    cached = _cache_fresh(DB_STATS_CACHE, DB_STATS_CACHE_LOCK, "all", DB_STATS_TTL)
    if cached is not None:
        return cached

    def hii_totals():
        engine = _hii_engine()
        if engine is None:
            return None
        from sqlalchemy import text

        with engine.connect() as conn:
            return conn.execute(
                text(
                    """
                    SELECT (SELECT COUNT(*) FROM hii_rainfall) AS rain_n,
                           (SELECT COUNT(*) FROM hii_waterlevel) AS wl_n,
                           (SELECT COUNT(*) FROM hii_rain_stations) AS rain_s,
                           (SELECT COUNT(*) FROM hii_wl_stations) AS wl_s,
                           (SELECT MIN(timestamp) FROM hii_rainfall) AS rain_lo,
                           (SELECT MAX(timestamp) FROM hii_rainfall) AS rain_hi,
                           (SELECT MIN(timestamp) FROM hii_waterlevel) AS wl_lo,
                           (SELECT MAX(timestamp) FROM hii_waterlevel) AS wl_hi
                    """
                )
            ).one()

    def compute():
        # Heavy: full-table counts and coverage over ~1.7M rows. Runs at most
        # once per TTL thanks to the single-flight; concurrent misses wait for
        # this one result instead of piling identical queries onto Postgres.
        stats = scraper.db_adapter.get_database_stats()
        if stats is None:
            raise RuntimeError("Database statistics unavailable")

        hii = None
        try:
            hii = hii_totals()
        except Exception as e:
            logger.warning(f"HII stats unavailable: {e}")

        def as_dt(value):
            if isinstance(value, str):
                return datetime.fromisoformat(value)
            return value

        first_ts = stats["first_timestamp"]
        last_ts = stats["last_timestamp"]
        rain_n = wl_n = hii_stations = 0
        if hii is not None:
            rain_n, wl_n = hii.rain_n or 0, hii.wl_n or 0
            hii_stations = (hii.rain_s or 0) + (hii.wl_s or 0)
            for lo in (as_dt(hii.rain_lo), as_dt(hii.wl_lo)):
                if lo is not None and lo < first_ts:
                    first_ts = lo
            for hi in (as_dt(hii.rain_hi), as_dt(hii.wl_hi)):
                if hi is not None and hi > last_ts:
                    last_ts = hi

        return {
            # Whole-DB totals (RID + HII feeds); breakdown fields alongside
            "total_measurements": stats["total_measurements"] + rain_n + wl_n,
            "rid_measurements": stats["total_measurements"],
            "hii_rainfall_measurements": rain_n,
            "hii_waterlevel_measurements": wl_n,
            "station_count": stats["station_count"] + hii_stations,
            "rid_station_count": stats["station_count"],
            "hii_station_count": hii_stations,
            "first_timestamp": first_ts.isoformat(),
            "last_timestamp": last_ts.isoformat(),
            "days_spanned": (last_ts.date() - first_ts.date()).days + 1,
            "coverage_percent": stats["coverage_percent"],
        }

    try:
        data, _ = await _cached_swr(
            DB_STATS_CACHE,
            DB_STATS_CACHE_LOCK,
            _DB_STATS_COMPUTE_LOCK,
            "all",
            DB_STATS_TTL,
            compute,
        )
        return data
    except RuntimeError as e:
        raise HTTPException(status_code=503, detail=str(e))
    except Exception as e:
        logger.error(f"Error fetching database stats: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/scrape/trigger", dependencies=[Depends(require_admin_key)])
async def trigger_scraping(background_tasks: BackgroundTasks):
    """Trigger manual data scraping"""
    increment_counter("api_requests", labels={"endpoint": "scrape_trigger"})

    if app_state["is_scraping"]:
        raise HTTPException(status_code=409, detail="Scraping already in progress")

    scraper = app_state["scraper"]
    if not scraper:
        raise HTTPException(status_code=503, detail="Scraper not initialized")

    def run_scraping():
        """Background task to run scraping"""
        try:
            app_state["is_scraping"] = True
            logger.info("Manual scraping triggered via API")

            result = scraper.run_scraping_cycle()

            # Update stats
            app_state["scraping_stats"]["total_runs"] += 1
            app_state["scraping_stats"]["last_run"] = datetime.now()

            if result:
                app_state["scraping_stats"]["successful_runs"] += 1
                increment_counter("manual_scraping_successful")
            else:
                app_state["scraping_stats"]["failed_runs"] += 1
                increment_counter("manual_scraping_failed")

        except Exception as e:
            app_state["scraping_stats"]["failed_runs"] += 1
            increment_counter("manual_scraping_failed")
            logger.error(f"Manual scraping failed: {e}")
        finally:
            app_state["is_scraping"] = False

    background_tasks.add_task(run_scraping)

    return {"message": "Scraping triggered", "status": "started"}


@app.get("/scraping/status", response_model=ScrapingStatusResponse)
async def get_scraping_status():
    """Get current scraping status"""
    increment_counter("api_requests", labels={"endpoint": "scraping_status"})

    stats = app_state["scraping_stats"]

    return ScrapingStatusResponse(
        is_running=app_state["is_scraping"],
        last_run=stats["last_run"],
        next_run=stats["next_run"],
        total_runs=stats["total_runs"],
        successful_runs=stats["successful_runs"],
        failed_runs=stats["failed_runs"],
    )


@app.get("/config", dependencies=[Depends(require_admin_key)])
async def get_config():
    """Get current configuration (sensitive data masked)"""
    increment_counter("api_requests", labels={"endpoint": "config"})

    config = Config.get_all_settings()

    # Mask sensitive information
    for key in config:
        if "password" in key.lower() or "secret" in key.lower():
            if config[key]:
                config[key] = "*" * 8

    return config


if __name__ == "__main__":
    import uvicorn

    # Setup logging
    setup_logging(
        log_level=Config.LOG_LEVEL,
        log_file=Config.LOG_FILE,
        enable_console=True,
        enable_colors=True,
    )

    # Run the API server
    uvicorn.run(
        "web_api:app", host="0.0.0.0", port=8000, reload=False, log_config=None
    )  # Use our custom logging