Logan CusanoandClaude Opus 5.5
2c2cf9d9be
start_stream now asks Discord (cached REST snapshots, S2) whether each bot
it could pick may View, Connect and Speak in the voice channel, before the
lease. Covers all four entry points: bot POST /discord/streams, web picker,
nodes discord_join, and the reconciler's re-place.
- pick_token: denied bots skipped; confirmed-ok sorts before unknown;
unknown is allowed (fail open); all denied -> missing_permissions naming
each bot, missing permission, trace reason and fix. The reuse branch
falls through when its token is denied.
- Bounded: discord_perm_check_concurrency at once, within
discord_perm_check_budget_seconds; unanswered = unknown.
- Non-voice channel -> 400 bad_channel (not missing_permissions).
- systems/{id}.preferred_token_id is now read (plan finding 1), as a hint.
- Pool bots at discord_pool_bot_guild_cap are never offered as invites;
pool_full when nothing else is left (owner Q3).
- handle_actual drops cached access on missing_permissions, channel_full,
not_in_guild, bad_channel.
- New GET /discord/nodes/{node_id}/guilds/{guild_id}/voice-check (org
member, names only) and GET /discord/guilds/{guild_id}/voice-check
(service key or admin).
- PUT /tokens/{id}/prefer/{system_id}: org owners for their own systems
and own BYO bots; shared systems and pool bots stay admin-only (owner Q2).
Keeps both pointers single-valued; audited.
- Kill switch: DISCORD_PERM_CHECK_ENABLED=false.
- tests/conftest.py: autouse fixture so no test ever reaches discord.com.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
515 lines
23 KiB
Python
515 lines
23 KiB
Python
"""
|
|
Radio-bot token leases (server-26#204, node-26#25).
|
|
|
|
A pool "radio bot" is a Discord bot token that an edge node logs in with to
|
|
join voice channels. The rules this module enforces, all inside one Firestore
|
|
transaction per change so two concurrent requests can never both win:
|
|
|
|
* A token is leased to **at most one node** at a time (``lease.node_id``).
|
|
Discord allows one gateway session per token to be useful; two nodes logged
|
|
in on one token fight over the same voice connections.
|
|
* One leased token can serve several streams on its node, but **one per
|
|
guild** — Discord allows one voice connection per bot user per guild.
|
|
* Re-requesting a stream that already holds a lease on the same node reuses
|
|
it (``/radio play`` twice, a reconcile retry, a re-join) instead of taking a
|
|
second token.
|
|
* Release is **ordered**: stopping a stream only marks the token
|
|
``releasing``. The token returns to the pool once the node reports (MQTT
|
|
``nodes/{id}/discord/actual``) that it applied a desired-state version at
|
|
least as new as the one that dropped the token *and* has no gateway session
|
|
on it any more (``settle_releases``) — or, as a last resort, once the node
|
|
has been offline longer than the edge's own MQTT-loss fail-safe
|
|
(``release_node_leases``), at which point the edge has already logged out.
|
|
|
|
Firestore layout:
|
|
|
|
bot_tokens/{token_id}
|
|
token, name, org_id (None = platform pool, else BYO for that org),
|
|
guild_ids (list | None — None means membership never checked),
|
|
application_id, bot_user_id, valid, validated_at, check_error,
|
|
lease: None | {node_id, lease_id, since, streams: [stream_id],
|
|
releasing: bool, release_seq: int | None}
|
|
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/{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
|
|
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from app.config import settings
|
|
from app.internal import firestore as fstore
|
|
|
|
TOKENS = "bot_tokens"
|
|
STREAMS = "discord_streams"
|
|
|
|
# Connect (1<<20) + Speak (1<<21) + View Channel (1<<10). Nothing else — a
|
|
# radio bot never reads or sends messages, so it never asks to.
|
|
RADIO_BOT_PERMISSIONS = (1 << 10) | (1 << 20) | (1 << 21)
|
|
|
|
|
|
def invite_url(application_id: Optional[str], guild_id: Optional[str] = None) -> Optional[str]:
|
|
if not application_id:
|
|
return None
|
|
url = (
|
|
"https://discord.com/oauth2/authorize"
|
|
f"?client_id={application_id}&scope=bot&permissions={RADIO_BOT_PERMISSIONS}"
|
|
)
|
|
if guild_id:
|
|
url += f"&guild_id={guild_id}&disable_guild_select=true"
|
|
return url
|
|
|
|
|
|
# Old ids are ``guild:channel``; new ones add ``:<hex>``. 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]
|
|
|
|
|
|
def invitable(tok: dict) -> bool:
|
|
"""May DRB suggest inviting this bot to one more server? A pool bot
|
|
(``org_id`` None) at ``discord_pool_bot_guild_cap`` servers is full: an
|
|
unverified Discord bot can't join more than 100, and the cap leaves room
|
|
(owner decision Q3, 2026-10-05: cap 90, add bots when full). An org's own
|
|
BYO bot is the org's business, never capped here."""
|
|
if tok.get("org_id") is not None:
|
|
return True
|
|
members = tok.get("guild_ids")
|
|
return members is None or len(members) < settings.discord_pool_bot_guild_cap
|
|
|
|
|
|
# Permission-check outcomes the lease code acts on (server-26#225 S3). The
|
|
# full per-bot result is built by discord_streams.check_bots; pick_token reads
|
|
# only ``status`` (plus name/missing/reason/fix for the refusal).
|
|
PERM_OK = "ok"
|
|
PERM_DENIED = "denied"
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@dataclass
|
|
class LeaseResult:
|
|
ok: bool
|
|
token_id: Optional[str] = None
|
|
lease_id: Optional[str] = None
|
|
reused: bool = False
|
|
# 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) |
|
|
# missing_permissions (every bot that could take it lacks a Discord
|
|
# permission in the channel — ``detail["denied"]`` names each) |
|
|
# pool_full (no bot is in this server and every pool bot that could be
|
|
# invited is at discord_pool_bot_guild_cap)
|
|
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:
|
|
return list(((tok.get("lease") or {}).get("streams")) or [])
|
|
|
|
|
|
def _legacy_mirror(lease: Optional[dict]) -> dict:
|
|
if not lease:
|
|
return {"lease": None, "in_use": False, "assigned_node_id": None, "assigned_at": None}
|
|
return {
|
|
"lease": lease,
|
|
"in_use": True,
|
|
"assigned_node_id": lease["node_id"],
|
|
"assigned_at": lease.get("since"),
|
|
}
|
|
|
|
|
|
def _perm_status(perms: Optional[dict], tok_id: str) -> Optional[str]:
|
|
return ((perms or {}).get(tok_id) or {}).get("status")
|
|
|
|
|
|
def _denied_row(tok_id: str, tok: dict, perms: dict) -> dict:
|
|
p = perms.get(tok_id) or {}
|
|
return {"name": tok.get("name") or tok_id, "missing": list(p.get("missing") or []),
|
|
"reason": p.get("reason"), "fix": p.get("fix")}
|
|
|
|
|
|
def _candidate_order(tok_id: str, tok: dict, node_id: str, org_id: Optional[str],
|
|
preferred: Optional[str], perms: Optional[dict] = None) -> tuple:
|
|
lease = tok.get("lease") or {}
|
|
return (
|
|
0 if tok_id == preferred else 1, # the system's preferred bot
|
|
0 if org_id and tok.get("org_id") == org_id else 1, # BYO before pool
|
|
0 if _perm_status(perms, tok_id) == PERM_OK else 1, # confirmed-ok before unknown
|
|
0 if lease.get("node_id") == node_id else 1, # reuse a session
|
|
0 if tok.get("guild_ids") is not None else 1, # known member first
|
|
str(tok.get("name") or ""),
|
|
tok_id,
|
|
)
|
|
|
|
|
|
def pick_token(
|
|
tokens: dict[str, dict],
|
|
*,
|
|
node_id: str,
|
|
guild_id: str,
|
|
org_id: Optional[str],
|
|
preferred_token_id: Optional[str] = None,
|
|
exclude_stream_id: Optional[str] = None,
|
|
soft_preferred_token_id: Optional[str] = None,
|
|
perms: Optional[dict] = None,
|
|
) -> LeaseResult:
|
|
"""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.
|
|
``soft_preferred_token_id`` (the system's preferred bot,
|
|
``systems/{id}.preferred_token_id``) is a hint: tried first, but any other
|
|
bot that fits is used when it can't.
|
|
|
|
``perms`` ({token_id: {status, missing, reason, fix}}, server-26#225) is
|
|
the Discord permission check for this voice channel. A ``denied`` bot is
|
|
skipped; a confirmed ``ok`` bot sorts before one whose check was
|
|
``unknown`` (or absent), and unknown is allowed (fail open: a Discord
|
|
outage must never block a start). None = no check (kill switch).
|
|
"""
|
|
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, perms=perms)
|
|
usable: list[tuple] = []
|
|
not_member: list[dict] = []
|
|
denied: list[dict] = []
|
|
pool_full = False
|
|
for tok_id, tok in tokens.items():
|
|
if tok.get("valid") is False:
|
|
continue
|
|
owner = tok.get("org_id")
|
|
if owner is not None and owner != org_id:
|
|
continue # another org's branded bot — never leased across orgs
|
|
lease = tok.get("lease") or {}
|
|
if lease and lease.get("node_id") and lease.get("node_id") != node_id:
|
|
continue # live (or still releasing) on another node
|
|
busy_guilds = {guild_of(s) for s in _lease_streams(tok) if s != exclude_stream_id}
|
|
if guild_id in busy_guilds:
|
|
continue # this bot already has a voice connection in this guild
|
|
members = tok.get("guild_ids")
|
|
if (members is not None and guild_id not in members) or _perm_status(perms, tok_id) == "not_member":
|
|
if invitable(tok):
|
|
not_member.append({"name": tok.get("name") or tok_id,
|
|
"invite_url": invite_url(tok.get("application_id"), guild_id)})
|
|
else:
|
|
pool_full = True
|
|
continue
|
|
if _perm_status(perms, tok_id) == "invalid":
|
|
continue # Discord answered 401 during the check
|
|
if _perm_status(perms, tok_id) == PERM_DENIED:
|
|
denied.append(_denied_row(tok_id, tok, perms or {}))
|
|
continue
|
|
usable.append((_candidate_order(tok_id, tok, node_id, org_id, soft_preferred_token_id, perms), tok_id))
|
|
if usable:
|
|
usable.sort()
|
|
return LeaseResult(ok=True, token_id=usable[0][1])
|
|
if denied:
|
|
# Fixing a permission is the shortest way to radio; inviting another
|
|
# bot is the other, so the invite links ride along.
|
|
return LeaseResult(ok=False, error="missing_permissions", invite=not_member,
|
|
detail={"denied": denied})
|
|
if not_member:
|
|
return LeaseResult(ok=False, error="no_bot_in_guild", invite=not_member)
|
|
if pool_full:
|
|
return LeaseResult(ok=False, error="pool_full")
|
|
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],
|
|
perms: Optional[dict] = None) -> 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)
|
|
or _perm_status(perms, tok_id) == "invalid"):
|
|
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) or _perm_status(perms, tok_id) == "not_member":
|
|
if not invitable(tok):
|
|
return LeaseResult(ok=False, error="pool_full", detail={"token_name": name})
|
|
return LeaseResult(ok=False, error="no_bot_in_guild",
|
|
invite=[{"name": name, "invite_url": invite_url(tok.get("application_id"), guild_id)}])
|
|
if _perm_status(perms, tok_id) == PERM_DENIED:
|
|
return LeaseResult(ok=False, error="missing_permissions",
|
|
detail={"token_name": name, "denied": [_denied_row(tok_id, tok, perms or {})]})
|
|
return LeaseResult(ok=True, token_id=tok_id)
|
|
|
|
|
|
async def lease_for_stream(
|
|
stream_id: str,
|
|
*,
|
|
node_id: str,
|
|
org_id: Optional[str],
|
|
preferred_token_id: Optional[str] = None,
|
|
stream_fields: Optional[dict] = None,
|
|
soft_preferred_token_id: Optional[str] = None,
|
|
perms: Optional[dict] = None,
|
|
) -> LeaseResult:
|
|
"""Bind ``stream_id`` to a token leased to ``node_id``, atomically.
|
|
|
|
Also writes ``stream_fields`` (desired=on, node_id, token_id, lease_id...)
|
|
onto the stream doc in the same transaction, so a stream doc can never name
|
|
a token whose lease does not list it.
|
|
|
|
``perms`` / ``soft_preferred_token_id``: see ``pick_token``. A stream's
|
|
current token is reused only while its permission check is not
|
|
``denied`` — otherwise the stream moves to a bot that can speak there.
|
|
"""
|
|
guild_id = guild_of(stream_id)
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
stream_ref = db.collection(STREAMS).document(stream_id)
|
|
stream_snap = stream_ref.get(transaction=txn)
|
|
stream = stream_snap.to_dict() if stream_snap.exists else {}
|
|
tokens = {s.id: s.to_dict() for s in db.collection(TOKENS).stream(transaction=txn)}
|
|
|
|
prev_tok_id = stream.get("token_id")
|
|
prev_tok = tokens.get(prev_tok_id) if prev_tok_id else None
|
|
if (
|
|
prev_tok
|
|
and stream.get("node_id") == node_id
|
|
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
|
|
and _perm_status(perms, prev_tok_id) not in (PERM_DENIED, "invalid", "not_member")
|
|
# 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"):
|
|
lease = {**lease, "releasing": False, "release_seq": None}
|
|
txn.update(db.collection(TOKENS).document(prev_tok_id), _legacy_mirror(lease))
|
|
txn.set(stream_ref, {**(stream_fields or {}), "node_id": node_id,
|
|
"token_id": prev_tok_id, "lease_id": lease["lease_id"]}, merge=True)
|
|
return LeaseResult(ok=True, token_id=prev_tok_id, lease_id=lease["lease_id"], reused=True)
|
|
|
|
result = pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id,
|
|
preferred_token_id=preferred_token_id, exclude_stream_id=stream_id,
|
|
soft_preferred_token_id=soft_preferred_token_id, perms=perms)
|
|
if not result.ok:
|
|
return result
|
|
|
|
# Detach from the previous token (moved node, or token went invalid).
|
|
if prev_tok and stream_id in _lease_streams(prev_tok) and prev_tok_id != result.token_id:
|
|
txn.update(db.collection(TOKENS).document(prev_tok_id),
|
|
_legacy_mirror(_without_stream(prev_tok["lease"], stream_id)))
|
|
|
|
tok = tokens[result.token_id]
|
|
lease = tok.get("lease") or {}
|
|
if lease.get("node_id") == node_id:
|
|
streams = [s for s in _lease_streams(tok) if s != stream_id] + [stream_id]
|
|
lease = {**lease, "streams": streams, "releasing": False, "release_seq": None}
|
|
else:
|
|
lease = {"node_id": node_id, "lease_id": uuid.uuid4().hex, "since": _now().isoformat(),
|
|
"streams": [stream_id], "releasing": False, "release_seq": None}
|
|
txn.update(db.collection(TOKENS).document(result.token_id), _legacy_mirror(lease))
|
|
txn.set(stream_ref, {**(stream_fields or {}), "node_id": node_id,
|
|
"token_id": result.token_id, "lease_id": lease["lease_id"]}, merge=True)
|
|
result.lease_id = lease["lease_id"]
|
|
return result
|
|
|
|
return await fstore.run_transaction(_txn)
|
|
|
|
|
|
def _without_stream(lease: dict, stream_id: str) -> dict:
|
|
streams = [s for s in (lease.get("streams") or []) if s != stream_id]
|
|
# The token stays leased to the node — it may still be logged in there —
|
|
# until settle_releases sees the node ack. ``releasing`` only once empty.
|
|
return {**lease, "streams": streams, "releasing": not streams,
|
|
"release_seq": lease.get("release_seq") if streams else None}
|
|
|
|
|
|
async def detach_stream(stream_id: str, stream_updates: Optional[dict] = None) -> Optional[str]:
|
|
"""Drop ``stream_id`` from its token's lease (token -> ``releasing`` if that
|
|
was its last stream) and apply ``stream_updates`` to the stream doc, in one
|
|
transaction. Returns the node_id the token is (still) leased to, if any."""
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
stream_ref = db.collection(STREAMS).document(stream_id)
|
|
snap = stream_ref.get(transaction=txn)
|
|
if not snap.exists:
|
|
return None
|
|
stream = snap.to_dict()
|
|
tok_id = stream.get("token_id")
|
|
tok_node = None
|
|
if tok_id:
|
|
tok_ref = db.collection(TOKENS).document(tok_id)
|
|
tsnap = tok_ref.get(transaction=txn)
|
|
tok = tsnap.to_dict() if tsnap.exists else None
|
|
if tok and tok.get("lease") and stream_id in _lease_streams(tok):
|
|
txn.update(tok_ref, _legacy_mirror(_without_stream(tok["lease"], stream_id)))
|
|
tok_node = tok["lease"].get("node_id")
|
|
txn.set(stream_ref, {**(stream_updates or {}), "token_id": None, "lease_id": None}, merge=True)
|
|
return tok_node or stream.get("node_id")
|
|
|
|
return await fstore.run_transaction(_txn)
|
|
|
|
|
|
async def delete_stopped_stream(stream_id: str) -> bool:
|
|
"""Delete a stream doc, but only while it is stopped and holds no token.
|
|
|
|
True when the doc is gone (deleted now, or already absent). False when it
|
|
is desired-on again or names a token — a concurrent start re-leased it,
|
|
and deleting it would leave that token's lease listing a stream nobody can
|
|
stop. ``lease_for_stream`` writes the stream doc in its own transaction,
|
|
so reading it here is enough to serialise against it."""
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
ref = db.collection(STREAMS).document(stream_id)
|
|
snap = ref.get(transaction=txn)
|
|
if not snap.exists:
|
|
return True
|
|
stream = snap.to_dict() or {}
|
|
if stream.get("desired") == "on" or stream.get("token_id"):
|
|
return False
|
|
txn.delete(ref)
|
|
return True
|
|
|
|
return await fstore.run_transaction(_txn)
|
|
|
|
|
|
async def stamp_release_seq(node_id: str, seq: int) -> None:
|
|
"""Record the desired-state version that dropped each releasing token, so a
|
|
stale actual report (from before the node saw the drop) can never free it."""
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
tokens = [(s.id, s.to_dict()) for s in db.collection(TOKENS).stream(transaction=txn)]
|
|
for tok_id, tok in tokens:
|
|
lease = tok.get("lease") or {}
|
|
if lease.get("node_id") == node_id and lease.get("releasing") and lease.get("release_seq") is None:
|
|
txn.update(db.collection(TOKENS).document(tok_id),
|
|
_legacy_mirror({**lease, "release_seq": seq}))
|
|
|
|
await fstore.run_transaction(_txn)
|
|
|
|
|
|
async def settle_releases(node_id: str, sessions: set, applied_seq: Optional[int]) -> list[str]:
|
|
"""Free tokens the node has confirmed it no longer runs. Returns freed ids."""
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
freed = []
|
|
tokens = [(s.id, s.to_dict()) for s in db.collection(TOKENS).stream(transaction=txn)]
|
|
for tok_id, tok in tokens:
|
|
lease = tok.get("lease") or {}
|
|
if lease.get("node_id") != node_id or not lease.get("releasing") or lease.get("streams"):
|
|
continue
|
|
rseq = lease.get("release_seq")
|
|
if rseq is None or applied_seq is None or applied_seq < rseq:
|
|
continue # the node may not have seen the drop yet
|
|
if tok_id in sessions:
|
|
continue # still logged in on the node
|
|
txn.update(db.collection(TOKENS).document(tok_id), _legacy_mirror(None))
|
|
freed.append(tok_id)
|
|
return freed
|
|
|
|
return await fstore.run_transaction(_txn)
|
|
|
|
|
|
async def release_node_leases(node_id: str) -> list[str]:
|
|
"""Last-resort release for a node that has been offline longer than the
|
|
edge's MQTT-loss fail-safe (the edge logs every radio bot out by itself
|
|
after DISCORD_MQTT_LOSS_STOP_SECONDS without the broker). Returns the
|
|
stream ids that lost their token so the caller can re-place them."""
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
orphaned = []
|
|
tokens = [(s.id, s.to_dict()) for s in db.collection(TOKENS).stream(transaction=txn)]
|
|
for tok_id, tok in tokens:
|
|
lease = tok.get("lease") or {}
|
|
if lease.get("node_id") != node_id:
|
|
continue
|
|
orphaned.extend(lease.get("streams") or [])
|
|
txn.update(db.collection(TOKENS).document(tok_id), _legacy_mirror(None))
|
|
return orphaned
|
|
|
|
return await fstore.run_transaction(_txn)
|
|
|
|
|
|
async def force_release_token(token_id: str) -> Optional[dict]:
|
|
"""Admin override (/tokens/flush, delete). Frees the token and detaches
|
|
every stream that named it, in one transaction, so no stream doc is left
|
|
pointing at a token whose lease no longer lists it. Returns the old lease
|
|
(``None`` if the token was not leased or does not exist); the caller
|
|
republishes desired state for ``lease["node_id"]`` so the edge logs out."""
|
|
db = fstore.db
|
|
|
|
def _txn(txn):
|
|
ref = db.collection(TOKENS).document(token_id)
|
|
snap = ref.get(transaction=txn)
|
|
if not snap.exists:
|
|
return None
|
|
lease = snap.to_dict().get("lease")
|
|
stream_refs = []
|
|
for sid in (lease or {}).get("streams") or []:
|
|
sref = db.collection(STREAMS).document(sid)
|
|
ssnap = sref.get(transaction=txn)
|
|
if ssnap.exists and (ssnap.to_dict() or {}).get("token_id") == token_id:
|
|
stream_refs.append(sref)
|
|
txn.update(ref, _legacy_mirror(None))
|
|
for sref in stream_refs:
|
|
txn.set(sref, {"token_id": None, "lease_id": None}, merge=True)
|
|
return lease
|
|
|
|
return await fstore.run_transaction(_txn)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token validation / guild membership / leaving a server moved to
|
|
# app/internal/discord_rest.py (server-26#225 S2): inspect_token,
|
|
# refresh_membership, leave_guild, TokenRejected.
|
|
# ---------------------------------------------------------------------------
|