From c92bdfe84124514217575dc21fb645ab41b5b0d8 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 4 Oct 2026 22:40:43 -0400 Subject: [PATCH 1/2] Discord: log node-reported stream state; remove a radio channel or a server - handle_actual logs every node-reported state TRANSITION (not heartbeats) with stream/node/guild/channel/token ids and the node's error text: info for connected/connecting/stopped, warning for error, and a dedicated "radio bot X failed to log in on node N: " line for login_failed. The node's reason code is now stored in actual.code and shown next to the error on Settings > Discord and the node page. - Remove a radio channel permanently: streams.delete_stream stops it through desired state first (node republished without it, token -> releasing per #216, freed on the node's ack), then deletes the doc in a transaction that refuses if the channel was restarted in between. DELETE /discord/streams/{id} (platform admin) and DELETE /discord/nodes/{node}/streams/{id} (admin or org owner, same org + node only; operators can still stop, not remove). Audited. - Remove a server: DELETE /discord/guilds/{id} (admin) deletes every radio channel in it the same way, then tombstones discord_guilds (org_id None + unlinked_at, #218 semantics, plus removed_at) so it never falls back to DISCORD_UNLINKED_GUILD_ORG_ID; the admin page hides it, and /drb link or the admin link brings it back. Optional leave_radio_bots makes pool bots and the server's org's own BYO bots leave via Discord REST; the DRB server bot stays until kicked, and the UI says so. Audited. - UI: Remove (with confirm) on Radio channels, Servers and the node page list. Refs #213, #221 Co-Authored-By: Claude Opus 5.5 --- .../app/internal/discord_guild_link.py | 3 + drb-c2-core/app/internal/discord_leases.py | 63 +++ drb-c2-core/app/internal/discord_streams.py | 93 ++++- drb-c2-core/app/routers/discord.py | 67 +++- drb-c2-core/app/routers/discord_web.py | 42 +- drb-c2-core/tests/fake_firestore.py | 7 + .../tests/test_discord_remove_and_logging.py | 362 ++++++++++++++++++ drb-frontend/components/NodeDiscordRadio.tsx | 55 ++- .../components/discord/ServersPanel.tsx | 115 +++++- .../components/discord/StreamsPanel.tsx | 27 +- drb-frontend/lib/c2api.ts | 21 + drb-frontend/lib/discord.ts | 2 + drb-frontend/lib/discordPicker.ts | 4 +- 13 files changed, 836 insertions(+), 25 deletions(-) create mode 100644 drb-c2-core/tests/test_discord_remove_and_logging.py diff --git a/drb-c2-core/app/internal/discord_guild_link.py b/drb-c2-core/app/internal/discord_guild_link.py index c846457..af9da2d 100644 --- a/drb-c2-core/app/internal/discord_guild_link.py +++ b/drb-c2-core/app/internal/discord_guild_link.py @@ -132,6 +132,9 @@ def _redeem_txn(code: str, guild_id: str, guild_name: Optional[str], discord_use "linked_at": ((guild or {}).get("linked_at") if current == org_id else None) or iso, "unlinked_at": None, "unlinked_by": None, + # A server an admin removed (streams.remove_guild) comes back on link. + "removed_at": None, + "removed_by": None, }, merge=True) return {"org_id": org_id, "already_linked": current == org_id, "created_by_uid": doc.get("created_by_uid")} diff --git a/drb-c2-core/app/internal/discord_leases.py b/drb-c2-core/app/internal/discord_leases.py index 1c4c31b..aca4b53 100644 --- a/drb-c2-core/app/internal/discord_leases.py +++ b/drb-c2-core/app/internal/discord_leases.py @@ -257,6 +257,30 @@ async def detach_stream(stream_id: str, stream_updates: Optional[dict] = None) - return await fstore.run_transaction(_txn) +async def delete_stopped_stream(stream_id: str) -> bool: + """Delete a stream doc, but only while it is stopped and holds no token. + + True when the doc is gone (deleted now, or already absent). False when it + is desired-on again or names a token — a concurrent start re-leased it, + and deleting it would leave that token's lease listing a stream nobody can + stop. ``lease_for_stream`` writes the stream doc in its own transaction, + so reading it here is enough to serialise against it.""" + db = fstore.db + + def _txn(txn): + ref = db.collection(STREAMS).document(stream_id) + snap = ref.get(transaction=txn) + if not snap.exists: + return True + stream = snap.to_dict() or {} + if stream.get("desired") == "on" or stream.get("token_id"): + return False + txn.delete(ref) + return True + + return await fstore.run_transaction(_txn) + + async def stamp_release_seq(node_id: str, seq: int) -> None: """Record the desired-state version that dropped each releasing token, so a stale actual report (from before the node saw the drop) can never free it.""" @@ -411,3 +435,42 @@ async def refresh_membership(token_ids: list[str]) -> int: changed += 1 await fstore.doc_update(TOKENS, tok_id, updates) return changed + + +async def leave_guild(guild_id: str, org_id: Optional[str]) -> list[dict]: + """Make the radio bots DRB holds tokens for leave one Discord server + (Discord REST ``DELETE /users/@me/guilds/{id}``, the bot's own token). + + Only platform-pool bots and ``org_id``'s own BYO bots — never another + org's branded bot. Bots whose cached membership says they are not in the + server are skipped; unknown membership is tried (404 = not a member). + The DRB *server* bot (the discord-bot service) is not in this pool, so it + stays in the server until someone kicks it. Returns one row per bot tried: + ``{token_id, name, left, error}``. Never raises for a Discord failure.""" + import httpx + guild_id = str(guild_id) + out: list[dict] = [] + for tok in await fstore.collection_list(TOKENS): + tok_id = tok.get("token_id") + if not tok_id or not tok.get("token"): + continue + if tok.get("org_id") is not None and tok.get("org_id") != org_id: + continue + members = tok.get("guild_ids") + if members is not None and guild_id not in members: + continue + row = {"token_id": tok_id, "name": tok.get("name") or tok_id, "left": False, "error": None} + try: + async with httpx.AsyncClient(timeout=10, base_url=DISCORD_API, + headers={"Authorization": f"Bot {tok['token']}"}) as client: + r = await client.delete(f"/users/@me/guilds/{guild_id}") + if r.status_code in (200, 204, 404): # 404: Unknown Guild — not a member + row["left"] = True + else: + row["error"] = f"Discord answered {r.status_code}" + except Exception as e: # network — report, do not fail the removal + row["error"] = str(e)[:200] + if row["left"] and members is not None: + await fstore.doc_update(TOKENS, tok_id, {"guild_ids": [g for g in members if g != guild_id]}) + out.append(row) + return out diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index dbe7556..f42cfa4 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -13,9 +13,11 @@ that should carry radio. Fields: (Idle auto-dismiss is not built yet — see the PR.) desired: "on" | "off" node_id, token_id, lease_id — placement (written with the lease, atomically) - actual: {state, error, since, at} — reported by the node, never guessed here + actual: {state, error, code, since, at} — reported by the node, never guessed here state: pending | connecting | connected | error | stopping | node_offline | paused_locally + code: the node's machine-readable reason on error (login_failed, + connect_failed, not_in_guild, bad_channel, no_audio) or None tg_filter: None — phase-2 seam (per-channel talkgroup gate); always None now now_playing_message_id, created_by_discord_id, created_at, updated_at @@ -324,10 +326,68 @@ async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: }) if node_id: await publish_desired(node_id) - logger.info(f"Discord stream {stream_id} stopped ({reason})") + logger.info(f"Discord stream {stream_id} stopped ({reason}) node={node_id or existing.get('node_id')}" + f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}") return True +async def delete_stream(stream_id: str, reason: str = "removed") -> Optional[dict]: + """Permanently remove a radio channel (server-26#213 follow-up). + + Order matters (server-26#216 lease semantics): the stream is first stopped + through the normal desired-state path — detached from its token (token -> + ``releasing``) and the node's list republished without it — and only then + is the doc deleted. The token returns to the pool exactly as for /leave, + when the node acks a list without it and no longer has a session on it + (``settle_releases`` reads the token lease, never the stream doc). + + The delete itself is a transaction that refuses if the stream was started + again in between (desired on, or a token re-leased onto it): deleting it + then would strand that token's lease on a stream that no longer exists. + Returns the doc as it was before removal, or None if it did not exist. + """ + existing = await fstore.doc_get(STREAMS, stream_id) + if not existing: + return None + if existing.get("desired") == "on" or existing.get("token_id"): + await stop_stream(stream_id, reason) + if not await leases.delete_stopped_stream(stream_id): + raise StreamError("restarted", "That radio channel was started again while it was being removed. " + "Stop it and try again.") + logger.info(f"Discord stream {stream_id} removed ({reason}) node={existing.get('node_id')}" + f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}" + f" org={existing.get('org_id')}") + return existing + + +async def remove_guild(guild_id: str, by: str) -> dict: + """Forget a Discord server: remove every radio channel in it (stop first, + then delete — ``delete_stream``) and tombstone its ``discord_guilds`` doc. + + The doc is NOT deleted: with no doc, ``resolve_guild_org`` would hand the + server to ``DISCORD_UNLINKED_GUILD_ORG_ID`` the next time anyone typed a + command there. It is unlinked the way PR #218's unlink does it + (``org_id`` None + ``unlinked_at``, so no fallback) and stamped + ``removed_at`` so the admin page stops listing it. Linking it again (``/drb + link`` or the admin link) clears ``removed_at``.""" + guild_id = str(guild_id) + link = await fstore.doc_get(GUILDS, guild_id) or {} + removed = [] + for s in await fstore.collection_list(STREAMS, guild_id=guild_id): + if s.get("stream_id") and await delete_stream(s["stream_id"], "server_removed"): + removed.append(s["stream_id"]) + now = _now_iso() + await fstore.doc_set(GUILDS, guild_id, { + "guild_id": guild_id, "org_id": None, + "unlinked_at": now, "unlinked_by": by, + "removed_at": now, "removed_by": by, + }, merge=True) + logger.info(f"Discord guild {guild_id} removed by {by}: {len(removed)} radio channel(s) deleted," + f" was org={link.get('org_id')!r}") + return {"guild_id": guild_id, "previous_org_id": link.get("org_id"), "streams_removed": removed, + "guild_name": link.get("guild_name")} + + async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None) -> list[str]: """Stop every live stream in ONE Discord server (optionally one channel). The /leave fix for server-26#203: never touches another server's streams.""" @@ -358,6 +418,26 @@ async def stop_node(node_id: str, reason: str) -> list[str]: _VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"} +def _log_transition(node_id: str, doc: dict, prev_state: Optional[str], state: str, + error: Optional[str], code: Optional[str]) -> None: + """One log line per node-reported state CHANGE (never per heartbeat — the + edge re-reports every ~60s). Errors are warnings so they surface in the + c2-core log without turning debug on; a radio bot that cannot log in gets + its own line naming the token, since that is the one failure that takes + every channel on the token down at once.""" + where = (f"stream={doc.get('stream_id')} node={node_id} guild={doc.get('guild_id')}" + f" channel={doc.get('channel_id')} token={doc.get('token_id')}") + if state == "error": + if code == "login_failed": + logger.warning(f"Radio bot {doc.get('token_id')} failed to log in on node {node_id}: {error} ({where})") + else: + logger.warning(f"Discord stream {prev_state or 'pending'} -> error [{code or 'unknown'}]:" + f" {error} ({where})") + else: + logger.info(f"Discord stream {prev_state or 'pending'} -> {state} ({where})" + + (f": {error}" if error else "")) + + async def handle_actual(node_id: str, payload: dict) -> None: """``nodes/{node_id}/discord/actual``. A node may only report on streams currently placed on it — anything else is ignored, so a compromised node @@ -377,15 +457,18 @@ async def handle_actual(node_id: str, payload: dict) -> None: state = st.get("state") if st.get("state") in _VALID_STATES else "error" error = st.get("error") error = str(error)[:300] if error else None - if st.get("code") == "not_in_guild" and doc.get("token_id"): + code = st.get("code") + code = str(code)[:40] if code else None + if code == "not_in_guild" and doc.get("token_id"): recheck.add(doc["token_id"]) prev = doc.get("actual") or {} - if prev.get("state") == state and prev.get("error") == error: + if prev.get("state") == state and prev.get("error") == error and prev.get("code") == code: since = prev.get("since") or now else: since = now + _log_transition(node_id, doc, prev.get("state"), state, error, code) await fstore.doc_update(STREAMS, str(stream_id), { - "actual": {"state": state, "error": error, "since": since, "at": now}, + "actual": {"state": state, "error": error, "code": code, "since": since, "at": now}, }) if recheck: await leases.refresh_membership(sorted(recheck)) diff --git a/drb-c2-core/app/routers/discord.py b/drb-c2-core/app/routers/discord.py index 3ae3258..a51be4f 100644 --- a/drb-c2-core/app/routers/discord.py +++ b/drb-c2-core/app/routers/discord.py @@ -9,6 +9,8 @@ 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. + * DELETE /discord/streams/{stream_id}, DELETE /discord/guilds/{id} — platform + admin only: permanently remove a radio channel / a whole server. * 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. """ @@ -18,7 +20,7 @@ import secrets from datetime import datetime, timezone from typing import Optional -from fastapi import APIRouter, Depends, HTTPException, Security +from fastapi import APIRouter, Depends, HTTPException, Path, Security from fastapi.responses import JSONResponse from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel @@ -27,11 +29,15 @@ 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.audit import write_audit +from app.internal.auth import describe_actor, 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) +# guild:channel snowflakes (leases.stream_id_for) — anything else is a 422, +# never a Firestore path. +STREAM_ID = r"^[0-9]{1,25}:[0-9]{1,25}$" def _error(e: streams.StreamError) -> JSONResponse: @@ -162,6 +168,8 @@ async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(req doc = { "guild_id": guild_id, "org_id": body.org_id, "linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(), + # Re-linking a server that was unlinked or removed brings it back. + "unlinked_at": None, "unlinked_by": None, "removed_at": None, "removed_by": None, } if body.guild_name and body.guild_name.strip(): doc["guild_name"] = body.guild_name.strip()[:100] @@ -175,6 +183,59 @@ async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(req return {"ok": True, "guild_id": guild_id, "org_id": body.org_id, "stopped": stopped} +@router.delete("/streams/{stream_id}") +async def delete_stream(stream_id: str = Path(..., pattern=STREAM_ID), + decoded: dict = Depends(require_admin_token)): + """Permanently remove one radio channel: stopped through the desired-state + path first (node logs the bot out, lease released on its ack), then the + doc is deleted. Platform admin only; org owners use the node page route.""" + try: + doc = await streams.delete_stream(stream_id, "removed_from_web_admin") + except streams.StreamError as e: + return _error(e) + if not doc: + raise HTTPException(404, "Radio channel not found.") + 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": doc.get("node_id"), "was_desired": doc.get("desired"), + "via": "settings_discord", + }) + return {"ok": True, "stream_id": stream_id} + + +@router.delete("/guilds/{guild_id}") +async def remove_guild(guild_id: str, leave_radio_bots: bool = False, + decoded: dict = Depends(require_admin_token)): + """Remove a Discord server from DRB: delete its radio channels (each + stopped first), then unlink it with no fallback org (streams.remove_guild). + ``leave_radio_bots`` also makes the pool radio bots (and the server's org's + own BYO bots) leave it via Discord's REST API. The DRB server bot is not + one of those and stays until someone kicks it in Discord.""" + if not guild_id.isdigit(): + raise HTTPException(400, "guild_id must be a Discord server id (digits only).") + known = await fstore.doc_get(streams.GUILDS, guild_id) + if not known and not await fstore.collection_list(streams.STREAMS, guild_id=guild_id): + raise HTTPException(404, "Discord server not found.") + uid, email = describe_actor(decoded) + acting_org = await streams.resolve_guild_org(guild_id) + try: + out = await streams.remove_guild(guild_id, uid) + except streams.StreamError as e: + return _error(e) + left: list = [] + if leave_radio_bots: + left = await leases.leave_guild(guild_id, out["previous_org_id"] or acting_org) + await write_audit(uid, email, "discord.guild.remove", details={ + "guild_id": guild_id, "previous_org_id": out["previous_org_id"], "acting_org_id": acting_org, + "streams_removed": out["streams_removed"], "leave_radio_bots": leave_radio_bots, + "radio_bots_left": [r["token_id"] for r in left if r["left"]], + "radio_bots_failed": [r["token_id"] for r in left if not r["left"]], + }) + return {"ok": True, "guild_id": guild_id, "streams_removed": out["streams_removed"], + "radio_bots": [{"name": r["name"], "left": r["left"], "error": r["error"]} for r in left]} + + # --------------------------------------------------------------------------- # Admin view — the web Discord page (server-26#213 item 2) # --------------------------------------------------------------------------- @@ -201,6 +262,8 @@ async def admin_overview(_: dict = Depends(require_admin_token)): guilds: dict[str, dict] = {} for gid, link in links.items(): + if link.get("removed_at"): + continue # removed from DRB (DELETE /discord/guilds/{id}); re-listed if it gets a channel again 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"), # Explicitly unlinked servers get no fallback org (mirrors diff --git a/drb-c2-core/app/routers/discord_web.py b/drb-c2-core/app/routers/discord_web.py index 8174590..22957bc 100644 --- a/drb-c2-core/app/routers/discord_web.py +++ b/drb-c2-core/app/routers/discord_web.py @@ -24,13 +24,15 @@ Auth: 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 +from fastapi import APIRouter, Depends, HTTPException, Path from fastapi.responses import JSONResponse from pydantic import BaseModel, Field @@ -38,7 +40,9 @@ 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, @@ -110,6 +114,13 @@ async def _node_for_caller(node_id: str, decoded: dict) -> tuple[dict, bool]: 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") @@ -209,6 +220,7 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to "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), @@ -291,3 +303,31 @@ async def stop_node_streams(node_id: str, body: NodeStreamStop, decoded: dict = 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} diff --git a/drb-c2-core/tests/fake_firestore.py b/drb-c2-core/tests/fake_firestore.py index 56cdccc..8f1685f 100644 --- a/drb-c2-core/tests/fake_firestore.py +++ b/drb-c2-core/tests/fake_firestore.py @@ -75,6 +75,9 @@ class _Txn: def update(self, ref: _DocRef, data: dict): self.writes.append(("update", ref.coll, ref.id, copy.deepcopy(data), True)) + def delete(self, ref: _DocRef): + self.writes.append(("delete", ref.coll, ref.id, {}, False)) + class FakeFirestore: def __init__(self): @@ -91,6 +94,10 @@ class FakeFirestore: def _write(self, kind: str, coll: str, doc_id: str, data: dict, merge: bool): docs = self.data.setdefault(coll, {}) + if kind == "delete": + docs.pop(doc_id, None) + self.versions[(coll, doc_id)] = self.versions.get((coll, doc_id), 0) + 1 + return if kind == "update" and doc_id not in docs: raise KeyError(f"No document to update: {coll}/{doc_id}") if merge and doc_id in docs: diff --git a/drb-c2-core/tests/test_discord_remove_and_logging.py b/drb-c2-core/tests/test_discord_remove_and_logging.py new file mode 100644 index 0000000..b4dacdd --- /dev/null +++ b/drb-c2-core/tests/test_discord_remove_and_logging.py @@ -0,0 +1,362 @@ +""" +Discord radio channels: server-side state logging, permanent removal of a +radio channel, and removal of a whole Discord server (server-26#213 / #221 +follow-up). + +Removal must go through desired state first — the node is told to drop the +channel (and logs the bot out), the token is only ``releasing`` until the node +acks — and only then is the doc deleted. Deleting first would leave the +token's lease naming a stream nobody can stop. +""" +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient + +from app.config import settings +from app.internal import discord_guild_link as glink +from app.internal import discord_leases as leases +from app.internal import discord_streams as streams +from app.internal.auth import require_admin_token, require_firebase_token +from app.main import app +from app.routers import discord_web as web +from tests.fake_firestore import FakeFirestore + +ORG, OTHER, SYS = "org-a", "org-b", "sys-1" +ADMIN = {"uid": "u-admin", "role": "admin", "email": "admin@example.test"} +OWNER = {"uid": "u-own", "org_id": ORG, "org_role": "owner"} +OPERATOR = {"uid": "u-op", "org_id": ORG, "org_role": "member", "role": "operator", "owned_node_ids": ["n1"]} +STRANGER = {"uid": "u-x", "org_id": OTHER, "org_role": "owner"} + +client = TestClient(app) + + +class _Log: + def __init__(self): + self.lines: list[tuple[str, str]] = [] + + def __getattr__(self, level): + return lambda msg, *a, **k: self.lines.append((level, str(msg))) + + +@pytest.fixture +def fs(monkeypatch): + fake = FakeFirestore.install(monkeypatch) + fake.published = [] + fake.on_publish = None + + def _publish(node_id, payload): + if fake.on_publish: + fake.on_publish(node_id, payload) + fake.published.append((node_id, payload)) + return True + + from app.internal.mqtt_handler import mqtt_handler + monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish) + streams._publish_locks.clear() + + async def _no_discord(token_ids): + return 0 + monkeypatch.setattr(leases, "refresh_membership", _no_discord) + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "") + + for nid, org in (("n1", ORG), ("n2", ORG), ("nx", OTHER)): + fake.put("nodes", nid, {"node_id": nid, "name": f"Node {nid}", "org_id": org, "status": "online", + "approval_status": "approved", "assigned_system_id": SYS}) + fake.put("systems", SYS, {"system_id": SYS, "name": "County P25", "org_id": ORG}) + fake.put("organizations", ORG, {"org_id": ORG, "name": "Alpha"}) + fake.put("discord_guilds", "100", {"guild_id": "100", "org_id": ORG, "guild_name": "Fire Co. 7", + "bot_present": True, + "voice_channels": [{"channel_id": "1001", "name": "Dispatch"}, + {"channel_id": "1002", "name": "Fireground"}]}) + for tid, org in (("t1", None), ("t2", None)): + fake.put("bot_tokens", tid, {"token_id": tid, "name": f"Radio {tid}", "token": f"secret-{tid}", + "org_id": org, "guild_ids": ["100", "200"], "valid": True, + "application_id": f"app-{tid}", "lease": None}) + yield fake + app.dependency_overrides.clear() + + +async def _start(channel="1001", node_id="n1", guild="100", org=ORG): + return await streams.start_stream(org_id=org, guild_id=guild, channel_id=channel, system_id=SYS, + node_id=node_id) + + +def _lease(fs, tid): + return (fs.get("bot_tokens", tid) or {}).get("lease") + + +def _audit(fs, action): + return [a for a in (fs.data.get("audit_log") or {}).values() if a["action"] == action] + + +# --------------------------------------------------------------------------- +# 1. Logging node-reported state transitions +# --------------------------------------------------------------------------- + +async def test_actual_logs_transitions_not_heartbeats_and_keeps_the_code(fs, monkeypatch): + log = _Log() + monkeypatch.setattr(streams, "logger", log) + doc = await _start() + sid, tok = doc["stream_id"], doc["token_id"] + + report = {"streams": {sid: {"state": "connected", "error": None, "code": None}}} + await streams.handle_actual("n1", report) + await streams.handle_actual("n1", report) # heartbeat: same state + infos = [m for lvl, m in log.lines if lvl == "info" and "pending -> connected" in m] + assert len(infos) == 1 + assert f"stream={sid}" in infos[0] and "node=n1" in infos[0] + assert "guild=100" in infos[0] and "channel=1001" in infos[0] and f"token={tok}" in infos[0] + + err = {"streams": {sid: {"state": "error", "code": "connect_failed", + "error": "Could not connect to the voice channel (TimeoutError: )."}}} + await streams.handle_actual("n1", err) + await streams.handle_actual("n1", err) + warns = [m for lvl, m in log.lines if lvl == "warning"] + assert len(warns) == 1 + assert "connected -> error [connect_failed]" in warns[0] and "TimeoutError" in warns[0] + actual = fs.get("discord_streams", sid)["actual"] + assert actual["code"] == "connect_failed" and actual["error"].startswith("Could not connect") + + +async def test_login_failure_reason_is_logged_against_the_token(fs, monkeypatch): + log = _Log() + monkeypatch.setattr(streams, "logger", log) + doc = await _start() + await streams.handle_actual("n1", {"streams": {doc["stream_id"]: { + "state": "error", "code": "login_failed", + "error": "The radio bot could not log in to Discord (LoginFailure: Improper token has been passed.)."}}}) + warns = [m for lvl, m in log.lines if lvl == "warning"] + assert len(warns) == 1 + assert warns[0].startswith(f"Radio bot {doc['token_id']} failed to log in on node n1:") + assert "Improper token" in warns[0] + + +async def test_actual_from_a_node_not_carrying_the_stream_is_neither_stored_nor_logged(fs, monkeypatch): + log = _Log() + monkeypatch.setattr(streams, "logger", log) + doc = await _start(node_id="n1") + await streams.handle_actual("n2", {"streams": {doc["stream_id"]: {"state": "error", "error": "spoof"}}}) + assert fs.get("discord_streams", doc["stream_id"])["actual"]["state"] == "pending" + assert not [m for _, m in log.lines if "spoof" in m] + + +# --------------------------------------------------------------------------- +# 2. Remove a radio channel +# --------------------------------------------------------------------------- + +async def test_delete_stops_through_desired_state_before_deleting(fs): + doc = await _start() + sid, tok = doc["stream_id"], doc["token_id"] + seen = [] + + def _on_publish(node_id, payload): + # The node must be told to drop the channel while the doc still exists. + cur = fs.get("discord_streams", sid) + seen.append((node_id, [s["stream_id"] for s in payload["streams"]], cur and cur["desired"])) + fs.on_publish = _on_publish + + before = await streams.delete_stream(sid, "removed") + assert before["stream_id"] == sid + assert seen == [("n1", [], "off")] # stop published first, doc still there + assert fs.get("discord_streams", sid) is None # then deleted + + # Lease follows #216: releasing until the node acks a list without it. + lease = _lease(fs, tok) + assert lease["releasing"] is True and lease["streams"] == [] + await streams.handle_actual("n1", {"applied_seq": lease["release_seq"], "sessions": [tok], "streams": {}}) + assert _lease(fs, tok) is not None # still logged in on the node + await streams.handle_actual("n1", {"applied_seq": lease["release_seq"], "sessions": [], "streams": {}}) + assert _lease(fs, tok) is None # node logged out -> back in the pool + + +async def test_delete_of_an_already_stopped_stream_does_not_republish(fs): + doc = await _start() + await streams.stop_stream(doc["stream_id"]) + n = len(fs.published) + assert await streams.delete_stream(doc["stream_id"]) + assert len(fs.published) == n + assert fs.get("discord_streams", doc["stream_id"]) is None + assert await streams.delete_stream(doc["stream_id"]) is None + + +async def test_delete_refuses_when_the_stream_is_restarted_in_between(fs, monkeypatch): + doc = await _start() + real_stop = streams.stop_stream + + async def _stop_then_restart(stream_id, reason="stopped"): + await real_stop(stream_id, reason) + await _start() # /join lands between stop and delete + monkeypatch.setattr(streams, "stop_stream", _stop_then_restart) + + with pytest.raises(streams.StreamError) as e: + await streams.delete_stream(doc["stream_id"]) + assert e.value.code == "restarted" and e.value.status == 409 + cur = fs.get("discord_streams", doc["stream_id"]) + assert cur["desired"] == "on" and doc["stream_id"] in _lease(fs, cur["token_id"])["streams"] + + +def test_admin_route_deletes_and_audits(fs): + import asyncio + doc = asyncio.run(_start()) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + assert client.delete("/discord/streams/not-a-stream").status_code == 422 + assert client.delete("/discord/streams/100:9999").status_code == 404 + r = client.delete(f"/discord/streams/{doc['stream_id']}") + assert r.status_code == 200 and r.json()["ok"] is True + assert fs.get("discord_streams", doc["stream_id"]) is None + (entry,) = _audit(fs, "discord.stream.delete") + assert entry["actor_uid"] == "u-admin" + assert entry["details"]["stream_id"] == doc["stream_id"] and entry["details"]["was_desired"] == "on" + assert entry["details"]["org_id"] == ORG and entry["details"]["node_id"] == "n1" + + +def test_admin_route_requires_platform_admin(fs): + def deny(): + raise HTTPException(403, "Admin access required") + app.dependency_overrides[require_admin_token] = deny + assert client.delete("/discord/streams/100:1001").status_code == 403 + assert client.delete("/discord/guilds/100").status_code == 403 + + +async def test_node_page_remove_is_owner_or_admin_and_org_scoped(fs): + a = await _start("1001", node_id="n1") + b = await _start("1002", node_id="n2") + + # An operator can stop but not permanently remove. + with pytest.raises(HTTPException) as e: + await web.remove_node_stream("n1", a["stream_id"], OPERATOR) + assert e.value.status_code == 403 + # Another org's owner cannot even see the node. + with pytest.raises(HTTPException) as e: + await web.remove_node_stream("n1", a["stream_id"], STRANGER) + assert e.value.status_code == 404 + # A stream on another node of the same org is not reachable through n1's URL. + with pytest.raises(HTTPException) as e: + await web.remove_node_stream("n1", b["stream_id"], OWNER) + assert e.value.status_code == 404 + + assert (await web.remove_node_stream("n1", a["stream_id"], OWNER))["ok"] is True + assert fs.get("discord_streams", a["stream_id"]) is None + assert fs.get("discord_streams", b["stream_id"])["desired"] == "on" # untouched + (entry,) = _audit(fs, "discord.stream.delete") + assert entry["actor_uid"] == "u-own" and entry["details"]["via"] == "node_page" + + assert (await web.remove_node_stream("n2", b["stream_id"], ADMIN))["ok"] is True + + +async def test_join_options_says_who_may_remove(fs): + assert (await web.join_options("n1", OWNER))["can_remove"] is True + assert (await web.join_options("n1", ADMIN))["can_remove"] is True + opts = await web.join_options("n1", OPERATOR) + assert opts["can_control"] is True and opts["can_remove"] is False + + +# --------------------------------------------------------------------------- +# 3. Remove a Discord server +# --------------------------------------------------------------------------- + +async def test_remove_guild_deletes_streams_and_unlinks_without_fallback(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", ORG) + a = await _start("1001", node_id="n1") + b = await _start("1002", node_id="n1") + await streams.stop_stream(b["stream_id"]) # one already stopped + out = await streams.remove_guild("100", "u-admin") + + assert sorted(out["streams_removed"]) == sorted([a["stream_id"], b["stream_id"]]) + assert fs.data["discord_streams"] == {} + g = fs.get("discord_guilds", "100") + assert g["org_id"] is None and g["unlinked_at"] and g["removed_at"] and g["removed_by"] == "u-admin" + # No fallback even though DISCORD_UNLINKED_GUILD_ORG_ID is set. + assert await streams.resolve_guild_org("100") is None + # Leases follow #216: releasing, freed on the node's ack. + toks = {a["token_id"], b["token_id"]} + assert len(toks) == 2 # one bot per guild voice connection + leased = [_lease(fs, t) for t in toks] + assert all(lease["releasing"] for lease in leased) + seq = max(lease["release_seq"] for lease in leased) + await streams.handle_actual("n1", {"applied_seq": seq, "sessions": [], "streams": {}}) + assert [_lease(fs, t) for t in toks] == [None, None] + + +def test_remove_guild_route_hides_server_and_relink_restores_it(fs): + import asyncio + asyncio.run(_start("1001")) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + assert client.delete("/discord/guilds/abc").status_code == 400 + assert client.delete("/discord/guilds/999").status_code == 404 + + r = client.delete("/discord/guilds/100") + assert r.status_code == 200 and r.json()["streams_removed"] == ["100:1001"] + assert r.json()["radio_bots"] == [] # not asked to leave + body = client.get("/discord/admin/overview").json() + assert body["guilds"] == [] and body["streams"] == [] + (entry,) = _audit(fs, "discord.guild.remove") + assert entry["details"]["previous_org_id"] == ORG and entry["details"]["streams_removed"] == ["100:1001"] + + # Admin link brings it back. + assert client.put("/discord/guilds/100/link", json={"org_id": ORG}).status_code == 200 + g = fs.get("discord_guilds", "100") + assert g["removed_at"] is None and g["unlinked_at"] is None + assert [x["guild_id"] for x in client.get("/discord/admin/overview").json()["guilds"]] == ["100"] + + +async def test_redeeming_a_link_code_restores_a_removed_server(fs): + await streams.remove_guild("100", "u-admin") + minted = await glink.mint_code(ORG, "u-own") + await glink.redeem_code(minted["code"], "100", "Fire Co. 7", "4242") + g = fs.get("discord_guilds", "100") + assert g["org_id"] == ORG and g["removed_at"] is None + + +def test_remove_guild_can_make_pool_and_own_radio_bots_leave(fs, monkeypatch): + import httpx + fs.put("bot_tokens", "byo-mine", {"token_id": "byo-mine", "name": "Ours", "token": "secret-mine", + "org_id": ORG, "guild_ids": ["100"], "valid": True}) + fs.put("bot_tokens", "byo-theirs", {"token_id": "byo-theirs", "name": "Theirs", "token": "secret-theirs", + "org_id": OTHER, "guild_ids": ["100"], "valid": True}) + fs.put("bot_tokens", "elsewhere", {"token_id": "elsewhere", "name": "Elsewhere", "token": "secret-else", + "org_id": None, "guild_ids": ["300"], "valid": True}) + calls = [] + + class _Resp: + def __init__(self, code): + self.status_code = code + + class _Client: + def __init__(self, *, headers, **kw): + self.auth = headers["Authorization"] + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def delete(self, url): + calls.append((self.auth, url)) + return _Resp(500 if self.auth == "Bot secret-t2" else 204) + + monkeypatch.setattr(httpx, "AsyncClient", _Client) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + r = client.delete("/discord/guilds/100?leave_radio_bots=true") + assert r.status_code == 200 + tried = sorted(a for a, _ in calls) + # pool bots in the server + this org's BYO; never another org's bot, never a bot not in it + assert tried == ["Bot secret-mine", "Bot secret-t1", "Bot secret-t2"] + assert all(u == "/users/@me/guilds/100" for _, u in calls) + rows = {x["name"]: x for x in r.json()["radio_bots"]} + assert rows["Radio t1"]["left"] is True and rows["Radio t2"]["left"] is False + assert "secret" not in r.text + assert fs.get("bot_tokens", "t1")["guild_ids"] == ["200"] + assert fs.get("bot_tokens", "t2")["guild_ids"] == ["100", "200"] # failed: membership unchanged + assert fs.get("bot_tokens", "byo-theirs")["guild_ids"] == ["100"] + (entry,) = _audit(fs, "discord.guild.remove") + assert sorted(entry["details"]["radio_bots_left"]) == ["byo-mine", "t1"] + assert entry["details"]["radio_bots_failed"] == ["t2"] + + +def test_node_page_routes_need_a_signed_in_user(fs): + def deny(): + raise HTTPException(401, "Missing authorization") + app.dependency_overrides[require_firebase_token] = deny + assert client.delete("/discord/nodes/n1/streams/100:1001").status_code == 401 diff --git a/drb-frontend/components/NodeDiscordRadio.tsx b/drb-frontend/components/NodeDiscordRadio.tsx index 7e65002..0409d93 100644 --- a/drb-frontend/components/NodeDiscordRadio.tsx +++ b/drb-frontend/components/NodeDiscordRadio.tsx @@ -354,8 +354,24 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) { } } + async function remove(s: DiscordStreamView) { + const name = s.channel_name ?? "this channel"; + if (!confirm(`Remove radio from ${name} permanently? It is stopped first (the node logs its radio bot out), then deleted from DRB.`)) return; + setBusy(`rm:${s.stream_id}`); + setActionError(null); + try { + await c2api.removeNodeDiscordStream(nodeId, s.stream_id); + await load(); + } catch (err) { + setActionError(parseC2Error(err).message); + } finally { + setBusy(null); + } + } + const streams = options?.streams ?? []; const canControl = options?.can_control ?? false; + const canRemove = options?.can_remove ?? false; return (
@@ -405,7 +421,7 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) { Server · channel Mode Desired → actual - {canControl && } + {(canControl || canRemove) && } @@ -419,19 +435,34 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) { {s.actual?.error && ( -
{s.actual.error}
+
+ {s.actual.error} + {s.actual.code && [{s.actual.code}]} +
)} - {canControl && ( - - + {(canControl || canRemove) && ( + + {canControl && ( + + )} + {canRemove && ( + + )} )} diff --git a/drb-frontend/components/discord/ServersPanel.tsx b/drb-frontend/components/discord/ServersPanel.tsx index b7932b6..351f8a6 100644 --- a/drb-frontend/components/discord/ServersPanel.tsx +++ b/drb-frontend/components/discord/ServersPanel.tsx @@ -10,6 +10,8 @@ import { DIM, Field, INPUT, Panel, Status, Table, TD } from "./ui"; // radio channel without being linked (it acts for DISCORD_UNLINKED_GUILD_ORG_ID). // Linking is PUT /discord/guilds/{id}/link (platform admin). The owner-side // "/drb link code:" flow is server-26#213 item 1 and is not built here. +// Remove is DELETE /discord/guilds/{id}: its radio channels are stopped then +// deleted, and the server is unlinked with no fallback org. function LinkForm({ overview, @@ -83,6 +85,63 @@ function LinkForm({ ); } +type RemoveResult = { guild: string; streams: number; bots: { name: string; left: boolean; error: string | null }[] }; + +function RemoveForm({ + guild, + onDone, + onCancel, +}: { + guild: DiscordGuild; + onDone: (r: RemoveResult) => void; + onCancel: () => void; +}) { + const [leaveBots, setLeaveBots] = useState(false); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(null); + const n = guild.live_streams; + + async function submit() { + setSaving(true); + setError(null); + try { + const r = await c2api.removeDiscordGuild(guild.guild_id, leaveBots); + onDone({ guild: guildLabel(guild), streams: r.streams_removed.length, bots: r.radio_bots }); + } catch (err) { + setError(parseC2Error(err).message); + } finally { + setSaving(false); + } + } + + return ( +
+

+ Remove {guildLabel(guild)} from DRB?{" "} + {n > 0 + ? `Its ${n} live radio channel${n === 1 ? " is" : "s are"} stopped first, then every radio channel in it is deleted.` + : "Every radio channel in it is deleted."}{" "} + It is unlinked and will not fall back to the default organization; linking it again brings it back. +

+ +

+ The DRB bot itself stays in the Discord server until someone kicks it there. +

+ {error &&

{error}

} +
+ + +
+
+ ); +} + export function ServersPanel({ overview, onChanged, @@ -92,6 +151,8 @@ export function ServersPanel({ }) { // "new" = blank link form; a guild id = editing that row's link. const [editing, setEditing] = useState(null); + const [removing, setRemoving] = useState(null); + const [removed, setRemoved] = useState(null); const fallback = overview.limits.unlinked_guild_org_id; const fallbackName = overview.limits.unlinked_guild_org_name || fallback; const max = overview.limits.guild_max_streams; @@ -119,6 +180,19 @@ export function ServersPanel({ } > {editing === "new" && setEditing(null)} />} + {removed && ( +
+ Removed {removed.guild}: {removed.streams} radio channel{removed.streams === 1 ? "" : "s"} deleted. + {removed.bots.length > 0 && ( + <> + {" "}Radio bots:{" "} + {removed.bots.map((b) => `${b.name} ${b.left ? "left" : `did not leave (${b.error ?? "unknown error"})`}`).join("; ")}. + + )}{" "} + The DRB bot is still in that Discord server until someone kicks it. + +
+ )} {overview.guilds.length === 0 ? (

No Discord servers yet. Link one, or run a radio command in a server the DRB bot is in.

) : ( @@ -147,13 +221,50 @@ export function ServersPanel({ Unlinked )} - - + ); + if (removing === g.guild_id) { + return [ + row, + + + setRemoving(null)} + onDone={async (r) => { + setRemoving(null); + setRemoved(r); + await onChanged(); + }} + /> + + , + ]; + } if (editing !== g.guild_id) return [row]; return [ row, diff --git a/drb-frontend/components/discord/StreamsPanel.tsx b/drb-frontend/components/discord/StreamsPanel.tsx index 7acb8d9..5eced8a 100644 --- a/drb-frontend/components/discord/StreamsPanel.tsx +++ b/drb-frontend/components/discord/StreamsPanel.tsx @@ -18,6 +18,8 @@ import { DIM, Field, INPUT, InviteLinks, Panel, Status, Table, TD } from "./ui"; // Radio channels = discord_streams docs, desired vs actual. Start/Stop/Add go // through the same routes the Discord bot uses (POST /discord/streams, // POST /discord/guilds/{id}/stop), which accept a platform-admin token. +// Remove is DELETE /discord/streams/{id} (admin): c2-core stops the channel +// through desired state first, then deletes it. type Invite = { name: string; invite_url: string | null }[]; @@ -214,7 +216,12 @@ export function StreamsPanel({ {s.mode === "adhoc" ? "ad-hoc" : s.mode ?? "—"} {label} - {isOn && s.actual?.error &&
{s.actual.error}
} + {isOn && s.actual?.error && ( +
+ {s.actual.error} + {s.actual.code && [{s.actual.code}]} +
+ )} {!isOn && s.stop_reason &&
{s.stop_reason}
} {err && (
@@ -224,7 +231,7 @@ export function StreamsPanel({ )} {s.token_name ?? —} - + {isOn ? ( ); diff --git a/drb-frontend/lib/c2api.ts b/drb-frontend/lib/c2api.ts index d8a5a4e..bfb3067 100644 --- a/drb-frontend/lib/c2api.ts +++ b/drb-frontend/lib/c2api.ts @@ -84,6 +84,21 @@ export const c2api = { channel_name?: string; preferred_token_id?: string; }) => request("/discord/streams", { method: "POST", body: JSON.stringify(body) }), + /** Permanently remove a radio channel (stopped through desired state first). Admin only. */ + deleteDiscordStream: (streamId: string) => + request<{ ok: boolean; stream_id: string }>(`/discord/streams/${encodeURIComponent(streamId)}`, { + method: "DELETE", + }), + /** Remove a Discord server from DRB: its radio channels are deleted and it is unlinked with no fallback org. */ + removeDiscordGuild: (guildId: string, leaveRadioBots: boolean) => + request<{ + ok: boolean; + guild_id: string; + streams_removed: string[]; + radio_bots: { name: string; left: boolean; error: string | null }[]; + }>(`/discord/guilds/${encodeURIComponent(guildId)}?leave_radio_bots=${leaveRadioBots ? "true" : "false"}`, { + method: "DELETE", + }), stopDiscordStream: (guildId: string, channelId: string) => request<{ stopped: string[] }>(`/discord/guilds/${encodeURIComponent(guildId)}/stop`, { method: "POST", @@ -116,6 +131,12 @@ export const c2api = { method: "POST", body: JSON.stringify(streamId ? { stream_id: streamId } : {}), }), + /** Permanently remove one radio channel from this node (org owner or admin). */ + removeNodeDiscordStream: (nodeId: string, streamId: string) => + request<{ ok: boolean; stream_id: string }>( + `/discord/nodes/${nodeId}/streams/${encodeURIComponent(streamId)}`, + { method: "DELETE" }, + ), deleteNode: (id: string) => request(`/nodes/${id}`, { method: "DELETE" }), diff --git a/drb-frontend/lib/discord.ts b/drb-frontend/lib/discord.ts index e86639d..59bf385 100644 --- a/drb-frontend/lib/discord.ts +++ b/drb-frontend/lib/discord.ts @@ -41,6 +41,8 @@ export interface RadioBotAddResult { export interface DiscordStreamActual { state?: string; error?: string | null; + /** The node's reason code on error: login_failed, connect_failed, not_in_guild, bad_channel, no_audio. */ + code?: string | null; since?: string | null; at?: string | null; } diff --git a/drb-frontend/lib/discordPicker.ts b/drb-frontend/lib/discordPicker.ts index 025cbc6..7079f98 100644 --- a/drb-frontend/lib/discordPicker.ts +++ b/drb-frontend/lib/discordPicker.ts @@ -22,7 +22,7 @@ export interface DiscordStreamView { mode: "persistent" | "adhoc" | null; node_id: string | null; node_name: string | null; - actual: { state: StreamState; error: string | null; since?: string; at?: string } | null; + actual: { state: StreamState; error: string | null; code?: string | null; since?: string; at?: string } | null; updated_at: string | null; } @@ -56,6 +56,8 @@ export interface DiscordJoinOptions { system_id: string | null; system_name: string | null; can_control: boolean; + /** Permanent removal: platform admin or the org owner only (operators can stop, not remove). */ + can_remove?: boolean; limits: { node_max: number; node_used: number; guild_max: number }; guilds: DiscordGuildOption[]; streams: DiscordStreamView[]; -- 2.54.0 From 25553fbac65ce130fe2b14dbf55928edc42c6fc5 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 4 Oct 2026 22:48:30 -0400 Subject: [PATCH 2/2] Discord remove server: audit a removal a restart interrupted DELETE /discord/guilds/{id} deletes channels one by one. If a channel was restarted mid-removal, delete_stream raised "restarted" and the route returned 409 with no audit entry, though the channels before it were already deleted for good. The error now carries the channels already removed and the route writes discord.guild.remove with outcome "incomplete" before returning the 409. The server stays linked. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/discord_streams.py | 11 ++++++++-- drb-c2-core/app/routers/discord.py | 5 +++++ .../tests/test_discord_remove_and_logging.py | 21 +++++++++++++++++++ 3 files changed, 35 insertions(+), 2 deletions(-) diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index f42cfa4..4786a71 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -374,8 +374,15 @@ async def remove_guild(guild_id: str, by: str) -> dict: link = await fstore.doc_get(GUILDS, guild_id) or {} removed = [] for s in await fstore.collection_list(STREAMS, guild_id=guild_id): - if s.get("stream_id") and await delete_stream(s["stream_id"], "server_removed"): - removed.append(s["stream_id"]) + try: + if s.get("stream_id") and await delete_stream(s["stream_id"], "server_removed"): + removed.append(s["stream_id"]) + except StreamError as e: + # A channel restarted mid-removal: stop here, leave the server + # linked, and tell the caller which channels are already gone so + # the partial removal is still audited. + e.removed = removed + raise now = _now_iso() await fstore.doc_set(GUILDS, guild_id, { "guild_id": guild_id, "org_id": None, diff --git a/drb-c2-core/app/routers/discord.py b/drb-c2-core/app/routers/discord.py index a51be4f..5e0e9b6 100644 --- a/drb-c2-core/app/routers/discord.py +++ b/drb-c2-core/app/routers/discord.py @@ -222,6 +222,11 @@ async def remove_guild(guild_id: str, leave_radio_bots: bool = False, try: out = await streams.remove_guild(guild_id, uid) except streams.StreamError as e: + # Channels deleted before the race are gone for good — audit them. + await write_audit(uid, email, "discord.guild.remove", details={ + "guild_id": guild_id, "acting_org_id": acting_org, "outcome": "incomplete", "error": e.code, + "streams_removed": getattr(e, "removed", []), "leave_radio_bots": leave_radio_bots, + }) return _error(e) left: list = [] if leave_radio_bots: diff --git a/drb-c2-core/tests/test_discord_remove_and_logging.py b/drb-c2-core/tests/test_discord_remove_and_logging.py index b4dacdd..b010c0c 100644 --- a/drb-c2-core/tests/test_discord_remove_and_logging.py +++ b/drb-c2-core/tests/test_discord_remove_and_logging.py @@ -278,6 +278,27 @@ async def test_remove_guild_deletes_streams_and_unlinks_without_fallback(fs, mon assert [_lease(fs, t) for t in toks] == [None, None] +def test_remove_guild_interrupted_by_a_restart_is_still_audited(fs, monkeypatch): + import asyncio + a = asyncio.run(_start("1001")) + b = asyncio.run(_start("1002", node_id="n2")) + real_stop = streams.stop_stream + + async def _stop(stream_id, reason="stopped"): + await real_stop(stream_id, reason) + if stream_id == b["stream_id"]: + await _start("1002", node_id="n2") # restarted between stop and delete + monkeypatch.setattr(streams, "stop_stream", _stop) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + + r = client.delete("/discord/guilds/100") + assert r.status_code == 409 and r.json()["code"] == "restarted" + (entry,) = _audit(fs, "discord.guild.remove") + assert entry["details"]["outcome"] == "incomplete" + assert entry["details"]["streams_removed"] == [a["stream_id"]] + assert fs.get("discord_guilds", "100")["org_id"] == ORG # left linked: a live channel remains + + def test_remove_guild_route_hides_server_and_relink_restores_it(fs): import asyncio asyncio.run(_start("1001")) -- 2.54.0