Add mobile delta sync for progress

Persist cursors for watched items and watch progress.

Use snapshot fallback when delta RPCs are unavailable.

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

@@ -9,6 +9,7 @@ import com.nuvio.app.features.trakt.WatchProgressSource
 import com.nuvio.app.features.trakt.shouldUseTraktProgress
 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.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -26,6 +27,8 @@ import kotlinx.serialization.json.Json
 private data class StoredWatchedPayload(
     val items: List<WatchedItem> = emptyList(),
     val lastSuccessfulPushEpochMs: Long = 0L,
+    val deltaCursorEventId: Long = 0L,
+    val deltaInitialized: Boolean = false,
 )
 
 internal enum class WatchedTraktHistorySync {
@@ -40,6 +43,9 @@ internal fun shouldMirrorWatchedMarkToTraktHistory(
 
 object WatchedRepository {
     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 log = Logger.withTag("WatchedRepository")
@@ -55,11 +61,10 @@ object WatchedRepository {
     private var currentProfileId: Int = 1
     private var itemsByKey: MutableMap<String, WatchedItem> = mutableMapOf()
     private var lastSuccessfulPushEpochMs: Long = 0L
+    private var deltaCursorEventId: Long = 0L
+    private var deltaInitialized: Boolean = false
     internal var syncAdapter: WatchedSyncAdapter = SupabaseWatchedSyncAdapter
 
-    private fun activePullSyncAdapter(): WatchedSyncAdapter =
-        if (shouldUseTraktWatchedSync()) TraktWatchedSyncAdapter else syncAdapter
-
     fun ensureLoaded() {
         if (hasLoaded) return
         loadFromDisk(ProfileRepository.activeProfileId)
@@ -75,6 +80,8 @@ object WatchedRepository {
         currentProfileId = 1
         itemsByKey.clear()
         lastSuccessfulPushEpochMs = 0L
+        deltaCursorEventId = 0L
+        deltaInitialized = false
         _uiState.value = WatchedUiState()
     }
 
@@ -89,12 +96,16 @@ object WatchedRepository {
                 json.decodeFromString<StoredWatchedPayload>(payload)
             }.getOrDefault(StoredWatchedPayload())
             lastSuccessfulPushEpochMs = storedPayload.lastSuccessfulPushEpochMs
+            deltaCursorEventId = storedPayload.deltaCursorEventId
+            deltaInitialized = storedPayload.deltaInitialized
             itemsByKey = storedPayload.items
                 .map(WatchedItem::normalizedMarkedAt)
                 .associateBy { watchedItemKey(it.type, it.id, it.season, it.episode) }
                 .toMutableMap()
         } else {
             lastSuccessfulPushEpochMs = 0L
+            deltaCursorEventId = 0L
+            deltaInitialized = false
         }
 
         publish()
@@ -110,25 +121,138 @@ object WatchedRepository {
             .toList()
         val lastPushEpochMs = lastSuccessfulPushEpochMs
         runCatching {
-            val serverItems = activePullSyncAdapter().pull(
-                profileId = profileId,
-                pageSize = watchedItemsPageSize,
-            )
-
-            itemsByKey = mergeWatchedItemsPreservingUnsynced(
-                serverItems = serverItems,
-                localItems = localBeforePull,
-                lastSuccessfulPushEpochMs = lastPushEpochMs,
-                pullStartedEpochMs = pullStartedEpochMs,
-            ).toMutableMap()
-            hasLoaded = true
-            publish()
-            persist()
+            if (shouldUseTraktWatchedSync()) {
+                pullFullFromAdapter(
+                    adapter = TraktWatchedSyncAdapter,
+                    profileId = profileId,
+                    localBeforePull = localBeforePull,
+                    lastPushEpochMs = lastPushEpochMs,
+                    pullStartedEpochMs = pullStartedEpochMs,
+                    resetDeltaState = true,
+                )
+            } else {
+                pullSupabaseDeltaFromServer(
+                    profileId = profileId,
+                    localBeforePull = localBeforePull,
+                    lastPushEpochMs = lastPushEpochMs,
+                    pullStartedEpochMs = pullStartedEpochMs,
+                )
+            }
         }.onFailure { e ->
             log.e(e) { "Failed to pull watched items from server" }
         }
     }
 
+    private suspend fun pullFullFromAdapter(
+        adapter: WatchedSyncAdapter,
+        profileId: Int,
+        localBeforePull: List<WatchedItem>,
+        lastPushEpochMs: Long,
+        pullStartedEpochMs: Long,
+        resetDeltaState: Boolean,
+    ) {
+        val serverItems = adapter.pull(
+            profileId = profileId,
+            pageSize = watchedItemsPageSize,
+        )
+
+        itemsByKey = mergeWatchedItemsPreservingUnsynced(
+            serverItems = serverItems,
+            localItems = localBeforePull,
+            lastSuccessfulPushEpochMs = lastPushEpochMs,
+            pullStartedEpochMs = pullStartedEpochMs,
+        ).toMutableMap()
+        if (resetDeltaState) {
+            deltaCursorEventId = 0L
+            deltaInitialized = false
+        }
+        hasLoaded = true
+        publish()
+        persist()
+    }
+
+    private suspend fun pullSupabaseDeltaFromServer(
+        profileId: Int,
+        localBeforePull: List<WatchedItem>,
+        lastPushEpochMs: Long,
+        pullStartedEpochMs: Long,
+    ) {
+        if (!deltaInitialized) {
+            val cursorBeforeSnapshot = syncAdapter.getDeltaCursor(profileId) ?: return
+            pullFullFromAdapter(
+                adapter = syncAdapter,
+                profileId = profileId,
+                localBeforePull = localBeforePull,
+                lastPushEpochMs = lastPushEpochMs,
+                pullStartedEpochMs = pullStartedEpochMs,
+                resetDeltaState = false,
+            )
+            deltaCursorEventId = cursorBeforeSnapshot
+            deltaInitialized = true
+            persist()
+            return
+        }
+
+        var cursor = deltaCursorEventId
+        var changed = false
+
+        while (true) {
+            val events = syncAdapter.pullDelta(
+                profileId = profileId,
+                sinceEventId = cursor,
+                limit = watchedItemsDeltaPageSize,
+            )
+            if (events.isEmpty()) break
+
+            applyWatchedDeltaEvents(
+                events = events,
+                lastPushEpochMs = lastPushEpochMs,
+                pullStartedEpochMs = pullStartedEpochMs,
+            )
+            cursor = maxOf(cursor, events.maxOf { it.eventId })
+            deltaCursorEventId = cursor
+            deltaInitialized = true
+            changed = true
+
+            if (events.size < watchedItemsDeltaPageSize) break
+        }
+
+        hasLoaded = true
+        if (changed) {
+            publish()
+            persist()
+        }
+    }
+
+    private fun applyWatchedDeltaEvents(
+        events: Collection<WatchedDeltaEvent>,
+        lastPushEpochMs: Long,
+        pullStartedEpochMs: Long,
+    ) {
+        events.forEach { event ->
+            val key = watchedItemKey(event.contentType, event.contentId, event.season, event.episode)
+            when (event.operation.lowercase()) {
+                watchedDeltaOperationUpsert -> {
+                    itemsByKey[key] = WatchedItem(
+                        id = event.contentId,
+                        type = event.contentType,
+                        name = event.title,
+                        season = event.season,
+                        episode = event.episode,
+                        markedAtEpochMs = normalizeWatchedMarkedAtEpochMs(event.watchedAt),
+                    )
+                }
+                watchedDeltaOperationDelete -> {
+                    val localItem = itemsByKey[key]
+                    if (localItem != null && shouldPreserveLocalWatchedItem(localItem, lastPushEpochMs, pullStartedEpochMs)) {
+                        return@forEach
+                    }
+                    itemsByKey.remove(key)
+                }
+            }
+        }
+    }
+
     fun toggleWatched(item: WatchedItem) {
         ensureLoaded()
         val key = watchedItemKey(item.type, item.id, item.season, item.episode)
@@ -303,6 +427,8 @@ object WatchedRepository {
                         .map(WatchedItem::normalizedMarkedAt)
                         .sortedByDescending { it.markedAtEpochMs },
                     lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
+                    deltaCursorEventId = deltaCursorEventId,
+                    deltaInitialized = deltaInitialized,
                 ),
             ),
         )
@@ -381,10 +507,7 @@ internal fun mergeWatchedItemsPreservingUnsynced(
         .forEach { localItem ->
             val key = watchedItemKey(localItem.type, localItem.id, localItem.season, localItem.episode)
             if (key in merged) return@forEach
-            val markedAt = localItem.markedAtEpochMs
-            val wasMarkedAfterLastPush = lastSuccessfulPushEpochMs > 0L && markedAt > lastSuccessfulPushEpochMs
-            val wasMarkedDuringPull = pullStartedEpochMs > 0L && markedAt >= pullStartedEpochMs
-            if (wasMarkedAfterLastPush || wasMarkedDuringPull) {
+            if (shouldPreserveLocalWatchedItem(localItem, lastSuccessfulPushEpochMs, pullStartedEpochMs)) {
                 merged[key] = localItem
             }
         }
@@ -392,6 +515,17 @@ internal fun mergeWatchedItemsPreservingUnsynced(
     return merged
 }
 
+internal fun shouldPreserveLocalWatchedItem(
+    localItem: WatchedItem,
+    lastSuccessfulPushEpochMs: Long,
+    pullStartedEpochMs: Long,
+): Boolean {
+    val markedAt = localItem.markedAtEpochMs
+    val wasMarkedAfterLastPush = lastSuccessfulPushEpochMs > 0L && markedAt > lastSuccessfulPushEpochMs
+    val wasMarkedDuringPull = pullStartedEpochMs > 0L && markedAt >= pullStartedEpochMs
+    return wasMarkedAfterLastPush || wasMarkedDuringPull
+}
+
 internal fun shouldUseTraktWatchedSync(
     isAuthenticated: Boolean,
     source: WatchProgressSource,

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/ProgressSyncAdapter.kt

@@ -13,6 +13,20 @@ data class ProgressSyncRecord(
     val lastWatched: Long = 0L,
 )
 
+data class ProgressDeltaEvent(
+    val eventId: Long,
+    val operation: String,
+    val progressKey: String,
+    val contentId: String,
+    val contentType: String,
+    val videoId: String,
+    val season: Int? = null,
+    val episode: Int? = null,
+    val position: Long = 0L,
+    val duration: Long = 0L,
+    val lastWatched: Long = 0L,
+)
+
 interface ProgressSyncAdapter {
     suspend fun pull(
         profileId: Int,
@@ -20,6 +34,14 @@ interface ProgressSyncAdapter {
         limit: Int? = null,
     ): List<ProgressSyncRecord>
 
+    suspend fun getDeltaCursor(profileId: Int): Long? = null
+
+    suspend fun pullDelta(
+        profileId: Int,
+        sinceEventId: Long,
+        limit: Int,
+    ): List<ProgressDeltaEvent> = emptyList()
+
     suspend fun push(
         profileId: Int,
         entries: Collection<WatchProgressEntry>,

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseProgressSyncAdapter.kt

@@ -17,6 +17,43 @@ object SupabaseProgressSyncAdapter : ProgressSyncAdapter {
         encodeDefaults = true
     }
 
+    override suspend fun getDeltaCursor(profileId: Int): Long {
+        val params = buildJsonObject {
+            put("p_profile_id", profileId)
+        }
+        return SupabaseProvider.client.postgrest
+            .rpc("sync_get_watch_progress_delta_cursor", params)
+            .decodeAs<Long>()
+    }
+
+    override suspend fun pullDelta(
+        profileId: Int,
+        sinceEventId: Long,
+        limit: Int,
+    ): List<ProgressDeltaEvent> {
+        val params = buildJsonObject {
+            put("p_profile_id", profileId)
+            put("p_since_event_id", sinceEventId)
+            put("p_limit", limit)
+        }
+        val result = SupabaseProvider.client.postgrest.rpc("sync_pull_watch_progress_delta", params)
+        return result.decodeList<WatchProgressDeltaSyncEntry>().map { event ->
+            ProgressDeltaEvent(
+                eventId = event.eventId,
+                operation = event.operation,
+                progressKey = event.progressKey,
+                contentId = event.contentId,
+                contentType = event.contentType,
+                videoId = event.videoId,
+                season = event.season,
+                episode = event.episode,
+                position = event.position,
+                duration = event.duration,
+                lastWatched = event.lastWatched,
+            )
+        }
+    }
+
     override suspend fun pull(
         profileId: Int,
         sinceLastWatched: Long?,
@@ -109,3 +146,18 @@ private data class WatchProgressSyncEntry(
     @SerialName("last_watched") val lastWatched: Long = 0,
     @SerialName("progress_key") val progressKey: String = "",
 )
+
+@Serializable
+private data class WatchProgressDeltaSyncEntry(
+    @SerialName("event_id") val eventId: Long,
+    val operation: String,
+    @SerialName("progress_key") val progressKey: String,
+    @SerialName("content_id") val contentId: String,
+    @SerialName("content_type") val contentType: String,
+    @SerialName("video_id") val videoId: String,
+    val season: Int? = null,
+    val episode: Int? = null,
+    val position: Long = 0,
+    val duration: Long = 0,
+    @SerialName("last_watched") val lastWatched: Long = 0,
+)

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseWatchedSyncAdapter.kt

@@ -18,6 +18,40 @@ object SupabaseWatchedSyncAdapter : WatchedSyncAdapter {
         encodeDefaults = true
     }
 
+    override suspend fun getDeltaCursor(profileId: Int): Long {
+        val params = buildJsonObject {
+            put("p_profile_id", profileId)
+        }
+        return SupabaseProvider.client.postgrest
+            .rpc("sync_get_watched_items_delta_cursor", params)
+            .decodeAs<Long>()
+    }
+
+    override suspend fun pullDelta(
+        profileId: Int,
+        sinceEventId: Long,
+        limit: Int,
+    ): List<WatchedDeltaEvent> {
+        val params = buildJsonObject {
+            put("p_profile_id", profileId)
+            put("p_since_event_id", sinceEventId)
+            put("p_limit", limit)
+        }
+        val result = SupabaseProvider.client.postgrest.rpc("sync_pull_watched_items_delta", params)
+        return result.decodeList<WatchedDeltaSyncItem>().map { event ->
+            WatchedDeltaEvent(
+                eventId = event.eventId,
+                operation = event.operation,
+                contentId = event.contentId,
+                contentType = event.contentType,
+                title = event.title,
+                season = event.season,
+                episode = event.episode,
+                watchedAt = event.watchedAt,
+            )
+        }
+    }
+
     override suspend fun pull(
         profileId: Int,
         pageSize: Int,
@@ -101,6 +135,18 @@ private data class WatchedSyncItem(
     @SerialName("watched_at") val watchedAt: Long = 0,
 )
 
+@Serializable
+private data class WatchedDeltaSyncItem(
+    @SerialName("event_id") val eventId: Long,
+    val operation: String,
+    @SerialName("content_id") val contentId: String,
+    @SerialName("content_type") val contentType: String,
+    val title: String = "",
+    val season: Int? = null,
+    val episode: Int? = null,
+    @SerialName("watched_at") val watchedAt: Long = 0,
+)
+
 @Serializable
 private data class WatchedDeleteKey(
     @SerialName("content_id") val contentId: String,

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/WatchedSyncAdapter.kt

@@ -2,12 +2,31 @@ package com.nuvio.app.features.watching.sync
 
 import com.nuvio.app.features.watched.WatchedItem
 
+data class WatchedDeltaEvent(
+    val eventId: Long,
+    val operation: String,
+    val contentId: String,
+    val contentType: String,
+    val title: String,
+    val season: Int?,
+    val episode: Int?,
+    val watchedAt: Long,
+)
+
 interface WatchedSyncAdapter {
     suspend fun pull(
         profileId: Int,
         pageSize: Int,
     ): List<WatchedItem>
 
+    suspend fun getDeltaCursor(profileId: Int): Long? = null
+
+    suspend fun pullDelta(
+        profileId: Int,
+        sinceEventId: Long,
+        limit: Int,
+    ): List<WatchedDeltaEvent> = emptyList()
+
     suspend fun push(
         profileId: Int,
         items: Collection<WatchedItem>,

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

@@ -16,6 +16,7 @@ import com.nuvio.app.features.trakt.TraktProgressRepository
 import com.nuvio.app.features.trakt.TraktSettingsRepository
 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
 import com.nuvio.app.features.watching.sync.ProgressSyncAdapter
 import com.nuvio.app.features.watching.sync.SupabaseProgressSyncAdapter
@@ -40,6 +41,9 @@ import kotlinx.coroutines.withTimeoutOrNull
 
 private const val WATCH_PROGRESS_METADATA_RESOLUTION_CONCURRENCY = 4
 private const val WATCH_PROGRESS_METADATA_RESOLUTION_LIMIT = 64
+private const val WATCH_PROGRESS_DELTA_PAGE_SIZE = 900
+private const val WATCH_PROGRESS_DELTA_OPERATION_UPSERT = "upsert"
+private const val WATCH_PROGRESS_DELTA_OPERATION_DELETE = "delete"
 
 private data class RemoteMetadataResolutionResult(
     val key: Pair<String, String>,
@@ -73,7 +77,9 @@ object WatchProgressRepository {
     private var entriesByVideoId: MutableMap<String, WatchProgressEntry> = mutableMapOf()
     private var metadataResolutionJob: Job? = null
     private var isPullingNuvioSyncFromServer = false
-    private var hasCompletedInitialNuvioSyncPull = false
+    private var lastSuccessfulPushEpochMs = 0L
+    private var deltaCursorEventId = 0L
+    private var deltaInitialized = false
     private var lastAddonMetadataReadyFingerprint: String? = null
     internal var syncAdapter: ProgressSyncAdapter = SupabaseProgressSyncAdapter
 
@@ -155,6 +161,9 @@ object WatchProgressRepository {
         currentProfileId = 1
         lastAddonMetadataReadyFingerprint = null
         entriesByVideoId.clear()
+        lastSuccessfulPushEpochMs = 0L
+        deltaCursorEventId = 0L
+        deltaInitialized = false
         TraktProgressRepository.clearLocalState()
         TraktSettingsRepository.clearLocalState()
         _uiState.value = WatchProgressUiState()
@@ -168,9 +177,17 @@ object WatchProgressRepository {
 
         val payload = WatchProgressStorage.loadPayload(profileId).orEmpty().trim()
         if (payload.isNotEmpty()) {
-            entriesByVideoId = WatchProgressCodec.decodeEntries(payload)
+            val storedPayload = WatchProgressCodec.decodePayload(payload)
+            lastSuccessfulPushEpochMs = storedPayload.lastSuccessfulPushEpochMs
+            deltaCursorEventId = storedPayload.deltaCursorEventId
+            deltaInitialized = storedPayload.deltaInitialized
+            entriesByVideoId = storedPayload.entries
                 .associateBy { it.videoId }
                 .toMutableMap()
+        } else {
+            lastSuccessfulPushEpochMs = 0L
+            deltaCursorEventId = 0L
+            deltaInitialized = false
         }
         publish()
         resolveRemoteMetadata()
@@ -203,37 +220,10 @@ object WatchProgressRepository {
             }
 
             runCatching {
-                val sinceLastWatched = entriesByVideoId.values
-                    .maxOfOrNull { entry -> entry.lastUpdatedEpochMs }
-                    ?.takeIf { hasCompletedInitialNuvioSyncPull }
-                val serverEntries = syncAdapter.pull(
+                pullSupabaseDeltaFromServer(
                     profileId = profileId,
-                    sinceLastWatched = sinceLastWatched,
+                    pullStartedEpochMs = WatchProgressClock.nowEpochMs(),
                 )
-                val isIncrementalPull = sinceLastWatched != null
-                if (isIncrementalPull && serverEntries.isEmpty()) {
-                    hasLoaded = true
-                    hasCompletedInitialNuvioSyncPull = true
-                    return@runCatching
-                }
-                val oldLocal = entriesByVideoId.toMap()
-                val newMap = if (isIncrementalPull) {
-                    entriesByVideoId.toMutableMap()
-                } else {
-                    mutableMapOf()
-                }
-
-                serverEntries.forEach { entry ->
-                    newMap[entry.videoId] = entry.toWatchProgressEntry(cached = oldLocal[entry.videoId])
-                }
-
-                entriesByVideoId = newMap
-                hasLoaded = true
-                hasCompletedInitialNuvioSyncPull = true
-                publish()
-                persist()
-
-                resolveRemoteMetadata()
             }.onFailure { e ->
                 if (e is CancellationException) throw e
                 log.e(e) { "Failed to pull watch progress from server" }
@@ -245,6 +235,140 @@ object WatchProgressRepository {
         }
     }
 
+    private suspend fun pullSupabaseDeltaFromServer(
+        profileId: Int,
+        pullStartedEpochMs: Long,
+    ) {
+        if (!deltaInitialized) {
+            val cursorBeforeSnapshot = try {
+                syncAdapter.getDeltaCursor(profileId)
+            } catch (error: CancellationException) {
+                throw error
+            } catch (error: Throwable) {
+                log.w { "Watch progress delta cursor unavailable, falling back to full pull: ${error.message}" }
+                null
+            }
+            if (cursorBeforeSnapshot == null) {
+                pullFullFromAdapter(
+                    profileId = profileId,
+                    pullStartedEpochMs = pullStartedEpochMs,
+                    resetDeltaState = true,
+                )
+                return
+            }
+
+            pullFullFromAdapter(
+                profileId = profileId,
+                pullStartedEpochMs = pullStartedEpochMs,
+                resetDeltaState = false,
+            )
+            deltaCursorEventId = cursorBeforeSnapshot
+            deltaInitialized = true
+            persist()
+            return
+        }
+
+        var cursor = deltaCursorEventId
+        var changed = false
+
+        while (true) {
+            val events = try {
+                syncAdapter.pullDelta(
+                    profileId = profileId,
+                    sinceEventId = cursor,
+                    limit = WATCH_PROGRESS_DELTA_PAGE_SIZE,
+                )
+            } catch (error: CancellationException) {
+                throw error
+            } catch (error: Throwable) {
+                log.w { "Watch progress delta pull unavailable, falling back to full pull: ${error.message}" }
+                pullFullFromAdapter(
+                    profileId = profileId,
+                    pullStartedEpochMs = pullStartedEpochMs,
+                    resetDeltaState = true,
+                )
+                return
+            }
+            if (events.isEmpty()) break
+
+            changed = applyWatchProgressDeltaEvents(
+                events = events,
+                pullStartedEpochMs = pullStartedEpochMs,
+            ) || changed
+            cursor = maxOf(cursor, events.maxOf { it.eventId })
+            deltaCursorEventId = cursor
+            deltaInitialized = true
+
+            if (events.size < WATCH_PROGRESS_DELTA_PAGE_SIZE) break
+        }
+
+        hasLoaded = true
+        if (changed) {
+            publish()
+            persist()
+            resolveRemoteMetadata()
+        }
+    }
+
+    private suspend fun pullFullFromAdapter(
+        profileId: Int,
+        pullStartedEpochMs: Long,
+        resetDeltaState: Boolean,
+    ) {
+        val serverEntries = syncAdapter.pull(profileId = profileId)
+        entriesByVideoId = mergeWatchProgressEntriesPreservingUnsynced(
+            serverEntries = serverEntries,
+            localEntries = entriesByVideoId.values,
+            lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
+            pullStartedEpochMs = pullStartedEpochMs,
+        ).toMutableMap()
+        if (resetDeltaState) {
+            deltaCursorEventId = 0L
+            deltaInitialized = false
+        }
+        hasLoaded = true
+        publish()
+        persist()
+        resolveRemoteMetadata()
+    }
+
+    private fun applyWatchProgressDeltaEvents(
+        events: Collection<ProgressDeltaEvent>,
+        pullStartedEpochMs: Long,
+    ): Boolean {
+        var changed = false
+        events.forEach { event ->
+            if (event.videoId.isBlank()) return@forEach
+            when (event.operation.lowercase()) {
+                WATCH_PROGRESS_DELTA_OPERATION_UPSERT -> {
+                    val current = entriesByVideoId[event.videoId]
+                    val updated = event.toProgressSyncRecord().toWatchProgressEntry(cached = current)
+                    if (current != updated) {
+                        entriesByVideoId[event.videoId] = updated
+                        changed = true
+                    }
+                }
+                WATCH_PROGRESS_DELTA_OPERATION_DELETE -> {
+                    val localEntry = entriesByVideoId[event.videoId]
+                    if (
+                        localEntry != null &&
+                        shouldPreserveLocalWatchProgressEntry(
+                            localEntry = localEntry,
+                            lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
+                            pullStartedEpochMs = pullStartedEpochMs,
+                        )
+                    ) {
+                        return@forEach
+                    }
+                    if (entriesByVideoId.remove(event.videoId) != null) {
+                        changed = true
+                    }
+                }
+            }
+        }
+        return changed
+    }
+
     private fun ProgressSyncRecord.toWatchProgressEntry(cached: WatchProgressEntry?): WatchProgressEntry =
         WatchProgressEntry(
             contentType = contentType,
@@ -271,6 +395,56 @@ object WatchProgressRepository {
             isCompleted = isWatchProgressComplete(position, duration, false),
         )
 
+    private fun ProgressDeltaEvent.toProgressSyncRecord(): ProgressSyncRecord =
+        ProgressSyncRecord(
+            contentId = contentId,
+            contentType = contentType,
+            videoId = videoId,
+            season = season,
+            episode = episode,
+            position = position,
+            duration = duration,
+            lastWatched = lastWatched,
+        )
+
+    private fun mergeWatchProgressEntriesPreservingUnsynced(
+        serverEntries: Collection<ProgressSyncRecord>,
+        localEntries: Collection<WatchProgressEntry>,
+        lastSuccessfulPushEpochMs: Long,
+        pullStartedEpochMs: Long,
+    ): Map<String, WatchProgressEntry> {
+        val localByVideoId = localEntries.associateBy { entry -> entry.videoId }
+        val merged = serverEntries.associate { record ->
+            record.videoId to record.toWatchProgressEntry(cached = localByVideoId[record.videoId])
+        }.toMutableMap()
+
+        localByVideoId.forEach { (videoId, localEntry) ->
+            val remoteEntry = merged[videoId]
+            val shouldPreserve = shouldPreserveLocalWatchProgressEntry(
+                localEntry = localEntry,
+                lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
+                pullStartedEpochMs = pullStartedEpochMs,
+            )
+            if (!shouldPreserve) return@forEach
+            if (remoteEntry == null || localEntry.lastUpdatedEpochMs > remoteEntry.lastUpdatedEpochMs) {
+                merged[videoId] = localEntry
+            }
+        }
+
+        return merged
+    }
+
+    private fun shouldPreserveLocalWatchProgressEntry(
+        localEntry: WatchProgressEntry,
+        lastSuccessfulPushEpochMs: Long,
+        pullStartedEpochMs: Long,
+    ): Boolean {
+        val updatedAt = localEntry.lastUpdatedEpochMs
+        val wasUpdatedAfterLastPush = lastSuccessfulPushEpochMs > 0L && updatedAt > lastSuccessfulPushEpochMs
+        val wasUpdatedDuringPull = pullStartedEpochMs > 0L && updatedAt >= pullStartedEpochMs
+        return wasUpdatedAfterLastPush || wasUpdatedDuringPull
+    }
+
     private fun retryMetadataResolutionWhenAddonMetaProvidersReady(state: AddonsUiState) {
         if (!hasLoaded || shouldUseTraktProgress()) return
 
@@ -609,6 +783,7 @@ object WatchProgressRepository {
             runCatching {
                 val profileId = ProfileRepository.activeProfileId
                 syncAdapter.push(profileId = profileId, entries = listOf(entry))
+                recordSuccessfulPush(profileId = profileId, entries = listOf(entry))
             }.onFailure { e ->
                 log.e(e) { "Failed to push watch progress scrobble" }
             }
@@ -645,10 +820,27 @@ object WatchProgressRepository {
     private fun persist() {
         WatchProgressStorage.savePayload(
             currentProfileId,
-            WatchProgressCodec.encodeEntries(entriesByVideoId.values),
+            WatchProgressCodec.encodePayload(
+                entries = entriesByVideoId.values,
+                lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
+                deltaCursorEventId = deltaCursorEventId,
+                deltaInitialized = deltaInitialized,
+            ),
         )
     }
 
+    private fun recordSuccessfulPush(profileId: Int, entries: Collection<WatchProgressEntry>) {
+        if (profileId != currentProfileId) return
+        val latestPushed = entries
+            .asSequence()
+            .map { entry -> entry.lastUpdatedEpochMs }
+            .maxOrNull()
+            ?: return
+        if (latestPushed <= lastSuccessfulPushEpochMs) return
+        lastSuccessfulPushEpochMs = latestPushed
+        persist()
+    }
+
     private fun shouldUseTraktProgress(): Boolean =
         shouldUseTraktProgressSource(
             isAuthenticated = TraktAuthRepository.isAuthenticated.value,

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRules.kt

@@ -15,8 +15,11 @@ import kotlinx.serialization.json.Json
 internal const val ContinueWatchingLimit = DefaultContinueWatchingLimit
 
 @Serializable
-private data class StoredWatchProgressPayload(
+internal data class StoredWatchProgressPayload(
     val entries: List<WatchProgressEntry> = emptyList(),
+    val lastSuccessfulPushEpochMs: Long = 0L,
+    val deltaCursorEventId: Long = 0L,
+    val deltaInitialized: Boolean = false,
 )
 
 internal object WatchProgressCodec {
@@ -26,15 +29,37 @@ internal object WatchProgressCodec {
     }
 
     fun decodeEntries(payload: String): List<WatchProgressEntry> =
+        decodePayload(payload).entries
+
+    fun decodePayload(payload: String): StoredWatchProgressPayload =
         runCatching {
-            json.decodeFromString<StoredWatchProgressPayload>(payload).entries
-                .map(WatchProgressEntry::normalizedCompletion)
-        }.getOrDefault(emptyList())
+            json.decodeFromString<StoredWatchProgressPayload>(payload).let { storedPayload ->
+                storedPayload.copy(
+                    entries = storedPayload.entries.map(WatchProgressEntry::normalizedCompletion),
+                )
+            }
+        }.getOrDefault(StoredWatchProgressPayload())
 
     fun encodeEntries(entries: Collection<WatchProgressEntry>): String =
+        encodePayload(
+            entries = entries,
+            lastSuccessfulPushEpochMs = 0L,
+            deltaCursorEventId = 0L,
+            deltaInitialized = false,
+        )
+
+    fun encodePayload(
+        entries: Collection<WatchProgressEntry>,
+        lastSuccessfulPushEpochMs: Long,
+        deltaCursorEventId: Long,
+        deltaInitialized: Boolean,
+    ): String =
         json.encodeToString(
             StoredWatchProgressPayload(
                 entries = entries.toList().sortedByDescending { it.lastUpdatedEpochMs },
+                lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
+                deltaCursorEventId = deltaCursorEventId,
+                deltaInitialized = deltaInitialized,
             ),
         )
 }