sync: reset queues only for successfully changed origins

This commit is contained in:
austinried
2026-11-14 12:05:28 +00:00
parent acc820e263
commit 548bf39136
4 changed files with 942 additions and 73 deletions
@@ -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
}
@@ -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<QueueOrigin>()
private var transactionOrigins: MutableSet<QueueOrigin>? = null
private val changedOrigins =
object : AbstractMutableSet<QueueOrigin>() {
override val size: Int get() = committedOrigins.size
override fun iterator(): MutableIterator<QueueOrigin> = committedOrigins.iterator()
override fun add(element: QueueOrigin): Boolean = (transactionOrigins ?: committedOrigins).add(element)
}
private suspend fun <T> commitOrigins(block: suspend () -> T): T {
check(transactionOrigins == null)
val pending = mutableSetOf<QueueOrigin>()
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,11 +181,17 @@ class SyncService(
batch.flatMap { song ->
song.artistPositions().map { (position, ref) -> SongArtistRef(song.sourceId, song.id, position, ref.id, ref.name) }
}
return db.useWriterConnection { connection ->
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)
connection.upsertChanged("songs", primaryKey, resolved) { it.columns() }
val nextStates = resolved.associate { it.id to it.syncState(albumArtists) }
val downloadedIds = connection.completedDownloadIds(batch.map { it.id })
val changedSongIds = mutableSetOf<String>()
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)
@@ -160,9 +214,190 @@ class SyncService(
}
}
connection.upsertChanged("song_artists", songArtistKey, artists) { it.columns() }
(connection.totalChanges() != before).also { changed ->
if (changed) connection.markCreditsStale(batch.first().sourceId)
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<String>) {
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<String>): Map<String, SongSyncState> {
val states = mutableMapOf<String, MutableSongSyncState>()
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<String>): Set<String> {
val result = mutableSetOf<String>()
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<String>): Map<String, Set<String>> {
val artists = mutableMapOf<String, MutableSet<String>>()
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<String>): Map<String, Set<String>> {
val result = mutableMapOf<String, MutableSet<String>>()
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<String>,
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<String>,
) {
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<String>,
) {
db.useWriterConnection { connection ->
connection.immediateTransaction {
connection.deleteIds(table, ids)
connection.persistPendingOrigins()
}
}
}
@@ -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<String, Long>()
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 <T> write(
table: String,
@@ -284,14 +561,18 @@ class SyncService(
rows: List<T>,
preserved: Set<String> = emptySet(),
markStale: Boolean = false,
onChanged: (T) -> Unit = {},
columns: (T) -> List<Pair<String, Any?>>,
): Boolean =
commitOrigins {
db.useWriterConnection { connection ->
connection.immediateTransaction {
val before = connection.totalChanges()
connection.upsertChanged(table, key, rows, preserved, columns)
(connection.totalChanges() != before).also { changed ->
connection.upsertChanged(table, key, rows, preserved, onChanged, columns)
val changed = connection.totalChanges() != before
if (changed && markStale) connection.markCreditsStale(source.id)
connection.persistPendingOrigins()
changed
}
}
}
@@ -304,12 +585,29 @@ class SyncService(
albums.flatMap { album ->
album.artistPositions().map { (position, ref) -> AlbumArtistRef(album.sourceId, album.id, position, ref.id, ref.name) }
}
return db.useWriterConnection { connection ->
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<String>()
val resolved = connection.resolveAlbumSortNames(albums)
connection.upsertChanged("albums", primaryKey, resolved) { it.columns() }
connection.upsertChanged("discs", discKey, discs) { it.columns() }
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)
@@ -321,8 +619,20 @@ class SyncService(
}
}
connection.upsertChanged("album_artists", albumArtistKey, artists) { it.columns() }
(connection.totalChanges() != before).also { changed ->
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,6 +640,8 @@ class SyncService(
private suspend fun deleteStaleDiscs(stale: Collection<DiscKey>) {
stale.chunked(DELETE_CHUNK).forEach { chunk ->
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 ->
@@ -342,6 +654,8 @@ class SyncService(
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<QueueOrigin> = emptySet(),
)
private data class SongSyncState(
val id: String,
val albumId: String?,
val artistIds: Set<String>,
val genres: Set<String>,
val starred: Boolean,
val sortName: String,
)
private class MutableSongSyncState {
var albumId: String? = null
var sortName: String = ""
val artistIds = mutableSetOf<String>()
val genres = mutableSetOf<String>()
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<String> =
.map { part -> id3v1Genre(part) ?: part }
.distinct()
private fun Song.syncState(albumArtists: Map<String, Set<String>>): 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<String> = split(';', '/', ',').map { it.trim() }.filter { it.isNotEmpty() }
@@ -583,6 +929,7 @@ private suspend fun <T> PooledConnection.upsertChanged(
key: List<String>,
rows: List<T>,
preserved: Set<String> = emptySet(),
onChanged: (T) -> Unit = {},
columns: (T) -> List<Pair<String, Any?>>,
) {
if (rows.isEmpty()) return
@@ -593,11 +940,12 @@ private suspend fun <T> 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()
}
@@ -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<Context>()
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 {
"<subsonic-response status=\"ok\" version=\"1.16.1\"><searchResult3>" +
"<song id=\"s1\" title=\"Song One\" albumId=\"al1\" artistId=\"ar1\" track=\"1\"/>" +
"</searchResult3></subsonic-response>"
const val TWO_SONGS =
"<subsonic-response status=\"ok\" version=\"1.16.1\"><searchResult3>" +
"<song id=\"s1\" title=\"Song One\" albumId=\"al1\" artistId=\"ar1\" track=\"1\"/>" +
"<song id=\"s2\" title=\"Song Two\" albumId=\"al1\" artistId=\"ar1\" track=\"2\"/>" +
"</searchResult3></subsonic-response>"
const val PLAYLISTS =
"<subsonic-response status=\"ok\" version=\"1.16.1\"><playlists>" +
"<playlist id=\"p1\" name=\"Playlist\" songCount=\"2\"/>" +
"</playlists></subsonic-response>"
const val PLAYLIST =
"<subsonic-response status=\"ok\" version=\"1.16.1\">" +
"<playlist id=\"p1\" name=\"Playlist\" songCount=\"2\">" +
"<entry id=\"s1\" title=\"Song One\" albumId=\"al1\" artistId=\"ar1\" track=\"1\"/>" +
"<entry id=\"s2\" title=\"Song Two\" albumId=\"al1\" artistId=\"ar1\" track=\"2\"/>" +
"</playlist></subsonic-response>"
}
@Test
@@ -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 ->