diff --git a/src/modules/platform-api/platform-api.controller.spec.ts b/src/modules/platform-api/platform-api.controller.spec.ts index 25077a1..ae4c352 100644 --- a/src/modules/platform-api/platform-api.controller.spec.ts +++ b/src/modules/platform-api/platform-api.controller.spec.ts @@ -199,3 +199,113 @@ describe('PlatformApiController - addTable', () => { }); }); }); + + +describe('PlatformApiController - deleteTables (batch)', () => { + let controller: PlatformApiController; + let platformApiService: { proxy: jest.Mock }; + let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock }; + + beforeEach(async () => { + platformApiService = { proxy: jest.fn() }; + inputsService = { + markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }), + unmarkTableDeleted: jest.fn().mockResolvedValue({}), + }; + + const module: TestingModule = await Test.createTestingModule({ + controllers: [PlatformApiController], + providers: [ + { provide: PlatformApiService, useValue: platformApiService }, + { provide: ElasticsearchService, useValue: {} }, + { provide: DynamoDBService, useValue: {} }, + { provide: CustomersService, useValue: {} }, + { provide: CatalogService, useValue: {} }, + { provide: InputsService, useValue: inputsService }, + { provide: PipelinesService, useValue: {} }, + { provide: DadosferaLogger, useValue: { logger } }, + ], + }).compile(); + + controller = module.get(PlatformApiController); + }); + + const pipelineWithJobs = () => ({ + jobs: [ + { job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } }, + { job_id: 'p_1', input: { connector: 'cdc', table_name: 'clientes' } }, + { job_id: 'p_2', input: { connector: 'cdc', table_name: 'produtos' } }, + ], + }); + + it('removes N tables in ONE platform call (connectors reconfigured once)', async () => { + platformApiService.proxy.mockImplementation((method: string) => + Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}), + ); + + await controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'produtos'] }, mockUser); + + // one mark per table + expect(inputsService.markTableDeleted).toHaveBeenCalledTimes(2); + + // exactly one DELETE to the batch endpoint, with BOTH job_ids + const deleteCalls = platformApiService.proxy.mock.calls.filter( + ([m, p]: any[]) => m === 'DELETE' && p === '/pipeline/pid/jobs', + ); + expect(deleteCalls).toHaveLength(1); + expect(deleteCalls[0][3]).toEqual({ + job_ids: ['p_0', 'p_2'], + delete_snowflake_tables: false, + }); + // never the per-job route + const perJob = platformApiService.proxy.mock.calls.some( + ([m, p]: any[]) => m === 'DELETE' && String(p).startsWith('/jobs/'), + ); + expect(perJob).toBe(false); + }); + + it('rolls back only this call\'s marks when the platform delete fails', async () => { + platformApiService.proxy.mockImplementation((method: string) => { + if (method === 'GET') return Promise.resolve(pipelineWithJobs()); + return Promise.reject(new Error('platform boom')); + }); + + await expect( + controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser), + ).rejects.toThrow(); + + // both marks rolled back, nothing else + expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(2); + const unmarked = inputsService.unmarkTableDeleted.mock.calls.map((c: any[]) => c[0].table_name).sort(); + expect(unmarked).toEqual(['clientes', 'pedidos']); + }); + + it('rolls back the marks made so far if a later mark fails (atomic)', async () => { + // second mark fails → first must be rolled back, no platform delete attempted + inputsService.markTableDeleted + .mockResolvedValueOnce({ is_deleted: true, deleted_at: 't' }) + .mockRejectedValueOnce(new Error('dynamo boom')); + + await expect( + controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser), + ).rejects.toThrow(); + + expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(1); + expect(inputsService.unmarkTableDeleted.mock.calls[0][0].table_name).toBe('pedidos'); + // never reached the platform delete + const attemptedDelete = platformApiService.proxy.mock.calls.some(([m]: any[]) => m === 'DELETE'); + expect(attemptedDelete).toBe(false); + }); + + it('404s when a requested table has no matching job', async () => { + platformApiService.proxy.mockImplementation((method: string) => + Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}), + ); + + await expect( + controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'ghost'] }, mockUser), + ).rejects.toThrow(); + // the successful mark (pedidos) must be rolled back + expect(inputsService.unmarkTableDeleted).toHaveBeenCalled(); + }); +}); diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 0a6c6cc..e547b0d 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -992,6 +992,75 @@ export class PlatformApiController { } } + @Delete('pipelines/:pipelineId/inputs/:inputId/tables') + @ApiOperation({ summary: 'Batch-remove tables from an input; reconfigures the CDC connectors once' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE) + @UseGuards(PipelineExecutionGuard) + async deleteTables( + @Param('pipelineId') pipelineId: string, + @Param('inputId') inputId: string, + @Body() body: { table_names: string[] }, + @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; + } + } + @Post('pipelines/:pipelineId/inputs/:inputId/tables') @ApiOperation({ summary: 'Add a CDC table to an input and dispatch its platform jobs' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)