app.py

"""Write gateway for the Reachy Arcade global leaderboard.

Validates the caller's HF OAuth token via whoami, then merges their new score
into the Dataset's scores.json (one global best per game).
"""
import asyncio
import json
import logging
import os
from datetime import datetime, timezone
from typing import Optional

import httpx
from fastapi import FastAPI, Header, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from huggingface_hub import HfApi, hf_hub_download
from pydantic import BaseModel, Field

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("arcade-api")

DATASET_REPO = "cdeplanne/reachy-arcade-scores"
SCORES_FILENAME = "scores.json"
PERSONAL_BESTS_FILENAME = "personal_bests.json"
GAME_KEYS = {"spaceship", "pong", "duckrun", "flappy", "boss", "rhythm"}
NAME_MAX_LEN = 4

HF_TOKEN = os.environ.get("HF_TOKEN")
if not HF_TOKEN:
    log.warning("HF_TOKEN env var is unset — writes will fail until it's configured.")

api = HfApi(token=HF_TOKEN)
# One global mutex avoids interleaved read-modify-write on scores.json.
_write_lock = asyncio.Lock()


class ScoreIn(BaseModel):
    game: str = Field(..., description="One of the GAME_KEYS")
    score: int = Field(..., ge=0, le=10_000_000)
    name: str = Field("AAAA", min_length=1, max_length=8)


app = FastAPI(title="Reachy Arcade API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://cdeplanne-reachy-arcade.static.hf.space",
        "http://localhost:8765",
    ],
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["*"],
)


@app.get("/")
def health():
    return {"ok": True, "service": "reachy-arcade-api"}


async def _whoami(token: str) -> Optional[str]:
    """Return the HF username for `token`, or None if invalid."""
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(
            "https://huggingface.co/api/whoami-v2",
            headers={"Authorization": f"Bearer {token}"},
        )
    if r.status_code != 200:
        return None
    data = r.json()
    return data.get("name") or data.get("preferred_username") or None


def _load_json(filename: str) -> dict:
    """Pull the latest copy of `filename` from the Dataset, or {} if missing."""
    try:
        path = hf_hub_download(
            repo_id=DATASET_REPO,
            repo_type="dataset",
            filename=filename,
            force_download=True,  # always grab the live version
            token=HF_TOKEN,
        )
        with open(path) as f:
            return json.load(f)
    except Exception as e:
        log.warning("could not load %s (%s); starting empty", filename, e)
        return {}


def _save_json(filename: str, data: dict, commit_message: str) -> None:
    payload = json.dumps(data, separators=(",", ":")).encode("utf-8")
    api.upload_file(
        path_or_fileobj=payload,
        path_in_repo=filename,
        repo_id=DATASET_REPO,
        repo_type="dataset",
        commit_message=commit_message,
    )


@app.post("/score")
async def post_score(body: ScoreIn, authorization: str = Header(default="")):
    if not HF_TOKEN:
        raise HTTPException(503, "Backend missing HF_TOKEN")
    if not authorization.lower().startswith("bearer "):
        raise HTTPException(401, "Missing bearer token")
    user_token = authorization.split(" ", 1)[1].strip()
    if not user_token:
        raise HTTPException(401, "Empty bearer token")

    username = await _whoami(user_token)
    if not username:
        raise HTTPException(401, "Invalid HF token")

    if body.game not in GAME_KEYS:
        raise HTTPException(400, f"Unknown game '{body.game}'")

    clean_name = (body.name or "AAAA").strip().upper()[:NAME_MAX_LEN] or "AAAA"

    now_iso = datetime.now(timezone.utc).isoformat(timespec="seconds")

    async with _write_lock:
        # ── personal best (per-user) ────────────────────────────────────────
        personal = _load_json(PERSONAL_BESTS_FILENAME)
        user_pb = personal.get(username) or {}
        prev_pb = int((user_pb.get(body.game) or {}).get("score") or 0)
        wrote_personal = False
        if body.score > prev_pb:
            user_pb[body.game] = {
                "score": body.score,
                "name": clean_name,
                "updated": now_iso,
            }
            personal[username] = user_pb
            wrote_personal = True

        # ── global best (across all users) ──────────────────────────────────
        scores = _load_json(SCORES_FILENAME)
        prev_global = scores.get(body.game) or {}
        prev_global_score = int(prev_global.get("score") or 0)
        wrote_global = False
        if body.score > prev_global_score:
            scores[body.game] = {
                "score": body.score,
                "name": clean_name,
                "owner": username,
                "updated": now_iso,
            }
            wrote_global = True

        if not wrote_personal and not wrote_global:
            return {
                "status": "not_best",
                "personal_best": prev_pb,
                "global_best": prev_global_score,
                "global_owner": prev_global.get("owner"),
            }

        try:
            if wrote_personal:
                _save_json(PERSONAL_BESTS_FILENAME, personal, f"PB: {username} {body.game}={body.score}")
            if wrote_global:
                _save_json(SCORES_FILENAME, scores, f"Global: {body.game}={body.score} by {username}")
        except Exception as e:
            log.exception("save failed")
            raise HTTPException(500, f"Could not save: {e}")

    return {
        "status": "saved",
        "personal_best": wrote_personal,
        "global_best": wrote_global,
        "game": body.game,
        "score": body.score,
        "owner": username,
    }