library: reconcile queues after starred and download changes

This commit is contained in:
austinried
2026-11-14 12:05:28 +00:00
parent 481f12c892
commit ed19ca2ece
11 changed files with 371 additions and 17 deletions
@@ -19,9 +19,12 @@ import com.subtracks.data.model.DownloadList
import com.subtracks.data.model.DownloadStatus
import com.subtracks.data.model.DownloadedSong
import com.subtracks.data.model.ListDownloadStatus
import com.subtracks.data.model.QueueKind
import com.subtracks.data.model.QueueOrigin
import com.subtracks.data.model.SongDownload
import com.subtracks.data.model.artistImageKey
import com.subtracks.data.model.coverArtKey
import com.subtracks.data.repo.QueueResetPoint
import com.subtracks.log.Log
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
@@ -55,6 +58,7 @@ class DownloadRepository(
private val downloadsDir: File,
private val artworkStore: ArtworkStore,
private val artworkFetcher: ArtworkFetcher,
private val queueRepository: QueueRepository,
private val showMessage: (UiMessage) -> Unit = {},
private val dispatcher: CoroutineDispatcher = Dispatchers.IO,
private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + dispatcher),
@@ -67,6 +71,9 @@ class DownloadRepository(
private var started = false
private val statesFlow = MutableStateFlow<Map<String, SongDownload>>(emptyMap())
private val completedBySource = mutableMapOf<Long, Map<String, Long>>()
private val completedBaselineMutex = Mutex()
private val pendingResetPoints = mutableMapOf<Long, QueueResetPoint>()
private val attemptedBios = ConcurrentHashMap.newKeySet<Pair<Long, String>>()
private val failedArtistImages = ConcurrentHashMap.newKeySet<String>()
@@ -76,8 +83,21 @@ class DownloadRepository(
downloadsDir.mkdirs()
File(downloadsDir, NO_MEDIA).createNewFile()
scope.launch {
downloadStates().collect { rows ->
var resetPoint = queueRepository.captureResetPoint()
downloadStates().collect { (sourceId, rows) ->
statesFlow.value = rows
if (sourceId != null) {
val completed = completedDownloads(rows)
completedBaselineMutex.withLock {
val previous = completedBySource[sourceId]
if (previous != null && completed != previous) {
val point = pendingResetPoints.remove(sourceId) ?: resetPoint
queueRepository.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Downloaded, sourceId, "")), point)
completedBySource[sourceId] = completed
}
}
}
resetPoint = queueRepository.captureResetPoint()
loaded.complete(Unit)
}
}
@@ -96,6 +116,25 @@ class DownloadRepository(
fun states(): StateFlow<Map<String, SongDownload>> = statesFlow
internal suspend fun captureForReset(sourceId: Long) =
completedBaselineMutex.withLock {
if (sourceId !in pendingResetPoints) pendingResetPoints[sourceId] = queueRepository.captureResetPoint()
}
private suspend fun ensureCompletedBaselines(rows: List<SongDownload>) =
completedBaselineMutex.withLock {
rows.groupBy { it.sourceId }.forEach { (sourceId, sourceRows) ->
if (sourceId !in completedBySource) {
completedBySource[sourceId] = completedDownloads(sourceRows.associateBy { it.songId })
}
}
}
private fun completedDownloads(rows: Map<String, SongDownload>): Map<String, Long> =
rows.values
.filter { it.status == DownloadStatus.Completed }
.associate { it.songId to it.downloadedAt }
@OptIn(ExperimentalCoroutinesApi::class)
fun activeDownloads(): Flow<List<SongDownload>> =
sourceRepository.activeSourceId().flatMapLatest { sourceId ->
@@ -351,6 +390,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)
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 }) {
@@ -371,12 +411,9 @@ class DownloadRepository(
suspend fun removeSource(sourceId: Long) =
withContext(dispatcher) {
mutex.withLock {
db
.downloadDao()
.all()
.filter { it.sourceId == sourceId }
.mapNotNull { it.engineId }
.let(engine::cancel)
val rows = db.downloadDao().all().filter { it.sourceId == sourceId }
if (rows.any { it.status == DownloadStatus.Completed }) captureForReset(sourceId)
rows.mapNotNull { it.engineId }.let(engine::cancel)
db.downloadDao().deleteSource(sourceId)
db.libraryDao().clearDownloadedArtistArt(sourceId)
dir(sourceId).deleteRecursively()
@@ -388,6 +425,7 @@ class DownloadRepository(
val promoted =
mutex.withLock {
val rows = db.downloadDao().all()
ensureCompletedBaselines(rows)
reconcileRows(rows)
sweep(rows)
ensurePolling()
@@ -397,11 +435,16 @@ class DownloadRepository(
}
@OptIn(ExperimentalCoroutinesApi::class)
private fun downloadStates(): Flow<Map<String, SongDownload>> =
private fun downloadStates(): Flow<Pair<Long?, Map<String, SongDownload>>> =
sourceRepository
.activeSourceId()
.flatMapLatest { sourceId -> if (sourceId == null) flowOf(emptyList()) else db.downloadDao().downloads(sourceId) }
.map { rows -> rows.associateBy { it.songId } }
.flatMapLatest { sourceId ->
if (sourceId == null) {
flowOf(null to emptyMap())
} else {
db.downloadDao().downloads(sourceId).map { rows -> sourceId to rows.associateBy { it.songId } }
}
}
private suspend fun alreadyDownloadingOrDownloaded(
sourceId: Long,
@@ -585,6 +628,7 @@ class DownloadRepository(
songId: String,
): Boolean {
val row = db.downloadDao().find(sourceId, songId) ?: return false
if (row.status == DownloadStatus.Completed) captureForReset(sourceId)
row.engineId?.let { engine.cancel(listOf(it)) }
file(sourceId, songId).delete()
db.downloadDao().delete(sourceId, songId)
@@ -668,6 +712,14 @@ class DownloadRepository(
row.downloadedAt
},
)
if (updated != row &&
(
updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed ||
updated.downloadedAt != row.downloadedAt
)
) {
captureForReset(row.sourceId)
}
if (updated != row) db.downloadDao().upsert(updated)
if (updated.status != row.status &&
(updated.status == DownloadStatus.Completed || row.status == DownloadStatus.Completed)
@@ -688,6 +740,7 @@ class DownloadRepository(
row: SongDownload,
error: DownloadError,
) {
if (row.status == DownloadStatus.Completed) captureForReset(row.sourceId)
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))
@@ -19,6 +19,8 @@ import com.subtracks.data.model.ArtistBio
import com.subtracks.data.model.Disc
import com.subtracks.data.model.Playlist
import com.subtracks.data.model.PlaylistSongItem
import com.subtracks.data.model.QueueKind
import com.subtracks.data.model.QueueOrigin
import com.subtracks.data.model.Song
import com.subtracks.data.prefs.AlbumSort
import com.subtracks.data.prefs.ArtistSort
@@ -50,6 +52,7 @@ class LibraryRepository(
private val sourceRepository: SourceRepository,
private val serverActions: ServerActionSink,
private val userPreferences: UserPreferences,
private val queueRepository: QueueRepository,
private val showMessage: (UiMessage) -> Unit = {},
private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO),
) {
@@ -452,10 +455,18 @@ class LibraryRepository(
starred: Long?,
) {
val dao = db.libraryDao()
if (type == StarType.Song) {
val previous = song(sourceId, id).first()?.starred
if (previous == starred) return
val point = queueRepository.captureResetPoint()
dao.setSongStar(sourceId, id, starred)
queueRepository.resetChangedOrigins(setOf(QueueOrigin(QueueKind.Starred, sourceId, "")), point)
} else {
when (type) {
StarType.Song -> dao.setSongStar(sourceId, id, starred)
StarType.Album -> dao.setAlbumStar(sourceId, id, starred)
StarType.Artist -> dao.setArtistStar(sourceId, id, starred)
StarType.Song -> Unit
}
}
}
@@ -96,7 +96,7 @@ fun appModule(
scope = get(),
)
}
single { LibraryRepository(get(), get(), get(), get(), toast, scope = get(named("io"))) }
single { LibraryRepository(get(), get(), get(), get(), get(), toast, scope = get(named("io"))) }
single<ArtworkSeedStore> { ArtworkSeedRepository(get()) }
single { SyncManager(get(), get(), get(), get(), preferences = get(), showMessage = toast, scope = get()) }
single {
@@ -120,6 +120,7 @@ fun appModule(
downloadsDir = downloadsRoot(context),
artworkStore = get(),
artworkFetcher = get(),
queueRepository = get(),
showMessage = toast,
dispatcher = Dispatchers.IO,
scope = get(named("io")),
@@ -21,6 +21,7 @@ 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.Song
import com.subtracks.data.model.SongDownload
import com.subtracks.data.model.Source
@@ -61,6 +62,7 @@ class DownloadRepositoryTest {
private lateinit var sources: SourceRepository
private lateinit var engine: FakeDownloadEngine
private lateinit var repository: DownloadRepository
private lateinit var queues: QueueRepository
private lateinit var artwork: ArtworkStore
private lateinit var dir: File
private lateinit var scope: CoroutineScope
@@ -89,6 +91,7 @@ class DownloadRepositoryTest {
sources = SourceRepository(db, OkHttpClient(), prefs, artwork, scope = scope)
engine = FakeDownloadEngine()
dir = File(context.cacheDir, "downloads-${System.nanoTime()}")
queues = QueueRepository(db)
repository =
DownloadRepository(
db,
@@ -97,6 +100,7 @@ class DownloadRepositoryTest {
dir,
artworkStore = artwork,
artworkFetcher = fetcher,
queueRepository = queues,
showMessage = { messages += it.resolve(resources) },
scope = scope,
)
@@ -361,6 +365,34 @@ class DownloadRepositoryTest {
)
}
@Test
fun deleteAllCapturesTheCurrentQueuePositionBeforeRemovingDownloads() =
runTest {
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))
queues.replace(listOf(queues.downloadedEntry(1), queues.albumEntry(1, "al1")))
}
writeFile(1, "s1", "audio")
writeFile(1, "s2", "audio")
repository.start()
await { repository.localUri("s1") != null && repository.localUri("s2") != null }
runBlocking { repository.awaitLoaded() }
runBlocking {
queues.setCursor(4)
queues.setPosition(42_000L)
}
val beforeReset = runBlocking { queues.snapshot().resetVersion }
runBlocking { repository.removeSource(1) }
await { runBlocking { queues.snapshot().resetVersion > beforeReset } }
assertEquals(2L, runBlocking { queues.cursor() })
assertEquals(42_000L, runBlocking { queues.cursorPositionMs() })
}
@Test
fun reconcileDeletesFilesThatBelongToNoDownload() =
runTest {
@@ -613,6 +645,19 @@ class DownloadRepositoryTest {
assertNotNull(artwork.uri(1, coverArtKey(1, ALBUM_ART, false)))
}
@Test
fun artworkSurvivesTheDownloadedSongGoingMissing() =
runTest {
seedLibrary()
runBlocking { repository.download(1, "s1") }
assertNotNull(artwork.uri(1, coverArtKey(1, ALBUM_ART, false)))
runBlocking { db.libraryDao().deleteSongs(1, listOf("s1")) }
runBlocking { repository.reconcile() }
assertNotNull(artwork.uri(1, coverArtKey(1, ALBUM_ART, false)))
}
@Test
fun removingTheLastDownloadOfAnAlbumRemovesItsArtwork() =
runTest {
@@ -1354,6 +1399,213 @@ class DownloadRepositoryTest {
}
}
@Test
fun switchingSourcesDoesNotResetAnEditedDownloadedQueue() =
runTest {
seedLibrary()
runBlocking {
db.sourcesDao().upsertSource(Source(2, "other", "http://other/", isActive = false, createdAt = 1))
db.libraryDao().upsertSongs(
listOf(
Song(
2,
"o1",
"other-album",
"other-artist",
"Other",
"Other Album",
"Other Artist",
100,
track = 1,
disc = 1,
starred = null,
genre = null,
),
),
)
db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
db.downloadDao().upsert(SongDownload(2, "o1", DownloadStatus.Completed, downloadedAt = 1))
}
writeFile(1, "s1", "one")
writeFile(1, "s2", "two")
writeFile(2, "o1", "other")
repository.start()
await { repository.localUri("s1") != null }
runBlocking {
queues.replace(listOf(queues.downloadedEntry(1)))
queues.removeAt(queues.snapshot(), 1)
}
val edited = runBlocking { queues.snapshot().entries.map { it.entry } }
assertEquals(1, edited.size)
assertNotNull(edited.single().rangeStart)
runBlocking { sources.selectSource(2) }
await { repository.localUri("o1") != null }
assertEquals(edited, runBlocking { queues.snapshot().entries.map { it.entry } })
}
@Test
fun aDownloadCompletedWhileItsSourceIsInactiveResetsOnReturn() =
runTest {
seedLibrary()
runBlocking {
db.sourcesDao().upsertSource(Source(2, "other", "http://other/", isActive = false, createdAt = 1))
db.libraryDao().upsertSongs(
listOf(
Song(1, "s3", "al1", "ar1", "Song 3", "Album", "Artist", 100, track = 3, disc = 1, starred = null, genre = null),
Song(
2,
"o1",
"other-album",
"other-artist",
"Other",
"Other Album",
"Other Artist",
100,
track = 1,
disc = 1,
starred = null,
genre = null,
),
),
)
db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 1))
db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 2))
db.downloadDao().upsert(SongDownload(2, "o1", DownloadStatus.Completed, downloadedAt = 1))
}
writeFile(1, "s1", "one")
writeFile(1, "s2", "two")
writeFile(2, "o1", "other")
repository.start()
await { repository.localUri("s1") != null }
runBlocking {
queues.replace(listOf(queues.downloadedEntry(1)))
queues.removeAt(queues.snapshot(), 1)
sources.selectSource(2)
}
await { repository.localUri("o1") != null }
assertNotNull(
runBlocking {
queues
.snapshot()
.entries
.single()
.entry.rangeStart
},
)
runBlocking {
repository.captureForReset(1)
db.downloadDao().upsert(SongDownload(1, "s3", DownloadStatus.Completed, downloadedAt = 3))
writeFile(1, "s3", "three")
repository.reconcile()
sources.selectSource(1)
}
await { repository.localUri("s3") != null }
assertNull(
runBlocking {
queues
.snapshot()
.entries
.single()
.entry.rangeStart
},
)
}
@Test
fun backfillingDownloadTimeResetsAnEditedDownloadedQueue() =
runTest {
seedLibrary()
runBlocking {
db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, engineId = 42, downloadedAt = 0))
db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 1))
}
writeFile(1, "s1", "one")
writeFile(1, "s2", "two")
repository.start()
await { repository.localUri("s1") != null && repository.localUri("s2") != null }
runBlocking {
queues.replace(listOf(queues.downloadedEntry(1)))
queues.removeAt(queues.snapshot(), 0)
}
assertNotNull(
runBlocking {
queues
.snapshot()
.entries
.single()
.entry.rangeStart
},
)
engine.complete(42, total = 10)
runBlocking { repository.reconcile() }
await { runBlocking { (row(1, "s1")?.downloadedAt ?: 0L) > 0L } }
Thread.sleep(200)
assertNull(
runBlocking {
queues
.snapshot()
.entries
.single()
.entry.rangeStart
},
)
}
@Test
fun aMultiRowRecoveryKeepsTheResetPointFromBeforeTheWholeBatch() =
runTest {
seedLibrary()
seedSongs(3..3)
runBlocking {
db.downloadDao().upsert(SongDownload(1, "s3", DownloadStatus.Completed, downloadedAt = 1))
queues.replace(listOf(queues.downloadedEntry(1)))
queues.setCursor(0)
repository.captureForReset(1)
db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Completed, downloadedAt = 2))
repository.captureForReset(1)
db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed, downloadedAt = 3))
repository.captureForReset(1)
}
val field = DownloadRepository::class.java.getDeclaredField("pendingResetPoints")
field.isAccessible = true
val pending = field.get(repository) as Map<*, *>
val point = pending[1L] as QueueResetPoint
assertEquals("s3", point.current?.songId)
}
@Test
fun startupRecoveryResetsAnEditedDownloadedQueueWhenAFileIsMissing() =
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")
repository.start()
runBlocking { repository.reconcile() }
await {
runBlocking {
queues
.snapshot()
.entries
.single()
.entry.rangeStart == null
}
}
}
private fun seedPlaylist(songIds: List<String>) {
runBlocking {
db.libraryDao().upsertPlaylists(listOf(Playlist(1, "pl1", "Playlist", null, null, songIds.size.toLong(), 0)))
@@ -72,6 +72,7 @@ class LibraryRepositoryTest {
sourceRepository,
NetworkServerActionSink(sourceRepository),
fakeUserPreferences(),
queueRepository = QueueRepository(db),
showMessage = { messages += it.resolve(resources) },
scope = scope,
)
@@ -538,6 +538,7 @@ class SyncManagerTest {
downloadsDir,
artworkStore = ArtworkStore(File(context.cacheDir, "art")),
artworkFetcher = { ByteArray(0) },
queueRepository = queueRepository,
scope = scope,
)
File(downloadsDir, "1").mkdirs()
@@ -119,6 +119,7 @@ class PlaybackControllerTest {
downloadsDir,
artworkStore = ArtworkStore(File(context.cacheDir, "art")),
artworkFetcher = { ByteArray(0) },
queueRepository = queues,
scope = repoScope,
).also { it.start() }
handle = FakePlayerHandle()
@@ -3928,6 +3929,34 @@ class PlaybackControllerTest {
assertEquals(2L, controller.state.value.position)
}
@Test
fun aDownloadInsertedBeforeCurrentKeepsTheActualCurrentSong() {
seedAlbum(3, sourceId = 1)
seedSong("s0", "al1")
markDownloaded(1, "s1", downloadedAt = 200)
markDownloaded(1, "s2", downloadedAt = 100)
controller.playDownloaded(1, 1)
await { handle.currentItem?.id == "s2" }
markDownloaded(1, "s0", downloadedAt = 300)
await { controller.state.value.position == 2L && handle.currentItem?.id == "s2" }
}
@Test
fun deletingTheCurrentDownloadDetachesThatSong() {
seedAlbum(3, sourceId = 1)
markDownloaded(1, "s1", downloadedAt = 200)
markDownloaded(1, "s2", downloadedAt = 100)
controller.playDownloaded(1, 1)
await { handle.currentItem?.id == "s2" }
runBlocking { downloads.remove(1, "s2") }
await { handle.items.take(2).map { it.id } == listOf("s2", "s1") }
assertEquals("s2", handle.currentItem?.id)
}
@Test
fun aDetachedTrackSurvivesMediaRefresh() {
seedPlaylist(5)
@@ -4369,6 +4398,7 @@ class PlaybackControllerTest {
val deadline = System.nanoTime() + TEST_TIMEOUT_MS * 1_000_000
while (System.nanoTime() < deadline) {
runBlocking {
downloads.captureForReset(sourceId)
db.downloadDao().upsert(SongDownload(sourceId, songId, DownloadStatus.Completed, downloadedAt = downloadedAt))
file.parentFile?.mkdirs()
file.writeBytes(byteArrayOf(1))
@@ -42,6 +42,7 @@ class ViewModelRepositoryFixture {
val downloadEngine = FakeDownloadEngine()
val downloadsDir = File(context.cacheDir, "downloads-${System.nanoTime()}")
val sourceRepository = SourceRepository(db, OkHttpClient(), fakeUserPreferences(), artworkStore, scope = repoScope)
val queueRepository = QueueRepository(db)
val downloadRepository =
DownloadRepository(
db,
@@ -50,9 +51,9 @@ class ViewModelRepositoryFixture {
downloadsDir,
artworkStore = artworkStore,
artworkFetcher = { ByteArray(0) },
queueRepository = queueRepository,
scope = repoScope,
)
val queueRepository = QueueRepository(db)
val controller =
PlaybackController(
sourceRepository,
@@ -68,6 +69,7 @@ class ViewModelRepositoryFixture {
sourceRepository,
NetworkServerActionSink(sourceRepository),
fakeUserPreferences(),
queueRepository = queueRepository,
scope = repoScope,
)
@@ -91,9 +91,10 @@ class NowPlayingRouteTest {
File(context.cacheDir, "downloads-${System.nanoTime()}"),
artworkStore = artworkStore,
artworkFetcher = { ByteArray(0) },
queueRepository = queues,
scope = repoScope,
).also { it.start() }
library = LibraryRepository(db, sources, serverActions, prefs, scope = repoScope)
library = LibraryRepository(db, sources, serverActions, prefs, queueRepository = queues, scope = repoScope)
handle = FakePlayerHandle()
controller =
PlaybackController(
@@ -83,6 +83,7 @@ class QueueViewModelTest {
downloadsDir,
artworkStore = artwork,
artworkFetcher = { ByteArray(0) },
queueRepository = queues,
scope = repoScope,
).also { it.start() }
controller =
@@ -67,6 +67,7 @@ class AddSourceViewModelTest {
File(context.cacheDir, "downloads-${System.nanoTime()}"),
artworkStore = artwork,
artworkFetcher = { ByteArray(0) },
queueRepository = QueueRepository(db),
scope = repoScope,
)
server = MockWebServer()