""" Discord REST client for C2 (server-26#225 slice S2). C2 calls Discord's REST API with the radio-bot tokens it holds: to check a token when an admin adds it, to re-read which servers a bot is in, to make bots leave a server, and (from S3 on) to read a server's roles, the bot's own member record and the channel list, so C2 can tell whether a bot may join and speak in a voice channel before it hands the bot to a node. Everything here is built so that a Discord problem can never become a C2 problem: * **One shared ``httpx.AsyncClient``** with the User-Agent Discord asks bots to send (``DiscordBot (url, version)``). The token travels per request in the ``Authorization`` header and is never logged, never put in an error string and never used as a dict key (``_token_key`` hashes it). * **Rate limits per token.** Buckets are keyed by the ``X-RateLimit-Bucket`` Discord returns plus the route's major parameter (the guild or channel id). A request that would have to wait more than ``discord_rest_max_retry_after_seconds`` is not sent: its result is *unknown*. A 429 is retried once, only when ``retry_after`` is within that limit. A global 429 pauses that token. * **A process-wide breaker on invalid requests.** Discord bans the egress IP for an hour after 10,000 responses of 401, 403 or 429 (except 429s with ``X-RateLimit-Scope: shared``) in 10 minutes. The main DRB bot shares that IP. Once ``discord_rest_invalid_budget`` (default 4000) is reached inside ``discord_rest_invalid_window_seconds``, nothing more is sent and every result is unknown until the window drains. * **Unknown is not bad.** Callers fail open on an unknown result: a Discord outage must never block a stream start or starve the reconciler. Status mapping for the per-server snapshot (``guild_snapshot``): 401 token invalid: ``bot_tokens.valid = False``; never retried 404 10004 / 10007 not a member: the server is dropped from ``guild_ids`` 403 50001 unknown, then confirmed with ONE read of the bot's server list; ``guild_ids`` is never changed from the 403 alone anything else unknown Snapshots are cached in-process per (token_id, guild_id), one c2-core process being the same assumption ``_RateLimiter`` and ``_publish_locks`` already make. Results that let a bot in live ``discord_access_ttl_seconds``; not-member and invalid results live ``discord_access_negative_ttl_seconds``; unknown results live ``discord_access_unknown_ttl_seconds``, so the reconciler's 30 s loop can't turn into a REST loop against Discord. Concurrent loads of one key share one request (single flight). ``invalidate(token_id=, guild_id=)`` drops entries. """ from __future__ import annotations import asyncio import hashlib import re import time from collections import deque from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Awaitable, Callable, Optional import httpx from app.config import settings from app.internal import firestore as fstore from app.internal.logger import logger DISCORD_API = "https://discord.com/api/v10" USER_AGENT = "DiscordBot (https://drb.cusano.net, 1)" TOKENS = "bot_tokens" # GET /users/@me/guilds returns at most 200 per page. 50 pages is 10,000 # servers, well past what an unsharded bot can be in; the cap only stops a # misbehaving response from looping forever. GUILDS_PAGE = 200 MAX_GUILD_PAGES = 50 # Discord JSON error codes this module acts on. UNKNOWN_GUILD = 10004 UNKNOWN_MEMBER = 10007 MISSING_ACCESS = 50001 # Snapshot / access outcomes. ``denied`` is not produced here; the permission # check built on these snapshots (S3) uses it, with the same negative TTL. OK = "ok" DENIED = "denied" NOT_MEMBER = "not_member" INVALID = "invalid" UNKNOWN = "unknown" class TokenRejected(Exception): """Discord answered 401: the string is not a working bot token.""" def _now() -> datetime: return datetime.now(timezone.utc) def _token_key(token: str) -> str: """A stable, non-reversible handle for a token, for rate-limit state.""" return hashlib.sha256(token.encode()).hexdigest()[:16] # --------------------------------------------------------------------------- # Results # --------------------------------------------------------------------------- @dataclass class RestResult: """One Discord REST answer. ``status`` is None when nothing came back (not sent, network error): the result is unknown.""" status: Optional[int] data: Any = None code: Optional[int] = None # Discord's JSON error code, on an error error: Optional[str] = None # short reason; never holds the token @property def ok(self) -> bool: return self.status is not None and 200 <= self.status < 300 @property def unknown(self) -> bool: """No usable answer: not sent, network error, rate limited or 5xx.""" return self.status is None or self.status == 429 or self.status >= 500 # --------------------------------------------------------------------------- # Rate limits and the invalid-request breaker # --------------------------------------------------------------------------- _ID_RE = re.compile(r"\d{5,}") _MAJOR_RE = re.compile(r"^/(?:guilds|channels|webhooks)/(\d+)") def _route(method: str, path: str) -> tuple[str, str]: """(route key, major parameter) for a path. Snowflakes in the path are collapsed so ``/guilds/1/channels`` and ``/guilds/2/channels`` share a route; the major parameter keeps their buckets apart.""" m = _MAJOR_RE.match(path) return f"{method.upper()} {_ID_RE.sub('{id}', path)}", (m.group(1) if m else "") class _RateLimits: def __init__(self) -> None: self.route_bucket: dict[tuple[str, str], str] = {} self.buckets: dict[tuple[str, str, str], tuple[int, float]] = {} self.paused_until: dict[str, float] = {} def wait_for(self, tok: str, method: str, path: str, now: float) -> float: """Seconds to wait before this request may be sent (0 = go now).""" wait = max(0.0, self.paused_until.get(tok, 0.0) - now) route, major = _route(method, path) bucket = self.route_bucket.get((tok, route)) if bucket: remaining, reset_at = self.buckets.get((tok, bucket, major), (1, 0.0)) if remaining <= 0 and reset_at > now: wait = max(wait, reset_at - now) return wait def update(self, tok: str, method: str, path: str, headers: httpx.Headers, now: float) -> None: bucket = headers.get("x-ratelimit-bucket") if not bucket: return route, major = _route(method, path) self.route_bucket[(tok, route)] = bucket try: remaining = int(headers.get("x-ratelimit-remaining", "1")) reset_after = float(headers.get("x-ratelimit-reset-after", "0")) except ValueError: return self.buckets[(tok, bucket, major)] = (remaining, now + reset_after) def pause(self, tok: str, seconds: float, now: float) -> None: self.paused_until[tok] = max(self.paused_until.get(tok, 0.0), now + seconds) class _Breaker: """Counts 401 / 403 / non-shared 429 responses in a sliding window.""" def __init__(self) -> None: self.hits: deque[float] = deque() self.open_logged = False def _prune(self, now: float) -> None: horizon = now - settings.discord_rest_invalid_window_seconds while self.hits and self.hits[0] <= horizon: self.hits.popleft() def record(self, now: float) -> None: self.hits.append(now) def tripped(self, now: float) -> bool: self._prune(now) tripped = len(self.hits) >= settings.discord_rest_invalid_budget if tripped and not self.open_logged: logger.warning( f"Discord REST breaker open: {len(self.hits)} invalid responses in " f"{settings.discord_rest_invalid_window_seconds}s; REST checks report unknown " "until the window drains (protects the shared egress IP from a Discord ban)") self.open_logged = True elif not tripped: self.open_logged = False return tripped _limits = _RateLimits() _breaker = _Breaker() _sleep = asyncio.sleep # tests replace this, not asyncio.sleep itself # --------------------------------------------------------------------------- # The shared HTTP client # --------------------------------------------------------------------------- _client_obj: Optional[httpx.AsyncClient] = None _client_loop: Optional[asyncio.AbstractEventLoop] = None _transport: Optional[httpx.AsyncBaseTransport] = None def _client() -> httpx.AsyncClient: """The one shared client, recreated only if the event loop changed (the connection pool belongs to a loop; tests run one loop per test).""" global _client_obj, _client_loop loop = asyncio.get_running_loop() if _client_obj is None or _client_loop is not loop or _client_obj.is_closed: _client_obj = httpx.AsyncClient( base_url=DISCORD_API, timeout=settings.discord_rest_timeout_seconds, headers={"User-Agent": USER_AGENT}, transport=_transport, ) _client_loop = loop return _client_obj def set_transport(transport: Optional[httpx.AsyncBaseTransport]) -> None: """Tests: route every request through ``transport`` (e.g. httpx.MockTransport).""" global _transport, _client_obj _transport = transport _client_obj = None def reset() -> None: """Forget all in-process state: client, rate limits, breaker, cache.""" global _limits, _breaker, _client_obj _limits = _RateLimits() _breaker = _Breaker() _client_obj = None access_cache.clear() def _retry_after(r: httpx.Response) -> tuple[float, bool]: """(seconds, is_global) from a 429's body or headers.""" body: Any = None try: body = r.json() except Exception: body = None seconds: Optional[float] = None is_global = r.headers.get("x-ratelimit-global", "").lower() == "true" if isinstance(body, dict): if isinstance(body.get("retry_after"), (int, float)): seconds = float(body["retry_after"]) is_global = is_global or bool(body.get("global")) if seconds is None: try: seconds = float(r.headers.get("retry-after", "1")) except ValueError: seconds = 1.0 return max(0.0, seconds), is_global async def request(token: str, method: str, path: str, *, params: Optional[dict] = None) -> RestResult: """Send one request as the bot ``token``. Never raises for a Discord or network failure: those come back as an unknown RestResult.""" tok = _token_key(token) max_wait = settings.discord_rest_max_retry_after_seconds for attempt in (0, 1): now = time.monotonic() if _breaker.tripped(now): return RestResult(None, error="Discord REST paused (invalid-request budget reached)") wait = _limits.wait_for(tok, method, path, now) if wait > max_wait: return RestResult(None, error=f"rate limited for {wait:.0f}s") if wait > 0: await _sleep(wait) try: r = await _client().request(method, path, params=params, headers={"Authorization": f"Bot {token}"}) except httpx.HTTPError as e: # The exception text holds at most the URL, never the header. return RestResult(None, error=f"{type(e).__name__}: {str(e)[:120]}") now = time.monotonic() _limits.update(tok, method, path, r.headers, now) status = r.status_code if status in (401, 403) or (status == 429 and r.headers.get("x-ratelimit-scope") != "shared"): _breaker.record(now) if status == 429: seconds, is_global = _retry_after(r) if is_global: _limits.pause(tok, seconds, now) if attempt == 0 and seconds <= max_wait: await _sleep(seconds) continue return RestResult(429, error=f"rate limited (retry after {seconds:.1f}s)") data: Any = None if r.content: try: data = r.json() except ValueError: data = None code = data.get("code") if (status >= 400 and isinstance(data, dict)) else None err = None if 200 <= status < 300 else f"Discord answered {status}" + (f" ({code})" if code else "") return RestResult(status, data, code if isinstance(code, int) else None, err) return RestResult(429, error="rate limited") # pragma: no cover - loop always returns # --------------------------------------------------------------------------- # Cache (in-process, single flight) # --------------------------------------------------------------------------- def ttl_for(outcome: str) -> float: """How long a result with this outcome stays cached.""" if outcome == OK: return float(settings.discord_access_ttl_seconds) if outcome in (DENIED, NOT_MEMBER, INVALID): return float(settings.discord_access_negative_ttl_seconds) return float(settings.discord_access_unknown_ttl_seconds) class AccessCache: """TTL cache keyed by tuples that start ``(token_id, guild_id, ...)``. ``get_or_load`` runs one load per key at a time; concurrent callers await the same load. A load that overlaps an ``invalidate`` touching its key is returned to its callers but not stored, so an invalidation is never undone by a load that started before it.""" def __init__(self) -> None: self._entries: dict[tuple, tuple[float, Any]] = {} self._inflight: dict[tuple, asyncio.Task] = {} self._gen: dict[tuple, int] = {} def clear(self) -> None: self._entries.clear() self._inflight.clear() self._gen.clear() def peek(self, key: tuple) -> Any: hit = self._entries.get(key) if hit and hit[0] > time.monotonic(): return hit[1] return None async def get_or_load(self, key: tuple, loader: Callable[[], Awaitable[Any]], ttl_of: Callable[[Any], float], *, refresh: bool = False) -> Any: if not refresh: hit = self._entries.get(key) if hit and hit[0] > time.monotonic(): return hit[1] task = self._inflight.get(key) loop = asyncio.get_running_loop() if task is None or task.done() or task.get_loop() is not loop: gen = self._gen.get(key, 0) async def _load() -> Any: try: value = await loader() if self._gen.get(key, 0) == gen: self._entries[key] = (time.monotonic() + ttl_of(value), value) return value finally: if self._inflight.get(key) is task_ref[0]: self._inflight.pop(key, None) task_ref: list = [None] task = loop.create_task(_load()) task_ref[0] = task self._inflight[key] = task return await asyncio.shield(task) def invalidate(self, token_id: Optional[str] = None, guild_id: Optional[str] = None) -> int: """Drop entries for a token, a guild, both, or (no args) everything. Returns how many cached entries were dropped.""" def match(k: tuple) -> bool: return ((token_id is None or k[0] == token_id) and (guild_id is None or (len(k) > 1 and k[1] == str(guild_id)))) dropped = [k for k in self._entries if match(k)] for k in dropped: del self._entries[k] for k in set(self._inflight) | set(self._entries) | set(dropped): if match(k): self._gen[k] = self._gen.get(k, 0) + 1 for k in [k for k in self._inflight if match(k)]: self._inflight.pop(k, None) return len(dropped) access_cache = AccessCache() def invalidate(token_id: Optional[str] = None, guild_id: Optional[str] = None) -> int: return access_cache.invalidate(token_id=token_id, guild_id=guild_id) # --------------------------------------------------------------------------- # Token inspection / membership (moved from discord_leases, server-26#204) # --------------------------------------------------------------------------- async def list_guild_ids(token: str) -> RestResult: """Every server this bot is in, following ``after=`` pages (the old code read one page of 200 and silently dropped the rest). On success ``data`` is the list of guild id strings; any failed page makes the whole result that page's failure, never a partial list.""" out: list[str] = [] after: Optional[str] = None for _ in range(MAX_GUILD_PAGES): params: dict = {"limit": GUILDS_PAGE} if after: params["after"] = after r = await request(token, "GET", "/users/@me/guilds", params=params) if not r.ok: return r page = r.data if isinstance(r.data, list) else [] ids = [str(g["id"]) for g in page if isinstance(g, dict) and g.get("id") is not None] out.extend(ids) if len(page) < GUILDS_PAGE or not ids: return RestResult(200, out) after = max(ids, key=int) logger.warning(f"Discord /users/@me/guilds: stopped after {MAX_GUILD_PAGES} pages") return RestResult(None, error=f"more than {MAX_GUILD_PAGES * GUILDS_PAGE} servers") async def inspect_token(token: str) -> dict: """Ask Discord who this token is and which servers it is in. Returns ``{bot_user_id, bot_name, application_id, guild_ids, valid, validated_at, check_error}``. Raises TokenRejected on a 401. Any other failure (Discord down, rate limit) is reported in ``check_error`` with ``valid``/``guild_ids`` = None — unknown, not bad: a token is never rejected because Discord had a bad minute. """ out: dict = {"valid": None, "guild_ids": None, "check_error": None, "validated_at": _now().isoformat()} me = await request(token, "GET", "/users/@me") if me.status == 401: raise TokenRejected("Discord rejected this bot token (401).") if not me.ok or not isinstance(me.data, dict): out["check_error"] = (me.error or "unexpected answer from Discord")[:200] return out out["bot_user_id"] = str(me.data.get("id")) out["bot_name"] = me.data.get("username") app = await request(token, "GET", "/oauth2/applications/@me") if app.ok and isinstance(app.data, dict) and app.data.get("id") is not None: out["application_id"] = str(app.data["id"]) guilds = await list_guild_ids(token) if guilds.status == 401: raise TokenRejected("Discord rejected this bot token (401).") if not guilds.ok: out["check_error"] = (guilds.error or "could not list servers")[:200] return out out["guild_ids"] = guilds.data out["valid"] = True return out async def refresh_membership(token_ids: list[str]) -> int: """Re-read guild membership for these tokens. Returns how many changed. Every token re-read has its cached snapshots dropped.""" changed = 0 for tok_id in token_ids: tok = await fstore.doc_get(TOKENS, tok_id) if not tok or not tok.get("token"): continue try: info = await inspect_token(tok["token"]) except TokenRejected as e: await fstore.doc_update(TOKENS, tok_id, {"valid": False, "check_error": str(e), "validated_at": _now().isoformat()}) invalidate(token_id=tok_id) changed += 1 continue updates = {k: v for k, v in info.items() if v is not None or k == "check_error"} if updates.get("guild_ids") != tok.get("guild_ids") or updates.get("valid") != tok.get("valid"): changed += 1 await fstore.doc_update(TOKENS, tok_id, updates) invalidate(token_id=tok_id) return changed async def leave_guild(guild_id: str, org_id: Optional[str]) -> list[dict]: """Make the radio bots DRB holds tokens for leave one Discord server (Discord REST ``DELETE /users/@me/guilds/{id}``, the bot's own token). Only platform-pool bots and ``org_id``'s own BYO bots — never another org's branded bot. Bots whose cached membership says they are not in the server are skipped; unknown membership is tried (404 = not a member). The DRB *server* bot (the discord-bot service) is not in this pool, so it stays in the server until someone kicks it. Returns one row per bot tried: ``{token_id, name, left, error}``. Never raises for a Discord failure.""" guild_id = str(guild_id) out: list[dict] = [] for tok in await fstore.collection_list(TOKENS): tok_id = tok.get("token_id") if not tok_id or not tok.get("token"): continue if tok.get("org_id") is not None and tok.get("org_id") != org_id: continue members = tok.get("guild_ids") if members is not None and guild_id not in members: continue row = {"token_id": tok_id, "name": tok.get("name") or tok_id, "left": False, "error": None} r = await request(tok["token"], "DELETE", f"/users/@me/guilds/{guild_id}") if r.status in (200, 204, 404): # 404: Unknown Guild — not a member row["left"] = True else: row["error"] = r.error or f"Discord answered {r.status}" if row["left"]: invalidate(token_id=tok_id, guild_id=guild_id) if members is not None: await fstore.doc_update(TOKENS, tok_id, {"guild_ids": [g for g in members if g != guild_id]}) out.append(row) return out # --------------------------------------------------------------------------- # Per-(token, guild) snapshot — what S3's permission check reads # --------------------------------------------------------------------------- @dataclass class GuildSnapshot: """What one bot can see of one server. ``status`` is OK, NOT_MEMBER, INVALID or UNKNOWN; ``guild``/``member``/``channels`` are set only on OK. ``guild`` is Discord's guild object (``owner_id``, ``roles``), ``member`` the bot's own member object (``roles``, ``communication_disabled_until``), ``channels`` the server's channel list.""" token_id: str guild_id: str status: str bot_user_id: Optional[str] = None guild: Optional[dict] = None member: Optional[dict] = None channels: Optional[list] = None error: Optional[str] = None fetched_at: float = field(default_factory=time.time) async def _mark_invalid(token_id: str) -> None: await fstore.doc_update(TOKENS, token_id, { "valid": False, "check_error": "Discord rejected this bot token (401).", "validated_at": _now().isoformat()}) async def _drop_membership(token_id: str, guild_id: str) -> None: tok = await fstore.doc_get(TOKENS, token_id) or {} members = tok.get("guild_ids") if isinstance(members, list) and guild_id in members: await fstore.doc_update(TOKENS, token_id, {"guild_ids": [g for g in members if g != guild_id]}) async def _confirm_membership(token_id: str, token: str, guild_id: str) -> Optional[bool]: """After a 403 Missing Access: one read of the bot's server list decides. True/False = member or not (``guild_ids`` written from the full list); None = still unknown.""" r = await list_guild_ids(token) if r.status == 401: await _mark_invalid(token_id) return None if not r.ok: return None await fstore.doc_update(TOKENS, token_id, {"guild_ids": r.data, "validated_at": _now().isoformat()}) return guild_id in r.data async def _classify(snap: GuildSnapshot, r: RestResult, token: str) -> Optional[GuildSnapshot]: """None when ``r`` is a success; otherwise the finished snapshot.""" if r.ok: return None snap.error = r.error if r.status == 401: await _mark_invalid(snap.token_id) snap.status = INVALID elif r.status == 404 and r.code in (UNKNOWN_GUILD, UNKNOWN_MEMBER): await _drop_membership(snap.token_id, snap.guild_id) snap.status = NOT_MEMBER elif r.status == 403 and r.code == MISSING_ACCESS: member = await _confirm_membership(snap.token_id, token, snap.guild_id) snap.status = NOT_MEMBER if member is False else UNKNOWN else: snap.status = UNKNOWN return snap async def _load_snapshot(token_doc: dict, guild_id: str) -> GuildSnapshot: token_id = str(token_doc.get("token_id")) snap = GuildSnapshot(token_id=token_id, guild_id=guild_id, status=UNKNOWN, bot_user_id=token_doc.get("bot_user_id")) token = token_doc.get("token") if not token: snap.error = "no token stored" return snap if token_doc.get("valid") is False: snap.status, snap.error = INVALID, "token marked invalid" return snap if not snap.bot_user_id: me = await request(token, "GET", "/users/@me") done = await _classify(snap, me, token) if done: return done snap.bot_user_id = str((me.data or {}).get("id") or "") or None if not snap.bot_user_id: snap.error = "Discord did not say who this bot is" return snap await fstore.doc_update(TOKENS, token_id, {"bot_user_id": snap.bot_user_id}) g = await request(token, "GET", f"/guilds/{guild_id}") done = await _classify(snap, g, token) if done: return done m, c = await asyncio.gather( request(token, "GET", f"/guilds/{guild_id}/members/{snap.bot_user_id}"), request(token, "GET", f"/guilds/{guild_id}/channels"), ) for r in (m, c): done = await _classify(snap, r, token) if done: return done if not isinstance(g.data, dict) or not isinstance(m.data, dict) or not isinstance(c.data, list): snap.error = "unexpected answer from Discord" return snap snap.guild, snap.member, snap.channels = g.data, m.data, c.data snap.status, snap.error = OK, None return snap async def guild_snapshot(token_doc: dict, guild_id: str, *, refresh: bool = False) -> GuildSnapshot: """Cached snapshot of one server as one bot sees it: three GETs (``/guilds/{g}``, ``/guilds/{g}/members/{bot}`` — no privileged intent needed — and ``/guilds/{g}/channels``). ``refresh=True`` skips the cache (a concurrent load is still shared). Never raises for a Discord failure.""" guild_id = str(guild_id) token_id = str(token_doc.get("token_id")) key = (token_id, guild_id, "snapshot") async def _load() -> GuildSnapshot: # Fail open on anything else too (a Firestore write inside the # classification, a malformed body): the caller gets UNKNOWN, never # an exception, and a shielded load never dies unretrieved. try: return await _load_snapshot(token_doc, guild_id) except Exception as e: logger.warning(f"Discord snapshot for bot {token_id} in {guild_id} failed: {type(e).__name__}") return GuildSnapshot(token_id=token_id, guild_id=guild_id, status=UNKNOWN, bot_user_id=token_doc.get("bot_user_id"), error=f"check failed ({type(e).__name__})") return await access_cache.get_or_load(key, _load, lambda s: ttl_for(s.status), refresh=refresh) # --------------------------------------------------------------------------- # Voice-channel directory from a radio bot's view (server-26#225 S5) # --------------------------------------------------------------------------- DIRECTORY_MAX_CHANNELS = 200 # same cap as PUT /discord/guilds/{id}/directory _DIRECTORY_NAME_MAX = 100 VOICE_CHANNEL = 2 CATEGORY_CHANNEL = 4 def _int_or(value: Any, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): return default def directory_from_snapshot(snap: Optional[GuildSnapshot]) -> Optional[dict]: """A server's name and voice channels as one radio bot's snapshot sees them, in the shape the DRB server bot pushes (``PUT /discord/guilds/{id}/directory``): ``{name, voice_channels: [{channel_id, name, category, position}]}``, in Discord's sidebar order (channels with no category first, then each category by position; channels by position, then id). None unless the snapshot is OK. Every voice channel in the list is kept, including any this bot can't View, exactly like the server bot's gateway list: whether a bot may join is the permission check's answer (discord_streams.evaluate_access), and the picker shows it per channel. Stage channels are not voice channels here either.""" if snap is None or snap.status != OK or not isinstance(snap.channels, list): return None chans = [c for c in snap.channels if isinstance(c, dict) and c.get("id") is not None] cats = {str(c["id"]): c for c in chans if c.get("type") == CATEGORY_CHANNEL} def parent(c: dict) -> Optional[dict]: return cats.get(str(c.get("parent_id") or "")) def order(c: dict) -> tuple: p = parent(c) top = (0, 0, 0) if p is None else (1, _int_or(p.get("position")), _int_or(p.get("id"))) return (*top, _int_or(c.get("position")), _int_or(c.get("id"))) out: list[dict] = [] for c in sorted((c for c in chans if c.get("type") == VOICE_CHANNEL), key=order)[:DIRECTORY_MAX_CHANNELS]: p = parent(c) category = str(p.get("name") or "")[:_DIRECTORY_NAME_MAX] if p else "" out.append({ "channel_id": str(c["id"]), "name": str(c.get("name") or "")[:_DIRECTORY_NAME_MAX], "category": category or None, "position": len(out), }) name = snap.guild.get("name") if isinstance(snap.guild, dict) else None return {"name": str(name)[:_DIRECTORY_NAME_MAX] if name else None, "voice_channels": out} @dataclass class DirectoryResult: """``guild_directory``'s answer. ``status``: OK (``name`` and ``voice_channels`` from ``token_id``'s snapshot), NOT_MEMBER (every bot asked turned out not to be in the server, or its token is invalid) or UNKNOWN (no usable answer — keep what is stored).""" guild_id: str status: str name: Optional[str] = None voice_channels: Optional[list] = None token_id: Optional[str] = None fetched_at: Optional[float] = None async def guild_directory(token_docs: list[dict], guild_id: str, *, refresh: bool = False, max_bots: int = 3) -> DirectoryResult: """The voice-channel list of one server, read from the first of ``token_docs`` (radio bots believed to be in it, best first) whose cached snapshot is OK. It is the same snapshot the permission check reads, so listing channels costs no extra Discord request. At most ``max_bots`` bots are asked, one after another. Never raises for a Discord failure. The caller decides which server and which bots: this never looks a server up by itself, so it can't discover one (a server DRB doesn't already know is never listed from a radio bot's membership).""" guild_id = str(guild_id) statuses: list[str] = [] for doc in token_docs[:max(1, max_bots)]: snap = await guild_snapshot(doc, guild_id, refresh=refresh) statuses.append(snap.status) d = directory_from_snapshot(snap) if d is not None: return DirectoryResult(guild_id=guild_id, status=OK, name=d["name"], voice_channels=d["voice_channels"], token_id=snap.token_id, fetched_at=snap.fetched_at) if statuses and all(s in (NOT_MEMBER, INVALID) for s in statuses): return DirectoryResult(guild_id=guild_id, status=NOT_MEMBER) return DirectoryResult(guild_id=guild_id, status=UNKNOWN)