feat: add uploadMultipartFromStream method for multi-part encrypted uploads from a Readable stream

Author: larryrider

src/services/network/network-facade.service.ts

@@ -6,17 +6,26 @@ import {
   DownloadFileFunction,
   EncryptFileFunction,
   UploadFileFunction,
+  UploadFileMultipartFunction,
 } from '@internxt/sdk/dist/network';
 import { Environment } from '@internxt/inxt-js';
 import { randomBytes } from 'node:crypto';
 import { Readable, Transform } from 'node:stream';
-import { DownloadOptions, UploadOptions, UploadProgressCallback, DownloadProgressCallback } from '../../types/network.types';
+import {
+  DownloadOptions,
+  UploadOptions,
+  UploadProgressCallback,
+  DownloadProgressCallback,
+  UploadMultipartOptions,
+  UploadTask,
+} from '../../types/network.types';
 import { CryptoService } from '../crypto.service';
 import { UploadService } from './upload.service';
 import { DownloadService } from './download.service';
 import { ValidationService } from '../validation.service';
 import { HashStream } from '../../utils/hash.utils';
 import { RangeOptions } from '../../utils/network.utils';
+import { queue, QueueObject } from 'async';
 
 export class NetworkFacade {
   private readonly cryptoLib: Network.Crypto;
@@ -183,4 +192,138 @@ export class NetworkFacade {
 
     return [uploadOperation(), abortable];
   }
+
+  /**
+   * Performs a multi-part upload encrypting the stream content
+   *
+   * @param bucketId The bucket where the file will be uploaded
+   * @param mnemonic The plain mnemonic of the user
+   * @param size The total size of the stream content
+   * @param from The source ReadStream to upload from
+   * @param options The upload options
+   * @returns A promise to execute the upload and an abort controller to cancel the upload
+   */
+  async uploadMultipartFromStream(
+    bucketId: string,
+    mnemonic: string,
+    size: number,
+    from: Readable,
+    options: UploadMultipartOptions,
+  ): Promise<[Promise<{ fileId: string; hash: Buffer }>, AbortController]> {
+    const hashStream = new HashStream();
+    const abortable = options?.abortController ?? new AbortController();
+    let encryptionTransform: Transform;
+    let hash: Buffer;
+
+    const partsUploadedBytes: Record<number, number> = {};
+    const fileParts: { PartNumber: number; ETag: string }[] = [];
+
+    const onProgress = (partId: number, loadedBytes: number) => {
+      if (!options?.progressCallback) return;
+      partsUploadedBytes[partId] = loadedBytes;
+      const currentTotalLoadedBytes = Object.values(partsUploadedBytes).reduce((a, p) => a + p, 0);
+      const reportedProgress = Math.round((currentTotalLoadedBytes / size) * 100);
+      options.progressCallback(reportedProgress);
+    };
+
+    const encryptFile: EncryptFileFunction = async (_, key, iv) => {
+      const encryptionCipher = this.cryptoService.getEncryptionTransform(
+        Buffer.from(key as ArrayBuffer),
+        Buffer.from(iv as ArrayBuffer),
+      );
+      const streamInParts = this.cryptoService.encryptStreamInParts(from, encryptionCipher, size, options.parts);
+      encryptionTransform = streamInParts.pipe(hashStream);
+    };
+
+    const uploadFileMultipart: UploadFileMultipartFunction = async (urls: string[]) => {
+      let partIndex = 0;
+      const limitConcurrency = 6;
+
+      const worker = async (upload: UploadTask) => {
+        const { etag } = await this.uploadService.uploadFile(upload.urlToUpload, upload.contentToUpload, {
+          abortController: abortable,
+          progressCallback: (loadedBytes: number) => {
+            onProgress(upload.index, loadedBytes);
+          },
+        });
+
+        fileParts.push({
+          ETag: etag,
+          PartNumber: upload.index + 1,
+        });
+      };
+
+      const uploadQueue: QueueObject<UploadTask> = queue<UploadTask>(function (task, callback) {
+        worker(task)
+          .then(() => {
+            callback();
+          })
+          .catch((e) => {
+            callback(e);
+          });
+      }, limitConcurrency);
+
+      for await (const chunk of encryptionTransform) {
+        const part: Buffer = chunk;
+
+        if (uploadQueue.running() === limitConcurrency) {
+          await uploadQueue.unsaturated();
+        }
+
+        if (abortable.signal.aborted) {
+          throw new Error('Upload cancelled by user');
+        }
+
+        let errorAlreadyThrown = false;
+
+        uploadQueue
+          .pushAsync({
+            contentToUpload: part,
+            urlToUpload: urls[partIndex],
+            index: partIndex++,
+          })
+          .catch((err: Error) => {
+            if (errorAlreadyThrown) return;
+
+            errorAlreadyThrown = true;
+            if (err) {
+              uploadQueue.kill();
+              if (!abortable?.signal.aborted) {
+                abortable.abort();
+              }
+            }
+          });
+      }
+
+      while (uploadQueue.running() > 0 || uploadQueue.length() > 0) {
+        await uploadQueue.drain();
+      }
+
+      hash = hashStream.getHash();
+      return {
+        hash: hash.toString('hex'),
+        parts: fileParts.sort((pA, pB) => pA.PartNumber - pB.PartNumber),
+      };
+    };
+
+    const uploadOperation = async () => {
+      const uploadResult = await NetworkUpload.uploadMultipartFile(
+        this.network,
+        this.cryptoLib,
+        bucketId,
+        mnemonic,
+        size,
+        encryptFile,
+        uploadFileMultipart,
+        options.parts,
+      );
+
+      return {
+        fileId: uploadResult,
+        hash: hash,
+      };
+    };
+
+    return [uploadOperation(), abortable];
+  }
 }