From fc5e0f48477366641b97efb0541053cc1e061be5 Mon Sep 17 00:00:00 2001 From: salmonik Date: Tue, 27 Oct 2026 18:14:07 +0100 Subject: 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. --- src/commands/music.rs | 355 +++++++++++---------------- src/commands/utilities.rs | 2 +- src/lang.rs | 17 +- src/main.rs | 16 +- src/music.rs | 602 ++++++++++++++++++++++++++++++++++++++++++++++ src/tasks/mod.rs | 67 ++---- 6 files changed, 790 insertions(+), 269 deletions(-) create mode 100644 src/music.rs (limited to 'src') 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, #[description = "Bezpośredni plik z twojego systemu"] file: Option, #[description = "Chujowa kompresja!?"] compressed: Option, + #[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, + #[description = "Bezpośredni plik z twojego systemu"] file: Option, + #[description = "Chujowa kompresja!?"] compressed: Option, + #[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, + file: Option, + compressed: Option, + position: Option, ) -> 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, @@ -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, file: Option, 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| { + 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( + ¤t.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> { - 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>>, + /// Active music queue and current track per guild. + pub music: Arc, /// TTS engine: piper model + audio cache. pub tts: Arc, } @@ -198,7 +199,11 @@ async fn main() -> Result<(), Box> { ) .await?; - let music_tracks: Arc>> = 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> { 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> { 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, +} + +/// 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, +} + +#[derive(Debug, Clone, Default)] +pub struct QueueSnapshot { + pub current: Option, + pub upcoming: Vec, +} + +// ── 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, + queue: VecDeque, + /// Voice channel the bot is (or should be) connected to. + voice: serenity::ChannelId, + /// Text channel where auto-advance announcements are posted. + announce: Option, +} + +/// Shared, per-guild music state. Stored in [`crate::Data`]. +pub struct MusicState { + guilds: Mutex>, + /// One transition lock per guild. Held across the whole stop/prepare/play + /// sequence so concurrent `/music` commands cannot interleave. + locks: Mutex>>>, + voice: VoiceManager, + http: Arc, + reqwest: reqwest::Client, +} + +impl MusicState { + pub fn new( + voice: VoiceManager, + http: Arc, + reqwest: reqwest::Client, + ) -> Arc { + Arc::new(Self { + guilds: Mutex::new(HashMap::new()), + locks: Mutex::new(HashMap::new()), + voice, + http, + reqwest, + }) + } + + async fn guild_lock(&self, guild: u64) -> Arc> { + 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, + guild: u64, + voice_channel: serenity::ChannelId, + announce: Option, + request: TrackRequest, + position: QueuePosition, + ) -> Result { + 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, guild: u64) -> 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 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, 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, + guild: u64, + voice_channel: serenity::ChannelId, + announce: Option, + 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, guild: u64) -> Result { + let mut last_error: Option = 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, 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, 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, + guild: u64, + token: Uuid, +} + +#[async_trait::async_trait] +impl songbird::EventHandler for TrackEndHandler { + async fn act(&self, _ctx: &songbird::EventContext<'_>) -> Option { + 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 { + 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, - pub voice_manager: VoiceManager, pub copyparty: CopypartyClient, - pub music_tracks: Arc>>, + pub music: Arc, } /// 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 ); } -- cgit v1.2.3