path: root/src/voice/mod.rs

use std::collections::HashMap;
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::copyparty::redact_url;
use crate::error::{Error, Result};
use crate::lang;
use crate::recording::{Recorder, RecorderEventHandler};

/// Sample rate (Hz) songbird decodes incoming voice into — drives ring sizing
/// and the ffmpeg `s16le` encode. Must match the `DecodeConfig` set in `main.rs`.
const RECORDING_SAMPLE_RATE: u32 = 48000;

#[derive(Clone)]
pub struct VoiceManager {
    pub songbird: Arc<Songbird>,
    /// Live recorder per guild, if any.
    ///
    /// Entries exist only while the bot is connected (and recording) in that
    /// guild, so the ~86 MB ring only exists while actually recording. Keying
    /// by guild prevents one guild's connection from replacing or stopping
    /// another guild's buffer.
    pub recorders: Arc<Mutex<HashMap<GuildId, Arc<Recorder>>>>,
    pub recording_enabled: bool,
    pub recording_window_seconds: u64,
}

impl VoiceManager {
    pub async fn new(
        ctx: &serenity::client::Context,
        recording_enabled: bool,
        recording_window_seconds: u64,
    ) -> Result<Self> {
        let songbird = songbird::get(ctx)
            .await
            .ok_or_else(|| Error::Custom(
                "Songbird not registered via register_songbird()".into(),
            ))?;
        Ok(Self {
            songbird,
            recorders: Arc::new(Mutex::new(HashMap::new())),
            recording_enabled,
            recording_window_seconds,
        })
    }

    pub fn recording_enabled(&self) -> bool {
        self.recording_enabled
    }

    /// Construct a `Recorder` for `guild_id`, register its `VoiceTick` handler
    /// on `call`, and store the strong `Arc`. Replacing an existing entry drops
    /// the previous ring; its handler's `Weak` then no-ops.
    pub async fn start_recording(&self, guild_id: GuildId, call: &Arc<Mutex<Call>>) {
        if !self.recording_enabled {
            return;
        }
        let recorder = Arc::new(Recorder::new(
            RECORDING_SAMPLE_RATE,
            self.recording_window_seconds,
        ));
        let handler = RecorderEventHandler::new(&recorder);
        {
            let mut call_lock = call.lock().await;
            call_lock.add_global_event(
                songbird::Event::Core(songbird::CoreEvent::VoiceTick),
                handler,
            );
        }
        self.recorders
            .lock()
            .await
            .insert(guild_id, recorder);
        info!(
            guild = %guild_id,
            window_secs = self.recording_window_seconds,
            "recording: started rolling buffer for this voice connection"
        );
    }

    /// Drop this guild's live recorder (frees the ring immediately). The
    /// songbird `Call` may still exist briefly; the handler's `Weak` then no-ops.
    pub async fn stop_recording(&self, guild_id: GuildId) {
        let prev = self.recorders.lock().await.remove(&guild_id);
        if prev.is_some() {
            info!(guild = %guild_id, "recording: stopped, ring buffer freed");
        }
    }

    /// Clone of this guild's live recorder, if any — used by `/record`.
    pub async fn get_recorder(&self, guild_id: GuildId) -> Option<Arc<Recorder>> {
        self.recorders.lock().await.get(&guild_id).cloned()
    }

    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;
                    // Reliable setup point: we hold the `Call` right after a
                    // successful join, so register the VoiceTick handler here.
                    // (The bot's own VoiceStateUpdate also reconciles, but may
                    // race songbird's Call creation — this never misses.)
                    if self.recording_enabled {
                        self.start_recording(guild_id, &call).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);

                    // Stale voice state: remove the guild from songbird and
                    // wait a few seconds so Discord's voice gateway can fully
                    // process the disconnect before we retry.
                    let _ = self.songbird.remove(guild_id).await;
                    tokio::time::sleep(Duration::from_secs(3)).await;
                }
            }
        }
        Err(Error::SongbirdJoin(last_err.unwrap()))
    }

    pub async fn disconnect(&self, guild_id: GuildId) -> Result<()> {
        // Free the ring before tearing down the Call — the handler's `Weak`
        // will then no-op any in-flight ticks until songbird drops it.
        self.stop_recording(guild_id).await;
        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([
            "--ignore-config",
            "-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 = %redact_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)
        .timeout(Duration::from_secs(300))
        .send()
        .await
        .map_err(|e| Error::Custom(format!("HTTP GET failed for {}: {e}", redact_url(url))))?;
    let bytes = resp
        .bytes()
        .await
        .map_err(|e| Error::Custom(format!("Failed to read body for {}: {e}", redact_url(url))))?;
    Ok(bytes.to_vec())
}

// ── 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)
    }
}

/// Decode a local audio file via ffmpeg to raw f32 PCM at 48 kHz stereo.
/// No reconnect flags (those are for HTTP streams only).
pub async fn ffmpeg_decode_file(path: &str) -> Result<songbird::input::Input> {
    use std::process::Stdio;

    let mut child = tokio::process::Command::new("ffmpeg")
        .args([
            "-i",
            path,
            "-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(format!("ffmpeg decode spawn: {e}")))?;

    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| Error::Custom("ffmpeg stdout not captured".into()))?;

    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())
}

/// 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 = %redact_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"),
    }
}