mirror of
https://github.com/tapframe/NuvioStreaming.git
synced 2026-07-28 23:33:02 +00:00
fix(trakt): preserve continue watching sync
This commit is contained in:
tapframe
parent
2309970f20
commit
ebffbb4ebd
5 changed files with 263 additions and 123 deletions
|
|
@ -227,7 +227,7 @@ object SyncManager {
|
|||
|
||||
private fun pullForegroundForProfile(profileId: Int) {
|
||||
scope.launch {
|
||||
log.i { "pullForegroundForProfile($profileId) — syncing watch progress, library, collections, and home settings" }
|
||||
log.i { "pullForegroundForProfile($profileId) — syncing watch progress, watched items, library, collections, and home settings" }
|
||||
|
||||
runCatching { TraktCredentialSync.pullFromRemote(profileId) }
|
||||
.onFailure { log.e(it) { "Foreground Trakt credential pull failed" } }
|
||||
|
|
@ -238,10 +238,15 @@ object SyncManager {
|
|||
}
|
||||
|
||||
launch {
|
||||
runCatching { WatchProgressRepository.pullFromServer(profileId) }
|
||||
runCatching { WatchProgressRepository.forceSnapshotRefreshFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Foreground watch progress pull failed" } }
|
||||
}
|
||||
|
||||
launch {
|
||||
runCatching { WatchedRepository.forceSnapshotRefreshFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Foreground watched items pull failed" } }
|
||||
}
|
||||
|
||||
launch {
|
||||
runCatching { CollectionSyncService.pullFromServer(profileId) }
|
||||
.onFailure { log.e(it) { "Foreground collections pull failed" } }
|
||||
|
|
|
|||
|
|
@ -79,15 +79,15 @@ import com.nuvio.app.features.collection.CollectionRepository
|
|||
import com.nuvio.app.features.profiles.ProfileRepository
|
||||
import com.nuvio.app.features.home.components.HomeCollectionRowSection
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.combine
|
||||
import kotlinx.coroutines.flow.emptyFlow
|
||||
import kotlinx.coroutines.flow.flowOn
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Semaphore
|
||||
import kotlinx.coroutines.sync.withPermit
|
||||
import kotlinx.coroutines.withContext
|
||||
import kotlinx.coroutines.yield
|
||||
import com.nuvio.app.features.trakt.TraktEpisodeMappingService
|
||||
import com.nuvio.app.features.home.components.continueWatchingHeroViewportReserveHeight
|
||||
|
|
@ -415,48 +415,77 @@ fun HomeScreen(
|
|||
val semaphore = Semaphore(NEXT_UP_RESOLUTION_CONCURRENCY)
|
||||
val freshResults = mutableMapOf<String, Pair<Long, ContinueWatchingItem>>()
|
||||
val processedFreshContentIds = mutableSetOf<String>()
|
||||
val candidateBatches = resolutionCandidates.chunked(NEXT_UP_RESOLUTION_BATCH_SIZE)
|
||||
|
||||
for (batch in candidateBatches) {
|
||||
val batchResults = batch.map { completedEntry ->
|
||||
async {
|
||||
semaphore.withPermit {
|
||||
resolveHomeNextUpCandidate(
|
||||
completedEntry = completedEntry,
|
||||
watchProgressEntries = watchProgressUiState.entries,
|
||||
watchedItems = watchedUiState.items,
|
||||
todayIsoDate = todayIsoDate,
|
||||
preferFurthestEpisode = continueWatchingPreferences.upNextFromFurthestEpisode,
|
||||
showUnairedNextUp = continueWatchingPreferences.showUnairedNextUp,
|
||||
dismissedNextUpKeys = continueWatchingPreferences.dismissedNextUpKeys,
|
||||
isTraktProgressActive = isTraktProgressActive,
|
||||
)
|
||||
suspend fun resolveCandidatesStreaming(
|
||||
candidates: List<CompletedSeriesCandidate>,
|
||||
) {
|
||||
if (candidates.isEmpty()) return
|
||||
|
||||
val results = Channel<HomeNextUpCandidateResolution>(Channel.UNLIMITED)
|
||||
candidates.forEach { completedEntry ->
|
||||
launch {
|
||||
val resolved = try {
|
||||
semaphore.withPermit {
|
||||
resolveHomeNextUpCandidate(
|
||||
completedEntry = completedEntry,
|
||||
watchProgressEntries = watchProgressUiState.entries,
|
||||
watchedItems = watchedUiState.items,
|
||||
todayIsoDate = todayIsoDate,
|
||||
preferFurthestEpisode = continueWatchingPreferences.upNextFromFurthestEpisode,
|
||||
showUnairedNextUp = continueWatchingPreferences.showUnairedNextUp,
|
||||
dismissedNextUpKeys = continueWatchingPreferences.dismissedNextUpKeys,
|
||||
isTraktProgressActive = isTraktProgressActive,
|
||||
)
|
||||
}
|
||||
} catch (error: Throwable) {
|
||||
if (error is CancellationException) throw error
|
||||
null
|
||||
}
|
||||
results.send(
|
||||
HomeNextUpCandidateResolution(
|
||||
candidate = completedEntry,
|
||||
resolved = resolved,
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
repeat(candidates.size) {
|
||||
val resolution = results.receive()
|
||||
processedFreshContentIds += resolution.candidate.content.id
|
||||
|
||||
var changed = false
|
||||
resolution.resolved?.let { (contentId, item) ->
|
||||
if (cachedResolvedNextUpItems.size + freshResults.size < HomeContinueWatchingMaxRecentProgressItems) {
|
||||
val previous = freshResults.put(contentId, item)
|
||||
changed = previous != item
|
||||
}
|
||||
}
|
||||
}.awaitAll()
|
||||
batch.forEach { candidate -> processedFreshContentIds += candidate.content.id }
|
||||
|
||||
val resolvedBeforeBatch = freshResults.size
|
||||
batchResults.filterNotNull().forEach { (contentId, item) ->
|
||||
freshResults[contentId] = item
|
||||
}
|
||||
val batchResolvedCount = freshResults.size - resolvedBeforeBatch
|
||||
if (batchResolvedCount > 0) {
|
||||
val progressiveResults = cachedResolvedNextUpItems + freshResults
|
||||
withContext(Dispatchers.Main) {
|
||||
nextUpItemsBySeries = progressiveResults
|
||||
processedNextUpContentIds = (
|
||||
cachedResolvedNextUpItems.keys +
|
||||
processedFreshContentIds
|
||||
).toSet()
|
||||
if (changed) {
|
||||
val progressiveResults = cachedResolvedNextUpItems + freshResults
|
||||
withContext(Dispatchers.Main) {
|
||||
nextUpItemsBySeries = progressiveResults
|
||||
processedNextUpContentIds = (
|
||||
cachedResolvedNextUpItems.keys +
|
||||
processedFreshContentIds
|
||||
).toSet()
|
||||
}
|
||||
saveContinueWatchingSnapshots(
|
||||
profileId = activeProfileId,
|
||||
nextUpItemsBySeries = progressiveResults,
|
||||
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
|
||||
todayIsoDate = todayIsoDate,
|
||||
seedLastWatchedMap = seedLastWatchedMap,
|
||||
)
|
||||
}
|
||||
yield()
|
||||
}
|
||||
|
||||
if (cachedResolvedNextUpItems.size + freshResults.size >= HomeContinueWatchingMaxRecentProgressItems) {
|
||||
break
|
||||
}
|
||||
results.close()
|
||||
}
|
||||
|
||||
resolveCandidatesStreaming(resolutionCandidates)
|
||||
|
||||
val results = cachedResolvedNextUpItems + freshResults
|
||||
withContext(Dispatchers.Main) {
|
||||
nextUpItemsBySeries = results
|
||||
|
|
@ -478,55 +507,23 @@ fun HomeScreen(
|
|||
return@withContext
|
||||
}
|
||||
|
||||
val deferredCandidateBatches = deferredResolutionCandidates.chunked(NEXT_UP_RESOLUTION_BATCH_SIZE)
|
||||
for (batch in deferredCandidateBatches) {
|
||||
if (cachedResolvedNextUpItems.size + freshResults.size >= HomeContinueWatchingMaxRecentProgressItems) {
|
||||
break
|
||||
}
|
||||
resolveCandidatesStreaming(deferredResolutionCandidates)
|
||||
|
||||
val batchResults = batch.map { completedEntry ->
|
||||
async {
|
||||
semaphore.withPermit {
|
||||
resolveHomeNextUpCandidate(
|
||||
completedEntry = completedEntry,
|
||||
watchProgressEntries = watchProgressUiState.entries,
|
||||
watchedItems = watchedUiState.items,
|
||||
todayIsoDate = todayIsoDate,
|
||||
preferFurthestEpisode = continueWatchingPreferences.upNextFromFurthestEpisode,
|
||||
showUnairedNextUp = continueWatchingPreferences.showUnairedNextUp,
|
||||
dismissedNextUpKeys = continueWatchingPreferences.dismissedNextUpKeys,
|
||||
isTraktProgressActive = isTraktProgressActive,
|
||||
)
|
||||
}
|
||||
}
|
||||
}.awaitAll()
|
||||
batch.forEach { candidate -> processedFreshContentIds += candidate.content.id }
|
||||
|
||||
val resolvedBeforeBatch = freshResults.size
|
||||
batchResults.filterNotNull().forEach { (contentId, item) ->
|
||||
if (cachedResolvedNextUpItems.size + freshResults.size < HomeContinueWatchingMaxRecentProgressItems) {
|
||||
freshResults[contentId] = item
|
||||
}
|
||||
}
|
||||
if (freshResults.size > resolvedBeforeBatch) {
|
||||
val progressiveResults = cachedResolvedNextUpItems + freshResults
|
||||
withContext(Dispatchers.Main) {
|
||||
nextUpItemsBySeries = progressiveResults
|
||||
processedNextUpContentIds = (
|
||||
cachedResolvedNextUpItems.keys +
|
||||
processedFreshContentIds
|
||||
).toSet()
|
||||
}
|
||||
saveContinueWatchingSnapshots(
|
||||
profileId = activeProfileId,
|
||||
nextUpItemsBySeries = progressiveResults,
|
||||
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
|
||||
todayIsoDate = todayIsoDate,
|
||||
seedLastWatchedMap = seedLastWatchedMap,
|
||||
)
|
||||
}
|
||||
yield()
|
||||
val deferredResults = cachedResolvedNextUpItems + freshResults
|
||||
withContext(Dispatchers.Main) {
|
||||
nextUpItemsBySeries = deferredResults
|
||||
processedNextUpContentIds = (
|
||||
cachedResolvedNextUpItems.keys +
|
||||
processedFreshContentIds
|
||||
).toSet()
|
||||
}
|
||||
saveContinueWatchingSnapshots(
|
||||
profileId = activeProfileId,
|
||||
nextUpItemsBySeries = deferredResults,
|
||||
visibleContinueWatchingEntries = visibleContinueWatchingEntries,
|
||||
todayIsoDate = todayIsoDate,
|
||||
seedLastWatchedMap = seedLastWatchedMap,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -783,7 +780,6 @@ internal const val HomeNextUpInitialResolutionLimit = 32
|
|||
private const val MILLIS_PER_DAY = 24L * 60L * 60L * 1000L
|
||||
private const val OPTIMISTIC_NEXT_UP_SEED_WINDOW_MS = 3L * 60L * 1000L
|
||||
private const val NEXT_UP_RESOLUTION_CONCURRENCY = 4
|
||||
private const val NEXT_UP_RESOLUTION_BATCH_SIZE = NEXT_UP_RESOLUTION_CONCURRENCY
|
||||
|
||||
internal data class HomeProgressDerivedState(
|
||||
val watchProgressUiState: WatchProgressUiState = WatchProgressUiState(),
|
||||
|
|
@ -1321,6 +1317,11 @@ private data class HomeContinueWatchingCandidate(
|
|||
val isProgressEntry: Boolean,
|
||||
)
|
||||
|
||||
private data class HomeNextUpCandidateResolution(
|
||||
val candidate: CompletedSeriesCandidate,
|
||||
val resolved: Pair<String, Pair<Long, ContinueWatchingItem>>?,
|
||||
)
|
||||
|
||||
private fun saveContinueWatchingSnapshots(
|
||||
profileId: Int,
|
||||
nextUpItemsBySeries: Map<String, Pair<Long, ContinueWatchingItem>>,
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import kotlinx.coroutines.delay
|
|||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.flow.update
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.Semaphore
|
||||
|
|
@ -37,6 +38,7 @@ import nuvio.composeapp.generated.resources.*
|
|||
import org.jetbrains.compose.resources.getString
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlin.math.abs
|
||||
|
||||
private const val BASE_URL = "https://api.trakt.tv"
|
||||
private const val TRAKT_COMPLETION_PERCENT_THRESHOLD = 90f
|
||||
|
|
@ -53,6 +55,7 @@ private const val METADATA_HYDRATION_LIMIT = 110
|
|||
private const val REFRESH_BASE_INTERVAL_MS = 60L * 1000L
|
||||
private const val EPISODE_PROGRESS_CACHE_TTL_MS = 30L * 60L * 1000L
|
||||
private const val EPISODE_PROGRESS_FETCH_THROTTLE_MS = 60L * 1000L
|
||||
private const val OPTIMISTIC_PROGRESS_TTL_MS = 3L * 60L * 1000L
|
||||
private const val MILLIS_PER_DAY = 24L * 60L * 60L * 1000L
|
||||
private const val AMBIGUOUS_ID_MARKER = "__ambiguous__"
|
||||
|
||||
|
|
@ -64,6 +67,11 @@ data class TraktProgressUiState(
|
|||
)
|
||||
|
||||
object TraktProgressRepository {
|
||||
private data class OptimisticProgressEntry(
|
||||
val progress: WatchProgressEntry,
|
||||
val expiresAtMs: Long,
|
||||
)
|
||||
|
||||
private val log = Logger.withTag("TraktProgress")
|
||||
private val json = Json { ignoreUnknownKeys = true }
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
|
|
@ -73,11 +81,13 @@ object TraktProgressRepository {
|
|||
val uiState: StateFlow<TraktProgressUiState> = _uiState.asStateFlow()
|
||||
|
||||
private val hiddenProgressShowIds = MutableStateFlow<Set<String>>(emptySet())
|
||||
private val optimisticProgress = MutableStateFlow<Map<String, OptimisticProgressEntry>>(emptyMap())
|
||||
|
||||
private var hasLoaded = false
|
||||
private var refreshRequestId: Long = 0L
|
||||
private val refreshJobMutex = Mutex()
|
||||
private var inFlightRefresh: Deferred<Unit>? = null
|
||||
private var remoteEntriesSnapshot: List<WatchProgressEntry> = emptyList()
|
||||
private var refreshIntervalMs = REFRESH_BASE_INTERVAL_MS
|
||||
private var lastKnownMoviesWatchedAt: String? = null
|
||||
private var lastKnownEpisodeActivityFingerprint: String? = null
|
||||
|
|
@ -138,6 +148,8 @@ object TraktProgressRepository {
|
|||
fun onProfileChanged() {
|
||||
invalidateInFlightRefreshes()
|
||||
hasLoaded = false
|
||||
remoteEntriesSnapshot = emptyList()
|
||||
optimisticProgress.value = emptyMap()
|
||||
hiddenProgressShowIds.value = emptySet()
|
||||
resetActivitySnapshot()
|
||||
resetShowProgressCaches()
|
||||
|
|
@ -148,6 +160,8 @@ object TraktProgressRepository {
|
|||
fun clearLocalState() {
|
||||
invalidateInFlightRefreshes()
|
||||
hasLoaded = false
|
||||
remoteEntriesSnapshot = emptyList()
|
||||
optimisticProgress.value = emptyMap()
|
||||
hiddenProgressShowIds.value = emptySet()
|
||||
resetActivitySnapshot()
|
||||
resetShowProgressCaches()
|
||||
|
|
@ -179,6 +193,18 @@ object TraktProgressRepository {
|
|||
}
|
||||
}
|
||||
|
||||
suspend fun invalidateAndRefresh() {
|
||||
ensureLoaded()
|
||||
invalidateInFlightRefreshes()
|
||||
resetActivitySnapshot()
|
||||
episodeProgressMutex.withLock {
|
||||
episodeProgressFetchedAtMsByContentId.clear()
|
||||
episodeProgressLastAttemptAtMsByContentId.clear()
|
||||
inFlightEpisodeProgressContentIds.clear()
|
||||
}
|
||||
refreshNow()
|
||||
}
|
||||
|
||||
suspend fun refreshEpisodeProgress(
|
||||
contentId: String,
|
||||
forceRefresh: Boolean = false,
|
||||
|
|
@ -208,14 +234,14 @@ object TraktProgressRepository {
|
|||
|
||||
try {
|
||||
val entries = fetchEpisodeProgressEntries(headers = headers, contentId = normalizedContentId)
|
||||
val existingEntries = _uiState.value.entries
|
||||
val merged = mergeNewestByVideoId(existingEntries + entries)
|
||||
_uiState.value = _uiState.value.copy(entries = merged.sortedByDescending { it.lastUpdatedEpochMs })
|
||||
remoteEntriesSnapshot = mergeNewestByVideoId(remoteEntriesSnapshot + entries)
|
||||
reconcileOptimisticProgress(remoteEntriesSnapshot)
|
||||
publishProgressState(entries = remoteEntriesSnapshot)
|
||||
episodeProgressMutex.withLock {
|
||||
episodeProgressFetchedAtMsByContentId[cacheKey] = TraktPlatformClock.nowEpochMs()
|
||||
}
|
||||
if (entries.isNotEmpty()) {
|
||||
launchHydration(requestId = refreshRequestId, entries = entries)
|
||||
launchHydration(requestId = refreshRequestId, entries = _uiState.value.entries)
|
||||
}
|
||||
} finally {
|
||||
episodeProgressMutex.withLock {
|
||||
|
|
@ -295,11 +321,13 @@ object TraktProgressRepository {
|
|||
val requestId = nextRefreshRequestId()
|
||||
val headers = TraktAuthRepository.authorizedHeaders()
|
||||
if (headers == null) {
|
||||
remoteEntriesSnapshot = emptyList()
|
||||
optimisticProgress.value = emptyMap()
|
||||
_uiState.value = TraktProgressUiState()
|
||||
return
|
||||
}
|
||||
|
||||
_uiState.value = _uiState.value.copy(isLoading = true, errorMessage = null)
|
||||
publishProgressState(isLoading = true, errorMessage = null)
|
||||
|
||||
val playbackEntries = runCatching {
|
||||
fetchPlaybackEntries(headers)
|
||||
|
|
@ -321,14 +349,14 @@ object TraktProgressRepository {
|
|||
// Merge new playback entries into the existing state rather than replacing it wholesale.
|
||||
// This prevents the CW list from briefly losing "next up" seeds (like One Piece) for the
|
||||
// ~2.8s gap between fetchPlaybackEntries completing and the full sync finishing.
|
||||
val existingEntries = _uiState.value.entries
|
||||
val mergedWithPlayback = if (existingEntries.isEmpty()) {
|
||||
val existingEntries = remoteEntriesSnapshot
|
||||
remoteEntriesSnapshot = if (existingEntries.isEmpty()) {
|
||||
playbackEntries
|
||||
} else {
|
||||
mergeNewestByVideoId(existingEntries + playbackEntries)
|
||||
}
|
||||
_uiState.value = _uiState.value.copy(
|
||||
entries = mergedWithPlayback,
|
||||
publishProgressState(
|
||||
entries = remoteEntriesSnapshot,
|
||||
isLoading = true,
|
||||
errorMessage = null,
|
||||
hasLoadedRemoteProgress = false,
|
||||
|
|
@ -351,7 +379,7 @@ object TraktProgressRepository {
|
|||
|
||||
if (completedEntries == null) {
|
||||
if (!isLatestRefreshRequest(requestId)) return
|
||||
_uiState.value = _uiState.value.copy(
|
||||
publishProgressState(
|
||||
isLoading = false,
|
||||
errorMessage = null,
|
||||
hasLoadedRemoteProgress = false,
|
||||
|
|
@ -361,18 +389,19 @@ object TraktProgressRepository {
|
|||
|
||||
if (!isLatestRefreshRequest(requestId)) return
|
||||
|
||||
val merged = mergeNewestByVideoId(playbackEntries + completedEntries)
|
||||
val sortedMerged = merged.sortedByDescending { it.lastUpdatedEpochMs }
|
||||
remoteEntriesSnapshot = mergeNewestByVideoId(playbackEntries + completedEntries)
|
||||
reconcileOptimisticProgress(remoteEntriesSnapshot)
|
||||
|
||||
_uiState.value = _uiState.value.copy(
|
||||
entries = sortedMerged,
|
||||
publishProgressState(
|
||||
entries = remoteEntriesSnapshot,
|
||||
isLoading = false,
|
||||
errorMessage = null,
|
||||
hasLoadedRemoteProgress = true,
|
||||
)
|
||||
|
||||
if (sortedMerged.isNotEmpty()) {
|
||||
launchHydration(requestId = requestId, entries = sortedMerged)
|
||||
|
||||
val visibleEntries = _uiState.value.entries
|
||||
if (visibleEntries.isNotEmpty()) {
|
||||
launchHydration(requestId = requestId, entries = visibleEntries)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -390,27 +419,129 @@ object TraktProgressRepository {
|
|||
|
||||
if (!isLatestRefreshRequest(requestId)) return@launch
|
||||
|
||||
val merged = mergeEntriesPreferRichMetadata(
|
||||
current = _uiState.value.entries,
|
||||
hydrated = hydrated,
|
||||
)
|
||||
val sortedMerged = merged.sortedByDescending { it.lastUpdatedEpochMs }
|
||||
_uiState.value = _uiState.value.copy(
|
||||
entries = sortedMerged,
|
||||
val hydratedByVideoId = hydrated.associateBy { it.videoId }
|
||||
val remoteVideoIds = remoteEntriesSnapshot.mapTo(mutableSetOf()) { it.videoId }
|
||||
val remoteHydrated = hydrated.filter { it.videoId in remoteVideoIds }
|
||||
if (remoteHydrated.isNotEmpty()) {
|
||||
remoteEntriesSnapshot = mergeEntriesPreferRichMetadata(
|
||||
current = remoteEntriesSnapshot,
|
||||
hydrated = remoteHydrated,
|
||||
)
|
||||
}
|
||||
|
||||
optimisticProgress.update { current ->
|
||||
current.mapValues { (videoId, optimistic) ->
|
||||
val hydratedEntry = hydratedByVideoId[videoId] ?: return@mapValues optimistic
|
||||
val normalizedHydrated = hydratedEntry.normalizedCompletion()
|
||||
if (shouldReplaceEntry(existing = optimistic.progress, candidate = normalizedHydrated)) {
|
||||
optimistic.copy(progress = normalizedHydrated)
|
||||
} else {
|
||||
optimistic
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
publishProgressState(
|
||||
isLoading = false,
|
||||
errorMessage = null,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private fun publishProgressState(
|
||||
entries: List<WatchProgressEntry> = remoteEntriesSnapshot,
|
||||
isLoading: Boolean = _uiState.value.isLoading,
|
||||
errorMessage: String? = _uiState.value.errorMessage,
|
||||
hasLoadedRemoteProgress: Boolean = _uiState.value.hasLoadedRemoteProgress,
|
||||
) {
|
||||
_uiState.value = TraktProgressUiState(
|
||||
entries = mergeWithActiveOptimistic(entries),
|
||||
isLoading = isLoading,
|
||||
errorMessage = errorMessage,
|
||||
hasLoadedRemoteProgress = hasLoadedRemoteProgress,
|
||||
)
|
||||
}
|
||||
|
||||
private fun mergeWithActiveOptimistic(entries: List<WatchProgressEntry>): List<WatchProgressEntry> {
|
||||
val optimisticEntries = activeOptimisticEntries()
|
||||
if (optimisticEntries.isEmpty()) {
|
||||
return entries.sortedByDescending { it.lastUpdatedEpochMs }
|
||||
}
|
||||
return mergeNewestByVideoId(entries + optimisticEntries)
|
||||
}
|
||||
|
||||
private fun activeOptimisticEntries(
|
||||
now: Long = TraktPlatformClock.nowEpochMs(),
|
||||
): List<WatchProgressEntry> {
|
||||
val current = optimisticProgress.value
|
||||
if (current.isEmpty()) return emptyList()
|
||||
val active = current.filterValues { entry -> entry.expiresAtMs > now }
|
||||
if (active.size != current.size) {
|
||||
optimisticProgress.value = active
|
||||
}
|
||||
return active.values.map { it.progress }
|
||||
}
|
||||
|
||||
private fun putOptimisticProgress(entry: WatchProgressEntry) {
|
||||
val now = TraktPlatformClock.nowEpochMs()
|
||||
optimisticProgress.update { current ->
|
||||
val active = current
|
||||
.filterValues { optimistic -> optimistic.expiresAtMs > now }
|
||||
.toMutableMap()
|
||||
val existing = active[entry.videoId]?.progress
|
||||
if (existing == null || shouldReplaceProgressSnapshotEntry(existing = existing, candidate = entry)) {
|
||||
active[entry.videoId] = OptimisticProgressEntry(
|
||||
progress = entry,
|
||||
expiresAtMs = now + OPTIMISTIC_PROGRESS_TTL_MS,
|
||||
)
|
||||
}
|
||||
active
|
||||
}
|
||||
}
|
||||
|
||||
private fun removeOptimisticProgress(
|
||||
shouldRemove: (WatchProgressEntry) -> Boolean,
|
||||
) {
|
||||
optimisticProgress.update { current ->
|
||||
current.filterValues { optimistic -> !shouldRemove(optimistic.progress) }
|
||||
}
|
||||
}
|
||||
|
||||
private fun reconcileOptimisticProgress(remoteEntries: List<WatchProgressEntry>) {
|
||||
if (remoteEntries.isEmpty() || optimisticProgress.value.isEmpty()) return
|
||||
val remoteByVideoId = remoteEntries.associateBy { it.videoId }
|
||||
val now = TraktPlatformClock.nowEpochMs()
|
||||
optimisticProgress.update { current ->
|
||||
current.filter { (videoId, optimistic) ->
|
||||
if (optimistic.expiresAtMs <= now) return@filter false
|
||||
val remoteEntry = remoteByVideoId[videoId] ?: return@filter true
|
||||
!remoteConfirmsOptimisticEntry(
|
||||
remote = remoteEntry,
|
||||
optimistic = optimistic.progress,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun remoteConfirmsOptimisticEntry(
|
||||
remote: WatchProgressEntry,
|
||||
optimistic: WatchProgressEntry,
|
||||
): Boolean {
|
||||
val normalizedRemote = remote.normalizedCompletion()
|
||||
val normalizedOptimistic = optimistic.normalizedCompletion()
|
||||
val remoteNewEnough = normalizedRemote.lastUpdatedEpochMs >= normalizedOptimistic.lastUpdatedEpochMs - 60_000L
|
||||
if (normalizedOptimistic.isEffectivelyCompleted) {
|
||||
return normalizedRemote.isEffectivelyCompleted && remoteNewEnough
|
||||
}
|
||||
|
||||
val closeEnough = abs(normalizedRemote.progressFraction - normalizedOptimistic.progressFraction) <= 0.03f
|
||||
return closeEnough && remoteNewEnough
|
||||
}
|
||||
|
||||
fun applyOptimisticProgress(entry: WatchProgressEntry) {
|
||||
if (!TraktAuthRepository.isAuthenticated.value) return
|
||||
val current = _uiState.value.entries.associateBy { it.videoId }.toMutableMap()
|
||||
val normalizedEntry = entry.normalizedCompletion()
|
||||
val existing = current[normalizedEntry.videoId]
|
||||
if (existing == null || normalizedEntry.lastUpdatedEpochMs >= existing.lastUpdatedEpochMs) {
|
||||
current[normalizedEntry.videoId] = normalizedEntry
|
||||
}
|
||||
putOptimisticProgress(normalizedEntry)
|
||||
if (normalizedEntry.isCompleted && normalizedEntry.seasonNumber != null && normalizedEntry.episodeNumber != null) {
|
||||
optimisticallyAddWatchedEpisode(
|
||||
contentId = normalizedEntry.parentMetaId,
|
||||
|
|
@ -418,14 +549,15 @@ object TraktProgressRepository {
|
|||
episode = normalizedEntry.episodeNumber,
|
||||
)
|
||||
}
|
||||
_uiState.value = _uiState.value.copy(entries = current.values.sortedByDescending { it.lastUpdatedEpochMs })
|
||||
publishProgressState()
|
||||
}
|
||||
|
||||
fun applyOptimisticRemoval(videoId: String) {
|
||||
if (!TraktAuthRepository.isAuthenticated.value) return
|
||||
if (videoId.isBlank()) return
|
||||
val filtered = _uiState.value.entries.filterNot { it.videoId == videoId }
|
||||
_uiState.value = _uiState.value.copy(entries = filtered)
|
||||
removeOptimisticProgress { it.videoId == videoId }
|
||||
remoteEntriesSnapshot = remoteEntriesSnapshot.filterNot { it.videoId == videoId }
|
||||
publishProgressState()
|
||||
}
|
||||
|
||||
fun applyOptimisticRemoval(
|
||||
|
|
@ -436,7 +568,7 @@ object TraktProgressRepository {
|
|||
if (!TraktAuthRepository.isAuthenticated.value) return
|
||||
val normalizedContentId = contentId.trim()
|
||||
if (normalizedContentId.isBlank()) return
|
||||
val filtered = _uiState.value.entries.filterNot { entry ->
|
||||
val shouldRemove: (WatchProgressEntry) -> Boolean = { entry ->
|
||||
if (entry.parentMetaId != normalizedContentId) {
|
||||
false
|
||||
} else if (seasonNumber != null && episodeNumber != null) {
|
||||
|
|
@ -445,6 +577,8 @@ object TraktProgressRepository {
|
|||
true
|
||||
}
|
||||
}
|
||||
removeOptimisticProgress(shouldRemove)
|
||||
remoteEntriesSnapshot = remoteEntriesSnapshot.filterNot(shouldRemove)
|
||||
if (seasonNumber != null && episodeNumber != null) {
|
||||
optimisticallyRemoveWatchedEpisode(
|
||||
contentId = normalizedContentId,
|
||||
|
|
@ -452,7 +586,7 @@ object TraktProgressRepository {
|
|||
episode = episodeNumber,
|
||||
)
|
||||
}
|
||||
_uiState.value = _uiState.value.copy(entries = filtered)
|
||||
publishProgressState()
|
||||
}
|
||||
|
||||
suspend fun removeProgress(
|
||||
|
|
|
|||
|
|
@ -237,7 +237,7 @@ internal object TraktScrobbleRepository {
|
|||
)
|
||||
|
||||
if (action == "stop") {
|
||||
runCatching { TraktProgressRepository.refreshNow() }
|
||||
runCatching { TraktProgressRepository.invalidateAndRefresh() }
|
||||
.onFailure { error ->
|
||||
if (error is CancellationException) throw error
|
||||
log.w { "Failed to refresh Trakt progress after stop: ${error.message}" }
|
||||
|
|
|
|||
|
|
@ -283,7 +283,7 @@ object WatchProgressRepository {
|
|||
|
||||
if (shouldUseTraktProgress()) {
|
||||
log.d { "Force refreshing Trakt watch progress for profile $profileId" }
|
||||
runCatching { TraktProgressRepository.refreshNow() }
|
||||
runCatching { TraktProgressRepository.invalidateAndRefresh() }
|
||||
.onFailure { error ->
|
||||
if (error is CancellationException) throw error
|
||||
log.e(error) { "Failed to force refresh Trakt progress" }
|
||||
|
|
|
|||
Loading…
Reference in a new issue