Feature/graphql download queue subscription send only updates (#1011)

* Emit only download changes instead of full status

The download subscription emitted the full download status, which, depending on how big the queue was, took forever because the graphql subscription does not support data loader batching, causing it to run into the n+1 problem

* Rename "DownloadManager#status" to "DownloadManager#updates"

* Add initial queue to download subscription type

Adds the current queue at the time of sending the initial message.
This field is null for all following messages after the initial one

* Optionally limit and omit download updates

To prevent the n+1 dataloader issue, the max number of updates included in the download subscription can be limited.
This way, the problem will be circumvented and instead, the latest download status should be (re-)fetched via the download status query, which does not run into this problem.

* Formatting

server/src/main/kotlin/suwayomi/tachidesk/manga/impl/download/DownloadManager.kt

@@ -21,7 +21,6 @@ import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.sample
@@ -36,6 +35,9 @@ import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Downloading
 import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Error
 import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Queued
 import suwayomi.tachidesk.manga.impl.download.model.DownloadStatus
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdate
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdateType
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdates
 import suwayomi.tachidesk.manga.impl.download.model.Status
 import suwayomi.tachidesk.manga.model.dataclass.ChapterDataClass
 import suwayomi.tachidesk.manga.model.dataclass.MangaDataClass
@@ -47,6 +49,7 @@ import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.CopyOnWriteArraySet
 import kotlin.reflect.jvm.jvmName
 import kotlin.time.Duration.Companion.seconds
 
@@ -57,6 +60,7 @@ object DownloadManager {
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
     private val clients = ConcurrentHashMap<String, WsContext>()
     private val downloadQueue = CopyOnWriteArrayList<DownloadChapter>()
+    private val downloadUpdates = CopyOnWriteArraySet<DownloadUpdate>()
     private val downloaders = ConcurrentHashMap<String, Downloader>()
 
     private const val DOWNLOAD_QUEUE_KEY = "downloadQueueKey"
@@ -81,6 +85,13 @@ object DownloadManager {
         scope.launch { saveQueueFlow.emit(Unit) }
     }
 
+    private fun handleDownloadUpdate(
+        immediate: Boolean,
+        download: DownloadUpdate? = null,
+    ) {
+        notifyAllClients(immediate, listOfNotNull(download))
+    }
+
     fun restoreAndResumeDownloads() {
         scope.launch {
             logger.debug { "restoreAndResumeDownloads: Restore download queue..." }
@@ -124,9 +135,15 @@ object DownloadManager {
 
     private val notifyFlow = MutableSharedFlow<Unit>(extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
 
+    @Deprecated("Replaced with updatesFlow", replaceWith = ReplaceWith("updatesFlow"))
     private val statusFlow = MutableSharedFlow<DownloadStatus>()
+
+    @Deprecated("Replaced with updates", replaceWith = ReplaceWith("updates"))
     val status = statusFlow.onStart { emit(getStatus()) }
 
+    private val updatesFlow = MutableSharedFlow<DownloadUpdates>()
+    val updates = updatesFlow.onStart { emit(getDownloadUpdates(addInitial = true)) }
+
     init {
         scope.launch {
             notifyFlow.sample(1.seconds).collect {
@@ -147,12 +164,21 @@ object DownloadManager {
         }
     }
 
-    private fun notifyAllClients(immediate: Boolean = false) {
+    private fun notifyAllClients(
+        immediate: Boolean = false,
+        downloads: List<DownloadUpdate> = emptyList(),
+    ) {
+        downloadUpdates.addAll(downloads)
+
         if (immediate) {
             val status = getStatus()
+            val updates = getDownloadUpdates()
+
+            downloadUpdates.clear()
 
             scope.launch {
                 statusFlow.emit(status)
+                updatesFlow.emit(updates)
                 sendStatusToAllClients(status)
             }
 
@@ -164,7 +190,7 @@ object DownloadManager {
         }
     }
 
-    private fun getStatus(): DownloadStatus =
+    fun getStatus(): DownloadStatus =
         DownloadStatus(
             if (downloadQueue.none { it.state == Downloading }) {
                 Status.Stopped
@@ -174,6 +200,17 @@ object DownloadManager {
             downloadQueue.toList(),
         )
 
+    private fun getDownloadUpdates(addInitial: Boolean = false): DownloadUpdates =
+        DownloadUpdates(
+            if (downloadQueue.none { it.state == Downloading }) {
+                Status.Stopped
+            } else {
+                Status.Started
+            },
+            downloadUpdates.toList(),
+            if (addInitial) downloadQueue.toList() else null,
+        )
+
     private val downloaderWatch = MutableSharedFlow<Unit>(extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
 
     init {
@@ -234,7 +271,7 @@ object DownloadManager {
                 scope = scope,
                 sourceId = sourceId,
                 downloadQueue = downloadQueue,
-                notifier = ::notifyAllClients,
+                notifier = ::handleDownloadUpdate,
                 onComplete = ::refreshDownloaders,
                 onDownloadFinished = ::triggerSaveDownloadQueue,
             )
@@ -303,7 +340,7 @@ object DownloadManager {
         val addedChapters = inputs.mapNotNull { addToQueue(it.first, it.second) }
         if (addedChapters.isNotEmpty()) {
             start()
-            notifyAllClients(true)
+            notifyAllClients(false, addedChapters.map { DownloadUpdate(DownloadUpdateType.QUEUED, it) })
         }
         scope.launch {
             downloaderWatch.emit(Unit)
@@ -328,6 +365,7 @@ object DownloadManager {
                     manga.id,
                     chapter,
                     manga,
+                    downloadQueue.size,
                 )
             downloadQueue.add(newDownloadChapter)
             triggerSaveDownloadQueue()
@@ -374,7 +412,7 @@ object DownloadManager {
         downloadQueue.removeAll(chapterDownloads)
         triggerSaveDownloadQueue()
 
-        notifyAllClients()
+        notifyAllClients(false, chapterDownloads.toList().map { DownloadUpdate(DownloadUpdateType.DEQUEUED, it) })
     }
 
     fun reorder(
@@ -410,6 +448,8 @@ object DownloadManager {
 
         downloadQueue -= download
         downloadQueue.add(to, download)
+        download.position = to
+        notifyAllClients(false, listOf(DownloadUpdate(DownloadUpdateType.POSITION, download)))
         triggerSaveDownloadQueue()
     }
 
@@ -439,9 +479,10 @@ object DownloadManager {
         logger.debug { "clear" }
 
         stop()
+        val removedDownloads = downloadQueue.toList().map { DownloadUpdate(DownloadUpdateType.DEQUEUED, it) }
         downloadQueue.clear()
         triggerSaveDownloadQueue()
-        notifyAllClients()
+        notifyAllClients(false, removedDownloads)
     }
 }
 

server/src/main/kotlin/suwayomi/tachidesk/manga/impl/download/Downloader.kt

@@ -27,6 +27,10 @@ import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Downloading
 import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Error
 import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Finished
 import suwayomi.tachidesk.manga.impl.download.model.DownloadState.Queued
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdate
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdateType.ERROR
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdateType.FINISHED
+import suwayomi.tachidesk.manga.impl.download.model.DownloadUpdateType.PROGRESS
 import suwayomi.tachidesk.manga.model.table.ChapterTable
 import java.util.concurrent.CopyOnWriteArrayList
 
@@ -34,7 +38,7 @@ class Downloader(
     private val scope: CoroutineScope,
     val sourceId: String,
     private val downloadQueue: CopyOnWriteArrayList<DownloadChapter>,
-    private val notifier: (immediate: Boolean) -> Unit,
+    private val notifier: (immediate: Boolean, download: DownloadUpdate?) -> Unit,
     private val onComplete: () -> Unit,
     private val onDownloadFinished: () -> Unit,
 ) {
@@ -49,10 +53,11 @@ class Downloader(
     class PauseDownloadException : Exception("Pause download")
 
     private suspend fun step(
-        download: DownloadChapter?,
+        downloadUpdate: DownloadUpdate?,
         immediate: Boolean,
     ) {
-        notifier(immediate)
+        val download = downloadUpdate?.downloadChapter
+        notifier(immediate, downloadUpdate)
         currentCoroutineContext().ensureActive()
         if (download != null && download != availableSourceDownloads.firstOrNull { it.state != Error }) {
             if (download in downloadQueue) {
@@ -81,7 +86,7 @@ class Downloader(
                         }
                     }
             logger.debug { "started" }
-            notifier(false)
+            notifier(false, null)
         }
     }
 
@@ -90,14 +95,14 @@ class Downloader(
         logger.debug { "stopped" }
     }
 
-    private suspend fun finishDownload(
+    private fun finishDownload(
         logger: KLogger,
         download: DownloadChapter,
     ) {
-        downloadQueue.removeIf { it.mangaId == download.mangaId && it.chapterIndex == download.chapterIndex }
-        step(null, false)
-        logger.debug { "finished" }
+        downloadQueue -= download
+        notifier(true, DownloadUpdate(FINISHED, download))
         onDownloadFinished()
+        logger.debug { "finished" }
     }
 
     private suspend fun run() {
@@ -122,12 +127,13 @@ class Downloader(
 
             try {
                 download.state = Downloading
-                step(download, true)
+                step(DownloadUpdate(PROGRESS, download), true)
 
                 download.chapter = getChapterDownloadReadyByIndex(download.chapterIndex, download.mangaId)
-                step(download, false)
 
-                ChapterDownloadHelper.download(download.mangaId, download.chapter.id, download, scope, this::step)
+                ChapterDownloadHelper.download(download.mangaId, download.chapter.id, download, scope) { downloadChapter, immediate ->
+                    step(downloadChapter?.let { DownloadUpdate(PROGRESS, downloadChapter) }, immediate)
+                }
                 download.state = Finished
                 transaction {
                     ChapterTable.update(
@@ -136,7 +142,6 @@ class Downloader(
                         it[isDownloaded] = true
                     }
                 }
-                step(download, true)
                 finishDownload(downloadLogger, download)
             } catch (e: CancellationException) {
                 logger.debug("Downloader was stopped")
@@ -149,7 +154,7 @@ class Downloader(
                 download.tries++
                 download.state = Error
             } finally {
-                notifier(false)
+                notifier(false, DownloadUpdate(ERROR, download))
             }
         }
     }

server/src/main/kotlin/suwayomi/tachidesk/manga/impl/download/model/DownloadChapter.kt

@@ -16,6 +16,7 @@ class DownloadChapter(
     val mangaId: Int,
     var chapter: ChapterDataClass,
     var manga: MangaDataClass,
+    var position: Int,
     var state: DownloadState = Queued,
     var progress: Float = 0f,
     var tries: Int = 0,

server/src/main/kotlin/suwayomi/tachidesk/manga/impl/download/model/DownloadStatus.kt

@@ -16,3 +16,9 @@ data class DownloadStatus(
     val status: Status,
     val queue: List<DownloadChapter>,
 )
+
+data class DownloadUpdates(
+    val status: Status,
+    val updates: List<DownloadUpdate>,
+    val initial: List<DownloadChapter>?,
+)

server/src/main/kotlin/suwayomi/tachidesk/manga/impl/download/model/DownloadUpdate.kt

@@ -0,0 +1,15 @@
+package suwayomi.tachidesk.manga.impl.download.model
+
+enum class DownloadUpdateType {
+    QUEUED,
+    DEQUEUED,
+    PROGRESS,
+    FINISHED,
+    ERROR,
+    POSITION,
+}
+
+data class DownloadUpdate(
+    val type: DownloadUpdateType,
+    val downloadChapter: DownloadChapter,
+)