wrongmove/api/perf_routes.py

"""Frontend performance metrics ingestion endpoint."""
from __future__ import annotations

from fastapi import APIRouter, Response
from pydantic import BaseModel, Field, field_validator

import api.metrics as app_metrics

ALLOWED_METRICS = {"worker_roundtrip", "worker_compute", "main_thread", "feature_count"}
MAX_BATCH_SIZE = 100


class PerfSample(BaseModel):
    metric: str
    operation: str = Field(max_length=50)
    value: float = Field(ge=0, le=3600)

    @field_validator("metric")
    @classmethod
    def validate_metric(cls, v: str) -> str:
        if v not in ALLOWED_METRICS:
            raise ValueError(f"Unknown metric: {v}")
        return v


perf_router = APIRouter(tags=["perf"])


@perf_router.post("/api/perf")
async def record_perf(samples: list[PerfSample]) -> Response:
    if len(samples) > MAX_BATCH_SIZE:
        samples = samples[:MAX_BATCH_SIZE]

    for s in samples:
        attrs = {"operation": s.operation}
        if s.metric == "worker_roundtrip":
            app_metrics.frontend_worker_roundtrip.record(s.value, attrs)
        elif s.metric == "worker_compute":
            app_metrics.frontend_worker_compute.record(s.value, attrs)
        elif s.metric == "main_thread":
            app_metrics.frontend_main_thread.record(s.value, attrs)
        elif s.metric == "feature_count":
            app_metrics.frontend_feature_count.record(s.value)

    return Response(status_code=204)