Logan CusanoandClaude Opus 5.5
a976464e72
A server linked to an org now shows in the node picker while a radio bot the org may use is in it, even when the DRB server bot is absent or left. Its voice-channel list is read from that radio bot's cached REST snapshot (discord_rest.guild_directory, the same snapshot the permission check reads, so no extra Discord request): inline when no list is stored, in the background when the list is older than discord_directory_refresh_seconds. The server bot's own fresh list still wins (back-compat) and its directory push now drops the server's cached snapshots. Guards: a server acting only through the unlinked-server fallback org still needs the server bot; REST never creates a discord_guilds doc, never writes org_id or bot_present, and never writes onto a server unlinked meanwhile. Also fixes server-26#251: an interactive start (/join, admin page, web picker, node discord_join) re-reads a cached "denied" snapshot older than 10 s once before refusing, so "fix it in Discord, then try again" works on the next try. The reconciler never re-reads (no REST loop). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
535 lines
25 KiB
Python
535 lines
25 KiB
Python
"""
|
|
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 kept on ``discord_guilds/{id}``.
|
|
|
|
Where the directory comes from (server-26#225 S5 — the main bot is optional):
|
|
|
|
* the DRB server bot, when it is in the server: it pushes its gateway view
|
|
(``PUT /discord/guilds/{id}/directory``) on start, on changes and every
|
|
15 minutes (``directory_source: "main_bot"``);
|
|
* otherwise a radio bot that is in the server: C2 reads the channel list
|
|
over Discord REST with that bot's token (discord_rest.guild_directory —
|
|
the same cached snapshot the permission check reads, so no extra
|
|
request) and stores it (``directory_source: "radio_bot"``,
|
|
``radio_bot_present``). That happens when the picker is opened and the
|
|
stored list is missing (inline, inside the permission-check time budget)
|
|
or older than ``discord_directory_refresh_seconds`` (in the background).
|
|
|
|
So C2 does call Discord when a signed-in user opens the picker — with a
|
|
radio bot's token, for a server already linked to the caller's org, through
|
|
the rate-limited, cached client. No token string ever reaches the browser.
|
|
|
|
Which servers an org sees (``_org_guilds``):
|
|
|
|
* linked to the org: shown unless the server bot said it left AND no
|
|
radio bot the org may use is in it (``bot_tokens.guild_ids``);
|
|
* acting for the org only through the unlinked-server fallback: shown only
|
|
while the server bot is there, exactly as before. A radio bot's
|
|
membership never discovers a server, and REST never creates a
|
|
``discord_guilds`` doc.
|
|
|
|
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 in that server's directory, 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
|
|
|
|
import asyncio
|
|
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_rest
|
|
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
|
|
MAIN_BOT = "main_bot"
|
|
RADIO_BOT = "radio_bot"
|
|
# The server bot re-pushes every 15 min. A list it pushed more than three
|
|
# pushes ago means it is down (it never said it left): a radio bot's view
|
|
# takes over until it is back.
|
|
MAIN_BOT_STALE_SECONDS = 45 * 60
|
|
# How many member radio bots are asked, one after another, for the list.
|
|
DIRECTORY_BOTS = 3
|
|
|
|
|
|
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).
|
|
Cached Discord snapshots of the server are dropped (a channel was added,
|
|
removed or renamed), so the next permission check reads it fresh."""
|
|
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]
|
|
data["directory_source"] = MAIN_BOT
|
|
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)
|
|
discord_rest.invalidate(guild_id=guild_id)
|
|
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 _load_tokens() -> dict[str, dict]:
|
|
return {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
|
|
|
|
|
|
def _member_radio_bots(tokens: dict[str, dict], guild_id: str, org_id: Optional[str]) -> list[str]:
|
|
"""Radio bots this org may use (pool, or the org's own) that Discord last
|
|
said are in the server, best first. Unknown membership (``guild_ids``
|
|
None) does not count: presence has to be positive."""
|
|
out = [tid for tid, t in tokens.items()
|
|
if t.get("token") and t.get("valid") is not False and t.get("org_id") in (None, org_id)
|
|
and isinstance(t.get("guild_ids"), list) and guild_id in t["guild_ids"]]
|
|
out.sort(key=lambda tid: leases._candidate_order(tid, tokens[tid], "", org_id, None))
|
|
return out
|
|
|
|
|
|
async def _org_guilds(org_id: str, tokens: Optional[dict] = None) -> list[dict]:
|
|
"""``discord_guilds`` docs this org may stream into, with a bot there.
|
|
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.
|
|
|
|
A linked server is shown while the server bot is there or a radio bot the
|
|
org may use is (server-26#225 S5: the main bot is optional). A server
|
|
that acts for the org only through the fallback still needs the server
|
|
bot: a radio bot's membership never discovers one."""
|
|
fallback = settings.discord_unlinked_guild_org_id or None
|
|
if tokens is None:
|
|
tokens = await _load_tokens()
|
|
out = []
|
|
for g in await fstore.collection_list(streams.GUILDS):
|
|
gid = str(g.get("guild_id") or "")
|
|
if not gid:
|
|
continue
|
|
if g.get("org_id"):
|
|
if g["org_id"] != org_id:
|
|
continue
|
|
if g.get("bot_present") is False and not _member_radio_bots(tokens, gid, org_id):
|
|
continue
|
|
else:
|
|
if g.get("unlinked_at") or not fallback or fallback != org_id or g.get("bot_present") is False:
|
|
continue
|
|
out.append(g)
|
|
out.sort(key=lambda g: ((_guild_name(g) or "").lower(), g["guild_id"]))
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Directory from a radio bot (server-26#225 S5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _age_seconds(ts) -> Optional[float]:
|
|
when = streams._parse_ts(ts)
|
|
return (datetime.now(timezone.utc) - when).total_seconds() if when else None
|
|
|
|
|
|
def _directory_due(g: dict) -> Optional[str]:
|
|
"""None when the stored list is fine; "missing" when there is none;
|
|
"stale" when it is old. Only for a server linked to an org — a fallback
|
|
server's list is the server bot's alone."""
|
|
if not g.get("org_id"):
|
|
return None
|
|
if g.get("voice_channels") is None:
|
|
return "missing"
|
|
from_main = g.get("bot_present") is True and (g.get("directory_source") or MAIN_BOT) == MAIN_BOT
|
|
limit = MAIN_BOT_STALE_SECONDS if from_main else settings.discord_directory_refresh_seconds
|
|
age = _age_seconds(g.get("directory_at"))
|
|
return "stale" if age is None or age > limit else None
|
|
|
|
|
|
def _main_bot_list_fresh(g: dict) -> bool:
|
|
"""The server bot is there and its own list is current: it wins."""
|
|
return (g.get("bot_present") is True and (g.get("directory_source") or MAIN_BOT) == MAIN_BOT
|
|
and _directory_due(g) is None)
|
|
|
|
|
|
async def _refresh_directory(g: dict, org_id: str, tokens: dict[str, dict]) -> Optional[dict]:
|
|
"""Read one linked server's voice channels from a member radio bot and
|
|
store them. Returns the fields written (also merged into ``g``), or None
|
|
when nothing was learned. Writes only onto the existing doc, and only
|
|
while it is still linked to ``org_id``: never creates a doc, never
|
|
touches ``org_id`` or the server bot's ``bot_present``."""
|
|
gid = g["guild_id"]
|
|
members = _member_radio_bots(tokens, gid, org_id)
|
|
if not members:
|
|
return None
|
|
docs = [{**tokens[t], "token_id": t} for t in members]
|
|
res = await discord_rest.guild_directory(docs, gid, max_bots=DIRECTORY_BOTS)
|
|
now = _now_iso()
|
|
if res.status == discord_rest.OK:
|
|
updates: dict = {"voice_channels": res.voice_channels, "directory_at": now,
|
|
"directory_source": RADIO_BOT, "radio_bot_present": True, "radio_bot_checked_at": now}
|
|
if res.name:
|
|
updates["guild_name"] = res.name
|
|
elif res.status == discord_rest.NOT_MEMBER:
|
|
updates = {"radio_bot_present": False, "radio_bot_checked_at": now}
|
|
else:
|
|
return None
|
|
cur = await fstore.doc_get(streams.GUILDS, gid)
|
|
if not cur or cur.get("org_id") != org_id:
|
|
return None # unlinked / moved / removed meanwhile
|
|
if _main_bot_list_fresh(cur):
|
|
return None # the server bot pushed a fresh list meanwhile; it wins
|
|
await fstore.doc_update(streams.GUILDS, gid, updates)
|
|
g.update(updates)
|
|
logger.info(f"Discord directory for {gid} from radio bot {res.token_id or '-'}: "
|
|
f"{res.status}, {len(res.voice_channels or [])} voice channel(s)")
|
|
return updates
|
|
|
|
|
|
_directory_tasks: dict[str, asyncio.Task] = {}
|
|
|
|
|
|
async def _refresh_quietly(g: dict, org_id: str, tokens: dict[str, dict]) -> None:
|
|
try:
|
|
await _refresh_directory(g, org_id, tokens)
|
|
except Exception as e: # a background refresh must never surface anywhere
|
|
logger.warning(f"Discord directory refresh for {g.get('guild_id')} failed: {type(e).__name__}")
|
|
|
|
|
|
def _refresh_in_background(g: dict, org_id: str, tokens: dict[str, dict]) -> None:
|
|
gid = g["guild_id"]
|
|
running = _directory_tasks.get(gid)
|
|
if running is not None and not running.done():
|
|
return
|
|
task = asyncio.create_task(_refresh_quietly(dict(g), org_id, tokens))
|
|
_directory_tasks[gid] = task
|
|
task.add_done_callback(lambda t, gid=gid: _directory_tasks.pop(gid, None)
|
|
if _directory_tasks.get(gid) is t else None)
|
|
|
|
|
|
async def _fill_directories(guilds: list[dict], org_id: str, tokens: dict[str, dict]) -> None:
|
|
"""Missing lists are read now (all servers at once, inside the
|
|
permission-check budget — an answer that comes later is cached and
|
|
stored by the next look); stale ones are refreshed in the background and
|
|
the stored list is shown meanwhile."""
|
|
now: list[dict] = []
|
|
for g in guilds:
|
|
due = _directory_due(g)
|
|
if not due or not _member_radio_bots(tokens, g["guild_id"], org_id):
|
|
continue
|
|
if due == "missing":
|
|
now.append(g)
|
|
else:
|
|
_refresh_in_background(g, org_id, tokens)
|
|
if not now:
|
|
return
|
|
tasks = [asyncio.create_task(_refresh_quietly(g, org_id, tokens)) for g in now]
|
|
await asyncio.wait(tasks, timeout=settings.discord_perm_check_budget_seconds)
|
|
|
|
|
|
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, token_names: Optional[dict] = None) -> 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_id": s.get("system_id"),
|
|
"system_name": s.get("system_name"),
|
|
"token_name": (token_names or {}).get(s.get("token_id") or ""),
|
|
"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
|
|
|
|
tokens = await _load_tokens()
|
|
guilds = await _org_guilds(org_id, tokens) if org_id else []
|
|
if org_id:
|
|
await _fill_directories(guilds, org_id, tokens)
|
|
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 {}
|
|
# Several streams can share one voice channel (each its own bot + system).
|
|
by_channel: dict[tuple, list] = {}
|
|
for s in sorted(live, key=lambda s: (s.get("created_at") or "", s.get("stream_id") or "")):
|
|
by_channel.setdefault((str(s.get("guild_id")), str(s.get("channel_id"))), []).append(s)
|
|
token_names = {tid: t.get("name") or tid for tid, t in tokens.items()}
|
|
|
|
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())):
|
|
on = by_channel.get((gid, str(c.get("channel_id"))), [])
|
|
# ``stream``: the one a pick here would touch — this system's (a
|
|
# pick restarts / moves it). Every other stream is left alone.
|
|
mine = next((s for s in on if s.get("system_id") == system_id), None)
|
|
channels.append({
|
|
"channel_id": str(c.get("channel_id")),
|
|
"name": c.get("name"),
|
|
"category": c.get("category"),
|
|
"stream": _stream_view(mine, nodes, token_names) if mine else None,
|
|
"streams": [_stream_view(s, nodes, token_names) for s in on],
|
|
})
|
|
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"),
|
|
# Who listed the channels: "main_bot" (DRB server bot) or "radio_bot" (S5).
|
|
"directory_source": (g.get("directory_source") or (MAIN_BOT if g.get("voice_channels") is not None
|
|
else None)),
|
|
"main_bot_present": g.get("bot_present") is True,
|
|
"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, token_names) 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)
|
|
tokens = await _load_tokens()
|
|
guild = next((g for g in await _org_guilds(org_id, tokens) if g["guild_id"] == guild_id), None)
|
|
if not guild:
|
|
raise HTTPException(404, "That Discord server is not linked to this organization.")
|
|
|
|
def _find(g: dict) -> Optional[dict]:
|
|
return next((c for c in g.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
|
|
|
|
channel = _find(guild)
|
|
if not channel and guild.get("org_id") and not _main_bot_list_fresh(guild):
|
|
# No list yet, or a radio bot's list that may predate this channel:
|
|
# read it once more (cached snapshot) before saying no.
|
|
try:
|
|
await asyncio.wait_for(_refresh_directory(guild, org_id, tokens),
|
|
timeout=settings.discord_perm_check_budget_seconds)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
channel = _find(guild)
|
|
if not channel:
|
|
raise HTTPException(400, "That is not a voice channel DRB can see in this server. "
|
|
"Refresh the list and pick again.")
|
|
|
|
# Same system already on in this channel = that stream (moved here if it
|
|
# is on another node); anything else is a NEW stream beside the others.
|
|
existing = await streams.find_channel_stream(guild_id, channel_id, system_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,
|
|
interactive=True,
|
|
)
|
|
except streams.StreamError as e:
|
|
return JSONResponse(status_code=e.status,
|
|
content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite})
|
|
logger.info(f"Web: radio channel {doc['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[:suffix] (leases.STREAM_ID_PATTERN). 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=leases.STREAM_ID_PATTERN)
|
|
|
|
|
|
@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=leases.STREAM_ID_PATTERN),
|
|
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}
|