music: add per-guild queue with now/next/last, skip and queue

Tracks are queued in memory per guild, defaulting to append (last).
Playback auto-advances on TrackEvent::End/Error with a per-track token
so manual stops and skips cannot double-advance. /music queue shows the
current track followed by pending items. /stop clears the queue.

Cron audio now goes through the same MusicState instead of poking the
track-handle map directly.

6 files changed, 790 insertions, 269 deletions

diff --git a/src/commands/music.rs b/src/commands/music.rs
index 8739a5b..8841329 100644
--- a/src/commands/music.rs
+++ b/src/commands/music.rs
@@ -1,5 +1,5 @@
-use poise::serenity_prelude as serenity;
 use super::utilities::{FileType, extract_urls, file_type};
+use poise::serenity_prelude as serenity;
 
 use tracing::{error, info};
 
@@ -8,29 +8,16 @@ use crate::Data;
 use crate::copyparty::{html_decode, percent_decode};
 use crate::error::{Error, Result};
 use crate::lang;
+use crate::music::{EnqueueOutcome, QueuePosition, SourceKind, TrackRequest, classify_url};
 
 type Context<'a> = poise::Context<'a, Data, Error>;
 
 // ── Source classification ──────────────────────────────────────────
 
-/// What kind of audio source are we dealing with?
-///
-/// This mirrors the Python original's split: `EXTERNAL_URL_PATTERN`
-/// (youtube/soundcloud/bandcamp) goes through yt-dlp download, while
-/// *every other* URL is streamed directly via ffmpeg (radio, HLS,
-/// playlists, direct media files).
-#[derive(Debug, Clone, PartialEq)]
-enum SourceKind {
-    /// YouTube, SoundCloud, Twitch, Bandcamp — platforms that don't serve
-    /// raw audio at a direct URL. Needs yt-dlp to extract + download the
-    /// media to memory before decoding.
-    YtDlp,
-    /// Any HTTP(S) URL: internet radio (Icecast/Shoutcast), HLS `.m3u8`,
-    /// `.m3u`/`.pls` playlists, Copyparty files, Discord attachments, or
-    /// directly-accessible media files. Streamed via an ffmpeg subprocess
-    /// — NO full download. ffmpeg handles playlist parsing, Icecast
-    /// metadata, reconnection, and codec/container detection universally.
-    Stream,
+/// Music playback and queue management.
+#[poise::command(slash_command, subcommands("grajcuj", "skip", "queue"))]
+pub async fn music(_ctx: Context<'_>) -> Result<()> {
+    Ok(())
 }
 
 /// Play music from Copyparty, URL, or file upload
@@ -42,14 +29,44 @@ pub async fn grajcuj(
     melodia: Option<String>,
     #[description = "Bezpośredni plik z twojego systemu"] file: Option<serenity::Attachment>,
     #[description = "Chujowa kompresja!?"] compressed: Option<bool>,
+    #[description = "Gdzie dodać utwór w kolejce (domyślnie: last)"] position: Option<
+        QueuePosition,
+    >,
+) -> Result<()> {
+    play_command(ctx, melodia, file, compressed, position).await
+}
+
+/// Play music from Copyparty, URL, or file upload (top-level alias)
+#[poise::command(slash_command, rename = "grajcuj")]
+pub async fn grajcuj_top(
+    ctx: Context<'_>,
+    #[description = "Plik do odtworzenia"]
+    #[autocomplete = "audio_file"]
+    melodia: Option<String>,
+    #[description = "Bezpośredni plik z twojego systemu"] file: Option<serenity::Attachment>,
+    #[description = "Chujowa kompresja!?"] compressed: Option<bool>,
+    #[description = "Gdzie dodać utwór w kolejce (domyślnie: last)"] position: Option<
+        QueuePosition,
+    >,
+) -> Result<()> {
+    play_command(ctx, melodia, file, compressed, position).await
+}
+
+async fn play_command(
+    ctx: Context<'_>,
+    melodia: Option<String>,
+    file: Option<serenity::Attachment>,
+    compressed: Option<bool>,
+    position: Option<QueuePosition>,
 ) -> Result<()> {
     let compressed = compressed.unwrap_or(false);
+    let position = position.unwrap_or(QueuePosition::Last);
     eprintln!(
-        "[DEBUG] play() reached, melodia={:?}, file={:?}, compressed={compressed}",
+        "[DEBUG] play() reached, melodia={:?}, file={:?}, compressed={compressed}, position={position:?}",
         melodia,
         file.as_ref().map(|f| &f.filename)
     );
-    let result = play_impl(ctx, melodia, file, compressed).await;
+    let result = play_impl(ctx, melodia, file, compressed, position).await;
     if let Err(e) = &result {
         eprintln!("[DEBUG] play() error: {e}");
         error!("play command failed: {e}");
@@ -69,38 +86,6 @@ fn clean_name(raw: &str) -> String {
 
 // ── URL classification helpers ──────────────────────────────────────
 
-/// URLs that require yt-dlp extraction (platforms that don't serve
-/// raw audio at a direct URL). These MUST be downloaded first.
-///
-/// Matches the Python original's `EXTERNAL_URL_PATTERN`:
-/// `youtube|youtu.be|soundcloud|bandcamp` (plus twitch).
-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?
-/// (Used for logging / display only — dispatch is controlled by `SourceKind`.)
-fn is_live_stream_hint(url: &str) -> bool {
-    let lower = url.to_lowercase();
-    // Common Icecast/Shoutcast mount patterns
-    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/")   // common Icecast port
-        || lower.contains(":8080/") // common alt port
-}
-
 async fn resolve_source(
     ctx: Context<'_>,
     melodia: Option<String>,
@@ -154,11 +139,7 @@ async fn resolve_source(
 
         // yt-dlp platforms → download; everything else (radio, HLS,
         // playlists, Copyparty, direct media) → ffmpeg stream.
-        let kind = if needs_ytdlp(&url) {
-            SourceKind::YtDlp
-        } else {
-            SourceKind::Stream
-        };
+        let kind = classify_url(&url);
         return Ok(ResolvedSource {
             url,
             display_name,
@@ -174,8 +155,10 @@ async fn resolve_source(
         .search(
             &name,
             5,
-            &["mp3", "wav", "ogg", "flac", "m4a", "opus", "aac", "wma", "oga",
-              "mp4", "webm", "mkv", "avi", "mov", "flv", "wmv", "m4v", "mpg", "mpeg", "3gp", "ts"],
+            &[
+                "mp3", "wav", "ogg", "flac", "m4a", "opus", "aac", "wma", "oga", "mp4", "webm",
+                "mkv", "avi", "mov", "flv", "wmv", "m4v", "mpg", "mpeg", "3gp", "ts",
+            ],
         )
         .await
         .map_err(|e| {
@@ -202,34 +185,6 @@ async fn resolve_source(
     })
 }
 
-/// 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..];
-    // Strip `?pw=...` or `&pw=...`
-    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.
-/// yt-dlp only needs `?v=ID`; extra `&` params can cause failures.
-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()
-}
-
 // ── Playlist resolution is handled by ffmpeg natively ──────────────
 // ffmpeg reads `.m3u`/`.m3u8`/`.pls` playlists directly (HLS segment
 // fetching, PLS/M3U entry parsing) — no manual resolution needed.
@@ -240,6 +195,7 @@ async fn play_impl(
     melodia: Option<String>,
     file: Option<serenity::Attachment>,
     compressed: bool,
+    position: QueuePosition,
 ) -> Result<()> {
     ctx.defer().await?;
 
@@ -259,132 +215,93 @@ async fn play_impl(
     let file_reupload = file.as_ref().map(|f| (f.url.clone(), f.filename.clone()));
     let src = resolve_source(ctx, melodia, file).await?;
 
-    let call;
-    let input;
-
-    if compressed {
-        // ── Compressed path: download → compress → decode ──────────
-        // Defer stopping current track until compression is done.
-        let audio_bytes = match src.kind {
-            SourceKind::YtDlp => {
-                let clean_url = sanitize_yt_url(&src.url);
-                info!(url = %clean_url, "downloading via yt-dlp for compression");
-                crate::voice::ytdlp_download(&clean_url).await?
-            }
-            SourceKind::Stream => {
-                if is_live_stream_hint(&src.url) {
-                    ctx.say(lang::L_MUSIC_CANT_COMPRESS_LIVE).await?;
-                    // Fall through to normal streaming below
-                    let live = true;
-                    info!(url = %redact_url(&src.url), live, "live stream — compression skipped, streaming via ffmpeg");
-                    // Stop current, connect, stream
-                    {
-                        let mut map = ctx.data().music_tracks.lock().await;
-                        if let Some(handle) = map.remove(&guild_id.get()) {
-                            let _ = handle.stop();
-                        }
-                    }
-                    let (call_res, input_res) = tokio::join!(
-                        ctx.data().voice_manager.connect(guild_id, channel_id),
-                        crate::voice::ffmpeg_stream_input(&src.url)
-                    );
-                    call = call_res?;
-                    input = input_res?;
-                    // skip the rest of the compressed path
-                    let handle = {
-                        let mut call_lock = call.lock().await;
-                        call_lock.play_input(input)
-                    };
-                    ctx.data()
-                        .music_tracks
-                        .lock()
-                        .await
-                        .insert(guild_id.get(), handle);
-                    send_music_msg(ctx, &src.display_name, file_reupload).await?;
-                    return Ok(());
-                }
-                info!(url = %redact_url(&src.url), "downloading for compression");
-                crate::voice::download_url(&ctx.data().http_client, &src.url).await?
-            }
-        };
+    let request = TrackRequest {
+        url: src.url,
+        display_name: src.display_name.clone(),
+        kind: src.kind,
+        compressed,
+        reupload: file_reupload.clone(),
+        requester: Some(ctx.author().id),
+    };
 
-        info!("compressing audio ({} bytes)", audio_bytes.len());
-        let compressed_bytes =
-            crate::commands::utilities::compress_audio_bytes(&audio_bytes, &src.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).
-        // Write to temp file, then stream through ffmpeg to raw PCM.
-        let compressed_path = format!(
-            "{}/play_compressed_{}.ogg",
-            crate::config::TEMP_PATH,
-            uuid::Uuid::new_v4()
-        );
-        let _ = tokio::fs::create_dir_all(crate::config::TEMP_PATH).await;
-        tokio::fs::write(&compressed_path, &compressed_bytes).await?;
-        let decoded = crate::voice::ffmpeg_decode_file(&compressed_path).await?;
-
-        // Deferred stop: only interrupt current track now that we're ready.
-        {
-            let mut map = ctx.data().music_tracks.lock().await;
-            if let Some(handle) = map.remove(&guild_id.get()) {
-                let _ = handle.stop();
-            }
+    let outcome = ctx
+        .data()
+        .music
+        .enqueue(
+            guild_id.get(),
+            channel_id,
+            Some(ctx.channel_id()),
+            request,
+            position,
+        )
+        .await?;
+
+    match outcome {
+        EnqueueOutcome::Started => {
+            send_music_msg(ctx, &src.display_name, file_reupload).await?;
         }
-        call = ctx
-            .data()
-            .voice_manager
-            .connect(guild_id, channel_id)
-            .await?;
-        input = decoded;
-    } else {
-        // ── Normal path: stop first, then stream/download ──────────
-        {
-            let mut map = ctx.data().music_tracks.lock().await;
-            if let Some(handle) = map.remove(&guild_id.get()) {
-                let _ = handle.stop();
-            }
+        EnqueueOutcome::Queued { position } => {
+            ctx.say(lang::fmt_music_queued(&src.display_name, position))
+                .await?;
         }
+    }
+    Ok(())
+}
 
-        match src.kind {
-            SourceKind::YtDlp => {
-                let clean_url = sanitize_yt_url(&src.url);
-                info!(url = %clean_url, "downloading via yt-dlp");
-                let audio_bytes = crate::voice::ytdlp_download(&clean_url).await?;
-                let (call_res, input_res) = tokio::join!(
-                    ctx.data().voice_manager.connect(guild_id, channel_id),
-                    tokio::task::spawn_blocking(move || crate::voice::decode_audio(&audio_bytes))
-                );
-                call = call_res?;
-                input = input_res??;
-            }
-            SourceKind::Stream => {
-                let live = is_live_stream_hint(&src.url);
-                info!(url = %redact_url(&src.url), live, "streaming via ffmpeg subprocess");
-                let (call_res, input_res) = tokio::join!(
-                    ctx.data().voice_manager.connect(guild_id, channel_id),
-                    crate::voice::ffmpeg_stream_input(&src.url)
-                );
-                call = call_res?;
-                input = input_res?;
-            }
-        }
+/// Skip the current track and play the next queued one
+#[poise::command(slash_command)]
+pub async fn skip(ctx: Context<'_>) -> Result<()> {
+    ctx.defer().await?;
+    let guild_id = ctx.guild_id().ok_or_else(lang::err_not_in_guild)?;
+    let played = ctx.data().music.skip(guild_id.get()).await?;
+    let content = if played {
+        lang::L_MUSIC_SKIPPED
+    } else {
+        lang::L_MUSIC_QUEUE_EMPTY
+    };
+    ctx.send(
+        poise::CreateReply::default()
+            .content(content)
+            .ephemeral(true),
+    )
+    .await?;
+    Ok(())
+}
+
+/// Show the current track and the pending queue
+#[poise::command(slash_command)]
+pub async fn queue(ctx: Context<'_>) -> Result<()> {
+    let guild_id = ctx.guild_id().ok_or_else(lang::err_not_in_guild)?;
+    let snapshot = ctx.data().music.snapshot(guild_id.get()).await;
+
+    if snapshot.current.is_none() && snapshot.upcoming.is_empty() {
+        ctx.say(lang::L_MUSIC_QUEUE_EMPTY).await?;
+        return Ok(());
     }
 
-    let handle = {
-        let mut call_lock = call.lock().await;
-        call_lock.play_input(input)
+    let who = |requester: Option<serenity::UserId>| {
+        requester
+            .map(|id| format!(" <@{}>", id.get()))
+            .unwrap_or_default()
     };
 
-    ctx.data()
-        .music_tracks
-        .lock()
-        .await
-        .insert(guild_id.get(), handle);
+    let mut lines = Vec::with_capacity(snapshot.upcoming.len() + 2);
+    lines.push(lang::L_MUSIC_QUEUE_HEADER.to_string());
+    if let Some(current) = &snapshot.current {
+        lines.push(lang::fmt_music_queue_current(
+            &current.display_name,
+            &who(current.requester),
+        ));
+    }
+    for (index, item) in snapshot.upcoming.iter().enumerate() {
+        lines.push(lang::fmt_music_queue_item(
+            index + 1,
+            &item.display_name,
+            &who(item.requester),
+        ));
+    }
 
-    send_music_msg(ctx, &src.display_name, file_reupload).await?;
+    ctx.say(lines.join("\n")).await?;
     Ok(())
 }
 
@@ -392,8 +309,8 @@ async fn play_impl(
 #[poise::command(slash_command)]
 pub async fn stop(ctx: Context<'_>) -> Result<()> {
     let guild_id = ctx.guild_id().ok_or_else(lang::err_not_in_guild)?;
+    ctx.data().music.stop(guild_id.get()).await;
     ctx.data().voice_manager.disconnect(guild_id).await?;
-    ctx.data().music_tracks.lock().await.remove(&guild_id.get());
     ctx.send(
         poise::CreateReply::default()
             .content(lang::L_MUSIC_STOPPED)
@@ -443,9 +360,14 @@ async fn send_music_msg(
             Ok(resp) => match resp.bytes().await {
                 Ok(bytes) => {
                     if ctx
-                        .send(poise::CreateReply::default().content(lang::fmt_music_playing(display_name)).attachment(
-                            serenity::CreateAttachment::bytes(bytes.to_vec(), filename.clone()),
-                        ))
+                        .send(
+                            poise::CreateReply::default()
+                                .content(lang::fmt_music_playing(display_name))
+                                .attachment(serenity::CreateAttachment::bytes(
+                                    bytes.to_vec(),
+                                    filename.clone(),
+                                )),
+                        )
                         .await
                         .is_ok()
                     {
@@ -466,20 +388,21 @@ async fn send_music_msg(
 #[poise::command(context_menu_command = "Odtwórz")]
 pub async fn message_grajcuj(ctx: Context<'_>, message: serenity::Message) -> Result<()> {
     // Files > links: prefer audio/video attachments first.
-    if let Some(att) = message.attachments.iter().find(|a| {
-        matches!(file_type(&a.filename), FileType::Audio | FileType::Video)
-    }) {
-        return play_impl(ctx, None, Some(att.clone()), false).await;
+    if let Some(att) = message
+        .attachments
+        .iter()
+        .find(|a| matches!(file_type(&a.filename), FileType::Audio | FileType::Video))
+    {
+        return play_impl(ctx, None, Some(att.clone()), false, QueuePosition::Now).await;
     }
     // Fallback: first URL in message content.
     if let Some(url) = extract_urls(&message.content).into_iter().next() {
-        return play_impl(ctx, Some(url), None, false).await;
+        return play_impl(ctx, Some(url), None, false, QueuePosition::Now).await;
     }
     ctx.say(lang::L_MUSIC_NO_AUDIO_ATTACHMENTS).await?;
     Ok(())
 }
 
-
 pub fn commands() -> Vec<poise::Command<Data, Error>> {
-    vec![grajcuj(), stop(), join(), message_grajcuj()]
+    vec![music(), grajcuj_top(), stop(), join(), message_grajcuj()]
 }
diff --git a/src/commands/utilities.rs b/src/commands/utilities.rs
index 87a5089..25d4bcf 100644
--- a/src/commands/utilities.rs
+++ b/src/commands/utilities.rs
@@ -2,7 +2,7 @@ use poise::serenity_prelude as serenity;
 use tracing::warn;
 
 use super::autocomplete::any_file;
-use super::music::sanitize_yt_url;
+use crate::music::sanitize_yt_url;
 use crate::Data;
 use crate::copyparty::{html_decode, percent_decode};
 use crate::error::{Error, Result};
diff --git a/src/lang.rs b/src/lang.rs
index 4bba9b5..de6d23f 100644
--- a/src/lang.rs
+++ b/src/lang.rs
@@ -91,8 +91,6 @@ pub fn fmt_math_result(r: &str) -> String {
 pub const L_MUSIC_NEED_NAME: &str = "Daj mi nazwę pliku, URL albo plik";
 pub const L_MUSIC_NO_AUDIO_ON_CP: &str = "Brak powiązanego audio na copyparty";
 pub const L_MUSIC_NO_URL_IN_RESULT: &str = "Wynik nie ma URL";
-pub const L_MUSIC_CANT_COMPRESS_LIVE: &str =
-    "Nie można skompresować transmisji na żywo, odtwarzam normalnie";
 pub const L_MUSIC_STOPPED: &str = "wyszed";
 pub const L_MUSIC_NO_AUDIO_ATTACHMENTS: &str = "Brak plików audio w wiadomości";
 pub const L_VOICE_USER_NOT_IN_VC: &str = "You are not in a voice channel";
@@ -122,6 +120,21 @@ pub fn fmt_music_search_failed(e: &dyn std::fmt::Display) -> String {
 pub fn fmt_music_playing(name: &str) -> String {
     format!("Grajcowanie: `{name}`")
 }
+pub const L_MUSIC_QUEUE_EMPTY: &str = "Kolejka jest pusta";
+pub const L_MUSIC_QUEUE_HEADER: &str = "**Kolejka:**";
+pub const L_MUSIC_SKIPPED: &str = "Pominięto";
+pub fn fmt_music_queued(name: &str, position: usize) -> String {
+    format!("W kolejce (#{position}): `{name}`")
+}
+pub fn fmt_music_track_failed(name: &str, e: &dyn std::fmt::Display) -> String {
+    format!("Nie udało się odtworzyć `{name}`: `{e}`")
+}
+pub fn fmt_music_queue_current(name: &str, who: &str) -> String {
+    format!("▶ `{name}`{who}")
+}
+pub fn fmt_music_queue_item(index: usize, name: &str, who: &str) -> String {
+    format!("{index}. `{name}`{who}")
+}
 pub fn fmt_voice_joined(cid: &dyn std::fmt::Display) -> String {
     format!("Dołączon <#{cid}>")
 }
diff --git a/src/main.rs b/src/main.rs
index df40481..4e308b8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -13,6 +13,7 @@ mod database;
 mod error;
 mod events;
 mod lang;
+mod music;
 mod openrouter;
 mod recording;
 mod tasks;
@@ -49,8 +50,8 @@ pub struct Data {
     /// Populated by background tasks when the first search times out, so the
     /// next keystroke's autocomplete can serve real results immediately.
     pub ac_search_cache: AcSearchCache,
-    /// Active music track per guild (only one music source at a time).
-    pub music_tracks: Arc<tokio::sync::Mutex<HashMap<u64, songbird::tracks::TrackHandle>>>,
+    /// Active music queue and current track per guild.
+    pub music: Arc<music::MusicState>,
     /// TTS engine: piper model + audio cache.
     pub tts: Arc<tts::TtsState>,
 }
@@ -198,7 +199,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 )
                 .await?;
 
-                let music_tracks: Arc<tokio::sync::Mutex<HashMap<u64, songbird::tracks::TrackHandle>>> = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
+                let music = music::MusicState::new(
+                    vm.clone(),
+                    ctx.http.clone(),
+                    http_client.clone(),
+                );
 
                 let tts = Arc::new(tts::TtsState::new(config.piper_model_path()));
 
@@ -213,7 +218,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     triggers: tokio::sync::Mutex::new(triggers),
                     ac_cache: tokio::sync::Mutex::new(HashMap::new()),
                     ac_search_cache: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
-                    music_tracks: music_tracks.clone(),
+                    music,
                 };
 
                 // Spawn cron scheduler loop.
@@ -221,9 +226,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     let cron_ctx = crate::tasks::CronContext {
                         pool: data.db_pool.clone(),
                         http: ctx.http.clone(),
-                        voice_manager: data.voice_manager.clone(),
                         copyparty: data.copyparty.clone(),
-                        music_tracks: music_tracks.clone(),
+                        music: data.music.clone(),
                     };
                     tokio::spawn(async move {
                         crate::tasks::cron_scheduler_loop(cron_ctx).await;
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
+    }
+}
diff --git a/src/tasks/mod.rs b/src/tasks/mod.rs
index d4cf9b2..ab0d062 100644
--- a/src/tasks/mod.rs
+++ b/src/tasks/mod.rs
@@ -1,4 +1,3 @@
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use chrono::{DateTime, Duration, Local, TimeZone, Timelike, Utc};
@@ -10,16 +9,15 @@ use tracing::{error, info};
 
 use crate::copyparty::CopypartyClient;
 use crate::database::{self, DbPool};
-use crate::voice::VoiceManager;
+use crate::music::{MusicState, SourceKind, TrackRequest};
 
 /// Shared resources the cron scheduler needs.
 #[derive(Clone)]
 pub struct CronContext {
     pub pool: DbPool,
     pub http: Arc<serenity::Http>,
-    pub voice_manager: VoiceManager,
     pub copyparty: CopypartyClient,
-    pub music_tracks: Arc<tokio::sync::Mutex<HashMap<u64, songbird::tracks::TrackHandle>>>,
+    pub music: Arc<MusicState>,
 }
 
 /// Normalize user-friendly cron expressions to the 7-field format the `cron` crate expects.
@@ -139,14 +137,9 @@ async fn execute_audio_job(ctx: &CronContext, job: &crate::database::CronJob) {
         // Legacy relative path — build full URL with password
         ctx.copyparty.build_url(audio_url)
     };
-    info!("Cron audio job {} streaming via ffmpeg: {url}", job.id);
-    let guild_channel_fut = channel_id.to_channel(&ctx.http);
-    let input_fut = crate::voice::ffmpeg_stream_input(&url);
-
-    let (channel_result, input_result) = tokio::join!(guild_channel_fut, input_fut);
-
-    let guild_id = match channel_result {
-        Ok(serenity::Channel::Guild(ch)) => ch.guild_id,
+    info!("Cron audio job {} playing: {url}", job.id);
+    let channel = match channel_id.to_channel(&ctx.http).await {
+        Ok(serenity::Channel::Guild(ch)) => ch,
         other => {
             error!(
                 "Cron audio job {}: channel {} is not a guild voice channel ({:?})",
@@ -158,45 +151,31 @@ async fn execute_audio_job(ctx: &CronContext, job: &crate::database::CronJob) {
         }
     };
 
-    let input = match input_result {
-        Ok(i) => i,
-        Err(e) => {
-            error!(
-                "Cron audio job {} ffmpeg stream failed ({url}): {e}",
-                job.id
-            );
-            return;
-        }
+    let request = TrackRequest {
+        url: url.clone(),
+        display_name: audio_url
+            .rsplit('/')
+            .next()
+            .unwrap_or(audio_url)
+            .to_string(),
+        kind: SourceKind::Stream,
+        compressed: false,
+        reupload: None,
+        requester: None,
     };
 
-    // Stop any existing track in this guild.
+    if let Err(e) = ctx
+        .music
+        .play_now(channel.guild_id.get(), channel_id, None, request)
+        .await
     {
-        let mut map = ctx.music_tracks.lock().await;
-        if let Some(handle) = map.remove(&guild_id.get()) {
-            let _ = handle.stop();
-        }
+        error!("Cron audio job {} playback failed: {e}", job.id);
+        return;
     }
 
-    // Connect to voice and play.
-    let call = match ctx.voice_manager.connect(guild_id, channel_id).await {
-        Ok(c) => c,
-        Err(e) => {
-            error!("Cron audio job {} voice connect failed: {e}", job.id);
-            return;
-        }
-    };
-
-    let handle = {
-        let mut call_lock = call.lock().await;
-        call_lock.play_input(input)
-    };
-
-    // Store the handle so the track stays alive.
-    ctx.music_tracks.lock().await.insert(guild_id.get(), handle);
-
     info!(
         "Cron audio job {} playing '{}' (url: {url}) in guild {}",
-        job.id, audio_url, guild_id
+        job.id, audio_url, channel.guild_id
     );
 }