fix: keep pending Loom conversions pointed at the owner's original after a move

apps/web/__tests__/unit/loom-import-backfill.test.ts

@@ -1,3 +1,5 @@
+import type { SQL } from "drizzle-orm";
+import { MySqlDialect } from "drizzle-orm/mysql-core";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const mocks = vi.hoisted(() => ({
@@ -301,6 +303,55 @@ describe("backfillLoomImport", () => {
 		expect(mocks.reassign).not.toHaveBeenCalled();
 	});
 
+	it("repairs a pending conversion left on the previous owner's path", async () => {
+		mocks.selects = [
+			[
+				video({
+					ownerId: "zephyaaaaaaaaaa",
+					metadata: {
+						loomDeferredProcessing: {
+							status: "pending",
+							rawFileKey: "lukeaaaaaaaaaaa/videoaaaaaaaaaa/raw-upload.webm",
+						},
+					},
+				}),
+			],
+			[lukeFolder("folderaaaaaaaaa")],
+		];
+		await expect(
+			backfillLoomImport(org, { videoId: "videoaaaaaaaaaa" }),
+		).resolves.toMatchObject({ status: "updated", changes: ["original"] });
+		expect(mocks.updates).toHaveLength(1);
+		const query = new MySqlDialect().sqlToQuery(
+			mocks.updates[0]?.metadata as SQL,
+		);
+		expect(query.sql).toContain(
+			"JSON_SET(?, '$.loomDeferredProcessing.rawFileKey', ?)",
+		);
+		expect(query.params).toContain(
+			"zephyaaaaaaaaaa/videoaaaaaaaaaa/raw-upload.webm",
+		);
+	});
+
+	it("leaves a conversion already on the owner's path alone", async () => {
+		mocks.selects = [
+			[
+				video({
+					metadata: {
+						loomDeferredProcessing: {
+							status: "pending",
+							rawFileKey: "lukeaaaaaaaaaaa/videoaaaaaaaaaa/raw-upload.webm",
+						},
+					},
+				}),
+			],
+			[lukeFolder("folderaaaaaaaaa")],
+		];
+		await expect(
+			backfillLoomImport(org, { videoId: "videoaaaaaaaaaa" }),
+		).resolves.toMatchObject({ status: "unchanged" });
+	});
+
 	it("reports moves it could not make or finish cleaning up", async () => {
 		mocks.selects = [[video()], [lukeFolder("folderaaaaaaaaa")]];
 		mocks.reassign.mockResolvedValue({

apps/web/__tests__/unit/loom-operator-tools.test.ts

@@ -1,4 +1,6 @@
 import type { HttpApi } from "@effect/platform";
+import type { SQL } from "drizzle-orm";
+import { MySqlDialect } from "drizzle-orm/mysql-core";
 import { Effect, type Layer as EffectLayer, Option } from "effect";
 import { NextRequest } from "next/server";
 import { beforeEach, describe, expect, it, vi } from "vitest";
@@ -154,6 +156,7 @@ import {
 	encryptEditTranscriptObject,
 } from "@/lib/edit-transcript-storage";
 import { reassignLoomImport } from "@/lib/loom-import-reassign";
+import { getOwnerOriginalKey } from "@/lib/loom-original-key";
 
 const call = (
 	handler: (request: Request) => Promise<Response>,
@@ -293,7 +296,11 @@ describe("reassignLoomImport", () => {
 			],
 		]);
 		expect(mocks.updates).toEqual([
-			{ ownerId: ids.to, public: false, folderId: null },
+			expect.objectContaining({
+				ownerId: ids.to,
+				public: false,
+				folderId: null,
+			}),
 		]);
 		expect(mocks.deletes).toBe(2);
 		expect(mocks.removed).toEqual([mocks.objects.map(({ Key }) => Key)]);
@@ -310,6 +317,19 @@ describe("reassignLoomImport", () => {
 			[importedVideo({ source: { type: "webMP4", outputKey: "edited.mp4" } })],
 			"edited video",
 		],
+		[
+			[
+				importedVideo({
+					metadata: {
+						loomDeferredProcessing: {
+							status: "started",
+							rawFileKey: `${ids.from}/${ids.video}/raw-upload.webm`,
+						},
+					},
+				}),
+			],
+			"conversion running",
+		],
 		[[importedVideo(), []], "new owner not a member"],
 	])("leaves unsafe moves alone: %#", async (selects, reason) => {
 		mocks.selects = (selects as unknown[]).map((row) =>
@@ -341,6 +361,34 @@ describe("reassignLoomImport", () => {
 		expect(mocks.claims).toHaveLength(2);
 	});
 
+	it("points a pending conversion at the new owner's copy", async () => {
+		mocks.selects = [
+			[
+				importedVideo({
+					metadata: {
+						loomDeferredProcessing: {
+							status: "pending",
+							rawFileKey: `someoneaaaaaaaa/${ids.video}/raw-upload.webm`,
+						},
+					},
+				}),
+			],
+			[{ userId: ids.to }],
+		];
+		await reassignLoomImport(input);
+		const query = new MySqlDialect().sqlToQuery(
+			mocks.updates[0]?.metadata as SQL,
+		);
+		expect(query.sql).toContain("'$.loomDeferredProcessing.rawFileKey', ?)");
+		expect(query.params).toEqual([`${ids.to}/${ids.video}/raw-upload.webm`]);
+	});
+
+	it("leaves metadata alone for videos without a pending conversion", async () => {
+		mocks.selects = [[importedVideo()], [{ userId: ids.to }]];
+		await reassignLoomImport(input);
+		expect(mocks.updates[0]).not.toHaveProperty("metadata");
+	});
+
 	it("keeps sharing and visibility when asked", async () => {
 		mocks.selects = [[importedVideo()], [{ userId: ids.to }]];
 		await expect(
@@ -350,8 +398,9 @@ describe("reassignLoomImport", () => {
 			}),
 		).resolves.toMatchObject({ status: "moved" });
 		expect(mocks.updates).toEqual([
-			{ ownerId: ids.to, folderId: "folderaaaaaaaaa" },
+			expect.objectContaining({ ownerId: ids.to, folderId: "folderaaaaaaaaa" }),
 		]);
+		expect(mocks.updates[0]).not.toHaveProperty("public");
 		expect(mocks.deletes).toBe(0);
 	});
 
@@ -458,3 +507,15 @@ describe("POST /api/internal/loom-imports/reassign", () => {
 		});
 	});
 });
+
+describe("getOwnerOriginalKey", () => {
+	it.each([
+		["a/vid/raw-upload.webm", "b/vid/raw-upload.webm"],
+		["a/vid/nested/file.webm", "b/vid/nested/file.webm"],
+		["a/other/raw-upload.webm", null],
+		["a/vid/", null],
+		[undefined, null],
+	])("%s -> %s", (key, expected) => {
+		expect(getOwnerOriginalKey(key, "b", "vid")).toBe(expected);
+	});
+});

apps/web/lib/loom-deferred-processing.ts

@@ -3,6 +3,7 @@ import { videos, videoUploads } from "@cap/database/schema";
 import type { VideoMetadata } from "@cap/database/types";
 import type { Video } from "@cap/web-domain";
 import { and, eq, sql } from "drizzle-orm";
+import { LOOM_REASSIGN_CLAIM_TTL_MS } from "@/lib/loom-original-key";
 import { startVideoProcessingWorkflow } from "@/lib/video-processing";
 
 export const DEFERRED_PROCESSING_RECLAIM_MS = 30 * 60 * 1000;
@@ -61,6 +62,7 @@ export async function startDeferredLoomProcessing(
 			and(
 				eq(videos.id, videoId),
 				sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.rawFileKey')) = ${rawFileKey}`,
+				sql`(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim') IS NULL OR JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim.at')) < ${new Date(now.getTime() - LOOM_REASSIGN_CLAIM_TTL_MS).toISOString()})`,
 				sql`(JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.status')) = 'pending' OR (JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.status')) = 'started' AND JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.startedAt')) < ${reclaimBefore}))`,
 			),
 		);

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

@@ -17,6 +17,7 @@ import {
 	normalizeImportEmail,
 } from "@/lib/loom-csv-import";
 import { reassignLoomImport } from "@/lib/loom-import-reassign";
+import { getOwnerOriginalKey } from "@/lib/loom-original-key";
 import { provisionOrganizationInvitee } from "@/lib/organization-provisioning";
 
 export type LoomBackfillItem = {
@@ -58,6 +59,7 @@ export async function backfillLoomImport(
 			ownerId: videos.ownerId,
 			folderId: videos.folderId,
 			effectiveCreatedAt: videos.effectiveCreatedAt,
+			metadata: videos.metadata,
 			orgOwnerId: organizations.ownerId,
 		})
 		.from(videos)
@@ -153,6 +155,29 @@ export async function backfillLoomImport(
 		changes.push("folder");
 	}
 
+	const deferredKey = video.metadata?.loomDeferredProcessing?.rawFileKey;
+	const repairedKey = getOwnerOriginalKey(deferredKey, ownerId, item.videoId);
+	if (
+		ownerId === video.ownerId &&
+		video.metadata?.loomDeferredProcessing?.status === "pending" &&
+		repairedKey &&
+		repairedKey !== deferredKey
+	) {
+		await db()
+			.update(videos)
+			.set({
+				metadata: sql`JSON_SET(${videos.metadata}, '$.loomDeferredProcessing.rawFileKey', ${repairedKey})`,
+			})
+			.where(
+				and(
+					eq(videos.id, id),
+					eq(videos.ownerId, ownerId),
+					sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.rawFileKey')) = ${deferredKey}`,
+				),
+			);
+		changes.push("original");
+	}
+
 	if (date && video.effectiveCreatedAt?.getTime() !== date.getTime()) {
 		await db()
 			.update(videos)

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

@@ -19,6 +19,10 @@ import {
 	decryptEditTranscriptObject,
 	encryptEditTranscriptObject,
 } from "@/lib/edit-transcript-storage";
+import {
+	getOwnerOriginalKey,
+	LOOM_REASSIGN_CLAIM_TTL_MS,
+} from "@/lib/loom-original-key";
 import { runPromise } from "@/lib/server";
 
 export type LoomReassignOptions = {
@@ -35,8 +39,6 @@ export type LoomReassignOutcome =
 	  }
 	| { videoId: string; status: "skipped" | "failed"; reason: string };
 
-const CLAIM_TTL_MS = 60 * 60 * 1000;
-
 function getAffectedRows(result: unknown) {
 	const header = Array.isArray(result) ? result[0] : result;
 	return (header as { affectedRows?: number } | undefined)?.affectedRows ?? 0;
@@ -104,6 +106,7 @@ export async function reassignLoomImport({
 			ownerId: videos.ownerId,
 			orgId: videos.orgId,
 			source: videos.source,
+			metadata: videos.metadata,
 			bucketId: videos.bucket,
 			storageIntegrationId: videos.storageIntegrationId,
 			uploadPhase: videoUploads.phase,
@@ -132,6 +135,8 @@ export async function reassignLoomImport({
 		return { videoId, status: "skipped", reason: "import not finished" };
 	if (video.source && "outputKey" in video.source && video.source.outputKey)
 		return { videoId, status: "skipped", reason: "edited video" };
+	if (video.metadata?.loomDeferredProcessing?.status === "started")
+		return { videoId, status: "skipped", reason: "conversion running" };
 
 	const [member] = await db()
 		.select({ userId: organizationMembers.userId })
@@ -160,7 +165,7 @@ export async function reassignLoomImport({
 			and(
 				eq(videos.id, id),
 				eq(videos.ownerId, User.UserId.make(fromUserId)),
-				sql`(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim') IS NULL OR JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim.at')) < ${new Date(claimedAt.getTime() - CLAIM_TTL_MS).toISOString()})`,
+				sql`(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim') IS NULL OR JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim.at')) < ${new Date(claimedAt.getTime() - LOOM_REASSIGN_CLAIM_TTL_MS).toISOString()})`,
 			),
 		);
 	if (getAffectedRows(claim) !== 1)
@@ -174,6 +179,7 @@ export async function reassignLoomImport({
 			toUserId,
 			token,
 			options,
+			originalKey: video.metadata?.loomDeferredProcessing?.rawFileKey,
 		});
 	} finally {
 		await db()
@@ -198,6 +204,7 @@ async function moveClaimedVideo({
 	toUserId,
 	token,
 	options,
+	originalKey,
 }: {
 	id: Video.VideoId;
 	videoId: string;
@@ -206,6 +213,7 @@ async function moveClaimedVideo({
 	toUserId: string;
 	token: string;
 	options: LoomReassignOptions;
+	originalKey: unknown;
 }): Promise<LoomReassignOutcome> {
 	const source = await listObjects(videoId, fromUserId, organizationId);
 	if (source.objects.length === 0)
@@ -269,10 +277,16 @@ async function moveClaimedVideo({
 	}
 	const copiedObjects = written.length;
 
+	const movedOriginalKey = getOwnerOriginalKey(originalKey, toUserId, videoId);
 	const updated = await db().transaction(async (tx) => {
 		const result = await tx
 			.update(videos)
 			.set({
+				...(movedOriginalKey
+					? {
+							metadata: sql`JSON_SET(${videos.metadata}, '$.loomDeferredProcessing.rawFileKey', ${movedOriginalKey})`,
+						}
+					: {}),
 				ownerId: User.UserId.make(toUserId),
 				folderId: options.folderId
 					? Folder.FolderId.make(options.folderId)
@@ -284,6 +298,8 @@ async function moveClaimedVideo({
 					eq(videos.id, id),
 					eq(videos.ownerId, User.UserId.make(fromUserId)),
 					sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomReassignClaim.token')) = ${token}`,
+					sql`COALESCE(JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.status')), '') <> 'started'`,
+					sql`JSON_UNQUOTE(JSON_EXTRACT(${videos.metadata}, '$.loomDeferredProcessing.rawFileKey')) <=> ${typeof originalKey === "string" ? originalKey : null}`,
 				),
 			);
 		if (getAffectedRows(result) !== 1) return false;
@@ -297,7 +313,11 @@ async function moveClaimedVideo({
 		await runPromise(
 			source.bucket.deleteObjects(written.map((key) => ({ Key: key }))),
 		).catch(() => undefined);
-		return { videoId, status: "failed", reason: "move claim was lost" };
+		return {
+			videoId,
+			status: "failed",
+			reason: "video changed during the move",
+		};
 	}
 
 	try {

apps/web/lib/loom-original-key.ts

@@ -0,0 +1,13 @@
+export const LOOM_REASSIGN_CLAIM_TTL_MS = 60 * 60 * 1000;
+
+export function getOwnerOriginalKey(
+	key: unknown,
+	ownerId: string,
+	videoId: string,
+) {
+	if (typeof key !== "string") return null;
+	const [, keyVideoId, ...rest] = key.split("/");
+	const fileName = rest.join("/");
+	if (keyVideoId !== videoId || !fileName) return null;
+	return `${ownerId}/${videoId}/${fileName}`;
+}