Compare commits

...
Author SHA1 Message Date
marcos-silva-rodrigues 901518d26c FIX: remove full table 2026-04-09 17:38:28 -03:00
marcos-silva-rodrigues 4ab8147149 FEAT: remove endpoints 2026-04-09 17:23:58 -03:00
2 changed files with 0 additions and 271 deletions
@@ -533,12 +533,6 @@ export class PipelinesService implements OnModuleInit {
jobSyncMode['target_load_type'] = table.type;
}
if (table.type === "full_load" && pipelineType === "application") {
hasUpdateSyncMode = true;
jobSyncMode['target_load_type'] = "FULL_TABLE";
}
if(hasUpdateSyncMode) {
jobUpdate["sync_mode"] = jobSyncMode;
}
@@ -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<void> {
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<void>> = [];
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<void>>,
targetKey: string,
jobId: string,
): Promise<void> {
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')