Pipelines Refactor

src/modules/pipelines/client.service.ts

@@ -0,0 +1,277 @@
+import {
+  BadRequestException,
+  Inject,
+  InternalServerErrorException,
+  OnModuleInit,
+} from '@nestjs/common';
+import { ClientGrpc, Payload } from '@nestjs/microservices';
+import { PipelineServicesNames, PipelinesServiceInterface } from 'protospack';
+import { lastValueFrom } from 'rxjs';
+import {
+  objectCamelToSnake,
+  objectSnakeToCamel,
+} from 'src/utils/CaseConverter';
+
+import {
+  ICreatePipelineDto,
+  IGetPipelineLogsRequest,
+  IIdRequest,
+  IUpdatePipelineRequest,
+} from './interfaces';
+
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+import { PipelinesClientConfiguration } from './pipelines-client';
+
+export class PipelinesClientService implements OnModuleInit {
+  private pipelineService: PipelinesServiceInterface;
+  logger: DadosferaLogger;
+  constructor(
+    @Inject(DadosferaLogger)
+    dadosferaLogger: DadosferaLogger,
+    @Inject(PipelinesClientConfiguration.name)
+    private readonly grpcClient: ClientGrpc,
+  ) {
+    this.logger = dadosferaLogger.logger;
+  }
+
+  onModuleInit() {
+    this.pipelineService =
+      this.grpcClient.getService<PipelinesServiceInterface>(
+        PipelineServicesNames.PipelineService,
+      );
+  }
+
+  async create(@Payload() createPipelineDto: ICreatePipelineDto) {
+    this.logger.info('PipelinesClientService - Create');
+    const createPipelineResponse = await lastValueFrom(
+      this.pipelineService.Create(objectSnakeToCamel(createPipelineDto)),
+    )
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((error: { details: string }) => {
+        if (error.details.includes('INVALID_REQUEST')) {
+          // eslint-disable-next-line @typescript-eslint/no-unused-vars
+          const [errorType, message] = error.details.split('|');
+          throw new BadRequestException(message);
+        }
+        throw new InternalServerErrorException(error.details);
+      });
+
+    return createPipelineResponse;
+  }
+
+  async findAll(data: IIdRequest) {
+    this.logger.info('PipelinesClientService - FindAll');
+    const findAllPipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService.FindAll(objectSnakeToCamel(data)).subscribe({
+        next(x) {
+          resolve(x.pipelines);
+        },
+        error(err) {
+          reject(err);
+        },
+        complete() {
+          // console.log('done');
+        },
+      });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return findAllPipelineResponse;
+  }
+
+  async findOne(data: IIdRequest) {
+    this.logger.info('PipelinesClientService - FindOne');
+
+    const findOnePipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService.FindOne(objectSnakeToCamel(data)).subscribe({
+        next(x) {
+          resolve(x);
+        },
+        error(err) {
+          reject(err);
+        },
+        complete() {
+          // console.log('done');
+        },
+      });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return findOnePipelineResponse;
+  }
+
+  async update(UpdatePipelineRequest: IUpdatePipelineRequest) {
+    this.logger.info('PipelinesClientService - Update');
+
+    const updatePipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService
+        .Update(objectSnakeToCamel(UpdatePipelineRequest))
+        .subscribe({
+          next(x) {
+            resolve(x);
+          },
+          error(err) {
+            reject(err);
+          },
+          complete() {
+            //console.log('done');
+          },
+        });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return updatePipelineResponse;
+  }
+
+  async remove(data: IIdRequest) {
+    this.logger.info('PipelinesClientService - Remove');
+
+    const removePipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService.remove(objectSnakeToCamel(data)).subscribe({
+        next(x) {
+          resolve(objectCamelToSnake(x));
+        },
+        error(err) {
+          reject(err);
+        },
+        complete() {
+          // this.logger.info('Done');
+        },
+      });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return removePipelineResponse;
+  }
+
+  async getPipelineLogsMessages(data: IGetPipelineLogsRequest) {
+    this.logger.info('PipelinesClientService - GetPipelineLogsMessages');
+
+    const logsPipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService.getPipelineLogs(data).subscribe({
+        next(x) {
+          resolve(x);
+        },
+        error(err) {
+          reject(err);
+        },
+        complete() {
+          // this.logger.info('Done');
+        },
+      });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return logsPipelineResponse;
+  }
+
+  async getPipelineStatus(data) {
+    this.logger.info('PipelinesClientService - GetPipelineStatus');
+
+    const statusPipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService
+        .getPipelineStatus(objectSnakeToCamel(data))
+        .subscribe({
+          next(x) {
+            const statusArray = x.status.sort((a, b) => {
+              if (a.id < b.id) {
+                return 1;
+              } else {
+                return -1;
+              }
+            });
+            resolve({ status: statusArray });
+          },
+          error(err) {
+            reject(err);
+          },
+          complete() {
+            // this.logger.info('Done');
+          },
+        });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return statusPipelineResponse;
+  }
+
+  async runPipeline({ id, info }: IIdRequest) {
+    this.logger.info('PipelinesClientService - RunPipeline');
+    const statusPipelineResponse = await new Promise((resolve, reject) => {
+      this.pipelineService
+        .triggerPipeline(objectSnakeToCamel({ id, info }))
+        .subscribe({
+          next(x) {
+            if (x.status == false) {
+              reject(
+                'This pipeline is not ready yet to execute, Try again later!',
+              );
+            }
+            resolve(x);
+          },
+          error(err) {
+            this.logger.error('Observable Error');
+            reject(err);
+          },
+          complete() {
+            //this.logger.info('Done');
+          },
+        });
+    })
+      .then((res) => {
+        this.logger.info('Done');
+        return res;
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+
+    return statusPipelineResponse;
+  }
+}

src/modules/pipelines/interfaces.d.ts

@@ -0,0 +1,36 @@
+import { Info } from 'protospack/dist/lib/interfaces';
+
+export interface ICreatePipelineDto {
+  input: IdRequest;
+  transformations: IdRequest[];
+  output: IdRequest;
+  tags: string[];
+  name: string;
+  description: string;
+  info: Info;
+}
+
+export interface IdRequest {
+  id: string;
+}
+
+export interface IIdRequest {
+  id: string;
+  info: Info;
+}
+
+export interface IUpdatePipelineRequest {
+  input: IdRequest;
+  transformations: IdRequest[];
+  output: IdRequest;
+  tags: string[];
+  name: string;
+  description: string;
+  id: string;
+  info: Info;
+}
+
+export interface IGetPipelineLogsRequest {
+  id: string;
+  details: string;
+}

src/modules/pipelines/pipelines-client.ts

@@ -0,0 +1,29 @@
+import {
+  ClientsProviderAsyncOptions,
+  GrpcOptions,
+  Transport,
+} from '@nestjs/microservices';
+import { PipelinePackages, PipelineProtoFilePath } from 'protospack';
+import { credentials } from '@grpc/grpc-js';
+
+export class PipelinesClientConfiguration {
+  public name = 'PipelinesClientConfiguration';
+  private config: GrpcOptions = {
+    transport: Transport.GRPC,
+    options: {
+      url: process.env.PIFACTORY_URL,
+      package: PipelinePackages,
+      credentials: process.env.LOCAL_ENV ? undefined : credentials.createSsl(),
+      protoPath: PipelineProtoFilePath,
+      loader: {
+        enums: String,
+        objects: true,
+        arrays: true,
+      },
+    },
+  };
+  providerOptions: ClientsProviderAsyncOptions = {
+    name: this.name,
+    ...this.config,
+  };
+}

src/modules/pipelines/pipelines.controller.ts

@@ -9,7 +9,7 @@ import {
   Put,
 } from '@nestjs/common';
 import { ApiTags } from '@nestjs/swagger';
-import { PipelinesClientService } from 'src/clients/pipelines/client.service';
+import { PipelinesClientService } from './client.service';
 import { AuthenticateCondition } from 'src/authentication/authentication.decorator';
 import { PERMISSIONS } from '../../authentication/permissions.enum';
 import { PipelinesService } from './pipelines.service';

src/modules/pipelines/pipelines.module.ts

@@ -0,0 +1,19 @@
+import { Module } from '@nestjs/common';
+import { ClientsModule } from '@nestjs/microservices';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+
+import { PipelinesController } from './pipelines.controller';
+import { PipelinesService } from './pipelines.service';
+
+import { PipelinesClientConfiguration } from './pipelines-client';
+import { PipelinesClientService } from './client.service';
+
+const client = new PipelinesClientConfiguration();
+
+@Module({
+  imports: [ClientsModule.register([client.providerOptions])],
+  controllers: [PipelinesController],
+  providers: [PipelinesService, PipelinesClientService, DadosferaLogger],
+  exports: [PipelinesService],
+})
+export class PipelinesModule {}

src/modules/pipelines/pipelines.service.ts

@@ -1,8 +1,8 @@
 import { HttpException, HttpStatus, Injectable } from '@nestjs/common';
 import { DecodeGrpcStruct } from 'protospack';
 import { Info } from 'protospack/dist/lib/interfaces';
-import { PipelinesClientService } from 'src/clients/pipelines/client.service';
-import { IIdRequest } from 'src/clients/pipelines/interfaces';
+import { PipelinesClientService } from './client.service';
+import { IIdRequest } from './interfaces';
 import { objectCamelToSnake } from 'src/utils/CaseConverter';
 
 @Injectable()