mirror of
https://github.com/tapframe/NuvioStreaming.git
synced 2026-08-09 00:17:59 +00:00
fix: serialize watched state snapshots (NUVIO-MOBILE-PW NUVIO-MOBILE-PY NUVIO-MOBILE-Q2)
This commit is contained in:
tapframe
parent
e0cbf447dc
commit
15fcb8d25e
3 changed files with 231 additions and 99 deletions
|
|
@ -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<String, WatchedItem>()
|
||||
private val providerItems = mutableMapOf<TrackingProviderId, MutableMap<String, WatchedItem>>()
|
||||
private val dirtyNuvioKeys = mutableSetOf<String>()
|
||||
|
||||
fun <T> read(
|
||||
block: (
|
||||
nuvioItems: Map<String, WatchedItem>,
|
||||
providerItems: Map<TrackingProviderId, Map<String, WatchedItem>>,
|
||||
dirtyNuvioKeys: Set<String>,
|
||||
) -> T,
|
||||
): T = synchronized(lock) {
|
||||
block(nuvioItems, providerItems, dirtyNuvioKeys)
|
||||
}
|
||||
|
||||
fun <T> update(
|
||||
block: (
|
||||
nuvioItems: MutableMap<String, WatchedItem>,
|
||||
providerItems: MutableMap<TrackingProviderId, MutableMap<String, WatchedItem>>,
|
||||
dirtyNuvioKeys: MutableSet<String>,
|
||||
) -> T,
|
||||
): T = synchronized(lock) {
|
||||
block(nuvioItems, providerItems, dirtyNuvioKeys)
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String, WatchedItem> = mutableMapOf()
|
||||
private var providerItemsByKey: MutableMap<TrackingProviderId, MutableMap<String, WatchedItem>> = mutableMapOf()
|
||||
private val itemsStore = WatchedItemsStore()
|
||||
private var nuvioFullyWatchedSeriesKeys: Set<String> = emptySet()
|
||||
private var providerFullyWatchedSeriesKeys: MutableMap<TrackingProviderId, Set<String>> = mutableMapOf()
|
||||
private var expandedSiblingKeys: Set<String> = emptySet()
|
||||
|
|
@ -155,7 +154,6 @@ object WatchedRepository {
|
|||
private var loadedProviders: MutableSet<TrackingProviderId> = mutableSetOf()
|
||||
private var nuvioHasLoadedRemote: Boolean = false
|
||||
private var providersLoadedFromRemote: MutableSet<TrackingProviderId> = mutableSetOf()
|
||||
private var nuvioDirtyWatchedKeys: MutableSet<String> = 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<String, WatchedItem> =
|
||||
source.providerId
|
||||
?.let { providerId -> providerItemsByKey.getOrPut(providerId, ::mutableMapOf) }
|
||||
?: nuvioItemsByKey
|
||||
private fun itemsForSourceSnapshot(source: WatchProgressSource): List<WatchedItem> =
|
||||
itemsStore.read { nuvioItems, providerItems, _ ->
|
||||
val items = source.providerId
|
||||
?.let { providerId -> providerItems[providerId]?.values.orEmpty() }
|
||||
?: nuvioItems.values
|
||||
items.toList()
|
||||
}
|
||||
|
||||
private fun fullyWatchedSeriesKeysForSource(source: WatchProgressSource): Set<String> =
|
||||
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<WatchedItem>,
|
||||
) {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue