From 8af410aede31b2aeaa221778286d5f556bb6cce1 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 5 Oct 2026 00:30:40 -0400 Subject: [PATCH] Discord: several radio streams per voice channel; adding never overwrites (#213) Owner-reported on live: adding a radio channel for a server + voice channel that already had one (different label, system, even bot) overwrote the existing stream and made its bot leave, because stream_id was guild:channel. - New streams get guild:channel:<12 hex> ids (leases.new_stream_id); old guild:channel docs keep working for read/stop/remove/node reports. Route validators accept both (leases.STREAM_ID_PATTERN). - start_stream: no stream_id = ADD. Same system already on in that channel is idempotent (already_on, label/mode untouched); a stopped one of that system is restarted; anything else is a new doc. stream_id = explicit restart/edit (admin Start button, reconciler), incl. moving to another bot. - preferred_token_id is strict: bot already in this server -> 409 preferred_busy naming the stream; on another node -> preferred_other_node; unusable -> preferred_unavailable. No silent swap. pool_exhausted now says one-bot-per-server and to add + invite another bot. - One-token-per-guild was already enforced inside the lease transaction; now covered by race tests. Per-guild asyncio lock serialises the already-on / server-cap check with the create (guild half of #223). - Stop by stream_id (POST /discord/guilds/{id}/stop {stream_id}); admin UI Stop/Start/Remove act on the row's stream; node picker lists every stream in a channel and adds beside other systems. - Bot: /leave stream: (autocomplete) stops one; /leave channel: with several streams asks which unless all:True; bare /leave still stops the server. /status and /join name each stream's bot. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/discord_leases.py | 65 ++- drb-c2-core/app/internal/discord_streams.py | 170 ++++++- drb-c2-core/app/routers/discord.py | 37 +- drb-c2-core/app/routers/discord_web.py | 37 +- drb-c2-core/tests/test_discord_leases.py | 62 ++- .../tests/test_discord_multi_stream.py | 440 ++++++++++++++++++ .../tests/test_discord_remove_and_logging.py | 7 +- drb-c2-core/tests/test_discord_web_picker.py | 23 +- drb-frontend/components/NodeDiscordRadio.tsx | 26 +- .../components/discord/StreamsPanel.tsx | 62 ++- drb-frontend/lib/c2api.ts | 9 +- drb-frontend/lib/discord.ts | 2 + drb-frontend/lib/discordPicker.ts | 6 + drb-server-discord-bot/app/commands/radio.py | 74 ++- .../app/internal/c2_client.py | 6 +- 15 files changed, 913 insertions(+), 113 deletions(-) create mode 100644 drb-c2-core/tests/test_discord_multi_stream.py diff --git a/drb-c2-core/app/internal/discord_leases.py b/drb-c2-core/app/internal/discord_leases.py index aca4b53..0b3c2ba 100644 --- a/drb-c2-core/app/internal/discord_leases.py +++ b/drb-c2-core/app/internal/discord_leases.py @@ -32,7 +32,15 @@ Firestore layout: in_use / assigned_node_id / assigned_at — legacy mirrors of ``lease``, still written so the old /tokens page and any reader of them stay right. - discord_streams/{guild_id}:{channel_id} — see discord_streams.py. + discord_streams/{stream_id} — see discord_streams.py. ``stream_id`` is + ``{guild_id}:{channel_id}:{suffix}`` (server-26#213 follow-up: several + streams may share one voice channel, each with its own bot); docs created + before that are plain ``{guild_id}:{channel_id}`` and keep working. + +Several streams in ONE guild therefore always hold DIFFERENT tokens: a token's +lease lists its streams and ``pick_token`` skips any token that already has a +stream in the guild. Both the read and the write happen inside the lease +transaction, so two concurrent starts in one guild can never share a token. """ from __future__ import annotations @@ -63,10 +71,23 @@ def invite_url(application_id: Optional[str], guild_id: Optional[str] = None) -> return url +# Old ids are ``guild:channel``; new ones add ``:``. Safe as a Firestore +# doc id (no "/") and as a URL path segment. Every route validates against it. +STREAM_ID_PATTERN = r"^[0-9]{1,25}:[0-9]{1,25}(?::[0-9a-f]{8,32})?$" + + def stream_id_for(guild_id: str, channel_id: str) -> str: + """The LEGACY one-stream-per-channel id. Only for reading docs created + before several streams could share a channel — never for a new stream.""" return f"{guild_id}:{channel_id}" +def new_stream_id(guild_id: str, channel_id: str) -> str: + """A fresh id for a new stream. Unique per stream, so adding a radio + channel can never land on (and overwrite) an existing one.""" + return f"{guild_id}:{channel_id}:{uuid.uuid4().hex[:12]}" + + def guild_of(stream_id: str) -> str: return stream_id.split(":", 1)[0] @@ -81,10 +102,15 @@ class LeaseResult: token_id: Optional[str] = None lease_id: Optional[str] = None reused: bool = False - # error codes: pool_empty | no_bot_in_guild | pool_exhausted + # error codes: pool_empty | no_bot_in_guild | pool_exhausted | + # preferred_busy (the chosen bot already has a stream in this guild) | + # preferred_other_node (the chosen bot is logged in on another node) | + # preferred_unavailable (unknown, invalid, or another org's bot) error: Optional[str] = None # bots that would work if invited to this guild: [{name, invite_url}] invite: list = field(default_factory=list) + # preferred_* errors: {token_name, busy_stream_id, node_id} + detail: dict = field(default_factory=dict) def _lease_streams(tok: dict) -> list: @@ -124,9 +150,15 @@ def pick_token( preferred_token_id: Optional[str] = None, exclude_stream_id: Optional[str] = None, ) -> LeaseResult: - """Pure choice over a snapshot of bot_tokens. No I/O — unit tested directly.""" + """Pure choice over a snapshot of bot_tokens. No I/O — unit tested directly. + + A ``preferred_token_id`` is a demand, not a hint: if that bot cannot take + the stream the result says why instead of quietly using another bot.""" if not tokens: return LeaseResult(ok=False, error="pool_empty") + if preferred_token_id: + return _pick_preferred(tokens, preferred_token_id, node_id=node_id, guild_id=guild_id, + org_id=org_id, exclude_stream_id=exclude_stream_id) usable: list[tuple] = [] not_member: list[dict] = [] for tok_id, tok in tokens.items(): @@ -155,6 +187,31 @@ def pick_token( return LeaseResult(ok=False, error="pool_exhausted") +def _pick_preferred(tokens: dict[str, dict], tok_id: str, *, node_id: str, guild_id: str, + org_id: Optional[str], exclude_stream_id: Optional[str]) -> LeaseResult: + tok = tokens.get(tok_id) + owner = (tok or {}).get("org_id") + if not tok or tok.get("valid") is False or (owner is not None and owner != org_id): + return LeaseResult(ok=False, error="preferred_unavailable", + detail={"token_name": (tok or {}).get("name") or tok_id}) + name = tok.get("name") or tok_id + busy = [s for s in _lease_streams(tok) if s != exclude_stream_id and guild_of(s) == guild_id] + if busy: + # One voice connection per bot per guild (Discord): never move the + # bot off another stream to serve this one. + return LeaseResult(ok=False, error="preferred_busy", + detail={"token_name": name, "busy_stream_id": busy[0]}) + lease = tok.get("lease") or {} + if lease.get("node_id") and lease["node_id"] != node_id: + return LeaseResult(ok=False, error="preferred_other_node", + detail={"token_name": name, "node_id": lease["node_id"]}) + members = tok.get("guild_ids") + if members is not None and guild_id not in members: + return LeaseResult(ok=False, error="no_bot_in_guild", + invite=[{"name": name, "invite_url": invite_url(tok.get("application_id"), guild_id)}]) + return LeaseResult(ok=True, token_id=tok_id) + + async def lease_for_stream( stream_id: str, *, @@ -186,6 +243,8 @@ async def lease_for_stream( and (prev_tok.get("lease") or {}).get("node_id") == node_id and stream_id in _lease_streams(prev_tok) and prev_tok.get("valid") is not False + # An explicit edit to a different bot moves the stream to it. + and (not preferred_token_id or preferred_token_id == prev_tok_id) ): lease = prev_tok["lease"] if lease.get("releasing"): diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index 4786a71..0c1ff65 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -1,8 +1,14 @@ """ Discord radio channels as desired state (server-26#203, #204; node-26#25). -One ``discord_streams/{guild_id}:{channel_id}`` doc per Discord voice channel -that should carry radio. Fields: +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, @@ -201,12 +207,14 @@ async def publish_desired(node_id: str) -> bool: # --------------------------------------------------------------------------- class StreamError(Exception): - def __init__(self, code: str, message: str, invite: Optional[list] = None, status: int = 409): + 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 = { @@ -214,16 +222,69 @@ ERROR_TEXT = { "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. Stop another radio channel or add a bot.", + "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.", + "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) -> "StreamError": + code = res.error or "no_node" + d = res.detail or {} + bot = d.get("token_name") or "That radio bot" + 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: str, - channel_id: str, + guild_id: Optional[str] = None, + channel_id: Optional[str] = None, system_id: str, mode: str = "adhoc", channel_name: Optional[str] = None, @@ -232,16 +293,67 @@ async def start_stream( node_id: Optional[str] = None, preferred_token_id: Optional[str] = None, max_streams: Optional[int] = None, + stream_id: Optional[str] = None, ) -> dict: - """Make ``guild_id:channel_id`` carry ``system_id``. Idempotent: re-running - it for a channel that is already on reuses the node and the lease.""" + """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. + """ 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) - stream_id = leases.stream_id_for(guild_id, channel_id) - existing = await fstore.doc_get(STREAMS, stream_id) or {} + 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"] @@ -263,6 +375,12 @@ async def start_stream( cur = existing.get("node_id") if cur and existing.get("system_id") == system_id: candidates.sort(key=lambda n: 0 if n["node_id"] == cur else 1) + if preferred_token_id: + # A chosen bot can only run on the node it is logged in on. + tok = await fstore.doc_get(leases.TOKENS, preferred_token_id) or {} + held_on = (tok.get("lease") or {}).get("node_id") + if held_on: + candidates.sort(key=lambda n: 0 if n["node_id"] == held_on else 1) if not candidates: raise StreamError("no_node", ERROR_TEXT["no_node"]) @@ -306,13 +424,16 @@ async def start_stream( 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)") + 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} + "delivered": delivered, "already_on": already_on} + if res.error in ("preferred_busy", "preferred_unavailable"): + raise await _lease_error(res) # true on every node — no point trying the rest last = res - code = last.error if last and last.error else "no_node" - raise StreamError(code, ERROR_TEXT.get(code, code), invite=last.invite if last else []) + if last is None: + raise StreamError("no_node", ERROR_TEXT["no_node"]) + raise await _lease_error(last) async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: @@ -395,14 +516,19 @@ async def remove_guild(guild_id: str, by: str) -> dict: "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.""" +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 s.get("channel_id") != str(channel_id): + 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"]) @@ -456,8 +582,8 @@ async def handle_actual(node_id: str, payload: dict) -> None: recheck: set = set() if isinstance(reported, dict): for stream_id, st in list(reported.items())[:50]: - if not isinstance(st, dict): - continue + 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 @@ -552,8 +678,8 @@ async def reconcile_once() -> None: continue org_id = s.get("org_id") try: - await start_stream(org_id=org_id, guild_id=s["guild_id"], channel_id=s["channel_id"], - system_id=s["system_id"], mode="persistent") + 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) diff --git a/drb-c2-core/app/routers/discord.py b/drb-c2-core/app/routers/discord.py index 5e0e9b6..9be1dbf 100644 --- a/drb-c2-core/app/routers/discord.py +++ b/drb-c2-core/app/routers/discord.py @@ -23,7 +23,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Path, Security from fastapi.responses import JSONResponse from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer -from pydantic import BaseModel +from pydantic import BaseModel, Field from app.config import settings from app.internal import discord_leases as leases @@ -35,14 +35,14 @@ 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}$" +# guild:channel[:suffix] (leases.STREAM_ID_PATTERN) — old one-per-channel ids +# and new per-stream ids. Anything else is a 422, never a Firestore path. +STREAM_ID = leases.STREAM_ID_PATTERN def _error(e: streams.StreamError) -> JSONResponse: return JSONResponse(status_code=e.status, - content={"detail": e.message, "code": e.code, "invite": e.invite}) + content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite}) async def _org_or_403(guild_id: str) -> str: @@ -54,7 +54,8 @@ async def _org_or_403(guild_id: str) -> str: def _public_stream(s: dict) -> dict: keep = ("stream_id", "guild_id", "channel_id", "channel_name", "system_id", "system_name", "mode", - "desired", "node_id", "token_id", "actual", "stop_reason", "updated_at", "delivered", "reused") + "desired", "node_id", "token_id", "token_name", "actual", "stop_reason", "updated_at", "delivered", + "reused", "already_on") return {k: s.get(k) for k in keep if k in s} @@ -75,8 +76,13 @@ async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or nodes = await fstore.collection_list("nodes", org_id=org_id) gstreams = [s for s in await fstore.collection_list(streams.STREAMS, guild_id=guild_id) if s.get("desired") == "on"] + all_tokens = await fstore.collection_list(leases.TOKENS) + names = {t.get("token_id"): t.get("name") for t in all_tokens} + # Several streams can share a voice channel; the bot names each by its bot. + gstreams = [{**s, "token_name": names.get(s.get("token_id"))} for s in gstreams] + gstreams.sort(key=lambda s: (s.get("channel_name") or "", s.get("channel_id") or "", s.get("created_at") or "")) bots = [] - for t in await fstore.collection_list(leases.TOKENS): + for t in all_tokens: if t.get("org_id") not in (None, org_id) or t.get("valid") is False: continue lease = t.get("lease") or {} @@ -104,6 +110,9 @@ class StreamStart(BaseModel): guild_id: str channel_id: str system_id: str + # Set = restart / edit THAT stream. Unset = add a stream to the channel + # (never overwrites another system's stream there — several may share it). + stream_id: Optional[str] = Field(None, pattern=STREAM_ID) mode: str = "adhoc" preferred_token_id: Optional[str] = None channel_name: Optional[str] = None @@ -125,7 +134,7 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_ org_id=org_id, guild_id=body.guild_id, channel_id=body.channel_id, system_id=body.system_id, mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name, created_by=body.created_by, preferred_token_id=body.preferred_token_id, - max_streams=settings.discord_guild_max_streams, + max_streams=settings.discord_guild_max_streams, stream_id=body.stream_id, ) except streams.StreamError as e: return _error(e) @@ -133,7 +142,8 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_ @router.get("/streams/{stream_id}") -async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_admin)): +async def get_stream(stream_id: str = Path(..., pattern=STREAM_ID), + _: dict = Depends(require_service_key_or_admin)): doc = await fstore.doc_get(streams.STREAMS, stream_id) if not doc: raise HTTPException(404, "Stream not found.") @@ -141,13 +151,17 @@ async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_ad class GuildStop(BaseModel): + # Neither: every live stream in the server. channel_id: every stream in + # that voice channel. stream_id: just that one (it must be this server's). channel_id: Optional[str] = None + stream_id: Optional[str] = Field(None, pattern=STREAM_ID) reason: str = "stopped" @router.post("/guilds/{guild_id}/stop") async def stop_guild(guild_id: str, body: GuildStop, _: dict = Depends(require_service_key_or_admin)): - stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id) + stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id, + stream_id=body.stream_id) return {"stopped": stopped} @@ -293,8 +307,9 @@ async def admin_overview(_: dict = Depends(require_admin_token)): g["acting_org_name"] = (orgs.get(acting) or {}).get("name") if acting else None g["live_streams"] = sum(1 for r in rows if r.get("guild_id") == g["guild_id"] and r.get("desired") == "on") + # Streams sharing a voice channel sit together, oldest first. rows.sort(key=lambda r: (0 if r.get("desired") == "on" else 1, r.get("guild_name") or r.get("guild_id") or "", - r.get("channel_name") or r.get("channel_id") or "")) + r.get("channel_id") or "", r.get("created_at") or "", r.get("stream_id") or "")) systems = await fstore.collection_list("systems") # Names for every node a stream or a radio-bot lease points at. referenced = {r.get("node_id") for r in rows} | {(t.get("lease") or {}).get("node_id") for t in tokens.values()} diff --git a/drb-c2-core/app/routers/discord_web.py b/drb-c2-core/app/routers/discord_web.py index 22957bc..8dd8df4 100644 --- a/drb-c2-core/app/routers/discord_web.py +++ b/drb-c2-core/app/routers/discord_web.py @@ -156,14 +156,16 @@ def _bot_status(tokens: dict, *, node_id: str, guild_id: str, org_id: str) -> di "invite": [i for i in res.invite if i.get("invite_url")][:5]} -def _stream_view(s: dict, node_names: dict) -> dict: +def _stream_view(s: dict, node_names: dict, token_names: Optional[dict] = None) -> dict: return { "stream_id": s.get("stream_id"), "guild_id": s.get("guild_id"), "channel_id": s.get("channel_id"), "guild_name": s.get("guild_name"), "channel_name": s.get("channel_name"), + "system_id": s.get("system_id"), "system_name": s.get("system_name"), + "token_name": (token_names or {}).get(s.get("token_id") or ""), "mode": s.get("mode"), "node_id": s.get("node_id"), "node_name": node_names.get(s.get("node_id") or "") or s.get("node_id"), @@ -184,8 +186,12 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to if s.get("org_id") == org_id] nodes = {n.get("node_id"): n.get("name") for n in await fstore.collection_list("nodes", org_id=org_id) if n.get("node_id")} if org_id else {} - by_channel = {s.get("stream_id"): s for s in live} + # Several streams can share one voice channel (each its own bot + system). + by_channel: dict[tuple, list] = {} + for s in sorted(live, key=lambda s: (s.get("created_at") or "", s.get("stream_id") or "")): + by_channel.setdefault((str(s.get("guild_id")), str(s.get("channel_id"))), []).append(s) tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")} + token_names = {tid: t.get("name") or tid for tid, t in tokens.items()} out_guilds = [] for g in guilds: @@ -193,12 +199,16 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to channels = [] for c in sorted(g.get("voice_channels") or [], key=lambda c: (c.get("position") or 0, (c.get("name") or "").lower())): - s = by_channel.get(leases.stream_id_for(gid, str(c.get("channel_id")))) + on = by_channel.get((gid, str(c.get("channel_id"))), []) + # ``stream``: the one a pick here would touch — this system's (a + # pick restarts / moves it). Every other stream is left alone. + mine = next((s for s in on if s.get("system_id") == system_id), None) channels.append({ "channel_id": str(c.get("channel_id")), "name": c.get("name"), "category": c.get("category"), - "stream": _stream_view(s, nodes) if s else None, + "stream": _stream_view(mine, nodes, token_names) if mine else None, + "streams": [_stream_view(s, nodes, token_names) for s in on], }) out_guilds.append({ "guild_id": gid, @@ -227,7 +237,7 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to "guild_max": settings.discord_guild_max_streams, }, "guilds": out_guilds, - "streams": [_stream_view(s, nodes) for s in here], + "streams": [_stream_view(s, nodes, token_names) for s in here], } @@ -259,8 +269,9 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict = raise HTTPException(400, "That is not a voice channel DRB can see in this server. " "Refresh the list and pick again.") - stream_id = leases.stream_id_for(guild_id, channel_id) - existing = await fstore.doc_get(streams.STREAMS, stream_id) or {} + # Same system already on in this channel = that stream (moved here if it + # is on another node); anything else is a NEW stream beside the others. + existing = await streams.find_channel_stream(guild_id, channel_id, system_id) or {} already_here = existing.get("desired") == "on" and existing.get("node_id") == node_id if not already_here: used = sum(1 for s in await fstore.collection_list(streams.STREAMS, node_id=node_id) @@ -279,16 +290,16 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict = ) except streams.StreamError as e: return JSONResponse(status_code=e.status, - content={"detail": e.message, "code": e.code, "invite": e.invite}) - logger.info(f"Web: radio channel {stream_id} pinned to node {node_id} ({body.mode})") + content={**e.extra, "detail": e.message, "code": e.code, "invite": e.invite}) + logger.info(f"Web: radio channel {doc['stream_id']} pinned to node {node_id} ({body.mode})") return {"ok": True, "stream_id": doc["stream_id"], "token_id": doc.get("token_id"), "delivered": doc.get("delivered"), "reused": doc.get("reused"), "actual": doc.get("actual")} class NodeStreamStop(BaseModel): - # guild:channel snowflakes (leases.stream_id_for). Validated so a malformed - # id (e.g. one with "/") is a 422, not a Firestore path error / 500. - stream_id: Optional[str] = Field(None, pattern=r"^[0-9]{1,25}:[0-9]{1,25}$") + # guild:channel[:suffix] (leases.STREAM_ID_PATTERN). Validated so a + # malformed id (e.g. one with "/") is a 422, not a Firestore path error / 500. + stream_id: Optional[str] = Field(None, pattern=leases.STREAM_ID_PATTERN) @router.post("/nodes/{node_id}/streams/stop") @@ -307,7 +318,7 @@ async def stop_node_streams(node_id: str, body: NodeStreamStop, decoded: dict = @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}$"), + stream_id: str = Path(..., pattern=leases.STREAM_ID_PATTERN), decoded: dict = Depends(require_firebase_token)): """Permanently remove a radio channel this node carries (or last carried): stopped through desired state first so the node logs the bot out and the diff --git a/drb-c2-core/tests/test_discord_leases.py b/drb-c2-core/tests/test_discord_leases.py index ec4092f..9ae9ba9 100644 --- a/drb-c2-core/tests/test_discord_leases.py +++ b/drb-c2-core/tests/test_discord_leases.py @@ -34,6 +34,7 @@ def fs(monkeypatch): from app.internal.mqtt_handler import mqtt_handler monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish) streams._publish_locks.clear() + streams._guild_locks.clear() async def _no_discord(token_ids): return 0 @@ -70,6 +71,15 @@ def _lease(fake, token_id): return (fake.get("bot_tokens", token_id) or {}).get("lease") +def _sid(fake, guild="g1", channel="c1"): + """The id of the one stream start_stream created in this channel (new + streams get a unique ``guild:channel:suffix`` id — server-26#213).""" + ids = [k for k, d in (fake.data.get("discord_streams") or {}).items() + if d.get("guild_id") == guild and d.get("channel_id") == channel] + assert len(ids) == 1, ids + return ids[0] + + # --------------------------------------------------------------------------- # pick_token — pure choice # --------------------------------------------------------------------------- @@ -118,7 +128,8 @@ async def test_rejoin_reuses_the_lease_instead_of_taking_a_second_token(fs): assert again["token_id"] == first["token_id"] and again["reused"] is True leased = [t for t in ("t1", "t2") if _lease(fs, t)] assert leased == [first["token_id"]] # server-26#204 item 2 - assert _lease(fs, first["token_id"])["streams"] == ["g1:c1"] + assert _lease(fs, first["token_id"])["streams"] == [first["stream_id"]] + assert again["stream_id"] == first["stream_id"] and again["already_on"] is True async def test_preferred_token_is_honoured(fs): @@ -144,9 +155,9 @@ async def test_join_never_moves_another_guilds_stream(fs): _system(fs); _node(fs, "n1"); _token(fs, "t1") await _start("gA", "c1") await _start("gB", "c1") - a = fs.get("discord_streams", "gA:c1") + a = fs.get("discord_streams", _sid(fs, "gA", "c1")) assert a["desired"] == "on" and a["token_id"] == "t1" - assert sorted(_lease(fs, "t1")["streams"]) == ["gA:c1", "gB:c1"] + assert sorted(_lease(fs, "t1")["streams"]) == sorted([_sid(fs, "gA", "c1"), _sid(fs, "gB", "c1")]) async def test_desired_state_never_carries_the_token_string(fs): @@ -236,7 +247,7 @@ async def test_two_guilds_same_node_racing_share_one_session(fs): async def test_stop_marks_releasing_and_holds_until_the_node_acks(fs): _system(fs); _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1") await _start() - await streams.stop_stream("g1:c1") + await streams.stop_stream(_sid(fs, "g1", "c1")) lease = _lease(fs, "t1") assert lease["releasing"] is True and lease["streams"] == [] drop_seq = lease["release_seq"] @@ -260,25 +271,26 @@ async def test_stop_marks_releasing_and_holds_until_the_node_acks(fs): async def test_restart_before_release_reuses_the_logged_in_session(fs): _system(fs); _node(fs, "n1"); _token(fs, "t1"); _token(fs, "t2") first = await _start() - await streams.stop_stream("g1:c1") + await streams.stop_stream(_sid(fs, "g1", "c1")) again = await _start() assert again["token_id"] == first["token_id"] lease = _lease(fs, first["token_id"]) - assert lease["releasing"] is False and lease["streams"] == ["g1:c1"] + assert lease["releasing"] is False and lease["streams"] == [first["stream_id"]] + assert again["stream_id"] == first["stream_id"] # restarted, not a second doc async def test_handle_actual_settles_and_ignores_other_nodes_streams(fs): _system(fs); _node(fs, "n1"); _node(fs, "n2"); _token(fs, "t1"); _token(fs, "t2") - await _start("g1", "c1", node_id="n1") - await _start("g2", "c1", node_id="n2") + a = await _start("g1", "c1", node_id="n1") + b = await _start("g2", "c1", node_id="n2") await streams.handle_actual("n1", { "applied_seq": 1, "sessions": ["t1"], - "streams": {"g1:c1": {"state": "connected"}, "g2:c1": {"state": "error", "error": "spoof"}}, + "streams": {a["stream_id"]: {"state": "connected"}, b["stream_id"]: {"state": "error", "error": "spoof"}}, }) - assert fs.get("discord_streams", "g1:c1")["actual"]["state"] == "connected" - assert fs.get("discord_streams", "g2:c1")["actual"]["state"] == "pending" # not n1's to report + assert fs.get("discord_streams", _sid(fs, "g1", "c1"))["actual"]["state"] == "connected" + assert fs.get("discord_streams", _sid(fs, "g2", "c1"))["actual"]["state"] == "pending" # not n1's to report - await streams.stop_stream("g1:c1") + await streams.stop_stream(_sid(fs, "g1", "c1")) seq = _lease(fs, "t1")["release_seq"] await streams.handle_actual("n1", {"applied_seq": seq, "sessions": [], "streams": {}}) assert _lease(fs, "t1") is None @@ -295,7 +307,7 @@ async def test_failed_publish_is_retried_by_the_reconciler(fs): n = len(fs.published) await streams.reconcile_once() assert len(fs.published) > n and fs.get("discord_nodes", "n1")["published"] is True - assert fs.published[-1][1]["streams"][0]["stream_id"] == "g1:c1" + assert fs.published[-1][1]["streams"][0]["stream_id"] == doc["stream_id"] async def test_concurrent_publishes_keep_the_retained_list_newest(fs): @@ -315,9 +327,9 @@ async def test_leave_in_one_guild_never_touches_another(fs): await _start("gA", "c1") await _start("gB", "c1") stopped = await streams.stop_guild("gA", "leave") - assert stopped == ["gA:c1"] - assert fs.get("discord_streams", "gB:c1")["desired"] == "on" - assert fs.get("discord_streams", "gA:c1")["desired"] == "off" + assert stopped == [_sid(fs, "gA", "c1")] + assert fs.get("discord_streams", _sid(fs, "gB", "c1"))["desired"] == "on" + assert fs.get("discord_streams", _sid(fs, "gA", "c1"))["desired"] == "off" async def test_guild_org_resolution(fs, monkeypatch): @@ -340,7 +352,7 @@ async def test_offline_node_keeps_its_lease_inside_the_grace(fs, monkeypatch): _node(fs, "n1", status="offline", age=120) await streams.reconcile_once() assert _lease(fs, "t1")["node_id"] == "n1" - assert fs.get("discord_streams", "g1:c1")["actual"]["state"] == "node_offline" + assert fs.get("discord_streams", _sid(fs, "g1", "c1"))["actual"]["state"] == "node_offline" async def test_persistent_stream_moves_after_grace_adhoc_stops(fs, monkeypatch): @@ -351,10 +363,10 @@ async def test_persistent_stream_moves_after_grace_adhoc_stops(fs, monkeypatch): _node(fs, "n1", status="offline", age=900) _node(fs, "n2") await streams.reconcile_once() - moved = fs.get("discord_streams", "g1:c1") + moved = fs.get("discord_streams", _sid(fs, "g1", "c1")) assert moved["desired"] == "on" and moved["node_id"] == "n2" and moved["token_id"] assert _lease(fs, moved["token_id"])["node_id"] == "n2" - assert fs.get("discord_streams", "g2:c1")["desired"] == "off" + assert fs.get("discord_streams", _sid(fs, "g2", "c1"))["desired"] == "off" assert all((_lease(fs, t) or {}).get("node_id") != "n1" for t in ("t1", "t2")) @@ -365,7 +377,7 @@ async def test_leave_while_node_offline_frees_token_after_grace(fs, monkeypatch) _system(fs); _node(fs, "n1"); _token(fs, "t1") await _start() _node(fs, "n1", status="offline", age=120) - await streams.stop_stream("g1:c1") + await streams.stop_stream(_sid(fs, "g1", "c1")) await streams.reconcile_once() assert _lease(fs, "t1")["releasing"] is True # inside the grace: held _node(fs, "n1", status="offline", age=900) @@ -377,7 +389,7 @@ async def test_deleted_node_lease_freed_after_grace_from_last_report(fs, monkeyp monkeypatch.setattr(streams.settings, "discord_lease_offline_release_seconds", 600) _system(fs); _node(fs, "n1"); _token(fs, "t1") await _start() - await streams.stop_stream("g1:c1") + await streams.stop_stream(_sid(fs, "g1", "c1")) fs.put("discord_nodes", "n1", {**(fs.get("discord_nodes", "n1") or {}), "actual_at": datetime.now(timezone.utc).isoformat()}) await fs.doc_delete("nodes", "n1") @@ -401,12 +413,12 @@ async def test_release_seq_never_names_a_list_that_still_holds_the_stream(fs, mo out = await real(node_id) if not fired: fired.append(1) - await leases.detach_stream("g1:c1", {"desired": "off"}) + await leases.detach_stream(_sid(fs, "g1", "c1"), {"desired": "off"}) return out monkeypatch.setattr(streams, "desired_for_node", _racy) await streams.publish_desired("n1") racy_seq = fs.published[-1][1]["seq"] - assert any(d["stream_id"] == "g1:c1" for d in fs.published[-1][1]["streams"]) + assert any(d["stream_id"] == _sid(fs, "g1", "c1") for d in fs.published[-1][1]["streams"]) rseq = _lease(fs, "t1")["release_seq"] assert rseq is None or rseq > racy_seq assert await leases.settle_releases("n1", set(), racy_seq) == [] @@ -417,7 +429,7 @@ async def test_retuned_node_gives_up_a_persistent_stream(fs): await _start(mode="persistent") _node(fs, "n1", system="sys-2") await streams.reconcile_once() - s = fs.get("discord_streams", "g1:c1") + s = fs.get("discord_streams", _sid(fs, "g1", "c1")) assert s["desired"] == "on" and s["token_id"] is None and s["actual"]["state"] == "error" assert _lease(fs, "t1")["releasing"] is True @@ -432,7 +444,7 @@ async def test_force_release_detaches_streams(fs): old = await leases.force_release_token("t1") assert old["node_id"] == "n1" assert _lease(fs, "t1") is None - assert fs.get("discord_streams", "g1:c1")["token_id"] is None + assert fs.get("discord_streams", _sid(fs, "g1", "c1"))["token_id"] is None def test_node_token_pickup_only_for_the_lease_holder(fs): diff --git a/drb-c2-core/tests/test_discord_multi_stream.py b/drb-c2-core/tests/test_discord_multi_stream.py new file mode 100644 index 0000000..ad5bb17 --- /dev/null +++ b/drb-c2-core/tests/test_discord_multi_stream.py @@ -0,0 +1,440 @@ +""" +Several radio streams in ONE Discord voice channel (server-26#213 follow-up, +owner decision 2026-10-05). + +Owner-reported bug on live: adding a radio channel for a server + voice +channel that already had one — different label, system, even a different +preferred bot — overwrote the existing stream and made its bot leave, because +the stream id was ``guild:channel``. Now every new stream gets its own id; +each stream has its own bot (Discord: one voice connection per bot per +server), and a chosen bot that is already in the server is refused by name. +""" +import asyncio +import threading + +import pytest +from fastapi.testclient import TestClient + +from app.config import settings +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_service_key_or_admin +from app.main import app +from app.routers import discord_web as web +from tests.fake_firestore import FakeFirestore + +ORG = "org-a" +P25, FIRE = "sys-p25", "sys-fire" +ADMIN = {"uid": "u-admin", "role": "admin", "email": "admin@example.test"} +OWNER = {"uid": "u-own", "org_id": ORG, "org_role": "owner"} +client = TestClient(app) + + +@pytest.fixture +def fs(monkeypatch): + fake = FakeFirestore.install(monkeypatch) + fake.published = [] + from app.internal.mqtt_handler import mqtt_handler + monkeypatch.setattr(mqtt_handler, "publish_discord_desired", + lambda n, p: fake.published.append((n, p)) or True) + streams._publish_locks.clear() + streams._guild_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", "") + monkeypatch.setattr(settings, "discord_guild_max_streams", 4) + monkeypatch.setattr(settings, "discord_node_max_streams", 4) + + # One node per system: a node plays the one system it is tuned to. + for nid, system in (("n-p25", P25), ("n-fire", FIRE)): + fake.put("nodes", nid, {"node_id": nid, "name": f"Node {nid}", "org_id": ORG, "status": "online", + "approval_status": "approved", "assigned_system_id": system}) + fake.put("systems", P25, {"system_id": P25, "name": "County P25", "org_id": ORG}) + fake.put("systems", FIRE, {"system_id": FIRE, "name": "City Fire", "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 in ("t1", "t2", "t3"): + fake.put("bot_tokens", tid, {"token_id": tid, "name": f"Radio {tid}", "token": f"secret-{tid}", + "org_id": None, "guild_ids": ["100", "200"], "valid": True, + "application_id": f"app-{tid}", "lease": None}) + yield fake + app.dependency_overrides.clear() + + +def _lease(fs, tid): + return (fs.get("bot_tokens", tid) or {}).get("lease") + + +def _live(fs, channel="1001"): + return sorted((d for d in (fs.data.get("discord_streams") or {}).values() + if d.get("channel_id") == channel and d.get("desired") == "on"), + key=lambda d: d.get("created_at") or "") + + +def _post(body): + app.dependency_overrides[require_service_key_or_admin] = lambda: ADMIN + return client.post("/discord/streams", json={"guild_id": "100", "channel_id": "1001", **body}) + + +def _legacy(fs, token="t1", node="n-p25", system=P25, channel="1001"): + """A stream doc as it was created before this change: id guild:channel.""" + sid = f"100:{channel}" + fs.put("discord_streams", sid, { + "stream_id": sid, "org_id": ORG, "guild_id": "100", "channel_id": channel, "channel_name": "Old label", + "system_id": system, "system_name": "County P25", "mode": "persistent", "desired": "on", + "node_id": node, "token_id": token, "lease_id": "L-old", "actual": {"state": "connected"}, + "created_at": "2026-10-01T00:00:00+00:00", + }) + fs.put("bot_tokens", token, {**fs.get("bot_tokens", token), + "lease": {"node_id": node, "lease_id": "L-old", "since": "x", + "streams": [sid], "releasing": False, "release_seq": None}}) + return sid + + +# --------------------------------------------------------------------------- +# The reported bug +# --------------------------------------------------------------------------- + +def test_adding_a_second_stream_to_a_channel_never_touches_the_first(fs): + """Same server + voice channel, different label, system and preferred bot: + a second row, each with its own bot — the first keeps playing.""" + a = _post({"system_id": P25, "channel_name": "County", "preferred_token_id": "t1"}) + assert a.status_code == 200, a.text + first = fs.get("discord_streams", a.json()["stream_id"]) + + b = _post({"system_id": FIRE, "channel_name": "Fire", "preferred_token_id": "t2"}) + assert b.status_code == 200, b.text + assert b.json()["stream_id"] != a.json()["stream_id"] + assert a.json()["stream_id"].startswith("100:1001:") and b.json()["stream_id"].startswith("100:1001:") + + assert fs.get("discord_streams", a.json()["stream_id"]) == first # byte-for-byte untouched + live = _live(fs) + assert [(s["system_id"], s["channel_name"], s["token_id"], s["node_id"]) for s in live] == [ + (P25, "County", "t1", "n-p25"), (FIRE, "Fire", "t2", "n-fire")] + assert _lease(fs, "t1")["streams"] == [a.json()["stream_id"]] # t1 never left + assert _lease(fs, "t2")["streams"] == [b.json()["stream_id"]] + # Each node is told only about its own stream. + last = {nid: p for nid, p in fs.published} + assert [s["stream_id"] for s in last["n-p25"]["streams"]] == [a.json()["stream_id"]] + assert [s["stream_id"] for s in last["n-fire"]["streams"]] == [b.json()["stream_id"]] + + +def test_any_free_bot_never_reuses_a_bot_already_in_the_server(fs): + """Two streams in one channel on ONE node (same token pool): different bots.""" + fs.put("nodes", "n-fire", {**fs.get("nodes", "n-fire"), "assigned_system_id": P25}) + fs.put("systems", "sys-3", {"system_id": "sys-3", "name": "Third", "org_id": ORG}) + a = asyncio.run(streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25, + node_id="n-p25")) + b = asyncio.run(streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id="sys-3", + node_id="n-p25")) + assert a["stream_id"] != b["stream_id"] and a["token_id"] != b["token_id"] + assert sorted(_lease(fs, a["token_id"])["streams"] + _lease(fs, b["token_id"])["streams"]) == \ + sorted([a["stream_id"], b["stream_id"]]) + + +def test_preferred_bot_already_in_the_server_is_refused_by_name(fs): + a = _post({"system_id": P25, "channel_name": "County", "preferred_token_id": "t1"}) + first = fs.get("discord_streams", a.json()["stream_id"]) + + r = _post({"system_id": FIRE, "channel_name": "Fire", "preferred_token_id": "t1"}) + assert r.status_code == 409 + body = r.json() + assert body["code"] == "preferred_busy" and body["busy_stream_id"] == a.json()["stream_id"] + assert "Radio t1" in body["detail"] and "County P25" in body["detail"] and "#County" in body["detail"] + # Not a silent swap: nothing moved, no second stream left behind half-made. + assert fs.get("discord_streams", a.json()["stream_id"]) == first + assert len(_live(fs)) == 1 and _lease(fs, "t1")["streams"] == [a.json()["stream_id"]] + assert all(_lease(fs, t) is None for t in ("t2", "t3")) + + # Same bot in ANOTHER channel of the same server: still one voice connection per server. + app.dependency_overrides[require_service_key_or_admin] = lambda: ADMIN + r2 = client.post("/discord/streams", json={"guild_id": "100", "channel_id": "1002", "system_id": P25, + "preferred_token_id": "t1"}) + assert r2.status_code == 409 and r2.json()["code"] == "preferred_busy" + + +def test_preferred_bot_on_another_node_is_refused_not_swapped(fs): + # t1 is logged in on n-p25 (serving another server); a FIRE stream needs n-fire. + fs.put("bot_tokens", "t1", {**fs.get("bot_tokens", "t1"), + "lease": {"node_id": "n-p25", "lease_id": "L", "streams": ["200:2001:aaaaaaaaaaaa"], + "releasing": False, "release_seq": None}}) + r = _post({"system_id": FIRE, "preferred_token_id": "t1"}) + assert r.status_code == 409 and r.json()["code"] == "preferred_other_node" + assert "Node n-p25" in r.json()["detail"] + assert _live(fs) == [] + + +def test_pool_exhausted_for_the_server_says_to_add_and_invite_a_bot(fs): + for tid in ("t2", "t3"): + fs.put("bot_tokens", tid, {**fs.get("bot_tokens", tid), "valid": False}) + assert _post({"system_id": P25}).status_code == 200 + r = _post({"system_id": FIRE}) + assert r.status_code == 409 and r.json()["code"] == "pool_exhausted" + assert "one voice channel per server" in r.json()["detail"] and "invite it" in r.json()["detail"] + + +def test_same_system_twice_in_a_channel_is_the_same_stream(fs): + a = _post({"system_id": P25, "channel_name": "County", "mode": "persistent"}).json() + again = _post({"system_id": P25, "channel_name": "Renamed", "mode": "adhoc"}) + assert again.status_code == 200 + assert again.json()["stream_id"] == a["stream_id"] and again.json()["already_on"] is True + doc = fs.get("discord_streams", a["stream_id"]) + assert doc["channel_name"] == "County" and doc["mode"] == "persistent" # a re-add is not an edit + assert len(_live(fs)) == 1 + # ...and naming a DIFFERENT bot for it is a clear conflict, not a swap. + r = _post({"system_id": P25, "preferred_token_id": "t3" if a["token_id"] != "t3" else "t2"}) + assert r.status_code == 409 and r.json()["code"] == "duplicate" + + +def test_stop_by_stream_id_leaves_the_other_stream_in_the_channel_on(fs): + a = _post({"system_id": P25}).json() + b = _post({"system_id": FIRE}).json() + r = client.post("/discord/guilds/100/stop", json={"stream_id": b["stream_id"], "reason": "test"}) + assert r.status_code == 200 and r.json()["stopped"] == [b["stream_id"]] + assert [s["stream_id"] for s in _live(fs)] == [a["stream_id"]] + # A stream id from another server matches nothing in this one. + assert client.post("/discord/guilds/999/stop", json={"stream_id": a["stream_id"]}).json()["stopped"] == [] + assert client.post("/discord/guilds/100/stop", json={"stream_id": "100/1001"}).status_code == 422 + # Channel-wide stop still stops every stream in the channel. + b2 = _post({"system_id": FIRE}).json() + assert b2["stream_id"] == b["stream_id"] # restarted, not a new doc + r = client.post("/discord/guilds/100/stop", json={"channel_id": "1001"}) + assert sorted(r.json()["stopped"]) == sorted([a["stream_id"], b["stream_id"]]) + + +def test_restart_and_edit_by_stream_id(fs): + a = _post({"system_id": P25, "channel_name": "County"}).json() + b = _post({"system_id": FIRE, "channel_name": "Fire"}).json() + client.post("/discord/guilds/100/stop", json={"stream_id": a["stream_id"]}) + # The admin page's Start button names the stream: it restarts THAT one. + r = _post({"system_id": P25, "stream_id": a["stream_id"]}) + assert r.status_code == 200 and r.json()["stream_id"] == a["stream_id"] + assert fs.get("discord_streams", b["stream_id"])["token_id"] == b["token_id"] + # Explicit edit to a free bot moves the stream to it (and only it). + free = next(t for t in ("t1", "t2", "t3") if t not in (r.json()["token_id"], b["token_id"])) + e = _post({"system_id": P25, "stream_id": a["stream_id"], "preferred_token_id": free}) + assert e.status_code == 200 and e.json()["token_id"] == free + assert a["stream_id"] not in (_lease(fs, r.json()["token_id"]) or {}).get("streams", []) + # Editing to the bot the OTHER stream in this server uses is refused. + e2 = _post({"system_id": P25, "stream_id": a["stream_id"], "preferred_token_id": b["token_id"]}) + assert e2.status_code == 409 and e2.json()["code"] == "preferred_busy" + # A stream id must belong to the server/channel it is posted with. + bad = client.post("/discord/streams", json={"guild_id": "100", "channel_id": "1002", "system_id": P25, + "stream_id": a["stream_id"]}) + assert bad.status_code == 400 + missing = _post({"system_id": P25, "stream_id": "100:1001:0123456789ab"}) + assert missing.status_code == 404 + + +async def test_reconciler_replaces_each_stream_in_a_shared_channel_by_id(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_lease_offline_release_seconds", 600) + fs.put("nodes", "n-p25b", {"node_id": "n-p25b", "name": "Node n-p25b", "org_id": ORG, "status": "online", + "approval_status": "approved", "assigned_system_id": P25}) + a = await streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25, + mode="persistent", node_id="n-p25") + b = await streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=FIRE, + mode="persistent") + from datetime import datetime, timezone, timedelta + old = (datetime.now(timezone.utc) - timedelta(seconds=900)).isoformat() + fs.put("nodes", "n-p25", {**fs.get("nodes", "n-p25"), "status": "offline", "last_seen": old}) + await streams.reconcile_once() + moved = fs.get("discord_streams", a["stream_id"]) + assert moved["desired"] == "on" and moved["node_id"] == "n-p25b" and moved["token_id"] + assert moved["token_id"] != fs.get("discord_streams", b["stream_id"])["token_id"] + assert fs.get("discord_streams", b["stream_id"])["node_id"] == "n-fire" + assert len(_live(fs)) == 2 # no third doc + + +# --------------------------------------------------------------------------- +# Races — the per-server bot rule holds under concurrency +# --------------------------------------------------------------------------- + +def _hold_both_at_commit(fake): + barrier = threading.Barrier(2, timeout=5) + + def _hook(attempt): + if attempt == 0: + barrier.wait() + fake.before_commit = _hook + + +async def test_two_streams_one_channel_racing_never_share_a_bot(fs): + """Two NEW streams in one server on one node, leased concurrently: the + transaction re-reads the pool, so they end up on two different bots.""" + s1, s2 = leases.new_stream_id("100", "1001"), leases.new_stream_id("100", "1001") + _hold_both_at_commit(fs) + r1, r2 = await asyncio.gather( + leases.lease_for_stream(s1, node_id="n-p25", org_id=ORG), + leases.lease_for_stream(s2, node_id="n-p25", org_id=ORG), + ) + assert fs.retries >= 1 # the race really happened + assert r1.ok and r2.ok and r1.token_id != r2.token_id + for tid in (r1.token_id, r2.token_id): + assert len([s for s in _lease(fs, tid)["streams"] if leases.guild_of(s) == "100"]) == 1 + + +async def test_racing_for_the_last_free_bot_in_a_server_one_loses_cleanly(fs): + fs.put("bot_tokens", "t3", {**fs.get("bot_tokens", "t3"), "valid": False}) + fs.put("bot_tokens", "t2", {**fs.get("bot_tokens", "t2"), + "lease": {"node_id": "n-p25", "lease_id": "L2", "streams": ["100:1002:bbbbbbbbbbbb"], + "releasing": False, "release_seq": None}}) + s1, s2 = leases.new_stream_id("100", "1001"), leases.new_stream_id("100", "1001") + _hold_both_at_commit(fs) + r1, r2 = await asyncio.gather( + leases.lease_for_stream(s1, node_id="n-p25", org_id=ORG), + leases.lease_for_stream(s2, node_id="n-p25", org_id=ORG), + ) + assert sorted([r1.ok, r2.ok]) == [False, True] + loser = r2 if r1.ok else r1 + assert loser.error == "pool_exhausted" + assert len(_lease(fs, "t1")["streams"]) == 1 + + +async def test_racing_same_preferred_bot_for_two_streams_in_one_server(fs): + s1, s2 = leases.new_stream_id("100", "1001"), leases.new_stream_id("100", "1002") + _hold_both_at_commit(fs) + r1, r2 = await asyncio.gather( + leases.lease_for_stream(s1, node_id="n-p25", org_id=ORG, preferred_token_id="t1"), + leases.lease_for_stream(s2, node_id="n-p25", org_id=ORG, preferred_token_id="t1"), + ) + assert sorted([r1.ok, r2.ok]) == [False, True] + loser = r2 if r1.ok else r1 + assert loser.error == "preferred_busy" and loser.detail["busy_stream_id"] in (s1, s2) + assert len(_lease(fs, "t1")["streams"]) == 1 + + +def _slow_find(monkeypatch): + """Yield to the other requests right after every stream-list read (the + "already on?" and "server at its cap?" checks), so unserialised starts + would all read before any of them writes — as real Firestore reads allow.""" + from app.internal import firestore as fstore + real = fstore.collection_list + + async def _list(coll, **filters): + out = await real(coll, **filters) + if coll == streams.STREAMS: + await asyncio.sleep(0.05) + return out + monkeypatch.setattr(fstore, "collection_list", _list) + + +async def test_concurrent_adds_of_one_system_to_one_channel_make_one_stream(fs, monkeypatch): + _slow_find(monkeypatch) + r1, r2 = await asyncio.gather( + streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25), + streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=P25), + ) + assert r1["stream_id"] == r2["stream_id"] + assert len(_live(fs)) == 1 + assert len([t for t in ("t1", "t2", "t3") if _lease(fs, t)]) == 1 + + +async def test_concurrent_adds_respect_the_server_cap(fs, monkeypatch): + _slow_find(monkeypatch) + fs.put("systems", "sys-3", {"system_id": "sys-3", "name": "Third", "org_id": ORG}) + fs.put("nodes", "n-3", {"node_id": "n-3", "name": "Node n-3", "org_id": ORG, "status": "online", + "approval_status": "approved", "assigned_system_id": "sys-3"}) + results = await asyncio.gather(*( + streams.start_stream(org_id=ORG, guild_id="100", channel_id="1001", system_id=sid, max_streams=2) + for sid in (P25, FIRE, "sys-3")), return_exceptions=True) + errors = [r for r in results if isinstance(r, streams.StreamError)] + assert len(errors) == 1 and errors[0].code == "limit" + assert len(_live(fs)) == 2 + + +# --------------------------------------------------------------------------- +# Old guild:channel ids keep working +# --------------------------------------------------------------------------- + +def test_old_id_stop_and_admin_remove(fs): + sid = _legacy(fs) + app.dependency_overrides[require_service_key_or_admin] = lambda: ADMIN + app.dependency_overrides[require_admin_token] = lambda: ADMIN + assert client.get(f"/discord/streams/{sid}").json()["stream_id"] == sid + r = client.post("/discord/guilds/100/stop", json={"stream_id": sid}) + assert r.json()["stopped"] == [sid] + assert _lease(fs, "t1")["releasing"] is True and _lease(fs, "t1")["streams"] == [] + r = client.delete(f"/discord/streams/{sid}") + assert r.status_code == 200 and fs.get("discord_streams", sid) is None + + +def test_old_id_doc_is_not_overwritten_by_an_add_and_is_restarted_by_its_own_system(fs): + sid = _legacy(fs) + b = _post({"system_id": FIRE, "channel_name": "Fire"}).json() + assert b["stream_id"] != sid and b["token_id"] != "t1" + old = fs.get("discord_streams", sid) + assert old["desired"] == "on" and old["token_id"] == "t1" and old["channel_name"] == "Old label" + # Re-adding the old stream's own system lands on the old doc (idempotent). + again = _post({"system_id": P25}).json() + assert again["stream_id"] == sid and again["already_on"] is True + + +async def test_old_id_node_page_stop_and_remove(fs): + sid = _legacy(fs) + r = await web.stop_node_streams("n-p25", web.NodeStreamStop(stream_id=sid), OWNER) + assert r["stopped"] == [sid] + r = await web.remove_node_stream("n-p25", sid, OWNER) + assert r["ok"] and fs.get("discord_streams", sid) is None + + +async def test_old_id_still_reported_on_by_the_node(fs): + sid = _legacy(fs) + await streams.handle_actual("n-p25", {"streams": {sid: {"state": "error", "error": "x", "code": "no_audio"}, + "100/evil": {"state": "connected"}}}) + assert fs.get("discord_streams", sid)["actual"]["code"] == "no_audio" + + +# --------------------------------------------------------------------------- +# Node page + admin page flows +# --------------------------------------------------------------------------- + +async def test_node_page_adds_beside_another_systems_stream(fs): + p25 = await web.start_node_stream("n-p25", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER) + fire = await web.start_node_stream("n-fire", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER) + assert p25["stream_id"] != fire["stream_id"] and p25["token_id"] != fire["token_id"] + assert fs.get("discord_streams", p25["stream_id"])["node_id"] == "n-p25" # not moved + + opts = await web.join_options("n-fire", OWNER) + ch = next(c for c in opts["guilds"][0]["voice_channels"] if c["channel_id"] == "1001") + assert [s["system_name"] for s in ch["streams"]] == ["County P25", "City Fire"] + assert {s["token_name"] for s in ch["streams"]} == {f"Radio {p25['token_id']}", f"Radio {fire['token_id']}"} + assert ch["stream"]["stream_id"] == fire["stream_id"] # this node's system's one + assert [s["stream_id"] for s in opts["streams"]] == [fire["stream_id"]] + assert opts["guilds"][0]["streams_on"] == 2 + + # Re-picking on the same node is idempotent (and free under the node cap). + again = await web.start_node_stream("n-fire", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER) + assert again["stream_id"] == fire["stream_id"] + # Removing one leaves the other. + await web.remove_node_stream("n-fire", fire["stream_id"], OWNER) + assert [s["stream_id"] for s in _live(fs)] == [p25["stream_id"]] + + +def test_admin_overview_lists_one_row_per_stream_with_its_bot(fs): + app.dependency_overrides[require_admin_token] = lambda: ADMIN + a = _post({"system_id": P25, "channel_name": "County"}).json() + b = _post({"system_id": FIRE, "channel_name": "Fire"}).json() + body = client.get("/discord/admin/overview").json() + rows = [(r["stream_id"], r["channel_id"], r["system_name"], r["token_name"]) for r in body["streams"]] + assert rows == [(a["stream_id"], "1001", "County P25", f"Radio {a['token_id']}"), + (b["stream_id"], "1001", "City Fire", f"Radio {b['token_id']}")] + assert body["guilds"][0]["live_streams"] == 2 + # Remove one by id: the other row stays and stays on. + assert client.delete(f"/discord/streams/{a['stream_id']}").status_code == 200 + body = client.get("/discord/admin/overview").json() + assert [(r["stream_id"], r["desired"]) for r in body["streams"]] == [(b["stream_id"], "on")] + + +def test_bot_overview_names_each_streams_bot(fs): + a = _post({"system_id": P25}).json() + b = _post({"system_id": FIRE}).json() + body = client.get("/discord/guilds/100/overview").json() + assert {(s["stream_id"], s["token_name"]) for s in body["streams"]} == { + (a["stream_id"], f"Radio {a['token_id']}"), (b["stream_id"], f"Radio {b['token_id']}")} + busy = {x["token_id"]: x["busy_here"] for x in body["bots"]} + assert busy[a["token_id"]] and busy[b["token_id"]] 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 b010c0c..7b2b910 100644 --- a/drb-c2-core/tests/test_discord_remove_and_logging.py +++ b/drb-c2-core/tests/test_discord_remove_and_logging.py @@ -53,6 +53,7 @@ def fs(monkeypatch): from app.internal.mqtt_handler import mqtt_handler monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish) streams._publish_locks.clear() + streams._guild_locks.clear() async def _no_discord(token_ids): return 0 @@ -301,18 +302,18 @@ def test_remove_guild_interrupted_by_a_restart_is_still_audited(fs, monkeypatch) def test_remove_guild_route_hides_server_and_relink_restores_it(fs): import asyncio - asyncio.run(_start("1001")) + sid = asyncio.run(_start("1001"))["stream_id"] 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.status_code == 200 and r.json()["streams_removed"] == [sid] 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"] + assert entry["details"]["previous_org_id"] == ORG and entry["details"]["streams_removed"] == [sid] # Admin link brings it back. assert client.put("/discord/guilds/100/link", json={"org_id": ORG}).status_code == 200 diff --git a/drb-c2-core/tests/test_discord_web_picker.py b/drb-c2-core/tests/test_discord_web_picker.py index d2bf74d..24eb44c 100644 --- a/drb-c2-core/tests/test_discord_web_picker.py +++ b/drb-c2-core/tests/test_discord_web_picker.py @@ -39,6 +39,7 @@ def fs(monkeypatch): from app.internal.mqtt_handler import mqtt_handler monkeypatch.setattr(mqtt_handler, "publish_discord_desired", lambda n, p: published.append((n, p)) or True) streams._publish_locks.clear() + streams._guild_locks.clear() async def _no_discord(token_ids): return 0 @@ -149,14 +150,14 @@ def test_options_reports_missing_bot_with_invite(fs): def test_start_pins_stream_to_node_with_names(fs): r = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER)) - assert r["ok"] and r["stream_id"] == "100:1001" and r["token_id"] == "t1" - doc = fs.get("discord_streams", "100:1001") + assert r["ok"] and r["stream_id"].startswith("100:1001:") and r["token_id"] == "t1" + doc = fs.get("discord_streams", r["stream_id"]) assert doc["node_id"] == "n1" and doc["org_id"] == ORG and doc["system_id"] == SYS assert doc["channel_name"] == "Dispatch" and doc["guild_name"] == "Fire Co. 7" assert doc["created_by_discord_id"] == "web:u-own" assert fs.published and fs.published[-1][0] == "n1" opts = run(web.join_options("n1", OWNER)) - assert [s["stream_id"] for s in opts["streams"]] == ["100:1001"] + assert [s["stream_id"] for s in opts["streams"]] == [r["stream_id"]] assert opts["guilds"][0]["voice_channels"][0]["stream"]["node_name"] == "Node n1" @@ -191,13 +192,13 @@ def test_start_lease_refusal_carries_invite(fs): def test_stop_one_and_scoped_to_node(fs): - run(web.start_node_stream("n1", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER)) + sid = run(web.start_node_stream("n1", web.NodeStreamStart(guild_id="100", channel_id="1001"), OWNER))["stream_id"] with pytest.raises(HTTPException) as e: - run(web.stop_node_streams("n2", web.NodeStreamStop(stream_id="100:1001"), ADMIN)) + run(web.stop_node_streams("n2", web.NodeStreamStop(stream_id=sid), ADMIN)) assert e.value.status_code == 404 - r = run(web.stop_node_streams("n1", web.NodeStreamStop(stream_id="100:1001"), OWNER)) - assert r["stopped"] == ["100:1001"] - assert fs.get("discord_streams", "100:1001")["desired"] == "off" + r = run(web.stop_node_streams("n1", web.NodeStreamStop(stream_id=sid), OWNER)) + assert r["stopped"] == [sid] + assert fs.get("discord_streams", sid)["desired"] == "off" def test_legacy_raw_id_join_refuses_other_orgs_server(fs): @@ -210,8 +211,10 @@ def test_legacy_raw_id_join_refuses_other_orgs_server(fs): def test_stop_rejects_malformed_stream_id(): # "a/b" would otherwise reach Firestore as a document path and 500. from pydantic import ValidationError - for bad in ("100/1001", "../x", "100:1001/extra", ""): + for bad in ("100/1001", "../x", "100:1001/extra", "", "100:1001:ab/cd", "100:1001:XYZ", "100:1001:"): with pytest.raises(ValidationError): web.NodeStreamStop(stream_id=bad) - assert web.NodeStreamStop(stream_id="100:1001").stream_id == "100:1001" + assert web.NodeStreamStop(stream_id="100:1001").stream_id == "100:1001" # legacy id + new = leases.new_stream_id("100", "1001") + assert web.NodeStreamStop(stream_id=new).stream_id == new # per-stream id assert web.NodeStreamStop().stream_id is None diff --git a/drb-frontend/components/NodeDiscordRadio.tsx b/drb-frontend/components/NodeDiscordRadio.tsx index 0409d93..2cc019e 100644 --- a/drb-frontend/components/NodeDiscordRadio.tsx +++ b/drb-frontend/components/NodeDiscordRadio.tsx @@ -5,6 +5,10 @@ // the Discord servers linked to this node's org and the voice channels the // DRB bot reported in them; a pick becomes a radio channel pinned to this node // (desired state — the node reconciles and reports back, shown per row). +// A voice channel can carry several radio streams, each its own bot playing +// its own system (server-26#213 follow-up): picking a channel that already +// plays another system ADDS this node's system beside it; only a stream of +// this node's own system there is restarted / moved here. // Layout follows design-canvas/discord-system-mockup.html "Radio channels". import { useCallback, useEffect, useMemo, useState } from "react"; @@ -64,6 +68,11 @@ function modeLabel(mode: DiscordStreamView["mode"]) { return mode === "persistent" ? "persistent" : "ad-hoc"; } +function streamLabel(s: DiscordStreamView) { + const sys = s.system_name ?? "radio"; + return s.token_name ? `${sys} (${s.token_name})` : sys; +} + // --------------------------------------------------------------------------- // Picker // --------------------------------------------------------------------------- @@ -199,6 +208,7 @@ function DiscordPicker({
{grp.channels.map((c) => { const here = c.stream?.node_id === nodeId; + const others = (c.streams ?? []).filter((x) => x.stream_id !== c.stream?.stream_id); return (