feat: guided Loom bulk import with live progress, 2,000 videos per CSV

apps/desktop-gpui/src/editor_preparing/audio.rs

@@ -174,13 +174,16 @@ mod tests {
 
     impl TestDirectory {
         fn new() -> Self {
+            static NEXT_DIRECTORY: std::sync::atomic::AtomicUsize =
+                std::sync::atomic::AtomicUsize::new(0);
             let nonce = std::time::SystemTime::now()
                 .duration_since(std::time::UNIX_EPOCH)
                 .unwrap()
                 .as_nanos();
             let path = std::env::temp_dir().join(format!(
-                "cap-gpui-preparing-audio-{}-{nonce}",
-                std::process::id()
+                "cap-gpui-preparing-audio-{}-{nonce}-{}",
+                std::process::id(),
+                NEXT_DIRECTORY.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
             ));
             std::fs::create_dir(&path).unwrap();
             Self(path)

apps/media-server/src/__tests__/lib/idle-timeout.test.ts

@@ -0,0 +1,37 @@
+import { describe, expect, test } from "bun:test";
+import { withIdleTimeout } from "../../lib/media-common";
+
+const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe("withIdleTimeout", () => {
+	test("lets slow work finish for as long as it keeps reporting progress", async () => {
+		const result = await withIdleTimeout(async (touch) => {
+			for (let step = 0; step < 8; step++) {
+				await wait(30);
+				touch();
+			}
+			return "done";
+		}, 80);
+		expect(result).toBe("done");
+	});
+
+	test("stops work that goes quiet and runs its cleanup", async () => {
+		let cleaned = false;
+		const started = performance.now();
+		await expect(
+			withIdleTimeout(
+				async (touch) => {
+					touch();
+					await wait(1_000);
+					return "late";
+				},
+				60,
+				() => {
+					cleaned = true;
+				},
+			),
+		).rejects.toThrow("Stopped making progress");
+		expect(performance.now() - started).toBeLessThan(500);
+		expect(cleaned).toBe(true);
+	});
+});

apps/media-server/src/__tests__/lib/job-manager.test.ts

@@ -7,11 +7,14 @@ import {
 	createJob,
 	deleteJob,
 	getJob,
+	JOB_PROGRESS_STALL_MS,
 	type JobProgress,
+	markJobProgress,
 	type RecordingWorkerAcknowledgement,
 	sendWebhook,
 	touchJob,
 	updateJob,
+	watchJobProgress,
 } from "../../lib/job-manager";
 
 const createdJobs: string[] = [];
@@ -521,4 +524,59 @@ describe("job cleanup", () => {
 		expect(currentJob?.phase).toBe("error");
 		expect(currentJob?.error).toContain("maximum lifetime of 60 minutes");
 	});
+
+	test("keeps a job that is still making progress running for hours", () => {
+		const job = createTrackedJob("job-long-but-moving");
+		const now = Date.now();
+		job.phase = "processing";
+		watchJobProgress(job.jobId);
+		job.createdAt = now - 3 * 60 * 60 * 1000;
+		job.updatedAt = now;
+		job.progressAt = now - 60_000;
+
+		expect(cleanupExpiredJobs()).toBe(0);
+		expect(getJob(job.jobId)?.phase).toBe("processing");
+	});
+
+	test("fails a job that stopped making progress, however short it is", () => {
+		const job = createTrackedJob("job-short-but-stuck");
+		const now = Date.now();
+		job.phase = "processing";
+		watchJobProgress(job.jobId);
+		job.createdAt = now - 20 * 60 * 1000;
+		job.updatedAt = now;
+		job.progressAt = now - JOB_PROGRESS_STALL_MS - 60_000;
+
+		expect(cleanupExpiredJobs()).toBe(1);
+		expect(getJob(job.jobId)).toMatchObject({
+			phase: "error",
+			message: "Processing failed (stalled)",
+		});
+		expect(getJob(job.jobId)?.error).toContain("stopped making progress");
+	});
+
+	test("counts rising progress and new phases as progress, not heartbeats", () => {
+		const job = createTrackedJob("job-progress-signals");
+		job.phase = "processing";
+		job.progress = 20;
+		watchJobProgress(job.jobId);
+		const old = Date.now() - 10 * 60 * 1000;
+
+		job.progressAt = old;
+		touchJob(job.jobId);
+		updateJob(job.jobId, { message: "Still here" });
+		updateJob(job.jobId, { progress: 20 });
+		expect(job.progressAt).toBe(old);
+
+		updateJob(job.jobId, { progress: 21 });
+		expect(job.progressAt).toBeGreaterThan(old);
+
+		job.progressAt = old;
+		updateJob(job.jobId, { phase: "uploading" });
+		expect(job.progressAt).toBeGreaterThan(old);
+
+		job.progressAt = old;
+		markJobProgress(job.jobId);
+		expect(job.progressAt).toBeGreaterThan(old);
+	});
 });

apps/media-server/src/__tests__/lib/media-video.integration.test.ts

@@ -32,6 +32,7 @@ import {
 	repairContainer,
 	uploadFileToS3,
 	uploadFileToStorage,
+	uploadTimeoutMs,
 	uploadToS3,
 } from "../../lib/media-video";
 
@@ -1008,6 +1009,80 @@ describe("processVideo integration tests", () => {
 		expect(existsSync(tempFile.path)).toBe(false);
 	}, 60000);
 
+	test("lets an encode run past its idle limit while frames keep advancing", async () => {
+		const dir = mkdtempSync(join(tmpdir(), "advancing-encode-"));
+		const longInput = join(dir, "input.mp4");
+		execFileSync("ffmpeg", [
+			"-hide_banner",
+			"-v",
+			"error",
+			"-f",
+			"lavfi",
+			"-i",
+			"testsrc2=size=1280x720:rate=30",
+			"-t",
+			"60",
+			"-c:v",
+			"libx264",
+			"-preset",
+			"ultrafast",
+			"-pix_fmt",
+			"yuv420p",
+			"-y",
+			longInput,
+		]);
+		const idleTimeoutMs = 1_500;
+		try {
+			const metadata = await probeVideo(`file://${longInput}`);
+			const progressUpdates: number[] = [];
+			const started = performance.now();
+
+			const tempFile = await processVideo(
+				longInput,
+				metadata,
+				{ maxWidth: 960, maxHeight: 540, preset: "medium", idleTimeoutMs },
+				(progress) => progressUpdates.push(progress),
+			);
+			tempFiles.push(tempFile.path);
+
+			expect(performance.now() - started).toBeGreaterThan(idleTimeoutMs * 2);
+			expect(progressUpdates.length).toBeGreaterThan(2);
+			for (let index = 1; index < progressUpdates.length; index++) {
+				expect(progressUpdates[index]).toBeGreaterThan(
+					progressUpdates[index - 1] ?? -1,
+				);
+			}
+			await tempFile.cleanup();
+		} finally {
+			rmSync(dir, { recursive: true, force: true });
+		}
+	}, 120000);
+
+	test("stops an encode whose input stops producing frames", async () => {
+		const metadata = await probeVideo(`file://${TEST_VIDEO_WITH_AUDIO}`);
+		const dir = mkdtempSync(join(tmpdir(), "stalled-encode-"));
+		const stalledInput = join(dir, "input.mp4");
+		execFileSync("mkfifo", [stalledInput]);
+		try {
+			const started = performance.now();
+			await expect(
+				processVideo(stalledInput, metadata, {
+					remuxOnly: true,
+					idleTimeoutMs: 1_000,
+				}),
+			).rejects.toThrow("Stopped making progress");
+			expect(performance.now() - started).toBeLessThan(15_000);
+		} finally {
+			rmSync(dir, { recursive: true, force: true });
+		}
+	}, 30000);
+
+	test("gives large uploads time in proportion to their size", () => {
+		expect(uploadTimeoutMs(0)).toBe(10 * 60 * 1000);
+		expect(uploadTimeoutMs(200 * 1024 * 1024)).toBe(10 * 60 * 1000);
+		expect(uploadTimeoutMs(3 * 1024 ** 3)).toBe(3 * 1024 * 1000);
+	});
+
 	test("respects CRF setting", async () => {
 		const metadata = await probeVideo(`file://${TEST_VIDEO_WITH_AUDIO}`);
 

apps/media-server/src/__tests__/lib/video-input-validation.integration.test.ts

@@ -208,4 +208,37 @@ describe("raw recording input validation", () => {
 			"timed out",
 		);
 	});
+
+	test("reports decoding progress while it checks a clean recording", async () => {
+		let reports = 0;
+		await expect(
+			validateVideoInput(clean, undefined, undefined, {
+				idleTimeoutMs: 5_000,
+				onProgress: () => reports++,
+			}),
+		).resolves.toBeUndefined();
+		expect(reports).toBeGreaterThan(0);
+	});
+
+	test("still rejects a damaged recording while tracking progress", async () => {
+		await expect(
+			validateVideoInput(corrupt, undefined, undefined, {
+				idleTimeoutMs: 5_000,
+				onProgress: () => {},
+			}),
+		).rejects.toThrow("original upload has been preserved");
+	});
+
+	test("stops a check whose input stops producing frames", async () => {
+		const stalled = join(directory, "stalled.webm");
+		execFileSync("mkfifo", [stalled]);
+		let reports = 0;
+		await expect(
+			validateVideoInput(stalled, undefined, undefined, {
+				idleTimeoutMs: 1_000,
+				onProgress: () => reports++,
+			}),
+		).rejects.toThrow("Stopped making progress");
+		expect(reports).toBe(0);
+	}, 30_000);
 });

apps/media-server/src/__tests__/routes/video-progress-webhooks.test.ts

@@ -0,0 +1,98 @@
+import { afterAll, afterEach, describe, expect, spyOn, test } from "bun:test";
+import { copyFileSync, mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import app from "../../app";
+import * as jobManager from "../../lib/job-manager";
+import { deleteJob, getJob } from "../../lib/job-manager";
+import * as mediaVideo from "../../lib/media-video";
+
+const fixture = join(import.meta.dir, "..", "fixtures", "test-with-audio.mp4");
+const directory = mkdtempSync(join(tmpdir(), "cap-progress-webhooks-"));
+const received: { videoId: string; phase: string; message?: string }[] = [];
+const receiver = Bun.serve({
+	port: 0,
+	async fetch(request) {
+		received.push(await request.json());
+		return Response.json({ success: true });
+	},
+});
+
+afterAll(() => {
+	receiver.stop(true);
+	rmSync(directory, { recursive: true, force: true });
+});
+
+const spies: { mockRestore: () => void }[] = [];
+afterEach(() => {
+	for (const spy of spies.splice(0)) spy.mockRestore();
+});
+
+async function encodeWithRapidProgress(priority: "normal" | "bulk") {
+	const videoId = `progress-${priority}`;
+	const input = join(directory, `${priority}-input.mp4`);
+	const output = join(directory, `${priority}-output.mp4`);
+	copyFileSync(fixture, input);
+	copyFileSync(fixture, output);
+	spies.push(
+		spyOn(jobManager, "canAcceptNewVideoProcess").mockReturnValue(true),
+		spyOn(mediaVideo, "downloadVideoToTemp").mockResolvedValue({
+			path: input,
+			cleanup: async () => {},
+		}),
+		spyOn(mediaVideo, "processVideo").mockImplementation(
+			async (_input, _metadata, _options, onProgress) => {
+				for (let step = 1; step <= 40; step++) {
+					onProgress?.(step * 2.5, `Encoding: ${step * 2.5}%`);
+					await Bun.sleep(60);
+				}
+				return { path: output, cleanup: async () => {} };
+			},
+		),
+		spyOn(mediaVideo, "uploadFileToS3").mockResolvedValue({}),
+	);
+	const secret = process.env.MEDIA_SERVER_WEBHOOK_SECRET ?? "test-secret";
+	process.env.MEDIA_SERVER_WEBHOOK_SECRET = secret;
+	const response = await app.fetch(
+		new Request("http://localhost/video/process", {
+			method: "POST",
+			headers: {
+				"Content-Type": "application/json",
+				"x-media-server-secret": secret,
+			},
+			body: JSON.stringify({
+				videoId,
+				userId: "progress-test",
+				videoUrl: "https://example.com/raw.mp4",
+				outputPresignedUrl: "https://example.com/result.mp4",
+				webhookUrl: `http://127.0.0.1:${receiver.port}/progress`,
+				webhookSecret: secret,
+				priority,
+			}),
+		}),
+	);
+	expect(response.status).toBe(200);
+	const { jobId } = (await response.json()) as { jobId: string };
+	const deadline = Date.now() + 15_000;
+	while (Date.now() < deadline && getJob(jobId)?.phase !== "complete") {
+		await Bun.sleep(20);
+	}
+	expect(getJob(jobId)?.phase).toBe("complete");
+	deleteJob(jobId);
+	return received.filter(
+		(update) =>
+			update.videoId === videoId && update.message?.startsWith("Encoding"),
+	).length;
+}
+
+describe("encoding progress webhooks", () => {
+	test("keeps regular uploads updating every second for the share page", async () => {
+		const sent = await encodeWithRapidProgress("normal");
+		expect(sent).toBeGreaterThanOrEqual(2);
+		expect(sent).toBeLessThanOrEqual(4);
+	}, 30_000);
+
+	test("sends bulk imports one update per five seconds", async () => {
+		expect(await encodeWithRapidProgress("bulk")).toBe(1);
+	}, 30_000);
+});