From 44a99bdb3af174e0fb4985badfced2c4a781ea4d Mon Sep 17 00:00:00 2001
From: wukko <me@wukko.me>
Date: Sat, 25 Jan 2025 01:25:53 +0600
Subject: [PATCH] web/queue: add remuxing progress & general improvements

and a bunch of other stuff:
- size and percentage in queue
- indeterminate progress bar
- if libav wasm freezes, the worker kill itself
- cleaner states
- cleaner props
---
 web/src/components/misc/DropReceiver.svelte   |  1 -
 .../components/queue/ProcessingQueue.svelte   | 16 +++--
 .../queue/ProcessingQueueItem.svelte          | 65 +++++++++++++------
 web/src/lib/libav.ts                          |  2 +-
 web/src/lib/queen-bee/run-worker.ts           | 36 +++++++++-
 web/src/lib/queen-bee/scheduler.ts            | 14 ++--
 web/src/lib/state/queen-bee/current-tasks.ts  | 12 ++--
 web/src/lib/state/queen-bee/queue.ts          |  8 ++-
 web/src/lib/types/queen-bee.ts                |  5 +-
 web/src/lib/types/queue.ts                    |  2 +-
 web/src/lib/types/workers.ts                  |  4 +-
 web/src/lib/workers/remux.ts                  |  5 +-
 12 files changed, 117 insertions(+), 53 deletions(-)

diff --git a/web/src/components/misc/DropReceiver.svelte b/web/src/components/misc/DropReceiver.svelte
index 4a51407a..e8de49b9 100644
--- a/web/src/components/misc/DropReceiver.svelte
+++ b/web/src/components/misc/DropReceiver.svelte
@@ -16,7 +16,6 @@
     };
 
     const dragOverHandler = (ev: DragEvent) => {
-        console.log("dragged over omg")
         draggedOver = true;
         ev.preventDefault();
     };
diff --git a/web/src/components/queue/ProcessingQueue.svelte b/web/src/components/queue/ProcessingQueue.svelte
index 2e89c500..ac8af3fb 100644
--- a/web/src/components/queue/ProcessingQueue.svelte
+++ b/web/src/components/queue/ProcessingQueue.svelte
@@ -26,7 +26,7 @@
 
     $: queueLength = Object.keys($queue).length;
     $: completedQueueItems = queueItems.filter(([id, item]) => {
-        return item.state === "done"
+        return item.state === "done";
     }).length;
 
     // TODO: toggle this only when progress is unknown
@@ -42,7 +42,11 @@
 </script>
 
 <div id="processing-queue" class:expanded>
-    <ProcessingStatus progress={(completedQueueItems / queueLength) * 100} {indeterminate} expandAction={popover?.showPopover} />
+    <ProcessingStatus
+        progress={(completedQueueItems / queueLength) * 100}
+        {indeterminate}
+        expandAction={popover?.showPopover}
+    />
 
     <PopoverContainer
         bind:this={popover}
@@ -71,10 +75,10 @@
             {#each queueItems as [id, item]}
                 <ProcessingQueueItem
                     {id}
-                    mediaType={item.mediaType}
-                    filename={item.filename}
-                    state={item.state}
-                    resultFile={item.state === "done" ? item.resultFile : undefined}
+                    info={item}
+                    runningWorker={
+                        item.state === "running" ? $currentTasks[item.runningWorker] : undefined
+                    }
                 />
             {/each}
             {#if queueLength === 0}
diff --git a/web/src/components/queue/ProcessingQueueItem.svelte b/web/src/components/queue/ProcessingQueueItem.svelte
index 75d6743b..ce4ac82a 100644
--- a/web/src/components/queue/ProcessingQueueItem.svelte
+++ b/web/src/components/queue/ProcessingQueueItem.svelte
@@ -1,5 +1,11 @@
 <script lang="ts">
-    import { queue, removeItem } from "$lib/state/queen-bee/queue";
+    import { downloadFile } from "$lib/download";
+    import { removeItem } from "$lib/state/queen-bee/queue";
+
+    import type { CobaltQueueItem } from "$lib/types/queue";
+    import type { CobaltCurrentTaskItem } from "$lib/types/queen-bee";
+
+    import Skeleton from "$components/misc/Skeleton.svelte";
 
     import IconX from "@tabler/icons-svelte/IconX.svelte";
     import IconDownload from "@tabler/icons-svelte/IconDownload.svelte";
@@ -7,9 +13,6 @@
     import IconMovie from "@tabler/icons-svelte/IconMovie.svelte";
     import IconMusic from "@tabler/icons-svelte/IconMusic.svelte";
     import IconPhoto from "@tabler/icons-svelte/IconPhoto.svelte";
-    import { downloadFile } from "$lib/download";
-    import type { CobaltQueueItemState } from "$lib/types/queue";
-    import type { CobaltPipelineResultFileType } from "$lib/types/workers";
 
     const itemIcons = {
         video: IconMovie,
@@ -18,13 +21,12 @@
     };
 
     export let id: string;
-    export let mediaType: CobaltPipelineResultFileType;
-    export let filename: string;
-    export let state: CobaltQueueItemState;
-    export let resultFile: File | undefined;
+    export let info: CobaltQueueItem;
+    export let runningWorker: CobaltCurrentTaskItem | undefined;
 
-    // TODO: use a real value
-    const progress = 0;
+    $: state = info.state;
+
+    $: progress = runningWorker?.progress;
 
     const download = (file: File) =>
         downloadFile({
@@ -36,27 +38,47 @@
     <div class="processing-info">
         <div class="file-title">
             <div class="processing-type">
-                <svelte:component this={itemIcons[mediaType]} />
+                <svelte:component this={itemIcons[info.mediaType]} />
             </div>
             <span>
-                {filename}
+                {info.filename}
             </span>
         </div>
         {#if state === "running"}
             <div class="file-progress">
-                <div
-                    class="progress"
-                    style="width: {Math.min(100, progress)}%"
-                ></div>
+                {#if progress?.percentage}
+                    <div
+                        class="progress"
+                        style="width: {Math.min(100, progress?.percentage || 0)}%"
+                    ></div>
+                {:else}
+                    <Skeleton height="6px" width="100%" class="elevated indeterminate-progress" />
+                {/if}
             </div>
         {/if}
-        <div class="file-status">{id}: {state}</div>
+        <div class="file-status">
+            {#if info.state === "done"}
+                done: {info.resultFile?.size} bytes
+            {:else if info.state === "running"}
+                {#if progress && progress.percentage}
+                    processing: {Math.ceil(progress.percentage)}%, {progress.size} bytes
+                {:else if progress && progress.size}
+                    processing: {progress.size}
+                {:else}
+                    processing...
+                {/if}
+            {:else if info.state === "error"}
+                error: {info.errorCode}
+            {:else}
+                queued
+            {/if}
+        </div>
     </div>
     <div class="file-actions">
-        {#if state === "done" && resultFile}
+        {#if info.state === "done" && info.resultFile}
             <button
                 class="action-button"
-                on:click={() => download(resultFile)}
+                on:click={() => download(info.resultFile)}
             >
                 <IconDownload />
             </button>
@@ -112,6 +134,11 @@
     .file-progress .progress {
         height: 6px;
         border-radius: 10px;
+        transition: width 0.1s;
+    }
+
+    .file-progress :global(.indeterminate-progress) {
+        display: block;
     }
 
     .file-progress .progress {
diff --git a/web/src/lib/libav.ts b/web/src/lib/libav.ts
index af5c2e8a..40da0d2e 100644
--- a/web/src/lib/libav.ts
+++ b/web/src/lib/libav.ts
@@ -38,7 +38,7 @@ export default class LibAVWrapper {
 
         try {
             await libav.ffprobe([
-                //'-v', 'quiet',
+                '-v', 'quiet',
                 '-print_format', 'json',
                 '-show_format',
                 '-show_streams',
diff --git a/web/src/lib/queen-bee/run-worker.ts b/web/src/lib/queen-bee/run-worker.ts
index 8163865a..4247ac2b 100644
--- a/web/src/lib/queen-bee/run-worker.ts
+++ b/web/src/lib/queen-bee/run-worker.ts
@@ -1,8 +1,8 @@
 import RemuxWorker from "$lib/workers/remux?worker";
-//import RemoveBgWorker from "$lib/workers/removebg?worker";
 
 import type { CobaltPipelineItem } from "$lib/types/workers";
 import { itemDone, itemError } from "$lib/state/queen-bee/queue";
+import { updateWorkerProgress } from "$lib/state/queen-bee/current-tasks";
 
 const workerError = (parentId: string, workerId: string, worker: Worker, error: string) => {
     itemError(parentId, workerId, error);
@@ -26,13 +26,43 @@ export const runRemuxWorker = async (workerId: string, parentId: string, file: F
         workerError(parentId, workerId, worker, "internal error");
     };
 
+    // sometimes chrome refuses to start libav wasm,
+    // so we check the health and kill self if it doesn't spawn
+
+    let bumpAttempts = 0;
+    const startCheck = setInterval(() => {
+        bumpAttempts++;
+
+        if (bumpAttempts === 8) {
+            worker.terminate();
+            console.error("worker didn't start after 4 seconds, so it was killed");
+
+            // TODO: proper error code
+            return workerError(parentId, workerId, worker, "worker didn't start");
+        }
+    }, 500);
+
+    let totalDuration: number | null = null;
+
     worker.onmessage = (event) => {
         const eventData = event.data.cobaltRemuxWorker;
         if (!eventData) return;
 
-        console.log(eventData);
+        // temporary debug logging
+        console.log(JSON.stringify(eventData, null, 2));
 
-        // TODO: calculate & use progress again
+        clearInterval(startCheck);
+
+        if (eventData.progress) {
+            if (eventData.progress.duration) {
+                totalDuration = eventData.progress.duration;
+            }
+
+            updateWorkerProgress(workerId, {
+                percentage: totalDuration ? (eventData.progress.durationProcessed / totalDuration) * 100 : 0,
+                size: eventData.progress.size,
+            })
+        }
 
         if (eventData.render) {
             return workerSuccess(
diff --git a/web/src/lib/queen-bee/scheduler.ts b/web/src/lib/queen-bee/scheduler.ts
index 468415ae..581cab0e 100644
--- a/web/src/lib/queen-bee/scheduler.ts
+++ b/web/src/lib/queen-bee/scheduler.ts
@@ -23,16 +23,16 @@ export const checkTasks = () => {
                 // one parent & pipeline
                 const pipelineItem = task.pipeline[i];
 
-                startWorker(pipelineItem);
-
-                addWorkerToQueue({
-                    id: pipelineItem.workerId,
+                addWorkerToQueue(pipelineItem.workerId, {
                     parentId: task.id,
-                    step: i + 1,
-                    totalSteps: task.pipeline.length,
                 });
 
-                itemRunning(task.id, i);
+                itemRunning(
+                    task.id,
+                    pipelineItem.workerId
+                );
+
+                startWorker(pipelineItem);
                 break;
             }
             break;
diff --git a/web/src/lib/state/queen-bee/current-tasks.ts b/web/src/lib/state/queen-bee/current-tasks.ts
index e1c26d85..aa413702 100644
--- a/web/src/lib/state/queen-bee/current-tasks.ts
+++ b/web/src/lib/state/queen-bee/current-tasks.ts
@@ -10,9 +10,9 @@ const currentTasks = readable<CobaltCurrentTasks>(
     (_, _update) => { update = _update }
 );
 
-export function addWorkerToQueue(item: CobaltCurrentTaskItem) {
+export function addWorkerToQueue(workerId: string, item: CobaltCurrentTaskItem) {
     update(tasks => {
-        tasks[item.id] = item;
+        tasks[workerId] = item;
         return tasks;
     });
 }
@@ -24,10 +24,10 @@ export function removeWorkerFromQueue(id: string) {
     });
 }
 
-export function updateWorkerProgress(id: string, progress: CobaltWorkerProgress) {
-    update(tasks => {
-        tasks[id].progress = progress;
-        return tasks;
+export function updateWorkerProgress(workerId: string, progress: CobaltWorkerProgress) {
+    update(allTasks => {
+        allTasks[workerId].progress = progress;
+        return allTasks;
     });
 }
 
diff --git a/web/src/lib/state/queen-bee/queue.ts b/web/src/lib/state/queen-bee/queue.ts
index 4d784844..8ccdc8c7 100644
--- a/web/src/lib/state/queen-bee/queue.ts
+++ b/web/src/lib/state/queen-bee/queue.ts
@@ -51,13 +51,13 @@ export function itemDone(id: string, workerId: string, file: File) {
     checkTasks();
 }
 
-export function itemRunning(id: string, step: number) {
+export function itemRunning(id: string, workerId: string) {
     update(queueData => {
         if (queueData[id]) {
             queueData[id] = {
                 ...queueData[id],
                 state: "running",
-                currentStep: step,
+                runningWorker: workerId,
             }
         }
         return queueData;
@@ -68,6 +68,10 @@ export function itemRunning(id: string, step: number) {
 
 export function removeItem(id: string) {
     update(queueData => {
+        for (const worker in queueData[id].pipeline) {
+            removeWorkerFromQueue(queueData[id].pipeline[worker].workerId);
+        }
+
         delete queueData[id];
         return queueData;
     });
diff --git a/web/src/lib/types/queen-bee.ts b/web/src/lib/types/queen-bee.ts
index 3684ac5a..a69345ee 100644
--- a/web/src/lib/types/queen-bee.ts
+++ b/web/src/lib/types/queen-bee.ts
@@ -1,10 +1,7 @@
 import type { CobaltWorkerProgress } from "$lib/types/workers";
 
 export type CobaltCurrentTaskItem = {
-    id: string,
-    parentId: string, // parent id is queue id to which this pipeline worker belongs to
-    step: number,
-    totalSteps: number,
+    parentId: string,
     progress?: CobaltWorkerProgress,
 }
 
diff --git a/web/src/lib/types/queue.ts b/web/src/lib/types/queue.ts
index 91caac90..859afee7 100644
--- a/web/src/lib/types/queue.ts
+++ b/web/src/lib/types/queue.ts
@@ -17,7 +17,7 @@ export type CobaltQueueItemWaiting = CobaltQueueBaseItem & {
 
 export type CobaltQueueItemRunning = CobaltQueueBaseItem & {
     state: "running",
-    currentStep: number,
+    runningWorker: string,
 };
 
 export type CobaltQueueItemDone = CobaltQueueBaseItem & {
diff --git a/web/src/lib/types/workers.ts b/web/src/lib/types/workers.ts
index 3bbc79c5..0dabeb1c 100644
--- a/web/src/lib/types/workers.ts
+++ b/web/src/lib/types/workers.ts
@@ -4,9 +4,9 @@ export type CobaltWorkerType = "remux" | "removebg";
 export type CobaltPipelineResultFileType = typeof resultFileTypes[number];
 
 export type CobaltWorkerProgress = {
-    indeterminate: boolean,
+    percentage?: number,
     speed?: number,
-    percentage: number,
+    size: number,
 }
 
 export type CobaltWorkerArgs = {
diff --git a/web/src/lib/workers/remux.ts b/web/src/lib/workers/remux.ts
index 502d03c0..4f9fd402 100644
--- a/web/src/lib/workers/remux.ts
+++ b/web/src/lib/workers/remux.ts
@@ -15,6 +15,9 @@ const ff = new LibAVWrapper((progress) => {
             progress: {
                 durationProcessed: progress.out_time_sec,
                 speed: progress.speed,
+                size: progress.total_size,
+                currentFrame: progress.frame,
+                fps: progress.fps,
             }
         }
     })
@@ -45,7 +48,7 @@ const remux = async (file: File) => {
 
         self.postMessage({
             cobaltRemuxWorker: {
-                progressInfo: {
+                progress: {
                     duration: Number(file_info.format.duration),
                 }
             }