feat: chunked (S3 multipart) uploads for R2/S3 video backend

Self-hosted instances on the R2/S3 backend could only upload a video as a
single PUT, which fails behind a Cloudflare proxy/tunnel (100MB request-body
cap) and is capped at 5GiB with no resilience. Bunny already avoids this via
tus; this brings the R2/S3 path to parity.

Files larger than a threshold (default 90MiB) are now split into parts
(default 32MiB, min 5MiB) and uploaded directly browser->R2 via presigned
UploadPart URLs, then reassembled server-side with CompleteMultipartUpload.
Each request stays under the 100MB cap, lifts the size ceiling well past
5GiB, and adds per-chunk retry. Files at/under the threshold keep the
existing single-PUT path unchanged. Bunny path is untouched.

Thresholds are env-overridable via OPENFRAME_R2_MULTIPART_THRESHOLD_BYTES
and OPENFRAME_R2_MULTIPART_PART_SIZE_BYTES.

Verified end-to-end against real Cloudflare R2 and a local MinIO behind an
nginx 90MB cap (single 141MB PUT 413s on master; 32MB parts pass here).

Closes #22
This commit is contained in:
yusufipk
2026-07-10 19:19:07 +07:00
parent 82932c6b22
commit 880d0ac0fa
9 changed files with 513 additions and 12 deletions
+6
View File
@@ -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.
@@ -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<number>();
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');
}
}
@@ -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)
+150 -6
View File
@@ -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<string> {
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<T>(fn: () => Promise<T>, delays: number[]): Promise<T> {
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<void> {
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<void> {
const totalBytes = file.size;
const partSize = multipart.partSizeBytes;
const loadedPerPart = new Array<number>(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<number | null> {
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;
+33
View File
@@ -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
);
}
+2
View File
@@ -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,
},
});
}
+92
View File
@@ -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<string> {
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<string> {
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<void> {
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<void> {
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,
@@ -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;
+1
View File
@@ -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")