maestro/src/clients/inputs/client.service.ts

import {
  OnModuleInit,
  Inject,
  HttpException,
  HttpStatus,
  NotFoundException,
  InternalServerErrorException,
} from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import {
  DecodeGrpcStruct,
  EncodeJsonToGrpcStruct,
  InputCreateS3Request,
  InputNewCreateRequest,
  InputService,
  TestConnectionGetColumnsRequest,
  TestConnectionRequest,
} from 'protospack';
import axios, { AxiosRequestConfig } from 'axios';
import { lastValueFrom } from 'rxjs';
import { InputModel } from 'src/modules/inputs/dtos/input.model';
import {
  objectCamelToSnake,
  objectSnakeToCamel,
} from 'src/utils/CaseConverter';
import { mustache } from 'src/utils/mustache';
import { DadosferaLogger } from 'dadosfera-logs';
import { OauthSecrets } from 'src/utils/OauthSecrets';
import { URLSearchParams } from 'url';
import { IIdRequest, UpdateInputRequest } from './interfaces';

export class InputsClientService implements OnModuleInit {
  private inputService: InputService;
  logger: any;
  constructor(
    @Inject(DadosferaLogger)
    dadosferaLogger: DadosferaLogger,
    @Inject('INPUTS_PACKAGE') private readonly grpcClient: ClientGrpc,
    @Inject('OAUTH_SECRETS') private readonly oauthSecrets: OauthSecrets,
  ) {
    this.logger = dadosferaLogger.logger;
  }

  async onModuleInit() {
    this.inputService =
      this.grpcClient.getService<InputService>('InputService');
  }

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

    const input = await lastValueFrom(
      this.inputService.NewCreate(objectSnakeToCamel(createInputDto)),
    );
    objectCamelToSnake(input);
    return input;
  }

  async createGeneric(createInputGeneric) {
    this.logger.info('InputClientService - Create');
    const { info, ...data } = createInputGeneric;
    // data.credentials = await this.getInputTokens(data);
    if (data.options) delete data.options;
    if (data.credentials.oauth_code) delete data.credentials.oauth_code;
    const grpcPayload = {
      input: EncodeJsonToGrpcStruct(data),
      info,
    };
    objectSnakeToCamel(grpcPayload);

    const structReturn = await lastValueFrom(
      this.inputService.Create(grpcPayload),
    ).catch((err: { details: string }) => {
      if (err.details === 'Item Not found!') {
        throw new NotFoundException('Input not found');
      }
      throw new InternalServerErrorException(err.details);
    });
    objectCamelToSnake(structReturn);
    const inputCreated = DecodeGrpcStruct(structReturn.input);
    return { input: inputCreated };
  }

  async createS3Inputs(createInputDto: InputCreateS3Request) {
    this.logger.info('InputClientService - Create');
    objectSnakeToCamel(createInputDto);
    const createInputResponse = await lastValueFrom(
      this.inputService.CreateS3(createInputDto),
    ).catch((e) => {
      throw new HttpException(e.details, 500);
    });
    this.logger.info('Done');
    objectCamelToSnake(createInputResponse);
    return createInputResponse;
  }

  async findOne(data: IIdRequest) {
    this.logger.info('InputClientService - FindOne');
    const findOneInputResponse = await lastValueFrom(
      this.inputService.FindOne(objectSnakeToCamel(data)),
    ).catch((e) => {
      this.logger.error(e.details);
      throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
    });
    objectCamelToSnake(findOneInputResponse);
    return findOneInputResponse;
  }

  async findAll(data) {
    this.logger.info('InputClientService - FindAll');
    const findAllInputResponse = await lastValueFrom(
      this.inputService.FindAll(objectSnakeToCamel(data)),
    ).catch((e) => {
      this.logger.error(e.details);
      throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
    });
    findAllInputResponse.inputs.forEach((input) => objectCamelToSnake(input));

    return findAllInputResponse;
  }

  async update(updateInputDTO: UpdateInputRequest) {
    this.logger.info('InputClientService - Update');
    const updateInputResponse = await new Promise((resolve, reject) => {
      this.inputService.Update(objectSnakeToCamel(updateInputDTO)).subscribe({
        next(x) {
          resolve(objectCamelToSnake(x));
        },
        error(err) {
          reject(err);
        },
        complete() {
          // console.log('done');
        },
      });
    })
      .then((res) => {
        this.logger.info('Done');
        return res;
      })
      .catch((err) => {
        this.logger.error(err.message);
        throw new Error(err);
      });

    return updateInputResponse;
  }

  async remove(idRequest: IIdRequest) {
    this.logger.info('InputClientService - Remove');

    const removeInputResponse = await new Promise((resolve, reject) => {
      this.inputService.Remove(objectSnakeToCamel(idRequest)).subscribe({
        next(x) {
          resolve(objectCamelToSnake(x));
        },
        error(err) {
          reject(err);
        },
        complete() {
          // console.log('done');
        },
      });
    })
      .then((res) => {
        this.logger.info('Done');
        return res;
      })
      .catch((err) => {
        this.logger.error(err.message);
        throw new Error(err);
      });

    return removeInputResponse;
  }

  async testConnection(data: TestConnectionRequest) {
    this.logger.info('InputClientService - TestConnection');
    const testConnectionResponse = await new Promise((resolve, reject) => {
      this.inputService.NewTestConnection(objectSnakeToCamel(data)).subscribe({
        next(x) {
          resolve(objectCamelToSnake(x));
        },
        error(err) {
          reject(err);
        },
        complete() {
          // console.log('done');
        },
      });
    })
      .then((res) => {
        this.logger.info('Done');
        return res;
      })
      .catch((err) => {
        this.logger.error(err.message);
        throw new Error(err);
      });

    return testConnectionResponse;
  }

  async getColumns(data: TestConnectionGetColumnsRequest) {
    this.logger.info('InputClientService - TestConnection/Get-Columns');
    const getColumnsResponse = await new Promise((resolve, reject) => {
      this.inputService.GetColumns(objectSnakeToCamel(data)).subscribe({
        next(x) {
          resolve(objectCamelToSnake(x));
        },
        error(err) {
          reject(err);
        },
        complete() {
          // console.log('done');
        },
      });
    })
      .then((res) => {
        this.logger.info('Done');
        return res;
      })
      .catch((err) => {
        this.logger.error(err.message);
        throw new Error(err);
      });

    return getColumnsResponse;
  }
  async getAuthSecrets(plugin) {
    switch (plugin) {
      case 'hubspot':
        return {
          client_id: this.oauthSecrets.hubspot.client_id || '',
          client_secret: this.oauthSecrets.hubspot.client_secret || '',
        };
    }
    return {};
  }
  async getInputTokens(input: InputModel) {
    const { credentials } = input;
    let credentialsTokens = {};
    switch (credentials.connection_type) {
      case 'oauth':
        credentialsTokens = await this.getOauthTokens(input);
        break;
    }
    return { ...credentials, ...credentialsTokens };
  }
  async getOauthTokens(input: InputModel) {
    const tokens: Record<string, any> = {};
    const { options, credentials } = input;
    const { oauth } = options;
    const { get_tokens_url_params, get_tokens_set_response } = oauth;
    const secrets = await this.getAuthSecrets(input.plugin);
    const get_tokens_url_params_string = mustache(get_tokens_url_params, {
      ...secrets,
      ...credentials,
    });
    let get_tokens_url_params_obj: Record<string, any> = {};
    try {
      get_tokens_url_params_obj = JSON.parse(get_tokens_url_params_string);
    } catch (error) {
      throw new HttpException('Erro transformando get_tokens_url_params', 400);
    }
    let params = {};
    switch (oauth.content_type) {
      case 'application/x-www-form-urlencoded':
        params = new URLSearchParams(get_tokens_url_params_obj);
        params = params.toString();
        break;
      default:
        break;
    }
    const axiosRequestConfig: AxiosRequestConfig = {
      url: oauth.get_tokens_url,
      method: oauth.get_tokens_method || 'POST',
      data: params,
      headers: { 'content-type': oauth.content_type || 'application/json' },
    };

    const { data } = await axios(axiosRequestConfig).catch((err) => {
      this.logger.error(err.response.data);
      throw new HttpException(err.response.data.message, err.response.status);
    });
    for (const key in get_tokens_set_response) {
      const responseKey = get_tokens_set_response[key];
      tokens[key] = data[responseKey];
    }
    const { redirect_uri } = get_tokens_url_params_obj;
    tokens.redirect_uri = redirect_uri;
    return tokens;
  }

  async getAvailableEntities(data) {
    this.logger.info('InputClientService - GetAvailableEntities');
    objectSnakeToCamel(data);
    const response = await lastValueFrom(
      this.inputService.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;
  }
}