From b0d06a1d8c60b2087823c9676116294057175ff3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Felix=20F=C3=B6rtsch?= <mail@felixfoertsch.de>
Date: Wed, 15 Apr 2026 08:12:59 +0200
Subject: [PATCH] execute: drain newly-approved jobs mid-run, use strict shared
 parseId
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Queue previously processed a snapshot of pending jobs — anything approved
after Run-all clicked sat idle until the user clicked again. Now, when
the local queue drains, re-poll the DB once for newly-approved jobs
before exiting.

Also swap the looser local parseId (Number.parseInt accepted '42abc')
for the strict shared parseId in server/lib/validate.ts.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 server/api/execute.ts | 32 ++++++++++++++++++++++----------
 1 file changed, 22 insertions(+), 10 deletions(-)

diff --git a/server/api/execute.ts b/server/api/execute.ts
index 3146e7c..a622bd5 100644
--- a/server/api/execute.ts
+++ b/server/api/execute.ts
@@ -3,6 +3,7 @@ import { Hono } from "hono";
 import { stream } from "hono/streaming";
 import { getDb } from "../db/index";
 import { log, error as logError, warn } from "../lib/log";
+import { parseId } from "../lib/validate";
 import { predictExtractedFiles } from "../services/ffmpeg";
 import {
 	getScheduleConfig,
@@ -31,12 +32,18 @@ function emitQueueStatus(
 	for (const l of jobListeners) l(line);
 }
 
-async function runSequential(jobs: Job[]): Promise<void> {
+async function runSequential(initial: Job[]): Promise<void> {
 	if (queueRunning) return;
 	queueRunning = true;
 	try {
 		let first = true;
-		for (const job of jobs) {
+		const queue: Job[] = [...initial];
+		const seen = new Set<number>(queue.map((j) => j.id));
+
+		while (queue.length > 0) {
+			// biome-ignore lint/style/noNonNullAssertion: length checked above
+			const job = queue.shift()!;
+
 			// Pause outside the processing window
 			if (!isInProcessWindow()) {
 				emitQueueStatus("paused", {
@@ -70,6 +77,19 @@ async function runSequential(jobs: Job[]): Promise<void> {
 			} catch (err) {
 				logError(`Job ${job.id} failed:`, err);
 			}
+
+			// When the local queue drains, re-check the DB for jobs that were
+			// approved mid-run. Without this they'd sit pending until the user
+			// manually clicks "Run all" again.
+			if (queue.length === 0) {
+				const more = db.prepare("SELECT * FROM jobs WHERE status = 'pending' ORDER BY created_at").all() as Job[];
+				for (const m of more) {
+					if (!seen.has(m.id)) {
+						queue.push(m);
+						seen.add(m.id);
+					}
+				}
+			}
 		}
 	} finally {
 		queueRunning = false;
@@ -137,14 +157,6 @@ function loadJobRow(jobId: number) {
 	return { job: row as unknown as Job, item };
 }
 
-// ─── Param helpers ────────────────────────────────────────────────────────────
-
-function parseId(raw: string | undefined): number | null {
-	if (!raw) return null;
-	const n = Number.parseInt(raw, 10);
-	return Number.isFinite(n) && n > 0 ? n : null;
-}
-
 // ─── Start all pending ────────────────────────────────────────────────────────
 
 app.post("/start", (c) => {