diff --git a/apps/web/__tests__/unit/loom-import-backfill.test.ts b/apps/web/__tests__/unit/loom-import-backfill.test.ts index d8ec6f3aaf..37e62a6ca1 100644 --- a/apps/web/__tests__/unit/loom-import-backfill.test.ts +++ b/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({ diff --git a/apps/web/__tests__/unit/loom-operator-tools.test.ts b/apps/web/__tests__/unit/loom-operator-tools.test.ts index 98738ee1b6..9acf167052 100644 --- a/apps/web/__tests__/unit/loom-operator-tools.test.ts +++ b/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, @@ -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); + }); +}); diff --git a/apps/web/lib/loom-deferred-processing.ts b/apps/web/lib/loom-deferred-processing.ts index 7a1ed19b7d..908aee4820 100644 --- a/apps/web/lib/loom-deferred-processing.ts +++ b/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}))`, ), ); diff --git a/apps/web/lib/loom-import-backfill.ts b/apps/web/lib/loom-import-backfill.ts index f7bee4584a..e9d267f608 100644 --- a/apps/web/lib/loom-import-backfill.ts +++ b/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) diff --git a/apps/web/lib/loom-import-reassign.ts b/apps/web/lib/loom-import-reassign.ts index 4e5c9a7a29..5e91a08e35 100644 --- a/apps/web/lib/loom-import-reassign.ts +++ b/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 { 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 { diff --git a/apps/web/lib/loom-original-key.ts b/apps/web/lib/loom-original-key.ts new file mode 100644 index 0000000000..def6f2d4dc --- /dev/null +++ b/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}`; +}