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

import {
  BadRequestException,
  Inject,
  InternalServerErrorException,
  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/authentication/user.decorator';
import { TransformationsService } from '../transformations/transformations.service';

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 InternalServerErrorException(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) {
    this.logger.info('PipelinesClientService - Update');

    const updatePipelineResponse = await lastValueFrom(
      this.pipelineWriteService.PipelineV2Update(UpdatePipelineRequest),
    );
    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,
    };
    const { pipeline } = await lastValueFrom(
      this.pipelineReadService.PipelineV2FindOne({ id }, metadata),
    );
    await lastValueFrom(
      this.pipelineWriteService.PipelineV2Remove({ id }, metadata),
    );

    //{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 pipelineProperties = 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 = JSON.parse(res.connector);
    const config_controls = connector.config_controls || [];

    for (const key in pipelineProperties) {
      const i = config_controls
        .map((cc) => cc.name)
        .indexOf(`properties.${key}`);
      if (i > -1) {
        config_controls[i] = {
          ...config_controls[i],
          default_value: pipelineProperties[key],
        };
      }
    }

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