Optimize API calls and enhance concurrency in TraktRepository

app/src/main/kotlin/com/arflix/tv/data/api/TmdbApi.kt

@@ -230,7 +230,8 @@ data class TmdbTvDetails(
     @SerializedName("number_of_episodes") val numberOfEpisodes: Int = 0,
     @SerializedName("episode_run_time") val episodeRunTime: List<Int> = emptyList(),
     val status: String? = null,
-    val genres: List<TmdbGenre> = emptyList()
+    val genres: List<TmdbGenre> = emptyList(),
+    val seasons: List<TmdbTvSeason> = emptyList()
 )
 
 data class TmdbSeasonDetails(
@@ -272,3 +273,13 @@ data class TmdbReview(val id: String = "", val author: String = "", @SerializedN
 data class TmdbAuthorDetails(val name: String = "", val username: String = "", @SerializedName("avatar_path") val avatarPath: String? = null, val rating: Float? = null)
 data class TmdbFindResponse(@SerializedName("movie_results") val movieResults: List<TmdbFindItem> = emptyList(), @SerializedName("tv_results") val tvResults: List<TmdbFindItem> = emptyList())
 data class TmdbFindItem(val id: Int = 0, val popularity: Float = 0f)
+
+data class TmdbTvSeason(
+    val id: Int = 0,
+    @SerializedName("season_number") val seasonNumber: Int = 1,
+    @SerializedName("episode_count") val episodeCount: Int = 0,
+    val name: String? = null,
+    val overview: String? = null,
+    @SerializedName("poster_path") val posterPath: String? = null,
+    @SerializedName("air_date") val airDate: String? = null
+)

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

@@ -23,6 +23,12 @@ import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.suspendCancellableCoroutine
+import okhttp3.Call
+import okhttp3.Callback
+import okhttp3.Response
+import java.io.IOException
+import kotlin.coroutines.resume
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
@@ -966,9 +972,11 @@ class IptvRepository @Inject constructor(
     }
 
     /**
-     * Lightweight EPG refresh for specific channels using Xtream short EPG API.
-     * Only fetches EPG for the given channel IDs. Updates cachedNowNext in place.
-     * Returns the updated nowNext entries for those channels, or null if not an Xtream provider.
+     * Refreshes short Xtream EPG data for the specified channel IDs.
+     *
+     * Updates the repository's in-memory `cachedNowNext` entries for channels that have Xtream stream identifiers.
+     *
+     * @return A map of channel ID to `IptvNowNext` containing the updated EPG entries for those channels, or `null` if Xtream credentials are not available or no EPG data was retrieved.
      */
     suspend fun refreshEpgForChannels(channelIds: Set<String>): Map<String, IptvNowNext>? {
         if (channelIds.isEmpty()) return null
@@ -998,38 +1006,11 @@ class IptvRepository @Inject constructor(
 
             System.err.println("[EPG-Refresh] Fetching short EPG for ${xtreamChannels.size} favorite channels")
 
-            val allListings = java.util.Collections.synchronizedList(mutableListOf<XtreamEpgListing>())
-            val errorCount = java.util.concurrent.atomic.AtomicInteger(0)
-            // Use a small thread pool — this is just favorites (typically <20 channels)
-            val executor = java.util.concurrent.Executors.newFixedThreadPool(10.coerceAtMost(xtreamChannels.size))
-
-            for (ch in xtreamChannels) {
-                val sid = resolveXtreamStreamId(ch) ?: continue
-                executor.submit {
-                    val url = "${creds.baseUrl}/player_api.php?username=${creds.username}" +
-                        "&password=${creds.password}&action=get_short_epg&stream_id=$sid&limit=12"
-                    try {
-                        var resp: XtreamEpgResponse? = requestJson(url, XtreamEpgResponse::class.java)
-                        var listings = resp?.epgListings
-                        if (listings.isNullOrEmpty()) {
-                            val fallbackUrl = "${creds.baseUrl}/player_api.php?username=${creds.username}" +
-                                "&password=${creds.password}&action=get_short_epg&stream_id=$sid"
-                            resp = requestJson(fallbackUrl, XtreamEpgResponse::class.java)
-                            listings = resp?.epgListings
-                        }
-                        listings?.let { allListings.addAll(it) }
-                    } catch (_: Exception) { errorCount.incrementAndGet() }
-                }
+            val streamIds = xtreamChannels.mapNotNull { resolveXtreamStreamId(it) }
+            var errors = 0
+            val allListings = fetchXtreamEpgListingsAsync(creds, streamIds) { _, hadError ->
+                if (hadError) errors++
             }
-
-            try {
-                executor.shutdown()
-                executor.awaitTermination(20, java.util.concurrent.TimeUnit.SECONDS)
-            } catch (_: Exception) {
-                executor.shutdownNow()
-            }
-
-            val errors = errorCount.get()
             System.err.println("[EPG-Refresh] Done: ${allListings.size} listings, $errors errors")
 
             if (allListings.isEmpty()) return@withContext null
@@ -2926,7 +2907,7 @@ class IptvRepository @Inject constructor(
         }.distinct()
     }
 
-    private fun fetchXtreamLiveChannels(
+    private suspend fun fetchXtreamLiveChannels(
         creds: XtreamCredentials,
         onProgress: (IptvLoadProgress) -> Unit
     ): List<IptvChannel> {
@@ -2969,24 +2950,49 @@ class IptvRepository @Inject constructor(
         }
     }
 
-    private fun <T> requestJson(
+    private suspend fun <T> requestJson(
         url: String,
         type: Type,
         client: OkHttpClient = iptvHttpClient
-    ): T? {
+    ): T? = suspendCancellableCoroutine { continuation ->
         val request = Request.Builder()
             .url(url)
             .header("User-Agent", "VLC/3.0.20 LibVLC/3.0.20")
             .header("Accept", "application/json,*/*")
             .get()
             .build()
-        val response = client.newCall(request).execute()
-        response.use {
-            if (!it.isSuccessful) return null
-            val body = it.body?.string() ?: return null
-            if (body.isBlank()) return null
-            return runCatching { gson.fromJson<T>(body, type) }.getOrNull()
+
+        val call = client.newCall(request)
+
+        continuation.invokeOnCancellation {
+            call.cancel()
         }
+
+        call.enqueue(object : Callback {
+            override fun onFailure(call: Call, e: IOException) {
+                if (continuation.isActive) continuation.resume(null)
+            }
+
+            override fun onResponse(call: Call, response: Response) {
+                if (!continuation.isActive) {
+                    response.close()
+                    return
+                }
+                response.use {
+                    if (!it.isSuccessful) {
+                        continuation.resume(null)
+                        return
+                    }
+                    val body = it.body?.string()
+                    if (body.isNullOrBlank()) {
+                        continuation.resume(null)
+                        return
+                    }
+                    val result = runCatching { gson.fromJson<T>(body, type) }.getOrNull()
+                    continuation.resume(result)
+                }
+            }
+        })
     }
 
     private fun fetchAndParseM3uOnce(
@@ -3157,8 +3163,10 @@ class IptvRepository @Inject constructor(
      * Returns null if the API is not supported or fails (caller should fall back to XMLTV).
      */
     /**
-     * Extract the Xtream stream ID from an IptvChannel.
-     * Uses the explicit field if set, otherwise parses from the "xtream:123" id format.
+     * Determine the Xtream numeric stream identifier for a channel.
+     *
+     * @param ch The channel to inspect; may contain an explicit `xtreamStreamId` or an `id` with the `xtream:{id}` form.
+     * @return The numeric Xtream stream id if present, `null` otherwise.
      */
     private fun resolveXtreamStreamId(ch: IptvChannel): Int? {
         ch.xtreamStreamId?.let { return it }
@@ -3168,7 +3176,15 @@ class IptvRepository @Inject constructor(
         return null
     }
 
-    private fun fetchXtreamShortEpg(
+    /**
+     * Fetches short EPG listings from an Xtream provider and converts them into now/next program snapshots per channel.
+     *
+     * @param creds Xtream credentials used to query the provider's short EPG endpoints.
+     * @param channels The channels to resolve short EPG for; only channels with resolvable Xtream stream IDs are queried.
+     * @param onProgress Callback invoked with load progress updates.
+     * @return A map from IPTV channel ID to its derived IptvNowNext when listings were successfully retrieved and considered reliable, or `null` if no listings were available or the fetch was deemed unreliable (e.g., excessive errors).
+     */
+    private suspend fun fetchXtreamShortEpg(
         creds: XtreamCredentials,
         channels: List<IptvChannel>,
         onProgress: (IptvLoadProgress) -> Unit
@@ -3207,56 +3223,19 @@ class IptvRepository @Inject constructor(
         System.err.println("[EPG] Xtream short EPG: fetching ${toFetch.size}/${xtreamChannels.size} channels")
         if (toFetch.isEmpty()) return null
 
-        // Parallel fetch using a thread pool (20 concurrent connections)
-        val allListings = java.util.Collections.synchronizedList(mutableListOf<XtreamEpgListing>())
-        val errorCount = java.util.concurrent.atomic.AtomicInteger(0)
-        val fetchedCount = java.util.concurrent.atomic.AtomicInteger(0)
+        var errors = 0
+        var fetched = 0
         val total = toFetch.size
-        val executor = java.util.concurrent.Executors.newFixedThreadPool(20)
-        val futures = mutableListOf<java.util.concurrent.Future<*>>()
-
-        val sampleLogged = java.util.concurrent.atomic.AtomicBoolean(false)
-        for (ch in toFetch) {
-            val sid = resolveXtreamStreamId(ch) ?: continue
-            futures.add(executor.submit {
-                val url = "${creds.baseUrl}/player_api.php?username=${creds.username}" +
-                    "&password=${creds.password}&action=get_short_epg&stream_id=$sid&limit=12"
-                try {
-                    var resp: XtreamEpgResponse? = requestJson(url, XtreamEpgResponse::class.java)
-                    var listings = resp?.epgListings
-                    // Fallback: some providers don't support limit param - retry without it
-                    if (listings.isNullOrEmpty()) {
-                        val fallbackUrl = "${creds.baseUrl}/player_api.php?username=${creds.username}" +
-                            "&password=${creds.password}&action=get_short_epg&stream_id=$sid"
-                        resp = requestJson(fallbackUrl, XtreamEpgResponse::class.java)
-                        listings = resp?.epgListings
-                    }
-                    if (listings != null) {
-                        allListings.addAll(listings)
-                        if (listings.isNotEmpty() && sampleLogged.compareAndSet(false, true)) {
-                            val sample = listings.first()
-                            System.err.println("[EPG] Sample response for stream_id=$sid: channelId=${sample.channelId} epgId=${sample.epgId} streamId=${sample.streamId} start=${sample.start} startTs=${sample.startTimestamp} title=${sample.title?.take(40)}")
-                        }
-                    }
-                } catch (_: Exception) { errorCount.incrementAndGet() }
-                val done = fetchedCount.incrementAndGet()
-                if (done % 50 == 0) {
-                    val pct = (90 + ((done.toLong() * 8L) / total.toLong())).toInt().coerceIn(90, 98)
-                    onProgress(IptvLoadProgress("Loading EPG... $done/$total channels", pct))
-                }
-            })
-        }
-
-        // Wait for all to complete (with timeout)
-        try {
-            executor.shutdown()
-            executor.awaitTermination(60, java.util.concurrent.TimeUnit.SECONDS)
-        } catch (_: Exception) {
-            executor.shutdownNow()
+        val streamIds = toFetch.mapNotNull { resolveXtreamStreamId(it) }
+
+        val allListings = fetchXtreamEpgListingsAsync(creds, streamIds) { _, hadError ->
+            fetched++
+            if (hadError) errors++
+            if (fetched % 50 == 0) {
+                val pct = (90 + ((fetched.toLong() * 8L) / total.toLong())).toInt().coerceIn(90, 98)
+                onProgress(IptvLoadProgress("Loading EPG... $fetched/$total channels", pct))
+            }
         }
-
-        val errors = errorCount.get()
-        val fetched = fetchedCount.get()
         System.err.println("[EPG] Xtream short EPG done: ${allListings.size} listings, $fetched fetched, $errors errors")
 
         if (errors > fetched / 2 && fetched > 20) {
@@ -3271,8 +3250,68 @@ class IptvRepository @Inject constructor(
 
 
     /**
-     * Build IptvNowNext map from Xtream EPG listings.
-     * Groups listings by channel, sorts by start time, assigns now/next/later/upcoming.
+     * Fetches short EPG listings for the given Xtream stream IDs in parallel.
+     *
+     * Requests the Xtream `get_short_epg` endpoint for each stream ID (first with a `limit=12`,
+     * then a fallback without `limit` if the first response is empty). Records one sample log
+     * for the first non-empty response observed and invokes `onStreamProcessed` for each stream
+     * to report whether that stream encountered an error.
+     *
+     * @param creds Xtream credentials and base URL used to construct API requests.
+     * @param streamIds The list of Xtream stream IDs to query.
+     * @param onStreamProcessed Callback invoked once per stream ID with `(streamId, hadError)`,
+     *   where `hadError` is `true` if the request sequence for that stream failed.
+     * @return A flattened list of all `XtreamEpgListing` objects returned by the provider
+     *   (empty if no listings were retrieved).
+     */
+    @OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+    private suspend fun fetchXtreamEpgListingsAsync(
+        creds: XtreamCredentials,
+        streamIds: List<Int>,
+        timeoutMillis: Long = 60_000L,
+        onStreamProcessed: (Int, Boolean) -> Unit = { _, _ -> }
+    ): List<XtreamEpgListing> {
+        val result = withTimeoutOrNull(timeoutMillis) {
+            withContext(Dispatchers.IO.limitedParallelism(20)) {
+                val sampleLogged = java.util.concurrent.atomic.AtomicBoolean(false)
+                streamIds.map { sid ->
+                    async {
+                        var hadError = false
+                        val url = "${creds.baseUrl}/player_api.php?username=${creds.username}" +
+                            "&password=${creds.password}&action=get_short_epg&stream_id=$sid&limit=12"
+                        var listings: List<XtreamEpgListing>? = null
+                        try {
+                            var resp: XtreamEpgResponse? = requestJson(url, XtreamEpgResponse::class.java)
+                            listings = resp?.epgListings
+                            if (listings.isNullOrEmpty()) {
+                                val fallbackUrl = "${creds.baseUrl}/player_api.php?username=${creds.username}" +
+                                    "&password=${creds.password}&action=get_short_epg&stream_id=$sid"
+                                resp = requestJson(fallbackUrl, XtreamEpgResponse::class.java)
+                                listings = resp?.epgListings
+                            }
+                            if (!listings.isNullOrEmpty() && sampleLogged.compareAndSet(false, true)) {
+                                val sample = listings.first()
+                                System.err.println("[EPG] Sample response for stream_id=$sid: channelId=${sample.channelId} epgId=${sample.epgId} streamId=${sample.streamId} start=${sample.start} startTs=${sample.startTimestamp} title=${sample.title?.take(40)}")
+                            }
+                        } catch (_: Exception) { hadError = true }
+                        onStreamProcessed(sid, hadError)
+                        listings ?: emptyList()
+                    }
+                }.awaitAll().flatten()
+            }
+        }
+        return result ?: emptyList()
+    }
+
+    /**
+     * Constructs a mapping of IPTV channel IDs to their current and upcoming program windows from a list of Xtream short EPG listings.
+     *
+     * The function groups listings by resolved channel (using `epgIdToChannelIds` and `streamIdToChannelIds`), orders programs by start time, and populates `now`, `next`, `later`, `upcoming`, and `recent` slots for each channel.
+     *
+     * @param listings Xtream short EPG listings to convert into program windows.
+     * @param epgIdToChannelIds Map from EPG identifier to the list of IPTV channel IDs that share that EPG id.
+     * @param streamIdToChannelIds Map from Xtream stream identifier to the list of IPTV channel IDs that correspond to that stream.
+     * @return A map keyed by IPTV channel ID with values of `IptvNowNext`. Each `IptvNowNext` may contain `now`, `next`, `later`, a truncated `upcoming` list (at most 12 items), and a `recent` list of programs that ended within the recent cutoff window.
      */
     private fun buildNowNextFromXtreamListings(
         listings: List<XtreamEpgListing>,