Merge pull request 'Sonos queue resync + cold-start prefetcher gate' (#83) from dev into main

android/app/src/main/java/com/fabledsword/minstrel/player/AudioPrefetcher.kt

@@ -58,59 +58,98 @@ class AudioPrefetcher @Inject constructor(
     private val activeJobs = mutableMapOf<String, Job>()
     private val mutex = Mutex()
 
+    private data class ReconcileInput(
+        val queue: List<Pair<String, String>>,
+        val index: Int,
+        val window: Int,
+        val isPlaying: Boolean,
+    )
+
     init {
         scope.launch {
             combine(
                 playerController.uiState.map { it.queue.map { t -> t.id to t.streamUrl } },
                 playerController.uiState.map { it.queueIndex },
                 authStore.cacheSettings.map { it.prefetchWindow },
-            ) { queue, index, window -> Triple(queue, index, window) }
+                playerController.uiState.map { it.isPlaying },
+            ) { queue, index, window, isPlaying ->
+                ReconcileInput(queue, index, window, isPlaying)
+            }
                 .distinctUntilChanged()
-                .collect { (queue, index, window) -> reconcile(queue, index, window) }
+                .collect { input ->
+                    reconcile(input.queue, input.index, input.window, input.isPlaying)
+                }
         }
     }
 
+    /**
+     * Apply the prefetch window to the current queue.
+     *
+     * Cancellation of out-of-window jobs always runs -- a queue mutation
+     * or skip should free bandwidth from stale prefetches immediately.
+     * Starting new prefetches is gated on [isPlaying]: until the current
+     * track is actually playing, every byte of upstream bandwidth should
+     * land on it, not on upcoming-track prefetches. Without this gate a
+     * cold start fanned out 4-6 concurrent CacheWriter jobs against the
+     * same OkHttp client as the playback DataSource and the user waited
+     * ~25 s for the first audio to start; with the gate the current
+     * track gets the full pipe to its first STATE_READY, then the
+     * prefetcher fills in the next window.
+     */
     private suspend fun reconcile(
         queue: List<Pair<String, String>>,
         index: Int,
         window: Int,
+        isPlaying: Boolean,
     ) {
         mutex.withLock {
-            if (index < 0 || queue.isEmpty() || window <= 0) {
+            val targets = computeTargets(queue, index, window)
+            if (targets.isEmpty()) {
                 cancelAllLocked()
                 return
             }
-            // Exclude the currently-playing track (it's loaded by the
-            // player itself) and walk `window` tracks forward.
-            val firstIdx = (index + 1).coerceAtMost(queue.size)
-            val lastIdx = (index + window).coerceAtMost(queue.size - 1)
-            if (firstIdx > lastIdx) {
-                cancelAllLocked()
-                return
-            }
-            val targets = queue.subList(firstIdx, lastIdx + 1)
             val targetIds = targets.mapTo(mutableSetOf()) { it.first }
+            // Cancellation always runs so a queue mutation or skip frees
+            // the pipe immediately, even while paused.
+            cancelOutOfWindowLocked(targetIds)
+            if (isPlaying) startInWindowLocked(targets)
+        }
+    }
 
-            // Cancel jobs for tracks that have slid out of the window.
+    private fun computeTargets(
-            activeJobs.entries
+        queue: List<Pair<String, String>>,
-                .filter { it.key !in targetIds }
+        index: Int,
-                .toList()
+        window: Int,
-                .forEach { (id, job) ->
+    ): List<Pair<String, String>> {
-                    job.cancel()
+        // Exclude the currently-playing track (it's loaded by the player
-                    activeJobs.remove(id)
+        // itself) and walk `window` tracks forward.
-                }
+        val firstIdx = index + 1
+        val lastIdx = (index + window).coerceAtMost(queue.size - 1)
+        val isValid = index >= 0 && queue.isNotEmpty() && window > 0 && firstIdx <= lastIdx
+        return if (isValid) queue.subList(firstIdx, lastIdx + 1) else emptyList()
+    }
 
-            // Start prefetches for new arrivals. Skip blank URLs (these
+    private fun cancelOutOfWindowLocked(targetIds: Set<String>) {
-            // come from minimal TrackRefs synthesized from playlist rows
+        activeJobs.entries
-            // when the upstream track was removed from the library).
+            .filter { it.key !in targetIds }
-            for ((trackId, streamUrl) in targets) {
+            .toList()
-                if (trackId in activeJobs || streamUrl.isBlank()) continue
+            .forEach { (id, job) ->
-                val job = scope.launch(Dispatchers.IO) {
+                job.cancel()
-                    runCatching { prefetchOne(trackId, streamUrl) }
+                activeJobs.remove(id)
-                    mutex.withLock { activeJobs.remove(trackId) }
-                }
-                activeJobs[trackId] = job
             }
+    }
+
+    private fun startInWindowLocked(targets: List<Pair<String, String>>) {
+        // Skip blank URLs (these come from minimal TrackRefs synthesized
+        // from playlist rows when the upstream track was removed from
+        // the library).
+        for ((trackId, streamUrl) in targets) {
+            if (trackId in activeJobs || streamUrl.isBlank()) continue
+            val job = scope.launch(Dispatchers.IO) {
+                runCatching { prefetchOne(trackId, streamUrl) }
+                mutex.withLock { activeJobs.remove(trackId) }
+            }
+            activeJobs[trackId] = job
         }
     }
 

android/app/src/main/java/com/fabledsword/minstrel/player/MinstrelForwardingPlayer.kt

@@ -8,6 +8,7 @@ import androidx.lifecycle.DefaultLifecycleObserver
 import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.ProcessLifecycleOwner
 import androidx.media3.common.ForwardingPlayer
+import androidx.media3.common.MediaItem
 import androidx.media3.common.Player
 import com.fabledsword.minstrel.player.output.ActiveUpnp
 import com.fabledsword.minstrel.player.output.ActiveUpnpHolder
@@ -124,6 +125,43 @@ class MinstrelForwardingPlayer(
     private fun isLoadingUpnp(): Boolean =
         holder.target.value != null && holder.active.value == null
 
+    // ─── setMediaItems intercepts ──────────────────────────────────────
+    // When PlayerController.setQueue replaces the queue while Sonos is the
+    // active route, the wrapped delegate's queue gets the new items but
+    // Sonos's native queue still holds the OLD tracks -- and the play()
+    // that PlayerController fires immediately after setMediaItems would
+    // resume the old Sonos queue (user reported on-device: "player view
+    // updates but Sonos queue does not"). We clear active + set target
+    // synchronously here so the next play() in the same IPC sequence
+    // drops via isLoadingUpnp() = true; the OutputPickerController
+    // observes the uiState.queue change and runs the resync (re-clears
+    // Sonos's native queue + AddURIToQueue the new tracks + Play).
+
+    override fun setMediaItems(mediaItems: List<MediaItem>) {
+        super.setMediaItems(mediaItems)
+        markPendingResyncIfRemote()
+    }
+
+    override fun setMediaItems(mediaItems: List<MediaItem>, resetPosition: Boolean) {
+        super.setMediaItems(mediaItems, resetPosition)
+        markPendingResyncIfRemote()
+    }
+
+    override fun setMediaItems(mediaItems: List<MediaItem>, startIndex: Int, startPositionMs: Long) {
+        super.setMediaItems(mediaItems, startIndex, startPositionMs)
+        markPendingResyncIfRemote()
+    }
+
+    private fun markPendingResyncIfRemote() {
+        val wasActive = holder.active.value ?: return
+        Timber.w(
+            "setMediaItems while UPnP active (%s) -- marking pending resync",
+            wasActive.routeName,
+        )
+        holder.set(null)
+        holder.setTarget(wasActive.routeId)
+    }
+
     override fun play() {
         if (isLoadingUpnp()) {
             Timber.w("ForwardingPlayer.play() dropped -- UPnP loading")

android/app/src/main/java/com/fabledsword/minstrel/player/output/OutputPickerController.kt

@@ -139,6 +139,14 @@ class OutputPickerController @Inject constructor(
         ) = refresh()
     }
 
+    // Last-synced queue identity. Used by observeQueueChangesForSonosResync
+    // to detect when the user has mutated the queue (full replacement,
+    // playNext insert, or radio-append) while UPnP is active and apply the
+    // minimum-incremental set of Sonos SOAP operations to bring its native
+    // queue back in sync. Stored as the full id list (not a join-key) so we
+    // can run the longest-common-prefix / common-suffix diff.
+    private var lastSyncedQueueIds: List<String>? = null
+
     init {
         // Two-arg addCallback registers with no discovery flag —
         // androidx.mediarouter 1.7.0's default passive behavior:
@@ -154,6 +162,206 @@ class OutputPickerController @Inject constructor(
         scope.launch {
             playerFactory.dropEvents.collect { handleRemoteDrop() }
         }
+        scope.launch { observeQueueChangesForSonosResync() }
+    }
+
+    /**
+     * When the user plays a different playlist while Sonos is active,
+     * PlayerController.setQueue replaces the local queue but Sonos's
+     * native queue still holds the OLD tracks. MinstrelForwardingPlayer's
+     * setMediaItems override clears holder.active + sets target so the
+     * imminent play() call drops (drops via isLoadingUpnp() = true). Then
+     * this collector observes the uiState.queue change and re-runs
+     * loadQueueOnSonos to push the new tracks to Sonos.
+     *
+     * Discrimination: selectUpnp's initial-load path doesn't change
+     * uiState.queue (the queue was already populated before route
+     * selection), so this collector doesn't fire during that window. Only
+     * a fresh setQueue from PlayerController bumps the joined-ids key.
+     */
+    private suspend fun observeQueueChangesForSonosResync() {
+        playerController.uiState.collect { state ->
+            val newIds = state.queue.map { it.id }
+            val oldIds = lastSyncedQueueIds
+            if (newIds == oldIds) return@collect
+            lastSyncedQueueIds = newIds
+            // Route can be in target (setMediaItems-induced clearing already
+            // ran in ForwardingPlayer) OR in active (queue changed via
+            // addMediaItem / removeMediaItems etc. which don't hit the
+            // markPending hook).
+            val routeId = activeUpnpHolder.target.value
+                ?: activeUpnpHolder.active.value?.routeId
+                ?: return@collect
+            if (state.queue.isEmpty()) {
+                Timber.w("Sonos resync skipped: empty queue (clearing target)")
+                activeUpnpHolder.setTarget(null)
+                return@collect
+            }
+            scope.launch {
+                resyncSonosQueue(routeId, oldIds.orEmpty(), state.queue, state.queueIndex)
+            }
+        }
+    }
+
+    /**
+     * Bring Sonos's native queue back in sync with the local queue after a
+     * mutation. Tries an incremental SOAP diff first (RemoveTrackRangeFromQueue
+     * + AddURIToQueue at the insertion point) so playback continues without
+     * interruption -- that's what playNext / radio-append want. Falls back to
+     * the full removeAllTracks + reload path when the diff implies the current
+     * Sonos track was deleted (e.g. user switched playlists), which is what
+     * the user-reported "Sonos queue does not update" bug needed.
+     */
+    private suspend fun resyncSonosQueue(
+        routeId: String,
+        oldIds: List<String>,
+        newQueue: List<TrackRef>,
+        newCurrentIndex: Int,
+    ) = selectUpnpMutex.withLock {
+        val upnpRoute = upnpDiscovery.routes.value.firstOrNull { it.id == routeId }
+        val transport = upnpDiscovery.transportFor(routeId)
+        if (upnpRoute == null || transport == null) {
+            Timber.w(
+                "Sonos resync: route or transport gone for %s, dropping to local",
+                routeId,
+            )
+            activeUpnpHolder.setTarget(null)
+            selectedUpnpRouteIdInternal.value = null
+            return@withLock
+        }
+        val handledIncrementally = runCatching {
+            tryIncrementalResync(transport, oldIds, newQueue)
+        }.getOrElse { e ->
+            Timber.w(e, "Sonos incremental resync errored; falling back to full reload")
+            false
+        }
+        if (handledIncrementally) {
+            // Active was never cleared on the incremental path; clear any
+            // target that markPendingResyncIfRemote set (it didn't, for
+            // incremental cases that don't go through setMediaItems, but
+            // belt-and-suspenders).
+            activeUpnpHolder.setTarget(null)
+            return@withLock
+        }
+        // Full rebuild: ensure active is cleared so transport calls drop
+        // (markPendingResyncIfRemote may already have done this on the
+        // setMediaItems path).
+        if (activeUpnpHolder.active.value != null) {
+            activeUpnpHolder.set(null)
+            activeUpnpHolder.setTarget(routeId)
+        }
+        val outputRoute = OutputRoute.fromUpnpRoute(upnpRoute)
+        val rendering = renderingClientFor(routeId)
+        Timber.w("Sonos resync: full reload of %d tracks on %s", newQueue.size, outputRoute.name)
+        runCatching {
+            loadQueueOnSonos(transport, outputRoute, newQueue, newCurrentIndex)
+            activeUpnpHolder.set(
+                ActiveUpnp(
+                    routeId = routeId,
+                    routeName = outputRoute.name,
+                    avTransport = transport,
+                    rendering = rendering,
+                ),
+            )
+            activeUpnpHolder.setTarget(null)
+        }.onFailure { e ->
+            Timber.w(e, "Sonos resync (full) failed -- dropping to local")
+            activeUpnpHolder.setTarget(null)
+            selectedUpnpRouteIdInternal.value = null
+        }
+    }
+
+    /**
+     * Diff-based incremental Sonos queue sync. Returns true when the new
+     * queue can be produced from the old one with a remove-then-insert at
+     * the same middle slice -- the common-prefix and common-suffix portions
+     * stay untouched, and the current Sonos track must lie in the preserved
+     * prefix (otherwise the diff would orphan playback). Returns false to
+     * signal the caller to fall back to a full reload.
+     */
+    private suspend fun tryIncrementalResync(
+        transport: AVTransportClient,
+        oldIds: List<String>,
+        newQueue: List<TrackRef>,
+    ): Boolean {
+        val newIds = newQueue.map { it.id }
+        if (oldIds == newIds) return true
+        val prefixLen = commonPrefixLength(oldIds, newIds)
+        val suffixLen = commonSuffixLength(
+            oldIds.subList(prefixLen, oldIds.size),
+            newIds.subList(prefixLen, newIds.size),
+        )
+        val removedCount = oldIds.size - prefixLen - suffixLen
+        val addedCount = newIds.size - prefixLen - suffixLen
+        // Sonos's current track number is 1-based; compare against the
+        // preserved-prefix range as 0-based. If the current track is in
+        // the removed slice, incremental can't preserve playback -- caller
+        // falls back to full rebuild.
+        val currentSonosIdx0 = remoteState.trackNumber - 1
+        val canApply = currentSonosIdx0 in 0 until prefixLen
+        if (canApply) {
+            applyQueueDiff(transport, newQueue, prefixLen, removedCount, addedCount)
+        } else {
+            Timber.w(
+                "Sonos incremental: current track %d not in preserved prefix [0,%d); full rebuild",
+                currentSonosIdx0,
+                prefixLen,
+            )
+        }
+        return canApply
+    }
+
+    private suspend fun applyQueueDiff(
+        transport: AVTransportClient,
+        newQueue: List<TrackRef>,
+        prefixLen: Int,
+        removedCount: Int,
+        addedCount: Int,
+    ) {
+        if (removedCount > 0) {
+            Timber.w(
+                "Sonos incremental: RemoveTrackRangeFromQueue start=%d count=%d",
+                prefixLen + 1,
+                removedCount,
+            )
+            transport.removeTrackRangeFromQueue(
+                startingIndex = prefixLen + 1,
+                numberOfTracks = removedCount,
+            )
+        }
+        if (addedCount == 0) return
+        Timber.w(
+            "Sonos incremental: AddURIToQueue x%d starting at position %d",
+            addedCount,
+            prefixLen + 1,
+        )
+        for (i in 0 until addedCount) {
+            val ref = newQueue[prefixLen + i]
+            val token = streamTokens.mint(ref.id)
+            transport.addURIToQueue(
+                uri = token.url,
+                mime = token.mime,
+                title = token.title,
+                enqueuedURIPosition = prefixLen + i + 1,
+            )
+            if (i > 0) delay(EXTEND_THROTTLE_MS)
+        }
+    }
+
+    private fun commonPrefixLength(a: List<String>, b: List<String>): Int {
+        val limit = minOf(a.size, b.size)
+        for (i in 0 until limit) {
+            if (a[i] != b[i]) return i
+        }
+        return limit
+    }
+
+    private fun commonSuffixLength(a: List<String>, b: List<String>): Int {
+        val limit = minOf(a.size, b.size)
+        for (i in 0 until limit) {
+            if (a[a.size - 1 - i] != b[b.size - 1 - i]) return i
+        }
+        return limit
     }
 
     /**

android/app/src/main/java/com/fabledsword/minstrel/player/output/upnp/AVTransportClient.kt

@@ -62,6 +62,31 @@ class AVTransportClient(
         )
     }
 
+    /**
+     * Remove a contiguous range of tracks from the renderer's native queue.
+     * Sonos-specific extension to AVTransport; `UpdateID=0` skips the queue-
+     * version check so this works without first calling GetQueue to learn
+     * the current update id.
+     *
+     * [startingIndex] is 1-based per Sonos convention; [numberOfTracks] is
+     * the count to remove. Used by OutputPickerController's incremental
+     * queue resync path (radio-append: remove tail, then AddURIToQueue
+     * the new items).
+     */
+    suspend fun removeTrackRangeFromQueue(startingIndex: Int, numberOfTracks: Int) {
+        soap.call(
+            controlUrl = controlUrl,
+            serviceType = SERVICE_TYPE,
+            action = "RemoveTrackRangeFromQueue",
+            args = mapOf(
+                "InstanceID" to "0",
+                "UpdateID" to "0",
+                "StartingIndex" to startingIndex.toString(),
+                "NumberOfTracks" to numberOfTracks.toString(),
+            ),
+        )
+    }
+
     /**
      * Append a track to the renderer's queue. Sonos returns the assigned
      * track number + new total queue length in the response, but we don't