fix: contain Trakt watched rate limits (NUVIO-MOBILE-Q1)

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watched/WatchedRepository.kt

@@ -101,6 +101,18 @@ internal fun replaceWatchedItemsForSource(
     target.putAll(replacement)
 }
 
+internal suspend fun <T> watchedProviderRefreshOrNull(
+    refresh: suspend () -> T,
+    onFailure: (Throwable) -> Unit,
+): T? = try {
+    refresh()
+} catch (error: CancellationException) {
+    throw error
+} catch (error: Throwable) {
+    onFailure(error)
+    null
+}
+
 object WatchedRepository {
     private data class WatchedRefreshOperation(
         val profileId: Int,
@@ -1054,16 +1066,21 @@ object WatchedRepository {
                 .collectLatest { extraKeys ->
                     val keysChanged = providerExtraWatchedKeys[providerId] != extraKeys
                     if (keysChanged) {
-                        providerExtraWatchedKeys[providerId] = extraKeys
-                        // Re-pull items from provider to reflect snapshot changes
-                        // (e.g. after remote episode removal)
-                        val freshItems = adapter.pull(
-                            profileId = currentProfileId,
-                            pageSize = watchedItemsPageSize,
-                        )
+                        val freshItems = watchedProviderRefreshOrNull(
+                            refresh = {
+                                adapter.pull(
+                                    profileId = currentProfileId,
+                                    pageSize = watchedItemsPageSize,
+                                )
+                            },
+                            onFailure = { error ->
+                                log.w(error) { "Failed to refresh watched items from ${providerId.storageId}" }
+                            },
+                        ) ?: return@collectLatest
                         val itemsByKey = freshItems.associateBy { item ->
                             watchedItemKey(item.type, item.id, item.season, item.episode)
                         }.toMutableMap()
+                        providerExtraWatchedKeys[providerId] = extraKeys
                         providerItemsByKey[providerId] = itemsByKey
                         publish()
                     }

composeApp/src/commonMain/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapter.kt

@@ -14,6 +14,7 @@ import com.nuvio.app.features.watched.normalizeWatchedMarkedAtEpochMs
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.encodeToString
@@ -24,6 +25,8 @@ private const val WATCHED_PAGE_LIMIT = 250
 private const val WATCHED_MAX_PAGES = 1_000
 private const val WATCHED_SHOWS_EXTENDED = "progress"
 internal const val TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES = 8 * 1024 * 1024
+private const val TRAKT_WATCHED_MAX_ATTEMPTS = 2
+private const val TRAKT_WATCHED_MAX_RETRY_DELAY_MS = 60_000L
 
 internal fun interface TraktWatchedHttpEngine {
     suspend fun get(
@@ -35,15 +38,48 @@ internal fun interface TraktWatchedHttpEngine {
 
 internal class TraktWatchedPageClient(
     private val engine: TraktWatchedHttpEngine,
+    private val sleep: suspend (Long) -> Unit = { delayMs -> delay(delayMs) },
 ) {
-    suspend fun get(url: String, headers: Map<String, String>): RawHttpResponse =
-        engine.get(
-            url = url,
-            headers = headers,
-            maxResponseBodyBytes = TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES,
-        )
+    suspend fun get(url: String, headers: Map<String, String>): RawHttpResponse {
+        repeat(TRAKT_WATCHED_MAX_ATTEMPTS) { attempt ->
+            val response = engine.get(
+                url = url,
+                headers = headers,
+                maxResponseBodyBytes = TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES,
+            )
+            if (response.status in 200..299) return response
+            if (!isTransientTraktWatchedStatus(response.status) || attempt == TRAKT_WATCHED_MAX_ATTEMPTS - 1) {
+                throw TraktWatchedHttpException(response.status)
+            }
+            sleep(
+                traktWatchedRetryDelayMs(
+                    attempt = attempt,
+                    retryAfterSeconds = response.headers.entries
+                        .firstOrNull { (name, _) -> name.equals("retry-after", ignoreCase = true) }
+                        ?.value,
+                ),
+            )
+        }
+        error("Trakt watched request exhausted without a response")
+    }
 }
 
+internal class TraktWatchedHttpException(
+    val status: Int,
+) : Exception("Trakt watched request failed: $status")
+
+internal fun isTransientTraktWatchedStatus(status: Int): Boolean =
+    status == 429 || status in 500..599
+
+internal fun traktWatchedRetryDelayMs(attempt: Int, retryAfterSeconds: String?): Long =
+    retryAfterSeconds
+        ?.trim()
+        ?.toLongOrNull()
+        ?.coerceAtLeast(0L)
+        ?.times(1_000L)
+        ?.coerceAtMost(TRAKT_WATCHED_MAX_RETRY_DELAY_MS)
+        ?: (1_000L shl attempt.coerceIn(0, 5)).coerceAtMost(TRAKT_WATCHED_MAX_RETRY_DELAY_MS)
+
 private val platformTraktWatchedHttpEngine = TraktWatchedHttpEngine { url, headers, maxResponseBodyBytes ->
     httpRequestRaw(
         method = "GET",
@@ -153,9 +189,6 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider {
                 url = "$BASE_URL/sync/watched/movies?page=$page&limit=$WATCHED_PAGE_LIMIT",
                 headers = headers,
             )
-            if (response.status !in 200..299) {
-                error("Trakt watched movies request failed: ${response.status}")
-            }
             val pageItems = json.decodeFromString<List<TraktWatchedMovieDto>>(response.body)
             if (pageItems.isEmpty()) break
             items.addAll(pageItems)
@@ -177,9 +210,6 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider {
                 url = "$BASE_URL/sync/watched/shows?page=$page&limit=$WATCHED_PAGE_LIMIT&extended=$WATCHED_SHOWS_EXTENDED",
                 headers = headers,
             )
-            if (response.status !in 200..299) {
-                error("Trakt watched shows request failed: ${response.status}")
-            }
             val pageItems = json.decodeFromString<List<TraktWatchedShowDto>>(response.body)
             if (pageItems.isEmpty()) break
             items.addAll(pageItems)

composeApp/src/commonTest/kotlin/com/nuvio/app/features/watched/WatchedRepositoryTest.kt

@@ -4,12 +4,41 @@ import com.nuvio.app.features.details.MetaDetails
 import com.nuvio.app.features.details.MetaVideo
 import com.nuvio.app.features.tracking.TrackingProviderId
 import com.nuvio.app.features.tracking.WatchProgressSource
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.runBlocking
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
+import kotlin.test.assertNull
 import kotlin.test.assertTrue
 
 class WatchedRepositoryTest {
+    @Test
+    fun providerRefreshFailure_isContainedWithoutReplacingState() = runBlocking {
+        val failure = IllegalStateException("rate limited")
+        var observedFailure: Throwable? = null
+
+        val result = watchedProviderRefreshOrNull(
+            refresh = { throw failure },
+            onFailure = { observedFailure = it },
+        )
+
+        assertNull(result)
+        assertEquals(failure, observedFailure)
+    }
+
+    @Test
+    fun providerRefreshCancellation_isNotContained() = runBlocking {
+        assertFailsWith<CancellationException> {
+            watchedProviderRefreshOrNull(
+                refresh = { throw CancellationException("cancelled") },
+                onFailure = {},
+            )
+        }
+        Unit
+    }
+
     @Test
     fun watchedItemKey_isTypeAware() {
         assertEquals("movie:tt1:-1:-1", watchedItemKey(type = "movie", id = "tt1"))

composeApp/src/commonTest/kotlin/com/nuvio/app/features/watching/sync/TraktWatchedSyncAdapterTest.kt

@@ -5,6 +5,7 @@ import com.nuvio.app.features.addons.RawHttpResponse
 import kotlinx.coroutines.runBlocking
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
 
 class TraktWatchedSyncAdapterTest {
@@ -35,4 +36,56 @@ class TraktWatchedSyncAdapterTest {
         assertFalse(response.body.length < body.length)
         assertEquals(body, response.body)
     }
+
+    @Test
+    fun `rate limited watched request retries once after retry after`() = runBlocking {
+        var attempts = 0
+        val delays = mutableListOf<Long>()
+        val client = TraktWatchedPageClient(
+            engine = TraktWatchedHttpEngine { _, _, _ ->
+                attempts += 1
+                response(
+                    status = if (attempts == 1) 429 else 200,
+                    headers = if (attempts == 1) mapOf("Retry-After" to "2") else emptyMap(),
+                )
+            },
+            sleep = delays::add,
+        )
+
+        val result = client.get("https://api.trakt.tv/sync/watched/shows", emptyMap())
+
+        assertEquals(200, result.status)
+        assertEquals(2, attempts)
+        assertEquals(listOf(2_000L), delays)
+    }
+
+    @Test
+    fun `repeated rate limit becomes a bounded typed failure`() = runBlocking {
+        var attempts = 0
+        val client = TraktWatchedPageClient(
+            engine = TraktWatchedHttpEngine { _, _, _ ->
+                attempts += 1
+                response(status = 429)
+            },
+            sleep = {},
+        )
+
+        val error = assertFailsWith<TraktWatchedHttpException> {
+            client.get("https://api.trakt.tv/sync/watched/shows", emptyMap())
+        }
+
+        assertEquals(429, error.status)
+        assertEquals(2, attempts)
+    }
+
+    private fun response(
+        status: Int,
+        headers: Map<String, String> = emptyMap(),
+    ): RawHttpResponse = RawHttpResponse(
+        status = status,
+        statusText = "",
+        url = "https://api.trakt.tv/sync/watched/shows",
+        body = "[]",
+        headers = headers,
+    )
 }