diff --git a/apps/web/__tests__/unit/loom-csv.test.ts b/apps/web/__tests__/unit/loom-csv.test.ts new file mode 100644 index 00000000000..90d8db8f19e --- /dev/null +++ b/apps/web/__tests__/unit/loom-csv.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from "vitest"; +import { parseConciergeLoomCsv } from "@/lib/loom-csv"; + +describe("concierge Loom CSV", () => { + it("keeps quoted workspace names and source row numbers", () => { + const rows = parseConciergeLoomCsv( + "\uFEFFuser_email,space_name,loom_video_url\r\n" + + "owner@example.com," + + JSON.stringify("Sales, Europe") + + ",https://www.loom.com/share/0123456789abcdef\r\n", + ); + expect(rows).toEqual([ + { + rowNumber: 2, + loomUrl: "https://www.loom.com/share/0123456789abcdef", + userEmail: "owner@example.com", + spaceName: "Sales, Europe", + }, + ]); + }); + + it("rejects a file without the mapping columns", () => { + expect(() => parseConciergeLoomCsv("url,email\nlink,person@x.com")).toThrow( + "loom_video_url and user_email", + ); + }); +}); diff --git a/apps/web/__tests__/unit/loom-import-ui.test.ts b/apps/web/__tests__/unit/loom-import-ui.test.ts index d92f8e90913..07fd55309d5 100644 --- a/apps/web/__tests__/unit/loom-import-ui.test.ts +++ b/apps/web/__tests__/unit/loom-import-ui.test.ts @@ -96,6 +96,9 @@ vi.mock("next/link", () => ({ }: React.PropsWithChildren>) => React.createElement("a", props, children), })); +vi.mock("@cap/env", () => ({ + buildEnv: { NEXT_PUBLIC_IS_CAP: "true" }, +})); vi.mock("@/actions/loom", () => ({ getLoomImportFolders: mocks.folders, importFromLoom: mocks.import, @@ -186,6 +189,11 @@ describe("Loom importer component", () => { it("shows the inherited subfolder and submits it then returns there", async () => { await render({ folderId: Folder.FolderId.make("child") }); await ready(); + expect( + Array.from(container.querySelectorAll("a")).find( + (link) => link.getAttribute("href") === "/dashboard/migrations/loom", + )?.textContent, + ).toContain("Want Cap to move your whole Loom workspace for you?"); expect( getByRole(container, "combobox", { name: "Import to" }).textContent, ).toContain("My Caps / Course / Live Calls - Two"); diff --git a/apps/web/__tests__/unit/loom-migration-state.test.ts b/apps/web/__tests__/unit/loom-migration-state.test.ts new file mode 100644 index 00000000000..87b1f90a544 --- /dev/null +++ b/apps/web/__tests__/unit/loom-migration-state.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from "vitest"; +import { validateOperatorMigrationUpdate } from "@/lib/loom-migration-state"; + +const base = { + currentStatus: "in_progress" as const, + nextStatus: "completed" as const, + message: "Migration finished", + verified: true, + expectedVideoCount: 12, + importedVideoCount: 12, +}; + +describe("Loom migration completion", () => { + it("requires a verified library and reconciled counts", () => { + expect(() => + validateOperatorMigrationUpdate({ ...base, verified: false }), + ).toThrow("Verify the migrated library"); + expect(() => + validateOperatorMigrationUpdate({ ...base, importedVideoCount: 11 }), + ).toThrow("Reconcile the expected video count"); + expect(() => + validateOperatorMigrationUpdate({ ...base, expectedVideoCount: null }), + ).toThrow("Enter the agreed source video count"); + }); + + it("requires a useful message when Cap asks for information", () => { + expect(() => + validateOperatorMigrationUpdate({ + ...base, + nextStatus: "needs_information", + message: " ", + }), + ).toThrow("Tell the customer"); + }); + + it("accepts completion after verification", () => { + expect(validateOperatorMigrationUpdate(base)).toBe("Migration finished"); + }); +}); diff --git a/apps/web/actions/loom-concierge.tsx b/apps/web/actions/loom-concierge.tsx new file mode 100644 index 00000000000..007584ce806 --- /dev/null +++ b/apps/web/actions/loom-concierge.tsx @@ -0,0 +1,345 @@ +"use server"; + +import { db } from "@cap/database"; +import { sendEmail } from "@cap/database/emails/config"; +import { + LoomMigrationRequestEmail, + LoomMigrationStatusEmail, +} from "@cap/database/emails/loom-migration"; +import { nanoId } from "@cap/database/helpers"; +import { + loomMigrationImports, + loomMigrationRequests, + organizations, + users, + videoUploads, +} from "@cap/database/schema"; +import { serverEnv } from "@cap/env"; +import type { Organisation } from "@cap/web-domain"; +import { and, eq, inArray, isNotNull, lte, or } from "drizzle-orm"; +import { revalidatePath } from "next/cache"; +import { + getCustomerMigrationRequests, + getOperatorMigrationQueue, + requireCustomerMigrationAccess, + requireMigrationOperator, + requireProCustomerMigrationAccess, +} from "@/lib/loom-concierge"; +import { + LOOM_MIGRATION_STATUS_LABELS, + type LoomMigrationStatus, + normalizeMigrationText, + validateOperatorMigrationUpdate, +} from "@/lib/loom-migration-state"; +import { isOrganizationOwnerPro } from "@/lib/org-pro"; + +const SUPPORT_EMAIL = "hello@cap.so"; + +function affectedRows(result: unknown) { + if (Array.isArray(result)) { + return ( + (result[0] as { affectedRows?: number } | undefined)?.affectedRows ?? 0 + ); + } + return (result as { affectedRows?: number } | undefined)?.affectedRows ?? 0; +} + +function refreshMigrationPages() { + revalidatePath("/dashboard/migrations/loom"); + revalidatePath("/dashboard/admin/loom-migrations"); +} + +async function findMigration(requestId: string) { + if (typeof requestId !== "string" || !/^[A-Za-z0-9_-]{15}$/.test(requestId)) { + throw new Error("Invalid migration request."); + } + const [request] = await db() + .select() + .from(loomMigrationRequests) + .where(eq(loomMigrationRequests.id, requestId)) + .limit(1); + if (!request) throw new Error("Migration request not found."); + return request; +} + +export async function getLoomMigrationDashboard( + organizationId: Organisation.OrganisationId, +) { + await requireCustomerMigrationAccess(organizationId); + const [isPro, requests] = await Promise.all([ + isOrganizationOwnerPro(organizationId), + getCustomerMigrationRequests(organizationId), + ]); + return { isPro, requests }; +} + +export async function requestLoomMigration({ + organizationId, + workspaceName, + note, +}: { + organizationId: Organisation.OrganisationId; + workspaceName: string; + note: string; +}) { + const user = await requireProCustomerMigrationAccess(organizationId); + const normalizedWorkspaceName = normalizeMigrationText( + workspaceName, + "Workspace name", + 255, + ); + const normalizedNote = normalizeMigrationText(note, "Note", 2000); + const [existing] = await db() + .select({ id: loomMigrationRequests.id }) + .from(loomMigrationRequests) + .where(eq(loomMigrationRequests.activeOrganizationId, organizationId)) + .limit(1); + if (existing) return { id: existing.id, alreadyRequested: true }; + + const id = nanoId(); + try { + await db() + .insert(loomMigrationRequests) + .values({ + id, + organizationId, + activeOrganizationId: organizationId, + requestedByUserId: user.id, + workspaceName: normalizedWorkspaceName || null, + customerNote: normalizedNote || null, + }); + } catch (error) { + const [racedRequest] = await db() + .select({ id: loomMigrationRequests.id }) + .from(loomMigrationRequests) + .where(eq(loomMigrationRequests.activeOrganizationId, organizationId)) + .limit(1); + if (racedRequest) { + return { id: racedRequest.id, alreadyRequested: true }; + } + throw error; + } + + const [organization] = await db() + .select({ name: organizations.name }) + .from(organizations) + .where(eq(organizations.id, organizationId)) + .limit(1); + try { + const delivery = await sendEmail({ + email: SUPPORT_EMAIL, + subject: `New Loom migration request: ${organization?.name ?? "Cap workspace"}`, + react: ( + + ), + idempotencyKey: `loom-migration-request-${id}`, + }); + if (delivery?.error) { + console.error( + "Failed to send Loom migration request notice", + delivery.error, + ); + } + } catch (error) { + console.error("Failed to send Loom migration request notice", error); + } + + refreshMigrationPages(); + return { id, alreadyRequested: false }; +} + +export async function confirmLoomMigrationInvite(requestId: string) { + const request = await findMigration(requestId); + await requireCustomerMigrationAccess(request.organizationId); + if (request.status === "completed") { + throw new Error("This migration is already complete."); + } + if (request.invitedAt) return; + const result = await db() + .update(loomMigrationRequests) + .set({ invitedAt: new Date(), updatedAt: new Date() }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + ), + ); + if (affectedRows(result) === 0) { + throw new Error("The migration request changed. Refresh and try again."); + } + refreshMigrationPages(); +} + +export async function answerLoomMigrationQuestion({ + requestId, + answer, +}: { + requestId: string; + answer: string; +}) { + const request = await findMigration(requestId); + await requireCustomerMigrationAccess(request.organizationId); + const normalizedAnswer = normalizeMigrationText(answer, "Answer", 2000); + if (!normalizedAnswer) + throw new Error("Enter the information Cap asked for."); + const result = await db() + .update(loomMigrationRequests) + .set({ + customerReply: normalizedAnswer, + status: "pending", + updatedAt: new Date(), + }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + eq(loomMigrationRequests.status, "needs_information"), + ), + ); + if (affectedRows(result) === 0) { + throw new Error("The migration request changed. Refresh and try again."); + } + refreshMigrationPages(); +} + +export async function getLoomMigrationOperatorQueue() { + return getOperatorMigrationQueue(); +} + +export async function updateLoomMigrationStatus({ + requestId, + nextStatus, + message, + verified, + expectedVideoCount, + importedVideoCount, +}: { + requestId: string; + nextStatus: LoomMigrationStatus; + message: string; + verified: boolean; + expectedVideoCount: number | null; + importedVideoCount: number; +}) { + const operator = await requireMigrationOperator(); + const request = await findMigration(requestId); + const normalizedMessage = validateOperatorMigrationUpdate({ + currentStatus: request.status, + nextStatus, + message, + verified, + expectedVideoCount, + importedVideoCount, + }); + if ( + nextStatus === request.status && + (normalizedMessage || null) === request.capMessage && + expectedVideoCount === request.expectedVideoCount && + importedVideoCount === request.importedVideoCount + ) { + return; + } + if (nextStatus === "completed") { + const [unfinishedImport] = await db() + .select({ videoId: videoUploads.videoId }) + .from(loomMigrationImports) + .innerJoin( + videoUploads, + eq(videoUploads.videoId, loomMigrationImports.videoId), + ) + .where( + and( + eq(loomMigrationImports.requestId, requestId), + inArray(videoUploads.phase, [ + "uploading", + "processing", + "generating_thumbnail", + "error", + ]), + ), + ) + .limit(1); + if (unfinishedImport) { + throw new Error( + "Loom imports are still processing or have errors. Resolve them before completion.", + ); + } + } + const [organization] = await db() + .select({ name: organizations.name }) + .from(organizations) + .where(eq(organizations.id, request.organizationId)) + .limit(1); + const result = await db() + .update(loomMigrationRequests) + .set({ + status: nextStatus, + capMessage: normalizedMessage || null, + expectedVideoCount, + importedVideoCount, + activeImportCount: nextStatus === "completed" ? 0 : undefined, + activeImportLeaseToken: nextStatus === "completed" ? null : undefined, + activeImportLeaseUntil: nextStatus === "completed" ? null : undefined, + activeOrganizationId: + nextStatus === "completed" ? null : request.organizationId, + completedAt: nextStatus === "completed" ? new Date() : null, + lastOperatorUserId: operator.id, + lastOperatorAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + eq(loomMigrationRequests.status, request.status), + isNotNull(loomMigrationRequests.activeOrganizationId), + nextStatus === "completed" + ? or( + eq(loomMigrationRequests.activeImportCount, 0), + lte(loomMigrationRequests.activeImportLeaseUntil, new Date()), + ) + : undefined, + ), + ); + if (affectedRows(result) === 0) { + throw new Error( + "The migration request changed or imports are starting. Refresh and try again.", + ); + } + + const [requester] = await db() + .select({ email: users.email }) + .from(users) + .where(eq(users.id, request.requestedByUserId)) + .limit(1); + if (requester) { + try { + const delivery = await sendEmail({ + email: requester.email, + subject: `Loom migration: ${LOOM_MIGRATION_STATUS_LABELS[nextStatus]}`, + react: ( + + ), + replyTo: SUPPORT_EMAIL, + idempotencyKey: `loom-migration-status-${requestId}-${Date.now()}`, + }); + if (delivery?.error) { + console.error( + "Failed to send Loom migration status notice", + delivery.error, + ); + } + } catch (error) { + console.error("Failed to send Loom migration status notice", error); + } + } + refreshMigrationPages(); +} diff --git a/apps/web/actions/loom.ts b/apps/web/actions/loom.ts index b9e03dcecf9..a6ac424ab98 100644 --- a/apps/web/actions/loom.ts +++ b/apps/web/actions/loom.ts @@ -7,7 +7,10 @@ import { nanoId } from "@cap/database/helpers"; import { folders, importedVideos, + loomMigrationImports, + loomMigrationRequests, organizationMembers, + organizations, sharedVideos, spaceMembers, spaces, @@ -23,11 +26,12 @@ import { type Organisation, Space, SpaceMemberId, + type Storage as StorageDomain, type User, Video, } from "@cap/web-domain"; -import { and, asc, eq, isNull } from "drizzle-orm"; -import { Option } from "effect"; +import { and, asc, eq, isNotNull, isNull, sql } from "drizzle-orm"; +import { Effect, Option, Schedule } from "effect"; import { revalidatePath } from "next/cache"; import { start } from "workflow/api"; import { @@ -36,10 +40,19 @@ import { requireOrganizationSettingsManager, } from "@/actions/organization/authorization"; import { requireSpaceManager } from "@/actions/organization/space-authorization"; +import { + releaseConciergeImport, + renewConciergeImport, + requireMigrationOperator, + reserveConciergeImport, +} from "@/lib/loom-concierge"; import type { LoomImportDestination } from "@/lib/loom-import-destination"; +import { isOrganizationOwnerPro } from "@/lib/org-pro"; import { provisionOrganizationInvitee } from "@/lib/organization-provisioning"; import { canManageOrganizationSettings } from "@/lib/permissions/roles"; import { runPromise } from "@/lib/server"; +import { startVideoProcessingWorkflow } from "@/lib/video-processing"; +import { decodeStorageVideo } from "@/lib/video-storage"; import { importLoomVideoWorkflow } from "@/workflows/import-loom-video"; interface LoomUrlResponse { @@ -96,22 +109,26 @@ const MAX_LOOM_SPACE_NAME_LENGTH = 255; const LOOM_CSV_LIMIT_ERROR = `CSV imports are limited to ${MAX_LOOM_CSV_ROWS} rows at a time. Contact support to raise this limit.`; const LOOM_CSV_PERMISSION_ERROR = "Only organization admins and owners can import Loom videos from a CSV."; +const LOOM_VIDEO_ID_PATTERN = /^[A-Za-z0-9_-]{10,64}$/; function extractLoomVideoId(url: string): string | null { try { const parsed = new URL(url); - if (!parsed.hostname.includes("loom.com")) { + if ( + parsed.protocol !== "https:" || + (parsed.hostname !== "loom.com" && !parsed.hostname.endsWith(".loom.com")) + ) { return null; } const pathParts = parsed.pathname.split("/").filter(Boolean); const id = pathParts[pathParts.length - 1] ?? null; - if (!id || id.length < 10) { + if (!id || !LOOM_VIDEO_ID_PATTERN.test(id)) { return null; } - return id.split("?")[0] ?? null; + return id; } catch { return null; } @@ -119,7 +136,7 @@ function extractLoomVideoId(url: string): string | null { async function fetchLoomEndpoint( videoId: string, - endpoint: string, + endpoint: "transcoded-url" | "raw-url", includeBody = true, ): Promise { try { @@ -138,7 +155,7 @@ async function fetchLoomEndpoint( } const response = await fetch( - `https://www.loom.com/api/campaigns/sessions/${videoId}/${endpoint}`, + `https://www.loom.com/api/campaigns/sessions/${encodeURIComponent(videoId)}/${endpoint}`, options, ); @@ -200,7 +217,10 @@ function isDirectMp4Url(url: string): boolean { } async function getLoomDownloadUrl(loomVideoId: string): Promise { - const requestVariants: Array<{ endpoint: string; includeBody: boolean }> = [ + const requestVariants: Array<{ + endpoint: "transcoded-url" | "raw-url"; + includeBody: boolean; + }> = [ { endpoint: "transcoded-url", includeBody: true }, { endpoint: "raw-url", includeBody: true }, { endpoint: "transcoded-url", includeBody: false }, @@ -226,7 +246,7 @@ async function fetchLoomOEmbed( ): Promise<{ duration?: number; width?: number; height?: number } | null> { try { const response = await fetch( - `https://www.loom.com/v1/oembed?url=https://www.loom.com/share/${loomVideoId}`, + `https://www.loom.com/v1/oembed?url=${encodeURIComponent(`https://www.loom.com/share/${loomVideoId}`)}`, { headers: { Accept: "application/json" } }, ); if (!response.ok) return null; @@ -300,11 +320,15 @@ async function importLoomVideoForOwner({ orgId, ownerId, destination = {}, + migrationRequestId, + migrationLeaseToken, }: { loomUrl: string; orgId: Organisation.OrganisationId; ownerId: User.UserId; destination?: LoomImportDestination; + migrationRequestId?: string; + migrationLeaseToken?: string; }): Promise { const loomVideoId = extractLoomVideoId(loomUrl.trim()); if (!loomVideoId) { @@ -395,6 +419,27 @@ async function importLoomVideoForOwner({ `Loom Import - ${new Date().toLocaleDateString("en-US", { day: "numeric", month: "long", year: "numeric" })}`; await db().transaction(async (tx) => { + if (migrationRequestId) { + const [reservation] = await tx + .select({ + activeOrganizationId: loomMigrationRequests.activeOrganizationId, + leaseToken: loomMigrationRequests.activeImportLeaseToken, + leaseUntil: loomMigrationRequests.activeImportLeaseUntil, + }) + .from(loomMigrationRequests) + .where(eq(loomMigrationRequests.id, migrationRequestId)) + .limit(1) + .for("update"); + if ( + !migrationLeaseToken || + !reservation?.activeOrganizationId || + reservation.leaseToken !== migrationLeaseToken || + !reservation.leaseUntil || + reservation.leaseUntil <= new Date() + ) { + throw new Error("The concierge import expired. Retry this video."); + } + } await tx.insert(videos).values({ id: videoId, name, @@ -423,6 +468,19 @@ async function importLoomVideoForOwner({ source: "loom", sourceId: loomVideoId, }); + if (migrationRequestId) { + await tx.insert(loomMigrationImports).values({ + videoId, + requestId: migrationRequestId, + }); + await tx + .update(loomMigrationRequests) + .set({ + queuedVideoCount: sql`${loomMigrationRequests.queuedVideoCount} + 1`, + updatedAt: new Date(), + }) + .where(eq(loomMigrationRequests.id, migrationRequestId)); + } if (destination.spaceId === orgId) { await tx.insert(sharedVideos).values({ @@ -770,6 +828,22 @@ export async function importFromLoomCsv({ }; } + return processLoomCsvRows({ rows, orgId, actorId: user.id }); +} + +async function processLoomCsvRows({ + rows, + orgId, + actorId, + migrationRequestId, + migrationLeaseToken, +}: { + rows: LoomCsvImportRow[]; + orgId: Organisation.OrganisationId; + actorId: User.UserId; + migrationRequestId?: string; + migrationLeaseToken?: string; +}): Promise { const inputRows = Array.isArray(rows) ? rows : []; const normalizedRows = inputRows .map((row, index) => ({ @@ -814,6 +888,12 @@ export async function importFromLoomCsv({ const touchedSpaceIds = new Set(); for (const row of normalizedRows) { + if (migrationRequestId) { + if (!migrationLeaseToken) { + throw new Error("The concierge import is missing its reservation."); + } + await renewConciergeImport(migrationRequestId, migrationLeaseToken); + } if (!row.loomUrl) { results.push({ rowNumber: row.rowNumber, @@ -854,7 +934,7 @@ export async function importFromLoomCsv({ const provisionedMember = await provisionOrganizationInvitee({ organizationId: orgId, email: row.userEmail, - invitedByUserId: user.id, + invitedByUserId: actorId, role: "member", }); member = { @@ -878,6 +958,8 @@ export async function importFromLoomCsv({ loomUrl: row.loomUrl, orgId, ownerId: member.userId, + migrationRequestId, + migrationLeaseToken, }); let spaceName = row.spaceName || undefined; @@ -886,14 +968,14 @@ export async function importFromLoomCsv({ try { const space = await getOrCreateImportSpace({ orgId, - createdById: user.id, + createdById: actorId, name: row.spaceName, spaceCache, }); await addImportedVideoToSpace({ videoId: result.videoId, spaceId: space.id, - addedById: user.id, + addedById: actorId, }); await addImportOwnerToSpace({ spaceId: space.id, @@ -944,3 +1026,420 @@ export async function importFromLoomCsv({ error: importedCount > 0 ? undefined : "No Loom videos were imported.", }; } + +export async function importFromLoomCsvForConcierge({ + requestId, + rows, +}: { + requestId: string; + rows: LoomCsvImportRow[]; +}): Promise { + const operator = await requireMigrationOperator(); + if (typeof requestId !== "string" || !/^[A-Za-z0-9_-]{15}$/.test(requestId)) { + throw new Error("Invalid migration request."); + } + const [request] = await db() + .select({ + organizationId: loomMigrationRequests.organizationId, + status: loomMigrationRequests.status, + ownerId: organizations.ownerId, + }) + .from(loomMigrationRequests) + .innerJoin( + organizations, + eq(organizations.id, loomMigrationRequests.organizationId), + ) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + isNull(organizations.tombstoneAt), + ), + ) + .limit(1); + if (!request || request.status === "completed") { + throw new Error("The migration request is no longer active."); + } + if (!(await isOrganizationOwnerPro(request.organizationId))) { + throw new Error("The destination workspace needs Cap Pro."); + } + if (!Array.isArray(rows) || rows.length > 10) { + throw new Error("Concierge CSV imports run in batches of 10 videos."); + } + const leaseToken = await reserveConciergeImport(requestId, operator.id); + try { + const result = await processLoomCsvRows({ + rows, + orgId: request.organizationId, + actorId: request.ownerId, + migrationRequestId: requestId, + migrationLeaseToken: leaseToken, + }); + revalidatePath("/dashboard/migrations/loom"); + revalidatePath("/dashboard/admin/loom-migrations"); + return result; + } finally { + await releaseConciergeImport(requestId, leaseToken); + } +} + +export async function createConciergeLoomFileUpload({ + requestId, + loomUrl, + userEmail, + spaceName, + videoTitle, +}: { + requestId: string; + loomUrl: string; + userEmail: string; + spaceName: string; + videoTitle: string; +}): Promise<{ + videoId: Video.VideoId; + uploadTarget: StorageDomain.UploadTarget; +}> { + const operator = await requireMigrationOperator(); + if (typeof requestId !== "string" || !/^[A-Za-z0-9_-]{15}$/.test(requestId)) { + throw new Error("Invalid migration request."); + } + const loomVideoId = extractLoomVideoId(loomUrl.trim()); + if (!loomVideoId) throw new Error("Enter a valid Loom video URL."); + const normalizedEmail = normalizeImportEmail(userEmail); + if (!isValidImportEmail(normalizedEmail)) { + throw new Error("Enter a valid destination owner email."); + } + const normalizedSpaceName = normalizeImportSpaceName(spaceName); + if (!isValidImportSpaceName(normalizedSpaceName)) { + throw new Error("Space name is too long."); + } + const normalizedTitle = videoTitle.trim(); + if (!normalizedTitle || normalizedTitle.length > 255) { + throw new Error("Enter a video title of 255 characters or fewer."); + } + const [request] = await db() + .select({ + organizationId: loomMigrationRequests.organizationId, + ownerId: organizations.ownerId, + }) + .from(loomMigrationRequests) + .innerJoin( + organizations, + eq(organizations.id, loomMigrationRequests.organizationId), + ) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + isNull(organizations.tombstoneAt), + ), + ) + .limit(1); + if (!request) throw new Error("The migration request is no longer active."); + if (!(await isOrganizationOwnerPro(request.organizationId))) { + throw new Error("The destination workspace needs Cap Pro."); + } + const leaseToken = await reserveConciergeImport(requestId, operator.id); + try { + const [existing] = await db() + .select({ + id: importedVideos.id, + ownerId: videos.ownerId, + ownerEmail: users.email, + bucket: videos.bucket, + storageIntegrationId: videos.storageIntegrationId, + rawFileKey: videoUploads.rawFileKey, + phase: videoUploads.phase, + processingMessage: videoUploads.processingMessage, + }) + .from(importedVideos) + .leftJoin( + videos, + and( + eq(videos.id, importedVideos.id), + eq(videos.orgId, importedVideos.orgId), + ), + ) + .leftJoin(users, eq(users.id, videos.ownerId)) + .leftJoin(videoUploads, eq(videoUploads.videoId, importedVideos.id)) + .where( + and( + eq(importedVideos.orgId, request.organizationId), + eq(importedVideos.source, "loom"), + eq(importedVideos.sourceId, loomVideoId), + ), + ) + .limit(1); + if (existing) { + if ( + !existing.ownerId || + !existing.ownerEmail || + normalizeImportEmail(existing.ownerEmail) !== normalizedEmail || + !existing.rawFileKey || + (existing.phase !== "error" && + (existing.phase !== "uploading" || + existing.processingMessage !== "Uploading Loom video...")) + ) { + throw new Error( + "This Loom video is already in the Cap import inventory.", + ); + } + const retryUpload = await Storage.createUploadTargetForUser( + existing.ownerId, + existing.rawFileKey, + { + contentType: "video/mp4", + videoTitle: normalizedTitle, + method: "put", + fields: { "x-amz-meta-userid": existing.ownerId }, + }, + request.organizationId, + ).pipe(runPromise); + if ( + Option.getOrNull(retryUpload.bucketId) !== existing.bucket || + Option.getOrNull(retryUpload.storageIntegrationId) !== + existing.storageIntegrationId + ) { + throw new Error( + "The Cap storage destination changed. Review this upload before retrying.", + ); + } + const [mapped] = await db() + .select({ requestId: loomMigrationImports.requestId }) + .from(loomMigrationImports) + .where( + eq(loomMigrationImports.videoId, Video.VideoId.make(existing.id)), + ) + .limit(1); + if (mapped && mapped.requestId !== requestId) { + throw new Error( + "This Loom import belongs to another migration request.", + ); + } + if (!mapped) { + await db().transaction(async (tx) => { + const [reservation] = await tx + .select({ + activeOrganizationId: loomMigrationRequests.activeOrganizationId, + leaseToken: loomMigrationRequests.activeImportLeaseToken, + leaseUntil: loomMigrationRequests.activeImportLeaseUntil, + }) + .from(loomMigrationRequests) + .where(eq(loomMigrationRequests.id, requestId)) + .limit(1) + .for("update"); + if ( + !reservation?.activeOrganizationId || + reservation.leaseToken !== leaseToken || + !reservation.leaseUntil || + reservation.leaseUntil <= new Date() + ) { + throw new Error("The concierge import expired. Retry this video."); + } + await tx.insert(loomMigrationImports).values({ + videoId: Video.VideoId.make(existing.id), + requestId, + }); + await tx + .update(loomMigrationRequests) + .set({ + queuedVideoCount: sql`${loomMigrationRequests.queuedVideoCount} + 1`, + updatedAt: new Date(), + }) + .where(eq(loomMigrationRequests.id, requestId)); + }); + } + return { + videoId: Video.VideoId.make(existing.id), + uploadTarget: retryUpload.upload, + }; + } + let member = await getOrganizationMemberByEmail( + request.organizationId, + normalizedEmail, + ); + if (!member) { + const provisioned = await provisionOrganizationInvitee({ + organizationId: request.organizationId, + email: normalizedEmail, + invitedByUserId: request.ownerId, + role: "member", + }); + member = { userId: provisioned.userId, email: normalizedEmail }; + } + const space = normalizedSpaceName + ? await getOrCreateImportSpace({ + orgId: request.organizationId, + createdById: request.ownerId, + name: normalizedSpaceName, + spaceCache: new Map(), + }) + : null; + if (space) { + await addImportOwnerToSpace({ spaceId: space.id, userId: member.userId }); + } + const videoId = Video.VideoId.make(nanoId()); + const rawFileKey = `${member.userId}/${videoId}/raw-upload.mp4`; + const upload = await Storage.createUploadTargetForUser( + member.userId, + rawFileKey, + { + contentType: "video/mp4", + videoTitle: normalizedTitle, + method: "put", + fields: { "x-amz-meta-userid": member.userId }, + }, + request.organizationId, + ).pipe(runPromise); + await db().transaction(async (tx) => { + const [reservation] = await tx + .select({ + activeOrganizationId: loomMigrationRequests.activeOrganizationId, + leaseToken: loomMigrationRequests.activeImportLeaseToken, + leaseUntil: loomMigrationRequests.activeImportLeaseUntil, + }) + .from(loomMigrationRequests) + .where(eq(loomMigrationRequests.id, requestId)) + .limit(1) + .for("update"); + if ( + !reservation?.activeOrganizationId || + reservation.leaseToken !== leaseToken || + !reservation.leaseUntil || + reservation.leaseUntil <= new Date() + ) { + throw new Error("The concierge import expired. Retry this video."); + } + await tx.insert(videos).values({ + id: videoId, + name: normalizedTitle, + ownerId: member.userId, + orgId: request.organizationId, + source: { type: "webMP4" as const }, + bucket: Option.getOrNull(upload.bucketId), + storageIntegrationId: Option.getOrNull(upload.storageIntegrationId), + public: serverEnv().CAP_VIDEOS_DEFAULT_PUBLIC, + }); + await tx.insert(videoUploads).values({ + videoId, + mode: "singlepart", + phase: "uploading", + processingProgress: 0, + processingMessage: "Uploading Loom video...", + rawFileKey, + }); + await tx.insert(importedVideos).values({ + id: videoId, + orgId: request.organizationId, + source: "loom", + sourceId: loomVideoId, + }); + await tx.insert(loomMigrationImports).values({ videoId, requestId }); + await tx + .update(loomMigrationRequests) + .set({ + queuedVideoCount: sql`${loomMigrationRequests.queuedVideoCount} + 1`, + updatedAt: new Date(), + }) + .where(eq(loomMigrationRequests.id, requestId)); + if (space) { + await tx.insert(spaceVideos).values({ + id: nanoId(), + videoId, + spaceId: space.id, + addedById: request.ownerId, + }); + } + }); + revalidatePath("/dashboard/admin/loom-migrations"); + return { videoId, uploadTarget: upload.upload }; + } finally { + await releaseConciergeImport(requestId, leaseToken); + } +} + +export async function finishConciergeLoomFileUpload({ + requestId, + videoId, +}: { + requestId: string; + videoId: Video.VideoId; +}) { + const operator = await requireMigrationOperator(); + if (typeof requestId !== "string" || !/^[A-Za-z0-9_-]{15}$/.test(requestId)) { + throw new Error("Invalid migration request."); + } + const leaseToken = await reserveConciergeImport(requestId, operator.id); + try { + const [record] = await db() + .select({ + video: videos, + rawFileKey: videoUploads.rawFileKey, + phase: videoUploads.phase, + processingMessage: videoUploads.processingMessage, + }) + .from(videos) + .innerJoin( + importedVideos, + and( + eq(importedVideos.id, videos.id), + eq(importedVideos.orgId, videos.orgId), + ), + ) + .innerJoin(videoUploads, eq(videoUploads.videoId, videos.id)) + .innerJoin( + loomMigrationImports, + eq(loomMigrationImports.videoId, videos.id), + ) + .innerJoin( + loomMigrationRequests, + and( + eq(loomMigrationRequests.id, loomMigrationImports.requestId), + eq(loomMigrationRequests.organizationId, importedVideos.orgId), + ), + ) + .where( + and( + eq(videos.id, videoId), + eq(loomMigrationRequests.id, requestId), + eq(importedVideos.source, "loom"), + isNotNull(loomMigrationRequests.activeOrganizationId), + ), + ) + .limit(1); + if ( + !record || + !record.rawFileKey || + (record.phase !== "error" && + (record.phase !== "uploading" || + record.processingMessage !== "Uploading Loom video...")) + ) { + throw new Error("This Loom upload is no longer available."); + } + const [bucket] = await Storage.getAccessForVideo( + decodeStorageVideo(record.video), + ).pipe(runPromise); + const head = await bucket.headObject(record.rawFileKey).pipe( + Effect.retry({ + times: 3, + schedule: Schedule.exponential("100 millis"), + }), + runPromise, + ); + if ((head.ContentLength ?? 0) <= 0) + throw new Error("The uploaded file is empty."); + const status = await startVideoProcessingWorkflow({ + videoId, + userId: record.video.ownerId, + rawFileKey: record.rawFileKey, + bucketId: record.video.bucket, + processingMessage: "Processing Loom video...", + startFailureMessage: "Loom file processing could not start.", + }); + revalidatePath("/dashboard/migrations/loom"); + revalidatePath("/dashboard/admin/loom-migrations"); + return { status }; + } finally { + await releaseConciergeImport(requestId, leaseToken); + } +} diff --git a/apps/web/app/(org)/dashboard/_components/Navbar/Items.tsx b/apps/web/app/(org)/dashboard/_components/Navbar/Items.tsx index e91138d70e5..d5d0ebfe639 100644 --- a/apps/web/app/(org)/dashboard/_components/Navbar/Items.tsx +++ b/apps/web/app/(org)/dashboard/_components/Navbar/Items.tsx @@ -46,6 +46,7 @@ import { loomImportDestinationFromPathname, loomImportPageHref, } from "@/lib/loom-import-destination"; +import { MESSENGER_ADMIN_EMAIL } from "@/lib/messenger/constants"; import { canViewOrganizationSettings, getEffectiveOrganizationRole, @@ -78,6 +79,9 @@ const AdminNavItems = ({ toggleMobileNav }: Props) => { const showDeveloperDashboard = buildEnv.NEXT_PUBLIC_IS_CAP && DEVELOPER_DASHBOARD_ALLOWED_EMAILS.includes(user.email); + const showMigrationQueue = + buildEnv.NEXT_PUBLIC_IS_CAP && + user.email.toLowerCase() === MESSENGER_ADMIN_EMAIL; const manageNavigation = [ { @@ -107,6 +111,17 @@ const AdminNavItems = ({ toggleMobileNav }: Props) => { icon: , subNav: [], }, + ...(buildEnv.NEXT_PUBLIC_IS_CAP + ? [ + { + name: "Loom Migration", + href: "/dashboard/migrations/loom", + adminOnly: true, + icon: , + subNav: [], + }, + ] + : []), { name: "Organization Settings", href: `/dashboard/settings/organization`, @@ -127,6 +142,16 @@ const AdminNavItems = ({ toggleMobileNav }: Props) => { }, ] : []), + ...(showMigrationQueue + ? [ + { + name: "Migration Queue", + href: "/dashboard/admin/loom-migrations", + icon: , + subNav: [], + }, + ] + : []), ]; const [dialogOpen, setDialogOpen] = useState(false); diff --git a/apps/web/app/(org)/dashboard/admin/loom-migrations/LoomMigrationQueue.tsx b/apps/web/app/(org)/dashboard/admin/loom-migrations/LoomMigrationQueue.tsx new file mode 100644 index 00000000000..34f6f5e2c42 --- /dev/null +++ b/apps/web/app/(org)/dashboard/admin/loom-migrations/LoomMigrationQueue.tsx @@ -0,0 +1,529 @@ +"use client"; + +import { Button } from "@cap/ui"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { LoaderCircle } from "lucide-react"; +import { useEffect, useState } from "react"; +import { toast } from "sonner"; +import { + createConciergeLoomFileUpload, + finishConciergeLoomFileUpload, + importFromLoomCsvForConcierge, + type LoomCsvImportRow, + type LoomCsvImportRowResult, +} from "@/actions/loom"; +import { + getLoomMigrationOperatorQueue, + updateLoomMigrationStatus, +} from "@/actions/loom-concierge"; +import type { OperatorLoomMigrationView } from "@/lib/loom-concierge"; +import { parseConciergeLoomCsv } from "@/lib/loom-csv"; +import { + LOOM_MIGRATION_STATUS_LABELS, + type LoomMigrationStatus, +} from "@/lib/loom-migration-state"; +import { uploadWithTarget } from "@/utils/upload-target"; + +const BATCH_SIZE = 10; +const BATCH_DELAY_MS = 1500; + +function delay(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function QueueItem({ request }: { request: OperatorLoomMigrationView }) { + const queryClient = useQueryClient(); + const [nextStatus, setNextStatus] = useState( + request.status, + ); + const [message, setMessage] = useState(request.capMessage ?? ""); + const [expectedVideoCount, setExpectedVideoCount] = useState( + request.expectedVideoCount?.toString() ?? "", + ); + const [importedVideoCount, setImportedVideoCount] = useState( + request.importedVideoCount.toString(), + ); + const [verified, setVerified] = useState(false); + const [csvRows, setCsvRows] = useState([]); + const [processedRows, setProcessedRows] = useState(0); + const [startedRows, setStartedRows] = useState(0); + const [failedRows, setFailedRows] = useState([]); + const [fileLoomUrl, setFileLoomUrl] = useState(""); + const [fileOwnerEmail, setFileOwnerEmail] = useState(""); + const [fileSpaceName, setFileSpaceName] = useState(""); + const [fileTitle, setFileTitle] = useState(""); + const [file, setFile] = useState(null); + const [uploadPercent, setUploadPercent] = useState(0); + + useEffect(() => { + setNextStatus(request.status); + setMessage(request.capMessage ?? ""); + setExpectedVideoCount(request.expectedVideoCount?.toString() ?? ""); + setImportedVideoCount(request.importedVideoCount.toString()); + setVerified(false); + }, [ + request.status, + request.capMessage, + request.expectedVideoCount, + request.importedVideoCount, + ]); + + const refresh = () => + queryClient.invalidateQueries({ queryKey: ["loom-migration-queue"] }); + const statusMutation = useMutation({ + mutationFn: () => + updateLoomMigrationStatus({ + requestId: request.id, + nextStatus, + message, + verified, + expectedVideoCount: + expectedVideoCount.trim() === "" ? null : Number(expectedVideoCount), + importedVideoCount: Number(importedVideoCount), + }), + onSuccess: async () => { + await refresh(); + toast.success("Migration status updated."); + }, + onError: (error) => toast.error(error.message), + }); + const importMutation = useMutation({ + mutationFn: async () => { + if (csvRows.length === 0) throw new Error("Choose a Loom CSV first."); + setProcessedRows(0); + setStartedRows(0); + setFailedRows([]); + let started = 0; + const failures: LoomCsvImportRowResult[] = []; + for (let index = 0; index < csvRows.length; index += BATCH_SIZE) { + const batch = csvRows.slice(index, index + BATCH_SIZE); + const result = await importFromLoomCsvForConcierge({ + requestId: request.id, + rows: batch, + }); + started += result.importedCount; + failures.push(...result.results.filter((row) => !row.success)); + setProcessedRows(index + batch.length); + setStartedRows(started); + setFailedRows([...failures]); + if (index + BATCH_SIZE < csvRows.length) await delay(BATCH_DELAY_MS); + } + return { started, failures }; + }, + onSuccess: async (result) => { + await refresh(); + toast.success( + `${result.started} imports started. ${result.failures.length} rows need review. Wait for processing and reconcile the library before completion.`, + ); + }, + onError: async (error) => { + await refresh(); + toast.error( + `${error.message} Check the rows already started before retrying.`, + ); + }, + }); + const fileMutation = useMutation({ + mutationFn: async () => { + if (!file || !fileLoomUrl.trim() || !fileOwnerEmail.trim()) { + throw new Error("Choose an MP4 file, Loom URL, and destination owner."); + } + if (!file.name.toLowerCase().endsWith(".mp4")) { + throw new Error("Choose an MP4 downloaded from Loom."); + } + if (file.size === 0) throw new Error("The MP4 file is empty."); + const prepared = await createConciergeLoomFileUpload({ + requestId: request.id, + loomUrl: fileLoomUrl, + userEmail: fileOwnerEmail, + spaceName: fileSpaceName, + videoTitle: fileTitle.trim() || file.name.replace(/\.mp4$/i, ""), + }); + setUploadPercent(0); + await uploadWithTarget({ + target: prepared.uploadTarget, + body: file, + fileName: file.name, + onProgress: ({ loaded, total }) => { + if (total > 0) setUploadPercent(Math.round((loaded / total) * 100)); + }, + }); + return finishConciergeLoomFileUpload({ + requestId: request.id, + videoId: prepared.videoId, + }); + }, + onSuccess: async (result) => { + await refresh(); + setFile(null); + setUploadPercent(0); + toast.success( + result.status === "started" + ? "Loom file uploaded. Processing has started." + : "Loom file uploaded. Processing was already underway.", + ); + }, + onError: async (error) => { + await refresh(); + toast.error( + `${error.message} The same Loom URL can be retried after checking its upload.`, + ); + }, + }); + + return ( +
+
+
+

+ {request.organizationName} +

+

+ {request.requestedByEmail} · {request.organizationId} · Requested{" "} + {new Date(request.createdAt).toLocaleDateString()} +

+
+ + {LOOM_MIGRATION_STATUS_LABELS[request.status]} + +
+
+

Loom workspace: {request.workspaceName || "Not provided"}

+

+ Loom invite:{" "} + {request.invitedAt ? "Marked as sent" : "Waiting for customer"} +

+

Videos queued: {request.queuedVideoCount}

+

Verified videos: {request.importedVideoCount}

+

Imports starting: {request.activeImportCount}

+
+ {request.customerNote && ( +
+ Customer note: {request.customerNote} +
+ )} + {request.customerReply && ( +
+ Latest customer answer: {request.customerReply} +
+ )} + +
+

Import a mapped Loom CSV

+

+ Use loom_video_url,user_email,space_name. The destination is this Cap + organization; imports run in batches of 10. Started jobs still need + terminal processing checks. +

+ + { + const file = event.target.files?.[0]; + if (!file) return; + try { + setCsvRows(parseConciergeLoomCsv(await file.text())); + setProcessedRows(0); + setStartedRows(0); + setFailedRows([]); + } catch (error) { + setCsvRows([]); + toast.error( + error instanceof Error + ? error.message + : "Could not read the CSV.", + ); + } + }} + type="file" + /> + {csvRows.length > 0 && ( +

+ {csvRows.length} rows ready. Review the source inventory and owner + mapping before import. +

+ )} + + {processedRows > 0 && ( +

+ {startedRows} jobs started; {failedRows.length} rows failed to + start. +

+ )} + {failedRows.length > 0 && ( +
    + {failedRows.map((row) => ( +
  • + Row {row.rowNumber}: {row.error || "Could not start import."} +
  • + ))} +
+ )} +
+ +
{ + event.preventDefault(); + fileMutation.mutate(); + }} + > +

+ Upload a restricted Loom video +

+

+ Use this when the CSV import cannot download a video. Download an MP4 + from Loom with an account that has permission, then map it to its Loom + link and Cap owner. Restricted Library videos may need explicit + access. +

+
+ + + + +
+ + +
+ +
{ + event.preventDefault(); + statusMutation.mutate(); + }} + > +

Update customer status

+
+
+ + +
+
+ + setExpectedVideoCount(event.target.value)} + type="number" + value={expectedVideoCount} + /> +
+
+ + setImportedVideoCount(event.target.value)} + type="number" + value={importedVideoCount} + /> +
+
+
+ +