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

import { Body, HttpException, HttpStatus, Injectable } from '@nestjs/common';
import { Timeout } from '@nestjs/schedule';
import { InputsClientService } from 'src/clients/inputs/client.service';
import { IIdRequest, Info } from 'src/clients/inputs/interfaces';

@Injectable()
export class InputsService {
  constructor(private inputClient: InputsClientService) {}

  adjustInputPayload(payload) {
    return payload?.input_s3 || payload?.input_jdbc;
  }
  async create(@Body() data) {
    try {
      if (
        data.plugin == 'csv' ||
        data.plugin == 'json' ||
        data.plugin == 'parquet'
      ) {
        const { info, ...input } = data;
        const inputPayload = this.generateInputS3Payload(input);
        const createInputResponse = await this.inputClient.createS3Inputs({
          input: inputPayload,
          info,
        });
        return createInputResponse;
      } else {
        const createInputResponse: any = await this.inputClient.create(data);
        if (createInputResponse?.input?.input_jdbc) {
          return { input: createInputResponse?.input?.input_jdbc };
        }
        return createInputResponse;
      }
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

  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) {
    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 } = 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,
      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;
  }
}