1 files changed, 602 insertions, 0 deletions

diff --git a/src/music.rs b/src/music.rs
new file mode 100644
index 0000000..8dfc4f8
--- /dev/null
+++ b/src/music.rs
@@ -0,0 +1,602 @@
+//! 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>,
+}
+
+/// 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>,
+}
+
+/// Shared, per-guild music state. Stored in [`crate::Data`].
+pub struct MusicState {
+    guilds: Mutex<HashMap<u64, GuildMusic>>,
+    /// One transition lock per guild. Held across the whole stop/prepare/play
+    /// sequence so concurrent `/music` commands cannot interleave.
+    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 _transition = lock.lock().await;
+
+        let was_playing;
+        {
+            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,
+            });
+            gm.voice = voice_channel;
+            gm.announce = announce;
+            was_playing = gm.current.is_some();
+
+            match position {
+                QueuePosition::Now | QueuePosition::Next => gm.queue.push_front(request),
+                QueuePosition::Last => gm.queue.push_back(request),
+            }
+        }
+
+        if was_playing && position != QueuePosition::Now {
+            let position = {
+                let map = self.guilds.lock().await;
+                let len = map.get(&guild).map(|gm| gm.queue.len()).unwrap_or(1);
+                if position == QueuePosition::Next {
+                    1
+                } else {
+                    len
+                }
+            };
+            return Ok(EnqueueOutcome::Queued { position });
+        }
+
+        if was_playing {
+            // `Now`: drop the current track. The new request is at the front of
+            // the queue; play_next_locked will pick it up.
+            let current = {
+                let mut map = self.guilds.lock().await;
+                map.get_mut(&guild).and_then(|gm| gm.current.take())
+            };
+            if let Some(current) = current {
+                let _ = current.handle.stop();
+            }
+        }
+
+        let played = self.play_next_locked(guild).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 _transition = lock.lock().await;
+
+        let current = {
+            let mut map = self.guilds.lock().await;
+            map.get_mut(&guild).and_then(|gm| gm.current.take())
+        };
+        if let Some(current) = current {
+            let _ = current.handle.stop();
+        }
+
+        let has_queue = {
+            let map = self.guilds.lock().await;
+            map.get(&guild)
+                .map(|gm| !gm.queue.is_empty())
+                .unwrap_or(false)
+        };
+        if !has_queue {
+            return Ok(false);
+        }
+
+        self.play_next_locked(guild).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 _transition = lock.lock().await;
+
+        let current = {
+            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 _transition = lock.lock().await;
+
+        let current = {
+            let mut map = self.guilds.lock().await;
+            map.get_mut(&guild).and_then(|gm| gm.current.take())
+        };
+        if let Some(current) = current {
+            let _ = current.handle.stop();
+        }
+
+        {
+            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,
+            });
+            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);
+        }
+
+        if self.play_next_locked(guild).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. Caller must hold the guild
+    /// transition lock.
+    async fn play_next_locked(self: &Arc<Self>, guild: u64) -> Result<bool> {
+        let mut last_error: Option<Error> = None;
+        loop {
+            let request = {
+                let mut map = self.guilds.lock().await;
+                match map.get_mut(&guild).and_then(|gm| gm.queue.pop_front()) {
+                    Some(request) => request,
+                    None => {
+                        return match last_error {
+                            Some(e) => Err(e),
+                            None => Ok(false),
+                        };
+                    }
+                }
+            };
+
+            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");
+                    let message = lang::fmt_music_track_failed(&request.display_name, &e);
+                    self.announce_failure(guild, message).await;
+                    last_error = Some(e);
+                    continue;
+                }
+            };
+
+            let (voice_channel, announce) = {
+                let map = self.guilds.lock().await;
+                match map.get(&guild) {
+                    Some(gm) => (gm.voice, gm.announce),
+                    None => return Ok(false),
+                }
+            };
+
+            let call = match self.voice.get_call(serenity::GuildId::new(guild)) {
+                Some(call) => call,
+                None => {
+                    self.voice
+                        .connect(serenity::GuildId::new(guild), voice_channel)
+                        .await?
+                }
+            };
+
+            let handle = {
+                let mut call = call.lock().await;
+                call.play_input(input)
+            };
+            let token = Uuid::new_v4();
+            {
+                let mut map = self.guilds.lock().await;
+                if let Some(gm) = map.get_mut(&guild) {
+                    gm.current = Some(CurrentTrack {
+                        token,
+                        handle: handle.clone(),
+                        request: request.clone(),
+                    });
+                }
+            }
+
+            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,
+                },
+            );
+
+            self.announce_playing(announce, &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 lock = self.guild_lock(guild).await;
+        let _transition = lock.lock().await;
+
+        let is_current = {
+            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_locked(guild).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> {
+    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).
+        let compressed_path = format!(
+            "{}/play_compressed_{}.ogg",
+            crate::config::TEMP_PATH,
+            Uuid::new_v4()
+        );
+        let _ = tokio::fs::create_dir_all(crate::config::TEMP_PATH).await;
+        tokio::fs::write(&compressed_path, &compressed_bytes).await?;
+        let input = voice::ffmpeg_decode_file(&compressed_path).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
+    }
+}