feat(cdc): batch table removal — reconfigure connectors once

Removing N tables via Edit Objects previously looped a single-table DELETE per table (maestro deleteTable hardcodes job_ids:[one]), so the Debezium source + Snowflake sink connectors were rewritten/restarted once per table. platform-api's DELETE /pipeline/:id/jobs already batches (2 connector writes total for any N), but nothing above it used the array. New maestro DELETE /pipelines/:pipelineId/inputs/:inputId/tables takes { table_names: [] }: soft-deletes each in DynamoDB (tracking successes), resolves all table_names -> job_ids from the platform pipeline in one GET, then makes ONE DELETE /pipeline/:id/jobs with all job_ids. All-or-nothing: any failure (a later mark, an unmatched table, or the platform delete) rolls back only the marks made in this call. The single-table deleteTable route is kept (unchanged) — nothing else depends on removing it, and that's a separable cleanup. Tests: N tables -> one platform DELETE with all job_ids and no per-job call; rollback on platform failure; rollback + no delete when a later mark fails; 404 for an unmatched table. 8 controller specs pass; maestro builds. Co-Authored-By: WOZCODE <contact@withwoz.com>
2026-10-16 01:19:08 +00:00 · 2026-08-19 10:47:06 -03:00
parent 714c1334b9
commit ead3fa12dc
2 changed files with 179 additions and 0 deletions
@@ -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>(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();
+  });
+});
@@ -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)