Merge pull request 'Discord: channel list and presence without the main bot (#225 S5, fixes #251)' (#262) from feat/225-s5-channels into main
This commit was merged in pull request #262.
This commit is contained in:
@@ -655,3 +655,100 @@ async def guild_snapshot(token_doc: dict, guild_id: str, *, refresh: bool = Fals
|
||||
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)
|
||||
|
||||
@@ -53,6 +53,7 @@ A retained MQTT message therefore never stores a bot credential on the broker.
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional
|
||||
|
||||
@@ -352,14 +353,39 @@ async def check_bots(tokens: dict[str, dict], token_ids: list[str], guild_id: st
|
||||
for t in token_ids}
|
||||
|
||||
|
||||
# server-26#251: an interactive start (a person pressed Start / ran /join)
|
||||
# re-reads a bot's snapshot that says "denied" once it is this old, so a
|
||||
# permission fixed in Discord a moment ago is seen without waiting out the
|
||||
# cache. The reconciler never does (plan finding 4: no REST loop).
|
||||
RECHECK_DENIED_AFTER_SECONDS = 10.0
|
||||
|
||||
|
||||
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]:
|
||||
preferred_token_id: Optional[str], node_ids: set,
|
||||
recheck_denied: bool = False) -> 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)
|
||||
snaps = await gather_snapshots(tokens, ids, guild_id)
|
||||
|
||||
def verdict(t: str) -> dict:
|
||||
return evaluate_access(snaps[t], channel_id, bot=tokens[t].get("name") or t) if snaps.get(t) \
|
||||
else _unanswered()
|
||||
|
||||
out = {t: verdict(t) for t in ids}
|
||||
if recheck_denied:
|
||||
now = time.time()
|
||||
stale = [t for t in ids if out[t]["status"] == leases.PERM_DENIED
|
||||
and now - (getattr(snaps.get(t), "fetched_at", None) or now) >= RECHECK_DENIED_AFTER_SECONDS]
|
||||
if stale:
|
||||
fresh = await gather_snapshots(tokens, stale, guild_id, refresh=True)
|
||||
for t, snap in fresh.items():
|
||||
if snap is not None: # not answered in time: keep the cached verdict
|
||||
snaps[t] = snap
|
||||
out[t] = verdict(t)
|
||||
return out
|
||||
|
||||
|
||||
def _channel_ref(perms: Optional[dict], channel_name: Optional[str]) -> str:
|
||||
@@ -484,6 +510,7 @@ async def start_stream(
|
||||
preferred_token_id: Optional[str] = None,
|
||||
max_streams: Optional[int] = None,
|
||||
stream_id: Optional[str] = None,
|
||||
interactive: bool = False,
|
||||
) -> dict:
|
||||
"""Put ``system_id`` on a voice channel.
|
||||
|
||||
@@ -504,6 +531,12 @@ async def start_stream(
|
||||
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.
|
||||
|
||||
``interactive``: a person asked for this start (the routes pass True,
|
||||
the reconciler never does). A bot whose cached check says "denied" is
|
||||
then asked again if that answer is more than
|
||||
``RECHECK_DENIED_AFTER_SECONDS`` old (server-26#251), so "fix it in
|
||||
Discord, then try again" works on the next try.
|
||||
"""
|
||||
if mode not in ("persistent", "adhoc"):
|
||||
raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400)
|
||||
@@ -527,14 +560,14 @@ async def start_stream(
|
||||
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,
|
||||
existing=existing, interactive=interactive,
|
||||
)
|
||||
|
||||
|
||||
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:
|
||||
stream_id: Optional[str], existing: dict, interactive: bool = False) -> dict:
|
||||
already_on = False
|
||||
if stream_id:
|
||||
existing = await fstore.doc_get(STREAMS, stream_id) or existing # re-read under the lock
|
||||
@@ -612,7 +645,8 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
|
||||
|
||||
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)
|
||||
preferred_token_id=preferred_token_id, node_ids=node_ids,
|
||||
recheck_denied=interactive)
|
||||
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.",
|
||||
@@ -638,7 +672,7 @@ async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_i
|
||||
# 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)
|
||||
node_ids=node_ids, recheck_denied=interactive)
|
||||
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)
|
||||
|
||||
@@ -138,6 +138,7 @@ async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_
|
||||
mode=body.mode, channel_name=body.channel_name, guild_name=body.guild_name,
|
||||
created_by=body.created_by, preferred_token_id=body.preferred_token_id,
|
||||
max_streams=settings.discord_guild_max_streams, stream_id=body.stream_id,
|
||||
interactive=True, # /join, the admin page: re-read a stale "denied" (server-26#251)
|
||||
)
|
||||
except streams.StreamError as e:
|
||||
return _error(e)
|
||||
|
||||
@@ -6,15 +6,38 @@ The node page used to take raw guild/channel IDs. It now picks from:
|
||||
* the Discord servers the caller's org owns — ``discord_guilds`` docs that
|
||||
resolve to the node's org (an explicit link, or the unlinked-server
|
||||
fallback org, same rule the bot uses: streams.resolve_guild_org), and
|
||||
* their voice channels, from a directory the DRB server bot pushes
|
||||
(``PUT /discord/guilds/{id}/directory``). The bot already holds its
|
||||
gateway cache of every server it is in; C2 never calls Discord with a
|
||||
token on the browser's behalf and no token string ever reaches the browser.
|
||||
* their voice channels, from a directory kept on ``discord_guilds/{id}``.
|
||||
|
||||
Where the directory comes from (server-26#225 S5 — the main bot is optional):
|
||||
|
||||
* the DRB server bot, when it is in the server: it pushes its gateway view
|
||||
(``PUT /discord/guilds/{id}/directory``) on start, on changes and every
|
||||
15 minutes (``directory_source: "main_bot"``);
|
||||
* otherwise a radio bot that is in the server: C2 reads the channel list
|
||||
over Discord REST with that bot's token (discord_rest.guild_directory —
|
||||
the same cached snapshot the permission check reads, so no extra
|
||||
request) and stores it (``directory_source: "radio_bot"``,
|
||||
``radio_bot_present``). That happens when the picker is opened and the
|
||||
stored list is missing (inline, inside the permission-check time budget)
|
||||
or older than ``discord_directory_refresh_seconds`` (in the background).
|
||||
|
||||
So C2 does call Discord when a signed-in user opens the picker — with a
|
||||
radio bot's token, for a server already linked to the caller's org, through
|
||||
the rate-limited, cached client. No token string ever reaches the browser.
|
||||
|
||||
Which servers an org sees (``_org_guilds``):
|
||||
|
||||
* linked to the org: shown unless the server bot said it left AND no
|
||||
radio bot the org may use is in it (``bot_tokens.guild_ids``);
|
||||
* acting for the org only through the unlinked-server fallback: shown only
|
||||
while the server bot is there, exactly as before. A radio bot's
|
||||
membership never discovers a server, and REST never creates a
|
||||
``discord_guilds`` doc.
|
||||
|
||||
A pick becomes a ``discord_streams`` desired-state doc **pinned to this node**
|
||||
(streams.start_stream(node_id=...)), exactly what the old raw-ID path did, with
|
||||
the checks that path skipped: the server must belong to the node's org, the
|
||||
channel must be a voice channel the bot reported in that server, and the
|
||||
channel must be a voice channel in that server's directory, and the
|
||||
per-server and per-node stream caps apply.
|
||||
|
||||
Auth:
|
||||
@@ -29,6 +52,7 @@ Auth:
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
@@ -38,6 +62,7 @@ from pydantic import BaseModel, Field
|
||||
|
||||
from app.config import settings
|
||||
from app.internal import discord_leases as leases
|
||||
from app.internal import discord_rest
|
||||
from app.internal import discord_streams as streams
|
||||
from app.internal import firestore as fstore
|
||||
from app.internal.audit import write_audit
|
||||
@@ -54,6 +79,14 @@ router = APIRouter(prefix="/discord", tags=["discord"])
|
||||
|
||||
MAX_CHANNELS = 200
|
||||
_NAME_MAX = 100
|
||||
MAIN_BOT = "main_bot"
|
||||
RADIO_BOT = "radio_bot"
|
||||
# The server bot re-pushes every 15 min. A list it pushed more than three
|
||||
# pushes ago means it is down (it never said it left): a radio bot's view
|
||||
# takes over until it is back.
|
||||
MAIN_BOT_STALE_SECONDS = 45 * 60
|
||||
# How many member radio bots are asked, one after another, for the list.
|
||||
DIRECTORY_BOTS = 3
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
@@ -80,16 +113,20 @@ class GuildDirectory(BaseModel):
|
||||
@router.put("/guilds/{guild_id}/directory")
|
||||
async def put_guild_directory(guild_id: str, body: GuildDirectory, _: dict = Depends(require_service_key)):
|
||||
"""The server bot's view of one Discord server: its name and voice channels.
|
||||
Merged onto ``discord_guilds/{id}`` — never touches ``org_id`` (the link)."""
|
||||
Merged onto ``discord_guilds/{id}`` — never touches ``org_id`` (the link).
|
||||
Cached Discord snapshots of the server are dropped (a channel was added,
|
||||
removed or renamed), so the next permission check reads it fresh."""
|
||||
if not guild_id.isdigit():
|
||||
raise HTTPException(400, "guild_id must be a Discord snowflake.")
|
||||
data: dict = {"guild_id": guild_id, "bot_present": body.present, "directory_at": _now_iso()}
|
||||
if body.present:
|
||||
data["voice_channels"] = [c.model_dump() for c in body.voice_channels]
|
||||
data["directory_source"] = MAIN_BOT
|
||||
if body.name:
|
||||
# Same field the /drb link flow writes (feat/213-guild-link, PR #218).
|
||||
data["guild_name"] = body.name
|
||||
await fstore.doc_set(streams.GUILDS, guild_id, data)
|
||||
discord_rest.invalidate(guild_id=guild_id)
|
||||
return {"ok": True, "guild_id": guild_id, "voice_channels": len(body.voice_channels)}
|
||||
|
||||
|
||||
@@ -130,24 +167,156 @@ def _require_control(can: bool) -> None:
|
||||
raise HTTPException(403, "Only the organization owner, an admin, or this node's operator can do that.")
|
||||
|
||||
|
||||
async def _org_guilds(org_id: str) -> list[dict]:
|
||||
"""``discord_guilds`` docs this org may stream into, bot still present.
|
||||
async def _load_tokens() -> dict[str, dict]:
|
||||
return {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
|
||||
|
||||
|
||||
def _member_radio_bots(tokens: dict[str, dict], guild_id: str, org_id: Optional[str]) -> list[str]:
|
||||
"""Radio bots this org may use (pool, or the org's own) that Discord last
|
||||
said are in the server, best first. Unknown membership (``guild_ids``
|
||||
None) does not count: presence has to be positive."""
|
||||
out = [tid for tid, t in tokens.items()
|
||||
if t.get("token") and t.get("valid") is not False and t.get("org_id") in (None, org_id)
|
||||
and isinstance(t.get("guild_ids"), list) and guild_id in t["guild_ids"]]
|
||||
out.sort(key=lambda tid: leases._candidate_order(tid, tokens[tid], "", org_id, None))
|
||||
return out
|
||||
|
||||
|
||||
async def _org_guilds(org_id: str, tokens: Optional[dict] = None) -> list[dict]:
|
||||
"""``discord_guilds`` docs this org may stream into, with a bot there.
|
||||
Mirrors streams.resolve_guild_org: explicit link, else the unlinked-server
|
||||
fallback org — except a server an org explicitly unlinked (``unlinked_at``,
|
||||
PR #218), which gets no fallback."""
|
||||
PR #218), which gets no fallback.
|
||||
|
||||
A linked server is shown while the server bot is there or a radio bot the
|
||||
org may use is (server-26#225 S5: the main bot is optional). A server
|
||||
that acts for the org only through the fallback still needs the server
|
||||
bot: a radio bot's membership never discovers one."""
|
||||
fallback = settings.discord_unlinked_guild_org_id or None
|
||||
if tokens is None:
|
||||
tokens = await _load_tokens()
|
||||
out = []
|
||||
for g in await fstore.collection_list(streams.GUILDS):
|
||||
gid = str(g.get("guild_id") or "")
|
||||
if not gid or g.get("bot_present") is False:
|
||||
if not gid:
|
||||
continue
|
||||
owner = g.get("org_id") or (None if g.get("unlinked_at") else fallback)
|
||||
if owner and owner == org_id:
|
||||
out.append(g)
|
||||
if g.get("org_id"):
|
||||
if g["org_id"] != org_id:
|
||||
continue
|
||||
if g.get("bot_present") is False and not _member_radio_bots(tokens, gid, org_id):
|
||||
continue
|
||||
else:
|
||||
if g.get("unlinked_at") or not fallback or fallback != org_id or g.get("bot_present") is False:
|
||||
continue
|
||||
out.append(g)
|
||||
out.sort(key=lambda g: ((_guild_name(g) or "").lower(), g["guild_id"]))
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Directory from a radio bot (server-26#225 S5)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _age_seconds(ts) -> Optional[float]:
|
||||
when = streams._parse_ts(ts)
|
||||
return (datetime.now(timezone.utc) - when).total_seconds() if when else None
|
||||
|
||||
|
||||
def _directory_due(g: dict) -> Optional[str]:
|
||||
"""None when the stored list is fine; "missing" when there is none;
|
||||
"stale" when it is old. Only for a server linked to an org — a fallback
|
||||
server's list is the server bot's alone."""
|
||||
if not g.get("org_id"):
|
||||
return None
|
||||
if g.get("voice_channels") is None:
|
||||
return "missing"
|
||||
from_main = g.get("bot_present") is True and (g.get("directory_source") or MAIN_BOT) == MAIN_BOT
|
||||
limit = MAIN_BOT_STALE_SECONDS if from_main else settings.discord_directory_refresh_seconds
|
||||
age = _age_seconds(g.get("directory_at"))
|
||||
return "stale" if age is None or age > limit else None
|
||||
|
||||
|
||||
def _main_bot_list_fresh(g: dict) -> bool:
|
||||
"""The server bot is there and its own list is current: it wins."""
|
||||
return (g.get("bot_present") is True and (g.get("directory_source") or MAIN_BOT) == MAIN_BOT
|
||||
and _directory_due(g) is None)
|
||||
|
||||
|
||||
async def _refresh_directory(g: dict, org_id: str, tokens: dict[str, dict]) -> Optional[dict]:
|
||||
"""Read one linked server's voice channels from a member radio bot and
|
||||
store them. Returns the fields written (also merged into ``g``), or None
|
||||
when nothing was learned. Writes only onto the existing doc, and only
|
||||
while it is still linked to ``org_id``: never creates a doc, never
|
||||
touches ``org_id`` or the server bot's ``bot_present``."""
|
||||
gid = g["guild_id"]
|
||||
members = _member_radio_bots(tokens, gid, org_id)
|
||||
if not members:
|
||||
return None
|
||||
docs = [{**tokens[t], "token_id": t} for t in members]
|
||||
res = await discord_rest.guild_directory(docs, gid, max_bots=DIRECTORY_BOTS)
|
||||
now = _now_iso()
|
||||
if res.status == discord_rest.OK:
|
||||
updates: dict = {"voice_channels": res.voice_channels, "directory_at": now,
|
||||
"directory_source": RADIO_BOT, "radio_bot_present": True, "radio_bot_checked_at": now}
|
||||
if res.name:
|
||||
updates["guild_name"] = res.name
|
||||
elif res.status == discord_rest.NOT_MEMBER:
|
||||
updates = {"radio_bot_present": False, "radio_bot_checked_at": now}
|
||||
else:
|
||||
return None
|
||||
cur = await fstore.doc_get(streams.GUILDS, gid)
|
||||
if not cur or cur.get("org_id") != org_id:
|
||||
return None # unlinked / moved / removed meanwhile
|
||||
if _main_bot_list_fresh(cur):
|
||||
return None # the server bot pushed a fresh list meanwhile; it wins
|
||||
await fstore.doc_update(streams.GUILDS, gid, updates)
|
||||
g.update(updates)
|
||||
logger.info(f"Discord directory for {gid} from radio bot {res.token_id or '-'}: "
|
||||
f"{res.status}, {len(res.voice_channels or [])} voice channel(s)")
|
||||
return updates
|
||||
|
||||
|
||||
_directory_tasks: dict[str, asyncio.Task] = {}
|
||||
|
||||
|
||||
async def _refresh_quietly(g: dict, org_id: str, tokens: dict[str, dict]) -> None:
|
||||
try:
|
||||
await _refresh_directory(g, org_id, tokens)
|
||||
except Exception as e: # a background refresh must never surface anywhere
|
||||
logger.warning(f"Discord directory refresh for {g.get('guild_id')} failed: {type(e).__name__}")
|
||||
|
||||
|
||||
def _refresh_in_background(g: dict, org_id: str, tokens: dict[str, dict]) -> None:
|
||||
gid = g["guild_id"]
|
||||
running = _directory_tasks.get(gid)
|
||||
if running is not None and not running.done():
|
||||
return
|
||||
task = asyncio.create_task(_refresh_quietly(dict(g), org_id, tokens))
|
||||
_directory_tasks[gid] = task
|
||||
task.add_done_callback(lambda t, gid=gid: _directory_tasks.pop(gid, None)
|
||||
if _directory_tasks.get(gid) is t else None)
|
||||
|
||||
|
||||
async def _fill_directories(guilds: list[dict], org_id: str, tokens: dict[str, dict]) -> None:
|
||||
"""Missing lists are read now (all servers at once, inside the
|
||||
permission-check budget — an answer that comes later is cached and
|
||||
stored by the next look); stale ones are refreshed in the background and
|
||||
the stored list is shown meanwhile."""
|
||||
now: list[dict] = []
|
||||
for g in guilds:
|
||||
due = _directory_due(g)
|
||||
if not due or not _member_radio_bots(tokens, g["guild_id"], org_id):
|
||||
continue
|
||||
if due == "missing":
|
||||
now.append(g)
|
||||
else:
|
||||
_refresh_in_background(g, org_id, tokens)
|
||||
if not now:
|
||||
return
|
||||
tasks = [asyncio.create_task(_refresh_quietly(g, org_id, tokens)) for g in now]
|
||||
await asyncio.wait(tasks, timeout=settings.discord_perm_check_budget_seconds)
|
||||
|
||||
|
||||
def _bot_status(tokens: dict, *, node_id: str, guild_id: str, org_id: str) -> dict:
|
||||
res = leases.pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id)
|
||||
if res.ok:
|
||||
@@ -181,7 +350,10 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
|
||||
system_id = streams._effective_system(node)
|
||||
system = (await fstore.doc_get_cached("systems", system_id)) if system_id else None
|
||||
|
||||
guilds = await _org_guilds(org_id) if org_id else []
|
||||
tokens = await _load_tokens()
|
||||
guilds = await _org_guilds(org_id, tokens) if org_id else []
|
||||
if org_id:
|
||||
await _fill_directories(guilds, org_id, tokens)
|
||||
live = [s for s in await fstore.collection_list(streams.STREAMS, desired="on")
|
||||
if s.get("org_id") == org_id]
|
||||
nodes = {n.get("node_id"): n.get("name") for n in await fstore.collection_list("nodes", org_id=org_id)
|
||||
@@ -190,7 +362,6 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
|
||||
by_channel: dict[tuple, list] = {}
|
||||
for s in sorted(live, key=lambda s: (s.get("created_at") or "", s.get("stream_id") or "")):
|
||||
by_channel.setdefault((str(s.get("guild_id")), str(s.get("channel_id"))), []).append(s)
|
||||
tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
|
||||
token_names = {tid: t.get("name") or tid for tid, t in tokens.items()}
|
||||
|
||||
out_guilds = []
|
||||
@@ -216,6 +387,10 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to
|
||||
"linked": bool(g.get("org_id")),
|
||||
"synced": g.get("voice_channels") is not None,
|
||||
"synced_at": g.get("directory_at"),
|
||||
# Who listed the channels: "main_bot" (DRB server bot) or "radio_bot" (S5).
|
||||
"directory_source": (g.get("directory_source") or (MAIN_BOT if g.get("voice_channels") is not None
|
||||
else None)),
|
||||
"main_bot_present": g.get("bot_present") is True,
|
||||
"voice_channels": channels,
|
||||
"streams_on": sum(1 for s in live if s.get("guild_id") == gid),
|
||||
"bot": _bot_status(tokens, node_id=node_id, guild_id=gid, org_id=org_id),
|
||||
@@ -261,10 +436,24 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict =
|
||||
raise HTTPException(409, "This node is not listening to a system yet.")
|
||||
|
||||
guild_id, channel_id = str(body.guild_id), str(body.channel_id)
|
||||
guild = next((g for g in await _org_guilds(org_id) if g["guild_id"] == guild_id), None)
|
||||
tokens = await _load_tokens()
|
||||
guild = next((g for g in await _org_guilds(org_id, tokens) if g["guild_id"] == guild_id), None)
|
||||
if not guild:
|
||||
raise HTTPException(404, "That Discord server is not linked to this organization.")
|
||||
channel = next((c for c in guild.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
|
||||
|
||||
def _find(g: dict) -> Optional[dict]:
|
||||
return next((c for c in g.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
|
||||
|
||||
channel = _find(guild)
|
||||
if not channel and guild.get("org_id") and not _main_bot_list_fresh(guild):
|
||||
# No list yet, or a radio bot's list that may predate this channel:
|
||||
# read it once more (cached snapshot) before saying no.
|
||||
try:
|
||||
await asyncio.wait_for(_refresh_directory(guild, org_id, tokens),
|
||||
timeout=settings.discord_perm_check_budget_seconds)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
channel = _find(guild)
|
||||
if not channel:
|
||||
raise HTTPException(400, "That is not a voice channel DRB can see in this server. "
|
||||
"Refresh the list and pick again.")
|
||||
@@ -287,6 +476,7 @@ async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict =
|
||||
channel_name=channel.get("name"), guild_name=_guild_name(guild),
|
||||
created_by=f"web:{decoded.get('uid') or 'unknown'}",
|
||||
max_streams=settings.discord_guild_max_streams,
|
||||
interactive=True,
|
||||
)
|
||||
except streams.StreamError as e:
|
||||
return JSONResponse(status_code=e.status,
|
||||
|
||||
@@ -210,7 +210,7 @@ async def send_command(
|
||||
doc = await discord_streams.start_stream(
|
||||
org_id=node.get("org_id"), guild_id=cmd.guild_id, channel_id=cmd.channel_id,
|
||||
system_id=system_id, node_id=node_id, preferred_token_id=cmd.preferred_token_id,
|
||||
mode="adhoc",
|
||||
mode="adhoc", interactive=True,
|
||||
)
|
||||
except discord_streams.StreamError as e:
|
||||
raise HTTPException(e.status, e.message)
|
||||
|
||||
Reference in New Issue
Block a user