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

import { ConflictException, Inject, OnModuleInit } from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { PipelineServicesNames, PipelinesServiceInterface } from 'protospack';
import { lastValueFrom } from 'rxjs';

import { IIdRequest } 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 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;
  }
}