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

import {
  ForbiddenException,
  HttpException,
  HttpStatus,
  Inject,
  Injectable,
} from '@nestjs/common';
import CronParser, { CronExpression } from 'cron-parser';
import { ClientGrpc } from '@nestjs/microservices';
import DadosferaLogger from '@dadosfera/dadosfera-logs/dist';
import { lastValueFrom } from 'rxjs';
import { objectCamelToSnake } from 'src/utils/CaseConverter';
import { IIdRequest, UpdateInputRequest } from './dtos/old_interfaces';
import { Input } from '@dadosfera/protospack-v2';
import {
  GetAvailableEntitiesRequest,
  InputCreateGenericRequest,
  InputCreateS3Request,
  InputNewCreateRequest,
  TestConnectionRequest,
} from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
import { CreateInputReq } from './dtos/input.model';

@Injectable()

//TODO refactor this class to remove OLD_inputClient
export class InputsService {
  logger: DadosferaLogger;
  inputWriteService: Input.WriteService.InputWriteService;
  inputReadService: Input.ReadService.InputReadService;

  constructor(
    @Inject(DadosferaLogger)
    dadosferaLogger: DadosferaLogger,
    @Inject('InputsGrpcClient') private readonly grpcClient: ClientGrpc,
  ) {
    this.logger = dadosferaLogger.logger;
  }

  async onModuleInit() {
    this.inputWriteService =
      this.grpcClient.getService<Input.WriteService.InputWriteService>(
        Input.ProtoServices.InputWriteService,
      );
    this.inputReadService =
      this.grpcClient.getService<Input.ReadService.InputReadService>(
        Input.ProtoServices.InputReadService,
      );
  }

  OLD_inputClient = {
    newCreate: async (createInputDto: InputNewCreateRequest) => {
      this.logger.info('InputClientService - Create');

      const input = await lastValueFrom(
        this.inputWriteService.InputNewCreate(createInputDto),
      );
      objectCamelToSnake(input);
      return input;
    },
    createS3Inputs: async (createInputDto: InputCreateS3Request) => {
      this.logger.info('InputClientService - Create');
      createInputDto;
      const createInputResponse = await lastValueFrom(
        this.inputWriteService.InputCreateS3(createInputDto),
      ).catch((e) => {
        throw new HttpException(e.details, 500);
      });
      this.logger.info('Done');
      objectCamelToSnake(createInputResponse);
      return createInputResponse;
    },
    update: async (updateInputDTO: UpdateInputRequest) => {
      this.logger.info('InputClientService - Update');
      const updateInputResponse = await lastValueFrom(
        this.inputWriteService.InputUpdate(updateInputDTO),
      );

      return updateInputResponse;
    },
    testConnection: async (data: TestConnectionRequest) => {
      this.logger.info('InputClientService - TestConnection');
      const testConnectionResponse = await lastValueFrom(
        this.inputReadService.NewTestConnection(data),
      );

      return testConnectionResponse;
    },
    getAvailableEntities: async (data) => {
      this.logger.info('InputClientService - GetAvailableEntities');
      data;
      const response = await lastValueFrom(
        this.inputReadService.GetAvailableEntities(data),
      ).catch((err) => {
        this.logger.info(err.details);
        if (err.details && err.details.includes('400'))
          throw new HttpException('Plugin inválido', HttpStatus.BAD_REQUEST);
        throw new HttpException(err.details, HttpStatus.INTERNAL_SERVER_ERROR);
      });
      objectCamelToSnake(response);

      return response;
    },
  };

  secondsInADay = 60 * 60 * 24;
  secondsInAnHour = 60 * 60;

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

      const { credentials } = payload;
      return {
        ...payload,
        credentials: credentials && JSON.parse(credentials),
      };
    }
    return payload;
  }
  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 ForbiddenException(
        'Intervalo de tempo não pode ser inferior a um dia.',
      );
    } else if (secondsApart < this.secondsInAnHour) {
      throw new ForbiddenException(
        'Intervalo de tempo não pode ser inferior a uma hora.',
      );
    }
  }
  async create(data: { body: CreateInputReq; info: Info }) {
    // this.validateCron(data);
    const { body, info } = data;

    const inputCreateGenericRequest: InputCreateGenericRequest = {
      input: {
        ...body,
      },
      info,
    };

    const { input } = await lastValueFrom(
      this.inputWriteService.InputCreate(inputCreateGenericRequest),
    );
    const adjustedInput = this.adjustInputPayload(input);
    return { input: adjustedInput };
  }

  async getAvailableEntities(data: GetAvailableEntitiesRequest) {
    return lastValueFrom(this.inputReadService.GetAvailableEntities(data));
  }

  async findAll(info: Info) {
    const { inputs } = await lastValueFrom(
      this.inputReadService.InputFindAll({ info }),
    );
    if (!inputs) return { inputs: [] };

    const inputsReturn = inputs.map((i) => this.adjustInputPayload(i));
    return { inputs: inputsReturn };
  }

  async findOne(idRequest: IIdRequest) {
    const findOneInputResponse: any = await lastValueFrom(
      this.inputReadService.InputFindOne(idRequest),
    );
    findOneInputResponse.input = this.adjustInputPayload(
      findOneInputResponse.input,
    );
    return findOneInputResponse;
  }

  async update(id: string, data, info: Info) {
    // this.validateCron({ ...data, info });
    try {
      const updateInputResponse: any = await this.OLD_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) {
    return lastValueFrom(this.inputWriteService.InputRemove(idRequest));
  }

  async testConnection(data) {
    try {
      const testConnectionInputResponse =
        await this.OLD_inputClient.testConnection(data);

      return testConnectionInputResponse;
    } 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;
  }
}