diff --git a/README.md b/README.md index 028ca81..4432ce9 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,7 @@ # Maestro + Maestro é a API principal da Dadosfera. É responsável pela comunicação do Frontend com nossos microsserviços. ```mermaid diff --git a/src/main.ts b/src/main.ts index 1085fb3..85e460a 100644 --- a/src/main.ts +++ b/src/main.ts @@ -111,3 +111,4 @@ function configureSwagger(app: INestApplication) { ); } bootstrap(); + diff --git a/src/modules/auth/auth.controller.ts b/src/modules/auth/auth.controller.ts index 385d60c..40eaad9 100644 --- a/src/modules/auth/auth.controller.ts +++ b/src/modules/auth/auth.controller.ts @@ -478,6 +478,7 @@ export class AuthController { @Get('me') async getMe(@Req() req: Request, @Res() res: Response) { this.logger.info('GET /auth/me ') + this.logger.info(JSON.stringify(req.headers)); // Check for API key header first const apiKey = req.get('X-Api-key'); @@ -488,6 +489,7 @@ export class AuthController { const userDto = { id: api_key.user_id, name: api_key.username, + email: api_key.username, customer: { id: api_key.customer_id, name: api_key.customer_name, diff --git a/src/modules/auth/auth.service.ts b/src/modules/auth/auth.service.ts index 6d67b95..8539d0e 100644 --- a/src/modules/auth/auth.service.ts +++ b/src/modules/auth/auth.service.ts @@ -437,7 +437,8 @@ export class AuthClientService implements OnModuleInit { const userDto: UserDTO = { id: user.id, - name: user.username, + name: user.name, + email: user.email, jobTitle: user?.jobTitle || null, department: user?.department || null, hierarchy: user?.hierarchy || null, diff --git a/src/modules/auth/dtos/login.ts b/src/modules/auth/dtos/login.ts index 3a484c8..2e22363 100644 --- a/src/modules/auth/dtos/login.ts +++ b/src/modules/auth/dtos/login.ts @@ -144,6 +144,7 @@ export interface BulkEditResponse { export type UserDTO = { id: string, name: string, + email: string, jobTitle?: string, department?: string, hierarchy?: string, diff --git a/src/modules/catalog/catalog.service.ts b/src/modules/catalog/catalog.service.ts index 89dcf84..ebb888c 100644 --- a/src/modules/catalog/catalog.service.ts +++ b/src/modules/catalog/catalog.service.ts @@ -734,6 +734,64 @@ class CatalogService implements OnModuleInit { } } + async renameTableOnNimbus( + nimbusUrl: string, + nimbusId: number, + changes: { table_name?: string; table_schema?: string; display_name?: string }, + ): Promise { + const endpoint = `${nimbusUrl}/api/catalog/table-metadata/${nimbusId}`; + this.logger.info(`Renaming table-metadata ${nimbusId} on Nimbus`, { endpoint, changes }); + await axios.patch(endpoint, changes); + } + + async renameColumnMetadataOnNimbus( + nimbusUrl: string, + databaseName: string, + oldTableName: string, + oldTableSchema: string, + newTableName: string, + newTableSchema: string, + ): Promise { + const listEndpoint = `${nimbusUrl}/api/catalog/column-metadata/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`; + this.logger.info(`Fetching column-metadata records to rename`, { listEndpoint }); + const { data: columns } = await axios.get(listEndpoint); + + const filtered = Array.isArray(columns) ? columns : []; + + for (const column of filtered) { + const patchEndpoint = `${nimbusUrl}/api/catalog/column-metadata/${column.id}`; + await axios.patch(patchEndpoint, { + table_name: newTableName, + table_schema: newTableSchema, + }); + } + this.logger.info(`Renamed ${filtered.length} column-metadata records on Nimbus`); + } + + async renameDataPreviewOnNimbus( + nimbusUrl: string, + databaseName: string, + oldTableName: string, + oldTableSchema: string, + newTableName: string, + newTableSchema: string, + ): Promise { + const listEndpoint = `${nimbusUrl}/api/catalog/data-preview/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`; + this.logger.info(`Fetching data-preview records to rename`, { listEndpoint }); + const { data: previews } = await axios.get(listEndpoint); + + const filtered = Array.isArray(previews) ? previews : []; + + for (const preview of filtered) { + const patchEndpoint = `${nimbusUrl}/api/catalog/data-preview/${preview.id}`; + await axios.patch(patchEndpoint, { + table_name: newTableName, + table_schema: newTableSchema, + }); + } + this.logger.info(`Renamed ${filtered.length} data-preview records on Nimbus`); + } + async catalogDatasetItem(table_metadata_id: number, metadata: Metadata) { const customer_name_raw = metadata.get('customer_name'); diff --git a/src/modules/inputs/inputs.controller.ts b/src/modules/inputs/inputs.controller.ts index 82ddd36..a8cc143 100644 --- a/src/modules/inputs/inputs.controller.ts +++ b/src/modules/inputs/inputs.controller.ts @@ -99,6 +99,7 @@ export class InputsController { customer: info.customer, }); + this.logger.info(JSON.stringify(body)) const response = await this.inputService.create({ body, info }); return response; diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index d258215..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), ); @@ -205,18 +206,24 @@ export class InputsService { } async update(id: string, data, info: Info) { - this.validateCron({ ...data, 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/interfaces.ts b/src/modules/pipelinesV2/interfaces.ts index db9a201..252b94c 100644 --- a/src/modules/pipelinesV2/interfaces.ts +++ b/src/modules/pipelinesV2/interfaces.ts @@ -1,6 +1,15 @@ import { ApiProperty, ApiPropertyOptional, OmitType } from '@nestjs/swagger'; import { Info } from '@dadosfera/protospack/dist/lib/interfaces'; +export class PipelineInputsDTO { + @ApiProperty() + tables: Array<{ + name: string, + type: string, + + }> +} + export class IPipelineV2 { @ApiProperty() id: string; @@ -123,3 +132,30 @@ export class PipelineFindAllReq { @ApiPropertyOptional() type?: string | undefined; } + +export interface UpdateTableDTO { + name: string; + type: string; + columns: string[]; + destinations: { + raw: { + table_schema: string; + table_name: string; + }; + qualify: { + table_schema: string; + table_name: string; + }; + }; + identifier_columns: string[]; + reference_column: { + name: string; + type: string; + }; + memory: number; +} + +export interface UpdatePlatformInputRequest { + cron: string; + tables: Array; +} diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index 4f5ff1d..1c00bcd 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -43,11 +43,15 @@ import { IPipelineV2, IInitUploadCSVFile, PipelineFindAllReq, + UpdatePlatformInputRequest, } from './interfaces'; import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter'; import { LanguageEnum } from 'src/utils/languages.enum'; import { Language } from 'src/decorators/language.decorator'; import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator'; +import { TableColumns } from '../inputs/dtos/input.model'; +import { UpdateInputRequest } from '../inputs/dtos/old_interfaces'; +import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; @ApiTags('PipelinesV2') @ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }]) @@ -223,6 +227,7 @@ export class PipelinesController { .then((res) => { //{pipeline:{tables: {tables: [], input_id: ''}}} let tables = JSON.parse(res.pipeline.config.tables); + const input_id = tables?.input_id; if (tables?.tables) tables = tables.tables; Object.assign(res.pipeline, { transformations: res.pipeline.transformations @@ -231,6 +236,7 @@ export class PipelinesController { config: { cron: res.pipeline.config.cron, tables, + input_id }, properties: res.pipeline.properties ? JSON.parse(res.pipeline.properties) @@ -277,6 +283,46 @@ export class PipelinesController { return response; } + @Patch('/:pipelineId/inputs/:id') + @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + async updatePipelineInput( + @Language() language: LanguageEnum, + @Body() pipelineInputDTO: UpdatePlatformInputRequest, + @Param('id') inputId: string, + @Param('pipelineId') pipelineId: string, + @User() user: RequestUser, + ) { + 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, + user_id, + username, + language, + }); + + const response = await this.pipelinesClientService.updatePipelineInput( + pipelineId, + inputId, + pipelineInputDTO, + info, + user, + metadata, + ); + + this.logger.info('PipelinesController - update: OK', { user }); + return response; + } + @ApiInternalOnlyEndpoint() @Put('/:id') @ApiOperation({ diff --git a/src/modules/pipelinesV2/pipelines.module.ts b/src/modules/pipelinesV2/pipelines.module.ts index 7bc7cc2..d0007b8 100644 --- a/src/modules/pipelinesV2/pipelines.module.ts +++ b/src/modules/pipelinesV2/pipelines.module.ts @@ -11,6 +11,9 @@ import { PipelinesModule as OldPipelineModule } from 'src/modules/pipelines/pipe 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(); @@ -21,9 +24,11 @@ const client = new PipelinesClientConfiguration(); ConnectorModule, InputsModule, TransformationsModule, + 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 f10b8ee..db04131 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -1,3 +1,4 @@ +/* eslint-disable no-async-promise-executor */ import { BadRequestException, HttpException, @@ -16,7 +17,7 @@ import { lastValueFrom } from 'rxjs'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { PipelinesClientConfiguration } from './pipelines-client'; -import { ICreatePipelineV2Req } from './interfaces'; +import { ICreatePipelineV2Req, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces'; import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages'; import { Metadata } from '@grpc/grpc-js'; import { ConnectorClientService } from '../connector/client.service'; @@ -26,6 +27,11 @@ import { TransformationsService } from '../transformations/transformations.servi 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, 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; @@ -39,6 +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 nimbusService: NimbusService ) { this.logger = dadosferaLogger.logger; } @@ -138,6 +146,7 @@ export class PipelinesService implements OnModuleInit { const findOnePipelineResponse = await lastValueFrom( this.pipelineReadService.PipelineV2FindOne(data, metadata), ); + console.log('pipeline find one response', findOnePipelineResponse); this.logger.info('Done'); return findOnePipelineResponse; @@ -339,4 +348,129 @@ export class PipelinesService implements OnModuleInit { return res; } + + async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) { + this.logger.info('InputClientService - Update'); + + 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(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: `${pipelineId}_${index}`, + } + + if (table.memory) { + jobUpdate["memory"] = { + amount: table.memory * 1000 + } + } + + this.logger.info('Updating input reference for table: ' + table.name); + let hasUpdateSyncMode = false; + + const jobSyncMode = {} + + if (table.columns) { + hasUpdateSyncMode = true; + jobSyncMode['column_include_list'] = table.columns; + } + + if (table.reference_column) { + hasUpdateSyncMode = true; + jobSyncMode['incremental_column_name'] = table.reference_column.name; + jobSyncMode['incremental_column_type'] = table.reference_column.type; + } + + if (table.identifier_columns) { + hasUpdateSyncMode = true; + jobSyncMode['primary_keys'] = table.identifier_columns; + } + + if (table.type) { + hasUpdateSyncMode = true; + jobSyncMode['target_load_type'] = table.type; + + } + + if(hasUpdateSyncMode) { + jobUpdate["sync_mode"] = jobSyncMode; + } + + if (Object.keys(table.destinations).length > 1) { + let hasChanges = false + const jobRenameTables = { + raw: {}, + qualify: {} + } + + if (Object.keys(table.destinations.raw).length > 1) { + hasChanges = true; + jobRenameTables.raw = table.destinations.raw; + } + + if (Object.keys(table.destinations.qualify).length > 1) { + hasChanges = true; + jobRenameTables.qualify = table.destinations.qualify; + } + + if (hasChanges) { + jobUpdate['rename_tables'] = jobRenameTables; + } + } + + jobsUpdated.push(jobUpdate); + } + + this.logger.info('Request body:' + JSON.stringify({ + jobs_updated: jobsUpdated + })); + + const response = await this.platformAPI.proxy( + 'PUT', + `/pipeline/${pipelineId}/jobs`, + user, + { + job_updates: jobsUpdated + } + ) + this.logger.info('Platform api response: ' + JSON.stringify(response)); + } + } diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 7c1c025..8c42102 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -10,6 +10,7 @@ import { Query, Inject, BadRequestException, + HttpException, } from '@nestjs/common'; import { ApiTags, ApiOperation } from '@nestjs/swagger'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; @@ -25,6 +26,9 @@ import { ElasticsearchService } from '../../services/elasticsearch'; import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb'; import { CustomersService } from '../customers/customers.service'; import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation'; +import { CatalogService } from '../catalog/catalog.service'; +import { PackTheMetadata } from '../../utils/PackTheMetadata'; +import { ValidationTableDTO } from './platform-api.dto'; type ValidateTablesDTO = { @@ -34,6 +38,11 @@ type ValidateTablesDTO = { }> } +type RenameTablesBody = { + raw?: { table_name: string; table_schema: string }; + qualify?: { table_name: string; table_schema: string }; +} + @ApiTags('Platform API') @Controller('platform') export class PlatformApiController { @@ -44,6 +53,7 @@ export class PlatformApiController { private readonly elasticsearchService: ElasticsearchService, private readonly dynamoDBService: DynamoDBService, private readonly customersService: CustomersService, + private readonly catalogService: CatalogService, @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, ) { this.logger = dadosferaLogger.logger; @@ -75,6 +85,23 @@ export class PlatformApiController { return jobId?.replace(/-/g, '_') || ''; } + private async getJobByAnyConnectorType(normalizedJobId: string, user: RequestUser): Promise { + const connectorTypes = ['jdbc', 'singer', 's3']; + for (const type of connectorTypes) { + try { + const job = await this.platformApiService.proxy( + 'GET', + `/jobs/${type}/${normalizedJobId}`, + user, + ); + return job; + } catch (error) { + // Continue to next connector type + } + } + throw new HttpException(`Job ${normalizedJobId} not found in any connector type (jdbc, singer, s3)`, 404); + } + /** * Extract the pipeline ID (base UUID) from a job ID. * Job IDs have format "uuid-suffix" where suffix is the job index (e.g., "0", "1"). @@ -728,26 +755,10 @@ export class PlatformApiController { ); } - // ==================== Catalog ROUTES ==================== - - @Get('pipelines/catalog/tables') - @ApiOperation({ summary: 'Get all tables available' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) - async getAvailableTables( - @User() user: RequestUser, - @Query() query: Record, - ) { - return this.platformApiService.proxy( - 'GET', - '/catalog/tables', - user, - undefined, - query, - ); - } + // ==================== PIPELINE VALIDATION ==================== @Get('pipelines/catalog/schemas') - @ApiOperation({ summary: 'Get all schemas available' }) + @ApiOperation({ summary: 'Get available schemas' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getAvailableSchemas( @User() user: RequestUser, @@ -755,7 +766,7 @@ export class PlatformApiController { ) { return this.platformApiService.proxy( 'GET', - '/catalog/schemas', + `/catalog/schemas`, user, undefined, query, @@ -763,18 +774,18 @@ export class PlatformApiController { } @Post('pipelines/catalog/tables/validate') - @ApiOperation({ summary: 'Validate tables and schemas' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + @ApiOperation({ summary: 'Validate Table and Schema' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async validateTableAndSchema( + @Body() payload: ValidationTableDTO, @User() user: RequestUser, @Query() query: Record, - @Body() validateTablesDto: ValidateTablesDTO[] ) { return this.platformApiService.proxy( 'POST', - '/catalog/tables/validate', + `/catalog/tables/validate`, user, - validateTablesDto, + payload, query, ); } @@ -986,6 +997,192 @@ export class PlatformApiController { return result; } + @Post('jobs/:jobId/rename-tables') + @ApiOperation({ summary: 'Rename job output tables and sync to catalog' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + async renameJobTables( + @Param('jobId') jobId: string, + @Body() body: RenameTablesBody, + @User() user: RequestUser, + ) { + const normalizedJobId = this.normalizeJobId(jobId); + + const currentJob = await this.getJobByAnyConnectorType(normalizedJobId, user); + + const result = await this.platformApiService.proxy( + 'POST', + `/jobs/${normalizedJobId}/rename-tables`, + user, + body, + ); + + try { + await this.syncTableRenameToCatalog(jobId, body, currentJob, user); + } catch (error) { + this.logger.error('Catalog sync failed, rolling back Snowflake rename', { jobId, error: error.message }); + + const reverseBody = this.buildSnowflakeRollbackBody(body, currentJob.output_config || {}); + if (reverseBody) { + try { + await this.platformApiService.proxy('POST', `/jobs/${normalizedJobId}/rename-tables`, user, reverseBody); + this.logger.info('Snowflake rename rolled back', { jobId }); + } catch (rollbackError) { + this.logger.error('Snowflake rollback failed', { jobId, error: rollbackError.message }); + } + } + + throw new HttpException('Table rename failed: catalog sync error, Snowflake reverted', 500); + } + + return result; + } + + private buildSnowflakeRollbackBody( + body: RenameTablesBody, + outputConfig: any, + ): RenameTablesBody | null { + const reverse: RenameTablesBody = {}; + + if (body.raw) { + const nested = outputConfig.raw; + const oldTableName = nested?.table_name || outputConfig.table_name; + const oldTableSchema = nested?.table_schema || 'PUBLIC'; + if (oldTableName) reverse.raw = { table_name: oldTableName, table_schema: oldTableSchema }; + } + + if (body.qualify) { + const nested = outputConfig.qualify; + if (nested?.table_name) reverse.qualify = { table_name: nested.table_name, table_schema: nested.table_schema || 'STAGED' }; + } + + return Object.keys(reverse).length > 0 ? reverse : null; + } + + /** + * Sync table rename to Elasticsearch and Nimbus. + * + * For each target (raw, qualify): + * 1. Resolve old table name from output_config + * 2. Find the ES data asset by pipeline + table + schema + * 3. Update ES, Nimbus table-metadata, column-metadata, and data-preview + * 4. If any step fails, rollback all completed steps for that target + */ + private async syncTableRenameToCatalog( + jobId: string, + body: RenameTablesBody, + currentJob: any, + user: RequestUser, + ): Promise { + const pipelineId = this.extractPipelineIdFromJobId(jobId); + const outputConfig = currentJob.output_config || {}; + const nimbusUrl = this.catalogService._getNimbusUrl({ info: { customer: user.customer_name } }); + const databaseName = `DADOSFERA_PRD_${user.customer_name.toUpperCase()}`; + + const targets = this.buildRenameTargets(body, outputConfig); + + for (const { key, oldTableName, oldTableSchema, newValues } of targets) { + const rollbackSteps: Array<() => Promise> = []; + + try { + const dataAsset = await this.elasticsearchService.findDataAssetByTable( + user.customer_name, oldTableName, oldTableSchema, + ); + + if (!dataAsset) { + this.logger.warn(`No data asset found for ${key}`, { jobId, pipelineId, oldTableName, oldTableSchema }); + continue; + } + + const { _es_id: esAssetId, nimbus_id: nimbusId } = dataAsset; + const oldValues = { table_name: oldTableName, table_schema: oldTableSchema }; + + // ES update + const esFields = { name: newValues.table_name, table_name: newValues.table_name, table_schema: newValues.table_schema, display_name: newValues.table_name }; + await this.elasticsearchService.updateDataAsset(user.customer_name, esAssetId, esFields); + rollbackSteps.push(() => this.elasticsearchService.updateDataAsset( + user.customer_name, esAssetId, + { name: oldTableName, table_name: oldTableName, table_schema: oldTableSchema, display_name: oldTableName }, + )); + + const newTableNameUpper = newValues.table_name.toUpperCase(); + const newTableSchemaUpper = newValues.table_schema.toUpperCase(); + const oldTableNameUpper = oldTableName.toUpperCase(); + const oldTableSchemaUpper = oldTableSchema.toUpperCase(); + + // Nimbus table-metadata + if (nimbusId) { + await this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, { table_name: newTableNameUpper, table_schema: newTableSchemaUpper }); + rollbackSteps.push(() => this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, { table_name: oldTableNameUpper, table_schema: oldTableSchemaUpper })); + } + + // Nimbus column-metadata + await this.catalogService.renameColumnMetadataOnNimbus( + nimbusUrl, databaseName, oldTableNameUpper, oldTableSchemaUpper, newTableNameUpper, newTableSchemaUpper, + ); + rollbackSteps.push(() => this.catalogService.renameColumnMetadataOnNimbus( + nimbusUrl, databaseName, newTableNameUpper, newTableSchemaUpper, oldTableNameUpper, oldTableSchemaUpper, + )); + + // Nimbus data-preview + await this.catalogService.renameDataPreviewOnNimbus( + nimbusUrl, databaseName, oldTableNameUpper, oldTableSchemaUpper, newTableNameUpper, newTableSchemaUpper, + ); + rollbackSteps.push(() => this.catalogService.renameDataPreviewOnNimbus( + nimbusUrl, databaseName, newTableNameUpper, newTableSchemaUpper, oldTableNameUpper, oldTableSchemaUpper, + )); + + this.logger.info(`Synced catalog rename for ${key}`, { jobId, oldTableName, newTableName: newValues.table_name }); + } catch (error) { + this.logger.error(`Catalog sync failed for ${key}, rolling back catalog`, { jobId, error: error.message }); + await this.executeRollback(rollbackSteps, key, jobId); + throw error; + } + } + } + + private buildRenameTargets( + body: RenameTablesBody, + outputConfig: any, + ): Array<{ key: string; oldTableName: string; oldTableSchema: string; newValues: { table_name: string; table_schema: string } }> { + const DEFAULT_SCHEMAS = { raw: 'PUBLIC', qualify: 'STAGED' }; + const targets: Array<{ key: string; oldTableName: string; oldTableSchema: string; newValues: { table_name: string; table_schema: string } }> = []; + + for (const key of ['raw', 'qualify'] as const) { + if (!body[key]) continue; + + const nested = outputConfig[key]; + + // qualify: only sync if output_config.qualify already exists + if (key === 'qualify' && !nested?.table_name) continue; + + const oldTableName = nested?.table_name || outputConfig.table_name; + if (!oldTableName) continue; + + targets.push({ + key, + oldTableName, + oldTableSchema: nested?.table_schema || DEFAULT_SCHEMAS[key], + newValues: body[key], + }); + } + + return targets; + } + + private async executeRollback( + steps: Array<() => Promise>, + targetKey: string, + jobId: string, + ): Promise { + for (const rollback of steps.reverse()) { + try { + await rollback(); + } catch (error) { + this.logger.error(`Rollback failed for ${targetKey}`, { jobId, error: error.message }); + } + } + } + @Get('jobs/jdbc/configs/allowed_datatypes') @ApiOperation({ summary: 'Get allowed datatypes for JDBC' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) diff --git a/src/modules/platform-api/platform-api.dto.ts b/src/modules/platform-api/platform-api.dto.ts new file mode 100644 index 0000000..3f6b89f --- /dev/null +++ b/src/modules/platform-api/platform-api.dto.ts @@ -0,0 +1,9 @@ +import { ApiProperty } from "@nestjs/swagger"; + +export class ValidationTableDTO { + @ApiProperty() + tables: Array<{ + table_name: string; + table_schema: string; + }> +} \ No newline at end of file diff --git a/src/modules/platform-api/platform-api.module.ts b/src/modules/platform-api/platform-api.module.ts index c2b31e4..b78e680 100644 --- a/src/modules/platform-api/platform-api.module.ts +++ b/src/modules/platform-api/platform-api.module.ts @@ -7,9 +7,10 @@ import { PlatformApiService } from './platform-api.service'; import { ElasticsearchModule } from '../../services/elasticsearch'; import { DynamoDBModule } from '../../services/dynamodb'; import { CustomersModule } from '../customers/customers.module'; +import { CatalogModule } from '../catalog/catalog.module'; @Module({ - imports: [ElasticsearchModule, DynamoDBModule, CustomersModule], + imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule], controllers: [PlatformApiController], providers: [PlatformApiService, DadosferaLogger], exports: [PlatformApiService], diff --git a/src/modules/platform-api/platform-api.service.ts b/src/modules/platform-api/platform-api.service.ts index 97797a3..7fe9928 100644 --- a/src/modules/platform-api/platform-api.service.ts +++ b/src/modules/platform-api/platform-api.service.ts @@ -89,6 +89,12 @@ export class PlatformApiService { // Propagate non-2xx responses as HttpExceptions if (response.status >= 400) { + 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); } @@ -101,6 +107,8 @@ export class PlatformApiService { method: method.toUpperCase(), }); + this.logger.error(error) + if (error instanceof HttpException) { throw error; } diff --git a/src/services/elasticsearch/elasticsearch.service.ts b/src/services/elasticsearch/elasticsearch.service.ts index e2c1857..ab2791a 100644 --- a/src/services/elasticsearch/elasticsearch.service.ts +++ b/src/services/elasticsearch/elasticsearch.service.ts @@ -358,6 +358,81 @@ export class ElasticsearchService { } } + private getDataAssetIndex(customerName: string): string { + return `${customerName}_data_assets_catalog`; + } + + async findDataAssetByTable( + customerName: string, + tableName: string, + tableSchema: string, + ): Promise<{ id: string; nimbus_id: number | null; [key: string]: any } | null> { + const index = this.getDataAssetIndex(customerName); + + this.logger.info('Elasticsearch: Searching data asset', { + index, + tableName, + tableSchema, + }); + + try { + const response = await this.client.post(`/${index}/_search`, { + query: { + bool: { + must: [ + { term: { 'table_name.keyword': tableName.toUpperCase() } }, + { term: { 'table_schema.keyword': tableSchema.toUpperCase() } }, + ], + }, + }, + size: 1, + }); + + const hits = response.data.hits?.hits || []; + if (hits.length === 0) { + this.logger.warn('Elasticsearch: Data asset not found', { tableName, tableSchema, index }); + return null; + } + + return { ...hits[0]._source, _es_id: hits[0]._id }; + } catch (error) { + this.handleError('findDataAssetByTable', error, { tableName, tableSchema, index }); + throw error; + } + } + + async updateDataAsset( + customerName: string, + assetId: string, + updates: Record, + ): Promise { + const index = this.getDataAssetIndex(customerName); + + this.logger.info('Elasticsearch: Updating data asset', { + index, + assetId, + fields: Object.keys(updates), + }); + + try { + const response = await this.client.post( + `/${index}/_update/${assetId}`, + { doc: updates }, + { params: { refresh: 'wait_for' } }, + ); + + this.logger.info('Elasticsearch: Data asset updated', { + assetId, + result: response.data.result, + }); + + return response.data; + } catch (error) { + this.handleError('updateDataAsset', error, { assetId, index }); + throw error; + } + } + private handleError( operation: string, error: any, 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 + } + } + + } +}