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 15cc8b1a0..f009c9a3e 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 @@ -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(extraBufferCapacity = 4) + val watchHistoryEvents: SharedFlow = _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) { diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/WatchHistoryRepository.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/WatchHistoryRepository.kt index 9f9de6313..0d921b198 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/WatchHistoryRepository.kt +++ b/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, private val supabaseApi: SupabaseApi, - private val profileManager: ProfileManager + private val profileManager: ProfileManager, + private val realtimeSyncManagerProvider: Provider ) { 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() } + } } /** 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 5f85353d1..8800b56e6 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 @@ -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()) {