diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedItemsStore.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedItemsStore.kt new file mode 100644 index 000000000..06ce6bc41 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedItemsStore.kt @@ -0,0 +1,32 @@ +package com.nuvio.app.features.watched + +import com.nuvio.app.features.tracking.TrackingProviderId +import kotlinx.atomicfu.locks.SynchronizedObject +import kotlinx.atomicfu.locks.synchronized + +internal class WatchedItemsStore { + private val lock = SynchronizedObject() + private val nuvioItems = mutableMapOf() + private val providerItems = mutableMapOf>() + private val dirtyNuvioKeys = mutableSetOf() + + fun read( + block: ( + nuvioItems: Map, + providerItems: Map>, + dirtyNuvioKeys: Set, + ) -> T, + ): T = synchronized(lock) { + block(nuvioItems, providerItems, dirtyNuvioKeys) + } + + fun update( + block: ( + nuvioItems: MutableMap, + providerItems: MutableMap>, + dirtyNuvioKeys: MutableSet, + ) -> T, + ): T = synchronized(lock) { + block(nuvioItems, providerItems, dirtyNuvioKeys) + } +} diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt index e3f305c6b..5d3f179b2 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt @@ -145,8 +145,7 @@ object WatchedRepository { private var profileGeneration: Long = 0L private var activeSource: WatchProgressSource = WatchProgressSource.NUVIO_SYNC private var sourceGeneration: Long = 0L - private var nuvioItemsByKey: MutableMap = mutableMapOf() - private var providerItemsByKey: MutableMap> = mutableMapOf() + private val itemsStore = WatchedItemsStore() private var nuvioFullyWatchedSeriesKeys: Set = emptySet() private var providerFullyWatchedSeriesKeys: MutableMap> = mutableMapOf() private var expandedSiblingKeys: Set = emptySet() @@ -155,7 +154,6 @@ object WatchedRepository { private var loadedProviders: MutableSet = mutableSetOf() private var nuvioHasLoadedRemote: Boolean = false private var providersLoadedFromRemote: MutableSet = mutableSetOf() - private var nuvioDirtyWatchedKeys: MutableSet = mutableSetOf() private var lastSuccessfulPushEpochMs: Long = 0L private var deltaCursorEventId: Long = 0L private var deltaInitialized: Boolean = false @@ -197,8 +195,11 @@ object WatchedRepository { profileGeneration += 1L activeSource = WatchProgressSource.NUVIO_SYNC sourceGeneration += 1L - nuvioItemsByKey.clear() - providerItemsByKey.clear() + itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys -> + nuvioItems.clear() + providerItems.clear() + dirtyNuvioKeys.clear() + } nuvioFullyWatchedSeriesKeys = emptySet() providerFullyWatchedSeriesKeys.clear() expandedSiblingKeys = emptySet() @@ -207,7 +208,6 @@ object WatchedRepository { loadedProviders.clear() nuvioHasLoadedRemote = false providersLoadedFromRemote.clear() - nuvioDirtyWatchedKeys.clear() lastSuccessfulPushEpochMs = 0L deltaCursorEventId = 0L deltaInitialized = false @@ -221,8 +221,11 @@ object WatchedRepository { activeSource = WatchProgressSource.NUVIO_SYNC sourceGeneration += 1L hasLoaded = true - nuvioItemsByKey.clear() - providerItemsByKey.clear() + itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys -> + nuvioItems.clear() + providerItems.clear() + dirtyNuvioKeys.clear() + } nuvioFullyWatchedSeriesKeys = emptySet() providerFullyWatchedSeriesKeys.clear() expandedSiblingKeys = emptySet() @@ -231,7 +234,6 @@ object WatchedRepository { loadedProviders.clear() nuvioHasLoadedRemote = false providersLoadedFromRemote.clear() - nuvioDirtyWatchedKeys.clear() val payload = WatchedStorage.loadPayload(profileId).orEmpty().trim() if (payload.isNotEmpty()) { @@ -241,19 +243,19 @@ object WatchedRepository { lastSuccessfulPushEpochMs = storedPayload.lastSuccessfulPushEpochMs deltaCursorEventId = storedPayload.deltaCursorEventId deltaInitialized = storedPayload.deltaInitialized - nuvioItemsByKey = storedPayload.items + val restoredItems = storedPayload.items .map(WatchedItem::normalizedMarkedAt) .associateBy { watchedItemKey(it.type, it.id, it.season, it.episode) } - .toMutableMap() - nuvioDirtyWatchedKeys = storedPayload.dirtyWatchedKeys - .filterTo(mutableSetOf()) { key -> key in nuvioItemsByKey } + itemsStore.update { nuvioItems, _, dirtyNuvioKeys -> + nuvioItems.putAll(restoredItems) + dirtyNuvioKeys += storedPayload.dirtyWatchedKeys.filter { key -> key in restoredItems } + } nuvioFullyWatchedSeriesKeys = storedPayload.fullyWatchedSeriesKeys expandedSiblingKeys = storedPayload.expandedSiblingKeys } else { lastSuccessfulPushEpochMs = 0L deltaCursorEventId = 0L deltaInitialized = false - nuvioDirtyWatchedKeys.clear() nuvioFullyWatchedSeriesKeys = emptySet() } @@ -277,7 +279,9 @@ object WatchedRepository { } val previousSource = activeSource source.providerId?.let { providerId -> - providerItemsByKey.getOrPut(providerId, ::mutableMapOf).clear() + itemsStore.update { _, providerItems, _ -> + providerItems.getOrPut(providerId, ::mutableMapOf).clear() + } providerFullyWatchedSeriesKeys[providerId] = emptySet() providerExtraWatchedKeys.remove(providerId) loadedProviders -= providerId @@ -449,6 +453,7 @@ object WatchedRepository { pageSize = watchedItemsPageSize, ) val fullyWatchedSeriesKeys = adapter.pullFullyWatchedSeriesKeys(profileId) + val extraWatchedKeys = adapter.pullExtraWatchedKeys(profileId) val source = operation.sourceOperation.source log.i { "Watched adapter result source=$source provider=${source.providerId?.storageId ?: "nuvio"} " + @@ -469,35 +474,36 @@ object WatchedRepository { } return false } - val localAtApply = itemsForSource(operation.sourceOperation.source).values.toList() - - val mergedSnapshot = mergeWatchedSnapshot( - serverItems = serverItems, - localItems = localAtApply, - dirtyKeys = if (operation.sourceOperation.source == WatchProgressSource.NUVIO_SYNC) { - nuvioDirtyWatchedKeys - } else { - emptySet() - }, - ) - replaceWatchedItemsForSource( - source = operation.sourceOperation.source, - nuvioItems = nuvioItemsByKey, - providerItems = providerItemsByKey, - replacement = mergedSnapshot.items, - ) - fullyWatchedSeriesKeys?.let { keys -> - setFullyWatchedSeriesKeysForSource(operation.sourceOperation.source, keys) + itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys -> + val items = source.providerId + ?.let { providerId -> providerItems[providerId]?.values.orEmpty() } + ?: nuvioItems.values + val merged = mergeWatchedSnapshot( + serverItems = serverItems, + localItems = items.toList(), + dirtyKeys = if (source.providerId == null) dirtyNuvioKeys else emptySet(), + ) + replaceWatchedItemsForSource( + source = source, + nuvioItems = nuvioItems, + providerItems = providerItems, + replacement = merged.items, + ) + if (source.providerId == null) { + dirtyNuvioKeys.clear() + dirtyNuvioKeys += merged.dirtyKeys + } } - val extraWatchedKeys = adapter.pullExtraWatchedKeys(profileId) - operation.sourceOperation.source.providerId?.let { providerId -> + fullyWatchedSeriesKeys?.let { keys -> + setFullyWatchedSeriesKeysForSource(source, keys) + } + source.providerId?.let { providerId -> if (extraWatchedKeys.isNotEmpty()) { providerExtraWatchedKeys[providerId] = extraWatchedKeys } loadedProviders += providerId providersLoadedFromRemote += providerId } ?: run { - nuvioDirtyWatchedKeys = mergedSnapshot.dirtyKeys.toMutableSet() nuvioHasLoaded = true nuvioHasLoadedRemote = true if (resetDeltaState) { @@ -559,11 +565,13 @@ object WatchedRepository { if (!isActiveOperation(operation)) return false if (events.isEmpty()) break - applyWatchedDeltaEvents( - targetItems = nuvioItemsByKey, - dirtyKeys = nuvioDirtyWatchedKeys, - events = events, - ) + itemsStore.update { nuvioItems, _, dirtyNuvioKeys -> + applyWatchedDeltaEvents( + targetItems = nuvioItems, + dirtyKeys = dirtyNuvioKeys, + events = events, + ) + } cursor = maxOf(cursor, events.maxOf { it.eventId }) deltaCursorEventId = cursor deltaInitialized = true @@ -685,10 +693,13 @@ object WatchedRepository { return remoteType.isSeriesLikeWatchedType() && localType.isSeriesLikeWatchedType() } - private fun itemsForSource(source: WatchProgressSource): MutableMap = - source.providerId - ?.let { providerId -> providerItemsByKey.getOrPut(providerId, ::mutableMapOf) } - ?: nuvioItemsByKey + private fun itemsForSourceSnapshot(source: WatchProgressSource): List = + itemsStore.read { nuvioItems, providerItems, _ -> + val items = source.providerId + ?.let { providerId -> providerItems[providerId]?.values.orEmpty() } + ?: nuvioItems.values + items.toList() + } private fun fullyWatchedSeriesKeysForSource(source: WatchProgressSource): Set = source.providerId @@ -709,16 +720,23 @@ object WatchedRepository { private fun hasLoadedSource(source: WatchProgressSource): Boolean = source.providerId?.let(loadedProviders::contains) ?: nuvioHasLoaded - private fun itemCountForSource(source: WatchProgressSource): Int = source.providerId - ?.let { providerId -> providerItemsByKey[providerId]?.size ?: 0 } - ?: nuvioItemsByKey.size + private fun itemCountForSource(source: WatchProgressSource): Int = + itemsStore.read { nuvioItems, providerItems, _ -> + source.providerId + ?.let { providerId -> providerItems[providerId]?.size ?: 0 } + ?: nuvioItems.size + } fun toggleWatched(item: WatchedItem) { ensureLoaded() val source = activeSource - val targetItems = itemsForSource(source) val key = watchedItemKey(item.type, item.id, item.season, item.episode) - if (targetItems.containsKey(key)) { + val isMarked = itemsStore.read { nuvioItems, providerItems, _ -> + source.providerId + ?.let { providerId -> providerItems[providerId]?.containsKey(key) == true } + ?: nuvioItems.containsKey(key) + } + if (isMarked) { unmarkWatched(item) } else { markWatched(item) @@ -749,16 +767,20 @@ object WatchedRepository { ensureLoaded() if (items.isEmpty()) return val source = activeSource - val targetItems = itemsForSource(source) val markedAt = WatchedClock.nowEpochMs() val timestampedItems = items.map { watchedItem -> watchedItem.copy(markedAtEpochMs = markedAt) } - timestampedItems.forEach { watchedItem -> - val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode) - targetItems[key] = watchedItem - if (source.providerId == null) { - nuvioDirtyWatchedKeys += key + itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys -> + val targetItems = source.providerId + ?.let { providerId -> providerItems.getOrPut(providerId, ::mutableMapOf) } + ?: nuvioItems + timestampedItems.forEach { watchedItem -> + val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode) + targetItems[key] = watchedItem + if (source.providerId == null) { + dirtyNuvioKeys += key + } } } publish() @@ -802,25 +824,27 @@ object WatchedRepository { ensureLoaded() if (items.isEmpty()) return val source = activeSource - val targetItems = itemsForSource(source) - val removedItems = items.mapNotNull { watchedItem -> - val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode) - targetItems.remove(key)?.let { storeItem -> - // Preserve videoId from the original request (store items don't have it) - if (watchedItem.videoId != null && storeItem.videoId == null) { - storeItem.copy(videoId = watchedItem.videoId) - } else { - storeItem - } - }?.also { - if (source.providerId == null) { - nuvioDirtyWatchedKeys -= key - } - // Optimistically remove from extra keys so publish() doesn't re-add it - source.providerId?.let { providerId -> - providerExtraWatchedKeys[providerId]?.let { extraKeys -> - if (key in extraKeys) { - providerExtraWatchedKeys[providerId] = extraKeys - key + val removedItems = itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys -> + val targetItems = source.providerId + ?.let { providerId -> providerItems.getOrPut(providerId, ::mutableMapOf) } + ?: nuvioItems + items.mapNotNull { watchedItem -> + val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode) + targetItems.remove(key)?.let { storeItem -> + if (watchedItem.videoId != null && storeItem.videoId == null) { + storeItem.copy(videoId = watchedItem.videoId) + } else { + storeItem + } + }?.also { + if (source.providerId == null) { + dirtyNuvioKeys -= key + } + source.providerId?.let { providerId -> + providerExtraWatchedKeys[providerId]?.let { extraKeys -> + if (key in extraKeys) { + providerExtraWatchedKeys[providerId] = extraKeys - key + } } } } @@ -847,7 +871,13 @@ object WatchedRepository { episode: Int? = null, ): Boolean { ensureLoaded() - return itemsForSource(activeSource).containsKey(watchedItemKey(type, id, season, episode)) + val source = activeSource + val key = watchedItemKey(type, id, season, episode) + return itemsStore.read { nuvioItems, providerItems, _ -> + source.providerId + ?.let { providerId -> providerItems[providerId]?.containsKey(key) == true } + ?: nuvioItems.containsKey(key) + } } fun isFullyWatchedSeries(id: String, type: String): Boolean { @@ -874,7 +904,7 @@ object WatchedRepository { "Watched series reconciliation source=$activeSource content=${meta.type}:${meta.id} " + "episodes=${meta.videos.size} shouldMarkSeries=$shouldMarkSeriesWatched " + "hasSeriesMarker=$hasSeriesWatchedMarker " + - "matchingItems=${itemsForSource(activeSource).values.count { it.id == meta.id }}" + "matchingItems=${itemsForSourceSnapshot(activeSource).count { it.id == meta.id }}" } if (shouldMarkSeriesWatched) { if (!hasSeriesWatchedMarker) { @@ -1007,10 +1037,15 @@ object WatchedRepository { } private fun publish() { + val (nuvioItems, providerItems) = itemsStore.read { storedNuvioItems, storedProviderItems, _ -> + storedNuvioItems.values.toList() to storedProviderItems.mapValues { (_, itemsByKey) -> + itemsByKey.values.toList() + } + } val items = watchedItemsForSource( source = activeSource, - nuvioItems = nuvioItemsByKey.values, - providerItems = providerItemsByKey.mapValues { (_, itemsByKey) -> itemsByKey.values }, + nuvioItems = nuvioItems, + providerItems = providerItems, ) .map(WatchedItem::normalizedMarkedAt) .sortedByDescending { it.markedAtEpochMs } @@ -1081,7 +1116,9 @@ object WatchedRepository { watchedItemKey(item.type, item.id, item.season, item.episode) }.toMutableMap() providerExtraWatchedKeys[providerId] = extraKeys - providerItemsByKey[providerId] = itemsByKey + itemsStore.update { _, providerItems, _ -> + providerItems[providerId] = itemsByKey + } publish() } } @@ -1094,19 +1131,22 @@ object WatchedRepository { } private fun persistNuvio() { + val (items, dirtyKeys) = itemsStore.read { nuvioItems, _, dirtyNuvioKeys -> + nuvioItems.values + .map(WatchedItem::normalizedMarkedAt) + .sortedByDescending { it.markedAtEpochMs } to dirtyNuvioKeys.toSet() + } WatchedStorage.savePayload( currentProfileId, json.encodeToString( StoredWatchedPayload( - items = nuvioItemsByKey.values - .map(WatchedItem::normalizedMarkedAt) - .sortedByDescending { it.markedAtEpochMs }, + items = items, fullyWatchedSeriesKeys = nuvioFullyWatchedSeriesKeys, expandedSiblingKeys = expandedSiblingKeys, lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs, deltaCursorEventId = deltaCursorEventId, deltaInitialized = deltaInitialized, - dirtyWatchedKeys = nuvioDirtyWatchedKeys.toSet(), + dirtyWatchedKeys = dirtyKeys, ), ), ) @@ -1118,26 +1158,31 @@ object WatchedRepository { items: Collection, ) { if (profileId != currentProfileId || operationGeneration != profileGeneration) return - val acknowledgedDirtyKeys = acknowledgeSuccessfulWatchedPush( - currentItems = nuvioItemsByKey, - dirtyKeys = nuvioDirtyWatchedKeys, - pushedItems = items, - ) val latestPushed = items .asSequence() .map { item -> normalizeWatchedMarkedAtEpochMs(item.markedAtEpochMs) } .maxOrNull() ?: return - val updatedLastSuccessfulPushEpochMs = maxOf(lastSuccessfulPushEpochMs, latestPushed) - if ( - acknowledgedDirtyKeys == nuvioDirtyWatchedKeys && - updatedLastSuccessfulPushEpochMs == lastSuccessfulPushEpochMs - ) { - return + val changed = itemsStore.update { nuvioItems, _, dirtyNuvioKeys -> + val acknowledgedDirtyKeys = acknowledgeSuccessfulWatchedPush( + currentItems = nuvioItems, + dirtyKeys = dirtyNuvioKeys, + pushedItems = items, + ) + val updatedLastSuccessfulPushEpochMs = maxOf(lastSuccessfulPushEpochMs, latestPushed) + if ( + acknowledgedDirtyKeys == dirtyNuvioKeys && + updatedLastSuccessfulPushEpochMs == lastSuccessfulPushEpochMs + ) { + false + } else { + dirtyNuvioKeys.clear() + dirtyNuvioKeys += acknowledgedDirtyKeys + lastSuccessfulPushEpochMs = updatedLastSuccessfulPushEpochMs + true + } } - nuvioDirtyWatchedKeys = acknowledgedDirtyKeys.toMutableSet() - lastSuccessfulPushEpochMs = updatedLastSuccessfulPushEpochMs - persistNuvio() + if (changed) persistNuvio() } private suspend fun pushToTargetsForSource( diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedItemsStoreTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedItemsStoreTest.kt new file mode 100644 index 000000000..71a98c739 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedItemsStoreTest.kt @@ -0,0 +1,55 @@ +package com.nuvio.app.features.watched + +import com.nuvio.app.features.tracking.TrackingProviderId +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals + +class WatchedItemsStoreTest { + @Test + fun `concurrent updates publish coherent item snapshots`() = runBlocking { + val store = WatchedItemsStore() + + coroutineScope { + repeat(4) { writer -> + launch(Dispatchers.Default) { + repeat(500) { index -> + val key = "$writer:$index" + val item = WatchedItem( + id = key, + type = "movie", + name = key, + markedAtEpochMs = index.toLong(), + ) + store.update { nuvioItems, providerItems, dirtyNuvioKeys -> + nuvioItems[key] = item + providerItems + .getOrPut(TrackingProviderId.TRAKT, ::mutableMapOf)[key] = item + dirtyNuvioKeys += key + } + } + } + } + repeat(4) { + launch(Dispatchers.Default) { + repeat(500) { + store.read { nuvioItems, providerItems, dirtyNuvioKeys -> + val nuvioKeys = nuvioItems.keys.toSet() + assertEquals(nuvioKeys, providerItems[TrackingProviderId.TRAKT].orEmpty().keys) + assertEquals(nuvioKeys, dirtyNuvioKeys) + } + } + } + } + } + + store.read { nuvioItems, providerItems, dirtyNuvioKeys -> + assertEquals(2_000, nuvioItems.size) + assertEquals(nuvioItems.keys, providerItems[TrackingProviderId.TRAKT].orEmpty().keys) + assertEquals(nuvioItems.keys, dirtyNuvioKeys) + } + } +}