Wear playlist transfer — PR 1+2: contrato compartido + Room de playlists locales

app/src/main/java/com/theveloper/pixelplay/PixelPlayApplication.kt

@@ -16,6 +16,7 @@ import coil.ImageLoaderFactory
 import com.theveloper.pixelplay.data.preferences.UserPreferencesRepository
 import com.theveloper.pixelplay.data.diagnostics.AdvancedPerformanceDiagnosticsController
 import com.theveloper.pixelplay.data.repository.ArtistImageRepository
+import com.theveloper.pixelplay.data.service.wear.PlaylistWatchTransferCoordinator
 import com.theveloper.pixelplay.data.telegram.TelegramRepository
 import com.theveloper.pixelplay.presentation.viewmodel.LibraryStateHolder
 import com.theveloper.pixelplay.presentation.viewmodel.ThemeStateHolder
@@ -72,6 +73,9 @@ class PixelPlayApplication : Application(), ImageLoaderFactory, Configuration.Pr
     @Inject
     lateinit var advancedPerformanceDiagnosticsController: dagger.Lazy<AdvancedPerformanceDiagnosticsController>
 
+    @Inject
+    lateinit var playlistWatchTransferCoordinator: dagger.Lazy<PlaylistWatchTransferCoordinator>
+
     private val startupScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
 
     // AÑADE EL COMPANION OBJECT
@@ -130,6 +134,21 @@ class PixelPlayApplication : Application(), ImageLoaderFactory, Configuration.Pr
                 AlbumArtCacheManager.configuredCacheLimitMb = savedLimit.toLong()
             }
         }
+
+        startupScope.launch {
+            // Best-effort: a cold start not directly triggered by the user (e.g. the system
+            // reviving the process for an unrelated broadcast) may be too restricted to start the
+            // foreground service this resumes into — resumePersistedBatchIfNeeded() just skips
+            // resuming this time rather than crashing app startup over it; the persisted intent
+            // stays put for the next launch that can.
+            try {
+                playlistWatchTransferCoordinator.get().resumePersistedBatchIfNeeded()
+            } catch (e: kotlinx.coroutines.CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                Timber.w(e, "Failed to resume an interrupted playlist watch transfer")
+            }
+        }
     }
 
     override fun newImageLoader(): ImageLoader {

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

@@ -91,13 +91,26 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         }
     }
 
+    /**
+     * Substitutes the audio actually streamed to the watch with an already-transcoded file (see
+     * [WatchAudioTranscoder]), bypassing [isSongTransferEligible] and the song's own local-file
+     * resolution — the override file was just written locally by the transcoder, so it's
+     * unconditionally eligible regardless of what the original [Song]'s source was.
+     */
+    data class WatchAudioOverride(
+        val file: File,
+        val mimeType: String,
+        val bitrateBps: Int,
+    )
+
     fun startTransferToWatch(
         nodeId: String,
         requestId: String,
         songId: String,
         transferMode: String = WearTransferRequest.MODE_SAVE_TO_LIBRARY,
         startPositionMs: Long = 0L,
         autoPlay: Boolean = false,
+        audioOverride: WatchAudioOverride? = null,
     ) {
         transferStateStore.markRequested(
             requestId = requestId,
@@ -112,6 +125,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
                 transferMode = transferMode,
                 startPositionMs = startPositionMs,
                 autoPlay = autoPlay,
+                audioOverride = audioOverride,
             )
         }
     }
@@ -123,6 +137,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         transferMode: String,
         startPositionMs: Long,
         autoPlay: Boolean,
+        audioOverride: WatchAudioOverride? = null,
     ) {
         var openedSongSource: OpenedSongSource? = null
         try {
@@ -138,6 +153,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
             }
 
             if (
+                audioOverride == null &&
                 transferMode == WearTransferRequest.MODE_SAVE_TO_LIBRARY &&
                 !isSongTransferEligible(song)
             ) {
@@ -150,19 +166,23 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
                 return
             }
 
-            val songSource = openSongSource(
-                song = song,
-                allowProxyStreaming = transferMode == WearTransferRequest.MODE_TEMPORARY_PLAYBACK,
-            )
+            val songSource = if (audioOverride != null) {
+                openOverrideSongSource(audioOverride)
+            } else {
+                openSongSource(
+                    song = song,
+                    allowProxyStreaming = transferMode == WearTransferRequest.MODE_TEMPORARY_PLAYBACK,
+                )
+            }
             if (songSource == null) {
                 sendTransferMetadataError(
                     nodeId = nodeId,
                     requestId = requestId,
                     songId = song.id,
-                    errorMessage = if (transferMode == WearTransferRequest.MODE_TEMPORARY_PLAYBACK) {
-                        "Cannot stream audio source to watch"
-                    } else {
-                        "Cannot read audio file"
+                    errorMessage = when {
+                        audioOverride != null -> "Cannot read transcoded audio file"
+                        transferMode == WearTransferRequest.MODE_TEMPORARY_PLAYBACK -> "Cannot stream audio source to watch"
+                        else -> "Cannot read audio file"
                     },
                 )
                 return
@@ -182,9 +202,9 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
                 album = song.album,
                 albumId = song.albumId,
                 duration = song.duration,
-                mimeType = song.mimeType ?: "audio/mpeg",
+                mimeType = audioOverride?.mimeType ?: (song.mimeType ?: "audio/mpeg"),
                 fileSize = fileSize,
-                bitrate = song.bitrate ?: 0,
+                bitrate = audioOverride?.bitrateBps ?: (song.bitrate ?: 0),
                 sampleRate = song.sampleRate ?: 0,
                 isFavorite = song.isFavorite,
                 paletteSeedArgb = paletteSeedArgb,
@@ -325,6 +345,15 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
         return openHttpSongSource(streamUrl)
     }
 
+    private fun openOverrideSongSource(override: WatchAudioOverride): OpenedSongSource? {
+        val file = override.file.takeIf { it.isFile && it.canRead() && it.length() > 0L } ?: return null
+        return runCatching {
+            OpenedSongSource(inputStream = file.inputStream(), fileSize = file.length())
+        }.onFailure { error ->
+            Timber.tag(TAG).w(error, "Failed to open transcoded override file=%s", file.absolutePath)
+        }.getOrNull()
+    }
+
     private fun openDirectSongSource(song: Song): OpenedSongSource? {
         val directFile = song.path
             .takeIf { it.isNotBlank() }

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

@@ -1,5 +1,6 @@
 package com.theveloper.pixelplay.data.service.wear
 
+import com.theveloper.pixelplay.shared.WearPlaylistSyncAck
 import com.theveloper.pixelplay.shared.WearTransferProgress
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -8,8 +9,11 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
@@ -33,11 +37,36 @@ data class PhoneWatchTransferState(
         }
 }
 
+/**
+ * Aggregate state of a whole-playlist watch transfer, driven by [PlaylistWatchTransferCoordinator].
+ * [currentSongProgress] is the 0f..1f progress of whichever song [activeRequestId] refers to
+ * (weighted across transcode+transfer phases by the coordinator) — the per-song byte-level detail
+ * lives in [PhoneWatchTransferState], keyed by that same requestId.
+ */
+data class PhoneWatchBatchTransferState(
+    val batchId: String,
+    val playlistId: String,
+    val playlistName: String,
+    val totalSongCount: Int,
+    val completedSongCount: Int = 0,
+    val failedSongCount: Int = 0,
+    val status: String = WearTransferProgress.STATUS_TRANSFERRING,
+    val activeRequestId: String? = null,
+    val currentSongTitle: String = "",
+    val currentSongProgress: Float = 0f,
+    val errorMessage: String? = null,
+    val updatedAtMillis: Long = System.currentTimeMillis(),
+) {
+    val processedSongCount: Int get() = completedSongCount + failedSongCount
+}
+
 @Singleton
 class PhoneWatchTransferStateStore @Inject constructor() {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
     private val _transfers = MutableStateFlow<Map<String, PhoneWatchTransferState>>(emptyMap())
     val transfers: StateFlow<Map<String, PhoneWatchTransferState>> = _transfers.asStateFlow()
+    private val _batchTransfers = MutableStateFlow<Map<String, PhoneWatchBatchTransferState>>(emptyMap())
+    val batchTransfers: StateFlow<Map<String, PhoneWatchBatchTransferState>> = _batchTransfers.asStateFlow()
     private val _reachableWatchNodeIds = MutableStateFlow<Set<String>>(emptySet())
     val reachableWatchNodeIds: StateFlow<Set<String>> = _reachableWatchNodeIds.asStateFlow()
     private val _watchLibrarySyncedNodeIds = MutableStateFlow<Set<String>>(emptySet())
@@ -48,8 +77,19 @@ class PhoneWatchTransferStateStore @Inject constructor() {
     private val _watchSongIds = MutableStateFlow<Set<String>>(emptySet())
     val watchSongIds: StateFlow<Set<String>> = _watchSongIds.asStateFlow()
 
+    // Replay a handful rather than 0: the ack can in principle arrive and be emitted before
+    // PlaylistWatchTransferCoordinator starts collecting for it (right after messageClient's own
+    // send call returns), and a plain event stream with no replay would silently drop it in that
+    // case instead of just delivering it a moment "late" to a fresh collector.
+    private val _playlistSyncAcks = MutableSharedFlow<WearPlaylistSyncAck>(replay = 8)
+    val playlistSyncAcks: SharedFlow<WearPlaylistSyncAck> = _playlistSyncAcks.asSharedFlow()
+
     private val cleanupJobs = ConcurrentHashMap<String, Job>()
 
+    fun onPlaylistSyncAckReceived(ack: WearPlaylistSyncAck) {
+        _playlistSyncAcks.tryEmit(ack)
+    }
+
     fun markRequested(
         requestId: String,
         songId: String,
@@ -232,6 +272,133 @@ class PhoneWatchTransferStateStore @Inject constructor() {
         }
     }
 
+    // --- Playlist batch transfers, driven by PlaylistWatchTransferCoordinator ---
+
+    private val batchCleanupJobs = ConcurrentHashMap<String, Job>()
+
+    fun markBatchStarted(batchId: String, playlistId: String, playlistName: String, totalSongCount: Int) {
+        batchCleanupJobs.remove(batchId)?.cancel()
+        _batchTransfers.update { map ->
+            map + (batchId to PhoneWatchBatchTransferState(
+                batchId = batchId,
+                playlistId = playlistId,
+                playlistName = playlistName,
+                totalSongCount = totalSongCount,
+                status = WearTransferProgress.STATUS_TRANSFERRING,
+            ))
+        }
+    }
+
+    fun markBatchSongStarted(
+        batchId: String,
+        activeRequestId: String,
+        songTitle: String,
+        startingProgress: Float = 0f,
+    ) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                activeRequestId = activeRequestId,
+                currentSongTitle = songTitle,
+                currentSongProgress = startingProgress.coerceIn(0f, 1f),
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    /** [status] is informational only here — [progress] is what actually drives the notification/UI. */
+    fun markBatchSongProgress(batchId: String, status: String, progress: Float) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                currentSongProgress = progress.coerceIn(0f, 1f),
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    fun markBatchSongCompleted(batchId: String) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                completedSongCount = current.completedSongCount + 1,
+                activeRequestId = null,
+                currentSongProgress = 0f,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    fun markBatchSongFailed(batchId: String, errorMessage: String? = null) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                failedSongCount = current.failedSongCount + 1,
+                activeRequestId = null,
+                currentSongProgress = 0f,
+                errorMessage = errorMessage ?: current.errorMessage,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+    }
+
+    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, errorMessage: String) {
+        _batchTransfers.update { map ->
+            val current = map[batchId] ?: return@update map
+            map + (batchId to current.copy(
+                status = WearTransferProgress.STATUS_FAILED,
+                errorMessage = errorMessage,
+                activeRequestId = null,
+                updatedAtMillis = System.currentTimeMillis(),
+            ))
+        }
+        scheduleBatchTerminalCleanup(batchId)
+    }
+
+    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,
+                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 companion object {
         const val TERMINAL_STATE_VISIBILITY_MS = 3500L
     }