diff --git a/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt b/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt index 9f9441c3..8d0c631f 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt @@ -6,8 +6,8 @@ import com.subtracks.data.model.Source import com.subtracks.data.model.SubsonicConfig import com.subtracks.data.model.SubsonicSource import com.subtracks.data.prefs.UserPreferences +import com.subtracks.data.source.MusicSource import com.subtracks.data.source.subsonic.SubsonicClient -import com.subtracks.data.sync.SyncService import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob @@ -16,8 +16,6 @@ import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import okhttp3.HttpUrl.Companion.toHttpUrl import okhttp3.OkHttpClient @@ -29,7 +27,6 @@ class SourceRepository( private val prefs: UserPreferences, ) { private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - private val syncLock = Mutex() @Volatile private var active: SubsonicMusicSource? = null @@ -118,13 +115,10 @@ class SourceRepository( runCatching { client(address, username, password, useTokenAuth).check("ping") } } - suspend fun sync(): Result = - syncLock.withLock { - runCatching { - val config = db.sourcesDao().activeSubsonicConfigOnce() ?: error("No server configured") - SyncService(db, config.toMusicSource(prefs.maxBitrate.first(), prefs.streamFormat.first())).sync() - } - } + suspend fun activeMusicSource(): MusicSource? { + val config = db.sourcesDao().activeSubsonicConfigOnce() ?: return null + return config.toMusicSource(prefs.maxBitrate.first(), prefs.streamFormat.first()) + } private fun SubsonicConfig.toMusicSource( maxBitrate: Int, diff --git a/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt b/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt index bb4b4888..8278d6d2 100644 --- a/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt +++ b/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt @@ -25,6 +25,7 @@ class SubsonicSource( private val client: SubsonicClient, private val maxBitrate: Int = 0, private val streamFormat: String? = null, + private val maxPages: Int = MAX_PAGES, ) : MusicSource { private var emptyQuerySearchSupported: Boolean? = null @@ -45,7 +46,7 @@ class SubsonicSource( val (frequent, recent) = fetchRanks() var offset = 0 var pages = 0 - while (pages < MAX_PAGES) { + while (pages < maxPages) { var count = 0 entityBatches( method = "getAlbumList2", @@ -69,7 +70,8 @@ class SubsonicSource( offset += PAGE_SIZE pages++ } - }.flowOn(Dispatchers.IO) + checkPageCap(pages) + }.flowOn(Dispatchers.IO).buffer(1) override fun songs(): Flow> = flow { @@ -78,7 +80,7 @@ class SubsonicSource( } else { emitAll(albumSongs()) } - }.flowOn(Dispatchers.IO) + }.flowOn(Dispatchers.IO).buffer(1) override fun playlists(): Flow> = entityBatches( @@ -96,10 +98,14 @@ class SubsonicSource( method = "getPlaylist", params = mapOf("id" to playlistId), tag = "entry", - create = { attrs -> SubsonicXml.playlistSong(id, playlistId, position++, attrs) }, + create = { attrs -> + val entryId = attrs.attr("id") + SubsonicXml.playlistSong(id, playlistId, if (entryId.isEmpty()) -1L else position++, attrs) + }, + accept = { it.songId.isNotEmpty() }, ).collect { emit(it) } } - }.flowOn(Dispatchers.IO) + }.flowOn(Dispatchers.IO).buffer(1) fun streamUri(songId: String) = client.uri( @@ -132,7 +138,7 @@ class SubsonicSource( flow { var offset = 0 var pages = 0 - while (pages < MAX_PAGES) { + while (pages < maxPages) { var count = 0 entityBatches( method = "search3", @@ -155,13 +161,14 @@ class SubsonicSource( offset += PAGE_SIZE pages++ } + checkPageCap(pages) } private fun albumSongs(): Flow> = flow { var offset = 0 var pages = 0 - while (pages < MAX_PAGES) { + while (pages < maxPages) { val albums = ArrayList(PAGE_SIZE) entityBatches( method = "getAlbumList2", @@ -183,34 +190,39 @@ class SubsonicSource( offset += PAGE_SIZE pages++ } + checkPageCap(pages) } private suspend fun fetchRanks(): Pair, Map> = fetchRank("frequent") to fetchRank("recent") - private suspend fun fetchRank(type: String): Map { - val ranks = HashMap() - var offset = 0 - var pages = 0 - while (pages < MAX_PAGES) { - var index = 0 - entityBatches( - method = "getAlbumList2", - params = page(type, offset), - tag = "album", - create = { it.attr("id") }, - accept = { it.isNotEmpty() }, - ).collect { batch -> - batch.forEach { albumId -> - ranks[albumId] = (offset + index).toLong() - index++ + private suspend fun fetchRank(type: String): Map = + try { + val ranks = HashMap() + var offset = 0 + var pages = 0 + while (pages < maxPages) { + var index = 0 + entityBatches( + method = "getAlbumList2", + params = page(type, offset), + tag = "album", + create = { it.attr("id") }, + accept = { it.isNotEmpty() }, + ).collect { batch -> + batch.forEach { albumId -> + ranks[albumId] = (offset + index).toLong() + index++ + } } + if (index < PAGE_SIZE) break + offset += PAGE_SIZE + pages++ } - if (index < PAGE_SIZE) break - offset += PAGE_SIZE - pages++ + checkPageCap(pages) + ranks + } catch (_: SubsonicException) { + emptyMap() } - return ranks - } private fun supportsEmptyQuerySearch(): Boolean { emptyQuerySearchSupported?.let { return it } @@ -227,6 +239,10 @@ class SubsonicSource( return supported } + private fun checkPageCap(pages: Int) { + if (pages >= maxPages) throw SubsonicException(-1, "Library exceeds the $maxPages page cap") + } + private fun entityBatches( method: String, params: Map = emptyMap(), diff --git a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt index 0d27bb34..8e90af65 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt +++ b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt @@ -1,14 +1,17 @@ package com.subtracks.data.sync import android.util.Log +import com.subtracks.data.db.SubtracksDatabase import com.subtracks.data.repo.QueueRepository import com.subtracks.data.repo.SourceRepository +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex sealed interface SyncStatus { data object Idle : SyncStatus @@ -23,20 +26,38 @@ sealed interface SyncStatus { } class SyncManager( + private val db: SubtracksDatabase, private val sourceRepository: SourceRepository, private val queueRepository: QueueRepository, ) { private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + private val syncLock = Mutex() private val _status = MutableStateFlow(SyncStatus.Idle) val status: StateFlow = _status fun requestSync() { - scope.launch { runSync() } + if (!syncLock.tryLock()) return + scope.launch { + try { + runSync() + } finally { + syncLock.unlock() + } + } } private suspend fun runSync() { _status.value = SyncStatus.Running - val result = sourceRepository.sync() + val result = + try { + val source = sourceRepository.activeMusicSource() ?: error("No server configured") + SyncService(db, source).sync() + Result.success(Unit) + } catch (cancellation: CancellationException) { + throw cancellation + } catch (failure: Exception) { + Result.failure(failure) + } queueRepository.invalidateLibraryCache() _status.value = result.fold( diff --git a/app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt b/app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt index a3dea53e..9a91a425 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt +++ b/app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt @@ -117,17 +117,18 @@ class SyncService( } private suspend fun deleteStaleDiscs(stale: Set) { - if (stale.isEmpty()) return - db.useWriterConnection { connection -> - connection.immediateTransaction { - connection.usePrepared("DELETE FROM discs WHERE sourceId = ? AND albumId = ? AND disc = ?") { statement -> - stale.forEach { key -> - statement.bind(1, source.id) - statement.bind(2, key.albumId) - statement.bind(3, key.disc) - statement.step() - statement.reset() - statement.clearBindings() + stale.chunked(DELETE_CHUNK).forEach { chunk -> + db.useWriterConnection { connection -> + connection.immediateTransaction { + connection.usePrepared("DELETE FROM discs WHERE sourceId = ? AND albumId = ? AND disc = ?") { statement -> + chunk.forEach { key -> + statement.bind(1, source.id) + statement.bind(2, key.albumId) + statement.bind(3, key.disc) + statement.step() + statement.reset() + statement.clearBindings() + } } } } diff --git a/app/src/main/kotlin/com/subtracks/di/AppModule.kt b/app/src/main/kotlin/com/subtracks/di/AppModule.kt index 6e077cb6..01982ed9 100644 --- a/app/src/main/kotlin/com/subtracks/di/AppModule.kt +++ b/app/src/main/kotlin/com/subtracks/di/AppModule.kt @@ -34,7 +34,7 @@ fun appModule( single { SourceRepository(get(), get(), get()) } single { LibraryRepository(get(), get()) } single { ArtworkSeedRepository(get()) } - single { SyncManager(get(), get()) } + single { SyncManager(get(), get(), get()) } single { QueueRepository(get()) } single { MediaSessionConnection(context.applicationContext, get()) } single { PlaybackController(get(), get(), get()) } diff --git a/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt b/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt index 14405cf4..a78d08c2 100644 --- a/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt @@ -179,6 +179,49 @@ class SubsonicSourceTest { assertEquals(500, firstBatch.size) } + @Test + fun albumsAbortWhenThePageCapIsHit() = + runBlocking { + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + if (url.encodedPath != "/rest/getAlbumList2.view") return MockResponse().setResponseCode(404) + return if (url.queryParameter("type") == "newest") { + MockResponse().setBody(albumPage(0, 500)) + } else { + MockResponse().setBody(emptyAlbumList()) + } + } + } + + val failure = + runCatching { + SubsonicSource(1, client(), maxPages = 2).albums().toList() + } + + assertEquals(true, failure.exceptionOrNull() is SubsonicException) + } + + @Test + fun playlistEntriesWithoutIdsAreSkipped() = + runBlocking { + server.enqueue( + MockResponse().setBody( + "" + + "" + + "" + + "" + + "", + ), + ) + + val entries = SubsonicSource(1, client()).playlistSongs(listOf("p1")).toList().flatten() + + assertEquals(listOf("s1", "s2"), entries.map { it.songId }) + assertEquals(listOf(0L, 1L), entries.map { it.position }) + } + @Test fun streamUriRequestsTranscodingWhenABitrateIsSet() { val uri = SubsonicSource(1, client(), maxBitrate = 128).streamUri("s1").toString() diff --git a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt b/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt index 83424c51..b3e199a9 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt @@ -37,7 +37,7 @@ class SyncManagerTest { .build() sourceRepository = SourceRepository(db, OkHttpClient(), fakeUserPreferences()) queueRepository = QueueRepository(db) - manager = SyncManager(sourceRepository, queueRepository) + manager = SyncManager(db, sourceRepository, queueRepository) } @After diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index c096fbfc..14cdcd77 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -12,7 +12,7 @@ Everything needed to mirror a server library locally: - `MusicSource` abstraction with `SubsonicClient`, `SubsonicXml` and `SubsonicSource`. - Room 3 schema on bundled SQLite, so the SQLite version is identical on every device. -- `SyncService`: fetch first, then a single transaction of upserts with diff-based pruning. +- `SyncService`: stream each entity in batches, upserting per batch and pruning the IDs the server no longer lists; the sync is not atomic. - Unit tests (client, XML, sync), native integration tests against navidrome and gonic, and CI (`unit`, `lint`, `integration`) with warm caches. ## Phase 2 - UI vertical slice (done)