""" Discord radio channels as desired state (server-26#203, #204; node-26#25). One ``discord_streams/{stream_id}`` doc per radio stream. A Discord voice channel may carry SEVERAL streams (server-26#213 follow-up, owner decision 2026-10-05): each has its own radio bot and plays its own system, so a channel can hold e.g. County P25 on one bot and City Fire on another. ``stream_id`` is ``{guild_id}:{channel_id}:{suffix}`` (``discord_leases.new_stream_id``); docs from before are ``{guild_id}:{channel_id}`` and are read, stopped and removed exactly the same way. Adding a stream never reuses another stream's doc except the same system in the same channel (idempotent re-add / restart). Fields: org_id, guild_id, channel_id, channel_name, guild_name, system_id, system_name, mode: "persistent" | "adhoc" persistent — stays connected and is re-placed by the reconciler (node reboot, node lost, C2 restart) until someone stops it. adhoc — stopped instead of re-placed when its node goes away. (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, 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, missing_permissions, channel_full) 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 Before a token is leased, ``start_stream`` asks Discord (REST, cached — discord_rest.guild_snapshot) whether each bot it could pick may View, Connect and Speak in the voice channel (server-26#225 S3). A bot that is denied is never handed to a node; when every bot is denied the start is refused with the bot, the channel, the missing permission and what removed it. A check that can't be completed is "unknown" and allowed (fail open). Kill switch: ``DISCORD_PERM_CHECK_ENABLED=false``. The system's preferred bot (``systems/{id}.preferred_token_id``) is tried first when nobody named one. Flow: ``start_stream`` picks a node + leases a token -> ``publish_desired`` retains ``nodes/{node}/discord/desired`` (full list for that node, with a monotonically increasing ``seq``) -> the edge reconciles and publishes ``nodes/{node}/discord/actual`` -> ``handle_actual`` updates ``actual`` and settles token releases. ``reconcile_loop`` re-places persistent streams whose node went away and retries desired-state publishes that never reached the broker. The desired payload carries token **ids**, never token strings: the edge fetches the string over HTTPS (GET /discord/node/{node_id}/tokens/{id}) with its own node API key, and C2 hands it out only to the node holding the lease. A retained MQTT message therefore never stores a bot credential on the broker. """ from __future__ import annotations import asyncio from datetime import datetime, timezone, timedelta from typing import Optional from app.config import settings from app.internal import firestore as fstore from app.internal import discord_leases as leases from app.internal import discord_perms from app.internal import discord_rest from app.internal.logger import logger STREAMS = leases.STREAMS GUILDS = "discord_guilds" # discord_guilds/{guild_id}: {org_id, linked_by, linked_at} NODE_STATE = "discord_nodes" # discord_nodes/{node_id}: {seq, published, applied_seq, actual_at} ONLINE = ("online", "recording") def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _effective_system(node: dict) -> Optional[str]: if node.get("is_overridden") and node.get("override_system_id"): return node["override_system_id"] return node.get("assigned_system_id") def _parse_ts(value) -> Optional[datetime]: try: ts = datetime.fromisoformat(value) if isinstance(value, str) else value except (TypeError, ValueError): return None if isinstance(ts, datetime) and ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) return ts if isinstance(ts, datetime) else None # --------------------------------------------------------------------------- # Guild -> org # --------------------------------------------------------------------------- def acting_org_of(link: Optional[dict]) -> Optional[str]: """``resolve_guild_org`` for an already-read ``discord_guilds`` doc (None = no doc). Pure, so a link transaction can decide from the snapshot it read (discord_guild_link.link_guild_txn, server-26#225 S6).""" if link and link.get("org_id"): return link["org_id"] if link and link.get("unlinked_at"): return None return settings.discord_unlinked_guild_org_id or None async def resolve_guild_org(guild_id: str) -> Optional[str]: """The org a Discord server acts for: its ``discord_guilds`` link, else ``settings.discord_unlinked_guild_org_id`` (None when that is empty). A server an org explicitly unlinked (``unlinked_at`` set, server-26#213) gets no fallback: otherwise unlinking org B's server would quietly hand it to the founding org.""" return acting_org_of(await fstore.doc_get(GUILDS, str(guild_id))) async def _node_stream_counts(exclude_stream_id: Optional[str] = None) -> dict[str, int]: streams = await fstore.collection_list(STREAMS, desired="on") counts: dict[str, int] = {} for s in streams: if s.get("node_id") and s.get("stream_id") != exclude_stream_id: counts[s["node_id"]] = counts.get(s["node_id"], 0) + 1 return counts def _node_usable(n: dict, org_id: str, system_id: str) -> bool: return ( n.get("org_id") == org_id and n.get("status") in ONLINE and n.get("approval_status") in (None, "approved") and _effective_system(n) == system_id ) async def candidate_nodes(org_id: str, system_id: str, exclude_stream_id: Optional[str] = None) -> list[dict]: """Online, approved nodes of this org currently listening to ``system_id``, best first: nodes already carrying streams (one gateway session can serve several guilds) but under the per-node cap, then the rest.""" nodes = await fstore.collection_list("nodes", org_id=org_id) counts = await _node_stream_counts(exclude_stream_id) cap = settings.discord_node_max_streams out = [n for n in nodes if n.get("node_id") and _node_usable(n, org_id, system_id) and counts.get(n["node_id"], 0) < cap] out.sort(key=lambda n: (-counts.get(n["node_id"], 0), n.get("name") or n["node_id"])) return out # --------------------------------------------------------------------------- # Desired state -> node # --------------------------------------------------------------------------- # One c2-core process publishes desired state. Serialising bump->build->publish # per node is what keeps the RETAINED message the newest one: two unserialised # publishes could land seq 7 then seq 6, and a reconnecting edge would then be # handed the stale list. _publish_locks: dict[str, asyncio.Lock] = {} def _lock_for(node_id: str) -> asyncio.Lock: lock = _publish_locks.get(node_id) if lock is None: lock = _publish_locks[node_id] = asyncio.Lock() return lock async def _bump_seq(node_id: str) -> int: db = fstore.db def _txn(txn): ref = db.collection(NODE_STATE).document(node_id) snap = ref.get(transaction=txn) seq = int(((snap.to_dict() or {}) if snap.exists else {}).get("seq") or 0) + 1 txn.set(ref, {"node_id": node_id, "seq": seq}, merge=True) return seq return await fstore.run_transaction(_txn) async def desired_for_node(node_id: str) -> list[dict]: streams = await fstore.collection_list(STREAMS, node_id=node_id) out = [] for s in streams: if s.get("desired") != "on" or not s.get("token_id"): continue out.append({ "stream_id": s["stream_id"], "guild_id": s["guild_id"], "channel_id": s["channel_id"], "token_id": s["token_id"], "lease_id": s.get("lease_id"), "system_name": s.get("system_name"), "mode": s.get("mode", "adhoc"), # Phase-2 seam: the edge ignores this today (no TG gate yet). "tg_filter": s.get("tg_filter"), }) out.sort(key=lambda d: d["stream_id"]) return out async def publish_desired(node_id: str) -> bool: """Retain the node's full stream list under a fresh ``seq``. Releasing tokens are stamped with that seq BEFORE the publish, so the earliest ack that can free them is one for a list that already lacks them. Returns whether the broker took the message; a miss is recorded and the reconciler retries it (nothing is lost — the streams stay desired).""" from app.internal.mqtt_handler import mqtt_handler async with _lock_for(node_id): seq = await _bump_seq(node_id) # Stamp BEFORE building the list: a stream detached between the two # (stop_stream does not hold this lock) must not get this seq, because # the list built here could still contain it — an ack for it would then # free a token the node was still told to use. await leases.stamp_release_seq(node_id, seq) payload = {"seq": seq, "streams": await desired_for_node(node_id), "issued_at": _now_iso()} ok = mqtt_handler.publish_discord_desired(node_id, payload) await fstore.doc_set(NODE_STATE, node_id, {"published": bool(ok), "published_at": _now_iso()}) if not ok: logger.warning(f"Discord desired state for {node_id} (seq {seq}) not delivered — reconciler will retry") return bool(ok) # --------------------------------------------------------------------------- # Discord permission check (server-26#225 S3) # --------------------------------------------------------------------------- # Trace sentences (discord_perms) that only say a permission went with # another one. The fix is for the other one, so they are not what we report. _IMPLIED = ("without View Channel", "without Connect") def _join_labels(labels: list) -> str: if len(labels) <= 1: return "".join(labels) return ", ".join(labels[:-1]) + " or " + labels[-1] def _fix_text(bot: str, missing: list, reason: str) -> str: if "timed out" in reason: return (f"In Discord: remove {bot}'s timeout (right-click it in the member list → Remove Timeout), " "then try again.") allow = " and ".join(missing) if len(missing) <= 2 else ", ".join(missing[:-1]) + " and " + missing[-1] return f"In Discord: Edit Channel → Permissions → add {bot} → allow {allow}, then try again." def evaluate_access(snap: "discord_rest.GuildSnapshot", channel_id: str, *, bot: str) -> dict: """One bot's verdict for one voice channel, from its cached snapshot. ``status``: ok | denied | unknown | not_member | invalid | bad_channel (the channel exists but is not a voice channel). ``missing`` lists the edge's labels ("View Channel", "Connect", "Speak"); ``reason`` is the trace sentence for the first one that isn't just implied by another; ``fix`` is what to change in Discord. Never raises.""" out: dict = {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None, "user_limit": 0, "limit_bypass": False, "channel_name": None} if snap.status != discord_rest.OK: out["status"] = snap.status out["error"] = snap.error return out channel = next((c for c in snap.channels or [] if isinstance(c, dict) and str(c.get("id")) == str(channel_id)), None) if channel is None: # Deleted, or hidden from the bot's list: the edge's join is the judge. out["error"] = "channel not in the server's channel list" return out out["channel_name"] = channel.get("name") try: bits, trace = discord_perms.effective_permissions(snap.guild or {}, snap.member or {}, channel) except (ValueError, TypeError, AttributeError) as e: out["error"] = f"could not read permissions ({type(e).__name__})" return out verdict = discord_perms.check_voice(bits, channel, trace) out["user_limit"] = verdict["user_limit"] out["limit_bypass"] = verdict["limit_bypass"] if not verdict["is_voice"]: out["status"] = "bad_channel" return out if verdict["ok"]: out["status"] = leases.PERM_OK return out missing = verdict["missing"] reasons = verdict["reasons"] primary = [m for m in missing if not any(x in (reasons.get(m) or "") for x in _IMPLIED)] or missing out.update(status=leases.PERM_DENIED, missing=missing, reasons=reasons, reason=reasons.get(primary[0]) or "not granted", fix=_fix_text(bot, primary, reasons.get(primary[0]) or "")) return out def tokens_to_check(tokens: dict[str, dict], *, guild_id: str, org_id: Optional[str], node_ids: Optional[set] = None, exclude_stream_id: Optional[str] = None, preferred_token_id: Optional[str] = None) -> list[str]: """The bots ``pick_token`` could hand this stream to, best first — the only ones worth asking Discord about. ``node_ids`` None = any node.""" out = [] for tok_id, tok in tokens.items(): if tok.get("valid") is False or not tok.get("token"): continue if tok.get("org_id") is not None and tok.get("org_id") != org_id: continue # another org's bot is never asked about this org's server members = tok.get("guild_ids") if members is not None and guild_id not in members: continue if preferred_token_id: if tok_id == preferred_token_id: out.append(tok_id) continue lease_node = (tok.get("lease") or {}).get("node_id") if lease_node and node_ids is not None and lease_node not in node_ids: continue if any(leases.guild_of(s) == guild_id for s in leases._lease_streams(tok) if s != exclude_stream_id): continue out.append(tok_id) out.sort(key=lambda t: leases._candidate_order(t, tokens[t], "", org_id, None)) return out def _unanswered() -> dict: return {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None, "error": "Discord did not answer in time"} async def gather_snapshots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, *, refresh: bool = False, budget: Optional[float] = None) -> dict: """``discord_rest.guild_snapshot`` for each bot, at most ``discord_perm_check_concurrency`` at a time, all within ``discord_perm_check_budget_seconds``. A bot not answered in time maps to None — unknown (its load keeps running and fills the cache for next time: the cache shields it from this cancellation).""" out: dict = {} if not token_ids: return out sem = asyncio.Semaphore(max(1, settings.discord_perm_check_concurrency)) async def _one(tok_id: str) -> None: async with sem: out[tok_id] = await discord_rest.guild_snapshot({**tokens[tok_id], "token_id": tok_id}, guild_id, refresh=refresh) tasks = [asyncio.create_task(_one(t)) for t in token_ids] _, pending = await asyncio.wait(tasks, timeout=budget if budget is not None else settings.discord_perm_check_budget_seconds) for t in pending: t.cancel() for t in tasks: if t.done() and not t.cancelled() and t.exception() is not None: logger.warning(f"Discord permission check failed: {type(t.exception()).__name__}") return {t: out.get(t) for t in token_ids} async def check_bots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, channel_id: str, *, refresh: bool = False, budget: Optional[float] = None) -> dict[str, dict]: """``evaluate_access`` for each bot in one voice channel. Never raises.""" snaps = await gather_snapshots(tokens, token_ids, guild_id, refresh=refresh, budget=budget) return {t: (evaluate_access(snaps[t], channel_id, bot=tokens[t].get("name") or t) if snaps.get(t) else _unanswered()) for t in token_ids} async def _perm_check(*, org_id: str, guild_id: str, channel_id: str, stream_id: str, preferred_token_id: Optional[str], node_ids: set) -> Optional[dict]: if not settings.discord_perm_check_enabled: return None tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")} ids = tokens_to_check(tokens, guild_id=guild_id, org_id=org_id, node_ids=node_ids, exclude_stream_id=stream_id, preferred_token_id=preferred_token_id) return await check_bots(tokens, ids, guild_id, channel_id) def _channel_ref(perms: Optional[dict], channel_name: Optional[str]) -> str: name = next((p.get("channel_name") for p in (perms or {}).values() if p.get("channel_name")), None) name = name or channel_name return f"#{name}" if name else "that voice channel" def _missing_permissions_error(denied: list, invite: list, chan: str) -> "StreamError": def line(d: dict) -> str: return f"{d['name']} can't {_join_labels(d.get('missing') or ['Speak'])} in {chan}: {d.get('reason')}." if len(denied) == 1: d = denied[0] text = f"{line(d)} {d.get('fix') or ''}".strip() else: shown = denied[:3] text = f"No radio bot can play in {chan}:\n" + "\n".join(f"- {line(d)}" for d in shown) if len(denied) > len(shown): text += f"\n- …and {len(denied) - len(shown)} more." text += "\nIn Discord: Edit Channel → Permissions → add one of these bots → allow the missing " \ "permission, then try again." if invite: text += " Or invite another radio bot with a link below." return StreamError("missing_permissions", text, invite=invite, extra={"denied": denied}) # --------------------------------------------------------------------------- # Start / stop # --------------------------------------------------------------------------- class StreamError(Exception): def __init__(self, code: str, message: str, invite: Optional[list] = None, status: int = 409, extra: Optional[dict] = None): super().__init__(message) self.code = code self.message = message self.invite = invite or [] self.status = status self.extra = extra or {} ERROR_TEXT = { "no_org": "This Discord server is not linked to a DRB organization yet.", "no_node": "No online node in your organization is listening to that system right now.", "pool_empty": "No radio bots have been added to DRB yet. A platform admin adds them under Radio bots.", "no_bot_in_guild": "No free radio bot is a member of this server. Invite one with the link below, then try again.", "pool_exhausted": ("Every radio bot that can join this server is already busy — a bot can be in only one voice " "channel per server, and a bot carrying radio on another node can't be shared. Add another " "radio bot to DRB and invite it to this server, or stop a radio channel here."), "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.", "preferred_unavailable": "That radio bot can't be used here (it was removed, failed validation, or belongs to " "another organization). Pick another bot, or let DRB choose any free one.", "pool_full": ("No radio bot is in this server, and every DRB radio bot that could be invited is already in its " "maximum number of servers. A platform admin needs to add another radio bot."), "missing_permissions": "No radio bot has the Discord permissions it needs in that voice channel.", "bad_channel": "That is not a voice channel. Radio can only play in a voice channel.", "not_found": "That radio channel no longer exists.", } # One c2-core process. Serialising start_stream per guild makes "is this # system already playing in this channel?" + "is the server at its cap?" + # create a real check-then-write: two concurrent adds of the same system to # one channel land on ONE stream, and two adds can't both take the last slot. _guild_locks: dict[str, asyncio.Lock] = {} def _guild_lock(guild_id: str) -> asyncio.Lock: lock = _guild_locks.get(guild_id) if lock is None: lock = _guild_locks[guild_id] = asyncio.Lock() return lock def _stream_label(s: dict) -> str: where = f"#{s['channel_name']}" if s.get("channel_name") else f"channel {s.get('channel_id')}" return f"{s.get('system_name') or s.get('system_id') or 'radio'} in {where}" async def find_channel_stream(guild_id: str, channel_id: str, system_id: str) -> Optional[dict]: """The stream already carrying ``system_id`` in this voice channel: a live one first, else the most recently updated stopped one (restarted instead of piling up a new doc per /join + /leave). None if there is neither.""" same = [s for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)) if str(s.get("channel_id")) == str(channel_id) and s.get("system_id") == system_id and s.get("stream_id")] live = sorted((s for s in same if s.get("desired") == "on"), key=lambda s: s.get("created_at") or "") if live: return live[0] same.sort(key=lambda s: s.get("updated_at") or "", reverse=True) return same[0] if same else None async def _lease_error(res: leases.LeaseResult, chan: str = "that voice channel") -> "StreamError": code = res.error or "no_node" d = res.detail or {} bot = d.get("token_name") or "That radio bot" if code == "missing_permissions" and d.get("denied"): return _missing_permissions_error(d["denied"], res.invite, chan) if code == "preferred_busy": busy = (await fstore.doc_get(STREAMS, d["busy_stream_id"]) if d.get("busy_stream_id") else None) or {} what = _stream_label(busy) if busy else "another radio channel" return StreamError(code, f"{bot} is already playing {what} in this server. A bot can be in only one voice " "channel per server — pick another bot, or let DRB choose any free one.", extra={"busy_stream_id": d.get("busy_stream_id")}) if code == "preferred_other_node": node = (await fstore.doc_get("nodes", d["node_id"]) if d.get("node_id") else None) or {} return StreamError(code, f"{bot} is carrying radio on another node ({node.get('name') or d.get('node_id')}) " "that can't serve this channel. Pick another bot, or let DRB choose any free one.") return StreamError(code, ERROR_TEXT.get(code, code), invite=res.invite) async def start_stream( *, org_id: str, guild_id: Optional[str] = None, channel_id: Optional[str] = None, system_id: str, mode: str = "adhoc", channel_name: Optional[str] = None, guild_name: Optional[str] = None, created_by: Optional[str] = None, node_id: Optional[str] = None, preferred_token_id: Optional[str] = None, max_streams: Optional[int] = None, stream_id: Optional[str] = None, ) -> dict: """Put ``system_id`` on a voice channel. * ``stream_id`` given — restart / edit THAT stream (the reconciler, the admin page's Start button). Its guild and channel come from the doc. * otherwise — ADD a stream to ``guild_id:channel_id``. If this system is already on in that channel the existing stream is returned (idempotent re-add, ``already_on``); a stopped one with this system is restarted; anything else gets a brand-new stream id, so another system's stream in the same channel is never touched. ``preferred_token_id`` is strict: a bot already in this server, on another node, or unusable is refused with a reason, never silently swapped. With none given, the system's ``preferred_token_id`` is tried first (a hint: another bot is used when it can't take the stream). Every bot that could be picked is permission-checked against the voice channel first (``_perm_check``; server-26#225). Every entry point — the bot's /join, the web picker, the node command and the reconciler — comes through here, so none can hand a node a bot Discord will refuse. """ if mode not in ("persistent", "adhoc"): raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400) if not org_id: raise StreamError("no_org", ERROR_TEXT["no_org"], status=403) existing: dict = {} if stream_id: existing = await fstore.doc_get(STREAMS, stream_id) or {} if not existing: raise StreamError("not_found", ERROR_TEXT["not_found"], status=404) doc_guild, doc_channel = str(existing.get("guild_id")), str(existing.get("channel_id")) if (guild_id and str(guild_id) != doc_guild) or (channel_id and str(channel_id) != doc_channel): raise StreamError("bad_stream", "That radio channel belongs to a different server or voice channel.", status=400) guild_id, channel_id = doc_guild, doc_channel if not guild_id or not channel_id: raise StreamError("bad_stream", "guild_id and channel_id are required.", status=400) guild_id, channel_id = str(guild_id), str(channel_id) async with _guild_lock(guild_id): return await _start_locked( org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id, mode=mode, channel_name=channel_name, guild_name=guild_name, created_by=created_by, node_id=node_id, preferred_token_id=preferred_token_id, max_streams=max_streams, stream_id=stream_id, existing=existing, ) async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_id: str, mode: str, channel_name: Optional[str], guild_name: Optional[str], created_by: Optional[str], node_id: Optional[str], preferred_token_id: Optional[str], max_streams: Optional[int], stream_id: Optional[str], existing: dict) -> dict: already_on = False if stream_id: existing = await fstore.doc_get(STREAMS, stream_id) or existing # re-read under the lock else: existing = await find_channel_stream(guild_id, channel_id, system_id) or {} if existing.get("desired") == "on": already_on = True if preferred_token_id and existing.get("token_id") and existing["token_id"] != preferred_token_id: tok = await fstore.doc_get(leases.TOKENS, existing["token_id"]) or {} raise StreamError("duplicate", f"{_stream_label(existing)} is already playing on radio bot " f"{tok.get('name') or existing['token_id']}. Remove that one " "first, or pick another system.", extra={"busy_stream_id": existing["stream_id"]}) # A re-add is not an edit: keep the running stream's label and mode. channel_name, mode = None, existing.get("mode") or mode stream_id = existing.get("stream_id") or leases.new_stream_id(guild_id, channel_id) if max_streams is not None and existing.get("desired") != "on": live = [s for s in await fstore.collection_list(STREAMS, guild_id=guild_id) if s.get("desired") == "on"] if len(live) >= max_streams: raise StreamError("limit", ERROR_TEXT["limit"]) system = await fstore.doc_get_cached("systems", system_id) or {} # Finding 1 (#225 plan): "preferred for this system" was written by # PUT /tokens/{id}/prefer/{system} and never read. It is a hint, not a # demand — an explicit preferred_token_id still wins and stays strict. # Read uncached: the systems cache lives 5 min, and a preference the # owner just set should apply to the next start. soft_pref = None if not preferred_token_id: soft_pref = ((await fstore.doc_get("systems", system_id)) or {}).get("preferred_token_id") or None if node_id: node = await fstore.doc_get("nodes", node_id) # A pinned node still has to be this org's and approved; it may be # listening to anything (the web "Join Discord" button pins a node). if not node or node.get("org_id") != org_id or node.get("approval_status") not in (None, "approved"): raise StreamError("no_node", ERROR_TEXT["no_node"]) candidates = [node] else: candidates = await candidate_nodes(org_id, system_id, exclude_stream_id=stream_id) cur = existing.get("node_id") if existing.get("system_id") == system_id else None held_on = None if preferred_token_id or soft_pref: # A bot can only run on the node it is logged in on. tok = await fstore.doc_get(leases.TOKENS, preferred_token_id or soft_pref) or {} held_on = (tok.get("lease") or {}).get("node_id") # Keep the current node when it is still a valid choice (reuse the # lease) — unless a bot was demanded, whose node then comes first. order = [held_on, cur] if preferred_token_id else [cur, held_on] for nid in reversed(order): if nid: candidates.sort(key=lambda n, nid=nid: 0 if n["node_id"] == nid else 1) if not candidates: raise StreamError("no_node", ERROR_TEXT["no_node"]) now = _now_iso() fields = { "stream_id": stream_id, "org_id": org_id, "guild_id": guild_id, "channel_id": channel_id, "system_id": system_id, "system_name": system.get("name") or system_id, "mode": mode, "desired": "on", "tg_filter": None, "stop_reason": None, "updated_at": now, } prev_actual = existing.get("actual") or {} if existing.get("desired") != "on" or prev_actual.get("state") not in ("connected", "connecting"): fields["actual"] = {"state": "pending", "error": None, "since": now, "at": now} if channel_name: fields["channel_name"] = channel_name if guild_name: fields["guild_name"] = guild_name if not existing: fields["created_at"] = now fields["created_by_discord_id"] = created_by fields["now_playing_message_id"] = None node_ids = {n["node_id"] for n in candidates} perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id, preferred_token_id=preferred_token_id, node_ids=node_ids) if perms and any(p.get("status") == "bad_channel" for p in perms.values()): raise StreamError("bad_channel", f"{_channel_ref(perms, channel_name)} is not a voice channel. " "Radio can only play in a voice channel — pick one from the list.", status=400) last: Optional[leases.LeaseResult] = None perm_refusal: Optional[leases.LeaseResult] = None prev_node = existing.get("node_id") refreshed = False for node in candidates: res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, preferred_token_id=preferred_token_id, stream_fields=fields, soft_preferred_token_id=soft_pref, perms=perms) if not res.ok and res.error in ("no_bot_in_guild", "pool_full") and not refreshed: # Membership is cached on the token doc; someone may have just # invited a bot. Re-read it from Discord once, then retry. refreshed = True tokens = await fstore.collection_list(leases.TOKENS) stale = [t["token_id"] for t in tokens if t.get("token_id") and t.get("guild_ids") is not None and guild_id not in t["guild_ids"]] if await discord_rest.refresh_membership(stale): if perms is not None: # Bots that just turned out to be members get checked too. perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id, preferred_token_id=preferred_token_id, node_ids=node_ids) res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, preferred_token_id=preferred_token_id, stream_fields=fields, soft_preferred_token_id=soft_pref, perms=perms) if res.ok: delivered = await publish_desired(node["node_id"]) if prev_node and prev_node != node["node_id"]: await publish_desired(prev_node) logger.info(f"Discord stream {stream_id} -> node {node['node_id']} token {res.token_id}" f" ({'reused' if res.reused else 'new'} lease{', already on' if already_on else ''})") return {**existing, **fields, "node_id": node["node_id"], "node_name": node.get("name"), "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused, "delivered": delivered, "already_on": already_on} chan = _channel_ref(perms, channel_name or existing.get("channel_name")) if res.error in ("preferred_busy", "preferred_unavailable") or ( preferred_token_id and res.error == "missing_permissions"): raise await _lease_error(res, chan) # true on every node — no point trying the rest if res.error == "missing_permissions": perm_refusal = res # more useful than another node's "all busy" last = res if last is None: raise StreamError("no_node", ERROR_TEXT["no_node"]) raise await _lease_error(perm_refusal or last, _channel_ref(perms, channel_name or existing.get("channel_name"))) async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: existing = await fstore.doc_get(STREAMS, stream_id) if not existing: return False now = _now_iso() node_id = await leases.detach_stream(stream_id, { "desired": "off", "updated_at": now, "stop_reason": reason, "actual": {**(existing.get("actual") or {}), "state": "stopping", "at": now}, }) if node_id: await publish_desired(node_id) 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): 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, "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, stream_id: Optional[str] = None) -> list[str]: """Stop live streams in ONE Discord server: all of them, every stream in one voice channel, or one stream by id. The /leave fix for server-26#203: never touches another server's streams (a ``stream_id`` from another server simply matches nothing).""" stopped = [] for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)): if s.get("desired") != "on": continue if channel_id and str(s.get("channel_id")) != str(channel_id): continue if stream_id and s.get("stream_id") != stream_id: continue await stop_stream(s["stream_id"], reason) stopped.append(s["stream_id"]) return stopped async def stop_node(node_id: str, reason: str) -> list[str]: stopped = [] for s in await fstore.collection_list(STREAMS, node_id=node_id): if s.get("desired") == "on": await stop_stream(s["stream_id"], reason) stopped.append(s["stream_id"]) return stopped # --------------------------------------------------------------------------- # Actual state <- node # --------------------------------------------------------------------------- _VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"} # Edge error codes that mean the cached permission/membership view is stale. _INVALIDATE_ON = {"missing_permissions", "channel_full", "not_in_guild", "bad_channel"} 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 cannot rewrite another node's (or org's) stream state.""" if not isinstance(payload, dict): return reported = payload.get("streams") or {} now = _now_iso() recheck: set = set() if isinstance(reported, dict): for stream_id, st in list(reported.items())[:50]: if not isinstance(st, dict) or "/" in str(stream_id): continue # "/" would be a Firestore sub-path, never a stream id doc = await fstore.doc_get(STREAMS, str(stream_id)) if not doc or doc.get("node_id") != node_id or doc.get("desired") != "on": continue 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 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"]) if code in _INVALIDATE_ON and doc.get("token_id"): # The edge saw Discord refuse; a cached "ok" must not keep # handing this bot the same channel (server-26#225). discord_rest.invalidate(token_id=doc["token_id"], guild_id=str(doc.get("guild_id"))) prev = doc.get("actual") or {} 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, "code": code, "since": since, "at": now}, }) if recheck: await discord_rest.refresh_membership(sorted(recheck)) if "sessions" in payload: sessions = {str(t) for t in (payload.get("sessions") or [])} applied = payload.get("applied_seq") applied = int(applied) if isinstance(applied, (int, float)) and not isinstance(applied, bool) else None freed = await leases.settle_releases(node_id, sessions, applied) for tok in freed: logger.info(f"Radio bot token {tok} released — node {node_id} confirmed it logged out") await fstore.doc_set(NODE_STATE, node_id, {"applied_seq": applied, "actual_at": now}) # --------------------------------------------------------------------------- # Self-healing: re-place persistent streams whose node is gone # --------------------------------------------------------------------------- async def release_offline_node(node_id: str) -> None: """Node offline past the lease grace: free its tokens (the edge already logged them out — see release_node_leases) and unplace its streams.""" orphaned = await leases.release_node_leases(node_id) if not orphaned: return now = _now_iso() for sid in orphaned: await fstore.doc_set(STREAMS, sid, { "token_id": None, "lease_id": None, "actual": {"state": "node_offline", "error": "The node carrying this channel went offline.", "since": now, "at": now}, }) await publish_desired(node_id) # retained: the node sees an accurate (empty) list when it returns logger.warning(f"Node {node_id} offline past lease grace — released {len(orphaned)} radio stream(s)") async def _unplace(s: dict, why: str) -> None: """Take a stream off its node (token detached, node republished) but leave it desired, so the reconciler can place it somewhere valid.""" now = _now_iso() old = await leases.detach_stream(s["stream_id"], { "node_id": None, "actual": {"state": "error", "error": why, "since": now, "at": now}, }) if old: await publish_desired(old) async def reconcile_once() -> None: nodes = {n["node_id"]: n for n in await fstore.collection_list("nodes") if n.get("node_id")} grace = timedelta(seconds=settings.discord_lease_offline_release_seconds) now = datetime.now(timezone.utc) released: set = set() for s in await fstore.collection_list(STREAMS, desired="on"): node = nodes.get(s.get("node_id") or "") if node and s.get("token_id"): if node.get("status") in ONLINE: if _effective_system(node) == s.get("system_id"): continue # healthy placement # Node retuned to another system under a live stream. if s.get("mode") != "persistent": await stop_stream(s["stream_id"], "node_retuned") continue await _unplace(s, "The node carrying this channel switched systems; finding another.") else: ls = _parse_ts(node.get("last_seen")) if ls and now - ls < grace: await _mark(s, "node_offline", "The node carrying this channel is offline; waiting for it.") continue # hold the lease through the grace window if node["node_id"] not in released: await release_offline_node(node["node_id"]) released.add(node["node_id"]) if s.get("mode") != "persistent": await stop_stream(s["stream_id"], "node_lost") continue org_id = s.get("org_id") try: await start_stream(org_id=org_id, stream_id=s["stream_id"], system_id=s["system_id"], mode="persistent") except StreamError as e: await _mark(s, "error", e.message) # Leases with NO live stream left (``releasing``) on a node that is gone: # /leave while the node was down, or the node was deleted. The pass above # only walks desired=on streams, so without this the token would stay out # of the pool until that node came back — forever for a dead node. for tok in await fstore.collection_list(leases.TOKENS): nid = (tok.get("lease") or {}).get("node_id") if not nid or nid in released: continue node = nodes.get(nid) if node and node.get("status") in ONLINE: continue if node is not None: heard = _parse_ts(node.get("last_seen")) else: # Deleted node (its MQTT client is revoked, so the edge fail-safe # logs out). Last heard = its last actual report, else lease start. st = await fstore.doc_get(NODE_STATE, nid) or {} heard = _parse_ts(st.get("actual_at")) or _parse_ts((tok.get("lease") or {}).get("since")) if heard and now - heard < grace: continue await release_offline_node(nid) released.add(nid) # Desired-state publishes the broker never took (MQTT was down). for st in await fstore.collection_list(NODE_STATE, published=False): if st.get("node_id"): await publish_desired(st["node_id"]) async def _mark(s: dict, state: str, error: str) -> None: prev = s.get("actual") or {} if prev.get("state") == state and prev.get("error") == error: return now = _now_iso() await fstore.doc_set(STREAMS, s["stream_id"], {"actual": {"state": state, "error": error, "since": now, "at": now}}) async def reconcile_loop() -> None: logger.info("Discord stream reconciler started.") while True: await asyncio.sleep(settings.discord_reconcile_interval_seconds) try: await reconcile_once() except Exception as e: logger.error(f"Discord reconcile error: {e}")