Creepso/NuvioStreaming
1
0
Fork 0
mirror of https://github.com/tapframe/NuvioStreaming.git synced 2026-07-27 06:52:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: update library sync logic while in active pull state

Browse source
This commit is contained in:
tapframe 2026-07-14 15:31:22 +05:30
parent b3be03fb7f
commit f74ae8182e
5 changed files with 176 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
composeApp/src/commonMain/kotlin/com/nuvio/app/core/sync/SyncManager.kt
View file

@ -33,7 +33,7 @@ import kotlinx.coroutines.launch
private const val FOREGROUND_PULL_DELAY_MS = 2500L
private const val FOREGROUND_PULL_MIN_INTERVAL_MS = 30 * 60_000L
private const val PERIODIC_NUVIO_SYNC_PULL_INTERVAL_MS = 60_000L
private const val PERIODIC_NUVIO_SYNC_PULL_INTERVAL_MS = 240_000L
internal enum class ProfileSyncStep {
Addons,
@ -130,8 +130,6 @@ internal enum class ProfileSyncRequestResult {
Replaced,
}
internal fun shouldQueueCoalescedForegroundPull(force: Boolean): Boolean = force
internal class ProfileSyncRequestGate {
private data class PendingRequest(
val scope: CoroutineScope,
@ -298,11 +296,8 @@ object SyncManager {
delay(FOREGROUND_PULL_DELAY_MS)
}
if (!force && hasRecentFullPull(profileId)) return@launch
startFullProfilePull(
profileId = profileId,
reason = "foreground",
queueIfCoalesced = shouldQueueCoalescedForegroundPull(force),
)
if (ProfileRepository.activeProfileId != profileId) return@launch
pullForegroundForProfile(profileId)
} finally {
synchronized(pullStateLock) {
if (foregroundPullJob === requestJob) {
@ -325,6 +320,47 @@ object SyncManager {
TraktPlatformClock.nowEpochMs() - lastFullPullAtMs < FOREGROUND_PULL_MIN_INTERVAL_MS
}
private suspend fun pullForegroundForProfile(profileId: Int) {
log.i { "Foreground sync started profile=$profileId" }
runCatching { ProfileRepository.pullProfiles() }
.onFailure { log.e(it) { "Foreground profiles pull failed" } }
runCatching { ProfileSettingsSync.pull(profileId) }
.onFailure { log.e(it) { "Foreground profile settings pull failed" } }
coroutineScope {
launch {
runCatching { AddonRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground addons pull failed" } }
}
if (AppFeaturePolicy.pluginsEnabled) {
launch {
runCatching { PluginRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground plugins pull failed" } }
}
}
launch {
runCatching { LibraryRepository.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground library pull failed" } }
}
launch {
runCatching {
WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = true)
}.onFailure { log.e(it) { "Foreground active watch source pull failed" } }
}
launch {
runCatching { CollectionSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground collections pull failed" } }
}
launch {
runCatching { HomeCatalogSettingsSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Foreground home catalog settings pull failed" } }
}
}
log.i { "Foreground sync completed profile=$profileId" }
}
private fun startFullProfilePull(
profileId: Int,
reason: String,
@ -449,15 +485,6 @@ object SyncManager {
val authState = AuthRepository.state.value
if (authState !is AuthState.Authenticated || authState.isAnonymous) return
if (surface == "profile_settings") {
startFullProfilePull(
profileId = profileId,
reason = "realtime_profile_settings",
queueIfCoalesced = true,
)
return
}
accountScopeSnapshot().launch {
log.i { "requestRealtimeSurfacePull($profileId, $surface)" }
when (surface) {
@ -480,6 +507,10 @@ object SyncManager {
WatchProgressSourceCoordinator.refreshActiveSource(profileId = profileId, force = false)
}.onFailure { log.e(it) { "Realtime active watch source pull failed" } }
}
"profile_settings" -> {
runCatching { ProfileSettingsSync.pull(profileId) }
.onFailure { log.e(it) { "Realtime profile settings pull failed" } }
}
"collections" -> {
runCatching { CollectionSyncService.pullFromServer(profileId) }
.onFailure { log.e(it) { "Realtime collections pull failed" } }

57
composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryLocalState.kt
View file

@ -16,8 +16,7 @@ internal data class LibraryLocalSnapshot(
val hasLoaded: Boolean,
val isLoading: Boolean,
val items: List<LibraryItem>,
val isPullingNuvioSyncFromServer: Boolean,
val hasCompletedInitialNuvioSyncPull: Boolean,
val hasPendingPush: Boolean,
)
internal data class LibraryStateTransition(
@ -61,8 +60,7 @@ internal class LibraryLocalState {
private var contentRevision = 0L
private var isLoading = false
private var itemsById: MutableMap<String, LibraryItem> = mutableMapOf()
private var isPullingNuvioSyncFromServer = false
private var hasCompletedInitialNuvioSyncPull = false
private var hasPendingPush = false
private var pushJob: Job? = null
fun snapshot(): LibraryLocalSnapshot = synchronized(lock) {
@ -85,6 +83,10 @@ internal class LibraryLocalState {
isCurrentLocked(snapshot)
}
fun isContentCurrent(snapshot: LibraryLocalSnapshot): Boolean = synchronized(lock) {
isContentCurrentLocked(snapshot)
}
fun runIfCurrent(snapshot: LibraryLocalSnapshot, block: () -> Unit): Boolean = synchronized(lock) {
if (!isCurrentLocked(snapshot)) {
false
@ -122,8 +124,7 @@ internal class LibraryLocalState {
hasLoaded = false
isLoading = true
itemsById = mutableMapOf()
isPullingNuvioSyncFromServer = false
hasCompletedInitialNuvioSyncPull = false
hasPendingPush = false
LibraryStateTransition(
snapshot = snapshotLocked(),
detachedPushJob = detachedPushJob,
@ -156,8 +157,7 @@ internal class LibraryLocalState {
hasLoaded = false
isLoading = false
itemsById = mutableMapOf()
isPullingNuvioSyncFromServer = false
hasCompletedInitialNuvioSyncPull = false
hasPendingPush = false
LibraryStateTransition(
snapshot = snapshotLocked(),
detachedPushJob = detachedPushJob,
@ -166,34 +166,22 @@ internal class LibraryLocalState {
fun markPullStarted(token: LibraryProfileToken): LibraryLocalSnapshot? = synchronized(lock) {
if (!isCurrentLocked(token)) return@synchronized null
isPullingNuvioSyncFromServer = true
revision += 1L
snapshotLocked()
}
fun finishPull(
token: LibraryProfileToken,
completedSuccessfully: Boolean,
): LibraryLocalSnapshot? = synchronized(lock) {
if (!isCurrentLocked(token)) return@synchronized null
isPullingNuvioSyncFromServer = false
if (completedSuccessfully) {
hasCompletedInitialNuvioSyncPull = true
}
revision += 1L
snapshotLocked()
}
fun applyServerItems(
token: LibraryProfileToken,
pullSnapshot: LibraryLocalSnapshot,
serverItems: Collection<LibraryItem>,
): LibraryServerItemsApplyResult? = synchronized(lock) {
if (!isCurrentLocked(token)) return@synchronized null
if (!isCurrentLocked(pullSnapshot.token)) return@synchronized null
val preserveLocalItems = serverItems.isEmpty() && itemsById.isNotEmpty()
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) }
contentRevision += 1L
hasPendingPush = false
}
hasLoaded = true
isLoading = false
@ -208,6 +196,7 @@ internal class LibraryLocalState {
itemsById[libraryItemKey(item.id, item.type)] = item
revision += 1L
contentRevision += 1L
hasPendingPush = true
snapshotLocked()
}
@ -221,6 +210,7 @@ internal class LibraryLocalState {
}
revision += 1L
contentRevision += 1L
hasPendingPush = true
LibraryLocalToggleResult(
snapshot = snapshotLocked(),
isSaved = isSaved,
@ -234,6 +224,7 @@ internal class LibraryLocalState {
if (affectedCount > 0) {
revision += 1L
contentRevision += 1L
hasPendingPush = true
}
LibraryLocalMutation(
snapshot = snapshotLocked(),
@ -246,6 +237,7 @@ internal class LibraryLocalState {
if (affectedCount > 0) {
revision += 1L
contentRevision += 1L
hasPendingPush = true
}
LibraryLocalMutation(
snapshot = snapshotLocked(),
@ -269,7 +261,7 @@ internal class LibraryLocalState {
snapshot: LibraryLocalSnapshot,
job: Job,
): LibraryPushJobInstallResult = synchronized(lock) {
if (!isCurrentLocked(snapshot)) {
if (!isContentCurrentLocked(snapshot)) {
LibraryPushJobInstallResult(installed = false, detachedPushJob = null)
} else {
val detachedPushJob = pushJob
@ -284,6 +276,14 @@ internal class LibraryLocalState {
}
}
fun markPushCompleted(snapshot: LibraryLocalSnapshot) {
synchronized(lock) {
if (isContentCurrentLocked(snapshot)) {
hasPendingPush = false
}
}
}
private fun tokenLocked(): LibraryProfileToken =
LibraryProfileToken(
profileId = currentProfileId,
@ -298,8 +298,7 @@ internal class LibraryLocalState {
hasLoaded = hasLoaded,
isLoading = isLoading,
items = itemsById.values.toList(),
isPullingNuvioSyncFromServer = isPullingNuvioSyncFromServer,
hasCompletedInitialNuvioSyncPull = hasCompletedInitialNuvioSyncPull,
hasPendingPush = hasPendingPush,
)
private fun isCurrentLocked(token: LibraryProfileToken): Boolean =

31
composeApp/src/commonMain/kotlin/com/nuvio/app/features/library/LibraryRepository.kt
View file

@ -239,20 +239,18 @@ object LibraryRepository {
nuvioPullMutex.withLock {
val serializedToken = activeOperationToken(profileId) ?: return@withLock
if (localState.markPullStarted(serializedToken) == null) return@withLock
val pullSnapshot = localState.markPullStarted(serializedToken) ?: return@withLock
var completedSuccessfully = false
var appliedItems = false
try {
val serverItems = pullAllLibrarySyncItems(profileId).map { it.toLibraryItem() }
val applyResult = localState.applyServerItems(serializedToken, serverItems)
val applyResult = localState.applyServerItems(pullSnapshot, serverItems)
?: return@withLock
completedSuccessfully = true
appliedItems = true
if (applyResult.preservedLocalItems) {
log.w {
"Remote library is empty while local has ${applyResult.snapshot.items.size} entries; " +
"preserving local library"
"Preserving ${applyResult.snapshot.items.size} local library items because the remote " +
"snapshot is empty or local changes are pending"
}
} else {
persist(applyResult.snapshot)
@ -261,11 +259,6 @@ object LibraryRepository {
throw error
} catch (error: Throwable) {
log.e(error) { "Failed to pull library from server" }
} finally {
localState.finishPull(
token = serializedToken,
completedSuccessfully = completedSuccessfully,
)
}
if (appliedItems) publish()
@ -466,23 +459,14 @@ object LibraryRepository {
log.w { "Skipping library push: anonymous auth user=${authState.userId} profile=$profileId" }
return
}
if (snapshot.isPullingNuvioSyncFromServer) {
log.i { "Skipping library push: server pull is active profile=$profileId localItems=$itemCount" }
return
}
if (!snapshot.hasCompletedInitialNuvioSyncPull) {
log.w { "Skipping library push: initial Nuvio sync pull not completed profile=$profileId localItems=$itemCount" }
return
}
val pushJob = syncScope.launch(start = CoroutineStart.LAZY) {
delay(500)
if (!localState.isCurrent(snapshot)) {
if (!localState.isContentCurrent(snapshot)) {
val current = localState.snapshot()
log.w {
"Skipping stale debounced library push: scheduled=${snapshot.token} " +
"current=${current.token} scheduledRevision=${snapshot.revision} " +
"currentRevision=${current.revision}"
"current=${current.token} scheduledContentRevision=${snapshot.contentRevision} " +
"currentContentRevision=${current.contentRevision}"
}
return@launch
}
@ -502,6 +486,7 @@ object LibraryRepository {
true
}.onSuccess { pushed ->
if (pushed) {
localState.markPushCompleted(snapshot)
log.i { "Library push completed profile=$profileId itemCount=$itemCount" }
}
}.onFailure { e ->

6
composeApp/src/commonTest/kotlin/com/nuvio/app/core/sync/SyncManagerTest.kt
View file

@ -10,12 +10,6 @@ import kotlin.test.assertTrue
class SyncManagerTest {
@Test
fun `forced foreground recovery queues behind an active profile sync`() {
assertFalse(shouldQueueCoalescedForegroundPull(force = false))
assertTrue(shouldQueueCoalescedForegroundPull(force = true))
}
@Test
fun `source prerequisites finish before source dependent pulls`() = runBlocking {
val events = mutableListOf<String>()

92
composeApp/src/commonTest/kotlin/com/nuvio/app/features/library/LibraryRepositoryTest.kt
View file

@ -188,6 +188,98 @@ class LibraryRepositoryTest {
assertEquals(setOf("first", "second"), state.snapshot().items.map { it.id }.toSet())
}
@Test
fun `pull lifecycle does not invalidate pending local push`() {
val state = LibraryLocalState()
val token = state.beginProfileLoad(profileId = 1).snapshot.token
state.completeProfileLoad(
token = token,
activeProfileId = 1,
items = emptyList(),
)
val localSnapshot = state.upsert(libraryItem(id = "local", savedAtEpochMs = 1L))
state.markPullStarted(token)
assertTrue(state.isContentCurrent(localSnapshot))
}
@Test
fun `server pull preserves local items awaiting push`() {
val state = LibraryLocalState()
val token = state.beginProfileLoad(profileId = 1).snapshot.token
state.completeProfileLoad(
token = token,
activeProfileId = 1,
items = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)),
)
state.upsert(libraryItem(id = "local-new", savedAtEpochMs = 2L))
val pullSnapshot = assertNotNull(state.markPullStarted(token))
val result = assertNotNull(
state.applyServerItems(
pullSnapshot = pullSnapshot,
serverItems = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)),
),
)
assertTrue(result.preservedLocalItems)
assertEquals(
setOf("remote-old", "local-new"),
result.snapshot.items.map { it.id }.toSet(),
)
}
@Test
fun `server pull preserves a local mutation made while pulling`() {
val state = LibraryLocalState()
val token = state.beginProfileLoad(profileId = 1).snapshot.token
state.completeProfileLoad(
token = token,
activeProfileId = 1,
items = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)),
)
val pullSnapshot = assertNotNull(state.markPullStarted(token))
state.upsert(libraryItem(id = "local-new", savedAtEpochMs = 2L))
val result = assertNotNull(
state.applyServerItems(
pullSnapshot = pullSnapshot,
serverItems = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)),
),
)
assertTrue(result.preservedLocalItems)
assertEquals(
setOf("remote-old", "local-new"),
result.snapshot.items.map { it.id }.toSet(),
)
}
@Test
fun `server pull applies after local push completes`() {
val state = LibraryLocalState()
val token = state.beginProfileLoad(profileId = 1).snapshot.token
state.completeProfileLoad(
token = token,
activeProfileId = 1,
items = listOf(libraryItem(id = "remote-old", savedAtEpochMs = 1L)),
)
val localSnapshot = state.upsert(libraryItem(id = "local-new", savedAtEpochMs = 2L))
state.markPushCompleted(localSnapshot)
val pullSnapshot = assertNotNull(state.markPullStarted(token))
val result = assertNotNull(
state.applyServerItems(
pullSnapshot = pullSnapshot,
serverItems = listOf(libraryItem(id = "remote-new", savedAtEpochMs = 3L)),
),
)
assertFalse(result.preservedLocalItems)
assertEquals(listOf("remote-new"), result.snapshot.items.map { it.id })
}
private fun libraryItem(id: String, savedAtEpochMs: Long): LibraryItem =
LibraryItem(
id = id,