server-26/drb-c2-core/app/internal/discord_streams.py

"""
Discord radio channels as desired state (server-26#203, #204; node-26#25).

One ``discord_streams/{stream_id}`` doc per radio stream. A Discord voice
channel may carry SEVERAL streams (server-26#213 follow-up, owner decision
2026-10-05): each has its own radio bot and plays its own system, so a channel
can hold e.g. County P25 on one bot and City Fire on another. ``stream_id`` is
``{guild_id}:{channel_id}:{suffix}`` (``discord_leases.new_stream_id``); docs
from before are ``{guild_id}:{channel_id}`` and are read, stopped and removed
exactly the same way. Adding a stream never reuses another stream's doc except
the same system in the same channel (idempotent re-add / restart). Fields:

  org_id, guild_id, channel_id, channel_name, guild_name,
  system_id, system_name,
  mode: "persistent" | "adhoc"
      persistent — stays connected and is re-placed by the reconciler (node
                   reboot, node lost, C2 restart) until someone stops it.
      adhoc      — stopped instead of re-placed when its node goes away.
                   (Idle auto-dismiss is not built yet — see the PR.)
  desired: "on" | "off"
  node_id, token_id, lease_id — placement (written with the lease, atomically)
  actual: {state, error, code, since, at} — reported by the node, never guessed here
      state: pending | connecting | connected | error | stopping |
             node_offline | paused_locally
      code:  the node's machine-readable reason on error (login_failed,
             connect_failed, not_in_guild, bad_channel, no_audio,
             missing_permissions, channel_full) or None
  tg_filter: None — phase-2 seam (per-channel talkgroup gate); always None now
  now_playing_message_id, created_by_discord_id, created_at, updated_at

Before a token is leased, ``start_stream`` asks Discord (REST, cached —
discord_rest.guild_snapshot) whether each bot it could pick may View, Connect
and Speak in the voice channel (server-26#225 S3). A bot that is denied is
never handed to a node; when every bot is denied the start is refused with
the bot, the channel, the missing permission and what removed it. A check
that can't be completed is "unknown" and allowed (fail open). Kill switch:
``DISCORD_PERM_CHECK_ENABLED=false``. The system's preferred bot
(``systems/{id}.preferred_token_id``) is tried first when nobody named one.

Flow: ``start_stream`` picks a node + leases a token -> ``publish_desired``
retains ``nodes/{node}/discord/desired`` (full list for that node, with a
monotonically increasing ``seq``) -> the edge reconciles and publishes
``nodes/{node}/discord/actual`` -> ``handle_actual`` updates ``actual`` and
settles token releases. ``reconcile_loop`` re-places persistent streams whose
node went away and retries desired-state publishes that never reached the
broker.

The desired payload carries token **ids**, never token strings: the edge
fetches the string over HTTPS (GET /discord/node/{node_id}/tokens/{id}) with
its own node API key, and C2 hands it out only to the node holding the lease.
A retained MQTT message therefore never stores a bot credential on the broker.
"""
from __future__ import annotations

import asyncio
from datetime import datetime, timezone, timedelta
from typing import Optional

from app.config import settings
from app.internal import firestore as fstore
from app.internal import discord_leases as leases
from app.internal import discord_perms
from app.internal import discord_rest
from app.internal.logger import logger

STREAMS = leases.STREAMS
GUILDS = "discord_guilds"      # discord_guilds/{guild_id}: {org_id, linked_by, linked_at}
NODE_STATE = "discord_nodes"   # discord_nodes/{node_id}: {seq, published, applied_seq, actual_at}
ONLINE = ("online", "recording")


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _effective_system(node: dict) -> Optional[str]:
    if node.get("is_overridden") and node.get("override_system_id"):
        return node["override_system_id"]
    return node.get("assigned_system_id")


def _parse_ts(value) -> Optional[datetime]:
    try:
        ts = datetime.fromisoformat(value) if isinstance(value, str) else value
    except (TypeError, ValueError):
        return None
    if isinstance(ts, datetime) and ts.tzinfo is None:
        ts = ts.replace(tzinfo=timezone.utc)
    return ts if isinstance(ts, datetime) else None


# ---------------------------------------------------------------------------
# Guild -> org
# ---------------------------------------------------------------------------

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)


# ---------------------------------------------------------------------------
# Discord permission check (server-26#225 S3)
# ---------------------------------------------------------------------------

# Trace sentences (discord_perms) that only say a permission went with
# another one. The fix is for the other one, so they are not what we report.
_IMPLIED = ("without View Channel", "without Connect")


def _join_labels(labels: list) -> str:
    if len(labels) <= 1:
        return "".join(labels)
    return ", ".join(labels[:-1]) + " or " + labels[-1]


def _fix_text(bot: str, missing: list, reason: str) -> str:
    if "timed out" in reason:
        return (f"In Discord: remove {bot}'s timeout (right-click it in the member list → Remove Timeout), "
                "then try again.")
    allow = " and ".join(missing) if len(missing) <= 2 else ", ".join(missing[:-1]) + " and " + missing[-1]
    return f"In Discord: Edit Channel → Permissions → add {bot} → allow {allow}, then try again."


def evaluate_access(snap: "discord_rest.GuildSnapshot", channel_id: str, *, bot: str) -> dict:
    """One bot's verdict for one voice channel, from its cached snapshot.

    ``status``: ok | denied | unknown | not_member | invalid | bad_channel
    (the channel exists but is not a voice channel). ``missing`` lists the
    edge's labels ("View Channel", "Connect", "Speak"); ``reason`` is the
    trace sentence for the first one that isn't just implied by another;
    ``fix`` is what to change in Discord. Never raises."""
    out: dict = {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None,
                 "user_limit": 0, "limit_bypass": False, "channel_name": None}
    if snap.status != discord_rest.OK:
        out["status"] = snap.status
        out["error"] = snap.error
        return out
    channel = next((c for c in snap.channels or []
                    if isinstance(c, dict) and str(c.get("id")) == str(channel_id)), None)
    if channel is None:
        # Deleted, or hidden from the bot's list: the edge's join is the judge.
        out["error"] = "channel not in the server's channel list"
        return out
    out["channel_name"] = channel.get("name")
    try:
        bits, trace = discord_perms.effective_permissions(snap.guild or {}, snap.member or {}, channel)
    except (ValueError, TypeError, AttributeError) as e:
        out["error"] = f"could not read permissions ({type(e).__name__})"
        return out
    verdict = discord_perms.check_voice(bits, channel, trace)
    out["user_limit"] = verdict["user_limit"]
    out["limit_bypass"] = verdict["limit_bypass"]
    if not verdict["is_voice"]:
        out["status"] = "bad_channel"
        return out
    if verdict["ok"]:
        out["status"] = leases.PERM_OK
        return out
    missing = verdict["missing"]
    reasons = verdict["reasons"]
    primary = [m for m in missing if not any(x in (reasons.get(m) or "") for x in _IMPLIED)] or missing
    out.update(status=leases.PERM_DENIED, missing=missing, reasons=reasons,
               reason=reasons.get(primary[0]) or "not granted",
               fix=_fix_text(bot, primary, reasons.get(primary[0]) or ""))
    return out


def tokens_to_check(tokens: dict[str, dict], *, guild_id: str, org_id: Optional[str],
                    node_ids: Optional[set] = None, exclude_stream_id: Optional[str] = None,
                    preferred_token_id: Optional[str] = None) -> list[str]:
    """The bots ``pick_token`` could hand this stream to, best first — the
    only ones worth asking Discord about. ``node_ids`` None = any node."""
    out = []
    for tok_id, tok in tokens.items():
        if tok.get("valid") is False or not tok.get("token"):
            continue
        if tok.get("org_id") is not None and tok.get("org_id") != org_id:
            continue  # another org's bot is never asked about this org's server
        members = tok.get("guild_ids")
        if members is not None and guild_id not in members:
            continue
        if preferred_token_id:
            if tok_id == preferred_token_id:
                out.append(tok_id)
            continue
        lease_node = (tok.get("lease") or {}).get("node_id")
        if lease_node and node_ids is not None and lease_node not in node_ids:
            continue
        if any(leases.guild_of(s) == guild_id
               for s in leases._lease_streams(tok) if s != exclude_stream_id):
            continue
        out.append(tok_id)
    out.sort(key=lambda t: leases._candidate_order(t, tokens[t], "", org_id, None))
    return out


def _unanswered() -> dict:
    return {"status": discord_rest.UNKNOWN, "missing": [], "reason": None, "fix": None,
            "error": "Discord did not answer in time"}


async def gather_snapshots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, *,
                           refresh: bool = False, budget: Optional[float] = None) -> dict:
    """``discord_rest.guild_snapshot`` for each bot, at most
    ``discord_perm_check_concurrency`` at a time, all within
    ``discord_perm_check_budget_seconds``. A bot not answered in time maps
    to None — unknown (its load keeps running and fills the cache for next
    time: the cache shields it from this cancellation)."""
    out: dict = {}
    if not token_ids:
        return out
    sem = asyncio.Semaphore(max(1, settings.discord_perm_check_concurrency))

    async def _one(tok_id: str) -> None:
        async with sem:
            out[tok_id] = await discord_rest.guild_snapshot({**tokens[tok_id], "token_id": tok_id}, guild_id,
                                                            refresh=refresh)

    tasks = [asyncio.create_task(_one(t)) for t in token_ids]
    _, pending = await asyncio.wait(tasks, timeout=budget if budget is not None
                                    else settings.discord_perm_check_budget_seconds)
    for t in pending:
        t.cancel()
    for t in tasks:
        if t.done() and not t.cancelled() and t.exception() is not None:
            logger.warning(f"Discord permission check failed: {type(t.exception()).__name__}")
    return {t: out.get(t) for t in token_ids}


async def check_bots(tokens: dict[str, dict], token_ids: list[str], guild_id: str, channel_id: str,
                     *, refresh: bool = False, budget: Optional[float] = None) -> dict[str, dict]:
    """``evaluate_access`` for each bot in one voice channel. Never raises."""
    snaps = await gather_snapshots(tokens, token_ids, guild_id, refresh=refresh, budget=budget)
    return {t: (evaluate_access(snaps[t], channel_id, bot=tokens[t].get("name") or t) if snaps.get(t)
                else _unanswered())
            for t in token_ids}


async def _perm_check(*, org_id: str, guild_id: str, channel_id: str, stream_id: str,
                      preferred_token_id: Optional[str], node_ids: set) -> Optional[dict]:
    if not settings.discord_perm_check_enabled:
        return None
    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
    ids = tokens_to_check(tokens, guild_id=guild_id, org_id=org_id, node_ids=node_ids,
                          exclude_stream_id=stream_id, preferred_token_id=preferred_token_id)
    return await check_bots(tokens, ids, guild_id, channel_id)


def _channel_ref(perms: Optional[dict], channel_name: Optional[str]) -> str:
    name = next((p.get("channel_name") for p in (perms or {}).values() if p.get("channel_name")), None)
    name = name or channel_name
    return f"#{name}" if name else "that voice channel"


def _missing_permissions_error(denied: list, invite: list, chan: str) -> "StreamError":
    def line(d: dict) -> str:
        return f"{d['name']} can't {_join_labels(d.get('missing') or ['Speak'])} in {chan}: {d.get('reason')}."
    if len(denied) == 1:
        d = denied[0]
        text = f"{line(d)} {d.get('fix') or ''}".strip()
    else:
        shown = denied[:3]
        text = f"No radio bot can play in {chan}:\n" + "\n".join(f"- {line(d)}" for d in shown)
        if len(denied) > len(shown):
            text += f"\n- …and {len(denied) - len(shown)} more."
        text += "\nIn Discord: Edit Channel → Permissions → add one of these bots → allow the missing " \
                "permission, then try again."
    if invite:
        text += " Or invite another radio bot with a link below."
    return StreamError("missing_permissions", text, invite=invite, extra={"denied": denied})


# ---------------------------------------------------------------------------
# Start / stop
# ---------------------------------------------------------------------------

class StreamError(Exception):
    def __init__(self, code: str, message: str, invite: Optional[list] = None, status: int = 409,
                 extra: Optional[dict] = None):
        super().__init__(message)
        self.code = code
        self.message = message
        self.invite = invite or []
        self.status = status
        self.extra = extra or {}


ERROR_TEXT = {
    "no_org": "This Discord server is not linked to a DRB organization yet.",
    "no_node": "No online node in your organization is listening to that system right now.",
    "pool_empty": "No radio bots have been added to DRB yet. A platform admin adds them under Radio bots.",
    "no_bot_in_guild": "No free radio bot is a member of this server. Invite one with the link below, then try again.",
    "pool_exhausted": ("Every radio bot that can join this server is already busy — a bot can be in only one voice "
                       "channel per server, and a bot carrying radio on another node can't be shared. Add another "
                       "radio bot to DRB and invite it to this server, or stop a radio channel here."),
    "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.",
    "preferred_unavailable": "That radio bot can't be used here (it was removed, failed validation, or belongs to "
                             "another organization). Pick another bot, or let DRB choose any free one.",
    "pool_full": ("No radio bot is in this server, and every DRB radio bot that could be invited is already in its "
                  "maximum number of servers. A platform admin needs to add another radio bot."),
    "missing_permissions": "No radio bot has the Discord permissions it needs in that voice channel.",
    "bad_channel": "That is not a voice channel. Radio can only play in a voice channel.",
    "not_found": "That radio channel no longer exists.",
}

# One c2-core process. Serialising start_stream per guild makes "is this
# system already playing in this channel?" + "is the server at its cap?" +
# create a real check-then-write: two concurrent adds of the same system to
# one channel land on ONE stream, and two adds can't both take the last slot.
_guild_locks: dict[str, asyncio.Lock] = {}


def _guild_lock(guild_id: str) -> asyncio.Lock:
    lock = _guild_locks.get(guild_id)
    if lock is None:
        lock = _guild_locks[guild_id] = asyncio.Lock()
    return lock


def _stream_label(s: dict) -> str:
    where = f"#{s['channel_name']}" if s.get("channel_name") else f"channel {s.get('channel_id')}"
    return f"{s.get('system_name') or s.get('system_id') or 'radio'} in {where}"


async def find_channel_stream(guild_id: str, channel_id: str, system_id: str) -> Optional[dict]:
    """The stream already carrying ``system_id`` in this voice channel: a live
    one first, else the most recently updated stopped one (restarted instead
    of piling up a new doc per /join + /leave). None if there is neither."""
    same = [s for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id))
            if str(s.get("channel_id")) == str(channel_id) and s.get("system_id") == system_id and s.get("stream_id")]
    live = sorted((s for s in same if s.get("desired") == "on"), key=lambda s: s.get("created_at") or "")
    if live:
        return live[0]
    same.sort(key=lambda s: s.get("updated_at") or "", reverse=True)
    return same[0] if same else None


async def _lease_error(res: leases.LeaseResult, chan: str = "that voice channel") -> "StreamError":
    code = res.error or "no_node"
    d = res.detail or {}
    bot = d.get("token_name") or "That radio bot"
    if code == "missing_permissions" and d.get("denied"):
        return _missing_permissions_error(d["denied"], res.invite, chan)
    if code == "preferred_busy":
        busy = (await fstore.doc_get(STREAMS, d["busy_stream_id"]) if d.get("busy_stream_id") else None) or {}
        what = _stream_label(busy) if busy else "another radio channel"
        return StreamError(code, f"{bot} is already playing {what} in this server. A bot can be in only one voice "
                                 "channel per server — pick another bot, or let DRB choose any free one.",
                           extra={"busy_stream_id": d.get("busy_stream_id")})
    if code == "preferred_other_node":
        node = (await fstore.doc_get("nodes", d["node_id"]) if d.get("node_id") else None) or {}
        return StreamError(code, f"{bot} is carrying radio on another node ({node.get('name') or d.get('node_id')}) "
                                 "that can't serve this channel. Pick another bot, or let DRB choose any free one.")
    return StreamError(code, ERROR_TEXT.get(code, code), invite=res.invite)


async def start_stream(
    *,
    org_id: str,
    guild_id: Optional[str] = None,
    channel_id: Optional[str] = None,
    system_id: str,
    mode: str = "adhoc",
    channel_name: Optional[str] = None,
    guild_name: Optional[str] = None,
    created_by: Optional[str] = None,
    node_id: Optional[str] = None,
    preferred_token_id: Optional[str] = None,
    max_streams: Optional[int] = None,
    stream_id: Optional[str] = None,
) -> dict:
    """Put ``system_id`` on a voice channel.

    * ``stream_id`` given — restart / edit THAT stream (the reconciler, the
      admin page's Start button). Its guild and channel come from the doc.
    * otherwise — ADD a stream to ``guild_id:channel_id``. If this system is
      already on in that channel the existing stream is returned (idempotent
      re-add, ``already_on``); a stopped one with this system is restarted;
      anything else gets a brand-new stream id, so another system's stream in
      the same channel is never touched.

    ``preferred_token_id`` is strict: a bot already in this server, on another
    node, or unusable is refused with a reason, never silently swapped. With
    none given, the system's ``preferred_token_id`` is tried first (a hint:
    another bot is used when it can't take the stream).

    Every bot that could be picked is permission-checked against the voice
    channel first (``_perm_check``; server-26#225). Every entry point — the
    bot's /join, the web picker, the node command and the reconciler —
    comes through here, so none can hand a node a bot Discord will refuse.
    """
    if mode not in ("persistent", "adhoc"):
        raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400)
    if not org_id:
        raise StreamError("no_org", ERROR_TEXT["no_org"], status=403)
    existing: dict = {}
    if stream_id:
        existing = await fstore.doc_get(STREAMS, stream_id) or {}
        if not existing:
            raise StreamError("not_found", ERROR_TEXT["not_found"], status=404)
        doc_guild, doc_channel = str(existing.get("guild_id")), str(existing.get("channel_id"))
        if (guild_id and str(guild_id) != doc_guild) or (channel_id and str(channel_id) != doc_channel):
            raise StreamError("bad_stream", "That radio channel belongs to a different server or voice channel.",
                              status=400)
        guild_id, channel_id = doc_guild, doc_channel
    if not guild_id or not channel_id:
        raise StreamError("bad_stream", "guild_id and channel_id are required.", status=400)
    guild_id, channel_id = str(guild_id), str(channel_id)
    async with _guild_lock(guild_id):
        return await _start_locked(
            org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id, mode=mode,
            channel_name=channel_name, guild_name=guild_name, created_by=created_by, node_id=node_id,
            preferred_token_id=preferred_token_id, max_streams=max_streams, stream_id=stream_id,
            existing=existing,
        )


async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_id: str, mode: str,
                        channel_name: Optional[str], guild_name: Optional[str], created_by: Optional[str],
                        node_id: Optional[str], preferred_token_id: Optional[str], max_streams: Optional[int],
                        stream_id: Optional[str], existing: dict) -> dict:
    already_on = False
    if stream_id:
        existing = await fstore.doc_get(STREAMS, stream_id) or existing   # re-read under the lock
    else:
        existing = await find_channel_stream(guild_id, channel_id, system_id) or {}
        if existing.get("desired") == "on":
            already_on = True
            if preferred_token_id and existing.get("token_id") and existing["token_id"] != preferred_token_id:
                tok = await fstore.doc_get(leases.TOKENS, existing["token_id"]) or {}
                raise StreamError("duplicate", f"{_stream_label(existing)} is already playing on radio bot "
                                               f"{tok.get('name') or existing['token_id']}. Remove that one "
                                               "first, or pick another system.",
                                  extra={"busy_stream_id": existing["stream_id"]})
            # A re-add is not an edit: keep the running stream's label and mode.
            channel_name, mode = None, existing.get("mode") or mode
        stream_id = existing.get("stream_id") or leases.new_stream_id(guild_id, channel_id)

    if max_streams is not None and existing.get("desired") != "on":
        live = [s for s in await fstore.collection_list(STREAMS, guild_id=guild_id) if s.get("desired") == "on"]
        if len(live) >= max_streams:
            raise StreamError("limit", ERROR_TEXT["limit"])

    system = await fstore.doc_get_cached("systems", system_id) or {}
    # Finding 1 (#225 plan): "preferred for this system" was written by
    # PUT /tokens/{id}/prefer/{system} and never read. It is a hint, not a
    # demand — an explicit preferred_token_id still wins and stays strict.
    # Read uncached: the systems cache lives 5 min, and a preference the
    # owner just set should apply to the next start.
    soft_pref = None
    if not preferred_token_id:
        soft_pref = ((await fstore.doc_get("systems", system_id)) or {}).get("preferred_token_id") or None

    if node_id:
        node = await fstore.doc_get("nodes", node_id)
        # A pinned node still has to be this org's and approved; it may be
        # listening to anything (the web "Join Discord" button pins a node).
        if not node or node.get("org_id") != org_id or node.get("approval_status") not in (None, "approved"):
            raise StreamError("no_node", ERROR_TEXT["no_node"])
        candidates = [node]
    else:
        candidates = await candidate_nodes(org_id, system_id, exclude_stream_id=stream_id)
        cur = existing.get("node_id") if existing.get("system_id") == system_id else None
        held_on = None
        if preferred_token_id or soft_pref:
            # A bot can only run on the node it is logged in on.
            tok = await fstore.doc_get(leases.TOKENS, preferred_token_id or soft_pref) or {}
            held_on = (tok.get("lease") or {}).get("node_id")
        # Keep the current node when it is still a valid choice (reuse the
        # lease) — unless a bot was demanded, whose node then comes first.
        order = [held_on, cur] if preferred_token_id else [cur, held_on]
        for nid in reversed(order):
            if nid:
                candidates.sort(key=lambda n, nid=nid: 0 if n["node_id"] == nid else 1)
    if not candidates:
        raise StreamError("no_node", ERROR_TEXT["no_node"])

    now = _now_iso()
    fields = {
        "stream_id": stream_id, "org_id": org_id, "guild_id": guild_id, "channel_id": channel_id,
        "system_id": system_id, "system_name": system.get("name") or system_id,
        "mode": mode, "desired": "on", "tg_filter": None, "stop_reason": None,
        "updated_at": now,
    }
    prev_actual = existing.get("actual") or {}
    if existing.get("desired") != "on" or prev_actual.get("state") not in ("connected", "connecting"):
        fields["actual"] = {"state": "pending", "error": None, "since": now, "at": now}
    if channel_name:
        fields["channel_name"] = channel_name
    if guild_name:
        fields["guild_name"] = guild_name
    if not existing:
        fields["created_at"] = now
        fields["created_by_discord_id"] = created_by
        fields["now_playing_message_id"] = None

    node_ids = {n["node_id"] for n in candidates}
    perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id, stream_id=stream_id,
                              preferred_token_id=preferred_token_id, node_ids=node_ids)
    if perms and any(p.get("status") == "bad_channel" for p in perms.values()):
        raise StreamError("bad_channel", f"{_channel_ref(perms, channel_name)} is not a voice channel. "
                                         "Radio can only play in a voice channel — pick one from the list.",
                          status=400)

    last: Optional[leases.LeaseResult] = None
    perm_refusal: Optional[leases.LeaseResult] = None
    prev_node = existing.get("node_id")
    refreshed = False
    for node in candidates:
        res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id,
                                            preferred_token_id=preferred_token_id, stream_fields=fields,
                                            soft_preferred_token_id=soft_pref, perms=perms)
        if not res.ok and res.error == "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 discord_rest.refresh_membership(stale):
                if perms is not None:
                    # Bots that just turned out to be members get checked too.
                    perms = await _perm_check(org_id=org_id, guild_id=guild_id, channel_id=channel_id,
                                              stream_id=stream_id, preferred_token_id=preferred_token_id,
                                              node_ids=node_ids)
                res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id,
                                                    preferred_token_id=preferred_token_id, stream_fields=fields,
                                                    soft_preferred_token_id=soft_pref, perms=perms)
        if res.ok:
            delivered = await publish_desired(node["node_id"])
            if prev_node and prev_node != node["node_id"]:
                await publish_desired(prev_node)
            logger.info(f"Discord stream {stream_id} -> node {node['node_id']} token {res.token_id}"
                        f" ({'reused' if res.reused else 'new'} lease{', already on' if already_on else ''})")
            return {**existing, **fields, "node_id": node["node_id"], "node_name": node.get("name"),
                    "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused,
                    "delivered": delivered, "already_on": already_on}
        chan = _channel_ref(perms, channel_name or existing.get("channel_name"))
        if res.error in ("preferred_busy", "preferred_unavailable") or (
                preferred_token_id and res.error == "missing_permissions"):
            raise await _lease_error(res, chan)           # true on every node — no point trying the rest
        if res.error == "missing_permissions":
            perm_refusal = res                            # more useful than another node's "all busy"
        last = res
    if last is None:
        raise StreamError("no_node", ERROR_TEXT["no_node"])
    raise await _lease_error(perm_refusal or last,
                             _channel_ref(perms, channel_name or existing.get("channel_name")))


async def stop_stream(stream_id: str, reason: str = "stopped") -> bool:
    existing = await fstore.doc_get(STREAMS, stream_id)
    if not existing:
        return False
    now = _now_iso()
    node_id = await leases.detach_stream(stream_id, {
        "desired": "off", "updated_at": now, "stop_reason": reason,
        "actual": {**(existing.get("actual") or {}), "state": "stopping", "at": now},
    })
    if node_id:
        await publish_desired(node_id)
    logger.info(f"Discord stream {stream_id} stopped ({reason}) node={node_id or existing.get('node_id')}"
                f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}")
    return True


async def delete_stream(stream_id: str, reason: str = "removed") -> Optional[dict]:
    """Permanently remove a radio channel (server-26#213 follow-up).

    Order matters (server-26#216 lease semantics): the stream is first stopped
    through the normal desired-state path — detached from its token (token ->
    ``releasing``) and the node's list republished without it — and only then
    is the doc deleted. The token returns to the pool exactly as for /leave,
    when the node acks a list without it and no longer has a session on it
    (``settle_releases`` reads the token lease, never the stream doc).

    The delete itself is a transaction that refuses if the stream was started
    again in between (desired on, or a token re-leased onto it): deleting it
    then would strand that token's lease on a stream that no longer exists.
    Returns the doc as it was before removal, or None if it did not exist.
    """
    existing = await fstore.doc_get(STREAMS, stream_id)
    if not existing:
        return None
    if existing.get("desired") == "on" or existing.get("token_id"):
        await stop_stream(stream_id, reason)
    if not await leases.delete_stopped_stream(stream_id):
        raise StreamError("restarted", "That radio channel was started again while it was being removed. "
                                       "Stop it and try again.")
    logger.info(f"Discord stream {stream_id} removed ({reason}) node={existing.get('node_id')}"
                f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}"
                f" org={existing.get('org_id')}")
    return existing


async def remove_guild(guild_id: str, by: str) -> dict:
    """Forget a Discord server: remove every radio channel in it (stop first,
    then delete — ``delete_stream``) and tombstone its ``discord_guilds`` doc.

    The doc is NOT deleted: with no doc, ``resolve_guild_org`` would hand the
    server to ``DISCORD_UNLINKED_GUILD_ORG_ID`` the next time anyone typed a
    command there. It is unlinked the way PR #218's unlink does it
    (``org_id`` None + ``unlinked_at``, so no fallback) and stamped
    ``removed_at`` so the admin page stops listing it. Linking it again (``/drb
    link`` or the admin link) clears ``removed_at``."""
    guild_id = str(guild_id)
    link = await fstore.doc_get(GUILDS, guild_id) or {}
    removed = []
    for s in await fstore.collection_list(STREAMS, guild_id=guild_id):
        try:
            if s.get("stream_id") and await delete_stream(s["stream_id"], "server_removed"):
                removed.append(s["stream_id"])
        except StreamError as e:
            # A channel restarted mid-removal: stop here, leave the server
            # linked, and tell the caller which channels are already gone so
            # the partial removal is still audited.
            e.removed = removed
            raise
    now = _now_iso()
    await fstore.doc_set(GUILDS, guild_id, {
        "guild_id": guild_id, "org_id": None,
        "unlinked_at": now, "unlinked_by": by,
        "removed_at": now, "removed_by": by,
    }, merge=True)
    logger.info(f"Discord guild {guild_id} removed by {by}: {len(removed)} radio channel(s) deleted,"
                f" was org={link.get('org_id')!r}")
    return {"guild_id": guild_id, "previous_org_id": link.get("org_id"), "streams_removed": removed,
            "guild_name": link.get("guild_name")}


async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None,
                     stream_id: Optional[str] = None) -> list[str]:
    """Stop live streams in ONE Discord server: all of them, every stream in
    one voice channel, or one stream by id. The /leave fix for server-26#203:
    never touches another server's streams (a ``stream_id`` from another
    server simply matches nothing)."""
    stopped = []
    for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)):
        if s.get("desired") != "on":
            continue
        if channel_id and str(s.get("channel_id")) != str(channel_id):
            continue
        if stream_id and s.get("stream_id") != stream_id:
            continue
        await stop_stream(s["stream_id"], reason)
        stopped.append(s["stream_id"])
    return stopped


async def stop_node(node_id: str, reason: str) -> list[str]:
    stopped = []
    for s in await fstore.collection_list(STREAMS, node_id=node_id):
        if s.get("desired") == "on":
            await stop_stream(s["stream_id"], reason)
            stopped.append(s["stream_id"])
    return stopped


# ---------------------------------------------------------------------------
# Actual state <- node
# ---------------------------------------------------------------------------

_VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"}
# Edge error codes that mean the cached permission/membership view is stale.
_INVALIDATE_ON = {"missing_permissions", "channel_full", "not_in_guild", "bad_channel"}


def _log_transition(node_id: str, doc: dict, prev_state: Optional[str], state: str,
                    error: Optional[str], code: Optional[str]) -> None:
    """One log line per node-reported state CHANGE (never per heartbeat — the
    edge re-reports every ~60s). Errors are warnings so they surface in the
    c2-core log without turning debug on; a radio bot that cannot log in gets
    its own line naming the token, since that is the one failure that takes
    every channel on the token down at once."""
    where = (f"stream={doc.get('stream_id')} node={node_id} guild={doc.get('guild_id')}"
             f" channel={doc.get('channel_id')} token={doc.get('token_id')}")
    if state == "error":
        if code == "login_failed":
            logger.warning(f"Radio bot {doc.get('token_id')} failed to log in on node {node_id}: {error} ({where})")
        else:
            logger.warning(f"Discord stream {prev_state or 'pending'} -> error [{code or 'unknown'}]:"
                           f" {error} ({where})")
    else:
        logger.info(f"Discord stream {prev_state or 'pending'} -> {state} ({where})"
                    + (f": {error}" if error else ""))


async def handle_actual(node_id: str, payload: dict) -> None:
    """``nodes/{node_id}/discord/actual``. A node may only report on streams
    currently placed on it — anything else is ignored, so a compromised node
    cannot rewrite another node's (or org's) stream state."""
    if not isinstance(payload, dict):
        return
    reported = payload.get("streams") or {}
    now = _now_iso()
    recheck: set = set()
    if isinstance(reported, dict):
        for stream_id, st in list(reported.items())[:50]:
            if not isinstance(st, dict) or "/" in str(stream_id):
                continue  # "/" would be a Firestore sub-path, never a stream id
            doc = await fstore.doc_get(STREAMS, str(stream_id))
            if not doc or doc.get("node_id") != node_id or doc.get("desired") != "on":
                continue
            state = st.get("state") if st.get("state") in _VALID_STATES else "error"
            error = st.get("error")
            error = str(error)[:300] if error else None
            code = st.get("code")
            code = str(code)[:40] if code else None
            if code == "not_in_guild" and doc.get("token_id"):
                recheck.add(doc["token_id"])
            if code in _INVALIDATE_ON and doc.get("token_id"):
                # The edge saw Discord refuse; a cached "ok" must not keep
                # handing this bot the same channel (server-26#225).
                discord_rest.invalidate(token_id=doc["token_id"], guild_id=str(doc.get("guild_id")))
            prev = doc.get("actual") or {}
            if prev.get("state") == state and prev.get("error") == error and prev.get("code") == code:
                since = prev.get("since") or now
            else:
                since = now
                _log_transition(node_id, doc, prev.get("state"), state, error, code)
            await fstore.doc_update(STREAMS, str(stream_id), {
                "actual": {"state": state, "error": error, "code": code, "since": since, "at": now},
            })
    if recheck:
        await discord_rest.refresh_membership(sorted(recheck))

    if "sessions" in payload:
        sessions = {str(t) for t in (payload.get("sessions") or [])}
        applied = payload.get("applied_seq")
        applied = int(applied) if isinstance(applied, (int, float)) and not isinstance(applied, bool) else None
        freed = await leases.settle_releases(node_id, sessions, applied)
        for tok in freed:
            logger.info(f"Radio bot token {tok} released — node {node_id} confirmed it logged out")
        await fstore.doc_set(NODE_STATE, node_id, {"applied_seq": applied, "actual_at": now})


# ---------------------------------------------------------------------------
# Self-healing: re-place persistent streams whose node is gone
# ---------------------------------------------------------------------------

async def release_offline_node(node_id: str) -> None:
    """Node offline past the lease grace: free its tokens (the edge already
    logged them out — see release_node_leases) and unplace its streams."""
    orphaned = await leases.release_node_leases(node_id)
    if not orphaned:
        return
    now = _now_iso()
    for sid in orphaned:
        await fstore.doc_set(STREAMS, sid, {
            "token_id": None, "lease_id": None,
            "actual": {"state": "node_offline", "error": "The node carrying this channel went offline.",
                       "since": now, "at": now},
        })
    await publish_desired(node_id)  # retained: the node sees an accurate (empty) list when it returns
    logger.warning(f"Node {node_id} offline past lease grace — released {len(orphaned)} radio stream(s)")


async def _unplace(s: dict, why: str) -> None:
    """Take a stream off its node (token detached, node republished) but leave
    it desired, so the reconciler can place it somewhere valid."""
    now = _now_iso()
    old = await leases.detach_stream(s["stream_id"], {
        "node_id": None,
        "actual": {"state": "error", "error": why, "since": now, "at": now},
    })
    if old:
        await publish_desired(old)


async def reconcile_once() -> None:
    nodes = {n["node_id"]: n for n in await fstore.collection_list("nodes") if n.get("node_id")}
    grace = timedelta(seconds=settings.discord_lease_offline_release_seconds)
    now = datetime.now(timezone.utc)
    released: set = set()
    for s in await fstore.collection_list(STREAMS, desired="on"):
        node = nodes.get(s.get("node_id") or "")
        if node and s.get("token_id"):
            if node.get("status") in ONLINE:
                if _effective_system(node) == s.get("system_id"):
                    continue  # healthy placement
                # Node retuned to another system under a live stream.
                if s.get("mode") != "persistent":
                    await stop_stream(s["stream_id"], "node_retuned")
                    continue
                await _unplace(s, "The node carrying this channel switched systems; finding another.")
            else:
                ls = _parse_ts(node.get("last_seen"))
                if ls and now - ls < grace:
                    await _mark(s, "node_offline", "The node carrying this channel is offline; waiting for it.")
                    continue  # hold the lease through the grace window
                if node["node_id"] not in released:
                    await release_offline_node(node["node_id"])
                    released.add(node["node_id"])
        if s.get("mode") != "persistent":
            await stop_stream(s["stream_id"], "node_lost")
            continue
        org_id = s.get("org_id")
        try:
            await start_stream(org_id=org_id, stream_id=s["stream_id"], system_id=s["system_id"],
                               mode="persistent")
        except StreamError as e:
            await _mark(s, "error", e.message)

    # Leases with NO live stream left (``releasing``) on a node that is gone:
    # /leave while the node was down, or the node was deleted. The pass above
    # only walks desired=on streams, so without this the token would stay out
    # of the pool until that node came back — forever for a dead node.
    for tok in await fstore.collection_list(leases.TOKENS):
        nid = (tok.get("lease") or {}).get("node_id")
        if not nid or nid in released:
            continue
        node = nodes.get(nid)
        if node and node.get("status") in ONLINE:
            continue
        if node is not None:
            heard = _parse_ts(node.get("last_seen"))
        else:
            # Deleted node (its MQTT client is revoked, so the edge fail-safe
            # logs out). Last heard = its last actual report, else lease start.
            st = await fstore.doc_get(NODE_STATE, nid) or {}
            heard = _parse_ts(st.get("actual_at")) or _parse_ts((tok.get("lease") or {}).get("since"))
        if heard and now - heard < grace:
            continue
        await release_offline_node(nid)
        released.add(nid)

    # Desired-state publishes the broker never took (MQTT was down).
    for st in await fstore.collection_list(NODE_STATE, published=False):
        if st.get("node_id"):
            await publish_desired(st["node_id"])


async def _mark(s: dict, state: str, error: str) -> None:
    prev = s.get("actual") or {}
    if prev.get("state") == state and prev.get("error") == error:
        return
    now = _now_iso()
    await fstore.doc_set(STREAMS, s["stream_id"], {"actual": {"state": state, "error": error,
                                                              "since": now, "at": now}})


async def reconcile_loop() -> None:
    logger.info("Discord stream reconciler started.")
    while True:
        await asyncio.sleep(settings.discord_reconcile_interval_seconds)
        try:
            await reconcile_once()
        except Exception as e:
            logger.error(f"Discord reconcile error: {e}")