"""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}