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

"""
Discord radio channels — the HTTP face of internal/discord_streams.py.

Every bot-facing route is scoped to ONE Discord server (``guild_id`` in the
path or body) and acts for the org that server is linked to
(server-26#203): /leave in server A can no longer touch server B, and the
system/node/bot lists a server sees are its own org's.

Auth:
  * bot routes — the Discord bot's service key, or a platform admin.
  * PUT /discord/guilds/{id}/link — platform admin only.
  * GET /discord/node/{node_id}/tokens/{token_id} — that node's own API key;
    the token string is handed only to the node currently holding its lease.
"""
from __future__ import annotations

import secrets
from datetime import datetime, timezone
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Security
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel

from app.config import settings
from app.internal import discord_leases as leases
from app.internal import discord_streams as streams
from app.internal import firestore as fstore
from app.internal.auth import require_admin_token, require_service_key_or_admin
from app.internal.logger import logger

router = APIRouter(prefix="/discord", tags=["discord"])
_bearer = HTTPBearer(auto_error=False)


def _error(e: streams.StreamError) -> JSONResponse:
    return JSONResponse(status_code=e.status,
                        content={"detail": e.message, "code": e.code, "invite": e.invite})


async def _org_or_403(guild_id: str) -> str:
    org_id = await streams.resolve_guild_org(guild_id)
    if not org_id:
        raise HTTPException(403, streams.ERROR_TEXT["no_org"])
    return org_id


def _public_stream(s: dict) -> dict:
    keep = ("stream_id", "guild_id", "channel_id", "channel_name", "system_id", "system_name", "mode",
            "desired", "node_id", "token_id", "actual", "stop_reason", "updated_at", "delivered", "reused")
    return {k: s.get(k) for k in keep if k in s}


# ---------------------------------------------------------------------------
# Bot routes
# ---------------------------------------------------------------------------

@router.get("/guilds/{guild_id}/overview")
async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or_admin)):
    """Everything one Discord server may see: its org's systems and nodes, its
    own streams, and which radio bots could serve it. No token strings."""
    org_id = await streams.resolve_guild_org(guild_id)
    if not org_id:
        return {"guild_id": guild_id, "org_id": None, "linked": False,
                "systems": [], "nodes": [], "streams": [], "bots": []}
    link = await fstore.doc_get(streams.GUILDS, guild_id)
    systems = await fstore.collection_list("systems", org_id=org_id)
    nodes = await fstore.collection_list("nodes", org_id=org_id)
    gstreams = [s for s in await fstore.collection_list(streams.STREAMS, guild_id=guild_id)
                if s.get("desired") == "on"]
    bots = []
    for t in await fstore.collection_list(leases.TOKENS):
        if t.get("org_id") not in (None, org_id) or t.get("valid") is False:
            continue
        lease = t.get("lease") or {}
        member = t.get("guild_ids")
        bots.append({
            "token_id": t.get("token_id"),
            "name": t.get("name"),
            "busy_here": any(leases.guild_of(s) == guild_id for s in lease.get("streams") or []),
            "member": None if member is None else guild_id in member,
            "invite_url": leases.invite_url(t.get("application_id"), guild_id),
        })
    return {
        "guild_id": guild_id,
        "org_id": org_id,
        "linked": bool(link and link.get("org_id")),
        "systems": [{"system_id": s.get("system_id"), "name": s.get("name")} for s in systems],
        "nodes": [{"node_id": n.get("node_id"), "name": n.get("name"), "status": n.get("status"),
                   "system_id": streams._effective_system(n)} for n in nodes],
        "streams": [_public_stream(s) for s in gstreams],
        "bots": bots,
    }


class StreamStart(BaseModel):
    guild_id: str
    channel_id: str
    system_id: str
    mode: str = "adhoc"
    preferred_token_id: Optional[str] = None
    channel_name: Optional[str] = None
    guild_name: Optional[str] = None
    created_by: Optional[str] = None


@router.post("/streams")
async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_or_admin)):
    # The bot always sends snowflakes; the web admin page types them in.
    if not (str(body.guild_id).isdigit() and str(body.channel_id).isdigit()):
        raise HTTPException(400, "guild_id and channel_id must be Discord ids (digits only).")
    org_id = await _org_or_403(body.guild_id)
    system = await fstore.doc_get_cached("systems", body.system_id)
    if not system or system.get("org_id") not in (None, org_id):
        raise HTTPException(404, "System not found.")
    try:
        doc = await streams.start_stream(
            org_id=org_id, guild_id=body.guild_id, channel_id=body.channel_id, system_id=body.system_id,
            mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name,
            created_by=body.created_by, preferred_token_id=body.preferred_token_id,
            max_streams=settings.discord_guild_max_streams,
        )
    except streams.StreamError as e:
        return _error(e)
    return _public_stream(doc)


@router.get("/streams/{stream_id}")
async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_admin)):
    doc = await fstore.doc_get(streams.STREAMS, stream_id)
    if not doc:
        raise HTTPException(404, "Stream not found.")
    return _public_stream(doc)


class GuildStop(BaseModel):
    channel_id: Optional[str] = None
    reason: str = "stopped"


@router.post("/guilds/{guild_id}/stop")
async def stop_guild(guild_id: str, body: GuildStop, _: dict = Depends(require_service_key_or_admin)):
    stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id)
    return {"stopped": stopped}


class GuildLink(BaseModel):
    org_id: str
    # Display label only (the admin page); Discord never reads it.
    guild_name: Optional[str] = None


@router.put("/guilds/{guild_id}/link")
async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(require_admin_token)):
    if not guild_id.isdigit():
        raise HTTPException(400, "guild_id must be a Discord server id (digits only).")
    org = await fstore.doc_get("organizations", body.org_id)
    if not org:
        raise HTTPException(404, "Organization not found.")
    previous_org = await streams.resolve_guild_org(guild_id)
    doc = {
        "guild_id": guild_id, "org_id": body.org_id,
        "linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(),
    }
    if body.guild_name and body.guild_name.strip():
        doc["guild_name"] = body.guild_name.strip()[:100]
    await fstore.doc_set(streams.GUILDS, guild_id, doc)
    # Live channels in this server were placed on the previous org's nodes;
    # left running, the reconciler would keep re-placing them there, so one
    # org's equipment would keep serving a server that now belongs to another.
    stopped: list[str] = []
    if previous_org and previous_org != body.org_id:
        stopped = await streams.stop_guild(guild_id, "server linked to another organization")
    return {"ok": True, "guild_id": guild_id, "org_id": body.org_id, "stopped": stopped}


# ---------------------------------------------------------------------------
# Admin view — the web Discord page (server-26#213 item 2)
# ---------------------------------------------------------------------------

@router.get("/admin/overview")
async def admin_overview(_: dict = Depends(require_admin_token)):
    """Every Discord server DRB knows about, every radio channel (desired vs
    actual) and what the page needs to act on them. Platform admin only.

    A server is "known" when it is linked (``discord_guilds``) or when it has
    ever had a radio channel — an unlinked server that ran ``/join`` acts for
    ``DISCORD_UNLINKED_GUILD_ORG_ID`` and should be visible so it can be linked.
    Radio bots come from ``GET /tokens``; no token string is returned here.
    Discord user ids (``created_by_discord_id``) and the linking admin's uid
    are deliberately left out.
    """
    links = {str(g.get("guild_id")): g for g in await fstore.collection_list(streams.GUILDS)
             if g.get("guild_id")}
    all_streams = await fstore.collection_list(streams.STREAMS)
    orgs = {o.get("org_id"): o for o in await fstore.collection_list("organizations") if o.get("org_id")}
    nodes = {n.get("node_id"): n for n in await fstore.collection_list("nodes") if n.get("node_id")}
    tokens = {t.get("token_id"): t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
    fallback = settings.discord_unlinked_guild_org_id or None

    guilds: dict[str, dict] = {}
    for gid, link in links.items():
        guilds[gid] = {"guild_id": gid, "guild_name": link.get("guild_name"), "linked": bool(link.get("org_id")),
                       "org_id": link.get("org_id"), "linked_at": link.get("linked_at")}
    rows = []
    for s in all_streams:
        gid = str(s.get("guild_id") or leases.guild_of(s.get("stream_id") or ""))
        g = guilds.setdefault(gid, {"guild_id": gid, "guild_name": None, "linked": False,
                                    "org_id": None, "linked_at": None})
        if s.get("guild_name") and not g.get("guild_name"):
            g["guild_name"] = s["guild_name"]
        node = nodes.get(s.get("node_id")) or {}
        tok = tokens.get(s.get("token_id")) or {}
        rows.append({**_public_stream(s), "org_id": s.get("org_id"), "guild_name": s.get("guild_name"),
                     "created_at": s.get("created_at"), "node_name": node.get("name"),
                     "node_status": node.get("status"), "token_name": tok.get("name")})
    for g in guilds.values():
        acting = g["org_id"] if g["linked"] else fallback
        g["acting_org_id"] = acting
        g["acting_org_name"] = (orgs.get(acting) or {}).get("name") if acting else None
        g["live_streams"] = sum(1 for r in rows if r["guild_id"] == g["guild_id"] and r.get("desired") == "on")

    rows.sort(key=lambda r: (0 if r.get("desired") == "on" else 1, r.get("guild_name") or r.get("guild_id") or "",
                             r.get("channel_name") or r.get("channel_id") or ""))
    systems = await fstore.collection_list("systems")
    # Names for every node a stream or a radio-bot lease points at.
    referenced = {r.get("node_id") for r in rows} | {(t.get("lease") or {}).get("node_id") for t in tokens.values()}
    node_names = {nid: nodes[nid].get("name") or nid for nid in referenced if nid in nodes}
    return {
        "guilds": sorted(guilds.values(), key=lambda g: (not g["linked"], g.get("guild_name") or g["guild_id"])),
        "streams": rows,
        "orgs": sorted(({"org_id": oid, "name": o.get("name") or oid} for oid, o in orgs.items()),
                       key=lambda o: o["name"].lower()),
        "node_names": node_names,
        "systems": [{"system_id": s.get("system_id"), "name": s.get("name"), "org_id": s.get("org_id")}
                    for s in systems if s.get("system_id")],
        "limits": {"guild_max_streams": settings.discord_guild_max_streams,
                   "node_max_streams": settings.discord_node_max_streams,
                   "unlinked_guild_org_id": fallback,
                   "unlinked_guild_org_name": (orgs.get(fallback) or {}).get("name") if fallback else None},
    }


# ---------------------------------------------------------------------------
# Node route — token pickup
# ---------------------------------------------------------------------------

@router.get("/node/{node_id}/tokens/{token_id}")
async def node_token(
    node_id: str,
    token_id: str,
    credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer),
):
    """The bot token string, for the node that holds its lease and nobody else."""
    if not credentials:
        raise HTTPException(401, "Missing authorization")
    key_doc = await fstore.doc_get("node_keys", node_id)
    stored = (key_doc or {}).get("api_key") or ""
    if not stored or not secrets.compare_digest(stored, credentials.credentials):
        raise HTTPException(401, "Invalid node API key")
    tok = await fstore.doc_get(leases.TOKENS, token_id)
    lease = (tok or {}).get("lease") or {}
    if not tok or lease.get("node_id") != node_id or lease.get("releasing"):
        logger.warning(f"Node {node_id} asked for token {token_id} it does not hold a lease on")
        raise HTTPException(403, "This node does not hold a lease on that token.")
    return {"token_id": token_id, "token": tok["token"], "lease_id": lease.get("lease_id"),
            "name": tok.get("name")}