Fix/cloud sync reliability overhaul

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.

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

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,18 +52,30 @@ 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
             flushJob?.cancel()
             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()
+                }
             }
         }
     }

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<Unit> = 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")
     }
 }

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")

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 {

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)