diff --git a/.github/workflows/deploy-manually.yml b/.github/workflows/deploy-manually.yml index b204e48..dea0b1d 100644 --- a/.github/workflows/deploy-manually.yml +++ b/.github/workflows/deploy-manually.yml @@ -134,7 +134,7 @@ jobs: eb deploy - name: Remove Docker's Trash - continue-on-error: true + if: always() run: | - docker system prune - docker rmi -f $(docker images -aq) + docker system prune --volumes -a -f + docker system df diff --git a/.nvmrc b/.nvmrc index 25bf17f..1667c1a 100644 --- a/.nvmrc +++ b/.nvmrc @@ -1 +1 @@ -18 \ No newline at end of file +18.3.0 \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index f80c060..6f073d4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,7 +1,8 @@ -FROM node:14.15.4-alpine3.12 as packages +FROM node:18.3.0-alpine3.15 as packages WORKDIR /packages ARG NPM_TOKEN COPY package.json . +COPY package-lock.json . COPY .npmrc . RUN npm install RUN rm -f ./.npmrc diff --git a/package-lock.json b/package-lock.json index 0ab3ab4..52d832f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -19,7 +19,7 @@ "@nestjs/platform-express": "^8.4.3", "@nestjs/schedule": "^1.1.0", "@nestjs/swagger": "^5.2.1", - "@victorradael/protospack": "2.4.2-1", + "@victorradael/protospack": "2.5.0", "axios": "^0.25.0", "cron-parser": "^4.4.0", "dotenv": "^14.2.0", @@ -56,6 +56,9 @@ "ts-node": "^10.7.0", "tsconfig-paths": "^3.14.1", "typescript": "^4.6.3" + }, + "engines": { + "node": "18.3.0" } }, "node_modules/@ampproject/remapping": { @@ -2357,9 +2360,9 @@ } }, "node_modules/@victorradael/protospack": { - "version": "2.4.2-1", - "resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.4.2-1.tgz", - "integrity": "sha512-wdcGaK0DIEJeOHga1br11kq264JNdhdB1G2toy/u6TvAk90wwBKHNs/cZko0p9WFVuqHQOVehoHj7WJbJ3K25Q==", + "version": "2.5.0", + "resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.5.0.tgz", + "integrity": "sha512-DORlm3uHAYVH9dlcmwyOxjA4rk1UZWFyYUpD02k0bFfrH+uzSWe8qiPtTgZym9P8U9D8bXYRWeLlroasBj2Rdw==", "dependencies": { "rxjs": "^7.5.5" } @@ -10932,9 +10935,9 @@ } }, "@victorradael/protospack": { - "version": "2.4.2-1", - "resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.4.2-1.tgz", - "integrity": "sha512-wdcGaK0DIEJeOHga1br11kq264JNdhdB1G2toy/u6TvAk90wwBKHNs/cZko0p9WFVuqHQOVehoHj7WJbJ3K25Q==", + "version": "2.5.0", + "resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.5.0.tgz", + "integrity": "sha512-DORlm3uHAYVH9dlcmwyOxjA4rk1UZWFyYUpD02k0bFfrH+uzSWe8qiPtTgZym9P8U9D8bXYRWeLlroasBj2Rdw==", "requires": { "rxjs": "^7.5.5" } diff --git a/package.json b/package.json index 5889917..b31a45c 100644 --- a/package.json +++ b/package.json @@ -5,6 +5,9 @@ "author": "", "private": true, "license": "UNLICENSED", + "engines": { + "node": "18.3.0" + }, "scripts": { "prebuild": "rimraf dist", "build": "nest build", @@ -31,7 +34,7 @@ "@nestjs/platform-express": "^8.4.3", "@nestjs/schedule": "^1.1.0", "@nestjs/swagger": "^5.2.1", - "@victorradael/protospack": "2.4.2-1", + "@victorradael/protospack": "2.5.0", "axios": "^0.25.0", "cron-parser": "^4.4.0", "dotenv": "^14.2.0", diff --git a/src/clients/inputs/client.service.ts b/src/clients/inputs/client.service.ts index dee162e..968bb4b 100644 --- a/src/clients/inputs/client.service.ts +++ b/src/clients/inputs/client.service.ts @@ -1,17 +1,27 @@ -import { OnModuleInit, Inject } from '@nestjs/common'; +import { + OnModuleInit, + Inject, + HttpException, + HttpStatus, +} from '@nestjs/common'; import { ClientGrpc } from '@nestjs/microservices'; import { - InputCreateResponse, + DecodeGrpcStruct, + EncodeJsonToGrpcStruct, InputCreateS3Request, InputNewCreateRequest, InputService, TestConnectionGetColumnsRequest, TestConnectionRequest, } from '@victorradael/protospack'; +import axios 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 { IIdRequest, UpdateInputRequest } from './interfaces'; export class InputsClientService implements OnModuleInit { @@ -25,102 +35,68 @@ export class InputsClientService implements OnModuleInit { this.grpcClient.getService('InputService'); } - async create(createInputDto: InputNewCreateRequest) { + async newCreate(createInputDto: InputNewCreateRequest) { console.log('InputClientService', 'Create'); - const createInputResponse = await new Promise((resolve, reject) => { - this.inputService - .NewCreate(objectSnakeToCamel(createInputDto)) - .subscribe({ - next(x) { - resolve(objectCamelToSnake(x)); - }, - error(err) { - console.log('Observable Error'); - reject(err); - }, - complete() { - console.log('done'); - }, - }); - }).catch((err) => { - throw new Error(err); - }); + const input = await lastValueFrom( + this.inputService.NewCreate(objectSnakeToCamel(createInputDto)), + ); + objectCamelToSnake(input); + return input; + } - return createInputResponse; + async createGeneric(createInputGeneric) { + console.log('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), + ); + objectCamelToSnake(structReturn); + const inputCreated = DecodeGrpcStruct(structReturn.input); + return { input: inputCreated }; } async createS3Inputs(createInputDto: InputCreateS3Request) { console.log('InputClientService', 'Create'); - - const createInputResponse = await new Promise((resolve, reject) => { - this.inputService.CreateS3(objectSnakeToCamel(createInputDto)).subscribe({ - next(x) { - resolve(objectCamelToSnake(x)); - }, - error(err) { - console.log('Observable Error'); - reject(err); - }, - complete() { - console.log('done'); - }, - }); - }) - .then((res) => res) - .catch((err) => { - throw new Error(err); - }); - + objectSnakeToCamel(createInputDto); + const createInputResponse = await lastValueFrom( + this.inputService.CreateS3(createInputDto), + ).catch((e) => { + throw new HttpException(e.details, 500); + }); + console.log('done'); + objectCamelToSnake(createInputResponse); return createInputResponse; } async findOne(data: IIdRequest) { console.log('InputClientService', 'FindOne'); - - const findOneInputResponse = await new Promise((resolve, reject) => { - this.inputService.FindOne(objectSnakeToCamel(data)).subscribe({ - next(x) { - resolve(objectCamelToSnake(x)); - }, - error(err) { - console.log('Observable Error'); - reject(err); - }, - complete() { - console.log('done'); - }, - }); - }) - .then((res) => res) - .catch((err) => { - throw new Error(err); - }); - + const findOneInputResponse = await lastValueFrom( + this.inputService.FindOne(objectSnakeToCamel(data)), + ).catch((e) => { + throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR); + }); + objectCamelToSnake(findOneInputResponse); return findOneInputResponse; } async findAll(data) { console.log('InputClientService', 'FindAll'); - - const findAllInputResponse = await new Promise((resolve, reject) => { - this.inputService.FindAll(objectSnakeToCamel(data)).subscribe({ - next(x) { - resolve(objectCamelToSnake(x)); - }, - error(err) { - console.log('Observable Error'); - reject(err); - }, - complete() { - console.log('done'); - }, - }); - }) - .then((res) => res) - .catch((err) => { - throw new Error(err); - }); + const findAllInputResponse = await lastValueFrom( + this.inputService.FindAll(objectSnakeToCamel(data)), + ).catch((e) => { + throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR); + }); + findAllInputResponse.inputs.forEach((input) => objectCamelToSnake(input)); return findAllInputResponse; } @@ -221,4 +197,82 @@ export class InputsClientService implements OnModuleInit { return getColumnsResponse; } + async getAuthSecrets(plugin) { + switch (plugin) { + case 'hubspot': + return { + client_id: process.env.HUBSPOT_CLIENT_ID || '', + client_secret: process.env.HUBSPOT_CLIENT_SECRET || '', + }; + } + return {}; + } + async getInputTokens(input: InputModel) { + const { credentials, options } = input; + const credentialsTokens: any = {}; + switch (credentials.connection_type) { + case 'oauth': + const { oauth } = options; + const { get_tokens_url_params, get_tokens_set_response } = oauth; + const secrets = await this.getAuthSecrets(input.plugin); + if (oauth.content_type === 'application/x-www-form-urlencoded') { + const requestParams = new URLSearchParams(); + const get_tokens_url_params_string = mustache(get_tokens_url_params, { + ...secrets, + ...credentials, + }); + let get_tokens_url_params_obj: Record = {}; + 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, + ); + } + const { redirect_uri } = get_tokens_url_params_obj; + for (const key in get_tokens_url_params_obj) { + const value = get_tokens_url_params_obj[key]; + requestParams.append(key, value); + } + const { data } = await axios + .post(options.oauth.get_tokens_url, requestParams) + .catch((err) => { + console.log(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]; + credentialsTokens[key] = data[responseKey]; + } + credentialsTokens.redirect_uri = redirect_uri; + } + break; + } + return { ...credentials, ...credentialsTokens }; + } + // getOauthTokens(input){ + + // } + + async getAvailableEntities(data) { + console.log('InputClientService', 'GetAvailableEntities'); + objectSnakeToCamel(data); + const response = await lastValueFrom( + this.inputService.GetAvailableEntities(data), + ).catch((err) => { + console.log(err); + 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; + } } diff --git a/src/modules/inputs/dtos/input.model.ts b/src/modules/inputs/dtos/input.model.ts new file mode 100644 index 0000000..73f0d19 --- /dev/null +++ b/src/modules/inputs/dtos/input.model.ts @@ -0,0 +1,202 @@ +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; + +export class InputModel { + category: string; + plugin: string; + name: string; + cron?: string; + credentials?: { + connection_type: string; + oauth: string; + oauth_code: string; + start_date: string; + }; + options?: { + oauth?: { + get_tokens_url: string; + get_tokens_url_params: string; + get_tokens_set_response: Record; + content_type: string; + }; + }; +} + +class GoogleAnalyticsClientSecrets { + @ApiProperty() + type: string; + @ApiProperty() + project_id: string; + @ApiProperty() + private_key_id: string; + @ApiProperty() + private_key: string; + @ApiProperty() + client_email: string; + @ApiProperty() + client_id: string; + @ApiProperty() + auth_uri: string; + @ApiProperty() + token_uri: string; + @ApiProperty() + auth_provider_x509_cert_url: string; + @ApiProperty() + client_x509_cert_url: string; +} +class OauthObject { + @ApiPropertyOptional() + get_tokens_url: string; + @ApiPropertyOptional() + get_tokens_url_params: string; + @ApiPropertyOptional() + get_tokens_set_response: Record; + @ApiPropertyOptional() + content_type: string; +} +class InputOptions { + @ApiPropertyOptional() + oauth: OauthObject; + @ApiPropertyOptional() + skip_select_columns: true; + @ApiPropertyOptional() + skip_select_entities: false; + @ApiPropertyOptional() + skip_transformation: true; +} + +class CredentialsJdbc { + @ApiPropertyOptional() + jdbc_user: string; + @ApiPropertyOptional() + jdbc_password: string; + @ApiPropertyOptional() + database: string; + @ApiPropertyOptional() + endpoint: string; + @ApiPropertyOptional() + port: string; + @ApiPropertyOptional() + engine: string; + @ApiPropertyOptional() + schema: string; +} +class FileFormatParams { + @ApiPropertyOptional() + file_format: string; + @ApiPropertyOptional() + encoding: string; +} + +class Credentials extends CredentialsJdbc { + @ApiProperty() + connection_type: string; + + @ApiPropertyOptional() + client_aws_access_key_id: string; + @ApiPropertyOptional() + client_aws_secret_access_key: string; + @ApiPropertyOptional() + client_bucket: string; + @ApiPropertyOptional() + file_to_extract: string; + @ApiPropertyOptional() + file_format_params: FileFormatParams; + + @ApiPropertyOptional() + view_id: string; + @ApiPropertyOptional() + client_secrets: GoogleAnalyticsClientSecrets; + @ApiPropertyOptional() + start_date: string; + @ApiPropertyOptional() + end_date: string; + + @ApiPropertyOptional() + oauth_code: string; +} +class Column { + @ApiProperty() + name: string; + @ApiProperty() + type: string; +} +class TableColumns { + @ApiProperty() + name: string; + @ApiProperty() + columns: Column[]; + @ApiProperty() + references: Column[]; +} +export class GetAvailableEntitiesReq { + @ApiProperty() + plugin: string; +} + +export class GetAvailableEntitiesRes { + @ApiProperty() + entities: string[]; +} + +export class TestConnectionGetColumnsReq { + @ApiProperty() + plugin: string; + @ApiProperty() + tables: string[]; + @ApiPropertyOptional() + credentials: CredentialsJdbc; + @ApiPropertyOptional() + id: string; +} +export class TestConnectionGetColumnsRes { + @ApiProperty() + tables: TableColumns[]; +} + +export class TestConnectionReq { + @ApiProperty() + plugin: string; + @ApiProperty() + credentials: CredentialsJdbc; +} +export class TestConnectionRes { + @ApiProperty() + connection_state: boolean; + @ApiProperty() + total_entities: number; + @ApiProperty() + database_tables: string[]; +} +export class CreateInputReq { + @ApiProperty() + plugin: string; + @ApiProperty() + category: string; + @ApiProperty() + credentials: Credentials; + @ApiProperty() + name: string; + @ApiPropertyOptional() + options: InputOptions; +} + +export class Input { + @ApiProperty() + id: string; + @ApiProperty() + category: string; + @ApiProperty() + plugin: string; + @ApiProperty() + name: string; + @ApiProperty() + cron: string; + @ApiProperty() + credentials: Credentials; + @ApiProperty() + client_id: string; + @ApiProperty() + created_at: string; + @ApiProperty() + updated_at: string; +} diff --git a/src/modules/inputs/inputs.controller.ts b/src/modules/inputs/inputs.controller.ts index 918aa1a..6ddf90a 100644 --- a/src/modules/inputs/inputs.controller.ts +++ b/src/modules/inputs/inputs.controller.ts @@ -10,14 +10,40 @@ import { import { InputsService } from './inputs.service'; import { InputsClientService } from 'src/clients/inputs/client.service'; import { UpdateInputRequest } from 'src/clients/inputs/interfaces'; -import { InputNewCreateRequest } from '@victorradael/protospack'; +import { ApiOkResponse, ApiTags } from '@nestjs/swagger'; +import { + CreateInputReq, + GetAvailableEntitiesReq, + GetAvailableEntitiesRes, + Input, + TestConnectionGetColumnsReq, + TestConnectionGetColumnsRes, + TestConnectionReq, + TestConnectionRes, +} from './dtos/input.model'; +@ApiTags('inputs') @Controller('inputs') export class InputsController { - constructor(private inputsClientService: InputsClientService) {} + inputService: InputsService; + constructor(private inputsClientService: InputsClientService) { + this.inputService = new InputsService(this.inputsClientService); + } + + @Get('available-entities/:plugin') + @ApiOkResponse({ type: GetAvailableEntitiesRes }) + async getAvailableEntities( + @Body() body, + @Param() params: GetAvailableEntitiesReq, + ) { + const { info } = body; + const { plugin } = params; + return await this.inputService.getAvailableEntities({ info, plugin }); + } @Post('/test-connection') - async testConnection(@Body() data) { + @ApiOkResponse({ type: TestConnectionRes }) + async testConnection(@Body() data: TestConnectionReq) { console.log(`/test-connection`, 'ON TEST CONNECTION ROUTE'); const inputService = new InputsService(this.inputsClientService); @@ -27,7 +53,8 @@ export class InputsController { } @Post('/test-connection/get-columns') - async getColumns(@Body() data) { + @ApiOkResponse({ type: TestConnectionGetColumnsRes }) + async getColumns(@Body() data: TestConnectionGetColumnsReq) { console.log( `/test-connection/get-columns`, 'ON TEST CONNECTION GET COLUMNS ROUTE', @@ -40,7 +67,8 @@ export class InputsController { } @Post() - async create(@Body() createInputDto: InputNewCreateRequest) { + @ApiOkResponse({ type: Input }) + async create(@Body() createInputDto: CreateInputReq) { console.log(`/input`, 'ON CREATE ROUTE'); const inputService = new InputsService(this.inputsClientService); diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 03898bd..2d41455 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -1,5 +1,6 @@ import { Body, HttpException, HttpStatus, Injectable } from '@nestjs/common'; import { Timeout } from '@nestjs/schedule'; +import { DecodeGrpcStruct } from '@victorradael/protospack'; import CronParser, { CronExpression } from 'cron-parser'; import { InputsClientService } from 'src/clients/inputs/client.service'; import { IIdRequest, Info } from 'src/clients/inputs/interfaces'; @@ -11,6 +12,8 @@ export class InputsService { 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) { @@ -47,31 +50,34 @@ export class InputsService { } async create(@Body() data) { this.validateCron(data); - try { - if ( - data.plugin == 'csv' || - data.plugin == 'json' || - data.plugin == 'parquet' - ) { + let response; + switch (data.plugin.toLowerCase()) { + case 'csv': + case 'json': + case 'parquet': const { info, ...input } = data; const inputPayload = this.generateInputS3Payload(input); - const createInputResponse = await this.inputClient.createS3Inputs({ + response = 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); + 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 getAvailableEntities(data): Promise<{ entities: string[] }> { + return await this.inputClient.getAvailableEntities(data); } - async findAll(body) { try { const findAllInputResponse: any = await this.inputClient.findAll(body); @@ -153,7 +159,7 @@ export class InputsService { } generateInputS3Payload(payload) { - const { credentials, plugin } = payload; + const { credentials, plugin, cron } = payload; if (!credentials) return payload; const { @@ -164,8 +170,10 @@ export class InputsService { client_bucket, file_to_extract, } = credentials; + const formatedPayload = { plugin, + cron, source_bucket: client_bucket, source_prefix: file_to_extract, auth_parameters: { diff --git a/src/modules/pipelines/pipelines.service.ts b/src/modules/pipelines/pipelines.service.ts index 4755a00..f0987f0 100644 --- a/src/modules/pipelines/pipelines.service.ts +++ b/src/modules/pipelines/pipelines.service.ts @@ -1,4 +1,5 @@ import { HttpException, HttpStatus, Injectable } from '@nestjs/common'; +import { DecodeGrpcStruct } from '@victorradael/protospack'; import { PipelinesClientService } from 'src/clients/pipelines/client.service'; import { IIdRequest, Info } from 'src/clients/pipelines/interfaces'; import { objectCamelToSnake } from 'src/utils/CaseConverter'; @@ -8,7 +9,11 @@ export class PipelinesService { constructor(private pipelineClient: PipelinesClientService) {} adjustPayload(payload) { - payload.input = payload.input?.input_s3 || payload.input?.input_jdbc; + if (payload.input?.input_generic) { + payload.input = DecodeGrpcStruct(payload.input?.input_generic); + } else { + payload.input = payload.input?.input_s3 || payload.input?.input_jdbc; + } } async create(createPipelineDto) { diff --git a/src/utils/mustache.ts b/src/utils/mustache.ts new file mode 100644 index 0000000..0f6064a --- /dev/null +++ b/src/utils/mustache.ts @@ -0,0 +1,7 @@ +export const mustache = (string: string, data: Record = {}) => { + return Object.entries(data).reduce((res, [key, value]) => { + const mainRe = new RegExp(`(?