maestro/src/modules/inputs/inputs.service.ts

import { Body, HttpException, HttpStatus, Injectable } from '@nestjs/common';
import { Timeout } from '@nestjs/schedule';
import { DecodeGrpcStruct } from 'protospack';
import CronParser, { CronExpression } from 'cron-parser';
import { InputsClientService } from 'src/clients/inputs/client.service';
import { IIdRequest, Info } from 'src/clients/inputs/interfaces';

@Injectable()
export class InputsService {
  constructor(private inputClient: InputsClientService) { }
  secondsInADay = 60 * 60 * 24;
  secondsInAnHour = 60 * 60;

  adjustInputPayload(payload) {
    if (payload?.input_generic) return DecodeGrpcStruct(payload.input_generic);
    if (!payload.input_s3 && !payload.input_jdbc) return payload;
    return payload?.input_s3 || payload?.input_jdbc;
  }
  getDifferenceInSeconds(date1: Date, date2: Date) {
    const diffInMs = Math.abs(date2.getTime() - date1.getTime());
    return diffInMs / 1000;
  }
  validateCron(data) {
    const { info, cron } = data;
    const { customer_tier } = info;
    if (!cron) return;
    let interval: CronExpression;
    try {
      interval = CronParser.parseExpression(cron);
    } catch (error) {
      throw new HttpException(
        'Intervalo de tempo inválido',
        HttpStatus.BAD_REQUEST,
      );
    }
    const nextDate = interval.next().toDate();
    const afterNextDate = interval.next().toDate();
    const secondsApart = this.getDifferenceInSeconds(nextDate, afterNextDate);
    if (customer_tier === 'BASIC' && secondsApart < this.secondsInADay) {
      throw new HttpException(
        'Intervalo de tempo não pode ser inferior a um dia.',
        HttpStatus.FORBIDDEN,
      );
    } else if (secondsApart < this.secondsInAnHour) {
      throw new HttpException(
        'Intervalo de tempo não pode ser inferior a uma hora.',
        HttpStatus.FORBIDDEN,
      );
    }
  }
  async create(@Body() data) {
    this.validateCron(data);
    let response;
    switch (data.plugin.toLowerCase()) {
      case 'csv':
      case 'json':
      case 'parquet':
        const { info, ...input } = data;
        const inputPayload = this.generateInputS3Payload(input);
        response = await this.inputClient.createS3Inputs({
          input: inputPayload,
          info,
        });
        break;
      case 'oracle':
      case 'mysql':
      case 'postgresql':
      case 'sqlserver':
        response = await this.inputClient.newCreate(data);
        break;
      default:
        response = await this.inputClient.createGeneric(data);
        break;
    }
    const adjustedInput = this.adjustInputPayload(response.input);
    return { ...response, input: adjustedInput };
  }
  async reCreate(data) {
    const { info, cron } = data;
    if (cron) this.validateCron({ info, cron });
    const response = await this.inputClient.createGeneric(data);
    const adjustedInput = this.adjustInputPayload(response.input);
    return { ...response, input: adjustedInput };
  }
  async getAvailableEntities(data): Promise<{ entities: string[] }> {
    return await this.inputClient.getAvailableEntities(data);
  }
  async findAll(body) {
    try {
      const findAllInputResponse: any = await this.inputClient.findAll(body);
      if (findAllInputResponse?.inputs?.length) {
        findAllInputResponse.inputs = findAllInputResponse.inputs.map((input) =>
          this.adjustInputPayload(input),
        );
      }
      return findAllInputResponse;
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  async findOne(idRequest: IIdRequest) {
    try {
      const findOneInputResponse: any = await this.inputClient.findOne(
        idRequest,
      );
      findOneInputResponse.input = this.adjustInputPayload(
        findOneInputResponse.input,
      );
      return findOneInputResponse;
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  async update(id: string, data, info: Info) {
    this.validateCron({ ...data, info });
    try {
      const updateInputResponse: any = await this.inputClient.update({
        id,
        info,
        ...data,
      });

      updateInputResponse.input = this.adjustInputPayload(
        updateInputResponse?.input,
      );
      return updateInputResponse;
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  async remove(idRequest: IIdRequest) {
    try {
      const removeInputResponse = await this.inputClient.remove(idRequest);

      return removeInputResponse;
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  @Timeout(60000 * 10) // Timeout set for 10 minutes
  async testConnection(data) {
    try {
      const testConnectionInputResponse = await this.inputClient.testConnection(
        data,
      );

      return testConnectionInputResponse;
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  async getColumns(data) {
    try {
      const testConnectionGetColumnsResponse =
        await this.inputClient.getColumns(data);

      return testConnectionGetColumnsResponse;
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  generateInputS3Payload(payload) {
    const { credentials, plugin, cron } = payload;
    if (!credentials) return payload;

    const {
      client_aws_access_key_id,
      client_aws_secret_access_key,
      file_format_params,
      format_file_params,
      client_bucket,
      file_to_extract,
    } = credentials;

    const formatedPayload = {
      plugin,
      cron,
      source_bucket: client_bucket,
      source_prefix: file_to_extract,
      auth_parameters: {
        aws_access_key_id: client_aws_access_key_id,
        aws_secret_access_key: client_aws_secret_access_key,
      },
      file_format_params: file_format_params || format_file_params,
    };
    return formatedPayload;
  }
}