Merge f510c9a876 into b60995324b

composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryLocalState.kt

@@ -135,6 +135,7 @@ internal class LibraryLocalState {
         token: LibraryProfileToken,
         activeProfileId: Int,
         items: Collection<LibraryItem>,
+        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<LibraryItem>? =
+    items.takeIf { hasLoaded && hasPendingPush }
+
 private fun libraryItemKey(id: String, type: String): String =
     "${type.trim().lowercase()}:${id.trim()}"

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<LibraryItem> = 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<StoredLibraryPayload>(payload).items
-            }.getOrDefault(emptyList())
+                json.decodeFromString<StoredLibraryPayload>(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"

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<StoredLibraryPayload>(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,