Implement Downloads GQL and fix url changes requiring restarts.

2025-12-10 06:42:05 +01:00 · 2025-10-14 14:19:47 -04:00
parent bef31a01eb
commit 5b7228c687
41 changed files with 688 additions and 229 deletions
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/SharedDomainComponent.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/SharedDomainComponent.kt
@@ -118,11 +118,6 @@ interface SharedDomainComponent : CoreComponent {
    val libraryUpdateServiceFactory: LibraryUpdateService
        get() = LibraryUpdateService(serverPreferences, http)

-    @get:AppScope
-    @get:Provides
-    val downloadServiceFactory: DownloadService
-        get() = DownloadService(serverPreferences, http)
-
    @get:AppScope
    @get:Provides
    val serverListenersFactory: ServerListeners
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/base/WebsocketService.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/base/WebsocketService.kt
@@ -45,8 +45,8 @@ abstract class WebsocketService(
    fun init() {
        errorConnectionCount = 0
        job?.cancel()
-        job = serverUrl
-            .mapLatest { serverUrl ->
+        job = client
+            .mapLatest { client ->
                status.value = Status.STARTING
                while (true) {
                    if (errorConnectionCount > 3) {
@@ -54,6 +54,7 @@ abstract class WebsocketService(
                        throw CancellationException("Finish")
                    }
                    runCatching {
+                        val serverUrl = serverUrl.value
                        client.ws(
                            host = serverUrl.host,
                            port = serverUrl.port,
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadChapter.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadChapter.kt
@@ -6,19 +6,24 @@

 package ca.gosyer.jui.domain.download.model

-import androidx.compose.runtime.Immutable
-import ca.gosyer.jui.domain.chapter.model.Chapter
-import ca.gosyer.jui.domain.manga.model.Manga
-import kotlinx.serialization.Serializable
-
-@Serializable
-@Immutable
-data class DownloadChapter(
-    val chapterIndex: Int,
-    val mangaId: Long,
-    val chapter: Chapter,
-    val manga: Manga,
-    val state: DownloadState = DownloadState.Queued,
-    val progress: Float = 0f,
-    val tries: Int = 0,
+data class DownloadQueueItem(
+    val position: Int,
+    val progress: Float,
+    val state: DownloadState,
+    val tries: Int,
+    val chapter: DownloadChapter,
+    val manga: DownloadManga
+)
+
+data class DownloadChapter(
+    val id: Long,
+    val name: String,
+    val pageCount: Int,
+)
+
+data class DownloadManga(
+    val id: Long,
+    val title: String,
+    val thumbnailUrl: String?,
+    val thumbnailUrlLastFetched: Long = 0,
 )
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadState.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadState.kt
@@ -6,16 +6,9 @@

 package ca.gosyer.jui.domain.download.model

-import androidx.compose.runtime.Stable
-import kotlinx.serialization.Serializable
-
-@Serializable
-@Stable
-enum class DownloadState(
-    val state: Int,
-) {
-    Queued(0),
-    Downloading(1),
-    Finished(2),
-    Error(3),
+enum class DownloadState {
+    QUEUED,
+    DOWNLOADING,
+    FINISHED,
+    ERROR
 }
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadStatus.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadStatus.kt
@@ -6,12 +6,7 @@

 package ca.gosyer.jui.domain.download.model

-import androidx.compose.runtime.Immutable
-import kotlinx.serialization.Serializable
-
-@Serializable
-@Immutable
 data class DownloadStatus(
-    val status: DownloaderStatus,
-    val queue: List<DownloadChapter>,
+    val status: DownloaderState,
+    val queue: List<DownloadQueueItem>,
 )
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadUpdate.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadUpdate.kt
@@ -0,0 +1,17 @@
+package ca.gosyer.jui.domain.download.model
+
+data class DownloadUpdate(
+    val type: DownloadUpdateType? = null,
+    val download: DownloadQueueItem? = null
+)
+
+enum class DownloadUpdateType {
+    QUEUED,
+    DEQUEUED,
+    PAUSED,
+    STOPPED,
+    PROGRESS,
+    FINISHED,
+    ERROR,
+    POSITION
+}
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadUpdates.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloadUpdates.kt
@@ -0,0 +1,8 @@
+package ca.gosyer.jui.domain.download.model
+
+data class DownloadUpdates(
+    val initial: List<DownloadQueueItem>? = null,
+    val omittedUpdates: Boolean,
+    val state: DownloaderState,
+    val updates: List<DownloadUpdate>? = null
+)
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloaderStatus.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/model/DownloaderStatus.kt
@@ -6,12 +6,7 @@

 package ca.gosyer.jui.domain.download.model

-import androidx.compose.runtime.Stable
-import kotlinx.serialization.Serializable
-
-@Serializable
-@Stable
-enum class DownloaderStatus {
-    Started,
-    Stopped,
+enum class DownloaderState {
+    STARTED,
+    STOPPED
 }
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/service/DownloadRepository.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/service/DownloadRepository.kt
@@ -6,6 +6,8 @@

 package ca.gosyer.jui.domain.download.service

+import ca.gosyer.jui.domain.download.model.DownloadStatus
+import ca.gosyer.jui.domain.download.model.DownloadUpdates
 import kotlinx.coroutines.flow.Flow

 interface DownloadRepository {
@@ -25,4 +27,8 @@ interface DownloadRepository {
    ): Flow<Unit>

    fun batchDownload(chapterIds: List<Long>): Flow<Unit>
+
+    fun downloadSubscription(): Flow<DownloadUpdates>
+
+    fun downloadStatus(): Flow<DownloadStatus>
 }
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/service/DownloadService.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/download/service/DownloadService.kt
@@ -7,49 +7,144 @@
 package ca.gosyer.jui.domain.download.service

 import ca.gosyer.jui.domain.base.WebsocketService
-import ca.gosyer.jui.domain.download.model.DownloadChapter
-import ca.gosyer.jui.domain.download.model.DownloadStatus
-import ca.gosyer.jui.domain.download.model.DownloaderStatus
-import ca.gosyer.jui.domain.server.Http
-import ca.gosyer.jui.domain.server.service.ServerPreferences
-import io.ktor.websocket.Frame
-import io.ktor.websocket.readText
+import ca.gosyer.jui.domain.download.model.DownloadQueueItem
+import ca.gosyer.jui.domain.download.model.DownloadUpdateType
+import ca.gosyer.jui.domain.download.model.DownloaderState
+import com.diamondedge.logging.logging
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.flow.update
 import me.tatarka.inject.annotations.Inject

@Inject
 class DownloadService(
-    serverPreferences: ServerPreferences,
-    client: Http,
-) : WebsocketService(serverPreferences, client) {
-    override val status: MutableStateFlow<Status>
-        get() = DownloadService.status
+    private val downloadRepository: DownloadRepository,
+) {
+    private val log = logging()

-    override val query: String
-        get() = "/api/v1/downloads"
+    fun getSubscription(): Flow<Unit> {
+        return downloadRepository.downloadSubscription()
+            .onStart {
+                log.info { "Starting download status subscription" }
+                status.value = WebsocketService.Status.STARTING
+            }
+            .catch { error ->
+                log.error(error) { "Error in download status subscription" }
+                status.value = WebsocketService.Status.STOPPED
+            }
+            .map { updates ->
+                status.value = WebsocketService.Status.RUNNING
+                if (updates.omittedUpdates) {
+                    log.info { "Omitted updates detected, fetching fresh download status" }
+                    fetchDownloadStatus()
+                    return@map
+                }
+                if (updates.initial != null) {
+                    downloadQueue.value = updates.initial
+                }
+                downloaderStatus.value = updates.state
+                updates.updates?.forEach { update ->
+                    when (update.type) {
+                        DownloadUpdateType.QUEUED -> {
+                            update.download?.let { download ->
+                                downloadQueue.update {
+                                    it.toMutableList().apply {
+                                        add(download.position.coerceAtMost(it.size), download)
+                                    }
+                                }
+                            }
+                        }
+                        DownloadUpdateType.DEQUEUED -> {
+                            downloadQueue.update {
+                                it.filter { it.chapter.id != update.download?.chapter?.id }
+                            }
+                        }
+                        DownloadUpdateType.PAUSED -> {
+                            downloaderStatus.value = DownloaderState.STOPPED
+                        }
+                        DownloadUpdateType.STOPPED -> {
+                            downloaderStatus.value = DownloaderState.STOPPED
+                        }
+                        DownloadUpdateType.ERROR -> {
+                            update.download?.let { download ->
+                                downloadQueue.update {
+                                    it.map { chapter ->
+                                        if (chapter.chapter.id == download.chapter.id) {
+                                            chapter.copy(state = download.state)
+                                        } else {
+                                            chapter
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                        DownloadUpdateType.PROGRESS -> {
+                            update.download?.let { download ->
+                                downloadQueue.update {
+                                    it.map { chapter ->
+                                        if (chapter.chapter.id == download.chapter.id) {
+                                            chapter.copy(progress = download.progress)
+                                        } else {
+                                            chapter
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                        DownloadUpdateType.FINISHED -> {
+                            downloadQueue.update {
+                                it.filter { it.chapter.id != update.download?.chapter?.id }
+                            }
+                        }
+                        DownloadUpdateType.POSITION -> {
+                            update.download?.let { download ->
+                                downloadQueue.update {
+                                    val index = it.indexOfFirst { it.chapter.id == download.chapter.id }
+                                    if (index != -1) {
+                                        it.toMutableList().apply {
+                                            removeAt(index)
+                                            add(download.position.coerceAtMost(it.size), download)
+                                        }.toList()
+                                    } else it
+                                }

-    override suspend fun onReceived(frame: Frame.Text) {
-        val status = json.decodeFromString<DownloadStatus>(frame.readText())
-        downloaderStatus.value = status.status
-        downloadQueue.value = status.queue
+                            }
+                        }
+                        null -> {
+                            // todo Handle null case
+                        }
+                    }
+                }
+            }
+    }
+
+    private suspend fun fetchDownloadStatus() {
+        val status = downloadRepository.downloadStatus().firstOrNull()
+        if (status != null) {
+            downloadQueue.value = status.queue
+            downloaderStatus.value = status.status
+        }
    }

    companion object {
-        val status = MutableStateFlow(Status.STARTING)
-        val downloadQueue = MutableStateFlow(emptyList<DownloadChapter>())
-        val downloaderStatus = MutableStateFlow(DownloaderStatus.Stopped)
+        val status = MutableStateFlow(WebsocketService.Status.STARTING)
+        val downloadQueue = MutableStateFlow(emptyList<DownloadQueueItem>())
+        val downloaderStatus = MutableStateFlow(DownloaderState.STOPPED)

        fun registerWatch(mangaId: Long) =
            downloadQueue
                .map {
-                    it.filter { it.mangaId == mangaId }
+                    it.filter { it.manga.id == mangaId }
                }

        fun registerWatches(mangaIds: Set<Long>) =
            downloadQueue
                .map {
-                    it.filter { it.mangaId in mangaIds }
+                    it.filter { it.manga.id in mangaIds }
                }
    }
 }
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/server/HttpClient.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/server/HttpClient.kt
@@ -29,49 +29,64 @@ import io.ktor.client.plugins.logging.Logging
 import io.ktor.client.plugins.websocket.WebSockets
 import io.ktor.http.URLBuilder
 import io.ktor.http.URLProtocol
+import io.ktor.http.Url
 import io.ktor.serialization.kotlinx.json.json
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.stateIn
 import kotlinx.serialization.json.Json
 import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
 import io.ktor.client.plugins.auth.Auth as AuthPlugin

-typealias Http = HttpClient
+typealias Http = StateFlow<HttpClient>

 expect val Engine: HttpClientEngineFactory<HttpClientEngineConfig>

 expect fun HttpClientConfig<HttpClientEngineConfig>.configurePlatform()

-fun httpClient(
-    serverPreferences: ServerPreferences,
-    json: Json,
-): Http =
-    HttpClient(Engine) {
+private fun getHttpClient(
+    serverUrl: Url,
+    proxy: Proxy,
+    proxyHttpHost: String,
+    proxyHttpPort: Int,
+    proxySocksHost: String,
+    proxySocksPort: Int,
+    auth: Auth,
+    authUsername: String,
+    authPassword: String,
+    json: Json
+): HttpClient {
+    return HttpClient(Engine) {
        configurePlatform()

        expectSuccess = true

        defaultRequest {
-            url(serverPreferences.serverUrl().get().toString())
+            url(serverUrl.toString())
        }

        engine {
-            proxy = when (serverPreferences.proxy().get()) {
+            this.proxy = when (proxy) {
                Proxy.NO_PROXY -> null

                Proxy.HTTP_PROXY -> ProxyBuilder.http(
                    URLBuilder(
-                        host = serverPreferences.proxyHttpHost().get(),
-                        port = serverPreferences.proxyHttpPort().get(),
+                        host = proxyHttpHost,
+                        port = proxyHttpPort,
                    ).build(),
                )

                Proxy.SOCKS_PROXY -> ProxyBuilder.socks(
-                    serverPreferences.proxySocksHost().get(),
-                    serverPreferences.proxySocksPort().get(),
+                    proxySocksHost,
+                    proxySocksPort,
                )
            }
        }
-        when (serverPreferences.auth().get()) {
+        when (auth) {
            Auth.NONE -> Unit

            Auth.BASIC -> AuthPlugin {
@@ -81,8 +96,8 @@ fun httpClient(
                    }
                    credentials {
                        BasicAuthCredentials(
-                            serverPreferences.authUsername().get(),
-                            serverPreferences.authPassword().get(),
+                            authUsername,
+                            authPassword,
                        )
                    }
                }
@@ -92,8 +107,8 @@ fun httpClient(
                digest {
                    credentials {
                        DigestAuthCredentials(
-                            serverPreferences.authUsername().get(),
-                            serverPreferences.authPassword().get(),
+                            authUsername,
+                            authPassword,
                        )
                    }
                }
@@ -123,3 +138,48 @@ fun httpClient(
            }
        }
    }
+}
+
+@OptIn(DelicateCoroutinesApi::class)
+fun httpClient(
+    serverPreferences: ServerPreferences,
+    json: Json,
+): Http = combine(
+    serverPreferences.serverUrl().stateIn(GlobalScope),
+    serverPreferences.proxy().stateIn(GlobalScope),
+    serverPreferences.proxyHttpHost().stateIn(GlobalScope),
+    serverPreferences.proxyHttpPort().stateIn(GlobalScope),
+    serverPreferences.proxySocksHost().stateIn(GlobalScope),
+    serverPreferences.proxySocksPort().stateIn(GlobalScope),
+    serverPreferences.auth().stateIn(GlobalScope),
+    serverPreferences.authUsername().stateIn(GlobalScope),
+    serverPreferences.authPassword().stateIn(GlobalScope),
+) {
+    getHttpClient(
+        it[0] as Url,
+        it[1] as Proxy,
+        it[2] as String,
+        it[3] as Int,
+        it[4] as String,
+        it[5] as Int,
+        it[6] as Auth,
+        it[7] as String,
+        it[8] as String,
+        json,
+    )
+}.stateIn(
+    GlobalScope,
+    SharingStarted.Eagerly,
+    getHttpClient(
+        serverPreferences.serverUrl().get(),
+        serverPreferences.proxy().get(),
+        serverPreferences.proxyHttpHost().get(),
+        serverPreferences.proxyHttpPort().get(),
+        serverPreferences.proxySocksHost().get(),
+        serverPreferences.proxySocksPort().get(),
+        serverPreferences.auth().get(),
+        serverPreferences.authUsername().get(),
+        serverPreferences.authPassword().get(),
+        json,
+    )
+)
--- a/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/updates/interactor/UpdateChecker.kt
+++ b/domain/src/commonMain/kotlin/ca/gosyer/jui/domain/updates/interactor/UpdateChecker.kt
@@ -39,7 +39,7 @@ class UpdateChecker(
    fun asFlow(manualFetch: Boolean) =
        flow {
            if (!manualFetch && !updatePreferences.enabled().get()) return@flow
-            val latestRelease = client.get(
+            val latestRelease = client.value.get(
                "https://api.github.com/repos/$GITHUB_REPO/releases/latest",
            ).body<GithubRelease>()