diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt index 916059799..4f459078a 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt @@ -33,7 +33,7 @@ import kotlinx.coroutines.launch 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 +private const val PERIODIC_NUVIO_SYNC_PULL_INTERVAL_MS = 240_000L internal enum class ProfileSyncStep { Addons, @@ -130,8 +130,6 @@ internal enum class ProfileSyncRequestResult { Replaced, } -internal fun shouldQueueCoalescedForegroundPull(force: Boolean): Boolean = force - internal class ProfileSyncRequestGate { private data class PendingRequest( val scope: CoroutineScope, @@ -298,11 +296,8 @@ object SyncManager { delay(FOREGROUND_PULL_DELAY_MS) } if (!force && hasRecentFullPull(profileId)) return@launch - startFullProfilePull( - profileId = profileId, - reason = "foreground", - queueIfCoalesced = shouldQueueCoalescedForegroundPull(force), - ) + if (ProfileRepository.activeProfileId != profileId) return@launch + pullForegroundForProfile(profileId) } finally { synchronized(pullStateLock) { if (foregroundPullJob === requestJob) { @@ -325,6 +320,47 @@ object SyncManager { TraktPlatformClock.nowEpochMs() - lastFullPullAtMs < FOREGROUND_PULL_MIN_INTERVAL_MS } + private suspend fun pullForegroundForProfile(profileId: Int) { + log.i { "Foreground sync started profile=$profileId" } + + runCatching { ProfileRepository.pullProfiles() } + .onFailure { log.e(it) { "Foreground profiles pull failed" } } + runCatching { ProfileSettingsSync.pull(profileId) } + .onFailure { log.e(it) { "Foreground profile settings pull failed" } } + + coroutineScope { + launch { + runCatching { AddonRepository.pullFromServer(profileId) } + .onFailure { log.e(it) { "Foreground addons pull failed" } } + } + if (AppFeaturePolicy.pluginsEnabled) { + launch { + runCatching { PluginRepository.pullFromServer(profileId) } + .onFailure { log.e(it) { "Foreground plugins pull failed" } } + } + } + launch { + runCatching { LibraryRepository.pullFromServer(profileId) } + .onFailure { log.e(it) { "Foreground library pull failed" } } + } + launch { + runCatching { + WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = true) + }.onFailure { log.e(it) { "Foreground active watch source 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" } } + } + } + + log.i { "Foreground sync completed profile=$profileId" } + } + private fun startFullProfilePull( profileId: Int, reason: String, @@ -449,15 +485,6 @@ object SyncManager { val authState = AuthRepository.state.value if (authState !is AuthState.Authenticated || authState.isAnonymous) return - if (surface == "profile_settings") { - startFullProfilePull( - profileId = profileId, - reason = "realtime_profile_settings", - queueIfCoalesced = true, - ) - return - } - accountScopeSnapshot().launch { log.i { "requestRealtimeSurfacePull($profileId, $surface)" } when (surface) { @@ -480,6 +507,10 @@ object SyncManager { WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = false) }.onFailure { log.e(it) { "Realtime active watch source pull failed" } } } + "profile_settings" -> { + runCatching { ProfileSettingsSync.pull(profileId) } + .onFailure { log.e(it) { "Realtime profile settings pull failed" } } + } "collections" -> { runCatching { CollectionSyncService.pullFromServer(profileId) } .onFailure { log.e(it) { "Realtime collections pull failed" } } diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryLocalState.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryLocalState.kt index a0de3fe5b..2d48b5fac 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryLocalState.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryLocalState.kt @@ -16,8 +16,7 @@ internal data class LibraryLocalSnapshot( val hasLoaded: Boolean, val isLoading: Boolean, val items: List, - val isPullingNuvioSyncFromServer: Boolean, - val hasCompletedInitialNuvioSyncPull: Boolean, + val hasPendingPush: Boolean, ) internal data class LibraryStateTransition( @@ -61,8 +60,7 @@ internal class LibraryLocalState { private var contentRevision = 0L private var isLoading = false private var itemsById: MutableMap = mutableMapOf() - private var isPullingNuvioSyncFromServer = false - private var hasCompletedInitialNuvioSyncPull = false + private var hasPendingPush = false private var pushJob: Job? = null fun snapshot(): LibraryLocalSnapshot = synchronized(lock) { @@ -85,6 +83,10 @@ internal class LibraryLocalState { isCurrentLocked(snapshot) } + fun isContentCurrent(snapshot: LibraryLocalSnapshot): Boolean = synchronized(lock) { + isContentCurrentLocked(snapshot) + } + fun runIfCurrent(snapshot: LibraryLocalSnapshot, block: () -> Unit): Boolean = synchronized(lock) { if (!isCurrentLocked(snapshot)) { false @@ -122,8 +124,7 @@ internal class LibraryLocalState { hasLoaded = false isLoading = true itemsById = mutableMapOf() - isPullingNuvioSyncFromServer = false - hasCompletedInitialNuvioSyncPull = false + hasPendingPush = false LibraryStateTransition( snapshot = snapshotLocked(), detachedPushJob = detachedPushJob, @@ -156,8 +157,7 @@ internal class LibraryLocalState { hasLoaded = false isLoading = false itemsById = mutableMapOf() - isPullingNuvioSyncFromServer = false - hasCompletedInitialNuvioSyncPull = false + hasPendingPush = false LibraryStateTransition( snapshot = snapshotLocked(), detachedPushJob = detachedPushJob, @@ -166,34 +166,22 @@ internal class LibraryLocalState { fun markPullStarted(token: LibraryProfileToken): LibraryLocalSnapshot? = synchronized(lock) { if (!isCurrentLocked(token)) return@synchronized null - isPullingNuvioSyncFromServer = true - revision += 1L - snapshotLocked() - } - - fun finishPull( - token: LibraryProfileToken, - completedSuccessfully: Boolean, - ): LibraryLocalSnapshot? = synchronized(lock) { - if (!isCurrentLocked(token)) return@synchronized null - isPullingNuvioSyncFromServer = false - if (completedSuccessfully) { - hasCompletedInitialNuvioSyncPull = true - } - revision += 1L snapshotLocked() } fun applyServerItems( - token: LibraryProfileToken, + pullSnapshot: LibraryLocalSnapshot, serverItems: Collection, ): LibraryServerItemsApplyResult? = synchronized(lock) { - if (!isCurrentLocked(token)) return@synchronized null + if (!isCurrentLocked(pullSnapshot.token)) return@synchronized null - val preserveLocalItems = serverItems.isEmpty() && itemsById.isNotEmpty() + val localContentChanged = contentRevision != pullSnapshot.contentRevision + val hasLocalChanges = pullSnapshot.hasPendingPush || hasPendingPush || localContentChanged + val preserveLocalItems = itemsById.isNotEmpty() && (serverItems.isEmpty() || hasLocalChanges) if (!preserveLocalItems) { itemsById = serverItems.associateByTo(mutableMapOf()) { libraryItemKey(it.id, it.type) } contentRevision += 1L + hasPendingPush = false } hasLoaded = true isLoading = false @@ -208,6 +196,7 @@ internal class LibraryLocalState { itemsById[libraryItemKey(item.id, item.type)] = item revision += 1L contentRevision += 1L + hasPendingPush = true snapshotLocked() } @@ -221,6 +210,7 @@ internal class LibraryLocalState { } revision += 1L contentRevision += 1L + hasPendingPush = true LibraryLocalToggleResult( snapshot = snapshotLocked(), isSaved = isSaved, @@ -234,6 +224,7 @@ internal class LibraryLocalState { if (affectedCount > 0) { revision += 1L contentRevision += 1L + hasPendingPush = true } LibraryLocalMutation( snapshot = snapshotLocked(), @@ -246,6 +237,7 @@ internal class LibraryLocalState { if (affectedCount > 0) { revision += 1L contentRevision += 1L + hasPendingPush = true } LibraryLocalMutation( snapshot = snapshotLocked(), @@ -269,7 +261,7 @@ internal class LibraryLocalState { snapshot: LibraryLocalSnapshot, job: Job, ): LibraryPushJobInstallResult = synchronized(lock) { - if (!isCurrentLocked(snapshot)) { + if (!isContentCurrentLocked(snapshot)) { LibraryPushJobInstallResult(installed = false, detachedPushJob = null) } else { val detachedPushJob = pushJob @@ -284,6 +276,14 @@ internal class LibraryLocalState { } } + fun markPushCompleted(snapshot: LibraryLocalSnapshot) { + synchronized(lock) { + if (isContentCurrentLocked(snapshot)) { + hasPendingPush = false + } + } + } + private fun tokenLocked(): LibraryProfileToken = LibraryProfileToken( profileId = currentProfileId, @@ -298,8 +298,7 @@ internal class LibraryLocalState { hasLoaded = hasLoaded, isLoading = isLoading, items = itemsById.values.toList(), - isPullingNuvioSyncFromServer = isPullingNuvioSyncFromServer, - hasCompletedInitialNuvioSyncPull = hasCompletedInitialNuvioSyncPull, + hasPendingPush = hasPendingPush, ) private fun isCurrentLocked(token: LibraryProfileToken): Boolean = diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryRepository.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryRepository.kt index 2b9d476aa..06c102f8c 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryRepository.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryRepository.kt @@ -239,20 +239,18 @@ object LibraryRepository { nuvioPullMutex.withLock { val serializedToken = activeOperationToken(profileId) ?: return@withLock - if (localState.markPullStarted(serializedToken) == null) return@withLock + val pullSnapshot = localState.markPullStarted(serializedToken) ?: return@withLock - var completedSuccessfully = false var appliedItems = false try { val serverItems = pullAllLibrarySyncItems(profileId).map { it.toLibraryItem() } - val applyResult = localState.applyServerItems(serializedToken, serverItems) + val applyResult = localState.applyServerItems(pullSnapshot, serverItems) ?: return@withLock - completedSuccessfully = true appliedItems = true if (applyResult.preservedLocalItems) { log.w { - "Remote library is empty while local has ${applyResult.snapshot.items.size} entries; " + - "preserving local library" + "Preserving ${applyResult.snapshot.items.size} local library items because the remote " + + "snapshot is empty or local changes are pending" } } else { persist(applyResult.snapshot) @@ -261,11 +259,6 @@ object LibraryRepository { throw error } catch (error: Throwable) { log.e(error) { "Failed to pull library from server" } - } finally { - localState.finishPull( - token = serializedToken, - completedSuccessfully = completedSuccessfully, - ) } if (appliedItems) publish() @@ -466,23 +459,14 @@ object LibraryRepository { log.w { "Skipping library push: anonymous auth user=${authState.userId} profile=$profileId" } return } - if (snapshot.isPullingNuvioSyncFromServer) { - log.i { "Skipping library push: server pull is active profile=$profileId localItems=$itemCount" } - return - } - if (!snapshot.hasCompletedInitialNuvioSyncPull) { - log.w { "Skipping library push: initial Nuvio sync pull not completed profile=$profileId localItems=$itemCount" } - return - } - val pushJob = syncScope.launch(start = CoroutineStart.LAZY) { delay(500) - if (!localState.isCurrent(snapshot)) { + if (!localState.isContentCurrent(snapshot)) { val current = localState.snapshot() log.w { "Skipping stale debounced library push: scheduled=${snapshot.token} " + - "current=${current.token} scheduledRevision=${snapshot.revision} " + - "currentRevision=${current.revision}" + "current=${current.token} scheduledContentRevision=${snapshot.contentRevision} " + + "currentContentRevision=${current.contentRevision}" } return@launch } @@ -502,6 +486,7 @@ object LibraryRepository { true }.onSuccess { pushed -> if (pushed) { + localState.markPushCompleted(snapshot) log.i { "Library push completed profile=$profileId itemCount=$itemCount" } } }.onFailure { e -> diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt index 4b46c2c40..7eb617ac4 100644 --- a/composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt @@ -10,12 +10,6 @@ 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() diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryRepositoryTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryRepositoryTest.kt index 5d16bc6c7..0a5add232 100644 --- a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryRepositoryTest.kt +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryRepositoryTest.kt @@ -188,6 +188,98 @@ class LibraryRepositoryTest { assertEquals(setOf("first", "second"), state.snapshot().items.map { it.id }.toSet()) } + @Test + fun `pull lifecycle does not invalidate pending local push`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = emptyList(), + ) + val localSnapshot = state.upsert(libraryItem(id = "local", savedAtEpochMs = 1L)) + + state.markPullStarted(token) + + assertTrue(state.isContentCurrent(localSnapshot)) + } + + @Test + fun `server pull preserves local items awaiting push`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)), + ) + state.upsert(libraryItem(id = "local-new", savedAtEpochMs = 2L)) + val pullSnapshot = assertNotNull(state.markPullStarted(token)) + + val result = assertNotNull( + state.applyServerItems( + pullSnapshot = pullSnapshot, + serverItems = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)), + ), + ) + + assertTrue(result.preservedLocalItems) + assertEquals( + setOf("remote-old", "local-new"), + result.snapshot.items.map { it.id }.toSet(), + ) + } + + @Test + fun `server pull preserves a local mutation made while pulling`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)), + ) + val pullSnapshot = assertNotNull(state.markPullStarted(token)) + state.upsert(libraryItem(id = "local-new", savedAtEpochMs = 2L)) + + val result = assertNotNull( + state.applyServerItems( + pullSnapshot = pullSnapshot, + serverItems = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)), + ), + ) + + assertTrue(result.preservedLocalItems) + assertEquals( + setOf("remote-old", "local-new"), + result.snapshot.items.map { it.id }.toSet(), + ) + } + + @Test + fun `server pull applies after local push completes`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)), + ) + val localSnapshot = state.upsert(libraryItem(id = "local-new", savedAtEpochMs = 2L)) + state.markPushCompleted(localSnapshot) + val pullSnapshot = assertNotNull(state.markPullStarted(token)) + + val result = assertNotNull( + state.applyServerItems( + pullSnapshot = pullSnapshot, + serverItems = listOf(libraryItem(id = "remote-new", savedAtEpochMs = 3L)), + ), + ) + + assertFalse(result.preservedLocalItems) + assertEquals(listOf("remote-new"), result.snapshot.items.map { it.id }) + } + private fun libraryItem(id: String, savedAtEpochMs: Long): LibraryItem = LibraryItem( id = id,