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/music.rs | 602 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 602 insertions(+) create mode 100644 src/music.rs (limited to 'src/music.rs') 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 + } +} -- cgit v1.2.3