app/insights_api/main.py

"""HTTP surface for insights-api. No outbound calls."""

import time

from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from pydantic import BaseModel, Field

from insights_api.buffer import RingBuffer
from insights_api.metrics import render_metrics

app = FastAPI(title="insights-api", version="0.1.0")
_started = time.time()
_accepted = 0
_buffer = RingBuffer(1024)


class EventIn(BaseModel):
    product_id: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")
    event_type: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")
    ts: int = Field(ge=0)


def reset_state(capacity=1024):
    global _accepted, _buffer, _started
    _accepted = 0
    _buffer = RingBuffer(capacity)
    _started = time.time()


def _valid_product_id(product_id):
    if not product_id or len(product_id) > 64:
        return False
    allowed = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-")
    return all(char in allowed for char in product_id)


@app.get("/healthz")
def healthz():
    return {"status": "ok"}


@app.get("/readyz")
def readyz():
    return {"status": "ready"}


@app.post("/v1/events", status_code=202)
def post_event(body: EventIn):
    global _accepted
    _buffer.add(body.model_dump())
    _accepted += 1
    return {"accepted": True}


@app.get("/v1/insights/{product_id}")
def insights(product_id: str):
    if not _valid_product_id(product_id):
        raise HTTPException(status_code=400, detail="invalid product_id")
    return _buffer.counts_for(product_id)


@app.get("/metrics")
def metrics():
    body = render_metrics(len(_buffer), _accepted, time.time() - _started)
    return Response(content=body, media_type="text/plain; version=0.0.4; charset=utf-8")