token_api.py

from __future__ import annotations

import os
import secrets
from datetime import timedelta

from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from livekit import api

load_dotenv()

app = FastAPI(title="Bedtime LiveKit Token API", version="0.1.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class AndroidTokenRequest(BaseModel):
    room_name: str | None = Field(default=None, description="Optional room to join")
    participant_name: str = Field(default="Kid Listener")
    participant_identity: str | None = Field(default=None)
    metadata: str | None = Field(default=None)


class AndroidTokenResponse(BaseModel):
    server_url: str
    room_name: str
    participant_name: str
    participant_identity: str
    participant_token: str


def _require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise HTTPException(status_code=500, detail=f"Missing required environment variable: {name}")
    return value


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/api/token/android", response_model=AndroidTokenResponse)
def create_android_token(payload: AndroidTokenRequest) -> AndroidTokenResponse:
    livekit_url = _require_env("LIVEKIT_URL")
    api_key = _require_env("LIVEKIT_API_KEY")
    api_secret = _require_env("LIVEKIT_API_SECRET")

    room_name = payload.room_name or f"bedtime-room-{secrets.token_hex(3)}"
    participant_identity = payload.participant_identity or f"android-{secrets.token_hex(4)}"

    token = (
        api.AccessToken(api_key, api_secret)
        .with_identity(participant_identity)
        .with_name(payload.participant_name)
        .with_metadata(payload.metadata or "android-bedtime-client")
        .with_ttl(timedelta(minutes=30))
        .with_grants(
            api.VideoGrants(
                room_join=True,
                room=room_name,
                can_publish=True,
                can_subscribe=True,
                can_publish_data=True,
            )
        )
        .to_jwt()
    )

    return AndroidTokenResponse(
        server_url=livekit_url,
        room_name=room_name,
        participant_name=payload.participant_name,
        participant_identity=participant_identity,
        participant_token=token,
    )