maestro/src/modules/pipelines/client.service.ts

import {
  BadRequestException,
  ConflictException,
  Inject,
  InternalServerErrorException,
  OnModuleInit,
} from '@nestjs/common';
import { ClientGrpc, Payload } from '@nestjs/microservices';
import { PipelineServicesNames, PipelinesServiceInterface } from 'protospack';
import { lastValueFrom } from 'rxjs';
import { 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)),
    ).catch((error: { details: string }) => {
      if (error.details.includes('INVALID_REQUEST')) {
        const [, message] = error.details.split('|');
        throw new BadRequestException(message);
      }
      throw new InternalServerErrorException(error.details);
    });
    this.logger.info('Done');

    return createPipelineResponse;
  }

  async findAll(data) {
    this.logger.info('PipelinesClientService - FindAll');
    const findAllPipelineResponse = await lastValueFrom(
      this.pipelineService.FindAll(data),
    ).catch((err) => {
      this.logger.error(err.message);
      throw new Error(err);
    });
    this.logger.info('Done');

    return findAllPipelineResponse;
  }

  async findOne(data: IIdRequest) {
    this.logger.info('PipelinesClientService - FindOne');

    const findOnePipelineResponse = await lastValueFrom(
      this.pipelineService.FindOne(data),
    ).catch((err) => {
      this.logger.error(err.message);
      throw new Error(err);
    });
    this.logger.info('Done');

    return findOnePipelineResponse;
  }

  async update(UpdatePipelineRequest: IUpdatePipelineRequest) {
    this.logger.info('PipelinesClientService - Update');

    const updatePipelineResponse = await lastValueFrom(
      this.pipelineService.Update(UpdatePipelineRequest),
    ).catch((err) => {
      this.logger.error(err.message);
      throw new Error(err);
    });
    this.logger.info('Done');

    return updatePipelineResponse;
  }

  async remove(data: IIdRequest) {
    this.logger.info('PipelinesClientService - Remove');

    const removePipelineResponse = await lastValueFrom(
      this.pipelineService.remove(data),
    ).catch((err) => {
      this.logger.error(err.message);
      throw new Error(err);
    });
    this.logger.info('Done');

    return removePipelineResponse;
  }

  async getPipelineLogsMessages(data: IGetPipelineLogsRequest) {
    this.logger.info('PipelinesClientService - GetPipelineLogsMessages');

    const logsPipelineResponse = await lastValueFrom(
      this.pipelineService.getPipelineLogs(data),
    ).catch((err) => {
      this.logger.error(err.message);
      throw new Error(err);
    });
    this.logger.info('Done');

    return logsPipelineResponse;
  }

  async getPipelineStatus(data) {
    this.logger.info('PipelinesClientService - GetPipelineStatus');

    const statusPipelineResponse = await lastValueFrom(
      this.pipelineService.getPipelineStatus(data),
    )
      .then((res) => {
        const statusArray =
          res.status?.sort((a, b) => {
            if (a.id < b.id) {
              return 1;
            } else {
              return -1;
            }
          }) || [];
        return { status: statusArray };
      })
      .catch((err) => {
        this.logger.error(err.message);
        throw new Error(err);
      });
    this.logger.info('Done');

    return statusPipelineResponse;
  }

  async runPipeline({ id, info }: IIdRequest) {
    this.logger.info('PipelinesClientService - RunPipeline');
    const statusPipelineResponse = await lastValueFrom(
      this.pipelineService.triggerPipeline({ id, info }),
    ).catch((err) => {
      this.logger.error(err.message);
      throw new Error(err);
    });

    if (statusPipelineResponse.status == false) {
      throw new ConflictException(
        'This pipeline is not ready yet to execute, Try again later!',
      );
    }

    this.logger.info('Done');
    return statusPipelineResponse;
  }
}