mirror of
https://github.com/dadosfera/maestro.git
synced 2026-09-01 04:08:16 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
901518d26c | ||
|
|
4ab8147149 |
@@ -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')
|
||||
|
||||
Reference in New Issue
Block a user