gitea_admin/maestro
1
0
Fork 0
mirror of https://github.com/dadosfera/maestro.git synced 2026-10-15 03:09:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

UPDATE: one platform route for table removal; table orchestration moves to PipelineTablesService

Browse Source 
Review (maestro #510): batch and CDC tables are both removed through
DELETE /pipeline/{id}/jobs (the platform dispatches by type: Airflow refresh
vs Kafka Connect reconfigure), so the connector 'if' in the controller is
gone. deleteTable/deleteTables/addTable are now thin controller methods over
PipelineTablesService (mark -> resolve jobs -> platform -> rollback), with
typed request bodies.

Behavior change for batch: the platform refuses to remove the LAST table of
a pipeline (400 'Cannot remove all jobs'), where DELETE /jobs/{id} allowed it.

Co-Authored-By: WOZCODE <contact@withwoz.com>
This commit is contained in:
Rafael
2026-08-29 12:57:49 -03:00
co-authored by WOZCODE
parent 23f0554311
commit 2347f65b78
6 changed files with 513 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/platform-api/platform-api.controller.ts
+10 -166
View File
@@ -11,7 +11,6 @@ import {
Inject,
BadRequestException,
HttpException,
NotFoundException,
UseGuards,
} from '@nestjs/common';
import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger';
@@ -31,10 +30,9 @@ 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';
import { AddCdcTableBody, DeleteTableBody, DeleteTablesBody, ValidationTableDTO } from './platform-api.dto';
import { InputsService } from '../inputs/inputs.service';
import { toCdcTable } from '../inputs/cdc-table.mapper';
import { PipelinesService } from '../pipelinesV2/pipelines.service';
import { PipelineTablesService } from './pipeline-tables.service';
import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
@@ -63,7 +61,7 @@ export class PlatformApiController {
private readonly customersService: CustomersService,
private readonly catalogService: CatalogService,
private readonly inputsService: InputsService,
private readonly pipelinesClientService: PipelinesService,
private readonly pipelineTablesService: PipelineTablesService,
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
) {
this.logger = dadosferaLogger.logger;
@@ -989,60 +987,16 @@ export class PlatformApiController {
}
@Delete('pipelines/:pipelineId/inputs/:inputId')
@ApiOperation({ summary: 'Mark a table as deleted and delete its associated job via platform-api' })
@ApiOperation({ summary: 'Mark a table as deleted and remove its job via platform-api' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
@UseGuards(PipelineExecutionGuard)
async deleteTable(
@Param('pipelineId') pipelineId: string,
@Param('inputId') inputId: string,
@Body() body: { table_name: string },
@Body() body: DeleteTableBody,
@User() user: RequestUser,
) {
const tableName = body.table_name;
const info = {
customer_id: user.customer_id,
customer: user.customer_name,
user_id: user.user_id,
};
this.logger.info('deleteTable: marking table as deleted', { inputId, tableName });
const updatedInput: any = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info });
this.logger.info('deleteTable: table marked as deleted', { inputId, tableName });
try {
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
this.logger.info('deleteTable: fetching pipeline from platform-api', { pipelineId, normalizedPipelineId });
const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
this.logger.info('deleteTable: pipeline fetched', { jobCount: platformPipeline?.jobs?.length });
const job = platformPipeline?.jobs?.find((j: any) => j.input?.table_name === tableName);
if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`);
this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id });
if (job.input?.connector === 'cdc') {
// CDC: reconfigure the Debezium/Kafka-Connect connector (stop
// replicating this table); keep the landed Snowflake data.
await this.platformApiService.proxy(
'DELETE',
`/pipeline/${normalizedPipelineId}/jobs`,
user,
{ job_ids: [job.job_id], delete_snowflake_tables: false },
);
} else {
await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user);
}
this.logger.info('deleteTable: job deleted', { jobId: job.job_id });
return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at };
} catch (error) {
this.logger.error('deleteTable: platform-api delete failed, attempting rollback', { tableName, error: error.message });
try {
await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: tableName, info });
} catch (rollbackError) {
this.logger.error('deleteTable: rollback failed', { tableName, error: rollbackError.message });
}
throw error;
}
return this.pipelineTablesService.removeTable(pipelineId, inputId, body.table_name, user);
}
@Delete('pipelines/:pipelineId/inputs/:inputId/tables')
@@ -1052,66 +1006,14 @@ export class PlatformApiController {
async deleteTables(
@Param('pipelineId') pipelineId: string,
@Param('inputId') inputId: string,
@Body() body: { table_names: string[] },
@Body() body: DeleteTablesBody,
@User() user: RequestUser,
) {
const tableNames = body.table_names ?? [];
if (tableNames.length === 0) {
throw new BadRequestException('table_names must be a non-empty array');
}
const info = {
customer_id: user.customer_id,
customer: user.customer_name,
user_id: user.user_id,
};
// 1) Soft-delete each table in DynamoDB, tracking which succeeded so a later
// failure only rolls back the marks made in THIS call.
const marked: string[] = [];
const rollback = async () => {
for (const name of marked) {
try {
await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info });
} catch (rollbackError) {
this.logger.error('deleteTables: rollback failed', { tableName: name, error: rollbackError.message });
}
}
};
try {
for (const name of tableNames) {
await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info });
marked.push(name);
}
// 2) Resolve all table_names -> job_ids from the platform pipeline (one GET).
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
const jobs = platformPipeline?.jobs ?? [];
const jobIds: string[] = [];
for (const name of tableNames) {
const job = jobs.find((j: any) => j.input?.table_name === name);
if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`);
jobIds.push(job.job_id);
}
// 3) Remove them all in ONE platform call so the Debezium source + Snowflake
// sink connectors are reconfigured a single time, not once per table.
await this.platformApiService.proxy(
'DELETE',
`/pipeline/${normalizedPipelineId}/jobs`,
user,
{ job_ids: jobIds, delete_snowflake_tables: false },
);
this.logger.info('deleteTables: jobs deleted', { jobIds });
return { table_names: tableNames, deleted: true };
} catch (error) {
this.logger.error('deleteTables: failed, rolling back this call\'s marks', { tableNames, error: error.message });
await rollback();
throw error;
}
return this.pipelineTablesService.removeTables(pipelineId, inputId, tableNames, user);
}
@Post('pipelines/:pipelineId/inputs/:inputId/tables')
@@ -1121,68 +1023,10 @@ export class PlatformApiController {
async addTable(
@Param('pipelineId') pipelineId: string,
@Param('inputId') inputId: string,
@Body() body: {
table_name: string;
table_schema: string;
primary_keys: string[];
destinations: {
raw: { table_schema: string; table_name: string };
qualify: { table_schema: string; table_name: string };
};
// Iceberg destination only (protospack CdcTable.iceberg_table_name);
// absent for snowflake, back-compat.
iceberg_table_name?: string;
// Per-table deduped (qualify) Iceberg table name (protospack
// CdcTable.iceberg_qualify_table_name); absent => same as the raw name.
iceberg_qualify_table_name?: string;
// Source column schema for iceberg deduped table pre-create (protospack CdcTable.columns).
columns?: { name: string; type: string; is_primary_key: boolean }[];
// Columns the user chose to ignore -> Debezium column.exclude.list.
column_exclude_list?: string[];
},
@Body() body: AddCdcTableBody,
@User() user: RequestUser,
) {
const info = {
customer_id: user.customer_id,
customer: user.customer_name,
user_id: user.user_id,
};
const cdcTable = toCdcTable(body);
this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name });
await this.inputsService.addCdcTable({ client_id: user.customer_id, id: inputId, table: cdcTable, info });
this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name });
try {
const customInfo = {
customer_id: user.customer_id,
user_id: user.user_id,
customer: user.customer_name,
};
const res = await this.pipelinesClientService.addCdcJobs({
pipeline_id: pipelineId,
input_id: inputId,
tables: [{
table_schema: body.table_schema,
table_name: body.table_name,
primary_keys: body.primary_keys,
destinations: body.destinations,
}],
info: customInfo,
});
return res;
} catch (error) {
this.logger.error('addTable: platform AddJobs failed, rolling back the DynamoDB row', { tableName: body.table_name, error: error.message });
try {
await this.inputsService.removeCdcTable({ client_id: user.customer_id, id: inputId, table_name: body.table_name, info });
} catch (rbErr) {
this.logger.error('addTable: rollback failed', { error: rbErr.message });
}
throw error;
}
return this.pipelineTablesService.addTable(pipelineId, inputId, body, user);
}
// ==================== JOBS - JDBC SYNC MODE ROUTES ====================