diff --git a/drb-c2-core/app/internal/discord_rest.py b/drb-c2-core/app/internal/discord_rest.py index 5129fe7..515d264 100644 --- a/drb-c2-core/app/internal/discord_rest.py +++ b/drb-c2-core/app/internal/discord_rest.py @@ -655,3 +655,100 @@ async def guild_snapshot(token_doc: dict, guild_id: str, *, refresh: bool = Fals error=f"check failed ({type(e).__name__})") return await access_cache.get_or_load(key, _load, lambda s: ttl_for(s.status), refresh=refresh) + + +# --------------------------------------------------------------------------- +# Voice-channel directory from a radio bot's view (server-26#225 S5) +# --------------------------------------------------------------------------- + +DIRECTORY_MAX_CHANNELS = 200 # same cap as PUT /discord/guilds/{id}/directory +_DIRECTORY_NAME_MAX = 100 +VOICE_CHANNEL = 2 +CATEGORY_CHANNEL = 4 + + +def _int_or(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def directory_from_snapshot(snap: Optional[GuildSnapshot]) -> Optional[dict]: + """A server's name and voice channels as one radio bot's snapshot sees + them, in the shape the DRB server bot pushes (``PUT + /discord/guilds/{id}/directory``): ``{name, voice_channels: [{channel_id, + name, category, position}]}``, in Discord's sidebar order (channels with + no category first, then each category by position; channels by position, + then id). None unless the snapshot is OK. + + Every voice channel in the list is kept, including any this bot can't + View, exactly like the server bot's gateway list: whether a bot may join + is the permission check's answer (discord_streams.evaluate_access), and + the picker shows it per channel. Stage channels are not voice channels + here either.""" + if snap is None or snap.status != OK or not isinstance(snap.channels, list): + return None + chans = [c for c in snap.channels if isinstance(c, dict) and c.get("id") is not None] + cats = {str(c["id"]): c for c in chans if c.get("type") == CATEGORY_CHANNEL} + + def parent(c: dict) -> Optional[dict]: + return cats.get(str(c.get("parent_id") or "")) + + def order(c: dict) -> tuple: + p = parent(c) + top = (0, 0, 0) if p is None else (1, _int_or(p.get("position")), _int_or(p.get("id"))) + return (*top, _int_or(c.get("position")), _int_or(c.get("id"))) + + out: list[dict] = [] + for c in sorted((c for c in chans if c.get("type") == VOICE_CHANNEL), key=order)[:DIRECTORY_MAX_CHANNELS]: + p = parent(c) + category = str(p.get("name") or "")[:_DIRECTORY_NAME_MAX] if p else "" + out.append({ + "channel_id": str(c["id"]), + "name": str(c.get("name") or "")[:_DIRECTORY_NAME_MAX], + "category": category or None, + "position": len(out), + }) + name = snap.guild.get("name") if isinstance(snap.guild, dict) else None + return {"name": str(name)[:_DIRECTORY_NAME_MAX] if name else None, "voice_channels": out} + + +@dataclass +class DirectoryResult: + """``guild_directory``'s answer. ``status``: OK (``name`` and + ``voice_channels`` from ``token_id``'s snapshot), NOT_MEMBER (every bot + asked turned out not to be in the server, or its token is invalid) or + UNKNOWN (no usable answer — keep what is stored).""" + guild_id: str + status: str + name: Optional[str] = None + voice_channels: Optional[list] = None + token_id: Optional[str] = None + fetched_at: Optional[float] = None + + +async def guild_directory(token_docs: list[dict], guild_id: str, *, refresh: bool = False, + max_bots: int = 3) -> DirectoryResult: + """The voice-channel list of one server, read from the first of + ``token_docs`` (radio bots believed to be in it, best first) whose cached + snapshot is OK. It is the same snapshot the permission check reads, so + listing channels costs no extra Discord request. At most ``max_bots`` + bots are asked, one after another. Never raises for a Discord failure. + + The caller decides which server and which bots: this never looks a + server up by itself, so it can't discover one (a server DRB doesn't + already know is never listed from a radio bot's membership).""" + guild_id = str(guild_id) + statuses: list[str] = [] + for doc in token_docs[:max(1, max_bots)]: + snap = await guild_snapshot(doc, guild_id, refresh=refresh) + statuses.append(snap.status) + d = directory_from_snapshot(snap) + if d is not None: + return DirectoryResult(guild_id=guild_id, status=OK, name=d["name"], + voice_channels=d["voice_channels"], token_id=snap.token_id, + fetched_at=snap.fetched_at) + if statuses and all(s in (NOT_MEMBER, INVALID) for s in statuses): + return DirectoryResult(guild_id=guild_id, status=NOT_MEMBER) + return DirectoryResult(guild_id=guild_id, status=UNKNOWN) diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index 4950d3c..f7c23da 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -53,6 +53,7 @@ A retained MQTT message therefore never stores a bot credential on the broker. from __future__ import annotations import asyncio +import time from datetime import datetime, timezone, timedelta from typing import Optional @@ -352,14 +353,39 @@ async def check_bots(tokens: dict[str, dict], token_ids: list[str], guild_id: st for t in token_ids} +# server-26#251: an interactive start (a person pressed Start / ran /join) +# re-reads a bot's snapshot that says "denied" once it is this old, so a +# permission fixed in Discord a moment ago is seen without waiting out the +# cache. The reconciler never does (plan finding 4: no REST loop). +RECHECK_DENIED_AFTER_SECONDS = 10.0 + + async def _perm_check(*, org_id: str, guild_id: str, channel_id: str, stream_id: str, - preferred_token_id: Optional[str], node_ids: set) -> Optional[dict]: + preferred_token_id: Optional[str], node_ids: set, + recheck_denied: bool = False) -> Optional[dict]: if not settings.discord_perm_check_enabled: return None tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")} ids = tokens_to_check(tokens, guild_id=guild_id, org_id=org_id, node_ids=node_ids, exclude_stream_id=stream_id, preferred_token_id=preferred_token_id) - return await check_bots(tokens, ids, guild_id, channel_id) + snaps = await gather_snapshots(tokens, ids, guild_id) + + def verdict(t: str) -> dict: + return evaluate_access(snaps[t], channel_id, bot=tokens[t].get("name") or t) if snaps.get(t) \ + else _unanswered() + + out = {t: verdict(t) for t in ids} + if recheck_denied: + now = time.time() + stale = [t for t in ids if out[t]["status"] == leases.PERM_DENIED + and now - (getattr(snaps.get(t), "fetched_at", None) or now) >= RECHECK_DENIED_AFTER_SECONDS] + if stale: + fresh = await gather_snapshots(tokens, stale, guild_id, refresh=True) + for t, snap in fresh.items(): + if snap is not None: # not answered in time: keep the cached verdict + snaps[t] = snap + out[t] = verdict(t) + return out def _channel_ref(perms: Optional[dict], channel_name: Optional[str]) -> str: @@ -484,6 +510,7 @@ async def start_stream( preferred_token_id: Optional[str] = None, max_streams: Optional[int] = None, stream_id: Optional[str] = None, + interactive: bool = False, ) -> dict: """Put ``system_id`` on a voice channel. @@ -504,6 +531,12 @@ async def start_stream( channel first (``_perm_check``; server-26#225). Every entry point — the bot's /join, the web picker, the node command and the reconciler — comes through here, so none can hand a node a bot Discord will refuse. + + ``interactive``: a person asked for this start (the routes pass True, + the reconciler never does). A bot whose cached check says "denied" is + then asked again if that answer is more than + ``RECHECK_DENIED_AFTER_SECONDS`` old (server-26#251), so "fix it in + Discord, then try again" works on the next try. """ if mode not in ("persistent", "adhoc"): raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400) @@ -527,14 +560,14 @@ async def start_stream( org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id, mode=mode, channel_name=channel_name, guild_name=guild_name, created_by=created_by, node_id=node_id, preferred_token_id=preferred_token_id, max_streams=max_streams, stream_id=stream_id, - existing=existing, + existing=existing, interactive=interactive, ) async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_id: str, mode: str, channel_name: Optional[str], guild_name: Optional[str], created_by: Optional[str], node_id: Optional[str], preferred_token_id: Optional[str], max_streams: Optional[int], - stream_id: Optional[str], existing: dict) -> dict: + stream_id: Optional[str], existing: dict, interactive: bool = False) -> dict: already_on = False if stream_id: existing = await fstore.doc_get(STREAMS, stream_id) or existing # re-read under the lock @@ -612,7 +645,8 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i node_ids = {n["node_id"] for n in candidates} perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id, - preferred_token_id=preferred_token_id, node_ids=node_ids) + preferred_token_id=preferred_token_id, node_ids=node_ids, + recheck_denied=interactive) if perms and any(p.get("status") == "bad_channel" for p in perms.values()): raise StreamError("bad_channel", f"{_channel_ref(perms, channel_name)} is not a voice channel. " "Radio can only play in a voice channel — pick one from the list.", @@ -638,7 +672,7 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i # Bots that just turned out to be members get checked too. perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id, preferred_token_id=preferred_token_id, - node_ids=node_ids) + node_ids=node_ids, recheck_denied=interactive) res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, preferred_token_id=preferred_token_id, stream_fields=fields, soft_preferred_token_id=soft_pref, perms=perms) diff --git a/drb-c2-core/app/routers/discord.py b/drb-c2-core/app/routers/discord.py index 70129e2..7ccf1d0 100644 --- a/drb-c2-core/app/routers/discord.py +++ b/drb-c2-core/app/routers/discord.py @@ -138,6 +138,7 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_ 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, stream_id=body.stream_id, + interactive=True, # /join, the admin page: re-read a stale "denied" (server-26#251) ) except streams.StreamError as e: return _error(e) diff --git a/drb-c2-core/app/routers/discord_web.py b/drb-c2-core/app/routers/discord_web.py index 8dd8df4..6d09b76 100644 --- a/drb-c2-core/app/routers/discord_web.py +++ b/drb-c2-core/app/routers/discord_web.py @@ -6,15 +6,38 @@ 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. + * 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 the bot reported in that server, and the +channel must be a voice channel in that server's directory, and the per-server and per-node stream caps apply. Auth: @@ -29,6 +52,7 @@ Auth: """ from __future__ import annotations +import asyncio from datetime import datetime, timezone from typing import Optional @@ -38,6 +62,7 @@ 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 @@ -54,6 +79,14 @@ 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: @@ -80,16 +113,20 @@ class GuildDirectory(BaseModel): @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).""" + 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)} @@ -130,24 +167,156 @@ def _require_control(can: bool) -> None: 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. +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.""" + 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 or g.get("bot_present") is False: + if not gid: continue - owner = g.get("org_id") or (None if g.get("unlinked_at") else fallback) - if owner and owner == org_id: - out.append(g) + 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: @@ -181,7 +350,10 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to 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 [] + 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) @@ -190,7 +362,6 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to 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) - tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")} token_names = {tid: t.get("name") or tid for tid, t in tokens.items()} out_guilds = [] @@ -216,6 +387,10 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to "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), @@ -261,10 +436,24 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict = 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) + 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.") - channel = next((c for c in guild.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None) + + 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.") @@ -287,6 +476,7 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict = 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, diff --git a/drb-c2-core/app/routers/nodes.py b/drb-c2-core/app/routers/nodes.py index 3707366..85da6b2 100644 --- a/drb-c2-core/app/routers/nodes.py +++ b/drb-c2-core/app/routers/nodes.py @@ -210,7 +210,7 @@ async def send_command( doc = await discord_streams.start_stream( org_id=node.get("org_id"), guild_id=cmd.guild_id, channel_id=cmd.channel_id, system_id=system_id, node_id=node_id, preferred_token_id=cmd.preferred_token_id, - mode="adhoc", + mode="adhoc", interactive=True, ) except discord_streams.StreamError as e: raise HTTPException(e.status, e.message) diff --git a/drb-c2-core/tests/test_discord_directory.py b/drb-c2-core/tests/test_discord_directory.py new file mode 100644 index 0000000..37b74b7 --- /dev/null +++ b/drb-c2-core/tests/test_discord_directory.py @@ -0,0 +1,392 @@ +""" +Channel list and presence without the main bot (server-26#225 S5), plus the +interactive re-check of a cached "denied" answer (server-26#251). + +* discord_rest.directory_from_snapshot / guild_directory: a radio bot's REST + snapshot becomes the same directory the DRB server bot pushes. +* _org_guilds: a linked server stays visible with only a radio bot in it; a + fallback-org server still needs the server bot; REST never discovers a + server or creates a ``discord_guilds`` doc. +* join_options / start_node_stream: a missing list is read inline from a + member radio bot, a stale one in the background; the server bot's fresh + list is never overwritten; nothing is written onto a server that was + unlinked meanwhile. +* start_stream(interactive=True) re-reads a "denied" snapshot older than + RECHECK_DENIED_AFTER_SECONDS once; the reconciler never does. +""" +import asyncio +import time +from datetime import datetime, timedelta, timezone + +import pytest +from fastapi import HTTPException + +from app.config import settings +from app.internal import discord_rest +from app.internal import discord_streams as streams +from app.routers import discord as discord_router +from app.models import CommandPayload +from app.routers import discord_web as web +from app.routers import nodes as nodes_router +from tests.test_discord_perm_check import ( # noqa: F401 (fs = fixture) + ADMIN, GUILD, ORG, OTHER, OWNER, SYS, _guild, _snap, _token, fs, run, +) + +LINKED = "300" # linked to ORG, never seen by the server bot + + +def _ago(seconds: float) -> str: + return (datetime.now(timezone.utc) - timedelta(seconds=seconds)).isoformat() + + +def _rest_channels(): + return [ + {"id": "9", "name": "Radio", "type": 4, "position": 1}, + {"id": "8", "name": "Admin", "type": 4, "position": 0}, + {"id": "3001", "name": "dispatch", "type": 2, "position": 1, "parent_id": "9", "permission_overwrites": []}, + {"id": "3002", "name": "fireground", "type": 2, "position": 0, "parent_id": "9", "permission_overwrites": []}, + {"id": "3003", "name": "staff", "type": 2, "position": 0, "parent_id": "8", "permission_overwrites": []}, + {"id": "3004", "name": "lobby", "type": 2, "position": 5, "permission_overwrites": []}, + {"id": "3005", "name": "general", "type": 0, "position": 0, "permission_overwrites": []}, + {"id": "3006", "name": "town hall", "type": 13, "position": 0, "permission_overwrites": []}, + ] + + +def _rest_snap(tok_id, guild_id=LINKED, status=discord_rest.OK): + if status != discord_rest.OK: + return discord_rest.GuildSnapshot(token_id=tok_id, guild_id=guild_id, status=status, error="x") + return discord_rest.GuildSnapshot( + token_id=tok_id, guild_id=guild_id, status=discord_rest.OK, bot_user_id=f"u-{tok_id}", + guild={**_guild(), "id": guild_id, "name": "Station 9"}, member={"roles": []}, + channels=_rest_channels()) + + +@pytest.fixture +def dfs(fs, monkeypatch): + """The perm-check fixture plus a linked server only radio bots know.""" + fs.put("discord_guilds", LINKED, {"guild_id": LINKED, "org_id": ORG, "linked_at": _ago(60)}) + fs.dir_mode = {} # token_id -> "ok" | "not_member" | "unknown" + + async def _snapshot(token_doc, guild_id, *, refresh=False): + tid = token_doc["token_id"] + fs.snap_calls.append((tid, guild_id, refresh)) + if guild_id != LINKED: + mode = fs.snap_mode.get(tid, "ok") + return _snap(tid, muted=(mode == "muted")) + mode = fs.dir_mode.get(tid, "ok") + status = {"ok": discord_rest.OK, "not_member": discord_rest.NOT_MEMBER}.get(mode, discord_rest.UNKNOWN) + return _rest_snap(tid, status=status) + monkeypatch.setattr(discord_rest, "guild_snapshot", _snapshot) + web._directory_tasks.clear() + return fs + + +async def _drain(): + while web._directory_tasks: + await asyncio.gather(*list(web._directory_tasks.values()), return_exceptions=True) + + +def _ids(opts): + return [g["guild_id"] for g in opts["guilds"]] + + +# --------------------------------------------------------------------------- +# discord_rest — directory from a snapshot +# --------------------------------------------------------------------------- + +def test_directory_from_snapshot_matches_the_server_bots_shape_and_order(): + d = discord_rest.directory_from_snapshot(_rest_snap("t1")) + assert d["name"] == "Station 9" + assert [(c["channel_id"], c["category"], c["position"]) for c in d["voice_channels"]] == [ + ("3004", None, 0), # no category first + ("3003", "Admin", 1), # categories by position + ("3002", "Radio", 2), # channels by position inside one + ("3001", "Radio", 3), + ] # text and stage channels are not listed + assert set(d["voice_channels"][0]) == {"channel_id", "name", "category", "position"} + + +def test_directory_from_snapshot_is_none_unless_ok(): + assert discord_rest.directory_from_snapshot(None) is None + for status in (discord_rest.NOT_MEMBER, discord_rest.UNKNOWN, discord_rest.INVALID): + assert discord_rest.directory_from_snapshot(_rest_snap("t1", status=status)) is None + + +def test_directory_caps_channels_and_name_length(): + snap = _rest_snap("t1") + snap.channels = [{"id": str(5000 + i), "name": "x" * 300, "type": 2, "position": i} for i in range(250)] + d = discord_rest.directory_from_snapshot(snap) + assert len(d["voice_channels"]) == discord_rest.DIRECTORY_MAX_CHANNELS + assert len(d["voice_channels"][0]["name"]) == 100 + + +async def test_guild_directory_tries_the_next_bot_and_reports_not_member(dfs): + _token(dfs, "a", guilds=(LINKED,)) + _token(dfs, "b", guilds=(LINKED,)) + docs = [{**dfs.get("bot_tokens", t)} for t in ("a", "b")] + dfs.dir_mode["a"] = "not_member" + got = await discord_rest.guild_directory(docs, LINKED) + assert got.status == discord_rest.OK and got.token_id == "b" and len(got.voice_channels) == 4 + dfs.dir_mode["b"] = "not_member" + assert (await discord_rest.guild_directory(docs, LINKED)).status == discord_rest.NOT_MEMBER + dfs.dir_mode["b"] = "unknown" + assert (await discord_rest.guild_directory(docs, LINKED)).status == discord_rest.UNKNOWN + assert (await discord_rest.guild_directory([], LINKED)).status == discord_rest.UNKNOWN + + +# --------------------------------------------------------------------------- +# Which servers an org sees +# --------------------------------------------------------------------------- + +def test_linked_server_with_only_a_radio_bot_is_listed_with_its_channels(dfs): + """Owner requirement 1: invite only the radio bot, no main bot.""" + _token(dfs, "byo", org=ORG, guilds=(LINKED,)) + opts = run(web.join_options("n1", OWNER)) + g = next(g for g in opts["guilds"] if g["guild_id"] == LINKED) + assert g["synced"] is True and g["directory_source"] == "radio_bot" and g["main_bot_present"] is False + assert [c["name"] for c in g["voice_channels"]] == ["lobby", "staff", "fireground", "dispatch"] + assert g["name"] == "Station 9" and g["linked"] is True and g["bot"]["ok"] is True + doc = dfs.get("discord_guilds", LINKED) + assert doc["org_id"] == ORG and "bot_present" not in doc # never the server bot's field + assert doc["radio_bot_present"] is True and doc["directory_source"] == "radio_bot" + assert doc["guild_name"] == "Station 9" and len(doc["voice_channels"]) == 4 + assert "secret" not in repr(opts) + + +def test_linked_server_without_any_radio_bot_keeps_todays_behaviour(dfs): + opts = run(web.join_options("n1", OWNER)) + g = next(g for g in opts["guilds"] if g["guild_id"] == LINKED) + assert g["synced"] is False and g["bot"]["ok"] is False # invite links, as before + assert not [c for c in dfs.snap_calls if c[1] == LINKED] + + +@pytest.mark.parametrize("tok", [ + {"org": OTHER}, # another org's branded bot + {"guilds": None}, # membership never checked + {"valid": False}, +]) +def test_only_a_usable_member_bot_keeps_a_server_without_the_main_bot(dfs, tok): + dfs.put("discord_guilds", LINKED, {**dfs.get("discord_guilds", LINKED), "bot_present": False}) + kw = {"guilds": (LINKED,), **tok} + _token(dfs, "x", **kw) + assert LINKED not in _ids(run(web.join_options("n1", OWNER))) + assert not [c for c in dfs.snap_calls if c[1] == LINKED] + + +def test_server_bot_leaving_keeps_server_while_a_radio_bot_is_in_it(dfs): + _token(dfs, "t1", guilds=(GUILD,)) + run(web.put_guild_directory(GUILD, web.GuildDirectory(present=False), {"service": True})) + opts = run(web.join_options("n1", OWNER)) + assert GUILD in _ids(opts) + g = next(g for g in opts["guilds"] if g["guild_id"] == GUILD) + assert g["main_bot_present"] is False + + +def test_fallback_server_is_never_discovered_by_a_radio_bot(dfs, monkeypatch): + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", ORG) + dfs.put("discord_guilds", "400", {"guild_id": "400", "bot_present": False, "voice_channels": []}) + _token(dfs, "t1", guilds=("400", "401", LINKED)) # also in a server DRB has no doc for + ids = _ids(run(web.join_options("n1", OWNER))) + assert "400" not in ids and "401" not in ids + assert dfs.get("discord_guilds", "401") is None # REST never creates a doc + assert not [c for c in dfs.snap_calls if c[1] in ("400", "401")] + + +def test_fallback_server_list_is_never_read_over_rest(dfs, monkeypatch): + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", ORG) + dfs.put("discord_guilds", "400", {"guild_id": "400", "bot_present": True, "voice_channels": None, + "directory_at": _ago(99999)}) + _token(dfs, "t1", guilds=("400",)) + assert "400" in _ids(run(web.join_options("n1", OWNER))) + assert not [c for c in dfs.snap_calls if c[1] == "400"] + + +# --------------------------------------------------------------------------- +# When the list is read, and who wins +# --------------------------------------------------------------------------- + +def test_fresh_main_bot_list_is_never_read_over_rest(dfs): + _token(dfs, "t1", guilds=(LINKED,)) + dfs.put("discord_guilds", LINKED, {**dfs.get("discord_guilds", LINKED), "bot_present": True, + "directory_source": "main_bot", "directory_at": _ago(600), + "voice_channels": [{"channel_id": "3001", "name": "dispatch"}]}) + opts = run(web.join_options("n1", OWNER)) + g = next(g for g in opts["guilds"] if g["guild_id"] == LINKED) + assert g["directory_source"] == "main_bot" and len(g["voice_channels"]) == 1 + assert not [c for c in dfs.snap_calls if c[1] == LINKED] + + +def test_stale_list_is_refreshed_in_the_background(dfs): + _token(dfs, "t1", guilds=(LINKED,)) + old = [{"channel_id": "3001", "name": "old name"}] + dfs.put("discord_guilds", LINKED, {**dfs.get("discord_guilds", LINKED), "voice_channels": old, + "directory_source": "radio_bot", + "directory_at": _ago(settings.discord_directory_refresh_seconds + 5)}) + + async def go(): + opts = await web.join_options("n1", OWNER) + await _drain() + return opts + opts = run(go()) + g = next(g for g in opts["guilds"] if g["guild_id"] == LINKED) + assert [c["name"] for c in g["voice_channels"]] == ["old name"] # stored list shown meanwhile + assert len(dfs.get("discord_guilds", LINKED)["voice_channels"]) == 4 # and replaced right after + + +def test_dead_main_bot_list_is_taken_over_by_a_radio_bot(dfs): + _token(dfs, "t1", guilds=(LINKED,)) + dfs.put("discord_guilds", LINKED, {**dfs.get("discord_guilds", LINKED), "bot_present": True, + "directory_at": _ago(web.MAIN_BOT_STALE_SECONDS + 60), + "voice_channels": [{"channel_id": "3001", "name": "dispatch"}]}) + + async def go(): + await web.join_options("n1", OWNER) + await _drain() + run(go()) + doc = dfs.get("discord_guilds", LINKED) + assert doc["directory_source"] == "radio_bot" and len(doc["voice_channels"]) == 4 + assert doc["bot_present"] is True # not ours to change + + +def test_not_member_marks_presence_and_keeps_the_list(dfs): + _token(dfs, "t1", guilds=(LINKED,)) + keep = [{"channel_id": "3001", "name": "dispatch"}] + dfs.put("discord_guilds", LINKED, {**dfs.get("discord_guilds", LINKED), "voice_channels": keep, + "directory_source": "radio_bot", "directory_at": _ago(99999)}) + dfs.dir_mode["t1"] = "not_member" + + async def go(): + await web.join_options("n1", OWNER) + await _drain() + run(go()) + doc = dfs.get("discord_guilds", LINKED) + assert doc["radio_bot_present"] is False and doc["voice_channels"] == keep + + +def test_refresh_never_writes_onto_a_server_unlinked_meanwhile(dfs, monkeypatch): + _token(dfs, "t1", guilds=(LINKED,)) + real = discord_rest.guild_directory + + async def _unlink_then_answer(docs, guild_id, **kw): + dfs.put("discord_guilds", LINKED, {"guild_id": LINKED, "org_id": None, "unlinked_at": _ago(0)}) + return await real(docs, guild_id, **kw) + monkeypatch.setattr(discord_rest, "guild_directory", _unlink_then_answer) + g = dict(dfs.get("discord_guilds", LINKED), org_id=ORG) + assert run(web._refresh_directory(g, ORG, {"t1": dfs.get("bot_tokens", "t1")})) is None + assert "voice_channels" not in dfs.get("discord_guilds", LINKED) + + +def test_slow_discord_never_holds_the_picker_past_the_budget(dfs, monkeypatch): + _token(dfs, "t1", guilds=(LINKED,)) + monkeypatch.setattr(settings, "discord_perm_check_budget_seconds", 0.05) + + async def _slow(docs, guild_id, **kw): + await asyncio.sleep(5) + monkeypatch.setattr(discord_rest, "guild_directory", _slow) + + async def go(): + t0 = time.monotonic() + opts = await web.join_options("n1", OWNER) + return opts, time.monotonic() - t0 + opts, took = run(go()) + assert took < 1.0 + assert next(g for g in opts["guilds"] if g["guild_id"] == LINKED)["synced"] is False + + +def test_directory_push_marks_main_bot_and_drops_cached_snapshots(dfs, monkeypatch): + dropped = [] + monkeypatch.setattr(discord_rest, "invalidate", lambda token_id=None, guild_id=None: dropped.append( + (token_id, guild_id)) or 0) + run(web.put_guild_directory(LINKED, web.GuildDirectory(name="S9", voice_channels=[ + web.VoiceChannelIn(channel_id="3001", name="dispatch")]), {"service": True})) + doc = dfs.get("discord_guilds", LINKED) + assert doc["directory_source"] == "main_bot" and doc["bot_present"] is True and doc["org_id"] == ORG + assert dropped == [(None, LINKED)] + + +# --------------------------------------------------------------------------- +# Starting from the picker without the main bot +# --------------------------------------------------------------------------- + +def test_start_checks_the_channel_against_the_radio_bots_list(dfs): + _token(dfs, "byo", org=ORG, guilds=(LINKED,)) + doc = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id=LINKED, channel_id="3001"), OWNER)) + assert doc["ok"] is True and doc["token_id"] == "byo" + s = dfs.get("discord_streams", doc["stream_id"]) + assert s["channel_name"] == "dispatch" and s["guild_name"] == "Station 9" and s["node_id"] == "n1" + + +def test_start_refuses_a_channel_the_radio_bot_does_not_list(dfs): + _token(dfs, "byo", org=ORG, guilds=(LINKED,)) + for bad in ("3005", "3999"): # a text channel; no such channel + with pytest.raises(HTTPException) as e: + run(web.start_node_stream("n1", web.NodeStreamStart(guild_id=LINKED, channel_id=bad), OWNER)) + assert e.value.status_code == 400 + + +# --------------------------------------------------------------------------- +# server-26#251 — an interactive start re-reads a stale "denied" +# --------------------------------------------------------------------------- + +@pytest.fixture +def aging(fs, monkeypatch): + """Snapshots that are ``fs.age`` seconds old unless refreshed; ``muted`` + until ``fs.fixed`` (the owner fixed the permission in Discord).""" + fs.age, fs.fixed = 60.0, False + + async def _snapshot(token_doc, guild_id, *, refresh=False): + tid = token_doc["token_id"] + fs.snap_calls.append((tid, guild_id, refresh)) + s = _snap(tid, muted=not (refresh and fs.fixed)) + s.fetched_at = time.time() - (0 if refresh else fs.age) + return s + monkeypatch.setattr(discord_rest, "guild_snapshot", _snapshot) + _token(fs, "t1", name="DRB Radio 2") + return fs + + +def _join(): + return discord_router.start_stream(discord_router.StreamStart(guild_id=GUILD, channel_id="1001", + system_id=SYS), {"service": True}) + + +def test_join_after_fixing_the_permission_succeeds_at_once(aging): + aging.fixed = True + r = run(_join()) + assert r["token_id"] == "t1" + assert ("t1", GUILD, True) in aging.snap_calls + + +def test_still_denied_after_the_recheck_is_refused(aging): + r = run(_join()) + assert r.status_code == 409 and b"can't Speak in #dispatch" in r.body + assert [c for c in aging.snap_calls if c[2]] == [("t1", GUILD, True)] # once + + +def test_a_fresh_denied_answer_is_not_asked_again(aging): + aging.age = 1.0 + aging.fixed = True + r = run(_join()) + assert r.status_code == 409 + assert not [c for c in aging.snap_calls if c[2]] + + +async def test_reconciler_never_rechecks(aging): + aging.fixed = True + with pytest.raises(streams.StreamError) as e: + await streams.start_stream(org_id=ORG, guild_id=GUILD, channel_id="1001", system_id=SYS) + assert e.value.code == "missing_permissions" + assert not [c for c in aging.snap_calls if c[2]] + + +def test_web_picker_start_is_interactive(aging): + aging.fixed = True + doc = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id=GUILD, channel_id="1001"), OWNER)) + assert doc["ok"] is True and doc["token_id"] == "t1" + + +def test_node_discord_join_command_is_interactive(aging): + aging.fixed = True + out = run(nodes_router.send_command("n1", CommandPayload(action="discord_join", guild_id=GUILD, + channel_id="1001"), ADMIN)) + assert out["ok"] is True and out["token_id"] == "t1" diff --git a/drb-c2-core/tests/test_discord_web_picker.py b/drb-c2-core/tests/test_discord_web_picker.py index 708c0a9..48da518 100644 --- a/drb-c2-core/tests/test_discord_web_picker.py +++ b/drb-c2-core/tests/test_discord_web_picker.py @@ -80,7 +80,10 @@ def test_directory_push_keeps_org_link(fs): assert [c["channel_id"] for c in g["voice_channels"]] == ["1003"] -def test_directory_bot_removed_hides_guild(fs): +def test_directory_bot_removed_hides_guild_without_a_radio_bot(fs): + # server-26#225 S5: the server bot leaving hides a linked server only when + # no radio bot the org may use is in it either. + fs.put("bot_tokens", "t1", {**fs.get("bot_tokens", "t1"), "guild_ids": ["200"]}) run(web.put_guild_directory("100", web.GuildDirectory(present=False), {"service": True})) opts = run(web.join_options("n1", OWNER)) assert opts["guilds"] == [] diff --git a/drb-frontend/lib/discordPicker.ts b/drb-frontend/lib/discordPicker.ts index 0e259ae..d615731 100644 --- a/drb-frontend/lib/discordPicker.ts +++ b/drb-frontend/lib/discordPicker.ts @@ -50,6 +50,10 @@ export interface DiscordGuildOption { linked: boolean; synced: boolean; synced_at: string | null; + /** Who listed the channels (server-26#225 S5): the DRB server bot, or a radio bot read over REST. */ + directory_source?: "main_bot" | "radio_bot" | null; + /** The DRB server bot is in this server. Not required: a radio bot alone is enough. */ + main_bot_present?: boolean; voice_channels: DiscordVoiceChannel[]; streams_on: number; bot: { ok: boolean; code: string | null; message: string | null; invite: DiscordInvite[] };