From 1f15dc31a9a12f229a335405d8a88ae6d0a813a2 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:46:11 +0100 Subject: [PATCH 1/7] feat: add Cap Pro Loom concierge migration --- apps/web/__tests__/unit/loom-csv.test.ts | 25 + .../unit/loom-migration-state.test.ts | 39 + apps/web/actions/loom-concierge.tsx | 331 ++ apps/web/actions/loom.ts | 429 +- .../dashboard/_components/Navbar/Items.tsx | 25 + .../loom-migrations/LoomMigrationQueue.tsx | 527 ++ .../dashboard/admin/loom-migrations/page.tsx | 17 + .../app/(org)/dashboard/import/ImportPage.tsx | 38 +- .../dashboard/import/loom/ImportLoomPage.tsx | 61 +- .../migrations/loom/LoomMigrationPage.tsx | 398 ++ .../(org)/dashboard/migrations/loom/page.tsx | 10 + apps/web/content/docs/migrating-to-cap.mdx | 16 +- apps/web/lib/loom-concierge.ts | 134 + apps/web/lib/loom-csv.ts | 72 + apps/web/lib/loom-migration-state.ts | 80 + packages/database/emails/loom-migration.tsx | 70 + .../migrations/0046_loom_concierge.sql | 25 + .../migrations/meta/0046_snapshot.json | 4735 +++++++++++++++++ .../database/migrations/meta/_journal.json | 7 + packages/database/schema.ts | 51 + 20 files changed, 7031 insertions(+), 59 deletions(-) create mode 100644 apps/web/__tests__/unit/loom-csv.test.ts create mode 100644 apps/web/__tests__/unit/loom-migration-state.test.ts create mode 100644 apps/web/actions/loom-concierge.tsx create mode 100644 apps/web/app/(org)/dashboard/admin/loom-migrations/LoomMigrationQueue.tsx create mode 100644 apps/web/app/(org)/dashboard/admin/loom-migrations/page.tsx create mode 100644 apps/web/app/(org)/dashboard/migrations/loom/LoomMigrationPage.tsx create mode 100644 apps/web/app/(org)/dashboard/migrations/loom/page.tsx create mode 100644 apps/web/lib/loom-concierge.ts create mode 100644 apps/web/lib/loom-csv.ts create mode 100644 apps/web/lib/loom-migration-state.ts create mode 100644 packages/database/emails/loom-migration.tsx create mode 100644 packages/database/migrations/0046_loom_concierge.sql create mode 100644 packages/database/migrations/meta/0046_snapshot.json 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..5338285db28 --- /dev/null +++ b/apps/web/__tests__/unit/loom-csv.test.ts @@ -0,0 +1,25 @@ +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,"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-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..b1a91d30f14 --- /dev/null +++ b/apps/web/actions/loom-concierge.tsx @@ -0,0 +1,331 @@ +"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 { + importedVideos, + 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 } 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(importedVideos) + .innerJoin(videoUploads, eq(videoUploads.videoId, importedVideos.id)) + .where( + and( + eq(importedVideos.orgId, request.organizationId), + eq(importedVideos.source, "loom"), + 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, + 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), + ), + ); + if (affectedRows(result) === 0) { + throw new Error("The migration request changed. 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..7cb49e13fac 100644 --- a/apps/web/actions/loom.ts +++ b/apps/web/actions/loom.ts @@ -7,7 +7,9 @@ import { nanoId } from "@cap/database/helpers"; import { folders, importedVideos, + loomMigrationRequests, organizationMembers, + organizations, sharedVideos, spaceMembers, spaces, @@ -23,11 +25,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 +39,14 @@ import { requireOrganizationSettingsManager, } from "@/actions/organization/authorization"; import { requireSpaceManager } from "@/actions/organization/space-authorization"; +import { requireMigrationOperator } 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 { @@ -97,10 +104,22 @@ const LOOM_CSV_LIMIT_ERROR = `CSV imports are limited to ${MAX_LOOM_CSV_ROWS} ro const LOOM_CSV_PERMISSION_ERROR = "Only organization admins and owners can import Loom videos from a CSV."; +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 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; } @@ -770,6 +789,18 @@ export async function importFromLoomCsv({ }; } + return processLoomCsvRows({ rows, orgId, actorId: user.id }); +} + +async function processLoomCsvRows({ + rows, + orgId, + actorId, +}: { + rows: LoomCsvImportRow[]; + orgId: Organisation.OrganisationId; + actorId: User.UserId; +}): Promise { const inputRows = Array.isArray(rows) ? rows : []; const normalizedRows = inputRows .map((row, index) => ({ @@ -854,7 +885,7 @@ export async function importFromLoomCsv({ const provisionedMember = await provisionOrganizationInvitee({ organizationId: orgId, email: row.userEmail, - invitedByUserId: user.id, + invitedByUserId: actorId, role: "member", }); member = { @@ -886,14 +917,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 +975,389 @@ 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."); + } + const claimed = await db() + .update(loomMigrationRequests) + .set({ + status: "in_progress", + lastOperatorUserId: operator.id, + lastOperatorAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + ), + ); + if (affectedRows(claimed) === 0) { + throw new Error("The migration request changed. Refresh and try again."); + } + const result = await processLoomCsvRows({ + rows, + orgId: request.organizationId, + actorId: request.ownerId, + }); + if (result.importedCount > 0) { + const updated = await db() + .update(loomMigrationRequests) + .set({ + queuedVideoCount: sql`${loomMigrationRequests.queuedVideoCount} + ${result.importedCount}`, + lastOperatorUserId: operator.id, + lastOperatorAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + ), + ); + if (affectedRows(updated) === 0) { + throw new Error( + "The migration request changed after jobs started. Review them before retrying.", + ); + } + } + revalidatePath("/dashboard/migrations/loom"); + revalidatePath("/dashboard/admin/loom-migrations"); + return result; +} + +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 [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.", + ); + } + 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) => { + 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, + }); + if (space) { + await tx.insert(spaceVideos).values({ + id: nanoId(), + videoId, + spaceId: space.id, + addedById: request.ownerId, + }); + } + }); + await db() + .update(loomMigrationRequests) + .set({ + status: "in_progress", + lastOperatorUserId: operator.id, + lastOperatorAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + ), + ); + revalidatePath("/dashboard/admin/loom-migrations"); + return { videoId, uploadTarget: upload.upload }; +} + +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 [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( + loomMigrationRequests, + 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.", + }); + if (status === "started") { + await db() + .update(loomMigrationRequests) + .set({ + queuedVideoCount: sql`${loomMigrationRequests.queuedVideoCount} + 1`, + lastOperatorUserId: operator.id, + lastOperatorAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq(loomMigrationRequests.id, requestId), + isNotNull(loomMigrationRequests.activeOrganizationId), + ), + ); + } + revalidatePath("/dashboard/migrations/loom"); + revalidatePath("/dashboard/admin/loom-migrations"); + return { status }; +} 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..d256a9aded3 --- /dev/null +++ b/apps/web/app/(org)/dashboard/admin/loom-migrations/LoomMigrationQueue.tsx @@ -0,0 +1,527 @@ +"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"} +

+

Jobs started: {request.queuedVideoCount}

+

Verified videos: {request.importedVideoCount}

+
+ {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} + /> +
+
+
+ +