api/performance_rum.py

"""RUM performance API for the existing FastAPI backend.

Routes:
  POST /api/performance/ingest       public, bounded and rate-limited
  GET  /api/performance/admin/summary admin-only aggregated data

The module uses the existing DATABASE_URL PostgreSQL connection and never
accepts or returns prompt/chat contents.
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
import math
import os
import re
import time
from datetime import datetime, timezone
from typing import Annotated, Any

from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, Field, field_validator

from .auth_guard import AuthRole, require_role

router = APIRouter(prefix="/api/performance", tags=["performance"])
admin_dependency = Depends(require_role(AuthRole.ADMIN))

_RELEASE_RE = re.compile(r"^[0-9a-f]{7,40}$")
_ALLOWED_METRICS = frozenset({
    "CLS", "FCP", "INP", "LCP", "TTFB",
    "codemirror:core", "codemirror:first-interactive",
    "codemirror:language", "agent:tool-executor",
})
_ALLOWED_CONNECTIONS = frozenset({"slow-2g", "2g", "3g", "4g", "wifi", "unknown"})
_ALLOWED_NAVIGATION = frozenset({"navigate", "reload", "back-forward", "prerender", "unknown"})
_RATE_WINDOW_SECONDS = 60
_RATE_LIMIT = 120
_rate_buckets: dict[str, tuple[float, int]] = {}
_logger = logging.getLogger("performance_rum")


class MetricSample(BaseModel):
    model_config = ConfigDict(extra="forbid")

    metric: str = Field(min_length=2, max_length=40)
    value: float = Field(ge=0, lt=86_400_000)
    release: str = Field(min_length=7, max_length=40)
    device: str
    connection_type: str | None = Field(default=None, max_length=12)
    navigation_type: str | None = Field(default=None, max_length=20)

    @field_validator("metric")
    @classmethod
    def validate_metric(cls, value: str) -> str:
        if value not in _ALLOWED_METRICS:
            raise ValueError("unsupported metric")
        return value

    @field_validator("value")
    @classmethod
    def validate_value(cls, value: float) -> float:
        if not math.isfinite(value):
            raise ValueError("value must be finite")
        return round(value, 3)

    @field_validator("release")
    @classmethod
    def validate_release(cls, value: str) -> str:
        value = value.strip().lower()
        if not _RELEASE_RE.fullmatch(value):
            raise ValueError("release must be a git SHA")
        return value

    @field_validator("device")
    @classmethod
    def validate_device(cls, value: str) -> str:
        if value not in {"mobile", "desktop"}:
            raise ValueError("device must be mobile or desktop")
        return value

    @field_validator("connection_type")
    @classmethod
    def validate_connection(cls, value: str | None) -> str | None:
        return value if value in _ALLOWED_CONNECTIONS else ("unknown" if value else None)

    @field_validator("navigation_type")
    @classmethod
    def validate_navigation(cls, value: str | None) -> str | None:
        return value if value in _ALLOWED_NAVIGATION else ("unknown" if value else None)


def _bucket_start(now: datetime, minutes: int = 15) -> datetime:
    now = now.astimezone(timezone.utc).replace(second=0, microsecond=0)
    return now.replace(minute=(now.minute // minutes) * minutes)


def _rate_key(request: Request, salt: str) -> str:
    # Digest only; the raw IP and user-agent are never persisted or logged.
    ip = request.headers.get("cf-connecting-ip", "")
    ua = request.headers.get("user-agent", "")[:160]
    return hashlib.sha256(f"{salt}:{ip}:{ua}".encode()).hexdigest()


def _allow_rate(key: str) -> bool:
    now = time.monotonic()
    started, count = _rate_buckets.get(key, (now, 0))
    if now - started >= _RATE_WINDOW_SECONDS:
        _rate_buckets[key] = (now, 1)
        return True
    if count >= _RATE_LIMIT:
        return False
    _rate_buckets[key] = (started, count + 1)
    return True


def _db_url() -> str:
    value = os.getenv("DATABASE_URL", "").strip()
    if not value.startswith(("postgresql://", "postgres://")):
        raise HTTPException(status_code=503, detail="PostgreSQL non configurato")
    return value



def _execute(sql: str, params: dict[str, Any], *, fetch: bool = False) -> list[dict[str, Any]]:
    import psycopg2
    import psycopg2.extras

    conn = psycopg2.connect(
        _db_url(),
        connect_timeout=5,
        sslmode="require",
        application_name="baida-rum",
        options="-c statement_timeout=5000",
    )
    try:
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute("SET statement_timeout = '5000ms'")
            cur.execute(sql, params)
            rows = [dict(row) for row in cur.fetchall()] if fetch else []
        conn.commit()
        return rows
    finally:
        conn.close()


@router.post("/ingest", status_code=status.HTTP_202_ACCEPTED)
async def ingest_metric(
    sample: MetricSample,
    request: Request,
    x_rum_key: Annotated[str | None, Header(alias="X-Rum-Key")] = None,
) -> dict[str, bool]:
    expected_key = os.getenv("RUM_INGEST_KEY", "")
    if expected_key and x_rum_key != expected_key:
        raise HTTPException(status_code=401, detail="invalid rum key")
    if not _allow_rate(_rate_key(request, os.getenv("RUM_HASH_SALT", "rum"))):
        raise HTTPException(status_code=429, detail="rate limit exceeded")

    try:
        await asyncio.to_thread(
            _execute,
            """
            insert into public.rum_samples
              (metric, value, release, device, connection_type, navigation_type, bucket_start)
            values (%(metric)s, %(value)s, %(release)s, %(device)s, %(connection_type)s,
                    %(navigation_type)s, %(bucket_start)s)
            """,
            {
                "metric": sample.metric,
                "value": sample.value,
                "release": sample.release,
                "device": sample.device,
                "connection_type": sample.connection_type,
                "navigation_type": sample.navigation_type,
                "bucket_start": _bucket_start(datetime.now(timezone.utc)),
            },
        )
    except HTTPException:
        raise
    except Exception as exc:
        _logger.error(
            "RUM database operation failed type=%s metric=%s device=%s",
            type(exc).__name__, sample.metric, sample.device,
        )
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="RUM database temporaneamente non disponibile",
        ) from exc
    return {"accepted": True}



@router.get("/admin/summary", dependencies=[admin_dependency])
async def performance_summary(
    from_time: datetime,
    to_time: datetime,
    release: str = "all",
    device: str = "mobile",
) -> dict[str, Any]:
    if device not in {"mobile", "desktop"}:
        raise HTTPException(status_code=400, detail="invalid device")
    if to_time <= from_time or (to_time - from_time).days > 31:
        raise HTTPException(status_code=400, detail="invalid time range")
    if release != "all" and not _RELEASE_RE.fullmatch(release.lower()):
        raise HTTPException(status_code=400, detail="invalid release")

    try:
        rows = await asyncio.to_thread(
            _execute,
            """
            select metric, release, device,
                   count(*)::integer as sample_count,
                   percentile_cont(0.50) within group (order by value) as p50,
                   percentile_cont(0.75) within group (order by value) as p75,
                   percentile_cont(0.95) within group (order by value) as p95
            from public.rum_samples
            where bucket_start >= %(from_time)s
              and bucket_start < %(to_time)s
              and device = %(device)s
              and (%(release)s = 'all' or release = %(release)s)
            group by metric, release, device
            order by metric, release
            """,
            {"from_time": from_time, "to_time": to_time, "device": device, "release": release.lower()},
            fetch=True,
        )
    except HTTPException:
        raise
    except Exception as exc:
        _logger.error(
            "RUM summary database operation failed type=%s device=%s",
            type(exc).__name__, device,
        )
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="RUM database temporaneamente non disponibile",
        ) from exc
    for row in rows:
        if row["sample_count"] < 20:
            row["p50"] = row["p75"] = row["p95"] = None
            row["insufficient_sample"] = True
        else:
            row["insufficient_sample"] = False
    return {"ok": True, "generated_at": datetime.now(timezone.utc).isoformat(), "items": rows}