Creepso/NuvioStreaming
1
0
Fork 0
mirror of https://github.com/tapframe/NuvioStreaming.git synced 2026-07-26 22:42:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(simkl): enforce compliant sync cadence

Browse source
This commit is contained in:
tapframe 2026-07-22 14:04:19 +05:30
parent b1a7f0757e
commit c1022e2eb5
11 changed files with 315 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
composeApp/src/commonMain/kotlin/com/nuvio/app/App.kt
View file

@ -1219,7 +1219,7 @@ private fun MainAppContent(
val activeProfileId = profileState.activeProfile?.profileIndex ?: return@LaunchedEffect
SyncManager.pullAllForProfile(activeProfileId)
AppForegroundMonitor.events().collect {
SyncManager.requestForegroundPull(activeProfileId, force = true)
SyncManager.requestForegroundPull(activeProfileId)
}
}
var resumePromptItem by remember { mutableStateOf<ContinueWatchingItem?>(null) }

5
composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt
View file

@ -346,6 +346,11 @@ object SyncManager {
}
}
synchronized(pullStateLock) {
lastFullPullAtMs = EpisodeReleaseDatePlatform.nowEpochMs()
lastFullPullProfileId = profileId
}
log.i { "Foreground sync completed profile=$profileId" }
}

27
composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryRepository.kt
View file

@ -13,6 +13,7 @@ import com.nuvio.app.features.tracking.TrackingLibraryProvider
import com.nuvio.app.features.tracking.TrackingLibraryTab
import com.nuvio.app.features.tracking.TrackingLibraryTabKind
import com.nuvio.app.features.tracking.TrackingProviderRegistry
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import com.nuvio.app.features.tracking.TrackingSettingsRepository
import com.nuvio.app.features.tracking.effectiveLibrarySourceMode as resolveEffectiveLibrarySourceMode
import com.nuvio.app.features.tracking.providerId
@ -97,7 +98,11 @@ object LibraryRepository {
TrackingProviderRegistry.connectedProviderIds.collectLatest {
TrackingProviderRegistry.connectedLibraryProviders().forEach(TrackingLibraryProvider::prepare)
activeLibraryProvider()?.let { provider ->
refreshLibraryProvider(provider, "authentication change")
refreshLibraryProvider(
provider = provider,
reason = "authentication change",
intent = TrackingRefreshIntent.INVALIDATED,
)
}
publish()
}
@ -206,14 +211,21 @@ object LibraryRepository {
) != null
}
suspend fun pullFromServer(profileId: Int) {
suspend fun pullFromServer(
profileId: Int,
refreshIntent: TrackingRefreshIntent = TrackingRefreshIntent.AUTOMATIC,
) {
val operationToken = activeOperationToken(profileId) ?: run {
log.d { "Skipping library pull for inactive profile $profileId" }
return
}
activeLibraryProvider()?.let { provider ->
refreshLibraryProvider(provider, "explicit pull")
refreshLibraryProvider(
provider = provider,
reason = "explicit pull",
intent = refreshIntent,
)
if (!isActiveOperation(operationToken)) return
publish()
return
@ -566,7 +578,11 @@ object LibraryRepository {
private fun refreshLibraryProviderAsync(provider: TrackingLibraryProvider) {
syncScope.launch {
refreshLibraryProvider(provider, "background refresh")
refreshLibraryProvider(
provider = provider,
reason = "background refresh",
intent = TrackingRefreshIntent.AUTOMATIC,
)
publish()
}
}
@ -574,9 +590,10 @@ object LibraryRepository {
private suspend fun refreshLibraryProvider(
provider: TrackingLibraryProvider,
reason: String,
intent: TrackingRefreshIntent,
) {
try {
provider.refresh()
provider.refresh(intent)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {

26
composeApp/src/commonMain/kotlin/com/nuvio/app/features/simkl/SimklApplicationAdapters.kt
View file

@ -13,6 +13,7 @@ import com.nuvio.app.features.tracking.TrackingListStatus
import com.nuvio.app.features.tracking.TrackingProviderId
import com.nuvio.app.features.tracking.TrackingProgressProvider
import com.nuvio.app.features.tracking.TrackingProgressSnapshot
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import com.nuvio.app.features.tracking.TrackingWatchedProvider
import com.nuvio.app.features.watched.WatchedItem
import com.nuvio.app.features.watchprogress.WatchProgressEntry
@ -52,8 +53,8 @@ object SimklLibraryRepository {
publish(SimklSyncRepository.state.value)
}
suspend fun refreshNow() {
SimklSyncRepository.refreshNow()
suspend fun refresh(intent: TrackingRefreshIntent) {
SimklSyncRepository.refresh(intent)
publish(SimklSyncRepository.state.value)
}
@ -93,7 +94,7 @@ object SimklLibraryRepository {
check(result.isComplete) {
"Simkl could not match ${result.notFoundCount} of ${result.attemptedCount} library items"
}
refreshNow()
refresh(TrackingRefreshIntent.INVALIDATED)
}
private fun publish(syncState: SimklSyncUiState) {
@ -114,7 +115,8 @@ object SimklTrackingLibraryProvider : TrackingLibraryProvider {
override fun onProfileChanged() = SimklLibraryRepository.ensureLoaded()
override suspend fun refresh() = SimklLibraryRepository.refreshNow()
override suspend fun refresh(intent: TrackingRefreshIntent) =
SimklLibraryRepository.refresh(intent)
override fun snapshot(): TrackingLibrarySnapshot {
val state = SimklLibraryRepository.uiState.value
@ -179,13 +181,13 @@ object SimklWatchedSyncAdapter : TrackingWatchedProvider {
override val providerId: TrackingProviderId = TrackingProviderId.SIMKL
override suspend fun pull(profileId: Int, pageSize: Int): List<WatchedItem> {
if (profileId != ProfileRepository.activeProfileId) return emptyList()
SimklSyncRepository.ensureFresh()
SimklSyncRepository.refresh(TrackingRefreshIntent.AUTOMATIC)
return SimklSyncRepository.state.value.snapshot.toSimklWatchedProjection().items
}
override suspend fun pullFullyWatchedSeriesKeys(profileId: Int): Set<String>? {
if (profileId != ProfileRepository.activeProfileId) return null
SimklSyncRepository.ensureFresh()
SimklSyncRepository.refresh(TrackingRefreshIntent.AUTOMATIC)
return SimklSyncRepository.state.value.snapshot.toSimklWatchedProjection().fullyWatchedSeriesKeys
}
@ -258,8 +260,8 @@ object SimklProgressRepository {
publish(SimklSyncRepository.state.value)
}
suspend fun refreshNow() {
SimklSyncRepository.refreshNow()
suspend fun refresh(intent: TrackingRefreshIntent) {
SimklSyncRepository.refresh(intent)
publish(SimklSyncRepository.state.value)
}
@ -290,7 +292,9 @@ object SimklProgressRepository {
}
}
SimklSyncRepository.commitPlaybackRemoval(removed)
if (removed.isNotEmpty()) SimklSyncRepository.refreshAsync()
if (removed.isNotEmpty()) {
SimklSyncRepository.refreshAsync(TrackingRefreshIntent.INVALIDATED)
}
}
private fun publish(syncState: SimklSyncUiState) {
@ -312,7 +316,9 @@ object SimklTrackingProgressProvider : TrackingProgressProvider {
override fun onProfileChanged() = SimklProgressRepository.ensureLoaded()
override suspend fun refresh(force: Boolean, sourceChanged: Boolean) =
SimklProgressRepository.refreshNow()
SimklProgressRepository.refresh(
if (sourceChanged) TrackingRefreshIntent.INVALIDATED else TrackingRefreshIntent.AUTOMATIC,
)
override fun snapshot(): TrackingProgressSnapshot {
val state = SimklProgressRepository.uiState.value

3
composeApp/src/commonMain/kotlin/com/nuvio/app/features/simkl/SimklAuthRepository.kt
View file

@ -7,6 +7,7 @@ import com.nuvio.app.features.tracking.TrackingCapability
import com.nuvio.app.features.tracking.TrackingProviderDescriptor
import com.nuvio.app.features.tracking.TrackingProviderId
import com.nuvio.app.features.tracking.TrackingProviderRegistry
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
@ -262,7 +263,7 @@ object SimklAuthRepository : TrackingAuthProvider {
persistMetadata()
publish(isLoading = false, error = null)
fetchAndStoreUserSettings()
SimklSyncRepository.refreshAsync()
SimklSyncRepository.refreshAsync(TrackingRefreshIntent.INVALIDATED)
}
private suspend fun fetchAndStoreUserSettings(): String? {

5
composeApp/src/commonMain/kotlin/com/nuvio/app/features/simkl/SimklMutationRepository.kt
View file

@ -12,6 +12,7 @@ import com.nuvio.app.features.tracking.TrackingMediaReference
import com.nuvio.app.features.tracking.TrackingMutationResult
import com.nuvio.app.features.tracking.TrackingProviderId
import com.nuvio.app.features.tracking.TrackingProviderRegistry
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import com.nuvio.app.features.tracking.TrackingScrobbleAction
import com.nuvio.app.features.tracking.TrackingScrobbleEvent
import com.nuvio.app.features.tracking.TrackingScrobbler
@ -118,7 +119,9 @@ object SimklMutationRepository : TrackingListWriter, TrackingHistoryWriter, Trac
private val service by lazy {
SimklMutationService(
client = SimklApi.client,
onMutationCommitted = SimklSyncRepository::refreshAsync,
onMutationCommitted = {
SimklSyncRepository.refreshAsync(TrackingRefreshIntent.INVALIDATED)
},
)
}

58
composeApp/src/commonMain/kotlin/com/nuvio/app/features/simkl/SimklRefreshPolicy.kt Normal file
View file

@ -0,0 +1,58 @@
package com.nuvio.app.features.simkl
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import kotlinx.atomicfu.atomic
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
internal const val SIMKL_AUTOMATIC_REFRESH_INTERVAL_MS = 15L * 60L * 1_000L
internal fun shouldRunSimklRefresh(
intent: TrackingRefreshIntent,
lastCheckedAtEpochMs: Long?,
nowEpochMs: Long,
hasError: Boolean,
automaticIntervalMs: Long = SIMKL_AUTOMATIC_REFRESH_INTERVAL_MS,
): Boolean {
if (intent != TrackingRefreshIntent.AUTOMATIC) return true
if (hasError || lastCheckedAtEpochMs == null) return true
val elapsedMs = nowEpochMs - lastCheckedAtEpochMs
return elapsedMs < 0L || elapsedMs >= automaticIntervalMs
}
/**
* Serializes provider refreshes and coalesces callers that overlap for the same profile generation.
*
* Library, watched, and progress are projections of one Simkl snapshot. They may request the same
* refresh concurrently, but only the first request should reach the network.
*/
internal class SimklRefreshGate {
private val mutex = Mutex()
private val completionSequence = atomic(0L)
private var lastCompletedProfileGeneration: Long? = null
suspend fun runIfNeeded(
profileGeneration: Long,
shouldRun: () -> Boolean,
block: suspend () -> Unit,
) {
val observedSequence = completionSequence.value
mutex.withLock {
if (
completionSequence.value != observedSequence &&
lastCompletedProfileGeneration == profileGeneration
) {
return
}
if (!shouldRun()) return
try {
block()
} finally {
lastCompletedProfileGeneration = profileGeneration
completionSequence.incrementAndGet()
}
}
}
}

92
composeApp/src/commonMain/kotlin/com/nuvio/app/features/simkl/SimklSyncRepository.kt
View file

@ -5,6 +5,7 @@ import com.nuvio.app.features.profiles.ProfileRepository
import com.nuvio.app.features.tracking.TrackingProfileStore
import com.nuvio.app.features.tracking.TrackingProviderId
import com.nuvio.app.features.tracking.TrackingProviderRegistry
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
@ -13,8 +14,6 @@ import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
@ -29,7 +28,7 @@ object SimklSyncRepository : TrackingProfileStore {
explicitNulls = false
}
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val refreshMutex = Mutex()
private val refreshGate = SimklRefreshGate()
private val engine = SimklSyncEngine(
remote = SimklApiSyncRemote(),
nowEpochMs = SimklPlatformClock::nowEpochMs,
@ -60,51 +59,60 @@ object SimklSyncRepository : TrackingProfileStore {
_state.value = SimklSyncUiState(snapshot = snapshot, hasLoaded = true)
}
fun refreshAsync() {
scope.launch { refreshNow() }
fun refreshAsync(intent: TrackingRefreshIntent) {
scope.launch { refresh(intent) }
}
suspend fun ensureFresh() {
suspend fun refresh(intent: TrackingRefreshIntent) {
ensureLoaded()
val lastChecked = state.value.snapshot.lastCheckedAtEpochMs
if (lastChecked != null && SimklPlatformClock.nowEpochMs() - lastChecked < CACHE_TTL_MS) return
refreshNow()
}
suspend fun refreshNow() {
ensureLoaded()
refreshMutex.withLock {
if (!SimklAuthRepository.isAuthenticated.value) return
val profileId = ProfileRepository.activeProfileId
val generation = profileGeneration
val previous = _state.value
_state.value = previous.copy(isLoading = true, errorMessage = null)
val result = try {
engine.synchronize(previous.snapshot)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.w { "Simkl sync failed: ${error.message}" }
if (generation == profileGeneration && profileId == ProfileRepository.activeProfileId) {
_state.value = previous.copy(
isLoading = false,
hasLoaded = true,
errorMessage = error.message ?: "Unable to sync Simkl",
val requestedGeneration = profileGeneration
refreshGate.runIfNeeded(
profileGeneration = requestedGeneration,
shouldRun = {
val current = _state.value
requestedGeneration == profileGeneration &&
SimklAuthRepository.isAuthenticated.value &&
shouldRunSimklRefresh(
intent = intent,
lastCheckedAtEpochMs = current.snapshot.lastCheckedAtEpochMs,
nowEpochMs = SimklPlatformClock.nowEpochMs(),
hasError = current.errorMessage != null,
)
}
return
}
if (generation != profileGeneration || profileId != ProfileRepository.activeProfileId) return
SimklSyncStorage.savePayload(json.encodeToString(result))
_state.value = SimklSyncUiState(
snapshot = result,
hasLoaded = true,
)
},
) {
refreshSnapshot(requestedGeneration)
}
}
private suspend fun refreshSnapshot(generation: Long) {
val profileId = ProfileRepository.activeProfileId
val previous = _state.value
_state.value = previous.copy(isLoading = true, errorMessage = null)
val result = try {
engine.synchronize(previous.snapshot)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.w { "Simkl sync failed: ${error.message}" }
if (generation == profileGeneration && profileId == ProfileRepository.activeProfileId) {
_state.value = previous.copy(
isLoading = false,
hasLoaded = true,
errorMessage = error.message ?: "Unable to sync Simkl",
)
}
return
}
if (generation != profileGeneration || profileId != ProfileRepository.activeProfileId) return
SimklSyncStorage.savePayload(json.encodeToString(result))
_state.value = SimklSyncUiState(
snapshot = result,
hasLoaded = true,
)
}
internal fun commitPlaybackRemoval(sessionIds: Set<Long>) {
if (sessionIds.isEmpty()) return
ensureLoaded()
@ -133,6 +141,4 @@ object SimklSyncRepository : TrackingProfileStore {
override fun removeStoredProfile(profileId: Int) {
SimklSyncStorage.removeProfile(profileId)
}
private const val CACHE_TTL_MS = 60_000L
}

14
composeApp/src/commonMain/kotlin/com/nuvio/app/features/tracking/TrackingReads.kt
View file

@ -28,6 +28,18 @@ data class TrackingLibrarySnapshot(
val errorMessage: String? = null,
)
/** Why a provider refresh was requested. Providers own the matching cache policy. */
enum class TrackingRefreshIntent {
/** Lifecycle, startup, reconnect, or other application-driven refresh. */
AUTOMATIC,
/** An explicit user action, such as Sync now or pull-to-refresh. */
USER_INITIATED,
/** A successful write or authentication change made the local snapshot stale. */
INVALIDATED,
}
/**
* Provider-owned projection of remote library state into application models.
*
@ -42,7 +54,7 @@ interface TrackingLibraryProvider {
fun prepare() = Unit
fun onProfileChanged() = Unit
fun clearLocalState() = Unit
suspend fun refresh()
suspend fun refresh(intent: TrackingRefreshIntent)
fun snapshot(): TrackingLibrarySnapshot
fun contains(contentId: String, contentType: String? = null): Boolean
fun find(contentId: String): LibraryItem?

8
composeApp/src/commonMain/kotlin/com/nuvio/app/features/trakt/TraktTrackingLibraryProvider.kt
View file

@ -7,6 +7,7 @@ import com.nuvio.app.features.tracking.TrackingLibrarySnapshot
import com.nuvio.app.features.tracking.TrackingLibraryTab
import com.nuvio.app.features.tracking.TrackingLibraryTabKind
import com.nuvio.app.features.tracking.TrackingProviderId
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
@ -22,7 +23,12 @@ object TraktTrackingLibraryProvider : TrackingLibraryProvider {
override fun clearLocalState() = TraktLibraryRepository.clearLocalState()
override suspend fun refresh() = TraktLibraryRepository.refreshNow()
override suspend fun refresh(intent: TrackingRefreshIntent) = when (intent) {
TrackingRefreshIntent.AUTOMATIC -> TraktLibraryRepository.ensureFresh()
TrackingRefreshIntent.USER_INITIATED,
TrackingRefreshIntent.INVALIDATED,
-> TraktLibraryRepository.refreshNow()
}
override fun snapshot(): TrackingLibrarySnapshot {
val state = TraktLibraryRepository.uiState.value

138
composeApp/src/commonTest/kotlin/com/nuvio/app/features/simkl/SimklRefreshPolicyTest.kt Normal file
View file

@ -0,0 +1,138 @@
package com.nuvio.app.features.simkl
import com.nuvio.app.features.tracking.TrackingRefreshIntent
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
class SimklRefreshPolicyTest {
@Test
fun `automatic refresh is throttled for fifteen minutes`() {
val lastCheckedAt = 1_000L
assertFalse(
shouldRunSimklRefresh(
intent = TrackingRefreshIntent.AUTOMATIC,
lastCheckedAtEpochMs = lastCheckedAt,
nowEpochMs = lastCheckedAt + SIMKL_AUTOMATIC_REFRESH_INTERVAL_MS - 1L,
hasError = false,
),
)
assertTrue(
shouldRunSimklRefresh(
intent = TrackingRefreshIntent.AUTOMATIC,
lastCheckedAtEpochMs = lastCheckedAt,
nowEpochMs = lastCheckedAt + SIMKL_AUTOMATIC_REFRESH_INTERVAL_MS,
hasError = false,
),
)
}
@Test
fun `missing stale or failed snapshots refresh automatically`() {
assertTrue(
shouldRunSimklRefresh(
intent = TrackingRefreshIntent.AUTOMATIC,
lastCheckedAtEpochMs = null,
nowEpochMs = 1_000L,
hasError = false,
),
)
assertTrue(
shouldRunSimklRefresh(
intent = TrackingRefreshIntent.AUTOMATIC,
lastCheckedAtEpochMs = 1_000L,
nowEpochMs = 1_001L,
hasError = true,
),
)
assertTrue(
shouldRunSimklRefresh(
intent = TrackingRefreshIntent.AUTOMATIC,
lastCheckedAtEpochMs = 2_000L,
nowEpochMs = 1_000L,
hasError = false,
),
)
}
@Test
fun `manual and invalidated refreshes bypass automatic freshness`() {
TrackingRefreshIntent.entries
.filterNot { intent -> intent == TrackingRefreshIntent.AUTOMATIC }
.forEach { intent ->
assertTrue(
shouldRunSimklRefresh(
intent = intent,
lastCheckedAtEpochMs = 1_000L,
nowEpochMs = 1_001L,
hasError = false,
),
)
}
}
@Test
fun `overlapping refreshes for one profile generation are coalesced`() = runBlocking {
val gate = SimklRefreshGate()
val firstEntered = CompletableDeferred<Unit>()
val releaseFirst = CompletableDeferred<Unit>()
val secondStarted = CompletableDeferred<Unit>()
var executions = 0
val first = async {
gate.runIfNeeded(profileGeneration = 7L, shouldRun = { true }) {
executions += 1
firstEntered.complete(Unit)
releaseFirst.await()
}
}
firstEntered.await()
val second = async {
secondStarted.complete(Unit)
gate.runIfNeeded(profileGeneration = 7L, shouldRun = { true }) {
executions += 1
}
}
secondStarted.await()
releaseFirst.complete(Unit)
first.await()
second.await()
assertEquals(1, executions)
}
@Test
fun `a new profile generation is not coalesced with an old profile refresh`() = runBlocking {
val gate = SimklRefreshGate()
val firstEntered = CompletableDeferred<Unit>()
val releaseFirst = CompletableDeferred<Unit>()
val secondStarted = CompletableDeferred<Unit>()
val executedGenerations = mutableListOf<Long>()
val first = async {
gate.runIfNeeded(profileGeneration = 1L, shouldRun = { true }) {
executedGenerations += 1L
firstEntered.complete(Unit)
releaseFirst.await()
}
}
firstEntered.await()
val second = async {
secondStarted.complete(Unit)
gate.runIfNeeded(profileGeneration = 2L, shouldRun = { true }) {
executedGenerations += 2L
}
}
secondStarted.await()
releaseFirst.complete(Unit)
first.await()
second.await()
assertEquals(listOf(1L, 2L), executedGenerations)
}
}