feat(android): Phase 18 — MutationReplayer drains offline write queue
Closes the last MVP infrastructure gap. Queued like-toggles,
Lidarr requests, and quarantine-unflags now actually reach the
server instead of accumulating in cached_mutations forever.
New:
- cache/mutations/MutationReplayer.kt — @Singleton. On
construction subscribes to AuthStore.sessionCookie and runs a
drain pass on every signed-in transition (cold start with
persisted cookie OR fresh sign-in). Reads pending rows in
FIFO order via CachedMutationDao.getAll, dispatches each by
kind to the raw Retrofit API:
LIKE_TOGGLE → LikesApi.like / unlike
REQUEST_CREATE → DiscoverApi.createRequest
QUARANTINE_UNFLAG → QuarantineApi.unflag
Crucially, uses the raw API interfaces — going through the
Repository wrappers would re-enqueue on failure, creating an
infinite-loop. Successful rows are deleted; failed rows stay in
place with attempts + lastAttemptAt updated. Unknown kinds are
dropped (claim success) so a stale schema entry can't wedge the
queue. Single in-flight via Mutex so back-to-back cookie events
coalesce.
Modified:
- MinstrelApplication.kt — adds @Inject lateinit var
mutationReplayer (same construct-the-singleton trick used for
ResumeController and SyncController). Without the @Inject Hilt
never instantiates the replayer and its init {} cookie observer
never subscribes.
Closes Phase 18 + every known MVP infrastructure gap. Remaining
known follow-ups (NOT MVP blockers):
- WorkManager-driven connectivity-listener replayer so queued
writes drain even with the app backgrounded. Current trigger
set (app open + sign-in) covers the common path.
- Exponential backoff + max-attempts cap so permanently-failing
rows eventually fail visibly rather than silently retrying
forever. Retry-forever is cheap given small queue sizes.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -6,6 +6,7 @@ import androidx.work.Configuration
|
||||
import coil3.ImageLoader
|
||||
import coil3.SingletonImageLoader
|
||||
import coil3.network.okhttp.OkHttpNetworkFetcherFactory
|
||||
import com.fabledsword.minstrel.cache.mutations.MutationReplayer
|
||||
import com.fabledsword.minstrel.cache.sync.SyncController
|
||||
import com.fabledsword.minstrel.di.ApplicationScope
|
||||
import com.fabledsword.minstrel.player.ResumeController
|
||||
@@ -48,6 +49,15 @@ class MinstrelApplication :
|
||||
*/
|
||||
@Suppress("unused") @Inject lateinit var syncController: SyncController
|
||||
|
||||
/**
|
||||
* Same construct-the-singleton trick — MutationReplayer's init
|
||||
* block subscribes to AuthStore.sessionCookie and drains the
|
||||
* offline write queue on every signed-in transition (sign-in OR
|
||||
* cold start with persisted cookie). Without this @Inject the
|
||||
* queue would only enqueue, never replay.
|
||||
*/
|
||||
@Suppress("unused") @Inject lateinit var mutationReplayer: MutationReplayer
|
||||
|
||||
@Inject @ApplicationScope lateinit var appScope: CoroutineScope
|
||||
|
||||
override fun onCreate() {
|
||||
|
||||
+167
@@ -0,0 +1,167 @@
|
||||
package com.fabledsword.minstrel.cache.mutations
|
||||
|
||||
import com.fabledsword.minstrel.api.endpoints.DiscoverApi
|
||||
import com.fabledsword.minstrel.api.endpoints.LikesApi
|
||||
import com.fabledsword.minstrel.api.endpoints.QuarantineApi
|
||||
import com.fabledsword.minstrel.auth.AuthStore
|
||||
import com.fabledsword.minstrel.cache.db.dao.CachedMutationDao
|
||||
import com.fabledsword.minstrel.cache.db.entities.CachedMutationEntity
|
||||
import com.fabledsword.minstrel.di.ApplicationScope
|
||||
import com.fabledsword.minstrel.likes.data.LikesRepository
|
||||
import com.fabledsword.minstrel.models.wire.CreateRequestBody
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.flow.filterNotNull
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.datetime.Clock
|
||||
import kotlinx.serialization.json.Json
|
||||
import retrofit2.Retrofit
|
||||
import retrofit2.create
|
||||
import javax.inject.Inject
|
||||
import javax.inject.Singleton
|
||||
|
||||
/**
|
||||
* Drains the offline MutationQueue. Reads pending rows in FIFO order,
|
||||
* re-fires each call against the raw Retrofit API (NOT the high-level
|
||||
* Repository, which would re-enqueue on failure → infinite loop), and
|
||||
* deletes the row on 2xx success.
|
||||
*
|
||||
* Triggered by:
|
||||
* - app start (MinstrelApplication @Inject forces the singleton's
|
||||
* init block to subscribe)
|
||||
* - every time AuthStore.sessionCookie transitions to a non-null
|
||||
* value (sign-in, app launch with persisted cookie)
|
||||
*
|
||||
* Single in-flight via the `mutex` — back-to-back triggers collapse
|
||||
* into one drain pass. A proper connectivity-listener / WorkManager
|
||||
* job that re-tries on Wi-Fi-came-back is a follow-up; for MVP the
|
||||
* sign-in + app-open triggers cover the common offline → online
|
||||
* transition (operator unlocks phone, opens app).
|
||||
*
|
||||
* Failed rows stay in the queue with their `attempts` and
|
||||
* `lastAttemptAt` updated. They get another chance on the next
|
||||
* trigger; no exponential backoff or max-attempts yet — the rows are
|
||||
* small (low single-digit count for typical use) so retry-forever is
|
||||
* cheap.
|
||||
*/
|
||||
@Singleton
|
||||
class MutationReplayer @Inject constructor(
|
||||
private val dao: CachedMutationDao,
|
||||
private val authStore: AuthStore,
|
||||
private val json: Json,
|
||||
@ApplicationScope private val scope: CoroutineScope,
|
||||
retrofit: Retrofit,
|
||||
) {
|
||||
private val likesApi: LikesApi = retrofit.create()
|
||||
private val discoverApi: DiscoverApi = retrofit.create()
|
||||
private val quarantineApi: QuarantineApi = retrofit.create()
|
||||
|
||||
private val mutex = Mutex()
|
||||
|
||||
init {
|
||||
scope.launch {
|
||||
authStore.sessionCookie
|
||||
.filterNotNull()
|
||||
.distinctUntilChanged()
|
||||
.collect { drainSafe() }
|
||||
}
|
||||
}
|
||||
|
||||
/** Public entry-point; coalesces concurrent triggers via a Mutex. */
|
||||
suspend fun drainSafe() {
|
||||
if (!mutex.tryLock()) return
|
||||
try {
|
||||
drain()
|
||||
} finally {
|
||||
mutex.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun drain() {
|
||||
val rows = dao.getAll()
|
||||
for (row in rows) {
|
||||
val sent = runCatching { dispatch(row) }.getOrDefault(false)
|
||||
if (sent) {
|
||||
dao.delete(row.id)
|
||||
} else {
|
||||
dao.recordAttempt(row.id, Clock.System.now())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns true if the server accepted the call, false on any
|
||||
* failure (transport, server error, decode). The replayer treats
|
||||
* false as "leave the row for next pass" — it doesn't try to
|
||||
* distinguish permanent rejections from transient ones; a row
|
||||
* that consistently fails will accumulate attempt count and
|
||||
* lastAttemptAt for diagnostics.
|
||||
*/
|
||||
private suspend fun dispatch(row: CachedMutationEntity): Boolean = when (row.kind) {
|
||||
MutationKind.LIKE_TOGGLE -> dispatchLikeToggle(row.payload)
|
||||
MutationKind.REQUEST_CREATE -> dispatchRequestCreate(row.payload)
|
||||
MutationKind.QUARANTINE_UNFLAG -> dispatchQuarantineUnflag(row.payload)
|
||||
else -> {
|
||||
// Unknown kind — drop the row by claiming success so a
|
||||
// stale schema entry can't wedge the queue forever.
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun dispatchLikeToggle(payload: String): Boolean {
|
||||
val decoded = json.decodeFromString(LikeTogglePayload.serializer(), payload)
|
||||
val kindPath = when (decoded.entityType) {
|
||||
LikesRepository.ENTITY_ARTIST -> "artists"
|
||||
LikesRepository.ENTITY_ALBUM -> "albums"
|
||||
LikesRepository.ENTITY_TRACK -> "tracks"
|
||||
else -> return true // unknown variant — drop.
|
||||
}
|
||||
return try {
|
||||
if (decoded.desiredState) {
|
||||
likesApi.like(kindPath, decoded.entityId)
|
||||
} else {
|
||||
likesApi.unlike(kindPath, decoded.entityId)
|
||||
}
|
||||
true
|
||||
} catch (
|
||||
@Suppress("TooGenericExceptionCaught") e: Throwable,
|
||||
) {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun dispatchRequestCreate(payload: String): Boolean {
|
||||
val decoded = json.decodeFromString(RequestCreatePayload.serializer(), payload)
|
||||
val body = CreateRequestBody(
|
||||
kind = decoded.kind,
|
||||
artistMbid = decoded.artistMbid,
|
||||
artistName = decoded.artistName,
|
||||
albumMbid = decoded.albumMbid,
|
||||
albumTitle = decoded.albumTitle,
|
||||
trackMbid = decoded.trackMbid,
|
||||
trackTitle = decoded.trackTitle,
|
||||
)
|
||||
return try {
|
||||
discoverApi.createRequest(body)
|
||||
true
|
||||
} catch (
|
||||
@Suppress("TooGenericExceptionCaught") e: Throwable,
|
||||
) {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun dispatchQuarantineUnflag(payload: String): Boolean {
|
||||
val decoded = json.decodeFromString(QuarantineUnflagPayload.serializer(), payload)
|
||||
return try {
|
||||
quarantineApi.unflag(decoded.trackId)
|
||||
true
|
||||
} catch (
|
||||
@Suppress("TooGenericExceptionCaught") e: Throwable,
|
||||
) {
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user