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

/* eslint-disable no-async-promise-executor */
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, UpdatePlatformInputRequest, UpdateTableDTO } 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';
import { PlatformApiService } from '../platform-api/platform-api.service';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
import { TableUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
import { AxiosError } from 'axios';
import { NimbusService } from 'src/services/nimbus/nimbus.service';

type RollbackPromise = () => Promise<any>;

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,
    private readonly platformAPI: PlatformApiService,
    private readonly nimbusService: NimbusService
  ) {
    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),
    );
    console.log('pipeline find one response', findOnePipelineResponse);
    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;
  }

  async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) {
    this.logger.info('InputClientService - Update');

    const {
      input: oldInput
    } = await this.inputsService.findOne({
      id: inputId,
      info: info
    });

    this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
    const pipelineIdFormat = pipelineId.split('-').join('_');
    const rollback: RollbackPromise[] = [];

    const updateInputResponse = await this.inputsService.update(
      inputId,
      updateInputDTO,
      info
    );

    const inputRollback = () => {
      this.logger.info("exec rollback to input: " + JSON.stringify(oldInput));
      return this.inputsService.rollbackUpdate(
        {
          id: inputId,
          dataAssetUpdate: updateInputResponse.dataAssetUpdate,
          tables: oldInput.tables,
          info
        }
      ) as Promise<any>;
    }

    rollback.push(inputRollback);

    this.logger.info("Input Update Response: " + JSON.stringify(updateInputResponse))

    const nimbusUpdates = updateInputResponse?.tablesUpdate || [];

    nimbusUpdates.forEach(update => {
      const nimbusRollback = () => {
        return this.nimbusService.renameTable(
          info.customer,
          update.database,
          {
            table_name: update.table_name,
            table_schema: update.table_schema
          },
          {
            table_name: update.old_table_name,
            table_schema: update.old_table_schema
          }
        );
      }
      rollback.push(nimbusRollback);
    });

    try {
      await this.updateNimbus(info.customer, nimbusUpdates);
    } catch (error) {
      this.logger.error(error);
      if (error instanceof AxiosError) {
        this.logger.error(JSON.stringify(error.response.data));
      }
      await this.executeRenameRollback(rollback);

      throw new Error("Error Nimbus updating tables");
    }

    try {
      await this.updatePlatformJobs(
        pipelineIdFormat,
        updateInputDTO,
        user
      );
    } catch (error) {
      this.logger.error(error);
      await this.executeRenameRollback(rollback)
      throw new Error("Error Platform API updating jobs");
    }

    return updateInputResponse;
  }

  private async executeRenameRollback(request: RollbackPromise[]) {
    this.logger.info('rollback steps: ' + request.length)
    const result = await Promise.allSettled(request.map(func => func()));
    result.forEach(promise => {
      this.logger.info("Promise finish with status: " + promise.status)

      if (promise.status === "rejected") {
        this.logger.error("reject with: " + JSON.stringify(promise.reason || {}))
      }

      if (promise.status === "fulfilled") {
        this.logger.info("success with: " + JSON.stringify(promise.value || {}))
      }
    });

  }

  private async updateNimbus(customer: string, changes: TableUpdate[]) {
    // throw new Error("teste error nimbus");
    this.logger.info('Nimbus Changes: ' + JSON.stringify(changes));
    if(!changes || changes.length === 0) return;

    const requests = changes.map(change => {
      return this.nimbusService.renameTable(customer, change.database, {
        table_name: change.old_table_name,
        table_schema: change.old_table_schema
      }, {
        table_name: change.table_name,
        table_schema: change.table_schema
      });
    })

    const values = await Promise.allSettled(requests);

    const success = values.map(request => request.status === "fulfilled")

    this.logger.info("Updates with succes: " + success.length);

    values.forEach(promise => {
      this.logger.info("Promise finish with status: " + promise.status)

      if (promise.status === "rejected") {
        this.logger.error("reject with: " + JSON.stringify(promise.reason || {}));
        throw new Error(promise.reason );
      }

      if (promise.status === "fulfilled") {
        this.logger.info("success with: " + JSON.stringify(promise.value || {}));
      }
    });

  }

  async updatePlatformJobs(pipelineId: string, updateInputDTO: UpdatePlatformInputRequest, user: RequestUser) {
    const jobsUpdated = [];

    for (const [index, table] of updateInputDTO.tables.entries()) {
      const jobUpdate = {
        job_id: `${pipelineId}_${index}`,
      }

      if (table.type !== "incremental_with_qualify") {
        delete table.destinations?.qualify;
      }

      if (table.memory) {
        jobUpdate["memory"] = {
          amount: table.memory * 1000
        }
      }
      
      this.logger.info('Updating input reference for table: ' + table.name);
      let hasUpdateSyncMode = false;

      const jobSyncMode = {}

      if (table.columns) {
        hasUpdateSyncMode = true;
        jobSyncMode['column_include_list'] = table.columns;
      }

      if (table.reference_column) {
        hasUpdateSyncMode = true;
        jobSyncMode['incremental_column_name'] = table.reference_column.name;
        jobSyncMode['incremental_column_type'] = table.reference_column.type;
      }

      if (table.identifier_columns) {
        hasUpdateSyncMode = true;
        jobSyncMode['primary_keys'] = table.identifier_columns;
      }

      if (table.type) {
        hasUpdateSyncMode = true;
        jobSyncMode['target_load_type'] = table.type;

      }

      if(hasUpdateSyncMode) {
        jobUpdate["sync_mode"] = jobSyncMode;
      }

      if (Object.keys(table.destinations).length > 1) {
        let hasChanges = false
        const jobRenameTables = {
            raw: {},
            qualify: {}
        }

        if (Object.keys(table.destinations.raw).length > 1) {
          hasChanges = true;
          jobRenameTables.raw = table.destinations.raw;
        }

        if (Object.keys(table.destinations.qualify).length > 1) {
          hasChanges = true;
          jobRenameTables.qualify = table.destinations.qualify;
        }

        if (hasChanges) {
          jobUpdate['rename_tables'] = jobRenameTables;
        }
      }

      jobsUpdated.push(jobUpdate);
    }

    this.logger.info('Request body:' + JSON.stringify({
      jobs_updated: jobsUpdated
    }));

    const response = await this.platformAPI.proxy(
      'PUT',
        `/pipeline/${pipelineId}/jobs`,
      user,
      {
        job_updates: jobsUpdated
      }
    )
    this.logger.info('Platform api response: ' + JSON.stringify(response));
  }

}