feat: 🎨 handle multipart uploads

Joystream · May 16, 2024 · b45260b · b45260b
1 parent c269c10
commit b45260b
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 13 deletions.
diff --git a/storage-node/src/services/storageProviders/IConnectionHandler.ts b/storage-node/src/services/storageProviders/IConnectionHandler.ts
@@ -7,10 +7,10 @@ export interface IConnectionHandler {
   /**
    * Asynchronously uploads a file to the remote bucket.
    * @param filename - The key of the file in the remote bucket.
-   * @param filestream - The file stream to upload.
+   * @param filePath - The file path of the file to upload.
    * @returns A promise that resolves when the upload is complete or rejects with an error.
    */
-  uploadFileToRemoteBucket(filename: string, filestream: ColossusFileStream): Promise<any>
+  uploadFileToRemoteBucket(filename: string, filePath: string): Promise<any>
 
   /**
    * Asynchronously retrieves a file from the remote bucket.

diff --git a/storage-node/src/services/storageProviders/awsConnectionHandler.ts b/storage-node/src/services/storageProviders/awsConnectionHandler.ts
@@ -1,12 +1,14 @@
 import { IConnectionHandler, ColossusFileStream } from './IConnectionHandler'
 import {
+  CreateMultipartUploadCommand,
   GetObjectCommand,
   ListObjectsCommand,
   ListObjectsCommandInput,
   PutObjectCommand,
   S3Client,
 } from '@aws-sdk/client-s3'
 import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
+import fs from 'fs'
 
 export type AwsConnectionHandlerParams = {
   accessKeyId: string
@@ -19,6 +21,9 @@ export class AwsConnectionHandler implements IConnectionHandler {
   private client: S3Client
   private bucket: string
 
+  // Official doc at https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html: Upload an object in a single operation by using the AWS SDKs, REST API, or AWS CLI – With a single PUT operation, you can upload a single object up to 5 GB in size.
+  private multiPartThresholdGB = 5
+
   constructor(opts: AwsConnectionHandlerParams) {
     this.client = new S3Client({
       credentials: {
@@ -34,18 +39,24 @@ export class AwsConnectionHandler implements IConnectionHandler {
     // Response status code info: https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
     return response.$metadata.httpStatusCode === 200
   }
+  private isMultiPartNeeded(filePath: string): boolean {
+    const stats = fs.statSync(filePath)
+    const fileSizeInBytes = stats.size
+    const fileSizeInGigabytes = fileSizeInBytes / (1024 * 1024 * 1024)
+    return fileSizeInGigabytes > this.multiPartThresholdGB
+  }
 
-  async uploadFileToRemoteBucket(filename: string, filestream: ColossusFileStream): Promise<any> {
-    // Setting up S3 upload parameters
-
+  async uploadFileToRemoteBucket(filename: string, filePath: string): Promise<any> {
     const input = {
       Bucket: this.bucket,
-      Key: filename, // File name you want to save as in S3
-      Body: filestream,
+      Key: filename,
+      Body: filePath,
     }
 
-    // Uploading files to the bucket
-    const command = new PutObjectCommand(input)
+    // Uploading files to the bucket: multipart
+    const command = this.isMultiPartNeeded(filePath)
+      ? new CreateMultipartUploadCommand(input)
+      : new PutObjectCommand(input)
     const response = await this.client.send(command)
     if (!this.isSuccessfulResponse(response)) {
       throw new Error('Failed to upload file to S3')

diff --git a/storage-node/src/services/sync/acceptPendingObjects.ts b/storage-node/src/services/sync/acceptPendingObjects.ts
@@ -185,7 +185,7 @@ export class AcceptPendingObjectsService {
           await moveFile(currentPath, newPath)
         } else {
           const connection = getStorageProviderConnection()!
-          await connection.uploadFileToRemoteBucket(dataObjectId, fs.createReadStream(currentPath)) // NOTE: consider converting to non blocking promise
+          await connection.uploadFileToRemoteBucket(dataObjectId, currentPath) // NOTE: consider converting to non blocking promise
           await fsPromises.unlink(currentPath) // delete the file from the local storage after successful upload
         }
         registerNewDataObjectId(dataObjectId)

diff --git a/storage-node/src/services/sync/tasks/StorageProviderSyncTask.ts b/storage-node/src/services/sync/tasks/StorageProviderSyncTask.ts
@@ -37,8 +37,7 @@ export class ProviderSyncTask extends DownloadFileTask {
       // TODO: I have added a HashFileVerificationError to the utils file, but it is not used here, we should establish what to do in case the file is corrupted
       await withRandomUrls(operatorUrls, async (chosenBaseUrl) => {
         await this.tryDownloadTemp(chosenBaseUrl, this.dataObjectId)
-        const fileStream = fs.createReadStream(tempFilePath)
-        await this.connection.uploadFileToRemoteBucket(this.dataObjectId, fileStream) // NOTE: consider converting to non blocking promise
+        await this.connection.uploadFileToRemoteBucket(this.dataObjectId, tempFilePath) // NOTE: consider converting to non blocking promise
       })
     } catch (err) {
       logger.error(`Sync - error when synching asset ${this.dataObjectId} with remote storage provider: ${err}`, {

diff --git a/storage-node/src/services/webApi/app.ts b/storage-node/src/services/webApi/app.ts
@@ -91,7 +91,6 @@ export async function createApp(config: AppConfig): Promise<Express> {
           // For multipart forms, the max number of file fields (Default: Infinity)
           files: 1,
           // For multipart forms, the max file size (in bytes) (Default: Infinity)
-          // TODO CS3 : This should be set to the maximum file size allowed by the storage provider.
           fileSize: config.maxFileSize,
         },
       },