fix: sync playback resume position across devices in real-time (#91)

app/src/main/kotlin/com/arflix/tv/data/repository/RealtimeSyncManager.kt

@@ -3,6 +3,9 @@ package com.arflix.tv.data.repository
 import android.util.Log
 import com.arflix.tv.util.Constants
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import okhttp3.*
 import org.json.JSONArray
 import org.json.JSONObject
@@ -14,13 +17,25 @@ import javax.inject.Singleton
 
 /**
  * Manages a Supabase Realtime WebSocket connection to receive instant
- * notifications when `account_sync_state` changes on any device.
+ * notifications when `account_sync_state` or `watch_history` changes on any device.
  *
- * When a change is detected, it triggers [CloudSyncRepository.pullFromCloud]
- * so the local app state updates within 1-2 seconds of a remote push.
+ * Two channels are joined on the same socket:
  *
- * Also runs a periodic fallback sync every 5 minutes in case the WebSocket
- * disconnects or misses an event.
+ * 1. `realtime:account_sync` \u2014 listens for UPDATEs on `account_sync_state`. On
+ *    change, the manager triggers [CloudSyncRepository.pullFromCloud] to reapply the
+ *    full JSON snapshot (addons, profiles, catalogs, IPTV config, preferences).
+ *
+ * 2. `realtime:watch_history` \u2014 listens for INSERTs/UPDATEs on `watch_history` so
+ *    the Home screen's Continue Watching row can refresh on other devices within
+ *    seconds of a progress update, instead of waiting for the user to reopen Home.
+ *    Because watch_history is a high-write table during playback (~every 10s), the
+ *    manager debounces events into [watchHistoryEvents] and the subscriber (HomeViewModel)
+ *    is expected to collect the flow and call refreshContinueWatchingOnly(force = true).
+ *    We do NOT trigger a full pullFromCloud on these events \u2014 that's wasteful for
+ *    what is effectively a single-row position update. Fixes issue #91.
+ *
+ * A periodic fallback sync runs every 90 seconds (lowered from 5 minutes) in case the
+ * WebSocket disconnects or misses an event on an unstable connection.
  */
 @Singleton
 class RealtimeSyncManager @Inject constructor(
@@ -31,8 +46,15 @@ class RealtimeSyncManager @Inject constructor(
         private const val TAG = "RealtimeSync"
         private const val HEARTBEAT_INTERVAL_MS = 30_000L
         private const val RECONNECT_DELAY_MS = 5_000L
-        private const val PERIODIC_SYNC_INTERVAL_MS = 5 * 60 * 1000L // 5 minutes
-        private const val DEBOUNCE_MS = 2_000L // Debounce pulls to avoid rapid-fire
+        // Periodic fallback: was 5 minutes. Lowered to 90s so users on flaky connections
+        // still get fresh playback resume positions within a reasonable time even if
+        // the WebSocket silently misses an event (#91).
+        private const val PERIODIC_SYNC_INTERVAL_MS = 90_000L
+        private const val DEBOUNCE_MS = 2_000L // Debounce full snapshot pulls to avoid rapid-fire
+        // Watch-history events fire every ~10s during playback. Coalesce bursts so we
+        // don't hammer Home's Continue Watching refresh on every tick while a user is
+        // actively watching something on the other device.
+        private const val WATCH_HISTORY_DEBOUNCE_MS = 5_000L
     }
 
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
@@ -44,19 +66,40 @@ class RealtimeSyncManager @Inject constructor(
     private var periodicSyncJob: Job? = null
     private var reconnectJob: Job? = null
     private var pendingPullJob: Job? = null
+    private var pendingWatchHistoryEmitJob: Job? = null
 
     // Track our own pushes to avoid pulling back what we just pushed
     @Volatile
     private var lastPushTimestamp = 0L
 
+    // Track our own watch_history writes (they happen every ~10s during local
+    // playback) so device A doesn't pointlessly refresh its own Continue Watching
+    // row in response to its own updates.
+    @Volatile
+    private var lastLocalWatchHistoryWriteTimestamp = 0L
+
     // User JWT for authenticated Realtime subscriptions
     @Volatile
     private var currentAccessToken: String? = null
 
+    // Event stream for watch_history realtime notifications. HomeViewModel collects
+    // this and triggers refreshContinueWatchingOnly(force = true) on each emission.
+    private val _watchHistoryEvents = MutableSharedFlow<Unit>(extraBufferCapacity = 4)
+    val watchHistoryEvents: SharedFlow<Unit> = _watchHistoryEvents.asSharedFlow()
+
     fun markPush() {
         lastPushTimestamp = System.currentTimeMillis()
     }
 
+    /**
+     * Called by WatchHistoryRepository.saveProgress so incoming watch_history realtime
+     * events from our own write can be ignored (device A shouldn't refresh its own CW
+     * row when it was device A that just wrote the update).
+     */
+    fun markLocalWatchHistoryWrite() {
+        lastLocalWatchHistoryWriteTimestamp = System.currentTimeMillis()
+    }
+
     /**
      * Start listening for realtime changes. Call once after login.
      */
@@ -79,6 +122,7 @@ class RealtimeSyncManager @Inject constructor(
         periodicSyncJob?.cancel()
         reconnectJob?.cancel()
         pendingPullJob?.cancel()
+        pendingWatchHistoryEmitJob?.cancel()
     }
 
     // ── WebSocket Connection ────────────────────────────────────────
@@ -155,9 +199,9 @@ class RealtimeSyncManager @Inject constructor(
     }
 
     private fun joinChannel(ws: WebSocket, userId: String) {
-        // Subscribe to postgres_changes on account_sync_state table filtered by user_id
-        // access_token is required for Supabase RLS to authenticate the subscription
-        val joinMsg = JSONObject().apply {
+        // Channel 1: account_sync_state UPDATEs (full snapshot pulls).
+        // access_token is required for Supabase RLS to authenticate the subscription.
+        val accountSyncJoin = JSONObject().apply {
             put("topic", "realtime:account_sync")
             put("event", "phx_join")
             put("payload", JSONObject().apply {
@@ -175,24 +219,67 @@ class RealtimeSyncManager @Inject constructor(
             })
             put("ref", msgRef.getAndIncrement().toString())
         }
-        ws.send(joinMsg.toString())
-        Log.i(TAG, "Joined channel for user $userId")
+        ws.send(accountSyncJoin.toString())
+
+        // Channel 2: watch_history INSERT + UPDATE events for cross-device Continue
+        // Watching refresh (#91). We subscribe to both INSERT (first watch of a new
+        // item) and UPDATE (position/progress changes) so either event refreshes the
+        // other device's Home row.
+        val watchHistoryJoin = JSONObject().apply {
+            put("topic", "realtime:watch_history")
+            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", "watch_history")
+                            put("filter", "user_id=eq.$userId")
+                        })
+                        put(JSONObject().apply {
+                            put("event", "UPDATE")
+                            put("schema", "public")
+                            put("table", "watch_history")
+                            put("filter", "user_id=eq.$userId")
+                        })
+                    })
+                })
+                currentAccessToken?.let { put("access_token", it) }
+            })
+            put("ref", msgRef.getAndIncrement().toString())
+        }
+        ws.send(watchHistoryJoin.toString())
+
+        Log.i(TAG, "Joined account_sync + watch_history channels for user $userId")
     }
 
     private fun handleMessage(text: String) {
         try {
             val msg = JSONObject(text)
             val event = msg.optString("event", "")
+            val topic = msg.optString("topic", "")
 
             when (event) {
                 "postgres_changes" -> {
-                    // A change was detected on account_sync_state
-                    Log.i(TAG, "Received sync change notification")
-                    debouncedPull()
+                    // Route based on which channel sent the event.
+                    when (topic) {
+                        "realtime:account_sync" -> {
+                            Log.i(TAG, "Received account_sync change")
+                            debouncedPull()
+                        }
+                        "realtime:watch_history" -> {
+                            Log.i(TAG, "Received watch_history change")
+                            debouncedWatchHistoryEmit()
+                        }
+                        else -> {
+                            Log.w(TAG, "postgres_changes on unknown topic: $topic")
+                        }
+                    }
                 }
                 "phx_reply" -> {
                     val status = msg.optJSONObject("payload")?.optString("status")
-                    Log.d(TAG, "Channel reply: $status")
+                    Log.d(TAG, "Channel reply ($topic): $status")
                 }
                 "phx_error" -> {
                     Log.w(TAG, "Channel error: $text")
@@ -201,7 +288,7 @@ class RealtimeSyncManager @Inject constructor(
                     // System messages like subscription confirmation
                     val payload = msg.optJSONObject("payload")
                     if (payload?.optString("status") == "ok") {
-                        Log.i(TAG, "Subscription confirmed")
+                        Log.i(TAG, "Subscription confirmed ($topic)")
                     }
                 }
             }
@@ -226,6 +313,23 @@ class RealtimeSyncManager @Inject constructor(
         }
     }
 
+    private fun debouncedWatchHistoryEmit() {
+        // Skip if the event is almost certainly from our own recent write. This
+        // catches the common case where the user is actively watching on device A
+        // and A's periodic watch_history UPDATE fires back to A as a realtime event.
+        if (System.currentTimeMillis() - lastLocalWatchHistoryWriteTimestamp < 3_000L) {
+            Log.d(TAG, "Skipping watch_history emit - recent local write")
+            return
+        }
+
+        pendingWatchHistoryEmitJob?.cancel()
+        pendingWatchHistoryEmitJob = scope.launch {
+            delay(WATCH_HISTORY_DEBOUNCE_MS)
+            Log.i(TAG, "Emitting watch_history event for Home refresh")
+            _watchHistoryEvents.tryEmit(Unit)
+        }
+    }
+
     // ── Heartbeat ───────────────────────────────────────────────────
 
     private fun startHeartbeat(ws: WebSocket) {

app/src/main/kotlin/com/arflix/tv/data/repository/WatchHistoryRepository.kt

@@ -46,7 +46,8 @@ data class WatchHistoryEntry(
 class WatchHistoryRepository @Inject constructor(
     private val authRepositoryProvider: Provider<AuthRepository>,
     private val supabaseApi: SupabaseApi,
-    private val profileManager: ProfileManager
+    private val profileManager: ProfileManager,
+    private val realtimeSyncManagerProvider: Provider<RealtimeSyncManager>
 ) {
     private fun profileHistorySource(base: String): String {
         // Use stable profile ID so rename/case changes do not split history.
@@ -128,19 +129,29 @@ class WatchHistoryRepository @Inject constructor(
 
         // Upsert - update if exists, insert if not.
         // Retry without stream_* fields if the Supabase schema hasn't been migrated yet.
+        var saved = false
         try {
             executeSupabaseCall("save watch progress") { auth ->
                 supabaseApi.upsertWatchHistory(auth = auth, item = entry.toRecord())
             }
+            saved = true
         } catch (e: HttpException) {
             runCatching {
                 val fallback = entry.copy(stream_key = null, stream_addon_id = null, stream_title = null)
                 executeSupabaseCall("save watch progress fallback") { auth ->
                     supabaseApi.upsertWatchHistory(auth = auth, item = fallback.toRecord())
                 }
+                saved = true
             }
         } catch (_: Exception) {
         }
+
+        // Mark the local write timestamp on the realtime manager so the incoming
+        // watch_history realtime event (which fires back to our own device) doesn't
+        // trigger a redundant Home Continue Watching refresh. See issue #91 fix.
+        if (saved) {
+            runCatching { realtimeSyncManagerProvider.get().markLocalWatchHistoryWrite() }
+        }
     }
 
     /**

app/src/main/kotlin/com/arflix/tv/ui/screens/home/HomeViewModel.kt

@@ -96,6 +96,7 @@ class HomeViewModel @Inject constructor(
     private val watchlistRepository: WatchlistRepository,
     private val cloudSyncRepository: CloudSyncRepository,
     private val launcherContinueWatchingRepository: LauncherContinueWatchingRepository,
+    private val realtimeSyncManager: com.arflix.tv.data.repository.RealtimeSyncManager,
     @ApplicationContext private val context: Context
 ) : ViewModel() {
     private val imageLoader: ImageLoader by lazy(LazyThreadSafetyMode.NONE) {
@@ -651,6 +652,18 @@ class HomeViewModel @Inject constructor(
                 _uiState.value = _uiState.value.copy(trailerAutoPlay = trailerEnabled)
             } catch (_: Exception) {}
         }
+
+        // Subscribe to realtime watch_history events so Continue Watching refreshes on
+        // this device within a few seconds of another device updating the shared
+        // Supabase watch_history table. Before this, mid-playback updates from device A
+        // were only visible on device B after a manual Home ON_RESUME or the 5-minute
+        // periodic sync — so switching devices mid-episode always showed the stale
+        // resume position. Fixes #91.
+        viewModelScope.launch {
+            realtimeSyncManager.watchHistoryEvents.collect {
+                refreshContinueWatchingOnly(force = true)
+            }
+        }
         // Restore logo URL cache from disk for instant clearlogos on cold start
         restoreLogoCacheFromDisk()
         if (logoCache.isNotEmpty()) {