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: <reason>" 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 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
fabdf5839e
commit
c92bdfe841
@@ -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")}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user