From 9ec7704d7baf4b01cedc02f2e52ed1858e9429c6 Mon Sep 17 00:00:00 2001 From: Kingkor Roy Tirtho Date: Sat, 5 Sep 2026 09:38:44 +0600 Subject: [PATCH] feat(jam-session): enhance jam session functionality with improved playback synchronization and error handling --- .../core/audioplayer/AudioPlayer.android.kt | 19 +- .../krtirtho/spotube/media/PlaybackService.kt | 11 +- .../dev/krtirtho/spotube/core/di/Modules.kt | 15 +- .../krtirtho/spotube/core/jam/JamProtocol.kt | 51 ++- .../spotube/core/jam/JamSessionService.kt | 223 ++++++++++-- .../spotube/core/jam/QueueSyncManager.kt | 294 ++++++++++++--- .../core/playback/CollectionPlaybackHelper.kt | 16 + .../core/remote/RemotePlaybackController.kt | 51 ++- .../modules/devices/PlayDestinationPicker.kt | 87 +++-- .../krtirtho/spotube/modules/jam/JamScreen.kt | 343 +++++++++++++++++- .../spotube/modules/jam/JamViewModel.kt | 198 ++++++++++ composeApp/src/commonMain/rust/webrtc_p2p.rs | 47 ++- 12 files changed, 1226 insertions(+), 129 deletions(-) diff --git a/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/core/audioplayer/AudioPlayer.android.kt b/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/core/audioplayer/AudioPlayer.android.kt index 899cd58f..a002f398 100644 --- a/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/core/audioplayer/AudioPlayer.android.kt +++ b/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/core/audioplayer/AudioPlayer.android.kt @@ -20,6 +20,7 @@ package dev.krtirtho.spotube.core.audioplayer import android.content.Context import android.content.Intent import android.os.Build +import android.util.Log import androidx.media3.common.AudioAttributes import androidx.media3.common.C import androidx.media3.common.MediaMetadata @@ -51,11 +52,19 @@ actual class AudioPlayer actual constructor(context: Any) : AudioPlayerInterface private val appContext: Context = (context as Context).applicationContext private fun ensureServiceStarted() { - val intent = Intent(appContext, PlaybackService::class.java) - if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) { - appContext.startForegroundService(intent) - } else { - appContext.startService(intent) + // On Android 12+ starting a foreground service from the background throws + // (ForegroundServiceStartNotAllowedException) — e.g. when a jam session or + // remote control applies playback while the app is backgrounded. Never let + // that crash the app; playback itself runs in-process without the service. + try { + val intent = Intent(appContext, PlaybackService::class.java) + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) { + appContext.startForegroundService(intent) + } else { + appContext.startService(intent) + } + } catch (e: Exception) { + Log.w("AudioPlayer", "Failed to start playback service", e) } } diff --git a/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/media/PlaybackService.kt b/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/media/PlaybackService.kt index 6db62be6..3c49b84c 100644 --- a/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/media/PlaybackService.kt +++ b/composeApp/src/androidMain/kotlin/dev/krtirtho/spotube/media/PlaybackService.kt @@ -86,7 +86,16 @@ class PlaybackService : MediaLibraryService(), KoinComponent { .setOngoing(true) .build() - startForeground(NOTIFICATION_ID, notification) + // On Android 15+ a service created from the background (e.g. by a jam + // sync or queue restoration) hits this in startForeground instead of at + // the startForegroundService call site. Don't crash: playback keeps + // running in-process, just without the notification until the app is + // foregrounded and the service can start properly. + try { + startForeground(NOTIFICATION_ID, notification) + } catch (e: Exception) { + Log.w(TAG, "startForeground not allowed; continuing without notification", e) + } librarySession = MediaLibrarySession.Builder(this, audioPlayer.player, LibrarySessionCallback()) diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/di/Modules.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/di/Modules.kt index 1b3d4f3c..8bf06e4f 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/di/Modules.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/di/Modules.kt @@ -183,7 +183,16 @@ val sharedModules = module { viewModelOf(::BlacklistViewModel) viewModel { DevicesViewModel(get()) } viewModelOf(::RemoteControlViewModel) - viewModelOf(::JamViewModel) + viewModel { + JamViewModel( + jamSession = get(), + deepLinks = get(), + shareService = get(), + settingsProvider = get(), + audioPlayer = get(), + audioPlayerQueue = get(), + ) + } // Album singleOf(::AlbumRepository) @@ -229,8 +238,8 @@ val sharedModules = module { single { RemoteControlService(get(), get(), get()) } withOptions { createdAtStart() } - single { RemotePlaybackController(get(), get(), get(), get()) } - single { JamSessionService(get(), get()) } + single { RemotePlaybackController(get(), get(), get(), get(), get()) } + single { JamSessionService(get(), get(), get()) } singleOf(::JamDeepLinkService) singleOf(::AudioPlayerQueueRepository) { bind() } single { diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamProtocol.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamProtocol.kt index ee8808c8..01fafa94 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamProtocol.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamProtocol.kt @@ -17,8 +17,10 @@ package dev.krtirtho.spotube.core.jam +import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.track.MetadataTrack import dev.krtirtho.spotube.core.audioplayer.LoopState import dev.krtirtho.spotube.core.audioplayer.MediaItem +import dev.krtirtho.spotube.core.audioplayer.QueueEntry import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable @@ -72,6 +74,13 @@ sealed class JamMessage { @SerialName("participantList") data class ParticipantList(val participants: List) : JamMessage() + @Serializable + @SerialName("kick") + data class Kick( + val participantId: String, + val reason: String = "kicked", + ) : JamMessage() + @Serializable @SerialName("leave") data class Leave(val reason: String = "user_left") : JamMessage() @@ -123,6 +132,7 @@ sealed class PlaybackCmd { @Serializable data class JamMediaItem( val url: String, + val trackId: String = "", val title: String, val artist: String, val album: String, @@ -131,6 +141,45 @@ data class JamMediaItem( val protocol: String, ) { companion object { + fun fromQueueEntry(entry: QueueEntry): JamMediaItem = when (entry) { + is QueueEntry.StreamingTrack -> JamMediaItem( + url = "", + trackId = entry.track.id, + title = entry.track.title, + artist = entry.track.artists.joinToString(", ") { it.name }, + album = entry.track.album?.title.orEmpty(), + durationMs = entry.track.durationMs, + coverUrl = entry.track.thumbnails?.maxByOrNull { it.width * it.height }?.url + ?: entry.track.album?.thumbnails?.maxByOrNull { it.width * it.height }?.url + .orEmpty(), + protocol = entry.protocol.name, + ) + + is QueueEntry.LocalTrack -> JamMediaItem( + url = entry.url, + trackId = "", + title = entry.name, + artist = entry.artists.joinToString(", "), + album = entry.album.orEmpty(), + durationMs = entry.duration, + coverUrl = "", + protocol = "PROGRESSIVE", + ) + } + + fun fromTrack(track: MetadataTrack): JamMediaItem = JamMediaItem( + url = "", + trackId = track.id, + title = track.title, + artist = track.artists.joinToString(", ") { it.name }, + album = track.album?.title.orEmpty(), + durationMs = track.durationMs, + coverUrl = track.thumbnails?.maxByOrNull { it.width * it.height }?.url + ?: track.album?.thumbnails?.maxByOrNull { it.width * it.height }?.url + .orEmpty(), + protocol = "PROGRESSIVE", + ) + fun fromMediaItem(item: MediaItem): JamMediaItem = JamMediaItem( url = item.url, title = item.title, @@ -149,7 +198,7 @@ data class JamMediaItem( coverURL = item.coverUrl, url = item.url, protocol = dev.krtirtho.plugin_interfaces.plugin_apis.audio.StreamProtocol - .valueOf(item.protocol), + .valueOf(item.protocol.ifBlank { "PROGRESSIVE" }), ) } } diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamSessionService.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamSessionService.kt index b23b25f6..d14cbca6 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamSessionService.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/JamSessionService.kt @@ -19,6 +19,7 @@ package dev.krtirtho.spotube.core.jam import co.touchlab.kermit.Logger import dev.krtirtho.spotube.core.audioplayer.AudioPlayerInterface +import dev.krtirtho.spotube.core.audioplayer.AudioPlayerQueue import dev.krtirtho.spotube.core.di.injectLogger import dev.krtirtho.spotube.modules.settings.SettingsProvider import kotlinx.coroutines.CoroutineScope @@ -49,13 +50,32 @@ data class JamInvite( val sdp: String, ) +/** + * Owns the peer connections of a jam session (star topology: host relays state + * to all guests) and the hello/welcome handshake, participant bookkeeping and + * kick/ban. Playback & queue synchronization itself is delegated to + * [QueueSyncManager], which runs while a session is active. + */ class JamSessionService( private val audioPlayer: AudioPlayerInterface, + private val audioPlayerQueue: AudioPlayerQueue, private val settingsProvider: SettingsProvider, ) : KoinComponent { val logger by injectLogger() private val log = Logger.withTag("JamSessionService") + /** + * Playback/queue synchronization. Owned by this service (not a Koin bean) so + * the two don't form a circular dependency; it's started/stopped with the + * session lifecycle. + */ + private val queueSyncManager = QueueSyncManager( + audioPlayer = audioPlayer, + audioPlayerQueue = audioPlayerQueue, + jamSession = this, + settingsProvider = settingsProvider, + ) + private val json = Json { ignoreUnknownKeys = true classDiscriminator = "type" @@ -91,23 +111,36 @@ class JamSessionService( /** Host side: guests whose handshake completed. Keyed by invite id. */ private val connectedGuests = mutableMapOf() + /** Host side: guest device ids, used for bans. */ + private val guestDeviceIds = mutableMapOf() + + /** Host side: latest RTCPeerConnection state per guest ("connecting", "connected", "failed"...). */ + private val guestConnectionStates = mutableMapOf() + + /** Host side: device ids banned for this session. */ + private val bannedDeviceIds = mutableSetOf() + /** Guest side: the single connection to the host. */ private var hostConnection: WebrtcPeerConnection? = null + private var hostDisplayName: String = "Host" + private var guestDisplayName: String = "Guest" + suspend fun createSession(): String { log.i { "Creating jam session" } - val hostName = resolveParticipantName(defaultPrefix = "Host") + hostDisplayName = resolveParticipantName(defaultPrefix = "Host") _role.value = JamRole.Host _localParticipantId.value = "host" _participants.value = listOf( JamParticipant( id = "host", - displayName = hostName, + displayName = hostDisplayName, isHost = true, ) ) _isActive.value = true + queueSyncManager.start() return generateInvite().sdp } @@ -174,12 +207,13 @@ class JamSessionService( ) } log.i { "Guest $resolvedId ($peerName) joined" } + broadcastParticipantList() return true } suspend fun joinSession(offerSdp: String, hostName: String? = null): String { log.i { "Joining jam session" } - val participantName = resolveParticipantName(defaultPrefix = "Guest") + guestDisplayName = resolveParticipantName(defaultPrefix = "Guest") val pc = createWebrtcPeerConnection( iceServers = defaultIceServers(), @@ -188,7 +222,7 @@ class JamSessionService( hostConnection = pc _role.value = JamRole.Guest - _localParticipantId.value = "guest" + _localParticipantId.value = null _participants.value = listOf( JamParticipant( id = "host", @@ -197,6 +231,7 @@ class JamSessionService( ) ) _isActive.value = true + queueSyncManager.start() // The data channel arrives in-band from the host's offer via on_data_channel; // we only answer here. @@ -210,27 +245,52 @@ class JamSessionService( val payload = json.encodeToString(JamMessage.serializer(), message) when (_role.value) { JamRole.Host -> { - val targets = if (guestId != null) { - listOfNotNull(connectedGuests[guestId]) - } else { - connectedGuests.values.toList() - } - targets.forEach { pc -> + if (guestId != null) { + val pc = connectedGuests[guestId] ?: return runCatching { pc.sendData(CHANNEL_LABEL, payload) } - .onFailure { e -> log.w(e) { "Failed to send to guest" } } + .onFailure { e -> + log.w(e) { "Failed to send to guest $guestId" } + onSendFailure(guestId) + } + } else { + val dead = mutableListOf() + connectedGuests.forEach { (id, pc) -> + runCatching { pc.sendData(CHANNEL_LABEL, payload) } + .onFailure { e -> + log.w(e) { "Failed to send to guest $id" } + dead += id + } + } + dead.forEach { id -> onSendFailure(id) } } } JamRole.Guest -> { - hostConnection?.sendData(CHANNEL_LABEL, payload) + runCatching { hostConnection?.sendData(CHANNEL_LABEL, payload) } + .onFailure { e -> + log.w(e) { "Failed to send to host" } + } } null -> log.w { "sendMessage called while no session is active" } } } + /** + * A send to a guest failed. If that guest's connection has already given up + * (failed/closed), drop them from the session; while the connection is merely + * "connecting" the channel may simply not be open yet, so keep them. + */ + private fun onSendFailure(guestId: String) { + val state = guestConnectionStates[guestId] + if (state == "failed" || state == "closed" || state == "disconnected") { + scope.launch { removeGuest(guestId) } + } + } + suspend fun leave() { log.i { "Leaving jam session" } + queueSyncManager.stop() runCatching { sendMessage(JamMessage.Leave()) } shutdownAll() _role.value = null @@ -238,6 +298,9 @@ class JamSessionService( _isActive.value = false _isConnected.value = false _localParticipantId.value = null + guestDeviceIds.clear() + guestConnectionStates.clear() + bannedDeviceIds.clear() } suspend fun broadcastPlaybackCommand(command: PlaybackCmd) { @@ -265,12 +328,44 @@ class JamSessionService( sendMessage(JamMessage.SuggestPlaylist(tracks)) } + // ---------- Host moderation ---------- + + suspend fun kickParticipant(participantId: String, reason: String = "kicked by host") { + if (_role.value != JamRole.Host) return + log.i { "Kicking participant $participantId" } + sendMessage(JamMessage.Kick(participantId, reason), guestId = participantId) + removeGuest(participantId) + } + + suspend fun banParticipant(participantId: String) { + if (_role.value != JamRole.Host) return + val deviceId = guestDeviceIds[participantId] + if (deviceId != null) { + bannedDeviceIds += deviceId + log.i { "Banning device $deviceId (participant $participantId)" } + } + kickParticipant(participantId, "banned by host") + } + + private suspend fun removeGuest(guestId: String) { + val pc = connectedGuests.remove(guestId) + runCatching { pc?.shutdown() } + guestDeviceIds.remove(guestId) + guestConnectionStates.remove(guestId) + _participants.update { current -> + current.filterNot { it.id == guestId } + } + broadcastParticipantList() + } + + private suspend fun broadcastParticipantList() { + if (_role.value != JamRole.Host) return + sendMessage(JamMessage.ParticipantList(_participants.value)) + } + /** * ICE servers for global peer-to-peer jam sessions: multiple STUN servers for * NAT traversal plus a TURN relay for symmetric NATs and strict firewalls. - * Unreachable servers no longer stall offer/answer creation — the webrtc - * driver completes gathering once every STUN client has answered or timed out, - * and [WebrtcPeerConnection] bounds the wait anyway. */ private fun defaultIceServers(): List = listOf( IceServerConfig( @@ -295,6 +390,11 @@ class JamSessionService( ?: "$defaultPrefix-${randomShortId()}" } + private fun localDeviceId(): String { + return settingsProvider.settingsState.value?.remoteControlDeviceId + ?: "device-${randomShortId()}" + } + /** * Per-guest handler so messages received on a guest's connection can be * attributed back to that guest (needed for kick-on-leave and targeted sends). @@ -310,6 +410,10 @@ class JamSessionService( override fun onConnectionStateChange(state: String) { log.i { "[$guestId] Connection state: $state" } + guestConnectionStates[guestId] = state + if (state == "failed" || state == "closed") { + scope.launch { removeGuest(guestId) } + } } override fun onDataChannelOpen(label: String) { @@ -323,6 +427,9 @@ class JamSessionService( override fun onDataChannelClose(label: String) { log.i { "[$guestId] Data channel closed" } + if (_role.value == JamRole.Host) { + scope.launch { removeGuest(guestId) } + } } } @@ -337,11 +444,19 @@ class JamSessionService( override fun onConnectionStateChange(state: String) { log.i { "Connection state: $state" } + if (state == "failed" || state == "closed") { + scope.launch { leave() } + } } override fun onDataChannelOpen(label: String) { log.i { "Data channel '$label' open" } _isConnected.value = true + // Introduce ourselves so the host can fill in our name and hand us + // our participant id. + scope.launch { + sendMessage(JamMessage.Hello(guestDisplayName, localDeviceId())) + } } override fun onDataChannelMessage(label: String, data: String) { @@ -350,6 +465,7 @@ class JamSessionService( override fun onDataChannelClose(label: String) { log.i { "Data channel closed" } + scope.launch { leave() } } } @@ -362,15 +478,44 @@ class JamSessionService( _incomingSuggestions.tryEmit(message) } + is JamMessage.Hello -> { + if (_role.value == JamRole.Host && fromGuestId != null) { + handleHello(fromGuestId, message) + } + } + + is JamMessage.Welcome -> { + if (_role.value == JamRole.Guest) { + _localParticipantId.value = message.participantId + _participants.update { current -> + current.map { participant -> + if (participant.isHost) { + participant.copy(displayName = message.hostName.ifBlank { participant.displayName }) + } else { + participant + } + } + } + log.i { "Welcome: joined as ${message.participantId}" } + } + } + + is JamMessage.ParticipantList -> { + if (_role.value == JamRole.Guest) { + _participants.value = message.participants + } + } + + is JamMessage.Kick -> { + if (_role.value == JamRole.Guest) { + log.i { "Kicked by host: ${message.reason}" } + scope.launch { leave() } + } + } + is JamMessage.Leave -> { if (_role.value == JamRole.Host && fromGuestId != null) { - val leavingPc = connectedGuests.remove(fromGuestId) - scope.launch { - runCatching { leavingPc?.shutdown() } - } - _participants.update { current -> - current.filterNot { it.id == fromGuestId } - } + scope.launch { removeGuest(fromGuestId) } } else if (_role.value == JamRole.Guest) { scope.launch { leave() } } @@ -383,6 +528,40 @@ class JamSessionService( } } + private fun handleHello(guestId: String, hello: JamMessage.Hello) { + val deviceId = hello.deviceId + if (deviceId in bannedDeviceIds) { + log.w { "Rejecting banned device $deviceId" } + scope.launch { + sendMessage( + JamMessage.Kick(guestId, "banned by host"), + guestId = guestId, + ) + removeGuest(guestId) + } + return + } + guestDeviceIds[guestId] = deviceId + _participants.update { current -> + current.map { participant -> + if (participant.id == guestId) { + participant.copy(displayName = hello.displayName.ifBlank { participant.displayName }) + } else { + participant + } + } + } + scope.launch { + sendMessage( + JamMessage.Welcome(hostDisplayName, guestId), + guestId = guestId, + ) + broadcastParticipantList() + // Give the newly joined guest the current queue + playback state. + queueSyncManager.broadcastNow() + } + } + private suspend fun shutdownAll() { pendingInvites.values.forEach { runCatching { it.shutdown() } } connectedGuests.values.forEach { runCatching { it.shutdown() } } diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/QueueSyncManager.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/QueueSyncManager.kt index 550de713..104cf223 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/QueueSyncManager.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/jam/QueueSyncManager.kt @@ -18,46 +18,61 @@ package dev.krtirtho.spotube.core.jam import co.touchlab.kermit.Logger +import dev.krtirtho.plugin_interfaces.plugin_apis.audio.StreamProtocol +import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.artist.MetadataArtist +import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.track.MetadataTrack import dev.krtirtho.spotube.core.audioplayer.AudioPlayerInterface +import dev.krtirtho.spotube.core.audioplayer.AudioPlayerQueue +import dev.krtirtho.spotube.core.audioplayer.MediaItem import dev.krtirtho.spotube.core.audioplayer.PlayerState +import dev.krtirtho.spotube.core.audioplayer.QueueEntry +import dev.krtirtho.spotube.modules.settings.SettingsProvider import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.serialization.json.Json -import kotlinx.coroutines.flow.first /** - * Manages queue synchronization between the host and the jam session. + * Keeps playback in sync across a jam session (star topology). * - * On the host: observes local playback state and broadcasts queue updates to guests. - * On the guest: receives queue updates and applies them to local playback. + * On the **host**: applies incoming playback commands and guest suggestions to the + * host's player, and broadcasts the current queue + playback state to all guests + * (on queue changes and periodically, so play/pause/seek/position propagate). * - * Conflict resolution: the host has authority. When a guest receives a queue state, - * it replaces the local queue. (and (The guest's local queue is essentially read-only - * during a jam session.) + * On the **guest**: mirrors the host's queue into the local player and applies + * playback commands. The guest's queue is read-only — the host has authority. */ class QueueSyncManager( private val audioPlayer: AudioPlayerInterface, + private val audioPlayerQueue: AudioPlayerQueue, private val jamSession: JamSessionService, - private val scope: CoroutineScope, + private val settingsProvider: SettingsProvider, ) { private val log = Logger.withTag("QueueSyncManager") - private val json = Json { - ignoreUnknownKeys = true - classDiscriminator = "type" - encodeDefaults = true - } + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) private val _isSyncing = MutableStateFlow(false) val isSyncing: StateFlow = _isSyncing.asStateFlow() private var hostBroadcastJob: Job? = null + private var hostCommandJob: Job? = null private var guestApplyJob: Job? = null private var guestCommandJob: Job? = null + /** Guest side: the last applied queue snapshot, used to detect real queue changes. */ + private var lastAppliedItems: List = emptyList() + + /** Guest side: tracks the player was told to start playing from. */ + private var lastAppliedCurrentIndex = -1 + fun start() { if (_isSyncing.value) return _isSyncing.value = true @@ -75,43 +90,87 @@ class QueueSyncManager( fun stop() { _isSyncing.value = false hostBroadcastJob?.cancel() + hostCommandJob?.cancel() guestApplyJob?.cancel() guestCommandJob?.cancel() hostBroadcastJob = null + hostCommandJob = null guestApplyJob = null guestCommandJob = null + lastAppliedItems = emptyList() + lastAppliedCurrentIndex = -1 } + // ---------- Host side ---------- + private fun startHostSync() { + // Apply commands/suggestions coming from guests. + hostCommandJob = scope.launch { + jamSession.role.first { it != null } + if (jamSession.role.value != JamRole.Host) return@launch + + jamSession.incomingMessages.collect { message -> + when (message) { + is JamMessage.PlaybackCommand -> applyPlaybackCommand(message.command) + is JamMessage.SuggestTrack -> acceptSuggestion(listOf(message.mediaItem)) + is JamMessage.SuggestPlaylist -> acceptSuggestion(message.tracks) + else -> {} + } + } + } + + // Broadcast state on queue changes and periodically. hostBroadcastJob = scope.launch { jamSession.role.first { it != null } if (jamSession.role.value != JamRole.Host) return@launch - jamSession.broadcastQueueState( - items = audioPlayer.playlistFlow.value.map(JamMediaItem::fromMediaItem), - currentIndex = audioPlayer.playlistFlow.value.indexOf( - audioPlayer.currentMediaItemFlow.value - ).coerceAtLeast(0), - isPlaying = audioPlayer.playerStateFlow.value == PlayerState.PLAYING, - positionMs = audioPlayer.positionFlow.value.inWholeMilliseconds, - ) - - audioPlayer.playlistFlow.collect { playlist -> - audioPlayer.playerStateFlow.value.let { state -> - audioPlayer.positionFlow.value.let { position -> - jamSession.broadcastQueueState( - items = playlist.map(JamMediaItem::fromMediaItem), - currentIndex = playlist.indexOf(audioPlayer.currentMediaItemFlow.value) - .coerceAtLeast(0), - isPlaying = state == PlayerState.PLAYING, - positionMs = position.inWholeMilliseconds, - ) - } + // Queue changes (separate coroutine — collect() never returns). + launch { + audioPlayerQueue.queueFlow.collect { + broadcastCurrentState() } } + + // Periodic tick so play/pause/seek/position propagate to guests. + while (isActive) { + delay(2_000) + broadcastCurrentState() + } } } + /** Immediately pushes the current queue + playback state to all guests. */ + suspend fun broadcastNow() { + if (jamSession.role.value == JamRole.Host) { + broadcastCurrentState() + } + } + + private suspend fun broadcastCurrentState() { + val queue = audioPlayerQueue.getQueue() + val current = audioPlayerQueue.getCurrentQueueEntry() + val currentIndex = if (current != null) { + queue.indexOfFirst { it.matchesEntry(current) } + } else { + -1 + } + jamSession.broadcastQueueState( + items = queue.map(JamMediaItem::fromQueueEntry), + currentIndex = currentIndex.coerceAtLeast(0), + isPlaying = audioPlayer.playerStateFlow.value == PlayerState.PLAYING, + positionMs = audioPlayer.positionFlow.value.inWholeMilliseconds, + ) + } + + private suspend fun acceptSuggestion(items: List) { + if (items.isEmpty()) return + val entries = items.map { it.toQueueEntry() } + log.i { "Accepting ${entries.size} suggested item(s) into the jam queue" } + audioPlayerQueue.addAllToQueue(entries) + } + + // ---------- Guest side ---------- + private fun startGuestSync() { guestApplyJob = scope.launch { jamSession.incomingMessages.collect { message -> @@ -130,33 +189,162 @@ class QueueSyncManager( private suspend fun applyQueueState(state: JamMessage.QueueState) { log.d { "Applying queue state: ${state.items.size} items, current=${state.currentIndex}" } - val mediaItems = state.items.map(JamMediaItem::toMediaItem) - audioPlayer.load( - playlist = mediaItems, - autoPlay = state.isPlaying, - startPosition = state.currentIndex.coerceAtLeast(0), + + // Items that carry neither a track id nor a usable URL can't be played + // on this device — skip them instead of crashing the player. + val playableItems = state.items.filter { it.trackId.isNotBlank() || it.url.isNotBlank() } + + val queueChanged = playableItems != lastAppliedItems + if (queueChanged) { + lastAppliedItems = playableItems + lastAppliedCurrentIndex = state.currentIndex + val mediaItems = playableItems.map { it.toPlayableMediaItem() } + runCatching { + audioPlayer.load( + playlist = mediaItems, + autoPlay = state.isPlaying, + startPosition = state.currentIndex.coerceIn(0, mediaItems.lastIndex.coerceAtLeast(0)), + ) + }.onFailure { e -> + log.e(e) { "Failed to apply jam queue to local player" } + } + return + } + + // Same queue: just sync playback state. Avoid seeking on every tick unless + // the drift is meaningful. + if (state.currentIndex != lastAppliedCurrentIndex) { + lastAppliedCurrentIndex = state.currentIndex + runCatching { audioPlayer.jumpTo(state.currentIndex.coerceAtLeast(0)) } + .onFailure { e -> log.w(e) { "Failed to jump to index ${state.currentIndex}" } } + } + val currentState = audioPlayer.playerStateFlow.value + if (state.isPlaying && currentState != PlayerState.PLAYING) { + audioPlayer.play() + } else if (!state.isPlaying && currentState == PlayerState.PLAYING) { + audioPlayer.pause() + } + val driftMs = kotlin.math.abs( + audioPlayer.positionFlow.value.inWholeMilliseconds - state.positionMs ) + if (driftMs > POSITION_SYNC_THRESHOLD_MS) { + runCatching { audioPlayer.seekTo(kotlin.time.Duration.parse("${state.positionMs}ms")) } + .onFailure { e -> log.w(e) { "Failed to sync position" } } + } } private suspend fun applyPlaybackCommand(command: PlaybackCmd) { log.d { "Applying playback command: $command" } - when (command) { - PlaybackCmd.Play -> audioPlayer.play() - PlaybackCmd.Pause -> audioPlayer.pause() - PlaybackCmd.Toggle -> { - if (audioPlayer.playerStateFlow.value == PlayerState.PLAYING) { - audioPlayer.pause() - } else { - audioPlayer.play() + runCatching { + when (command) { + PlaybackCmd.Play -> audioPlayer.play() + PlaybackCmd.Pause -> audioPlayer.pause() + PlaybackCmd.Toggle -> { + if (audioPlayer.playerStateFlow.value == PlayerState.PLAYING) { + audioPlayer.pause() + } else { + audioPlayer.play() + } } + + is PlaybackCmd.Seek -> audioPlayer.seekTo(kotlin.time.Duration.parse("${command.positionMs}ms")) + PlaybackCmd.SkipNext -> audioPlayer.skipToNext() + PlaybackCmd.SkipPrevious -> audioPlayer.skipToPrevious() + is PlaybackCmd.SetVolume -> audioPlayer.setVolume(command.volume) + is PlaybackCmd.SetLoop -> audioPlayer.loop(JamLoopMapping.fromString(command.loop)) + is PlaybackCmd.SetShuffle -> audioPlayer.shuffle(command.enabled) + is PlaybackCmd.JumpTo -> audioPlayer.jumpTo(command.index) } - is PlaybackCmd.Seek -> audioPlayer.seekTo(kotlin.time.Duration.parse("${command.positionMs}ms")) - PlaybackCmd.SkipNext -> audioPlayer.skipToNext() - PlaybackCmd.SkipPrevious -> audioPlayer.skipToPrevious() - is PlaybackCmd.SetVolume -> audioPlayer.setVolume(command.volume) - is PlaybackCmd.SetLoop -> audioPlayer.loop(JamLoopMapping.fromString(command.loop)) - is PlaybackCmd.SetShuffle -> audioPlayer.shuffle(command.enabled) - is PlaybackCmd.JumpTo -> audioPlayer.jumpTo(command.index) + }.onFailure { e -> + log.w(e) { "Failed to apply playback command: $command" } } } + + // ---------- Conversions ---------- + + /** Host side: turn a suggested item into a playable queue entry. */ + private fun JamMediaItem.toQueueEntry(): QueueEntry = when { + trackId.isNotBlank() -> QueueEntry.StreamingTrack( + track = MetadataTrack( + id = trackId, + title = title, + durationMs = durationMs, + trackNumber = null, + discNumber = null, + artists = listOf( + MetadataArtist.Basic(id = "", name = artist, thumbnails = emptyList(), externalUri = null) + ), + album = null, + thumbnails = null, + explicit = null, + popularity = null, + isrcCode = null, + externalUri = null, + ), + url = "", + protocol = runCatching { StreamProtocol.valueOf(protocol.ifBlank { "PROGRESSIVE" }) } + .getOrDefault(StreamProtocol.PROGRESSIVE), + ) + + else -> QueueEntry.LocalTrack( + name = title, + artists = artist.split(',').map { it.trim() }.filter { it.isNotEmpty() }, + duration = durationMs, + album = album.ifBlank { null }, + coverBytes = null, + url = url, + ) + } + + /** + * Guest side: build a playable MediaItem. Streaming tracks have their stream + * URL resolved through this device's own playback proxy (the host never sends + * usable URLs — each guest must fetch from its own plugins). + */ + private suspend fun JamMediaItem.toPlayableMediaItem(): MediaItem { + if (trackId.isNotBlank()) { + val proxyUrl = buildStreamingUrl(trackId, protocol) + return MediaItem( + title = title, + artist = artist, + album = album, + duration = kotlin.time.Duration.parse("${durationMs}ms"), + coverURL = coverUrl, + url = proxyUrl, + protocol = runCatching { StreamProtocol.valueOf(protocol.ifBlank { "PROGRESSIVE" }) } + .getOrDefault(StreamProtocol.PROGRESSIVE), + ) + } + return JamMediaItem.toMediaItem(this) + } + + private suspend fun buildStreamingUrl(trackId: String, protocol: String): String { + val port = settingsProvider.settingsState + .first() + ?.playbackProxyServerPort ?: return "" + val baseUrl = "http://127.0.0.1:$port" + val streamProtocol = runCatching { StreamProtocol.valueOf(protocol.ifBlank { "PROGRESSIVE" }) } + .getOrDefault(StreamProtocol.PROGRESSIVE) + return when (streamProtocol) { + StreamProtocol.HLS, StreamProtocol.DASH -> "${baseUrl.trimEnd('/')}/manifest/$trackId" + StreamProtocol.PROGRESSIVE -> "${baseUrl.trimEnd('/')}/stream/$trackId" + } + } + + private fun QueueEntry.matchesEntry(other: QueueEntry): Boolean { + return when { + this is QueueEntry.StreamingTrack && other is QueueEntry.StreamingTrack -> + this.track.id == other.track.id + + this is QueueEntry.LocalTrack && other is QueueEntry.LocalTrack -> + this.url == other.url && this.name == other.name + + else -> false + } + } + + companion object { + /** Seek the guest only when its position drifts more than this from the host. */ + private const val POSITION_SYNC_THRESHOLD_MS = 3_000L + } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/playback/CollectionPlaybackHelper.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/playback/CollectionPlaybackHelper.kt index 7441309c..da55fb38 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/playback/CollectionPlaybackHelper.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/playback/CollectionPlaybackHelper.kt @@ -26,6 +26,7 @@ import dev.krtirtho.spotube.modules.artist.ArtistRepository import dev.krtirtho.spotube.modules.blacklist.BlacklistRepository import dev.krtirtho.spotube.modules.playlist.PlaylistRepository import dev.krtirtho.spotube.modules.saved_tracks.SavedTracksRepository +import dev.krtirtho.spotube.core.remote.RemoteCollectionType class CollectionPlaybackHelper( private val albumRepository: AlbumRepository, @@ -264,6 +265,21 @@ class CollectionPlaybackHelper( } } + /** + * Resolves the tracks of a collection without loading them into the local + * queue — used to suggest a collection into a jam session from a guest. + */ + suspend fun resolveCollectionTracks(type: RemoteCollectionType, id: String): List = + when (type) { + RemoteCollectionType.Playlist -> fetchAllPlaylistTracks(id).asTracks() + RemoteCollectionType.Album -> fetchAllAlbumTracks(id).asTracks() + RemoteCollectionType.ArtistTopTracks -> fetchArtistTopTracks(id).asTracks() + RemoteCollectionType.SavedTracks -> fetchAllSavedTracks().asTracks() + } + + private fun List.asTracks(): List = + mapNotNull { (it as? QueueEntry.StreamingTrack)?.track } + private suspend fun fetchAllSavedTracks(): List { val allTracks = mutableListOf() var pagination = savedTracksRepository.getSavedTracks() diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/remote/RemotePlaybackController.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/remote/RemotePlaybackController.kt index a0ba5f5b..01a77c4b 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/remote/RemotePlaybackController.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/core/remote/RemotePlaybackController.kt @@ -21,6 +21,9 @@ import co.touchlab.kermit.Logger import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.track.MetadataTrack import dev.krtirtho.spotube.core.audioplayer.AudioPlayerQueue import dev.krtirtho.spotube.core.audioplayer.QueueEntry +import dev.krtirtho.spotube.core.jam.JamMediaItem +import dev.krtirtho.spotube.core.jam.JamRole +import dev.krtirtho.spotube.core.jam.JamSessionService import dev.krtirtho.spotube.core.playback.CollectionPlaybackHelper import dev.krtirtho.spotube.modules.blacklist.BlacklistRepository import kotlinx.coroutines.CoroutineScope @@ -84,6 +87,7 @@ class RemotePlaybackController( private val collectionPlaybackHelper: CollectionPlaybackHelper, private val audioPlayerQueue: AudioPlayerQueue, private val blacklistRepository: BlacklistRepository, + private val jamSession: JamSessionService, ) : KoinComponent { private val logger = Logger.withTag("RemotePlaybackController") private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) @@ -154,6 +158,49 @@ class RemotePlaybackController( _pendingRequest.value = null } + /** + * Routes the pending request into the active jam session. On the host the jam + * queue IS the local queue, so the action runs locally; on a guest the content + * is suggested to the host, which accepts it into the shared queue. + */ + fun playOnJam() { + val request = _pendingRequest.value ?: return + _pendingRequest.value = null + scope.launch { + try { + when (jamSession.role.value) { + JamRole.Host -> executeLocally(request) + JamRole.Guest -> suggestToJam(request) + null -> {} + } + } catch (e: Exception) { + logger.e(e) { "Failed to send content to jam session" } + } + } + } + + private suspend fun suggestToJam(request: PlaybackDestinationRequest) { + when (request) { + is PlaybackDestinationRequest.Collection -> { + val tracks = collectionPlaybackHelper.resolveCollectionTracks(request.type, request.id) + if (tracks.isNotEmpty()) { + jamSession.suggestPlaylist(tracks.map { it.toJamMediaItem() }) + logger.i { "Suggested ${tracks.size} track(s) to the jam session" } + } + } + + is PlaybackDestinationRequest.Track -> { + jamSession.suggestTrack(request.track.toJamMediaItem()) + } + + is PlaybackDestinationRequest.Tracks -> { + if (request.tracks.isNotEmpty()) { + jamSession.suggestPlaylist(request.tracks.map { it.toJamMediaItem() }) + } + } + } + } + // ---------- Internals ---------- private fun request(request: PlaybackDestinationRequest) { @@ -321,4 +368,6 @@ class RemotePlaybackController( album?.id == other.album?.id && artists.map { it.id.ifBlank { it.name } } == other.artists.map { it.id.ifBlank { it.name } } } -} \ No newline at end of file +} + +private fun MetadataTrack.toJamMediaItem(): JamMediaItem = JamMediaItem.fromTrack(this) \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/devices/PlayDestinationPicker.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/devices/PlayDestinationPicker.kt index a5b9f4b9..6fd70573 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/devices/PlayDestinationPicker.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/devices/PlayDestinationPicker.kt @@ -30,6 +30,7 @@ import androidx.compose.ui.Modifier import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.unit.dp import androidx.lifecycle.compose.collectAsStateWithLifecycle +import dev.krtirtho.spotube.core.jam.JamSessionService import dev.krtirtho.spotube.core.remote.ConnectionState import dev.krtirtho.spotube.core.remote.PlaybackDestinationAction import dev.krtirtho.spotube.core.remote.RemoteControlClient @@ -39,19 +40,22 @@ import dev.krtirtho.spotube.core.ui.base.ThemedDialog import dev.krtirtho.spotube.resources.iconsax.Iconsax import dev.krtirtho.spotube.resources.iconsax.IconsaxCd import dev.krtirtho.spotube.resources.iconsax.IconsaxMirroringScreen +import dev.krtirtho.spotube.resources.iconsax.IconsaxMusicPlaylist import org.koin.compose.koinInject /** - * Globally hosted dialog shown when a remote device is connected and the user - * tries to play / add to queue / play next. Lets the user choose between the - * local device and the connected remote device(s). + * Globally hosted dialog shown when the user tries to play / add to queue / + * play next and there is more than one place it could go (a connected remote + * device and/or an active jam session). Lets the user choose the destination. */ @Composable fun PlayDestinationPickerHost() { val controller = koinInject() val remoteControlClient = koinInject() + val jamSession = koinInject() val request by controller.pendingRequest.collectAsStateWithLifecycle() val connectionState by remoteControlClient.connectionState.collectAsStateWithLifecycle() + val jamActive by jamSession.isActive.collectAsStateWithLifecycle() val pendingRequest = request ?: return @@ -111,30 +115,59 @@ fun PlayDestinationPickerHost() { }, ) - ListRowTile( - onClick = controller::playOnRemote, - modifier = Modifier.fillMaxWidth(), - leading = { - Icon( - imageVector = Iconsax.IconsaxMirroringScreen, - contentDescription = null, - tint = MaterialTheme.colorScheme.primary, - ) - }, - title = { - Text( - text = remoteDeviceName, - style = MaterialTheme.typography.bodyLarge, - ) - }, - subtitle = { - Text( - text = "$actionLabel on the connected device", - style = MaterialTheme.typography.bodySmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - ) - }, - ) + if (connectionState is ConnectionState.Connected) { + ListRowTile( + onClick = controller::playOnRemote, + modifier = Modifier.fillMaxWidth(), + leading = { + Icon( + imageVector = Iconsax.IconsaxMirroringScreen, + contentDescription = null, + tint = MaterialTheme.colorScheme.primary, + ) + }, + title = { + Text( + text = remoteDeviceName, + style = MaterialTheme.typography.bodyLarge, + ) + }, + subtitle = { + Text( + text = "$actionLabel on the connected device", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + }, + ) + } + + if (jamActive) { + ListRowTile( + onClick = controller::playOnJam, + modifier = Modifier.fillMaxWidth(), + leading = { + Icon( + imageVector = Iconsax.IconsaxMusicPlaylist, + contentDescription = null, + tint = MaterialTheme.colorScheme.primary, + ) + }, + title = { + Text( + text = "Jam Session", + style = MaterialTheme.typography.bodyLarge, + ) + }, + subtitle = { + Text( + text = "$actionLabel in the shared jam queue", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + }, + ) + } } }, actions = { diff --git a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/jam/JamScreen.kt b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/jam/JamScreen.kt index 84b0e737..adc4d246 100644 --- a/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/jam/JamScreen.kt +++ b/composeApp/src/commonMain/kotlin/dev/krtirtho/spotube/modules/jam/JamScreen.kt @@ -17,17 +17,25 @@ package dev.krtirtho.spotube.modules.jam +import androidx.compose.foundation.clickable import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Column import androidx.compose.foundation.layout.Row import androidx.compose.foundation.layout.fillMaxSize import androidx.compose.foundation.layout.fillMaxWidth +import androidx.compose.foundation.layout.height +import androidx.compose.foundation.layout.heightIn import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.layout.size +import androidx.compose.foundation.lazy.LazyColumn +import androidx.compose.foundation.lazy.itemsIndexed import androidx.compose.foundation.rememberScrollState +import androidx.compose.foundation.shape.CircleShape import androidx.compose.foundation.text.selection.SelectionContainer import androidx.compose.foundation.verticalScroll import androidx.compose.material3.Button import androidx.compose.material3.HorizontalDivider +import androidx.compose.material3.Icon import androidx.compose.material3.MaterialTheme import androidx.compose.material3.OutlinedButton import androidx.compose.material3.OutlinedTextField @@ -45,15 +53,31 @@ import androidx.compose.runtime.saveable.rememberSaveable import androidx.compose.runtime.setValue import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier +import androidx.compose.ui.draw.clip +import androidx.compose.ui.graphics.graphicsLayer +import androidx.compose.ui.layout.ContentScale import androidx.compose.ui.platform.LocalClipboardManager import androidx.compose.ui.text.AnnotatedString +import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.unit.dp import androidx.lifecycle.compose.collectAsStateWithLifecycle +import coil3.compose.AsyncImage import dev.krtirtho.spotube.core.jam.JamRole import dev.krtirtho.spotube.core.navigation.NavigationCommands +import dev.krtirtho.spotube.core.ui.base.IconButton +import dev.krtirtho.spotube.core.ui.base.LocalBaseUITheme +import dev.krtirtho.spotube.core.ui.base.copyShape import dev.krtirtho.spotube.core.ui.component.ApplicationMainBar import dev.krtirtho.spotube.modules.shell.LocalAppShellBottomInset +import dev.krtirtho.spotube.resources.iconsax.Iconsax +import dev.krtirtho.spotube.resources.iconsax.IconsaxArrowDown4 +import dev.krtirtho.spotube.resources.iconsax.IconsaxNext +import dev.krtirtho.spotube.resources.iconsax.IconsaxPause +import dev.krtirtho.spotube.resources.iconsax.IconsaxPlay +import dev.krtirtho.spotube.resources.iconsax.IconsaxPrevious +import dev.krtirtho.spotube.resources.iconsax.IconsaxRepeateMusic +import dev.krtirtho.spotube.resources.iconsax.IconsaxShuffle import org.koin.compose.viewmodel.koinViewModel @Composable @@ -99,16 +123,34 @@ fun JamScreen( state.role == JamRole.Host -> HostSessionView( state = state, + playerState = viewModel.jamPlayerState.collectAsStateWithLifecycle().value, onNewInvite = viewModel::generateNewInvite, onSubmitAnswer = viewModel::submitAnswerPasted, onShare = viewModel::share, onLeave = viewModel::leave, + onTogglePlayPause = viewModel::togglePlayPause, + onSkipNext = viewModel::skipNext, + onSkipPrevious = viewModel::skipPrevious, + onSeek = viewModel::seek, + onJumpTo = viewModel::jumpTo, + onToggleShuffle = viewModel::toggleShuffle, + onCycleLoop = viewModel::cycleLoopMode, + onKick = viewModel::kickParticipant, + onBan = viewModel::banParticipant, ) else -> GuestSessionView( state = state, + playerState = viewModel.jamPlayerState.collectAsStateWithLifecycle().value, onShare = viewModel::share, onLeave = viewModel::leave, + onTogglePlayPause = viewModel::togglePlayPause, + onSkipNext = viewModel::skipNext, + onSkipPrevious = viewModel::skipPrevious, + onSeek = viewModel::seek, + onJumpTo = viewModel::jumpTo, + onToggleShuffle = viewModel::toggleShuffle, + onCycleLoop = viewModel::cycleLoopMode, ) } } @@ -220,16 +262,35 @@ private fun IncomingInviteView( @Composable private fun HostSessionView( state: JamUiState, + playerState: JamPlayerUiState, onNewInvite: () -> Unit, onSubmitAnswer: (String) -> Unit, onShare: (String) -> Unit, onLeave: () -> Unit, + onTogglePlayPause: () -> Unit, + onSkipNext: () -> Unit, + onSkipPrevious: () -> Unit, + onSeek: (Long) -> Unit, + onJumpTo: (Int) -> Unit, + onToggleShuffle: () -> Unit, + onCycleLoop: () -> Unit, + onKick: (String) -> Unit, + onBan: (String) -> Unit, ) { val clipboard = LocalClipboardManager.current var pastedAnswer by rememberSaveable { mutableStateOf("") } Column(verticalArrangement = Arrangement.spacedBy(16.dp)) { - ParticipantsSection(state.participants) + ParticipantsSection(state.participants, isHost = true, onKick = onKick, onBan = onBan) + + JamNowPlayingView( + playerState = playerState, + onTogglePlayPause = onTogglePlayPause, + onSkipNext = onSkipNext, + onSkipPrevious = onSkipPrevious, + onToggleShuffle = onToggleShuffle, + onCycleLoop = onCycleLoop, + ) HorizontalDivider() @@ -279,6 +340,13 @@ private fun HostSessionView( Text("Accept Answer") } + HorizontalDivider() + + JamQueueView( + queue = playerState.queue, + onJumpTo = onJumpTo, + ) + LeaveButton(onLeave) } } @@ -286,21 +354,37 @@ private fun HostSessionView( @Composable private fun GuestSessionView( state: JamUiState, + playerState: JamPlayerUiState, onShare: (String) -> Unit, onLeave: () -> Unit, + onTogglePlayPause: () -> Unit, + onSkipNext: () -> Unit, + onSkipPrevious: () -> Unit, + onSeek: (Long) -> Unit, + onJumpTo: (Int) -> Unit, + onToggleShuffle: () -> Unit, + onCycleLoop: () -> Unit, ) { val clipboard = LocalClipboardManager.current Column(verticalArrangement = Arrangement.spacedBy(16.dp)) { - ParticipantsSection(state.participants) + ParticipantsSection(state.participants, isHost = false, onKick = {}, onBan = {}) val answerLink = state.answerLink when { state.isConnected -> { - Text( - text = "Connected to the session", - style = MaterialTheme.typography.bodyMedium, - color = MaterialTheme.colorScheme.primary, + JamNowPlayingView( + playerState = playerState, + onTogglePlayPause = onTogglePlayPause, + onSkipNext = onSkipNext, + onSkipPrevious = onSkipPrevious, + onToggleShuffle = onToggleShuffle, + onCycleLoop = onCycleLoop, + ) + + JamQueueView( + queue = playerState.queue, + onJumpTo = onJumpTo, ) } @@ -331,7 +415,231 @@ private fun GuestSessionView( } @Composable -private fun ParticipantsSection(participants: List) { +private fun JamNowPlayingView( + playerState: JamPlayerUiState, + onTogglePlayPause: () -> Unit, + onSkipNext: () -> Unit, + onSkipPrevious: () -> Unit, + onToggleShuffle: () -> Unit, + onCycleLoop: () -> Unit, +) { + Column(verticalArrangement = Arrangement.spacedBy(12.dp)) { + Row( + modifier = Modifier.fillMaxWidth(), + verticalAlignment = Alignment.CenterVertically, + horizontalArrangement = Arrangement.spacedBy(12.dp), + ) { + AsyncImage( + model = playerState.currentCoverUrl?.takeIf { it.isNotBlank() }, + contentDescription = null, + contentScale = ContentScale.Crop, + modifier = Modifier + .size(64.dp) + .clip(MaterialTheme.shapes.medium), + ) + Column(modifier = Modifier.weight(1f)) { + Text( + text = playerState.currentTitle ?: "Nothing playing", + style = MaterialTheme.typography.bodyLarge, + fontWeight = FontWeight.SemiBold, + maxLines = 1, + overflow = TextOverflow.Ellipsis, + ) + Text( + text = playerState.currentArtist ?: "—", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + maxLines = 1, + overflow = TextOverflow.Ellipsis, + ) + } + } + + Row( + modifier = Modifier.fillMaxWidth(), + horizontalArrangement = Arrangement.SpaceBetween, + verticalAlignment = Alignment.CenterVertically, + ) { + Text( + text = formatJamDuration(playerState.positionMs), + style = MaterialTheme.typography.labelSmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + Text( + text = formatJamDuration(playerState.durationMs), + style = MaterialTheme.typography.labelSmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + + Row( + modifier = Modifier.fillMaxWidth(), + horizontalArrangement = Arrangement.SpaceEvenly, + verticalAlignment = Alignment.CenterVertically, + ) { + IconButton( + onClick = onToggleShuffle, + theme = LocalBaseUITheme.current.iconButtons.ghost.copyShape(CircleShape), + ) { + Icon( + imageVector = Iconsax.IconsaxShuffle, + contentDescription = "Shuffle", + tint = if (playerState.shuffleEnabled) { + MaterialTheme.colorScheme.primary + } else { + MaterialTheme.colorScheme.onSurfaceVariant + }, + ) + } + IconButton( + onClick = onSkipPrevious, + theme = LocalBaseUITheme.current.iconButtons.ghost.copyShape(CircleShape), + ) { + Icon(Iconsax.IconsaxPrevious, contentDescription = "Previous") + } + IconButton( + onClick = onTogglePlayPause, + theme = LocalBaseUITheme.current.iconButtons.primary.copyShape(CircleShape), + modifier = Modifier.size(64.dp), + ) { + Icon( + imageVector = if (playerState.isPlaying) { + Iconsax.IconsaxPause + } else { + Iconsax.IconsaxPlay + }, + contentDescription = if (playerState.isPlaying) "Pause" else "Play", + modifier = Modifier.size(32.dp), + ) + } + IconButton( + onClick = onSkipNext, + theme = LocalBaseUITheme.current.iconButtons.ghost.copyShape(CircleShape), + ) { + Icon(Iconsax.IconsaxNext, contentDescription = "Next") + } + IconButton( + onClick = onCycleLoop, + theme = LocalBaseUITheme.current.iconButtons.ghost.copyShape(CircleShape), + ) { + Icon( + imageVector = Iconsax.IconsaxRepeateMusic, + contentDescription = "Loop mode", + tint = if (playerState.loopMode != "none") { + MaterialTheme.colorScheme.primary + } else { + MaterialTheme.colorScheme.onSurfaceVariant + }, + ) + } + } + } +} + +@Composable +private fun JamQueueView( + queue: List, + onJumpTo: (Int) -> Unit, +) { + var expanded by rememberSaveable { mutableStateOf(false) } + + Column(verticalArrangement = Arrangement.spacedBy(8.dp)) { + Row( + modifier = Modifier + .fillMaxWidth() + .clickable { expanded = !expanded }, + verticalAlignment = Alignment.CenterVertically, + ) { + Text( + text = "Queue (${queue.size})", + style = MaterialTheme.typography.titleSmall, + modifier = Modifier.weight(1f), + ) + Icon( + imageVector = Iconsax.IconsaxArrowDown4, + contentDescription = if (expanded) "Collapse queue" else "Expand queue", + modifier = Modifier + .size(20.dp) + .graphicsLayer { rotationZ = if (expanded) 180f else 0f }, + tint = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + + if (queue.isEmpty()) { + Text( + text = "The queue is empty. Add tracks from anywhere in the app — the jam queue is shared.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } else if (expanded) { + LazyColumn( + modifier = Modifier + .fillMaxWidth() + .heightIn(max = 280.dp), + verticalArrangement = Arrangement.spacedBy(2.dp), + ) { + itemsIndexed(queue) { index, item -> + Row( + modifier = Modifier + .fillMaxWidth() + .clickable { onJumpTo(index) } + .padding(vertical = 6.dp, horizontal = 4.dp), + verticalAlignment = Alignment.CenterVertically, + horizontalArrangement = Arrangement.spacedBy(10.dp), + ) { + AsyncImage( + model = item.coverUrl.takeIf { it.isNotBlank() }, + contentDescription = null, + contentScale = ContentScale.Crop, + modifier = Modifier + .size(40.dp) + .clip(MaterialTheme.shapes.small), + ) + Column(modifier = Modifier.weight(1f)) { + Text( + text = item.title, + style = MaterialTheme.typography.bodyMedium, + maxLines = 1, + overflow = TextOverflow.Ellipsis, + color = if (item.isCurrent) { + MaterialTheme.colorScheme.primary + } else { + MaterialTheme.colorScheme.onSurface + }, + ) + Text( + text = item.artist, + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + maxLines = 1, + overflow = TextOverflow.Ellipsis, + ) + } + Text( + text = formatJamDuration(item.durationMs), + style = MaterialTheme.typography.labelSmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + } + } + } else { + Text( + text = "Tap to view the shared queue.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + } +} + +@Composable +private fun ParticipantsSection( + participants: List, + isHost: Boolean, + onKick: (String) -> Unit, + onBan: (String) -> Unit, +) { Column(verticalArrangement = Arrangement.spacedBy(8.dp)) { Text( text = "Participants (${participants.size})", @@ -357,11 +665,32 @@ private fun ParticipantsSection(participants: List = emptyList(), + val currentIndex: Int = -1, + val currentTitle: String? = null, + val currentArtist: String? = null, + val currentCoverUrl: String? = null, + val isPlaying: Boolean = false, + val positionMs: Long = 0, + val durationMs: Long = 0, + val shuffleEnabled: Boolean = false, + val loopMode: String = "none", +) + class JamViewModel( private val jamSession: JamSessionService, private val deepLinks: JamDeepLinkService, private val shareService: ShareService, private val settingsProvider: SettingsProvider, + private val audioPlayer: AudioPlayerInterface, + private val audioPlayerQueue: AudioPlayerQueue, ) : ViewModel() { private val _uiState = MutableStateFlow(JamUiState()) @@ -98,6 +136,144 @@ class JamViewModel( } } + /** + * The jam player state: the shared queue + current playback, built from the + * local player (the host's queue IS the jam queue; on guests the synced + * mirror lives in the local player). + */ + val jamPlayerState: StateFlow = combine( + audioPlayerQueue.queueFlow, + audioPlayerQueue.currentQueueEntryFlow, + audioPlayer.playlistFlow, + audioPlayer.currentMediaItemFlow, + audioPlayer.playerStateFlow, + audioPlayer.positionFlow, + audioPlayer.durationFlow, + audioPlayer.loopStateFlow, + audioPlayer.shuffleModeFlow, + ) { values -> + val queue: List = values[0] as List + val currentEntry: QueueEntry? = values[1] as QueueEntry? + val playlist: List = values[2] as List + val currentItem: MediaItem? = values[3] as MediaItem? + val playerState: PlayerState = values[4] as PlayerState + val position: kotlin.time.Duration = values[5] as kotlin.time.Duration + val duration: kotlin.time.Duration = values[6] as kotlin.time.Duration + val loop: LoopState = values[7] as LoopState + val shuffle: Boolean = values[8] as Boolean + + val isHost = jamSession.role.value == JamRole.Host + + val items: List + val currentIndex: Int + val currentTitle: String? + val currentArtist: String? + val currentCoverUrl: String? + + if (isHost) { + val queueItems = queue.map { JamMediaItem.fromQueueEntry(it) } + val index = if (currentEntry != null) { + queue.indexOfFirst { entry -> entry.matchesQueueEntry(currentEntry) } + } else { + -1 + } + items = queueItems.mapIndexed { i, item -> + item.toUiItem(i == index) + } + currentIndex = index + currentTitle = queueItems.getOrNull(index)?.title + currentArtist = queueItems.getOrNull(index)?.artist + currentCoverUrl = queueItems.getOrNull(index)?.coverUrl + } else { + val index = playlist.indexOf(currentItem) + items = playlist.mapIndexed { i, item -> + JamMediaItem.fromMediaItem(item).toUiItem(i == index) + } + currentIndex = index + currentTitle = currentItem?.title + currentArtist = currentItem?.artist + currentCoverUrl = currentItem?.coverURL + } + + JamPlayerUiState( + queue = items, + currentIndex = currentIndex, + currentTitle = currentTitle, + currentArtist = currentArtist, + currentCoverUrl = currentCoverUrl, + isPlaying = playerState == PlayerState.PLAYING, + positionMs = position.inWholeMilliseconds, + durationMs = duration.inWholeMilliseconds, + shuffleEnabled = shuffle, + loopMode = loop.name.lowercase(), + ) + }.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), JamPlayerUiState()) + + // ---------- Playback controls ---------- + + fun togglePlayPause() = sendOrApply(PlaybackCmd.Toggle) + + fun skipNext() = sendOrApply(PlaybackCmd.SkipNext) + + fun skipPrevious() = sendOrApply(PlaybackCmd.SkipPrevious) + + fun seek(positionMs: Long) = sendOrApply(PlaybackCmd.Seek(positionMs)) + + fun jumpTo(index: Int) = sendOrApply(PlaybackCmd.JumpTo(index)) + + fun toggleShuffle() = sendOrApply(PlaybackCmd.SetShuffle(!jamPlayerState.value.shuffleEnabled)) + + fun cycleLoopMode() { + val next = when (jamPlayerState.value.loopMode) { + "none" -> "one" + "one" -> "all" + else -> "none" + } + sendOrApply(PlaybackCmd.SetLoop(next)) + } + + private fun sendOrApply(command: PlaybackCmd) { + viewModelScope.launch { + if (jamSession.role.value == JamRole.Host) { + applyCommandLocally(command) + } else { + jamSession.sendMessage(JamMessage.PlaybackCommand(command)) + } + } + } + + private suspend fun applyCommandLocally(command: PlaybackCmd) { + when (command) { + PlaybackCmd.Play -> audioPlayer.play() + PlaybackCmd.Pause -> audioPlayer.pause() + PlaybackCmd.Toggle -> { + if (audioPlayer.playerStateFlow.value == PlayerState.PLAYING) { + audioPlayer.pause() + } else { + audioPlayer.play() + } + } + + is PlaybackCmd.Seek -> audioPlayer.seekTo(kotlin.time.Duration.parse("${command.positionMs}ms")) + PlaybackCmd.SkipNext -> audioPlayer.skipToNext() + PlaybackCmd.SkipPrevious -> audioPlayer.skipToPrevious() + is PlaybackCmd.SetVolume -> audioPlayer.setVolume(command.volume) + is PlaybackCmd.SetLoop -> audioPlayer.loop(JamLoopMapping.fromString(command.loop)) + is PlaybackCmd.SetShuffle -> audioPlayer.shuffle(command.enabled) + is PlaybackCmd.JumpTo -> audioPlayer.jumpTo(command.index) + } + } + + // ---------- Host moderation ---------- + + fun kickParticipant(participantId: String) { + viewModelScope.launch { jamSession.kickParticipant(participantId) } + } + + fun banParticipant(participantId: String) { + viewModelScope.launch { jamSession.banParticipant(participantId) } + } + fun createSession() { viewModelScope.launch { runCatching { @@ -228,4 +404,26 @@ class JamViewModel( private fun localName(): String = settingsProvider.settingsState.value?.jamParticipantName.orEmpty() +} + +private fun JamMediaItem.toUiItem(isCurrent: Boolean): JamQueueUiItem = JamQueueUiItem( + id = if (trackId.isNotBlank()) trackId else url, + title = title, + artist = artist, + album = album, + durationMs = durationMs, + coverUrl = coverUrl, + isCurrent = isCurrent, +) + +private fun QueueEntry.matchesQueueEntry(other: QueueEntry): Boolean { + return when { + this is QueueEntry.StreamingTrack && other is QueueEntry.StreamingTrack -> + this.track.id == other.track.id + + this is QueueEntry.LocalTrack && other is QueueEntry.LocalTrack -> + this.url == other.url && this.name == other.name + + else -> false + } } \ No newline at end of file diff --git a/composeApp/src/commonMain/rust/webrtc_p2p.rs b/composeApp/src/commonMain/rust/webrtc_p2p.rs index 3b60d756..c40e9a69 100644 --- a/composeApp/src/commonMain/rust/webrtc_p2p.rs +++ b/composeApp/src/commonMain/rust/webrtc_p2p.rs @@ -78,7 +78,7 @@ struct DataChannelEntry { pub struct WebrtcPeerConnection { pc: Arc, handler: Arc, - channels: Mutex>, + channels: Arc>>, gather_rx: Mutex>, } @@ -122,9 +122,11 @@ pub async fn create_webrtc_peer_connection( setting_engine.set_multicast_dns_mode(MulticastDnsMode::Disabled); let (gather_tx, gather_rx) = channel::<()>(1); + let channels = Arc::new(Mutex::new(Vec::new())); let pc_handler = Arc::new(PeerHandlerBridge { handler: Arc::clone(&handler), gather_tx, + channels: Arc::clone(&channels), }); let pc = PeerConnectionBuilder::new() @@ -140,7 +142,7 @@ pub async fn create_webrtc_peer_connection( Ok(Arc::new(WebrtcPeerConnection { pc: Arc::new(pc) as Arc, handler, - channels: Mutex::new(Vec::new()), + channels, gather_rx: Mutex::new(gather_rx), })) } @@ -150,17 +152,34 @@ impl WebrtcPeerConnection { /// candidates (non-trickle exchange). Must be called after `set_local_description`, /// which is what starts gathering. /// - /// Bounded by a timeout so a stalled gatherer (e.g. a platform that never reports - /// completion) can never hang `create_offer`/`create_answer` forever — the SDP - /// with the candidates gathered so far is returned instead. + /// Robust against a stalled gatherer (e.g. an unreachable STUN server): once at + /// least one candidate has landed in the local description, a short grace period + /// is enough — the SDP must never leave candidate-less. Hard cap at 5s. async fn wait_for_ice_gathering(&self) { let mut gather_rx = self.gather_rx.lock().clone(); - match tokio::time::timeout(Duration::from_secs(5), gather_rx.recv()).await { - Ok(_) => {} - Err(_) => { + let started = std::time::Instant::now(); + + loop { + let elapsed = started.elapsed(); + if elapsed >= Duration::from_secs(5) { log::warn!( - "ICE gathering did not complete within 5s; returning SDP with the candidates gathered so far" + "ICE gathering did not complete within 5s; using the candidates gathered so far" ); + return; + } + + match tokio::time::timeout(Duration::from_millis(100), gather_rx.recv()).await { + Ok(Some(())) => return, // gathering complete + Ok(None) => return, // handler dropped + Err(_) => {} // timed out, keep waiting + } + + // Grace period once candidates are present, so the SDP always carries them. + if elapsed >= Duration::from_secs(1) { + let sdp = self.pc.local_description().await.map(|d| d.sdp); + if sdp.as_deref().map_or(false, |s| s.contains("a=candidate:")) { + return; + } } } } @@ -251,6 +270,7 @@ impl WebrtcPeerConnection { struct PeerHandlerBridge { handler: Arc, gather_tx: webrtc::runtime::Sender<()>, + channels: Arc>>, } #[async_trait::async_trait] @@ -272,6 +292,15 @@ impl PeerConnectionEventHandler for PeerHandlerBridge { } async fn on_data_channel(&self, dc: Arc) { + // Register in-band (remote-initiated) channels so send_data() can find + // them — without this, the answering peer can never send anything. + let label = match dc.label().await { + Ok(l) => l, + Err(_) => return, + }; + self.channels + .lock() + .push(DataChannelEntry { dc: Arc::clone(&dc), label }); spawn_data_channel_poll_loop(dc, Arc::clone(&self.handler)); } }