From 9e83efd5fdb4fcada3191eea7c0c272e0bca26a2 Mon Sep 17 00:00:00 2001 From: tapframe <85391825+tapframe@users.noreply.github.com> Date: Mon, 1 Jun 2026 15:05:16 +0530 Subject: [PATCH] Add mobile delta sync for progress Persist cursors for watched items and watch progress. Use snapshot fallback when delta RPCs are unavailable. --- .../app/features/watched/WatchedRepository.kt | 176 ++++++++++-- .../watching/sync/ProgressSyncAdapter.kt | 22 ++ .../sync/SupabaseProgressSyncAdapter.kt | 52 ++++ .../sync/SupabaseWatchedSyncAdapter.kt | 46 ++++ .../watching/sync/WatchedSyncAdapter.kt | 19 ++ .../watchprogress/WatchProgressRepository.kt | 256 +++++++++++++++--- .../watchprogress/WatchProgressRules.kt | 33 ++- 7 files changed, 547 insertions(+), 57 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt index 66050e9b..3010a162 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt +++ b/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 = 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 = 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(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, + 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, + 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, + 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, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/ProgressSyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/ProgressSyncAdapter.kt index 3109c4fc..dfb8d48e 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/ProgressSyncAdapter.kt +++ b/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 + suspend fun getDeltaCursor(profileId: Int): Long? = null + + suspend fun pullDelta( + profileId: Int, + sinceEventId: Long, + limit: Int, + ): List = emptyList() + suspend fun push( profileId: Int, entries: Collection, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseProgressSyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseProgressSyncAdapter.kt index 6743de34..8521e4cd 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseProgressSyncAdapter.kt +++ b/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() + } + + override suspend fun pullDelta( + profileId: Int, + sinceEventId: Long, + limit: Int, + ): List { + 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().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, +) diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseWatchedSyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseWatchedSyncAdapter.kt index cab9e553..a9b52395 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/SupabaseWatchedSyncAdapter.kt +++ b/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() + } + + override suspend fun pullDelta( + profileId: Int, + sinceEventId: Long, + limit: Int, + ): List { + 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().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, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/WatchedSyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/WatchedSyncAdapter.kt index f86349c2..1e77f43f 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/WatchedSyncAdapter.kt +++ b/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 + suspend fun getDeltaCursor(profileId: Int): Long? = null + + suspend fun pullDelta( + profileId: Int, + sinceEventId: Long, + limit: Int, + ): List = emptyList() + suspend fun push( profileId: Int, items: Collection, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRepository.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRepository.kt index 37d8977b..74fa816b 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRepository.kt +++ b/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, @@ -73,7 +77,9 @@ object WatchProgressRepository { private var entriesByVideoId: MutableMap = 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, + 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, + localEntries: Collection, + lastSuccessfulPushEpochMs: Long, + pullStartedEpochMs: Long, + ): Map { + 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) { + 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, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRules.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRules.kt index bd002706..bcaedb77 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watchprogress/WatchProgressRules.kt +++ b/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 = 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 = + decodePayload(payload).entries + + fun decodePayload(payload: String): StoredWatchProgressPayload = runCatching { - json.decodeFromString(payload).entries - .map(WatchProgressEntry::normalizedCompletion) - }.getOrDefault(emptyList()) + json.decodeFromString(payload).let { storedPayload -> + storedPayload.copy( + entries = storedPayload.entries.map(WatchProgressEntry::normalizedCompletion), + ) + } + }.getOrDefault(StoredWatchProgressPayload()) fun encodeEntries(entries: Collection): String = + encodePayload( + entries = entries, + lastSuccessfulPushEpochMs = 0L, + deltaCursorEventId = 0L, + deltaInitialized = false, + ) + + fun encodePayload( + entries: Collection, + lastSuccessfulPushEpochMs: Long, + deltaCursorEventId: Long, + deltaInitialized: Boolean, + ): String = json.encodeToString( StoredWatchProgressPayload( entries = entries.toList().sortedByDescending { it.lastUpdatedEpochMs }, + lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs, + deltaCursorEventId = deltaCursorEventId, + deltaInitialized = deltaInitialized, ), ) }