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) -> Result<Self> {
        let songbird = songbird::get(ctx)
            .await
            .ok_or_else(|| Error::Custom(
                "Songbird not registered via register_songbird()".into(),
            ))?;
        Ok(Self { 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);

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

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