feat(storage/s3): use @aws-sdk/lib-storage for uploads

Author: LouisHaftmann

lib/storage/drivers/s3.ts

@@ -6,15 +6,15 @@ import {
   GetObjectCommand,
   HeadBucketCommand,
   ListObjectsV2Command,
-  PutObjectCommand,
   S3Client,
 } from '@aws-sdk/client-s3'
-import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
+import { Upload } from '@aws-sdk/lib-storage'
 
+import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
 import * as R from 'remeda'
 import { z } from 'zod'
 import { parseEnv, StorageDriver } from '~/lib/storage/defineStorageDriver'
-import { createTempDir, streamToBuffer } from '~/lib/utils'
+import { createTempDir } from '~/lib/utils'
 
 export class S3StorageDriver extends StorageDriver {
   s3
@@ -96,16 +96,20 @@ export class S3StorageDriver extends StorageDriver {
   }
 
   async uploadPart(opts: { uploadId: string; partNumber: number; data: ReadableStream }) {
-    await this.s3.send(
-      new PutObjectCommand({
+    const upload = new Upload({
+      client: this.s3,
+      params: {
         Bucket: this.bucket,
         Key: this.getUploadPartObjectName({
           uploadId: opts.uploadId,
           partNumber: opts.partNumber,
         }),
-        Body: await streamToBuffer(opts.data),
-      }),
-    )
+        Body: opts.data,
+      },
+      partSize: 64 * 1024 * 1024, // 64 MB
+      queueSize: 1,
+    })
+    await upload.done()
   }
 
   async completeMultipartUpload(opts: {
@@ -147,13 +151,17 @@ export class S3StorageDriver extends StorageDriver {
     await outputTempFile.close()
 
     const readStream = createReadStream(outputTempFilePath)
-    await this.s3.send(
-      new PutObjectCommand({
+    const upload = new Upload({
+      client: this.s3,
+      params: {
         Bucket: this.bucket,
         Key: this.addBaseFolderPrefix(opts.finalOutputObjectName),
         Body: readStream,
-      }),
-    )
+      },
+      partSize: 64 * 1024 * 1024, // 64 MB
+      queueSize: 1,
+    })
+    await upload.done()
 
     await Promise.all([
       this.cleanupMultipartUpload(opts.uploadId),

lib/utils.ts

@@ -3,19 +3,6 @@ import { promises as fs } from 'node:fs'
 import path from 'node:path'
 import { ENV } from './env'
 
-export async function streamToBuffer(stream: ReadableStream<Uint8Array>): Promise<Buffer> {
-  const reader = stream.getReader()
-  const chunks: Uint8Array[] = []
-
-  while (true) {
-    const { done, value } = await reader.read()
-    if (done) break
-    chunks.push(value)
-  }
-
-  return Buffer.concat(chunks)
-}
-
 export function getObjectNameFromKey(key: string, version: string) {
   return hash('sha1', Buffer.from(`${key}-${version}`))
 }

package.json

@@ -26,6 +26,7 @@
   "dependencies": {
     "@antfu/utils": "^9.2.0",
     "@aws-sdk/client-s3": "^3.740.0",
+    "@aws-sdk/lib-storage": "^3.846.0",
     "@aws-sdk/s3-request-presigner": "^3.740.0",
     "@google-cloud/storage": "^7.15.0",
     "@types/better-sqlite3": "^7.6.12",