path: root/src/music.rs

//! Per-guild music queue and playback pipeline.
//!
//! Playback used to live entirely in `commands::music`, tied to a poise
//! `Context` (voice lookup, replies, ...). A queue needs to keep playing after
//! the slash command has returned, so the stateful part lives here instead:
//! the current track, the pending queue, and the voice/announce channels.
//!
//! The command layer resolves a user request into a [`TrackRequest`] (URL,
//! display name, source kind) and hands it to [`MusicState`]. Everything after
//! that — stopping/interrupting, preparing inputs, connecting, announcing, and
//! advancing on `TrackEvent::End` — is handled here without a `Context`.

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;

use poise::serenity_prelude as serenity;
use songbird::input::Input;
use songbird::tracks::TrackHandle;
use tokio::sync::Mutex;
use tracing::{error, info, warn};
use uuid::Uuid;

use crate::error::{Error, Result};
use crate::lang;
use crate::voice::{self, VoiceManager};

// ── Request types ─────────────────────────────────────────────────────

/// What kind of audio source are we dealing with?
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SourceKind {
    /// YouTube, SoundCloud, Twitch, Bandcamp — needs yt-dlp extraction.
    YtDlp,
    /// Any HTTP(S) URL streamed directly through ffmpeg.
    Stream,
}

/// Everything needed to (re)start one track, independent of the command
/// invocation that created it.
#[derive(Debug, Clone)]
pub struct TrackRequest {
    pub url: String,
    pub display_name: String,
    pub kind: SourceKind,
    pub compressed: bool,
    /// `(attachment_url, filename)` when the track came from an upload. Used to
    /// re-upload the file alongside the "playing" message.
    pub reupload: Option<(String, String)>,
    pub requester: Option<serenity::UserId>,
    /// Suppress the "playing" and failure announcements for this track.
    /// Cron audio uses this so scheduled playback stays silent.
    pub silent: bool,
}

/// Where an enqueued track goes relative to the current playback.
#[derive(Debug, Clone, Copy, PartialEq, Eq, poise::ChoiceParameter)]
pub enum QueuePosition {
    /// Interrupt the current track and play this right away. Pending tracks
    /// stay queued behind it.
    #[name = "now"]
    Now,
    /// Insert directly after the current track.
    #[name = "next"]
    Next,
    /// Append to the end of the queue (the command default).
    #[name = "last"]
    Last,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EnqueueOutcome {
    /// Playback started immediately (nothing was playing, or `Now` was used).
    Started,
    /// The track was put into the queue; `position` is 1-based.
    Queued { position: usize },
}

#[derive(Debug, Clone)]
pub struct QueueItem {
    pub display_name: String,
    pub requester: Option<serenity::UserId>,
}

#[derive(Debug, Clone, Default)]
pub struct QueueSnapshot {
    pub current: Option<QueueItem>,
    pub upcoming: Vec<QueueItem>,
}

// ── State ─────────────────────────────────────────────────────────────

struct CurrentTrack {
    /// Identifies this particular playback. Lets a late `End` event from a
    /// manually stopped track avoid advancing the queue twice.
    token: Uuid,
    handle: TrackHandle,
    request: TrackRequest,
}

struct GuildMusic {
    current: Option<CurrentTrack>,
    queue: VecDeque<TrackRequest>,
    /// Voice channel the bot is (or should be) connected to.
    voice: serenity::ChannelId,
    /// Text channel where auto-advance announcements are posted.
    announce: Option<serenity::ChannelId>,
    /// Bumped by interrupts (`now`, skip, stop). In-flight preparation for an
    /// older generation is discarded instead of starting over the new track.
    generation: u64,
}

/// Shared, per-guild music state. Stored in [`crate::Data`].
pub struct MusicState {
    guilds: Mutex<HashMap<u64, GuildMusic>>,
    /// One state lock per guild. Held only for state mutations; downloads,
    /// compression, and voice connects happen outside it so one slow track
    /// cannot stall every other `/music` command in the guild.
    locks: Mutex<HashMap<u64, Arc<Mutex<()>>>>,
    voice: VoiceManager,
    http: Arc<serenity::Http>,
    reqwest: reqwest::Client,
}

impl MusicState {
    pub fn new(
        voice: VoiceManager,
        http: Arc<serenity::Http>,
        reqwest: reqwest::Client,
    ) -> Arc<Self> {
        Arc::new(Self {
            guilds: Mutex::new(HashMap::new()),
            locks: Mutex::new(HashMap::new()),
            voice,
            http,
            reqwest,
        })
    }

    async fn guild_lock(&self, guild: u64) -> Arc<Mutex<()>> {
        let mut locks = self.locks.lock().await;
        locks
            .entry(guild)
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    }

    /// Add a track to the queue and start it if nothing is playing.
    pub async fn enqueue(
        self: &Arc<Self>,
        guild: u64,
        voice_channel: serenity::ChannelId,
        announce: Option<serenity::ChannelId>,
        request: TrackRequest,
        position: QueuePosition,
    ) -> Result<EnqueueOutcome> {
        let lock = self.guild_lock(guild).await;
        let (was_playing, queued_position, interrupted) = {
            let _transition = lock.lock().await;
            let mut map = self.guilds.lock().await;
            let gm = map.entry(guild).or_insert_with(|| GuildMusic {
                current: None,
                queue: VecDeque::new(),
                voice: voice_channel,
                announce,
                generation: 0,
            });
            gm.voice = voice_channel;
            gm.announce = announce;
            let was_playing = gm.current.is_some();

            match position {
                QueuePosition::Now => gm.queue.push_front(request),
                QueuePosition::Next => gm.queue.push_front(request),
                QueuePosition::Last => gm.queue.push_back(request),
            }
            let queued_position = match position {
                QueuePosition::Next => 1,
                QueuePosition::Last => gm.queue.len(),
                QueuePosition::Now => 1,
            };

            let interrupted = if was_playing && position == QueuePosition::Now {
                gm.generation += 1;
                gm.current.take()
            } else {
                None
            };
            (was_playing, queued_position, interrupted)
        };

        if let Some(current) = interrupted {
            let _ = current.handle.stop();
        }

        if was_playing && position != QueuePosition::Now {
            return Ok(EnqueueOutcome::Queued {
                position: queued_position,
            });
        }

        // The command layer posts the "now playing" message for direct
        // enqueues, so don't announce here as well. Preparation happens
        // outside the guild lock.
        let played = self.play_next(guild, false).await?;
        Ok(if played {
            EnqueueOutcome::Started
        } else {
            EnqueueOutcome::Queued { position: 1 }
        })
    }

    /// Stop the current track and immediately start the next queued one.
    /// Returns `true` when a track started, `false` when the queue was empty.
    pub async fn skip(self: &Arc<Self>, guild: u64) -> Result<bool> {
        let lock = self.guild_lock(guild).await;
        let (current, has_queue) = {
            let _transition = lock.lock().await;
            let mut map = self.guilds.lock().await;
            let Some(gm) = map.get_mut(&guild) else {
                return Ok(false);
            };
            gm.generation += 1;
            (gm.current.take(), !gm.queue.is_empty())
        };
        if let Some(current) = current {
            let _ = current.handle.stop();
        }
        if !has_queue {
            return Ok(false);
        }

        self.play_next(guild, true).await
    }

    /// Stop playback and drop the queue (does not disconnect from voice).
    pub async fn stop(self: &Arc<Self>, guild: u64) {
        let lock = self.guild_lock(guild).await;
        let current = {
            let _transition = lock.lock().await;
            let mut map = self.guilds.lock().await;
            map.remove(&guild).and_then(|gm| gm.current)
        };
        if let Some(current) = current {
            let _ = current.handle.stop();
        }
    }

    /// Play a track immediately, interrupting whatever is playing. Used by the
    /// cron scheduler (no poise context, and the target channel is a voice
    /// channel rather than a text channel).
    pub async fn play_now(
        self: &Arc<Self>,
        guild: u64,
        voice_channel: serenity::ChannelId,
        announce: Option<serenity::ChannelId>,
        request: TrackRequest,
    ) -> Result<()> {
        let lock = self.guild_lock(guild).await;
        let current = {
            let _transition = lock.lock().await;
            let mut map = self.guilds.lock().await;
            let gm = map.entry(guild).or_insert_with(|| GuildMusic {
                current: None,
                queue: VecDeque::new(),
                voice: voice_channel,
                announce,
                generation: 0,
            });
            gm.generation += 1;
            gm.voice = voice_channel;
            // Keep the last text channel when the caller has nowhere to
            // announce (cron's channel is a voice channel).
            if announce.is_some() {
                gm.announce = announce;
            }
            gm.queue.push_front(request);
            gm.current.take()
        };
        if let Some(current) = current {
            let _ = current.handle.stop();
        }

        if self.play_next(guild, true).await? {
            Ok(())
        } else {
            Err(Error::Custom(lang::L_MUSIC_QUEUE_EMPTY.into()))
        }
    }

    /// Current track followed by the pending queue, for `/music queue`.
    pub async fn snapshot(&self, guild: u64) -> QueueSnapshot {
        let map = self.guilds.lock().await;
        let Some(gm) = map.get(&guild) else {
            return QueueSnapshot::default();
        };
        QueueSnapshot {
            current: gm.current.as_ref().map(|current| QueueItem {
                display_name: current.request.display_name.clone(),
                requester: current.request.requester,
            }),
            upcoming: gm
                .queue
                .iter()
                .map(|req| QueueItem {
                    display_name: req.display_name.clone(),
                    requester: req.requester,
                })
                .collect(),
        }
    }

    /// Pop and play the next playable track.
    ///
    /// Input preparation and the voice connect happen outside the guild state
    /// lock, so a slow download cannot block other commands. A generation
    /// token makes an in-flight track abort if it was interrupted meanwhile.
    async fn play_next(self: &Arc<Self>, guild: u64, announce: bool) -> Result<bool> {
        let mut last_error: Option<Error> = None;
        loop {
            let (request, voice_channel, announce_channel, generation) = {
                let lock = self.guild_lock(guild).await;
                let _transition = lock.lock().await;
                let mut map = self.guilds.lock().await;
                let Some(gm) = map.get_mut(&guild) else {
                    return Ok(false);
                };
                let Some(request) = gm.queue.pop_front() else {
                    return match last_error {
                        Some(e) => Err(e),
                        None => Ok(false),
                    };
                };
                (request, gm.voice, gm.announce, gm.generation)
            };

            let input = match build_input(&request, &self.reqwest).await {
                Ok(input) => input,
                Err(e) => {
                    warn!(track = %request.display_name, error = %e, "failed to prepare queued track");
                    if !request.silent {
                        let message = lang::fmt_music_track_failed(&request.display_name, &e);
                        self.announce_failure(guild, message).await;
                    }
                    last_error = Some(e);
                    continue;
                }
            };

            let call = match self.voice.get_call(serenity::GuildId::new(guild)) {
                Some(call) => call,
                None => match self
                    .voice
                    .connect(serenity::GuildId::new(guild), voice_channel)
                    .await
                {
                    Ok(call) => call,
                    Err(e) => {
                        // Put the track back unless an interrupt superseded it.
                        let lock = self.guild_lock(guild).await;
                        let _transition = lock.lock().await;
                        let mut map = self.guilds.lock().await;
                        if let Some(gm) = map.get_mut(&guild)
                            && gm.generation == generation
                        {
                            gm.queue.push_front(request);
                        }
                        return Err(e);
                    }
                },
            };

            let (handle, token) = {
                let lock = self.guild_lock(guild).await;
                let _transition = lock.lock().await;
                let mut map = self.guilds.lock().await;
                let Some(gm) = map.get_mut(&guild) else {
                    return Ok(false);
                };
                if gm.generation != generation {
                    // Interrupted while preparing; the replacement request is
                    // already at the front of the queue.
                    continue;
                }
                if gm.current.is_some() {
                    // Another preparation won the race; let it handle this.
                    gm.queue.push_front(request);
                    return Ok(false);
                }
                let handle = {
                    let mut call = call.lock().await;
                    call.play_input(input)
                };
                let token = Uuid::new_v4();
                gm.current = Some(CurrentTrack {
                    token,
                    handle: handle.clone(),
                    request: request.clone(),
                });
                (handle, token)
            };

            let _ = handle.add_event(
                songbird::Event::Track(songbird::TrackEvent::End),
                TrackEndHandler {
                    state: self.clone(),
                    guild,
                    token,
                },
            );
            let _ = handle.add_event(
                songbird::Event::Track(songbird::TrackEvent::Error),
                TrackEndHandler {
                    state: self.clone(),
                    guild,
                    token,
                },
            );

            if announce && !request.silent {
                self.announce_playing(announce_channel, &request).await;
            }
            return Ok(true);
        }
    }

    /// Called from the `End` event of a track. Only advances if the track that
    /// ended is still the current one.
    async fn on_track_end(self: Arc<Self>, guild: u64, token: Uuid) {
        let is_current = {
            let lock = self.guild_lock(guild).await;
            let _transition = lock.lock().await;
            let mut map = self.guilds.lock().await;
            match map.get_mut(&guild) {
                Some(gm) => match &gm.current {
                    Some(current) if current.token == token => {
                        gm.current = None;
                        true
                    }
                    _ => false,
                },
                None => false,
            }
        };

        if is_current {
            if let Err(e) = self.play_next(guild, true).await {
                error!(guild, error = %e, "failed to advance music queue");
            }
        }
    }

    async fn announce_playing(&self, channel: Option<serenity::ChannelId>, request: &TrackRequest) {
        let Some(channel) = channel else {
            return;
        };
        let mut message =
            serenity::CreateMessage::new().content(lang::fmt_music_playing(&request.display_name));

        if let Some((url, filename)) = &request.reupload {
            match self.reqwest.get(url).send().await {
                Ok(response) => match response.bytes().await {
                    Ok(bytes) => {
                        message = message.add_file(serenity::CreateAttachment::bytes(
                            bytes.to_vec(),
                            filename.clone(),
                        ));
                    }
                    Err(e) => error!("music announce: failed to read attachment bytes: {e}"),
                },
                Err(e) => error!("music announce: failed to download attachment: {e}"),
            }
        }

        if let Err(e) = channel.send_message(&*self.http, message).await {
            error!(channel = %channel, error = %e, "music announce: send failed");
        }
    }

    async fn announce_failure(&self, guild: u64, message: String) {
        let channel = {
            let map = self.guilds.lock().await;
            map.get(&guild).and_then(|gm| gm.announce)
        };
        let Some(channel) = channel else {
            return;
        };
        let message = serenity::CreateMessage::new().content(message);
        if let Err(e) = channel.send_message(&*self.http, message).await {
            error!(channel = %channel, error = %e, "music failure announce: send failed");
        }
    }
}

struct TrackEndHandler {
    state: Arc<MusicState>,
    guild: u64,
    token: Uuid,
}

#[async_trait::async_trait]
impl songbird::EventHandler for TrackEndHandler {
    async fn act(&self, _ctx: &songbird::EventContext<'_>) -> Option<songbird::Event> {
        let state = self.state.clone();
        let guild = self.guild;
        let token = self.token;
        tokio::spawn(async move {
            state.on_track_end(guild, token).await;
        });
        Some(songbird::Event::Cancel)
    }
}

// ── Input construction ────────────────────────────────────────────────

/// URLs that require yt-dlp extraction (platforms that don't serve raw audio
/// at a direct URL). These MUST be downloaded first.
fn needs_ytdlp(url: &str) -> bool {
    let lower = url.to_lowercase();
    lower.contains("youtube.com/")
        || lower.contains("youtu.be/")
        || lower.contains("twitch.tv/")
        || lower.contains("soundcloud.com/")
        || lower.contains("bandcamp.com/")
}

/// Heuristic: does this URL look like a live internet radio stream?
fn is_live_stream_hint(url: &str) -> bool {
    let lower = url.to_lowercase();
    lower.contains("/stream")
        || lower.contains("/live")
        || lower.contains("/radio")
        || lower.contains(";stream")
        || lower.contains("icecast")
        || lower.contains("shoutcast")
        || lower.ends_with(".m3u8")
        || lower.ends_with(".m3u")
        || lower.ends_with(".pls")
        || lower.contains(":8000/")
        || lower.contains(":8080/")
}

/// Strip `?pw=...` / `&pw=...` query params from a URL for safe logging.
fn redact_url(url: &str) -> &str {
    let Some(q) = url.find('?') else {
        return url;
    };
    let before = &url[..q];
    let after = &url[q..];
    if after.strip_prefix("?pw=").is_some() {
        return before;
    }
    if let Some(idx) = after.find("&pw=") {
        return &url[..q + idx];
    }
    url
}

/// Strip tracking/playlist params (`&...`) from YouTube URLs.
pub(crate) fn sanitize_yt_url(url: &str) -> String {
    let lower = url.to_lowercase();
    if (lower.contains("youtube.com/") || lower.contains("youtu.be/"))
        && let Some(pos) = url.find('&')
    {
        return url[..pos].to_string();
    }
    url.to_string()
}

async fn build_input(request: &TrackRequest, http: &reqwest::Client) -> Result<Input> {
    // yt-dlp would block forever collecting a live stream into memory (and
    // `--match-filter !is_live` rejects the remaining cases).
    if request.kind == SourceKind::YtDlp && is_live_stream_hint(&request.url) {
        return Err(Error::Custom(
            "live streams are not supported through yt-dlp".into(),
        ));
    }
    if request.compressed {
        let audio_bytes = match request.kind {
            SourceKind::YtDlp => {
                let clean_url = sanitize_yt_url(&request.url);
                info!(url = %clean_url, "downloading via yt-dlp for compression");
                voice::ytdlp_download(&clean_url).await?
            }
            SourceKind::Stream => {
                if is_live_stream_hint(&request.url) {
                    info!(
                        url = %redact_url(&request.url),
                        "live stream — compression skipped, streaming via ffmpeg"
                    );
                    return voice::ffmpeg_stream_input(&request.url).await;
                }
                info!(url = %redact_url(&request.url), "downloading for compression");
                voice::download_url(http, &request.url).await?
            }
        };

        info!("compressing audio ({} bytes)", audio_bytes.len());
        let compressed_bytes =
            crate::commands::utilities::compress_audio_bytes(&audio_bytes, &request.display_name)
                .await
                .ok_or_else(|| Error::Custom("compression failed".into()))?;
        info!("compressed to {} bytes", compressed_bytes.len());

        // Decode the compressed OGG via ffmpeg (symphonia lacks Opus support).
        // Piping keeps the intermediate out of TEMP_PATH entirely.
        let input = voice::ffmpeg_decode_bytes(compressed_bytes).await?;
        Ok(input)
    } else {
        match request.kind {
            SourceKind::YtDlp => {
                let clean_url = sanitize_yt_url(&request.url);
                info!(url = %clean_url, "downloading via yt-dlp");
                let audio_bytes = voice::ytdlp_download(&clean_url).await?;
                tokio::task::spawn_blocking(move || voice::decode_audio(&audio_bytes)).await?
            }
            SourceKind::Stream => {
                let live = is_live_stream_hint(&request.url);
                info!(url = %redact_url(&request.url), live, "streaming via ffmpeg subprocess");
                voice::ffmpeg_stream_input(&request.url).await
            }
        }
    }
}

/// Classify a raw URL for the command layer (yt-dlp vs. direct ffmpeg stream).
pub fn classify_url(url: &str) -> SourceKind {
    if needs_ytdlp(url) {
        SourceKind::YtDlp
    } else {
        SourceKind::Stream
    }
}