From 86f44ac9cee40a95b105383b33258a1159971c74 Mon Sep 17 00:00:00 2001 From: Barnabas Balogh Date: Mon, 22 Jun 2026 17:26:50 +0000 Subject: [PATCH] refactor(data): deduplicate concurrent data fetches with SingleFlight --- .../purefin/core/concurrency/SingleFlight.kt | 97 +++++++++++++++++++ .../catalog/InMemoryAppContentRepository.kt | 53 +++++----- .../catalog/InMemoryLocalMediaRepository.kt | 69 ++++++------- 3 files changed, 161 insertions(+), 58 deletions(-) create mode 100644 core/src/main/java/hu/bbara/purefin/core/concurrency/SingleFlight.kt diff --git a/core/src/main/java/hu/bbara/purefin/core/concurrency/SingleFlight.kt b/core/src/main/java/hu/bbara/purefin/core/concurrency/SingleFlight.kt new file mode 100644 index 00000000..257b65a5 --- /dev/null +++ b/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> = ConcurrentHashMap() + private val cache: MutableMap> = ConcurrentHashMap() + private val mutex = Mutex() + + suspend fun run(key: String, block: suspend () -> T): T { + val now = System.currentTimeMillis() + + @Suppress("UNCHECKED_CAST") + val cached = cache[key] as? CacheEntry + if (cached != null && now - cached.timestamp < TTL_MS) { + return cached.value + } + + val resolution: Resolution = mutex.withLock { + @Suppress("UNCHECKED_CAST") + val cachedNow = cache[key] as? CacheEntry + if (cachedNow != null && System.currentTimeMillis() - cachedNow.timestamp < TTL_MS) { + return@withLock Resolution.Cached(cachedNow.value) + } + @Suppress("UNCHECKED_CAST") + val existing = inFlight[key] as? Deferred + if (existing != null) { + return@withLock Resolution.Join(existing) + } + val newDeferred = CompletableDeferred() + 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 { + data class Cached(val value: T) : Resolution() + data class Join(val deferred: Deferred) : Resolution() + data class Start(val deferred: CompletableDeferred) : Resolution() + } + + private data class CacheEntry(val value: T, val timestamp: Long) + + companion object { + const val TTL_MS: Long = 15_000L + } +} diff --git a/data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryAppContentRepository.kt b/data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryAppContentRepository.kt index f86f038e..8c006169 100644 --- a/data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryAppContentRepository.kt +++ b/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, 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? { - 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? = + 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 diff --git a/data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryLocalMediaRepository.kt b/data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryLocalMediaRepository.kt index d7d66d0d..c6d52f0a 100644 --- a/data/src/main/java/hu/bbara/purefin/data/catalog/InMemoryLocalMediaRepository.kt +++ b/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 = 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