Impl/wear playlist transfer

app/src/main/java/com/theveloper/pixelplay/data/service/wear/PhoneDirectWatchTransferCoordinator.kt

@@ -98,6 +98,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         transferMode: String = WearTransferRequest.MODE_SAVE_TO_LIBRARY,
         startPositionMs: Long = 0L,
         autoPlay: Boolean = false,
+        overrideAudioFile: File? = null,
     ) {
         transferStateStore.markRequested(
             requestId = requestId,
@@ -112,6 +113,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
                 transferMode = transferMode,
                 startPositionMs = startPositionMs,
                 autoPlay = autoPlay,
+                overrideAudioFile = overrideAudioFile,
             )
         }
     }
@@ -123,6 +125,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         transferMode: String,
         startPositionMs: Long,
         autoPlay: Boolean,
+        overrideAudioFile: File? = null,
     ) {
         var openedSongSource: OpenedSongSource? = null
         try {
@@ -150,10 +153,19 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
                 return
             }
 
-            val songSource = openSongSource(
-                song = song,
-                allowProxyStreaming = transferMode == WearTransferRequest.MODE_TEMPORARY_PLAYBACK,
-            )
+            val songSource = if (overrideAudioFile != null) {
+                runCatching {
+                    OpenedSongSource(
+                        inputStream = overrideAudioFile.inputStream(),
+                        fileSize = overrideAudioFile.length(),
+                    )
+                }.getOrNull()
+            } else {
+                openSongSource(
+                    song = song,
+                    allowProxyStreaming = transferMode == WearTransferRequest.MODE_TEMPORARY_PLAYBACK,
+                )
+            }
             if (songSource == null) {
                 sendTransferMetadataError(
                     nodeId = nodeId,
@@ -182,9 +194,9 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
                 album = song.album,
                 albumId = song.albumId,
                 duration = song.duration,
-                mimeType = song.mimeType ?: "audio/mpeg",
+                mimeType = if (overrideAudioFile != null) TRANSCODED_MIME_TYPE else (song.mimeType ?: "audio/mpeg"),
                 fileSize = fileSize,
-                bitrate = song.bitrate ?: 0,
+                bitrate = if (overrideAudioFile != null) WatchAudioTranscoder.TARGET_BITRATE_BPS else (song.bitrate ?: 0),
                 sampleRate = song.sampleRate ?: 0,
                 isFavorite = song.isFavorite,
                 paletteSeedArgb = paletteSeedArgb,
@@ -920,20 +932,23 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         status: String,
         error: String? = null,
     ) {
+        // Local bookkeeping only: a phone-side "completed" just means the phone finished
+        // writing its own stream. The watch still has to validate the file and ack it before
+        // this song is trustworthy as "present on watch" — see WearCommandReceiver's handling
+        // of the watch's own TRANSFER_PROGRESS reply for where STATUS_COMPLETED actually lands.
+        val localStatus = if (status == WearTransferProgress.STATUS_COMPLETED) {
+            WearTransferProgress.STATUS_AWAITING_WATCH_ACK
+        } else {
+            status
+        }
         transferStateStore.markProgress(
             requestId = requestId,
             songId = songId,
             bytesTransferred = bytesTransferred,
             totalBytes = totalBytes,
-            status = status,
+            status = localStatus,
             error = error,
         )
-        if (status == WearTransferProgress.STATUS_COMPLETED) {
-            transferStateStore.markSongPresentOnWatch(
-                nodeId = nodeId,
-                songId = songId,
-            )
-        }
         val progress = WearTransferProgress(
             requestId = requestId,
             songId = songId,
@@ -961,5 +976,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         const val TRANSFER_ARTWORK_QUALITY = 95
         const val TRANSFER_ARTWORK_MAX_BYTES = 1_500_000
         const val METADATA_GUARD_DELAY_MS = 250L
+        // Kept in sync with WatchAudioTranscoder's AAC-LC output.
+        const val TRANSCODED_MIME_TYPE = "audio/mp4"
     }
 }

app/src/main/java/com/theveloper/pixelplay/data/service/wear/PhoneWatchTransferStateStore.kt

@@ -23,6 +23,7 @@ data class PhoneWatchTransferState(
     val totalBytes: Long = 0L,
     val status: String = WearTransferProgress.STATUS_TRANSFERRING,
     val error: String? = null,
+    val errorCode: String? = null,
     val updatedAtMillis: Long = System.currentTimeMillis(),
 ) {
     val progress: Float
@@ -33,6 +34,35 @@ data class PhoneWatchTransferState(
         }
 }
 
+/**
+ * Aggregate state of a whole-playlist transfer. [activeRequestId] is the requestId of the song
+ * currently in flight (if any) — the UI uses it to avoid showing that same song a second time as
+ * an individual [PhoneWatchTransferState] row while its batch row is already visible.
+ */
+data class PhoneWatchBatchTransferState(
+    val batchId: String,
+    val playlistId: String,
+    val playlistName: String,
+    val totalSongs: Int,
+    val completedSongs: Int = 0,
+    val failedSongCount: Int = 0,
+    val lastFailureErrorCode: String? = null,
+    val activeRequestId: String? = null,
+    val currentSongTitle: String = "",
+    val currentSongProgress: Float = 0f,
+    val status: String = WearTransferProgress.STATUS_TRANSFERRING,
+    val error: String? = null,
+    val updatedAtMillis: Long = System.currentTimeMillis(),
+) {
+    val overallProgress: Float
+        get() = if (totalSongs > 0) {
+            ((completedSongs.toFloat() + currentSongProgress.coerceIn(0f, 1f)) / totalSongs.toFloat())
+                .coerceIn(0f, 1f)
+        } else {
+            0f
+        }
+}
+
 @Singleton
 class PhoneWatchTransferStateStore @Inject constructor() {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
@@ -47,8 +77,14 @@ class PhoneWatchTransferStateStore @Inject constructor() {
     private val watchSongIdsByNodeId = ConcurrentHashMap<String, Set<String>>()
     private val _watchSongIds = MutableStateFlow<Set<String>>(emptySet())
     val watchSongIds: StateFlow<Set<String>> = _watchSongIds.asStateFlow()
+    private val freeStorageBytesByNodeId = ConcurrentHashMap<String, Long>()
+    private val _watchFreeStorageBytesByNodeId = MutableStateFlow<Map<String, Long>>(emptyMap())
+    val watchFreeStorageBytesByNodeId: StateFlow<Map<String, Long>> = _watchFreeStorageBytesByNodeId.asStateFlow()
+    private val _batchTransfers = MutableStateFlow<Map<String, PhoneWatchBatchTransferState>>(emptyMap())
+    val batchTransfers: StateFlow<Map<String, PhoneWatchBatchTransferState>> = _batchTransfers.asStateFlow()
 
     private val cleanupJobs = ConcurrentHashMap<String, Job>()
+    private val batchCleanupJobs = ConcurrentHashMap<String, Job>()
 
     fun markRequested(
         requestId: String,
@@ -108,6 +144,7 @@ class PhoneWatchTransferStateStore @Inject constructor() {
         totalBytes: Long,
         status: String,
         error: String? = null,
+        errorCode: String? = null,
         songTitle: String? = null,
     ) {
         _transfers.update { map ->
@@ -121,6 +158,7 @@ class PhoneWatchTransferStateStore @Inject constructor() {
                     totalBytes = maxOf(current.totalBytes, totalBytes),
                     status = status,
                     error = error,
+                    errorCode = errorCode,
                     updatedAtMillis = now,
                 )
             } else {
@@ -132,6 +170,7 @@ class PhoneWatchTransferStateStore @Inject constructor() {
                     totalBytes = totalBytes,
                     status = status,
                     error = error,
+                    errorCode = errorCode,
                     updatedAtMillis = now,
                 )
             }
@@ -164,6 +203,12 @@ class PhoneWatchTransferStateStore @Inject constructor() {
         updateWatchLibraryResolution()
     }
 
+    fun updateWatchFreeStorageBytes(nodeId: String, freeStorageBytes: Long) {
+        if (nodeId.isBlank()) return
+        freeStorageBytesByNodeId[nodeId] = freeStorageBytes
+        _watchFreeStorageBytesByNodeId.value = freeStorageBytesByNodeId.toMap()
+    }
+
     fun markSongPresentOnWatch(nodeId: String, songId: String) {
         if (nodeId.isBlank() || songId.isBlank()) return
         val existingSongIds = watchSongIdsByNodeId[nodeId].orEmpty()
@@ -172,6 +217,15 @@ class PhoneWatchTransferStateStore @Inject constructor() {
         _watchSongIds.value = watchSongIdsByNodeId.values.flatten().toSet()
     }
 
+    /** Corrects a wrongly-optimistic presence mark once the watch reports it never actually landed. */
+    fun clearSongPresentOnWatch(nodeId: String, songId: String) {
+        if (nodeId.isBlank() || songId.isBlank()) return
+        val existing = watchSongIdsByNodeId[nodeId] ?: return
+        if (songId !in existing) return
+        watchSongIdsByNodeId[nodeId] = existing - songId
+        _watchSongIds.value = watchSongIdsByNodeId.values.flatten().toSet()
+    }
+
     fun markCancelled(requestId: String, error: String? = null) {
         cleanupJobs.remove(requestId)?.cancel()
         _transfers.update { map ->
@@ -192,6 +246,12 @@ class PhoneWatchTransferStateStore @Inject constructor() {
                 watchSongIdsByNodeId.remove(nodeId)
             }
         }
+        freeStorageBytesByNodeId.keys.toList().forEach { nodeId ->
+            if (nodeId !in nodeIds) {
+                freeStorageBytesByNodeId.remove(nodeId)
+            }
+        }
+        _watchFreeStorageBytesByNodeId.value = freeStorageBytesByNodeId.toMap()
         _watchLibrarySyncedNodeIds.value = _watchLibrarySyncedNodeIds.value.intersect(nodeIds)
         _watchSongIds.value = watchSongIdsByNodeId.values.flatten().toSet()
         updateWatchLibraryResolution()
@@ -206,6 +266,135 @@ class PhoneWatchTransferStateStore @Inject constructor() {
         }
     }
 
+    fun markBatchStarted(batchId: String, playlistId: String, playlistName: String, totalSongs: Int) {
+        batchCleanupJobs.remove(batchId)?.cancel()
+        _batchTransfers.update { map ->
+            map + (batchId to PhoneWatchBatchTransferState(
+                batchId = batchId,
+                playlistId = playlistId,
+                playlistName = playlistName,
+                totalSongs = totalSongs,
+            ))
+        }
+    }
+
+    /**
+     * [initialProgress] lets a caller re-target [activeRequestId] (e.g. moving from a transcode
+     * request to the per-node transfer request for the same song) without snapping the visible
+     * progress back to 0 — see [PlaylistWatchTransferCoordinator]'s transcode/transfer weighting.
+     */
+    fun markBatchSongStarted(
+        batchId: String,
+        requestId: String,
+        songTitle: String,
+        initialProgress: Float = 0f,
+    ) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                activeRequestId = requestId,
+                currentSongTitle = songTitle,
+                currentSongProgress = initialProgress,
+                status = WearTransferProgress.STATUS_TRANSFERRING,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    fun markBatchSongProgress(batchId: String, status: String, progress: Float) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                status = status,
+                currentSongProgress = progress,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    fun markBatchSongCompleted(batchId: String) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                completedSongs = current.completedSongs + 1,
+                activeRequestId = null,
+                currentSongProgress = 0f,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    fun markBatchSongFailed(batchId: String, errorCode: String?) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                failedSongCount = current.failedSongCount + 1,
+                activeRequestId = null,
+                currentSongProgress = 0f,
+                lastFailureErrorCode = errorCode ?: current.lastFailureErrorCode,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    fun markBatchCompleted(batchId: String) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                status = WearTransferProgress.STATUS_COMPLETED,
+                activeRequestId = null,
+                currentSongProgress = 0f,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+        scheduleBatchTerminalCleanup(batchId)
+    }
+
+    fun markBatchCancelled(batchId: String) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                status = WearTransferProgress.STATUS_CANCELLED,
+                activeRequestId = null,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+        scheduleBatchTerminalCleanup(batchId)
+    }
+
+    fun markBatchFailed(batchId: String, error: String?) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                status = WearTransferProgress.STATUS_FAILED,
+                error = error,
+                activeRequestId = null,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+        scheduleBatchTerminalCleanup(batchId)
+    }
+
+    private fun scheduleBatchTerminalCleanup(batchId: String) {
+        batchCleanupJobs.remove(batchId)?.cancel()
+        batchCleanupJobs[batchId] = scope.launch {
+            delay(TERMINAL_STATE_VISIBILITY_MS)
+            _batchTransfers.update { map ->
+                val current = map[batchId]
+                if (current != null &&
+                    (current.status == WearTransferProgress.STATUS_COMPLETED ||
+                        current.status == WearTransferProgress.STATUS_FAILED ||
+                        current.status == WearTransferProgress.STATUS_CANCELLED)
+                ) {
+                    map - batchId
+                } else {
+                    map
+                }
+            }
+            batchCleanupJobs.remove(batchId)
+        }
+    }
+
     private fun updateWatchLibraryResolution() {
         val reachableNodeIds = _reachableWatchNodeIds.value
         _isWatchLibraryResolved.value = reachableNodeIds.isEmpty() ||