Check upload phase and resolve video source

Author: richiemcilroy

apps/web/__tests__/integration/transcribe.test.ts

@@ -8,6 +8,12 @@ vi.mock("@cap/env", () => ({
 }));
 
 const mockStart = vi.hoisted(() => vi.fn());
+const schemaMocks = vi.hoisted(() => ({
+	videos: { id: "id", settings: "settings" },
+	organizations: { id: "id", settings: "settings" },
+	s3Buckets: { id: "id" },
+	videoUploads: { videoId: "videoId", phase: "phase" },
+}));
 
 vi.mock("workflow/api", () => ({
 	start: mockStart,
@@ -18,19 +24,27 @@ vi.mock("@/workflows/transcribe", () => ({
 }));
 
 let mockQueryResult: unknown[] = [];
+let mockUploadQueryResult: unknown[] = [];
 
 vi.mock("@cap/database", () => ({
 	db: () => ({
 		select: () => ({
-			from: () => ({
-				leftJoin: () => ({
-					leftJoin: () => ({
-						where: vi
-							.fn()
-							.mockImplementation(() => Promise.resolve(mockQueryResult)),
-					}),
-				}),
-			}),
+			from: (table: unknown) =>
+				table === schemaMocks.videoUploads
+					? {
+							where: vi.fn().mockReturnValue({
+								limit: vi.fn().mockResolvedValue(mockUploadQueryResult),
+							}),
+						}
+					: {
+							leftJoin: () => ({
+								leftJoin: () => ({
+									where: vi
+										.fn()
+										.mockImplementation(() => Promise.resolve(mockQueryResult)),
+								}),
+							}),
+						},
 		}),
 		update: () => ({
 			set: () => ({
@@ -41,9 +55,10 @@ vi.mock("@cap/database", () => ({
 }));
 
 vi.mock("@cap/database/schema", () => ({
-	videos: { id: "id", settings: "settings" },
-	organizations: { id: "id", settings: "settings" },
-	s3Buckets: { id: "id" },
+	videos: schemaMocks.videos,
+	organizations: schemaMocks.organizations,
+	s3Buckets: schemaMocks.s3Buckets,
+	videoUploads: schemaMocks.videoUploads,
 }));
 
 vi.mock("drizzle-orm", () => ({
@@ -58,6 +73,7 @@ describe("transcribeVideo", () => {
 	beforeEach(() => {
 		vi.clearAllMocks();
 		mockQueryResult = [];
+		mockUploadQueryResult = [];
 	});
 
 	describe("input validation", () => {
@@ -266,6 +282,19 @@ describe("transcribeVideo", () => {
 			mockStart.mockResolvedValue({ id: "workflow-run-123" });
 		});
 
+		it("does not trigger while upload is still active", async () => {
+			mockUploadQueryResult = [{ phase: "processing" }];
+
+			const result = await transcribeVideo(
+				"video-123" as Video.VideoId,
+				"user-456",
+			);
+
+			expect(result.success).toBe(true);
+			expect(result.message).toBe("Video upload is still in progress");
+			expect(mockStart).not.toHaveBeenCalled();
+		});
+
 		it("triggers workflow for valid video", async () => {
 			const result = await transcribeVideo(
 				"video-123" as Video.VideoId,

apps/web/lib/transcribe.ts

@@ -1,5 +1,10 @@
 import { db } from "@cap/database";
-import { organizations, s3Buckets, videos } from "@cap/database/schema";
+import {
+	organizations,
+	s3Buckets,
+	videos,
+	videoUploads,
+} from "@cap/database/schema";
 import { serverEnv } from "@cap/env";
 import type { Video } from "@cap/web-domain";
 import { eq } from "drizzle-orm";
@@ -95,6 +100,23 @@ export async function transcribeVideo(
 		};
 	}
 
+	const upload = await db()
+		.select({ phase: videoUploads.phase })
+		.from(videoUploads)
+		.where(eq(videoUploads.videoId, videoId))
+		.limit(1);
+
+	if (
+		upload[0]?.phase === "uploading" ||
+		upload[0]?.phase === "processing" ||
+		upload[0]?.phase === "generating_thumbnail"
+	) {
+		return {
+			success: true,
+			message: "Video upload is still in progress",
+		};
+	}
+
 	try {
 		console.log(
 			`[transcribeVideo] Triggering transcription workflow for video ${videoId}`,

apps/web/workflows/transcribe.ts

@@ -1,6 +1,12 @@
 import { promises as fs } from "node:fs";
 import { db } from "@cap/database";
-import { organizations, s3Buckets, users, videos } from "@cap/database/schema";
+import {
+	organizations,
+	s3Buckets,
+	users,
+	videos,
+	videoUploads,
+} from "@cap/database/schema";
 import type { VideoMetadata } from "@cap/database/types";
 import { serverEnv } from "@cap/env";
 import { userIsPro } from "@cap/utils";
@@ -175,16 +181,7 @@ async function extractAudio(
 		Option.fromNullable(bucketId),
 	).pipe(runPromise);
 
-	const videoKey = `${userId}/${videoId}/result.mp4`;
-	const videoUrl = await bucket.getSignedObjectUrl(videoKey).pipe(runPromise);
-
-	const response = await fetch(videoUrl, {
-		method: "GET",
-		headers: { range: "bytes=0-0" },
-	});
-	if (!response.ok) {
-		throw new Error("Video file not accessible");
-	}
+	const videoUrl = await resolveVideoSourceUrl(videoId, userId, bucketId);
 
 	const useMediaServer = isMediaServerConfigured();
 
@@ -228,6 +225,45 @@ async function extractAudio(
 	return audioSignedUrl;
 }
 
+async function resolveVideoSourceUrl(
+	videoId: string,
+	userId: string,
+	bucketId: S3Bucket.S3BucketId | null,
+): Promise<string> {
+	const [bucket] = await S3Buckets.getBucketAccess(
+		Option.fromNullable(bucketId),
+	).pipe(runPromise);
+
+	const upload = await db()
+		.select({ rawFileKey: videoUploads.rawFileKey })
+		.from(videoUploads)
+		.where(eq(videoUploads.videoId, videoId as Video.VideoId))
+		.limit(1);
+
+	const candidateKeys = [
+		`${userId}/${videoId}/result.mp4`,
+		upload[0]?.rawFileKey,
+	].filter(
+		(value, index, values): value is string =>
+			Boolean(value) && values.indexOf(value) === index,
+	);
+
+	for (const key of candidateKeys) {
+		const url = await bucket.getSignedObjectUrl(key).pipe(runPromise);
+		const response = await fetch(url, {
+			method: "GET",
+			headers: { range: "bytes=0-0" },
+		});
+
+		if (response.ok) {
+			console.log(`[transcribe] Using video source ${key}`);
+			return url;
+		}
+	}
+
+	throw new Error("Video file not accessible");
+}
+
 async function transcribeWithDeepgram(audioUrl: string): Promise<string> {
 	"use step";