Creepso/NuvioStreaming
1
0
Fork 0
mirror of https://github.com/tapframe/NuvioStreaming.git synced 2026-08-03 18:16:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: sync watch progress source

Browse source
This commit is contained in:
tapframe 2026-07-10 01:21:55 +05:30
parent d085bf8f7d
commit 6d5281cdd2
27 changed files with 3047 additions and 636 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
composeApp/build.gradle.kts
View file

@ -316,6 +316,7 @@ kotlin {
}
minSdk = libs.versions.android.minSdk.get().toInt()
androidResources.enable = true
withHostTest {}
compilerOptions {
jvmTarget.set(JvmTarget.JVM_11)

2
composeApp/src/androidMain/kotlin/com/nuvio/app/core/storage/PlatformLocalAccountDataCleaner.android.kt
View file

@ -17,6 +17,7 @@ internal actual object PlatformLocalAccountDataCleaner {
"nuvio_debrid_settings",
"nuvio_mdblist_settings",
"nuvio_downloads",
"nuvio_auth",
"nuvio_trakt_auth",
"nuvio_trakt_library",
"nuvio_trakt_settings",
@ -24,6 +25,7 @@ internal actual object PlatformLocalAccountDataCleaner {
"nuvio_stream_link_cache",
"nuvio_stream_badge_settings",
"nuvio_continue_watching_preferences",
"nuvio_cw_enrichment",
"nuvio_episode_release_notifications",
"nuvio_episode_release_notifications_platform",
"nuvio_watch_progress",

18
composeApp/src/androidMain/kotlin/com/nuvio/app/features/trakt/TraktSettingsStorage.android.kt
View file

@ -7,6 +7,7 @@ import com.nuvio.app.core.storage.ProfileScopedKey
internal actual object TraktSettingsStorage {
private const val preferencesName = "nuvio_trakt_settings"
private const val payloadKey = "trakt_settings_payload"
private const val pendingWatchProgressSourceKey = "pending_watch_progress_source"
private var preferences: SharedPreferences? = null
@ -23,4 +24,21 @@ internal actual object TraktSettingsStorage {
?.putString(ProfileScopedKey.of(payloadKey), payload)
?.apply()
}
actual fun loadPendingWatchProgressSourcePayload(profileId: Int): String? =
preferences?.getString(ProfileScopedKey.of(pendingWatchProgressSourceKey, profileId), null)
actual fun savePendingWatchProgressSourcePayload(profileId: Int, payload: String) {
preferences
?.edit()
?.putString(ProfileScopedKey.of(pendingWatchProgressSourceKey, profileId), payload)
?.commit()
}
actual fun clearPendingWatchProgressSourcePayload(profileId: Int) {
preferences
?.edit()
?.remove(ProfileScopedKey.of(pendingWatchProgressSourceKey, profileId))
?.apply()
}
}

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

@ -209,6 +209,7 @@ import com.nuvio.app.features.watchprogress.ContinueWatchingPreferencesRepositor
import com.nuvio.app.features.watchprogress.ResumePromptRepository
import com.nuvio.app.features.watchprogress.WatchProgressPlaybackSession
import com.nuvio.app.features.watchprogress.WatchProgressRepository
import com.nuvio.app.features.watchprogress.WatchProgressSourceCoordinator
import com.nuvio.app.features.watchprogress.nextUpDismissKey
import com.nuvio.app.features.watchprogress.toContinueWatchingItem
import com.nuvio.app.features.watching.application.WatchingActions
@ -744,6 +745,7 @@ private fun MainAppContent(
var offlineLaunchRouteHandled by rememberSaveable { mutableStateOf(false) }
var networkToastBaselineReady by rememberSaveable { mutableStateOf(false) }
var lastNetworkToastCondition by rememberSaveable { mutableStateOf(NetworkCondition.Unknown.name) }
var watchSourceReconnectPending by remember { mutableStateOf(false) }
fun handleRootTabClick(tab: AppScreenTab) {
if (selectedTab != tab) {
@ -879,6 +881,42 @@ private fun MainAppContent(
lastNetworkToastCondition = condition.name
}
LaunchedEffect(
networkStatusUiState.condition,
(authState as? AuthState.Authenticated)?.userId,
profileState.activeProfile?.profileIndex,
) {
when (networkStatusUiState.condition) {
NetworkCondition.NoInternet,
NetworkCondition.ServersUnreachable,
-> watchSourceReconnectPending = true
NetworkCondition.Online -> {
if (!watchSourceReconnectPending) return@LaunchedEffect
val profileId = profileState.activeProfile?.profileIndex
?: ProfileRepository.activeProfileId
val authenticatedState = authState as? AuthState.Authenticated
if (authenticatedState != null && !authenticatedState.isAnonymous) {
SyncManager.requestForegroundPull(profileId = profileId, force = true)
watchSourceReconnectPending = false
} else {
val result = WatchProgressSourceCoordinator.refreshActiveSource(
profileId = profileId,
force = true,
)
if (result.succeeded) {
watchSourceReconnectPending = false
}
}
}
NetworkCondition.Unknown,
NetworkCondition.Checking,
-> Unit
}
}
LaunchedEffect(
initialHomeReady,
offlineLaunchRouteHandled,
@ -962,6 +1000,7 @@ private fun MainAppContent(
if (authenticatedState.isAnonymous) return@LaunchedEffect
val activeProfileId = profileState.activeProfile?.profileIndex ?: return@LaunchedEffect
SyncManager.pullAllForProfile(activeProfileId)
AppForegroundMonitor.events().collect {
SyncManager.requestForegroundPull(activeProfileId, force = true)
}

56
composeApp/src/commonMain/kotlin/com/nuvio/app/core/auth/AuthRepository.kt
View file

@ -10,6 +10,7 @@ import io.github.jan.supabase.exceptions.RestException
import io.github.jan.supabase.functions.functions
import kotlin.uuid.ExperimentalUuidApi
import kotlin.uuid.Uuid
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
@ -130,19 +131,44 @@ object AuthRepository {
_error.value = e.message ?: getString(Res.string.auth_sign_in_failed)
}
suspend fun signOut(): Result<Unit> = runCatching {
suspend fun signOut(): Result<Unit> {
_error.value = null
val wasAnonymous = AuthStorage.loadAnonymousUserId() != null
AuthStorage.clearAnonymousUserId()
val anonymousRead = runCatching { AuthStorage.loadAnonymousUserId() }
val wasAnonymous = anonymousRead.getOrNull() != null
val anonymousClear = runCatching { AuthStorage.clearAnonymousUserId() }
validatedRemoteUserId = null
if (!wasAnonymous) {
SupabaseProvider.client.auth.signOut()
val remoteSignOut = if (wasAnonymous) {
Result.success(Unit)
} else {
runCatching { SupabaseProvider.client.auth.signOut() }
}
val fallbackSessionClear = if (remoteSignOut.isFailure) {
runCatching { SupabaseProvider.client.auth.clearSession() }
.onFailure { error -> log.w(error) { "Failed to clear Supabase session after sign-out failure" } }
} else {
Result.success(Unit)
}
val localCleanup = runCatching { LocalAccountDataCleaner.wipe() }
_state.value = AuthState.Unauthenticated
LocalAccountDataCleaner.wipe()
}.onFailure { e ->
log.e(e) { "Sign-out failed" }
_error.value = e.message ?: getString(Res.string.auth_sign_out_failed)
val failure = anonymousRead.exceptionOrNull()
?: anonymousClear.exceptionOrNull()
?: remoteSignOut.exceptionOrNull()
?: fallbackSessionClear.exceptionOrNull()
?: localCleanup.exceptionOrNull()
val cancellation = remoteSignOut.exceptionOrNull() as? CancellationException
?: fallbackSessionClear.exceptionOrNull() as? CancellationException
if (cancellation != null) throw cancellation
return if (failure == null) {
Result.success(Unit)
} else {
log.e(failure) { "Sign-out did not complete cleanly; all local cleanup steps were attempted" }
_error.value = failure.message ?: runCatching {
getString(Res.string.auth_sign_out_failed)
}.getOrDefault("Sign out failed")
Result.failure(failure)
}
}
suspend fun signOutIfSessionInvalid(error: Throwable, source: String): Boolean {
@ -162,8 +188,11 @@ object AuthRepository {
}.onFailure { e ->
log.w(e) { "Failed to clear Supabase session after remote invalidation; continuing local reset" }
}
val localCleanup = runCatching { LocalAccountDataCleaner.wipe() }
_state.value = AuthState.Unauthenticated
LocalAccountDataCleaner.wipe()
localCleanup.onFailure { error ->
log.e(error) { "Local account cleanup failed after remote session invalidation" }
}
}
suspend fun deleteAccount(): Result<Unit> = runCatching {
@ -171,8 +200,11 @@ object AuthRepository {
SupabaseProvider.client.functions.invoke("delete-account")
SupabaseProvider.client.auth.signOut()
validatedRemoteUserId = null
_state.value = AuthState.Unauthenticated
LocalAccountDataCleaner.wipe()
try {
LocalAccountDataCleaner.wipe()
} finally {
_state.value = AuthState.Unauthenticated
}
}.onFailure { e ->
log.e(e) { "Account deletion failed" }
_error.value = e.message ?: getString(Res.string.auth_account_deletion_failed)

12
composeApp/src/commonMain/kotlin/com/nuvio/app/core/storage/LocalAccountDataCleaner.kt
View file

@ -1,6 +1,8 @@
package com.nuvio.app.core.storage
import com.nuvio.app.core.build.AppFeaturePolicy
import com.nuvio.app.core.sync.SyncManager
import com.nuvio.app.core.sync.ProfileSettingsSync
import com.nuvio.app.features.addons.AddonRepository
import com.nuvio.app.features.catalog.CatalogRepository
import com.nuvio.app.features.collection.CollectionMobileSettingsRepository
@ -27,11 +29,19 @@ import com.nuvio.app.features.trakt.TraktAuthRepository
import com.nuvio.app.features.trakt.TraktSettingsRepository
import com.nuvio.app.core.ui.PosterCardStyleRepository
import com.nuvio.app.features.watchprogress.ContinueWatchingPreferencesRepository
import com.nuvio.app.features.watchprogress.ContinueWatchingEnrichmentCache
import com.nuvio.app.features.watchprogress.WatchProgressRepository
import com.nuvio.app.features.watchprogress.WatchProgressSourceCoordinator
import com.nuvio.app.features.watched.WatchedRepository
internal object LocalAccountDataCleaner {
fun wipe() {
SyncManager.cancelAccountSync()
WatchProgressSourceCoordinator.clearLocalState()
ProfileSettingsSync.clearAccountState()
ContinueWatchingEnrichmentCache.clearLocalState()
WatchProgressRepository.clearLocalState()
WatchedRepository.clearLocalState()
PlatformLocalAccountDataCleaner.wipe()
ProfileRepository.clearInMemory()
@ -43,8 +53,6 @@ internal object LocalAccountDataCleaner {
HomeCatalogSettingsRepository.clearLocalState()
MetaScreenSettingsRepository.clearLocalState()
LibraryRepository.clearLocalState()
WatchProgressRepository.clearLocalState()
WatchedRepository.clearLocalState()
ContinueWatchingPreferencesRepository.clearLocalState()
EpisodeReleaseNotificationsRepository.clearLocalState()
CollectionMobileSettingsRepository.clearLocalState()

320
composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/ProfileSettingsSync.kt
View file

@ -27,6 +27,7 @@ import com.nuvio.app.features.tmdb.TmdbSettingsStorage
import com.nuvio.app.features.tmdb.TmdbSettingsRepository
import com.nuvio.app.features.trakt.TraktCommentsStorage
import com.nuvio.app.features.trakt.TraktCommentsSettings
import com.nuvio.app.features.trakt.ProfileSettingsWatchSourceOutbox
import com.nuvio.app.features.trakt.TraktSettingsStorage
import com.nuvio.app.features.trakt.TraktSettingsRepository
import com.nuvio.app.features.watchprogress.ContinueWatchingPreferencesStorage
@ -34,6 +35,9 @@ import com.nuvio.app.features.watchprogress.ContinueWatchingPreferencesRepositor
import io.github.jan.supabase.postgrest.postgrest
import io.github.jan.supabase.postgrest.rpc
import kotlin.concurrent.Volatile
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.FlowPreview
@ -45,6 +49,7 @@ import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.drop
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
@ -60,10 +65,22 @@ import kotlinx.serialization.json.put
private const val PUSH_DEBOUNCE_MS = 1500L
private data class ObservedProfileSettingsChange(
val signature: String,
val accountId: String?,
)
private data class SkippedProfileSettingsPush(
val signature: String,
val accountId: String?,
val profileId: Int,
)
object ProfileSettingsSync {
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val log = Logger.withTag("ProfileSettingsSync")
private val syncMutex = Mutex()
private val observeLock = SynchronizedObject()
private val json = Json {
ignoreUnknownKeys = true
encodeDefaults = true
@ -76,27 +93,62 @@ object ProfileSettingsSync {
private var isServerSyncInFlight: Boolean = false
@Volatile
private var skipNextPushSignature: String? = null
private var skipNextPush: SkippedProfileSettingsPush? = null
@Volatile
private var pushEnabledAccountId: String? = null
@Volatile
private var pendingLocalPush: SkippedProfileSettingsPush? = null
private var observeJob: Job? = null
private var pendingPushRetryJob: Job? = null
fun startObserving() {
if (observeJob?.isActive == true) return
fun startObserving() = synchronized(observeLock) {
if (observeJob?.isActive == true) return@synchronized
ensureRepositoriesLoaded()
observeLocalChangesAndPush()
}
fun clearAccountState() {
synchronized(observeLock) {
observeJob?.cancel()
observeJob = null
}
skipNextPush = null
pushEnabledAccountId = null
pendingLocalPush = null
pendingPushRetryJob?.cancel()
pendingPushRetryJob = null
}
suspend fun pull(profileId: Int): Boolean {
ensureRepositoriesLoaded()
startObserving()
val accountId = currentCloudAccountId() ?: return false
return syncMutex.withLock {
if (ProfileRepository.activeProfileId != profileId) {
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) {
log.d { "pull(profileId=$profileId) — skipped because profile is no longer active" }
return@withLock false
}
isServerSyncInFlight = true
try {
pushEnabledAccountId = accountId
hydrateDurableWatchSourcePush(profileId = profileId, accountId = accountId)
if (
!pushCurrentStateLocked(
profileId = profileId,
accountId = accountId,
forceCurrentState = false,
)
) {
schedulePendingPushRetry()
return@withLock false
}
val observedSignatureAtStart = currentObservedStateSignature()
val localBlob = exportSettingsBlob()
if (ProfileRepository.activeProfileId != profileId) return@withLock false
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) {
throw CancellationException("Profile settings pull target changed")
}
val localSignature = buildSignature(localBlob)
val params = buildJsonObject {
@ -104,61 +156,128 @@ object ProfileSettingsSync {
put("p_platform", MOBILE_SYNC_PLATFORM)
}
val result = SupabaseProvider.client.postgrest.rpc("sync_pull_profile_settings_blob", params)
if (ProfileRepository.activeProfileId != profileId) return@withLock false
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) {
throw CancellationException("Profile settings pull target changed")
}
val response = result.decodeList<SettingsBlobResponse>().firstOrNull()
val remoteJson = response?.settingsJson
if (remoteJson == null) {
log.i { "pull(profileId=$profileId) — no remote settings blob found" }
if (localSignature != defaultSignature()) {
pushToRemoteLocked(profileId, localBlob)
val pendingDuringPull = pendingLocalPush?.let { pending ->
pending.accountId == accountId && pending.profileId == profileId
} == true
val durableSourceChangeDuringPull =
ProfileSettingsWatchSourceOutbox.pendingFor(accountId, profileId) != null
if (
pendingDuringPull ||
durableSourceChangeDuringPull ||
currentObservedStateSignature() != observedSignatureAtStart
) {
if (
!pushCurrentStateLocked(
profileId = profileId,
accountId = accountId,
forceCurrentState = true,
)
) {
schedulePendingPushRetry()
}
return@withLock false
}
if (remoteJson == null) {
log.i { "pull(profileId=$profileId) — no remote settings blob found" }
if (localSignature != defaultSignature()) {
pushToRemoteLocked(profileId, localBlob, accountId)
}
pushEnabledAccountId = accountId
return@withLock false
}
val remoteBlob = try {
json.decodeFromJsonElement(MobileProfileSettingsBlob.serializer(), remoteJson)
} catch (error: Throwable) {
log.e(error) { "pull(profileId=$profileId) — failed to decode remote settings blob" }
throw error
}
var restoredPendingSourceAfterRemoteApply = false
isApplyingRemoteBlob = true
try {
val remoteBlob = runCatching {
json.decodeFromJsonElement(MobileProfileSettingsBlob.serializer(), remoteJson)
}.getOrElse { error ->
log.e(error) { "pull(profileId=$profileId) — failed to decode remote settings blob" }
return@withLock false
}
val remoteSignature = buildSignature(remoteBlob)
if (remoteSignature == localSignature) {
log.d { "pull(profileId=$profileId) — remote matches local" }
pushEnabledAccountId = accountId
return@withLock false
}
if (ProfileRepository.activeProfileId != profileId) return@withLock false
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) {
throw CancellationException("Profile settings pull target changed")
}
applyRemoteBlob(remoteBlob)
skipNextPushSignature = currentObservedStateSignature()
ProfileSettingsWatchSourceOutbox.pendingFor(accountId, profileId)?.let { pendingSource ->
if (TraktSettingsRepository.uiState.value.watchProgressSource != pendingSource.source) {
TraktSettingsRepository.setWatchProgressSource(pendingSource.source, profileId)
}
restoredPendingSourceAfterRemoteApply = true
}
skipNextPush = SkippedProfileSettingsPush(
signature = currentObservedStateSignature(),
accountId = currentCloudAccountId(),
profileId = profileId,
)
} finally {
isApplyingRemoteBlob = false
}
if (restoredPendingSourceAfterRemoteApply) {
pendingLocalPush = SkippedProfileSettingsPush(
signature = currentObservedStateSignature(),
accountId = accountId,
profileId = profileId,
)
if (
!pushCurrentStateLocked(
profileId = profileId,
accountId = accountId,
forceCurrentState = true,
)
) {
schedulePendingPushRetry()
}
return@withLock false
}
log.i { "pull(profileId=$profileId) — applied remote settings blob" }
pushEnabledAccountId = accountId
true
} catch (error: CancellationException) {
throw error
} catch (error: Exception) {
log.e(error) { "pull(profileId=$profileId) — FAILED" }
false
throw error
} finally {
isServerSyncInFlight = false
}
}
}
suspend fun pushCurrentProfileToRemote() {
suspend fun pushCurrentProfileToRemote(): Boolean {
ensureRepositoriesLoaded()
syncMutex.withLock {
runCatching {
val accountId = currentCloudAccountId() ?: return false
return syncMutex.withLock {
try {
val profileId = ProfileRepository.activeProfileId
val blob = exportSettingsBlob()
if (ProfileRepository.activeProfileId != profileId) return@runCatching
pushToRemoteLocked(profileId, blob)
}.onFailure { error ->
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) return@withLock false
pushCurrentStateLocked(
profileId = profileId,
accountId = accountId,
forceCurrentState = true,
)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.e(error) { "pushCurrentProfileToRemote() — FAILED" }
false
}
}
}
@ -184,24 +303,84 @@ object ProfileSettingsSync {
)
observeJob = scope.launch {
combine(signatureFlows) { currentObservedStateSignature() }
combine(signatureFlows) {
ObservedProfileSettingsChange(
signature = currentObservedStateSignature(),
accountId = currentCloudAccountId(),
)
}
.drop(1)
.distinctUntilChanged()
.onEach { change ->
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return@onEach
if (change.accountId == null || change.accountId != authState.userId) return@onEach
val observedChange = SkippedProfileSettingsPush(
signature = change.signature,
accountId = change.accountId,
profileId = ProfileRepository.activeProfileId,
)
if (skipNextPush != observedChange) {
pendingLocalPush = observedChange
schedulePendingPushRetry()
}
}
.debounce(PUSH_DEBOUNCE_MS)
.collect { signature ->
.collect { change ->
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return@collect
if (isApplyingRemoteBlob || isServerSyncInFlight) return@collect
if (signature == skipNextPushSignature) {
skipNextPushSignature = null
if (change.accountId == null || change.accountId != authState.userId) return@collect
val profileId = ProfileRepository.activeProfileId
val observedChange = SkippedProfileSettingsPush(
signature = change.signature,
accountId = change.accountId,
profileId = profileId,
)
if (skipNextPush == observedChange) {
skipNextPush = null
if (pendingLocalPush == observedChange) {
pendingLocalPush = null
}
return@collect
}
pendingLocalPush = observedChange
if (pushEnabledAccountId != change.accountId) return@collect
if (isApplyingRemoteBlob || isServerSyncInFlight) return@collect
pushCurrentProfileToRemote()
}
}
}
private suspend fun pushToRemoteLocked(profileId: Int, blob: MobileProfileSettingsBlob) {
private fun schedulePendingPushRetry() {
if (pendingPushRetryJob?.isActive == true) return
pendingPushRetryJob = scope.launch {
var retryDelayMs = 5_000L
while (pendingLocalPush != null) {
delay(retryDelayMs)
val pending = pendingLocalPush ?: break
if (
pending.accountId == currentCloudAccountId() &&
pending.profileId == ProfileRepository.activeProfileId &&
pushEnabledAccountId == pending.accountId &&
!isApplyingRemoteBlob &&
!isServerSyncInFlight &&
pushCurrentProfileToRemote()
) {
break
}
retryDelayMs = (retryDelayMs * 2L).coerceAtMost(60_000L)
}
}
}
private suspend fun pushToRemoteLocked(
profileId: Int,
blob: MobileProfileSettingsBlob,
accountId: String,
) {
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) {
throw CancellationException("Profile settings push target changed")
}
val params = buildJsonObject {
put("p_profile_id", profileId)
put("p_platform", MOBILE_SYNC_PLATFORM)
@ -209,9 +388,74 @@ object ProfileSettingsSync {
putSyncOriginClientId()
}
SupabaseProvider.client.postgrest.rpc("sync_push_profile_settings_blob", params)
if (!isCurrentSyncTarget(profileId = profileId, accountId = accountId)) {
throw CancellationException("Profile settings push target changed")
}
log.d { "pushToRemoteLocked(profileId=$profileId) — success" }
}
private fun hydrateDurableWatchSourcePush(profileId: Int, accountId: String) {
val durableChange = ProfileSettingsWatchSourceOutbox.pendingFor(accountId, profileId) ?: return
if (TraktSettingsRepository.uiState.value.watchProgressSource != durableChange.source) {
TraktSettingsRepository.setWatchProgressSource(durableChange.source, profileId)
}
pendingLocalPush = SkippedProfileSettingsPush(
signature = currentObservedStateSignature(),
accountId = accountId,
profileId = profileId,
)
schedulePendingPushRetry()
}
private suspend fun pushCurrentStateLocked(
profileId: Int,
accountId: String,
forceCurrentState: Boolean,
): Boolean {
val durableChange = ProfileSettingsWatchSourceOutbox.pendingFor(accountId, profileId)
val inMemoryChange = pendingLocalPush?.takeIf { pending ->
pending.accountId == accountId && pending.profileId == profileId
}
if (!forceCurrentState && durableChange == null && inMemoryChange == null) return true
val signature = currentObservedStateSignature()
pushToRemoteLocked(profileId, exportSettingsBlob(), accountId)
if (currentObservedStateSignature() != signature) {
pendingLocalPush = SkippedProfileSettingsPush(
signature = currentObservedStateSignature(),
accountId = accountId,
profileId = profileId,
)
schedulePendingPushRetry()
return false
}
val pushedChange = SkippedProfileSettingsPush(
signature = signature,
accountId = accountId,
profileId = profileId,
)
if (pendingLocalPush == pushedChange) {
pendingLocalPush = null
}
if (
durableChange != null &&
TraktSettingsRepository.uiState.value.watchProgressSource == durableChange.source
) {
ProfileSettingsWatchSourceOutbox.clearIfMatches(durableChange)
}
val durablePushRemains = ProfileSettingsWatchSourceOutbox.pendingFor(accountId, profileId) != null
val memoryPushRemains = pendingLocalPush?.let { pending ->
pending.accountId == accountId && pending.profileId == profileId
} == true
if (durablePushRemains || memoryPushRemains) {
schedulePendingPushRetry()
return false
}
return true
}
private fun exportSettingsBlob(): MobileProfileSettingsBlob {
ensureRepositoriesLoaded()
return MobileProfileSettingsBlob(
@ -315,6 +559,14 @@ object ProfileSettingsSync {
"trakt_comments=${TraktCommentsSettings.enabled.value}",
"episode_release_alerts=${EpisodeReleaseNotificationsRepository.uiState.value.isEnabled}",
).joinToString(separator = "||")
private fun currentCloudAccountId(): String? =
(AuthRepository.state.value as? AuthState.Authenticated)
?.takeUnless { it.isAnonymous }
?.userId
private fun isCurrentSyncTarget(profileId: Int, accountId: String): Boolean =
ProfileRepository.activeProfileId == profileId && currentCloudAccountId() == accountId
}
@Serializable

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

@ -17,12 +17,16 @@ import com.nuvio.app.features.trakt.TraktPlatformClock
import com.nuvio.app.features.trakt.TraktSettingsRepository
import com.nuvio.app.features.trakt.effectiveLibrarySourceMode
import com.nuvio.app.features.trakt.shouldUseTraktProgress
import com.nuvio.app.features.watched.WatchedRepository
import com.nuvio.app.features.watchprogress.WatchProgressRepository
import com.nuvio.app.features.watchprogress.WatchProgressSourceCoordinator
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
@ -31,86 +35,350 @@ private const val FOREGROUND_PULL_DELAY_MS = 2500L
private const val FOREGROUND_PULL_MIN_INTERVAL_MS = 30 * 60_000L
private const val PERIODIC_NUVIO_SYNC_PULL_INTERVAL_MS = 60_000L
internal enum class ProfileSyncStep {
Addons,
Plugins,
ProfileSettings,
TraktCredentials,
Library,
ActiveWatchSource,
Collections,
HomeCatalogSettings,
}
internal data class ProfileSyncOperations(
val pullAddons: suspend (Int) -> Unit,
val pullPlugins: suspend (Int) -> Unit,
val pullProfileSettings: suspend (Int) -> Unit,
val pullTraktCredentials: suspend (Int) -> Unit,
val pullLibrary: suspend (Int) -> Unit,
val refreshActiveWatchSource: suspend (Int) -> Unit,
val pullCollections: suspend (Int) -> Unit,
val pullHomeCatalogSettings: suspend (Int) -> Unit,
)
internal data class ProfileSyncResult(
val failedSteps: Set<ProfileSyncStep>,
) {
val succeeded: Boolean
get() = failedSteps.isEmpty()
}
internal suspend fun runOrderedProfileSync(
profileId: Int,
pluginsEnabled: Boolean,
operations: ProfileSyncOperations,
onFailure: (ProfileSyncStep, Throwable) -> Unit = { _, _ -> },
): ProfileSyncResult {
val failureLock = SynchronizedObject()
val failedSteps = mutableSetOf<ProfileSyncStep>()
suspend fun runStep(
step: ProfileSyncStep,
operation: suspend (Int) -> Unit,
) {
try {
operation(profileId)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
synchronized(failureLock) {
failedSteps += step
}
onFailure(step, error)
}
}
runStep(ProfileSyncStep.Addons, operations.pullAddons)
if (pluginsEnabled) {
runStep(ProfileSyncStep.Plugins, operations.pullPlugins)
}
coroutineScope {
val settingsJob = launch {
runStep(ProfileSyncStep.ProfileSettings, operations.pullProfileSettings)
}
val credentialsJob = launch {
runStep(ProfileSyncStep.TraktCredentials, operations.pullTraktCredentials)
}
settingsJob.join()
credentialsJob.join()
}
coroutineScope {
launch {
runStep(ProfileSyncStep.Library, operations.pullLibrary)
}
launch {
runStep(ProfileSyncStep.ActiveWatchSource, operations.refreshActiveWatchSource)
}
launch {
runStep(ProfileSyncStep.Collections, operations.pullCollections)
}
launch {
runStep(ProfileSyncStep.HomeCatalogSettings, operations.pullHomeCatalogSettings)
}
}
return ProfileSyncResult(
failedSteps = synchronized(failureLock) { failedSteps.toSet() },
)
}
internal enum class ProfileSyncRequestResult {
Started,
Coalesced,
Replaced,
}
internal fun shouldQueueCoalescedForegroundPull(force: Boolean): Boolean = force
internal class ProfileSyncRequestGate {
private data class PendingRequest(
val scope: CoroutineScope,
val profileId: Int,
val block: suspend () -> Unit,
)
private val lock = SynchronizedObject()
private var activeProfileId: Int? = null
private var activeJob: Job? = null
private var pendingRequest: PendingRequest? = null
fun launch(
scope: CoroutineScope,
profileId: Int,
queueIfCoalesced: Boolean = false,
block: suspend () -> Unit,
): ProfileSyncRequestResult {
lateinit var newJob: Job
var previousJob: Job? = null
val result = synchronized(lock) {
val active = activeJob?.takeUnless(Job::isCompleted)
if (active != null && activeProfileId == profileId) {
if (queueIfCoalesced) {
pendingRequest = PendingRequest(scope = scope, profileId = profileId, block = block)
}
return ProfileSyncRequestResult.Coalesced
}
previousJob = active
pendingRequest = null
val requestResult = if (active == null) {
ProfileSyncRequestResult.Started
} else {
ProfileSyncRequestResult.Replaced
}
newJob = scope.launch(start = CoroutineStart.LAZY) {
block()
}
activeProfileId = profileId
activeJob = newJob
newJob.invokeOnCompletion {
var pending: PendingRequest? = null
synchronized(lock) {
if (activeJob === newJob) {
activeJob = null
activeProfileId = null
pending = pendingRequest
pendingRequest = null
}
}
pending?.let { request ->
launch(
scope = request.scope,
profileId = request.profileId,
queueIfCoalesced = false,
block = request.block,
)
}
}
requestResult
}
previousJob?.cancel()
newJob.start()
return result
}
fun cancel() {
val job = synchronized(lock) {
activeJob.also {
activeJob = null
activeProfileId = null
pendingRequest = null
}
}
job?.cancel()
}
}
object SyncManager {
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val log = Logger.withTag("SyncManager")
private val fullSyncRequestGate = ProfileSyncRequestGate()
private val accountScopeLock = SynchronizedObject()
private var accountScopeJob: Job = SupervisorJob()
private var accountScope = CoroutineScope(accountScopeJob + Dispatchers.Default)
private val pullStateLock = SynchronizedObject()
private var foregroundPullJob: Job? = null
private var foregroundPullProfileId: Int? = null
private var periodicNuvioSyncPullJob: Job? = null
private var periodicNuvioSyncProfileId: Int? = null
private var lastForegroundPullAtMs: Long = 0L
private var lastFullPullAtMs: Long = 0L
private var lastFullPullProfileId: Int? = null
private val profileSyncOperations = ProfileSyncOperations(
pullAddons = { profileId -> AddonRepository.pullFromServer(profileId) },
pullPlugins = { profileId -> PluginRepository.pullFromServer(profileId) },
pullProfileSettings = { profileId -> ProfileSettingsSync.pull(profileId) },
pullTraktCredentials = { profileId -> TraktCredentialSync.pullFromRemoteOrThrow(profileId) },
pullLibrary = { profileId -> LibraryRepository.pullFromServer(profileId) },
refreshActiveWatchSource = { profileId ->
val result = WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = true)
check(result.succeeded) {
"Active watch source refresh was incomplete: " +
"progress=${result.progressRefreshed} watched=${result.watchedHistoryRefreshed}"
}
},
pullCollections = { profileId -> CollectionSyncService.pullFromServer(profileId) },
pullHomeCatalogSettings = { profileId -> HomeCatalogSettingsSyncService.pullFromServer(profileId) },
)
fun pullAllForProfile(profileId: Int) {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated) return
if (authState.isAnonymous) return
startFullProfilePull(profileId = profileId, reason = "requested")
}
scope.launch {
log.i { "pullAllForProfile($profileId) — auth=${authState.isAnonymous}" }
log.i { "pullAllForProfile — pulling addons first (await)..." }
runCatching { AddonRepository.pullFromServer(profileId) }
.onSuccess { log.i { "pullAllForProfile — addons pull completed" } }
.onFailure { log.e(it) { "Addon pull failed" } }
if (AppFeaturePolicy.pluginsEnabled) {
log.i { "pullAllForProfile — pulling plugins (await)..." }
runCatching { PluginRepository.pullFromServer(profileId) }
.onSuccess { log.i { "pullAllForProfile — plugins pull completed" } }
.onFailure { log.e(it) { "Plugin pull failed" } }
internal fun cancelAccountSync() {
fullSyncRequestGate.cancel()
val previousAccountJob = synchronized(accountScopeLock) {
accountScopeJob.also {
accountScopeJob = SupervisorJob()
accountScope = CoroutineScope(accountScopeJob + Dispatchers.Default)
}
runCatching { TraktCredentialSync.pullFromRemote(profileId) }
.onSuccess { applied -> log.i { "pullAllForProfile — Trakt credential pull completed applied=$applied" } }
.onFailure { log.e(it) { "Trakt credential pull failed" } }
log.i { "pullAllForProfile — launching remaining pulls in parallel" }
launch {
runCatching { LibraryRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Library pull failed" } }
}
launch {
runCatching { WatchProgressRepository.forceSnapshotRefreshFromServer(profileId) }
.onFailure { log.e(it) { "WatchProgress pull failed" } }
}
launch {
runCatching { WatchedRepository.forceSnapshotRefreshFromServer(profileId) }
.onFailure { log.e(it) { "Watched pull failed" } }
}
launch {
runCatching { ProfileSettingsSync.pull(profileId) }
.onFailure { log.e(it) { "ProfileSettings pull failed" } }
}
launch {
runCatching { CollectionSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Collections pull failed" } }
}
launch {
runCatching { HomeCatalogSettingsSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "HomeCatalogSettings pull failed" } }
}
log.i { "pullAllForProfile($profileId) — all pulls launched" }
}
previousAccountJob.cancel()
val foregroundJob = synchronized(pullStateLock) {
foregroundPullJob.also {
foregroundPullJob = null
foregroundPullProfileId = null
lastFullPullAtMs = 0L
lastFullPullProfileId = null
}
}
foregroundJob?.cancel()
stopPeriodicNuvioSyncPull()
}
private fun accountScopeSnapshot(): CoroutineScope = synchronized(accountScopeLock) {
accountScope
}
fun requestForegroundPull(profileId: Int, force: Boolean = false) {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return
val now = TraktPlatformClock.nowEpochMs()
if (!force && foregroundPullJob?.isActive == true) return
if (!force && now - lastForegroundPullAtMs < FOREGROUND_PULL_MIN_INTERVAL_MS) return
foregroundPullJob = scope.launch {
if (!force) {
delay(FOREGROUND_PULL_DELAY_MS)
if (!force && hasRecentFullPull(profileId)) {
return
}
lateinit var requestJob: Job
var previousJob: Job? = null
synchronized(pullStateLock) {
if (
!force &&
foregroundPullJob?.isCompleted == false &&
foregroundPullProfileId == profileId
) {
return
}
previousJob = foregroundPullJob
requestJob = accountScopeSnapshot().launch(start = CoroutineStart.LAZY) {
try {
if (!force) {
delay(FOREGROUND_PULL_DELAY_MS)
}
if (!force && hasRecentFullPull(profileId)) return@launch
startFullProfilePull(
profileId = profileId,
reason = "foreground",
queueIfCoalesced = shouldQueueCoalescedForegroundPull(force),
)
} finally {
synchronized(pullStateLock) {
if (foregroundPullJob === requestJob) {
foregroundPullJob = null
foregroundPullProfileId = null
}
}
}
}
foregroundPullProfileId = profileId
foregroundPullJob = requestJob
}
previousJob?.cancel()
requestJob.start()
}
private fun hasRecentFullPull(profileId: Int): Boolean =
synchronized(pullStateLock) {
lastFullPullProfileId == profileId &&
TraktPlatformClock.nowEpochMs() - lastFullPullAtMs < FOREGROUND_PULL_MIN_INTERVAL_MS
}
private fun startFullProfilePull(
profileId: Int,
reason: String,
queueIfCoalesced: Boolean = false,
) {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return
if (ProfileRepository.activeProfileId != profileId) return
val result = fullSyncRequestGate.launch(
scope = accountScopeSnapshot(),
profileId = profileId,
queueIfCoalesced = queueIfCoalesced,
) {
val currentAuthState = AuthRepository.state.value
if (currentAuthState !is AuthState.Authenticated || currentAuthState.isAnonymous) return@launch
if (ProfileRepository.activeProfileId != profileId) return@launch
lastForegroundPullAtMs = TraktPlatformClock.nowEpochMs()
pullForegroundForProfile(profileId)
log.i { "Full profile sync started profile=$profileId reason=$reason" }
WatchProgressSourceCoordinator.pauseAutomaticTransitions()
val syncResult = try {
runOrderedProfileSync(
profileId = profileId,
pluginsEnabled = AppFeaturePolicy.pluginsEnabled,
operations = profileSyncOperations,
onFailure = { step, error ->
log.e(error) { "Full profile sync step failed profile=$profileId step=$step" }
},
)
} finally {
WatchProgressSourceCoordinator.resumeAutomaticTransitions()
}
if (syncResult.succeeded) {
synchronized(pullStateLock) {
lastFullPullAtMs = TraktPlatformClock.nowEpochMs()
lastFullPullProfileId = profileId
}
} else {
log.w {
"Full profile sync incomplete profile=$profileId reason=$reason " +
"failedSteps=${syncResult.failedSteps}"
}
}
log.i { "Full profile sync completed profile=$profileId reason=$reason" }
}
when (result) {
ProfileSyncRequestResult.Started -> Unit
ProfileSyncRequestResult.Coalesced -> {
log.d { "Full profile sync coalesced profile=$profileId reason=$reason" }
}
ProfileSyncRequestResult.Replaced -> {
log.d { "Full profile sync replaced stale profile request with profile=$profileId reason=$reason" }
}
}
}
@ -124,7 +392,7 @@ object SyncManager {
stopPeriodicNuvioSyncPull()
periodicNuvioSyncProfileId = profileId
periodicNuvioSyncPullJob = scope.launch {
periodicNuvioSyncPullJob = accountScopeSnapshot().launch {
while (isActive) {
delay(PERIODIC_NUVIO_SYNC_PULL_INTERVAL_MS)
@ -163,8 +431,9 @@ object SyncManager {
.onFailure { log.e(it) { "Periodic Nuvio library pull failed" } }
}
if (shouldPullWatchProgress) {
runCatching { WatchProgressRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Periodic Nuvio watch progress pull failed" } }
runCatching {
WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = false)
}.onFailure { log.e(it) { "Periodic Nuvio watch source pull failed" } }
}
}
}
@ -180,7 +449,16 @@ object SyncManager {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return
scope.launch {
if (surface == "profile_settings") {
startFullProfilePull(
profileId = profileId,
reason = "realtime_profile_settings",
queueIfCoalesced = true,
)
return
}
accountScopeSnapshot().launch {
log.i { "requestRealtimeSurfacePull($profileId, $surface)" }
when (surface) {
"addons" -> {
@ -197,17 +475,10 @@ object SyncManager {
runCatching { LibraryRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Realtime library pull failed" } }
}
"watch_progress" -> {
runCatching { WatchProgressRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Realtime watch progress pull failed" } }
}
"watched_items" -> {
runCatching { WatchedRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Realtime watched items pull failed" } }
}
"profile_settings" -> {
runCatching { ProfileSettingsSync.pull(profileId) }
.onFailure { log.e(it) { "Realtime profile settings pull failed" } }
"watch_progress", "watched_items" -> {
runCatching {
WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = false)
}.onFailure { log.e(it) { "Realtime active watch source pull failed" } }
}
"collections" -> {
runCatching { CollectionSyncService.pullFromServer(profileId) }
@ -224,38 +495,4 @@ object SyncManager {
}
}
}
private fun pullForegroundForProfile(profileId: Int) {
scope.launch {
log.i { "pullForegroundForProfile($profileId) — syncing watch progress, watched items, library, collections, and home settings" }
runCatching { TraktCredentialSync.pullFromRemote(profileId) }
.onFailure { log.e(it) { "Foreground Trakt credential pull failed" } }
launch {
runCatching { LibraryRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground library pull failed" } }
}
launch {
runCatching { WatchProgressRepository.forceSnapshotRefreshFromServer(profileId) }
.onFailure { log.e(it) { "Foreground watch progress pull failed" } }
}
launch {
runCatching { WatchedRepository.forceSnapshotRefreshFromServer(profileId) }
.onFailure { log.e(it) { "Foreground watched items pull failed" } }
}
launch {
runCatching { CollectionSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground collections pull failed" } }
}
launch {
runCatching { HomeCatalogSettingsSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground home catalog settings pull failed" } }
}
}
}
}

78
composeApp/src/commonMain/kotlin/com/nuvio/app/features/home/HomeScreen.kt
View file

@ -13,6 +13,8 @@ import androidx.compose.runtime.setValue
import androidx.compose.ui.Modifier
import androidx.compose.ui.unit.dp
import androidx.lifecycle.compose.collectAsStateWithLifecycle
import com.nuvio.app.core.auth.AuthRepository
import com.nuvio.app.core.auth.AuthState
import com.nuvio.app.core.network.NetworkCondition
import com.nuvio.app.core.network.NetworkStatusRepository
import com.nuvio.app.core.ui.LocalNuvioBottomNavigationOverlayPadding
@ -39,11 +41,10 @@ import com.nuvio.app.features.home.components.HomeHeroSection
import com.nuvio.app.features.home.components.HomeSkeletonHero
import com.nuvio.app.features.home.components.HomeSkeletonRow
import com.nuvio.app.features.home.components.HomeContinueWatchingSectionBottomPadding
import com.nuvio.app.features.trakt.TraktAuthRepository
import com.nuvio.app.features.trakt.TRAKT_CONTINUE_WATCHING_DAYS_CAP_ALL
import com.nuvio.app.features.trakt.TraktSettingsRepository
import com.nuvio.app.features.trakt.WatchProgressSource
import com.nuvio.app.features.trakt.normalizeTraktContinueWatchingDaysCap
import com.nuvio.app.features.trakt.shouldUseTraktProgress
import com.nuvio.app.features.watched.WatchedItem
import com.nuvio.app.features.watched.WatchedRepository
import com.nuvio.app.features.watched.episodePlaybackId
@ -63,6 +64,7 @@ import com.nuvio.app.features.watchprogress.shouldUseAsCompletedSeedForContinueW
import com.nuvio.app.features.watchprogress.WatchProgressClock
import com.nuvio.app.features.watchprogress.WatchProgressEntry
import com.nuvio.app.features.watchprogress.WatchProgressRepository
import com.nuvio.app.features.watchprogress.WatchProgressSourceCoordinator
import com.nuvio.app.features.watchprogress.WatchProgressSourceTraktPlayback
import com.nuvio.app.features.watchprogress.buildContinueWatchingEpisodeSubtitle
import com.nuvio.app.features.watchprogress.continueWatchingEntries
@ -110,6 +112,10 @@ fun HomeScreen(
ContinueWatchingPreferencesRepository.ensureLoaded()
WatchedRepository.ensureLoaded()
WatchProgressRepository.ensureLoaded()
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) {
WatchProgressSourceCoordinator.ensureStarted()
}
}
val addonsUiState by AddonRepository.uiState.collectAsStateWithLifecycle()
@ -124,16 +130,13 @@ fun HomeScreen(
val watchedUiState by WatchedRepository.uiState.collectAsStateWithLifecycle()
val fullyWatchedSeriesKeys by WatchedRepository.fullyWatchedSeriesKeys.collectAsStateWithLifecycle()
val watchProgressUiState by WatchProgressRepository.uiState.collectAsStateWithLifecycle()
val effectiveWatchProgressSource by WatchProgressRepository.activeSourceState.collectAsStateWithLifecycle()
val cloudLibraryUiState by CloudLibraryRepository.uiState.collectAsStateWithLifecycle()
val networkStatusUiState by NetworkStatusRepository.uiState.collectAsStateWithLifecycle()
val traktSettingsUiState by remember {
TraktSettingsRepository.ensureLoaded()
TraktSettingsRepository.uiState
}.collectAsStateWithLifecycle()
val isTraktAuthenticated by remember {
TraktAuthRepository.ensureLoaded()
TraktAuthRepository.isAuthenticated
}.collectAsStateWithLifecycle()
var observedOfflineState by remember { mutableStateOf(false) }
LaunchedEffect(scrollToTopRequests) {
@ -163,15 +166,7 @@ fun HomeScreen(
}
}
val isTraktProgressActive = remember(
isTraktAuthenticated,
traktSettingsUiState.watchProgressSource,
) {
shouldUseTraktProgress(
isAuthenticated = isTraktAuthenticated,
source = traktSettingsUiState.watchProgressSource,
)
}
val isTraktProgressActive = effectiveWatchProgressSource == WatchProgressSource.TRAKT
val effectiveWatchProgressEntries = remember(
watchProgressUiState.entries,
@ -283,19 +278,25 @@ fun HomeScreen(
}
val profileState by ProfileRepository.state.collectAsStateWithLifecycle()
val activeProfileId = profileState.activeProfile?.profileIndex ?: 1
val cwCacheClearVersion by ContinueWatchingEnrichmentCache.cacheCleared.collectAsStateWithLifecycle()
val cwCacheGeneration by ContinueWatchingEnrichmentCache.generation.collectAsStateWithLifecycle()
var nextUpItemsBySeries by remember(activeProfileId) { mutableStateOf<Map<String, Pair<Long, ContinueWatchingItem>>>(emptyMap()) }
var processedNextUpContentIds by remember(activeProfileId) { mutableStateOf<Set<String>>(emptySet()) }
var nextUpItemsBySeries by remember(activeProfileId, effectiveWatchProgressSource) {
mutableStateOf<Map<String, Pair<Long, ContinueWatchingItem>>>(emptyMap())
}
var processedNextUpContentIds by remember(activeProfileId, effectiveWatchProgressSource) {
mutableStateOf<Set<String>>(emptySet())
}
LaunchedEffect(activeProfileId, cwCacheClearVersion) {
if (cwCacheClearVersion == 0) return@LaunchedEffect
LaunchedEffect(activeProfileId, effectiveWatchProgressSource, cwCacheGeneration) {
nextUpItemsBySeries = emptyMap()
processedNextUpContentIds = emptySet()
}
val cachedSnapshots = remember(activeProfileId, cwCacheClearVersion) {
ContinueWatchingEnrichmentCache.getSnapshots(activeProfileId)
val cachedSnapshots = remember(activeProfileId, effectiveWatchProgressSource, cwCacheGeneration) {
ContinueWatchingEnrichmentCache.getSnapshots(
profileId = activeProfileId,
source = effectiveWatchProgressSource,
)
}
val shouldValidateMissingNextUpSeeds = remember(
isTraktProgressActive,
@ -459,13 +460,25 @@ fun HomeScreen(
continueWatchingPreferences.upNextFromFurthestEpisode,
isRefreshingEnabledAddons,
watchProgressSeedKey,
visibleContinueWatchingEntries,
watchedUiState.items,
watchedUiState.isLoaded,
activeProfileId,
effectiveWatchProgressSource,
cwCacheGeneration,
) {
if (completedSeriesCandidates.isEmpty()) {
nextUpItemsBySeries = emptyMap()
processedNextUpContentIds = emptySet()
saveContinueWatchingSnapshots(
profileId = activeProfileId,
source = effectiveWatchProgressSource,
cacheGeneration = cwCacheGeneration,
nextUpItemsBySeries = emptyMap(),
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
todayIsoDate = CurrentDateProvider.todayIsoDate(),
seedLastWatchedMap = emptyMap(),
)
return@LaunchedEffect
}
@ -505,6 +518,8 @@ fun HomeScreen(
}
saveContinueWatchingSnapshots(
profileId = activeProfileId,
source = effectiveWatchProgressSource,
cacheGeneration = cwCacheGeneration,
nextUpItemsBySeries = cachedResolvedNextUpItems,
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
todayIsoDate = CurrentDateProvider.todayIsoDate(),
@ -579,6 +594,8 @@ fun HomeScreen(
}
saveContinueWatchingSnapshots(
profileId = activeProfileId,
source = effectiveWatchProgressSource,
cacheGeneration = cwCacheGeneration,
nextUpItemsBySeries = progressiveResults,
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
todayIsoDate = todayIsoDate,
@ -603,6 +620,8 @@ fun HomeScreen(
saveContinueWatchingSnapshots(
profileId = activeProfileId,
source = effectiveWatchProgressSource,
cacheGeneration = cwCacheGeneration,
nextUpItemsBySeries = results,
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
todayIsoDate = todayIsoDate,
@ -625,6 +644,8 @@ fun HomeScreen(
}
saveContinueWatchingSnapshots(
profileId = activeProfileId,
source = effectiveWatchProgressSource,
cacheGeneration = cwCacheGeneration,
nextUpItemsBySeries = deferredResults,
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
todayIsoDate = todayIsoDate,
@ -1284,6 +1305,8 @@ private data class HomeNextUpCandidateResolution(
private fun saveContinueWatchingSnapshots(
profileId: Int,
source: WatchProgressSource,
cacheGeneration: Int,
nextUpItemsBySeries: Map<String, Pair<Long, ContinueWatchingItem>>,
visibleContinueWatchingEntries: List<WatchProgressEntry>,
todayIsoDate: String,
@ -1338,11 +1361,22 @@ private fun saveContinueWatchingSnapshots(
}
ContinueWatchingEnrichmentCache.saveSnapshots(
profileId = profileId,
source = source,
generation = cacheGeneration,
nextUp = nextUpCache,
inProgress = inProgressCache,
)
}
internal fun effectiveContinueWatchingCacheSource(
isTraktProgressActive: Boolean,
): WatchProgressSource =
if (isTraktProgressActive) {
WatchProgressSource.TRAKT
} else {
WatchProgressSource.NUVIO_SYNC
}
private fun CompletedSeriesCandidate.toContinueWatchingSeed(meta: com.nuvio.app.features.details.MetaDetails) =
WatchProgressEntry(
contentType = content.type,

18
composeApp/src/commonMain/kotlin/com/nuvio/app/features/settings/TraktSettingsPage.kt
View file

@ -52,7 +52,7 @@ import com.nuvio.app.features.trakt.WatchProgressSource
import com.nuvio.app.features.trakt.TRAKT_CONTINUE_WATCHING_DAYS_CAP_ALL
import com.nuvio.app.features.trakt.normalizeTraktContinueWatchingDaysCap
import com.nuvio.app.features.trakt.traktBrandPainter
import com.nuvio.app.features.watchprogress.WatchProgressRepository
import com.nuvio.app.features.watchprogress.WatchProgressSourceCoordinator
import kotlinx.coroutines.launch
import nuvio.composeapp.generated.resources.Res
import nuvio.composeapp.generated.resources.action_cancel
@ -240,15 +240,19 @@ private fun TraktFeatureRows(
selectedSource = settingsUiState.watchProgressSource,
onSourceSelected = { source ->
scope.launch {
WatchProgressRepository.selectWatchProgressSource(
val result = WatchProgressSourceCoordinator.selectSource(
profileId = ProfileRepository.activeProfileId,
source = source,
)
}
statusMessage = if (source == WatchProgressSource.TRAKT) {
traktProgressSelectedMessage
} else {
nuvioProgressSelectedMessage
statusMessage = if (result.succeeded) {
if (result.requestedSource == WatchProgressSource.TRAKT) {
traktProgressSelectedMessage
} else {
nuvioProgressSelectedMessage
}
} else {
null
}
}
showWatchProgressDialog = false
},

47
composeApp/src/commonMain/kotlin/com/nuvio/app/features/trakt/TraktCredentialSync.kt
View file

@ -8,6 +8,7 @@ import com.nuvio.app.core.sync.putSyncOriginClientId
import com.nuvio.app.features.profiles.ProfileRepository
import io.github.jan.supabase.postgrest.postgrest
import io.github.jan.supabase.postgrest.rpc
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.serialization.SerialName
@ -63,24 +64,42 @@ object TraktCredentialSync {
}
suspend fun pullFromRemote(profileId: Int = ProfileRepository.activeProfileId): Boolean =
mutex.withLock {
try {
pullFromRemoteOrThrow(profileId)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.e(error) { "pullFromRemote(profileId=$profileId) failed" }
false
}
internal suspend fun pullFromRemoteOrThrow(
profileId: Int = ProfileRepository.activeProfileId,
): Boolean = mutex.withLock {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return@withLock false
val accountId = authState.userId
if (ProfileRepository.activeProfileId != profileId) return@withLock false
runCatching {
val params = buildJsonObject {
put("p_profile_id", profileId)
}
val result = SupabaseProvider.client.postgrest.rpc("sync_pull_provider_credentials", params)
val rows = result.decodeList<ProviderCredentialRow>()
val row = rows.firstOrNull { it.provider.equals(TRAKT_PROVIDER, ignoreCase = true) }
?: return@runCatching false
val remoteState = row.credentialJson.toTraktAuthState() ?: return@runCatching false
TraktAuthRepository.replaceStateFromSync(remoteState)
}.getOrElse { error ->
log.e(error) { "pullFromRemote(profileId=$profileId) failed" }
false
val params = buildJsonObject {
put("p_profile_id", profileId)
}
val result = SupabaseProvider.client.postgrest.rpc("sync_pull_provider_credentials", params)
val currentAuthState = AuthRepository.state.value
if (
currentAuthState !is AuthState.Authenticated ||
currentAuthState.isAnonymous ||
currentAuthState.userId != accountId ||
ProfileRepository.activeProfileId != profileId
) {
throw CancellationException("Trakt credential pull target changed")
}
val rows = result.decodeList<ProviderCredentialRow>()
val row = rows.firstOrNull { it.provider.equals(TRAKT_PROVIDER, ignoreCase = true) }
?: return@withLock false
val remoteState = row.credentialJson.toTraktAuthState()
?: error("Remote Trakt credential payload is invalid")
TraktAuthRepository.replaceStateFromSync(remoteState)
}
suspend fun deleteRemote(profileId: Int = ProfileRepository.activeProfileId): Boolean =

45
composeApp/src/commonMain/kotlin/com/nuvio/app/features/trakt/TraktSettingsRepository.kt
View file

@ -1,6 +1,9 @@
package com.nuvio.app.features.trakt
import com.nuvio.app.core.auth.AuthRepository
import com.nuvio.app.core.auth.AuthState
import com.nuvio.app.features.library.LibrarySourceMode
import com.nuvio.app.features.profiles.ProfileRepository
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
@ -94,11 +97,23 @@ object TraktSettingsRepository {
_uiState.value = TraktSettingsUiState()
}
fun setWatchProgressSource(source: WatchProgressSource) {
internal fun setWatchProgressSource(
source: WatchProgressSource,
profileId: Int = ProfileRepository.activeProfileId,
) {
ensureLoaded()
if (_uiState.value.watchProgressSource == source) return
_uiState.value = _uiState.value.copy(watchProgressSource = source)
persist()
val nextState = _uiState.value.copy(watchProgressSource = source)
persist(nextState)
val authState = AuthRepository.state.value
if (authState is AuthState.Authenticated && !authState.isAnonymous) {
ProfileSettingsWatchSourceOutbox.record(
accountId = authState.userId,
profileId = profileId,
source = source,
)
}
_uiState.value = nextState
}
fun setContinueWatchingDaysCap(days: Int) {
@ -148,14 +163,14 @@ object TraktSettingsRepository {
}
}
private fun persist() {
private fun persist(state: TraktSettingsUiState = _uiState.value) {
TraktSettingsStorage.savePayload(
json.encodeToString(
StoredTraktSettings(
watchProgressSource = _uiState.value.watchProgressSource.name,
continueWatchingDaysCap = _uiState.value.continueWatchingDaysCap,
librarySourceMode = _uiState.value.librarySourceMode.name,
moreLikeThisSource = _uiState.value.moreLikeThisSource.name,
watchProgressSource = state.watchProgressSource.name,
continueWatchingDaysCap = state.continueWatchingDaysCap,
librarySourceMode = state.librarySourceMode.name,
moreLikeThisSource = state.moreLikeThisSource.name,
),
),
)
@ -174,6 +189,20 @@ fun shouldUseTraktProgress(
source: WatchProgressSource,
): Boolean = isAuthenticated && source == WatchProgressSource.TRAKT
fun effectiveWatchProgressSource(
isTraktAuthenticated: Boolean,
requestedSource: WatchProgressSource,
): WatchProgressSource =
if (shouldUseTraktProgress(
isAuthenticated = isTraktAuthenticated,
source = requestedSource,
)
) {
WatchProgressSource.TRAKT
} else {
WatchProgressSource.NUVIO_SYNC
}
fun effectiveLibrarySourceMode(
isAuthenticated: Boolean,
source: LibrarySourceMode,

3
composeApp/src/commonMain/kotlin/com/nuvio/app/features/trakt/TraktSettingsStorage.kt
View file

@ -3,4 +3,7 @@ package com.nuvio.app.features.trakt
internal expect object TraktSettingsStorage {
fun loadPayload(): String?
fun savePayload(payload: String)
fun loadPendingWatchProgressSourcePayload(profileId: Int): String?
fun savePendingWatchProgressSourcePayload(profileId: Int, payload: String)
fun clearPendingWatchProgressSourcePayload(profileId: Int)
}

86
composeApp/src/commonMain/kotlin/com/nuvio/app/features/trakt/WatchProgressSourceSettingsOutbox.kt Normal file
View file

@ -0,0 +1,86 @@
package com.nuvio.app.features.trakt
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.serialization.Serializable
import kotlinx.serialization.decodeFromString
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
@Serializable
internal data class PendingWatchProgressSourceChange(
val accountId: String,
val profileId: Int,
val source: WatchProgressSource,
)
internal class WatchProgressSourceSettingsOutbox(
private val loadPayload: (profileId: Int) -> String?,
private val savePayload: (profileId: Int, payload: String) -> Unit,
private val clearPayload: (profileId: Int) -> Unit,
) {
private val lock = SynchronizedObject()
private val json = Json {
ignoreUnknownKeys = true
encodeDefaults = true
}
fun record(change: PendingWatchProgressSourceChange) = synchronized(lock) {
savePayload(change.profileId, json.encodeToString(change))
}
fun pendingFor(accountId: String, profileId: Int): PendingWatchProgressSourceChange? =
synchronized(lock) {
val payload = loadPayload(profileId).orEmpty().trim()
if (payload.isEmpty()) return@synchronized null
val change = runCatching {
json.decodeFromString<PendingWatchProgressSourceChange>(payload)
}.getOrNull()
if (change == null || change.accountId != accountId || change.profileId != profileId) {
clearPayload(profileId)
return@synchronized null
}
change
}
fun clearIfMatches(change: PendingWatchProgressSourceChange): Boolean = synchronized(lock) {
val current = loadPayload(change.profileId)
?.let { payload ->
runCatching {
json.decodeFromString<PendingWatchProgressSourceChange>(payload)
}.getOrNull()
}
if (current != change) return@synchronized false
clearPayload(change.profileId)
true
}
}
internal object ProfileSettingsWatchSourceOutbox {
private val delegate = WatchProgressSourceSettingsOutbox(
loadPayload = TraktSettingsStorage::loadPendingWatchProgressSourcePayload,
savePayload = TraktSettingsStorage::savePendingWatchProgressSourcePayload,
clearPayload = TraktSettingsStorage::clearPendingWatchProgressSourcePayload,
)
fun record(
accountId: String,
profileId: Int,
source: WatchProgressSource,
) {
delegate.record(
PendingWatchProgressSourceChange(
accountId = accountId,
profileId = profileId,
source = source,
),
)
}
fun pendingFor(accountId: String, profileId: Int): PendingWatchProgressSourceChange? =
delegate.pendingFor(accountId = accountId, profileId = profileId)
fun clearIfMatches(change: PendingWatchProgressSourceChange): Boolean =
delegate.clearIfMatches(change)
}

577
composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt
View file

@ -12,9 +12,12 @@ import com.nuvio.app.features.watching.sync.SupabaseWatchedSyncAdapter
import com.nuvio.app.features.watching.sync.TraktWatchedSyncAdapter
import com.nuvio.app.features.watching.sync.WatchedDeltaEvent
import com.nuvio.app.features.watching.sync.WatchedSyncAdapter
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
@ -44,13 +47,58 @@ internal fun shouldMirrorWatchedMarkToTraktHistory(
isTraktAuthenticated: Boolean,
): Boolean = sync == WatchedTraktHistorySync.Mirror && isTraktAuthenticated
internal data class WatchedSourceOperation(
val source: WatchProgressSource,
val generation: Long,
)
internal fun isWatchedSourceOperationCurrent(
operation: WatchedSourceOperation,
activeSource: WatchProgressSource,
activeGeneration: Long,
): Boolean = operation.source == activeSource && operation.generation == activeGeneration
internal fun watchedItemsForSource(
source: WatchProgressSource,
nuvioItems: Collection<WatchedItem>,
traktItems: Collection<WatchedItem>,
): Collection<WatchedItem> = when (source) {
WatchProgressSource.NUVIO_SYNC -> nuvioItems
WatchProgressSource.TRAKT -> traktItems
}
internal fun shouldPersistWatchedSource(source: WatchProgressSource): Boolean =
source == WatchProgressSource.NUVIO_SYNC
internal fun replaceWatchedItemsForSource(
source: WatchProgressSource,
nuvioItems: MutableMap<String, WatchedItem>,
traktItems: MutableMap<String, WatchedItem>,
replacement: Map<String, WatchedItem>,
) {
val target = when (source) {
WatchProgressSource.NUVIO_SYNC -> nuvioItems
WatchProgressSource.TRAKT -> traktItems
}
target.clear()
target.putAll(replacement)
}
object WatchedRepository {
private data class WatchedRefreshOperation(
val profileId: Int,
val profileGeneration: Long,
val sourceOperation: WatchedSourceOperation,
)
private const val watchedItemsPageSize = 900
private const val watchedItemsDeltaPageSize = 900
private const val watchedDeltaOperationUpsert = "upsert"
private const val watchedDeltaOperationDelete = "delete"
private val syncScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val accountScopeLock = SynchronizedObject()
private var accountScopeJob: Job = SupervisorJob()
private var accountScope = CoroutineScope(accountScopeJob + Dispatchers.Default)
private val log = Logger.withTag("WatchedRepository")
private val json = Json {
ignoreUnknownKeys = true
@ -65,15 +113,32 @@ object WatchedRepository {
private var hasLoaded = false
private var currentProfileId: Int = 1
private var profileGeneration: Long = 0L
private var itemsByKey: MutableMap<String, WatchedItem> = mutableMapOf()
private var activeSource: WatchProgressSource = WatchProgressSource.NUVIO_SYNC
private var sourceGeneration: Long = 0L
private var nuvioItemsByKey: MutableMap<String, WatchedItem> = mutableMapOf()
private var traktItemsByKey: MutableMap<String, WatchedItem> = mutableMapOf()
private var nuvioFullyWatchedSeriesKeys: Set<String> = emptySet()
private var traktFullyWatchedSeriesKeys: Set<String> = emptySet()
private var nuvioHasLoaded: Boolean = false
private var traktHasLoaded: Boolean = false
private var lastSuccessfulPushEpochMs: Long = 0L
private var deltaCursorEventId: Long = 0L
private var deltaInitialized: Boolean = false
internal var syncAdapter: WatchedSyncAdapter = SupabaseWatchedSyncAdapter
internal var traktSyncAdapter: WatchedSyncAdapter = TraktWatchedSyncAdapter
fun ensureLoaded() {
if (hasLoaded) return
loadFromDisk(ProfileRepository.activeProfileId)
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
if (!hasLoaded) {
loadFromDisk(ProfileRepository.activeProfileId)
activateEffectiveSource(
effectiveWatchedSource(
requestedSource = TraktSettingsRepository.uiState.value.watchProgressSource,
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
),
)
}
}
fun onProfileChanged(profileId: Int) {
@ -82,10 +147,24 @@ object WatchedRepository {
}
fun clearLocalState() {
val previousAccountJob = synchronized(accountScopeLock) {
accountScopeJob.also {
accountScopeJob = SupervisorJob()
accountScope = CoroutineScope(accountScopeJob + Dispatchers.Default)
}
}
previousAccountJob.cancel()
hasLoaded = false
currentProfileId = 1
profileGeneration += 1L
itemsByKey.clear()
activeSource = WatchProgressSource.NUVIO_SYNC
sourceGeneration += 1L
nuvioItemsByKey.clear()
traktItemsByKey.clear()
nuvioFullyWatchedSeriesKeys = emptySet()
traktFullyWatchedSeriesKeys = emptySet()
nuvioHasLoaded = false
traktHasLoaded = false
lastSuccessfulPushEpochMs = 0L
deltaCursorEventId = 0L
deltaInitialized = false
@ -96,8 +175,15 @@ object WatchedRepository {
private fun loadFromDisk(profileId: Int) {
currentProfileId = profileId
profileGeneration += 1L
activeSource = WatchProgressSource.NUVIO_SYNC
sourceGeneration += 1L
hasLoaded = true
itemsByKey.clear()
nuvioItemsByKey.clear()
traktItemsByKey.clear()
nuvioFullyWatchedSeriesKeys = emptySet()
traktFullyWatchedSeriesKeys = emptySet()
nuvioHasLoaded = true
traktHasLoaded = false
val payload = WatchedStorage.loadPayload(profileId).orEmpty().trim()
if (payload.isNotEmpty()) {
@ -107,179 +193,239 @@ object WatchedRepository {
lastSuccessfulPushEpochMs = storedPayload.lastSuccessfulPushEpochMs
deltaCursorEventId = storedPayload.deltaCursorEventId
deltaInitialized = storedPayload.deltaInitialized
itemsByKey = storedPayload.items
nuvioItemsByKey = storedPayload.items
.map(WatchedItem::normalizedMarkedAt)
.associateBy { watchedItemKey(it.type, it.id, it.season, it.episode) }
.toMutableMap()
_fullyWatchedSeriesKeys.value = storedPayload.fullyWatchedSeriesKeys
nuvioFullyWatchedSeriesKeys = storedPayload.fullyWatchedSeriesKeys
} else {
lastSuccessfulPushEpochMs = 0L
deltaCursorEventId = 0L
deltaInitialized = false
_fullyWatchedSeriesKeys.value = emptySet()
nuvioFullyWatchedSeriesKeys = emptySet()
}
publish()
}
private fun activeOperationGeneration(profileId: Int): Long? {
if (ProfileRepository.activeProfileId != profileId) return null
if (!hasLoaded || currentProfileId != profileId) {
loadFromDisk(profileId)
internal fun activateSource(source: WatchProgressSource): WatchProgressSource {
if (!hasLoaded) {
loadFromDisk(ProfileRepository.activeProfileId)
}
return profileGeneration
return activateEffectiveSource(source)
}
private fun isActiveOperation(profileId: Int, generation: Long): Boolean =
currentProfileId == profileId &&
profileGeneration == generation &&
ProfileRepository.activeProfileId == profileId
private fun activateEffectiveSource(source: WatchProgressSource): WatchProgressSource {
if (activeSource == source) return source
if (source == WatchProgressSource.TRAKT) {
traktItemsByKey.clear()
traktFullyWatchedSeriesKeys = emptySet()
traktHasLoaded = false
}
activeSource = source
sourceGeneration += 1L
publish()
return source
}
private fun newRefreshOperation(profileId: Int): WatchedRefreshOperation? {
if (ProfileRepository.activeProfileId != profileId) return null
if (!hasLoaded || currentProfileId != profileId) return null
return WatchedRefreshOperation(
profileId = profileId,
profileGeneration = profileGeneration,
sourceOperation = WatchedSourceOperation(
source = activeSource,
generation = sourceGeneration,
),
)
}
private fun isActiveOperation(operation: WatchedRefreshOperation): Boolean =
currentProfileId == operation.profileId &&
profileGeneration == operation.profileGeneration &&
ProfileRepository.activeProfileId == operation.profileId &&
isWatchedSourceOperationCurrent(
operation = operation.sourceOperation,
activeSource = activeSource,
activeGeneration = sourceGeneration,
)
suspend fun pullFromServer(profileId: Int) {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
val operationGeneration = activeOperationGeneration(profileId) ?: run {
log.d { "Skipping watched pull for inactive profile $profileId" }
return
}
val pullStartedEpochMs = WatchedClock.nowEpochMs()
val localBeforePull = itemsByKey.values
.map(WatchedItem::normalizedMarkedAt)
.toList()
val lastPushEpochMs = lastSuccessfulPushEpochMs
runCatching {
if (shouldUseTraktWatchedSync()) {
pullFullFromAdapter(
adapter = TraktWatchedSyncAdapter,
profileId = profileId,
localBeforePull = localBeforePull,
lastPushEpochMs = lastPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = true,
operationGeneration = operationGeneration,
)
} else {
pullSupabaseDeltaFromServer(
profileId = profileId,
localBeforePull = localBeforePull,
lastPushEpochMs = lastPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
operationGeneration = operationGeneration,
)
}
}.onFailure { e ->
log.e(e) { "Failed to pull watched items from server" }
}
refreshForSource(
profileId = profileId,
source = effectiveWatchedSource(
requestedSource = TraktSettingsRepository.uiState.value.watchProgressSource,
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
),
forceSnapshot = false,
)
}
suspend fun forceSnapshotRefreshFromServer(profileId: Int) {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
val operationGeneration = activeOperationGeneration(profileId) ?: run {
log.d { "Skipping watched snapshot pull for inactive profile $profileId" }
return
refreshForSource(
profileId = profileId,
source = effectiveWatchedSource(
requestedSource = TraktSettingsRepository.uiState.value.watchProgressSource,
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
),
forceSnapshot = true,
)
}
internal suspend fun refreshForSource(
profileId: Int,
source: WatchProgressSource,
forceSnapshot: Boolean = true,
): Boolean {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
if (ProfileRepository.activeProfileId != profileId) {
log.d { "Skipping watched refresh for inactive profile $profileId" }
return false
}
if (!hasLoaded || currentProfileId != profileId) {
loadFromDisk(profileId)
}
val effectiveSource = activateEffectiveSource(source)
val operation = newRefreshOperation(profileId) ?: return false
val pullStartedEpochMs = WatchedClock.nowEpochMs()
val localBeforePull = itemsByKey.values
.map(WatchedItem::normalizedMarkedAt)
.toList()
val lastPushEpochMs = lastSuccessfulPushEpochMs
runCatching {
if (shouldUseTraktWatchedSync()) {
pullFullFromAdapter(
adapter = TraktWatchedSyncAdapter,
return try {
if (effectiveSource == WatchProgressSource.TRAKT) {
pullSnapshotFromAdapter(
adapter = traktSyncAdapter,
operation = operation,
profileId = profileId,
localBeforePull = localBeforePull,
lastPushEpochMs = lastPushEpochMs,
lastPushEpochMs = 0L,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = true,
operationGeneration = operationGeneration,
)
return@runCatching
} else if (forceSnapshot) {
refreshNuvioSnapshot(
operation = operation,
profileId = profileId,
pullStartedEpochMs = pullStartedEpochMs,
)
} else {
pullSupabaseDeltaFromServer(
operation = operation,
profileId = profileId,
pullStartedEpochMs = pullStartedEpochMs,
)
}
val cursorBeforeSnapshot = try {
syncAdapter.getDeltaCursor(profileId)
} catch (error: CancellationException) {
throw error
} catch (_: Throwable) {
null
}
pullFullFromAdapter(
adapter = syncAdapter,
profileId = profileId,
localBeforePull = localBeforePull,
lastPushEpochMs = lastPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = cursorBeforeSnapshot == null,
operationGeneration = operationGeneration,
)
if (!isActiveOperation(profileId, operationGeneration)) return@runCatching
if (cursorBeforeSnapshot != null) {
deltaCursorEventId = cursorBeforeSnapshot
deltaInitialized = true
persist()
}
}.onFailure { e ->
if (e is CancellationException) throw e
log.e(e) { "Failed to pull watched items snapshot from server" }
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.e(error) { "Failed to refresh watched items from $effectiveSource" }
false
}
}
private suspend fun pullFullFromAdapter(
adapter: WatchedSyncAdapter,
private suspend fun refreshNuvioSnapshot(
operation: WatchedRefreshOperation,
profileId: Int,
pullStartedEpochMs: Long,
): Boolean {
val cursorBeforeSnapshot = try {
syncAdapter.getDeltaCursor(profileId)
} catch (error: CancellationException) {
throw error
} catch (_: Throwable) {
null
}
if (!isActiveOperation(operation)) return false
val applied = pullSnapshotFromAdapter(
adapter = syncAdapter,
operation = operation,
profileId = profileId,
lastPushEpochMs = lastSuccessfulPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = cursorBeforeSnapshot == null,
)
if (!applied || !isActiveOperation(operation)) return false
if (cursorBeforeSnapshot != null) {
deltaCursorEventId = cursorBeforeSnapshot
deltaInitialized = true
persistNuvio()
}
return true
}
private suspend fun pullSnapshotFromAdapter(
adapter: WatchedSyncAdapter,
operation: WatchedRefreshOperation,
profileId: Int,
localBeforePull: List<WatchedItem>,
lastPushEpochMs: Long,
pullStartedEpochMs: Long,
resetDeltaState: Boolean,
operationGeneration: Long,
) {
): Boolean {
val serverItems = adapter.pull(
profileId = profileId,
pageSize = watchedItemsPageSize,
)
if (!isActiveOperation(profileId, operationGeneration)) return
if (!isActiveOperation(operation)) return false
val localAtApply = itemsForSource(operation.sourceOperation.source).values.toList()
itemsByKey = mergeWatchedItemsPreservingUnsynced(
val mergedItems = mergeWatchedItemsPreservingUnsynced(
serverItems = serverItems,
localItems = localBeforePull,
localItems = localAtApply,
lastSuccessfulPushEpochMs = lastPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
preserveWhenNoSuccessfulPush = operation.sourceOperation.source == WatchProgressSource.NUVIO_SYNC,
).toMutableMap()
if (resetDeltaState) {
deltaCursorEventId = 0L
deltaInitialized = false
replaceWatchedItemsForSource(
source = operation.sourceOperation.source,
nuvioItems = nuvioItemsByKey,
traktItems = traktItemsByKey,
replacement = mergedItems,
)
when (operation.sourceOperation.source) {
WatchProgressSource.NUVIO_SYNC -> {
nuvioHasLoaded = true
if (resetDeltaState) {
deltaCursorEventId = 0L
deltaInitialized = false
}
}
WatchProgressSource.TRAKT -> {
traktHasLoaded = true
}
}
hasLoaded = true
publish()
persist()
if (shouldPersistWatchedSource(operation.sourceOperation.source)) {
persistNuvio()
}
return true
}
private suspend fun pullSupabaseDeltaFromServer(
operation: WatchedRefreshOperation,
profileId: Int,
localBeforePull: List<WatchedItem>,
lastPushEpochMs: Long,
pullStartedEpochMs: Long,
operationGeneration: Long,
) {
if (!isActiveOperation(profileId, operationGeneration)) return
): Boolean {
if (!isActiveOperation(operation)) return false
if (!deltaInitialized) {
val cursorBeforeSnapshot = syncAdapter.getDeltaCursor(profileId) ?: return
pullFullFromAdapter(
val cursorBeforeSnapshot = syncAdapter.getDeltaCursor(profileId) ?: return false
if (!isActiveOperation(operation)) return false
val applied = pullSnapshotFromAdapter(
adapter = syncAdapter,
operation = operation,
profileId = profileId,
localBeforePull = localBeforePull,
lastPushEpochMs = lastPushEpochMs,
lastPushEpochMs = lastSuccessfulPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = false,
operationGeneration = operationGeneration,
)
if (!isActiveOperation(profileId, operationGeneration)) return
if (!applied || !isActiveOperation(operation)) return false
deltaCursorEventId = cursorBeforeSnapshot
deltaInitialized = true
persist()
return
persistNuvio()
return true
}
var cursor = deltaCursorEventId
@ -291,10 +437,11 @@ object WatchedRepository {
sinceEventId = cursor,
limit = watchedItemsDeltaPageSize,
)
if (!isActiveOperation(profileId, operationGeneration)) return
if (!isActiveOperation(operation)) return false
if (events.isEmpty()) break
applyWatchedDeltaEvents(
targetItems = nuvioItemsByKey,
events = events,
pullStartedEpochMs = pullStartedEpochMs,
)
@ -306,14 +453,17 @@ object WatchedRepository {
if (events.size < watchedItemsDeltaPageSize) break
}
hasLoaded = true
if (!isActiveOperation(operation)) return false
nuvioHasLoaded = true
if (changed) {
publish()
persist()
persistNuvio()
}
return true
}
private fun applyWatchedDeltaEvents(
targetItems: MutableMap<String, WatchedItem>,
events: Collection<WatchedDeltaEvent>,
pullStartedEpochMs: Long,
) {
@ -329,7 +479,7 @@ object WatchedRepository {
when (event.operation.lowercase()) {
watchedDeltaOperationUpsert -> {
upsertCount += 1
itemsByKey[key] = WatchedItem(
targetItems[key] = WatchedItem(
id = event.contentId,
type = event.contentType,
name = event.title,
@ -340,16 +490,17 @@ object WatchedRepository {
}
watchedDeltaOperationDelete -> {
deleteCount += 1
val localItem = itemsByKey[key]
val localItem = targetItems[key]
if (localItem != null && wasWatchedItemMarkedDuringPull(localItem, pullStartedEpochMs)) {
preservedDuringPullCount += 1
return@forEach
}
val removedItem = itemsByKey.remove(key)
val removedItem = targetItems.remove(key)
if (removedItem != null) {
removedCount += 1
} else if (
removeWatchedItemByStableDeleteKey(
targetItems = targetItems,
contentId = event.contentId,
contentType = event.contentType,
season = event.season,
@ -374,19 +525,20 @@ object WatchedRepository {
}
private fun removeWatchedItemByStableDeleteKey(
targetItems: MutableMap<String, WatchedItem>,
contentId: String,
contentType: String,
season: Int?,
episode: Int?,
): Boolean {
val fallbackKey = itemsByKey.entries.firstOrNull { (_, item) ->
val fallbackKey = targetItems.entries.firstOrNull { (_, item) ->
item.id == contentId &&
watchedDeleteTypesCompatible(remoteType = contentType, localType = item.type) &&
item.season == season &&
item.episode == episode
}?.key ?: return false
itemsByKey.remove(fallbackKey)
targetItems.remove(fallbackKey)
log.w {
"Removed watched delta delete with fallback key contentId=$contentId contentType=$contentType " +
"season=$season episode=$episode matchedKey=$fallbackKey"
@ -399,10 +551,40 @@ object WatchedRepository {
return remoteType.isSeriesLikeWatchedType() && localType.isSeriesLikeWatchedType()
}
private fun itemsForSource(source: WatchProgressSource): MutableMap<String, WatchedItem> =
when (source) {
WatchProgressSource.NUVIO_SYNC -> nuvioItemsByKey
WatchProgressSource.TRAKT -> traktItemsByKey
}
private fun fullyWatchedSeriesKeysForSource(source: WatchProgressSource): Set<String> =
when (source) {
WatchProgressSource.NUVIO_SYNC -> nuvioFullyWatchedSeriesKeys
WatchProgressSource.TRAKT -> traktFullyWatchedSeriesKeys
}
private fun setFullyWatchedSeriesKeysForSource(
source: WatchProgressSource,
keys: Set<String>,
) {
when (source) {
WatchProgressSource.NUVIO_SYNC -> nuvioFullyWatchedSeriesKeys = keys
WatchProgressSource.TRAKT -> traktFullyWatchedSeriesKeys = keys
}
}
private fun hasLoadedSource(source: WatchProgressSource): Boolean =
when (source) {
WatchProgressSource.NUVIO_SYNC -> nuvioHasLoaded
WatchProgressSource.TRAKT -> traktHasLoaded
}
fun toggleWatched(item: WatchedItem) {
ensureLoaded()
val source = activeSource
val targetItems = itemsForSource(source)
val key = watchedItemKey(item.type, item.id, item.season, item.episode)
if (itemsByKey.containsKey(key)) {
if (targetItems.containsKey(key)) {
unmarkWatched(item)
} else {
markWatched(item)
@ -428,18 +610,26 @@ object WatchedRepository {
) {
ensureLoaded()
if (items.isEmpty()) return
val source = activeSource
val targetItems = itemsForSource(source)
val markedAt = WatchedClock.nowEpochMs()
val timestampedItems = items.map { watchedItem ->
watchedItem.copy(markedAtEpochMs = markedAt)
}
timestampedItems.forEach { watchedItem ->
val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode)
itemsByKey[key] = watchedItem
targetItems[key] = watchedItem
}
publish()
persist()
if (shouldPersistWatchedSource(source)) {
persistNuvio()
}
if (syncRemote) {
pushMarksToServer(timestampedItems, traktHistorySync)
pushMarksToServer(
items = timestampedItems,
traktHistorySync = traktHistorySync,
source = source,
)
}
}
@ -470,13 +660,17 @@ object WatchedRepository {
fun unmarkWatched(items: Collection<WatchedItem>) {
ensureLoaded()
if (items.isEmpty()) return
val source = activeSource
val targetItems = itemsForSource(source)
val removedItems = items.mapNotNull { watchedItem ->
itemsByKey.remove(watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode))
targetItems.remove(watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode))
}
if (removedItems.isNotEmpty()) {
publish()
persist()
pushDeleteToServer(removedItems)
if (shouldPersistWatchedSource(source)) {
persistNuvio()
}
pushDeleteToServer(items = removedItems, source = source)
}
}
@ -487,7 +681,7 @@ object WatchedRepository {
episode: Int? = null,
): Boolean {
ensureLoaded()
return itemsByKey.containsKey(watchedItemKey(type, id, season, episode))
return itemsForSource(activeSource).containsKey(watchedItemKey(type, id, season, episode))
}
fun reconcileSeriesWatchedState(
@ -557,28 +751,39 @@ object WatchedRepository {
key: String,
isFullyWatched: Boolean,
) {
val current = _fullyWatchedSeriesKeys.value
val source = activeSource
val current = fullyWatchedSeriesKeysForSource(source)
val updated = if (isFullyWatched) current + key else current - key
if (updated == current) return
_fullyWatchedSeriesKeys.value = updated
persist()
setFullyWatchedSeriesKeysForSource(source = source, keys = updated)
publish()
if (shouldPersistWatchedSource(source)) {
persistNuvio()
}
}
private fun pushMarksToServer(
items: Collection<WatchedItem>,
traktHistorySync: WatchedTraktHistorySync,
source: WatchProgressSource,
) {
val profileId = currentProfileId
syncScope.launch {
val operationGeneration = profileGeneration
accountScopeSnapshot().launch {
runCatching {
if (items.isEmpty()) return@runCatching
val pushed = pushToActiveTargets(
val pushed = pushToTargetsForSource(
profileId = profileId,
items = items,
traktHistorySync = traktHistorySync,
source = source,
)
if (pushed) {
recordSuccessfulPush(profileId = profileId, items = items)
if (pushed && shouldPersistWatchedSource(source)) {
recordSuccessfulPush(
profileId = profileId,
operationGeneration = operationGeneration,
items = items,
)
}
}.onFailure { e ->
log.e(e) { "Failed to push watched items" }
@ -586,12 +791,19 @@ object WatchedRepository {
}
}
private fun pushDeleteToServer(items: Collection<WatchedItem>) {
private fun pushDeleteToServer(
items: Collection<WatchedItem>,
source: WatchProgressSource,
) {
val profileId = currentProfileId
syncScope.launch {
accountScopeSnapshot().launch {
runCatching {
if (items.isEmpty()) return@runCatching
deleteFromActiveTargets(profileId = profileId, items = items)
deleteFromTargetsForSource(
profileId = profileId,
items = items,
source = source,
)
}.onFailure { e ->
log.e(e) { "Failed to push watched item delete" }
}
@ -599,27 +811,32 @@ object WatchedRepository {
}
private fun publish() {
val items = itemsByKey.values
val items = watchedItemsForSource(
source = activeSource,
nuvioItems = nuvioItemsByKey.values,
traktItems = traktItemsByKey.values,
)
.map(WatchedItem::normalizedMarkedAt)
.sortedByDescending { it.markedAtEpochMs }
_fullyWatchedSeriesKeys.value = fullyWatchedSeriesKeysForSource(activeSource)
_uiState.value = WatchedUiState(
items = items,
watchedKeys = items.mapTo(linkedSetOf()) {
watchedItemKey(it.type, it.id, it.season, it.episode)
},
isLoaded = true,
isLoaded = hasLoadedSource(activeSource),
)
}
private fun persist() {
private fun persistNuvio() {
WatchedStorage.savePayload(
currentProfileId,
json.encodeToString(
StoredWatchedPayload(
items = itemsByKey.values
items = nuvioItemsByKey.values
.map(WatchedItem::normalizedMarkedAt)
.sortedByDescending { it.markedAtEpochMs },
fullyWatchedSeriesKeys = _fullyWatchedSeriesKeys.value,
fullyWatchedSeriesKeys = nuvioFullyWatchedSeriesKeys,
lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
deltaCursorEventId = deltaCursorEventId,
deltaInitialized = deltaInitialized,
@ -628,8 +845,12 @@ object WatchedRepository {
)
}
private fun recordSuccessfulPush(profileId: Int, items: Collection<WatchedItem>) {
if (profileId != currentProfileId) return
private fun recordSuccessfulPush(
profileId: Int,
operationGeneration: Long,
items: Collection<WatchedItem>,
) {
if (profileId != currentProfileId || operationGeneration != profileGeneration) return
val latestPushed = items
.asSequence()
.map { item -> normalizeWatchedMarkedAtEpochMs(item.markedAtEpochMs) }
@ -637,52 +858,53 @@ object WatchedRepository {
?: return
if (latestPushed <= lastSuccessfulPushEpochMs) return
lastSuccessfulPushEpochMs = latestPushed
persist()
persistNuvio()
}
private fun shouldUseTraktWatchedSync(): Boolean =
shouldUseTraktWatchedSync(
isAuthenticated = TraktAuthRepository.isAuthenticated.value,
source = TraktSettingsRepository.uiState.value.watchProgressSource,
)
private suspend fun pushToActiveTargets(
private suspend fun pushToTargetsForSource(
profileId: Int,
items: Collection<WatchedItem>,
traktHistorySync: WatchedTraktHistorySync,
source: WatchProgressSource,
): Boolean {
val shouldMirrorToTrakt = shouldMirrorWatchedMarkToTraktHistory(
sync = traktHistorySync,
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
)
if (shouldUseTraktWatchedSync()) {
if (source == WatchProgressSource.TRAKT) {
if (!shouldMirrorToTrakt) return false
TraktWatchedSyncAdapter.push(profileId = profileId, items = items)
traktSyncAdapter.push(profileId = profileId, items = items)
return true
}
syncAdapter.push(profileId = profileId, items = items)
if (shouldMirrorToTrakt) {
TraktWatchedSyncAdapter.push(profileId = profileId, items = items)
traktSyncAdapter.push(profileId = profileId, items = items)
}
return true
}
private suspend fun deleteFromActiveTargets(
private suspend fun deleteFromTargetsForSource(
profileId: Int,
items: Collection<WatchedItem>,
source: WatchProgressSource,
) {
if (shouldUseTraktWatchedSync()) {
TraktWatchedSyncAdapter.delete(profileId = profileId, items = items)
if (source == WatchProgressSource.TRAKT) {
traktSyncAdapter.delete(profileId = profileId, items = items)
return
}
syncAdapter.delete(profileId = profileId, items = items)
if (TraktAuthRepository.isAuthenticated.value) {
TraktWatchedSyncAdapter.delete(profileId = profileId, items = items)
traktSyncAdapter.delete(profileId = profileId, items = items)
}
}
private fun accountScopeSnapshot(): CoroutineScope =
synchronized(accountScopeLock) {
accountScope
}
}
internal fun mergeWatchedItemsPreservingUnsynced(
@ -690,6 +912,7 @@ internal fun mergeWatchedItemsPreservingUnsynced(
localItems: Collection<WatchedItem>,
lastSuccessfulPushEpochMs: Long,
pullStartedEpochMs: Long,
preserveWhenNoSuccessfulPush: Boolean = true,
): Map<String, WatchedItem> {
val merged = serverItems
.map(WatchedItem::normalizedMarkedAt)
@ -701,7 +924,14 @@ internal fun mergeWatchedItemsPreservingUnsynced(
.forEach { localItem ->
val key = watchedItemKey(localItem.type, localItem.id, localItem.season, localItem.episode)
if (key in merged) return@forEach
if (shouldPreserveLocalWatchedItem(localItem, lastSuccessfulPushEpochMs, pullStartedEpochMs)) {
if (
shouldPreserveLocalWatchedItem(
localItem = localItem,
lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
pullStartedEpochMs = pullStartedEpochMs,
preserveWhenNoSuccessfulPush = preserveWhenNoSuccessfulPush,
)
) {
merged[key] = localItem
}
}
@ -713,9 +943,12 @@ internal fun shouldPreserveLocalWatchedItem(
localItem: WatchedItem,
lastSuccessfulPushEpochMs: Long,
pullStartedEpochMs: Long,
preserveWhenNoSuccessfulPush: Boolean = true,
): Boolean {
val markedAt = normalizeWatchedMarkedAtEpochMs(localItem.markedAtEpochMs)
val wasMarkedAfterLastPush = lastSuccessfulPushEpochMs > 0L && markedAt > lastSuccessfulPushEpochMs
val wasMarkedAfterLastPush =
(preserveWhenNoSuccessfulPush && lastSuccessfulPushEpochMs <= 0L) ||
(lastSuccessfulPushEpochMs > 0L && markedAt > lastSuccessfulPushEpochMs)
val wasMarkedDuringPull = pullStartedEpochMs > 0L && markedAt >= pullStartedEpochMs
return wasMarkedAfterLastPush || wasMarkedDuringPull
}
@ -736,5 +969,15 @@ internal fun shouldUseTraktWatchedSync(
source = source,
)
internal fun effectiveWatchedSource(
requestedSource: WatchProgressSource,
isTraktAuthenticated: Boolean,
): WatchProgressSource =
if (shouldUseTraktWatchedSync(isAuthenticated = isTraktAuthenticated, source = requestedSource)) {
WatchProgressSource.TRAKT
} else {
WatchProgressSource.NUVIO_SYNC
}
private fun String.isSeriesLikeWatchedType(): Boolean =
trim().lowercase() in setOf("series", "show", "tv", "tvshow")

135
composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/ContinueWatchingEnrichmentCache.kt
View file

@ -1,6 +1,9 @@
package com.nuvio.app.features.watchprogress
import com.nuvio.app.core.storage.ProfileScopedKey
import com.nuvio.app.features.trakt.WatchProgressSource
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
@ -65,18 +68,28 @@ internal object ContinueWatchingEnrichmentCache {
}
private const val storageKey = "cw_enrichment_cache"
private val lastPayloadHashByProfile = mutableMapOf<Int, Int>()
private val _cacheCleared = MutableStateFlow(0)
val cacheCleared: StateFlow<Int> = _cacheCleared.asStateFlow()
private val cacheLock = SynchronizedObject()
private val lastPayloadHashByScope = mutableMapOf<CacheScope, Int>()
private val _generation = MutableStateFlow(0)
val generation: StateFlow<Int> = _generation.asStateFlow()
fun getNextUpSnapshot(profileId: Int): List<CachedNextUpItem> =
loadPayload(profileId)?.nextUp ?: emptyList()
fun getNextUpSnapshot(
profileId: Int,
source: WatchProgressSource,
): List<CachedNextUpItem> =
loadPayload(profileId = profileId, source = source)?.nextUp ?: emptyList()
fun getInProgressSnapshot(profileId: Int): List<CachedInProgressItem> =
loadPayload(profileId)?.inProgress ?: emptyList()
fun getInProgressSnapshot(
profileId: Int,
source: WatchProgressSource,
): List<CachedInProgressItem> =
loadPayload(profileId = profileId, source = source)?.inProgress ?: emptyList()
fun getSnapshots(profileId: Int): Pair<List<CachedNextUpItem>, List<CachedInProgressItem>> {
val payload = loadPayload(profileId)
fun getSnapshots(
profileId: Int,
source: WatchProgressSource,
): Pair<List<CachedNextUpItem>, List<CachedInProgressItem>> {
val payload = loadPayload(profileId = profileId, source = source)
val nextUp = payload?.nextUp ?: emptyList()
val inProgress = payload?.inProgress ?: emptyList()
return nextUp to inProgress
@ -84,43 +97,111 @@ internal object ContinueWatchingEnrichmentCache {
fun saveSnapshots(
profileId: Int,
source: WatchProgressSource,
generation: Int,
nextUp: List<CachedNextUpItem>,
inProgress: List<CachedInProgressItem>,
force: Boolean = false,
) {
): Boolean = synchronized(cacheLock) {
if (generation != _generation.value) return@synchronized false
removeLegacyPayload(profileId)
val payload = CachedEnrichmentPayload(nextUp = nextUp, inProgress = inProgress)
val payloadHash = payload.hashCode()
if (!force && lastPayloadHashByProfile[profileId] == payloadHash) {
return
val scope = CacheScope(profileId = profileId, source = source)
if (!force && lastPayloadHashByScope[scope] == payloadHash) {
return@synchronized true
}
val encoded = runCatching {
json.encodeToString(payload)
}.getOrNull() ?: return
ContinueWatchingEnrichmentStorage.savePayload(profileScopedStorageKey(profileId), encoded)
lastPayloadHashByProfile[profileId] = payloadHash
}.getOrNull() ?: return@synchronized false
ContinueWatchingEnrichmentStorage.savePayload(
continueWatchingEnrichmentStorageKey(profileId = profileId, source = source),
encoded,
)
lastPayloadHashByScope[scope] = payloadHash
true
}
fun clearAll(profileId: Int) {
ContinueWatchingEnrichmentStorage.removePayload(profileScopedStorageKey(profileId))
lastPayloadHashByProfile.remove(profileId)
_cacheCleared.value += 1
fun invalidate(
profileId: Int,
source: WatchProgressSource,
) = synchronized(cacheLock) {
ContinueWatchingEnrichmentStorage.removePayload(
continueWatchingEnrichmentStorageKey(profileId = profileId, source = source),
)
removeLegacyPayload(profileId)
lastPayloadHashByScope.remove(CacheScope(profileId = profileId, source = source))
advanceGeneration()
}
fun onProfileChanged() {
_cacheCleared.value += 1
fun clearAll(profileId: Int) = synchronized(cacheLock) {
WatchProgressSource.entries.forEach { source ->
ContinueWatchingEnrichmentStorage.removePayload(
continueWatchingEnrichmentStorageKey(profileId = profileId, source = source),
)
lastPayloadHashByScope.remove(CacheScope(profileId = profileId, source = source))
}
removeLegacyPayload(profileId)
advanceGeneration()
}
private fun loadPayload(profileId: Int): CachedEnrichmentPayload? {
val raw = ContinueWatchingEnrichmentStorage.loadPayload(profileScopedStorageKey(profileId))
?: return null
return runCatching {
fun clearLocalState() = synchronized(cacheLock) {
lastPayloadHashByScope.clear()
advanceGeneration()
}
fun onProfileChanged() = synchronized(cacheLock) {
advanceGeneration()
}
private fun loadPayload(
profileId: Int,
source: WatchProgressSource,
): CachedEnrichmentPayload? = synchronized(cacheLock) {
removeLegacyPayload(profileId)
val scope = CacheScope(profileId = profileId, source = source)
val raw = ContinueWatchingEnrichmentStorage.loadPayload(
continueWatchingEnrichmentStorageKey(profileId = profileId, source = source),
) ?: run {
lastPayloadHashByScope.remove(scope)
return@synchronized null
}
runCatching {
json.decodeFromString<CachedEnrichmentPayload>(raw)
}.getOrNull()?.also { payload ->
lastPayloadHashByProfile[profileId] = payload.hashCode()
lastPayloadHashByScope[scope] = payload.hashCode()
} ?: run {
lastPayloadHashByScope.remove(scope)
ContinueWatchingEnrichmentStorage.removePayload(
continueWatchingEnrichmentStorageKey(profileId = profileId, source = source),
)
null
}
}
private fun profileScopedStorageKey(profileId: Int): String =
private fun removeLegacyPayload(profileId: Int) {
ContinueWatchingEnrichmentStorage.removePayload(legacyStorageKey(profileId))
}
private fun advanceGeneration() {
_generation.value += 1
}
private data class CacheScope(
val profileId: Int,
val source: WatchProgressSource,
)
internal fun continueWatchingEnrichmentStorageKey(
profileId: Int,
source: WatchProgressSource,
): String = ProfileScopedKey.of(
baseKey = "${storageKey}_${source.name.lowercase()}",
profileId = profileId,
)
internal fun legacyStorageKey(profileId: Int): String =
ProfileScopedKey.of(storageKey, profileId)
}

442
composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRepository.kt
View file

@ -15,9 +15,9 @@ import com.nuvio.app.features.trakt.TraktAuthRepository
import com.nuvio.app.features.trakt.TraktProgressRepository
import com.nuvio.app.features.trakt.TraktSettingsRepository
import com.nuvio.app.features.trakt.WatchProgressSource
import com.nuvio.app.features.trakt.effectiveWatchProgressSource
import com.nuvio.app.features.trakt.isTraktCompatibleId
import com.nuvio.app.features.trakt.resolveEffectiveContentId
import com.nuvio.app.features.trakt.shouldUseTraktProgress as shouldUseTraktProgressSource
import com.nuvio.app.features.watching.application.WatchingActions
import com.nuvio.app.features.watching.sync.ProgressDeltaEvent
import com.nuvio.app.features.watching.sync.ProgressSyncRecord
@ -38,7 +38,9 @@ import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.launch
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.withTimeoutOrNull
@ -77,6 +79,9 @@ private data class WatchProgressDeltaApplyResult(
object WatchProgressRepository {
private val syncScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val accountScopeLock = SynchronizedObject()
private var accountScopeJob: Job = SupervisorJob()
private var accountScope = CoroutineScope(accountScopeJob + Dispatchers.Default)
private val log = Logger.withTag("WatchProgressRepository")
private val _uiState = MutableStateFlow(WatchProgressUiState())
@ -85,10 +90,13 @@ object WatchProgressRepository {
private var hasLoaded = false
private var currentProfileId: Int = 1
private var profileGeneration: Long = 0L
private var activeSource: WatchProgressSource = WatchProgressSource.NUVIO_SYNC
private val _activeSourceState = MutableStateFlow(activeSource)
internal val activeSourceState: StateFlow<WatchProgressSource> = _activeSourceState.asStateFlow()
private val entriesLock = SynchronizedObject()
private var entriesByVideoId: MutableMap<String, WatchProgressEntry> = mutableMapOf()
private var metadataResolutionJob: Job? = null
private var isPullingNuvioSyncFromServer = false
private val nuvioPullMutex = Mutex()
private var lastSuccessfulPushEpochMs = 0L
private var deltaCursorEventId = 0L
private var deltaInitialized = false
@ -96,40 +104,6 @@ object WatchProgressRepository {
internal var syncAdapter: ProgressSyncAdapter = SupabaseProgressSyncAdapter
init {
syncScope.launch {
TraktAuthRepository.isAuthenticated.collectLatest { authenticated ->
if (shouldUseTraktProgressSource(
isAuthenticated = authenticated,
source = TraktSettingsRepository.uiState.value.watchProgressSource,
)
) {
runCatching { TraktProgressRepository.refreshNow() }
.onFailure { error ->
if (error is CancellationException) throw error
log.w { "Failed to refresh Trakt progress after auth: ${error.message}" }
}
}
publish()
}
}
syncScope.launch {
TraktSettingsRepository.uiState.collectLatest { settings ->
if (shouldUseTraktProgressSource(
isAuthenticated = TraktAuthRepository.isAuthenticated.value,
source = settings.watchProgressSource,
)
) {
runCatching { TraktProgressRepository.refreshNow() }
.onFailure { error ->
if (error is CancellationException) throw error
log.w { "Failed to refresh Trakt progress after source change: ${error.message}" }
}
}
publish()
}
}
syncScope.launch {
TraktProgressRepository.uiState.collectLatest {
if (shouldUseTraktProgress()) {
@ -150,28 +124,43 @@ object WatchProgressRepository {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
TraktProgressRepository.ensureLoaded()
if (hasLoaded) return
loadFromDisk(ProfileRepository.activeProfileId)
if (shouldUseTraktProgress()) {
TraktProgressRepository.refreshAsync()
if (!hasLoaded) {
updateActiveSource(
effectiveWatchProgressSource(
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
requestedSource = TraktSettingsRepository.uiState.value.watchProgressSource,
),
)
loadFromDisk(ProfileRepository.activeProfileId)
}
}
fun onProfileChanged(profileId: Int) {
if (profileId == currentProfileId && hasLoaded) return
TraktSettingsRepository.onProfileChanged()
updateActiveSource(
effectiveWatchProgressSource(
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
requestedSource = TraktSettingsRepository.uiState.value.watchProgressSource,
),
)
loadFromDisk(profileId)
TraktProgressRepository.onProfileChanged()
if (shouldUseTraktProgress()) {
TraktProgressRepository.refreshAsync()
}
}
fun clearLocalState() {
val previousAccountJob = synchronized(accountScopeLock) {
accountScopeJob.also {
accountScopeJob = SupervisorJob()
accountScope = CoroutineScope(accountScopeJob + Dispatchers.Default)
}
}
previousAccountJob.cancel()
metadataResolutionJob?.cancel()
hasLoaded = false
currentProfileId = 1
profileGeneration += 1L
updateActiveSource(WatchProgressSource.NUVIO_SYNC)
lastAddonMetadataReadyFingerprint = null
clearLocalEntries()
lastSuccessfulPushEpochMs = 0L
@ -224,195 +213,181 @@ object WatchProgressRepository {
ProfileRepository.activeProfileId == profileId
suspend fun pullFromServer(profileId: Int) {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
TraktProgressRepository.ensureLoaded()
val operationGeneration = activeOperationGeneration(profileId) ?: run {
log.d { "Skipping watch progress pull for inactive profile $profileId" }
return
}
val useTraktProgress = shouldUseTraktProgress()
if (!useTraktProgress && isPullingNuvioSyncFromServer) {
log.d { "Skipping watch progress pull for profile $profileId because a Nuvio sync pull is already running" }
return
}
if (!useTraktProgress) {
isPullingNuvioSyncFromServer = true
}
try {
if (useTraktProgress) {
log.d { "Pulling Trakt watch progress for profile $profileId" }
runCatching { TraktProgressRepository.refreshNow() }
.onFailure { e ->
if (e is CancellationException) throw e
log.e(e) { "Failed to pull Trakt progress" }
}
if (isActiveOperation(profileId, operationGeneration)) {
publish()
}
return
}
runCatching {
log.d { "Pulling Nuvio watch progress for profile $profileId" }
pullSupabaseDeltaFromServer(
profileId = profileId,
pullStartedEpochMs = WatchProgressClock.nowEpochMs(),
operationGeneration = operationGeneration,
)
}.onFailure { e ->
if (e is CancellationException) throw e
log.e(e) { "Failed to pull watch progress from server" }
}
} finally {
if (!useTraktProgress) {
isPullingNuvioSyncFromServer = false
}
}
refreshForSource(
profileId = profileId,
source = activeSource,
sourceChanged = false,
force = false,
)
}
suspend fun forceSnapshotRefreshFromServer(profileId: Int) {
ensureLoaded()
if (currentProfileId != profileId) {
loadFromDisk(profileId)
}
refreshForSource(
profileId = profileId,
source = activeSource,
sourceChanged = false,
force = true,
)
}
if (shouldUseTraktProgress()) {
log.d { "Force refreshing Trakt watch progress for profile $profileId" }
runCatching { TraktProgressRepository.invalidateAndRefresh() }
.onFailure { error ->
if (error is CancellationException) throw error
log.e(error) { "Failed to force refresh Trakt progress" }
}
suspend fun selectWatchProgressSource(profileId: Int, source: WatchProgressSource) {
WatchProgressSourceCoordinator.selectSource(profileId = profileId, source = source)
}
suspend fun clearLocalAndForceSnapshotRefreshFromServer(profileId: Int) {
ContinueWatchingEnrichmentCache.clearAll(profileId)
WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = true)
}
internal fun activateSource(source: WatchProgressSource) {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
TraktProgressRepository.ensureLoaded()
if (!hasLoaded) {
loadFromDisk(ProfileRepository.activeProfileId)
}
if (activeSource == source) {
publish()
return
}
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) {
log.d { "Skipping force watch progress refresh because Nuvio Sync is not authenticated" }
return
}
deltaCursorEventId = 0L
deltaInitialized = false
persist()
pullFromServer(profileId)
}
suspend fun selectWatchProgressSource(profileId: Int, source: WatchProgressSource) {
TraktSettingsRepository.ensureLoaded()
val previousSource = TraktSettingsRepository.uiState.value.watchProgressSource
if (previousSource == source) return
ensureLoaded()
if (currentProfileId != profileId) {
loadFromDisk(profileId)
}
ContinueWatchingEnrichmentCache.clearAll(profileId)
updateActiveSource(source)
metadataResolutionJob?.cancel()
TraktSettingsRepository.setWatchProgressSource(source)
val removedLocalEntries = removeLocalEntriesMatching { entry ->
isTraktCompatibleId(entry.parentMetaId)
}
if (removedLocalEntries) {
persist()
}
when (source) {
WatchProgressSource.TRAKT -> {
TraktProgressRepository.clearLocalState()
publish()
if (TraktAuthRepository.isAuthenticated.value) {
runCatching { TraktProgressRepository.invalidateAndRefresh() }
.onFailure { error ->
if (error is CancellationException) throw error
log.e(error) { "Failed to refresh Trakt progress after source selection" }
}
publish()
}
}
WatchProgressSource.NUVIO_SYNC -> {
publish()
forceSnapshotRefreshFromServer(profileId)
}
}
}
suspend fun clearLocalAndForceSnapshotRefreshFromServer(profileId: Int) {
ensureLoaded()
if (currentProfileId != profileId) {
loadFromDisk(profileId)
}
val operationGeneration = activeOperationGeneration(profileId) ?: run {
log.d { "Skipping clear and force watch progress refresh for inactive profile $profileId" }
return
}
metadataResolutionJob?.cancel()
clearLocalEntries()
lastSuccessfulPushEpochMs = 0L
deltaCursorEventId = 0L
deltaInitialized = false
publish()
persist()
if (shouldUseTraktProgress()) {
log.d { "Clearing local Trakt watch progress cache and force refreshing profile $profileId" }
if (source == WatchProgressSource.TRAKT) {
TraktProgressRepository.clearLocalState()
TraktProgressRepository.invalidateAndRefresh()
if (isActiveOperation(profileId, operationGeneration)) {
}
publish()
if (source == WatchProgressSource.NUVIO_SYNC) {
resolveRemoteMetadata()
}
}
internal suspend fun refreshForSource(
profileId: Int,
source: WatchProgressSource,
sourceChanged: Boolean,
force: Boolean,
): Boolean {
ensureLoaded()
if (currentProfileId != profileId) {
loadFromDisk(profileId)
}
val operationGeneration = activeOperationGeneration(profileId) ?: run {
log.d { "Skipping watch progress refresh for inactive profile $profileId" }
return false
}
activateSource(source)
return when (source) {
WatchProgressSource.TRAKT -> refreshTraktSource(
profileId = profileId,
operationGeneration = operationGeneration,
sourceChanged = sourceChanged,
force = force,
)
WatchProgressSource.NUVIO_SYNC -> refreshNuvioSource(
profileId = profileId,
operationGeneration = operationGeneration,
force = force,
)
}
}
private suspend fun refreshTraktSource(
profileId: Int,
operationGeneration: Long,
sourceChanged: Boolean,
force: Boolean,
): Boolean {
if (!TraktAuthRepository.isAuthenticated.value) {
log.d { "Skipping Trakt progress refresh because Trakt is not authenticated" }
return false
}
return try {
if (force || sourceChanged) {
TraktProgressRepository.invalidateAndRefresh()
} else {
TraktProgressRepository.refreshNow()
}
if (isActiveOperation(profileId, operationGeneration) && activeSource == WatchProgressSource.TRAKT) {
publish()
}
return
val state = TraktProgressRepository.uiState.value
state.hasLoadedRemoteProgress && state.errorMessage == null
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.e(error) { "Failed to refresh Trakt watch progress" }
false
}
}
private suspend fun refreshNuvioSource(
profileId: Int,
operationGeneration: Long,
force: Boolean,
): Boolean {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) {
log.d { "Cleared local watch progress but skipped remote refresh because Nuvio Sync is not authenticated" }
return
publish()
return true
}
if (isPullingNuvioSyncFromServer) {
log.d { "Cleared local watch progress but skipped remote refresh because a Nuvio sync pull is already running" }
return
}
isPullingNuvioSyncFromServer = true
try {
val pullStartedEpochMs = WatchProgressClock.nowEpochMs()
val cursorBeforeSnapshot = try {
syncAdapter.getDeltaCursor(profileId)
return nuvioPullMutex.withLock {
try {
val pullStartedEpochMs = WatchProgressClock.nowEpochMs()
if (force) {
pullNuvioSnapshotFromServer(
profileId = profileId,
pullStartedEpochMs = pullStartedEpochMs,
operationGeneration = operationGeneration,
)
} else {
pullSupabaseDeltaFromServer(
profileId = profileId,
pullStartedEpochMs = pullStartedEpochMs,
operationGeneration = operationGeneration,
)
}
true
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.w { "Watch progress delta cursor unavailable during clear refresh, falling back to snapshot reset: ${error.message}" }
null
log.e(error) { "Failed to refresh Nuvio watch progress" }
false
}
}
}
pullFullFromAdapter(
profileId = profileId,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = cursorBeforeSnapshot == null,
operationGeneration = operationGeneration,
preserveLocalEntries = false,
)
if (!isActiveOperation(profileId, operationGeneration)) return
private suspend fun pullNuvioSnapshotFromServer(
profileId: Int,
pullStartedEpochMs: Long,
operationGeneration: Long,
) {
val cursorBeforeSnapshot = try {
syncAdapter.getDeltaCursor(profileId)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.w { "Watch progress cursor unavailable during snapshot refresh: ${error.message}" }
null
}
if (cursorBeforeSnapshot != null) {
deltaCursorEventId = cursorBeforeSnapshot
deltaInitialized = true
persist()
}
} finally {
isPullingNuvioSyncFromServer = false
pullFullFromAdapter(
profileId = profileId,
pullStartedEpochMs = pullStartedEpochMs,
resetDeltaState = cursorBeforeSnapshot == null,
operationGeneration = operationGeneration,
preserveLocalEntries = true,
)
if (!isActiveOperation(profileId, operationGeneration)) return
if (cursorBeforeSnapshot != null) {
deltaCursorEventId = cursorBeforeSnapshot
deltaInitialized = true
persist()
}
}
@ -694,13 +669,14 @@ object WatchProgressRepository {
return merged
}
private fun shouldPreserveLocalWatchProgressEntry(
internal fun shouldPreserveLocalWatchProgressEntry(
localEntry: WatchProgressEntry,
lastSuccessfulPushEpochMs: Long,
pullStartedEpochMs: Long,
): Boolean {
val updatedAt = localEntry.lastUpdatedEpochMs
val wasUpdatedAfterLastPush = lastSuccessfulPushEpochMs > 0L && updatedAt > lastSuccessfulPushEpochMs
val wasUpdatedAfterLastPush =
lastSuccessfulPushEpochMs <= 0L || updatedAt > lastSuccessfulPushEpochMs
val wasUpdatedDuringPull = pullStartedEpochMs > 0L && updatedAt >= pullStartedEpochMs
return wasUpdatedAfterLastPush || wasUpdatedDuringPull
}
@ -1101,10 +1077,15 @@ object WatchProgressRepository {
}
private fun pushScrobbleToServer(entry: WatchProgressEntry, profileId: Int) {
syncScope.launch {
val operationGeneration = profileGeneration.takeIf { profileId == currentProfileId }
accountScopeSnapshot().launch {
runCatching {
syncAdapter.push(profileId = profileId, entries = listOf(entry))
recordSuccessfulPush(profileId = profileId, entries = listOf(entry))
recordSuccessfulPush(
profileId = profileId,
operationGeneration = operationGeneration,
entries = listOf(entry),
)
}.onFailure { e ->
log.e(e) { "Failed to push watch progress scrobble" }
}
@ -1114,7 +1095,7 @@ object WatchProgressRepository {
private fun pushDeleteToServer(entries: Collection<WatchProgressEntry>) {
if (shouldUseTraktProgress()) return
val profileId = currentProfileId
syncScope.launch {
accountScopeSnapshot().launch {
runCatching {
if (entries.isEmpty()) return@runCatching
syncAdapter.delete(profileId = profileId, entries = entries)
@ -1150,8 +1131,12 @@ object WatchProgressRepository {
)
}
private fun recordSuccessfulPush(profileId: Int, entries: Collection<WatchProgressEntry>) {
if (profileId != currentProfileId) return
private fun recordSuccessfulPush(
profileId: Int,
operationGeneration: Long?,
entries: Collection<WatchProgressEntry>,
) {
if (profileId != currentProfileId || operationGeneration != profileGeneration) return
val latestPushed = entries
.asSequence()
.map { entry -> entry.lastUpdatedEpochMs }
@ -1163,10 +1148,16 @@ object WatchProgressRepository {
}
private fun shouldUseTraktProgress(): Boolean =
shouldUseTraktProgressSource(
isAuthenticated = TraktAuthRepository.isAuthenticated.value,
source = TraktSettingsRepository.uiState.value.watchProgressSource,
)
activeSource == WatchProgressSource.TRAKT
private fun accountScopeSnapshot(): CoroutineScope = synchronized(accountScopeLock) {
accountScope
}
private fun updateActiveSource(source: WatchProgressSource) {
activeSource = source
_activeSourceState.value = source
}
private fun WatchProgressEntry.shouldAttemptTraktPlaybackDelete(): Boolean =
isTraktCompatibleId(parentMetaId)
@ -1225,19 +1216,6 @@ object WatchProgressRepository {
}
}
private fun removeLocalEntriesMatching(predicate: (WatchProgressEntry) -> Boolean): Boolean =
synchronized(entriesLock) {
val filteredEntries = entriesByVideoId
.filterValues { entry -> !predicate(entry) }
.toMutableMap()
if (filteredEntries.size == entriesByVideoId.size) {
false
} else {
entriesByVideoId = filteredEntries
true
}
}
private fun replaceLocalEntries(entries: Collection<WatchProgressEntry>) {
synchronized(entriesLock) {
entriesByVideoId = entries

485
composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressSourceCoordinator.kt Normal file
View file

@ -0,0 +1,485 @@
package com.nuvio.app.features.watchprogress
import co.touchlab.kermit.Logger
import com.nuvio.app.core.auth.AuthRepository
import com.nuvio.app.core.auth.AuthState
import com.nuvio.app.features.profiles.ProfileRepository
import com.nuvio.app.features.trakt.DEFAULT_WATCH_PROGRESS_SOURCE
import com.nuvio.app.features.trakt.TraktAuthRepository
import com.nuvio.app.features.trakt.TraktSettingsRepository
import com.nuvio.app.features.trakt.WatchProgressSource
import com.nuvio.app.features.trakt.effectiveWatchProgressSource
import com.nuvio.app.features.watched.WatchedRepository
import kotlinx.atomicfu.atomic
import kotlinx.atomicfu.locks.SynchronizedObject
import kotlinx.atomicfu.locks.synchronized
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
data class WatchProgressSourceTransitionState(
val profileId: Int? = null,
val requestedSource: WatchProgressSource = DEFAULT_WATCH_PROGRESS_SOURCE,
val effectiveSource: WatchProgressSource = WatchProgressSource.NUVIO_SYNC,
val isRefreshing: Boolean = false,
val lastRefreshSucceeded: Boolean? = null,
)
data class WatchProgressSourceTransitionResult(
val requestedSource: WatchProgressSource,
val effectiveSource: WatchProgressSource,
val progressRefreshed: Boolean,
val watchedHistoryRefreshed: Boolean,
) {
val succeeded: Boolean
get() = progressRefreshed && watchedHistoryRefreshed
}
internal data class WatchProgressSourceContext(
val profileId: Int,
val requestedSource: WatchProgressSource,
val effectiveSource: WatchProgressSource,
val isNuvioAuthenticated: Boolean,
)
internal fun resolveSerializedWatchProgressContext(
queuedContext: WatchProgressSourceContext,
currentContext: WatchProgressSourceContext,
): WatchProgressSourceContext? = currentContext.takeIf {
it.profileId == queuedContext.profileId
}
internal class WatchProgressSourceTransitionRunner(
private val currentAppliedSource: () -> WatchProgressSource? = { null },
private val invalidateCache: (profileId: Int, source: WatchProgressSource) -> Unit,
private val activateProgressSource: (WatchProgressSource) -> Unit,
private val activateWatchedSource: (WatchProgressSource) -> Unit,
private val refreshProgress: suspend (
profileId: Int,
source: WatchProgressSource,
sourceChanged: Boolean,
force: Boolean,
) -> Boolean,
private val refreshWatched: suspend (
profileId: Int,
source: WatchProgressSource,
force: Boolean,
) -> Boolean,
) {
private val transitionMutex = Mutex()
private val lifecycleLock = SynchronizedObject()
private var lifecycleGeneration: Long = 0L
private var lastAppliedContext: WatchProgressSourceContext? = null
fun currentGeneration(): Long = synchronized(lifecycleLock) {
lifecycleGeneration
}
suspend fun transition(
context: WatchProgressSourceContext,
refreshIfUnchanged: Boolean,
forceSnapshot: Boolean,
transitionGeneration: Long = currentGeneration(),
): WatchProgressSourceTransitionResult = transitionMutex.withLock {
val (previousContext, sourceChanged) = synchronized(lifecycleLock) {
ensureCurrentGeneration(transitionGeneration)
val previous = lastAppliedContext
val currentSourceDiffers = currentAppliedSource()
?.let { appliedSource -> appliedSource != context.effectiveSource }
?: false
val lastSuccessfulContextDiffers = previous?.let { successfulContext ->
successfulContext.profileId != context.profileId ||
successfulContext.effectiveSource != context.effectiveSource
} ?: false
val changed = currentSourceDiffers || lastSuccessfulContextDiffers
if (changed) {
invalidateCache(context.profileId, context.effectiveSource)
}
activateWatchedSource(context.effectiveSource)
activateProgressSource(context.effectiveSource)
previous to changed
}
if (!refreshIfUnchanged && !sourceChanged && previousContext == context) {
return@withLock WatchProgressSourceTransitionResult(
requestedSource = context.requestedSource,
effectiveSource = context.effectiveSource,
progressRefreshed = true,
watchedHistoryRefreshed = true,
)
}
val (progressRefreshed, watchedHistoryRefreshed) = coroutineScope {
val progress = synchronized(lifecycleLock) {
ensureCurrentGeneration(transitionGeneration)
async(start = CoroutineStart.UNDISPATCHED) {
runRefresh {
refreshProgress(
context.profileId,
context.effectiveSource,
sourceChanged,
forceSnapshot,
)
}
}
}
val watched = synchronized(lifecycleLock) {
ensureCurrentGeneration(transitionGeneration)
async(start = CoroutineStart.UNDISPATCHED) {
runRefresh {
refreshWatched(
context.profileId,
context.effectiveSource,
forceSnapshot,
)
}
}
}
progress.await() to watched.await()
}
if (progressRefreshed && watchedHistoryRefreshed) {
synchronized(lifecycleLock) {
ensureCurrentGeneration(transitionGeneration)
lastAppliedContext = context
}
} else {
synchronized(lifecycleLock) {
ensureCurrentGeneration(transitionGeneration)
}
}
WatchProgressSourceTransitionResult(
requestedSource = context.requestedSource,
effectiveSource = context.effectiveSource,
progressRefreshed = progressRefreshed,
watchedHistoryRefreshed = watchedHistoryRefreshed,
)
}
fun reset() = synchronized(lifecycleLock) {
lifecycleGeneration += 1L
lastAppliedContext = null
}
private fun ensureCurrentGeneration(expectedGeneration: Long) {
if (expectedGeneration != lifecycleGeneration) {
throw CancellationException("Watch progress source transition belongs to a cleared account")
}
}
private suspend fun runRefresh(block: suspend () -> Boolean): Boolean =
try {
block()
} catch (error: CancellationException) {
throw error
} catch (_: Throwable) {
false
}
}
object WatchProgressSourceCoordinator {
private val log = Logger.withTag("ProgressSourceCoordinator")
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val startLock = SynchronizedObject()
private val transitionStateMutex = Mutex()
private var observeJob: Job? = null
private var lifecycleGeneration: Long = 0L
private val automaticTransitionPauseCount = atomic(0)
private val _uiState = MutableStateFlow(WatchProgressSourceTransitionState())
val uiState: StateFlow<WatchProgressSourceTransitionState> = _uiState.asStateFlow()
private val runner = WatchProgressSourceTransitionRunner(
currentAppliedSource = { WatchProgressRepository.activeSourceState.value },
invalidateCache = ContinueWatchingEnrichmentCache::invalidate,
activateProgressSource = WatchProgressRepository::activateSource,
activateWatchedSource = { source -> WatchedRepository.activateSource(source) },
refreshProgress = WatchProgressRepository::refreshForSource,
refreshWatched = WatchedRepository::refreshForSource,
)
fun ensureStarted() {
val expectedGeneration = synchronized(startLock) { lifecycleGeneration }
ensureStartedForGeneration(expectedGeneration)
}
private fun ensureStartedForGeneration(expectedGeneration: Long) {
val generationIsCurrent = synchronized(startLock) {
expectedGeneration == lifecycleGeneration
}
if (!generationIsCurrent) return
ensureSourceStateLoaded()
synchronized(startLock) {
if (expectedGeneration != lifecycleGeneration) return
if (observeJob?.isActive == true) return
observeJob = scope.launch {
combine(
TraktSettingsRepository.uiState,
TraktAuthRepository.isAuthenticated,
AuthRepository.state,
ProfileRepository.state,
) { settings, isTraktAuthenticated, authState, profileState ->
buildContext(
profileId = profileState.activeProfile?.profileIndex
?: ProfileRepository.activeProfileId,
requestedSource = settings.watchProgressSource,
isTraktAuthenticated = isTraktAuthenticated,
authState = authState,
)
}
.distinctUntilChanged()
.collectLatest { context ->
if (automaticTransitionPauseCount.value > 0) return@collectLatest
runTransition(
context = context,
refreshIfUnchanged = false,
forceSnapshot = true,
)
}
}
}
}
private fun ensureSourceStateLoaded() {
TraktAuthRepository.ensureLoaded()
TraktSettingsRepository.ensureLoaded()
}
suspend fun selectSource(
profileId: Int,
source: WatchProgressSource,
): WatchProgressSourceTransitionResult {
val operationGeneration = synchronized(startLock) { lifecycleGeneration }
ensureSourceStateLoadedForGeneration(operationGeneration)
synchronized(startLock) {
ensureCoordinatorGeneration(operationGeneration)
TraktSettingsRepository.setWatchProgressSource(source, profileId)
}
val context = currentContext(profileId)
return try {
runTransition(
context = context,
refreshIfUnchanged = false,
forceSnapshot = true,
expectedCoordinatorGeneration = operationGeneration,
)
} finally {
ensureStartedForGeneration(operationGeneration)
}
}
suspend fun refreshActiveSource(
profileId: Int,
force: Boolean = true,
): WatchProgressSourceTransitionResult {
val operationGeneration = synchronized(startLock) { lifecycleGeneration }
ensureSourceStateLoadedForGeneration(operationGeneration)
val context = currentContext(profileId)
return try {
runTransition(
context = context,
refreshIfUnchanged = true,
forceSnapshot = force,
expectedCoordinatorGeneration = operationGeneration,
)
} finally {
ensureStartedForGeneration(operationGeneration)
}
}
private fun ensureSourceStateLoadedForGeneration(expectedGeneration: Long) {
synchronized(startLock) {
ensureCoordinatorGeneration(expectedGeneration)
}
ensureSourceStateLoaded()
synchronized(startLock) {
ensureCoordinatorGeneration(expectedGeneration)
}
}
private fun ensureCoordinatorGeneration(expectedGeneration: Long) {
if (expectedGeneration != lifecycleGeneration) {
throw CancellationException("Watch progress source operation belongs to a cleared account")
}
}
fun clearLocalState() {
synchronized(startLock) {
observeJob?.cancel()
observeJob = null
lifecycleGeneration += 1L
runner.reset()
automaticTransitionPauseCount.value = 0
_uiState.value = WatchProgressSourceTransitionState()
}
}
internal fun pauseAutomaticTransitions() {
automaticTransitionPauseCount.incrementAndGet()
}
internal fun resumeAutomaticTransitions() {
while (true) {
val current = automaticTransitionPauseCount.value
if (current == 0) return
if (automaticTransitionPauseCount.compareAndSet(current, current - 1)) {
if (current == 1) {
scope.launch {
val profileId = ProfileRepository.state.value.activeProfile?.profileIndex
?: ProfileRepository.activeProfileId
try {
runTransition(
context = currentContext(profileId),
refreshIfUnchanged = false,
forceSnapshot = true,
)
} catch (error: CancellationException) {
throw error
} catch (error: Throwable) {
log.e(error) { "Failed to replay source transition after profile sync" }
}
}
}
return
}
}
}
private suspend fun runTransition(
context: WatchProgressSourceContext,
refreshIfUnchanged: Boolean,
forceSnapshot: Boolean,
expectedCoordinatorGeneration: Long? = null,
): WatchProgressSourceTransitionResult {
currentCoroutineContext().ensureActive()
val transitionToken = synchronized(startLock) {
if (
expectedCoordinatorGeneration != null &&
expectedCoordinatorGeneration != lifecycleGeneration
) {
throw CancellationException("Watch progress source operation belongs to a cleared account")
}
TransitionToken(
coordinatorGeneration = lifecycleGeneration,
runnerGeneration = runner.currentGeneration(),
)
}
return transitionStateMutex.withLock {
currentCoroutineContext().ensureActive()
val activeProfileId = ProfileRepository.state.value.activeProfile?.profileIndex
?: ProfileRepository.activeProfileId
val resolvedContext = resolveSerializedWatchProgressContext(
queuedContext = context,
currentContext = currentContext(activeProfileId),
) ?: throw CancellationException("Watch progress source transition belongs to an inactive profile")
val started = synchronized(startLock) {
if (transitionToken.coordinatorGeneration != lifecycleGeneration) {
false
} else {
_uiState.value = WatchProgressSourceTransitionState(
profileId = resolvedContext.profileId,
requestedSource = resolvedContext.requestedSource,
effectiveSource = resolvedContext.effectiveSource,
isRefreshing = true,
lastRefreshSucceeded = null,
)
true
}
}
if (!started) {
throw CancellationException("Watch progress source transition belongs to a cleared account")
}
val result = try {
runner.transition(
context = resolvedContext,
refreshIfUnchanged = refreshIfUnchanged,
forceSnapshot = forceSnapshot,
transitionGeneration = transitionToken.runnerGeneration,
)
} catch (error: Throwable) {
synchronized(startLock) {
if (transitionToken.coordinatorGeneration == lifecycleGeneration) {
_uiState.value = _uiState.value.copy(
isRefreshing = false,
lastRefreshSucceeded = false,
)
}
}
throw error
}
val completed = synchronized(startLock) {
if (transitionToken.coordinatorGeneration != lifecycleGeneration) {
false
} else {
_uiState.value = _uiState.value.copy(
profileId = resolvedContext.profileId,
requestedSource = result.requestedSource,
effectiveSource = result.effectiveSource,
isRefreshing = false,
lastRefreshSucceeded = result.succeeded,
)
true
}
}
if (!completed) {
throw CancellationException("Watch progress source transition belongs to a cleared account")
}
if (!result.succeeded) {
log.w {
"Source refresh incomplete for profile ${resolvedContext.profileId}: " +
"source=${resolvedContext.effectiveSource} progress=${result.progressRefreshed} " +
"watched=${result.watchedHistoryRefreshed}"
}
}
result
}
}
private data class TransitionToken(
val coordinatorGeneration: Long,
val runnerGeneration: Long,
)
private fun buildContext(
profileId: Int,
requestedSource: WatchProgressSource,
isTraktAuthenticated: Boolean,
authState: AuthState,
): WatchProgressSourceContext = WatchProgressSourceContext(
profileId = profileId,
requestedSource = requestedSource,
effectiveSource = effectiveWatchProgressSource(
isTraktAuthenticated = isTraktAuthenticated,
requestedSource = requestedSource,
),
isNuvioAuthenticated = authState is AuthState.Authenticated && !authState.isAnonymous,
)
private fun currentContext(profileId: Int): WatchProgressSourceContext = buildContext(
profileId = profileId,
requestedSource = TraktSettingsRepository.uiState.value.watchProgressSource,
isTraktAuthenticated = TraktAuthRepository.isAuthenticated.value,
authState = AuthRepository.state.value,
)
}

190
composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt Normal file
View file

@ -0,0 +1,190 @@
package com.nuvio.app.core.sync
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.yield
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
class SyncManagerTest {
@Test
fun `forced foreground recovery queues behind an active profile sync`() {
assertFalse(shouldQueueCoalescedForegroundPull(force = false))
assertTrue(shouldQueueCoalescedForegroundPull(force = true))
}
@Test
fun `source prerequisites finish before source dependent pulls`() = runBlocking {
val events = mutableListOf<String>()
var profileSettingsApplied = false
var traktCredentialsApplied = false
runOrderedProfileSync(
profileId = 7,
pluginsEnabled = true,
operations = ProfileSyncOperations(
pullAddons = { events += "addons" },
pullPlugins = { events += "plugins" },
pullProfileSettings = {
events += "settings:start"
yield()
profileSettingsApplied = true
events += "settings:end"
},
pullTraktCredentials = {
events += "credentials:start"
yield()
traktCredentialsApplied = true
events += "credentials:end"
},
pullLibrary = {
assertTrue(profileSettingsApplied)
assertTrue(traktCredentialsApplied)
events += "library"
},
refreshActiveWatchSource = {
assertTrue(profileSettingsApplied)
assertTrue(traktCredentialsApplied)
events += "active-watch-source"
},
pullCollections = { events += "collections" },
pullHomeCatalogSettings = { events += "home-settings" },
),
onFailure = { _, error -> throw error },
)
val lastPrerequisite = maxOf(
events.indexOf("settings:end"),
events.indexOf("credentials:end"),
)
assertTrue(events.indexOf("library") > lastPrerequisite)
assertTrue(events.indexOf("active-watch-source") > lastPrerequisite)
assertEquals(1, events.count { it == "active-watch-source" })
}
@Test
fun `disabled plugins are skipped without changing sync ordering`() = runBlocking {
val events = mutableListOf<String>()
runOrderedProfileSync(
profileId = 2,
pluginsEnabled = false,
operations = recordingOperations(events),
onFailure = { _, error -> throw error },
)
assertTrue("plugins" !in events)
assertTrue(events.indexOf("settings") < events.indexOf("library"))
assertTrue(events.indexOf("credentials") < events.indexOf("active-watch-source"))
}
@Test
fun `duplicate active request for one profile is coalesced`() = runBlocking {
val gate = ProfileSyncRequestGate()
val firstStarted = CompletableDeferred<Unit>()
val releaseFirst = CompletableDeferred<Unit>()
var runCount = 0
val first = gate.launch(this, profileId = 4) {
runCount += 1
firstStarted.complete(Unit)
releaseFirst.await()
}
firstStarted.await()
val duplicate = gate.launch(this, profileId = 4) {
runCount += 1
}
assertEquals(ProfileSyncRequestResult.Started, first)
assertEquals(ProfileSyncRequestResult.Coalesced, duplicate)
assertEquals(1, runCount)
releaseFirst.complete(Unit)
yield()
gate.cancel()
}
@Test
fun `new profile replaces stale in flight request`() = runBlocking {
val gate = ProfileSyncRequestGate()
val firstStarted = CompletableDeferred<Unit>()
val firstCancelled = CompletableDeferred<Unit>()
val secondCompleted = CompletableDeferred<Unit>()
gate.launch(this, profileId = 1) {
firstStarted.complete(Unit)
try {
CompletableDeferred<Unit>().await()
} finally {
firstCancelled.complete(Unit)
}
}
firstStarted.await()
val replacement = gate.launch(this, profileId = 2) {
secondCompleted.complete(Unit)
}
assertEquals(ProfileSyncRequestResult.Replaced, replacement)
firstCancelled.await()
secondCompleted.await()
gate.cancel()
}
@Test
fun `failed step is reported by ordered sync result`() = runBlocking {
val result = runOrderedProfileSync(
profileId = 3,
pluginsEnabled = false,
operations = recordingOperations(mutableListOf()).copy(
refreshActiveWatchSource = { error("source refresh failed") },
),
)
assertFalse(result.succeeded)
assertEquals(setOf(ProfileSyncStep.ActiveWatchSource), result.failedSteps)
}
@Test
fun `realtime invalidation queued during active sync runs once afterwards`() = runBlocking {
val gate = ProfileSyncRequestGate()
val firstStarted = CompletableDeferred<Unit>()
val releaseFirst = CompletableDeferred<Unit>()
val replayCompleted = CompletableDeferred<Unit>()
var runCount = 0
gate.launch(this, profileId = 1) {
runCount += 1
firstStarted.complete(Unit)
releaseFirst.await()
}
firstStarted.await()
val queued = gate.launch(this, profileId = 1, queueIfCoalesced = true) {
runCount += 1
replayCompleted.complete(Unit)
}
assertEquals(ProfileSyncRequestResult.Coalesced, queued)
releaseFirst.complete(Unit)
replayCompleted.await()
assertEquals(2, runCount)
gate.cancel()
}
private fun recordingOperations(events: MutableList<String>): ProfileSyncOperations =
ProfileSyncOperations(
pullAddons = { events += "addons" },
pullPlugins = { events += "plugins" },
pullProfileSettings = { events += "settings" },
pullTraktCredentials = { events += "credentials" },
pullLibrary = { events += "library" },
refreshActiveWatchSource = { events += "active-watch-source" },
pullCollections = { events += "collections" },
pullHomeCatalogSettings = { events += "home-settings" },
)
}

13
composeApp/src/commonTest/kotlin/com/nuvio/app/features/home/HomeScreenTest.kt
View file

@ -11,6 +11,7 @@ import com.nuvio.app.features.watchprogress.ContinueWatchingItem
import com.nuvio.app.features.watchprogress.WatchProgressEntry
import com.nuvio.app.features.watched.WatchedItem
import com.nuvio.app.features.trakt.TRAKT_CONTINUE_WATCHING_DAYS_CAP_ALL
import com.nuvio.app.features.trakt.WatchProgressSource
import com.nuvio.app.features.watching.domain.WatchingContentRef
import kotlin.test.Test
import kotlin.test.assertEquals
@ -18,6 +19,18 @@ import kotlin.test.assertTrue
class HomeScreenTest {
@Test
fun `continue watching cache uses the effective progress source`() {
assertEquals(
WatchProgressSource.TRAKT,
effectiveContinueWatchingCacheSource(isTraktProgressActive = true),
)
assertEquals(
WatchProgressSource.NUVIO_SYNC,
effectiveContinueWatchingCacheSource(isTraktProgressActive = false),
)
}
@Test
fun `home trakt continue watching candidate limits match TV`() {
assertEquals(300, HomeContinueWatchingMaxRecentProgressItems)

52
composeApp/src/commonTest/kotlin/com/nuvio/app/features/trakt/TraktSettingsRepositoryTest.kt
View file

@ -4,10 +4,37 @@ import com.nuvio.app.features.library.LibrarySourceMode
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNull
import kotlin.test.assertTrue
class TraktSettingsRepositoryTest {
@Test
fun `watch source outbox survives restart and a stale push cannot clear a newer choice`() {
val disk = mutableMapOf<Int, String>()
fun newOutbox() = WatchProgressSourceSettingsOutbox(
loadPayload = disk::get,
savePayload = disk::set,
clearPayload = disk::remove,
)
val traktChoice = PendingWatchProgressSourceChange(
accountId = "account-a",
profileId = 2,
source = WatchProgressSource.TRAKT,
)
val nuvioChoice = traktChoice.copy(source = WatchProgressSource.NUVIO_SYNC)
newOutbox().record(traktChoice)
val afterProcessRestart = newOutbox()
assertEquals(traktChoice, afterProcessRestart.pendingFor("account-a", 2))
afterProcessRestart.record(nuvioChoice)
assertFalse(afterProcessRestart.clearIfMatches(traktChoice))
assertEquals(nuvioChoice, afterProcessRestart.pendingFor("account-a", 2))
assertTrue(afterProcessRestart.clearIfMatches(nuvioChoice))
assertNull(afterProcessRestart.pendingFor("account-a", 2))
}
@Test
fun `watch progress source defaults to Trakt for unset or invalid storage`() {
assertEquals(WatchProgressSource.TRAKT, WatchProgressSource.fromStorage(null))
@ -62,6 +89,31 @@ class TraktSettingsRepositoryTest {
assertTrue(shouldUseTraktProgress(isAuthenticated = true, source = WatchProgressSource.TRAKT))
}
@Test
fun `effective progress source falls back to Nuvio when Trakt is unavailable`() {
assertEquals(
WatchProgressSource.NUVIO_SYNC,
effectiveWatchProgressSource(
isTraktAuthenticated = false,
requestedSource = WatchProgressSource.TRAKT,
),
)
assertEquals(
WatchProgressSource.NUVIO_SYNC,
effectiveWatchProgressSource(
isTraktAuthenticated = true,
requestedSource = WatchProgressSource.NUVIO_SYNC,
),
)
assertEquals(
WatchProgressSource.TRAKT,
effectiveWatchProgressSource(
isTraktAuthenticated = true,
requestedSource = WatchProgressSource.TRAKT,
),
)
}
@Test
fun `effective library source uses Trakt only when authenticated and selected`() {
assertEquals(

145
composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedRepositoryTest.kt
View file

@ -2,6 +2,7 @@ package com.nuvio.app.features.watched
import com.nuvio.app.features.details.MetaDetails
import com.nuvio.app.features.details.MetaVideo
import com.nuvio.app.features.trakt.WatchProgressSource
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
@ -123,6 +124,35 @@ class WatchedRepositoryTest {
assertTrue(merged.isEmpty())
}
@Test
fun mergeWatchedItemsPreservingUnsynced_keeps_local_only_item_before_first_push() {
val localOnlyItem = watchedItem(id = "local-only", markedAtEpochMs = 1_000L)
val merged = mergeWatchedItemsPreservingUnsynced(
serverItems = emptyList(),
localItems = listOf(localOnlyItem),
lastSuccessfulPushEpochMs = 0L,
pullStartedEpochMs = 2_000L,
)
assertEquals(listOf(localOnlyItem), merged.values.toList())
}
@Test
fun traktSnapshot_drops_old_transient_items_missingFromRemote() {
val oldTraktItem = watchedItem(id = "old-trakt", markedAtEpochMs = 1_000L)
val merged = mergeWatchedItemsPreservingUnsynced(
serverItems = emptyList(),
localItems = listOf(oldTraktItem),
lastSuccessfulPushEpochMs = 0L,
pullStartedEpochMs = 2_000L,
preserveWhenNoSuccessfulPush = false,
)
assertTrue(merged.isEmpty())
}
@Test
fun playbackCompletionWatchedMarks_doNotMirrorToTraktHistory() {
assertFalse(
@ -144,4 +174,119 @@ class WatchedRepositoryTest {
),
)
}
@Test
fun watchedItemsForSource_keepsNuvioAndTraktSnapshotsIsolated() {
val nuvioItem = watchedItem(id = "nuvio", markedAtEpochMs = 1_000L)
val traktItem = watchedItem(id = "trakt", markedAtEpochMs = 2_000L)
assertEquals(
listOf(nuvioItem),
watchedItemsForSource(
source = WatchProgressSource.NUVIO_SYNC,
nuvioItems = listOf(nuvioItem),
traktItems = listOf(traktItem),
),
)
assertEquals(
listOf(traktItem),
watchedItemsForSource(
source = WatchProgressSource.TRAKT,
nuvioItems = listOf(nuvioItem),
traktItems = listOf(traktItem),
),
)
}
@Test
fun onlyNuvioWatchedStateIsPersisted() {
assertTrue(shouldPersistWatchedSource(WatchProgressSource.NUVIO_SYNC))
assertFalse(shouldPersistWatchedSource(WatchProgressSource.TRAKT))
}
@Test
fun replacingTraktSnapshot_doesNotOverwriteNuvioSnapshot() {
val nuvioItem = watchedItem(id = "nuvio", markedAtEpochMs = 1_000L)
val previousTraktItem = watchedItem(id = "old-trakt", markedAtEpochMs = 2_000L)
val refreshedTraktItem = watchedItem(id = "new-trakt", markedAtEpochMs = 3_000L)
val nuvioItems = mutableMapOf("nuvio" to nuvioItem)
val traktItems = mutableMapOf("old-trakt" to previousTraktItem)
replaceWatchedItemsForSource(
source = WatchProgressSource.TRAKT,
nuvioItems = nuvioItems,
traktItems = traktItems,
replacement = mapOf("new-trakt" to refreshedTraktItem),
)
assertEquals(mapOf("nuvio" to nuvioItem), nuvioItems)
assertEquals(mapOf("new-trakt" to refreshedTraktItem), traktItems)
}
@Test
fun effectiveWatchedSource_fallsBackToNuvioWhenTraktIsUnavailable() {
assertEquals(
WatchProgressSource.NUVIO_SYNC,
effectiveWatchedSource(
requestedSource = WatchProgressSource.TRAKT,
isTraktAuthenticated = false,
),
)
assertEquals(
WatchProgressSource.TRAKT,
effectiveWatchedSource(
requestedSource = WatchProgressSource.TRAKT,
isTraktAuthenticated = true,
),
)
}
@Test
fun sourceOperationGuard_rejectsResultFromSourceActiveBeforeSwitch() {
val traktOperation = WatchedSourceOperation(
source = WatchProgressSource.TRAKT,
generation = 4L,
)
assertTrue(
isWatchedSourceOperationCurrent(
operation = traktOperation,
activeSource = WatchProgressSource.TRAKT,
activeGeneration = 4L,
),
)
assertFalse(
isWatchedSourceOperationCurrent(
operation = traktOperation,
activeSource = WatchProgressSource.NUVIO_SYNC,
activeGeneration = 5L,
),
)
}
@Test
fun sourceOperationGuard_rejectsOlderRefreshAfterSwitchingAwayAndBack() {
val firstTraktOperation = WatchedSourceOperation(
source = WatchProgressSource.TRAKT,
generation = 7L,
)
assertFalse(
isWatchedSourceOperationCurrent(
operation = firstTraktOperation,
activeSource = WatchProgressSource.TRAKT,
activeGeneration = 9L,
),
)
}
private fun watchedItem(
id: String,
markedAtEpochMs: Long,
): WatchedItem = WatchedItem(
id = id,
type = "movie",
name = id,
markedAtEpochMs = markedAtEpochMs,
)
}

78
composeApp/src/commonTest/kotlin/com/nuvio/app/features/watchprogress/ContinueWatchingEnrichmentCacheTest.kt Normal file
View file

@ -0,0 +1,78 @@
package com.nuvio.app.features.watchprogress
import com.nuvio.app.features.trakt.WatchProgressSource
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNotEquals
import kotlin.test.assertTrue
class ContinueWatchingEnrichmentCacheTest {
@Test
fun `storage keys are scoped by profile and effective source`() {
val traktKey = ContinueWatchingEnrichmentCache.continueWatchingEnrichmentStorageKey(
profileId = 2,
source = WatchProgressSource.TRAKT,
)
val nuvioKey = ContinueWatchingEnrichmentCache.continueWatchingEnrichmentStorageKey(
profileId = 2,
source = WatchProgressSource.NUVIO_SYNC,
)
assertEquals("cw_enrichment_cache_trakt_2", traktKey)
assertEquals("cw_enrichment_cache_nuvio_sync_2", nuvioKey)
assertNotEquals(traktKey, nuvioKey)
assertEquals("cw_enrichment_cache_2", ContinueWatchingEnrichmentCache.legacyStorageKey(profileId = 2))
}
@Test
fun `stale resolver generation cannot write snapshots after invalidation`() {
val profileId = 4
val staleGeneration = ContinueWatchingEnrichmentCache.generation.value
ContinueWatchingEnrichmentCache.invalidate(
profileId = profileId,
source = WatchProgressSource.TRAKT,
)
assertFalse(
ContinueWatchingEnrichmentCache.saveSnapshots(
profileId = profileId,
source = WatchProgressSource.TRAKT,
generation = staleGeneration,
nextUp = emptyList(),
inProgress = emptyList(),
),
)
assertTrue(
ContinueWatchingEnrichmentCache.saveSnapshots(
profileId = profileId,
source = WatchProgressSource.TRAKT,
generation = ContinueWatchingEnrichmentCache.generation.value,
nextUp = emptyList(),
inProgress = emptyList(),
),
)
ContinueWatchingEnrichmentCache.clearAll(profileId)
}
@Test
fun `account clear invalidates resolver writes from the previous account`() {
val staleGeneration = ContinueWatchingEnrichmentCache.generation.value
ContinueWatchingEnrichmentCache.clearLocalState()
assertFalse(
ContinueWatchingEnrichmentCache.saveSnapshots(
profileId = 1,
source = WatchProgressSource.NUVIO_SYNC,
generation = staleGeneration,
nextUp = emptyList(),
inProgress = emptyList(),
),
)
ContinueWatchingEnrichmentCache.clearAll(profileId = 1)
}
}

11
composeApp/src/commonTest/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRulesTest.kt
View file

@ -316,6 +316,17 @@ class WatchProgressRulesTest {
assertEquals("movie", buildPlaybackVideoId(parentMetaId = "movie", seasonNumber = null, episodeNumber = null, fallbackVideoId = null))
}
@Test
fun `snapshot preserves local only progress before first successful push`() {
assertTrue(
WatchProgressRepository.shouldPreserveLocalWatchProgressEntry(
localEntry = entry(videoId = "local-only", lastUpdatedEpochMs = 1_000L),
lastSuccessfulPushEpochMs = 0L,
pullStartedEpochMs = 2_000L,
),
)
}
@Test
fun `up next continue watching uses actual episode id when available`() {
val item = entry(

347
composeApp/src/commonTest/kotlin/com/nuvio/app/features/watchprogress/WatchProgressSourceCoordinatorTest.kt Normal file
View file

@ -0,0 +1,347 @@
package com.nuvio.app.features.watchprogress
import com.nuvio.app.features.trakt.WatchProgressSource
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertFailsWith
import kotlin.test.assertTrue
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking
class WatchProgressSourceCoordinatorTest {
@Test
fun `serialized transition uses latest source for the same profile`() {
val queued = context(source = WatchProgressSource.NUVIO_SYNC)
val latest = context(source = WatchProgressSource.TRAKT)
assertEquals(
latest,
resolveSerializedWatchProgressContext(
queuedContext = queued,
currentContext = latest,
),
)
assertEquals(
null,
resolveSerializedWatchProgressContext(
queuedContext = queued,
currentContext = latest.copy(profileId = 2),
),
)
}
@Test
fun `source transition invalidates target cache and refreshes both read models`() = runBlocking {
val invalidations = mutableListOf<Pair<Int, WatchProgressSource>>()
val activatedProgress = mutableListOf<WatchProgressSource>()
val activatedWatched = mutableListOf<WatchProgressSource>()
val progressRefreshes = mutableListOf<RefreshCall>()
val watchedRefreshes = mutableListOf<RefreshCall>()
val runner = WatchProgressSourceTransitionRunner(
invalidateCache = { profileId, source -> invalidations += profileId to source },
activateProgressSource = activatedProgress::add,
activateWatchedSource = activatedWatched::add,
refreshProgress = { profileId, source, sourceChanged, force ->
progressRefreshes += RefreshCall(profileId, source, sourceChanged, force)
true
},
refreshWatched = { profileId, source, force ->
watchedRefreshes += RefreshCall(profileId, source, sourceChanged = false, force = force)
true
},
)
runner.transition(
context = context(source = WatchProgressSource.NUVIO_SYNC),
refreshIfUnchanged = false,
forceSnapshot = false,
)
invalidations.clear()
activatedProgress.clear()
activatedWatched.clear()
progressRefreshes.clear()
watchedRefreshes.clear()
val result = runner.transition(
context = context(source = WatchProgressSource.TRAKT),
refreshIfUnchanged = true,
forceSnapshot = true,
)
assertTrue(result.succeeded)
assertEquals(listOf(1 to WatchProgressSource.TRAKT), invalidations)
assertEquals(listOf(WatchProgressSource.TRAKT), activatedProgress)
assertEquals(listOf(WatchProgressSource.TRAKT), activatedWatched)
assertEquals(
listOf(RefreshCall(1, WatchProgressSource.TRAKT, sourceChanged = true, force = true)),
progressRefreshes,
)
assertEquals(
listOf(RefreshCall(1, WatchProgressSource.TRAKT, sourceChanged = false, force = true)),
watchedRefreshes,
)
}
@Test
fun `first coordinated transition detects a different already applied source`() = runBlocking {
val invalidations = mutableListOf<Pair<Int, WatchProgressSource>>()
var sourceChangedAtRefresh = false
val runner = WatchProgressSourceTransitionRunner(
currentAppliedSource = { WatchProgressSource.NUVIO_SYNC },
invalidateCache = { profileId, source -> invalidations += profileId to source },
activateProgressSource = {},
activateWatchedSource = {},
refreshProgress = { _, _, sourceChanged, _ ->
sourceChangedAtRefresh = sourceChanged
true
},
refreshWatched = { _, _, _ -> true },
)
runner.transition(
context = context(source = WatchProgressSource.TRAKT),
refreshIfUnchanged = false,
forceSnapshot = true,
)
assertTrue(sourceChangedAtRefresh)
assertEquals(listOf(1 to WatchProgressSource.TRAKT), invalidations)
}
@Test
fun `same observed context does not refetch without force`() = runBlocking {
var refreshCount = 0
var invalidationCount = 0
val runner = runner(
onInvalidate = { invalidationCount += 1 },
onRefresh = { refreshCount += 1 },
)
val context = context(source = WatchProgressSource.NUVIO_SYNC)
runner.transition(context = context, refreshIfUnchanged = false, forceSnapshot = false)
runner.transition(context = context, refreshIfUnchanged = false, forceSnapshot = false)
assertEquals(2, refreshCount, "the initial transition refreshes progress and watched once")
assertEquals(0, invalidationCount)
}
@Test
fun `authentication change refreshes same effective source without invalidating it`() = runBlocking {
var refreshCount = 0
var invalidationCount = 0
val runner = runner(
onInvalidate = { invalidationCount += 1 },
onRefresh = { refreshCount += 1 },
)
runner.transition(
context = context(
source = WatchProgressSource.NUVIO_SYNC,
isNuvioAuthenticated = false,
),
refreshIfUnchanged = false,
forceSnapshot = false,
)
runner.transition(
context = context(
source = WatchProgressSource.NUVIO_SYNC,
isNuvioAuthenticated = true,
),
refreshIfUnchanged = false,
forceSnapshot = false,
)
assertEquals(4, refreshCount)
assertEquals(0, invalidationCount)
}
@Test
fun `partial backend failure is surfaced`() = runBlocking {
val runner = WatchProgressSourceTransitionRunner(
invalidateCache = { _, _ -> },
activateProgressSource = {},
activateWatchedSource = {},
refreshProgress = { _, _, _, _ -> true },
refreshWatched = { _, _, _ -> false },
)
val result = runner.transition(
context = context(source = WatchProgressSource.TRAKT),
refreshIfUnchanged = true,
forceSnapshot = true,
)
assertTrue(result.progressRefreshed)
assertFalse(result.watchedHistoryRefreshed)
assertFalse(result.succeeded)
}
@Test
fun `failed transition retries the same observed context`() = runBlocking {
var watchedAttempts = 0
val runner = WatchProgressSourceTransitionRunner(
invalidateCache = { _, _ -> },
activateProgressSource = {},
activateWatchedSource = {},
refreshProgress = { _, _, _, _ -> true },
refreshWatched = { _, _, _ ->
watchedAttempts += 1
watchedAttempts > 1
},
)
val context = context(source = WatchProgressSource.TRAKT)
val first = runner.transition(
context = context,
refreshIfUnchanged = false,
forceSnapshot = true,
)
val retry = runner.transition(
context = context,
refreshIfUnchanged = false,
forceSnapshot = true,
)
assertFalse(first.succeeded)
assertTrue(retry.succeeded)
assertEquals(2, watchedAttempts)
}
@Test
fun `reverting after a failed source switch reactivates and refreshes the last successful source`() = runBlocking {
var appliedSource = WatchProgressSource.NUVIO_SYNC
val invalidations = mutableListOf<WatchProgressSource>()
val refreshedSources = mutableListOf<WatchProgressSource>()
var failTraktWatchedRefresh = true
val runner = WatchProgressSourceTransitionRunner(
currentAppliedSource = { appliedSource },
invalidateCache = { _, source -> invalidations += source },
activateProgressSource = { source -> appliedSource = source },
activateWatchedSource = {},
refreshProgress = { _, source, _, _ ->
refreshedSources += source
true
},
refreshWatched = { _, source, _ ->
refreshedSources += source
source != WatchProgressSource.TRAKT || !failTraktWatchedRefresh
},
)
assertTrue(
runner.transition(
context = context(source = WatchProgressSource.NUVIO_SYNC),
refreshIfUnchanged = true,
forceSnapshot = true,
).succeeded,
)
assertFalse(
runner.transition(
context = context(source = WatchProgressSource.TRAKT),
refreshIfUnchanged = false,
forceSnapshot = true,
).succeeded,
)
failTraktWatchedRefresh = false
val reverted = runner.transition(
context = context(source = WatchProgressSource.NUVIO_SYNC),
refreshIfUnchanged = false,
forceSnapshot = true,
)
assertTrue(reverted.succeeded)
assertEquals(WatchProgressSource.NUVIO_SYNC, appliedSource)
assertEquals(
listOf(WatchProgressSource.TRAKT, WatchProgressSource.NUVIO_SYNC),
invalidations,
)
assertEquals(
listOf(
WatchProgressSource.NUVIO_SYNC,
WatchProgressSource.NUVIO_SYNC,
WatchProgressSource.TRAKT,
WatchProgressSource.TRAKT,
WatchProgressSource.NUVIO_SYNC,
WatchProgressSource.NUVIO_SYNC,
),
refreshedSources,
)
}
@Test
fun `account reset rejects an in flight transition and forgets its context`() = runBlocking {
val refreshStarted = CompletableDeferred<Unit>()
val releaseRefresh = CompletableDeferred<Unit>()
var refreshCount = 0
val runner = WatchProgressSourceTransitionRunner(
invalidateCache = { _, _ -> },
activateProgressSource = {},
activateWatchedSource = {},
refreshProgress = { _, _, _, _ ->
refreshCount += 1
refreshStarted.complete(Unit)
releaseRefresh.await()
true
},
refreshWatched = { _, _, _ -> true },
)
val context = context(source = WatchProgressSource.TRAKT)
val staleTransition = async {
runner.transition(
context = context,
refreshIfUnchanged = false,
forceSnapshot = true,
)
}
refreshStarted.await()
runner.reset()
releaseRefresh.complete(Unit)
assertFailsWith<CancellationException> { staleTransition.await() }
runner.transition(
context = context,
refreshIfUnchanged = false,
forceSnapshot = true,
)
assertEquals(2, refreshCount)
}
private fun runner(
onInvalidate: () -> Unit,
onRefresh: () -> Unit,
): WatchProgressSourceTransitionRunner = WatchProgressSourceTransitionRunner(
invalidateCache = { _, _ -> onInvalidate() },
activateProgressSource = {},
activateWatchedSource = {},
refreshProgress = { _, _, _, _ ->
onRefresh()
true
},
refreshWatched = { _, _, _ ->
onRefresh()
true
},
)
private fun context(
source: WatchProgressSource,
isNuvioAuthenticated: Boolean = true,
): WatchProgressSourceContext = WatchProgressSourceContext(
profileId = 1,
requestedSource = source,
effectiveSource = source,
isNuvioAuthenticated = isNuvioAuthenticated,
)
private data class RefreshCall(
val profileId: Int,
val source: WatchProgressSource,
val sourceChanged: Boolean,
val force: Boolean,
)
}

7
composeApp/src/iosMain/kotlin/com/nuvio/app/core/storage/PlatformLocalAccountDataCleaner.ios.kt
View file

@ -6,6 +6,7 @@ internal actual object PlatformLocalAccountDataCleaner {
private val plainKeys = listOf(
"profile_payload",
"avatar_catalog_payload",
"anonymous_user_id",
)
private val profilePinCachePrefixes = listOf("profile_pin_cache_")
private val profileIndexedPrefixes = listOf(
@ -53,6 +54,7 @@ internal actual object PlatformLocalAccountDataCleaner {
"trakt_auth_payload",
"trakt_library_payload",
"trakt_settings_payload",
"pending_watch_progress_source",
"collection_mobile_settings_payload",
"collections_payload",
)
@ -76,7 +78,10 @@ internal actual object PlatformLocalAccountDataCleaner {
for (key in defaults.dictionaryRepresentation().keys) {
val keyString = key as? String ?: continue
if (keyString.startsWith("stream_link_")) {
if (
keyString.startsWith("stream_link_") ||
keyString.startsWith("cw_enrichment_cache_")
) {
defaults.removeObjectForKey(keyString)
}
}

19
composeApp/src/iosMain/kotlin/com/nuvio/app/features/trakt/TraktSettingsStorage.ios.kt
View file

@ -5,6 +5,7 @@ import platform.Foundation.NSUserDefaults
internal actual object TraktSettingsStorage {
private const val payloadKey = "trakt_settings_payload"
private const val pendingWatchProgressSourceKey = "pending_watch_progress_source"
actual fun loadPayload(): String? =
NSUserDefaults.standardUserDefaults.stringForKey(ProfileScopedKey.of(payloadKey))
@ -12,4 +13,22 @@ internal actual object TraktSettingsStorage {
actual fun savePayload(payload: String) {
NSUserDefaults.standardUserDefaults.setObject(payload, forKey = ProfileScopedKey.of(payloadKey))
}
actual fun loadPendingWatchProgressSourcePayload(profileId: Int): String? =
NSUserDefaults.standardUserDefaults.stringForKey(
ProfileScopedKey.of(pendingWatchProgressSourceKey, profileId),
)
actual fun savePendingWatchProgressSourcePayload(profileId: Int, payload: String) {
NSUserDefaults.standardUserDefaults.setObject(
payload,
forKey = ProfileScopedKey.of(pendingWatchProgressSourceKey, profileId),
)
}
actual fun clearPendingWatchProgressSourcePayload(profileId: Int) {
NSUserDefaults.standardUserDefaults.removeObjectForKey(
ProfileScopedKey.of(pendingWatchProgressSourceKey, profileId),
)
}
}