feat(storage-r2): client uploads using R2 multipart api (#14733)

### What?
Implements client-side uploads by splitting files into multiple parts
and uploading each one separately.
### Why?
To support uploading large files, without reaching the Worker's memory
limit.
### How?
By splitting the file into multiple small parts and uploading each one
separately. This leverages R2's support for multipart uploads:
https://developers.cloudflare.com/r2/objects/multipart-objects/
This feature was requested on Discord:
https://discord.com/channels/967097582721572934/1428654256562503740/1428654256562503740
---------
Co-authored-by: Ricardo Tavares <rtavares@cloudflare.com>
Co-authored-by: Paul Popus <paul@payloadcms.com>

Author: 3 people

packages/storage-r2/src/client/R2ClientUploadHandler.ts

@@ -0,0 +1,82 @@
+'use client'
+
+import{createClientUploadHandler}from'@payloadcms/plugin-cloud-storage/client'
+import{formatAdminURL}from'payload/shared'
+
+importtype{
+R2MultipartUpload,
+R2StorageClientUploadContext,
+R2StorageClientUploadHandlerParams,
+R2StorageMultipartUploadHandlerParams,
+R2UploadedPart,
+}from'../types.js'
+
+exportconstR2ClientUploadHandler=createClientUploadHandler<R2StorageClientUploadHandlerParams>({
+handler: async({
+ apiRoute,
+ collectionSlug,
+extra: { chunkSize =5*1024*1024, prefix =''},
+ file,
+ serverHandlerPath,
+ serverURL,
+}): Promise<R2StorageClientUploadContext|undefined>=>{
+constparams: R2StorageMultipartUploadHandlerParams={
+collection: collectionSlug,
+fileName: file.name,
+fileType: file.type,
+}
+constbaseURL=formatAdminURL({
+ apiRoute,
+path: serverHandlerPath,
+ serverURL,
+})
+
+constendpoint=`${baseURL}?${String(newURLSearchParams(params))}`
+
+constmultipart=awaitfetch(endpoint,{method: 'POST'})
+if(!multipart.ok){
+thrownewError('Failed to initialize multipart upload')
+}
+
+constmultipartUpload=(awaitmultipart.json())asPick<R2MultipartUpload,'key'|'uploadId'>
+constmultipartUploadedParts: R2UploadedPart[]=[]
+
+params.multipartId=multipartUpload.uploadId
+params.multipartKey=multipartUpload.key
+
+constpartTotal=Math.ceil(file.size/chunkSize)
+
+for(letpart=1;part<=partTotal;part++){
+constbytesEnd=Math.min(part*chunkSize,file.size)
+constbytesStart=(part-1)*chunkSize
+
+params.multipartNumber=String(part)
+
+constbody=file.slice(bytesStart,bytesEnd)
+constheaders={
+'Content-Length': String(body.size),
+'Content-Type': 'application/octet-stream',
+}
+constuploaded=awaitfetch(endpoint,{ body, headers,method: 'POST'})
+if(!uploaded.ok){
+thrownewError(`Failed to upload part ${part} / ${partTotal}`)
+}
+
+multipartUploadedParts.push((awaituploaded.json())asR2UploadedPart)
+
+if(part===partTotal){
+deleteparams.multipartNumber
+
+constbody=JSON.stringify(multipartUploadedParts)
+constheaders={'Content-Type': 'application/json'}
+constcomplete=awaitfetch(endpoint,{ body, headers,method: 'POST'})
+if(!complete.ok){
+thrownewError(`Failed to complete multipart upload`)
+}
+
+constkey=awaitcomplete.text()
+return{ key }
+}
+}
+},
+})

packages/storage-r2/src/exports/client.ts

@@ -0,0 +1 @@
+export{R2ClientUploadHandler}from'../client/R2ClientUploadHandler.js'

packages/storage-r2/src/handleMultiPartUpload.ts

@@ -0,0 +1,89 @@
+importtype{ClientUploadsAccess}from'@payloadcms/plugin-cloud-storage/types'
+importtype{PayloadHandler}from'payload'
+
+importpathfrom'path'
+import{APIError,Forbidden}from'payload'
+
+importtype{R2StorageOptions}from'./index.js'
+importtype{R2Bucket,R2StorageMultipartUploadHandlerParams}from'./types.js'
+
+typeArgs={
+access?: ClientUploadsAccess
+bucket: R2Bucket
+collections: R2StorageOptions['collections']
+}
+
+// Adapted from https://developers.cloudflare.com/r2/api/workers/workers-multipart-usage/
+exportconstgetHandleMultiPartUpload=
+({ access, bucket, collections }: Args): PayloadHandler=>
+async(req)=>{
+constparams=Object.fromEntries(req.searchParams)asR2StorageMultipartUploadHandlerParams
+constcollectionSlug=params.collection
+constfiletype=params.fileType
+
+constcollectionConfig=collections[collectionSlug]
+if(!collectionConfig){
+thrownewAPIError(`Collection ${collectionSlug} was not found in R2 Storage options`)
+}
+
+// Check custom access if provided, otherwise check collection's create access
+if(access){
+if(!(awaitaccess({ collectionSlug, req }))){
+thrownewForbidden(req.t)
+}
+}else{
+// Use the collection's create access control
+constcollection=req.payload.collections[collectionSlug]
+if(!collection){
+thrownewAPIError(`Collection ${collectionSlug} not found`)
+}
+
+constcreateAccess=collection.config.access?.create
+if(createAccess){
+consthasAccess=awaitcreateAccess({ req })
+if(!hasAccess){
+thrownewForbidden(req.t)
+}
+}elseif(!req.user){
+// No custom access and no user - deny by default
+thrownewForbidden(req.t)
+}
+}
+
+constprefix=(typeofcollectionConfig==='object'&&collectionConfig.prefix)||''
+constfileKey=path.posix.join(prefix,params.fileName)
+
+constmultipartId=params.multipartId
+constmultipartKey=params.multipartKey
+constmultipartNumber=parseInt(params.multipartNumber||'')
+
+if(multipartId&&multipartKey){
+constmultipartUpload=bucket.resumeMultipartUpload(multipartKey,multipartId)
+constrequest=reqasRequest
+
+if(isNaN(multipartNumber)){
+// Upload complete
+constobject=awaitmultipartUpload.complete((awaitrequest.json())asany)
+returnnewResponse(object.key,{status: 200})
+}else{
+// Upload part
+constuploadedPart=awaitmultipartUpload.uploadPart(
+multipartNumber,
+awaitrequest.arrayBuffer(),
+)
+returnResponse.json(uploadedPart)
+}
+}else{
+// Create multipart upload
+constmultipartUpload=awaitbucket.createMultipartUpload(fileKey,{
+httpMetadata: {
+contentType: filetype,
+},
+})
+
+returnResponse.json({
+key: multipartUpload.key,
+uploadId: multipartUpload.uploadId,
+})
+}
+}

packages/storage-r2/src/handleUpload.ts

@@ -18,7 +18,5 @@ export const getHandleUpload = ({ bucket, prefix = '' }: Args): HandleUpload =>
 awaitbucket.put(path.posix.join(data.prefix||prefix,file.filename),buffer,{
 httpMetadata: {contentType: file.mimeType},
 })
-
-returndata
 }
 }

packages/storage-r2/src/index.ts

@@ -1,16 +1,19 @@
 importtype{
 Adapter,
+ClientUploadsConfig,
 PluginOptionsasCloudStoragePluginOptions,
 CollectionOptions,
 GeneratedAdapter,
 }from'@payloadcms/plugin-cloud-storage/types'
 importtype{Config,Plugin,UploadCollectionSlug}from'payload'
 
 import{cloudStoragePlugin}from'@payloadcms/plugin-cloud-storage'
+import{initClientUploads}from'@payloadcms/plugin-cloud-storage/utilities'
 
-importtype{R2Bucket}from'./types.js'
+importtype{R2Bucket,R2StorageClientUploadHandlerParams}from'./types.js'
 
 import{getHandleDelete}from'./handleDelete.js'
+import{getHandleMultiPartUpload}from'./handleMultiPartUpload.js'
 import{getHandleUpload}from'./handleUpload.js'
 import{getHandler}from'./staticHandler.js'
 
@@ -27,6 +30,10 @@ export interface R2StorageOptions {
 alwaysInsertFields?: boolean
 
 bucket: R2Bucket
+/**
+ * Do uploads directly on the client, to bypass limits on Cloudflare/Vercel.
+ */
+clientUploads?: ClientUploadsConfig
 /**
  * Collection options to apply the R2 adapter to.
  */
@@ -43,6 +50,29 @@ export const r2Storage: R2StoragePlugin =
 
 constisPluginDisabled=r2StorageOptions.enabled===false
 
+initClientUploads<
+R2StorageClientUploadHandlerParams,
+R2StorageOptions['collections'][keyofR2StorageOptions['collections']]
+>({
+clientHandler: '@payloadcms/storage-r2/client#R2ClientUploadHandler',
+collections: r2StorageOptions.collections,
+config: incomingConfig,
+enabled: !isPluginDisabled&&Boolean(r2StorageOptions.clientUploads),
+extraClientHandlerProps: (collection)=>({
+prefix:
+(typeofcollection==='object'&&collection.prefix&&`${collection.prefix}/`)||'',
+}),
+serverHandler: getHandleMultiPartUpload({
+access:
+typeofr2StorageOptions.clientUploads==='object'
+ ? r2StorageOptions.clientUploads.access
+ : undefined,
+bucket: r2StorageOptions.bucket,
+collections: r2StorageOptions.collections,
+}),
+serverHandlerPath: '/storage-r2-multi-part-upload',
+})
+
 if(isPluginDisabled){
 returnincomingConfig
 }
@@ -85,10 +115,11 @@ export const r2Storage: R2StoragePlugin =
 })(config)
 }
 
-functionr2StorageInternal({ bucket }: R2StorageOptions): Adapter{
+functionr2StorageInternal({ bucket, clientUploads}: R2StorageOptions): Adapter{
 return({ collection, prefix }): GeneratedAdapter=>{
 return{
 name: 'r2',
+ clientUploads,
 handleDelete: getHandleDelete({ bucket }),
 handleUpload: getHandleUpload({
  bucket,

packages/storage-r2/src/staticHandler.ts

@@ -4,7 +4,7 @@ import type { CollectionConfig } from 'payload'
 importpathfrom'path'
 import{getRangeRequestInfo}from'payload/internal'
 
-importtype{R2Bucket}from'./types.js'
+importtype{R2Bucket,R2ObjectBody}from'./types.js'
 
 interfaceArgs{
 bucket: R2Bucket
@@ -15,7 +15,7 @@ interface Args {
 constisMiniflare=process.env.NODE_ENV==='development'
 
 exportconstgetHandler=({ bucket, collection, prefix =''}: Args): StaticHandler=>{
-returnasync(req,{headers: incomingHeaders,params: { filename }})=>{
+returnasync(req,{headers: incomingHeaders,params: {clientUploadContext,filename }})=>{
 try{
 constkey=path.posix.join(prefix,filename)
 
@@ -27,6 +27,11 @@ export const getHandler = ({ bucket, collection, prefix = '' }: Args): StaticHan
 
 constfileSize=headObj.size
 
+// Don't return large file uploads back to the client, or the Worker will run out of memory
+if(fileSize>50*1024*1024&&clientUploadContext){
+returnnewResponse(null,{status: 200})
+}
+
 // Handle range request
 constrangeHeader=req.headers.get('range')
 constrangeResult=getRangeRequestInfo({ fileSize, rangeHeader })
@@ -41,7 +46,7 @@ export const getHandler = ({ bucket, collection, prefix = '' }: Args): StaticHan
 // Get object with range if needed
 // Due to https://github.com/cloudflare/workers-sdk/issues/6047
 // We cannot send a Headers instance to Miniflare
-constobj=
+constobj: R2ObjectBody=
 rangeResult.type==='partial'&&!isMiniflare
  ? awaitbucket?.get(key,{
 range: {
@@ -63,7 +68,25 @@ export const getHandler = ({ bucket, collection, prefix = '' }: Args): StaticHan
 }
 
 // Add R2-specific headers
-if(!isMiniflare){
+if(isMiniflare){
+// In development with Miniflare, manually set headers from httpMetadata
+constmetadata=obj.httpMetadata
+if(metadata?.cacheControl){
+headers.set('Cache-Control',metadata.cacheControl)
+}
+if(metadata?.contentDisposition){
+headers.set('Content-Disposition',metadata.contentDisposition)
+}
+if(metadata?.contentEncoding){
+headers.set('Content-Encoding',metadata.contentEncoding)
+}
+if(metadata?.contentLanguage){
+headers.set('Content-Language',metadata.contentLanguage)
+}
+if(metadata?.contentType){
+headers.set('Content-Type',metadata.contentType)
+}
+}else{
 obj.writeHttpMetadata(headers)
 }
 

packages/storage-r2/src/types.ts

@@ -14,7 +14,7 @@ export interface R2GetOptions {
 }
 
 exportinterfaceR2Bucket{
-createMultipartUpload(key: string,options?: any): Promise<any>
+createMultipartUpload(key: string,options?: any): Promise<R2MultipartUpload>
 delete(keys: string|string[]): Promise<void>
 get(key: string,options?: R2GetOptions): Promise<any|null>
 head(key: string): Promise<any>
@@ -32,5 +32,60 @@ export interface R2Bucket {
 value: ArrayBuffer|ArrayBufferView|Blob|null|ReadableStream|string,
 options?: any,
 ): Promise<any>
-resumeMultipartUpload(key: string,uploadId: string): any
+resumeMultipartUpload(key: string,uploadId: string): R2MultipartUpload
+}
+
+interfaceR2HTTPMetadata{
+cacheControl?: string
+cacheExpiry?: Date
+contentDisposition?: string
+contentEncoding?: string
+contentLanguage?: string
+contentType?: string
+}
+
+exportinterfaceR2Object{
+readonlyetag: string
+readonlyhttpMetadata?: R2HTTPMetadata
+readonlykey: string
+readonlysize: number
+
+writeHttpMetadata(headers: Headers): void
+}
+exportinterfaceR2ObjectBodyextendsR2Object{
+getbody(): ReadableStream
+}
+
+exportinterfaceR2MultipartUpload{
+abort(): Promise<void>
+complete(uploadedParts: R2UploadedPart[]): Promise<R2Object>
+readonlykey: string
+readonlyuploadId: string
+uploadPart(
+partNumber: number,
+value: (ArrayBuffer|ArrayBufferView)|Blob|ReadableStream|string,
+options?: any,
+): Promise<R2UploadedPart>
+}
+
+exportinterfaceR2StorageClientUploadContext{
+key: string
+}
+exporttypeR2StorageClientUploadHandlerParams={
+chunkSize?: number
+prefix: string
+}
+
+exporttypeR2StorageMultipartUploadHandlerParams={
+collection: string
+fileName: string
+fileType: string
+multipartId?: string
+multipartKey?: string
+multipartNumber?: string
+}
+
+exportinterfaceR2UploadedPart{
+etag: string
+partNumber: number
 }