mirror of
https://github.com/tapframe/NuvioStreaming.git
synced 2026-07-27 06:52:18 +00:00
Add realtime sync invalidation handling
This commit is contained in:
tapframe
parent
6fa966cc3f
commit
dccbfc43c2
9 changed files with 361 additions and 14 deletions
|
|
@ -320,7 +320,7 @@ kotlin {
|
|||
implementation("com.squareup.okhttp3:okhttp:4.12.0")
|
||||
implementation("com.google.code.gson:gson:2.11.0")
|
||||
implementation("io.github.peerless2012:ass-media:0.4.0-beta01")
|
||||
implementation(libs.ktor.client.android)
|
||||
implementation(libs.ktor.client.okhttp)
|
||||
implementation(libs.androidx.media3.exoplayer.hls)
|
||||
implementation(libs.androidx.media3.exoplayer.dash)
|
||||
implementation(libs.androidx.media3.exoplayer.smoothstreaming)
|
||||
|
|
@ -368,6 +368,7 @@ kotlin {
|
|||
implementation(libs.supabase.postgrest)
|
||||
implementation(libs.supabase.auth)
|
||||
implementation(libs.supabase.functions)
|
||||
implementation(libs.supabase.realtime)
|
||||
implementation(libs.reorderable)
|
||||
}
|
||||
commonTest.dependencies {
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ import androidx.work.WorkManager
|
|||
import com.nuvio.app.core.storage.ProfileScopedKey
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.call.body
|
||||
import io.ktor.client.engine.android.Android
|
||||
import io.ktor.client.engine.okhttp.OkHttp
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import nuvio.composeapp.generated.resources.*
|
||||
import org.jetbrains.compose.resources.getString
|
||||
|
|
@ -53,7 +53,7 @@ internal actual object EpisodeReleaseNotificationPlatform {
|
|||
private var currentActivity: ComponentActivity? = null
|
||||
private var pendingPermissionContinuation: kotlin.coroutines.Continuation<Boolean>? = null
|
||||
private val httpClient by lazy {
|
||||
HttpClient(Android) {
|
||||
HttpClient(OkHttp) {
|
||||
install(HttpTimeout) {
|
||||
requestTimeoutMillis = 15_000
|
||||
connectTimeoutMillis = 15_000
|
||||
|
|
|
|||
|
|
@ -85,6 +85,7 @@ import com.nuvio.app.core.network.NetworkCondition
|
|||
import com.nuvio.app.core.network.NetworkStatusRepository
|
||||
import com.nuvio.app.core.sync.AppForegroundMonitor
|
||||
import com.nuvio.app.core.sync.ProfileSettingsSync
|
||||
import com.nuvio.app.core.sync.RealtimeSyncInvalidationService
|
||||
import com.nuvio.app.core.sync.SyncManager
|
||||
import com.nuvio.app.core.ui.NuvioNavigationBar
|
||||
import com.nuvio.app.core.ui.NuvioContinueWatchingActionSheet
|
||||
|
|
@ -907,6 +908,27 @@ private fun MainAppContent(
|
|||
}
|
||||
}
|
||||
|
||||
LaunchedEffect(authState, profileState.activeProfile?.profileIndex) {
|
||||
val authenticatedState = authState as? AuthState.Authenticated ?: return@LaunchedEffect
|
||||
if (authenticatedState.isAnonymous) return@LaunchedEffect
|
||||
|
||||
val activeProfileId = profileState.activeProfile?.profileIndex ?: return@LaunchedEffect
|
||||
RealtimeSyncInvalidationService.start(
|
||||
userId = authenticatedState.userId,
|
||||
profileId = activeProfileId,
|
||||
)
|
||||
}
|
||||
|
||||
DisposableEffect(authState, profileState.activeProfile?.profileIndex) {
|
||||
val authenticatedState = authState as? AuthState.Authenticated
|
||||
if (authenticatedState == null || authenticatedState.isAnonymous || profileState.activeProfile == null) {
|
||||
RealtimeSyncInvalidationService.stop()
|
||||
}
|
||||
onDispose {
|
||||
RealtimeSyncInvalidationService.stop()
|
||||
}
|
||||
}
|
||||
|
||||
LaunchedEffect(authState, profileState.activeProfile?.profileIndex) {
|
||||
val authenticatedState = authState as? AuthState.Authenticated ?: return@LaunchedEffect
|
||||
if (authenticatedState.isAnonymous) return@LaunchedEffect
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import io.github.jan.supabase.auth.Auth
|
|||
import io.github.jan.supabase.createSupabaseClient
|
||||
import io.github.jan.supabase.functions.Functions
|
||||
import io.github.jan.supabase.postgrest.Postgrest
|
||||
import io.github.jan.supabase.realtime.Realtime
|
||||
import io.ktor.client.plugins.defaultRequest
|
||||
import io.ktor.http.HttpHeaders
|
||||
|
||||
|
|
@ -25,6 +26,7 @@ object SupabaseProvider {
|
|||
install(Auth)
|
||||
install(Postgrest)
|
||||
install(Functions)
|
||||
install(Realtime)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,174 @@
|
|||
package com.nuvio.app.core.sync
|
||||
|
||||
import co.touchlab.kermit.Logger
|
||||
import com.nuvio.app.core.network.SupabaseProvider
|
||||
import io.github.jan.supabase.postgrest.query.filter.FilterOperator
|
||||
import io.github.jan.supabase.realtime.PostgresAction
|
||||
import io.github.jan.supabase.realtime.channel
|
||||
import io.github.jan.supabase.realtime.postgresChangeFlow
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.awaitCancellation
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.TimeoutCancellationException
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import kotlinx.coroutines.flow.launchIn
|
||||
import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.contentOrNull
|
||||
import kotlinx.serialization.json.intOrNull
|
||||
import kotlinx.serialization.json.jsonPrimitive
|
||||
|
||||
private const val REALTIME_INVALIDATION_COALESCE_MS = 500L
|
||||
private const val REALTIME_SUBSCRIBE_TIMEOUT_MS = 15_000L
|
||||
private const val REALTIME_RETRY_BASE_DELAY_MS = 1_000L
|
||||
private const val REALTIME_RETRY_MAX_DELAY_MS = 10_000L
|
||||
|
||||
object RealtimeSyncInvalidationService {
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
private val log = Logger.withTag("RealtimeSyncInvalidation")
|
||||
private val pendingMutex = Mutex()
|
||||
private val pendingSurfaces = mutableSetOf<String>()
|
||||
|
||||
private var subscriptionJob: Job? = null
|
||||
private var drainJob: Job? = null
|
||||
private var activeUserId: String? = null
|
||||
private var activeProfileId: Int? = null
|
||||
|
||||
fun start(userId: String, profileId: Int) {
|
||||
if (
|
||||
subscriptionJob?.isActive == true &&
|
||||
activeUserId == userId &&
|
||||
activeProfileId == profileId
|
||||
) {
|
||||
log.d { "Realtime sync already active for profile $profileId" }
|
||||
return
|
||||
}
|
||||
|
||||
stop()
|
||||
activeUserId = userId
|
||||
activeProfileId = profileId
|
||||
subscriptionJob = scope.launch {
|
||||
var attempt = 1
|
||||
while (isActive) {
|
||||
val channelName = "sync-invalidations:$userId:$profileId:$attempt"
|
||||
val channel = SupabaseProvider.client.channel(channelName)
|
||||
val realtime = channel.realtime
|
||||
val realtimeStatusJob = launch {
|
||||
realtime.status
|
||||
.collect { status ->
|
||||
log.i { "Realtime client status=$status channel=$channelName" }
|
||||
}
|
||||
}
|
||||
val channelStatusJob = launch {
|
||||
channel.status
|
||||
.collect { status ->
|
||||
log.i { "Realtime channel status=$status channel=$channelName" }
|
||||
}
|
||||
}
|
||||
val changesJob = channel.postgresChangeFlow<PostgresAction.Insert>(schema = "public") {
|
||||
table = "sync_invalidations"
|
||||
filter("user_id", FilterOperator.EQ, userId)
|
||||
}.onEach { action ->
|
||||
handleInsert(profileId, action.record)
|
||||
}.launchIn(this)
|
||||
|
||||
try {
|
||||
log.i { "Subscribing to sync invalidations channel=$channelName attempt=$attempt user=$userId profile=$profileId" }
|
||||
withTimeout(REALTIME_SUBSCRIBE_TIMEOUT_MS) {
|
||||
channel.subscribe(blockUntilSubscribed = true)
|
||||
}
|
||||
log.i { "Subscribed to sync invalidations channel=$channelName profile=$profileId" }
|
||||
awaitCancellation()
|
||||
} catch (error: TimeoutCancellationException) {
|
||||
log.e(error) {
|
||||
"Timed out subscribing to sync invalidations channel=$channelName " +
|
||||
"realtimeStatus=${realtime.status.value} channelStatus=${channel.status.value}"
|
||||
}
|
||||
} catch (error: CancellationException) {
|
||||
throw error
|
||||
} catch (error: Throwable) {
|
||||
log.e(error) {
|
||||
"Failed to subscribe to sync invalidations channel=$channelName " +
|
||||
"realtimeStatus=${realtime.status.value} channelStatus=${channel.status.value}"
|
||||
}
|
||||
} finally {
|
||||
changesJob.cancel()
|
||||
realtimeStatusJob.cancel()
|
||||
channelStatusJob.cancel()
|
||||
runCatching { channel.unsubscribe() }
|
||||
.onSuccess { log.i { "Unsubscribed from sync invalidations channel=$channelName" } }
|
||||
.onFailure { error -> log.w(error) { "Failed to unsubscribe from sync invalidations channel=$channelName" } }
|
||||
}
|
||||
|
||||
if (isActive) {
|
||||
val retryDelay = (REALTIME_RETRY_BASE_DELAY_MS * attempt)
|
||||
.coerceAtMost(REALTIME_RETRY_MAX_DELAY_MS)
|
||||
log.w { "Retrying sync invalidations subscription in ${retryDelay}ms profile=$profileId nextAttempt=${attempt + 1}" }
|
||||
delay(retryDelay)
|
||||
attempt += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun stop() {
|
||||
if (subscriptionJob != null || drainJob != null) {
|
||||
log.i { "Stopping realtime sync invalidation service for profile $activeProfileId" }
|
||||
}
|
||||
subscriptionJob?.cancel()
|
||||
drainJob?.cancel()
|
||||
subscriptionJob = null
|
||||
drainJob = null
|
||||
activeUserId = null
|
||||
activeProfileId = null
|
||||
pendingSurfaces.clear()
|
||||
}
|
||||
|
||||
private fun handleInsert(profileId: Int, record: JsonObject) {
|
||||
val eventId = record["id"]?.jsonPrimitive?.contentOrNull
|
||||
val createdAt = record["created_at"]?.jsonPrimitive?.contentOrNull
|
||||
val surface = record["surface"]?.jsonPrimitive?.contentOrNull ?: run {
|
||||
log.w { "Received sync invalidation without surface id=$eventId createdAt=$createdAt keys=${record.keys}" }
|
||||
return
|
||||
}
|
||||
val eventProfileId = record["profile_id"]?.jsonPrimitive?.intOrNull
|
||||
log.i {
|
||||
"Received sync invalidation id=$eventId surface=$surface eventProfile=$eventProfileId activeProfile=$profileId createdAt=$createdAt"
|
||||
}
|
||||
if (surface != "profiles" && eventProfileId != null && eventProfileId != profileId) {
|
||||
log.d { "Ignoring sync invalidation id=$eventId for inactive profile $eventProfileId" }
|
||||
return
|
||||
}
|
||||
enqueue(surface, profileId)
|
||||
}
|
||||
|
||||
private fun enqueue(surface: String, profileId: Int) {
|
||||
scope.launch {
|
||||
pendingMutex.withLock {
|
||||
pendingSurfaces += surface
|
||||
log.d { "Queued realtime surface pull surface=$surface profile=$profileId pending=${pendingSurfaces.size}" }
|
||||
if (drainJob?.isActive == true) return@withLock
|
||||
drainJob = scope.launch {
|
||||
delay(REALTIME_INVALIDATION_COALESCE_MS)
|
||||
val surfaces = pendingMutex.withLock {
|
||||
pendingSurfaces.toList().also {
|
||||
pendingSurfaces.clear()
|
||||
}
|
||||
}
|
||||
log.i { "Draining realtime surface pulls profile=$profileId surfaces=$surfaces" }
|
||||
surfaces.forEach { pendingSurface ->
|
||||
SyncManager.requestRealtimeSurfacePull(profileId, pendingSurface)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -100,6 +100,55 @@ object SyncManager {
|
|||
}
|
||||
}
|
||||
|
||||
fun requestRealtimeSurfacePull(profileId: Int, surface: String) {
|
||||
val authState = AuthRepository.state.value
|
||||
if (authState !is AuthState.Authenticated || authState.isAnonymous) return
|
||||
|
||||
scope.launch {
|
||||
log.i { "requestRealtimeSurfacePull($profileId, $surface)" }
|
||||
when (surface) {
|
||||
"addons" -> {
|
||||
runCatching { AddonRepository.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Realtime addons pull failed" } }
|
||||
}
|
||||
"plugins" -> {
|
||||
if (AppFeaturePolicy.pluginsEnabled) {
|
||||
runCatching { PluginRepository.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Realtime plugins pull failed" } }
|
||||
}
|
||||
}
|
||||
"library" -> {
|
||||
runCatching { LibraryRepository.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Realtime library pull failed" } }
|
||||
}
|
||||
"watch_progress" -> {
|
||||
runCatching { WatchProgressRepository.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Realtime watch progress pull failed" } }
|
||||
}
|
||||
"watched_items" -> {
|
||||
runCatching { WatchedRepository.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Realtime watched items 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" } }
|
||||
}
|
||||
"home_catalog_settings" -> {
|
||||
runCatching { HomeCatalogSettingsSyncService.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Realtime home catalog settings pull failed" } }
|
||||
}
|
||||
"profiles" -> {
|
||||
runCatching { ProfileRepository.pullProfiles() }
|
||||
.onFailure { log.e(it) { "Realtime profiles pull failed" } }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun pullForegroundForProfile(profileId: Int) {
|
||||
scope.launch {
|
||||
log.i { "pullForegroundForProfile($profileId) — syncing watch progress, library, collections, and home settings" }
|
||||
|
|
|
|||
|
|
@ -239,6 +239,7 @@ object LibraryRepository {
|
|||
ensureLoaded()
|
||||
|
||||
if (isTraktLibrarySourceActive()) {
|
||||
log.i { "toggleSaved routed to Trakt library source item=${item.id} type=${item.type} profile=$currentProfileId" }
|
||||
syncScope.launch {
|
||||
runCatching { TraktLibraryRepository.toggleWatchlist(item) }
|
||||
.onFailure { e ->
|
||||
|
|
@ -262,6 +263,7 @@ object LibraryRepository {
|
|||
|
||||
fun save(item: LibraryItem) {
|
||||
ensureLoaded()
|
||||
log.i { "Saving local library item item=${item.id} type=${item.type} profile=$currentProfileId" }
|
||||
itemsById[libraryItemKey(item.id, item.type)] = item.copy(savedAtEpochMs = LibraryClock.nowEpochMs())
|
||||
publish()
|
||||
persist()
|
||||
|
|
@ -273,6 +275,7 @@ object LibraryRepository {
|
|||
val before = itemsById.size
|
||||
itemsById.entries.removeAll { (_, item) -> item.id == id }
|
||||
if (itemsById.size != before) {
|
||||
log.i { "Removing local library item id=$id profile=$currentProfileId removed=${before - itemsById.size}" }
|
||||
publish()
|
||||
persist()
|
||||
pushToServer()
|
||||
|
|
@ -282,6 +285,7 @@ object LibraryRepository {
|
|||
private fun remove(id: String, type: String) {
|
||||
ensureLoaded()
|
||||
if (itemsById.remove(libraryItemKey(id, type)) != null) {
|
||||
log.i { "Removing local library item id=$id type=$type profile=$currentProfileId" }
|
||||
publish()
|
||||
persist()
|
||||
pushToServer()
|
||||
|
|
@ -347,6 +351,9 @@ object LibraryRepository {
|
|||
ensureLoaded()
|
||||
val localDesired = desiredMembership[LOCAL_LIBRARY_LIST_KEY] == true
|
||||
val currentlyInLocal = itemsById.containsKey(libraryItemKey(item.id, item.type))
|
||||
log.i {
|
||||
"Applying library membership item=${item.id} type=${item.type} profile=$currentProfileId localDesired=$localDesired currentlyInLocal=$currentlyInLocal traktAuthenticated=${TraktAuthRepository.isAuthenticated.value}"
|
||||
}
|
||||
if (localDesired != currentlyInLocal) {
|
||||
if (localDesired) {
|
||||
save(item)
|
||||
|
|
@ -379,24 +386,51 @@ object LibraryRepository {
|
|||
|
||||
private fun pushToServer() {
|
||||
val authState = AuthRepository.state.value
|
||||
if (authState !is AuthState.Authenticated || authState.isAnonymous) return
|
||||
if (isPullingNuvioSyncFromServer || !hasCompletedInitialNuvioSyncPull) return
|
||||
if (authState !is AuthState.Authenticated) {
|
||||
log.w { "Skipping library push: auth state is ${authState::class.simpleName} profile=$currentProfileId" }
|
||||
return
|
||||
}
|
||||
if (authState.isAnonymous) {
|
||||
log.w { "Skipping library push: anonymous auth user=${authState.userId} profile=$currentProfileId" }
|
||||
return
|
||||
}
|
||||
if (isPullingNuvioSyncFromServer) {
|
||||
log.i { "Skipping library push: server pull is active profile=$currentProfileId localItems=${itemsById.size}" }
|
||||
return
|
||||
}
|
||||
if (!hasCompletedInitialNuvioSyncPull) {
|
||||
log.w { "Skipping library push: initial Nuvio sync pull not completed profile=$currentProfileId localItems=${itemsById.size}" }
|
||||
return
|
||||
}
|
||||
|
||||
pushJob?.cancel()
|
||||
val profileId = currentProfileId
|
||||
pushJob = syncScope.launch {
|
||||
delay(500)
|
||||
if (profileId != currentProfileId) return@launch
|
||||
if (profileId != currentProfileId) {
|
||||
log.w { "Skipping debounced library push: profile changed scheduled=$profileId current=$currentProfileId" }
|
||||
return@launch
|
||||
}
|
||||
val itemCount = itemsById.size
|
||||
runCatching {
|
||||
val syncItems = itemsById.values.map { it.toSyncItem() }
|
||||
if (syncItems.isEmpty()) return@runCatching
|
||||
if (syncItems.isEmpty()) {
|
||||
log.w { "Skipping library push: sync payload is empty profile=$profileId" }
|
||||
return@runCatching false
|
||||
}
|
||||
val params = buildJsonObject {
|
||||
put("p_profile_id", profileId)
|
||||
put("p_items", json.encodeToJsonElement(syncItems))
|
||||
}
|
||||
log.i { "Pushing library to server profile=$profileId itemCount=${syncItems.size}" }
|
||||
SupabaseProvider.client.postgrest.rpc("sync_push_library", params)
|
||||
true
|
||||
}.onSuccess { pushed ->
|
||||
if (pushed) {
|
||||
log.i { "Library push completed profile=$profileId itemCount=$itemCount" }
|
||||
}
|
||||
}.onFailure { e ->
|
||||
log.e(e) { "Failed to push library to server" }
|
||||
log.e(e) { "Failed to push library to server profile=$profileId itemCount=$itemCount" }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -296,7 +296,6 @@ object WatchedRepository {
|
|||
|
||||
applyWatchedDeltaEvents(
|
||||
events = events,
|
||||
lastPushEpochMs = lastPushEpochMs,
|
||||
pullStartedEpochMs = pullStartedEpochMs,
|
||||
)
|
||||
cursor = maxOf(cursor, events.maxOf { it.eventId })
|
||||
|
|
@ -316,13 +315,20 @@ object WatchedRepository {
|
|||
|
||||
private fun applyWatchedDeltaEvents(
|
||||
events: Collection<WatchedDeltaEvent>,
|
||||
lastPushEpochMs: Long,
|
||||
pullStartedEpochMs: Long,
|
||||
) {
|
||||
var upsertCount = 0
|
||||
var deleteCount = 0
|
||||
var removedCount = 0
|
||||
var removedByFallbackKeyCount = 0
|
||||
var preservedDuringPullCount = 0
|
||||
var ignoredCount = 0
|
||||
|
||||
events.forEach { event ->
|
||||
val key = watchedItemKey(event.contentType, event.contentId, event.season, event.episode)
|
||||
when (event.operation.lowercase()) {
|
||||
watchedDeltaOperationUpsert -> {
|
||||
upsertCount += 1
|
||||
itemsByKey[key] = WatchedItem(
|
||||
id = event.contentId,
|
||||
type = event.contentType,
|
||||
|
|
@ -333,14 +339,64 @@ object WatchedRepository {
|
|||
)
|
||||
}
|
||||
watchedDeltaOperationDelete -> {
|
||||
deleteCount += 1
|
||||
val localItem = itemsByKey[key]
|
||||
if (localItem != null && shouldPreserveLocalWatchedItem(localItem, lastPushEpochMs, pullStartedEpochMs)) {
|
||||
if (localItem != null && wasWatchedItemMarkedDuringPull(localItem, pullStartedEpochMs)) {
|
||||
preservedDuringPullCount += 1
|
||||
return@forEach
|
||||
}
|
||||
itemsByKey.remove(key)
|
||||
val removedItem = itemsByKey.remove(key)
|
||||
if (removedItem != null) {
|
||||
removedCount += 1
|
||||
} else if (
|
||||
removeWatchedItemByStableDeleteKey(
|
||||
contentId = event.contentId,
|
||||
contentType = event.contentType,
|
||||
season = event.season,
|
||||
episode = event.episode,
|
||||
)
|
||||
) {
|
||||
removedCount += 1
|
||||
removedByFallbackKeyCount += 1
|
||||
}
|
||||
}
|
||||
else -> {
|
||||
ignoredCount += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
log.i {
|
||||
"Applied watched delta events total=${events.size} upserts=$upsertCount deletes=$deleteCount " +
|
||||
"removed=$removedCount removedByFallbackKey=$removedByFallbackKeyCount " +
|
||||
"preservedDuringPull=$preservedDuringPullCount ignored=$ignoredCount"
|
||||
}
|
||||
}
|
||||
|
||||
private fun removeWatchedItemByStableDeleteKey(
|
||||
contentId: String,
|
||||
contentType: String,
|
||||
season: Int?,
|
||||
episode: Int?,
|
||||
): Boolean {
|
||||
val fallbackKey = itemsByKey.entries.firstOrNull { (_, item) ->
|
||||
item.id == contentId &&
|
||||
watchedDeleteTypesCompatible(remoteType = contentType, localType = item.type) &&
|
||||
item.season == season &&
|
||||
item.episode == episode
|
||||
}?.key ?: return false
|
||||
|
||||
itemsByKey.remove(fallbackKey)
|
||||
log.w {
|
||||
"Removed watched delta delete with fallback key contentId=$contentId contentType=$contentType " +
|
||||
"season=$season episode=$episode matchedKey=$fallbackKey"
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
private fun watchedDeleteTypesCompatible(remoteType: String, localType: String): Boolean {
|
||||
if (remoteType.equals(localType, ignoreCase = true)) return true
|
||||
return remoteType.isSeriesLikeWatchedType() && localType.isSeriesLikeWatchedType()
|
||||
}
|
||||
|
||||
fun toggleWatched(item: WatchedItem) {
|
||||
|
|
@ -658,12 +714,20 @@ internal fun shouldPreserveLocalWatchedItem(
|
|||
lastSuccessfulPushEpochMs: Long,
|
||||
pullStartedEpochMs: Long,
|
||||
): Boolean {
|
||||
val markedAt = localItem.markedAtEpochMs
|
||||
val markedAt = normalizeWatchedMarkedAtEpochMs(localItem.markedAtEpochMs)
|
||||
val wasMarkedAfterLastPush = lastSuccessfulPushEpochMs > 0L && markedAt > lastSuccessfulPushEpochMs
|
||||
val wasMarkedDuringPull = pullStartedEpochMs > 0L && markedAt >= pullStartedEpochMs
|
||||
return wasMarkedAfterLastPush || wasMarkedDuringPull
|
||||
}
|
||||
|
||||
internal fun wasWatchedItemMarkedDuringPull(
|
||||
localItem: WatchedItem,
|
||||
pullStartedEpochMs: Long,
|
||||
): Boolean {
|
||||
val markedAt = normalizeWatchedMarkedAtEpochMs(localItem.markedAtEpochMs)
|
||||
return pullStartedEpochMs > 0L && markedAt >= pullStartedEpochMs
|
||||
}
|
||||
|
||||
internal fun shouldUseTraktWatchedSync(
|
||||
isAuthenticated: Boolean,
|
||||
source: WatchProgressSource,
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ coil-svg = { module = "io.coil-kt.coil3:coil-svg", version.ref = "coil" }
|
|||
kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinx-serialization" }
|
||||
kotlinx-atomicfu = { module = "org.jetbrains.kotlinx:atomicfu", version.ref = "atomicfu" }
|
||||
kmpalette-core = { module = "com.materialkolor.palette:core", version.ref = "kmpalette" }
|
||||
ktor-client-android = { module = "io.ktor:ktor-client-android", version.ref = "ktor" }
|
||||
ktor-client-okhttp = { module = "io.ktor:ktor-client-okhttp", version.ref = "ktor" }
|
||||
kermit = { module = "co.touchlab:kermit", version.ref = "kermit" }
|
||||
ktor-client-darwin = { module = "io.ktor:ktor-client-darwin", version.ref = "ktor" }
|
||||
androidx-media3-exoplayer = { module = "androidx.media3:media3-exoplayer", version.ref = "androidx-media3" }
|
||||
|
|
@ -79,6 +79,7 @@ mpv-android-lib = { module = "io.github.abdallahmehiz:mpv-android-lib", version.
|
|||
supabase-postgrest = { module = "io.github.jan-tennert.supabase:postgrest-kt", version.ref = "supabase" }
|
||||
supabase-auth = { module = "io.github.jan-tennert.supabase:auth-kt", version.ref = "supabase" }
|
||||
supabase-functions = { module = "io.github.jan-tennert.supabase:functions-kt", version.ref = "supabase" }
|
||||
supabase-realtime = { module = "io.github.jan-tennert.supabase:realtime-kt", version.ref = "supabase" }
|
||||
quickjs-kt = { module = "io.github.dokar3:quickjs-kt", version.ref = "quickjsKt" }
|
||||
ksoup = { module = "com.fleeksoft.ksoup:ksoup", version.ref = "ksoup" }
|
||||
reorderable = { module = "sh.calvin.reorderable:reorderable", version.ref = "reorderable" }
|
||||
|
|
|
|||
Loading…
Reference in a new issue