downloads: reconcile visible queue reset baselines

Consume captured reset points on the first collector emission and leave pending sources uninitialized during reconciliation.

Build ordered baselines from queue-visible completed downloads and ignore pruned-row transitions while retaining all-row state and recovery.

app/src/main/kotlin/com/subtracks/data/db/DownloadDao.kt

@@ -56,6 +56,24 @@ interface DownloadDao {
     @Query("SELECT * FROM song_downloads WHERE sourceId = :sourceId")
     fun downloads(sourceId: Long): Flow<List<SongDownload>>
 
+    @Query(
+        "SELECT d.* FROM song_downloads d " +
+            "JOIN songs ON songs.sourceId = d.sourceId AND songs.id = d.songId " +
+            "WHERE d.sourceId = :sourceId AND d.status = 'Completed' " +
+            "ORDER BY d.downloadedAt DESC, songs.sortName COLLATE NOCASE, songs.id",
+    )
+    suspend fun completedDownloads(sourceId: Long): List<SongDownload>
+
+    @Query(
+        "SELECT d.songId FROM song_downloads d " +
+            "JOIN songs ON songs.sourceId = d.sourceId AND songs.id = d.songId " +
+            "WHERE d.sourceId = :sourceId AND d.songId IN (:songIds)",
+    )
+    suspend fun queueVisibleSongIds(
+        sourceId: Long,
+        songIds: Collection<String>,
+    ): List<String>
+
     @Query("SELECT * FROM song_downloads WHERE status IN ('Queued', 'Running') AND (engineId IS NOT NULL OR sourceId = :sourceId)")
     fun activeDownloads(sourceId: Long): Flow<List<SongDownload>>
 

app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt

@@ -71,7 +71,8 @@ class DownloadRepository(
     private var started = false
 
     private val statesFlow = MutableStateFlow<Map<String, SongDownload>>(emptyMap())
-    private val completedBySource = mutableMapOf<Long, Map<String, Long>>()
+    private val completedBySource = mutableMapOf<Long, List<Pair<String, Long>>>()
+    private val completedRowsBySource = mutableMapOf<Long, Map<String, Long>>()
     private val completedBaselineMutex = Mutex()
     private val pendingResetPoints = mutableMapOf<Long, QueueResetPoint>()
     private val attemptedBios = ConcurrentHashMap.newKeySet<Pair<Long, String>>()
@@ -87,18 +88,34 @@ class DownloadRepository(
             downloadStates().collect { (sourceId, rows) ->
                 statesFlow.value = rows
                 if (sourceId != null) {
-                    val completed = completedDownloads(rows)
                     mutex.withLock {
+                        val completed = completedDownloads(sourceId)
+                        val completedRows = completedDownloadKeys(rows)
                         completedBaselineMutex.withLock {
                             val previous = completedBySource[sourceId]
-                            if (previous == null) {
+                            val previousRows = completedRowsBySource[sourceId]
-                                completedBySource[sourceId] = completed
+                            val pending = pendingResetPoints[sourceId]
-                            } else if (completed != previous) {
+                            when {
-                                val point = pendingResetPoints.remove(sourceId) ?: resetPoint
+                                pending != null -> {
-                                queueRepository.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), point)
+                                    pendingResetPoints.remove(sourceId)
-                                completedBySource[sourceId] = completed
+                                    if (previous == null || completed != previous) {
+                                        queueRepository.resetChangedOrigins(
+                                            setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")),
+                                            pending,
+                                        )
                                     }
                                 }
+
+                                previous != null && completed != previous && completedRows != previousRows -> {
+                                    queueRepository.resetChangedOrigins(
+                                        setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")),
+                                        resetPoint,
+                                    )
+                                }
+                            }
+                            completedBySource[sourceId] = completed
+                            completedRowsBySource[sourceId] = completedRows
+                        }
                     }
                 }
                 resetPoint = queueRepository.captureResetPoint()
@@ -125,16 +142,34 @@ class DownloadRepository(
             if (sourceId !in pendingResetPoints) pendingResetPoints[sourceId] = queueRepository.captureResetPoint()
         }
 
-    private suspend fun ensureCompletedBaselines(rows: List<SongDownload>) =
+    private suspend fun captureForReset(
+        sourceId: Long,
+        songIds: Collection<String>,
+    ) {
+        if (songIds.isEmpty()) return
+        if (db.downloadDao().queueVisibleSongIds(sourceId, songIds).isNotEmpty()) {
+            captureForReset(sourceId)
+        } else {
             completedBaselineMutex.withLock {
-            rows.groupBy { it.sourceId }.forEach { (sourceId, sourceRows) ->
+                if (sourceId !in pendingResetPoints) completedBySource[sourceId] = completedDownloads(sourceId)
-                if (sourceId !in completedBySource) {
-                    completedBySource[sourceId] = completedDownloads(sourceRows.associateBy { it.songId })
             }
         }
     }
 
-    private fun completedDownloads(rows: Map<String, SongDownload>): Map<String, Long> =
+    private suspend fun ensureCompletedBaselines(rows: List<SongDownload>) =
+        completedBaselineMutex.withLock {
+            rows.groupBy { it.sourceId }.forEach { (sourceId, sourceRows) ->
+                if (sourceId !in completedBySource && sourceId !in pendingResetPoints) {
+                    completedBySource[sourceId] = completedDownloads(sourceId)
+                    completedRowsBySource[sourceId] = completedDownloadKeys(sourceRows.associateBy { it.songId })
+                }
+            }
+        }
+
+    private suspend fun completedDownloads(sourceId: Long): List<Pair<String, Long>> =
+        db.downloadDao().completedDownloads(sourceId).map { it.songId to it.downloadedAt }
+
+    private fun completedDownloadKeys(rows: Map<String, SongDownload>): Map<String, Long> =
         rows.values
             .filter { it.status == DownloadStatus.Completed }
             .associate { it.songId to it.downloadedAt }
@@ -394,7 +429,7 @@ class DownloadRepository(
         val engineIds = rows.mapNotNull { it.engineId }
         if (engineIds.isNotEmpty()) engine.cancel(engineIds)
         rows.groupBy { it.sourceId }.forEach { (sourceId, group) ->
-            if (group.any { it.status == DownloadStatus.Completed }) captureForReset(sourceId)
+            captureForReset(sourceId, group.filter { it.status == DownloadStatus.Completed }.map { it.songId })
             group.forEach { file(sourceId, it.songId).delete() }
             group.map { it.songId }.chunked(DELETE_CHUNK).forEach { db.downloadDao().deleteSongs(sourceId, it) }
             if (group.any { it.status == DownloadStatus.Completed }) {
@@ -416,7 +451,7 @@ class DownloadRepository(
         withContext(dispatcher) {
             mutex.withLock {
                 val rows = db.downloadDao().all().filter { it.sourceId == sourceId }
-                if (rows.any { it.status == DownloadStatus.Completed }) captureForReset(sourceId)
+                captureForReset(sourceId, rows.filter { it.status == DownloadStatus.Completed }.map { it.songId })
                 rows.mapNotNull { it.engineId }.let(engine::cancel)
                 db.downloadDao().deleteSource(sourceId)
                 db.libraryDao().clearDownloadedArtistArt(sourceId)
@@ -632,7 +667,7 @@ class DownloadRepository(
         songId: String,
     ): Boolean {
         val row = db.downloadDao().find(sourceId, songId) ?: return false
-        if (row.status == DownloadStatus.Completed) captureForReset(sourceId)
+        if (row.status == DownloadStatus.Completed) captureForReset(sourceId, listOf(songId))
         row.engineId?.let { engine.cancel(listOf(it)) }
         file(sourceId, songId).delete()
         db.downloadDao().delete(sourceId, songId)
@@ -718,7 +753,7 @@ class DownloadRepository(
             )
         val completedBefore = row.takeIf { it.status == DownloadStatus.Completed }?.let { it.songId to it.downloadedAt }
         val completedAfter = updated.takeIf { it.status == DownloadStatus.Completed }?.let { it.songId to it.downloadedAt }
-        if (completedBefore != completedAfter) captureForReset(row.sourceId)
+        if (completedBefore != completedAfter) captureForReset(row.sourceId, listOf(row.songId))
         if (updated != row) db.downloadDao().upsert(updated)
         if (updated.status != row.status &&
             (updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed)
@@ -739,7 +774,7 @@ class DownloadRepository(
         row: SongDownload,
         error: DownloadError,
     ) {
-        if (row.status == DownloadStatus.Completed) captureForReset(row.sourceId)
+        if (row.status == DownloadStatus.Completed) captureForReset(row.sourceId, listOf(row.songId))
         row.engineId?.let { engine.cancel(listOf(it)) }
         file(row.sourceId, row.songId).delete()
         db.downloadDao().upsert(row.copy(status = DownloadStatus.Failed, engineId = null, error = error))

app/src/test/kotlin/com/subtracks/data/db/DownloadDaoTest.kt

@@ -0,0 +1,76 @@
+package com.subtracks.data.db
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.subtracks.data.model.DownloadStatus
+import com.subtracks.data.model.Song
+import com.subtracks.data.model.SongDownload
+import com.subtracks.data.model.Source
+import kotlinx.coroutines.test.runTest
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+
+@RunWith(AndroidJUnit4::class)
+class DownloadDaoTest {
+    private lateinit var db: SubtracksDatabase
+
+    @Before
+    fun setUp() {
+        db = createTestDatabase()
+    }
+
+    @After
+    fun tearDown() {
+        db.close()
+    }
+
+    @Test
+    fun completedDownloadsUseTheDownloadedQueueVisibilityAndOrder() =
+        runTest {
+            val sourceId =
+                db.sourcesDao().insertSource(
+                    Source(name = "source", address = "http://localhost", isActive = true, createdAt = 0),
+                )
+            db.libraryDao().upsertSongs(
+                listOf(
+                    song(sourceId, "s1", "Zulu"),
+                    song(sourceId, "s2", "Alpha"),
+                    song(sourceId, "s3", "Running"),
+                    song(sourceId, "s4", "Pruned"),
+                ),
+            )
+            db.downloadDao().upsert(SongDownload(sourceId, "s1", DownloadStatus.Completed, downloadedAt = 1))
+            db.downloadDao().upsert(SongDownload(sourceId, "s2", DownloadStatus.Completed, downloadedAt = 1))
+            db.downloadDao().upsert(SongDownload(sourceId, "s3", DownloadStatus.Running, downloadedAt = 2))
+            db.downloadDao().upsert(SongDownload(sourceId, "s4", DownloadStatus.Completed, downloadedAt = 3))
+            db.libraryDao().deleteSongs(sourceId, listOf("s4"))
+
+            assertEquals(listOf("s2", "s1"), db.downloadDao().completedDownloads(sourceId).map { it.songId })
+            assertEquals(
+                setOf("s1", "s2", "s3"),
+                db.downloadDao().queueVisibleSongIds(sourceId, listOf("s1", "s2", "s3", "s4")).toSet(),
+            )
+        }
+
+    private fun song(
+        sourceId: Long,
+        id: String,
+        sortName: String,
+    ) = Song(
+        sourceId = sourceId,
+        id = id,
+        albumId = "al1",
+        artistId = "ar1",
+        title = sortName,
+        sortName = sortName,
+        album = "Album",
+        artist = "Artist",
+        duration = 100,
+        track = 1,
+        disc = 1,
+        starred = null,
+        genre = null,
+    )
+}

app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt

@@ -29,9 +29,11 @@ import com.subtracks.data.model.SubsonicSource
 import com.subtracks.data.model.artistImageKey
 import com.subtracks.data.model.coverArtKey
 import com.subtracks.data.prefs.fakeUserPreferences
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
@@ -55,6 +57,7 @@ import java.io.File
 import java.io.IOException
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 
 @RunWith(AndroidJUnit4::class)
@@ -1375,6 +1378,46 @@ class DownloadRepositoryTest {
             )
         }
 
+    private fun useRepository(
+        scope: CoroutineScope,
+        dispatcher: CoroutineDispatcher,
+    ) {
+        repository =
+            DownloadRepository(
+                db,
+                sources,
+                engine,
+                dir,
+                artworkStore = artwork,
+                artworkFetcher = fetcher,
+                queueRepository = queues,
+                showMessage = { messages += it.resolve(resources) },
+                dispatcher = dispatcher,
+                scope = scope,
+            )
+    }
+
+    private fun withBlockedCollector(block: (CountDownLatch) -> Unit) {
+        val releaseCollector = CountDownLatch(1)
+        val executor = Executors.newSingleThreadExecutor()
+        val dispatcher = executor.asCoroutineDispatcher()
+        val collectorScope = CoroutineScope(SupervisorJob() + dispatcher)
+        val collectorBlocked = CountDownLatch(1)
+        executor.execute {
+            collectorBlocked.countDown()
+            releaseCollector.await()
+        }
+        assertTrue(collectorBlocked.await(5, TimeUnit.SECONDS))
+        useRepository(collectorScope, Dispatchers.Default)
+        try {
+            block(releaseCollector)
+        } finally {
+            releaseCollector.countDown()
+            cancelAndJoinBlocking(collectorScope)
+            dispatcher.close()
+        }
+    }
+
     private fun requestedSongIds(): List<String> = engine.requests.map { it.second.path.substringAfterLast('/') }
 
     private fun seedSongs(tracks: IntRange) {
@@ -1704,6 +1747,165 @@ class DownloadRepositoryTest {
             }
         }
 
+    @Test
+    fun aCapturedDeletionBeforeTheFirstEmissionResetsAnEditedDownloadedQueue() =
+        runTest {
+            seedLibrary()
+            runBlocking {
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
+                db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
+                queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 1)))
+            }
+            writeFile(1, "s1", "one")
+            writeFile(1, "s2", "two")
+            val beforeReset = runBlocking { queues.snapshot().resetVersion }
+            withBlockedCollector { releaseCollector ->
+                repository.start()
+                runBlocking {
+                    repository.remove(1, "s2")
+                    repository.reconcile()
+                }
+                releaseCollector.countDown()
+                runBlocking { repository.awaitLoaded() }
+
+                val snapshot = runBlocking { queues.snapshot() }
+                assertEquals(beforeReset + 1, snapshot.resetVersion)
+                assertNull(
+                    snapshot.entries
+                        .single()
+                        .entry.rangeStart,
+                )
+            }
+        }
+
+    @Test
+    fun aCapturedCompletionBeforeTheFirstEmissionResetsAnEditedDownloadedQueue() =
+        runTest {
+            seedLibrary()
+            runBlocking {
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
+                queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 0, rangeEnd = 0)))
+            }
+            writeFile(1, "s1", "one")
+            writeFile(1, "s2", "two")
+            val beforeReset = runBlocking { queues.snapshot().resetVersion }
+            withBlockedCollector { releaseCollector ->
+                repository.start()
+                runBlocking {
+                    repository.captureForReset(1)
+                    db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
+                    repository.reconcile()
+                }
+                releaseCollector.countDown()
+                runBlocking { repository.awaitLoaded() }
+
+                val snapshot = runBlocking { queues.snapshot() }
+                assertEquals(beforeReset + 1, snapshot.resetVersion)
+                assertNull(
+                    snapshot.entries
+                        .single()
+                        .entry.rangeStart,
+                )
+                assertEquals(2L, snapshot.size)
+            }
+        }
+
+    @Test
+    fun aPrunedDownloadCompletingDoesNotResetAnEditedDownloadedQueue() =
+        runTest {
+            val beforeReset = preparePrunedDownload(DownloadStatus.Running)
+            writeFile(1, "s3", "three")
+            engine.complete(42, total = 10)
+
+            runBlocking { repository.reconcile() }
+            await { repository.states().value["s3"]?.status == DownloadStatus.Completed }
+            Thread.sleep(200)
+
+            val snapshot = runBlocking { queues.snapshot() }
+            assertEquals(beforeReset, snapshot.resetVersion)
+            assertEquals(
+                1L,
+                snapshot.entries
+                    .single()
+                    .entry.rangeStart,
+            )
+        }
+
+    @Test
+    fun aPrunedCompletedDownloadFailingDoesNotResetAnEditedDownloadedQueue() =
+        runTest {
+            val beforeReset = preparePrunedDownload(DownloadStatus.Completed)
+            file(1, "s3").delete()
+
+            runBlocking { repository.reconcile() }
+            await { repository.states().value["s3"]?.status == DownloadStatus.Failed }
+            Thread.sleep(200)
+
+            val snapshot = runBlocking { queues.snapshot() }
+            assertEquals(beforeReset, snapshot.resetVersion)
+            assertEquals(
+                1L,
+                snapshot.entries
+                    .single()
+                    .entry.rangeStart,
+            )
+        }
+
+    @Test
+    fun removingAPrunedCompletedDownloadDoesNotResetAnEditedDownloadedQueue() =
+        runTest {
+            val beforeReset = preparePrunedDownload(DownloadStatus.Completed)
+
+            runBlocking { repository.remove(1, "s3") }
+            await { repository.states().value["s3"] == null }
+            Thread.sleep(200)
+
+            val snapshot = runBlocking { queues.snapshot() }
+            assertEquals(beforeReset, snapshot.resetVersion)
+            assertEquals(
+                1L,
+                snapshot.entries
+                    .single()
+                    .entry.rangeStart,
+            )
+        }
+
+    private fun preparePrunedDownload(status: DownloadStatus): Long {
+        seedLibrary()
+        seedSongs(3..3)
+        runBlocking {
+            db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
+            db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
+            db.downloadDao().upsert(
+                SongDownload(
+                    1,
+                    "s3",
+                    status,
+                    engineId = 42L.takeIf { status == DownloadStatus.Running },
+                    downloadedAt = if (status == DownloadStatus.Completed) 3L else 0L,
+                ),
+            )
+        }
+        writeFile(1, "s1", "one")
+        writeFile(1, "s2", "two")
+        writeFile(1, "s3", "three")
+        if (status == DownloadStatus.Running) engine.running(42, bytes = 1, total = 10)
+        repository.start()
+        await {
+            repository.localUri("s1") != null &&
+                repository.localUri("s2") != null &&
+                (status != DownloadStatus.Completed || repository.localUri("s3") != null)
+        }
+        return runBlocking {
+            repository.awaitLoaded()
+            repository.reconcile()
+            queues.replace(listOf(queues.downloadedEntry(1).copy(rangeStart = 1)))
+            db.libraryDao().deleteSongs(1, listOf("s3"))
+            queues.invalidateLibraryCache()
+            queues.snapshot().resetVersion
+        }
+    }
+
     private fun seedPlaylist(songIds: List<String>) {
         runBlocking {
             db.libraryDao().upsertPlaylists(listOf(Playlist(1, "pl1", "Playlist", null, null, songIds.size.toLong(), 0)))