library: recompute artist aggregates at the sync boundary

Replace the per-credit invalidation with one signal from the sync's write
transactions: if artists, albums or songs changed, the source is marked stale
and a single aggregate pass refreshes counts, plays and the credited flag in
one query. Identical syncs no longer run any of those scans, and a new credit
path cannot forget to invalidate the cache.

A changed sync still reads the credit unions once instead of separately for
counts and plays, and migration 33->34 remains a frozen snapshot rather than
sharing the runtime query text.

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

@@ -253,56 +253,44 @@ interface LibraryDao {
     )
     suspend fun recomputeAlbumPlayData(sourceId: Long)
 
-    // Every credited song, from both the lead ids and the relation tables, deduplicated so a song
+    // Counts, plays and the credited flag come from the same credit unions, so the artist badges,
-    // counts once per artist even when several credit paths point at it.
+    // page, shuffle and downloads cannot disagree.
-    @Query(
-        "WITH credited AS (SELECT DISTINCT * FROM ($CREDITED_SONG_ROWS)), totals AS (" +
-            "SELECT artistId AS id, SUM(playCount) AS playCount, MAX(played) AS played " +
-            "FROM credited GROUP BY artistId" +
-            ") " +
-            "UPDATE artists SET playCount = c.playCount, played = c.played " +
-            "FROM (SELECT artists.id AS id, COALESCE(totals.playCount, 0) AS playCount, totals.played AS played " +
-            "FROM artists LEFT JOIN totals ON totals.id = artists.id " +
-            "WHERE artists.sourceId = :sourceId) c " +
-            "WHERE artists.sourceId = :sourceId AND artists.id = c.id " +
-            "AND (artists.playCount != c.playCount OR artists.played IS NOT c.played)",
-    )
-    suspend fun recomputeArtistPlayData(sourceId: Long)
-
-    // Stored so the download badge query can report an artist's total without walking every song,
-    // and so the artist badges use the same counts as the artist page and the shuffle queue.
     @Query(
         "WITH credited AS (SELECT DISTINCT * FROM ($CREDITED_SONG_ROWS)), " +
             "credited_albums AS ($CREDITED_ALBUM_ROWS), " +
-            "counts AS (" +
+            "totals AS (" +
-            "SELECT artistId AS id, COUNT(DISTINCT songId) AS songCount " +
+            "SELECT artistId AS id, COUNT(DISTINCT songId) AS songCount, " +
+            "SUM(playCount) AS playCount, MAX(played) AS played " +
             "FROM credited GROUP BY artistId" +
-            "), album_counts AS (" +
+            "), album_totals AS (" +
             "SELECT artistId AS id, COUNT(DISTINCT albumId) AS albumCount " +
             "FROM credited_albums GROUP BY artistId" +
             ") " +
-            "UPDATE artists SET songCount = c.songCount, albumCount = c.albumCount, credited = c.credited " +
+            "UPDATE artists SET songCount = c.songCount, albumCount = c.albumCount, " +
-            "FROM (SELECT artists.id AS id, COALESCE(counts.songCount, 0) AS songCount, " +
+            "playCount = c.playCount, played = c.played, credited = c.credited " +
-            "COALESCE(album_counts.albumCount, 0) AS albumCount, " +
+            "FROM (SELECT artists.id AS id, COALESCE(totals.songCount, 0) AS songCount, " +
+            "COALESCE(album_totals.albumCount, 0) AS albumCount, " +
+            "COALESCE(totals.playCount, 0) AS playCount, totals.played AS played, " +
             "((NOT EXISTS (SELECT 1 FROM album_artists WHERE sourceId = artists.sourceId) " +
             "AND NOT EXISTS (SELECT 1 FROM song_artists WHERE sourceId = artists.sourceId)) " +
             "OR EXISTS (SELECT 1 FROM album_artists aa " +
             "WHERE aa.sourceId = artists.sourceId AND aa.artistId = artists.id) " +
             "OR EXISTS (SELECT 1 FROM song_artists sa " +
             "WHERE sa.sourceId = artists.sourceId AND sa.artistId = artists.id)) AS credited " +
-            "FROM artists LEFT JOIN counts ON counts.id = artists.id " +
+            "FROM artists LEFT JOIN totals ON totals.id = artists.id " +
-            "LEFT JOIN album_counts ON album_counts.id = artists.id " +
+            "LEFT JOIN album_totals ON album_totals.id = artists.id " +
             "WHERE artists.sourceId = :sourceId) c " +
             "WHERE artists.sourceId = :sourceId AND artists.id = c.id " +
             "AND (artists.songCount != c.songCount OR artists.albumCount != c.albumCount " +
+            "OR artists.playCount != c.playCount OR artists.played IS NOT c.played " +
             "OR artists.credited != c.credited)",
     )
-    suspend fun recomputeArtistSongCounts(sourceId: Long)
+    suspend fun recomputeArtistAggregates(sourceId: Long)
 
     @Transaction
-    suspend fun recomputePlayData(sourceId: Long) {
+    suspend fun recomputeAggregates(sourceId: Long) {
         recomputeAlbumPlayData(sourceId)
-        recomputeArtistPlayData(sourceId)
+        recomputeArtistAggregates(sourceId)
     }
 
     @Query(

app/src/main/kotlin/com/subtracks/data/sync/SyncService.kt

@@ -31,23 +31,17 @@ class SyncService(
     private var playlistSongsSynced = 0
     private var pruned = 0
 
-    private var creditsChanged = false
-
     suspend fun sync(): SyncSummary {
         val sources = db.sourcesDao()
         val wasStale = sources.creditsStale(source.id) ?: true
-        creditsChanged = false
+        val artistsChanged = syncArtists()
-        val artistsBefore = dbTotalChanges()
+        val albumsChanged = syncAlbums()
-        syncArtists()
+        val songsChanged = syncSongs()
-        val artistsChanged = dbTotalChanges() != artistsBefore
-        syncAlbums()
-        syncSongs()
         optimize()
-        if (creditsChanged || artistsChanged || wasStale) {
+        if (artistsChanged || albumsChanged || songsChanged || wasStale) {
-            recomputeArtistSongCounts()
+            recomputeAggregates()
             sources.setCreditsStale(source.id, false)
         }
-        aggregatePlayData()
         syncPlaylists()
         syncPlaylistSongs()
         return SyncSummary(
@@ -60,29 +54,35 @@ class SyncService(
         )
     }
 
-    private suspend fun syncArtists() {
+    private suspend fun syncArtists(): Boolean {
         val library = db.libraryDao()
         val seen = HashedIds()
+        var changed = false
         source.artists().collect { batch ->
             if (batch.isEmpty()) return@collect
-            write("artists", artistColumns, primaryKey, batch, preserved = setOf("albumCount")) { it.values() }
+            changed =
+                write("artists", artistColumns, primaryKey, batch, preserved = setOf("albumCount"), markStale = true) { it.values() } ||
+                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) })
+        return changed || pruned > prunedBefore
     }
 
-    private suspend fun syncAlbums() {
+    private suspend fun syncAlbums(): Boolean {
         val library = db.libraryDao()
         val seenAlbums = HashedIds()
         val seenDiscs = HashedIds()
+        var changed = false
         source.albums().collect { batch ->
             if (batch.isEmpty()) return@collect
             val discs =
                 batch.flatMap { album ->
                     album.discTitles.map { (disc, title) -> Disc(album.sourceId, album.id, disc, title) }
                 }
-            writeAlbumsAndDiscs(batch, discs)
+            changed = writeAlbumsAndDiscs(batch, discs) || changed
             batch.forEach { seenAlbums.add(idHash(it.id)) }
             discs.forEach { seenDiscs.add(discHash(it.albumId, it.disc)) }
             albumsSynced += batch.size
@@ -95,19 +95,20 @@ class SyncService(
                 { library.deleteAlbums(source.id, it) },
                 markStale = { db.sourcesDao().setCreditsStale(source.id, true) },
             )
-        if (pruned > prunedBefore) creditsChanged = true
         pruned +=
             pruneStaleDiscs(seenDiscs, { albumId, disc -> library.discKeysAfter(source.id, albumId, disc, PRUNE_PAGE) }) {
                 deleteStaleDiscs(it)
             }
+        return changed || pruned > prunedBefore
     }
 
-    private suspend fun syncSongs() {
+    private suspend fun syncSongs(): Boolean {
         val library = db.libraryDao()
         val seen = HashedIds()
+        var changed = false
         source.songs().collect { batch ->
             if (batch.isEmpty()) return@collect
-            writeSongs(batch)
+            changed = writeSongs(batch) || changed
             batch.forEach { seen.add(idHash(it.id)) }
             songsSynced += batch.size
         }
@@ -119,10 +120,10 @@ class SyncService(
                 { library.deleteSongs(source.id, it) },
                 markStale = { db.sourcesDao().setCreditsStale(source.id, true) },
             )
-        if (pruned > prunedBefore) creditsChanged = true
+        return changed || pruned > prunedBefore
     }
 
-    private suspend fun writeSongs(batch: List<Song>) {
+    private suspend fun writeSongs(batch: List<Song>): Boolean {
         val names = batch.associate { it.id to it.genreNames() }
         val genres =
             batch.flatMap { song ->
@@ -132,10 +133,9 @@ class SyncService(
             batch.flatMap { song ->
                 song.artistPositions().map { (position, ref) -> SongArtistRef(song.sourceId, song.id, position, ref.id, ref.name) }
             }
-        db.useWriterConnection { connection ->
+        return db.useWriterConnection { connection ->
             connection.immediateTransaction {
-                val songCreditsChanged = connection.songCreditsChanged(batch)
+                val before = connection.totalChanges()
-                if (songCreditsChanged) creditsChanged = true
                 val resolved = connection.resolveSongSortNames(batch)
                 connection.upsertChanged("songs", songColumns, primaryKey, resolved) { it.values() }
                 connection.usePrepared("DELETE FROM song_genres WHERE sourceId = ? AND songId = ? AND position >= ?") { statement ->
@@ -149,7 +149,6 @@ class SyncService(
                     }
                 }
                 connection.upsertChanged("song_genres", songGenreColumns, songGenreKey, genres) { it.values() }
-                val artistRowsBefore = connection.totalChanges()
                 connection.usePrepared("DELETE FROM song_artists WHERE sourceId = ? AND songId = ? AND position >= ?") { statement ->
                     batch.forEach { song ->
                         statement.bind(1, song.sourceId)
@@ -161,21 +160,15 @@ class SyncService(
                     }
                 }
                 connection.upsertChanged("song_artists", songArtistColumns, songArtistKey, artists) { it.values() }
-                val artistRowsChanged = connection.totalChanges() != artistRowsBefore
+                (connection.totalChanges() != before).also { changed ->
-                if (songCreditsChanged || artistRowsChanged) {
+                    if (changed) connection.markCreditsStale(batch.first().sourceId)
-                    connection.markCreditsStale(batch.first().sourceId)
-                    creditsChanged = true
                 }
             }
         }
     }
 
-    private suspend fun aggregatePlayData() {
+    private suspend fun recomputeAggregates() {
-        db.libraryDao().recomputePlayData(source.id)
+        db.libraryDao().recomputeAggregates(source.id)
-    }
-
-    private suspend fun recomputeArtistSongCounts() {
-        db.libraryDao().recomputeArtistSongCounts(source.id)
     }
 
     private suspend fun PooledConnection.totalChanges(): Long {
@@ -187,8 +180,6 @@ class SyncService(
         return value
     }
 
-    private suspend fun dbTotalChanges(): Long = db.useWriterConnection { it.totalChanges() }
-
     private suspend fun PooledConnection.markCreditsStale(sourceId: Long) {
         usePrepared("UPDATE sources SET creditsStale = 1 WHERE id = ? AND creditsStale = 0") { statement ->
             statement.bindLong(1, sourceId)
@@ -196,51 +187,6 @@ class SyncService(
         }
     }
 
-    private suspend fun PooledConnection.songCreditsChanged(batch: List<Song>): Boolean {
-        val expected = batch.associate { it.id to (it.albumId to it.artistId) }.toMutableMap()
-        if (expected.isEmpty()) return false
-        var changed = false
-        batch
-            .map { it.sourceId to it.id }
-            .chunked(SQL_VARIABLES)
-            .forEach { chunk ->
-                val placeholders = chunk.joinToString(",") { "?" }
-                usePrepared("SELECT id, albumId, artistId FROM songs WHERE sourceId = ? AND id IN ($placeholders)") { statement ->
-                    statement.bindLong(1, chunk.first().first)
-                    chunk.forEachIndexed { index, (_, id) -> statement.bindText(index + 2, id) }
-                    while (statement.step()) {
-                        val id = statement.getText(0)
-                        val albumId = if (statement.isNull(1)) null else statement.getText(1)
-                        val artistId = if (statement.isNull(2)) null else statement.getText(2)
-                        if (expected.remove(id) != (albumId to artistId)) changed = true
-                    }
-                }
-            }
-        return changed || expected.isNotEmpty()
-    }
-
-    private suspend fun PooledConnection.albumLeadsChanged(batch: List<Album>): Boolean {
-        val expected = batch.associate { it.id to it.artistId }.toMutableMap()
-        if (expected.isEmpty()) return false
-        var changed = false
-        batch
-            .map { it.sourceId to it.id }
-            .chunked(SQL_VARIABLES)
-            .forEach { chunk ->
-                val placeholders = chunk.joinToString(",") { "?" }
-                usePrepared("SELECT id, artistId FROM albums WHERE sourceId = ? AND id IN ($placeholders)") { statement ->
-                    statement.bindLong(1, chunk.first().first)
-                    chunk.forEachIndexed { index, (_, id) -> statement.bindText(index + 2, id) }
-                    while (statement.step()) {
-                        val id = statement.getText(0)
-                        val artistId = if (statement.isNull(1)) null else statement.getText(1)
-                        if (expected.remove(id) != artistId) changed = true
-                    }
-                }
-            }
-        return changed || expected.isNotEmpty()
-    }
-
     private suspend fun PooledConnection.resolveSongSortNames(batch: List<Song>): List<Song> {
         if (batch.isEmpty()) return batch
         val sortNames =
@@ -338,30 +284,33 @@ class SyncService(
         key: List<String>,
         rows: List<T>,
         preserved: Set<String> = emptySet(),
+        markStale: Boolean = false,
         values: (T) -> List<Any?>,
-    ) {
+    ): Boolean =
         db.useWriterConnection { connection ->
             connection.immediateTransaction {
+                val before = connection.totalChanges()
                 connection.upsertChanged(table, columns, key, rows, preserved, values)
+                (connection.totalChanges() != before).also { changed ->
+                    if (changed && markStale) connection.markCreditsStale(source.id)
+                }
             }
         }
-    }
 
     private suspend fun writeAlbumsAndDiscs(
         albums: List<Album>,
         discs: List<Disc>,
-    ) {
+    ): Boolean {
         val artists =
             albums.flatMap { album ->
                 album.artistPositions().map { (position, ref) -> AlbumArtistRef(album.sourceId, album.id, position, ref.id, ref.name) }
             }
-        db.useWriterConnection { connection ->
+        return db.useWriterConnection { connection ->
             connection.immediateTransaction {
-                val leadsChanged = connection.albumLeadsChanged(albums)
+                val before = connection.totalChanges()
                 val resolved = connection.resolveAlbumSortNames(albums)
                 connection.upsertChanged("albums", albumColumns, primaryKey, resolved) { it.values() }
                 connection.upsertChanged("discs", discColumns, discKey, discs) { it.values() }
-                val artistRowsBefore = connection.totalChanges()
                 connection.usePrepared("DELETE FROM album_artists WHERE sourceId = ? AND albumId = ? AND position >= ?") { statement ->
                     albums.forEach { album ->
                         statement.bind(1, album.sourceId)
@@ -373,9 +322,8 @@ class SyncService(
                     }
                 }
                 connection.upsertChanged("album_artists", albumArtistColumns, albumArtistKey, artists) { it.values() }
-                if (leadsChanged || connection.totalChanges() != artistRowsBefore) {
+                (connection.totalChanges() != before).also { changed ->
-                    connection.markCreditsStale(albums.first().sourceId)
+                    if (changed) connection.markCreditsStale(albums.first().sourceId)
-                    creditsChanged = true
                 }
             }
         }

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

@@ -189,8 +189,7 @@ class LibraryDaoTest {
                 ),
             )
 
-            dao.recomputeAlbumPlayData(sourceId)
+            dao.recomputeAggregates(sourceId)
-            dao.recomputeArtistPlayData(sourceId)
 
             assertEquals(3L, dao.albumOnce(sourceId, "al-1")!!.playCount)
             assertEquals(100L, dao.albumOnce(sourceId, "al-1")!!.played)

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

@@ -134,7 +134,7 @@ class QueryTimingMeasurementTest {
         timed("home recently starred") { dao.recentlyStarredSongs(1, 20).first() }
         timed("home genres") { dao.genresByMostPlayed(1).first() }
         timed("home decades") { dao.decades(1).first() }
-        timed("recompute play data") { dao.recomputePlayData(1) }
+        timed("recompute aggregates") { dao.recomputeAggregates(1) }
     }
 
     private suspend fun <T : Any> PagingSource<Int, T>.firstPage(): List<T> =