c2-core
- discord_streams/discord_leases wired in: POST /discord/streams,
POST /discord/guilds/{id}/stop, GET /discord/guilds/{id}/overview,
GET /discord/streams/{id}, PUT /discord/guilds/{id}/link (admin),
GET /discord/node/{node}/tokens/{id} (node key, lease holder only).
- Retained nodes/{id}/discord/desired (token ids, never token strings),
nodes/+/discord/actual subscribed and handled; per-node publish lock keeps
the retained list the newest; failed publishes retried by the reconciler.
- Tokens freed only after the node acks logout (or offline past
discord_lease_offline_release_seconds) - sweeper and C2 startup no longer
release leases.
- /nodes/{id}/command discord_join/leave map onto streams; CommandPayload
keeps preferred_token_id. /tokens: admin-only list, validate on add against
Discord, refresh, flush via transactional force release.
- Guild -> org via discord_guilds, unlinked guilds fall back to
DISCORD_UNLINKED_GUILD_ORG_ID (default "founding").
- tests/fake_firestore.py: optimistic-transaction fake; 27 lease/race tests.
discord bot
- /join, /leave, /status, autocomplete scoped to the calling guild; /leave
never touches another server; Manage Channels default permission;
/join edits its reply with the node's real result; /help copy fixed.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
185 lines
7.6 KiB
Python
185 lines
7.6 KiB
Python
"""
|
|
Discord radio channels — the HTTP face of internal/discord_streams.py.
|
|
|
|
Every bot-facing route is scoped to ONE Discord server (``guild_id`` in the
|
|
path or body) and acts for the org that server is linked to
|
|
(server-26#203): /leave in server A can no longer touch server B, and the
|
|
system/node/bot lists a server sees are its own org's.
|
|
|
|
Auth:
|
|
* bot routes — the Discord bot's service key, or a platform admin.
|
|
* PUT /discord/guilds/{id}/link — platform admin only.
|
|
* GET /discord/node/{node_id}/tokens/{token_id} — that node's own API key;
|
|
the token string is handed only to the node currently holding its lease.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from pydantic import BaseModel
|
|
|
|
from app.config import settings
|
|
from app.internal import discord_leases as leases
|
|
from app.internal import discord_streams as streams
|
|
from app.internal import firestore as fstore
|
|
from app.internal.auth import require_admin_token, require_service_key_or_admin
|
|
from app.internal.logger import logger
|
|
|
|
router = APIRouter(prefix="/discord", tags=["discord"])
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def _error(e: streams.StreamError) -> JSONResponse:
|
|
return JSONResponse(status_code=e.status,
|
|
content={"detail": e.message, "code": e.code, "invite": e.invite})
|
|
|
|
|
|
async def _org_or_403(guild_id: str) -> str:
|
|
org_id = await streams.resolve_guild_org(guild_id)
|
|
if not org_id:
|
|
raise HTTPException(403, streams.ERROR_TEXT["no_org"])
|
|
return org_id
|
|
|
|
|
|
def _public_stream(s: dict) -> dict:
|
|
keep = ("stream_id", "guild_id", "channel_id", "channel_name", "system_id", "system_name", "mode",
|
|
"desired", "node_id", "token_id", "actual", "stop_reason", "updated_at", "delivered", "reused")
|
|
return {k: s.get(k) for k in keep if k in s}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bot routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/guilds/{guild_id}/overview")
|
|
async def guild_overview(guild_id: str, _: dict = Depends(require_service_key_or_admin)):
|
|
"""Everything one Discord server may see: its org's systems and nodes, its
|
|
own streams, and which radio bots could serve it. No token strings."""
|
|
org_id = await streams.resolve_guild_org(guild_id)
|
|
if not org_id:
|
|
return {"guild_id": guild_id, "org_id": None, "linked": False,
|
|
"systems": [], "nodes": [], "streams": [], "bots": []}
|
|
link = await fstore.doc_get(streams.GUILDS, guild_id)
|
|
systems = await fstore.collection_list("systems", org_id=org_id)
|
|
nodes = await fstore.collection_list("nodes", org_id=org_id)
|
|
gstreams = [s for s in await fstore.collection_list(streams.STREAMS, guild_id=guild_id)
|
|
if s.get("desired") == "on"]
|
|
bots = []
|
|
for t in await fstore.collection_list(leases.TOKENS):
|
|
if t.get("org_id") not in (None, org_id) or t.get("valid") is False:
|
|
continue
|
|
lease = t.get("lease") or {}
|
|
member = t.get("guild_ids")
|
|
bots.append({
|
|
"token_id": t.get("token_id"),
|
|
"name": t.get("name"),
|
|
"busy_here": any(leases.guild_of(s) == guild_id for s in lease.get("streams") or []),
|
|
"member": None if member is None else guild_id in member,
|
|
"invite_url": leases.invite_url(t.get("application_id"), guild_id),
|
|
})
|
|
return {
|
|
"guild_id": guild_id,
|
|
"org_id": org_id,
|
|
"linked": bool(link and link.get("org_id")),
|
|
"systems": [{"system_id": s.get("system_id"), "name": s.get("name")} for s in systems],
|
|
"nodes": [{"node_id": n.get("node_id"), "name": n.get("name"), "status": n.get("status"),
|
|
"system_id": streams._effective_system(n)} for n in nodes],
|
|
"streams": [_public_stream(s) for s in gstreams],
|
|
"bots": bots,
|
|
}
|
|
|
|
|
|
class StreamStart(BaseModel):
|
|
guild_id: str
|
|
channel_id: str
|
|
system_id: str
|
|
mode: str = "adhoc"
|
|
preferred_token_id: Optional[str] = None
|
|
channel_name: Optional[str] = None
|
|
guild_name: Optional[str] = None
|
|
created_by: Optional[str] = None
|
|
|
|
|
|
@router.post("/streams")
|
|
async def start_stream(body: StreamStart, _: dict = Depends(require_service_key_or_admin)):
|
|
org_id = await _org_or_403(body.guild_id)
|
|
system = await fstore.doc_get_cached("systems", body.system_id)
|
|
if not system or system.get("org_id") not in (None, org_id):
|
|
raise HTTPException(404, "System not found.")
|
|
try:
|
|
doc = await streams.start_stream(
|
|
org_id=org_id, guild_id=body.guild_id, channel_id=body.channel_id, system_id=body.system_id,
|
|
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,
|
|
)
|
|
except streams.StreamError as e:
|
|
return _error(e)
|
|
return _public_stream(doc)
|
|
|
|
|
|
@router.get("/streams/{stream_id}")
|
|
async def get_stream(stream_id: str, _: dict = Depends(require_service_key_or_admin)):
|
|
doc = await fstore.doc_get(streams.STREAMS, stream_id)
|
|
if not doc:
|
|
raise HTTPException(404, "Stream not found.")
|
|
return _public_stream(doc)
|
|
|
|
|
|
class GuildStop(BaseModel):
|
|
channel_id: Optional[str] = None
|
|
reason: str = "stopped"
|
|
|
|
|
|
@router.post("/guilds/{guild_id}/stop")
|
|
async def stop_guild(guild_id: str, body: GuildStop, _: dict = Depends(require_service_key_or_admin)):
|
|
stopped = await streams.stop_guild(guild_id, body.reason[:60], channel_id=body.channel_id)
|
|
return {"stopped": stopped}
|
|
|
|
|
|
class GuildLink(BaseModel):
|
|
org_id: str
|
|
|
|
|
|
@router.put("/guilds/{guild_id}/link")
|
|
async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(require_admin_token)):
|
|
org = await fstore.doc_get("organizations", body.org_id)
|
|
if not org:
|
|
raise HTTPException(404, "Organization not found.")
|
|
await fstore.doc_set(streams.GUILDS, guild_id, {
|
|
"guild_id": guild_id, "org_id": body.org_id,
|
|
"linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
return {"ok": True, "guild_id": guild_id, "org_id": body.org_id}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Node route — token pickup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/node/{node_id}/tokens/{token_id}")
|
|
async def node_token(
|
|
node_id: str,
|
|
token_id: str,
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer),
|
|
):
|
|
"""The bot token string, for the node that holds its lease and nobody else."""
|
|
if not credentials:
|
|
raise HTTPException(401, "Missing authorization")
|
|
key_doc = await fstore.doc_get("node_keys", node_id)
|
|
stored = (key_doc or {}).get("api_key") or ""
|
|
if not stored or not secrets.compare_digest(stored, credentials.credentials):
|
|
raise HTTPException(401, "Invalid node API key")
|
|
tok = await fstore.doc_get(leases.TOKENS, token_id)
|
|
lease = (tok or {}).get("lease") or {}
|
|
if not tok or lease.get("node_id") != node_id or lease.get("releasing"):
|
|
logger.warning(f"Node {node_id} asked for token {token_id} it does not hold a lease on")
|
|
raise HTTPException(403, "This node does not hold a lease on that token.")
|
|
return {"token_id": token_id, "token": tok["token"], "lease_id": lease.get("lease_id"),
|
|
"name": tok.get("name")}
|