mirror of
https://github.com/tapframe/NuvioStreaming.git
synced 2026-08-02 01:28:25 +00:00
feat(library): add incremental delta sync
This commit is contained in:
tapframe
parent
d55ed7ae2a
commit
1b05c222b7
10 changed files with 1026 additions and 198 deletions
|
|
@ -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<LibraryItem>,
|
||||
val hasPendingPush: Boolean,
|
||||
)
|
||||
val deltaCursorEventId: Long,
|
||||
val deltaInitialized: Boolean,
|
||||
val pendingUpsertKeys: List<LibrarySyncKey>,
|
||||
val pendingDeleteKeys: List<LibrarySyncKey>,
|
||||
) {
|
||||
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<String, LibraryItem> = mutableMapOf()
|
||||
private var hasPendingPush = false
|
||||
private var deltaCursorEventId = 0L
|
||||
private var deltaInitialized = false
|
||||
private var pendingUpsertKeysByKey: MutableMap<String, LibrarySyncKey> = mutableMapOf()
|
||||
private var pendingDeleteKeysByKey: MutableMap<String, LibrarySyncKey> = 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<LibraryItem>,
|
||||
deltaCursorEventId: Long = 0L,
|
||||
deltaInitialized: Boolean = false,
|
||||
pendingUpsertKeys: Collection<LibrarySyncKey> = emptyList(),
|
||||
pendingDeleteKeys: Collection<LibrarySyncKey> = 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<LibraryItem>,
|
||||
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<LibraryDeltaEvent>,
|
||||
): 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<String, LibrarySyncKey>.filterToExistingItems(
|
||||
itemsById: Map<String, LibraryItem>,
|
||||
): MutableMap<String, LibrarySyncKey> =
|
||||
apply {
|
||||
keys.retainAll(itemsById.keys)
|
||||
}
|
||||
|
||||
internal fun libraryItemKey(id: String, type: String): String =
|
||||
"${type.trim().lowercase()}:${id.trim()}"
|
||||
|
|
|
|||
|
|
@ -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<LibraryItem> = 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<String> = 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<LibraryUiState> = _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<Int, Long>()
|
||||
private val lastPersistedRevisionByProfile = mutableMapOf<Int, Long>()
|
||||
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<StoredLibraryPayload>(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<LibrarySyncItem> {
|
||||
val allItems = mutableListOf<LibrarySyncItem>()
|
||||
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<LibrarySyncItem>()
|
||||
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()
|
||||
|
|
|
|||
|
|
@ -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<LibraryItem> = emptyList(),
|
||||
val deltaCursorEventId: Long = 0L,
|
||||
val deltaInitialized: Boolean = false,
|
||||
val pendingUpsertKeys: List<LibrarySyncKey> = emptyList(),
|
||||
val pendingDeleteKeys: List<LibrarySyncKey> = emptyList(),
|
||||
)
|
||||
|
||||
internal object LibraryStoragePayloadCodec {
|
||||
private val json = Json {
|
||||
ignoreUnknownKeys = true
|
||||
encodeDefaults = true
|
||||
}
|
||||
|
||||
fun decode(payload: String): StoredLibraryPayload =
|
||||
runCatching {
|
||||
json.decodeFromString<StoredLibraryPayload>(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,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
|
@ -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<String, LibraryItem>,
|
||||
val pendingUpsertKeysByKey: MutableMap<String, LibrarySyncKey>,
|
||||
val pendingDeleteKeysByKey: MutableMap<String, LibrarySyncKey>,
|
||||
val preservedLocalItems: Boolean,
|
||||
)
|
||||
|
||||
internal data class LibraryDeltaReconciliation(
|
||||
val itemsByKey: MutableMap<String, LibraryItem>,
|
||||
val changed: Boolean,
|
||||
val cursorEventId: Long,
|
||||
)
|
||||
|
||||
internal fun reconcileLibrarySnapshot(
|
||||
serverItems: Collection<LibraryItem>,
|
||||
localItemsByKey: Map<String, LibraryItem>,
|
||||
pendingUpsertKeysByKey: Map<String, LibrarySyncKey>,
|
||||
pendingDeleteKeysByKey: Map<String, LibrarySyncKey>,
|
||||
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<LibraryDeltaEvent>,
|
||||
currentItemsByKey: Map<String, LibraryItem>,
|
||||
pendingUpsertKeysByKey: Map<String, LibrarySyncKey>,
|
||||
pendingDeleteKeysByKey: Map<String, LibrarySyncKey>,
|
||||
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,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
|
@ -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<LibraryItem>
|
||||
|
||||
suspend fun getDeltaCursor(profileId: Int): Long
|
||||
|
||||
suspend fun pullDelta(
|
||||
profileId: Int,
|
||||
sinceEventId: Long,
|
||||
limit: Int,
|
||||
): List<LibraryDeltaEvent>
|
||||
|
||||
suspend fun pushItems(
|
||||
profileId: Int,
|
||||
items: Collection<LibraryItem>,
|
||||
)
|
||||
|
||||
suspend fun deleteItems(
|
||||
profileId: Int,
|
||||
keys: Collection<LibrarySyncKey>,
|
||||
)
|
||||
}
|
||||
|
||||
fun LibraryItem.toLibrarySyncKey(): LibrarySyncKey =
|
||||
LibrarySyncKey(
|
||||
contentId = id,
|
||||
contentType = type,
|
||||
)
|
||||
|
|
@ -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 <T> collectOffsetPages(
|
||||
pageSize: Int,
|
||||
fetchPage: suspend (limit: Int, offset: Int) -> List<T>,
|
||||
): List<T> {
|
||||
require(pageSize > 0)
|
||||
|
||||
val items = mutableListOf<T>()
|
||||
var offset = 0
|
||||
while (true) {
|
||||
val page = fetchPage(pageSize, offset)
|
||||
items += page
|
||||
if (page.size < pageSize) return items
|
||||
offset += pageSize
|
||||
}
|
||||
}
|
||||
|
||||
internal suspend fun <T> consumeCursorPages(
|
||||
initialCursor: Long,
|
||||
pageSize: Int,
|
||||
fetchPage: suspend (cursor: Long, limit: Int) -> List<T>,
|
||||
applyPage: suspend (page: List<T>, 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 <T> forEachMutationBatch(
|
||||
values: Collection<T>,
|
||||
batchSize: Int,
|
||||
sendBatch: suspend (List<T>) -> Unit,
|
||||
) {
|
||||
require(batchSize > 0)
|
||||
values.chunked(batchSize).forEach { batch ->
|
||||
sendBatch(batch)
|
||||
}
|
||||
}
|
||||
|
|
@ -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<LibraryItem> =
|
||||
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<LibrarySyncItem>()
|
||||
.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<Long>()
|
||||
}
|
||||
|
||||
override suspend fun pullDelta(
|
||||
profileId: Int,
|
||||
sinceEventId: Long,
|
||||
limit: Int,
|
||||
): List<LibraryDeltaEvent> {
|
||||
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<LibraryDeltaSyncItem>()
|
||||
.map { event ->
|
||||
LibraryDeltaEvent(
|
||||
eventId = event.eventId,
|
||||
operation = event.operation,
|
||||
item = event.toLibraryItem(),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun pushItems(
|
||||
profileId: Int,
|
||||
items: Collection<LibraryItem>,
|
||||
) {
|
||||
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<LibrarySyncKey>,
|
||||
) {
|
||||
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<String> = 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<String> = 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<String>
|
||||
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"
|
||||
}
|
||||
|
|
@ -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<LibraryItem>): 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,
|
||||
)
|
||||
}
|
||||
|
|
@ -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,
|
||||
)
|
||||
}
|
||||
|
|
@ -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<Int>()
|
||||
|
||||
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<Int>()
|
||||
|
||||
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<Long>()
|
||||
|
||||
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<Long>()
|
||||
|
||||
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<List<Int>>()
|
||||
val deleteBatches = mutableListOf<List<Int>>()
|
||||
|
||||
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())
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue