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

"""
Discord radio channels from the web node page (server-26#213 item 3, #109 item 13).

The node page used to take raw guild/channel IDs. It now picks from:

  * the Discord servers the caller's org owns — ``discord_guilds`` docs that
    resolve to the node's org (an explicit link, or the unlinked-server
    fallback org, same rule the bot uses: streams.resolve_guild_org), and
  * their voice channels, from a directory the DRB server bot pushes
    (``PUT /discord/guilds/{id}/directory``). The bot already holds its
    gateway cache of every server it is in; C2 never calls Discord with a
    token on the browser's behalf and no token string ever reaches the browser.

A pick becomes a ``discord_streams`` desired-state doc **pinned to this node**
(streams.start_stream(node_id=...)), exactly what the old raw-ID path did, with
the checks that path skipped: the server must belong to the node's org, the
channel must be a voice channel the bot reported in that server, and the
per-server and per-node stream caps apply.

Auth:
  * PUT  /discord/guilds/{id}/directory          — service key (the server bot) only.
  * GET  /discord/nodes/{node_id}/join-options   — any signed-in member of the node's org
                                                   (or a platform admin); ``can_control`` says
                                                   whether the caller may start/stop.
  * POST /discord/nodes/{node_id}/streams        — platform admin, the org's owner, or an
  * POST /discord/nodes/{node_id}/streams/stop     operator whose owned_node_ids lists the node.
  * DELETE /discord/nodes/{node_id}/streams/{id}  — platform admin or the org's owner only
                                                   (permanent: stop first, then delete).
"""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Path
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

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.audit import write_audit
from app.internal.auth import (
    describe_actor,
    get_org_role,
    get_role,
    require_firebase_token,
    require_service_key,
)
from app.internal.logger import logger

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

MAX_CHANNELS = 200
_NAME_MAX = 100


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


# ---------------------------------------------------------------------------
# Directory — pushed by the DRB server bot from its gateway cache
# ---------------------------------------------------------------------------

class VoiceChannelIn(BaseModel):
    channel_id: str = Field(..., pattern=r"^[0-9]{1,25}$")
    name: str = Field(..., max_length=_NAME_MAX)
    category: Optional[str] = Field(None, max_length=_NAME_MAX)
    position: int = 0


class GuildDirectory(BaseModel):
    name: Optional[str] = Field(None, max_length=_NAME_MAX)
    present: bool = True
    voice_channels: list[VoiceChannelIn] = Field(default_factory=list, max_length=MAX_CHANNELS)


@router.put("/guilds/{guild_id}/directory")
async def put_guild_directory(guild_id: str, body: GuildDirectory, _: dict = Depends(require_service_key)):
    """The server bot's view of one Discord server: its name and voice channels.
    Merged onto ``discord_guilds/{id}`` — never touches ``org_id`` (the link)."""
    if not guild_id.isdigit():
        raise HTTPException(400, "guild_id must be a Discord snowflake.")
    data: dict = {"guild_id": guild_id, "bot_present": body.present, "directory_at": _now_iso()}
    if body.present:
        data["voice_channels"] = [c.model_dump() for c in body.voice_channels]
        if body.name:
            # Same field the /drb link flow writes (feat/213-guild-link, PR #218).
            data["guild_name"] = body.name
    await fstore.doc_set(streams.GUILDS, guild_id, data)
    return {"ok": True, "guild_id": guild_id, "voice_channels": len(body.voice_channels)}


# ---------------------------------------------------------------------------
# Node page — picker
# ---------------------------------------------------------------------------

async def _node_for_caller(node_id: str, decoded: dict) -> tuple[dict, bool]:
    """The node, and whether this caller may start/stop its radio channels.
    A node outside the caller's org is a 404, never a 403 (no existence leak)."""
    node = await fstore.doc_get("nodes", node_id)
    if not node:
        raise HTTPException(404, f"Node '{node_id}' not found.")
    if get_role(decoded) == "admin":
        return node, True
    org_id = decoded.get("org_id")
    if not org_id or node.get("org_id") != org_id:
        raise HTTPException(404, f"Node '{node_id}' not found.")
    can = get_org_role(decoded) == "owner" or (
        get_role(decoded) == "operator" and node_id in (decoded.get("owned_node_ids") or [])
    )
    return node, can


def _can_remove(decoded: dict) -> bool:
    """Permanent removal is narrower than start/stop: platform admin or the
    org's owner (``_node_for_caller`` has already pinned the org), never a
    node operator."""
    return get_role(decoded) == "admin" or get_org_role(decoded) == "owner"


def _guild_name(g: dict) -> Optional[str]:
    return g.get("guild_name") or g.get("name")


def _require_control(can: bool) -> None:
    if not can:
        raise HTTPException(403, "Only the organization owner, an admin, or this node's operator can do that.")


async def _org_guilds(org_id: str) -> list[dict]:
    """``discord_guilds`` docs this org may stream into, bot still present.
    Mirrors streams.resolve_guild_org: explicit link, else the unlinked-server
    fallback org — except a server an org explicitly unlinked (``unlinked_at``,
    PR #218), which gets no fallback."""
    fallback = settings.discord_unlinked_guild_org_id or None
    out = []
    for g in await fstore.collection_list(streams.GUILDS):
        gid = str(g.get("guild_id") or "")
        if not gid or g.get("bot_present") is False:
            continue
        owner = g.get("org_id") or (None if g.get("unlinked_at") else fallback)
        if owner and owner == org_id:
            out.append(g)
    out.sort(key=lambda g: ((_guild_name(g) or "").lower(), g["guild_id"]))
    return out


def _bot_status(tokens: dict, *, node_id: str, guild_id: str, org_id: str) -> dict:
    res = leases.pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id)
    if res.ok:
        return {"ok": True, "code": None, "message": None, "invite": []}
    return {"ok": False, "code": res.error, "message": streams.ERROR_TEXT.get(res.error or "", res.error),
            "invite": [i for i in res.invite if i.get("invite_url")][:5]}


def _stream_view(s: dict, node_names: dict) -> dict:
    return {
        "stream_id": s.get("stream_id"),
        "guild_id": s.get("guild_id"),
        "channel_id": s.get("channel_id"),
        "guild_name": s.get("guild_name"),
        "channel_name": s.get("channel_name"),
        "system_name": s.get("system_name"),
        "mode": s.get("mode"),
        "node_id": s.get("node_id"),
        "node_name": node_names.get(s.get("node_id") or "") or s.get("node_id"),
        "actual": s.get("actual"),
        "updated_at": s.get("updated_at"),
    }


@router.get("/nodes/{node_id}/join-options")
async def join_options(node_id: str, decoded: dict = Depends(require_firebase_token)):
    node, can_control = await _node_for_caller(node_id, decoded)
    org_id = node.get("org_id")
    system_id = streams._effective_system(node)
    system = (await fstore.doc_get_cached("systems", system_id)) if system_id else None

    guilds = await _org_guilds(org_id) if org_id else []
    live = [s for s in await fstore.collection_list(streams.STREAMS, desired="on")
            if s.get("org_id") == org_id]
    nodes = {n.get("node_id"): n.get("name") for n in await fstore.collection_list("nodes", org_id=org_id)
             if n.get("node_id")} if org_id else {}
    by_channel = {s.get("stream_id"): s for s in live}
    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}

    out_guilds = []
    for g in guilds:
        gid = g["guild_id"]
        channels = []
        for c in sorted(g.get("voice_channels") or [],
                        key=lambda c: (c.get("position") or 0, (c.get("name") or "").lower())):
            s = by_channel.get(leases.stream_id_for(gid, str(c.get("channel_id"))))
            channels.append({
                "channel_id": str(c.get("channel_id")),
                "name": c.get("name"),
                "category": c.get("category"),
                "stream": _stream_view(s, nodes) if s else None,
            })
        out_guilds.append({
            "guild_id": gid,
            "name": _guild_name(g),
            "linked": bool(g.get("org_id")),
            "synced": g.get("voice_channels") is not None,
            "synced_at": g.get("directory_at"),
            "voice_channels": channels,
            "streams_on": sum(1 for s in live if s.get("guild_id") == gid),
            "bot": _bot_status(tokens, node_id=node_id, guild_id=gid, org_id=org_id),
        })

    here = [s for s in live if s.get("node_id") == node_id]
    here.sort(key=lambda s: (s.get("guild_name") or "", s.get("channel_name") or "", s.get("stream_id") or ""))
    return {
        "node_id": node_id,
        "node_name": node.get("name"),
        "node_online": node.get("status") in streams.ONLINE,
        "system_id": system_id,
        "system_name": (system or {}).get("name") or system_id,
        "can_control": can_control,
        "can_remove": _can_remove(decoded),
        "limits": {
            "node_max": settings.discord_node_max_streams,
            "node_used": len(here),
            "guild_max": settings.discord_guild_max_streams,
        },
        "guilds": out_guilds,
        "streams": [_stream_view(s, nodes) for s in here],
    }


class NodeStreamStart(BaseModel):
    guild_id: str
    channel_id: str
    mode: str = "adhoc"


@router.post("/nodes/{node_id}/streams")
async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict = Depends(require_firebase_token)):
    node, can_control = await _node_for_caller(node_id, decoded)
    _require_control(can_control)
    org_id = node.get("org_id")
    if not org_id:
        raise HTTPException(409, "This node is not in an organization yet.")
    if body.mode not in ("persistent", "adhoc"):
        raise HTTPException(400, "mode must be persistent or adhoc.")
    system_id = streams._effective_system(node)
    if not system_id:
        raise HTTPException(409, "This node is not listening to a system yet.")

    guild_id, channel_id = str(body.guild_id), str(body.channel_id)
    guild = next((g for g in await _org_guilds(org_id) if g["guild_id"] == guild_id), None)
    if not guild:
        raise HTTPException(404, "That Discord server is not linked to this organization.")
    channel = next((c for c in guild.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
    if not channel:
        raise HTTPException(400, "That is not a voice channel DRB can see in this server. "
                                 "Refresh the list and pick again.")

    stream_id = leases.stream_id_for(guild_id, channel_id)
    existing = await fstore.doc_get(streams.STREAMS, stream_id) or {}
    already_here = existing.get("desired") == "on" and existing.get("node_id") == node_id
    if not already_here:
        used = sum(1 for s in await fstore.collection_list(streams.STREAMS, node_id=node_id)
                   if s.get("desired") == "on")
        if used >= settings.discord_node_max_streams:
            raise HTTPException(409, f"This node already carries {used} radio channels, its maximum. "
                                     "Stop one first.")

    try:
        doc = await streams.start_stream(
            org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id,
            node_id=node_id, mode=body.mode,
            channel_name=channel.get("name"), guild_name=_guild_name(guild),
            created_by=f"web:{decoded.get('uid') or 'unknown'}",
            max_streams=settings.discord_guild_max_streams,
        )
    except streams.StreamError as e:
        return JSONResponse(status_code=e.status,
                            content={"detail": e.message, "code": e.code, "invite": e.invite})
    logger.info(f"Web: radio channel {stream_id} pinned to node {node_id} ({body.mode})")
    return {"ok": True, "stream_id": doc["stream_id"], "token_id": doc.get("token_id"),
            "delivered": doc.get("delivered"), "reused": doc.get("reused"), "actual": doc.get("actual")}


class NodeStreamStop(BaseModel):
    # guild:channel snowflakes (leases.stream_id_for). Validated so a malformed
    # id (e.g. one with "/") is a 422, not a Firestore path error / 500.
    stream_id: Optional[str] = Field(None, pattern=r"^[0-9]{1,25}:[0-9]{1,25}$")


@router.post("/nodes/{node_id}/streams/stop")
async def stop_node_streams(node_id: str, body: NodeStreamStop, decoded: dict = Depends(require_firebase_token)):
    """Stop one radio channel this node carries, or (no stream_id) all of them."""
    _, can_control = await _node_for_caller(node_id, decoded)
    _require_control(can_control)
    if body.stream_id:
        doc = await fstore.doc_get(streams.STREAMS, body.stream_id)
        if not doc or doc.get("node_id") != node_id or doc.get("desired") != "on":
            raise HTTPException(404, "This node is not carrying that radio channel.")
        await streams.stop_stream(body.stream_id, "stopped_from_web")
        return {"ok": True, "stopped": [body.stream_id]}
    return {"ok": True, "stopped": await streams.stop_node(node_id, "stopped_from_web")}


@router.delete("/nodes/{node_id}/streams/{stream_id}")
async def remove_node_stream(node_id: str,
                             stream_id: str = Path(..., pattern=r"^[0-9]{1,25}:[0-9]{1,25}$"),
                             decoded: dict = Depends(require_firebase_token)):
    """Permanently remove a radio channel this node carries (or last carried):
    stopped through desired state first so the node logs the bot out and the
    lease is released on its ack (server-26#216), then the doc is deleted."""
    node, _ = await _node_for_caller(node_id, decoded)
    if not _can_remove(decoded):
        raise HTTPException(403, "Only the organization owner or an admin can remove a radio channel.")
    doc = await fstore.doc_get(streams.STREAMS, stream_id)
    # Same org AND this node: an owner can never reach another org's channel
    # through their own node's URL.
    if not doc or doc.get("node_id") != node_id or doc.get("org_id") != node.get("org_id"):
        raise HTTPException(404, "This node is not carrying that radio channel.")
    try:
        await streams.delete_stream(stream_id, "removed_from_web")
    except streams.StreamError as e:
        return JSONResponse(status_code=e.status, content={"detail": e.message, "code": e.code})
    uid, email = describe_actor(decoded)
    await write_audit(uid, email, "discord.stream.delete", details={
        "stream_id": stream_id, "guild_id": doc.get("guild_id"), "channel_id": doc.get("channel_id"),
        "org_id": doc.get("org_id"), "node_id": node_id, "was_desired": doc.get("desired"),
        "via": "node_page",
    })
    return {"ok": True, "stream_id": stream_id}