//! 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, /// Suppress the "playing" and failure announcements for this track. /// Cron audio uses this so scheduled playback stays silent. pub silent: bool, } /// Where an enqueued track goes relative to the current playback. #[derive(Debug, Clone, Copy, PartialEq, Eq, poise::ChoiceParameter)] pub enum QueuePosition { /// Interrupt the current track and play this right away. Pending tracks /// stay queued behind it. #[name = "now"] Now, /// Insert directly after the current track. #[name = "next"] Next, /// Append to the end of the queue (the command default). #[name = "last"] Last, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EnqueueOutcome { /// Playback started immediately (nothing was playing, or `Now` was used). Started, /// The track was put into the queue; `position` is 1-based. Queued { position: usize }, } #[derive(Debug, Clone)] pub struct QueueItem { pub display_name: String, pub requester: Option, } #[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, /// Bumped by interrupts (`now`, skip, stop). In-flight preparation for an /// older generation is discarded instead of starting over the new track. generation: u64, } /// Shared, per-guild music state. Stored in [`crate::Data`]. pub struct MusicState { guilds: Mutex>, /// One state lock per guild. Held only for state mutations; downloads, /// compression, and voice connects happen outside it so one slow track /// cannot stall every other `/music` command in the guild. locks: Mutex>>>, 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 (was_playing, queued_position, interrupted) = { let _transition = lock.lock().await; let mut map = self.guilds.lock().await; let gm = map.entry(guild).or_insert_with(|| GuildMusic { current: None, queue: VecDeque::new(), voice: voice_channel, announce, generation: 0, }); gm.voice = voice_channel; gm.announce = announce; let was_playing = gm.current.is_some(); match position { QueuePosition::Now => gm.queue.push_front(request), QueuePosition::Next => gm.queue.push_front(request), QueuePosition::Last => gm.queue.push_back(request), } let queued_position = match position { QueuePosition::Next => 1, QueuePosition::Last => gm.queue.len(), QueuePosition::Now => 1, }; let interrupted = if was_playing && position == QueuePosition::Now { gm.generation += 1; gm.current.take() } else { None }; (was_playing, queued_position, interrupted) }; if let Some(current) = interrupted { let _ = current.handle.stop(); } if was_playing && position != QueuePosition::Now { return Ok(EnqueueOutcome::Queued { position: queued_position, }); } // The command layer posts the "now playing" message for direct // enqueues, so don't announce here as well. Preparation happens // outside the guild lock. let played = self.play_next(guild, false).await?; Ok(if played { EnqueueOutcome::Started } else { EnqueueOutcome::Queued { position: 1 } }) } /// Stop the current track and immediately start the next queued one. /// Returns `true` when a track started, `false` when the queue was empty. pub async fn skip(self: &Arc, guild: u64) -> Result { let lock = self.guild_lock(guild).await; let (current, has_queue) = { let _transition = lock.lock().await; let mut map = self.guilds.lock().await; let Some(gm) = map.get_mut(&guild) else { return Ok(false); }; gm.generation += 1; (gm.current.take(), !gm.queue.is_empty()) }; if let Some(current) = current { let _ = current.handle.stop(); } if !has_queue { return Ok(false); } self.play_next(guild, true).await } /// Stop playback and drop the queue (does not disconnect from voice). pub async fn stop(self: &Arc, guild: u64) { let lock = self.guild_lock(guild).await; let current = { let _transition = lock.lock().await; let mut map = self.guilds.lock().await; map.remove(&guild).and_then(|gm| gm.current) }; if let Some(current) = current { let _ = current.handle.stop(); } } /// Play a track immediately, interrupting whatever is playing. Used by the /// cron scheduler (no poise context, and the target channel is a voice /// channel rather than a text channel). pub async fn play_now( self: &Arc, guild: u64, voice_channel: serenity::ChannelId, announce: Option, request: TrackRequest, ) -> Result<()> { let lock = self.guild_lock(guild).await; let current = { let _transition = lock.lock().await; let mut map = self.guilds.lock().await; let gm = map.entry(guild).or_insert_with(|| GuildMusic { current: None, queue: VecDeque::new(), voice: voice_channel, announce, generation: 0, }); gm.generation += 1; gm.voice = voice_channel; // Keep the last text channel when the caller has nowhere to // announce (cron's channel is a voice channel). if announce.is_some() { gm.announce = announce; } gm.queue.push_front(request); gm.current.take() }; if let Some(current) = current { let _ = current.handle.stop(); } if self.play_next(guild, true).await? { Ok(()) } else { Err(Error::Custom(lang::L_MUSIC_QUEUE_EMPTY.into())) } } /// Current track followed by the pending queue, for `/music queue`. pub async fn snapshot(&self, guild: u64) -> QueueSnapshot { let map = self.guilds.lock().await; let Some(gm) = map.get(&guild) else { return QueueSnapshot::default(); }; QueueSnapshot { current: gm.current.as_ref().map(|current| QueueItem { display_name: current.request.display_name.clone(), requester: current.request.requester, }), upcoming: gm .queue .iter() .map(|req| QueueItem { display_name: req.display_name.clone(), requester: req.requester, }) .collect(), } } /// Pop and play the next playable track. /// /// Input preparation and the voice connect happen outside the guild state /// lock, so a slow download cannot block other commands. A generation /// token makes an in-flight track abort if it was interrupted meanwhile. async fn play_next(self: &Arc, guild: u64, announce: bool) -> Result { let mut last_error: Option = None; loop { let (request, voice_channel, announce_channel, generation) = { let lock = self.guild_lock(guild).await; let _transition = lock.lock().await; let mut map = self.guilds.lock().await; let Some(gm) = map.get_mut(&guild) else { return Ok(false); }; let Some(request) = gm.queue.pop_front() else { return match last_error { Some(e) => Err(e), None => Ok(false), }; }; (request, gm.voice, gm.announce, gm.generation) }; let input = match build_input(&request, &self.reqwest).await { Ok(input) => input, Err(e) => { warn!(track = %request.display_name, error = %e, "failed to prepare queued track"); if !request.silent { let message = lang::fmt_music_track_failed(&request.display_name, &e); self.announce_failure(guild, message).await; } last_error = Some(e); continue; } }; let call = match self.voice.get_call(serenity::GuildId::new(guild)) { Some(call) => call, None => match self .voice .connect(serenity::GuildId::new(guild), voice_channel) .await { Ok(call) => call, Err(e) => { // Put the track back unless an interrupt superseded it. let lock = self.guild_lock(guild).await; let _transition = lock.lock().await; let mut map = self.guilds.lock().await; if let Some(gm) = map.get_mut(&guild) && gm.generation == generation { gm.queue.push_front(request); } return Err(e); } }, }; let (handle, token) = { let lock = self.guild_lock(guild).await; let _transition = lock.lock().await; let mut map = self.guilds.lock().await; let Some(gm) = map.get_mut(&guild) else { return Ok(false); }; if gm.generation != generation { // Interrupted while preparing; the replacement request is // already at the front of the queue. continue; } if gm.current.is_some() { // Another preparation won the race; let it handle this. gm.queue.push_front(request); return Ok(false); } let handle = { let mut call = call.lock().await; call.play_input(input) }; let token = Uuid::new_v4(); gm.current = Some(CurrentTrack { token, handle: handle.clone(), request: request.clone(), }); (handle, token) }; let _ = handle.add_event( songbird::Event::Track(songbird::TrackEvent::End), TrackEndHandler { state: self.clone(), guild, token, }, ); let _ = handle.add_event( songbird::Event::Track(songbird::TrackEvent::Error), TrackEndHandler { state: self.clone(), guild, token, }, ); if announce && !request.silent { self.announce_playing(announce_channel, &request).await; } return Ok(true); } } /// Called from the `End` event of a track. Only advances if the track that /// ended is still the current one. async fn on_track_end(self: Arc, guild: u64, token: Uuid) { let is_current = { let lock = self.guild_lock(guild).await; let _transition = lock.lock().await; let mut map = self.guilds.lock().await; match map.get_mut(&guild) { Some(gm) => match &gm.current { Some(current) if current.token == token => { gm.current = None; true } _ => false, }, None => false, } }; if is_current { if let Err(e) = self.play_next(guild, true).await { error!(guild, error = %e, "failed to advance music queue"); } } } async fn announce_playing(&self, channel: Option, 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 { // yt-dlp would block forever collecting a live stream into memory (and // `--match-filter !is_live` rejects the remaining cases). if request.kind == SourceKind::YtDlp && is_live_stream_hint(&request.url) { return Err(Error::Custom( "live streams are not supported through yt-dlp".into(), )); } if request.compressed { let audio_bytes = match request.kind { SourceKind::YtDlp => { let clean_url = sanitize_yt_url(&request.url); info!(url = %clean_url, "downloading via yt-dlp for compression"); voice::ytdlp_download(&clean_url).await? } SourceKind::Stream => { if is_live_stream_hint(&request.url) { info!( url = %redact_url(&request.url), "live stream — compression skipped, streaming via ffmpeg" ); return voice::ffmpeg_stream_input(&request.url).await; } info!(url = %redact_url(&request.url), "downloading for compression"); voice::download_url(http, &request.url).await? } }; info!("compressing audio ({} bytes)", audio_bytes.len()); let compressed_bytes = crate::commands::utilities::compress_audio_bytes(&audio_bytes, &request.display_name) .await .ok_or_else(|| Error::Custom("compression failed".into()))?; info!("compressed to {} bytes", compressed_bytes.len()); // Decode the compressed OGG via ffmpeg (symphonia lacks Opus support). // Piping keeps the intermediate out of TEMP_PATH entirely. let input = voice::ffmpeg_decode_bytes(compressed_bytes).await?; Ok(input) } else { match request.kind { SourceKind::YtDlp => { let clean_url = sanitize_yt_url(&request.url); info!(url = %clean_url, "downloading via yt-dlp"); let audio_bytes = voice::ytdlp_download(&clean_url).await?; tokio::task::spawn_blocking(move || voice::decode_audio(&audio_bytes)).await? } SourceKind::Stream => { let live = is_live_stream_hint(&request.url); info!(url = %redact_url(&request.url), live, "streaming via ffmpeg subprocess"); voice::ffmpeg_stream_input(&request.url).await } } } } /// Classify a raw URL for the command layer (yt-dlp vs. direct ffmpeg stream). pub fn classify_url(url: &str) -> SourceKind { if needs_ytdlp(url) { SourceKind::YtDlp } else { SourceKind::Stream } }