FEAT: add upload init multi part upload

src/modules/pipelinesV2/interfaces.ts

@@ -67,9 +67,23 @@ export interface IGetPipelineLogsRequest {
   details: string;
 }
 
-export class IUploadCSVFile {
+export class IInitUploadCSVFile {
   @ApiProperty()
   file_name: string;
+
+  @ApiProperty()
+  parts: number;
+}
+
+export class ICompleteUploadCSVFile {
+  @ApiProperty()
+  upload_id: string;
+
+  @ApiProperty()
+  file_name: string;
+
+  @ApiProperty()
+  parts: { ETag: string; PartNumber: number }[];
 }
 
 export class ICreatePipelineCSVFile {

src/modules/pipelinesV2/pipelines.controller.ts

@@ -33,10 +33,11 @@ import { PackTheMetadata } from 'src/utils/ PackTheMetadata';
 
 import { PipelinesService as OldPipelineService } from 'src/modules/pipelines/pipelines.service';
 import {
+  ICompleteUploadCSVFile,
   ICreatePipelineCSVFile,
   ICreatePipelineV2Req,
   IPipelineV2,
-  IUploadCSVFile,
+  IInitUploadCSVFile,
 } from './interfaces';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
 
@@ -297,22 +298,27 @@ export class PipelinesController {
     this.logger.info('PipelinesController - delete: OK');
   }
 
-  @Post('/upload')
+  @Post('/init-upload')
   @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
-  async uploadFile(@User() user: RequestUser, @Body() body: IUploadCSVFile) {
-    this.logger.info('/upload - Upload Connector Route');
+  async initUploadFile(
+    @User() user: RequestUser,
+    @Body() body: IInitUploadCSVFile,
+  ) {
+    this.logger.info('/upload - Init Upload File Route');
     const metadata = PackTheMetadata({ ...user });
 
-    const { file_name } = body;
+    const { file_name, parts } = body;
 
-    const { url } = await this.pipelinesClientService.uploadFile(
-      {
-        name: file_name,
-      },
-      metadata,
-    );
+    const { urls, upload_id } =
+      await this.pipelinesClientService.initUploadFile(
+        {
+          name: file_name,
+          parts: String(parts),
+        },
+        metadata,
+      );
 
-    if (!url) {
+    if (!urls) {
       this.logger.info('pipeline/upload - Failed File pipeline');
       throw new HttpException(
         'Upload failed, try again in a few minutes, if the problem persists, contact support.',
@@ -320,7 +326,24 @@ export class PipelinesController {
       );
     }
 
-    return { url };
+    return { urls: JSON.parse(urls), upload_id };
+  }
+
+  @Post('/complete-upload')
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
+  async completeUploadFile(
+    @User() user: RequestUser,
+    @Body() body: ICompleteUploadCSVFile,
+  ) {
+    this.logger.info('/upload - Complete Upload File Route');
+    const metadata = PackTheMetadata({ ...user });
+
+    const { upload_id, parts, file_name } = body;
+
+    return await this.pipelinesClientService.completeUploadFile(
+      { upload_id, parts, file_name },
+      metadata,
+    );
   }
 
   @Post('/file')

src/modules/pipelinesV2/pipelines.service.ts

@@ -261,9 +261,10 @@ export class PipelinesService implements OnModuleInit {
     return { objects: JSON.parse(objects) };
   }
 
-  async uploadFile(uploadFile, metadata) {
+  async initUploadFile(uploadFile, metadata) {
     const body: Messages.PipelineV2UploadFileRequest = {
       name: uploadFile.name,
+      parts: uploadFile.parts,
     };
 
     return lastValueFrom(
@@ -289,4 +290,21 @@ export class PipelinesService implements OnModuleInit {
 
     return url;
   }
+
+  async completeUploadFile(uploadFile, metadata) {
+    const body: Messages.PipelineV2CompleteUploadFileRequest = {
+      upload_id: uploadFile.name,
+      parts: uploadFile.parts,
+      name: uploadFile.file_name,
+    };
+
+    return lastValueFrom(
+      this.pipelineWriteService.PipelineV2CompleteUploadFile(body, metadata),
+    ).catch((err) => {
+      throw new HttpException(
+        err.details,
+        err.code === 6 ? HttpStatus.CONFLICT : 400,
+      );
+    });
+  }
 }