Fix Downloaded quick filter (close #1076, close #1079)

app/src/main/kotlin/org/koitharu/kotatsu/core/util/ext/Flow.kt

@@ -9,6 +9,7 @@ import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onCompletion
@@ -129,3 +130,5 @@ fun <T1, T2, T3, T4, T5, T6, R> combine(
 suspend fun <T : Any> Flow<T?>.firstNotNull(): T = checkNotNull(first { x -> x != null })
 
 suspend fun <T : Any> Flow<T?>.firstNotNullOrNull(): T? = firstOrNull { x -> x != null }
+
+fun <T> Flow<Flow<T>>.flattenLatest() = flatMapLatest { it }

app/src/main/kotlin/org/koitharu/kotatsu/explore/data/MangaSourcesRepository.kt

@@ -27,6 +27,7 @@ import org.koitharu.kotatsu.core.parser.external.ExternalMangaSource
 import org.koitharu.kotatsu.core.prefs.AppSettings
 import org.koitharu.kotatsu.core.prefs.observeAsFlow
 import org.koitharu.kotatsu.core.ui.util.ReversibleHandle
+import org.koitharu.kotatsu.core.util.ext.flattenLatest
 import org.koitharu.kotatsu.parsers.model.ContentType
 import org.koitharu.kotatsu.parsers.model.MangaParserSource
 import org.koitharu.kotatsu.parsers.model.MangaSource
@@ -168,7 +169,7 @@ class MangaSourcesRepository @Inject constructor(
 		dao.observeEnabled(order).map {
 			it.toSources(skipNsfw, order)
 		}
-	}.flatMapLatest { it }
+	}.flattenLatest()
 		.onStart { assimilateNewSources() }
 		.combine(observeExternalSources()) { enabled, external ->
 			val list = ArrayList<MangaSourceInfo>(enabled.size + external.size)

app/src/main/kotlin/org/koitharu/kotatsu/favourites/domain/FavouritesRepository.kt

@@ -27,6 +27,7 @@ import javax.inject.Inject
 @Reusable
 class FavouritesRepository @Inject constructor(
 	private val db: MangaDatabase,
+	private val localObserver: LocalFavoritesObserver,
 ) {
 
 	suspend fun getAllManga(): List<Manga> {
@@ -40,6 +41,9 @@ class FavouritesRepository @Inject constructor(
 	}
 
 	fun observeAll(order: ListSortOrder, filterOptions: Set<ListFilterOption>, limit: Int): Flow<List<Manga>> {
+		if (ListFilterOption.Downloaded in filterOptions) {
+			return localObserver.observeAll(order, filterOptions - ListFilterOption.Downloaded, limit)
+		}
 		return db.getFavouritesDao().observeAll(order, filterOptions, limit)
 			.mapItems { it.toManga() }
 	}
@@ -55,6 +59,9 @@ class FavouritesRepository @Inject constructor(
 		filterOptions: Set<ListFilterOption>,
 		limit: Int
 	): Flow<List<Manga>> {
+		if (ListFilterOption.Downloaded in filterOptions) {
+			return localObserver.observeAll(categoryId, order, filterOptions - ListFilterOption.Downloaded, limit)
+		}
 		return db.getFavouritesDao().observeAll(categoryId, order, filterOptions, limit)
 			.mapItems { it.toManga() }
 	}

app/src/main/kotlin/org/koitharu/kotatsu/favourites/domain/LocalFavoritesObserver.kt

@@ -0,0 +1,42 @@
+package org.koitharu.kotatsu.favourites.domain
+
+import dagger.Reusable
+import kotlinx.coroutines.flow.Flow
+import org.koitharu.kotatsu.core.db.MangaDatabase
+import org.koitharu.kotatsu.core.db.entity.toManga
+import org.koitharu.kotatsu.core.db.entity.toMangaTags
+import org.koitharu.kotatsu.favourites.data.FavouriteManga
+import org.koitharu.kotatsu.list.domain.ListFilterOption
+import org.koitharu.kotatsu.list.domain.ListSortOrder
+import org.koitharu.kotatsu.local.data.LocalMangaRepository
+import org.koitharu.kotatsu.local.domain.LocalObserveMapper
+import org.koitharu.kotatsu.parsers.model.Manga
+import javax.inject.Inject
+
+@Reusable
+class LocalFavoritesObserver @Inject constructor(
+	localMangaRepository: LocalMangaRepository,
+	private val db: MangaDatabase,
+) : LocalObserveMapper<FavouriteManga, Manga>(localMangaRepository, limitStep = 10) {
+
+	fun observeAll(
+		order: ListSortOrder,
+		filterOptions: Set<ListFilterOption>,
+		limit: Int
+	): Flow<List<Manga>> = observe(limit) { newLimit ->
+		db.getFavouritesDao().observeAll(order, filterOptions, newLimit)
+	}
+
+	fun observeAll(
+		categoryId: Long,
+		order: ListSortOrder,
+		filterOptions: Set<ListFilterOption>,
+		limit: Int
+	): Flow<List<Manga>> = observe(limit) { newLimit ->
+		db.getFavouritesDao().observeAll(categoryId, order, filterOptions, newLimit)
+	}
+
+	override fun toManga(e: FavouriteManga) = e.manga.toManga(e.tags.toMangaTags())
+
+	override fun toResult(e: FavouriteManga, manga: Manga) = manga
+}

app/src/main/kotlin/org/koitharu/kotatsu/favourites/ui/list/FavouritesListViewModel.kt

@@ -4,26 +4,22 @@ import androidx.lifecycle.SavedStateHandle
 import androidx.lifecycle.viewModelScope
 import dagger.hilt.android.lifecycle.HiltViewModel
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.filterNotNull
-import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.plus
 import org.koitharu.kotatsu.R
-import org.koitharu.kotatsu.core.model.isLocal
 import org.koitharu.kotatsu.core.prefs.AppSettings
 import org.koitharu.kotatsu.core.prefs.ListMode
 import org.koitharu.kotatsu.core.prefs.observeAsFlow
 import org.koitharu.kotatsu.core.ui.util.ReversibleAction
 import org.koitharu.kotatsu.core.util.ext.call
+import org.koitharu.kotatsu.core.util.ext.flattenLatest
 import org.koitharu.kotatsu.download.ui.worker.DownloadWorker
 import org.koitharu.kotatsu.favourites.domain.FavoritesListQuickFilter
 import org.koitharu.kotatsu.favourites.domain.FavouritesRepository
@@ -39,12 +35,11 @@ import org.koitharu.kotatsu.list.ui.model.EmptyState
 import org.koitharu.kotatsu.list.ui.model.ListModel
 import org.koitharu.kotatsu.list.ui.model.LoadingState
 import org.koitharu.kotatsu.list.ui.model.toErrorState
-import org.koitharu.kotatsu.local.data.LocalMangaRepository
 import org.koitharu.kotatsu.parsers.model.Manga
 import java.util.concurrent.atomic.AtomicBoolean
 import javax.inject.Inject
 
-private const val PAGE_SIZE = 20
+private const val PAGE_SIZE = 16
 
 @HiltViewModel
 class FavouritesListViewModel @Inject constructor(
@@ -53,7 +48,6 @@ class FavouritesListViewModel @Inject constructor(
 	private val mangaListMapper: MangaListMapper,
 	private val markAsReadUseCase: MarkAsReadUseCase,
 	private val quickFilter: FavoritesListQuickFilter,
-	private val localMangaRepository: LocalMangaRepository,
 	settings: AppSettings,
 	downloadScheduler: DownloadWorker.Scheduler,
 ) : MangaListViewModel(settings, downloadScheduler), QuickFilterListener by quickFilter {
@@ -130,30 +124,32 @@ class FavouritesListViewModel @Inject constructor(
 	}
 
 	private suspend fun List<Manga>.mapList(mode: ListMode, filters: Set<ListFilterOption>): List<ListModel> {
-		val list = if (ListFilterOption.Downloaded in filters) {
-			mapToLocal()
-		} else {
-			this
-		}
-		if (list.isEmpty()) {
+		if (isEmpty()) {
 			return if (filters.isEmpty()) {
 				listOf(getEmptyState(hasFilters = false))
 			} else {
 				listOfNotNull(quickFilter.filterItem(filters), getEmptyState(hasFilters = true))
 			}
 		}
-		val result = ArrayList<ListModel>(list.size + 1)
+		val result = ArrayList<ListModel>(size + 1)
 		quickFilter.filterItem(filters)?.let(result::add)
-		mangaListMapper.toListModelList(result, list, mode)
+		mangaListMapper.toListModelList(result, this, mode)
 		return result
 	}
 
 	private fun observeFavorites() = if (categoryId == NO_ID) {
-		combine(sortOrder.filterNotNull(), quickFilter.appliedOptions.combineWithSettings(), limit, ::Triple)
-			.flatMapLatest { repository.observeAll(it.first, it.second - ListFilterOption.Downloaded, it.third) }
+		combine(
+			sortOrder.filterNotNull(),
+			quickFilter.appliedOptions.combineWithSettings(),
+			limit,
+		) { order, filters, limit ->
+			isReady.set(false)
+			repository.observeAll(order, filters, limit)
+		}.flattenLatest()
 	} else {
-		combine(quickFilter.appliedOptions, limit, ::Pair)
-			.flatMapLatest { repository.observeAll(categoryId, it.first - ListFilterOption.Downloaded, it.second) }
+		combine(quickFilter.appliedOptions.combineWithSettings(), limit) { filters, limit ->
+			repository.observeAll(categoryId, filters, limit)
+		}.flattenLatest()
 	}
 
 	private fun getEmptyState(hasFilters: Boolean) = if (hasFilters) {
@@ -175,16 +171,4 @@ class FavouritesListViewModel @Inject constructor(
 			actionStringRes = 0,
 		)
 	}
-
-	private suspend fun List<Manga>.mapToLocal(): List<Manga> = coroutineScope {
-		map {
-			async {
-				if (it.isLocal) {
-					it
-				} else {
-					localMangaRepository.findSavedManga(it)?.manga
-				}
-			}
-		}.awaitAll().filterNotNull()
-	}
 }

app/src/main/kotlin/org/koitharu/kotatsu/history/data/HistoryLocalObserver.kt

@@ -0,0 +1,35 @@
+package org.koitharu.kotatsu.history.data
+
+import dagger.Reusable
+import org.koitharu.kotatsu.core.db.MangaDatabase
+import org.koitharu.kotatsu.core.db.entity.toManga
+import org.koitharu.kotatsu.core.db.entity.toMangaTags
+import org.koitharu.kotatsu.history.domain.model.MangaWithHistory
+import org.koitharu.kotatsu.list.domain.ListFilterOption
+import org.koitharu.kotatsu.list.domain.ListSortOrder
+import org.koitharu.kotatsu.local.data.LocalMangaRepository
+import org.koitharu.kotatsu.local.domain.LocalObserveMapper
+import org.koitharu.kotatsu.parsers.model.Manga
+import javax.inject.Inject
+
+@Reusable
+class HistoryLocalObserver @Inject constructor(
+	localMangaRepository: LocalMangaRepository,
+	private val db: MangaDatabase,
+) : LocalObserveMapper<HistoryWithManga, MangaWithHistory>(localMangaRepository, limitStep = 10) {
+
+	fun observeAll(
+		order: ListSortOrder,
+		filterOptions: Set<ListFilterOption>,
+		limit: Int
+	) = observe(limit) { newLimit ->
+		db.getHistoryDao().observeAll(order, filterOptions, newLimit)
+	}
+
+	override fun toManga(e: HistoryWithManga) = e.manga.toManga(e.tags.toMangaTags())
+
+	override fun toResult(e: HistoryWithManga, manga: Manga) = MangaWithHistory(
+		manga = manga,
+		history = e.history.toMangaHistory(),
+	)
+}

app/src/main/kotlin/org/koitharu/kotatsu/history/data/HistoryRepository.kt

@@ -39,6 +39,7 @@ class HistoryRepository @Inject constructor(
 	private val settings: AppSettings,
 	private val scrobblers: Set<@JvmSuppressWildcards Scrobbler>,
 	private val mangaRepository: MangaDataRepository,
+	private val localObserver: HistoryLocalObserver,
 	private val newChaptersUseCaseProvider: Provider<CheckNewChaptersUseCase>,
 ) {
 
@@ -80,6 +81,9 @@ class HistoryRepository @Inject constructor(
 		filterOptions: Set<ListFilterOption>,
 		limit: Int
 	): Flow<List<MangaWithHistory>> {
+		if (ListFilterOption.Downloaded in filterOptions) {
+			return localObserver.observeAll(order, filterOptions - ListFilterOption.Downloaded, limit)
+		}
 		return db.getHistoryDao().observeAll(order, filterOptions, limit).mapItems {
 			MangaWithHistory(
 				it.manga.toManga(it.tags.toMangaTags()),

app/src/main/kotlin/org/koitharu/kotatsu/history/ui/HistoryListViewModel.kt

@@ -3,21 +3,16 @@ package org.koitharu.kotatsu.history.ui
 import androidx.lifecycle.viewModelScope
 import dagger.hilt.android.lifecycle.HiltViewModel
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.plus
 import org.koitharu.kotatsu.R
 import org.koitharu.kotatsu.core.model.MangaHistory
-import org.koitharu.kotatsu.core.model.isLocal
 import org.koitharu.kotatsu.core.prefs.AppSettings
 import org.koitharu.kotatsu.core.prefs.ListMode
 import org.koitharu.kotatsu.core.prefs.observeAsFlow
@@ -25,6 +20,7 @@ import org.koitharu.kotatsu.core.prefs.observeAsStateFlow
 import org.koitharu.kotatsu.core.ui.util.ReversibleAction
 import org.koitharu.kotatsu.core.util.ext.calculateTimeAgo
 import org.koitharu.kotatsu.core.util.ext.call
+import org.koitharu.kotatsu.core.util.ext.flattenLatest
 import org.koitharu.kotatsu.core.util.ext.onFirst
 import org.koitharu.kotatsu.download.ui.worker.DownloadWorker
 import org.koitharu.kotatsu.history.data.HistoryRepository
@@ -42,20 +38,18 @@ import org.koitharu.kotatsu.list.ui.model.ListHeader
 import org.koitharu.kotatsu.list.ui.model.ListModel
 import org.koitharu.kotatsu.list.ui.model.LoadingState
 import org.koitharu.kotatsu.list.ui.model.toErrorState
-import org.koitharu.kotatsu.local.data.LocalMangaRepository
 import org.koitharu.kotatsu.parsers.model.Manga
 import java.time.Instant
 import java.util.concurrent.atomic.AtomicBoolean
 import javax.inject.Inject
 
-private const val PAGE_SIZE = 20
+private const val PAGE_SIZE = 16
 
 @HiltViewModel
 class HistoryListViewModel @Inject constructor(
 	private val repository: HistoryRepository,
 	settings: AppSettings,
 	private val mangaListMapper: MangaListMapper,
-	private val localMangaRepository: LocalMangaRepository,
 	private val markAsReadUseCase: MarkAsReadUseCase,
 	private val quickFilter: HistoryListQuickFilter,
 	downloadScheduler: DownloadWorker.Scheduler,
@@ -144,21 +138,22 @@ class HistoryListViewModel @Inject constructor(
 		}
 	}
 
-	private fun observeHistory() = combine(sortOrder, quickFilter.appliedOptions.combineWithSettings(), limit, ::Triple)
-		.flatMapLatest { repository.observeAllWithHistory(it.first, it.second - ListFilterOption.Downloaded, it.third) }
+	private fun observeHistory() = combine(
+		sortOrder,
+		quickFilter.appliedOptions.combineWithSettings(),
+		limit,
+	) { order, filters, limit ->
+		isReady.set(false)
+		repository.observeAllWithHistory(order, filters, limit)
+	}.flattenLatest()
 
 	private suspend fun mapList(
-		historyList: List<MangaWithHistory>,
+		list: List<MangaWithHistory>,
 		grouped: Boolean,
 		mode: ListMode,
 		filters: Set<ListFilterOption>,
 		isIncognito: Boolean,
 	): List<ListModel> {
-		val list = if (ListFilterOption.Downloaded in filters) {
-			historyList.mapToLocal()
-		} else {
-			historyList
-		}
 		if (list.isEmpty()) {
 			return if (filters.isEmpty()) {
 				listOf(getEmptyState(hasFilters = false))
@@ -198,20 +193,6 @@ class HistoryListViewModel @Inject constructor(
 		return result
 	}
 
-	private suspend fun List<MangaWithHistory>.mapToLocal() = coroutineScope {
-		map {
-			async {
-				if (it.manga.isLocal) {
-					it
-				} else {
-					localMangaRepository.findSavedManga(it.manga)?.let { localManga ->
-						MangaWithHistory(localManga.manga, it.history)
-					}
-				}
-			}
-		}.awaitAll().filterNotNull()
-	}
-
 	private fun MangaHistory.header(order: ListSortOrder): ListHeader? = when (order) {
 		ListSortOrder.LAST_READ,
 		ListSortOrder.LONG_AGO_READ -> ListHeader(calculateTimeAgo(updatedAt))

app/src/main/kotlin/org/koitharu/kotatsu/local/domain/LocalObserveMapper.kt

@@ -0,0 +1,55 @@
+package org.koitharu.kotatsu.local.domain
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.transformLatest
+import org.koitharu.kotatsu.core.model.isLocal
+import org.koitharu.kotatsu.local.data.LocalMangaRepository
+import org.koitharu.kotatsu.parsers.model.Manga
+
+abstract class LocalObserveMapper<E, R>(
+	private val localMangaRepository: LocalMangaRepository,
+	private val limitStep: Int,
+) {
+
+	protected fun observe(limit: Int, observer: (limit: Int) -> Flow<List<E>>): Flow<List<R>> {
+		val floatingLimit = MutableStateFlow(limit)
+		return floatingLimit.flatMapLatest { l ->
+			observer(l)
+				.transformLatest { fullList ->
+					val mapped = fullList.mapToLocal()
+					if (mapped.size < limit && fullList.size == l) {
+						floatingLimit.value += limitStep
+					} else {
+						emit(mapped.take(limit))
+					}
+				}.distinctUntilChanged()
+
+		}
+	}
+
+	private suspend fun List<E>.mapToLocal(): List<R> = coroutineScope {
+		val dispatcher = Dispatchers.IO.limitedParallelism(6)
+		map {
+			async(dispatcher) {
+				val m = toManga(it)
+				val mapped = if (m.isLocal) {
+					m
+				} else {
+					localMangaRepository.findSavedManga(m)?.manga
+				}
+				mapped?.let { mm -> toResult(it, mm) }
+			}
+		}.awaitAll().filterNotNull()
+	}
+
+	protected abstract fun toManga(e: E): Manga
+
+	protected abstract fun toResult(e: E, manga: Manga): R
+}