From 1b05c222b74353573b75562efd005164de8a3c87 Mon Sep 17 00:00:00 2001 From: tapframe <85391825+tapframe@users.noreply.github.com> Date: Wed, 29 Jul 2026 01:06:46 +0530 Subject: [PATCH] feat(library): add incremental delta sync --- .../app/features/library/LibraryLocalState.kt | 150 +++++++-- .../app/features/library/LibraryRepository.kt | 288 ++++++++---------- .../features/library/LibraryStoragePayload.kt | 39 +++ .../features/library/LibrarySyncReconciler.kt | 89 ++++++ .../library/sync/LibrarySyncAdapter.kt | 48 +++ .../library/sync/LibrarySyncPaging.kt | 52 ++++ .../sync/SupabaseLibrarySyncAdapter.kt | 191 ++++++++++++ .../features/library/LibraryDeltaStateTest.kt | 152 +++++++++ .../library/LibrarySyncReconcilerTest.kt | 121 ++++++++ .../library/sync/LibrarySyncPagingTest.kt | 94 ++++++ 10 files changed, 1026 insertions(+), 198 deletions(-) create mode 100644 composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryStoragePayload.kt create mode 100644 composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibrarySyncReconciler.kt create mode 100644 composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncAdapter.kt create mode 100644 composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPaging.kt create mode 100644 composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/SupabaseLibrarySyncAdapter.kt create mode 100644 composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryDeltaStateTest.kt create mode 100644 composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibrarySyncReconcilerTest.kt create mode 100644 composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPagingTest.kt 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 2d48b5fa..d6e9a492 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 @@ -1,5 +1,8 @@ package com.nuvio.app.features.library +import com.nuvio.app.features.library.sync.LibraryDeltaEvent +import com.nuvio.app.features.library.sync.LibrarySyncKey +import com.nuvio.app.features.library.sync.toLibrarySyncKey import kotlinx.atomicfu.locks.SynchronizedObject import kotlinx.atomicfu.locks.synchronized import kotlinx.coroutines.Job @@ -16,8 +19,14 @@ internal data class LibraryLocalSnapshot( val hasLoaded: Boolean, val isLoading: Boolean, val items: List, - val hasPendingPush: Boolean, -) + val deltaCursorEventId: Long, + val deltaInitialized: Boolean, + val pendingUpsertKeys: List, + val pendingDeleteKeys: List, +) { + val hasPendingPush: Boolean + get() = pendingUpsertKeys.isNotEmpty() || pendingDeleteKeys.isNotEmpty() +} internal data class LibraryStateTransition( val snapshot: LibraryLocalSnapshot, @@ -60,7 +69,10 @@ internal class LibraryLocalState { private var contentRevision = 0L private var isLoading = false private var itemsById: MutableMap = mutableMapOf() - private var hasPendingPush = false + private var deltaCursorEventId = 0L + private var deltaInitialized = false + private var pendingUpsertKeysByKey: MutableMap = mutableMapOf() + private var pendingDeleteKeysByKey: MutableMap = mutableMapOf() private var pushJob: Job? = null fun snapshot(): LibraryLocalSnapshot = synchronized(lock) { @@ -124,7 +136,10 @@ internal class LibraryLocalState { hasLoaded = false isLoading = true itemsById = mutableMapOf() - hasPendingPush = false + deltaCursorEventId = 0L + deltaInitialized = false + pendingUpsertKeysByKey = mutableMapOf() + pendingDeleteKeysByKey = mutableMapOf() LibraryStateTransition( snapshot = snapshotLocked(), detachedPushJob = detachedPushJob, @@ -135,11 +150,24 @@ internal class LibraryLocalState { token: LibraryProfileToken, activeProfileId: Int, items: Collection, + deltaCursorEventId: Long = 0L, + deltaInitialized: Boolean = false, + pendingUpsertKeys: Collection = emptyList(), + pendingDeleteKeys: Collection = emptyList(), ): LibraryLocalSnapshot? = synchronized(lock) { if (activeProfileId != token.profileId || !isCurrentLocked(token)) { return@synchronized null } itemsById = items.associateByTo(mutableMapOf()) { libraryItemKey(it.id, it.type) } + this.deltaCursorEventId = deltaCursorEventId.coerceAtLeast(0L) + this.deltaInitialized = deltaInitialized + pendingUpsertKeysByKey = pendingUpsertKeys + .associateByTo(mutableMapOf()) { libraryItemKey(it.contentId, it.contentType) } + .filterToExistingItems(itemsById) + pendingDeleteKeysByKey = pendingDeleteKeys + .associateByTo(mutableMapOf()) { libraryItemKey(it.contentId, it.contentType) } + .apply { pendingUpsertKeysByKey.keys.forEach(::remove) } + pendingDeleteKeysByKey.keys.forEach(itemsById::remove) hasLoaded = true isLoading = false revision += 1L @@ -157,7 +185,10 @@ internal class LibraryLocalState { hasLoaded = false isLoading = false itemsById = mutableMapOf() - hasPendingPush = false + deltaCursorEventId = 0L + deltaInitialized = false + pendingUpsertKeysByKey = mutableMapOf() + pendingDeleteKeysByKey = mutableMapOf() LibraryStateTransition( snapshot = snapshotLocked(), detachedPushJob = detachedPushJob, @@ -172,45 +203,82 @@ internal class LibraryLocalState { fun applyServerItems( pullSnapshot: LibraryLocalSnapshot, serverItems: Collection, + cursorEventId: Long = 0L, ): LibraryServerItemsApplyResult? = synchronized(lock) { if (!isCurrentLocked(pullSnapshot.token)) return@synchronized null - 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) } + val reconciliation = reconcileLibrarySnapshot( + serverItems = serverItems, + localItemsByKey = itemsById, + pendingUpsertKeysByKey = pendingUpsertKeysByKey, + pendingDeleteKeysByKey = pendingDeleteKeysByKey, + preserveLegacyLocalWhenServerEmpty = !pullSnapshot.deltaInitialized, + ) + if (itemsById != reconciliation.itemsByKey) { contentRevision += 1L - hasPendingPush = false } + itemsById = reconciliation.itemsByKey + pendingUpsertKeysByKey = reconciliation.pendingUpsertKeysByKey + pendingDeleteKeysByKey = reconciliation.pendingDeleteKeysByKey + deltaCursorEventId = cursorEventId.coerceAtLeast(0L) + deltaInitialized = true hasLoaded = true isLoading = false revision += 1L LibraryServerItemsApplyResult( snapshot = snapshotLocked(), - preservedLocalItems = preserveLocalItems, + preservedLocalItems = reconciliation.preservedLocalItems, ) } + fun applyDeltaEvents( + token: LibraryProfileToken, + events: Collection, + ): LibraryLocalSnapshot? = synchronized(lock) { + if (!isCurrentLocked(token)) return@synchronized null + + val reconciliation = reconcileLibraryDelta( + events = events, + currentItemsByKey = itemsById, + pendingUpsertKeysByKey = pendingUpsertKeysByKey, + pendingDeleteKeysByKey = pendingDeleteKeysByKey, + currentCursorEventId = deltaCursorEventId, + ) + if (reconciliation.changed) { + itemsById = reconciliation.itemsByKey + contentRevision += 1L + } + deltaCursorEventId = reconciliation.cursorEventId + deltaInitialized = true + revision += 1L + snapshotLocked() + } + fun upsert(item: LibraryItem): LibraryLocalSnapshot = synchronized(lock) { - itemsById[libraryItemKey(item.id, item.type)] = item + val key = libraryItemKey(item.id, item.type) + itemsById[key] = item + pendingUpsertKeysByKey[key] = item.toLibrarySyncKey() + pendingDeleteKeysByKey.remove(key) revision += 1L contentRevision += 1L - hasPendingPush = true snapshotLocked() } fun toggle(item: LibraryItem): LibraryLocalToggleResult = synchronized(lock) { val key = libraryItemKey(item.id, item.type) - val isSaved = if (itemsById.remove(key) != null) { + val removedItem = itemsById.remove(key) + val isSaved = if (removedItem != null) { + pendingUpsertKeysByKey.remove(key) + pendingDeleteKeysByKey[key] = removedItem.toLibrarySyncKey() false } else { itemsById[key] = item + pendingUpsertKeysByKey[key] = item.toLibrarySyncKey() + pendingDeleteKeysByKey.remove(key) true } revision += 1L contentRevision += 1L - hasPendingPush = true LibraryLocalToggleResult( snapshot = snapshotLocked(), isSaved = isSaved, @@ -218,13 +286,17 @@ internal class LibraryLocalState { } fun removeById(id: String): LibraryLocalMutation = synchronized(lock) { - val before = itemsById.size - itemsById.entries.removeAll { (_, item) -> item.id == id } - val affectedCount = before - itemsById.size + val removedEntries = itemsById + .filterValues { item -> item.id == id } + removedEntries.forEach { (key, item) -> + itemsById.remove(key) + pendingUpsertKeysByKey.remove(key) + pendingDeleteKeysByKey[key] = item.toLibrarySyncKey() + } + val affectedCount = removedEntries.size if (affectedCount > 0) { revision += 1L contentRevision += 1L - hasPendingPush = true } LibraryLocalMutation( snapshot = snapshotLocked(), @@ -233,11 +305,14 @@ internal class LibraryLocalState { } fun remove(id: String, type: String): LibraryLocalMutation = synchronized(lock) { - val affectedCount = if (itemsById.remove(libraryItemKey(id, type)) != null) 1 else 0 - if (affectedCount > 0) { + val key = libraryItemKey(id, type) + val removedItem = itemsById.remove(key) + val affectedCount = if (removedItem != null) 1 else 0 + if (removedItem != null) { + pendingUpsertKeysByKey.remove(key) + pendingDeleteKeysByKey[key] = removedItem.toLibrarySyncKey() revision += 1L contentRevision += 1L - hasPendingPush = true } LibraryLocalMutation( snapshot = snapshotLocked(), @@ -261,7 +336,7 @@ internal class LibraryLocalState { snapshot: LibraryLocalSnapshot, job: Job, ): LibraryPushJobInstallResult = synchronized(lock) { - if (!isContentCurrentLocked(snapshot)) { + if (!isCurrentLocked(snapshot)) { LibraryPushJobInstallResult(installed = false, detachedPushJob = null) } else { val detachedPushJob = pushJob @@ -276,11 +351,14 @@ internal class LibraryLocalState { } } - fun markPushCompleted(snapshot: LibraryLocalSnapshot) { - synchronized(lock) { - if (isContentCurrentLocked(snapshot)) { - hasPendingPush = false - } + fun markPushCompleted(snapshot: LibraryLocalSnapshot): LibraryLocalSnapshot? = synchronized(lock) { + if (!isCurrentLocked(snapshot)) { + null + } else { + pendingUpsertKeysByKey.clear() + pendingDeleteKeysByKey.clear() + revision += 1L + snapshotLocked() } } @@ -298,7 +376,10 @@ internal class LibraryLocalState { hasLoaded = hasLoaded, isLoading = isLoading, items = itemsById.values.toList(), - hasPendingPush = hasPendingPush, + deltaCursorEventId = deltaCursorEventId, + deltaInitialized = deltaInitialized, + pendingUpsertKeys = pendingUpsertKeysByKey.values.toList(), + pendingDeleteKeys = pendingDeleteKeysByKey.values.toList(), ) private fun isCurrentLocked(token: LibraryProfileToken): Boolean = @@ -311,5 +392,12 @@ internal class LibraryLocalState { isCurrentLocked(snapshot.token) && contentRevision == snapshot.contentRevision } -private fun libraryItemKey(id: String, type: String): String = +private fun MutableMap.filterToExistingItems( + itemsById: Map, +): MutableMap = + apply { + keys.retainAll(itemsById.keys) + } + +internal 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 cb9fbebe..5dfb8d3b 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 @@ -4,9 +4,11 @@ import co.touchlab.kermit.Logger import com.nuvio.app.core.auth.AuthRepository import com.nuvio.app.core.ui.NuvioToastController import com.nuvio.app.core.auth.AuthState -import com.nuvio.app.core.network.SupabaseProvider -import com.nuvio.app.core.sync.putSyncOriginClientId -import com.nuvio.app.features.home.PosterShape +import com.nuvio.app.features.library.sync.LibrarySyncAdapter +import com.nuvio.app.features.library.sync.SupabaseLibrarySyncAdapter +import com.nuvio.app.features.library.sync.consumeCursorPages +import com.nuvio.app.features.library.sync.libraryDeltaPageSize +import com.nuvio.app.features.library.sync.librarySnapshotPageSize import com.nuvio.app.features.profiles.ProfileRepository import com.nuvio.app.features.trakt.TraktAuthRepository import com.nuvio.app.features.trakt.TraktLibraryRepository @@ -16,8 +18,6 @@ import com.nuvio.app.features.trakt.TraktMembershipChanges import com.nuvio.app.features.trakt.TraktSettingsRepository import com.nuvio.app.features.trakt.effectiveLibrarySourceMode as resolveEffectiveLibrarySourceMode import com.nuvio.app.features.trakt.shouldUseTraktLibrary -import io.github.jan.supabase.postgrest.postgrest -import io.github.jan.supabase.postgrest.rpc import kotlinx.atomicfu.locks.SynchronizedObject import kotlinx.atomicfu.locks.synchronized import kotlinx.coroutines.CancellationException @@ -36,14 +36,6 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock -import kotlinx.serialization.SerialName -import kotlinx.serialization.Serializable -import kotlinx.serialization.decodeFromString -import kotlinx.serialization.encodeToString -import kotlinx.serialization.json.Json -import kotlinx.serialization.json.buildJsonObject -import kotlinx.serialization.json.encodeToJsonElement -import kotlinx.serialization.json.put import nuvio.composeapp.generated.resources.Res import nuvio.composeapp.generated.resources.library_local_tab_title import nuvio.composeapp.generated.resources.library_other @@ -51,45 +43,21 @@ import nuvio.composeapp.generated.resources.trakt_lists_update_failed import org.jetbrains.compose.resources.StringResource import org.jetbrains.compose.resources.getString -@Serializable -private data class StoredLibraryPayload( - val items: List = emptyList(), -) - -@Serializable -private data class LibrarySyncItem( - @SerialName("content_id") val contentId: String, - @SerialName("content_type") val contentType: String, - val name: String = "", - val poster: String? = null, - @SerialName("poster_shape") val posterShape: String = "POSTER", - val background: String? = null, - val description: String? = null, - @SerialName("release_info") val releaseInfo: String? = null, - @SerialName("imdb_rating") val imdbRating: Float? = null, - val genres: List = emptyList(), - @SerialName("addon_base_url") val addonBaseUrl: String? = null, - @SerialName("added_at") val addedAt: Long = 0, -) - object LibraryRepository { - private const val PULL_PAGE_SIZE = 500 + private const val pushDebounceMs = 500L private val syncScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) private val log = Logger.withTag("LibraryRepository") - private val json = Json { - ignoreUnknownKeys = true - encodeDefaults = true - } private val _uiState = MutableStateFlow(LibraryUiState()) val uiState: StateFlow = _uiState.asStateFlow() private val localState = LibraryLocalState() private val loadLock = SynchronizedObject() - private val nuvioPullMutex = Mutex() + private val nuvioSyncMutex = Mutex() private val persistenceLock = SynchronizedObject() - private val lastPersistedContentRevisionByProfile = mutableMapOf() + private val lastPersistedRevisionByProfile = mutableMapOf() + internal var syncAdapter: LibrarySyncAdapter = SupabaseLibrarySyncAdapter init { syncScope.launch { @@ -177,7 +145,7 @@ object LibraryRepository { try { wipeStorage() } finally { - lastPersistedContentRevisionByProfile.clear() + lastPersistedRevisionByProfile.clear() } } } @@ -203,18 +171,20 @@ object LibraryRepository { private fun completeLoadFromDisk(token: LibraryProfileToken): Boolean { val payload = LibraryStorage.loadPayload(token.profileId).orEmpty().trim() - val items = if (payload.isNotEmpty()) { - runCatching { - json.decodeFromString(payload).items - }.getOrDefault(emptyList()) + val storedPayload = if (payload.isNotEmpty()) { + LibraryStoragePayloadCodec.decode(payload) } else { - emptyList() + StoredLibraryPayload() } return localState.completeProfileLoad( token = token, activeProfileId = ProfileRepository.activeProfileId, - items = items, + items = storedPayload.items, + deltaCursorEventId = storedPayload.deltaCursorEventId, + deltaInitialized = storedPayload.deltaInitialized, + pendingUpsertKeys = storedPayload.pendingUpsertKeys, + pendingDeleteKeys = storedPayload.pendingDeleteKeys, ) != null } @@ -223,6 +193,7 @@ object LibraryRepository { log.d { "Skipping library pull for inactive profile $profileId" } return } + var serializedOperationToken: LibraryProfileToken? = null if (isTraktLibrarySourceActive()) { try { @@ -237,32 +208,77 @@ object LibraryRepository { return } - nuvioPullMutex.withLock { + nuvioSyncMutex.withLock { val serializedToken = activeOperationToken(profileId) ?: return@withLock + serializedOperationToken = serializedToken val pullSnapshot = localState.markPullStarted(serializedToken) ?: return@withLock - var appliedItems = false try { - val serverItems = pullAllLibrarySyncItems(profileId).map { it.toLibraryItem() } - val applyResult = localState.applyServerItems(pullSnapshot, serverItems) - ?: return@withLock - appliedItems = true - if (applyResult.preservedLocalItems) { - log.w { - "Preserving ${applyResult.snapshot.items.size} local library items because the remote " + - "snapshot is empty or local changes are pending" - } - } else { + if (!pullSnapshot.deltaInitialized) { + val cursorBeforeSnapshot = syncAdapter.getDeltaCursor(profileId) + val serverItems = syncAdapter.pullSnapshot( + profileId = profileId, + pageSize = librarySnapshotPageSize, + ) + val applyResult = localState.applyServerItems( + pullSnapshot = pullSnapshot, + serverItems = serverItems, + cursorEventId = cursorBeforeSnapshot, + ) ?: return@withLock persist(applyResult.snapshot) + publish() + if (applyResult.preservedLocalItems) { + log.i { + "Merged pending local library changes during snapshot bootstrap " + + "profile=$profileId items=${applyResult.snapshot.items.size}" + } + } } + pullLibraryDelta(serializedToken, profileId) } catch (error: CancellationException) { throw error } catch (error: Throwable) { log.e(error) { "Failed to pull library from server" } } - - if (appliedItems) publish() } + val completedToken = serializedOperationToken ?: operationToken + val pendingSnapshot = localState.snapshot() + if (pendingSnapshot.token == completedToken && isActiveOperation(completedToken)) { + pushToServer(pendingSnapshot, delayMs = 0L) + } + } + + private suspend fun pullLibraryDelta( + token: LibraryProfileToken, + profileId: Int, + ) { + val initialSnapshot = localState.snapshot() + if (initialSnapshot.token != token) return + consumeCursorPages( + initialCursor = initialSnapshot.deltaCursorEventId, + pageSize = libraryDeltaPageSize, + fetchPage = { cursor, limit -> + if (isActiveOperation(token)) { + syncAdapter.pullDelta( + profileId = profileId, + sinceEventId = cursor, + limit = limit, + ) + } else { + emptyList() + } + }, + applyPage = { events, _ -> + if (!isActiveOperation(token)) { + null + } else { + localState.applyDeltaEvents(token, events)?.also { snapshot -> + persist(snapshot) + publish() + }?.deltaCursorEventId + } + }, + ) } private fun activeOperationToken(profileId: Int): LibraryProfileToken? { @@ -447,10 +463,13 @@ object LibraryRepository { applyMembershipChanges(item, desiredMembership) } - private fun pushToServer(snapshot: LibraryLocalSnapshot) { + private fun pushToServer( + snapshot: LibraryLocalSnapshot, + delayMs: Long = pushDebounceMs, + ) { + if (!snapshot.hasPendingPush) return val authState = AuthRepository.state.value val profileId = snapshot.token.profileId - val itemCount = snapshot.items.size if (authState !is AuthState.Authenticated) { log.w { "Skipping library push: auth state is ${authState::class.simpleName} profile=$profileId" } return @@ -460,38 +479,42 @@ object LibraryRepository { 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}" + delay(delayMs) + nuvioSyncMutex.withLock { + if (!localState.isCurrent(snapshot)) { + val current = localState.snapshot() + log.d { + "Skipping stale debounced library push scheduled=${snapshot.token} " + + "current=${current.token} scheduledRevision=${snapshot.revision} " + + "currentRevision=${current.revision}" + } + return@withLock } - return@launch - } - 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 + val currentAuthState = AuthRepository.state.value + if (currentAuthState !is AuthState.Authenticated || currentAuthState.isAnonymous) { + return@withLock } - val params = buildJsonObject { - put("p_profile_id", profileId) - put("p_items", json.encodeToJsonElement(syncItems)) - putSyncOriginClientId() + runCatching { + val itemsByKey = snapshot.items.associateBy { item -> + libraryItemKey(item.id, item.type) + } + val upsertItems = snapshot.pendingUpsertKeys.mapNotNull { key -> + itemsByKey[libraryItemKey(key.contentId, key.contentType)] + } + syncAdapter.pushItems(profileId, upsertItems) + syncAdapter.deleteItems(profileId, snapshot.pendingDeleteKeys) + localState.markPushCompleted(snapshot)?.let(::persist) + log.i { + "Library delta push completed profile=$profileId " + + "upserts=${upsertItems.size} deletes=${snapshot.pendingDeleteKeys.size}" + } + }.onFailure { error -> + if (error is CancellationException) throw error + log.e(error) { + "Failed to push library delta profile=$profileId " + + "upserts=${snapshot.pendingUpsertKeys.size} deletes=${snapshot.pendingDeleteKeys.size}" + } } - 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) - log.i { "Library push completed profile=$profileId itemCount=$itemCount" } - } - }.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) } @@ -505,27 +528,6 @@ object LibraryRepository { pushJob.start() } - private suspend fun pullAllLibrarySyncItems(profileId: Int): List { - val allItems = mutableListOf() - var offset = 0 - - while (true) { - val params = buildJsonObject { - put("p_profile_id", profileId) - put("p_limit", PULL_PAGE_SIZE) - put("p_offset", offset) - } - val result = SupabaseProvider.client.postgrest.rpc("sync_pull_library", params) - val page = result.decodeList() - allItems.addAll(page) - - if (page.size < PULL_PAGE_SIZE) break - offset += PULL_PAGE_SIZE - } - - return allItems - } - private fun publish() { val localSnapshot = localState.snapshot() if (isTraktLibrarySourceActive()) { @@ -584,18 +586,14 @@ object LibraryRepository { } private fun persist(snapshot: LibraryLocalSnapshot) { - val payload = json.encodeToString( - StoredLibraryPayload( - items = snapshot.items.sortedByDescending { it.savedAtEpochMs }, - ), - ) + val payload = LibraryStoragePayloadCodec.encode(snapshot) synchronized(persistenceLock) { val profileId = snapshot.token.profileId - val lastPersistedRevision = lastPersistedContentRevisionByProfile[profileId] ?: Long.MIN_VALUE - if (snapshot.contentRevision <= lastPersistedRevision) return@synchronized - localState.runIfContentCurrent(snapshot) { + val lastPersistedRevision = lastPersistedRevisionByProfile[profileId] ?: Long.MIN_VALUE + if (snapshot.revision <= lastPersistedRevision) return@synchronized + localState.runIfCurrent(snapshot) { LibraryStorage.savePayload(profileId, payload) - lastPersistedContentRevisionByProfile[profileId] = snapshot.contentRevision + lastPersistedRevisionByProfile[profileId] = snapshot.revision } } } @@ -656,50 +654,6 @@ internal fun libraryMembershipWithRemovedList( this[listKey] = false } -private fun LibrarySyncItem.toLibraryItem(): LibraryItem = LibraryItem( - id = contentId, - type = contentType, - name = name, - poster = poster, - banner = background, - description = description, - releaseInfo = releaseInfo, - imdbRating = imdbRating?.toString(), - genres = genres, - posterShape = posterShape.toPosterShape(), - addonBaseUrl = addonBaseUrl, - savedAtEpochMs = addedAt, -) - -private fun LibraryItem.toSyncItem(): LibrarySyncItem = LibrarySyncItem( - contentId = id, - contentType = type, - name = name, - poster = poster, - posterShape = posterShape.toSyncName(), - background = banner, - description = description, - releaseInfo = releaseInfo, - imdbRating = imdbRating?.toFloatOrNull(), - genres = genres, - addonBaseUrl = addonBaseUrl, - addedAt = savedAtEpochMs, -) - -private fun String.toPosterShape(): PosterShape = - when (trim().uppercase()) { - "LANDSCAPE" -> PosterShape.Landscape - "SQUARE" -> PosterShape.Square - else -> PosterShape.Poster - } - -private fun PosterShape.toSyncName(): String = - when (this) { - PosterShape.Poster -> "POSTER" - PosterShape.Square -> "SQUARE" - PosterShape.Landscape -> "LANDSCAPE" - } - internal fun String.toLibraryDisplayTitle(): String { val normalized = trim() if (normalized.isBlank()) return localizedLibraryOtherTitle() diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryStoragePayload.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryStoragePayload.kt new file mode 100644 index 00000000..4448813c --- /dev/null +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryStoragePayload.kt @@ -0,0 +1,39 @@ +package com.nuvio.app.features.library + +import com.nuvio.app.features.library.sync.LibrarySyncKey +import kotlinx.serialization.Serializable +import kotlinx.serialization.decodeFromString +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json + +@Serializable +internal data class StoredLibraryPayload( + val items: List = emptyList(), + val deltaCursorEventId: Long = 0L, + val deltaInitialized: Boolean = false, + val pendingUpsertKeys: List = emptyList(), + val pendingDeleteKeys: List = emptyList(), +) + +internal object LibraryStoragePayloadCodec { + private val json = Json { + ignoreUnknownKeys = true + encodeDefaults = true + } + + fun decode(payload: String): StoredLibraryPayload = + runCatching { + json.decodeFromString(payload) + }.getOrDefault(StoredLibraryPayload()) + + fun encode(snapshot: LibraryLocalSnapshot): String = + json.encodeToString( + StoredLibraryPayload( + items = snapshot.items.sortedByDescending(LibraryItem::savedAtEpochMs), + deltaCursorEventId = snapshot.deltaCursorEventId, + deltaInitialized = snapshot.deltaInitialized, + pendingUpsertKeys = snapshot.pendingUpsertKeys, + pendingDeleteKeys = snapshot.pendingDeleteKeys, + ), + ) +} diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibrarySyncReconciler.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibrarySyncReconciler.kt new file mode 100644 index 00000000..83641e52 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibrarySyncReconciler.kt @@ -0,0 +1,89 @@ +package com.nuvio.app.features.library + +import com.nuvio.app.features.library.sync.LibraryDeltaEvent +import com.nuvio.app.features.library.sync.LibrarySyncKey +import com.nuvio.app.features.library.sync.toLibrarySyncKey + +internal data class LibrarySnapshotReconciliation( + val itemsByKey: MutableMap, + val pendingUpsertKeysByKey: MutableMap, + val pendingDeleteKeysByKey: MutableMap, + val preservedLocalItems: Boolean, +) + +internal data class LibraryDeltaReconciliation( + val itemsByKey: MutableMap, + val changed: Boolean, + val cursorEventId: Long, +) + +internal fun reconcileLibrarySnapshot( + serverItems: Collection, + localItemsByKey: Map, + pendingUpsertKeysByKey: Map, + pendingDeleteKeysByKey: Map, + preserveLegacyLocalWhenServerEmpty: Boolean, +): LibrarySnapshotReconciliation { + val serverItemsByKey = serverItems.associateByTo(mutableMapOf()) { + libraryItemKey(it.id, it.type) + } + val pendingUpserts = pendingUpsertKeysByKey.toMutableMap() + val pendingDeletes = pendingDeleteKeysByKey.toMutableMap() + val migrateLegacyLocalItems = + preserveLegacyLocalWhenServerEmpty && + serverItemsByKey.isEmpty() && + localItemsByKey.isNotEmpty() && + pendingUpserts.isEmpty() && + pendingDeletes.isEmpty() + + if (migrateLegacyLocalItems) { + localItemsByKey.forEach { (key, item) -> + pendingUpserts[key] = item.toLibrarySyncKey() + } + } else { + pendingDeletes.keys.forEach(serverItemsByKey::remove) + pendingUpserts.keys.forEach { key -> + localItemsByKey[key]?.let { item -> serverItemsByKey[key] = item } + } + } + + return LibrarySnapshotReconciliation( + itemsByKey = if (migrateLegacyLocalItems) { + localItemsByKey.toMutableMap() + } else { + serverItemsByKey + }, + pendingUpsertKeysByKey = pendingUpserts, + pendingDeleteKeysByKey = pendingDeletes, + preservedLocalItems = migrateLegacyLocalItems || pendingUpserts.isNotEmpty() || pendingDeletes.isNotEmpty(), + ) +} + +internal fun reconcileLibraryDelta( + events: Collection, + currentItemsByKey: Map, + pendingUpsertKeysByKey: Map, + pendingDeleteKeysByKey: Map, + currentCursorEventId: Long, +): LibraryDeltaReconciliation { + val items = currentItemsByKey.toMutableMap() + + events.sortedBy(LibraryDeltaEvent::eventId).forEach { event -> + val key = libraryItemKey(event.item.id, event.item.type) + if (key in pendingUpsertKeysByKey || key in pendingDeleteKeysByKey) return@forEach + + when (event.operation.trim().lowercase()) { + "upsert" -> items[key] = event.item + "delete" -> items.remove(key) + } + } + + return LibraryDeltaReconciliation( + itemsByKey = items, + changed = items != currentItemsByKey, + cursorEventId = maxOf( + currentCursorEventId, + events.maxOfOrNull(LibraryDeltaEvent::eventId) ?: currentCursorEventId, + ), + ) +} diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncAdapter.kt new file mode 100644 index 00000000..fdfc53bb --- /dev/null +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncAdapter.kt @@ -0,0 +1,48 @@ +package com.nuvio.app.features.library.sync + +import com.nuvio.app.features.library.LibraryItem +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +@Serializable +data class LibrarySyncKey( + @SerialName("content_id") val contentId: String, + @SerialName("content_type") val contentType: String, +) + +data class LibraryDeltaEvent( + val eventId: Long, + val operation: String, + val item: LibraryItem, +) + +interface LibrarySyncAdapter { + suspend fun pullSnapshot( + profileId: Int, + pageSize: Int, + ): List + + suspend fun getDeltaCursor(profileId: Int): Long + + suspend fun pullDelta( + profileId: Int, + sinceEventId: Long, + limit: Int, + ): List + + suspend fun pushItems( + profileId: Int, + items: Collection, + ) + + suspend fun deleteItems( + profileId: Int, + keys: Collection, + ) +} + +fun LibraryItem.toLibrarySyncKey(): LibrarySyncKey = + LibrarySyncKey( + contentId = id, + contentType = type, + ) diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPaging.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPaging.kt new file mode 100644 index 00000000..ce1e0322 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPaging.kt @@ -0,0 +1,52 @@ +package com.nuvio.app.features.library.sync + +internal const val librarySnapshotPageSize = 500 +internal const val libraryDeltaPageSize = 500 +internal const val libraryMutationBatchSize = 500 + +internal suspend fun collectOffsetPages( + pageSize: Int, + fetchPage: suspend (limit: Int, offset: Int) -> List, +): List { + require(pageSize > 0) + + val items = mutableListOf() + var offset = 0 + while (true) { + val page = fetchPage(pageSize, offset) + items += page + if (page.size < pageSize) return items + offset += pageSize + } +} + +internal suspend fun consumeCursorPages( + initialCursor: Long, + pageSize: Int, + fetchPage: suspend (cursor: Long, limit: Int) -> List, + applyPage: suspend (page: List, cursor: Long) -> Long?, +): Long { + require(pageSize > 0) + + var cursor = initialCursor + while (true) { + val page = fetchPage(cursor, pageSize) + if (page.isEmpty()) return cursor + + val nextCursor = applyPage(page, cursor) ?: return cursor + check(nextCursor > cursor) + cursor = nextCursor + if (page.size < pageSize) return cursor + } +} + +internal suspend fun forEachMutationBatch( + values: Collection, + batchSize: Int, + sendBatch: suspend (List) -> Unit, +) { + require(batchSize > 0) + values.chunked(batchSize).forEach { batch -> + sendBatch(batch) + } +} diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/SupabaseLibrarySyncAdapter.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/SupabaseLibrarySyncAdapter.kt new file mode 100644 index 00000000..d7306411 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/sync/SupabaseLibrarySyncAdapter.kt @@ -0,0 +1,191 @@ +package com.nuvio.app.features.library.sync + +import com.nuvio.app.core.network.SupabaseProvider +import com.nuvio.app.core.sync.putSyncOriginClientId +import com.nuvio.app.features.home.PosterShape +import com.nuvio.app.features.library.LibraryItem +import io.github.jan.supabase.postgrest.postgrest +import io.github.jan.supabase.postgrest.rpc +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.encodeToJsonElement +import kotlinx.serialization.json.put + +object SupabaseLibrarySyncAdapter : LibrarySyncAdapter { + private val json = Json { + ignoreUnknownKeys = true + encodeDefaults = true + } + + override suspend fun pullSnapshot( + profileId: Int, + pageSize: Int, + ): List = + collectOffsetPages(pageSize) { limit, offset -> + val params = buildJsonObject { + put("p_profile_id", profileId) + put("p_limit", limit) + put("p_offset", offset) + } + SupabaseProvider.client.postgrest + .rpc("sync_pull_library", params) + .decodeList() + .map(LibrarySyncItem::toLibraryItem) + } + + override suspend fun getDeltaCursor(profileId: Int): Long { + val params = buildJsonObject { + put("p_profile_id", profileId) + } + return SupabaseProvider.client.postgrest + .rpc("sync_get_library_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) + } + return SupabaseProvider.client.postgrest + .rpc("sync_pull_library_delta", params) + .decodeList() + .map { event -> + LibraryDeltaEvent( + eventId = event.eventId, + operation = event.operation, + item = event.toLibraryItem(), + ) + } + } + + override suspend fun pushItems( + profileId: Int, + items: Collection, + ) { + forEachMutationBatch(items, libraryMutationBatchSize) { batch -> + val params = buildJsonObject { + put("p_profile_id", profileId) + put("p_items", json.encodeToJsonElement(batch.map(LibraryItem::toSyncItem))) + putSyncOriginClientId() + } + SupabaseProvider.client.postgrest.rpc("sync_push_library_items", params) + } + } + + override suspend fun deleteItems( + profileId: Int, + keys: Collection, + ) { + forEachMutationBatch(keys, libraryMutationBatchSize) { batch -> + val params = buildJsonObject { + put("p_profile_id", profileId) + put("p_keys", json.encodeToJsonElement(batch)) + putSyncOriginClientId() + } + SupabaseProvider.client.postgrest.rpc("sync_delete_library_items", params) + } + } +} + +@Serializable +private data class LibrarySyncItem( + @SerialName("content_id") override val contentId: String, + @SerialName("content_type") override val contentType: String, + override val name: String = "", + override val poster: String? = null, + @SerialName("poster_shape") override val posterShape: String = "POSTER", + override val background: String? = null, + override val description: String? = null, + @SerialName("release_info") override val releaseInfo: String? = null, + @SerialName("imdb_rating") override val imdbRating: Float? = null, + override val genres: List = emptyList(), + @SerialName("addon_base_url") override val addonBaseUrl: String? = null, + @SerialName("added_at") override val addedAt: Long = 0, +) : LibrarySyncFields + +@Serializable +private data class LibraryDeltaSyncItem( + @SerialName("event_id") val eventId: Long, + val operation: String, + @SerialName("content_id") override val contentId: String, + @SerialName("content_type") override val contentType: String, + override val name: String = "", + override val poster: String? = null, + @SerialName("poster_shape") override val posterShape: String = "POSTER", + override val background: String? = null, + override val description: String? = null, + @SerialName("release_info") override val releaseInfo: String? = null, + @SerialName("imdb_rating") override val imdbRating: Float? = null, + override val genres: List = emptyList(), + @SerialName("addon_base_url") override val addonBaseUrl: String? = null, + @SerialName("added_at") override val addedAt: Long = 0, +) : LibrarySyncFields + +private interface LibrarySyncFields { + val contentId: String + val contentType: String + val name: String + val poster: String? + val posterShape: String + val background: String? + val description: String? + val releaseInfo: String? + val imdbRating: Float? + val genres: List + val addonBaseUrl: String? + val addedAt: Long +} + +private fun LibrarySyncFields.toLibraryItem(): LibraryItem = + LibraryItem( + id = contentId, + type = contentType, + name = name, + poster = poster, + banner = background, + description = description, + releaseInfo = releaseInfo, + imdbRating = imdbRating?.toString(), + genres = genres, + posterShape = posterShape.toPosterShape(), + addonBaseUrl = addonBaseUrl, + savedAtEpochMs = addedAt, + ) + +private fun LibraryItem.toSyncItem(): LibrarySyncItem = + LibrarySyncItem( + contentId = id, + contentType = type, + name = name, + poster = poster, + posterShape = posterShape.toSyncName(), + background = banner, + description = description, + releaseInfo = releaseInfo, + imdbRating = imdbRating?.toFloatOrNull(), + genres = genres, + addonBaseUrl = addonBaseUrl, + addedAt = savedAtEpochMs, + ) + +private fun String.toPosterShape(): PosterShape = + when (trim().uppercase()) { + "LANDSCAPE" -> PosterShape.Landscape + "SQUARE" -> PosterShape.Square + else -> PosterShape.Poster + } + +private fun PosterShape.toSyncName(): String = + when (this) { + PosterShape.Poster -> "POSTER" + PosterShape.Square -> "SQUARE" + PosterShape.Landscape -> "LANDSCAPE" + } diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryDeltaStateTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryDeltaStateTest.kt new file mode 100644 index 00000000..d0a6618a --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryDeltaStateTest.kt @@ -0,0 +1,152 @@ +package com.nuvio.app.features.library + +import com.nuvio.app.features.library.sync.LibraryDeltaEvent +import com.nuvio.app.features.library.sync.LibrarySyncKey +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertTrue + +class LibraryDeltaStateTest { + + @Test + fun `removing final item records an explicit delete mutation`() { + val state = loadedState(listOf(libraryItem("item"))) + + val result = state.remove(id = "item", type = "movie") + + assertEquals(1, result.affectedCount) + assertTrue(result.snapshot.items.isEmpty()) + assertTrue(result.snapshot.pendingUpsertKeys.isEmpty()) + assertEquals(listOf("item"), result.snapshot.pendingDeleteKeys.map { it.contentId }) + } + + @Test + fun `successful push clears only the current mutation snapshot`() { + val state = loadedState(emptyList()) + val pushedSnapshot = state.upsert(libraryItem("first")) + state.upsert(libraryItem("second")) + + assertEquals(null, state.markPushCompleted(pushedSnapshot)) + assertEquals( + setOf("first", "second"), + state.snapshot().pendingUpsertKeys.mapTo(mutableSetOf()) { it.contentId }, + ) + } + + @Test + fun `delta deletion clears a remotely managed final item`() { + val state = loadedState(listOf(libraryItem("item"))) + val bootstrap = assertNotNull( + state.applyServerItems( + pullSnapshot = state.snapshot(), + serverItems = listOf(libraryItem("item")), + cursorEventId = 20L, + ), + ) + + val updated = assertNotNull( + state.applyDeltaEvents( + token = bootstrap.snapshot.token, + events = listOf( + LibraryDeltaEvent( + eventId = 21L, + operation = "delete", + item = libraryItem("item"), + ), + ), + ), + ) + + assertTrue(updated.items.isEmpty()) + assertEquals(21L, updated.deltaCursorEventId) + assertTrue(updated.deltaInitialized) + } + + @Test + fun `storage payload keeps cursor and pending deletes across reload`() { + val state = loadedState(listOf(libraryItem("item"))) + val bootstrap = assertNotNull( + state.applyServerItems( + pullSnapshot = state.snapshot(), + serverItems = listOf(libraryItem("item")), + cursorEventId = 42L, + ), + ) + val removed = state.remove(id = "item", type = "movie").snapshot + val stored = LibraryStoragePayloadCodec.decode( + LibraryStoragePayloadCodec.encode(removed), + ) + val reloaded = LibraryLocalState() + val token = reloaded.beginProfileLoad(profileId = 1).snapshot.token + val snapshot = assertNotNull( + reloaded.completeProfileLoad( + token = token, + activeProfileId = 1, + items = stored.items, + deltaCursorEventId = stored.deltaCursorEventId, + deltaInitialized = stored.deltaInitialized, + pendingUpsertKeys = stored.pendingUpsertKeys, + pendingDeleteKeys = stored.pendingDeleteKeys, + ), + ) + + assertTrue(bootstrap.snapshot.deltaInitialized) + assertEquals(42L, snapshot.deltaCursorEventId) + assertTrue(snapshot.deltaInitialized) + assertEquals(listOf("item"), snapshot.pendingDeleteKeys.map { it.contentId }) + } + + @Test + fun `legacy storage payload decodes with delta sync disabled`() { + val stored = LibraryStoragePayloadCodec.decode( + """{"items":[{"id":"legacy","type":"movie","name":"Legacy","savedAtEpochMs":1}]}""", + ) + + assertEquals(listOf("legacy"), stored.items.map(LibraryItem::id)) + assertEquals(0L, stored.deltaCursorEventId) + assertFalse(stored.deltaInitialized) + assertTrue(stored.pendingUpsertKeys.isEmpty()) + assertTrue(stored.pendingDeleteKeys.isEmpty()) + } + + @Test + fun `reloading a pending delete keeps the item removed`() { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + val snapshot = assertNotNull( + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = listOf(libraryItem("item")), + deltaInitialized = true, + pendingDeleteKeys = listOf( + LibrarySyncKey(contentId = "item", contentType = "movie"), + ), + ), + ) + + assertTrue(snapshot.items.isEmpty()) + assertEquals(listOf("item"), snapshot.pendingDeleteKeys.map { it.contentId }) + } + + private fun loadedState(items: List): LibraryLocalState { + val state = LibraryLocalState() + val token = state.beginProfileLoad(profileId = 1).snapshot.token + state.completeProfileLoad( + token = token, + activeProfileId = 1, + items = items, + ) + return state + } + + private fun libraryItem(id: String): LibraryItem = + LibraryItem( + id = id, + type = "movie", + name = id, + savedAtEpochMs = 1L, + ) +} diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibrarySyncReconcilerTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibrarySyncReconcilerTest.kt new file mode 100644 index 00000000..6a39c04d --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibrarySyncReconcilerTest.kt @@ -0,0 +1,121 @@ +package com.nuvio.app.features.library + +import com.nuvio.app.features.library.sync.LibraryDeltaEvent +import com.nuvio.app.features.library.sync.LibrarySyncKey +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class LibrarySyncReconcilerTest { + + @Test + fun `legacy empty snapshot queues local items for incremental migration`() { + val localItem = libraryItem(id = "local", savedAtEpochMs = 1L) + + val result = reconcileLibrarySnapshot( + serverItems = emptyList(), + localItemsByKey = mapOf(libraryItemKey(localItem.id, localItem.type) to localItem), + pendingUpsertKeysByKey = emptyMap(), + pendingDeleteKeysByKey = emptyMap(), + preserveLegacyLocalWhenServerEmpty = true, + ) + + assertEquals(listOf(localItem), result.itemsByKey.values.toList()) + assertEquals( + listOf(LibrarySyncKey(contentId = "local", contentType = "movie")), + result.pendingUpsertKeysByKey.values.toList(), + ) + assertTrue(result.preservedLocalItems) + } + + @Test + fun `snapshot merges pending upserts and deletes over remote state`() { + val localUpsert = libraryItem(id = "changed", savedAtEpochMs = 30L) + val remoteChanged = libraryItem(id = "changed", savedAtEpochMs = 10L) + val remoteKept = libraryItem(id = "kept", savedAtEpochMs = 20L) + val remoteDeleted = libraryItem(id = "deleted", savedAtEpochMs = 15L) + + val result = reconcileLibrarySnapshot( + serverItems = listOf(remoteChanged, remoteKept, remoteDeleted), + localItemsByKey = mapOf(libraryItemKey(localUpsert.id, localUpsert.type) to localUpsert), + pendingUpsertKeysByKey = mapOf( + libraryItemKey(localUpsert.id, localUpsert.type) to + LibrarySyncKey(localUpsert.id, localUpsert.type), + ), + pendingDeleteKeysByKey = mapOf( + libraryItemKey(remoteDeleted.id, remoteDeleted.type) to + LibrarySyncKey(remoteDeleted.id, remoteDeleted.type), + ), + preserveLegacyLocalWhenServerEmpty = false, + ) + + assertEquals( + setOf("changed", "kept"), + result.itemsByKey.values.mapTo(mutableSetOf(), LibraryItem::id), + ) + assertEquals(30L, result.itemsByKey[libraryItemKey("changed", "movie")]?.savedAtEpochMs) + assertTrue(result.preservedLocalItems) + } + + @Test + fun `delta applies ordered upserts and deletes and advances cursor`() { + val existing = libraryItem(id = "existing", savedAtEpochMs = 1L) + val added = libraryItem(id = "added", savedAtEpochMs = 2L) + + val result = reconcileLibraryDelta( + events = listOf( + LibraryDeltaEvent(eventId = 6L, operation = "delete", item = existing), + LibraryDeltaEvent(eventId = 7L, operation = "upsert", item = added), + ), + currentItemsByKey = mapOf(libraryItemKey(existing.id, existing.type) to existing), + pendingUpsertKeysByKey = emptyMap(), + pendingDeleteKeysByKey = emptyMap(), + currentCursorEventId = 5L, + ) + + assertEquals(listOf("added"), result.itemsByKey.values.map(LibraryItem::id)) + assertEquals(7L, result.cursorEventId) + assertTrue(result.changed) + } + + @Test + fun `delta preserves pending local mutations while advancing cursor`() { + val localUpsert = libraryItem(id = "local", savedAtEpochMs = 20L) + val remoteUpsert = libraryItem(id = "deleted-locally", savedAtEpochMs = 30L) + + val result = reconcileLibraryDelta( + events = listOf( + LibraryDeltaEvent(eventId = 11L, operation = "delete", item = localUpsert), + LibraryDeltaEvent(eventId = 12L, operation = "upsert", item = remoteUpsert), + ), + currentItemsByKey = mapOf( + libraryItemKey(localUpsert.id, localUpsert.type) to localUpsert, + ), + pendingUpsertKeysByKey = mapOf( + libraryItemKey(localUpsert.id, localUpsert.type) to + LibrarySyncKey(localUpsert.id, localUpsert.type), + ), + pendingDeleteKeysByKey = mapOf( + libraryItemKey(remoteUpsert.id, remoteUpsert.type) to + LibrarySyncKey(remoteUpsert.id, remoteUpsert.type), + ), + currentCursorEventId = 10L, + ) + + assertEquals(listOf("local"), result.itemsByKey.values.map(LibraryItem::id)) + assertEquals(12L, result.cursorEventId) + assertFalse(result.changed) + } + + private fun libraryItem( + id: String, + savedAtEpochMs: Long, + ): LibraryItem = + LibraryItem( + id = id, + type = "movie", + name = id, + savedAtEpochMs = savedAtEpochMs, + ) +} diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPagingTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPagingTest.kt new file mode 100644 index 00000000..643a909c --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/sync/LibrarySyncPagingTest.kt @@ -0,0 +1,94 @@ +package com.nuvio.app.features.library.sync + +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals + +class LibrarySyncPagingTest { + + @Test + fun `snapshot paging collects more than two full pages`() = runBlocking { + val source = (1..1_201).toList() + val offsets = mutableListOf() + + val result = collectOffsetPages(pageSize = librarySnapshotPageSize) { limit, offset -> + offsets += offset + source.drop(offset).take(limit) + } + + assertEquals(source, result) + assertEquals(listOf(0, 500, 1_000), offsets) + } + + @Test + fun `snapshot paging terminates after an exact page multiple`() = runBlocking { + val source = (1..1_000).toList() + val offsets = mutableListOf() + + val result = collectOffsetPages(pageSize = librarySnapshotPageSize) { limit, offset -> + offsets += offset + source.drop(offset).take(limit) + } + + assertEquals(source, result) + assertEquals(listOf(0, 500, 1_000), offsets) + } + + @Test + fun `delta paging consumes every page and advances the cursor`() = runBlocking { + val events = (1L..1_201L).toList() + val requestedCursors = mutableListOf() + + val cursor = consumeCursorPages( + initialCursor = 0L, + pageSize = libraryDeltaPageSize, + fetchPage = { currentCursor, limit -> + requestedCursors += currentCursor + events.filter { it > currentCursor }.take(limit) + }, + applyPage = { page, _ -> page.last() }, + ) + + assertEquals(1_201L, cursor) + assertEquals(listOf(0L, 500L, 1_000L), requestedCursors) + } + + @Test + fun `delta paging terminates after an exact page multiple`() = runBlocking { + val events = (1L..1_000L).toList() + val requestedCursors = mutableListOf() + + val cursor = consumeCursorPages( + initialCursor = 0L, + pageSize = libraryDeltaPageSize, + fetchPage = { currentCursor, limit -> + requestedCursors += currentCursor + events.filter { it > currentCursor }.take(limit) + }, + applyPage = { page, _ -> page.last() }, + ) + + assertEquals(1_000L, cursor) + assertEquals(listOf(0L, 500L, 1_000L), requestedCursors) + } + + @Test + fun `mutation batching preserves long upsert and delete sets`() = runBlocking { + val upserts = (1..1_201).toList() + val deletes = (1..1_001).toList() + val upsertBatches = mutableListOf>() + val deleteBatches = mutableListOf>() + + forEachMutationBatch(upserts, batchSize = libraryMutationBatchSize) { batch -> + upsertBatches += batch + } + forEachMutationBatch(deletes, batchSize = libraryMutationBatchSize) { batch -> + deleteBatches += batch + } + + assertEquals(listOf(500, 500, 201), upsertBatches.map { it.size }) + assertEquals(listOf(500, 500, 1), deleteBatches.map { it.size }) + assertEquals(upserts, upsertBatches.flatten()) + assertEquals(deletes, deleteBatches.flatten()) + } +}