[PB-5111] Feat: upload File service

src/services/network/upload/upload-file.service.ts

@@ -0,0 +1,146 @@
+import { logger } from '../../../utils/logger.utils';
+import {
+  DELAYS_MS,
+  MAX_CONCURRENT_UPLOADS,
+  MAX_RETRIES,
+  UploadFilesInBatchesParams,
+  UploadFileWithRetryParams,
+} from './upload.types';
+import { DriveFileService } from '../../drive/drive-file.service';
+import { dirname, extname } from 'node:path';
+import { isAlreadyExistsError } from '../../../utils/errors.utils';
+import { stat } from 'node:fs/promises';
+import { EncryptionVersion } from '@internxt/sdk/dist/drive/storage/types';
+import { createFileStreamWithBuffer, tryUploadThumbnail } from '../../../utils/thumbnail.utils';
+
+export class UploadFileService {
+  static readonly instance = new UploadFileService();
+
+  async uploadFilesInChunks({
+    network,
+    filesToUpload,
+    folderMap,
+    bucket,
+    destinationFolderUuid,
+    currentProgress,
+    emitProgress,
+  }: UploadFilesInBatchesParams): Promise<number> {
+    let bytesUploaded = 0;
+
+    const chunks = this.chunkArray(filesToUpload, MAX_CONCURRENT_UPLOADS);
+
+    for (const chunk of chunks) {
+      await Promise.allSettled(
+        chunk.map(async (file) => {
+          const parentPath = dirname(file.relativePath);
+          const parentFolderUuid =
+            parentPath === '.' || parentPath === '' ? destinationFolderUuid : folderMap.get(parentPath);
+
+          if (!parentFolderUuid) {
+            logger.warn(`Parent folder not found for ${file.relativePath}, skipping...`);
+            return null;
+          }
+          const createdFileUuid = await this.uploadFileWithRetry({
+            file,
+            network,
+            bucket,
+            parentFolderUuid,
+          });
+          if (createdFileUuid) {
+            bytesUploaded += file.size;
+            currentProgress.bytesUploaded += file.size;
+            currentProgress.itemsUploaded++;
+          }
+          emitProgress();
+        }),
+      );
+    }
+    return bytesUploaded;
+  }
+
+  async uploadFileWithRetry({
+    file,
+    network,
+    bucket,
+    parentFolderUuid,
+  }: UploadFileWithRetryParams): Promise<string | null> {
+    for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
+      try {
+        const stats = await stat(file.absolutePath);
+        if (!stats.size) {
+          logger.warn(`Skipping empty file: ${file.relativePath}`);
+          return null;
+        }
+
+        const fileType = extname(file.absolutePath).replaceAll('.', '');
+        const { fileStream, bufferStream } = createFileStreamWithBuffer({
+          path: file.absolutePath,
+          fileType,
+        });
+
+        const fileId = await new Promise<string>((resolve, reject) => {
+          network.uploadFile(
+            fileStream,
+            stats.size,
+            bucket,
+            (err: Error | null, res: string | null) => {
+              if (err) {
+                return reject(err);
+              }
+              resolve(res as string);
+            },
+            () => {},
+          );
+        });
+
+        const createdDriveFile = await DriveFileService.instance.createFile({
+          plainName: file.name,
+          type: fileType,
+          size: stats.size,
+          folderUuid: parentFolderUuid,
+          fileId,
+          bucket,
+          encryptVersion: EncryptionVersion.Aes03,
+          creationTime: stats.birthtime?.toISOString(),
+          modificationTime: stats.mtime?.toISOString(),
+        });
+
+        if (bufferStream) {
+          void tryUploadThumbnail({
+            bufferStream,
+            fileType,
+            userBucket: bucket,
+            fileUuid: createdDriveFile.uuid,
+            networkFacade: network,
+          });
+        }
+
+        return createdDriveFile.fileId;
+      } catch (error: unknown) {
+        if (isAlreadyExistsError(error)) {
+          const msg = `File ${file.name} already exists, skipping...`;
+          logger.info(msg);
+          return null;
+        }
+
+        if (attempt < MAX_RETRIES) {
+          const delay = DELAYS_MS[attempt];
+          const retryMsg = `Failed to upload file ${file.name}, retrying in ${delay}ms...`;
+          logger.warn(`${retryMsg} (attempt ${attempt + 1}/${MAX_RETRIES + 1})`);
+          await new Promise((resolve) => setTimeout(resolve, delay));
+        } else {
+          logger.error(`Failed to upload file ${file.name} after ${MAX_RETRIES + 1} attempts`);
+          return null;
+        }
+      }
+    }
+    return null;
+  }
+  private chunkArray<T>(array: T[], chunkSize: number): T[][] {
+    const chunks: T[][] = [];
+    for (let i = 0; i < array.length; i += chunkSize) {
+      chunks.push(array.slice(i, i + chunkSize));
+    }
+    return chunks;
+  }
+}

src/utils/thumbnail.utils.ts

@@ -1,3 +1,10 @@
+import { Readable } from 'node:stream';
+import { NetworkFacade } from '../services/network/network-facade.service';
+import { ThumbnailService } from '../services/thumbnail.service';
+import { ErrorUtils } from './errors.utils';
+import { BufferStream } from './stream.utils';
+import { createReadStream } from 'node:fs';
+
 export const ThumbnailConfig = {
   MaxWidth: 300,
   MaxHeight: 300,
@@ -41,3 +48,47 @@ export const isPDFThumbnailable = (fileType: string) => {
 export const isImageThumbnailable = (fileType: string) => {
   return fileType.trim().length > 0 && thumbnailableImageExtension.includes(fileType.trim().toLowerCase());
 };
+
+export const tryUploadThumbnail = async ({
+  bufferStream,
+  fileType,
+  userBucket,
+  fileUuid,
+  networkFacade,
+}: {
+  bufferStream: BufferStream;
+  fileType: string;
+  userBucket: string;
+  fileUuid: string;
+  networkFacade: NetworkFacade;
+}) => {
+  try {
+    const thumbnailBuffer = bufferStream.getBuffer();
+    if (thumbnailBuffer) {
+      await ThumbnailService.instance.uploadThumbnail(thumbnailBuffer, fileType, userBucket, fileUuid, networkFacade);
+    }
+  } catch (error) {
+    ErrorUtils.report(error);
+  }
+};
+
+export const createFileStreamWithBuffer = ({
+  path,
+  fileType,
+}: {
+  path: string;
+  fileType: string;
+}): {
+  bufferStream?: BufferStream;
+  fileStream: Readable;
+} => {
+  const readable: Readable = createReadStream(path);
+  if (isFileThumbnailable(fileType)) {
+    const bufferStream = new BufferStream();
+    return {
+      bufferStream,
+      fileStream: readable.pipe(bufferStream),
+    };
+  }
+  return { fileStream: readable };
+};