From 548bf3913670dd61ab6adc338e9fe0ac7c3d68e8 Mon Sep 17 00:00:00 2001 From: austinried <4966622+austinried@users.noreply.github.com> Date: Thu, 12 Nov 2026 07:55:18 +0000 Subject: [PATCH] sync: reset queues only for successfully changed origins --- .../com/subtracks/data/sync/SyncManager.kt | 25 +- .../com/subtracks/data/sync/SyncService.kt | 486 +++++++++++++++--- .../subtracks/data/sync/SyncManagerTest.kt | 230 +++++++++ .../subtracks/data/sync/SyncServiceTest.kt | 274 ++++++++++ 4 files changed, 942 insertions(+), 73 deletions(-) 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 5ea1e779..46597fe2 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt +++ b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt @@ -3,8 +3,10 @@ package com.subtracks.data.sync import com.subtracks.R import com.subtracks.UiMessage import com.subtracks.data.db.SubtracksDatabase +import com.subtracks.data.model.QueueOrigin import com.subtracks.data.prefs.UserPreferences import com.subtracks.data.repo.QueueRepository +import com.subtracks.data.repo.QueueResetPoint import com.subtracks.data.repo.SourceRepository import com.subtracks.data.source.ServerActionSink import com.subtracks.data.source.subsonic.SubsonicException @@ -81,17 +83,25 @@ class SyncManager( runCatching { preferences.setLastSyncAt(System.currentTimeMillis()) } .onFailure { Log.w("sync", "could not record the sync attempt", it) } val startedAt = System.nanoTime() + var summary: SyncSummary? = null + var resetPoint: QueueResetPoint? = null + var syncSourceId: Long? = null val result = try { val source = sourceRepository.activeMusicSource() ?: throw NoServerException() Log.i("sync", "started (source ${source.id})") serverActions.flush() - val summary = SyncService(db, source).sync() + resetPoint = queueRepository.captureResetPoint() + syncSourceId = source.id + val service = SyncService(db, source) + val completed = service.sync() + val pending = db.queueDao().pendingSyncOrigins(source.id).map { QueueOrigin(it.kind, it.sourceId, it.refId) } + summary = completed.copy(changedOrigins = pending.toSet() + completed.changedOrigins) Log.i( "sync", "finished in ${elapsedMs(startedAt)}ms: " + - "${summary.artists} artists, ${summary.albums} albums, ${summary.songs} songs, " + - "${summary.playlists} playlists, ${summary.playlistSongs} playlist tracks, ${summary.pruned} pruned", + "${completed.artists} artists, ${completed.albums} albums, ${completed.songs} songs, " + + "${completed.playlists} playlists, ${completed.playlistSongs} playlist tracks, ${completed.pruned} pruned", ) SyncStatus.Success } catch (cancellation: CancellationException) { @@ -112,7 +122,14 @@ class SyncManager( if (!silent) showMessage(message) SyncStatus.Failed(message) } - queueRepository.invalidateLibraryCache() + if (result is SyncStatus.Success) { + val completed = summary + val point = resetPoint + if (completed != null && point != null) { + queueRepository.resetChangedOrigins(completed.changedOrigins, point) + db.queueDao().clearPendingSyncOrigins(sourceId = checkNotNull(syncSourceId)) + } + } _status.value = result } 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 9efb0cc7..74fe376d 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt +++ b/app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt @@ -14,6 +14,8 @@ import com.subtracks.data.model.Disc import com.subtracks.data.model.DiscKey import com.subtracks.data.model.Playlist import com.subtracks.data.model.PlaylistSong +import com.subtracks.data.model.QueueKind +import com.subtracks.data.model.QueueOrigin import com.subtracks.data.model.Song import com.subtracks.data.model.SongArtistRef import com.subtracks.data.model.SongGenre @@ -30,6 +32,27 @@ class SyncService( private var playlistsSynced = 0 private var playlistSongsSynced = 0 private var pruned = 0 + private val committedOrigins = mutableSetOf() + private var transactionOrigins: MutableSet? = null + private val changedOrigins = + object : AbstractMutableSet() { + override val size: Int get() = committedOrigins.size + + override fun iterator(): MutableIterator = committedOrigins.iterator() + + override fun add(element: QueueOrigin): Boolean = (transactionOrigins ?: committedOrigins).add(element) + } + + private suspend fun commitOrigins(block: suspend () -> T): T { + check(transactionOrigins == null) + val pending = mutableSetOf() + transactionOrigins = pending + return try { + block().also { committedOrigins.addAll(pending) } + } finally { + transactionOrigins = null + } + } suspend fun sync(): SyncSummary { val sources = db.sourcesDao() @@ -51,6 +74,7 @@ class SyncService( playlists = playlistsSynced, playlistSongs = playlistSongsSynced, pruned = pruned, + changedOrigins = changedOrigins.toSet(), ) } @@ -61,13 +85,24 @@ class SyncService( source.artists().collect { batch -> if (batch.isEmpty()) return@collect changed = - write("artists", primaryKey, batch, preserved = setOf("albumCount"), markStale = true) { it.columns() } || + write( + "artists", + primaryKey, + batch, + preserved = setOf("albumCount"), + markStale = true, + onChanged = { changedOrigins += origin(QueueKind.Artist, it.id) }, + ) { it.columns() } || changed batch.forEach { seen.add(idHash(it.id)) } artistsSynced += batch.size } val prunedBefore = pruned - pruned += pruneStaleIds(seen, { library.artistIdsAfter(source.id, it, PRUNE_PAGE) }, { library.deleteArtists(source.id, it) }) + pruned += + pruneStaleIds(seen, { library.artistIdsAfter(source.id, it, PRUNE_PAGE) }, { + it.forEach { id -> changedOrigins += origin(QueueKind.Artist, id) } + deleteSyncedIds("artists", it) + }) return changed || pruned > prunedBefore } @@ -92,7 +127,18 @@ class SyncService( pruneStaleIds( seenAlbums, { library.albumIdsAfter(source.id, it, PRUNE_PAGE) }, - { library.deleteAlbums(source.id, it) }, + { + it.forEach { id -> changedOrigins += origin(QueueKind.Album, id) } + db.useWriterConnection { connection -> + connection.immediateTransaction { + val artistIds = connection.albumArtistIds(it.toSet()).values.flatten() + val trackArtistIds = connection.songArtistIds(it.toSet()).values.flatten() + changedOrigins += (artistIds + trackArtistIds).map { id -> origin(QueueKind.Artist, id) } + connection.deleteIds("albums", it) + connection.persistPendingOrigins() + } + } + }, markStale = { db.sourcesDao().setCreditsStale(source.id, true) }, ) pruned += @@ -117,7 +163,9 @@ class SyncService( pruneStaleIds( seen, { library.songIdsAfter(source.id, it, PRUNE_PAGE) }, - { library.deleteSongs(source.id, it) }, + { + deleteSongsAndRecordOrigins(it) + }, markStale = { db.sourcesDao().setCreditsStale(source.id, true) }, ) return changed || pruned > prunedBefore @@ -133,40 +181,227 @@ class SyncService( batch.flatMap { song -> song.artistPositions().map { (position, ref) -> SongArtistRef(song.sourceId, song.id, position, ref.id, ref.name) } } - return db.useWriterConnection { connection -> - connection.immediateTransaction { - val before = connection.totalChanges() - val resolved = connection.resolveSongSortNames(batch) - connection.upsertChanged("songs", primaryKey, resolved) { it.columns() } - connection.usePrepared("DELETE FROM song_genres WHERE sourceId = ? AND songId = ? AND position >= ?") { statement -> - batch.forEach { song -> - statement.bind(1, song.sourceId) - statement.bind(2, song.id) - statement.bind(3, names.getValue(song.id).size.toLong()) - statement.step() - statement.reset() - statement.clearBindings() + return commitOrigins { + db.useWriterConnection { connection -> + connection.immediateTransaction { + val before = connection.totalChanges() + val previousStates = connection.storedSongSyncStates(batch.map { it.id }) + val albumArtists = connection.albumArtistIds(batch.mapNotNull { it.albumId }.toSet()) + val resolved = connection.resolveSongSortNames(batch) + val nextStates = resolved.associate { it.id to it.syncState(albumArtists) } + val downloadedIds = connection.completedDownloadIds(batch.map { it.id }) + val changedSongIds = mutableSetOf() + connection.upsertChanged("songs", primaryKey, resolved, onChanged = { changedSongIds += it.id }) { it.columns() } + connection.usePrepared("DELETE FROM song_genres WHERE sourceId = ? AND songId = ? AND position >= ?") { statement -> + batch.forEach { song -> + statement.bind(1, song.sourceId) + statement.bind(2, song.id) + statement.bind(3, names.getValue(song.id).size.toLong()) + statement.step() + statement.reset() + statement.clearBindings() + } } - } - connection.upsertChanged("song_genres", songGenreKey, genres) { it.columns() } - connection.usePrepared("DELETE FROM song_artists WHERE sourceId = ? AND songId = ? AND position >= ?") { statement -> - batch.forEach { song -> - statement.bind(1, song.sourceId) - statement.bind(2, song.id) - statement.bind(3, song.artistPositions().size.toLong()) - statement.step() - statement.reset() - statement.clearBindings() + connection.upsertChanged("song_genres", songGenreKey, genres) { it.columns() } + connection.usePrepared("DELETE FROM song_artists WHERE sourceId = ? AND songId = ? AND position >= ?") { statement -> + batch.forEach { song -> + statement.bind(1, song.sourceId) + statement.bind(2, song.id) + statement.bind(3, song.artistPositions().size.toLong()) + statement.step() + statement.reset() + statement.clearBindings() + } } - } - connection.upsertChanged("song_artists", songArtistKey, artists) { it.columns() } - (connection.totalChanges() != before).also { changed -> + connection.upsertChanged("song_artists", songArtistKey, artists) { it.columns() } + batch.forEach { song -> + val previous = previousStates[song.id] + val next = nextStates.getValue(song.id) + if (song.id in changedSongIds || previous != next) { + recordSongOrigins(previous, next) + } + if (song.id in downloadedIds && (previous == null || previous.sortName != next.sortName)) { + changedOrigins += origin(QueueKind.Downloaded, "") + } + } + val changed = connection.totalChanges() != before if (changed) connection.markCreditsStale(batch.first().sourceId) + connection.persistPendingOrigins() + changed } } } } + private suspend fun deleteSongsAndRecordOrigins(ids: Collection) { + ids.forEach { changedOrigins += origin(QueueKind.Song, it) } + db.useWriterConnection { connection -> + connection.immediateTransaction { + connection.storedSongSyncStates(ids.toList()).values.forEach { recordSongOrigins(it, null) } + if (connection.completedDownloadIds(ids.toList()).isNotEmpty()) changedOrigins += origin(QueueKind.Downloaded, "") + connection.deleteIds("songs", ids) + connection.persistPendingOrigins() + } + } + } + + private suspend fun PooledConnection.recordSongOrigins( + previous: SongSyncState?, + next: SongSyncState?, + ) { + if ((previous == null) != (next == null)) { + val songId = (previous ?: next)!!.id + usePrepared("SELECT playlistId FROM playlist_songs WHERE sourceId = ? AND songId = ?") { statement -> + statement.bindLong(1, source.id) + statement.bindText(2, songId) + while (statement.step()) changedOrigins += origin(QueueKind.Playlist, statement.getText(0)) + } + } + val states = listOfNotNull(previous, next) + states.mapTo(changedOrigins) { origin(QueueKind.Song, it.id) } + states.mapNotNullTo(changedOrigins) { it.albumId?.let { id -> origin(QueueKind.Album, id) } } + states.flatMapTo(changedOrigins) { state -> state.artistIds.map { origin(QueueKind.Artist, it) } } + states.flatMapTo(changedOrigins) { state -> state.genres.map { origin(QueueKind.Genre, it) } } + if (states.any { it.starred }) changedOrigins += origin(QueueKind.Starred, "") + } + + private suspend fun PooledConnection.storedSongSyncStates(ids: List): Map { + val states = mutableMapOf() + readSourceIds( + ids, + { "SELECT id, albumId, artistId, sortName, starred FROM songs WHERE sourceId = ? AND id IN ($it)" }, + ) { statement -> + val state = MutableSongSyncState() + state.albumId = if (statement.isNull(1)) null else statement.getText(1) + if (!statement.isNull(2)) state.artistIds += statement.getText(2) + state.sortName = statement.getText(3) + state.starred = !statement.isNull(4) + states[statement.getText(0)] = state + } + readSourceIds(ids, { "SELECT songId, artistId FROM song_artists WHERE sourceId = ? AND songId IN ($it)" }) { statement -> + states[statement.getText(0)]?.artistIds?.plusAssign(statement.getText(1)) + } + readSourceIds(ids, { "SELECT songId, genre FROM song_genres WHERE sourceId = ? AND songId IN ($it)" }) { statement -> + states[statement.getText(0)]?.genres?.plusAssign(statement.getText(1)) + } + val albumArtists = albumArtistIds(states.values.mapNotNull { it.albumId }.toSet()) + return states.mapValues { (id, state) -> + SongSyncState( + id = id, + albumId = state.albumId, + artistIds = state.artistIds + state.albumId?.let { albumArtists[it].orEmpty() }.orEmpty(), + genres = state.genres, + starred = state.starred, + sortName = state.sortName, + ) + } + } + + private suspend fun PooledConnection.completedDownloadIds(ids: List): Set { + val result = mutableSetOf() + readSourceIds( + ids, + { "SELECT songId FROM song_downloads WHERE sourceId = ? AND status = 'Completed' AND songId IN ($it)" }, + ) { statement -> + result += statement.getText(0) + } + return result + } + + private suspend fun PooledConnection.albumArtistIds(albumIds: Set): Map> { + val artists = mutableMapOf>() + readSourceIds(albumIds, { "SELECT id, artistId FROM albums WHERE sourceId = ? AND artistId IS NOT NULL AND id IN ($it)" }) { + artists.getOrPut(it.getText(0)) { mutableSetOf() } += it.getText(1) + } + readSourceIds(albumIds, { "SELECT albumId, artistId FROM album_artists WHERE sourceId = ? AND albumId IN ($it)" }) { + artists.getOrPut(it.getText(0)) { mutableSetOf() } += it.getText(1) + } + return artists + } + + private suspend fun PooledConnection.songArtistIds(albumIds: Set): Map> { + val result = mutableMapOf>() + readSourceIds(albumIds, { + "SELECT albumId, artistId FROM songs WHERE sourceId = ? AND artistId IS NOT NULL AND albumId IN ($it)" + }) { statement -> + result.getOrPut(statement.getText(0)) { mutableSetOf() } += statement.getText(1) + } + albumIds.chunked(SQL_VARIABLES).forEach { chunk -> + val placeholders = chunk.joinToString(",") { "?" } + usePrepared( + "SELECT songs.albumId, song_artists.artistId FROM song_artists " + + "JOIN songs ON songs.sourceId = song_artists.sourceId AND songs.id = song_artists.songId " + + "WHERE song_artists.sourceId = ? AND songs.albumId IN ($placeholders)", + ) { statement -> + statement.bindLong(1, source.id) + chunk.forEachIndexed { index, id -> statement.bindText(index + 2, id) } + while (statement.step()) result.getOrPut(statement.getText(0)) { mutableSetOf() } += statement.getText(1) + } + } + return result + } + + private suspend fun PooledConnection.readSourceIds( + ids: Collection, + sql: (placeholders: String) -> String, + row: (SQLiteStatement) -> Unit, + ) { + ids.chunked(SQL_VARIABLES).forEach { chunk -> + usePrepared(sql(chunk.joinToString(",") { "?" })) { statement -> + statement.bindLong(1, source.id) + chunk.forEachIndexed { index, id -> statement.bindText(index + 2, id) } + while (statement.step()) row(statement) + } + } + } + + private fun origin( + kind: QueueKind, + refId: String, + ) = QueueOrigin(kind, source.id, refId) + + private suspend fun PooledConnection.persistPendingOrigins() { + val origins = transactionOrigins.orEmpty() + origins.chunked(SQL_VARIABLES).forEach { chunk -> + usePrepared("INSERT OR IGNORE INTO pending_sync_origins (sourceId, kind, refId) VALUES (?, ?, ?)") { statement -> + chunk.forEach { origin -> + statement.bindLong(1, origin.sourceId) + statement.bindText(2, origin.kind.name) + statement.bindText(3, origin.refId) + statement.step() + statement.reset() + statement.clearBindings() + } + } + } + } + + private suspend fun PooledConnection.deleteIds( + table: String, + ids: Collection, + ) { + ids.chunked(SQL_VARIABLES).forEach { chunk -> + val placeholders = chunk.joinToString(",") { "?" } + usePrepared("DELETE FROM $table WHERE sourceId = ? AND id IN ($placeholders)") { statement -> + statement.bindLong(1, source.id) + chunk.forEachIndexed { index, id -> statement.bindText(index + 2, id) } + statement.step() + } + } + } + + private suspend fun deleteSyncedIds( + table: String, + ids: Collection, + ) { + db.useWriterConnection { connection -> + connection.immediateTransaction { + connection.deleteIds(table, ids) + connection.persistPendingOrigins() + } + } + } + private suspend fun recomputeAggregates() { db.libraryDao().recomputeAggregates(source.id) } @@ -180,6 +415,15 @@ class SyncService( return value } + private suspend fun PooledConnection.changes(): Long { + var value = 0L + usePrepared("SELECT changes()") { statement -> + statement.step() + value = statement.getLong(0) + } + return value + } + private suspend fun PooledConnection.markCreditsStale(sourceId: Long) { usePrepared("UPDATE sources SET creditsStale = 1 WHERE id = ? AND creditsStale = 0") { statement -> statement.bindLong(1, sourceId) @@ -256,11 +500,15 @@ class SyncService( val seen = HashedIds() source.playlists().collect { batch -> if (batch.isEmpty()) return@collect - write("playlists", primaryKey, batch) { it.columns() } + write("playlists", primaryKey, batch, onChanged = { changedOrigins += origin(QueueKind.Playlist, it.id) }) { it.columns() } batch.forEach { seen.add(idHash(it.id)) } playlistsSynced += batch.size } - pruned += pruneStaleIds(seen, { library.playlistIdsAfter(source.id, it, PRUNE_PAGE) }, { library.deletePlaylists(source.id, it) }) + pruned += + pruneStaleIds(seen, { library.playlistIdsAfter(source.id, it, PRUNE_PAGE) }, { + it.forEach { id -> changedOrigins += origin(QueueKind.Playlist, id) } + deleteSyncedIds("playlists", it) + }) } private suspend fun syncPlaylistSongs() { @@ -269,14 +517,43 @@ class SyncService( val counts = HashMap() source.playlistSongs(playlistIds).collect { batch -> if (batch.isEmpty()) return@collect - write("playlist_songs", playlistSongKey, batch) { it.columns() } + write("playlist_songs", playlistSongKey, batch, onChanged = { changedOrigins += origin(QueueKind.Playlist, it.playlistId) }) { + it.columns() + } batch.forEach { counts[it.playlistId] = (counts[it.playlistId] ?: 0L) + 1L } playlistSongsSynced += batch.size } - playlistIds.forEach { library.deletePlaylistSongsFrom(source.id, it, counts[it] ?: 0L) } + playlistIds.forEach { playlistId -> + deletePlaylistSongsFrom(playlistId, counts[playlistId] ?: 0L) + } library.deleteOrphanPlaylistSongs(source.id) } + private suspend fun deletePlaylistSongsFrom( + playlistId: String, + position: Long, + ): Boolean = + commitOrigins { + db.useWriterConnection { connection -> + connection.immediateTransaction { + connection.usePrepared( + "DELETE FROM playlist_songs WHERE sourceId = ? AND playlistId = ? AND position >= ?", + ) { statement -> + statement.bindLong(1, source.id) + statement.bindText(2, playlistId) + statement.bindLong(3, position) + statement.step() + } + val changed = connection.changes() > 0 + if (changed) { + changedOrigins += origin(QueueKind.Playlist, playlistId) + connection.persistPendingOrigins() + } + changed + } + } + } + // One commit per batch, so Room notifies paged observers once per batch rather than per row. private suspend fun write( table: String, @@ -284,14 +561,18 @@ class SyncService( rows: List, preserved: Set = emptySet(), markStale: Boolean = false, + onChanged: (T) -> Unit = {}, columns: (T) -> List>, ): Boolean = - db.useWriterConnection { connection -> - connection.immediateTransaction { - val before = connection.totalChanges() - connection.upsertChanged(table, key, rows, preserved, columns) - (connection.totalChanges() != before).also { changed -> + commitOrigins { + db.useWriterConnection { connection -> + connection.immediateTransaction { + val before = connection.totalChanges() + connection.upsertChanged(table, key, rows, preserved, onChanged, columns) + val changed = connection.totalChanges() != before if (changed && markStale) connection.markCreditsStale(source.id) + connection.persistPendingOrigins() + changed } } } @@ -304,25 +585,54 @@ class SyncService( albums.flatMap { album -> album.artistPositions().map { (position, ref) -> AlbumArtistRef(album.sourceId, album.id, position, ref.id, ref.name) } } - return db.useWriterConnection { connection -> - connection.immediateTransaction { - val before = connection.totalChanges() - val resolved = connection.resolveAlbumSortNames(albums) - connection.upsertChanged("albums", primaryKey, resolved) { it.columns() } - connection.upsertChanged("discs", discKey, discs) { it.columns() } - connection.usePrepared("DELETE FROM album_artists WHERE sourceId = ? AND albumId = ? AND position >= ?") { statement -> - albums.forEach { album -> - statement.bind(1, album.sourceId) - statement.bind(2, album.id) - statement.bind(3, album.artistPositions().size.toLong()) - statement.step() - statement.reset() - statement.clearBindings() + return commitOrigins { + db.useWriterConnection { connection -> + connection.immediateTransaction { + val before = connection.totalChanges() + val previousArtists = connection.albumArtistIds(albums.map { it.id }.toSet()) + val previousTrackArtists = connection.songArtistIds(albums.map { it.id }.toSet()) + val changedAlbums = mutableSetOf() + val resolved = connection.resolveAlbumSortNames(albums) + connection.upsertChanged( + "albums", + primaryKey, + resolved, + onChanged = { + changedAlbums += it.id + changedOrigins += origin(QueueKind.Album, it.id) + }, + ) { it.columns() } + connection.upsertChanged( + "discs", + discKey, + discs, + onChanged = { changedOrigins += origin(QueueKind.Album, it.albumId) }, + ) { it.columns() } + connection.usePrepared("DELETE FROM album_artists WHERE sourceId = ? AND albumId = ? AND position >= ?") { statement -> + albums.forEach { album -> + statement.bind(1, album.sourceId) + statement.bind(2, album.id) + statement.bind(3, album.artistPositions().size.toLong()) + statement.step() + statement.reset() + statement.clearBindings() + } } - } - connection.upsertChanged("album_artists", albumArtistKey, artists) { it.columns() } - (connection.totalChanges() != before).also { changed -> + connection.upsertChanged("album_artists", albumArtistKey, artists) { it.columns() } + albums.forEach { album -> + val previous = previousArtists[album.id].orEmpty() + val current = (album.artistPositions().map { it.second.id } + album.artistId).filterNotNull().toSet() + if (album.id in changedAlbums || previous != current) { + changedOrigins += (previous + current).map { origin(QueueKind.Artist, it) } + } + } + changedAlbums.forEach { albumId -> + changedOrigins += previousTrackArtists[albumId].orEmpty().map { origin(QueueKind.Artist, it) } + } + val changed = connection.totalChanges() != before if (changed) connection.markCreditsStale(albums.first().sourceId) + connection.persistPendingOrigins() + changed } } } @@ -330,17 +640,21 @@ class SyncService( private suspend fun deleteStaleDiscs(stale: Collection) { 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() + commitOrigins { + chunk.forEach { changedOrigins += origin(QueueKind.Album, it.albumId) } + 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() + } } + connection.persistPendingOrigins() } } } @@ -361,7 +675,7 @@ class SyncService( if (ids.isEmpty()) return pruned val stale = ids.filterNot { seen.contains(idHash(it)) } if (stale.isNotEmpty()) markStale?.invoke() - stale.chunked(DELETE_CHUNK).forEach { delete(it) } + stale.chunked(DELETE_CHUNK).forEach { commitOrigins { delete(it) } } pruned += stale.size if (ids.size < PRUNE_PAGE) return pruned afterId = ids.last() @@ -397,8 +711,26 @@ data class SyncSummary( val playlists: Int, val playlistSongs: Int, val pruned: Int, + val changedOrigins: Set = emptySet(), ) +private data class SongSyncState( + val id: String, + val albumId: String?, + val artistIds: Set, + val genres: Set, + val starred: Boolean, + val sortName: String, +) + +private class MutableSongSyncState { + var albumId: String? = null + var sortName: String = "" + val artistIds = mutableSetOf() + val genres = mutableSetOf() + var starred = false +} + private const val DELETE_CHUNK = 500 private const val SQL_VARIABLES = 500 private const val PRUNE_PAGE = 500 @@ -446,6 +778,20 @@ private fun Song.genreNames(): List = .map { part -> id3v1Genre(part) ?: part } .distinct() +private fun Song.syncState(albumArtists: Map>): SongSyncState { + val artistIds = artistPositions().mapTo(mutableSetOf()) { it.second.id } + this.artistId?.let(artistIds::add) + albumId?.let { artistIds += albumArtists[it].orEmpty() } + return SongSyncState( + id = id, + albumId = albumId, + artistIds = artistIds, + genres = genreNames().toSet(), + starred = starred != null, + sortName = sortName, + ) +} + // Navidrome's default genre tag split (resources/mappings.yaml): semicolon, slash, comma. private fun String.genreParts(): List = split(';', '/', ',').map { it.trim() }.filter { it.isNotEmpty() } @@ -583,6 +929,7 @@ private suspend fun PooledConnection.upsertChanged( key: List, rows: List, preserved: Set = emptySet(), + onChanged: (T) -> Unit = {}, columns: (T) -> List>, ) { if (rows.isEmpty()) return @@ -593,11 +940,12 @@ private suspend fun PooledConnection.upsertChanged( "ON CONFLICT(${key.joinToString()}) DO UPDATE SET " + updates.joinToString { "$it = excluded.$it" } + " WHERE " + - updates.joinToString(" OR ") { "$table.$it COLLATE BINARY IS NOT excluded.$it" } + updates.joinToString(" OR ") { "$table.$it COLLATE BINARY IS NOT excluded.$it" } + + " RETURNING 1" usePrepared(sql) { statement -> rows.forEach { row -> columns(row).forEachIndexed { index, (_, value) -> statement.bind(index + 1, value) } - statement.step() + if (statement.step()) onChanged(row) statement.reset() statement.clearBindings() } 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 1b28c057..077a8dfb 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt @@ -5,18 +5,27 @@ import android.content.res.Resources import androidx.test.core.app.ApplicationProvider import androidx.test.ext.junit.runners.AndroidJUnit4 import com.subtracks.TEST_TIMEOUT_MS +import com.subtracks.awaitUntil import com.subtracks.cancelAndJoinBlocking import com.subtracks.data.db.SubtracksDatabase import com.subtracks.data.db.createTestDatabase import com.subtracks.data.download.ArtworkStore +import com.subtracks.data.download.FakeDownloadEngine +import com.subtracks.data.model.DownloadStatus +import com.subtracks.data.model.SongDownload +import com.subtracks.data.model.Source import com.subtracks.data.prefs.UserPreferences import com.subtracks.data.prefs.fakeUserPreferences +import com.subtracks.data.repo.DownloadRepository import com.subtracks.data.repo.NetworkServerActionSink import com.subtracks.data.repo.QueueRepository import com.subtracks.data.repo.SourceRepository +import com.subtracks.data.repo.replace +import com.subtracks.data.repo.setShuffle import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.flow.drop import kotlinx.coroutines.flow.first import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout @@ -28,6 +37,7 @@ import okhttp3.mockwebserver.RecordedRequest import okhttp3.mockwebserver.SocketPolicy import org.junit.After import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test @@ -100,6 +110,39 @@ class SyncManagerTest { assertTrue(messages.isEmpty()) } + @Test + fun aNoOpSyncPreservesTheQueueResolutionCache() = + runBlocking { + val healthy = healthyDispatcher() + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse = + when { + request.path?.startsWith("/rest/search3.view") == true -> MockResponse().setBody(TWO_SONGS) + request.path?.startsWith("/rest/getPlaylists.view") == true -> MockResponse().setBody(PLAYLISTS) + request.path?.startsWith("/rest/getPlaylist.view") == true -> MockResponse().setBody(PLAYLIST) + else -> healthy.dispatch(request) + } + } + sourceRepository.addSource("Local", server.url("/").toString(), "u", "p", useTokenAuth = false) + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Success } } + queueRepository.replace(listOf(queueRepository.playlistEntry(1, "p1"))) + val snapshot = queueRepository.snapshot() + assertEquals(2L, snapshot.size) + assertEquals(listOf("s1"), queueRepository.range(snapshot, 0, 0).map { it.item.song.id }) + assertEquals(listOf("s2"), queueRepository.range(snapshot, 1, 1).map { it.item.song.id }) + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.drop(1).first { it is SyncStatus.Success } } + val offsetSeeks = queueRepository.offsetSeeks + val keysetSeeks = queueRepository.keysetSeeks + assertEquals(listOf("s2"), queueRepository.range(snapshot, 1, 1).map { it.item.song.id }) + + assertEquals(offsetSeeks, queueRepository.offsetSeeks) + assertEquals(keysetSeeks + 1, queueRepository.keysetSeeks) + } + @Test fun aFailedSyncDoesNotAdvanceTheQueueContentVersion() = runBlocking { @@ -112,6 +155,19 @@ class SyncManagerTest { assertEquals(before, queueRepository.snapshot().version) } + @Test + fun aFailedSyncDoesNotResetTheQueue() = + runBlocking { + db.sourcesDao().upsertSource(Source(1, "Local", "http://localhost", true, 0)) + queueRepository.replace(listOf(queueRepository.albumEntry(1, "al1").copy(rangeStart = 0, rangeEnd = 0))) + val before = queueRepository.snapshot().entries.map { it.entry } + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Failed } } + + assertEquals(before, queueRepository.snapshot().entries.map { it.entry }) + } + @Test fun aSuccessfulSyncMirrorsTheLibraryIntoRoom() = runBlocking { @@ -169,6 +225,162 @@ class SyncManagerTest { assertEquals(listOf("al1"), db.libraryDao().albumIds(1)) } + @Test + fun aFailedSyncCarriesCommittedOriginsIntoTheNextSuccessfulReset() = + runBlocking { + val healthy = healthyDispatcher() + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse = + if (request.path?.startsWith("/rest/getPlaylists.view") == true) { + MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AT_START) + } else { + healthy.dispatch(request) + } + } + sourceRepository.addSource("Local", server.url("/").toString(), "u", "p", useTokenAuth = false) + val edited = queueRepository.albumEntry(1, "al1").copy(rangeStart = 0, rangeEnd = 0) + queueRepository.replace(listOf(edited)) + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Failed } } + assertEquals( + 0L, + queueRepository + .snapshot() + .entries + .single() + .entry.rangeStart, + ) + + assertTrue(db.queueDao().pendingSyncOrigins(1).isNotEmpty()) + manager = + SyncManager( + db, + sourceRepository, + queueRepository, + NetworkServerActionSink(sourceRepository), + preferences = preferences, + showMessage = { messages += it.resolve(resources) }, + scope = scope, + ) + server.dispatcher = healthy + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Success } } + + assertNull( + queueRepository + .snapshot() + .entries + .single() + .entry.rangeStart, + ) + assertTrue(db.queueDao().pendingSyncOrigins(1).isEmpty()) + } + + @Test + fun aFailedSyncPreservesResolvedQueueShuffle() = + runBlocking { + server.dispatcher = healthyDispatcher() + sourceRepository.addSource("Local", server.url("/").toString(), "u", "p", useTokenAuth = false) + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Success } } + queueRepository.replace(listOf(queueRepository.albumEntry(1, "al1"))) + queueRepository.setShuffle(42L) + assertTrue(queueRepository.snapshot().shuffled) + val healthy = healthyDispatcher() + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse = + when { + request.path?.startsWith("/rest/search3.view") == true -> { + MockResponse().setBody(TWO_SONGS) + } + + request.path?.startsWith("/rest/getPlaylists.view") == true -> { + MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AT_START) + } + + else -> { + healthy.dispatch(request) + } + } + } + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Failed } } + + val snapshot = queueRepository.snapshot() + assertTrue(snapshot.shuffled) + assertEquals(1L, snapshot.contextSize) + } + + @Test + fun aFailedSongPruneDoesNotResetAnEditedDownloadedQueue() = + runBlocking { + val healthy = healthyDispatcher() + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse = + if (request.path?.startsWith("/rest/search3.view") == true) { + MockResponse().setBody(TWO_SONGS) + } else { + healthy.dispatch(request) + } + } + sourceRepository.addSource("Local", server.url("/").toString(), "u", "p", useTokenAuth = false) + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Success } } + val context = ApplicationProvider.getApplicationContext() + val downloadsDir = File(context.cacheDir, "sync-downloads-${System.nanoTime()}") + val downloads = + DownloadRepository( + db, + sourceRepository, + FakeDownloadEngine(), + downloadsDir, + artworkStore = ArtworkStore(File(context.cacheDir, "art")), + artworkFetcher = { ByteArray(0) }, + scope = scope, + ) + File(downloadsDir, "1").mkdirs() + File(downloadsDir, "1/s2").writeText("audio") + db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 1)) + downloads.start() + downloads.awaitLoaded() + awaitUntil("waiting for download baseline") { downloads.localUri("s2") != null } + queueRepository.replace(listOf(queueRepository.downloadedEntry(1).copy(rangeStart = 1))) + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse = + when { + request.path?.startsWith("/rest/search3.view") == true -> { + MockResponse().setBody(SONG) + } + + request.path?.startsWith("/rest/getPlaylists.view") == true -> { + MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AT_START) + } + + else -> { + healthy.dispatch(request) + } + } + } + + manager.requestSync() + withTimeout(TEST_TIMEOUT_MS) { manager.status.first { it is SyncStatus.Failed } } + + assertEquals( + 1L, + queueRepository + .snapshot() + .entries + .single() + .entry.rangeStart, + ) + } + @Test fun requestsWhileRunningCoalesceIntoOneSync() = runBlocking { @@ -253,6 +465,24 @@ class SyncManagerTest { "" + "" + "" + + const val TWO_SONGS = + "" + + "" + + "" + + "" + + const val PLAYLISTS = + "" + + "" + + "" + + const val PLAYLIST = + "" + + "" + + "" + + "" + + "" } @Test diff --git a/app/src/test/kotlin/com/subtracks/data/sync/SyncServiceTest.kt b/app/src/test/kotlin/com/subtracks/data/sync/SyncServiceTest.kt index 54fa5b22..efad17f1 100644 --- a/app/src/test/kotlin/com/subtracks/data/sync/SyncServiceTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/sync/SyncServiceTest.kt @@ -18,6 +18,8 @@ import com.subtracks.data.model.DownloadStatus import com.subtracks.data.model.ListDownloadStatus import com.subtracks.data.model.Playlist import com.subtracks.data.model.PlaylistSong +import com.subtracks.data.model.QueueKind +import com.subtracks.data.model.QueueOrigin import com.subtracks.data.model.Song import com.subtracks.data.model.SongDownload import com.subtracks.data.model.Source @@ -1868,6 +1870,278 @@ class SyncServiceTest { } } + @Test + fun playlistMembershipChangesReportOnlyTheirOwnOrigins() = + runTest { + insertSource() + val source = + FakeMusicSource( + songs = listOf(song("s1"), song("s2")), + playlists = listOf(playlist("p1"), playlist("p2")), + playlistSongs = + listOf( + PlaylistSong(1, "p1", "s1", 0), + PlaylistSong(1, "p2", "s2", 0), + ), + ) + SyncService(db, source).sync() + + source.playlistSongs = + listOf( + PlaylistSong(1, "p1", "s2", 0), + PlaylistSong(1, "p2", "s2", 0), + ) + + val summary = SyncService(db, source).sync() + + assertEquals(setOf(QueueOrigin(QueueKind.Playlist, 1, "p1")), summary.changedOrigins) + + source.playlistSongs = listOf(PlaylistSong(1, "p2", "s2", 0)) + + val deletionSummary = SyncService(db, source).sync() + + assertEquals(setOf(QueueOrigin(QueueKind.Playlist, 1, "p1")), deletionSummary.changedOrigins) + } + + @Test + fun albumAndSongChangesReportOnlyAffectedOrigins() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1"), artist("a2"), artist("a3")), + albums = + listOf( + album("al1"), + album("al2").copy(artistId = "a2"), + album("al3").copy(artistId = "a3"), + ), + songs = + listOf( + song("s1").copy(genres = listOf("Rock")), + song("s2").copy(albumId = "al2", artistId = "a2", genres = listOf("Jazz")), + ), + ) + SyncService(db, source).sync() + + source.albums = + listOf( + album("al1").copy(name = "Renamed"), + album("al2").copy(artistId = "a2"), + album("al3").copy(artistId = "a3"), + ) + source.songs = + listOf( + song("s1").copy( + albumId = "al3", + artistId = "a3", + genres = listOf("Electronic"), + starred = 1234, + ), + song("s2").copy(albumId = "al2", artistId = "a2", genres = listOf("Jazz")), + ) + + val summary = SyncService(db, source).sync() + + assertEquals( + setOf( + QueueOrigin(QueueKind.Album, 1, "al1"), + QueueOrigin(QueueKind.Album, 1, "al3"), + QueueOrigin(QueueKind.Artist, 1, "a1"), + QueueOrigin(QueueKind.Artist, 1, "a3"), + QueueOrigin(QueueKind.Genre, 1, "Rock"), + QueueOrigin(QueueKind.Genre, 1, "Electronic"), + QueueOrigin(QueueKind.Starred, 1, ""), + QueueOrigin(QueueKind.Song, 1, "s1"), + ), + summary.changedOrigins, + ) + } + + @Test + fun deletedSongsReportOnlyTheirPreviousOrigins() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1"), artist("a2")), + albums = listOf(album("al1"), album("al2").copy(artistId = "a2")), + songs = + listOf( + song("s1").copy(genres = listOf("Rock"), starred = 1234), + song("s2").copy(albumId = "al2", artistId = "a2", genres = listOf("Jazz")), + ), + ) + SyncService(db, source).sync() + + source.songs = listOf(song("s2").copy(albumId = "al2", artistId = "a2", genres = listOf("Jazz"))) + + val summary = SyncService(db, source).sync() + + assertEquals( + setOf( + QueueOrigin(QueueKind.Album, 1, "al1"), + QueueOrigin(QueueKind.Artist, 1, "a1"), + QueueOrigin(QueueKind.Genre, 1, "Rock"), + QueueOrigin(QueueKind.Starred, 1, ""), + QueueOrigin(QueueKind.Song, 1, "s1"), + ), + summary.changedOrigins, + ) + } + + @Test + fun albumCreditChangesReportBothArtistOrigins() = + runTest { + insertSource() + val plural = listOf(ArtistRef("a2", "Artist a2")) + val source = + FakeMusicSource( + artists = listOf(artist("a1"), artist("a2"), artist("a3")), + albums = listOf(album("al1").copy(artists = plural)), + songs = listOf(song("s1")), + ) + SyncService(db, source).sync() + + source.albums = listOf(album("al1").copy(artistId = "a3", artists = plural, year = 2024)) + + val summary = SyncService(db, source).sync() + + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Artist, 1, "a1"))) + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Artist, 1, "a3"))) + } + + @Test + fun songPresenceChangesReportReferencingPlaylistOrigins() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1")), + albums = listOf(album("al1")), + songs = listOf(song("s1"), song("s2")), + playlists = listOf(playlist("p1")), + playlistSongs = listOf(PlaylistSong(1, "p1", "s1", 0), PlaylistSong(1, "p1", "s2", 1)), + ) + SyncService(db, source).sync() + + source.songs = listOf(song("s2")) + val deleted = SyncService(db, source).sync() + assertTrue(deleted.changedOrigins.contains(QueueOrigin(QueueKind.Playlist, 1, "p1"))) + + source.songs = listOf(song("s1"), song("s2")) + val restored = SyncService(db, source).sync() + assertTrue(restored.changedOrigins.contains(QueueOrigin(QueueKind.Playlist, 1, "p1"))) + } + + @Test + fun downloadedSongSortChangesReportDownloadedOrigins() = + runTest { + insertSource() + val source = + FakeMusicSource(artists = listOf(artist("a1")), albums = listOf(album("al1")), songs = listOf(song("s1"), song("s2"))) + SyncService(db, source).sync() + db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed)) + db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed)) + + source.songs = listOf(song("s1").copy(title = "Zzz", sortName = "Zzz"), song("s2").copy(title = "Zulu")) + val summary = SyncService(db, source).sync() + + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Downloaded, 1, ""))) + } + + @Test + fun albumChangesReportArtistsCreditedOnlyOnTracks() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1"), artist("a2")), + albums = listOf(album("al1")), + songs = listOf(song("s1").copy(artistId = "a2")), + ) + SyncService(db, source).sync() + + source.albums = listOf(album("al1").copy(year = 2024)) + val summary = SyncService(db, source).sync() + + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Artist, 1, "a2"))) + } + + @Test + fun albumPruningReportsArtistsCreditedOnlyOnRemainingTracks() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1"), artist("a2")), + albums = listOf(album("al1")), + songs = listOf(song("s1").copy(artistId = "a2")), + ) + SyncService(db, source).sync() + + source.albums = emptyList() + val summary = SyncService(db, source).sync() + + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Artist, 1, "a2"))) + } + + @Test + fun albumChangesIgnoreSongsWithoutTrackArtists() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1")), + albums = listOf(album("al1")), + songs = listOf(song("s1").copy(artistId = null)), + ) + SyncService(db, source).sync() + + source.albums = listOf(album("al1").copy(year = 2024)) + val summary = SyncService(db, source).sync() + + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Album, 1, "al1"))) + assertTrue(summary.changedOrigins.none { it.kind == QueueKind.Artist && it.refId.isBlank() }) + } + + @Test + fun committedOriginsArePersistedWithoutManagerRetention() = + runTest { + insertSource() + val source = FakeMusicSource(artists = listOf(artist("a1")), albums = listOf(album("al1")), songs = listOf(song("s1"))) + SyncService(db, source).sync() + + source.albums = listOf(album("al1").copy(name = "Changed")) + SyncService(db, source).sync() + + val pending = db.queueDao().pendingSyncOrigins(1) + assertTrue(pending.any { it.kind == QueueKind.Album && it.refId == "al1" }) + } + + @Test + fun albumPruningReportsArtistsCreditedOnlyByTheDeletedAlbum() = + runTest { + insertSource() + val source = + FakeMusicSource( + artists = listOf(artist("a1"), artist("a2")), + albums = listOf(album("al1"), album("al2")), + songs = + listOf( + song("s1").copy(artistId = "a2"), + song("s2").copy(albumId = "al2", artistId = "a2"), + ), + ) + SyncService(db, source).sync() + + source.albums = listOf(album("al2")) + source.songs = listOf(song("s2").copy(albumId = "al2", artistId = "a2")) + val summary = SyncService(db, source).sync() + + assertTrue(summary.changedOrigins.contains(QueueOrigin(QueueKind.Artist, 1, "a1"))) + } + private suspend fun totalChanges(): Long = db.useWriterConnection { connection -> connection.usePrepared("SELECT total_changes()") { statement ->