Logan CusanoandClaude Opus 5.5
419d7c4d8c
- Rewritten as RadioManager reconciling C2's retained
nodes/{id}/discord/desired (token ids) and publishing
nodes/{id}/discord/actual {applied_seq, sessions, streams}.
- One gateway session per leased token, one voice connection per
(token, guild); a token no stream needs is logged OUT (client.close()),
which is what lets C2 free it safely (item 1).
- Watchdog is a separate reconcile loop; recovering a dead session can no
longer cancel it (item 2). Presence kept across rejoins.
- No message-content intent and no @bot leave/join mention controls (item 3).
- Local dashboard: /api/discord/join returns 409 with an explanation instead
of a TypeError 500; /api/discord/leave pauses locally; GET /api/discord
shows state (item 4). Join/leave results are reported to C2 (item 5).
- Token strings fetched over HTTPS from C2 (lease holder only), memory only.
- MQTT-loss fail-safe: logs every bot out after
DISCORD_MQTT_LOSS_STOP_SECONDS (300, below C2's 600s lease grace).
- Legacy discord_join/discord_leave MQTT commands are ignored.
- Audio path unchanged: FFmpeg -f pulse on settings.pulse_source per
connection, never Icecast.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
485 lines
20 KiB
Python
485 lines
20 KiB
Python
"""
|
|
Discord radio streams on the edge, driven by desired state (node-26#25,
|
|
server-26#203/#204).
|
|
|
|
C2 retains ``nodes/{id}/discord/desired`` = the FULL list of streams this node
|
|
should carry: ``{seq, streams: [{stream_id, guild_id, channel_id, token_id,
|
|
system_name, ...}]}``. This module reconciles toward that list and publishes
|
|
``nodes/{id}/discord/actual`` = ``{applied_seq, sessions: [token_id],
|
|
streams: {stream_id: {state, error, code}}}`` after every change.
|
|
|
|
Rules this enforces by construction:
|
|
|
|
* One gateway session per leased token; a token no stream needs any more is
|
|
logged OUT (``client.close()``), not just disconnected from voice. C2 frees
|
|
the token only after an ``actual`` with ``applied_seq`` >= the drop and the
|
|
token absent from ``sessions`` — so one token is never logged in on two
|
|
nodes (node-26#25 item 1).
|
|
* One voice connection per (token, guild); one token can serve many guilds.
|
|
* The watchdog is a separate reconcile loop that never closes itself: closing
|
|
a session cannot cancel it (node-26#25 item 2).
|
|
* No message-content intent, no ``@bot leave/join`` mention controls — voice
|
|
is controlled only through C2 (node-26#25 item 3).
|
|
* MQTT-loss fail-safe: if the broker has been unreachable for
|
|
``discord_mqtt_loss_stop_seconds`` every radio bot is logged out, because C2
|
|
returns this node's tokens to the pool after its own (longer) grace.
|
|
|
|
Bot token strings never travel over MQTT: they are fetched over HTTPS from C2
|
|
(``GET /discord/node/{node}/tokens/{id}``, node API key), which hands a token
|
|
only to the node holding its lease, and are kept in memory only.
|
|
|
|
Audio: each voice connection plays the PulseAudio monitor (``-f pulse`` on
|
|
``settings.pulse_source``) — the accuracy path, never Icecast.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Awaitable, Callable, Optional
|
|
|
|
import discord
|
|
|
|
from app.config import settings
|
|
from app.internal import pulse
|
|
from app.internal.logger import logger
|
|
|
|
BOT_READY_TIMEOUT = 15 # seconds to wait for a Discord login to become ready
|
|
WATCHDOG_INTERVAL = 15 # seconds between reconcile passes
|
|
STREAM_RETRY_DELAY = 5 # seconds to back off before re-arming an audio source
|
|
REPORT_EVERY_TICKS = 4 # unconditional actual-state report every ~60s
|
|
_VOICE_TYPES = (discord.ChannelType.voice, discord.ChannelType.stage_voice)
|
|
|
|
|
|
@dataclass
|
|
class _Session:
|
|
token_id: str
|
|
client: object
|
|
task: Optional[asyncio.Task] = None
|
|
presence: Optional[str] = None
|
|
|
|
def alive(self) -> bool:
|
|
return self.task is not None and not self.task.done() and not self.client.is_closed()
|
|
|
|
|
|
@dataclass
|
|
class _Voice:
|
|
stream_id: str
|
|
token_id: str
|
|
guild_id: int
|
|
channel_id: int
|
|
vc: Optional[object] = None
|
|
state: str = "connecting"
|
|
error: Optional[str] = None
|
|
code: Optional[str] = None
|
|
|
|
|
|
def _default_client_factory():
|
|
intents = discord.Intents.none()
|
|
intents.guilds = True
|
|
intents.voice_states = True
|
|
return discord.Client(intents=intents)
|
|
|
|
|
|
def _default_audio_factory():
|
|
# before_options land ahead of -i: ffmpeg -f pulse -i drb_sink.monitor …
|
|
source = discord.FFmpegPCMAudio(settings.pulse_source, before_options="-f pulse")
|
|
return discord.PCMVolumeTransformer(source, volume=1.0)
|
|
|
|
|
|
async def _default_token_fetcher(token_id: str) -> Optional[str]:
|
|
import httpx
|
|
from app.internal import credentials
|
|
api_key = credentials.get_api_key()
|
|
if not settings.c2_url or not api_key:
|
|
logger.error("Cannot fetch a radio-bot token: C2_URL or the node API key is missing.")
|
|
return None
|
|
url = f"{settings.c2_url.rstrip('/')}/discord/node/{settings.node_id}/tokens/{token_id}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
r = await client.get(url, headers={"Authorization": f"Bearer {api_key}"})
|
|
if r.status_code != 200:
|
|
logger.error(f"C2 refused radio-bot token {token_id}: HTTP {r.status_code}")
|
|
return None
|
|
return r.json().get("token")
|
|
except Exception as e:
|
|
logger.error(f"Fetching radio-bot token {token_id} failed: {e}")
|
|
return None
|
|
|
|
|
|
async def _default_publisher(payload: dict) -> None:
|
|
from app.internal.mqtt_manager import mqtt_manager
|
|
mqtt_manager.publish_discord_actual(payload)
|
|
|
|
|
|
def _default_mqtt_up() -> bool:
|
|
from app.internal.mqtt_manager import mqtt_manager
|
|
return mqtt_manager.is_connected
|
|
|
|
|
|
class RadioManager:
|
|
def __init__(
|
|
self,
|
|
client_factory: Callable[[], object] = _default_client_factory,
|
|
token_fetcher: Callable[[str], Awaitable[Optional[str]]] = _default_token_fetcher,
|
|
publisher: Callable[[dict], Awaitable[None]] = _default_publisher,
|
|
audio_factory: Callable[[], object] = _default_audio_factory,
|
|
mqtt_up: Callable[[], bool] = _default_mqtt_up,
|
|
ready_timeout: float = BOT_READY_TIMEOUT,
|
|
):
|
|
self._client_factory = client_factory
|
|
self._fetch_token = token_fetcher
|
|
self._publish = publisher
|
|
self._audio_factory = audio_factory
|
|
self._mqtt_up = mqtt_up
|
|
self._ready_timeout = ready_timeout
|
|
|
|
self._desired: dict[str, dict] = {}
|
|
self._applied_seq: Optional[int] = None
|
|
self._sessions: dict[str, _Session] = {}
|
|
self._voices: dict[str, _Voice] = {}
|
|
self._session_errors: dict[str, str] = {}
|
|
self._tokens: dict[str, str] = {} # token_id -> string, memory only
|
|
self._lock = asyncio.Lock()
|
|
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
|
self._watchdog: Optional[asyncio.Task] = None
|
|
self._mqtt_down_since: Optional[float] = None
|
|
self._failsafe = False
|
|
self._paused = False
|
|
self._ticks = 0
|
|
|
|
# ------------------------------------------------------------------
|
|
# Inputs
|
|
# ------------------------------------------------------------------
|
|
|
|
async def apply_desired(self, payload: dict) -> None:
|
|
"""Handle a (retained) desired-state message from C2."""
|
|
if not isinstance(payload, dict):
|
|
return
|
|
seq = payload.get("seq")
|
|
seq = seq if isinstance(seq, int) and not isinstance(seq, bool) else None
|
|
if seq is not None and self._applied_seq is not None and seq < self._applied_seq:
|
|
logger.warning(f"Ignoring stale Discord desired state seq {seq} (applied {self._applied_seq})")
|
|
return
|
|
wanted = {}
|
|
for s in payload.get("streams") or []:
|
|
try:
|
|
wanted[str(s["stream_id"])] = {
|
|
"stream_id": str(s["stream_id"]), "guild_id": int(s["guild_id"]),
|
|
"channel_id": int(s["channel_id"]), "token_id": str(s["token_id"]),
|
|
"system_name": s.get("system_name"),
|
|
}
|
|
except (KeyError, TypeError, ValueError):
|
|
logger.error(f"Malformed Discord desired stream entry ignored: {s!r}")
|
|
self._ensure_watchdog()
|
|
self._desired = wanted
|
|
self._paused = False
|
|
await self.reconcile()
|
|
if seq is not None:
|
|
self._applied_seq = seq
|
|
await self.publish_actual()
|
|
|
|
async def pause_locally(self) -> None:
|
|
"""Local-dashboard stop: log every bot out until C2 sends a new list."""
|
|
self._paused = True
|
|
await self.reconcile()
|
|
await self.publish_actual()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Reconcile
|
|
# ------------------------------------------------------------------
|
|
|
|
def _target(self) -> dict[str, dict]:
|
|
return {} if (self._paused or self._failsafe) else self._desired
|
|
|
|
async def reconcile(self) -> None:
|
|
async with self._lock:
|
|
target = self._target()
|
|
# 1) voice connections no longer wanted (or re-pointed)
|
|
for sid, v in list(self._voices.items()):
|
|
d = target.get(sid)
|
|
if not d or (d["token_id"], d["guild_id"], d["channel_id"]) != (v.token_id, v.guild_id, v.channel_id):
|
|
await self._disconnect(v)
|
|
del self._voices[sid]
|
|
# 2) gateway sessions no stream needs: LOG OUT, not just leave voice
|
|
needed = {d["token_id"] for d in target.values()}
|
|
for tid in list(self._sessions):
|
|
if tid not in needed:
|
|
await self._close_session(tid)
|
|
for tid in list(self._tokens):
|
|
if tid not in needed:
|
|
self._tokens.pop(tid, None)
|
|
# 3) sessions + voice for everything wanted
|
|
for sid, d in sorted(target.items()):
|
|
v = self._voices.get(sid)
|
|
if v is None:
|
|
v = self._voices[sid] = _Voice(sid, d["token_id"], d["guild_id"], d["channel_id"])
|
|
sess = await self._ensure_session(d["token_id"])
|
|
if sess is None:
|
|
v.state, v.code = "error", "login_failed"
|
|
v.error = self._session_errors.get(d["token_id"], "The radio bot could not log in to Discord.")
|
|
continue
|
|
await self._ensure_voice(sess, v, d)
|
|
|
|
async def _ensure_session(self, token_id: str) -> Optional[_Session]:
|
|
sess = self._sessions.get(token_id)
|
|
if sess and sess.alive():
|
|
return sess
|
|
if sess:
|
|
await self._close_session(token_id)
|
|
token = self._tokens.get(token_id) or await self._fetch_token(token_id)
|
|
if not token:
|
|
self._session_errors[token_id] = "DRB would not hand this node the radio bot's token."
|
|
return None
|
|
client = self._client_factory()
|
|
sess = _Session(token_id, client)
|
|
ready = asyncio.Event()
|
|
|
|
async def on_ready():
|
|
ready.set()
|
|
client.event(on_ready)
|
|
sess.task = asyncio.create_task(client.start(token))
|
|
waiter = asyncio.create_task(ready.wait())
|
|
done, _ = await asyncio.wait({waiter, sess.task}, timeout=self._ready_timeout,
|
|
return_when=asyncio.FIRST_COMPLETED)
|
|
if waiter not in done:
|
|
waiter.cancel()
|
|
err = sess.task.exception() if sess.task.done() and not sess.task.cancelled() else None
|
|
msg = f"{type(err).__name__}: {err}" if err else f"not ready after {self._ready_timeout:.0f}s"
|
|
logger.error(f"Radio bot {token_id} failed to log in: {msg}")
|
|
self._session_errors[token_id] = f"The radio bot could not log in to Discord ({msg[:120]})."
|
|
await self._shutdown_client(sess)
|
|
if isinstance(err, discord.LoginFailure):
|
|
self._tokens.pop(token_id, None)
|
|
return None
|
|
self._tokens[token_id] = token
|
|
self._sessions[token_id] = sess
|
|
self._session_errors.pop(token_id, None)
|
|
logger.info(f"Radio bot {token_id} logged in as {getattr(client, 'user', None)}")
|
|
return sess
|
|
|
|
async def _ensure_voice(self, sess: _Session, v: _Voice, d: dict) -> None:
|
|
client = sess.client
|
|
guild = client.get_guild(v.guild_id)
|
|
if guild is None:
|
|
v.state, v.code = "error", "not_in_guild"
|
|
v.error = "This radio bot is not a member of that Discord server — invite it first."
|
|
return
|
|
channel = guild.get_channel(v.channel_id)
|
|
if channel is None or getattr(channel, "type", None) not in _VOICE_TYPES:
|
|
v.state, v.code = "error", "bad_channel"
|
|
v.error = "That channel is not a voice channel the radio bot can see."
|
|
return
|
|
vc = v.vc
|
|
if vc is not None and vc.is_connected() and getattr(vc.channel, "id", None) == v.channel_id:
|
|
if not vc.is_playing():
|
|
self._play(v)
|
|
v.state, v.error, v.code = "connected", None, None
|
|
await self._set_presence(sess, d.get("system_name"))
|
|
return
|
|
v.state = "connecting"
|
|
try:
|
|
existing = guild.voice_client
|
|
if existing is not None and existing.is_connected():
|
|
await existing.move_to(channel)
|
|
v.vc = existing
|
|
else:
|
|
if existing is not None:
|
|
await existing.disconnect(force=True)
|
|
v.vc = await channel.connect()
|
|
except Exception as e:
|
|
v.vc = None
|
|
v.state, v.code = "error", "connect_failed"
|
|
v.error = f"Could not connect to the voice channel ({type(e).__name__}: {str(e)[:120]})."
|
|
logger.error(f"Stream {v.stream_id}: voice connect failed: {e}")
|
|
return
|
|
# Bounded wait for the shared PulseAudio socket (op25 container creates it).
|
|
if not await pulse.wait_until_ready():
|
|
v.state, v.code = "error", "no_audio"
|
|
v.error = "The node's audio path (PulseAudio) is not up."
|
|
return
|
|
self._play(v, pulse_checked=True)
|
|
v.state, v.error, v.code = "connected", None, None
|
|
await self._set_presence(sess, d.get("system_name"))
|
|
logger.info(f"Stream {v.stream_id} connected (bot {v.token_id})")
|
|
|
|
async def _set_presence(self, sess: _Session, name: Optional[str]) -> None:
|
|
if not name or sess.presence == name:
|
|
return
|
|
try:
|
|
await sess.client.change_presence(
|
|
activity=discord.Activity(type=discord.ActivityType.listening, name=name))
|
|
sess.presence = name
|
|
except Exception as e:
|
|
logger.warning(f"Presence update failed: {e}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Audio
|
|
# ------------------------------------------------------------------
|
|
|
|
def _play(self, v: _Voice, pulse_checked: bool = False) -> None:
|
|
vc = v.vc
|
|
if vc is None or not vc.is_connected():
|
|
return
|
|
if not pulse_checked and not pulse.is_ready():
|
|
logger.error(f"PulseAudio socket {pulse.socket_path()} missing — retrying audio in {STREAM_RETRY_DELAY}s.")
|
|
self._schedule_replay(v.stream_id)
|
|
return
|
|
try:
|
|
vc.play(self._audio_factory(), after=lambda err, sid=v.stream_id: self._on_audio_end(sid, err))
|
|
except Exception as e:
|
|
logger.error(f"Stream {v.stream_id}: could not start audio: {e}")
|
|
self._schedule_replay(v.stream_id)
|
|
|
|
def _on_audio_end(self, stream_id: str, error) -> None:
|
|
"""discord.py calls this from its player thread."""
|
|
if error:
|
|
logger.error(f"Stream {stream_id} audio ended with error: {error}")
|
|
self._schedule_replay(stream_id, STREAM_RETRY_DELAY if error else 0.5)
|
|
|
|
def _schedule_replay(self, stream_id: str, delay: float = STREAM_RETRY_DELAY) -> None:
|
|
loop = self._loop
|
|
if loop is None or loop.is_closed():
|
|
return
|
|
|
|
async def _later():
|
|
await asyncio.sleep(delay)
|
|
v = self._voices.get(stream_id)
|
|
if v and v.vc and v.vc.is_connected() and not v.vc.is_playing():
|
|
self._play(v)
|
|
|
|
loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_later()))
|
|
|
|
# ------------------------------------------------------------------
|
|
# Teardown
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _disconnect(self, v: _Voice) -> None:
|
|
vc, v.vc = v.vc, None
|
|
if vc is None:
|
|
return
|
|
try:
|
|
if vc.is_playing():
|
|
vc.stop()
|
|
await vc.disconnect(force=True)
|
|
except Exception as e:
|
|
logger.warning(f"Stream {v.stream_id}: voice disconnect error: {e}")
|
|
|
|
async def _close_session(self, token_id: str) -> None:
|
|
sess = self._sessions.pop(token_id, None)
|
|
if sess is None:
|
|
return
|
|
for v in self._voices.values():
|
|
if v.token_id == token_id:
|
|
await self._disconnect(v)
|
|
await self._shutdown_client(sess)
|
|
logger.info(f"Radio bot {token_id} logged out")
|
|
|
|
async def _shutdown_client(self, sess: _Session) -> None:
|
|
try:
|
|
await sess.client.close()
|
|
except Exception as e:
|
|
logger.warning(f"Radio bot {sess.token_id} close error: {e}")
|
|
if sess.task and not sess.task.done():
|
|
sess.task.cancel()
|
|
try:
|
|
await sess.task
|
|
except BaseException:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
"""Node shutdown: log every radio bot out."""
|
|
if self._watchdog:
|
|
self._watchdog.cancel()
|
|
self._watchdog = None
|
|
async with self._lock:
|
|
for v in list(self._voices.values()):
|
|
await self._disconnect(v)
|
|
self._voices.clear()
|
|
for tid in list(self._sessions):
|
|
await self._close_session(tid)
|
|
self._tokens.clear()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Watchdog + MQTT-loss fail-safe
|
|
# ------------------------------------------------------------------
|
|
|
|
def _ensure_watchdog(self) -> None:
|
|
self._loop = asyncio.get_running_loop()
|
|
if self._watchdog is None or self._watchdog.done():
|
|
self._watchdog = asyncio.create_task(self._watchdog_loop())
|
|
|
|
async def _watchdog_loop(self) -> None:
|
|
while True:
|
|
await asyncio.sleep(WATCHDOG_INTERVAL)
|
|
try:
|
|
await self.tick()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Discord watchdog error: {e}")
|
|
|
|
async def tick(self, now: Optional[float] = None) -> None:
|
|
"""One watchdog pass: fail-safe check, then reconcile + report if needed."""
|
|
now = time.monotonic() if now is None else now
|
|
before = self.snapshot()
|
|
if self._mqtt_up():
|
|
self._mqtt_down_since = None
|
|
if self._failsafe:
|
|
logger.info("MQTT back — resuming Discord radio streams.")
|
|
self._failsafe = False
|
|
else:
|
|
if self._mqtt_down_since is None:
|
|
self._mqtt_down_since = now
|
|
elif (not self._failsafe and self._sessions
|
|
and now - self._mqtt_down_since >= settings.discord_mqtt_loss_stop_seconds):
|
|
logger.warning("MQTT unreachable past the fail-safe — logging every radio bot out "
|
|
"(C2 will hand these tokens to other nodes).")
|
|
self._failsafe = True
|
|
await self.reconcile()
|
|
self._ticks += 1
|
|
# On change, and every few passes regardless: a periodic report is what
|
|
# lets C2 settle a release whose first ack was lost.
|
|
if self.snapshot() != before or self._ticks % REPORT_EVERY_TICKS == 0:
|
|
await self.publish_actual()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Reporting
|
|
# ------------------------------------------------------------------
|
|
|
|
def snapshot(self) -> dict:
|
|
streams = {}
|
|
for sid in self._desired:
|
|
v = self._voices.get(sid)
|
|
if self._paused:
|
|
streams[sid] = {"state": "paused_locally", "error": None, "code": None}
|
|
elif v is None:
|
|
streams[sid] = {"state": "connecting", "error": None, "code": None}
|
|
else:
|
|
streams[sid] = {"state": v.state, "error": v.error, "code": v.code}
|
|
return {
|
|
"applied_seq": self._applied_seq,
|
|
"sessions": sorted(tid for tid, s in self._sessions.items() if s.alive()),
|
|
"streams": streams,
|
|
}
|
|
|
|
async def publish_actual(self) -> None:
|
|
try:
|
|
await self._publish(self.snapshot())
|
|
except Exception as e:
|
|
logger.warning(f"Publishing Discord actual state failed: {e}")
|
|
|
|
# metadata_watcher hooks — audio plays continuously, nothing to do per call.
|
|
def start_stream(self):
|
|
pass
|
|
|
|
def stop_stream(self):
|
|
pass
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return any(v.vc is not None and v.vc.is_connected() for v in self._voices.values())
|
|
|
|
|
|
radio_bot = RadioManager()
|