server-26/drb-c2-core/app/routers/tokens.py

import uuid
from fastapi import APIRouter, HTTPException, Depends, Query
from pydantic import BaseModel
from typing import Optional
from app.internal import firestore as fstore
from app.internal import discord_leases as leases
from app.internal import discord_rest
from app.internal.audit import write_audit
from app.internal.auth import (
    describe_actor,
    get_org_role,
    get_role,
    require_admin_token,
    require_firebase_token,
    resolve_org_scope,
)
from app.internal.logger import logger

router = APIRouter(prefix="/tokens", tags=["tokens"])


class TokenCreate(BaseModel):
    name: str        # friendly label e.g. "DRB Bot 1"
    token: str       # the actual Discord bot token
    # None = platform pool. An org id = that org's bring-your-own bot, never
    # leased to anyone else (discord_leases.pick_token).
    org_id: Optional[str] = None


def _masked(t: dict) -> dict:
    out = {**t, "token": "•••" + (t.get("token") or "")[-4:]}
    out["invite_url"] = leases.invite_url(t.get("application_id"))
    return out


async def _republish_after_release(lease: Optional[dict]) -> None:
    if lease and lease.get("node_id"):
        from app.internal.discord_streams import publish_desired
        await publish_desired(lease["node_id"])


# ---------------------------------------------------------------------------
# CRUD — platform admin only (server-26#204 item 6: the list used to be open
# to any signed-in user of any org)
# ---------------------------------------------------------------------------

@router.get("")
async def list_tokens(_: dict = Depends(require_admin_token)):
    """List all tokens. The actual token string is masked for safety."""
    return [_masked(t) for t in await fstore.collection_list("bot_tokens")]


@router.post("", status_code=201)
async def add_token(body: TokenCreate, _: dict = Depends(require_admin_token)):
    """Add a radio bot. The token is checked against Discord first
    (server-26#204 item 5): a string Discord answers 401 for is refused; if
    Discord is merely unreachable the token is stored with membership unknown."""
    try:
        info = await discord_rest.inspect_token(body.token.strip())
    except discord_rest.TokenRejected as e:
        raise HTTPException(400, str(e))
    token_id = str(uuid.uuid4())
    doc = {
        "token_id": token_id,
        "name": body.name,
        "token": body.token.strip(),
        "org_id": body.org_id,
        "lease": None,
        "in_use": False,
        "assigned_node_id": None,
        "assigned_at": None,
        **info,
    }
    await fstore.doc_set("bot_tokens", token_id, doc, merge=False)
    return {"token_id": token_id, "name": body.name, "valid": info.get("valid"),
            "guild_ids": info.get("guild_ids"), "check_error": info.get("check_error"),
            "invite_url": leases.invite_url(info.get("application_id"))}


@router.post("/{token_id}/refresh", status_code=200)
async def refresh_token(token_id: str, _: dict = Depends(require_admin_token)):
    """Re-check a bot with Discord (valid? which servers is it in?)."""
    if not await fstore.doc_get("bot_tokens", token_id):
        raise HTTPException(404, "Token not found.")
    await discord_rest.refresh_membership([token_id])
    return _masked(await fstore.doc_get("bot_tokens", token_id) or {})


@router.post("/flush", status_code=200)
async def flush_tokens(_: dict = Depends(require_admin_token)):
    """Force-release every leased token (admin last resort). Each holding node
    is sent a desired-state list without the token, so it logs the bot out;
    persistent streams that lost their token are re-placed by the reconciler."""
    released = 0
    for t in await fstore.collection_list("bot_tokens"):
        if not (t.get("lease") or t.get("in_use")):
            continue
        lease = await leases.force_release_token(t["token_id"])
        released += 1
        await _republish_after_release(lease)
    logger.warning(f"Admin flushed {released} radio-bot token lease(s)")
    return {"released": released}


def _forget_cached_system(system_id: str) -> None:
    """start_stream reads systems through fstore.doc_get_cached (5 min TTL):
    drop the entry so a new preference applies to the next start instead of
    up to five minutes later (server-26#225 S4)."""
    cache = getattr(fstore, "_doc_cache", None)
    path = getattr(fstore, "_path", None)
    if isinstance(cache, dict) and callable(path):
        cache.pop(f"{path('systems')}/{system_id}", None)


def _may_edit_system(decoded: dict, system: Optional[dict]) -> bool:
    """Platform admin: any system. Org owner: only their own org's systems —
    a shared system (no org_id) serves every org, so a preference there is
    platform-admin only (owner decision Q2, 2026-10-05)."""
    if get_role(decoded) == "admin":
        return True
    return bool(system and system.get("org_id") and system.get("org_id") == decoded.get("org_id"))


def _check_prefer_access(decoded: dict, token: Optional[dict], system: Optional[dict],
                         system_id: Optional[str]) -> None:
    """Raise unless the caller may point ``system`` at ``token``. Another
    org's bot or system is a 404, never a 403 (no existence leak)."""
    if get_role(decoded) == "admin":
        if not token:
            raise HTTPException(404, "Token not found.")
        if system_id is not None and not system:
            raise HTTPException(404, "System not found.")
        return
    org_id = decoded.get("org_id")
    if not org_id or get_org_role(decoded) != "owner":
        raise HTTPException(403, "Only the organization owner or a platform admin can choose a system's radio bot.")
    if not token or (token.get("org_id") is not None and token.get("org_id") != org_id):
        raise HTTPException(404, "Token not found.")
    if token.get("org_id") is None:
        raise HTTPException(403, "Only a platform admin can make a shared DRB radio bot preferred. "
                                 "Organization owners can choose their own organization's radio bots.")
    if system_id is None:
        return
    if not system or (system.get("org_id") and system.get("org_id") != org_id):
        raise HTTPException(404, "System not found.")
    if not system.get("org_id"):
        raise HTTPException(403, "That system is shared by every organization; only a platform admin can "
                                 "choose its radio bot.")


@router.put("/{token_id}/prefer/{system_id}", status_code=200)
async def set_preferred_system(
    token_id: str,
    system_id: str,
    decoded: dict = Depends(require_firebase_token),
):
    """
    Mark this token as the preferred bot for a system: ``start_stream`` tries
    it first for that system's radio channels when nobody named a bot
    (server-26#225 S3), and uses another bot when it can't take the stream.
    Pass system_id="_none" to clear the preference.

    Who (owner decision Q2, 2026-10-05): a platform admin for any bot and any
    system; an org owner for their OWN org's systems and their OWN
    bring-your-own bots. Shared systems (no org_id) and pool bots stay
    platform-admin only.
    """
    if "/" in token_id or "/" in system_id:
        raise HTTPException(400, "Bad id.")
    existing = await fstore.doc_get("bot_tokens", token_id)
    uid, email = describe_actor(decoded)
    if system_id == "_none":
        _check_prefer_access(decoded, existing, None, None)
        old_id = existing.get("preferred_for_system_id") or ""
        system_doc = await fstore.doc_get("systems", old_id) if old_id else None
        if system_doc and system_doc.get("preferred_token_id") == token_id:
            if not _may_edit_system(decoded, system_doc):
                raise HTTPException(403, "That system is shared by every organization; only a platform admin "
                                         "can change its radio bot.")
            await fstore.doc_set("systems", old_id, {"preferred_token_id": None})
            _forget_cached_system(old_id)
        await fstore.doc_set("bot_tokens", token_id, {"preferred_for_system_id": None})
        await write_audit(uid, email, "discord.token.prefer", details={
            "token_id": token_id, "system_id": None, "previous_system_id": old_id or None})
        return {"ok": True, "preferred_for_system_id": None}

    system_doc = await fstore.doc_get("systems", system_id)
    _check_prefer_access(decoded, existing, system_doc, system_id)

    # Keep both pointers single-valued: this bot stops being preferred for the
    # system it was preferred for (when the caller may edit that system), and
    # the bot this system preferred before loses its back-pointer.
    old_sys_id = existing.get("preferred_for_system_id")
    if old_sys_id and old_sys_id != system_id:
        old_sys = await fstore.doc_get("systems", old_sys_id)
        if old_sys and old_sys.get("preferred_token_id") == token_id and _may_edit_system(decoded, old_sys):
            await fstore.doc_set("systems", old_sys_id, {"preferred_token_id": None})
            _forget_cached_system(old_sys_id)
    old_tok_id = system_doc.get("preferred_token_id")
    if old_tok_id and old_tok_id != token_id:
        old_tok = await fstore.doc_get("bot_tokens", old_tok_id)
        if old_tok and old_tok.get("preferred_for_system_id") == system_id:
            await fstore.doc_set("bot_tokens", old_tok_id, {"preferred_for_system_id": None})

    await fstore.doc_set("systems", system_id, {"preferred_token_id": token_id})
    _forget_cached_system(system_id)
    await fstore.doc_set("bot_tokens", token_id, {"preferred_for_system_id": system_id})
    await write_audit(uid, email, "discord.token.prefer", details={
        "token_id": token_id, "system_id": system_id, "previous_token_id": old_tok_id,
        "org_id": system_doc.get("org_id")})
    return {"ok": True, "preferred_for_system_id": system_id}


@router.get("/org/preferred")
async def org_preferred_bots(
    org_id: Optional[str] = Query(None, description="Platform-admin only: act for this org."),
    decoded: dict = Depends(require_firebase_token),
):
    """What an org owner needs to choose a preferred radio bot per system
    (server-26#225 S4, owner decision Q2): the org's own systems, each with
    its preferred bot, and the org's own bring-your-own bots — the only bots
    an owner may choose (PUT /tokens/{id}/prefer/{system_id}). GET /tokens
    stays platform-admin only; this returns no pool bots and no token strings.
    A pool bot a platform admin chose for one of these systems is flagged,
    not named."""
    if get_role(decoded) != "admin" and get_org_role(decoded) != "owner":
        raise HTTPException(403, "Only the organization owner or a platform admin can choose a system's radio bot.")
    org = resolve_org_scope(decoded, org_id)
    systems = sorted(await fstore.collection_list("systems", org_id=org),
                     key=lambda s: str(s.get("name") or s.get("system_id") or "").lower())
    bots = sorted((t for t in await fstore.collection_list("bot_tokens", org_id=org) if t.get("token_id")),
                  key=lambda t: str(t.get("name") or "").lower())
    names = {t["token_id"]: t.get("name") or t["token_id"] for t in bots}
    out_systems = []
    for s in systems:
        if not s.get("system_id"):
            continue
        pref = s.get("preferred_token_id")
        out_systems.append({
            "system_id": s["system_id"],
            "name": s.get("name") or s["system_id"],
            "preferred_token_id": pref if pref in names else None,
            "preferred_is_platform_bot": bool(pref) and pref not in names,
            "preferred_name": names.get(pref) if pref in names else None,
        })
    return {
        "org_id": org,
        "systems": out_systems,
        "bots": [{
            "token_id": t["token_id"],
            "name": names[t["token_id"]],
            "valid": t.get("valid"),
            "server_count": len(t["guild_ids"]) if isinstance(t.get("guild_ids"), list) else None,
            "preferred_for_system_id": t.get("preferred_for_system_id"),
            "invite_url": leases.invite_url(t.get("application_id")),
        } for t in bots],
    }


@router.delete("/{token_id}", status_code=204)
async def delete_token(token_id: str, _: dict = Depends(require_admin_token)):
    existing = await fstore.doc_get("bot_tokens", token_id)
    if not existing:
        raise HTTPException(404, "Token not found.")
    if existing.get("lease"):
        raise HTTPException(409, "Token is currently leased to a node. Stop its radio channels (or flush) first.")
    await fstore.doc_delete("bot_tokens", token_id)