FEAT: Create Upload CSV Pipeline

src/modules/pipelinesV2/interfaces.ts

@@ -66,3 +66,23 @@ export interface IGetPipelineLogsRequest {
   id: string;
   details: string;
 }
+
+export class IUploadCSVFile {
+  @ApiProperty({ format: 'binary' })
+  file: string;
+
+  @ApiProperty()
+  name: string;
+
+  @ApiProperty()
+  description: string;
+
+  @ApiProperty()
+  encoding: string;
+
+  @ApiProperty()
+  sep: string;
+
+  @ApiProperty()
+  header: boolean;
+}

src/modules/pipelinesV2/pipelines.controller.ts

@@ -14,7 +14,7 @@ import {
   HttpStatus,
   Patch,
   UseInterceptors,
-  UploadedFiles,
+  UploadedFile,
 } from '@nestjs/common';
 import {
   ApiConsumes,
@@ -34,9 +34,13 @@ import { RequestUser, User } from 'src/authentication/user.decorator';
 import { PackTheMetadata } from 'src/utils/ PackTheMetadata';
 
 import { PipelinesService as OldPipelineService } from 'src/modules/pipelines/pipelines.service';
-import { ICreatePipelineV2Req, IPipelineV2 } from './interfaces';
+import {
+  ICreatePipelineV2Req,
+  IPipelineV2,
+  IUploadCSVFile,
+} from './interfaces';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
-import { FilesInterceptor } from '@nestjs/platform-express';
+import { FileInterceptor } from '@nestjs/platform-express';
 
 @UseFilters(new GrpcToHttpExceptionFilter())
 @ApiTags('PipelinesV2')
@@ -251,21 +255,53 @@ export class PipelinesController {
   @Post('/upload')
   @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
   @ApiConsumes('multipart/form-data')
-  @UseInterceptors(FilesInterceptor('file'))
-  async uploadFile(@User() user: RequestUser, @UploadedFiles() files) {
+  @UseInterceptors(
+    FileInterceptor('file', { limits: { fileSize: 50 * 1000 * 1000 + 1 } }),
+  )
+  async uploadFile(
+    @User() user: RequestUser,
+    @UploadedFile() file,
+    @Body() body: IUploadCSVFile,
+  ) {
     this.logger.info('/upload - Upload Connector Route');
     const metadata = PackTheMetadata({ ...user });
-    const promises = files.map(async (file) => {
-      return await this.pipelinesClientService.uploadFile(
-        {
-          file,
-          name: file.originalname.split('.')[0],
-        },
-        metadata,
-      );
-    });
+    let name = `${new Date().getTime()}_${file.originalname.split('.')[0]}`;
+    body.name ? (name = `${new Date().getTime()}_${body.name}`) : name;
+    const response = await this.pipelinesClientService.uploadFile(
+      {
+        file,
+        name,
+      },
+      metadata,
+    );
+
+    const { sep, header, encoding, description } = body;
+
+    const p = {
+      connection_id: process.env.UPLOAD_FILE_AGENT_CONNECTION,
+      connector_name: 'Amazon S3',
+      connector_plugin: 'aws_s3',
+      connector_version: '1.0.0',
+      image_url: 'https://assets.dadosfera.ai/images/connectors/csv.svg',
+      name: body.name || file.originalname.split('.')[0],
+      description,
+      transformations_ids: [],
+      tags: [],
+      cron: '@once',
+      config: { cron: '@once', tables: [] },
+      properties: {
+        engine: 'csv',
+        source_bucket: process.env.BUCKET_CUSTOMER_CSV_ASSETS,
+        source_prefix: name,
+        file_format_params: { sep, encoding, header: Boolean(header) },
+        is_a_upload_csv: true,
+      },
+      input_id: undefined,
+    };
+
+    const pipeline = await this.pipelinesClientService.create(p, metadata);
+    console.log(pipeline);
 
-    const response = await Promise.all(promises);
     return response;
   }
 }