refactor(data): deduplicate concurrent data fetches with SingleFlight

core/src/main/java/hu/bbara/purefin/core/concurrency/SingleFlight.kt

@@ -0,0 +1,97 @@
+package hu.bbara.purefin.core.concurrency
+
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import java.util.concurrent.ConcurrentHashMap
+import javax.inject.Inject
+import javax.inject.Singleton
+
+/**
+ * Coalesces concurrent and near-time-repeated suspend calls keyed by [key].
+ *
+ * Within a single [run] with a given [key]:
+ *  - If a call with the same [key] is already in flight, the new caller
+ *    awaits the same [Deferred] and receives the first caller's result.
+ *  - If a call with the same [key] completed successfully less than
+ *    [TTL_MS] ago, the cached result is returned without re-executing
+ *    [block].
+ *  - Otherwise [block] is executed, its result is cached for [TTL_MS]
+ *    and shared with any concurrent awaiters.
+ *
+ * Failures are never cached: a thrown exception (other than
+ * [CancellationException]) is propagated to all awaiters of the
+ * in-flight call, but a subsequent call with the same key will execute
+ *    [block] again immediately.
+ *
+ * [CancellationException] is always re-thrown so coroutine cancellation
+ * semantics are preserved.
+ */
+@Singleton
+class SingleFlight @Inject constructor() {
+    private val inFlight: MutableMap<String, Deferred<*>> = ConcurrentHashMap()
+    private val cache: MutableMap<String, CacheEntry<*>> = ConcurrentHashMap()
+    private val mutex = Mutex()
+
+    suspend fun <T> run(key: String, block: suspend () -> T): T {
+        val now = System.currentTimeMillis()
+
+        @Suppress("UNCHECKED_CAST")
+        val cached = cache[key] as? CacheEntry<T>
+        if (cached != null && now - cached.timestamp < TTL_MS) {
+            return cached.value
+        }
+
+        val resolution: Resolution<T> = mutex.withLock {
+            @Suppress("UNCHECKED_CAST")
+            val cachedNow = cache[key] as? CacheEntry<T>
+            if (cachedNow != null && System.currentTimeMillis() - cachedNow.timestamp < TTL_MS) {
+                return@withLock Resolution.Cached(cachedNow.value)
+            }
+            @Suppress("UNCHECKED_CAST")
+            val existing = inFlight[key] as? Deferred<T>
+            if (existing != null) {
+                return@withLock Resolution.Join(existing)
+            }
+            val newDeferred = CompletableDeferred<T>()
+            inFlight[key] = newDeferred
+            Resolution.Start(newDeferred)
+        }
+
+        return when (resolution) {
+            is Resolution.Cached -> resolution.value
+            is Resolution.Join -> resolution.deferred.await()
+            is Resolution.Start -> {
+                val deferred = resolution.deferred
+                try {
+                    val result = block()
+                    cache[key] = CacheEntry(result, System.currentTimeMillis())
+                    deferred.complete(result)
+                    result
+                } catch (error: CancellationException) {
+                    deferred.completeExceptionally(error)
+                    throw error
+                } catch (error: Throwable) {
+                    deferred.completeExceptionally(error)
+                    throw error
+                } finally {
+                    inFlight.remove(key, deferred)
+                }
+            }
+        }
+    }
+
+    private sealed class Resolution<T> {
+        data class Cached<T>(val value: T) : Resolution<T>()
+        data class Join<T>(val deferred: Deferred<T>) : Resolution<T>()
+        data class Start<T>(val deferred: CompletableDeferred<T>) : Resolution<T>()
+    }
+
+    private data class CacheEntry<T>(val value: T, val timestamp: Long)
+
+    companion object {
+        const val TTL_MS: Long = 15_000L
+    }
+}

data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryAppContentRepository.kt

@@ -1,6 +1,7 @@
 package hu.bbara.purefin.data.catalog
 
 import androidx.datastore.core.DataStore
+import hu.bbara.purefin.core.concurrency.SingleFlight
 import hu.bbara.purefin.core.data.HomeRepository
 import hu.bbara.purefin.core.data.NetworkMonitor
 import hu.bbara.purefin.core.data.UserSessionRepository
@@ -53,6 +54,7 @@ class InMemoryAppContentRepository @Inject constructor(
     private val homeCacheDataStore: DataStore<HomeCache>,
     private val onlineMediaRepository: InMemoryLocalMediaRepository,
     private val networkMonitor: NetworkMonitor,
+    private val singleFlight: SingleFlight,
 ) : HomeRepository {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
     private var cacheLoadJob: Job? = null
@@ -239,7 +241,7 @@ class InMemoryAppContentRepository @Inject constructor(
         )
     }
 
-    private suspend fun loadLibraries() {
+    private suspend fun loadLibraries() = singleFlight.run("AppContent:loadLibraries") {
         val librariesItem = runCatching { jellyfinApiClient.getLibraries() }
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load libraries")
@@ -292,20 +294,21 @@ class InMemoryAppContentRepository @Inject constructor(
         librariesState.value = filledLibraries
     }
 
-    override suspend fun loadLibraryContent(libraryId: UUID): List<MediaUiModel>? {
-        val library = librariesState.value.find { it.id == libraryId } ?: return emptyList()
-        // ETag hit — return null so the caller can keep its previously
-        // fetched copy. The library detail viewmodel uses this signal to
-        // preserve its in-memory cache across re-selections.
-        val items = jellyfinApiClient.getLibraryContent(libraryId) ?: return null
-        val url = serverUrl()
-        return when (library.type) {
-            LibraryKind.MOVIES -> items.map { MovieUiModel(it.toMovie(url)) }
-            LibraryKind.SERIES -> items.map { SeriesUiModel(it.toSeries(url)) }
+    override suspend fun loadLibraryContent(libraryId: UUID): List<MediaUiModel>? =
+        singleFlight.run("AppContent:loadLibraryContent:$libraryId") {
+            val library = librariesState.value.find { it.id == libraryId } ?: return@run emptyList()
+            // ETag hit — return null so the caller can keep its previously
+            // fetched copy. The library detail viewmodel uses this signal to
+            // preserve its in-memory cache across re-selections.
+            val items = jellyfinApiClient.getLibraryContent(libraryId) ?: return@run null
+            val url = serverUrl()
+            when (library.type) {
+                LibraryKind.MOVIES -> items.map { MovieUiModel(it.toMovie(url)) }
+                LibraryKind.SERIES -> items.map { SeriesUiModel(it.toSeries(url)) }
+            }
         }
-    }
 
-    private suspend fun loadSuggestions() {
+    private suspend fun loadSuggestions() = singleFlight.run("AppContent:loadSuggestions") {
         val url = serverUrl()
         // Head-first fetch: ask the server for just the first 2 items and
         // compare to the cached head. If the head is unchanged, the rest
@@ -316,15 +319,15 @@ class InMemoryAppContentRepository @Inject constructor(
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load suggestions head")
             }
-        if (headItems == null) return
-        if (headMatches(headItems, suggestionsState.value)) return
+        if (headItems == null) return@run
+        if (headMatches(headItems, suggestionsState.value)) return@run
 
         // Head changed (or first refresh) — fetch the full list.
         val suggestionsItems = runCatching { jellyfinApiClient.getSuggestions() }
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load suggestions")
             }
-        if (suggestionsItems == null) return
+        if (suggestionsItems == null) return@run
 
         suggestionsState.value = suggestionsItems.mapNotNull { item ->
             when (item.type) {
@@ -344,20 +347,20 @@ class InMemoryAppContentRepository @Inject constructor(
         }
     }
 
-    private suspend fun loadContinueWatching() {
+    private suspend fun loadContinueWatching() = singleFlight.run("AppContent:loadContinueWatching") {
         val url = serverUrl()
         val headItems = runCatching { jellyfinApiClient.getContinueWatchingHead() }
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load continue watching head")
             }
-        if (headItems == null) return
-        if (headMatches(headItems, continueWatchingState.value)) return
+        if (headItems == null) return@run
+        if (headMatches(headItems, continueWatchingState.value)) return@run
 
         val continueWatchingItems = runCatching { jellyfinApiClient.getContinueWatching() }
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load continue watching")
             }
-        if (continueWatchingItems == null) return
+        if (continueWatchingItems == null) return@run
 
         continueWatchingState.value = continueWatchingItems.mapNotNull { item ->
             when (item.type) {
@@ -376,20 +379,20 @@ class InMemoryAppContentRepository @Inject constructor(
         }
     }
 
-    private suspend fun loadNextUp() {
+    private suspend fun loadNextUp() = singleFlight.run("AppContent:loadNextUp") {
         val url = serverUrl()
         val headItems = runCatching { jellyfinApiClient.getNextUpHead() }
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load next up head")
             }
-        if (headItems == null) return
-        if (headMatches(headItems, nextUpState.value)) return
+        if (headItems == null) return@run
+        if (headMatches(headItems, nextUpState.value)) return@run
 
         val nextUpItems = runCatching { jellyfinApiClient.getNextUpEpisodes() }
             .getOrElse { error ->
                 handleRefreshFailure(error, "Unable to load next up")
             }
-        if (nextUpItems == null) return
+        if (nextUpItems == null) return@run
 
         nextUpState.value = nextUpItems.map { item ->
             Media.EpisodeMedia(episodeId = item.id, seriesId = item.seriesId!!)
@@ -400,7 +403,7 @@ class InMemoryAppContentRepository @Inject constructor(
         }
     }
 
-    private suspend fun loadLatestLibraryContent() {
+    private suspend fun loadLatestLibraryContent() = singleFlight.run("AppContent:loadLatestLibraryContent") {
         // Reuse the libraries already loaded by loadLibraries() so we don't refetch
         // the same /Users/{userId}/Views response a second time per refresh.
         val filteredLibraries = librariesState.value

data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryLocalMediaRepository.kt

@@ -2,6 +2,7 @@ package hu.bbara.purefin.data.catalog
 
 import hu.bbara.purefin.core.data.LocalMediaRepository
 import hu.bbara.purefin.core.data.UserSessionRepository
+import hu.bbara.purefin.core.concurrency.SingleFlight
 import hu.bbara.purefin.data.converter.toEpisode
 import hu.bbara.purefin.data.converter.toMovie
 import hu.bbara.purefin.data.converter.toSeason
@@ -32,6 +33,7 @@ import javax.inject.Singleton
 class InMemoryLocalMediaRepository @Inject constructor(
     private val userSessionRepository: UserSessionRepository,
     private val jellyfinApiClient: JellyfinApiClient,
+    private val singleFlight: SingleFlight,
 ) : LocalMediaRepository {
 
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
@@ -56,13 +58,13 @@ class InMemoryLocalMediaRepository @Inject constructor(
     override suspend fun getEpisode(id: UUID): Flow<Episode?> =
         episodesState.map { it[id] }.distinctUntilChanged()
 
-    override suspend fun loadMovie(id: UUID) {
-        if (moviesState.value.containsKey(id)) return
+    override suspend fun loadMovie(id: UUID) = singleFlight.run("LocalMedia:loadMovie:$id") {
+        if (moviesState.value.containsKey(id)) return@run
         try {
             jellyfinApiClient.getItemInfo(id)?.let { item ->
                 if (item.type != BaseItemKind.MOVIE) {
                     Timber.tag(TAG).d("Item is not an movie: ${item.type}")
-                    return
+                    return@run
                 }
                 val movie = item.toMovie(serverUrl.first())
                 moviesState.update { current -> current + (movie.id to movie) }
@@ -75,13 +77,13 @@ class InMemoryLocalMediaRepository @Inject constructor(
         }
     }
 
-    override suspend fun loadSeries(id: UUID) {
-        if (seriesState.value.containsKey(id)) return
+    override suspend fun loadSeries(id: UUID) = singleFlight.run("LocalMedia:loadSeries:$id") {
+        if (seriesState.value.containsKey(id)) return@run
         try {
             jellyfinApiClient.getItemInfo(id)?.let { item ->
                 if (item.type != BaseItemKind.SERIES) {
                     Timber.tag(TAG).d("Item is not an series: ${item.type}")
-                    return
+                    return@run
                 }
                 val series = item.toSeries(serverUrl.first())
                 seriesState.update { current -> current + (series.id to series) }
@@ -94,13 +96,13 @@ class InMemoryLocalMediaRepository @Inject constructor(
         }
     }
 
-    override suspend fun loadEpisode(id: UUID) {
-        if (episodesState.value.containsKey(id)) return
+    override suspend fun loadEpisode(id: UUID) = singleFlight.run("LocalMedia:loadEpisode:$id") {
+        if (episodesState.value.containsKey(id)) return@run
         try {
             jellyfinApiClient.getItemInfo(id)?.let { item ->
                 if (item.type != BaseItemKind.EPISODE) {
                     Timber.tag(TAG).d("Item is not an episode: ${item.type}")
-                    return
+                    return@run
                 }
                 val episode = item.toEpisode(serverUrl.first())
                 episodesState.update { current -> current + (episode.id to episode) }
@@ -125,7 +127,7 @@ class InMemoryLocalMediaRepository @Inject constructor(
         episodesState.update { current -> current + episodes.associateBy { it.id } }
     }
 
-    override suspend fun loadSeasons(seriesId: UUID) {
+    override suspend fun loadSeasons(seriesId: UUID) = singleFlight.run("LocalMedia:loadSeasons:$seriesId") {
         try {
             loadSeasonsInternal(seriesId)
         } catch (error: CancellationException) {
@@ -147,33 +149,34 @@ class InMemoryLocalMediaRepository @Inject constructor(
         seriesState.update { it + (updatedSeries.id to updatedSeries) }
     }
 
-    override suspend fun loadSeasonEpisodes(seriesId: UUID, seasonId: UUID) {
-        loadSeasons(seriesId)
+    override suspend fun loadSeasonEpisodes(seriesId: UUID, seasonId: UUID) =
+        singleFlight.run("LocalMedia:loadSeasonEpisodes:$seriesId:$seasonId") {
+            loadSeasons(seriesId)
 
-        val series = seriesState.value[seriesId] ?: throw RuntimeException("Series not found")
-        val season = series.seasons.firstOrNull { it.id == seasonId } ?: return
-        if (season.episodes.isNotEmpty() || season.episodeCount == 0) {
-            return
-        }
+            val series = seriesState.value[seriesId] ?: throw RuntimeException("Series not found")
+            val season = series.seasons.firstOrNull { it.id == seasonId } ?: return@run
+            if (season.episodes.isNotEmpty() || season.episodeCount == 0) {
+                return@run
+            }
 
-        val serverUrl = userSessionRepository.serverUrl.first()
-        val episodes = jellyfinApiClient.getEpisodesInSeason(seriesId, seasonId)
-            .map { it.toEpisode(serverUrl) }
-        seriesState.update { current ->
-            val currentSeries = current[seriesId] ?: return@update current
-            val updatedSeries = currentSeries.copy(
-                seasons = currentSeries.seasons.map {
-                    if (it.id == seasonId && it.episodes.isEmpty()) {
-                        it.copy(episodes = episodes)
-                    } else {
-                        it
+            val serverUrl = userSessionRepository.serverUrl.first()
+            val episodes = jellyfinApiClient.getEpisodesInSeason(seriesId, seasonId)
+                .map { it.toEpisode(serverUrl) }
+            seriesState.update { current ->
+                val currentSeries = current[seriesId] ?: return@update current
+                val updatedSeries = currentSeries.copy(
+                    seasons = currentSeries.seasons.map {
+                        if (it.id == seasonId && it.episodes.isEmpty()) {
+                            it.copy(episodes = episodes)
+                        } else {
+                            it
+                        }
                     }
-                }
-            )
-            current + (updatedSeries.id to updatedSeries)
+                )
+                current + (updatedSeries.id to updatedSeries)
+            }
+            episodesState.update { current -> current + episodes.associateBy { it.id } }
         }
-        episodesState.update { current -> current + episodes.associateBy { it.id } }
-    }
 
     override suspend fun updateWatchProgress(mediaId: UUID, positionMs: Long, durationMs: Long) {
         if (durationMs <= 0) return