diff --git a/src/app/integrations.rs b/src/app/integrations.rs index 0e73a53..01faa79 100644 --- a/src/app/integrations.rs +++ b/src/app/integrations.rs @@ -1,5 +1,7 @@ use super::modal::{SearchMode, SpotifyAuthStatus, SpotifyPlayerStatus}; -use super::spotify_state::{SpotifyPlaybackBackend, SpotifySearchPage}; +use super::spotify_state::{ + SpotifyPlaybackBackend, SpotifyRemoteSkipDirection, SpotifyRemoteSkipResult, SpotifySearchPage, +}; use super::{abort_task, App, SpotifyControlTarget}; use crate::config::{Config, SpotifyPlaybackMode}; @@ -748,6 +750,8 @@ impl App { self.config.save(); self.spotify.token_refreshed_at = Some(std::time::Instant::now()); self.start_playback_polling(); + self.spotify.remote_skip_paused_for_token_refresh = false; + self.start_next_spotify_remote_skip(); tracing::info!("spotify: access_token refreshed"); } Ok(Err(e)) => { @@ -767,61 +771,137 @@ impl App { } pub fn poll_spotify_play_result(&mut self) { - use crate::integrations::spotify::SpotifyError; if let Some(rx) = self.spotify.play_result_rx.take() { match rx.try_recv() { - Ok(Ok(())) => { - if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { - self.spotify.player_status = SpotifyPlayerStatus::Playing; - } + Ok(result) => self.handle_spotify_remote_command_result(result), + Err(std::sync::mpsc::TryRecvError::Empty) => { + self.spotify.play_result_rx = Some(rx); } - Ok(Err(SpotifyError::Unauthorized)) => { - if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { - self.spotify.active_backend = None; - } - self.spotify.player_status = SpotifyPlayerStatus::Error(crate::i18n::t( - "integrations.spotify.error.generic", - )); - tracing::warn!("spotify play: token expirado, renovando"); - self.spotify.token_refreshed_at = - Some(std::time::Instant::now() - std::time::Duration::from_secs(60 * 60)); + Err(std::sync::mpsc::TryRecvError::Disconnected) => {} + } + } + } + + fn handle_spotify_remote_command_result( + &mut self, + result: Result<(), crate::integrations::spotify::SpotifyError>, + ) { + use crate::integrations::spotify::SpotifyError; + match result { + Ok(()) => { + if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { + self.spotify.player_status = SpotifyPlayerStatus::Playing; } - Ok(Err(SpotifyError::DeviceUnavailable)) => { - if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { - self.spotify.active_backend = None; + } + Err(SpotifyError::Unauthorized) => { + if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { + self.spotify.active_backend = None; + } + self.spotify.player_status = SpotifyPlayerStatus::Error(crate::i18n::t( + "integrations.spotify.error.generic", + )); + tracing::warn!("spotify command: token expirado, renovando"); + self.spotify.token_refreshed_at = + Some(std::time::Instant::now() - std::time::Duration::from_secs(60 * 60)); + } + Err(SpotifyError::DeviceUnavailable) => { + if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { + self.spotify.active_backend = None; + } + if let Some(dead_id) = self.spotify.active_device_id.take() { + tracing::warn!( + device_id = dead_id, + "spotify command: device did not respond, evicting it" + ); + self.spotify.failed_device_ids.insert(dead_id.clone()); + self.spotify + .devices + .retain(|d| d.id.as_deref() != Some(&dead_id)); + } + self.spotify.active_device_id = resolve_active_spotify_device( + &self.spotify.devices, + self.spotify.active_device_id.as_deref(), + ); + let message = crate::i18n::t("integrations.spotify.error.device_gone"); + self.spotify.player_status = SpotifyPlayerStatus::Error(message.clone()); + self.notify_error(format!("Spotify: {message}")); + self.fetch_spotify_devices(); + } + Err(e) => { + tracing::warn!("spotify command: {e}"); + if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { + self.spotify.active_backend = None; + } + self.spotify.player_status = SpotifyPlayerStatus::Error(crate::i18n::t( + "integrations.spotify.error.generic", + )); + } + } + } + + pub fn poll_spotify_remote_skip_result(&mut self) { + let Some(rx) = self.spotify.remote_skip_result_rx.take() else { + return; + }; + match rx.try_recv() { + Ok(outcome) => { + self.spotify.remote_skip_task = None; + if !self + .spotify + .remote_skip_queue + .complete(outcome.id, &outcome.device_id) + { + tracing::debug!( + operation_id = outcome.id, + device_id = outcome.device_id, + "spotify remote skip: ignoring stale result" + ); + return; + } + match self.spotify_control_target() { + SpotifyControlTarget::Remote { device_id, .. } + if device_id == outcome.device_id => + { + let unauthorized = matches!( + &outcome.result, + Err(crate::integrations::spotify::SpotifyError::Unauthorized) + ); + if unauthorized { + self.spotify.remote_skip_paused_for_token_refresh = true; + } + self.handle_spotify_remote_command_result(outcome.result); + self.start_playback_polling(); + if !unauthorized { + self.start_next_spotify_remote_skip(); + } } - if let Some(dead_id) = self.spotify.active_device_id.take() { - tracing::warn!( - device_id = dead_id, - "spotify play_on_device: device did not respond, evicting it" + SpotifyControlTarget::Remote { .. } => { + tracing::debug!( + operation_id = outcome.id, + device_id = outcome.device_id, + "spotify remote skip: device changed, ignoring result state" ); - self.spotify.failed_device_ids.insert(dead_id.clone()); - self.spotify - .devices - .retain(|d| d.id.as_deref() != Some(&dead_id)); + self.start_playback_polling(); + self.start_next_spotify_remote_skip(); } - self.spotify.active_device_id = resolve_active_spotify_device( - &self.spotify.devices, - self.spotify.active_device_id.as_deref(), - ); - let message = crate::i18n::t("integrations.spotify.error.device_gone"); - self.spotify.player_status = SpotifyPlayerStatus::Error(message.clone()); - self.notify_error(format!("Spotify: {message}")); - self.fetch_spotify_devices(); - } - Ok(Err(e)) => { - tracing::warn!("spotify play_on_device: {e}"); - if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { - self.spotify.active_backend = None; + SpotifyControlTarget::Native | SpotifyControlTarget::None => { + tracing::debug!( + operation_id = outcome.id, + device_id = outcome.device_id, + "spotify remote skip: remote control is no longer active" + ); + self.cancel_spotify_remote_skips(); } - self.spotify.player_status = SpotifyPlayerStatus::Error(crate::i18n::t( - "integrations.spotify.error.generic", - )); - } - Err(std::sync::mpsc::TryRecvError::Empty) => { - self.spotify.play_result_rx = Some(rx); } - Err(std::sync::mpsc::TryRecvError::Disconnected) => {} + } + Err(std::sync::mpsc::TryRecvError::Empty) => { + self.spotify.remote_skip_result_rx = Some(rx); + } + Err(std::sync::mpsc::TryRecvError::Disconnected) => { + self.spotify.remote_skip_task = None; + tracing::warn!("spotify remote skip: result channel disconnected"); + self.spotify.remote_skip_queue.abandon_current(); + self.start_next_spotify_remote_skip(); } } } @@ -960,6 +1040,7 @@ impl App { abort_task(&mut self.spotify.playback_task); self.spotify.playback_rx = None; self.spotify.playback = None; + self.cancel_spotify_remote_skips(); if self.spotify.active_backend == Some(SpotifyPlaybackBackend::Remote) { self.spotify.active_backend = None; } @@ -1126,24 +1207,74 @@ impl App { self.spotify.player_status = SpotifyPlayerStatus::Loading; } - fn spotify_remote_skip(&mut self, token: String, device_id: String, forward: bool) { + fn spotify_remote_skip(&mut self, device_id: String, direction: SpotifyRemoteSkipDirection) { + self.spotify.remote_skip_queue.enqueue(device_id, direction); + self.start_next_spotify_remote_skip(); + } + + pub(super) fn cancel_spotify_remote_skips(&mut self) { + abort_task(&mut self.spotify.remote_skip_task); + self.spotify.remote_skip_queue.invalidate(); + self.spotify.remote_skip_paused_for_token_refresh = false; + self.spotify.remote_skip_result_rx = None; + } + + fn start_next_spotify_remote_skip(&mut self) { + if self.spotify.remote_skip_paused_for_token_refresh { + return; + } + let (token, active_device_id) = match self.spotify_control_target() { + SpotifyControlTarget::Remote { token, device_id } => (token, device_id), + SpotifyControlTarget::Native | SpotifyControlTarget::None => { + self.cancel_spotify_remote_skips(); + return; + } + }; + let operation = loop { + let Some(operation) = self.spotify.remote_skip_queue.begin_next() else { + return; + }; + if active_device_id == operation.device_id { + break operation; + } + tracing::debug!( + operation_id = operation.id, + device_id = operation.device_id, + "spotify remote skip: device changed, dropping queued operation" + ); + self.spotify + .remote_skip_queue + .complete(operation.id, &operation.device_id); + }; let (tx, rx) = std::sync::mpsc::channel(); - self.spotify.play_result_rx = Some(rx); - tokio::spawn(async move { - let result = if forward { - crate::integrations::spotify::devices::next_track(&token, &device_id).await - } else { - crate::integrations::spotify::devices::previous_track(&token, &device_id).await + self.spotify.remote_skip_result_rx = Some(rx); + let handle = tokio::spawn(async move { + let result = match operation.direction { + SpotifyRemoteSkipDirection::Next => { + crate::integrations::spotify::devices::next_track(&token, &operation.device_id) + .await + } + SpotifyRemoteSkipDirection::Previous => { + crate::integrations::spotify::devices::previous_track( + &token, + &operation.device_id, + ) + .await + } }; - let _ = tx.send(result); + let _ = tx.send(SpotifyRemoteSkipResult { + id: operation.id, + device_id: operation.device_id, + result, + }); }); - self.start_playback_polling(); + self.spotify.remote_skip_task = Some(handle); } pub(super) async fn spotify_play_next(&mut self) { match self.spotify_control_target() { - super::SpotifyControlTarget::Remote { token, device_id } => { - self.spotify_remote_skip(token, device_id, true); + super::SpotifyControlTarget::Remote { device_id, .. } => { + self.spotify_remote_skip(device_id, SpotifyRemoteSkipDirection::Next); } super::SpotifyControlTarget::Native => self.native_next(), super::SpotifyControlTarget::None => { @@ -1154,8 +1285,8 @@ impl App { pub(super) async fn spotify_play_previous(&mut self) { match self.spotify_control_target() { - super::SpotifyControlTarget::Remote { token, device_id } => { - self.spotify_remote_skip(token, device_id, false); + super::SpotifyControlTarget::Remote { device_id, .. } => { + self.spotify_remote_skip(device_id, SpotifyRemoteSkipDirection::Previous); } super::SpotifyControlTarget::Native => self.native_prev(), super::SpotifyControlTarget::None => { @@ -1680,6 +1811,162 @@ mod tests { } } + #[tokio::test] + async fn stopping_remote_polling_cancels_current_and_queued_skips() { + let mut app = App::new().await; + app.spotify + .remote_skip_queue + .enqueue("device".to_string(), SpotifyRemoteSkipDirection::Next); + app.spotify + .remote_skip_queue + .enqueue("device".to_string(), SpotifyRemoteSkipDirection::Previous); + let operation = app + .spotify + .remote_skip_queue + .begin_next() + .expect("remote skip starts"); + let (tx, rx) = std::sync::mpsc::channel(); + app.spotify.remote_skip_result_rx = Some(rx); + let task = tokio::spawn(async { std::future::pending::<()>().await }); + let abort_handle = task.abort_handle(); + app.spotify.remote_skip_task = Some(task); + + app.stop_playback_polling(); + tokio::task::yield_now().await; + + assert!(abort_handle.is_finished()); + assert!(app.spotify.remote_skip_task.is_none()); + assert!(app.spotify.remote_skip_result_rx.is_none()); + assert!(app.spotify.remote_skip_queue.begin_next().is_none()); + assert!( + tx.send(SpotifyRemoteSkipResult { + id: operation.id, + device_id: operation.device_id, + result: Ok(()), + }) + .is_err(), + "remote skip result receiver was dropped" + ); + } + + #[tokio::test] + async fn remote_skip_result_from_previous_device_does_not_update_player_state() { + let mut app = App::new().await; + app.spotify.access_token = Some("token".to_string()); + app.spotify.active_device_id = Some("new-device".to_string()); + app.spotify.active_backend = Some(SpotifyPlaybackBackend::Remote); + app.spotify.player_status = SpotifyPlayerStatus::Loading; + + app.spotify + .remote_skip_queue + .enqueue("old-device".to_string(), SpotifyRemoteSkipDirection::Next); + let operation = app + .spotify + .remote_skip_queue + .begin_next() + .expect("skip starts"); + let (tx, rx) = std::sync::mpsc::channel(); + app.spotify.remote_skip_result_rx = Some(rx); + tx.send(SpotifyRemoteSkipResult { + id: operation.id, + device_id: operation.device_id, + result: Err(crate::integrations::spotify::SpotifyError::Unauthorized), + }) + .expect("receiver is alive"); + + app.poll_spotify_remote_skip_result(); + + assert_eq!( + app.spotify.active_backend, + Some(SpotifyPlaybackBackend::Remote) + ); + assert!(matches!( + app.spotify.player_status, + SpotifyPlayerStatus::Loading + )); + } + + #[tokio::test] + async fn unauthorized_remote_skip_waits_for_token_refresh_before_dispatching_next() { + let mut app = App::new().await; + app.config.spotify.playback_mode = SpotifyPlaybackMode::Remote; + app.spotify.access_token = Some("expired-token".to_string()); + app.spotify.active_device_id = Some("device".to_string()); + app.spotify.active_backend = Some(SpotifyPlaybackBackend::Remote); + app.spotify + .remote_skip_queue + .enqueue("device".to_string(), SpotifyRemoteSkipDirection::Next); + app.spotify + .remote_skip_queue + .enqueue("device".to_string(), SpotifyRemoteSkipDirection::Previous); + let operation = app + .spotify + .remote_skip_queue + .begin_next() + .expect("first skip starts"); + let (tx, rx) = std::sync::mpsc::channel(); + app.spotify.remote_skip_result_rx = Some(rx); + tx.send(SpotifyRemoteSkipResult { + id: operation.id, + device_id: operation.device_id, + result: Err(crate::integrations::spotify::SpotifyError::Unauthorized), + }) + .expect("receiver is alive"); + + app.poll_spotify_remote_skip_result(); + + assert!(app.spotify.remote_skip_paused_for_token_refresh); + assert!(app.spotify.remote_skip_task.is_none()); + assert!(app.spotify.remote_skip_result_rx.is_none()); + app.spotify_play_next().await; + assert!( + app.spotify.remote_skip_task.is_none(), + "new input must not resume the queue with the expired token" + ); + let pending = app + .spotify + .remote_skip_queue + .begin_next() + .expect("queued skip waits for refreshed credentials"); + assert_eq!(pending.direction, SpotifyRemoteSkipDirection::Previous); + } + + #[tokio::test] + async fn remote_skip_result_in_native_mode_does_not_restart_polling_or_dispatch_queue() { + let mut app = App::new().await; + app.config.spotify.playback_mode = SpotifyPlaybackMode::Native; + app.spotify.access_token = Some("token".to_string()); + app.spotify.active_device_id = Some("device".to_string()); + app.spotify.active_backend = Some(SpotifyPlaybackBackend::Native); + + app.spotify + .remote_skip_queue + .enqueue("device".to_string(), SpotifyRemoteSkipDirection::Next); + app.spotify + .remote_skip_queue + .enqueue("device".to_string(), SpotifyRemoteSkipDirection::Previous); + let operation = app + .spotify + .remote_skip_queue + .begin_next() + .expect("skip starts"); + let (tx, rx) = std::sync::mpsc::channel(); + app.spotify.remote_skip_result_rx = Some(rx); + tx.send(SpotifyRemoteSkipResult { + id: operation.id, + device_id: operation.device_id, + result: Ok(()), + }) + .expect("receiver is alive"); + + app.poll_spotify_remote_skip_result(); + + assert!(app.spotify.playback_task.is_none()); + assert!(app.spotify.playback_rx.is_none()); + assert!(app.spotify.remote_skip_result_rx.is_none()); + assert!(app.spotify.remote_skip_queue.begin_next().is_none()); + } + #[tokio::test] async fn poll_playlist_tracks_does_not_backfill_playlist_total_from_loaded_tracks() { let mut app = App::new().await; diff --git a/src/app/mod.rs b/src/app/mod.rs index cb3e065..b199bcb 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -1071,6 +1071,21 @@ mod tests { app.spotify.playback_task = Some(tokio::spawn(async { std::future::pending::<()>().await; })); + app.spotify.remote_skip_queue.enqueue( + "device".to_string(), + spotify_state::SpotifyRemoteSkipDirection::Next, + ); + app.spotify.remote_skip_queue.enqueue( + "device".to_string(), + spotify_state::SpotifyRemoteSkipDirection::Previous, + ); + let skip_operation = app + .spotify + .remote_skip_queue + .begin_next() + .expect("remote skip starts"); + let (skip_tx, skip_rx) = std::sync::mpsc::channel(); + app.spotify.remote_skip_result_rx = Some(skip_rx); app.set_spotify_playback_mode(SpotifyPlaybackMode::Native); @@ -1082,6 +1097,18 @@ mod tests { assert!(app.spotify.playback.is_none()); assert!(app.spotify.playback_rx.is_none()); assert!(app.spotify.playback_task.is_none()); + assert!(app.spotify.remote_skip_result_rx.is_none()); + assert!(app.spotify.remote_skip_queue.begin_next().is_none()); + assert!( + skip_tx + .send(spotify_state::SpotifyRemoteSkipResult { + id: skip_operation.id, + device_id: skip_operation.device_id, + result: Ok(()), + }) + .is_err(), + "skip result receiver was dropped" + ); } #[tokio::test] diff --git a/src/app/spotify_state.rs b/src/app/spotify_state.rs index 4552825..66442ac 100644 --- a/src/app/spotify_state.rs +++ b/src/app/spotify_state.rs @@ -21,6 +21,137 @@ type PlaylistsResultRx = std::sync::mpsc::Receiver, bool), SpotifyError>>; type AlbumsResultRx = std::sync::mpsc::Receiver, bool), SpotifyError>>; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum SpotifyRemoteSkipDirection { + Next, + Previous, +} + +#[derive(Debug)] +pub(super) struct SpotifyRemoteSkipOperation { + pub(super) id: u64, + pub(super) device_id: String, + pub(super) direction: SpotifyRemoteSkipDirection, +} + +#[derive(Debug)] +pub(super) struct SpotifyRemoteSkipResult { + pub(super) id: u64, + pub(super) device_id: String, + pub(super) result: Result<(), SpotifyError>, +} + +#[derive(Debug, PartialEq, Eq)] +struct SpotifyRemoteSkipInFlight { + id: u64, + device_id: String, +} + +#[derive(Debug, Default)] +pub(super) struct SpotifyRemoteSkipQueue { + next_id: u64, + pending: VecDeque, + in_flight: Option, +} + +impl SpotifyRemoteSkipQueue { + pub(super) fn enqueue( + &mut self, + device_id: String, + direction: SpotifyRemoteSkipDirection, + ) -> u64 { + let id = self.next_id; + self.next_id = self + .next_id + .checked_add(1) + .expect("Spotify remote skip operation ID exhausted"); + self.pending.push_back(SpotifyRemoteSkipOperation { + id, + device_id, + direction, + }); + id + } + + pub(super) fn begin_next(&mut self) -> Option { + if self.in_flight.is_some() { + return None; + } + let operation = self.pending.pop_front()?; + self.in_flight = Some(SpotifyRemoteSkipInFlight { + id: operation.id, + device_id: operation.device_id.clone(), + }); + Some(operation) + } + + pub(super) fn complete(&mut self, id: u64, device_id: &str) -> bool { + let is_current = self + .in_flight + .as_ref() + .is_some_and(|current| current.id == id && current.device_id == device_id); + if is_current { + self.in_flight = None; + } + is_current + } + + pub(super) fn abandon_current(&mut self) { + self.in_flight = None; + } + + pub(super) fn invalidate(&mut self) { + self.pending.clear(); + self.in_flight = None; + } +} + +#[cfg(test)] +mod remote_skip_queue_tests { + use super::*; + + #[test] + fn remote_skips_are_started_one_at_a_time_in_input_order() { + let mut queue = SpotifyRemoteSkipQueue::default(); + let first_id = queue.enqueue("device".to_string(), SpotifyRemoteSkipDirection::Next); + let second_id = queue.enqueue("device".to_string(), SpotifyRemoteSkipDirection::Previous); + + let first = queue.begin_next().expect("first skip starts"); + assert_eq!(first.id, first_id); + assert_eq!(first.direction, SpotifyRemoteSkipDirection::Next); + assert!(queue.begin_next().is_none(), "only one skip may run"); + + assert!(queue.complete(first.id, &first.device_id)); + let second = queue.begin_next().expect("second skip starts next"); + assert_eq!(second.id, second_id); + assert_eq!(second.direction, SpotifyRemoteSkipDirection::Previous); + } + + #[test] + fn stale_result_cannot_release_the_current_operation() { + let mut queue = SpotifyRemoteSkipQueue::default(); + queue.enqueue("device".to_string(), SpotifyRemoteSkipDirection::Next); + let current = queue.begin_next().expect("skip starts"); + + assert!(!queue.complete(current.id + 1, ¤t.device_id)); + assert!(!queue.complete(current.id, "different-device")); + assert!(queue.begin_next().is_none()); + assert!(queue.complete(current.id, ¤t.device_id)); + } + + #[test] + fn invalidating_remote_skips_drops_current_and_pending_operations() { + let mut queue = SpotifyRemoteSkipQueue::default(); + queue.enqueue("device".to_string(), SpotifyRemoteSkipDirection::Next); + queue.enqueue("device".to_string(), SpotifyRemoteSkipDirection::Previous); + queue.begin_next().expect("first skip starts"); + + queue.invalidate(); + + assert!(queue.begin_next().is_none()); + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(super) enum SpotifyPlaybackBackend { Remote, @@ -82,6 +213,10 @@ pub struct SpotifyState { Option>>, pub(super) play_result_rx: Option>>, + pub(super) remote_skip_queue: SpotifyRemoteSkipQueue, + pub(super) remote_skip_paused_for_token_refresh: bool, + pub(super) remote_skip_task: Option>, + pub(super) remote_skip_result_rx: Option>, pub(super) save_track_rx: Option>>, pub playback_queue: VecDeque, @@ -165,6 +300,7 @@ impl SpotifyState { abort(&mut self.search_more_task); abort(&mut self.devices_task); abort(&mut self.playback_task); + abort(&mut self.remote_skip_task); abort(&mut self.token_refresh_task); abort(&mut self.radio_task); abort(&mut self.liked_task); @@ -225,6 +361,10 @@ impl Default for SpotifyState { token_refresh_task: None, token_refresh_rx: None, play_result_rx: None, + remote_skip_queue: SpotifyRemoteSkipQueue::default(), + remote_skip_paused_for_token_refresh: false, + remote_skip_task: None, + remote_skip_result_rx: None, save_track_rx: None, playback_queue: VecDeque::new(), radio_queue: VecDeque::new(), diff --git a/src/main.rs b/src/main.rs index 6f692e6..a87999c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -194,6 +194,7 @@ async fn run(tui: &mut terminal::Tui) -> Result<()> { app.poll_spotify_auth(); app.poll_token_refresh(); app.poll_spotify_play_result(); + app.poll_spotify_remote_skip_result(); app.poll_spotify_search(); app.poll_spotify_search_more(); app.poll_spotify_player_events();