From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- src/voice/mod.rs | 404 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 404 insertions(+) create mode 100644 src/voice/mod.rs (limited to 'src/voice/mod.rs') diff --git a/src/voice/mod.rs b/src/voice/mod.rs new file mode 100644 index 0000000..0d5b89e --- /dev/null +++ b/src/voice/mod.rs @@ -0,0 +1,404 @@ +use std::io::Cursor; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration; + +use serenity::all::{ChannelId, GuildId}; +use songbird::input::{AsyncAdapterStream, AsyncReadOnlySource, RawAdapter}; +use songbird::{Call, Songbird}; +use symphonia::core::audio::sample::Sample; +use symphonia::core::codecs::audio::AudioDecoderOptions; +use symphonia::core::formats::probe::Hint; +use symphonia::core::formats::{FormatOptions, TrackType}; +use symphonia::core::io::MediaSourceStream; +use symphonia::core::meta::MetadataOptions; +use tokio::sync::Mutex; +use tracing::{info, warn}; + +use crate::error::{Error, Result}; +use crate::lang; + +#[derive(Clone)] +pub struct VoiceManager { + pub songbird: Arc, +} + +impl VoiceManager { + pub async fn new(ctx: &serenity::client::Context) -> Self { + Self { + songbird: songbird::get(ctx) + .await + .expect("Songbird not registered via register_songbird()"), + } + } + + pub async fn connect( + &self, + guild_id: GuildId, + channel_id: ChannelId, + ) -> Result>> { + let mut last_err = None; + for attempt in 1..=3 { + match self.songbird.join(guild_id, channel_id).await { + Ok(call) => { + tokio::time::sleep(Duration::from_millis(500)).await; + return Ok(call); + } + Err(e) => { + warn!( + "Voice join attempt {attempt}/3 failed for guild {}: {e}", + guild_id + ); + if let songbird::error::JoinError::Driver(ref driver_err) = e { + warn!("Driver error details: {driver_err:?}"); + } + last_err = Some(e); + tokio::time::sleep(Duration::from_secs(1)).await; + } + } + } + Err(Error::SongbirdJoin(last_err.unwrap())) + } + + pub async fn disconnect(&self, guild_id: GuildId) -> Result<()> { + self.songbird + .remove(guild_id) + .await + .map_err(|e| Error::Custom(lang::fmt_voice_remove_failed(&e)))?; + Ok(()) + } + + pub fn get_call(&self, guild_id: GuildId) -> Option>> { + self.songbird.get(guild_id) + } +} + +// ── Audio decoding ───────────────────────────────────────────────────── + +/// Decode an in-memory audio buffer into raw f32 stereo PCM for songbird. +/// Supports mp3, m4a/aac, flac, ogg, wav, opus via symphonia auto-detection. +pub fn decode_audio(data: &[u8]) -> Result { + let mss = MediaSourceStream::new(Box::new(Cursor::new(data.to_vec())), Default::default()); + decode_stream(mss) +} + +fn decode_stream(mss: MediaSourceStream) -> Result { + let hint = Hint::new(); + let fmt_opts = FormatOptions::default(); + let meta_opts = MetadataOptions::default(); + + let mut format = symphonia::default::get_probe() + .probe(&hint, mss, fmt_opts, meta_opts) + .map_err(|e| Error::Custom(lang::fmt_voice_probe_failed(&e)))?; + + let track = format + .default_track(TrackType::Audio) + .ok_or_else(|| Error::Custom(lang::L_VOICE_NO_TRACK.into()))?; + let track_id = track.id; + + let codec_params = track + .codec_params + .as_ref() + .and_then(|p| p.audio()) + .ok_or_else(|| Error::Custom(lang::L_VOICE_NO_CODEC.into()))?; + + let sample_rate = codec_params.sample_rate.unwrap_or(48000); + let channels = codec_params + .channels + .clone() + .map(|c| c.count() as u32) + .unwrap_or(2); + + let dec_opts = AudioDecoderOptions::default(); + let mut decoder = symphonia::default::get_codecs() + .make_audio_decoder(codec_params, &dec_opts) + .map_err(|e| Error::Custom(lang::fmt_voice_decoder_failed(&e)))?; + + let mut pcm_f32: Vec = Vec::new(); + + loop { + match format.next_packet() { + Ok(Some(packet)) => { + if packet.track_id != track_id { + continue; + } + if let Ok(buf) = decoder.decode(&packet) { + let n = buf.samples_interleaved(); + let start = pcm_f32.len(); + pcm_f32.resize(start + n, f32::MID); + buf.copy_to_slice_interleaved(&mut pcm_f32[start..]); + } + } + Ok(None) => break, + Err(_) => break, + } + } + + let pcm_bytes: Vec = pcm_f32.iter().flat_map(|&s| s.to_le_bytes()).collect(); + let cursor = Cursor::new(pcm_bytes); + let adapter = RawAdapter::new(cursor, sample_rate, channels); + + Ok(adapter.into()) +} + +// ── yt-dlp helper ────────────────────────────────────────────────────── + +/// Download audio via yt-dlp, collect stdout into memory, return the raw bytes. +/// Works for both finite media and live streams (yt-dlp handles the difference). +pub async fn ytdlp_download(url: &str) -> Result> { + let output = tokio::process::Command::new("yt-dlp") + .args(["-f", "bestaudio", "-o", "-", "--no-playlist", url]) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .output() + .await + .map_err(|e| Error::Custom(lang::fmt_voice_ytdlp_launch(&e)))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + // Keep the full yt-dlp trace in the logs for debugging — the user + // only ever sees the short, translated `fmt_ytdlp_stderr` message. + warn!(url = %url, "yt-dlp failed:\n{}", stderr.trim_end()); + return Err(Error::Custom(lang::fmt_ytdlp_stderr(&stderr))); + } + + Ok(output.stdout) +} + +// ── Compression helpers ─────────────────────────────────────────────── + +/// Download a URL into memory via reqwest. +pub async fn download_url(client: &reqwest::Client, url: &str) -> Result> { + let resp = client + .get(url) + .send() + .await + .map_err(|e| Error::Custom(format!("HTTP GET failed for {url}: {e}")))?; + let bytes = resp + .bytes() + .await + .map_err(|e| Error::Custom(format!("Failed to read body for {url}: {e}")))?; + Ok(bytes.to_vec()) +} + +/// DEBUG: bypass encode/decode — just pipe through ffmpeg as PCM. +/// If this plays, the streaming pipeline works; bug is in Opus roundtrip. +pub async fn degrade_audio(input: &[u8]) -> Result { + use std::process::Stdio; + + let mut child = tokio::process::Command::new("ffmpeg") + .args([ + "-y", "-i", "pipe:0", "-vn", "-f", "f32le", "-ac", "2", "-ar", "8000", "-b:a", "500", + "pipe:1", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn() + .map_err(|e| Error::Custom(format!("ffmpeg passthrough spawn: {e}")))?; + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child + .stdout + .take() + .ok_or_else(|| Error::Custom("stdout not captured".into()))?; + + let owned_input = input.to_vec(); + tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + let _ = stdin.write_all(&owned_input).await; + }); + + let ffmpeg_child = FfmpegChild { + _child: child, + stdout, + }; + + let async_source = AsyncReadOnlySource::new(ffmpeg_child); + let adapter = AsyncAdapterStream::new(Box::new(async_source), 64 * 1024); + let raw = RawAdapter::new(adapter, 48000, 2); + Ok(raw.into()) +} + +// ── HTTP streaming via songbird's built-in HttpRequest (legacy) ───── + +/// Create a songbird `Input` that streams audio directly from an HTTP(S) URL. +/// Works for both finite files and infinite Icecast/Shoutcast live streams. +/// +/// Superseded by `ffmpeg_stream_input` for the `/play` command (which needs +/// HLS/playlist/Icecast-metadata handling that symphonia can't provide), but +/// retained for other callers that want lightweight symphonia-based streaming. +#[allow(dead_code)] +pub fn http_stream_input(client: reqwest::Client, url: &str) -> songbird::input::Input { + info!(%url, "streaming via HttpRequest"); + songbird::input::HttpRequest::new(client, url.to_string()).into() +} + +// ── Live streaming via ffmpeg subprocess ───────────────────────────── + +/// Adapter holding an ffmpeg child process and its stdout pipe. +/// +/// Implements `AsyncRead` by delegating to `ChildStdout`. The `Child` is kept +/// alive for the lifetime of the stream and killed on drop via +/// `kill_on_drop(true)`, so stopping playback terminates ffmpeg. +/// +/// `Sync`: the `Child` handle is only touched on `Drop` (a single, one-shot +/// operation); `ChildStdout` is only read by the single `AsyncAdapterSink` +/// task spawned inside `AsyncAdapterStream`. No field is accessed concurrently. +struct FfmpegChild { + _child: tokio::process::Child, + stdout: tokio::process::ChildStdout, +} + +impl tokio::io::AsyncRead for FfmpegChild { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.stdout).poll_read(cx, buf) + } +} + +/// Stream an HTTP(S) URL live via an ffmpeg subprocess, producing a songbird +/// `Input` of raw f32 PCM at 48 kHz stereo — NO full download. +/// +/// ffmpeg handles every streaming scenario the Python bot relied on it for: +/// Icecast/Shoutcast radio (incl. inline metadata), HLS `.m3u8` (segment +/// fetching), `.m3u`/`.pls` playlists, and direct media files. The +/// `-reconnect` flags let ffmpeg recover from dropped connections. +/// +/// Byte path: `ffmpeg stdout` → `AsyncReadOnlySource` (unseekable async +/// media source) → `AsyncAdapterStream` (async→sync ring buffer) → +/// `RawAdapter` (prepends `SbirdRaw` f32 header) → songbird `Input`. +pub async fn ffmpeg_stream_input(url: &str) -> Result { + use std::process::Stdio; + + let mut child = tokio::process::Command::new("ffmpeg") + .args([ + "-reconnect", + "1", + "-reconnect_streamed", + "1", + "-reconnect_delay_max", + "5", + "-i", + url, + "-vn", + "-f", + "f32le", + "-ar", + "48000", + "-ac", + "2", + "pipe:1", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .map_err(|e| Error::Custom(lang::fmt_voice_ffmpeg_launch(&e)))?; + + let stdout = child + .stdout + .take() + .ok_or_else(|| Error::Custom("ffmpeg stdout was not captured".into()))?; + + // Drain stderr in background so ffmpeg doesn't block; log first 4 KiB on drop. + if let Some(stderr) = child.stderr.take() { + let url_owned = url.to_string(); + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + let mut buf = Vec::with_capacity(4096); + let mut reader = tokio::io::BufReader::new(stderr); + let _ = reader.read_to_end(&mut buf).await; + let text = String::from_utf8_lossy(&buf); + if !text.is_empty() { + warn!(url = %url_owned, "ffmpeg stderr: {}", text.trim_end()); + } + }); + } + + let ffmpeg_child = FfmpegChild { + _child: child, + stdout, + }; + + let async_source = AsyncReadOnlySource::new(ffmpeg_child); + let adapter = AsyncAdapterStream::new(Box::new(async_source), 64 * 1024); + let raw = RawAdapter::new(adapter, 48000, 2); + Ok(raw.into()) +} + +/// Smoke-test an HTTP stream URL: creates a raw reqwest GET, reads the first +/// chunk directly from the async bytes stream (bypasses AsyncAdapterStream). +/// Logs whether data arrives — isolates the hang to songbird's sync adapter. +#[allow(dead_code)] +pub async fn smoke_test_stream(client: reqwest::Client, url: String) { + use futures_util::StreamExt; + + info!(%url, "smoke_test: sending GET"); + let resp = match client.get(url.as_str()).send().await { + Ok(r) => r, + Err(e) => { + warn!(%url, error = %e, "smoke_test: GET FAILED"); + return; + } + }; + info!(%url, status = %resp.status(), "smoke_test: response received"); + + let mut stream = resp.bytes_stream(); + info!(%url, "smoke_test: awaiting first chunk"); + match stream.next().await { + Some(Ok(chunk)) => { + info!(%url, len = chunk.len(), "smoke_test: first chunk arrived"); + } + Some(Err(e)) => { + warn!(%url, error = %e, "smoke_test: chunk error"); + } + None => { + info!(%url, "smoke_test: stream ended immediately (0 bytes)"); + } + } +} + +/// Diagnostic: fetch headers and full body for a URL, logging key metrics. +/// Helps debug symphonia probe EOF issues (Content-Length vs actual bytes). +/// +/// **Warning**: hangs forever on infinite streams (Icecast). Finite URLs only. +#[allow(dead_code)] +pub async fn debug_fetch(client: &reqwest::Client, url: &str) { + match client.get(url).send().await { + Ok(resp) => { + let status = resp.status(); + let headers = resp.headers().clone(); + let content_length = headers + .get("content-length") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()); + let content_type = headers.get("content-type").and_then(|v| v.to_str().ok()); + + info!( + %url, %status, + ?content_length, ?content_type, + "HTTP response headers" + ); + + match resp.bytes().await { + Ok(body) => { + let actual_len = body.len() as u64; + let reported_len = content_length.unwrap_or(0); + info!( + %url, reported = reported_len, actual = actual_len, + delta = actual_len as i64 - reported_len as i64, + "buffered .bytes(): reported vs actual" + ); + } + Err(e) => warn!(%url, error = %e, ".bytes() failed"), + } + } + Err(e) => warn!(%url, error = %e, "HTTP GET failed"), + } +} -- cgit v1.2.3