app.py

"""nova-connect: the web page and the room-ticket desk for a Nova Pi.

Runs as a Hugging Face Docker Space (port 7860) or any container host. It serves
the static UI and one endpoint: POST /api/session with the passcode returns a
LiveKit URL and a 10-minute token. It holds the LiveKit API secret; the browser
never sees it. Audio never passes through here -- browser and Pi meet at the SFU.

Configuration (Space secrets / environment):
  LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET  -- the SFU and its key pair
  NOVA_AGENT_NAME     -- the Pi's agent name, printed by the Pi at startup
                         ("LiveKit: registered as nova-<hostname>")
  NOVA_PASSCODE_HASH  -- from scripts/make_passcode.py
"""
from __future__ import annotations

import logging
import os
from dataclasses import dataclass
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

import token_service as ts

STATIC = Path(__file__).resolve().parent / "static"
logger = logging.getLogger("nova-connect")


@dataclass
class Settings:
    livekit_url: str
    api_key: str
    api_secret: str
    agent_name: str
    passcode_hash: str

    @classmethod
    def from_env(cls) -> Settings:
        e = os.environ.get
        return cls(
            livekit_url=e("LIVEKIT_URL", ""),
            api_key=e("LIVEKIT_API_KEY", ""),
            api_secret=e("LIVEKIT_API_SECRET", ""),
            agent_name=e("NOVA_AGENT_NAME", ""),
            passcode_hash=e("NOVA_PASSCODE_HASH", ""),
        )

    @property
    def configured(self) -> bool:
        return all(
            (self.livekit_url, self.api_key, self.api_secret, self.agent_name, self.passcode_hash)
        )


class SessionRequest(BaseModel):
    passcode: str


def _client_ip(request: Request) -> str:
    # Behind the HF / reverse proxy the socket peer is the proxy; the first
    # X-Forwarded-For hop is the caller.
    fwd = request.headers.get("x-forwarded-for", "")
    if fwd:
        return fwd.split(",")[0].strip()
    return request.client.host if request.client else "unknown"


def create_app(settings: Settings | None = None, *, limiter: ts.RateLimiter | None = None) -> FastAPI:
    settings = settings or Settings.from_env()
    limiter = limiter or ts.RateLimiter()
    app = FastAPI(title="nova-connect", docs_url=None, redoc_url=None)

    @app.get("/api/health")
    def health():
        return {"configured": settings.configured, "agent": settings.agent_name or None}

    @app.post("/api/session")
    def session(req: SessionRequest, request: Request):
        if not settings.configured:
            return JSONResponse({"error": "not_configured"}, status_code=503)
        if not limiter.allow(_client_ip(request)):
            return JSONResponse({"error": "rate_limited"}, status_code=429)
        if not ts.verify_passcode(req.passcode, settings.passcode_hash):
            return JSONResponse({"error": "bad_passcode"}, status_code=401)
        s = ts.mint_session(
            api_key=settings.api_key, api_secret=settings.api_secret, agent_name=settings.agent_name
        )
        logger.info("session issued room=%s", s.room)
        return {"url": settings.livekit_url, "token": s.token, "room": s.room, "expires_in": s.expires_in}

    @app.get("/")
    def index():
        return FileResponse(STATIC / "index.html")

    app.mount("/static", StaticFiles(directory=STATIC), name="static")
    return app


if __name__ == "__main__":
    import uvicorn

    logging.basicConfig(level=logging.INFO)
    uvicorn.run(create_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "7860")))