From 38a9e21f5f737e2e93859ed7beff7dd4c4be7c90 Mon Sep 17 00:00:00 2001 From: Rafael Date: Thu, 5 Mar 2026 17:29:12 -0300 Subject: [PATCH] feat: add rename-tables endpoint with catalog sync and rollback Add POST /platform/jobs/:jobId/rename-tables that renames Snowflake tables via platform-api and syncs the rename to Elasticsearch and Nimbus (table-metadata, column-metadata, data-preview). If catalog sync fails, all completed catalog steps are rolled back in reverse order and the Snowflake rename is reverted. - Support any connector type (jdbc, singer, s3) via getJobByAnyConnectorType - Resolve old table names from output_config (raw/qualify) - Skip qualify sync when output_config.qualify has no table_name - Add findDataAssetByPipelineAndTable and updateDataAsset to ElasticsearchService - Add renameTableOnNimbus, renameColumnMetadataOnNimbus, renameDataPreviewOnNimbus to CatalogService - Add upstream error logging to PlatformApiService Co-Authored-By: Claude Opus 4.6 --- src/modules/catalog/catalog.service.ts | 58 +++++ .../platform-api/platform-api.controller.ts | 207 ++++++++++++++++++ .../platform-api/platform-api.module.ts | 3 +- .../platform-api/platform-api.service.ts | 6 + .../elasticsearch/elasticsearch.service.ts | 78 +++++++ 5 files changed, 351 insertions(+), 1 deletion(-) 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/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 978f464..319dcd4 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,8 @@ 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'; type ValidateTablesDTO = { @@ -34,6 +37,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 +52,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 +84,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"). @@ -968,6 +994,187 @@ 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.findDataAssetByPipelineAndTable( + user.customer_name, pipelineId, 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 }, + )); + + // Nimbus table-metadata + if (nimbusId) { + await this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, newValues); + rollbackSteps.push(() => this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, oldValues)); + } + + // Nimbus column-metadata + await this.catalogService.renameColumnMetadataOnNimbus( + nimbusUrl, databaseName, oldTableName, oldTableSchema, newValues.table_name, newValues.table_schema, + ); + rollbackSteps.push(() => this.catalogService.renameColumnMetadataOnNimbus( + nimbusUrl, databaseName, newValues.table_name, newValues.table_schema, oldTableName, oldTableSchema, + )); + + // Nimbus data-preview + await this.catalogService.renameDataPreviewOnNimbus( + nimbusUrl, databaseName, oldTableName, oldTableSchema, newValues.table_name, newValues.table_schema, + ); + rollbackSteps.push(() => this.catalogService.renameDataPreviewOnNimbus( + nimbusUrl, databaseName, newValues.table_name, newValues.table_schema, oldTableName, oldTableSchema, + )); + + 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.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..f204678 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', { + status: response.status, + data: response.data, + path, + method: method.toUpperCase(), + }); throw new HttpException(response.data, response.status); } diff --git a/src/services/elasticsearch/elasticsearch.service.ts b/src/services/elasticsearch/elasticsearch.service.ts index e2c1857..8fff83e 100644 --- a/src/services/elasticsearch/elasticsearch.service.ts +++ b/src/services/elasticsearch/elasticsearch.service.ts @@ -358,6 +358,84 @@ export class ElasticsearchService { } } + private getDataAssetIndex(customerName: string): string { + return `${customerName}_data_assets_catalog`; + } + + async findDataAssetByPipelineAndTable( + customerName: string, + pipelineId: 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, + pipelineId, + tableName, + tableSchema, + }); + + try { + const response = await this.client.post(`/${index}/_search`, { + query: { + bool: { + must: [ + { term: { 'pipeline_id.keyword': pipelineId } }, + { term: { 'table_name.keyword': tableName } }, + { term: { 'table_schema.keyword': tableSchema } }, + ], + }, + }, + size: 1, + }); + + const hits = response.data.hits?.hits || []; + if (hits.length === 0) { + this.logger.warn('Elasticsearch: Data asset not found', { pipelineId, tableName, tableSchema, index }); + return null; + } + + return { ...hits[0]._source, _es_id: hits[0]._id }; + } catch (error) { + this.handleError('findDataAssetByPipelineAndTable', error, { pipelineId, 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,