Queue stars and scrobbles while the server is unreachable

Actions that cannot be delivered are persisted in a pending_actions
table and replayed once the server is reachable again. Stars coalesce so
only the latest state per item is sent, and replay happens on the next
successful action, when offline mode is turned off, or before a sync.

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

@@ -369,6 +369,21 @@ val MIGRATION_25_26 =
         }
     }
 
+val MIGRATION_26_27 =
+    object : Migration(26, 27) {
+        override suspend fun migrate(connection: SQLiteConnection) {
+            connection.execSQL(
+                "CREATE TABLE IF NOT EXISTS `pending_actions` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, " +
+                    "`sourceId` INTEGER NOT NULL, `kind` TEXT NOT NULL, `targetId` TEXT NOT NULL, " +
+                    "`starType` TEXT, `time` INTEGER NOT NULL DEFAULT 0, " +
+                    "FOREIGN KEY(`sourceId`) REFERENCES `sources`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )",
+            )
+            connection.execSQL(
+                "CREATE INDEX IF NOT EXISTS `index_pending_actions_sourceId` ON `pending_actions` (`sourceId`)",
+            )
+        }
+    }
+
 // Mirrors the FTS5 table and content-sync triggers Room generates for the @Fts5 entities.
 private suspend fun createSearchIndex(
     connection: SQLiteConnection,
@@ -428,4 +443,5 @@ val MIGRATIONS: Array<Migration> =
         MIGRATION_23_24,
         MIGRATION_24_25,
         MIGRATION_25_26,
+        MIGRATION_26_27,
     )

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

@@ -0,0 +1,28 @@
+package com.subtracks.data.db
+
+import androidx.room3.Dao
+import androidx.room3.Insert
+import androidx.room3.Query
+import com.subtracks.data.model.PendingAction
+
+@Dao
+interface PendingActionDao {
+    @Query("SELECT * FROM pending_actions WHERE sourceId = :sourceId ORDER BY id")
+    suspend fun pending(sourceId: Long): List<PendingAction>
+
+    @Insert
+    suspend fun insert(action: PendingAction): Long
+
+    @Query("DELETE FROM pending_actions WHERE id = :id")
+    suspend fun delete(id: Long)
+
+    @Query(
+        "DELETE FROM pending_actions WHERE sourceId = :sourceId " +
+            "AND kind IN ('Star', 'Unstar') AND starType = :starType AND targetId = :targetId",
+    )
+    suspend fun clearStar(
+        sourceId: Long,
+        starType: String,
+        targetId: String,
+    )
+}

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

@@ -11,6 +11,8 @@ import com.subtracks.data.model.ArtworkSeed
 import com.subtracks.data.model.Disc
 import com.subtracks.data.model.DownloadErrorConverter
 import com.subtracks.data.model.DownloadStatusConverter
+import com.subtracks.data.model.PendingAction
+import com.subtracks.data.model.PendingActionKindConverter
 import com.subtracks.data.model.PlaybackCursor
 import com.subtracks.data.model.Playlist
 import com.subtracks.data.model.PlaylistSearch
@@ -45,11 +47,17 @@ import com.subtracks.data.model.UpNextEntry
         PlaybackCursor::class,
         ArtworkSeed::class,
         SongDownload::class,
+        PendingAction::class,
     ],
-    version = 26,
+    version = 27,
     exportSchema = true,
 )
-@ColumnTypeConverters(QueueKindConverter::class, DownloadStatusConverter::class, DownloadErrorConverter::class)
+@ColumnTypeConverters(
+    QueueKindConverter::class,
+    DownloadStatusConverter::class,
+    DownloadErrorConverter::class,
+    PendingActionKindConverter::class,
+)
 abstract class SubtracksDatabase : RoomDatabase() {
     abstract fun libraryDao(): LibraryDao
 
@@ -60,4 +68,6 @@ abstract class SubtracksDatabase : RoomDatabase() {
     abstract fun artworkSeedDao(): ArtworkSeedDao
 
     abstract fun downloadDao(): DownloadDao
+
+    abstract fun pendingActionDao(): PendingActionDao
 }

app/src/main/kotlin/com/subtracks/data/model/Models.kt

@@ -287,6 +287,29 @@ data class SongGenre(
     val genre: String,
 )
 
+enum class PendingActionKind { Scrobble, Star, Unstar }
+
+@Entity(
+    tableName = "pending_actions",
+    foreignKeys = [
+        ForeignKey(
+            entity = Source::class,
+            parentColumns = ["id"],
+            childColumns = ["sourceId"],
+            onDelete = ForeignKey.CASCADE,
+        ),
+    ],
+    indices = [Index("sourceId")],
+)
+data class PendingAction(
+    @PrimaryKey(autoGenerate = true) val id: Long = 0,
+    val sourceId: Long,
+    val kind: PendingActionKind,
+    val targetId: String,
+    val starType: String? = null,
+    @ColumnInfo(defaultValue = "0") val time: Long = 0,
+)
+
 sealed interface SongItem {
     val song: Song
     val coverArt: String?
@@ -442,3 +465,12 @@ class QueueKindConverter {
     @ColumnTypeConverter
     fun toQueueKind(value: String): QueueKind = QueueKind.entries.firstOrNull { it.name == value } ?: QueueKind.Song
 }
+
+class PendingActionKindConverter {
+    @ColumnTypeConverter
+    fun fromPendingActionKind(kind: PendingActionKind): String = kind.name
+
+    @ColumnTypeConverter
+    fun toPendingActionKind(value: String): PendingActionKind =
+        PendingActionKind.entries.firstOrNull { it.name == value } ?: PendingActionKind.Scrobble
+}

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

@@ -25,4 +25,6 @@ class NetworkServerActionSink(
         val source = sources.activeMusicSource() ?: throw IllegalStateException("No active server")
         source.setStar(type, id, starred)
     }
+
+    override suspend fun flush() = Unit
 }

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

@@ -0,0 +1,135 @@
+package com.subtracks.data.repo
+
+import com.subtracks.data.db.SubtracksDatabase
+import com.subtracks.data.model.PendingAction
+import com.subtracks.data.model.PendingActionKind
+import com.subtracks.data.source.ServerActionSink
+import com.subtracks.data.source.StarType
+import com.subtracks.data.source.subsonic.SubsonicException
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.drop
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import java.io.IOException
+
+class PendingServerActionSink(
+    private val delegate: ServerActionSink,
+    private val db: SubtracksDatabase,
+    private val activeSourceId: suspend () -> Long?,
+    private val offline: StateFlow<Boolean>,
+    scope: CoroutineScope,
+) : ServerActionSink {
+    private val flushLock = Mutex()
+
+    init {
+        scope.launch {
+            // The first emission is the StateFlow's default, not necessarily the persisted value, and
+            // replaying onto the network must not race a cold start with offline mode on.
+            offline
+                .drop(1)
+                .collect { isOffline -> if (!isOffline) flush() }
+        }
+    }
+
+    override suspend fun nowPlaying(songId: String) {
+        if (offline.value) return
+        try {
+            delegate.nowPlaying(songId)
+            flush()
+        } catch (cancellation: CancellationException) {
+            throw cancellation
+        } catch (_: Exception) {
+        }
+    }
+
+    override suspend fun scrobble(
+        songId: String,
+        time: Long,
+    ) {
+        if (offline.value) {
+            queue(PendingActionKind.Scrobble, songId, null, time)
+            return
+        }
+        try {
+            delegate.scrobble(songId, time)
+            flush()
+        } catch (cancellation: CancellationException) {
+            throw cancellation
+        } catch (_: Exception) {
+            queue(PendingActionKind.Scrobble, songId, null, time)
+        }
+    }
+
+    override suspend fun setStar(
+        type: StarType,
+        id: String,
+        starred: Boolean,
+    ) {
+        val kind = if (starred) PendingActionKind.Star else PendingActionKind.Unstar
+        if (offline.value) {
+            if (!queue(kind, id, type.name, 0)) throw IllegalStateException("No active server")
+            return
+        }
+        try {
+            delegate.setStar(type, id, starred)
+            flush()
+        } catch (cancellation: CancellationException) {
+            throw cancellation
+        } catch (rejected: SubsonicException) {
+            // The server answered and refused; retrying the same request will not help.
+            throw rejected
+        } catch (failure: Exception) {
+            if (!queue(kind, id, type.name, 0)) throw failure
+        }
+    }
+
+    override suspend fun flush() {
+        flushLock.withLock {
+            if (offline.value) return
+            val sourceId = activeSourceId() ?: return
+            val dao = db.pendingActionDao()
+            for (action in dao.pending(sourceId)) {
+                try {
+                    deliver(action)
+                } catch (cancellation: CancellationException) {
+                    throw cancellation
+                } catch (_: IOException) {
+                    // Still unreachable: keep this and the remaining actions for the next attempt.
+                    return
+                } catch (_: Exception) {
+                    // The server answered and refused; drop the action rather than block the queue.
+                }
+                dao.delete(action.id)
+            }
+        }
+    }
+
+    private suspend fun deliver(action: PendingAction) {
+        when (action.kind) {
+            PendingActionKind.Scrobble -> {
+                delegate.scrobble(action.targetId, action.time)
+            }
+
+            PendingActionKind.Star, PendingActionKind.Unstar -> {
+                val type = StarType.valueOf(action.starType ?: error("Star action without a type"))
+                delegate.setStar(type, action.targetId, action.kind == PendingActionKind.Star)
+            }
+        }
+    }
+
+    private suspend fun queue(
+        kind: PendingActionKind,
+        targetId: String,
+        starType: String?,
+        time: Long,
+    ): Boolean {
+        val sourceId = activeSourceId() ?: return false
+        val dao = db.pendingActionDao()
+        if (starType != null) dao.clearStar(sourceId, starType, targetId)
+        dao.insert(PendingAction(sourceId = sourceId, kind = kind, targetId = targetId, starType = starType, time = time))
+        return true
+    }
+}

app/src/main/kotlin/com/subtracks/data/source/ServerActionSink.kt

@@ -1,9 +1,8 @@
 package com.subtracks.data.source
 
 /**
- * Writes the app pushes back to the server. Today they go straight over the
- * network, but funneling them through one seam leaves room to persist them and
- * replay later while the server is unreachable.
+ * Writes the app pushes back to the server. Actions that could not be delivered
+ * while the server was unreachable are persisted and replayed by [flush].
  */
 interface ServerActionSink {
     suspend fun nowPlaying(songId: String)
@@ -18,4 +17,6 @@ interface ServerActionSink {
         id: String,
         starred: Boolean,
     )
+
+    suspend fun flush()
 }

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

@@ -6,6 +6,7 @@ import com.subtracks.UiMessage
 import com.subtracks.data.db.SubtracksDatabase
 import com.subtracks.data.repo.QueueRepository
 import com.subtracks.data.repo.SourceRepository
+import com.subtracks.data.source.ServerActionSink
 import com.subtracks.data.source.subsonic.SubsonicException
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
@@ -32,6 +33,7 @@ class SyncManager(
     private val db: SubtracksDatabase,
     private val sourceRepository: SourceRepository,
     private val queueRepository: QueueRepository,
+    private val serverActions: ServerActionSink,
     private val showMessage: (UiMessage) -> Unit = {},
     private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
 ) {
@@ -60,6 +62,7 @@ class SyncManager(
         val result =
             try {
                 val source = sourceRepository.activeMusicSource() ?: throw NoServerException()
+                serverActions.flush()
                 SyncService(db, source).sync()
                 SyncStatus.Success
             } catch (cancellation: CancellationException) {

app/src/main/kotlin/com/subtracks/di/AppModule.kt

@@ -22,6 +22,7 @@ import com.subtracks.data.repo.ArtworkSeedStore
 import com.subtracks.data.repo.DownloadRepository
 import com.subtracks.data.repo.LibraryRepository
 import com.subtracks.data.repo.NetworkServerActionSink
+import com.subtracks.data.repo.PendingServerActionSink
 import com.subtracks.data.repo.QueueRepository
 import com.subtracks.data.repo.SourceRepository
 import com.subtracks.data.source.ServerActionSink
@@ -77,10 +78,18 @@ fun appModule(
             scope = get(),
         )
     }
-    single<ServerActionSink> { NetworkServerActionSink(get()) }
+    single<ServerActionSink> {
+        PendingServerActionSink(
+            delegate = NetworkServerActionSink(get()),
+            db = get(),
+            activeSourceId = get<SourceRepository>()::activeSourceIdOnce,
+            offline = get<SourceRepository>().offline,
+            scope = get(),
+        )
+    }
     single { LibraryRepository(get(), get(), get(), toast, scope = get(named("io"))) }
     single<ArtworkSeedStore> { ArtworkSeedRepository(get()) }
-    single { SyncManager(get(), get(), get(), showMessage = toast, scope = get()) }
+    single { SyncManager(get(), get(), get(), get(), showMessage = toast, scope = get()) }
     single { QueueRepository(get()) }
     single { ArtworkStore(downloadsRoot(context)) }
     single<ArtworkFetcher> { OkHttpArtworkFetcher(get()) }

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

@@ -0,0 +1,189 @@
+package com.subtracks.data.repo
+
+import android.content.Context
+import androidx.room3.Room
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.subtracks.cancelAndJoinBlocking
+import com.subtracks.data.db.SubtracksDatabase
+import com.subtracks.data.model.PendingActionKind
+import com.subtracks.data.model.Source
+import com.subtracks.data.source.ServerActionSink
+import com.subtracks.data.source.StarType
+import com.subtracks.data.source.subsonic.SubsonicException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.runBlocking
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertThrows
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import java.io.IOException
+import java.util.concurrent.CopyOnWriteArrayList
+
+@RunWith(AndroidJUnit4::class)
+class PendingServerActionSinkTest {
+    private sealed interface Call {
+        data class Star(
+            val type: StarType,
+            val id: String,
+            val starred: Boolean,
+        ) : Call
+
+        data class Scrobble(
+            val songId: String,
+            val time: Long,
+        ) : Call
+    }
+
+    private class RecordingSink : ServerActionSink {
+        val calls = CopyOnWriteArrayList<Call>()
+        var starFailure: Exception? = null
+        var scrobbleFailure: Exception? = null
+
+        override suspend fun nowPlaying(songId: String) = Unit
+
+        override suspend fun scrobble(
+            songId: String,
+            time: Long,
+        ) {
+            scrobbleFailure?.let { throw it }
+            calls += Call.Scrobble(songId, time)
+        }
+
+        override suspend fun setStar(
+            type: StarType,
+            id: String,
+            starred: Boolean,
+        ) {
+            starFailure?.let { throw it }
+            calls += Call.Star(type, id, starred)
+        }
+
+        override suspend fun flush() = Unit
+    }
+
+    private lateinit var db: SubtracksDatabase
+    private lateinit var scope: CoroutineScope
+    private lateinit var delegate: RecordingSink
+    private lateinit var sink: PendingServerActionSink
+    private val offline = MutableStateFlow(true)
+    private var sourceId = 0L
+
+    @Before
+    fun setUp() {
+        val context = ApplicationProvider.getApplicationContext<Context>()
+        db =
+            Room
+                .inMemoryDatabaseBuilder(context, SubtracksDatabase::class.java)
+                .setDriver(BundledSQLiteDriver())
+                .build()
+        scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        sourceId = runBlocking { db.sourcesDao().insertSource(Source(name = "server", address = "http://x/", createdAt = 0)) }
+        delegate = RecordingSink()
+        sink = PendingServerActionSink(delegate, db, activeSourceId = { sourceId }, offline = offline, scope = scope)
+    }
+
+    @After
+    fun tearDown() {
+        cancelAndJoinBlocking(scope)
+        db.close()
+    }
+
+    @Test
+    fun anOfflineStarIsQueuedAndReplayedWhenOnline() =
+        runBlocking {
+            sink.setStar(StarType.Song, "s1", true)
+
+            assertEquals(emptyList<Call>(), delegate.calls)
+            assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+
+            offline.value = false
+            sink.flush()
+
+            assertEquals(listOf(Call.Star(StarType.Song, "s1", true)), delegate.calls)
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
+
+    @Test
+    fun anOfflineUnstarCoalescesAwayAnEarlierStar() =
+        runBlocking {
+            sink.setStar(StarType.Song, "s1", true)
+            sink.setStar(StarType.Song, "s1", false)
+
+            val queued = db.pendingActionDao().pending(sourceId)
+            assertEquals(1, queued.size)
+            assertEquals(PendingActionKind.Unstar, queued.single().kind)
+
+            offline.value = false
+            sink.flush()
+
+            assertEquals(listOf(Call.Star(StarType.Song, "s1", false)), delegate.calls)
+        }
+
+    @Test
+    fun offlineScrobblesReplayInOrderWithTheirTimes() =
+        runBlocking {
+            sink.scrobble("s1", 1_000)
+            sink.scrobble("s2", 2_000)
+
+            offline.value = false
+            sink.flush()
+
+            assertEquals(listOf(Call.Scrobble("s1", 1_000), Call.Scrobble("s2", 2_000)), delegate.calls)
+        }
+
+    @Test
+    fun aConnectionFailureQueuesTheScrobbleAndFlushRetries() =
+        runBlocking {
+            offline.value = false
+            delegate.scrobbleFailure = IOException("down")
+
+            sink.scrobble("s1", 1_000)
+
+            assertEquals(emptyList<Call>(), delegate.calls)
+            assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+
+            delegate.scrobbleFailure = null
+            sink.flush()
+
+            assertEquals(listOf(Call.Scrobble("s1", 1_000)), delegate.calls)
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
+
+    @Test
+    fun aStarTheServerRejectsIsNotQueued() =
+        runBlocking {
+            offline.value = false
+            delegate.starFailure = SubsonicException(40, "nope")
+
+            assertThrows(SubsonicException::class.java) {
+                runBlocking { sink.setStar(StarType.Song, "s1", true) }
+            }
+
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
+
+    @Test
+    fun flushKeepsQueuedActionsWhileTheServerIsStillUnreachable() =
+        runBlocking {
+            sink.setStar(StarType.Song, "s1", true)
+
+            offline.value = false
+            delegate.starFailure = IOException("down")
+            sink.flush()
+
+            assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+
+            delegate.starFailure = null
+            sink.flush()
+
+            assertEquals(listOf(Call.Star(StarType.Song, "s1", true)), delegate.calls)
+            assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+        }
+}

app/src/test/kotlin/com/subtracks/data/sync/SyncManagerTest.kt

@@ -11,6 +11,7 @@ import com.subtracks.cancelAndJoinBlocking
 import com.subtracks.data.db.SubtracksDatabase
 import com.subtracks.data.download.ArtworkStore
 import com.subtracks.data.prefs.fakeUserPreferences
+import com.subtracks.data.repo.NetworkServerActionSink
 import com.subtracks.data.repo.QueueRepository
 import com.subtracks.data.repo.SourceRepository
 import kotlinx.coroutines.CoroutineScope
@@ -61,7 +62,15 @@ class SyncManagerTest {
         sourceRepository =
             SourceRepository(db, OkHttpClient(), fakeUserPreferences(), ArtworkStore(File(context.cacheDir, "art")), scope = scope)
         queueRepository = QueueRepository(db)
-        manager = SyncManager(db, sourceRepository, queueRepository, showMessage = { messages += it.resolve(resources) }, scope = scope)
+        manager =
+            SyncManager(
+                db,
+                sourceRepository,
+                queueRepository,
+                NetworkServerActionSink(sourceRepository),
+                showMessage = { messages += it.resolve(resources) },
+                scope = scope,
+            )
         server = MockWebServer()
         server.start()
     }

app/src/test/kotlin/com/subtracks/playback/ScrobblerTest.kt

@@ -48,6 +48,8 @@ class ScrobblerTest {
             id: String,
             starred: Boolean,
         ) = Unit
+
+        override suspend fun flush() = Unit
     }
 
     private fun TestScope.testScope() = CoroutineScope(UnconfinedTestDispatcher(testScheduler) + SupervisorJob())

app/src/test/kotlin/com/subtracks/ui/settings/AddSourceViewModelTest.kt

@@ -13,6 +13,7 @@ import com.subtracks.data.download.ArtworkStore
 import com.subtracks.data.download.FakeDownloadEngine
 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.sync.SyncManager
@@ -154,7 +155,8 @@ class AddSourceViewModelTest {
         }
 
     private fun viewModel(sourceId: Long? = null): AddSourceViewModel {
-        val syncManager = SyncManager(db, sourceRepository, QueueRepository(db), scope = repoScope)
+        val syncManager =
+            SyncManager(db, sourceRepository, QueueRepository(db), NetworkServerActionSink(sourceRepository), scope = repoScope)
         val resources = ApplicationProvider.getApplicationContext<Context>().resources
         return AddSourceViewModel(sourceRepository, syncManager, downloadRepository, sourceId, resources).also { createdViewModels += it }
     }