diff --git a/app/build.gradle.kts b/app/build.gradle.kts index e40b13833..e903de91b 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -188,6 +188,7 @@ dependencies { implementation("androidx.core:core-ktx:1.12.0") implementation("androidx.core:core-splashscreen:1.0.1") // Android 12+ Splash Screen implementation("androidx.lifecycle:lifecycle-runtime-ktx:2.7.0") + implementation("androidx.lifecycle:lifecycle-process:2.7.0") implementation("androidx.lifecycle:lifecycle-viewmodel-compose:2.7.0") // Provides collectAsStateWithLifecycle — pauses Flow collection while the // screen is off so we don't drive recompositions on invisible UI. diff --git a/app/src/main/kotlin/com/arflix/tv/ArflixApplication.kt b/app/src/main/kotlin/com/arflix/tv/ArflixApplication.kt index 4a5cbed68..d742cdc88 100644 --- a/app/src/main/kotlin/com/arflix/tv/ArflixApplication.kt +++ b/app/src/main/kotlin/com/arflix/tv/ArflixApplication.kt @@ -46,6 +46,9 @@ import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import com.arflix.tv.util.settingsDataStore +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleEventObserver +import androidx.lifecycle.ProcessLifecycleOwner import java.util.concurrent.TimeUnit import javax.inject.Inject @@ -102,6 +105,31 @@ class ArflixApplication : Application(), Configuration.Provider, ImageLoaderFact // Wire realtime push notification cloudSyncRepository.onPushCompleted = { realtimeSyncManager.markPush() } + // Track foreground/background transitions via ProcessLifecycleOwner. + // Unlike ActivityLifecycleCallbacks with an AtomicInteger counter, + // ProcessLifecycleOwner correctly handles configuration changes + // (rotation, theme changes) and activity-to-activity navigation where + // the new activity starts before the old one stops — it fires only on + // genuine app-level foreground/background transitions. + // When the app comes to foreground, retry any pending dirty push so user + // changes (CW dismissals, settings edits, addon changes) that failed during + // a previous background session are propagated immediately — instead of + // waiting up to 45s for the periodic sync tick. + ProcessLifecycleOwner.get().lifecycle.addObserver(LifecycleEventObserver { _, event -> + if (event == Lifecycle.Event.ON_START) { + appScope.launch(Dispatchers.IO) { + // Brief delay to let the UI settle before network work + delay(500L) + if (authRepository.getCurrentUserId().isNullOrBlank()) return@launch + if (cloudSyncRepository.isPushDirty) { + android.util.Log.i("ArflixApp", "Foreground: retrying dirty push") + runCatching { cloudSyncRepository.pushToCloud() } + .onFailure { android.util.Log.w("ArflixApp", "Foreground push retry failed: ${it.message}") } + } + } + } + }) + appScope.launch { runCatching { profileManager.initialize() } // Preload watchlist cache in background for instant display diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt index fa87c29d2..3aab6dcfa 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt @@ -18,6 +18,10 @@ class CloudSyncCoordinator @Inject constructor( private val cloudSyncRepository: CloudSyncRepository, private val authRepository: AuthRepository ) { + companion object { + private const val TAG = "CloudSyncCoordinator" + } + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) private val lifecycleLock = Any() private var collectorJob: Job? = null @@ -48,6 +52,16 @@ class CloudSyncCoordinator @Inject constructor( } } + /** + * Debounce invalidation events, then push to cloud. + * + * Retry logic lives entirely inside [CloudSyncRepository.pushToCloud] (up to 3 + * attempts with 1.5s gaps). The coordinator does not add its own retry loop to + * avoid stacking backoffs and holding [cloudSyncMutex] for tens of seconds + * across nested retry cycles. If pushToCloud fails after all internal retries, + * the dirty flag is set so the periodic sync (45s) or foreground-resume retry + * (ArflixApplication) picks it up. + */ private fun scheduleFlush(invalidation: CloudSyncInvalidation) { synchronized(lifecycleLock) { if (!started.get()) return @@ -55,11 +69,13 @@ class CloudSyncCoordinator @Inject constructor( flushJob = scope.launch { delay(debounceMsFor(invalidation.scope)) if (authRepository.getCurrentUserId().isNullOrBlank()) return@launch - runCatching { cloudSyncRepository.pushToCloud() } - .onFailure { error -> - Log.w("CloudSyncCoordinator", "Cloud push failed after ${invalidation.scope}: ${error.message}") - cloudSyncRepository.markLocalStateDirty() - } + + val result = runCatching { cloudSyncRepository.pushToCloud() } + if (result.isFailure) { + Log.w(TAG, "Push failed after ${invalidation.scope}: ${result.exceptionOrNull()?.message}") + // Mark dirty so periodic/foreground sync retries + cloudSyncRepository.markLocalStateDirty() + } } } } diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncRepository.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncRepository.kt index 8991e7e2e..d63914d92 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncRepository.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncRepository.kt @@ -18,6 +18,7 @@ import com.arflix.tv.util.settingsDataStore import com.google.gson.Gson import com.google.gson.reflect.TypeToken import dagger.hilt.android.qualifiers.ApplicationContext +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.first import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -465,6 +466,21 @@ class CloudSyncRepository @Inject constructor( // PUSH LOCAL STATE TO CLOUD // ══════════════════════════════════════════════════════════ + /** + * Push local state to cloud with automatic retry on transient network failure. + * + * Builds the JSON payload once (local serialization — no network I/O, so retry + * is pointless), then retries the network save ([saveAccountSyncPayload]) up to + * 2 additional times with 1.5s gap. This covers common network hiccups (DNS + * failover, TLS renegotiation) without making callers wait longer than ~4s + * total. If all attempts fail, [isPushDirty] stays `true` so the periodic sync + * or foreground-resume retry picks it up later. + * + * NOTE: Retries run inside [cloudSyncMutex] to prevent concurrent pushes from + * interleaving. This is acceptable because the mutex is released between calls + * to [saveAccountSyncPayload] via the delay, and the coordinator no longer adds + * its own outer retry loop (removed to avoid stacked backoff). + */ suspend fun pushToCloud(): Result = cloudSyncMutex.withLock { if (authRepository.getCurrentUserId().isNullOrBlank()) { AppLogger.breadcrumb( @@ -474,6 +490,9 @@ class CloudSyncRepository @Inject constructor( ) return@withLock Result.failure(IllegalStateException("Not logged in")) } + + // Build payload once — local serialization only (DataStore + in-memory reads), + // no network I/O. Retrying on build failure would waste time for no benefit. val payload = runCatching { buildCloudSnapshotJson() }.getOrElse { isPushDirty = true AppLogger.recordException( @@ -486,31 +505,50 @@ class CloudSyncRepository @Inject constructor( ) return@withLock Result.failure(it) } - val result = authRepository.saveAccountSyncPayload(payload) - if (result.isSuccess) { - isPushDirty = false - AppLogger.breadcrumb( - tag = "CloudSync", - message = "push_success size=${payloadSizeBucket(payload)}", - severity = "info" - ) - onPushCompleted?.invoke() - } else { - // Mark dirty so the next ON_RESUME or periodic sync retries the push. - // Without this, a single network hiccup would permanently diverge the - // cloud state until the user explicitly changes another setting. - isPushDirty = true - AppLogger.recordException( - throwable = result.exceptionOrNull() ?: IllegalStateException("Cloud push failed"), - context = mapOf( - "error_area" to "CloudSync", - "cloud_flow" to "push_save_payload", - "dirty" to isPushDirty.toString(), - "payload_size" to payloadSizeBucket(payload) + + // Retry the network save on transient failure; the payload is stable + // (built once above) so we don't need to rebuild between attempts. + val maxAttempts = 3 + val retryDelayMs = 1_500L + var lastError: Throwable? = null + + for (attempt in 1..maxAttempts) { + val result = authRepository.saveAccountSyncPayload(payload) + if (result.isSuccess) { + isPushDirty = false + AppLogger.breadcrumb( + tag = "CloudSync", + message = "push_success attempt=${attempt} size=${payloadSizeBucket(payload)}", + severity = "info" ) - ) + onPushCompleted?.invoke() + return@withLock result + } + + lastError = result.exceptionOrNull() + if (attempt < maxAttempts) { + AppLogger.breadcrumb( + tag = "CloudSync", + message = "push_save_attempt=${attempt}_failed_retrying", + severity = "warning" + ) + delay(retryDelayMs) + } } - result + + // All attempts exhausted — mark dirty so periodic/foreground sync retries + isPushDirty = true + val finalError = lastError ?: IllegalStateException("Cloud push failed after $maxAttempts attempts") + AppLogger.recordException( + throwable = finalError, + context = mapOf( + "error_area" to "CloudSync", + "cloud_flow" to "push_save_payload", + "attempts" to maxAttempts, + "dirty" to isPushDirty.toString() + ) + ) + Result.failure(finalError) } // ══════════════════════════════════════════════════════════ @@ -995,6 +1033,13 @@ class CloudSyncRepository @Inject constructor( traktRepository.clearAllProfileCaches() watchHistoryRepository.clearProfileCaches() - System.err.println("[CLOUD-SYNC] Full cloud restore applied successfully") + // Re-initialize the watched cache for the current profile immediately after + // clearing. Without this, clearAllProfileCaches() sets cacheInitialized = false, + // and the next UI read (e.g. fetchSeasonProgress, loadSeason, isEpisodeWatched) + // would either trigger a slow re-fetch or fall back to stale/empty data — + // causing watched badges to disappear and the season-watched revert bug. + runCatching { traktRepository.initializeWatchedCache() } + + System.err.println("[CLOUD-SYNC] Full cloud restore applied successfully after cache re-init") } } diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/RealtimeSyncManager.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/RealtimeSyncManager.kt index 4d3198931..236e8a4ec 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/RealtimeSyncManager.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/RealtimeSyncManager.kt @@ -219,13 +219,26 @@ class RealtimeSyncManager @Inject constructor( } private fun joinChannel(ws: WebSocket, userId: String) { - // Channel 1: account_sync_state UPDATEs + // Channel 1: account_sync_state INSERTs + UPDATEs. + // saveAccountSyncPayload() uses upsert() which does an INSERT when the + // row doesn't exist yet (first-ever push from a device). If we only + // subscribe to UPDATE, other devices won't get a realtime notification + // for the very first push — they'd have to wait up to 45s for the + // periodic sync to discover it. Subscribing to both INSERT and UPDATE + // ensures the WebSocket fires on every push, regardless of whether the + // row was just created or already existed. val accountSyncJoin = JSONObject().apply { put("topic", "realtime:account_sync") put("event", "phx_join") put("payload", JSONObject().apply { put("config", JSONObject().apply { put("postgres_changes", JSONArray().apply { + put(JSONObject().apply { + put("event", "INSERT") + put("schema", "public") + put("table", "account_sync_state") + put("filter", "user_id=eq.$userId") + }) put(JSONObject().apply { put("event", "UPDATE") put("schema", "public") diff --git a/app/src/main/kotlin/com/arflix/tv/ui/screens/details/DetailsViewModel.kt b/app/src/main/kotlin/com/arflix/tv/ui/screens/details/DetailsViewModel.kt index 33734ddb5..3266a24e3 100644 --- a/app/src/main/kotlin/com/arflix/tv/ui/screens/details/DetailsViewModel.kt +++ b/app/src/main/kotlin/com/arflix/tv/ui/screens/details/DetailsViewModel.kt @@ -317,9 +317,20 @@ class DetailsViewModel @Inject constructor( async { mediaRepository.getSeasonEpisodes(mediaId, seasonToLoad) } } else null - // For TV shows, fetch season progress (watched/total per season) + // For TV shows, fetch season progress (watched/total per season). + // initializeWatchedCache() runs inside the async block so it executes + // on the coroutine dispatcher (not the UI thread) and in parallel with + // the other deferreds above (item, watchlist, externalIds, resume, logo, + // episodes). If the cache is already warm, initializeWatchedCache is + // a no-op (it guards on cacheInitialized internally). If cold, it + // fetches from Supabase/Trakt before fetchSeasonProgress reads it — + // preventing the season-watched revert bug where an empty cache caused + // fallback to stale backend data. val seasonProgressDeferred = if (mediaType == MediaType.TV) { - async { fetchSeasonProgress(mediaId) } + async { + runCatching { traktRepository.initializeWatchedCache() } + fetchSeasonProgress(mediaId) + } } else null val requestMediaId = mediaId @@ -1587,12 +1598,15 @@ class DetailsViewModel @Inject constructor( traktRepository.markSeasonWatched(currentMediaId, season, episodeNumbers) } - // 2. Remove from watch history concurrently (all episodes at once) + // 2. Remove from watch history FIRST (synchronous), before Supabase writes. + // This avoids a race condition where removeFromHistory deletes the + // just-written watched records, causing watched status to be lost on re-entry. runCatching { watchHistoryRepository.removeFromHistory(currentMediaId, season, null) } - // 3. Concurrent Supabase writes for each episode (faster than sequential) + // 3. Concurrent Supabase writes for each episode (faster than sequential). + // These run AFTER removeFromHistory to ensure the watched records are the final state. episodeNumbers.map { epNum -> async { runCatching { diff --git a/app/src/main/kotlin/com/arflix/tv/ui/screens/home/HomeViewModel.kt b/app/src/main/kotlin/com/arflix/tv/ui/screens/home/HomeViewModel.kt index fe4480ec2..02cf745c8 100644 --- a/app/src/main/kotlin/com/arflix/tv/ui/screens/home/HomeViewModel.kt +++ b/app/src/main/kotlin/com/arflix/tv/ui/screens/home/HomeViewModel.kt @@ -1133,6 +1133,27 @@ class HomeViewModel @Inject constructor( } } } + + // CRITICAL: Pull the cloud snapshot before refreshing CW. When Device A + // removes a CW item, it does: + // 1. removeFromHistory() → Supabase DELETE (broadcast via WebSocket) + // 2. pushToCloud() → updates account_sync_state with dismissed CW + local CW + // Device B receives the DELETE instantly via WebSocket, but its DataStore + // still has the old dismissed CW set and local CW cache — so the dismissed + // filter doesn't catch the item and the local CW re-fetch shows it again. + // Pulling the cloud snapshot first ensures Device B's DataStore has the + // latest dismissed CW + local CW data before the CW refresh runs. + // The 5s debounce in RealtimeSyncManager gives Device A's pushToCloud() + // time to complete before this pull arrives. + runCatching { + cloudSyncRepository.pullFromCloud() + }.onSuccess { restoreResult -> + if (restoreResult == CloudSyncRepository.RestoreResult.RESTORED) { + // Cloud state changed — reload home data to pick up catalog/addon/settings changes too + loadHomeData() + } + } + // Full refresh: re-resolve from all sources (Trakt, local, Supabase). // This runs after the fast path so the progress bar updates immediately, // then the authoritative Trakt data (with correct subtitle/resume label)