diff --git a/.env.example b/.env.example index e9d4d96..fa63f41 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,12 @@ OPENFRAME_ENABLE_BUNNY_UPLOADS="true" OPENFRAME_ENABLE_S3_VIDEO_UPLOADS="false" # Max size per uploaded video file in bytes (default 5GB if unset) OPENFRAME_MAX_VIDEO_UPLOAD_BYTES="5368709120" +# Files larger than this use chunked (S3 multipart) uploads instead of a single PUT. +# Default 90MiB keeps each request under the common 100MB Cloudflare proxy/tunnel cap. +# Lower it if your proxy enforces a stricter request-body limit. +OPENFRAME_R2_MULTIPART_THRESHOLD_BYTES="94371840" +# Size of each multipart chunk in bytes (default 32MiB, minimum 5MiB). +OPENFRAME_R2_MULTIPART_PART_SIZE_BYTES="33554432" # Direct browser uploads require bucket CORS allowing PUT from your app origin(s). # Run once after creating the bucket: bun run r2:configure-cors # Or set CORS manually in Cloudflare R2 -> bucket -> Settings -> CORS policy. diff --git a/app/api/projects/[projectId]/videos/r2-complete/route.ts b/app/api/projects/[projectId]/videos/r2-complete/route.ts new file mode 100644 index 0000000..9d1c93f --- /dev/null +++ b/app/api/projects/[projectId]/videos/r2-complete/route.ts @@ -0,0 +1,168 @@ +import { NextRequest } from 'next/server'; +import { db } from '@/lib/db'; +import { auth, checkProjectAccess } from '@/lib/auth'; +import { apiErrors, successResponse, withCacheControl } from '@/lib/api-response'; +import { rateLimit } from '@/lib/rate-limit'; +import { parseR2UploadToken, verifyR2UploadToken } from '@/lib/r2-upload-token'; +import { abortMultipartVideoUpload, completeMultipartVideoUpload } from '@/lib/r2'; +import { isS3VideoUploadsEnabled } from '@/lib/feature-flags'; +import { objectKeyToVideoProxyPath } from '@/lib/video-upload-validation'; +import { releaseStorageReservation } from '@/lib/storage-quota'; +import { logError } from '@/lib/logger'; + +type RouteParams = { params: Promise<{ projectId: string }> }; + +async function getProjectWithEditAccess(projectId: string, userId: string) { + const project = await db.project.findUnique({ + where: { id: projectId }, + select: { + id: true, + ownerId: true, + workspaceId: true, + visibility: true, + workspace: { select: { ownerId: true } }, + }, + }); + + if (!project) return null; + + const access = await checkProjectAccess(project, userId, { intent: 'manage' }); + if (!access.canEdit) return null; + + return project; +} + +type IncomingPart = { partNumber: number; etag: string }; + +function parseParts(raw: unknown): IncomingPart[] | null { + if (!Array.isArray(raw) || raw.length === 0 || raw.length > 10000) { + return null; + } + + const parts: IncomingPart[] = []; + const seen = new Set(); + + for (const entry of raw) { + const partNumber = (entry as { partNumber?: unknown })?.partNumber; + const etag = (entry as { etag?: unknown })?.etag; + + if ( + typeof partNumber !== 'number' || + !Number.isInteger(partNumber) || + partNumber < 1 || + partNumber > 10000 || + seen.has(partNumber) + ) { + return null; + } + + if (typeof etag !== 'string' || etag.trim().length === 0) { + return null; + } + + seen.add(partNumber); + parts.push({ partNumber, etag: etag.trim() }); + } + + return parts; +} + +// POST /api/projects/[projectId]/videos/r2-complete +export async function POST(request: NextRequest, { params }: RouteParams) { + try { + const limited = await rateLimit(request, 'mutate'); + if (limited) return limited; + + const session = await auth(); + const { projectId } = await params; + + if (!session?.user?.id) { + return apiErrors.unauthorized(); + } + + if (!isS3VideoUploadsEnabled()) { + return apiErrors.badRequest('S3 video uploads are disabled by this host'); + } + + const project = await getProjectWithEditAccess(projectId, session.user.id); + if (!project) { + return apiErrors.forbidden('Access denied'); + } + + const body = await request.json().catch(() => null); + const objectKey = typeof body?.objectKey === 'string' ? body.objectKey.trim() : ''; + const uploadToken = typeof body?.uploadToken === 'string' ? body.uploadToken.trim() : ''; + const parts = parseParts(body?.parts); + + if (!objectKey || !uploadToken) { + return apiErrors.badRequest('objectKey and uploadToken are required'); + } + + if (!parts) { + return apiErrors.badRequest('parts must be a non-empty list of { partNumber, etag }'); + } + + const tokenPayload = parseR2UploadToken(uploadToken); + if (!tokenPayload) { + return apiErrors.forbidden('Invalid upload token'); + } + + const isValidUploadToken = verifyR2UploadToken(uploadToken, { + userId: session.user.id, + projectId, + objectKey, + sessionId: tokenPayload.sid, + tokenId: tokenPayload.jti, + }); + if (!isValidUploadToken) { + return apiErrors.forbidden('Invalid upload token'); + } + + const uploadSession = await db.videoUploadSession.findFirst({ + where: { + id: tokenPayload.sid, + status: 'INITIATED', + userId: session.user.id, + projectId, + objectKey, + uploadJti: tokenPayload.jti, + expiresAt: { gt: new Date() }, + }, + select: { + id: true, + multipartUploadId: true, + reservationId: true, + billedUserId: true, + }, + }); + if (!uploadSession || !uploadSession.multipartUploadId) { + return apiErrors.forbidden('Invalid upload token'); + } + + const proxyUrl = objectKeyToVideoProxyPath(objectKey); + if (!proxyUrl) { + return apiErrors.badRequest('Invalid object key'); + } + + try { + await completeMultipartVideoUpload(objectKey, uploadSession.multipartUploadId, parts); + } catch (error) { + logError('Failed to complete R2 multipart upload:', error); + await abortMultipartVideoUpload(objectKey, uploadSession.multipartUploadId).catch( + () => undefined + ); + await db.videoUploadSession.updateMany({ + where: { id: uploadSession.id, status: 'INITIATED' }, + data: { status: 'CANCELLED', consumedAt: new Date() }, + }); + await releaseStorageReservation(uploadSession.reservationId, uploadSession.billedUserId); + return apiErrors.internalError('Failed to complete multipart upload'); + } + + const response = successResponse({ objectKey, proxyUrl }); + return withCacheControl(response, 'private, no-store'); + } catch (error) { + logError('Error completing R2 multipart upload:', error); + return apiErrors.internalError('Failed to complete upload'); + } +} diff --git a/app/api/projects/[projectId]/videos/r2-init/route.ts b/app/api/projects/[projectId]/videos/r2-init/route.ts index 8944b15..ad2d2f5 100644 --- a/app/api/projects/[projectId]/videos/r2-init/route.ts +++ b/app/api/projects/[projectId]/videos/r2-init/route.ts @@ -10,12 +10,20 @@ import { verifyR2UploadToken, } from '@/lib/r2-upload-token'; import { + abortMultipartVideoUpload, + createMultipartVideoUpload, createPresignedImagePutUrl, + createPresignedUploadPartUrl, createPresignedVideoPutUrl, deleteR2Object, deleteVideoObject, } from '@/lib/r2'; -import { getMaxVideoUploadBytes, isS3VideoUploadsEnabled } from '@/lib/feature-flags'; +import { + getMaxVideoUploadBytes, + getR2MultipartPartSizeBytes, + getR2MultipartThresholdBytes, + isS3VideoUploadsEnabled, +} from '@/lib/feature-flags'; import { buildVideoObjectKey, getVideoExtensionFromMime, @@ -133,13 +141,52 @@ export async function POST(request: NextRequest, { params }: RouteParams) { const thumbnailObjectKey = `images/${thumbnailFilename}`; const thumbnailProxyUrl = `/api/upload/image/${thumbnailFilename}`; - let presignedPutUrl: string; + const useMultipart = sizeBytes > getR2MultipartThresholdBytes(); + + let presignedPutUrl = ''; let thumbnailPresignedPutUrl: string; + let multipartUploadId: string | null = null; + let multipart: { + uploadId: string; + partSizeBytes: number; + parts: Array<{ partNumber: number; url: string }>; + } | null = null; + try { - [presignedPutUrl, thumbnailPresignedPutUrl] = await Promise.all([ - createPresignedVideoPutUrl(objectKey, contentType, sizeBytes), - createPresignedImagePutUrl(thumbnailObjectKey, 'image/jpeg'), - ]); + if (useMultipart) { + const partSize = getR2MultipartPartSizeBytes(); + const partCount = Number((sizeBytes + partSize - BigInt(1)) / partSize); + + multipartUploadId = await createMultipartVideoUpload(objectKey, contentType); + + try { + const [parts, thumbnailUrl] = await Promise.all([ + Promise.all( + Array.from({ length: partCount }, async (_unused, index) => { + const partNumber = index + 1; + const url = await createPresignedUploadPartUrl( + objectKey, + multipartUploadId as string, + partNumber + ); + return { partNumber, url }; + }) + ), + createPresignedImagePutUrl(thumbnailObjectKey, 'image/jpeg'), + ]); + + multipart = { uploadId: multipartUploadId, partSizeBytes: Number(partSize), parts }; + thumbnailPresignedPutUrl = thumbnailUrl; + } catch (error) { + await abortMultipartVideoUpload(objectKey, multipartUploadId).catch(() => undefined); + throw error; + } + } else { + [presignedPutUrl, thumbnailPresignedPutUrl] = await Promise.all([ + createPresignedVideoPutUrl(objectKey, contentType, sizeBytes), + createPresignedImagePutUrl(thumbnailObjectKey, 'image/jpeg'), + ]); + } } catch (error) { await releaseStorageReservation(reserveResult.reservationId, project.workspace.ownerId); logError('Failed to create presigned video upload URL:', error); @@ -159,6 +206,7 @@ export async function POST(request: NextRequest, { params }: RouteParams) { reservationId: reserveResult.reservationId, uploadJti, expiresAt, + multipartUploadId, }); const uploadToken = createR2UploadToken({ @@ -180,6 +228,7 @@ export async function POST(request: NextRequest, { params }: RouteParams) { thumbnailPresignedPutUrl, thumbnailObjectKey, thumbnailProxyUrl, + multipart, }); return withCacheControl(response, 'private, no-store'); @@ -252,6 +301,7 @@ export async function DELETE(request: NextRequest, { params }: RouteParams) { reservationId: true, billedUserId: true, thumbnailObjectKey: true, + multipartUploadId: true, }, }); if (!uploadSession) { @@ -278,6 +328,9 @@ export async function DELETE(request: NextRequest, { params }: RouteParams) { try { await Promise.all([ + uploadSession.multipartUploadId + ? abortMultipartVideoUpload(objectKey, uploadSession.multipartUploadId) + : Promise.resolve(), deleteVideoObject(objectKey), uploadSession.thumbnailObjectKey.startsWith('images/') ? deleteR2Object(uploadSession.thumbnailObjectKey) diff --git a/lib/client/r2-video-upload.ts b/lib/client/r2-video-upload.ts index ee55e2c..153ab18 100644 --- a/lib/client/r2-video-upload.ts +++ b/lib/client/r2-video-upload.ts @@ -1,5 +1,13 @@ import { captureVideoThumbnail } from '@/lib/client/video-thumbnail'; +export type R2MultipartPart = { partNumber: number; url: string }; + +export type R2MultipartInit = { + uploadId: string; + partSizeBytes: number; + parts: R2MultipartPart[]; +}; + export type R2VideoInitResponse = { presignedPutUrl: string; objectKey: string; @@ -10,8 +18,11 @@ export type R2VideoInitResponse = { thumbnailPresignedPutUrl: string; thumbnailObjectKey: string; thumbnailProxyUrl: string; + multipart: R2MultipartInit | null; }; +const PART_RETRY_DELAYS = [0, 2000, 5000, 10000]; + export type R2VideoUploadResult = R2VideoInitResponse & { duration: number | null; thumbnailUrl: string | null; @@ -56,6 +67,128 @@ function uploadBytesWithProgress( }); } +function uploadPartWithProgress( + url: string, + body: Blob, + onPartProgress?: (loadedBytes: number) => void +): Promise { + return new Promise((resolve, reject) => { + const xhr = new XMLHttpRequest(); + xhr.open('PUT', url); + // Intentionally no Content-Type header: it is not part of the presigned + // UploadPart signature, and the part body is raw bytes. + + xhr.upload.onprogress = (event) => { + if (!onPartProgress || !event.lengthComputable) return; + onPartProgress(event.loaded); + }; + + xhr.onload = () => { + if (xhr.status >= 200 && xhr.status < 300) { + const etag = xhr.getResponseHeader('ETag'); + if (!etag) { + reject( + new Error( + 'Upload response missing ETag header. Configure bucket CORS to expose the ETag header.' + ) + ); + return; + } + resolve(etag); + return; + } + reject(new Error(`Chunk upload failed with status ${xhr.status}`)); + }; + + xhr.onerror = () => { + reject( + new Error( + 'Network error during upload. If you use direct S3/R2 uploads, configure bucket CORS to allow PUT from this site origin.' + ) + ); + }; + xhr.onabort = () => reject(new Error('Upload aborted')); + + xhr.send(body); + }); +} + +async function withRetry(fn: () => Promise, delays: number[]): Promise { + let lastError: unknown; + for (let attempt = 0; attempt < delays.length; attempt += 1) { + if (attempt > 0) { + await new Promise((resolve) => setTimeout(resolve, delays[attempt])); + } + try { + return await fn(); + } catch (error) { + lastError = error; + } + } + throw lastError instanceof Error ? lastError : new Error('Upload failed after retries'); +} + +async function completeMultipartUpload( + projectId: string, + objectKey: string, + uploadToken: string, + parts: Array<{ partNumber: number; etag: string }> +): Promise { + const res = await fetch(`/api/projects/${projectId}/videos/r2-complete`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ objectKey, uploadToken, parts }), + }); + + if (!res.ok) { + const payload = (await res.json().catch(() => null)) as { error?: string } | null; + throw new Error(payload?.error || 'Failed to complete multipart upload'); + } +} + +async function uploadVideoMultipart( + projectId: string, + file: File, + multipart: R2MultipartInit, + objectKey: string, + uploadToken: string, + onProgress?: UploadProgressHandler +): Promise { + const totalBytes = file.size; + const partSize = multipart.partSizeBytes; + const loadedPerPart = new Array(multipart.parts.length).fill(0); + + const reportProgress = () => { + if (!onProgress) return; + const loaded = loadedPerPart.reduce((sum, value) => sum + value, 0); + onProgress(Math.min(100, Math.round((loaded / totalBytes) * 100))); + }; + + const completedParts: Array<{ partNumber: number; etag: string }> = []; + + for (let index = 0; index < multipart.parts.length; index += 1) { + const part = multipart.parts[index]; + const start = (part.partNumber - 1) * partSize; + const end = Math.min(start + partSize, totalBytes); + const blob = file.slice(start, end); + + const etag = await withRetry( + () => + uploadPartWithProgress(part.url, blob, (loadedBytes) => { + loadedPerPart[index] = loadedBytes; + reportProgress(); + }), + PART_RETRY_DELAYS + ); + + loadedPerPart[index] = end - start; + reportProgress(); + completedParts.push({ partNumber: part.partNumber, etag }); + } + + await completeMultipartUpload(projectId, objectKey, uploadToken, completedParts); +} + async function readVideoDuration(file: File): Promise { return new Promise((resolve) => { const objectUrl = URL.createObjectURL(file); @@ -151,12 +284,23 @@ export async function uploadVideoToR2( }; try { - await uploadBytesWithProgress( - init.presignedPutUrl, - file, - init.contentType, - options?.onProgress - ); + if (init.multipart) { + await uploadVideoMultipart( + projectId, + file, + init.multipart, + init.objectKey, + init.uploadToken, + options?.onProgress + ); + } else { + await uploadBytesWithProgress( + init.presignedPutUrl, + file, + init.contentType, + options?.onProgress + ); + } } catch (error) { await cleanupPendingR2VideoUpload(projectId, cleanupInput); throw error; diff --git a/lib/feature-flags.ts b/lib/feature-flags.ts index 48fed40..00c1eb3 100644 --- a/lib/feature-flags.ts +++ b/lib/feature-flags.ts @@ -97,3 +97,36 @@ export function getMaxVideoUploadBytes(): bigint { export function isInviteCodeRequired() { return readBooleanEnv('OPENFRAME_REQUIRE_INVITE_CODE', true); } + +function parseBigIntEnv(name: string, defaultValue: bigint, minValue?: bigint): bigint { + const raw = process.env[name]?.trim(); + if (!raw) return defaultValue; + + try { + const parsed = BigInt(raw); + if (parsed <= BigInt(0)) return defaultValue; + if (minValue !== undefined && parsed < minValue) return minValue; + return parsed; + } catch { + return defaultValue; + } +} + +// Files larger than this use S3 multipart upload (chunked) instead of a single PUT. +// Default 90 MiB keeps each request under the common 100 MB Cloudflare proxy/tunnel cap. +export function getR2MultipartThresholdBytes(): bigint { + return parseBigIntEnv( + 'OPENFRAME_R2_MULTIPART_THRESHOLD_BYTES', + BigInt(90) * BigInt(1024) * BigInt(1024) + ); +} + +// Size of each multipart chunk. Clamped to the S3 minimum of 5 MiB for non-final parts. +export function getR2MultipartPartSizeBytes(): bigint { + const minPartSize = BigInt(5) * BigInt(1024) * BigInt(1024); + return parseBigIntEnv( + 'OPENFRAME_R2_MULTIPART_PART_SIZE_BYTES', + BigInt(32) * BigInt(1024) * BigInt(1024), + minPartSize + ); +} diff --git a/lib/r2-upload-session.ts b/lib/r2-upload-session.ts index 5a6df06..4c1613f 100644 --- a/lib/r2-upload-session.ts +++ b/lib/r2-upload-session.ts @@ -11,6 +11,7 @@ export type CreateR2UploadSessionInput = { reservationId: string | null; uploadJti: string; expiresAt: Date; + multipartUploadId?: string | null; }; export async function createR2UploadSession(input: CreateR2UploadSessionInput) { @@ -26,6 +27,7 @@ export async function createR2UploadSession(input: CreateR2UploadSessionInput) { reservationId: input.reservationId, uploadJti: input.uploadJti, expiresAt: input.expiresAt, + multipartUploadId: input.multipartUploadId ?? null, }, }); } diff --git a/lib/r2.ts b/lib/r2.ts index 9054b64..b4fb7f4 100644 --- a/lib/r2.ts +++ b/lib/r2.ts @@ -1,5 +1,8 @@ import { + AbortMultipartUploadCommand, + CompleteMultipartUploadCommand, CreateBucketCommand, + CreateMultipartUploadCommand, DeleteObjectCommand, GetObjectCommand, GetBucketCorsCommand, @@ -7,6 +10,7 @@ import { HeadObjectCommand, PutBucketCorsCommand, PutObjectCommand, + UploadPartCommand, S3Client, } from '@aws-sdk/client-s3'; import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; @@ -290,6 +294,94 @@ export async function createPresignedVideoPutUrl( return getSignedUrl(getOrCreateR2PresignClient(), command, { expiresIn: expiresInSeconds }); } +export async function createMultipartVideoUpload( + key: string, + contentType: string +): Promise { + if (!key.startsWith(VIDEO_OBJECT_KEY_PREFIX)) { + throw new Error('Invalid video object key'); + } + + const result = await r2Client.send( + new CreateMultipartUploadCommand({ + Bucket: R2_BUCKET_NAME, + Key: key, + ContentType: contentType, + }) + ); + + if (!result.UploadId) { + throw new Error('Failed to create multipart upload'); + } + + return result.UploadId; +} + +export async function createPresignedUploadPartUrl( + key: string, + uploadId: string, + partNumber: number, + expiresInSeconds = DEFAULT_PRESIGNED_PUT_TTL_SECONDS +): Promise { + if (!key.startsWith(VIDEO_OBJECT_KEY_PREFIX)) { + throw new Error('Invalid video object key'); + } + + if (!Number.isInteger(partNumber) || partNumber < 1 || partNumber > 10000) { + throw new Error('Invalid part number'); + } + + const command = new UploadPartCommand({ + Bucket: R2_BUCKET_NAME, + Key: key, + UploadId: uploadId, + PartNumber: partNumber, + }); + + return getSignedUrl(getOrCreateR2PresignClient(), command, { expiresIn: expiresInSeconds }); +} + +export async function completeMultipartVideoUpload( + key: string, + uploadId: string, + parts: Array<{ partNumber: number; etag: string }> +): Promise { + if (!key.startsWith(VIDEO_OBJECT_KEY_PREFIX)) { + throw new Error('Invalid video object key'); + } + + if (parts.length === 0) { + throw new Error('No parts provided for multipart completion'); + } + + const orderedParts = [...parts] + .sort((a, b) => a.partNumber - b.partNumber) + .map((part) => ({ PartNumber: part.partNumber, ETag: part.etag })); + + await r2Client.send( + new CompleteMultipartUploadCommand({ + Bucket: R2_BUCKET_NAME, + Key: key, + UploadId: uploadId, + MultipartUpload: { Parts: orderedParts }, + }) + ); +} + +export async function abortMultipartVideoUpload(key: string, uploadId: string): Promise { + if (!key.startsWith(VIDEO_OBJECT_KEY_PREFIX)) { + throw new Error('Invalid video object key'); + } + + await r2Client.send( + new AbortMultipartUploadCommand({ + Bucket: R2_BUCKET_NAME, + Key: key, + UploadId: uploadId, + }) + ); +} + export async function createPresignedImagePutUrl( key: string, contentType: string, diff --git a/prisma/migrations/20260627140000_add_video_upload_multipart_id/migration.sql b/prisma/migrations/20260627140000_add_video_upload_multipart_id/migration.sql new file mode 100644 index 0000000..3bd464b --- /dev/null +++ b/prisma/migrations/20260627140000_add_video_upload_multipart_id/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable: track the S3/R2 multipart upload id so chunked uploads can be completed/aborted. +ALTER TABLE "video_upload_sessions" ADD COLUMN "multipart_upload_id" TEXT; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 04b2dd6..8fc215b 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -724,6 +724,7 @@ model VideoUploadSession { declaredSizeBytes BigInt @map("declared_size_bytes") contentType String @map("content_type") reservationId String? @map("reservation_id") + multipartUploadId String? @map("multipart_upload_id") expiresAt DateTime @map("expires_at") status UploadSessionStatus @default(INITIATED) consumedAt DateTime? @map("consumed_at")