Read extension stores once at startup

Refreshing stores now fetches them all first and writes them in one transaction.

app/src/main/java/eu/kanade/tachiyomi/extension/util/ExtensionLoader.kt

@@ -15,6 +15,7 @@ import eu.kanade.tachiyomi.util.storage.copyAndSetReadOnlyTo
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.withContext
 import logcat.LogPriority
 import mihon.app.di.appGraph
@@ -115,7 +116,7 @@ internal object ExtensionLoader {
     ): List<Extension.Installed> {
         val trustExtension = context.appGraph.trustExtension
         val sourcePreferences = context.appGraph.sourcePreferences
-        val stores = context.appGraph.extensionStoreRepository.getAll()
+        val stores = context.appGraph.extensionStoreRepository.getAllAsFlow().first()
         val enabledContentWarnings = sourcePreferences.enabledContentWarnings.get()
         val applyContentWarningsToInstalled = sourcePreferences.applyContentWarningsToInstalled.get()
 

data/build.gradle.kts

@@ -32,6 +32,7 @@ dependencies {
     implementation(projects.sourceApi)
     implementation(projects.domain)
     implementation(projects.core.common)
+    implementation(projects.core.metro)
 
     implementation(libs.metro.runtime)
     implementation(libs.kotlinx.serialization.json)

data/src/main/java/mihon/data/extension/repository/ExtensionStoreRepositoryImpl.kt

@@ -7,11 +7,18 @@ import dev.zacsweers.metro.ContributesBinding
 import dev.zacsweers.metro.Inject
 import dev.zacsweers.metro.SingleIn
 import eu.kanade.tachiyomi.extension.model.Extension
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.WhileSubscribed
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.supervisorScope
 import logcat.LogPriority
+import mihon.core.metro.AppCoroutineScope
 import mihon.data.extension.service.ExtensionStoreService
 import mihon.domain.extension.model.ExtensionStore
 import mihon.domain.extension.repository.ExtensionStoreRepository
@@ -19,14 +26,33 @@ import tachiyomi.core.common.util.system.logcat
 import tachiyomi.data.Database
 import tachiyomi.data.subscribeToList
 import tachiyomi.data.subscribeToOne
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
 
 @Inject
 @SingleIn(AppScope::class)
 @ContributesBinding(AppScope::class)
 class ExtensionStoreRepositoryImpl(
+    @AppCoroutineScope scope: CoroutineScope,
     private val service: ExtensionStoreService,
     private val database: Database,
 ) : ExtensionStoreRepository {
+
+    private val stores = database.extension_storeQueries.getAll(::extensionStoreMapper)
+        .subscribeToList()
+        .catch {
+            logcat(LogPriority.ERROR, it) { "Failed to query extension stores" }
+            emit(emptyList())
+        }
+        .shareIn(
+            scope = scope,
+            started = SharingStarted.WhileSubscribed(
+                stopTimeout = 5.seconds,
+                replayExpiration = Duration.ZERO,
+            ),
+            replay = 1,
+        )
+
     override suspend fun insert(indexUrl: String): Result<Unit> {
         return service.fetch(indexUrl).mapCatching { upsert(it) }
     }
@@ -46,24 +72,26 @@ class ExtensionStoreRepositoryImpl(
 
     override suspend fun refreshAll() {
         try {
-            database.extension_storeQueries.getAll().awaitAsList().forEach { store ->
+            val refreshed = stores.first().mapNotNull { store ->
-                service.fetch(store.index_url)
+                service.fetch(store.indexUrl)
-                    .mapCatching {
-                        database.transaction {
-                            if (!database.extension_storeQueries.contains(store.index_url).awaitAsOne()) {
-                                return@transaction
-                            }
-                            upsert(it)
-                            if (store.index_url != it.indexUrl) {
-                                database.extension_storeQueries.delete(store.index_url)
-                            }
-                        }
-                    }
                     .onFailure {
                         logcat(LogPriority.ERROR, it) {
-                            "Failed to refresh extension store '${store.name} (${store.index_url})'"
+                            "Failed to refresh extension store '${store.name} (${store.indexUrl})'"
                         }
                     }
+                    .getOrNull()
+                    ?.let { store.indexUrl to it }
+            }
+            database.transaction {
+                refreshed.forEach { (indexUrl, store) ->
+                    if (!database.extension_storeQueries.contains(indexUrl).awaitAsOne()) {
+                        return@forEach
+                    }
+                    upsert(store)
+                    if (indexUrl != store.indexUrl) {
+                        database.extension_storeQueries.delete(indexUrl)
+                    }
+                }
             }
         } catch (e: Exception) {
             logcat(LogPriority.ERROR, e)
@@ -109,7 +137,7 @@ class ExtensionStoreRepositoryImpl(
     }
 
     override fun getAllAsFlow(): Flow<List<ExtensionStore>> {
-        return database.extension_storeQueries.getAll(::extensionStoreMapper).subscribeToList()
+        return stores
     }
 
     override fun getCountAsFlow(): Flow<Long> {