FIX: address review round 2 on the CDC refactor

- toCdcTable accepts the source table named either table_name (platform
  bodies) or name (create DTO), so call sites pass it point-free:
  body.tables.map(toCdcTable). The identity is still derived in one place.
- PipelineTablesService declares its logger like every other maestro service
  (logger: DadosferaLogger assigned from the injected instance).

Co-Authored-By: WOZCODE <contact@withwoz.com>

CLAUDE.md

@@ -0,0 +1,56 @@
+# maestro — development policy
+
+Maestro is the NestJS BFF between the Angular frontend and the gRPC services
+(pi-factory, in-factory) / platform-api. These rules come from code review and
+apply to every change; the same rules live in pi-factory and in-factory.
+
+## Layering
+- Controllers are thin: decorators, body validation, one call into a service,
+  response shape. Orchestration (multi-step calls, rollbacks, platform-api or
+  gRPC round-trips) lives in a `*.service.ts`. Example:
+  `platform-api/pipeline-tables.service.ts`.
+- One platform-api route per operation. If the platform already dispatches by
+  pipeline type (batch vs CDC), do not branch on the type here — call the
+  route that dispatches (e.g. `DELETE /pipeline/{id}/jobs` for table removal,
+  never `DELETE /jobs/{id}` from maestro).
+
+## Types
+- No `any` / `as any`. gRPC calls take the protospack request type exactly
+  (`AddCdcTableRequest`, `MarkTableDeletedRequest`, ...); DTOs and interfaces
+  are explicit classes/interfaces. Use `Record<string, T>` or `unknown` (with
+  narrowing) when a shape is genuinely open — never `any`.
+- Protospack payloads are built in one mapper per entity (e.g.
+  `inputs/cdc-table.mapper.ts`), so a field that mirrors another
+  (`CdcTable.name` == `table_name`) is derived in exactly one place.
+
+## Pipeline type
+- Decide CDC vs batch with `src/utils/cdc.ts` (`isCdcPlugin`, `isCdcJob`,
+  `isCdcPipeline`), never with inline `plugin.endsWith('_cdc')`, `!!x`, or
+  the absence of some other data (e.g. "no run history ⇒ editable").
+- The platform-api stamps `job.input.connector === 'cdc'` on CDC jobs; that
+  is the authoritative discriminator once a pipeline exists.
+
+## Dependencies
+- `@dadosfera/protospack-v2` is consumed from CodeArtifact, pinned to an exact
+  version (`npm i @dadosfera/protospack-v2@<version> --save-exact`). A `file:`
+  / tarball reference is for local development only and must never be
+  committed. Note: `^3.40.0-beta.N` resolves to the stable `3.40.0` —
+  prereleases must be pinned exactly.
+
+## Commits and releases
+- Deploys are cut by semantic-release with the eslint preset: the commit
+  title MUST start with `FIX:` (patch), `UPDATE:` or `FEAT:` (minor). A
+  lowercase `feat(scope): ...` merges without producing a version, so the
+  code never reaches stg/prd.
+- Pushing to `beta` deploys stg; `main` deploys prd.
+
+## Tests
+- Every service method with a rollback path has a spec covering the
+  happy path, the platform failure (rollback fires) and a rollback failure
+  (does not mask the original error). Run `npx jest <path>` for a folder,
+  `npx tsc --noEmit -p tsconfig.json` for types.
+- Several gRPC client configs read `process.env` at import time, so the full
+  suite needs the service URLs set (any `0.0.0.0:<port>` value works):
+  `DUC_URL=0.0.0.0:50051 INFACTORY_URL=0.0.0.0:50052 PIFACTORY_URL=0.0.0.0:50053 npx jest`.
+  A `Cannot read properties of undefined (reading 'startsWith')` at import is
+  this, not a broken test.

src/guards/pipeline-execution.guard.spec.ts

@@ -3,12 +3,23 @@ import { PipelineExecutionGuard } from './pipeline-execution.guard';
 
 const logger = { info: jest.fn(), error: jest.fn() };
 
-function buildGuard(proxyImpl: jest.Mock) {
-  const platformApiService: any = { proxy: proxyImpl };
-  return new PipelineExecutionGuard(
-    { logger } as any,
-    platformApiService,
+type Routes = {
+  pipeline?: unknown | Error;
+  runs?: unknown | Error;
+};
+
+// The guard makes up to two platform-api reads: GET /pipeline/{id} (type)
+// and, for batch only, GET /pipeline/{id}/pipeline_run (last run).
+function buildGuard(routes: Routes) {
+  const proxy = jest.fn((method: string, path: string) => {
+    const answer = path.endsWith('/pipeline_run') ? routes.runs : routes.pipeline;
+    return answer instanceof Error ? Promise.reject(answer) : Promise.resolve(answer);
+  });
+  const guard = new PipelineExecutionGuard(
+    { logger } as unknown as ConstructorParameters<typeof PipelineExecutionGuard>[0],
+    { proxy } as unknown as ConstructorParameters<typeof PipelineExecutionGuard>[1],
   );
+  return { guard, proxy };
 }
 
 function contextWith(pipelineId = 'abc-123') {
@@ -16,56 +27,96 @@ function contextWith(pipelineId = 'abc-123') {
     switchToHttp: () => ({
       getRequest: () => ({ params: { pipelineId }, user: {} }),
     }),
-  } as any;
+  } as unknown as Parameters<PipelineExecutionGuard['canActivate']>[0];
 }
 
+const cdcPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 't' } }] };
+const batchPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 't' } }] };
+
 describe('PipelineExecutionGuard', () => {
   afterEach(() => jest.clearAllMocks());
 
-  it('allows the edit when the pipeline has no run history (empty array)', async () => {
-    const guard = buildGuard(jest.fn().mockResolvedValue([]));
-    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  describe('CDC pipelines (explicit connector type)', () => {
+    it('allows the edit without consulting run history', async () => {
+      const { guard, proxy } = buildGuard({ pipeline: cdcPipeline });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+      expect(proxy).toHaveBeenCalledTimes(1);
+      expect(proxy).toHaveBeenCalledWith('GET', '/pipeline/abc_123', {});
+    });
+
+    it('is CDC when any job is a CDC job', async () => {
+      const mixed = { jobs: [...batchPipeline.jobs, ...cdcPipeline.jobs] };
+      const { guard, proxy } = buildGuard({ pipeline: mixed });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+      expect(proxy).toHaveBeenCalledTimes(1);
+    });
   });
 
-  it('allows the edit when the last run has no last_status', async () => {
-    const guard = buildGuard(jest.fn().mockResolvedValue([{}]));
-    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  describe('batch pipelines', () => {
+    it('allows the edit when the pipeline never ran (empty run history)', async () => {
+      const { guard } = buildGuard({ pipeline: batchPipeline, runs: [] });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+    });
+
+    it('allows the edit when the last run has no last_status', async () => {
+      const { guard } = buildGuard({ pipeline: batchPipeline, runs: [{}] });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+    });
+
+    it('allows the edit when the pipeline is not running', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: [{ last_status: 'SUCCEEDED' }],
+      });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+    });
+
+    it('blocks with the is-running message when the pipeline is running', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: [{ last_status: 'RUNNING' }],
+      });
+      await expect(guard.canActivate(contextWith())).rejects.toThrow(
+        'Pipeline is running, cannot update input now',
+      );
+    });
+
+    it('treats a pipeline with no jobs as batch and checks its runs', async () => {
+      const { guard, proxy } = buildGuard({ pipeline: { jobs: [] }, runs: [] });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+      expect(proxy).toHaveBeenCalledTimes(2);
+    });
+
+    it('does not double-wrap the is-running BadRequestException', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: [{ last_status: 'running' }],
+      });
+      await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf(
+        BadRequestException,
+      );
+      await expect(guard.canActivate(contextWith())).rejects.not.toThrow(
+        /Error checking pipeline status/,
+      );
+    });
   });
 
-  it('allows the edit when the pipeline is not running', async () => {
-    const guard = buildGuard(
-      jest.fn().mockResolvedValue([{ last_status: 'SUCCEEDED' }]),
-    );
-    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
-  });
+  describe('failures', () => {
+    it('fails closed when the pipeline read fails', async () => {
+      const { guard } = buildGuard({ pipeline: new Error('platform down') });
+      await expect(guard.canActivate(contextWith())).rejects.toThrow(
+        'Error checking pipeline status: platform down',
+      );
+    });
 
-  it('blocks with the is-running message when the pipeline is running', async () => {
-    const guard = buildGuard(
-      jest.fn().mockResolvedValue([{ last_status: 'RUNNING' }]),
-    );
-    await expect(guard.canActivate(contextWith())).rejects.toThrow(
-      'Pipeline is running, cannot update input now',
-    );
-  });
-
-  it('wraps a genuine status-check failure (fail closed)', async () => {
-    const guard = buildGuard(
-      jest.fn().mockRejectedValue(new Error('platform down')),
-    );
-    await expect(guard.canActivate(contextWith())).rejects.toThrow(
-      'Error checking pipeline status: platform down',
-    );
-  });
-
-  it('does not double-wrap the is-running BadRequestException', async () => {
-    const guard = buildGuard(
-      jest.fn().mockResolvedValue([{ last_status: 'running' }]),
-    );
-    await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf(
-      BadRequestException,
-    );
-    await expect(guard.canActivate(contextWith())).rejects.not.toThrow(
-      /Error checking pipeline status/,
-    );
+    it('fails closed when the run-history read fails', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: new Error('runs down'),
+      });
+      await expect(guard.canActivate(contextWith())).rejects.toThrow(
+        'Error checking pipeline status: runs down',
+      );
+    });
   });
 });

src/guards/pipeline-execution.guard.ts

@@ -16,6 +16,7 @@ import {
 import { PipelinesClientConfiguration } from 'src/modules/pipelinesV2/pipelines-client';
 import { PlatformApiService } from 'src/modules/platform-api/platform-api.service';
 import DadosferaLogger from '@dadosfera/dadosfera-logs';
+import { isCdcPipeline } from 'src/utils/cdc';
 
 @Injectable()
 export class PipelineExecutionGuard implements CanActivate {
@@ -39,7 +40,22 @@ export class PipelineExecutionGuard implements CanActivate {
       const user = request.user;
       const idRegex = /[^0-9a-zA-Z_$]+/g;
       const convertedId = pipelineId.replace(idRegex, '_');
-  
+
+      // Decide by the pipeline's explicit type (platform job
+      // `input.connector`), never by the absence of run history.
+      const pipeline = await this.platformApiService.proxy(
+        'GET',
+        `/pipeline/${convertedId}`,
+        user,
+      );
+
+      if (isCdcPipeline(pipeline)) {
+        // CDC pipelines have no batch runs; the platform-api gates connector
+        // edits itself (require_pipeline_running on add/remove tables).
+        this.logger.info('PipelineExecutionGuard: CDC pipeline, no run to block on');
+        return true;
+      }
+
       const status = await this.platformApiService.proxy(
         'GET',
         `/pipeline/${convertedId}/pipeline_run`,
@@ -50,8 +66,7 @@ export class PipelineExecutionGuard implements CanActivate {
 
       this.logger.info('Pipeline current status response:' +  JSON.stringify(currentStatus));
 
-      // No run history (e.g. CDC pipelines never record batch runs) means
-      // nothing is executing — allow the edit rather than crash on .last_status.
+      // Batch pipeline that never ran yet: nothing can be executing.
       if (!currentStatus?.last_status) {
         return true;
       }
@@ -59,9 +74,9 @@ export class PipelineExecutionGuard implements CanActivate {
       if (currentStatus.last_status.toLowerCase() === 'running') {
         this.logger.error('Pipeline is running, cannot update input now');
         throw new BadRequestException('Pipeline is running, cannot update input now');
-      } else {
-        return true;
       }
+
+      return true;
     } catch (error) {
       // Preserve the deliberate is-running rejection; only wrap genuine
       // status-check failures (fail closed on those for a destructive gate).
@@ -71,6 +86,5 @@ export class PipelineExecutionGuard implements CanActivate {
       this.logger.error('Error in PipelineExecutionGuard: ' + error.message);
       throw new BadRequestException('Error checking pipeline status: ' + error.message);
     }
-
   }
 }

src/modules/inputs/cdc-table.mapper.ts

@@ -0,0 +1,43 @@
+import {
+  CdcColumn,
+  CdcTable,
+} from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
+
+/** What a caller knows about a CDC table before it is stored. The source
+ *  table is named `table_name` on the platform-facing bodies and `name` on
+ *  the create DTO (CdcTableReq); either works — deriving one from the other
+ *  happens only here. */
+export interface CdcTableInput {
+  // Optional on the create DTO (CdcTableReq); the platform validates it.
+  table_schema?: string;
+  table_name?: string;
+  name?: string;
+  primary_keys?: string[];
+  iceberg_table_name?: string;
+  iceberg_qualify_table_name?: string;
+  columns?: CdcColumn[];
+  column_exclude_list?: string[];
+}
+
+/**
+ * The only place maestro builds a protospack `CdcTable`.
+ *
+ * `name` is the identity key shared with batch `NewTable` (in-factory keys
+ * update/soft-delete on it) and the proto keeps it required, so it mirrors
+ * `table_name` here and nowhere else. Follow-up (protospack): make `name`
+ * optional and let in-factory be its sole writer (it already backfills
+ * `name ?? table_name`).
+ */
+export function toCdcTable(table: CdcTableInput): CdcTable {
+  const tableName = table.table_name ?? table.name;
+  return {
+    table_schema: table.table_schema,
+    table_name: tableName,
+    name: tableName,
+    primary_keys: table.primary_keys ?? [],
+    iceberg_table_name: table.iceberg_table_name,
+    iceberg_qualify_table_name: table.iceberg_qualify_table_name,
+    columns: table.columns ?? [],
+    column_exclude_list: table.column_exclude_list ?? [],
+  };
+}

src/modules/inputs/dtos/input.model.ts

@@ -78,8 +78,10 @@ export class CdcColumnReq {
 export class CdcTableReq {
   @ApiProperty()
   name: string;
-  @ApiPropertyOptional()
-  table_schema?: string;
+  // Source database/schema. Required: Debezium addresses tables as
+  // `schema.table`, a CDC table without it cannot be replicated.
+  @ApiProperty()
+  table_schema: string;
   @ApiPropertyOptional({ type: [String] })
   primary_keys?: string[];
   // Per-table raw Iceberg table name override (iceberg destination only).

src/modules/inputs/inputs.service.ts

@@ -13,17 +13,21 @@ import { objectCamelToSnake } from 'src/utils/CaseConverter';
 import { IIdRequest, UpdateInputRequest } from './dtos/old_interfaces';
 import { Input } from '@dadosfera/protospack-v2';
 import {
+  AddCdcTableRequest,
   GetAvailableEntitiesRequest,
   InputCreateGenericRequest,
   InputCreateCdcRequest,
   InputCreateS3Request,
   InputNewCreateRequest,
   InputUpdateResponse,
+  MarkTableDeletedRequest,
+  RemoveCdcTableRequest,
   RollbackInputRequest,
   TestConnectionRequest,
 } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
 import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
 import { CreateCdcInputReq, CreateInputReq } from './dtos/input.model';
+import { toCdcTable } from './cdc-table.mapper';
 import { Metadata } from '@grpc/grpc-js';
 
 @Injectable()
@@ -192,16 +196,7 @@ export class InputsService {
         name: body.name,
         plugin: body.plugin,
         read_only: body.read_only ?? true,
-        tables: body.tables.map((t) => ({
-          table_schema: t.table_schema,
-          table_name: t.name,
-          name: t.name, // canonical identity == table_name (in-factory also backfills)
-          primary_keys: t.primary_keys ?? [],
-          iceberg_table_name: t.iceberg_table_name,
-          iceberg_qualify_table_name: t.iceberg_qualify_table_name,
-          columns: t.columns ?? [],
-          column_exclude_list: t.column_exclude_list ?? [],
-        })),
+        tables: body.tables.map(toCdcTable),
         destination: body.destination,
       },
       info,
@@ -320,19 +315,19 @@ export class InputsService {
     return formatedPayload;
   }
 
-  async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
+  async markTableDeleted(data: MarkTableDeletedRequest) {
     return lastValueFrom(this.inputWriteService.MarkTableDeleted(data));
   }
 
-  async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
-    return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data));
+  async unmarkTableDeleted(data: MarkTableDeletedRequest) {
+    return lastValueFrom(this.inputWriteService.UnmarkTableDeleted(data));
   }
 
-  async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) {
-    return lastValueFrom(this.inputWriteService.AddCdcTable(data as any));
+  async addCdcTable(data: AddCdcTableRequest) {
+    return lastValueFrom(this.inputWriteService.AddCdcTable(data));
   }
 
-  async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) {
-    return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data));
+  async removeCdcTable(data: RemoveCdcTableRequest) {
+    return lastValueFrom(this.inputWriteService.RemoveCdcTable(data));
   }
 }

src/modules/pipelinesV2/interfaces.ts

@@ -9,6 +9,33 @@ export class PipelineInputsDTO {
   }>
 }
 
+export class PipelineTableDestination {
+  @ApiPropertyOptional()
+  table_schema?: string;
+  @ApiPropertyOptional()
+  table_name?: string;
+}
+
+/** One entry of the create body's `config.tables`. Mirrors the frontend's
+ *  EntityWithColumnsNames; only the fields maestro/pi-factory read are typed,
+ *  the rest travels as-is. */
+export class PipelineTableConfig {
+  @ApiProperty()
+  name: string;
+  @ApiPropertyOptional({ type: () => PipelineTableDestination })
+  destinations?: Partial<Record<'raw' | 'qualify', PipelineTableDestination>>;
+  [extra: string]: unknown;
+}
+
+/** `{ cron, tables }` as sent by the frontend on create; JSON-serialized onto
+ *  the gRPC `config` string field. */
+export class PipelineConfig {
+  @ApiPropertyOptional()
+  cron?: string;
+  @ApiPropertyOptional({ type: [PipelineTableConfig] })
+  tables?: PipelineTableConfig[];
+}
+
 export class IPipelineV2 {
   @ApiProperty()
   id: string;
@@ -31,8 +58,8 @@ export class IPipelineV2 {
   tags?: string[];
   @ApiPropertyOptional()
   properties?: any;
-  @ApiPropertyOptional()
-  config?: any;
+  @ApiPropertyOptional({ type: () => PipelineConfig })
+  config?: PipelineConfig;
 
   @ApiProperty()
   connector_name: string;

src/modules/pipelinesV2/pipelines.service.ts

@@ -1,3 +1,4 @@
+import { isCdcPlugin } from 'src/utils/cdc';
 /* eslint-disable no-async-promise-executor */
 import {
   BadRequestException,
@@ -437,7 +438,7 @@ export class PipelinesService implements OnModuleInit {
     });
 
     this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
-    const isCdc = !!oldInput.plugin?.endsWith('_cdc');
+    const isCdc = isCdcPlugin(oldInput.plugin);
     const pipelineIdFormat = pipelineId.split('-').join('_');
     const rollback: RollbackPromise[] = [];
 

src/modules/platform-api/pipeline-tables.service.spec.ts

@@ -0,0 +1,243 @@
+// These service modules pull in gRPC client-config modules that read
+// process.env at load time; mock them (hoisted before imports) so the spec
+// needs no runtime env.
+jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
+jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));
+
+import { NotFoundException } from '@nestjs/common';
+import { PipelineTablesService } from './pipeline-tables.service';
+
+const logger = { info: jest.fn(), error: jest.fn() };
+const user = { customer_id: 'c1', customer_name: 'cust', user_id: 'u1' } as never;
+const info = { customer_id: 'c1', customer: 'cust', user_id: 'u1' };
+
+type Mocks = {
+  proxy: jest.Mock;
+  markTableDeleted: jest.Mock;
+  unmarkTableDeleted: jest.Mock;
+  addCdcTable: jest.Mock;
+  removeCdcTable: jest.Mock;
+  addCdcJobs: jest.Mock;
+};
+
+function build(): { service: PipelineTablesService; m: Mocks } {
+  const m: Mocks = {
+    proxy: jest.fn(),
+    markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }),
+    unmarkTableDeleted: jest.fn().mockResolvedValue({}),
+    addCdcTable: jest.fn().mockResolvedValue({ input: {} }),
+    removeCdcTable: jest.fn().mockResolvedValue({ input: {} }),
+    addCdcJobs: jest.fn(),
+  };
+  type Deps = ConstructorParameters<typeof PipelineTablesService>;
+  const service = new PipelineTablesService(
+    { proxy: m.proxy } as unknown as Deps[0],
+    {
+      markTableDeleted: m.markTableDeleted,
+      unmarkTableDeleted: m.unmarkTableDeleted,
+      addCdcTable: m.addCdcTable,
+      removeCdcTable: m.removeCdcTable,
+    } as unknown as Deps[1],
+    { addCdcJobs: m.addCdcJobs } as unknown as Deps[2],
+    { logger } as unknown as Deps[3],
+  );
+  return { service, m };
+}
+
+const pipelineWithJobs = (connector: string) => ({
+  jobs: [
+    { job_id: 'p_0', input: { connector, table_name: 'pedidos' } },
+    { job_id: 'p_1', input: { connector, table_name: 'clientes' } },
+    { job_id: 'p_2', input: { connector, table_name: 'produtos' } },
+  ],
+});
+
+const deleteCalls = (proxy: jest.Mock) =>
+  proxy.mock.calls.filter(([method]) => method === 'DELETE');
+
+describe('PipelineTablesService.removeTable', () => {
+  afterEach(() => jest.clearAllMocks());
+
+  it.each(['cdc', 'jdbc'])(
+    'removes a %s table through the single pipeline jobs route',
+    async (connector) => {
+      const { service, m } = build();
+      m.proxy.mockImplementation((method: string) =>
+        Promise.resolve(method === 'GET' ? pipelineWithJobs(connector) : {}),
+      );
+
+      const result = await service.removeTable('pi-d', 'iid', 'pedidos', user);
+
+      expect(m.markTableDeleted).toHaveBeenCalledWith({ input_id: 'iid', table_name: 'pedidos', info });
+      expect(m.proxy).toHaveBeenCalledWith('GET', '/pipeline/pi_d', user);
+      expect(deleteCalls(m.proxy)).toEqual([
+        ['DELETE', '/pipeline/pi_d/jobs', user, { job_ids: ['p_0'], delete_snowflake_tables: false }],
+      ]);
+      expect(result).toEqual({ name: 'pedidos', is_deleted: true, deleted_at: 't' });
+      expect(m.unmarkTableDeleted).not.toHaveBeenCalled();
+    },
+  );
+
+  it('rolls back the mark when the platform delete fails', async () => {
+    const { service, m } = build();
+    m.proxy.mockImplementation((method: string) =>
+      method === 'GET' ? Promise.resolve(pipelineWithJobs('cdc')) : Promise.reject(new Error('platform boom')),
+    );
+
+    await expect(service.removeTable('pid', 'iid', 'pedidos', user)).rejects.toThrow('platform boom');
+    expect(m.unmarkTableDeleted).toHaveBeenCalledWith({ input_id: 'iid', table_name: 'pedidos', info });
+  });
+
+  it('404s (and rolls back) when the table has no job', async () => {
+    const { service, m } = build();
+    m.proxy.mockResolvedValue(pipelineWithJobs('cdc'));
+
+    await expect(service.removeTable('pid', 'iid', 'ghost', user)).rejects.toBeInstanceOf(NotFoundException);
+    expect(deleteCalls(m.proxy)).toHaveLength(0);
+    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe('PipelineTablesService.removeTables', () => {
+  afterEach(() => jest.clearAllMocks());
+
+  it('removes N tables in ONE platform call (connectors reconfigured once)', async () => {
+    const { service, m } = build();
+    m.proxy.mockImplementation((method: string) =>
+      Promise.resolve(method === 'GET' ? pipelineWithJobs('cdc') : {}),
+    );
+
+    const result = await service.removeTables('pid', 'iid', ['pedidos', 'produtos'], user);
+
+    expect(m.markTableDeleted).toHaveBeenCalledTimes(2);
+    expect(deleteCalls(m.proxy)).toEqual([
+      ['DELETE', '/pipeline/pid/jobs', user, { job_ids: ['p_0', 'p_2'], delete_snowflake_tables: false }],
+    ]);
+    expect(result).toEqual({ table_names: ['pedidos', 'produtos'], deleted: true });
+  });
+
+  it("rolls back only this call's marks when the platform delete fails", async () => {
+    const { service, m } = build();
+    m.proxy.mockImplementation((method: string) =>
+      method === 'GET' ? Promise.resolve(pipelineWithJobs('cdc')) : Promise.reject(new Error('platform boom')),
+    );
+
+    await expect(service.removeTables('pid', 'iid', ['pedidos', 'clientes'], user)).rejects.toThrow();
+
+    const unmarked = m.unmarkTableDeleted.mock.calls.map(([arg]) => arg.table_name).sort();
+    expect(unmarked).toEqual(['clientes', 'pedidos']);
+  });
+
+  it('rolls back the marks made so far if a later mark fails (atomic)', async () => {
+    const { service, m } = build();
+    m.markTableDeleted
+      .mockResolvedValueOnce({ is_deleted: true, deleted_at: 't' })
+      .mockRejectedValueOnce(new Error('dynamo boom'));
+
+    await expect(service.removeTables('pid', 'iid', ['pedidos', 'clientes'], user)).rejects.toThrow('dynamo boom');
+
+    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(1);
+    expect(m.unmarkTableDeleted.mock.calls[0][0].table_name).toBe('pedidos');
+    expect(m.proxy).not.toHaveBeenCalled();
+  });
+
+  it('404s when a requested table has no matching job and rolls back the marks', async () => {
+    const { service, m } = build();
+    m.proxy.mockResolvedValue(pipelineWithJobs('cdc'));
+
+    await expect(service.removeTables('pid', 'iid', ['pedidos', 'ghost'], user)).rejects.toBeInstanceOf(NotFoundException);
+    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(2);
+  });
+
+  it('keeps going when a rollback step itself fails', async () => {
+    const { service, m } = build();
+    m.proxy.mockImplementation((method: string) =>
+      method === 'GET' ? Promise.resolve(pipelineWithJobs('cdc')) : Promise.reject(new Error('platform boom')),
+    );
+    m.unmarkTableDeleted.mockRejectedValueOnce(new Error('unmark boom'));
+
+    await expect(service.removeTables('pid', 'iid', ['pedidos', 'clientes'], user)).rejects.toThrow('platform boom');
+    expect(m.unmarkTableDeleted).toHaveBeenCalledTimes(2);
+  });
+});
+
+describe('PipelineTablesService.addTable', () => {
+  afterEach(() => jest.clearAllMocks());
+
+  const body = {
+    table_name: 'orders',
+    table_schema: 'public',
+    primary_keys: ['id'],
+    destinations: {
+      raw: { table_schema: 'raw', table_name: 'orders' },
+      qualify: { table_schema: 'qualify', table_name: 'orders' },
+    },
+  };
+
+  const storedTable = (extra: Record<string, unknown> = {}) => ({
+    table_schema: 'public',
+    table_name: 'orders',
+    name: 'orders',
+    primary_keys: ['id'],
+    iceberg_table_name: undefined,
+    iceberg_qualify_table_name: undefined,
+    columns: [],
+    column_exclude_list: [],
+    ...extra,
+  });
+
+  it('appends to DynamoDB, then dispatches the platform jobs', async () => {
+    const { service, m } = build();
+    m.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
+
+    const result = await service.addTable('pid', 'iid', body, user);
+
+    expect(m.addCdcTable).toHaveBeenCalledWith({ client_id: 'c1', id: 'iid', table: storedTable(), info });
+    expect(m.addCdcJobs).toHaveBeenCalledWith({
+      pipeline_id: 'pid',
+      input_id: 'iid',
+      tables: [{ table_schema: 'public', table_name: 'orders', primary_keys: ['id'], destinations: body.destinations }],
+      info,
+    });
+    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
+    expect(m.removeCdcTable).not.toHaveBeenCalled();
+    expect(m.addCdcTable.mock.invocationCallOrder[0]).toBeLessThan(m.addCdcJobs.mock.invocationCallOrder[0]);
+  });
+
+  it('carries the iceberg names and columns through to the stored CdcTable', async () => {
+    const { service, m } = build();
+    m.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] });
+    const columns = [
+      { name: 'id', type: 'int', is_primary_key: true },
+      { name: 'descr', type: 'varchar(255)', is_primary_key: false },
+    ];
+
+    await service.addTable('pid', 'iid', {
+      ...body,
+      iceberg_table_name: 'cdc_raw.public__orders',
+      iceberg_qualify_table_name: 'orders_dedup',
+      columns,
+      column_exclude_list: ['descr'],
+    }, user);
+
+    expect(m.addCdcTable).toHaveBeenCalledWith({
+      client_id: 'c1',
+      id: 'iid',
+      table: storedTable({
+        iceberg_table_name: 'cdc_raw.public__orders',
+        iceberg_qualify_table_name: 'orders_dedup',
+        columns,
+        column_exclude_list: ['descr'],
+      }),
+      info,
+    });
+  });
+
+  it('rolls back the DynamoDB row when AddJobs fails', async () => {
+    const { service, m } = build();
+    m.addCdcJobs.mockRejectedValue(new Error('grpc boom'));
+
+    await expect(service.addTable('pid', 'iid', body, user)).rejects.toThrow('grpc boom');
+    expect(m.removeCdcTable).toHaveBeenCalledWith({ client_id: 'c1', id: 'iid', table_name: 'orders', info });
+  });
+});

src/modules/platform-api/pipeline-tables.service.ts

@@ -0,0 +1,157 @@
+import { Inject, Injectable, NotFoundException } from '@nestjs/common';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
+
+import { RequestUser } from '../../decorators/user.decorator';
+import { InputsService } from '../inputs/inputs.service';
+import { toCdcTable } from '../inputs/cdc-table.mapper';
+import { PipelinesService } from '../pipelinesV2/pipelines.service';
+import { PlatformApiService } from './platform-api.service';
+import { AddCdcTableBody } from './platform-api.dto';
+
+interface PlatformJob {
+  job_id: string;
+  input?: { table_name?: string; connector?: string } | null;
+}
+
+interface PlatformPipeline {
+  jobs?: PlatformJob[] | null;
+}
+
+/**
+ * Add/remove tables of a pipeline: the DynamoDB input (soft-delete flags or
+ * a new CdcTable row) and the platform-api jobs move together, with the
+ * DynamoDB side rolled back when the platform side fails.
+ *
+ * Removal always goes through DELETE /pipeline/{id}/jobs — the one platform
+ * route that dispatches by pipeline type (Airflow refresh for batch, Kafka
+ * Connect reconfiguration for CDC). Landed destination data is kept
+ * (delete_snowflake_tables: false), matching the soft-delete in DynamoDB.
+ */
+@Injectable()
+export class PipelineTablesService {
+  logger: DadosferaLogger;
+
+  constructor(
+    private readonly platformApiService: PlatformApiService,
+    private readonly inputsService: InputsService,
+    private readonly pipelinesClientService: PipelinesService,
+    @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
+  ) {
+    this.logger = dadosferaLogger.logger;
+  }
+
+  async removeTable(pipelineId: string, inputId: string, tableName: string, user: RequestUser) {
+    const info = this.infoOf(user);
+
+    this.logger.info('removeTable: marking table as deleted', { inputId, tableName });
+    const marked = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info });
+
+    try {
+      const jobIds = await this.resolveJobIds(pipelineId, [tableName], user);
+      await this.removeJobs(pipelineId, jobIds, user);
+      return { name: tableName, is_deleted: marked.is_deleted ?? true, deleted_at: marked.deleted_at };
+    } catch (error) {
+      this.logger.error('removeTable: platform-api delete failed, rolling back the mark', { tableName, error: error.message });
+      await this.unmarkAll(inputId, [tableName], info);
+      throw error;
+    }
+  }
+
+  /** Removes N tables with ONE platform call, so CDC connectors are reconfigured once. */
+  async removeTables(pipelineId: string, inputId: string, tableNames: string[], user: RequestUser) {
+    const info = this.infoOf(user);
+
+    // Soft-delete each table, tracking which succeeded so a later failure
+    // only rolls back the marks made in THIS call.
+    const marked: string[] = [];
+    try {
+      for (const name of tableNames) {
+        await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info });
+        marked.push(name);
+      }
+
+      const jobIds = await this.resolveJobIds(pipelineId, tableNames, user);
+      await this.removeJobs(pipelineId, jobIds, user);
+      return { table_names: tableNames, deleted: true };
+    } catch (error) {
+      this.logger.error("removeTables: failed, rolling back this call's marks", { tableNames, error: error.message });
+      await this.unmarkAll(inputId, marked, info);
+      throw error;
+    }
+  }
+
+  /** Appends a CDC table to the DynamoDB input, then dispatches its platform jobs. */
+  async addTable(pipelineId: string, inputId: string, body: AddCdcTableBody, user: RequestUser) {
+    const info = this.infoOf(user);
+    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 });
+
+    try {
+      return 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,
+      });
+    } 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 (rollbackError) {
+        this.logger.error('addTable: rollback failed', { error: rollbackError.message });
+      }
+      throw error;
+    }
+  }
+
+  private infoOf(user: RequestUser): Info {
+    return { customer_id: user.customer_id, customer: user.customer_name, user_id: user.user_id };
+  }
+
+  /** Platform-API replaces '-' with '_' in pipeline IDs. */
+  private normalizePipelineId(id: string): string {
+    return id.replace(/-/g, '_');
+  }
+
+  private async resolveJobIds(pipelineId: string, tableNames: string[], user: RequestUser): Promise<string[]> {
+    const normalizedPipelineId = this.normalizePipelineId(pipelineId);
+    const pipeline: PlatformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
+    const jobs = pipeline?.jobs ?? [];
+
+    return tableNames.map((name) => {
+      const job = jobs.find((j) => j.input?.table_name === name);
+      if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`);
+      return job.job_id;
+    });
+  }
+
+  private async removeJobs(pipelineId: string, jobIds: string[], user: RequestUser) {
+    const normalizedPipelineId = this.normalizePipelineId(pipelineId);
+    this.logger.info('removeJobs: deleting jobs from platform-api', { jobIds });
+    await this.platformApiService.proxy(
+      'DELETE',
+      `/pipeline/${normalizedPipelineId}/jobs`,
+      user,
+      { job_ids: jobIds, delete_snowflake_tables: false },
+    );
+    this.logger.info('removeJobs: jobs deleted', { jobIds });
+  }
+
+  private async unmarkAll(inputId: string, tableNames: string[], info: Info) {
+    for (const name of tableNames) {
+      try {
+        await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info });
+      } catch (rollbackError) {
+        this.logger.error('rollback failed: table stays marked as deleted', { tableName: name, error: rollbackError.message });
+      }
+    }
+  }
+}

src/modules/platform-api/platform-api.controller.spec.ts

@@ -5,53 +5,43 @@
 jest.mock('../customers/customers.service', () => ({ CustomersService: class {} }));
 jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} }));
 jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
-jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));
+jest.mock('./pipeline-tables.service', () => ({ PipelineTablesService: class {} }));
 
+import { BadRequestException } from '@nestjs/common';
 import { Test, TestingModule } from '@nestjs/testing';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 
 import { PlatformApiController } from './platform-api.controller';
 import { PlatformApiService } from './platform-api.service';
+import { PipelineTablesService } from './pipeline-tables.service';
 import { ElasticsearchService } from '../../services/elasticsearch';
 import { DynamoDBService } from '../../services/dynamodb';
 import { CustomersService } from '../customers/customers.service';
 import { CatalogService } from '../catalog/catalog.service';
 import { InputsService } from '../inputs/inputs.service';
-import { PipelinesService } from '../pipelinesV2/pipelines.service';
 
-const logger = {
-  info: (...args) => args,
-  error: (...args) => args,
-};
+const logger = { info: jest.fn(), error: jest.fn() };
+const mockUser = { customer_id: 'c1', customer_name: 'cust', user_id: 'u1' } as never;
 
-const mockUser: any = {
-  customer_id: 'c1',
-  customer_name: 'cust',
-  user_id: 'u1',
-};
-
-describe('PlatformApiController - deleteTable', () => {
+// The table routes are thin: validate the body, delegate to
+// PipelineTablesService (covered in pipeline-tables.service.spec.ts).
+describe('PlatformApiController - table routes', () => {
   let controller: PlatformApiController;
-  let platformApiService: { proxy: jest.Mock };
-  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };
+  let tables: { removeTable: jest.Mock; removeTables: jest.Mock; addTable: jest.Mock };
 
   beforeEach(async () => {
-    platformApiService = { proxy: jest.fn() };
-    inputsService = {
-      markTableDeleted: jest.fn(),
-      unmarkTableDeleted: jest.fn(),
-    };
+    tables = { removeTable: jest.fn(), removeTables: jest.fn(), addTable: jest.fn() };
 
     const module: TestingModule = await Test.createTestingModule({
       controllers: [PlatformApiController],
       providers: [
-        { provide: PlatformApiService, useValue: platformApiService },
+        { provide: PlatformApiService, useValue: {} },
         { provide: ElasticsearchService, useValue: {} },
         { provide: DynamoDBService, useValue: {} },
         { provide: CustomersService, useValue: {} },
         { provide: CatalogService, useValue: {} },
-        { provide: InputsService, useValue: inputsService },
-        { provide: PipelinesService, useValue: {} },
+        { provide: InputsService, useValue: {} },
+        { provide: PipelineTablesService, useValue: tables },
         { provide: DadosferaLogger, useValue: { logger } },
       ],
     }).compile();
@@ -59,314 +49,37 @@ describe('PlatformApiController - deleteTable', () => {
     controller = module.get<PlatformApiController>(PlatformApiController);
   });
 
-  it('CDC removal reconfigures the connector', async () => {
-    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
-    platformApiService.proxy.mockImplementation((method: string, path: string) => {
-      if (method === 'GET') {
-        return Promise.resolve({
-          jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } }],
-        });
-      }
-      return Promise.resolve({});
-    });
-
-    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);
-
-    expect(platformApiService.proxy).toHaveBeenCalledWith(
-      'DELETE',
-      '/pipeline/pid/jobs',
-      mockUser,
-      { job_ids: ['p_0'], delete_snowflake_tables: false },
-    );
-    const deletedViaJobsRoute = platformApiService.proxy.mock.calls.some(
-      ([method, path]: any[]) => method === 'DELETE' && path === '/jobs/p_0',
-    );
-    expect(deletedViaJobsRoute).toBe(false);
+  it('deleteTable delegates', async () => {
+    tables.removeTable.mockResolvedValue({ name: 'pedidos', is_deleted: true });
+    await expect(controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser))
+      .resolves.toEqual({ name: 'pedidos', is_deleted: true });
+    expect(tables.removeTable).toHaveBeenCalledWith('pid', 'iid', 'pedidos', mockUser);
   });
 
-  it('batch removal unchanged', async () => {
-    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
-    platformApiService.proxy.mockImplementation((method: string, path: string) => {
-      if (method === 'GET') {
-        return Promise.resolve({
-          jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 'pedidos' } }],
-        });
-      }
-      return Promise.resolve({});
-    });
-
-    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);
-
-    expect(platformApiService.proxy).toHaveBeenCalledWith(
-      'DELETE',
-      '/jobs/p_0',
-      mockUser,
-    );
-    const reconfiguredConnector = platformApiService.proxy.mock.calls.some(
-      ([method, path]: any[]) => method === 'DELETE' && path === '/pipeline/pid/jobs',
-    );
-    expect(reconfiguredConnector).toBe(false);
-  });
-});
-
-describe('PlatformApiController - addTable', () => {
-  let controller: PlatformApiController;
-  let inputsService: { addCdcTable: jest.Mock; removeCdcTable: jest.Mock };
-  let pipelinesClientService: { addCdcJobs: jest.Mock };
-
-  const body = {
-    table_name: 'orders',
-    table_schema: 'public',
-    primary_keys: ['id'],
-    destinations: {
-      raw: { table_schema: 'raw', table_name: 'orders' },
-      qualify: { table_schema: 'qualify', table_name: 'orders' },
-    },
-  };
-
-  beforeEach(async () => {
-    inputsService = {
-      addCdcTable: jest.fn(),
-      removeCdcTable: jest.fn(),
-    };
-    pipelinesClientService = {
-      addCdcJobs: jest.fn(),
-    };
-
-    const module: TestingModule = await Test.createTestingModule({
-      controllers: [PlatformApiController],
-      providers: [
-        { provide: PlatformApiService, useValue: { proxy: jest.fn() } },
-        { provide: ElasticsearchService, useValue: {} },
-        { provide: DynamoDBService, useValue: {} },
-        { provide: CustomersService, useValue: {} },
-        { provide: CatalogService, useValue: {} },
-        { provide: InputsService, useValue: inputsService },
-        { provide: PipelinesService, useValue: pipelinesClientService },
-        { provide: DadosferaLogger, useValue: { logger } },
-      ],
-    }).compile();
-
-    controller = module.get<PlatformApiController>(PlatformApiController);
+  it('deleteTables delegates', async () => {
+    tables.removeTables.mockResolvedValue({ deleted: true });
+    await controller.deleteTables('pid', 'iid', { table_names: ['a', 'b'] }, mockUser);
+    expect(tables.removeTables).toHaveBeenCalledWith('pid', 'iid', ['a', 'b'], mockUser);
   });
 
-  it('addTable: DynamoDB append then platform AddJobs, returns job_ids', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
-
-    const result = await controller.addTable('pid', 'iid', body, mockUser);
-
-    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
-      table: {
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        name: 'orders',
-        iceberg_table_name: undefined,
-        iceberg_qualify_table_name: undefined,
-        columns: [],
-        column_exclude_list: [],
-      },
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-    expect(pipelinesClientService.addCdcJobs).toHaveBeenCalledWith({
-      pipeline_id: 'pid',
-      input_id: 'iid',
-      tables: [{
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        destinations: body.destinations,
-      }],
-      info: { customer_id: 'c1', user_id: 'u1', customer: 'cust' },
-    });
-    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
-    expect(inputsService.removeCdcTable).not.toHaveBeenCalled();
-
-    const addCdcTableOrder = inputsService.addCdcTable.mock.invocationCallOrder[0];
-    const addCdcJobsOrder = pipelinesClientService.addCdcJobs.mock.invocationCallOrder[0];
-    expect(addCdcTableOrder).toBeLessThan(addCdcJobsOrder);
+  it('deleteTables rejects an empty table_names before touching anything', async () => {
+    await expect(controller.deleteTables('pid', 'iid', { table_names: [] }, mockUser))
+      .rejects.toBeInstanceOf(BadRequestException);
+    expect(tables.removeTables).not.toHaveBeenCalled();
   });
 
-  it('addTable: carries iceberg_table_name on the added table through to AddCdcTable', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] });
-
-    const icebergBody = { ...body, iceberg_table_name: 'cdc_raw.public__orders' };
-
-    await controller.addTable('pid', 'iid', icebergBody, mockUser);
-
-    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
-      table: {
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        name: 'orders',
-        iceberg_table_name: 'cdc_raw.public__orders',
-        iceberg_qualify_table_name: undefined,
-        columns: [],
-        column_exclude_list: [],
-      },
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-  });
-
-  it('addTable: carries columns on the added table through to AddCdcTable', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_4'], skipped: [] });
-
-    const columnsBody = {
-      ...body,
-      columns: [
-        { name: 'id', type: 'int', is_primary_key: true },
-        { name: 'descr', type: 'varchar(255)', is_primary_key: false },
-      ],
-    };
-
-    await controller.addTable('pid', 'iid', columnsBody, mockUser);
-
-    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
-      table: {
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        name: 'orders',
-        iceberg_table_name: undefined,
-        iceberg_qualify_table_name: undefined,
-        columns: [
-          { name: 'id', type: 'int', is_primary_key: true },
-          { name: 'descr', type: 'varchar(255)', is_primary_key: false },
-        ],
-        column_exclude_list: [],
-      },
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-  });
-
-  it('addTable: rolls back the DynamoDB row when AddJobs fails', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockRejectedValue(new Error('platform down'));
-    inputsService.removeCdcTable.mockResolvedValue({});
-
-    await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down');
-
-    expect(inputsService.removeCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
+  it('addTable delegates', async () => {
+    const body = {
       table_name: 'orders',
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-  });
-});
-
-
-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({}),
+      table_schema: 'public',
+      primary_keys: ['id'],
+      destinations: {
+        raw: { table_schema: 'raw', table_name: 'orders' },
+        qualify: { table_schema: 'qualify', table_name: 'orders' },
+      },
     };
-
-    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();
+    tables.addTable.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
+    await expect(controller.addTable('pid', 'iid', body, mockUser)).resolves.toEqual({ job_ids: ['p_2'], skipped: [] });
+    expect(tables.addTable).toHaveBeenCalledWith('pid', 'iid', body, mockUser);
   });
 });

src/modules/platform-api/platform-api.controller.ts

@@ -11,7 +11,6 @@ import {
   Inject,
   BadRequestException,
   HttpException,
-  NotFoundException,
   UseGuards,
 } from '@nestjs/common';
 import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger';
@@ -31,9 +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 { PipelinesService } from '../pipelinesV2/pipelines.service';
+import { PipelineTablesService } from './pipeline-tables.service';
 import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
 
 
@@ -62,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;
@@ -988,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')
@@ -1051,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')
@@ -1120,77 +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 = {
-      table_schema: body.table_schema,
-      table_name: body.table_name,
-      primary_keys: body.primary_keys,
-      name: body.table_name,
-      iceberg_table_name: body.iceberg_table_name,
-      iceberg_qualify_table_name: body.iceberg_qualify_table_name,
-      columns: body.columns ?? [],
-      column_exclude_list: body.column_exclude_list ?? [],
-    };
-
-    this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name });
-    await this.inputsService.addCdcTable({ 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({ 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 ====================

src/modules/platform-api/platform-api.dto.ts

@@ -1,9 +1,66 @@
-import { ApiProperty } from "@nestjs/swagger";
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
 
 export class ValidationTableDTO {
-    @ApiProperty()
-    tables: Array<{
-        table_name: string;
-        table_schema: string;
-    }>
-}
+  @ApiProperty()
+  tables: Array<{
+    table_name: string;
+    table_schema: string;
+  }>;
+}
+
+export class DeleteTableBody {
+  @ApiProperty()
+  table_name: string;
+}
+
+export class DeleteTablesBody {
+  @ApiProperty({ type: [String] })
+  table_names: string[];
+}
+
+export class CdcTableDestination {
+  @ApiProperty()
+  table_schema: string;
+  @ApiProperty()
+  table_name: string;
+}
+
+export class CdcTableDestinations {
+  @ApiProperty({ type: CdcTableDestination })
+  raw: CdcTableDestination;
+  @ApiProperty({ type: CdcTableDestination })
+  qualify: CdcTableDestination;
+}
+
+export class CdcColumnBody {
+  @ApiProperty()
+  name: string;
+  @ApiProperty()
+  type: string;
+  @ApiProperty()
+  is_primary_key: boolean;
+}
+
+export class AddCdcTableBody {
+  @ApiProperty()
+  table_name: string;
+  @ApiProperty()
+  table_schema: string;
+  @ApiProperty({ type: [String] })
+  primary_keys: string[];
+  @ApiProperty({ type: CdcTableDestinations })
+  destinations: CdcTableDestinations;
+  // Iceberg destination only (protospack CdcTable.iceberg_table_name);
+  // absent for snowflake, back-compat.
+  @ApiPropertyOptional()
+  iceberg_table_name?: string;
+  // Per-table deduped (qualify) Iceberg table name; absent => same as raw.
+  @ApiPropertyOptional()
+  iceberg_qualify_table_name?: string;
+  // Source column schema for iceberg deduped table pre-create.
+  @ApiPropertyOptional({ type: [CdcColumnBody] })
+  columns?: CdcColumnBody[];
+  // Columns the user chose to ignore -> Debezium column.exclude.list.
+  @ApiPropertyOptional({ type: [String] })
+  column_exclude_list?: string[];
+}

src/modules/platform-api/platform-api.module.ts

@@ -4,6 +4,7 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 
 import { PlatformApiController } from './platform-api.controller';
 import { PlatformApiService } from './platform-api.service';
+import { PipelineTablesService } from './pipeline-tables.service';
 import { ElasticsearchModule } from '../../services/elasticsearch';
 import { DynamoDBModule } from '../../services/dynamodb';
 import { CustomersModule } from '../customers/customers.module';
@@ -21,7 +22,7 @@ import { PipelinesV2Module } from '../pipelinesV2/pipelines.module';
     forwardRef(() => PipelinesV2Module),
   ],
   controllers: [PlatformApiController],
-  providers: [PlatformApiService, DadosferaLogger],
+  providers: [PlatformApiService, PipelineTablesService, DadosferaLogger],
   exports: [PlatformApiService],
 })
 export class PlatformApiModule {}

src/utils/cdc.ts

@@ -0,0 +1,34 @@
+/**
+ * Single source of truth for "is this CDC?" across maestro.
+ *
+ * Two shapes carry the discriminator:
+ *  - a connection/input `plugin` (`mysql_cdc`, ...), used before a platform
+ *    pipeline exists (create flows, DynamoDB inputs);
+ *  - a platform job `input.connector === 'cdc'`, the explicit type the
+ *    platform-api stamps on every CDC job.
+ *
+ * Keep the plugin list in sync with pi-factory (`CDC_PLUGINS`) and in-factory
+ * (`cdcPlugins`).
+ */
+export const CDC_PLUGINS: readonly string[] = ['mysql_cdc', 'postgresql_cdc', 'oracle_cdc'];
+
+export function isCdcPlugin(plugin?: string | null): boolean {
+  return plugin != null && CDC_PLUGINS.includes(plugin);
+}
+
+export interface PlatformJobLike {
+  input?: { connector?: string } | null;
+}
+
+export interface PlatformPipelineLike {
+  jobs?: PlatformJobLike[] | null;
+}
+
+export function isCdcJob(job?: PlatformJobLike | null): boolean {
+  return job?.input?.connector === 'cdc';
+}
+
+/** A pipeline is CDC when any of its platform jobs is a CDC job. */
+export function isCdcPipeline(pipeline?: PlatformPipelineLike | null): boolean {
+  return (pipeline?.jobs ?? []).some((job) => isCdcJob(job));
+}