diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 5fd0b94..ff5539a 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -1012,225 +1012,6 @@ export class PlatformApiController { // ==================== JOBS - JDBC SYNC MODE ROUTES ==================== - @Get('jobs/jdbc/:jobId') - @ApiOperation({ summary: 'Get JDBC job details' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) - async getJdbcJob(@Param('jobId') jobId: string, @User() user: RequestUser) { - // Normalize job ID for Platform API (replace - with _) - const normalizedJobId = this.normalizeJobId(jobId); - return this.platformApiService.proxy('GET', `/jobs/jdbc/${normalizedJobId}`, user); - } - - @Post('jobs/jdbc/:jobId/sync-mode') - @ApiOperation({ summary: 'Update JDBC job sync mode' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) - async updateJdbcSyncMode( - @Param('jobId') jobId: string, - @Body() body: any, - @User() user: RequestUser, - ) { - // Normalize job ID for Platform API (replace - with _) - const normalizedJobId = this.normalizeJobId(jobId); - - const result = await this.platformApiService.proxy( - 'POST', - `/jobs/jdbc/${normalizedJobId}/sync-mode`, - user, - body, - ); - - // Sync to DynamoDB (pass raw jobId for pipeline extraction) - await this.syncJdbcSyncModeToDynamoDB(jobId, body, user); - - 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) @@ -1242,52 +1023,6 @@ export class PlatformApiController { ); } - // ==================== JOBS - SINGER REPLICATION ROUTES ==================== - - @Get('jobs/singer/:jobId') - @ApiOperation({ summary: 'Get Singer job details' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) - async getSingerJob(@Param('jobId') jobId: string, @User() user: RequestUser) { - // Normalize job ID for Platform API (replace - with _) - const normalizedJobId = this.normalizeJobId(jobId); - return this.platformApiService.proxy('GET', `/jobs/singer/${normalizedJobId}`, user); - } - - @Post('jobs/singer/:jobId/sync-mode') - @ApiOperation({ summary: 'Update Singer job sync mode' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) - async updateSingerSyncMode( - @Param('jobId') jobId: string, - @Body() body: any, - @User() user: RequestUser, - ) { - // Normalize job ID for Platform API (replace - with _) - const normalizedJobId = this.normalizeJobId(jobId); - - const result = await this.platformApiService.proxy( - 'POST', - `/jobs/singer/${normalizedJobId}/sync-mode`, - user, - body, - ); - - // Sync to DynamoDB (pass raw jobId for pipeline extraction) - await this.syncSingerSyncModeToDynamoDB(jobId, body, user); - - return result; - } - - // ==================== JOBS - S3 ROUTES ==================== - - @Get('jobs/s3/:jobId') - @ApiOperation({ summary: 'Get S3 job details' }) - @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) - async getS3Job(@Param('jobId') jobId: string, @User() user: RequestUser) { - // Normalize job ID for Platform API (replace - with _) - const normalizedJobId = this.normalizeJobId(jobId); - return this.platformApiService.proxy('GET', `/jobs/s3/${normalizedJobId}`, user); - } - // ==================== HEALTH ROUTE ==================== @Get('health')