perf: wake Loom imports from the worker webhook instead of polling

apps/web/__tests__/unit/media-processing-workflows.test.ts

@@ -2,6 +2,8 @@ import { Effect } from "effect";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const mocks = vi.hoisted(() => ({
+	createHook: vi.fn(),
+	hookDispose: vi.fn(),
 	rows: [] as unknown[][],
 	where: vi.fn(),
 	write: vi.fn(),
@@ -66,6 +68,7 @@ vi.mock("@/lib/ai-generation-entitlement", () => ({
 	isAiGenerationEnabledForUser: () => false,
 }));
 vi.mock("workflow", () => ({
+	createHook: mocks.createHook,
 	sleep: mocks.sleep,
 	FatalError: class FatalError extends Error {},
 	getWorkflowMetadata: () => ({ workflowRunId: "run-import" }),
@@ -110,6 +113,12 @@ function s3Error(name: string, httpStatusCode: number) {
 
 describe("media processing workflows", () => {
 	beforeEach(() => {
+		mocks.hookDispose.mockReset();
+		mocks.createHook.mockReset().mockImplementation(() => ({
+			then: (resolve: (value: unknown) => void) =>
+				resolve({ phase: "complete" }),
+			dispose: mocks.hookDispose,
+		}));
 		mocks.rows = [];
 		mocks.env = {};
 		mocks.loomSourceUrl = "https://cdn.loom.com/original.mp4";
@@ -257,7 +266,8 @@ describe("media processing workflows", () => {
 
 		it("sends HLS downloads to the importer and defers transcription", async () => {
 			mocks.loomSourceUrl = "https://luna.loom.com/id/x/playlist.m3u8";
-			mocks.rows = [[video], [pending], [], [metadata]];
+			mocks.rows = [[video], [], [metadata]];
+			mocks.sleep.mockImplementation(() => new Promise(() => undefined));
 			await expect(importLoomVideoWorkflow(payload)).resolves.toMatchObject({
 				success: true,
 			});
@@ -277,8 +287,12 @@ describe("media processing workflows", () => {
 			expect(body).toMatchObject({
 				videoUrl: "https://luna.loom.com/id/x/playlist.m3u8",
 				webhookUrl:
-					"https://cap.example.com/api/webhooks/media-server/progress?retryable=true&transcription=on-view&preview=on-view",
+					"https://cap.example.com/api/webhooks/media-server/progress?retryable=true&signal=loom-import%3Avideo%3Arun-import%3A0&transcription=on-view&preview=on-view",
+			});
+			expect(mocks.createHook).toHaveBeenCalledWith({
+				token: "loom-import:video:run-import:0",
 			});
+			expect(mocks.hookDispose).toHaveBeenCalledTimes(1);
 			expect(body.previewGifPresignedUrl).toBeUndefined();
 			expect(body.thumbnailPresignedUrl).toBeDefined();
 			expect(body.sourcePresignedUrl).toBeUndefined();

apps/web/__tests__/unit/media-server-progress.test.ts

@@ -17,6 +17,7 @@ const mocks = vi.hoisted(() => ({
 	transcribe: vi.fn(),
 	invalidateQuota: vi.fn(),
 	audio: vi.fn(),
+	signal: vi.fn(),
 	tables: {
 		videos: {
 			id: "videos.id",
@@ -75,6 +76,10 @@ vi.mock("@/lib/queue-video-transcription", () => ({
 	shouldQueueTranscriptionAfterMediaComplete: () => true,
 }));
 
+vi.mock("@/lib/loom-import-signal", () => ({
+	signalLoomImportFinished: mocks.signal,
+}));
+
 import { POST } from "@/app/api/webhooks/media-server/progress/route";
 import {
 	parseDesktopRecordingJob,
@@ -946,6 +951,56 @@ describe("Loom importer completions", () => {
 			expect(write.sql).not.toContain("loomPreviewGif");
 	});
 
+	it("wakes the waiting import once the finished state is saved", async () => {
+		const database = loomVideo();
+		mocks.signal.mockReset().mockImplementation(async () => {
+			expect(database.mutations).toContainEqual({
+				operation: "delete",
+				table: mocks.tables.uploads,
+			});
+		});
+		const token = `loom-import:${videoId}:run-1:0`;
+		const response = await request(
+			complete(),
+			"media-secret",
+			`?retryable=true&signal=${encodeURIComponent(token)}`,
+		);
+		expect(response.status).toBe(200);
+		expect(mocks.signal).toHaveBeenCalledWith(token, videoId, "complete");
+	});
+
+	it("wakes the waiting import when the worker fails", async () => {
+		loomVideo();
+		mocks.signal.mockReset().mockResolvedValue(undefined);
+		await request(
+			{
+				jobId: "importer-1",
+				videoId,
+				phase: "error",
+				progress: 40,
+				error: "Loom 404",
+			},
+			"media-secret",
+			"?retryable=true&signal=loom-import%3Aowned-video%3Arun-1%3A0",
+		);
+		expect(mocks.signal).toHaveBeenCalledWith(
+			"loom-import:owned-video:run-1:0",
+			videoId,
+			"error",
+		);
+	});
+
+	it("does not wake the import for progress updates", async () => {
+		loomVideo();
+		mocks.signal.mockReset().mockResolvedValue(undefined);
+		await request(
+			{ jobId: "importer-1", videoId, phase: "downloading", progress: 40 },
+			"media-secret",
+			"?retryable=true&signal=loom-import%3Aowned-video%3Arun-1%3A0",
+		);
+		expect(mocks.signal).not.toHaveBeenCalled();
+	});
+
 	it("keeps queueing transcription for media server completions", async () => {
 		const database = loomVideo();
 		await request(

apps/web/__tests__/unit/video-processing-status.test.ts

@@ -26,6 +26,7 @@ vi.mock("workflow", () => ({
 
 import {
 	readVideoProcessingStatus,
+	waitForSignalledVideoProcessing,
 	waitForVideoProcessing,
 } from "@/workflows/video-processing-status";
 
@@ -108,3 +109,75 @@ describe("durable video processing completion", () => {
 		);
 	});
 });
+
+describe("signalled video processing completion", () => {
+	beforeEach(() => {
+		mocks.rows = [];
+		mocks.now = 0;
+		vi.spyOn(Date, "now").mockImplementation(() => mocks.now);
+		mocks.where.mockReset().mockImplementation(async () => mocks.rows.shift());
+		mocks.sleep.mockReset().mockImplementation(async (delay: number) => {
+			mocks.now += delay;
+		});
+	});
+
+	afterEach(() => vi.restoreAllMocks());
+
+	it("reads the status once when the worker signals", async () => {
+		mocks.sleep.mockImplementation(() => new Promise(() => undefined));
+		mocks.rows = [[], [metadata]];
+		await expect(
+			waitForSignalledVideoProcessing("video", Promise.resolve()),
+		).resolves.toEqual(metadata);
+		expect(mocks.where).toHaveBeenCalledTimes(2);
+	});
+
+	it("checks every five minutes when no signal arrives", async () => {
+		mocks.rows = [[pending], [pending], [], [metadata]];
+		await expect(
+			waitForSignalledVideoProcessing("video", new Promise(() => undefined)),
+		).resolves.toEqual(metadata);
+		expect(mocks.sleep.mock.calls).toEqual([[300_000], [300_000], [300_000]]);
+	});
+
+	it("falls back to short polling if the signal beats the saved state", async () => {
+		const sleeps: number[] = [];
+		let signalled = false;
+		mocks.sleep.mockImplementation((delay: number) => {
+			sleeps.push(delay);
+			mocks.now += delay;
+			return delay === 300_000
+				? new Promise(() => undefined)
+				: Promise.resolve();
+		});
+		mocks.rows = [[pending], [pending], [], [metadata]];
+		await expect(
+			waitForSignalledVideoProcessing(
+				"video",
+				Promise.resolve().then(() => {
+					signalled = true;
+				}),
+			),
+		).resolves.toEqual(metadata);
+		expect(signalled).toBe(true);
+		expect(sleeps).toEqual([300_000, 5_000, 10_000]);
+	});
+
+	it("propagates worker failures after the signal", async () => {
+		mocks.sleep.mockImplementation(() => new Promise(() => undefined));
+		mocks.rows = [[{ ...pending, processingError: "Loom 404" }]];
+		await expect(
+			waitForSignalledVideoProcessing("video", Promise.resolve()),
+		).rejects.toThrow("Loom 404");
+	});
+
+	it("subscribes to the signal only once", async () => {
+		const then = vi.fn();
+		mocks.where.mockResolvedValue([pending]);
+		await expect(
+			waitForSignalledVideoProcessing("video", { then }),
+		).rejects.toThrow("Video processing timed out");
+		expect(then).toHaveBeenCalledTimes(1);
+		expect(mocks.now).toBeLessThanOrEqual(60 * 60 * 1000 + 300_000);
+	});
+});

apps/web/app/api/webhooks/media-server/progress/route.ts

@@ -15,6 +15,7 @@ import {
 } from "@/lib/desktop-recording-verification";
 import { queueDesktopSegmentsFinalization } from "@/lib/desktop-segments-finalization";
 import { invalidateGoogleDriveStorageQuotaCache } from "@/lib/google-drive-storage-quota";
+import { signalLoomImportFinished } from "@/lib/loom-import-signal";
 import {
 	queueVideoTranscription,
 	shouldQueueTranscriptionAfterMediaComplete,
@@ -381,6 +382,13 @@ export async function POST(request: NextRequest) {
 				.where(eq(videoUploads.videoId, payload.videoId as Video.VideoId));
 		}
 
+		if (dbPhase === "complete" || dbPhase === "error")
+			await signalLoomImportFinished(
+				request.nextUrl.searchParams.get("signal"),
+				payload.videoId,
+				payload.phase,
+			);
+
 		return NextResponse.json({ success: true });
 	} catch (error) {
 		console.error("[media-server-webhook] Error processing webhook:", error);

apps/web/lib/loom-import-signal-token.ts

@@ -0,0 +1,9 @@
+export const LOOM_IMPORT_SIGNAL_PREFIX = "loom-import:";
+
+export function getLoomImportSignalToken(videoId: string, attemptId: string) {
+	return `${LOOM_IMPORT_SIGNAL_PREFIX}${videoId}:${attemptId}`;
+}
+
+export function isLoomImportSignalToken(token: string, videoId: string) {
+	return token.startsWith(`${LOOM_IMPORT_SIGNAL_PREFIX}${videoId}:`);
+}

apps/web/lib/loom-import-signal.ts

@@ -0,0 +1,22 @@
+import "server-only";
+
+import { resumeHook } from "workflow/api";
+import { HookNotFoundError } from "workflow/errors";
+import { isLoomImportSignalToken } from "@/lib/loom-import-signal-token";
+
+export async function signalLoomImportFinished(
+	token: string | null,
+	videoId: string,
+	phase: string,
+) {
+	if (!token || !isLoomImportSignalToken(token, videoId)) return;
+	try {
+		await resumeHook(token, { phase });
+	} catch (error) {
+		if (HookNotFoundError.is(error)) return;
+		console.warn("[loom-import-signal] Could not wake the import", {
+			videoId,
+			error,
+		});
+	}
+}