S3 Multi-Part Upload: Part 2 Conclusion

# serverless.yml
provider:
    ....
    iamRoleStatements:
            ......
        - Effect: Allow
          Action:
            - s3:GetObject
            - s3:PutObject
            - s3:AbortMultipartUpload
          Resource:
            - arn:aws:s3:::${Bucket}/*
        - Effect: Allow
          Action:
            - s3:ListBucketMultipartUploads
          Resource:
            - arn:aws:s3:::${Bucket}

/**
 * Create a multipart upload for a given S3 bucket and key.
 *
 * @param {string} bucket - The S3 bucket name.
 * @param {string} key - The S3 object key.
 * @returns {Promise<string>} The upload ID for the multipart upload.
 * @throws {Error} If there's an issue with the multipart upload creation.
 */
const createMultipartUpload = async (bucket, key) => {
  if (typeof bucket !== 'string' || !bucket.trim()) {
    throw new Error('Invalid bucket name. Please provide a valid string value for the bucket.')
  }

  if (typeof key !== 'string' || !key.trim()) {
    throw new Error('Invalid object key. Please provide a valid string value for the key.')
  }

  try {
    const params = {
      Bucket: bucket,
      Key: key
    }

    // Use S3's createMultipartUpload with promise()
    const data = await S3.createMultipartUpload(params).promise()

    return data.UploadId
  } catch (error) {
    throw new Error(`Error creating multipart upload: ${error.message}`)
  }
}

// Below is the chunk result of the list of metadata of files fetched from s3.
chunkifyArray = [
  [
    {
      fileName: 'A.csv',
      ContentLength: 1728033, // 1.64798069 MiB
    },
    {
      fileName: 'B.csv',
      ContentLength: 53326970, // 50.856561661 MiB
    },
  ],
  {
    fileName: 'C.csv',
    ContentLength: 21646619, // 20.643824577 MiB
  },
  [
    {
      fileName: 'D.csv',
      ContentLength: 1728033, // 1.64798069 MiB
    },
    {
      fileName: 'E.csv',
      ContentLength: 5226970, // 4.984827042 MiB
    },
  ],
]

/**
 * Converts an array of file metadata into chunks based on a size threshold.
 *
 * @param {Object[]} data - Array of file metadata objects.
 * @param {number} THRESHOLDLIMIT_5MB - Size threshold for chunking.
 * @returns {Object[]} Array of chunks with grouped file metadata.
 */
const convertToChunks = async (data, THRESHOLDLIMIT_5MB) => {
  const chunkifyArray = [];
  let totalSize = 0;

  data.map((data) => {
    if (data.ContentLength) {
      totalSize += data.ContentLength;

      if (chunkifyArray.length === 0) {
        if (data.ContentLength >= THRESHOLDLIMIT_5MB) {
          chunkifyArray.push(data);
        } else {
          const obj = {
            content: [data],
            size: data.ContentLength,
          };
          chunkifyArray.push(obj);
        }
      } else {
        const currentRec = chunkifyArray[chunkifyArray.length - 1];

        if (
          currentRec.size === undefined ||
          (currentRec.size !== undefined && currentRec.size > THRESHOLDLIMIT_5MB)
        ) {
          if (data.ContentLength >= THRESHOLDLIMIT_5MB) {
            chunkifyArray.push(data);
          } else {
            const obj = {
              content: [data],
              size: data.ContentLength,
            };
            chunkifyArray.push(obj);
          }
        } else {
          // push into existing element of chunkifyArray
          chunkifyArray[chunkifyArray.length - 1].content.push(data);
          chunkifyArray[chunkifyArray.length - 1].size += data.ContentLength;
        }
      }
    }
  });

  return { chunkifyArray, totalSize } ;
};

/**
 * Uploads a part of a multipart upload to an S3 bucket.
 *
 * @param {Buffer | Uint8Array | string} body - The content of the part to upload.
 * @param {string} bucket - The name of the S3 bucket.
 * @param {string} key - The key (path) where the part will be stored in the bucket.
 * @param {number} partNumber - The part number for the multipart upload.
 * @param {string} uploadId - The ID of the multipart upload.
 * @returns {object} - The ETag and partNumber of the uploaded part.
 * @throws {Error} - If any validation or upload error occurs.
 */
const uploadMultiPartHelper = async (body, bucket, key, partNumber, uploadId) => {
  try {
    const params = {
      Body: body,
      Bucket: bucket,
      Key: key,
      PartNumber: partNumber,
      UploadId: uploadId
    }
    const data = await S3.uploadPart(params).promise()
    return {
      ETag: data.ETag,
      PartNumber: partNumber
    }
  } catch (error) {
    throw new Error(`Upload failed: ${error.message}`)
  }
}

/**
 * Validates the scenario where the total size of all files doesn't exceed 5 MiB.
 *
 * @param {Object[]} data - Array of file metadata objects.
 * @param {number} totalSize - The total size of all files.
 * @param {number} THRESHOLDLIMIT_5MB - Size threshold for validation.
 * @returns {boolean} Indicates whether the total size is under the threshold.
 */
const chunkDataWriteIntoStream = async (initalArrayOfFileMetaData, THRESHOLDLIMIT_5MB ) => {
  const { chunkifyArray, totalSize } = await convertToChunks( initalArrayOfFileMetaData, THRESHOLDLIMIT_5MB );
  if (totalSize < THRESHOLDLIMIT_5MB) {
    // If we're under the 5 MiB mark,
    // let's manually handle the upload using s3.upload() by combining all files into one.
  } else {
    // As the total size is bigger than 5MB, handle using multipart chunks upload...

    const uploadId = await createMultipartUpload(bucket, key);
        const respArr = []
    for (let i = 0; i < chunkifyArray.length; i++) {
      const partNumber = i + 1;
      const body = fetchCombinedRecordsFromMultipleFileObjects(chunkifyArray[i]); // create this function to fetch data from chunks.      
            // push the chunkifyArray[i] record into multipart stream having uploadID and partnumber 'i'.
            const uploadResponse = await uploadMultiPartHelper( body, bucket, key, partNumber, uploadId );
        respArr.push(uploadResponse)
        }
        return respArr 
        // respArr = [
        //    { 
      //      ETag: '',
      //      PartNumber: 0
      //    }
        //  ]
  }
};

/**
 * Completes a multipart upload to an S3 bucket and returns the uploaded object's location.
 *
 * @param {string} bucket - The name of the S3 bucket.
 * @param {string} key - The key or path for the uploaded object.
 * @param {Array<{ ETag: string, PartNumber: number }>} partArray - An array of parts with ETag and PartNumber.
 * @param {string} uploadId - The unique upload identifier for the multipart upload.
 * @returns {Promise<string>} A Promise that resolves to the uploaded object's location.
 * @throws {Error} Throws an error if the upload fails.
 */
const completeMultiPartUpload = async (bucket, key, partArray, uploadId) => {
  try {
    const params = {
      Bucket: bucket,
      Key: key,
      MultipartUpload: {
        Parts: partArray
      },
      UploadId: uploadId
    }

    const data = await S3.completeMultipartUpload(params).promise()
    return data.Location
  } catch (error) {
    throw new Error(error.message)
  }
}

/**
 * Aborts a multipart upload in an S3 bucket.
 *
 * @param {string} bucket - The S3 bucket name.
 * @param {string} key - The S3 object key.
 * @param {string} uploadId - The upload ID of the multipart upload.
 * @returns {Promise<Object>} A promise that resolves with the response from the S3 service.
 * @throws {Error} If any validation fails or an error occurs during the operation.
 */
const abortMultiPartHelper = async (bucket, key, uploadId) => {
  // Validation checks for bucket, key, and uploadId
  // It's like checking the keys before opening the treasure chest! 🔑💰

  try {
    const params = {
      Bucket: bucket,
      Key: key,
      UploadId: uploadId
    }
    const data = await S3.abortMultipartUpload(params).promise()
    return data
  } catch (error) {
    throw new Error(`Error during abortMultiPartHelper: ${error.message}`)
  }
}

/**
 * Aborts a specific multipart stream based on the provided uploadId.
 *
 * @param {string} bucket - The S3 bucket name.
 * @param {string} key - The S3 object key.
 * @param {string} uploadId - The unique ID of the multipart stream to be aborted.
 * @returns {Promise<Object>} Resolves with the response from S3.
 * @throws {Error} If any validation fails or an error occurs during the process.
 */
const abortSpecificStream = async (bucket, key, uploadId) => {
  try {
    const params = {
      Bucket: bucket,
      Key: key,
      UploadId: uploadId
    }
    const data = await S3.abortMultipartUpload(params).promise()
    return data 
  } catch (error) {
    throw new Error(`Error during abortSpecificStream: ${error.message}`)
  }
}

/**
 * Lists in-progress multipart uploads on a specific bucket.
 *
 * @param {string} bucket - The S3 bucket name.
 * @returns {Promise<Object>} A promise that resolves with the response containing in-progress multipart uploads.
 * @throws {Error} If any validation fails or an error occurs during the operation.
 */
const listMultiPartUploads = async (bucket) => {
  try {
    const params = {
      Bucket: bucket
    };
    const data = await S3.listMultipartUploads(params).promise();
    return data;
  } catch (error) {
    throw new Error(`Error during listMultiPartUploads: ${error.message}`);
  }
};
/**
 * Aborts all open multipart uploads for a given S3 bucket.
 *
 * @param {string} bucket - The S3 bucket name.
 * @returns {Promise<Object[]>} A promise resolving to an array containing information about aborted uploads.
 * @throws {Error} Throws an error if the operation encounters any issues.
 */
module.exports.abortMultiPart = async (bucket) => {
  try {
    // Fetches information about open multipart uploads
    const data = await listMultiPartUploads(bucket);
    const output = [];

    // Iterates through each open upload and aborts it
    for (const obj of data) {
      const key = obj.Key;
      const uploadId = obj.UploadId;
      // Aborts the multipart upload
      const response = await abortMultiPartHelper(bucket, key, uploadId);
      // Records the abort response for the upload
      output.push(response);
    }

    return output; // Returns an array containing information about aborted uploads
  } catch (e) {
    throw new Error(e.message); // Throws an error if any issues occur during the operation
  }
}