Logan CusanoandClaude Opus 5.5
8af410aede
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 <noreply@anthropic.com>
732 lines
35 KiB
Python
732 lines
35 KiB
Python
"""
|
|
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) 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
|
|
|
|
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.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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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."""
|
|
link = await fstore.doc_get(GUILDS, str(guild_id))
|
|
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 _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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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.",
|
|
"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: 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.
|
|
"""
|
|
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 {}
|
|
|
|
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)
|
|
# Keep the current node when it is still a valid choice (reuse the lease).
|
|
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"])
|
|
|
|
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
|
|
|
|
last: 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)
|
|
if not res.ok and res.error == "no_bot_in_guild" 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 leases.refresh_membership(stale):
|
|
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)
|
|
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}
|
|
if res.error in ("preferred_busy", "preferred_unavailable"):
|
|
raise await _lease_error(res) # true on every node — no point trying the rest
|
|
last = res
|
|
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:
|
|
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"}
|
|
|
|
|
|
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"])
|
|
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 leases.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}")
|