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

import { HttpException, Inject, OnModuleInit } from '@nestjs/common';
import { ClientGrpc, Payload } from '@nestjs/microservices';
import {
  PipelineServicesNames,
  PipelinesServiceInterface,
} from '@victorradael/protospack';
import {
  objectCamelToSnake,
  objectSnakeToCamel,
} from 'src/utils/CaseConverter';
import {
  ICreatePipelineDto,
  IGetPipelineLogsRequest,
  IIdRequest,
  IUpdatePipelineRequest,
} from './interfaces';

export class PipelinesClientService implements OnModuleInit {
  private pipelineService: PipelinesServiceInterface;
  constructor(
    @Inject('PIPELINES_PACKAGE') private readonly grpcClient: ClientGrpc,
  ) {}

  onModuleInit() {
    this.pipelineService =
      this.grpcClient.getService<PipelinesServiceInterface>(
        PipelineServicesNames.PipelineService,
      );
  }

  async create(@Payload() createPipelineDto: ICreatePipelineDto) {
    console.log('PipelinesClientService', 'Create');

    const createPipelineResponse = await new Promise((resolve, reject) => {
      this.pipelineService
        .Create(objectSnakeToCamel(createPipelineDto))
        .subscribe({
          next(x) {
            resolve(x);
          },
          error(err) {
            console.log('Observable Error');
            reject(err);
          },
          complete() {
            console.log('done');
          },
        });
    })
      .then((res) => res)
      .catch((err) => {
        if (err.details == 'Invalid Request') {
          throw new HttpException(err.details, 400);
        }
        throw new HttpException(err.details, 500);
      });

    return createPipelineResponse;
  }

  async findAll(data: IIdRequest) {
    console.log('PipelinesClientService', 'FindAll');

    const findAllPipelineResponse = await new Promise((resolve, reject) => {
      this.pipelineService.FindAll(objectSnakeToCamel(data)).subscribe({
        next(x) {
          resolve(x.pipelines);
        },
        error(err) {
          console.log('Observable Error');
          reject(err);
        },
        complete() {
          console.log('done');
        },
      });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return findAllPipelineResponse;
  }

  async findOne(data: IIdRequest) {
    console.log('PipelinesClientService', 'FindOne');

    const findOnePipelineResponse = await new Promise((resolve, reject) => {
      this.pipelineService.FindOne(objectSnakeToCamel(data)).subscribe({
        next(x) {
          resolve(x);
        },
        error(err) {
          console.log('Observable Error');
          reject(err);
        },
        complete() {
          console.log('done');
        },
      });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return findOnePipelineResponse;
  }

  async update(UpdatePipelineRequest: IUpdatePipelineRequest) {
    console.log('PipelinesClientService', 'Update');

    const updatePipelineResponse = await new Promise((resolve, reject) => {
      this.pipelineService
        .Update(objectSnakeToCamel(UpdatePipelineRequest))
        .subscribe({
          next(x) {
            resolve(x);
          },
          error(err) {
            console.log('Observable Error');
            reject(err);
          },
          complete() {
            console.log('done');
          },
        });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return updatePipelineResponse;
  }

  async remove(data: IIdRequest) {
    console.log('PipelinesClientService', 'Remove');

    const removePipelineResponse = await new Promise((resolve, reject) => {
      this.pipelineService.remove(objectSnakeToCamel(data)).subscribe({
        next(x) {
          resolve(objectCamelToSnake(x));
        },
        error(err) {
          console.log('Observable Error');
          reject(err);
        },
        complete() {
          console.log('done');
        },
      });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return removePipelineResponse;
  }

  async getPipelineLogsMessages(data: IGetPipelineLogsRequest) {
    console.log('PipelinesClientService', 'GetPipelineLogsMessages');

    const logsPipelineResponse = await new Promise((resolve, reject) => {
      this.pipelineService.getPipelineLogs(data).subscribe({
        next(x) {
          resolve(x);
        },
        error(err) {
          console.log('Observable Error');
          reject(err);
        },
        complete() {
          console.log('done');
        },
      });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return logsPipelineResponse;
  }

  async getPipelineStatus(data) {
    console.log('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) {
            console.log('Observable Error');
            reject(err);
          },
          complete() {
            console.log('done');
          },
        });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return statusPipelineResponse;
  }

  async runPipeline({ id, info }: IIdRequest) {
    console.log('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) {
            console.log('Observable Error');
            reject(err);
          },
          complete() {
            console.log('done');
          },
        });
    })
      .then((res) => res)
      .catch((err) => {
        throw new Error(err);
      });

    return statusPipelineResponse;
  }
}