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 2d48b5fac..c42c8060e 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 @@ -135,6 +135,7 @@ internal class LibraryLocalState { token: LibraryProfileToken, activeProfileId: Int, items: Collection, + hasPendingPush: Boolean = false, ): LibraryLocalSnapshot? = synchronized(lock) { if (activeProfileId != token.profileId || !isCurrentLocked(token)) { return@synchronized null @@ -142,6 +143,7 @@ internal class LibraryLocalState { itemsById = items.associateByTo(mutableMapOf()) { libraryItemKey(it.id, it.type) } hasLoaded = true isLoading = false + this.hasPendingPush = hasPendingPush revision += 1L contentRevision += 1L snapshotLocked() @@ -177,7 +179,7 @@ internal class LibraryLocalState { val localContentChanged = contentRevision != pullSnapshot.contentRevision val hasLocalChanges = pullSnapshot.hasPendingPush || hasPendingPush || localContentChanged - val preserveLocalItems = itemsById.isNotEmpty() && (serverItems.isEmpty() || hasLocalChanges) + val preserveLocalItems = hasLocalChanges || (itemsById.isNotEmpty() && serverItems.isEmpty()) if (!preserveLocalItems) { itemsById = serverItems.associateByTo(mutableMapOf()) { libraryItemKey(it.id, it.type) } contentRevision += 1L @@ -276,12 +278,12 @@ internal class LibraryLocalState { } } - fun markPushCompleted(snapshot: LibraryLocalSnapshot) { - synchronized(lock) { - if (isContentCurrentLocked(snapshot)) { - hasPendingPush = false - } - } + fun markPushCompleted(snapshot: LibraryLocalSnapshot): LibraryLocalSnapshot? = synchronized(lock) { + if (!isContentCurrentLocked(snapshot)) return@synchronized null + hasPendingPush = false + revision += 1L + contentRevision += 1L + snapshotLocked() } private fun tokenLocked(): LibraryProfileToken = @@ -311,5 +313,8 @@ internal class LibraryLocalState { isCurrentLocked(snapshot.token) && contentRevision == snapshot.contentRevision } +internal fun LibraryLocalSnapshot.pendingItemsForSync(): List? = + items.takeIf { hasLoaded && hasPendingPush } + private fun libraryItemKey(id: String, type: String): String = "${type.trim().lowercase()}:${id.trim()}" 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 cb9fbebe3..23a558967 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 @@ -52,8 +52,9 @@ import org.jetbrains.compose.resources.StringResource import org.jetbrains.compose.resources.getString @Serializable -private data class StoredLibraryPayload( +internal data class StoredLibraryPayload( val items: List = emptyList(), + val hasPendingPush: Boolean = false, ) @Serializable @@ -74,6 +75,8 @@ private data class LibrarySyncItem( object LibraryRepository { private const val PULL_PAGE_SIZE = 500 + private const val PUSH_DEBOUNCE_MS = 500L + private const val MAX_PUSH_ATTEMPTS = 4 private val syncScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) private val log = Logger.withTag("LibraryRepository") @@ -185,6 +188,7 @@ object LibraryRepository { private fun loadFromDisk(profileId: Int): Boolean { var shouldPublish = false + var restoredPendingSnapshot: LibraryLocalSnapshot? = null val loaded = synchronized(loadLock) { if (ProfileRepository.activeProfileId != profileId) return@synchronized false val current = localState.snapshot() @@ -194,28 +198,32 @@ object LibraryRepository { val transition = localState.beginProfileLoad(profileId) transition.detachedPushJob?.cancel() - shouldPublish = completeLoadFromDisk(transition.snapshot.token) + val loadedSnapshot = completeLoadFromDisk(transition.snapshot.token) + shouldPublish = loadedSnapshot != null + restoredPendingSnapshot = loadedSnapshot?.takeIf { it.hasPendingPush } shouldPublish } if (shouldPublish) publish() + restoredPendingSnapshot?.let(::pushToServer) return loaded } - private fun completeLoadFromDisk(token: LibraryProfileToken): Boolean { + private fun completeLoadFromDisk(token: LibraryProfileToken): LibraryLocalSnapshot? { val payload = LibraryStorage.loadPayload(token.profileId).orEmpty().trim() - val items = if (payload.isNotEmpty()) { + val storedPayload = if (payload.isNotEmpty()) { runCatching { - json.decodeFromString(payload).items - }.getOrDefault(emptyList()) + json.decodeFromString(payload) + }.getOrDefault(StoredLibraryPayload()) } else { - emptyList() + StoredLibraryPayload() } return localState.completeProfileLoad( token = token, activeProfileId = ProfileRepository.activeProfileId, - items = items, - ) != null + items = storedPayload.items, + hasPendingPush = storedPayload.hasPendingPush, + ) } suspend fun pullFromServer(profileId: Int) { @@ -223,6 +231,9 @@ object LibraryRepository { log.d { "Skipping library pull for inactive profile $profileId" } return } + localState.snapshot() + .takeIf { it.token == operationToken && it.hasPendingPush } + ?.let(::pushToServer) if (isTraktLibrarySourceActive()) { try { @@ -459,39 +470,53 @@ object LibraryRepository { log.w { "Skipping library push: anonymous auth user=${authState.userId} profile=$profileId" } return } - val pushJob = syncScope.launch(start = CoroutineStart.LAZY) { - delay(500) - if (!localState.isContentCurrent(snapshot)) { - val current = localState.snapshot() - log.w { - "Skipping stale debounced library push: scheduled=${snapshot.token} " + - "current=${current.token} scheduledContentRevision=${snapshot.contentRevision} " + - "currentContentRevision=${current.contentRevision}" - } - return@launch + val pendingItems = snapshot.pendingItemsForSync() + if (pendingItems == null) { + log.w { + "Skipping library push without a loaded local mutation " + + "profile=$profileId hasLoaded=${snapshot.hasLoaded} pending=${snapshot.hasPendingPush}" } - runCatching { - val syncItems = snapshot.items.map { it.toSyncItem() } - if (syncItems.isEmpty()) { - log.w { "Skipping library push: sync payload is empty profile=$profileId" } - return@runCatching false + return + } + val pushJob = syncScope.launch(start = CoroutineStart.LAZY) { + val syncItems = pendingItems.map { it.toSyncItem() } + var attempt = 0 + while (attempt < MAX_PUSH_ATTEMPTS) { + delay(libraryPushDelayMs(attempt, PUSH_DEBOUNCE_MS)) + if (!localState.isContentCurrent(snapshot)) { + val current = localState.snapshot() + log.w { + "Skipping stale debounced library push: scheduled=${snapshot.token} " + + "current=${current.token} scheduledContentRevision=${snapshot.contentRevision} " + + "currentContentRevision=${current.contentRevision}" + } + return@launch } - val params = buildJsonObject { - put("p_profile_id", profileId) - put("p_items", json.encodeToJsonElement(syncItems)) - putSyncOriginClientId() + val result = runCatching { + val params = buildJsonObject { + put("p_profile_id", profileId) + put("p_items", json.encodeToJsonElement(syncItems)) + putSyncOriginClientId() + } + log.i { + "Pushing library to server profile=$profileId itemCount=${syncItems.size} " + + "attempt=${attempt + 1}/$MAX_PUSH_ATTEMPTS" + } + SupabaseProvider.client.postgrest.rpc("sync_push_library", params) } - log.i { "Pushing library to server profile=$profileId itemCount=${syncItems.size}" } - SupabaseProvider.client.postgrest.rpc("sync_push_library", params) - true - }.onSuccess { pushed -> - if (pushed) { - localState.markPushCompleted(snapshot) + if (result.isSuccess) { + localState.markPushCompleted(snapshot)?.let(::persist) log.i { "Library push completed profile=$profileId itemCount=$itemCount" } + return@launch + } + + val error = result.exceptionOrNull() + if (error is CancellationException) throw error + attempt += 1 + log.e(error) { + "Failed to push library to server profile=$profileId itemCount=$itemCount " + + "attempt=$attempt/$MAX_PUSH_ATTEMPTS" } - }.onFailure { e -> - if (e is CancellationException) throw e - log.e(e) { "Failed to push library to server profile=$profileId itemCount=$itemCount" } } } pushJob.invokeOnCompletion { localState.clearPushJob(pushJob) } @@ -587,6 +612,7 @@ object LibraryRepository { val payload = json.encodeToString( StoredLibraryPayload( items = snapshot.items.sortedByDescending { it.savedAtEpochMs }, + hasPendingPush = snapshot.hasPendingPush, ), ) synchronized(persistenceLock) { @@ -623,6 +649,9 @@ object LibraryRepository { effectiveLibrarySourceMode() == LibrarySourceMode.TRAKT } +internal fun libraryPushDelayMs(attempt: Int, baseDelayMs: Long = 500L): Long = + baseDelayMs * (1L shl attempt.coerceIn(0, 3)) + internal const val LOCAL_LIBRARY_LIST_KEY = "local" private const val DEFAULT_LOCAL_LIBRARY_TAB_TITLE = "Nuvio Library" private const val DEFAULT_LIBRARY_OTHER_TITLE = "Other" 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 0a5add232..c517df379 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 @@ -5,6 +5,9 @@ import com.nuvio.app.features.home.PosterShape import com.nuvio.app.features.home.MetaPreview import com.nuvio.app.features.trakt.TraktListTab import com.nuvio.app.features.trakt.TraktListType +import kotlinx.serialization.decodeFromString +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertFalse @@ -280,6 +283,102 @@ class LibraryRepositoryTest { assertEquals(listOf("remote-new"), result.snapshot.items.map { it.id }) } + @Test + fun `server pull cannot restore items after pending delete all`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem(id = "last-item", savedAtEpochMs = 1L)), + ) + state.remove("last-item", "movie") + val pullSnapshot = assertNotNull(state.markPullStarted(token)) + + val result = assertNotNull( + state.applyServerItems( + pullSnapshot = pullSnapshot, + serverItems = listOf(libraryItem(id = "last-item", savedAtEpochMs = 1L)), + ), + ) + + assertTrue(result.preservedLocalItems) + assertTrue(result.snapshot.items.isEmpty()) + assertNotNull(result.snapshot.pendingItemsForSync()) + } + + @Test + fun `pending empty push survives storage round trip`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem(id = "last-item", savedAtEpochMs = 1L)), + ) + val pendingSnapshot = state.remove("last-item", "movie").snapshot + val payload = Json.encodeToString( + StoredLibraryPayload( + items = pendingSnapshot.items, + hasPendingPush = pendingSnapshot.hasPendingPush, + ), + ) + val restoredPayload = Json.decodeFromString(payload) + val restoredState = LibraryLocalState() + val restoredToken = restoredState.beginProfileLoad(profileId = 1).snapshot.token + val restoredSnapshot = assertNotNull( + restoredState.completeProfileLoad( + token = restoredToken, + activeProfileId = 1, + items = restoredPayload.items, + hasPendingPush = restoredPayload.hasPendingPush, + ), + ) + + assertTrue(restoredSnapshot.hasPendingPush) + assertEquals(emptyList(), restoredSnapshot.pendingItemsForSync()) + } + + @Test + fun `library push retry delay backs off with a cap`() { + assertEquals(500L, libraryPushDelayMs(attempt = 0)) + assertEquals(1_000L, libraryPushDelayMs(attempt = 1)) + assertEquals(4_000L, libraryPushDelayMs(attempt = 3)) + assertEquals(4_000L, libraryPushDelayMs(attempt = 10)) + } + + @Test + fun `removing the last local item produces an empty sync payload`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem(id = "last-item", savedAtEpochMs = 1L)), + ) + + val mutation = state.remove("last-item", "movie") + val pendingItems = assertNotNull(mutation.snapshot.pendingItemsForSync()) + + assertEquals(1, mutation.affectedCount) + assertTrue(pendingItems.isEmpty()) + } + + @Test + fun `loaded empty library without a local mutation is not pushed`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + val snapshot = assertNotNull( + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = emptyList(), + ), + ) + + assertNull(snapshot.pendingItemsForSync()) + } + private fun libraryItem(id: String, savedAtEpochMs: Long): LibraryItem = LibraryItem( id = id,