Custom playback integration phase 2: WIP

examples/app.toml

@@ -25,7 +25,7 @@ cover_img_length = 9
 cover_img_width = 5
 cover_img_pixels = 16
 seek_duration_secs = 5
-custom_queue = true
+custom_queue = false
 
 [device]
 name = "spotify-player"

spotify_player/src/cli/client.rs

@@ -463,7 +463,10 @@ async fn handle_playback_request(
             let client = client.clone();
             let state = state.clone();
             async move {
-                match client.handle_player_request(player_request, playback).await {
+                match client
+                    .handle_player_request(Some(&state), player_request, playback)
+                    .await
+                {
                     Ok(playback) => {
                         // update application's states
                         state.player.write().buffered_playback = playback;
@@ -480,7 +483,7 @@ async fn handle_playback_request(
     } else {
         // Handles the player request synchronously
         client
-            .handle_player_request(player_request, playback)
+            .handle_player_request(None, player_request, playback)
             .await?;
     }
     Ok(())

spotify_player/src/client/mod.rs

@@ -52,6 +52,9 @@ pub struct AppClient {
     user_client: Option<rspotify::AuthCodePkceSpotify>,
     #[cfg(feature = "streaming")]
     stream_conn: Arc<Mutex<Option<librespot_connect::Spirc>>>,
+    /// Sender half of the client request channel, used to send requests from
+    /// the streaming player-event task (e.g. batch transitions on `EndOfTrack`).
+    client_pub: Option<flume::Sender<ClientRequest>>,
 }
 
 impl Deref for AppClient {
@@ -108,9 +111,17 @@ impl AppClient {
 
             #[cfg(feature = "streaming")]
             stream_conn: Arc::new(Mutex::new(None)),
+            client_pub: None,
         })
     }
 
+    /// Set the client request sender. Must be called before starting a
+    /// streaming connection so that the player-event task can send
+    /// `ClientRequest`s (e.g. batch transitions).
+    pub fn set_client_pub(&mut self, sender: flume::Sender<ClientRequest>) {
+        self.client_pub = Some(sender);
+    }
+
     async fn token(&self) -> Result<String> {
         self.auto_reauth().await?;
         Ok(self
@@ -235,8 +246,13 @@ impl AppClient {
         session: librespot_core::Session,
         creds: librespot_core::authentication::Credentials,
     ) -> Result<()> {
+        let client_pub = self
+            .client_pub
+            .clone()
+            .context("client_pub must be set before creating a streaming connection")?;
         let new_conn =
-            crate::streaming::new_connection(self.clone(), state, session, creds).await?;
+            crate::streaming::new_connection(self.clone(), state, session, creds, client_pub)
+                .await?;
         let mut stream_conn = self.stream_conn.lock();
         // shutdown old streaming connection and replace it with a new connection
         if let Some(conn) = stream_conn.as_ref() {
@@ -251,6 +267,7 @@ impl AppClient {
     /// Handle a player request, return a new playback metadata on success
     pub async fn handle_player_request(
         &self,
+        state: Option<&SharedState>,
         request: PlayerRequest,
         mut playback: Option<PlaybackMetadata>,
     ) -> Result<Option<PlaybackMetadata>> {
@@ -261,6 +278,11 @@ impl AppClient {
                 // because `TransferPlayback` doesn't require an active playback
                 self.transfer_playback(&device_id, Some(force_play)).await?;
                 tracing::info!("Transferred playback to device with id={}", device_id);
+                // Transferring playback leaves the custom queue's context;
+                // clear it so stale state doesn't interfere.
+                if let Some(state) = state {
+                    state.player.write().custom_queue = None;
+                }
                 return Ok(None);
             }
             PlayerRequest::StartPlayback(p, shuffle) => {
@@ -401,7 +423,9 @@ impl AppClient {
             }
             ClientRequest::Player(request) => {
                 let playback = state.player.read().buffered_playback.clone();
-                let playback = self.handle_player_request(request, playback).await?;
+                let playback = self
+                    .handle_player_request(Some(state), request, playback)
+                    .await?;
                 state.player.write().buffered_playback = playback;
                 self.update_playback(state);
             }
@@ -586,7 +610,34 @@ impl AppClient {
             }
             ClientRequest::GetCurrentUserQueue => {
                 let queue = self.current_user_queue().await?;
-                state.player.write().queue = Some(queue);
+                let mut player = state.player.write();
+
+                // Queue consistency check — safety net that detects silent
+                // divergence between the custom queue and Spotify's actual
+                // queue state.  If the expected next track is missing from
+                // Spotify's queue, our local position is unreliable — clear
+                // the custom queue and fall back to Spotify-managed playback.
+                if let Some(custom_queue) = &player.custom_queue {
+                    if !custom_queue.in_batch_cooldown() {
+                        if let Some(expected) = custom_queue.expected_next_track() {
+                            let expected_uri = expected.uri();
+                            let found = queue
+                                .queue
+                                .iter()
+                                .any(|item| item.id().is_some_and(|id| id.uri() == expected_uri));
+                            if !found {
+                                tracing::warn!(
+                                    "Custom queue consistency check failed: \
+                                     expected next track {expected_uri} not found in Spotify queue; \
+                                     clearing custom queue."
+                                );
+                                player.custom_queue = None;
+                            }
+                        }
+                    }
+                }
+
+                player.queue = Some(queue);
             }
             ClientRequest::ReorderPlaylistItems {
                 playlist_id,
@@ -1616,6 +1667,36 @@ impl AppClient {
             player.playback = playback;
             player.playback_last_updated_time = Some(std::time::Instant::now());
 
+            // Clear custom_queue when there is no playback at all.
+            if player.playback.is_none() {
+                player.custom_queue = None;
+            } else if let Some(ref cq) = player.custom_queue {
+                // If Spotify reports a non-None context whose URI differs from
+                // the queue's source context, another context has taken over.
+                // context: None is expected for URIs playback — don't clear.
+                // Skip during cooldown after a batch transition to avoid
+                // premature clearing while Spotify's API catches up.
+                let in_cooldown = cq.in_batch_cooldown();
+                if !in_cooldown {
+                    if let Some(ref spotify_ctx) = player.playback.as_ref().unwrap().context {
+                        let queue_ctx_uri = cq.source_context().map(ContextId::uri);
+                        let spotify_uri = crate::utils::parse_uri(&spotify_ctx.uri);
+                        let matches = queue_ctx_uri
+                            .as_deref()
+                            .is_some_and(|u| u == spotify_uri.as_ref());
+                        if !matches {
+                            tracing::info!(
+                                "Spotify context changed (reported={}, expected={:?}); \
+                                 clearing custom queue",
+                                spotify_ctx.uri,
+                                queue_ctx_uri,
+                            );
+                            player.custom_queue = None;
+                        }
+                    }
+                }
+            }
+
             let curr_item = player.currently_playing();
 
             let curr_name = match curr_item {

spotify_player/src/config/mod.rs

@@ -390,7 +390,7 @@ impl Default for AppConfig {
             volume_scroll_step: 5,
             enable_mouse_scroll_volume: true,
 
-            custom_queue: true,
+            custom_queue: false,
         }
     }
 }

spotify_player/src/event/window.rs

@@ -1,5 +1,7 @@
 use super::page::handle_navigation_command;
 use super::*;
+#[cfg(feature = "streaming")]
+use crate::state::CustomQueue;
 use crate::{
     command::{
         construct_album_actions, construct_artist_actions, construct_playlist_actions,
@@ -258,6 +260,7 @@ fn handle_playlist_modify_command(
     Ok(false)
 }
 
+#[allow(clippy::needless_pass_by_value)]
 fn handle_command_for_track_table_window(
     command: Command,
     client_pub: &flume::Sender<ClientRequest>,
@@ -322,19 +325,55 @@ fn handle_command_for_track_table_window(
                 }
             }
 
-            let base_playback = match context_id {
-                None | Some(ContextId::Tracks(_)) => {
-                    Playback::URIs(tracks.iter().map(|t| t.id.clone().into()).collect(), None)
+            let configs = config::get_config();
+            let limit = configs.app_config.tracks_playback_limit;
+
+            // When streaming with the custom queue enabled and the context is a
+            // playlist/album/artist, build a CustomQueue that holds the **full**
+            // track list and always use URIs-based playback so the app controls
+            // the play order.
+            #[cfg(feature = "streaming")]
+            let use_custom_queue = state.should_use_custom_queue()
+                && matches!(
+                    context_id,
+                    Some(ContextId::Playlist(_) | ContextId::Album(_) | ContextId::Artist(_))
+                );
+            #[cfg(not(feature = "streaming"))]
+            let use_custom_queue = false;
+
+            let base_playback = if use_custom_queue {
+                Playback::URIs(tracks.iter().map(|t| t.id.clone().into()).collect(), None)
+            } else {
+                // Clear any stale custom queue when not using it.
+                state.player.write().custom_queue = None;
+                match &context_id {
+                    None | Some(ContextId::Tracks(_)) => {
+                        Playback::URIs(tracks.iter().map(|t| t.id.clone().into()).collect(), None)
+                    }
+                    Some(ContextId::Show(_)) => unreachable!(
+                        "show context should be handled by handle_command_for_episode_table_window"
+                    ),
+                    Some(context_id) => Playback::Context(context_id.clone(), None),
                 }
-                Some(ContextId::Show(_)) => unreachable!(
-                    "show context should be handled by handle_command_for_episode_table_window"
-                ),
-                Some(context_id) => Playback::Context(context_id, None),
             };
 
+            #[cfg(feature = "streaming")]
+            if use_custom_queue {
+                let track_ids: Vec<PlayableId<'static>> =
+                    tracks.iter().map(|t| t.id.clone().into()).collect();
+                let start_position = track_ids.iter().position(|t| t.uri() == uri).unwrap_or(0);
+                let queue = CustomQueue::new(
+                    track_ids,
+                    start_position,
+                    limit,
+                    context_id.clone(),
+                    false, // autoplay: wired up in Phase 5
+                );
+                state.player.write().custom_queue = Some(queue);
+            }
+
             client_pub.send(ClientRequest::Player(PlayerRequest::StartPlayback(
-                base_playback
-                    .uri_offset(uri, config::get_config().app_config.tracks_playback_limit),
+                base_playback.uri_offset(uri, limit),
                 None,
             )))?;
         }

spotify_player/src/main.rs

@@ -137,9 +137,10 @@ async fn start_app(state: &state::SharedState) -> Result<()> {
     }
 
     // create a Spotify API client
-    let client = client::AppClient::new()
+    let mut client = client::AppClient::new()
         .await
         .context("construct app client")?;
+    client.set_client_pub(client_pub.clone());
     client
         .new_session(Some(state), true)
         .await