OPEN THE RUST GATES

Signed-off-by: salmonik <raisersoft@proton.me>

1 files changed, 404 insertions, 0 deletions

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<Songbird>,
+}
+
+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<Arc<Mutex<Call>>> {
+        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<Arc<Mutex<Call>>> {
+        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<songbird::input::Input> {
+    let mss = MediaSourceStream::new(Box::new(Cursor::new(data.to_vec())), Default::default());
+    decode_stream(mss)
+}
+
+fn decode_stream(mss: MediaSourceStream) -> Result<songbird::input::Input> {
+    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<f32> = 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<u8> = 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<Vec<u8>> {
+    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<Vec<u8>> {
+    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<songbird::input::Input> {
+    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<std::io::Result<()>> {
+        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<songbird::input::Input> {
+    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::<u64>().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"),
+    }
+}