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

import {
  BadRequestException,
  HttpException,
  HttpStatus,
  Inject,
  OnModuleInit,
} from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import {
  Messages,
  WriteService,
  ReadService,
  ProtoServices,
} from '@dadosfera/protospack-v2/dist/lib/PipelineV2';
import { lastValueFrom } from 'rxjs';

import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
import { PipelinesClientConfiguration } from './pipelines-client';
import { ICreatePipelineV2Req } from './interfaces';
import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
import { Metadata } from '@grpc/grpc-js';
import { ConnectorClientService } from '../connector/client.service';
import { InputsService } from '../inputs/inputs.service';
import { RequestUser } from 'src/decorators/user.decorator';
import { TransformationsService } from '../transformations/transformations.service';
import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath';
import ErrorCodes from 'src/utils/errorCodes';
import ErrorBuilder from 'src/utils/ErrorBuilder';

export class PipelinesService implements OnModuleInit {
  logger: DadosferaLogger;
  pipelineWriteService: WriteService.PipelineV2WriteService;
  pipelineReadService: ReadService.PipelineV2ReadService;
  constructor(
    @Inject(DadosferaLogger)
    dadosferaLogger: DadosferaLogger,
    @Inject(PipelinesClientConfiguration.name)
    private readonly grpcClient: ClientGrpc,
    private readonly connectorService: ConnectorClientService,
    private readonly inputsService: InputsService,
    private readonly transformationsService: TransformationsService,
  ) {
    this.logger = dadosferaLogger.logger;
  }

  onModuleInit() {
    this.pipelineReadService =
      this.grpcClient.getService<ReadService.PipelineV2ReadService>(
        ProtoServices.PipelineV2ReadService,
      );

    this.pipelineWriteService =
      this.grpcClient.getService<WriteService.PipelineV2WriteService>(
        ProtoServices.PipelineV2WriteService,
      );
  }

  adjustInputPayload(payload) {
    const { input } = payload;
    if (input) {
      if (input.input_generic) {
        const { credentials } = input.input_generic;
        return {
          ...input.input_generic,
          credentials: credentials && JSON.parse(credentials),
        };
      }
      if (input.input_s3) return input.input_s3;
      if (input.input_jdbc) return input.input_jdbc;

      const { credentials } = input;
      return {
        ...input,
        credentials: credentials && JSON.parse(credentials),
      };
    }
    return input;
  }

  async create(
    body: ICreatePipelineV2Req,
    metadata,
  ): Promise<Messages.PipelineV2CreateResponse> {
    this.logger.info('PipelinesV2ClientService - Create');

    const pipelineV2CreateRequest: PipelineV2CreateRequest = {
      ...body,
      input_id: body.input_id,
      transformations_ids: body.transformations_ids,
      tags: body.tags,
      properties: body.properties && JSON.stringify(body.properties),
    };

    const createPipelineResponse = await lastValueFrom(
      this.pipelineWriteService.PipelineV2Create(
        pipelineV2CreateRequest,
        metadata,
      ),
    ).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 ErrorBuilder(error.details);
    });
    this.logger.info('Done');

    return createPipelineResponse;
  }

  async findAll(
    data: Messages.PipelineV2FindAllRequest,
    metadata,
  ): Promise<Messages.PipelineV2FindAllResponse> {
    this.logger.info('PipelinesClientService - FindAll');
    const findAllPipelineResponse = await lastValueFrom(
      this.pipelineReadService.PipelineV2FindAll(data, metadata),
    )
      .then((res) => {
        this.logger.info('Done');
        return res;
      })
      .catch((err) => {
        this.logger.error(err.message);
        throw new Error(err);
      });

    return findAllPipelineResponse;
  }

  async findOne(
    data: Messages.PipelineV2FindOneRequest,
    metadata,
  ): Promise<Messages.PipelineV2FindOneResponse> {
    this.logger.info('PipelinesClientService - FindOne');

    const findOnePipelineResponse = await lastValueFrom(
      this.pipelineReadService.PipelineV2FindOne(data, metadata),
    );
    this.logger.info('Done');

    return findOnePipelineResponse;
  }

  async update(
    UpdatePipelineRequest: Messages.PipelineV2UpdateRequest,
    metadata,
  ) {
    this.logger.info('PipelinesClientService - Update');

    const updatePipelineResponse = await lastValueFrom(
      this.pipelineWriteService.PipelineV2Update(
        UpdatePipelineRequest,
        metadata,
      ),
    );
    this.logger.info('Done');

    return updatePipelineResponse;
  }

  async remove(data: { id: string; metadata: Metadata; user: RequestUser }) {
    const { id, metadata, user } = data;
    const info = {
      user_id: user.user_id,
      customer_id: user.customer_id,
      customer: user.customer_name,
    };
    let existsInProduct = true;
    const { pipeline } = await lastValueFrom(
      this.pipelineReadService.PipelineV2FindOne({ id }, metadata),
    ).catch((e) => {
      if (e.details === ErrorCodes.PIPELINE.NOT_FOUND) existsInProduct = false;
      else throw new ErrorBuilder(e.details);

      return { pipeline: undefined };
    });
    await lastValueFrom(
      this.pipelineWriteService.PipelineV2Remove({ id }, metadata),
    );

    if (existsInProduct) {
      //{pipeline:{tables: {tables: [], input_id: ''}}}
      const input = pipeline.config.tables
        ? JSON.parse(pipeline.config.tables)
        : null;
      if (input)
        await this.inputsService
          .remove({
            id: input.input_id,
            info,
          })
          .catch((error) =>
            this.logger.error('Could not delete input', {
              data: { input, error },
            }),
          );

      const transformations: { id: string }[] = pipeline.transformations
        ? JSON.parse(pipeline.transformations)
        : null;
      if (transformations && transformations.length)
        for (const transformation of transformations) {
          await this.transformationsService
            .remove({
              id: transformation.id,
              info,
            })
            .catch((error) =>
              this.logger.error('Could not delete transformation', {
                data: { transformation, error },
              }),
            );
        }
    }
  }

  async findOneProperties(id: string, metadata: Metadata) {
    this.logger.info('PipelinesClientService - findOneProperties');
    const { pipeline } = await this.findOne({ id }, metadata);

    const properties = JSON.parse(pipeline.properties);

    const connectorId = `${pipeline.connector_plugin}-${pipeline.connector_version}`;

    const res = await this.connectorService.getConnectorDetails(
      connectorId,
      metadata.get('language')[0].toString(),
    );
    const connector: { [key: string]: any; config_controls: any[] } =
      JSON.parse(res.connector);
    const config_controls = (connector.config_controls || []).filter(
      (config_control) => {
        const path: string = config_control.name;
        if (!objHasPath({ object: { properties }, path })) return false;

        let default_value = getObjValueFromPath({
          object: { properties },
          path,
        });

        if (default_value != undefined) {
          if (
            path === 'properties.spreadsheet_id' &&
            typeof default_value === 'string'
          )
            default_value = 'https://docs.google.com/spreadsheets/d/'.concat(
              default_value,
            );
          config_control.default_value = default_value;
        }

        return true;
      },
    );

    return {
      pipeline: {
        ...pipeline,
        config_controls,
        properties,
      },
    };
  }

  async findOneObjects(id: string, metadata: Metadata) {
    this.logger.info('PipelinesClientService - FindOneObjects');

    const { objects } = await lastValueFrom(
      this.pipelineReadService.PipelineV2FindObjects({ id }, metadata),
    );
    this.logger.info('Done');

    return { objects: JSON.parse(objects) };
  }

  async initUploadFile(uploadFile, metadata) {
    const body: Messages.PipelineV2UploadFileRequest = {
      name: uploadFile.name,
      parts: uploadFile.parts,
    };

    return lastValueFrom(
      this.pipelineWriteService.PipelineV2UploadFile(body, metadata),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async downloadLogs(pipeline_run_id: string, metadata: Metadata) {
    this.logger.info('PipelinesClientService - DownloadLogs');

    const { url } = await lastValueFrom(
      this.pipelineReadService.PipelineV2DownloadLogs(
        { pipeline_run_id },
        metadata,
      ),
    );
    this.logger.info('Done');

    return url;
  }

  async completeUploadFile(uploadFile, metadata) {
    const body: Messages.PipelineV2CompleteUploadFileRequest = {
      upload_id: uploadFile.upload_id,
      parts: JSON.stringify(uploadFile.parts),
      name: uploadFile.file_name,
    };

    return lastValueFrom(
      this.pipelineWriteService.PipelineV2CompleteUploadFile(body, metadata),
    ).catch((err) => {
      throw new HttpException(
        err.details,
        err.code === 6 ? HttpStatus.CONFLICT : 400,
      );
    });
  }

  async getMonitoringDashboardUrl(metadata: Metadata) {
    this.logger.info('PipelinesClientService - getMonitoringDashboardUrl');

    const res = await lastValueFrom(
      this.pipelineReadService.PipelineV2GetDashboardUrl(
        {
          dashboard_id: '95',
          exp: '15m',
          metabase_customer_name: 'dadosferatech',
        },
        metadata,
      ),
    );
    this.logger.info('Done');

    return res;
  }
}