From 21c57e5620646e6f5cb75f9f0f0c9ce3d90f7e5c Mon Sep 17 00:00:00 2001 From: marcos-silva-rodrigues Date: Tue, 24 Mar 2026 12:28:20 -0300 Subject: [PATCH] FEAT: update nimbus after input --- docsfera.json | 18 +----- package-lock.json | 15 ++--- package.json | 2 +- src/modules/inputs/inputs.service.ts | 21 ++++--- .../pipelinesV2/pipelines.controller.ts | 3 +- src/modules/pipelinesV2/pipelines.module.ts | 7 ++- src/modules/pipelinesV2/pipelines.service.ts | 49 ++++++++++++---- .../platform-api/platform-api.service.ts | 6 +- src/services/nimbus/nimbus.module.ts | 9 +++ src/services/nimbus/nimbus.service.ts | 56 +++++++++++++++++++ 10 files changed, 140 insertions(+), 46 deletions(-) create mode 100644 src/services/nimbus/nimbus.module.ts create mode 100644 src/services/nimbus/nimbus.service.ts diff --git a/docsfera.json b/docsfera.json index bcdd005..4e2ed5f 100644 --- a/docsfera.json +++ b/docsfera.json @@ -3001,14 +3001,7 @@ "parameters": [], "responses": { "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } + "description": "" } }, "tags": [ @@ -3583,14 +3576,7 @@ ], "responses": { "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } + "description": "" } }, "tags": [ diff --git a/package-lock.json b/package-lock.json index b9f5209..eba19a9 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,7 +17,7 @@ "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", "@dadosfera/protospack": "2.5.3", - "@dadosfera/protospack-v2": "^3.38.0-beta.30", + "@dadosfera/protospack-v2": "3.38.0-beta.31", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", @@ -1744,9 +1744,10 @@ } }, "node_modules/@dadosfera/protospack-v2": { - "version": "3.38.0-beta.30", - "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.30.tgz", - "integrity": "sha512-EePtEV4Bjr47BCtQgeZP8WzappDv1mRWMro15ms+Vdev/FIVBydMew/BtK3spzI7KHWJniB0Ctsvpcdj2GiTDA==", + "version": "3.38.0-beta.31", + "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.31.tgz", + "integrity": "sha512-IxnVEaVvQNumN/PQsnF+9CsFvmn6/0YyfXauoPQ1bccvqU8Lo+hKZv4qTBrR+urd05OWyg+dJBv/QtW45b04XQ==", + "license": "ISC", "dependencies": { "@grpc/grpc-js": "^1.9.3", "rxjs": "^7.5.5" @@ -5533,9 +5534,9 @@ } }, "node_modules/axios": { - "version": "0.30.2", - "resolved": "https://registry.npmjs.org/axios/-/axios-0.30.2.tgz", - "integrity": "sha512-0pE4RQ4UQi1jKY6p7u6i1Tkzqmu+d+/tHS7Q7rKunWLB9WyilBTpHHpXzPNMDj5hTbK0B0PTLSz07yqMBiF6xg==", + "version": "0.30.3", + "resolved": "https://registry.npmjs.org/axios/-/axios-0.30.3.tgz", + "integrity": "sha512-5/tmEb6TmE/ax3mdXBc/Mi6YdPGxQsv+0p5YlciXWt3PHIn0VamqCXhRMtScnwY3lbgSXLneOuXAKUhgmSRpwg==", "license": "MIT", "dependencies": { "follow-redirects": "^1.15.4", diff --git a/package.json b/package.json index 4b109cd..7da4765 100644 --- a/package.json +++ b/package.json @@ -35,7 +35,7 @@ "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", "@dadosfera/protospack": "2.5.3", - "@dadosfera/protospack-v2": "^3.38.0-beta.30", + "@dadosfera/protospack-v2": "3.38.0-beta.31", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index cb9d25d..44c7ddd 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -17,6 +17,7 @@ import { InputCreateGenericRequest, InputCreateS3Request, InputNewCreateRequest, + InputUpdateResponse, TestConnectionRequest, } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; @@ -71,8 +72,8 @@ export class InputsService { objectCamelToSnake(createInputResponse); return createInputResponse; }, - update: async (updateInputDTO: UpdateInputRequest) => { - this.logger.info('InputClientService - Update'); + update: async (updateInputDTO: UpdateInputRequest): Promise => { + this.logger.info('InputClientService - Update' + JSON.stringify(updateInputDTO)); const updateInputResponse = await lastValueFrom( this.inputWriteService.InputUpdate(updateInputDTO), ); @@ -207,16 +208,22 @@ export class InputsService { async update(id: string, data, info: Info) { // this.validateCron({ ...data, info }); try { - const updateInputResponse: any = await this.OLD_inputClient.update({ + const { + nimbusUpdates, + input + } = await this.OLD_inputClient.update({ id, - info, ...data, + info, }); - updateInputResponse.input = this.adjustInputPayload( - updateInputResponse?.input, + const updateInputResponse = this.adjustInputPayload( + input, ); - return updateInputResponse; + return { + input: updateInputResponse, + nimbusUpdates + }; } catch (err) { throw new HttpException(err.message, HttpStatus.NOT_FOUND); } diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index 11e7b10..1c00bcd 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -294,13 +294,14 @@ export class PipelinesController { ) { this.logger.info('PipelinesController - update', { user }); - const { customer_id, customer_name, user_id, username } = user; const info: Info = { user_id: user.user_id, customer: user.customer_name, customer_id: user.customer_id, + pipeline_id: pipelineId }; + const metadata = PackTheMetadata({ customer_id, customer_name, diff --git a/src/modules/pipelinesV2/pipelines.module.ts b/src/modules/pipelinesV2/pipelines.module.ts index 9550aeb..d0007b8 100644 --- a/src/modules/pipelinesV2/pipelines.module.ts +++ b/src/modules/pipelinesV2/pipelines.module.ts @@ -12,6 +12,8 @@ import { ConnectorModule } from '../connector/connector.module'; import { InputsModule } from '../inputs/inputs.module'; import { TransformationsModule } from '../transformations/transformations.module'; import { PlatformApiModule } from '../platform-api/platform-api.module'; +import { NimbusServicesModule } from 'src/services/nimbus/nimbus.module'; +import { NimbusService } from 'src/services/nimbus/nimbus.service'; const client = new PipelinesClientConfiguration(); @@ -22,10 +24,11 @@ const client = new PipelinesClientConfiguration(); ConnectorModule, InputsModule, TransformationsModule, - PlatformApiModule + PlatformApiModule, + NimbusServicesModule ], controllers: [PipelinesController], - providers: [PipelinesService, DadosferaLogger], + providers: [PipelinesService, DadosferaLogger, NimbusService], exports: [PipelinesService], }) export class PipelinesV2Module {} diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index c0a1ad6..db04131 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -28,7 +28,10 @@ import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath'; import ErrorCodes from 'src/utils/errorCodes'; import ErrorBuilder from 'src/utils/ErrorBuilder'; import { PlatformApiService } from '../platform-api/platform-api.service'; -import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; +import { Info, NewTable } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; +import { NimbusUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; +import axios from 'axios'; +import { NimbusService } from 'src/services/nimbus/nimbus.service'; export class PipelinesService implements OnModuleInit { logger: DadosferaLogger; @@ -42,7 +45,8 @@ export class PipelinesService implements OnModuleInit { private readonly connectorService: ConnectorClientService, private readonly inputsService: InputsService, private readonly transformationsService: TransformationsService, - private readonly platformAPI: PlatformApiService + private readonly platformAPI: PlatformApiService, + private readonly nimbusService: NimbusService ) { this.logger = dadosferaLogger.logger; } @@ -350,18 +354,46 @@ export class PipelinesService implements OnModuleInit { this.logger.info('Update Dynamo Reference'); const pipelineIdFormat = pipelineId.split('-').join('_'); + console.log("updatePipelineInput", info, updateInputDTO) const updateInputResponse = await this.inputsService.update( inputId, updateInputDTO, info ) - this.logger.info('Dynamo Response', updateInputResponse); + this.logger.info(JSON.stringify(updateInputResponse)) + await this.updateNimbus(info.customer, updateInputResponse.nimbusUpdates); + + await this.updatePlatformJobs( + pipelineIdFormat, + updateInputDTO, + user + ); + + return updateInputResponse; + } + + private async updateNimbus(customer: string, changes: NimbusUpdate[]) { + this.logger.info('Nimbus Changes: ' + JSON.stringify(changes)); + if(!changes || changes.length === 0) return; + + const requests = changes.map(change => { + return this.nimbusService.renameTable(customer, change.database, change.old, change.new); + }) + + const values = await Promise.allSettled(requests); + + const success = values.map(request => request.status === "fulfilled") + + this.logger.info("Updates with succes: " + success.length); + } + + async updatePlatformJobs(pipelineId: string, updateInputDTO: UpdatePlatformInputRequest, user: RequestUser) { const jobsUpdated = []; for (const [index, table] of updateInputDTO.tables.entries()) { const jobUpdate = { - job_id: `${pipelineIdFormat}_${index}`, + job_id: `${pipelineId}_${index}`, } if (table.memory) { @@ -370,7 +402,7 @@ export class PipelinesService implements OnModuleInit { } } - this.logger.info('Updating input reference for table', table.name); + this.logger.info('Updating input reference for table: ' + table.name); let hasUpdateSyncMode = false; const jobSyncMode = {} @@ -431,17 +463,14 @@ export class PipelinesService implements OnModuleInit { })); const response = await this.platformAPI.proxy( - 'POST', - `/pipelines/${pipelineIdFormat}/batch-update`, + 'PUT', + `/pipeline/${pipelineId}/jobs`, user, { job_updates: jobsUpdated } ) this.logger.info('Platform api response: ' + JSON.stringify(response)); - - return updateInputResponse; - } } diff --git a/src/modules/platform-api/platform-api.service.ts b/src/modules/platform-api/platform-api.service.ts index f204678..7fe9928 100644 --- a/src/modules/platform-api/platform-api.service.ts +++ b/src/modules/platform-api/platform-api.service.ts @@ -89,12 +89,12 @@ export class PlatformApiService { // Propagate non-2xx responses as HttpExceptions if (response.status >= 400) { - this.logger.error('Platform API upstream error', { + this.logger.error('Platform API upstream error' + JSON.stringify({ status: response.status, data: response.data, path, method: method.toUpperCase(), - }); + })); throw new HttpException(response.data, response.status); } @@ -107,6 +107,8 @@ export class PlatformApiService { method: method.toUpperCase(), }); + this.logger.error(error) + if (error instanceof HttpException) { throw error; } diff --git a/src/services/nimbus/nimbus.module.ts b/src/services/nimbus/nimbus.module.ts new file mode 100644 index 0000000..de7b60f --- /dev/null +++ b/src/services/nimbus/nimbus.module.ts @@ -0,0 +1,9 @@ +import { Module } from "@nestjs/common"; +import { NimbusService } from "./nimbus.service"; +import DadosferaLogger from "@dadosfera/dadosfera-logs"; + +@Module({ + providers: [NimbusService, DadosferaLogger], + exports: [NimbusService], +}) +export class NimbusServicesModule {} diff --git a/src/services/nimbus/nimbus.service.ts b/src/services/nimbus/nimbus.service.ts new file mode 100644 index 0000000..031aae7 --- /dev/null +++ b/src/services/nimbus/nimbus.service.ts @@ -0,0 +1,56 @@ +import DadosferaLogger from "@dadosfera/dadosfera-logs"; +import { Inject, Injectable } from "@nestjs/common"; +import axios from "axios"; + +type TableUpdate = { + table_schema: string; + table_name: string; +} + +@Injectable() +export class NimbusService { + private logger: DadosferaLogger; + + constructor( + @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, + ) { + this.logger = dadosferaLogger.logger; + } + + private buildUrl(customerName: string) { + if (process.env.ENV === 'prd') { + return `https://nimbus-${customerName}.dadosfera.ai`; + } + + return `https://nimbus-${customerName}.${process.env.ENV.replace( + 'local', + 'stg', + )}.dadosfera.ai`; + } + + async renameTable(customerName: string, database: string, old: TableUpdate, update: TableUpdate) { + const nimbusUrl = this.buildUrl(customerName); + + const path = `/api/catalog/rename-tables/?database_name=${encodeURIComponent(database)}&table_name=${encodeURIComponent(old.table_name)}&table_schema=${encodeURIComponent(old.table_schema)}`; + + try { + this.logger.info("Request for PATCH" + nimbusUrl + path); + this.logger.info("Payload: " + JSON.stringify(update)); + const { data } = await axios.patch(nimbusUrl + path, { + table_name: update.table_name, + table_schema: update.table_schema + }) + + return data; + } catch (error) { + this.logger.error(error); + return { + message: error.message, + database, + old, + update + } + } + + } +}