Merge pull request #518 from dadosfera/feature/cdc-connector

FIX: CDC live status decodes int64 offsets as numbers

.github/workflows/test.yml

@@ -2,9 +2,21 @@ name: Test
 on:
   pull_request:
     branches:
+      - beta
       - main
 
 jobs:
+  # Blocks a local (file:/tarball/overlay) protospack-v2 dependency from
+  # reaching staging (beta) or prod (main).
+  protospack-dep-guard:
+    if: github.base_ref == 'beta' || github.base_ref == 'main'
+    runs-on: [self-hosted, prd]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Check protospack-v2 is consumed from the registry
+        run: node scripts/check-protospack-dep.js
+
   test:
     runs-on: [self-hosted, prd]
     env:

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.

Dockerfile

@@ -29,6 +29,7 @@ COPY . .
 # unit test specific build
 FROM ci_image AS test
 ENV DUC_URL=0.0.0.0:50051
+ENV INFACTORY_URL=0.0.0.0:50052
 ENTRYPOINT ["npm", "run", "test"]
 
 

package-lock.json

@@ -16,7 +16,7 @@
         "@aws-sdk/lib-dynamodb": "^3.414.0",
         "@aws-sdk/signature-v4": "^3.370.0",
         "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
-        "@dadosfera/protospack-v2": "^3.40.0-beta.14",
+        "@dadosfera/protospack-v2": "3.40.0-beta.21",
         "@grpc/grpc-js": "^1.9.3",
         "@grpc/proto-loader": "^0.7.9",
         "@nestjs/cli": "^9.5.0",
@@ -1735,9 +1735,10 @@
       }
     },
     "node_modules/@dadosfera/protospack-v2": {
-      "version": "3.40.0-beta.14",
-      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.14.tgz",
-      "integrity": "sha512-pv3pxq0x1XcBgf3ajD6QOFRLOduh8iEozKFA3AKlIW4gid+gT4iL0GcU2M+O7h0QFeO4JIzRZe/nEMN82nqk7A==",
+      "version": "3.40.0-beta.21",
+      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.21.tgz",
+      "integrity": "sha512-1mfFCXc8BFiYHcmFJSXt2KLqB3mQ1oPKQ/X39LCoB8buQHewvH1R7I7ZB67nrBxQ6XiTq6ojPT6PQRRgJtFgBg==",
+      "license": "ISC",
       "dependencies": {
         "@grpc/grpc-js": "^1.9.3",
         "rxjs": "^7.5.5"

package.json

@@ -34,7 +34,7 @@
     "@aws-sdk/lib-dynamodb": "^3.414.0",
     "@aws-sdk/signature-v4": "^3.370.0",
     "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
-    "@dadosfera/protospack-v2": "^3.40.0-beta.14",
+    "@dadosfera/protospack-v2": "3.40.0-beta.21",
     "@grpc/grpc-js": "^1.9.3",
     "@grpc/proto-loader": "^0.7.9",
     "@nestjs/cli": "^9.5.0",

scripts/check-protospack-dep.js

@@ -0,0 +1,76 @@
+#!/usr/bin/env node
+/*
+ * CI guard: fail if @dadosfera/protospack-v2 is consumed from a LOCAL ref
+ * (file:/link:/git/relative path/bare tarball) instead of the CodeArtifact
+ * registry.
+ *
+ * Only local consumption is blocked. Versions published to CodeArtifact —
+ * including alpha/beta/rc prereleases produced by the alpha/beta branches —
+ * are fine; those resolve to a registry URL in the lockfile. The thing that
+ * must NOT reach beta (staging) or main (prod) is a dependency wired to a
+ * local `npm pack` tarball / overlay. Runs in the PR test workflow for PRs
+ * targeting beta/main and exits non-zero on any local ref.
+ */
+const fs = require('fs');
+const path = require('path');
+
+const PKG = '@dadosfera/protospack-v2';
+const root = path.resolve(__dirname, '..');
+const pkg = JSON.parse(fs.readFileSync(path.join(root, 'package.json'), 'utf8'));
+
+const problems = [];
+
+// A dependency SPEC is local if it's a filesystem path, symlink, git ref, or a
+// bare tarball path. A plain semver (incl. prereleases like 3.35.0-beta.1)
+// resolves from the registry and is allowed.
+function isLocalSpec(spec) {
+  return /^(file:|link:|git[:+]|\.\.?\/|\/|~\/)/.test(spec) || spec.endsWith('.tgz');
+}
+
+const spec =
+  (pkg.dependencies && pkg.dependencies[PKG]) ||
+  (pkg.devDependencies && pkg.devDependencies[PKG]);
+
+if (!spec) {
+  problems.push(`${PKG} is not listed as a dependency at all.`);
+} else if (isLocalSpec(spec)) {
+  problems.push(`${PKG} points at a local path/tarball/git ref: "${spec}".`);
+}
+
+// Also catch a lockfile resolved to a LOCAL ref even if package.json looks
+// clean. A registry URL (https://.../-/*.tgz) is the normal published
+// resolution and is fine — only file: refs and bare local tarball paths
+// (no http host) are blocked. Prerelease VERSIONS are not flagged: an
+// alpha/beta/rc published to CodeArtifact resolves to a registry URL.
+const lockPath = path.join(root, 'package-lock.json');
+if (fs.existsSync(lockPath)) {
+  const lock = JSON.parse(fs.readFileSync(lockPath, 'utf8'));
+  const nodes = { ...(lock.packages || {}), ...(lock.dependencies || {}) };
+  for (const [name, node] of Object.entries(nodes)) {
+    if (!name.includes('protospack-v2') || !node) continue;
+    const resolved = node.resolved || '';
+    const isLocal =
+      resolved.startsWith('file:') ||
+      (resolved.endsWith('.tgz') && !/^https?:\/\//.test(resolved));
+    if (isLocal) {
+      problems.push(
+        `package-lock.json resolves ${PKG} to a local ref: "${resolved}".`,
+      );
+    }
+  }
+}
+
+if (problems.length) {
+  console.error('✗ protospack-v2 dependency guard FAILED:');
+  for (const p of problems) console.error('  - ' + p);
+  console.error(
+    '\nMerging to beta/main requires ' +
+      PKG +
+      ' to come from CodeArtifact, not a local tarball/overlay. Publish ' +
+      'protospack-v2 (a beta prerelease is fine for the beta branch) and ' +
+      'repoint this dependency before merging.',
+  );
+  process.exit(1);
+}
+
+console.log(`✓ ${PKG} is consumed from the registry: "${spec}"`);

src/authentication/permissions.enum.ts

@@ -722,6 +722,8 @@ export const DADOSFERA_MODULES_KEYS = {
   PII: 'pii',
   EMBED: 'embedded-analytics',
   EMBED_ASSIGNED: 'embed-assigned',
+  CATALOG: 'catalog',
+  COLLECT: 'collect',
 }
 
 export const DADOSFERA_MODULES: Array<DadosferaModule> = [

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

@@ -0,0 +1,122 @@
+import { BadRequestException } from '@nestjs/common';
+import { PipelineExecutionGuard } from './pipeline-execution.guard';
+
+const logger = { info: jest.fn(), error: jest.fn() };
+
+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') {
+  return {
+    switchToHttp: () => ({
+      getRequest: () => ({ params: { pipelineId }, user: {} }),
+    }),
+  } 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());
+
+  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);
+    });
+  });
+
+  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/,
+      );
+    });
+  });
+
+  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('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,27 +40,51 @@ 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`,
         user,
       );
 
-      const currentStatus = status[status.length - 1]
+      const currentStatus = status?.[status.length - 1];
 
       this.logger.info('Pipeline current status response:' +  JSON.stringify(currentStatus));
-  
+
+      // Batch pipeline that never ran yet: nothing can be executing.
+      if (!currentStatus?.last_status) {
+        return true;
+      }
+
       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).
+      if (error instanceof BadRequestException) {
+        throw error;
+      }
       this.logger.error('Error in PipelineExecutionGuard: ' + error.message);
       throw new BadRequestException('Error checking pipeline status: ' + error.message);
     }
-
   }
 }

src/modules/auth/auth.controller.ts

@@ -37,6 +37,7 @@ import {
   RequireAllPermissions,
 } from 'src/decorators/authentication.decorator';
 import { AuthClientService } from './auth.service';
+import { UserDTO } from './dtos/login';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { GrpcToHttpExceptionFilter } from '../../error/grpc-to-http-exception.filter';
 import { RequestUser, User } from 'src/decorators/user.decorator';
@@ -486,7 +487,7 @@ export class AuthController {
       this.logger.info('Authenticating via X-Api-key header');
       const { api_key } = await this.apiKeyService.get(apiKey);
 
-      const userDto = {
+      const userDto: UserDTO = {
         id: api_key.user_id,
         name: api_key.username,
         email: api_key.username,
@@ -494,7 +495,8 @@ export class AuthController {
           id: api_key.customer_id,
           name: api_key.customer_name,
           tier: api_key.customer_tier,
-        }
+        },
+        permissions: [],
       };
 
       return res.status(200).json(userDto);

src/modules/auth/auth.service.ts

@@ -447,6 +447,9 @@ export class AuthClientService implements OnModuleInit {
         name: payload.customer_name,
         tier: payload.customer_tier,
       },
+      // Raw permission seqids from the JWT. Consumers own the seqid->meaning
+      // mapping (e.g. Orchest's auth-server); Maestro reports them as-is.
+      permissions: payload.permissions ?? [],
     };
 
     return userDto;

src/modules/auth/dtos/login.ts

@@ -152,5 +152,6 @@ export type UserDTO = {
     id: string,
     name: string,
     tier: string,
-  }
+  },
+  permissions: number[],
 }

src/modules/catalog/catalog.controller.ts

@@ -9,6 +9,7 @@ import {
   Inject,
   NotFoundException,
   Param,
+  Patch,
   Post,
   Put,
   Query,
@@ -51,6 +52,7 @@ import {
   IUpdateDataRequest,
   TriggerCatalogReq,
   TriggerCatalogRes,
+  UpdateColumnsMetadataRequest,
 } from './dtos';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
 import { Language } from 'src/decorators/language.decorator';
@@ -85,6 +87,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async searchCatalog(
     @User() user: RequestUser,
     @Query() query: ICatalogAllRequest,
@@ -124,6 +129,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async dowloadAsserts(
     @User() user: RequestUser,
     @Query() query: ICatalogAllRequest,
@@ -167,6 +175,9 @@ export class CatalogController {
 
   @ApiInternalOnlyEndpoint()
   @Get('data-asset')
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async findByPipelineAndObject(@User() user: RequestUser, @Query() query) {
     const { username, user_id, customer_id, customer_name, permissions } = user;
     const { pipeline, object } = query;
@@ -225,6 +236,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async findAllTags(@Body() body) {
     this.logger.info(`/catalog - ON FIND ALL TAGS ROUTE`, {
       user: body.info.user_id,
@@ -293,6 +307,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async getDataAsset(
     @User() user: RequestUser,
     @Param('id') id: string,
@@ -404,6 +421,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async getDataAssetColumnsMetadata(
     @User() user: RequestUser,
     @Language() language: LanguageEnum,
@@ -430,11 +450,50 @@ export class CatalogController {
     return { columns_metadata };
   }
 
+  @Patch('data-asset/:id/columns-metadata')
+  @RequireSomePermission(
+    PERMISSIONS_GROUPS.CATALOG.permissions.UPDATE,
+    PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
+  )
+  async updateColumnsMetadata(
+    @User() user: RequestUser,
+    @Language() language: LanguageEnum,
+    @Param('id') id: string,
+    @Body(new ValidationPipe()) body: UpdateColumnsMetadataRequest,
+  ): Promise<{ success: boolean }> {
+    const { customer_name, customer_id, user_id, username } = user;
+
+    this.logger.info(`/catalog - update columns metadata`, {
+      user_id,
+      customer_name,
+      columns_count: body.columns.length,
+    });
+
+    const metadata = PackTheMetadata({
+      customer_name,
+      customer_id,
+      user_id,
+      username,
+      language,
+    });
+
+    await this.catalogService.updateColumnsDescriptions(
+      id,
+      body.columns,
+      metadata,
+    );
+
+    return { success: true };
+  }
+
   @Get('data-asset/:id/preview')
   @RequireSomePermission(
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async getDataAssetPreview(
     @User() user: RequestUser,
     @Language() language: LanguageEnum,
@@ -466,6 +525,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.GET,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async getDataAssetDocs(
     @User() user: RequestUser,
     @Language() language: LanguageEnum,
@@ -497,6 +559,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.UPDATE,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async updateDataAsset(
     @User() user: RequestUser,
     @Language() language: LanguageEnum,
@@ -532,6 +597,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.CERTIFY,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async updateDataAssetCertificationStatus(
     @User() user: RequestUser,
     @Language() language: LanguageEnum,
@@ -559,6 +627,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.UPDATE,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async manageDataAssetDocs(
     @User() user: RequestUser,
     @Headers() headers,
@@ -596,6 +667,9 @@ export class CatalogController {
 
   @ApiInternalOnlyEndpoint()
   @Put('data-asset/:id/manage-permissions')
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async manageDataAssetPermissions(
     @Param('id') id: string,
     @User() user: RequestUser,
@@ -618,6 +692,9 @@ export class CatalogController {
 
   @ApiInternalOnlyEndpoint()
   @Put('data-asset/:id/revoke-permissions')
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async revokeDataAssetPermissions(
     @Param('id') id: string,
     @User() user: RequestUser,
@@ -643,6 +720,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.CREATE,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async createDataAsset(
     @User() user: RequestUser,
     @Body() body: ICreateDataAsset,
@@ -667,6 +747,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.UPDATE,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async commentOnDataAsset(
     @Param('id') id: string,
     @User() user: RequestUser,
@@ -693,6 +776,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.DELETE,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async deleteDataAsset(@Param('id') id: string, @User() user: RequestUser) {
     const { customer_id, customer_name, user_id, username } = user;
     const metadata = PackTheMetadata({
@@ -714,6 +800,9 @@ export class CatalogController {
     PERMISSIONS_GROUPS.CATALOG.permissions.UPDATE,
     PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async deleteComment(
     @Param('id') id: string,
     @User() user: RequestUser,
@@ -867,6 +956,9 @@ export class CatalogController {
 
   @Get('nimbus-dashboards')
   @RequireAllPermissions(PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER)
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
   async getNimbusDashboards(
     @User() user: RequestUser,
     @Body() body: GetNimbusDashboardsRequest,

src/modules/catalog/catalog.service.ts

@@ -467,6 +467,16 @@ class CatalogService implements OnModuleInit {
     return result;
   }
 
+  async updateColumnsDescriptions(
+    id: string,
+    columns: { column_name: string; description: string }[],
+    metadata: Metadata,
+  ) {
+    await lastValueFrom(
+      this.catalogWriteService.UpdateColumnDescriptions({ id, columns }, metadata),
+    );
+  }
+
   async createDataDocs(body: CreateDataDocsDTO, metadata: Metadata) {
     if (body.asset_type === 'table' || body.asset_type === 'view') {
       return this.createDataDocsViaNimbus(body);
@@ -832,6 +842,8 @@ class CatalogService implements OnModuleInit {
               data_asset_id: table_metadata_id.toString(),
               customer_name: customer_name,
               data_asset_type: 'dataset',
+              column_metadata: [],
+              data_preview: '',
             },
           ],
         },

src/modules/catalog/dtos/index.ts

@@ -1,10 +1,13 @@
 import { ApiProperty, ApiPropertyOptional, PickType } from '@nestjs/swagger';
 import {
+  ArrayNotEmpty,
+  IsArray,
   IsEnum,
   IsNotEmpty,
-  IsOptional,
   IsString,
+  ValidateNested,
 } from 'class-validator';
+import { Type } from 'class-transformer';
 import { CreateDataAssetRequest } from '@dadosfera/protospack-v2/dist/lib/Catalog/interfaces/messages';
 
 export enum DataAssetShareType {
@@ -247,6 +250,26 @@ export class IUpdateCertificationStatusRequest {
   certification_status: CertificationStatus;
 }
 
+export class ColumnDescriptionDto {
+  @ApiProperty()
+  @IsString()
+  @IsNotEmpty()
+  column_name: string;
+
+  @ApiProperty()
+  @IsString()
+  description: string;
+}
+
+export class UpdateColumnsMetadataRequest {
+  @ApiProperty({ type: [ColumnDescriptionDto] })
+  @IsArray()
+  @ArrayNotEmpty()
+  @ValidateNested({ each: true })
+  @Type(() => ColumnDescriptionDto)
+  columns: ColumnDescriptionDto[];
+}
+
 export class ICreateDataAsset implements CreateDataAssetRequest {
   @ApiProperty()
   display_name: string;

src/modules/connection-test/connection-test.controller.ts

@@ -22,20 +22,26 @@ import {
   ConnectionTestListTablesRes,
   GetTableMetadataRes,
   GetTableMetadataReq,
+  ValidateCdcPrerequisitesReq,
+  ValidateCdcPrerequisitesRes,
   RefreshCatalogReq,
   RefreshCatalogRes,
   RefreshCatalogStatusReq,
 } from './dto/connection-test';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
-import { Authenticated } from 'src/decorators/authentication.decorator';
+import { Authenticated, RequireModule } from 'src/decorators/authentication.decorator';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
 import { ApiInternalOnlyController } from 'src/decorators/swagger.decorator';
+import { DADOSFERA_MODULES_KEYS } from 'src/authentication/permissions.enum';
 
 @ApiInternalOnlyController()
 @ApiTags('Connection Test')
 @Controller('connection-test')
 @UseFilters(new GrpcToHttpExceptionFilter())
 @Authenticated()
+@RequireModule(
+  DADOSFERA_MODULES_KEYS.COLLECT
+)
 export class ConnectionTestController {
   logger: any;
   constructor(
@@ -125,6 +131,23 @@ export class ConnectionTestController {
     );
   }
 
+  @Post('cdc-prerequisites')
+  @ApiOkResponse({ type: ValidateCdcPrerequisitesRes })
+  @HttpCode(HttpStatus.OK)
+  async validateCdcPrerequisites(
+    @User() user: RequestUser,
+    @Body(new ValidationPipe()) body: ValidateCdcPrerequisitesReq,
+  ) {
+    this.logger.info('/connection-test/cdc-prerequisites', {
+      user: user.user_id,
+      customer: user.customer_name,
+    });
+    return this.connectionTestService.validateCdcPrerequisites(
+      body,
+      user.customer_name,
+    );
+  }
+
   @Post('refresh-catalog')
   @ApiOkResponse({ type: RefreshCatalogRes })
   @HttpCode(HttpStatus.ACCEPTED)

src/modules/connection-test/connection-test.service.spec.ts

@@ -45,10 +45,24 @@ describe('ConnectionTestService catalog cache', () => {
     });
   });
 
-  it('keeps the existing tables response contract', async () => {
-    connectionsApiService.proxy.mockResolvedValue({
-      tables: [{ table_name: 'customers' }, { table_name: 'orders' }],
-    });
+  it('lists tables and enriches each with its cached primary keys', async () => {
+    connectionsApiService.proxy
+      // list-tables call (names only from the catalog cache)
+      .mockResolvedValueOnce({
+        tables: [{ table_name: 'customers' }, { table_name: 'orders' }],
+      })
+      // per-table columns calls: customers has a PK, orders has none
+      .mockResolvedValueOnce({
+        columns: [
+          { column_name: 'id', data_type: 'bigint', is_primary_key: true },
+          { column_name: 'name', data_type: 'text', is_primary_key: false },
+        ],
+      })
+      .mockResolvedValueOnce({
+        columns: [
+          { column_name: 'total', data_type: 'numeric', is_primary_key: false },
+        ],
+      });
 
     await expect(
       service.connectionTestListTables(
@@ -62,6 +76,10 @@ describe('ConnectionTestService catalog cache', () => {
     ).resolves.toEqual({
       operation_result: true,
       table_list: ['customers', 'orders'],
+      tables: [
+        { table_name: 'customers', primary_keys: ['id'] },
+        { table_name: 'orders', primary_keys: [] },
+      ],
     });
   });
 

src/modules/connection-test/connection-test.service.ts

@@ -13,6 +13,8 @@ import {
   ConnectionTestPingRes,
   GetTableMetadataReq,
   GetTableMetadataRes,
+  ValidateCdcPrerequisitesReq,
+  ValidateCdcPrerequisitesRes,
   RefreshCatalogReq,
   RefreshCatalogRes,
   RefreshCatalogStatusReq,
@@ -177,9 +179,33 @@ export class ConnectionTestService {
         `/schemas/${encodeURIComponent(body.schema)}/tables`,
       user,
     );
+    const table_names: string[] = result.tables.map((table) => table.table_name);
+    // CDC create needs the primary keys per table (used to build the deduped
+    // Iceberg identifier-fields). The catalog-cache list-tables endpoint returns
+    // only names, so fetch each table's columns from the cache and keep the ones
+    // flagged is_primary_key. Reads hit the stored catalog snapshot (populated by
+    // refresh-catalog), never the live connection.
+    const tables = await Promise.all(
+      table_names.map(async (table_name) => {
+        const columns = await this.connectionsApiService.proxy(
+          'GET',
+          `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
+            `/schemas/${encodeURIComponent(body.schema)}` +
+            `/tables/${encodeURIComponent(table_name)}/columns`,
+          user,
+        );
+        return {
+          table_name,
+          primary_keys: columns.columns
+            .filter((column) => column.is_primary_key)
+            .map((column) => column.column_name),
+        };
+      }),
+    );
     return {
       operation_result: true,
-      table_list: result.tables.map((table) => table.table_name),
+      table_list: table_names,
+      tables,
     };
   }
 
@@ -287,4 +313,18 @@ export class ConnectionTestService {
       date: body.date,
     };
   }
+
+  async validateCdcPrerequisites(
+    body: ValidateCdcPrerequisitesReq,
+    customer_name: string,
+  ): Promise<ValidateCdcPrerequisitesRes> {
+    const { plugin, connection_id } = body;
+    return lastValueFrom(
+      this.connectionTestReadClient.ValidateCdcPrerequisites({
+        connection_id,
+        customer_name,
+        plugin,
+      }),
+    );
+  }
 }

src/modules/connection-test/dto/connection-test.ts

@@ -102,11 +102,20 @@ export class ConnectionTestListTablesReq {
   schema: string;
 }
 
+export class ConnectionTestListTablesEntry {
+  @ApiProperty()
+  table_name: string;
+  @ApiProperty({ type: [String] })
+  primary_keys: string[];
+}
+
 export class ConnectionTestListTablesRes {
   @ApiProperty()
   operation_result: boolean;
   @ApiProperty()
   table_list: string[];
+  @ApiProperty({ type: [ConnectionTestListTablesEntry] })
+  tables: ConnectionTestListTablesEntry[];
 }
 
 export class GetTableMetadataReq {
@@ -134,13 +143,59 @@ export class GetTableMetadataRes {
   tables_metadata: TableMetadataDto[];
 }
 
+export class CdcCheckDto {
+  @ApiProperty()
+  name: string;
+  @ApiProperty()
+  expected: string;
+  @ApiProperty()
+  actual: string;
+  @ApiProperty()
+  passed: boolean;
+}
+
+export class ValidateCdcPrerequisitesReq {
+  @ApiProperty()
+  @IsString()
+  plugin: string;
+
+  @ApiProperty()
+  @IsString()
+  connection_id: string;
+}
+
+export class ValidateCdcPrerequisitesRes {
+  @ApiProperty()
+  operation_result: boolean;
+  @ApiProperty({ type: [CdcCheckDto] })
+  checks: CdcCheckDto[];
+}
+
 export class RefreshCatalogReq {
   @ApiProperty()
   @IsString()
   connection_id: string;
 
-  @ApiProperty({ enum: ['oracle', 'mysql', 'postgresql', 'sqlserver'] })
-  @IsIn(['oracle', 'mysql', 'postgresql', 'sqlserver'])
+  @ApiProperty({
+    enum: [
+      'oracle',
+      'mysql',
+      'postgresql',
+      'sqlserver',
+      'mysql_cdc',
+      'postgresql_cdc',
+      'oracle_cdc',
+    ],
+  })
+  @IsIn([
+    'oracle',
+    'mysql',
+    'postgresql',
+    'sqlserver',
+    'mysql_cdc',
+    'postgresql_cdc',
+    'oracle_cdc',
+  ])
   plugin: string;
 }
 

src/modules/connection/connection.controller.ts

@@ -16,8 +16,9 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import {
   Authenticated,
   RequireAllPermissions,
+  RequireModule,
 } from 'src/decorators/authentication.decorator';
-import { PERMISSIONS_GROUPS } from 'src/authentication/permissions.enum';
+import { DADOSFERA_MODULES_KEYS, PERMISSIONS_GROUPS } from 'src/authentication/permissions.enum';
 import { RequestUser, User } from 'src/decorators/user.decorator';
 import { ValidationPipe } from '../../pipes/object-validation.pipe';
 import {
@@ -39,6 +40,9 @@ const connectionPermissions = PERMISSIONS_GROUPS.CONNECTION.permissions;
 @ApiTags('connections')
 @Authenticated()
 @Controller('connections')
+@RequireModule(
+  DADOSFERA_MODULES_KEYS.COLLECT
+)
 export class ConnectionController {
   logger: any;
   constructor(

src/modules/connector/connector.controller.ts

@@ -25,9 +25,10 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import {
   Authenticated,
   RequireAllPermissions,
+  RequireModule,
   RequireSomePermission,
 } from 'src/decorators/authentication.decorator';
-import { PERMISSIONS_GROUPS } from 'src/authentication/permissions.enum';
+import { DADOSFERA_MODULES_KEYS, PERMISSIONS_GROUPS } from 'src/authentication/permissions.enum';
 import { Language } from 'src/decorators/language.decorator';
 import { LanguageEnum } from 'src/utils/languages.enum';
 import { ApiInternalOnlyController } from 'src/decorators/swagger.decorator';
@@ -99,6 +100,9 @@ export class ConnectorController {
     PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE,
     PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async getAllConnectors(
     @Language() language: LanguageEnum,
     @Query() queries: GetAllDto,
@@ -131,6 +135,9 @@ export class ConnectorController {
     PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE,
     PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async getConnectorsTags() {
     return await this.connectorClientService.getConnectorsTags();
   }
@@ -143,6 +150,9 @@ export class ConnectorController {
     PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE,
     PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async getConnector(
     @Language() language: LanguageEnum,
     @Param('plugin') plugin: string,
@@ -171,6 +181,9 @@ export class ConnectorController {
     PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE,
     PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE,
   )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async getConnectorDetails(
     @Language() language: LanguageEnum,
     @Param('plugin') plugin: string,
@@ -193,6 +206,9 @@ export class ConnectorController {
   @Put('/:plugin')
   @RequireAllPermissions(PERMISSIONS_GROUPS.CONNECTORS.permissions.UPDATE)
   @ApiConsumes('multipart/form-data')
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async updateConnector(
     @Param('plugin') plugin: string,
     @Body() body: UpdateDto,
@@ -214,6 +230,9 @@ export class ConnectorController {
 
   @Put('/:plugin/add-tag')
   @RequireAllPermissions(PERMISSIONS_GROUPS.CONNECTORS.permissions.UPDATE)
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async addTagOnConnector(
     @Param('plugin') plugin: string,
     @Body() body: AddTagDto,
@@ -241,6 +260,9 @@ export class ConnectorController {
 
   @Put('/:plugin/remove-tag')
   @RequireAllPermissions(PERMISSIONS_GROUPS.CONNECTORS.permissions.UPDATE)
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async removeTagOnConnector(
     @Param('plugin') plugin: string,
     @Body() body: RemoveTagDto,
@@ -269,6 +291,9 @@ export class ConnectorController {
 
   @Delete('/:plugin')
   @RequireAllPermissions(PERMISSIONS_GROUPS.CONNECTORS.permissions.DELETE)
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.COLLECT
+  )
   async deleteConnector(
     @Param('plugin') plugin: string,
     @Query('version') version: string,

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

@@ -65,3 +65,64 @@ export class CreateInputReq extends OmitType(Input, [
   'created_at',
   'updated_at',
 ]) {}
+
+export class CdcColumnReq {
+  @ApiProperty()
+  name: string;
+  @ApiProperty()
+  type: string;
+  @ApiProperty()
+  is_primary_key: boolean;
+}
+
+export class CdcTableReq {
+  @ApiProperty()
+  name: 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).
+  // Honored on the create/add path: the platform lowercases + sanitizes it
+  // authoritatively; empty/absent => the platform derives tb__<hash>__<table>.
+  @ApiPropertyOptional()
+  iceberg_table_name?: string;
+  // Per-table deduped (qualify) Iceberg table name override (iceberg dest only).
+  // Empty/absent => the deduped table takes the same name as the raw table.
+  @ApiPropertyOptional()
+  iceberg_qualify_table_name?: string;
+  @ApiPropertyOptional({ type: [CdcColumnReq] })
+  columns?: CdcColumnReq[];
+  // Columns the user chose to ignore -> Debezium column.exclude.list.
+  @ApiPropertyOptional({ type: [String] })
+  column_exclude_list?: string[];
+}
+
+export class IcebergDestinationReq {
+  @ApiProperty()
+  namespace: string;
+  // Pipeline-wide deduped (qualify) namespace. Absent => the platform derives
+  // the sibling of `namespace` (cdc_raw -> cdc_dedup).
+  @ApiPropertyOptional()
+  qualify_namespace?: string;
+}
+
+export class CdcDestinationReq {
+  @ApiPropertyOptional({ type: IcebergDestinationReq })
+  iceberg?: IcebergDestinationReq;
+}
+
+export class CreateCdcInputReq {
+  @ApiProperty()
+  name: string;
+  @ApiProperty()
+  plugin: string; // mysql_cdc (v1)
+  @ApiProperty({ type: [CdcTableReq] })
+  tables: CdcTableReq[];
+  @ApiPropertyOptional()
+  read_only?: boolean;
+  @ApiPropertyOptional({ type: CdcDestinationReq })
+  destination?: CdcDestinationReq;
+}

src/modules/inputs/inputs.controller.spec.ts

@@ -0,0 +1,75 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { InputsController } from './inputs.controller';
+import { InputsService } from './inputs.service';
+import DadosferaLogger from '@dadosfera/dadosfera-logs';
+import { CreateCdcInputReq } from './dtos/input.model';
+import { RequestUser } from 'src/decorators/user.decorator';
+
+describe('InputsController', () => {
+  let controller: InputsController;
+  let inputsService: { createCdc: jest.Mock };
+
+  beforeEach(async () => {
+    inputsService = {
+      createCdc: jest.fn().mockResolvedValue({ input: {} }),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [InputsController],
+      providers: [
+        {
+          provide: DadosferaLogger,
+          useValue: { logger: { info: jest.fn() } },
+        },
+        {
+          provide: InputsService,
+          useValue: inputsService,
+        },
+      ],
+    }).compile();
+
+    controller = module.get<InputsController>(InputsController);
+  });
+
+  it('should be defined', () => {
+    expect(controller).toBeDefined();
+  });
+
+  it('forwards destination.iceberg.namespace to InputsService.createCdc', async () => {
+    const body: CreateCdcInputReq = {
+      name: 'my-cdc-input',
+      plugin: 'mysql_cdc',
+      tables: [
+        {
+          name: 'orders',
+          table_schema: 'public',
+          iceberg_table_name: 'orders_iceberg',
+        },
+      ],
+      destination: {
+        iceberg: {
+          namespace: 'my_namespace',
+        },
+      },
+    };
+    const user: RequestUser = {
+      user_id: 'user-1',
+      customer_id: 'customer-1',
+      customer_name: 'customer',
+    } as RequestUser;
+
+    await controller.createCdc(body, user);
+
+    expect(inputsService.createCdc).toHaveBeenCalledWith(
+      expect.objectContaining({
+        body: expect.objectContaining({
+          destination: {
+            iceberg: {
+              namespace: 'my_namespace',
+            },
+          },
+        }),
+      }),
+    );
+  });
+});

src/modules/inputs/inputs.controller.ts

@@ -15,6 +15,7 @@ import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
 import { AuthenticateCondition } from 'src/decorators/authentication.decorator';
 import { ApiOkResponse, ApiTags } from '@nestjs/swagger';
 import {
+  CreateCdcInputReq,
   CreateInputReq,
   GetAvailableEntitiesReq,
   GetAvailableEntitiesRes,
@@ -105,6 +106,26 @@ export class InputsController {
     return response;
   }
 
+  @Post('cdc')
+  @ApiInternalOnlyEndpoint()
+  @ApiOkResponse({ type: Input })
+  async createCdc(
+    @Body() body: CreateCdcInputReq,
+    @User() user: RequestUser,
+  ) {
+    const info: Info = {
+      user_id: user.user_id,
+      customer: user.customer_name,
+      customer_id: user.customer_id,
+    };
+    this.logger.info(`/inputs/cdc - ON CREATE CDC INPUT ROUTE`, {
+      user: info.user_id,
+      customer: info.customer,
+    });
+
+    return this.inputService.createCdc({ body, info });
+  }
+
   @ApiInternalOnlyEndpoint()
   @Get()
   async findAll(@User() user: RequestUser) {

src/modules/inputs/inputs.service.spec.ts

@@ -0,0 +1,113 @@
+import { of } from 'rxjs';
+import { InputsService } from './inputs.service';
+import DadosferaLogger from '@dadosfera/dadosfera-logs/dist';
+import { CreateCdcInputReq } from './dtos/input.model';
+import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
+
+const info = { customer_id: 'cid', user_id: 'u' } as unknown as Info;
+
+describe('InputsService.createCdc', () => {
+  let service: InputsService;
+  let inputCreateCdcMock: jest.Mock;
+
+  beforeEach(async () => {
+    inputCreateCdcMock = jest
+      .fn()
+      .mockImplementation((req) => of({ input: req.input }));
+
+    const grpcClient: any = {
+      getService: jest.fn().mockReturnValue({
+        InputCreateCdc: inputCreateCdcMock,
+      }),
+    };
+
+    service = new InputsService(new DadosferaLogger(), grpcClient);
+    await service.onModuleInit();
+  });
+
+  it('forwards destination and per-table iceberg_table_name to the gRPC request', async () => {
+    const body: CreateCdcInputReq = {
+      name: 'CDC Iceberg Test',
+      plugin: 'mysql_cdc',
+      read_only: true,
+      destination: { iceberg: { namespace: 'cdc_raw' } },
+      tables: [
+        {
+          name: 'orders',
+          table_schema: 'mydb',
+          primary_keys: ['id'],
+          iceberg_table_name: 'cdc_raw.mydb__orders',
+        },
+      ],
+    };
+
+    await service.createCdc({ body, info });
+
+    expect(inputCreateCdcMock).toHaveBeenCalledTimes(1);
+    const sentRequest = inputCreateCdcMock.mock.calls[0][0];
+
+    expect(sentRequest.input).toEqual(
+      expect.objectContaining({
+        destination: { iceberg: { namespace: 'cdc_raw' } },
+      }),
+    );
+    expect(sentRequest.input.tables[0]).toEqual(
+      expect.objectContaining({
+        iceberg_table_name: 'cdc_raw.mydb__orders',
+      }),
+    );
+  });
+
+  it('forwards per-table columns to the gRPC request', async () => {
+    const body: CreateCdcInputReq = {
+      name: 'CDC Columns Test',
+      plugin: 'mysql_cdc',
+      read_only: true,
+      tables: [
+        {
+          name: 'orders',
+          table_schema: 'mydb',
+          primary_keys: ['id'],
+          columns: [
+            { name: 'id', type: 'int', is_primary_key: true },
+            { name: 'descr', type: 'varchar(255)', is_primary_key: false },
+          ],
+        },
+      ],
+    };
+
+    await service.createCdc({ body, info });
+
+    expect(inputCreateCdcMock).toHaveBeenCalledTimes(1);
+    const sentRequest = inputCreateCdcMock.mock.calls[0][0];
+
+    expect(sentRequest.input.tables[0].columns).toEqual([
+      { name: 'id', type: 'int', is_primary_key: true },
+      { name: 'descr', type: 'varchar(255)', is_primary_key: false },
+    ]);
+  });
+
+  it('back-compat: a body with no destination sends destination undefined, not an error', async () => {
+    const body: CreateCdcInputReq = {
+      name: 'CDC Legacy Test',
+      plugin: 'mysql_cdc',
+      read_only: true,
+      tables: [
+        {
+          name: 'pedidos',
+          table_schema: 'cadastros',
+          primary_keys: ['id'],
+        },
+      ],
+    };
+
+    const result = await service.createCdc({ body, info });
+
+    expect(inputCreateCdcMock).toHaveBeenCalledTimes(1);
+    const sentRequest = inputCreateCdcMock.mock.calls[0][0];
+
+    expect(sentRequest.input.destination).toBeUndefined();
+    expect(sentRequest.input.tables[0].iceberg_table_name).toBeUndefined();
+    expect(result.input).toBeDefined();
+  });
+});

src/modules/inputs/inputs.service.ts

@@ -13,16 +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 { CreateInputReq } from './dtos/input.model';
+import { CreateCdcInputReq, CreateInputReq } from './dtos/input.model';
+import { toCdcTable } from './cdc-table.mapper';
 import { Metadata } from '@grpc/grpc-js';
 
 @Injectable()
@@ -183,6 +188,26 @@ export class InputsService {
     return { input: adjustedInput };
   }
 
+  async createCdc(data: { body: CreateCdcInputReq; info: Info }) {
+    const { body, info } = data;
+
+    const inputCreateCdcRequest: InputCreateCdcRequest = {
+      input: {
+        name: body.name,
+        plugin: body.plugin,
+        read_only: body.read_only ?? true,
+        tables: body.tables.map(toCdcTable),
+        destination: body.destination,
+      },
+      info,
+    };
+
+    const { input } = await lastValueFrom(
+      this.inputWriteService.InputCreateCdc(inputCreateCdcRequest),
+    );
+    return { input };
+  }
+
   async getAvailableEntities(data: GetAvailableEntitiesRequest) {
     return lastValueFrom(this.inputReadService.GetAvailableEntities(data));
   }
@@ -290,11 +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: AddCdcTableRequest) {
+    return lastValueFrom(this.inputWriteService.AddCdcTable(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,6 +58,8 @@ export class IPipelineV2 {
   tags?: string[];
   @ApiPropertyOptional()
   properties?: any;
+  @ApiPropertyOptional({ type: () => PipelineConfig })
+  config?: PipelineConfig;
 
   @ApiProperty()
   connector_name: string;

src/modules/pipelinesV2/pipelines-client.ts

@@ -25,6 +25,10 @@ export class PipelinesClientConfiguration {
       loader: {
         keepCase: true,
         enums: String,
+        // int64 fields (PipelineV2SinkTableOffset.committed_offset) decode as
+        // plain JS numbers instead of Long.js objects, so they serialize as
+        // JSON numbers for the frontend. Safe: Kafka offsets fit in 2^53.
+        longs: Number,
         objects: true,
         arrays: true,
       },

src/modules/pipelinesV2/pipelines.controller.ts

@@ -25,9 +25,10 @@ import {
 } from '@nestjs/swagger';
 import {
   RequireAllPermissions,
+  RequireModule,
   RequireSomePermission,
 } from 'src/decorators/authentication.decorator';
-import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
+import { DADOSFERA_MODULES_KEYS, PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
 import { PipelinesService } from './pipelines.service';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { Messages } from '@dadosfera/protospack-v2/dist/lib/PipelineV2';
@@ -57,6 +58,9 @@ type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] };
 @ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
 @UseFilters(new GrpcToHttpExceptionFilter())
 @Controller('pipelinesV2')
+@RequireModule(
+  DADOSFERA_MODULES_KEYS.COLLECT
+)
 export class PipelinesController {
   logger: DadosferaLogger;
   constructor(
@@ -506,4 +510,78 @@ export class PipelinesController {
 
     return response;
   }
+
+  // ---- CDC pipeline operations (Kafka Connect backed) ----
+  // These operate on an existing pipeline, so they require UPDATE (not CREATE).
+
+  @Get(':id/live-status')
+  @ApiInternalOnlyEndpoint()
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
+  async getLiveStatus(
+    @Param('id') id: string,
+    @User() user: RequestUser,
+  ): Promise<Messages.PipelineV2GetLiveStatusResponse> {
+    this.logger.info('PipelinesController - getLiveStatus', { id });
+    const metadata = PackTheMetadata(user);
+    return this.pipelinesClientService.getLiveStatus(id, metadata);
+  }
+
+  @Post(':id/pause')
+  @HttpCode(HttpStatus.OK)
+  @ApiInternalOnlyEndpoint()
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async pause(
+    @Param('id') id: string,
+    @User() user: RequestUser,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesController - pause', { id });
+    const metadata = PackTheMetadata(user);
+    return this.pipelinesClientService.pause(id, metadata);
+  }
+
+  @Post(':id/unpause')
+  @HttpCode(HttpStatus.OK)
+  @ApiInternalOnlyEndpoint()
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async unpause(
+    @Param('id') id: string,
+    @User() user: RequestUser,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesController - unpause', { id });
+    const metadata = PackTheMetadata(user);
+    return this.pipelinesClientService.unpause(id, metadata);
+  }
+
+  @Post(':id/restart')
+  @HttpCode(HttpStatus.OK)
+  @ApiInternalOnlyEndpoint()
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async restart(
+    @Param('id') id: string,
+    @User() user: RequestUser,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesController - restart', { id });
+    const metadata = PackTheMetadata(user);
+    return this.pipelinesClientService.restart(id, metadata);
+  }
+
+  @Post(':id/jobs/:jobId/reset-state')
+  @HttpCode(HttpStatus.OK)
+  @ApiInternalOnlyEndpoint()
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async resetJobState(
+    @Param('id') id: string,
+    @Param('jobId') jobId: string,
+    @Body() body: { schedule_minutes?: number },
+    @User() user: RequestUser,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesController - resetJobState', { id, jobId });
+    const metadata = PackTheMetadata(user);
+    return this.pipelinesClientService.resetJobState(
+      id,
+      jobId,
+      body?.schedule_minutes,
+      metadata,
+    );
+  }
 }

src/modules/pipelinesV2/pipelines.module.ts

@@ -1,4 +1,4 @@
-import { Module } from '@nestjs/common';
+import { Module, forwardRef } from '@nestjs/common';
 import { ClientsModule } from '@nestjs/microservices';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 
@@ -23,7 +23,7 @@ const client = new PipelinesClientConfiguration();
     ConnectorModule,
     InputsModule,
     TransformationsModule,
-    PlatformApiModule,
+    forwardRef(() => PlatformApiModule),
     NimbusServicesModule,
     CatalogModule
   ],

src/modules/pipelinesV2/pipelines.service.spec.ts

@@ -0,0 +1,142 @@
+// These imported modules pull in gRPC client-config / service modules that read
+// process.env at load time; mock them (hoisted before imports) so the spec needs
+// no runtime env. Each mock severs an entire import subtree while still providing
+// a class usable as a value/DI token. Mirrors platform-api.controller.spec.ts.
+jest.mock('./pipelines-client', () => ({ PipelinesClientConfiguration: class {} }));
+jest.mock('../connector/client.service', () => ({ ConnectorClientService: class {} }));
+jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
+jest.mock('../transformations/transformations.service', () => ({ TransformationsService: class {} }));
+jest.mock('../platform-api/platform-api.service', () => ({ PlatformApiService: class {} }));
+jest.mock('src/services/nimbus/nimbus.service', () => ({ NimbusService: class {} }));
+jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} }));
+
+import { of } from 'rxjs';
+import { PipelinesService } from './pipelines.service';
+
+const logger = {
+  info: jest.fn(),
+  error: jest.fn(),
+};
+
+const cdcOldInput = { input: { plugin: 'mysql_cdc', tables: [] } };
+const batchOldInput = { input: { plugin: 'mysql', type: 'database', tables: [] } };
+
+const updateResponse = {
+  input: { type: 'database' },
+  tablesUpdate: [],
+  dataAssetUpdate: [],
+};
+
+const user: any = { customer_modules: [] };
+const updateInputDTO: any = { tables: [] };
+const info: any = { customer: 'cust' };
+const metadata: any = {};
+
+function buildService(oldInput: any) {
+  const inputsService: any = {
+    findOne: jest.fn().mockResolvedValue(oldInput),
+    update: jest.fn().mockResolvedValue(updateResponse),
+    rollbackUpdate: jest.fn().mockResolvedValue({}),
+  };
+  const nimbusService: any = { renameTable: jest.fn().mockResolvedValue({}) };
+
+  const service = new PipelinesService(
+    { logger } as any, // dadosferaLogger
+    {} as any, // grpcClient
+    {} as any, // connectorService
+    inputsService, // inputsService
+    {} as any, // transformationsService
+    {} as any, // platformAPI
+    nimbusService, // nimbusService
+    {} as any, // catalogService
+  );
+
+  const updatePlatformJobsSpy = jest
+    .spyOn(service, 'updatePlatformJobs')
+    .mockResolvedValue(undefined as any);
+
+  return { service, inputsService, updatePlatformJobsSpy };
+}
+
+describe('PipelinesService - updatePipelineInput', () => {
+  afterEach(() => jest.clearAllMocks());
+
+  it('CDC input skips updatePlatformJobs', async () => {
+    const { service, inputsService, updatePlatformJobsSpy } = buildService(cdcOldInput);
+
+    const result = await service.updatePipelineInput(
+      'pipeline-id',
+      'input-id',
+      updateInputDTO,
+      info,
+      user,
+      metadata,
+    );
+
+    expect(updatePlatformJobsSpy).not.toHaveBeenCalled();
+    expect(inputsService.update).toHaveBeenCalled();
+    expect(result).toBe(updateResponse);
+  });
+
+  it('batch input calls updatePlatformJobs', async () => {
+    const { service, inputsService, updatePlatformJobsSpy } = buildService(batchOldInput);
+
+    await service.updatePipelineInput(
+      'pipeline-id',
+      'input-id',
+      updateInputDTO,
+      info,
+      user,
+      metadata,
+    );
+
+    expect(updatePlatformJobsSpy).toHaveBeenCalled();
+    expect(inputsService.update).toHaveBeenCalled();
+  });
+});
+
+describe('PipelinesService - create', () => {
+  afterEach(() => jest.clearAllMocks());
+
+  // The body's `config` (carrying CDC destinations) must reach pi-factory as a
+  // JSON string — the gRPC proto field is a string, so an object would be
+  // stripped on the wire. Mirrors how `properties` is serialized.
+  it('serializes the body config into the gRPC create request', async () => {
+    const { service } = buildService(cdcOldInput);
+
+    let captured: any;
+    (service as any).pipelineWriteService = {
+      PipelineV2Create: (req: any) => {
+        captured = req;
+        // The service does `lastValueFrom(...)`; return a real Observable.
+        return of({ pipeline: {} });
+      },
+    };
+
+    const body: any = {
+      name: 'p',
+      input_id: 'i',
+      transformations_ids: [],
+      tags: [],
+      properties: { schema: 'cadastros' },
+      config: {
+        cron: '@once',
+        tables: [
+          {
+            name: 'pedidos',
+            destinations: {
+              raw: { table_schema: 'PUBLIC', table_name: 'pedidos_001' },
+            },
+          },
+        ],
+      },
+    };
+
+    await service.create(body, metadata);
+
+    expect(typeof captured.config).toBe('string');
+    expect(JSON.parse(captured.config).tables[0].destinations.raw.table_name).toBe(
+      'pedidos_001',
+    );
+  });
+});

src/modules/pipelinesV2/pipelines.service.ts

@@ -1,3 +1,4 @@
+import { isCdcPlugin } from 'src/utils/cdc';
 /* eslint-disable no-async-promise-executor */
 import {
   BadRequestException,
@@ -19,7 +20,7 @@ import { lastValueFrom } from 'rxjs';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { PipelinesClientConfiguration } from './pipelines-client';
 import { ICreatePipelineV2Req, IIdRequest, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces';
-import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
+import { PipelineV2CreateRequest, AddCdcJobsRequest, AddCdcJobsResponse } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
 import { Metadata } from '@grpc/grpc-js';
 import { ConnectorClientService } from '../connector/client.service';
 import { InputsService } from '../inputs/inputs.service';
@@ -105,6 +106,10 @@ export class PipelinesService implements OnModuleInit {
       transformations_ids: body.transformations_ids,
       tags: body.tags,
       properties: body.properties && JSON.stringify(body.properties),
+      // JSON-serialize the create body config so it survives the gRPC wire
+      // (the proto field is a string). pi-factory's CDC path reads
+      // config.tables[].destinations to honor a user-supplied raw table name.
+      config: body.config && JSON.stringify(body.config),
     };
 
     const createPipelineResponse = await lastValueFrom(
@@ -160,6 +165,62 @@ export class PipelinesService implements OnModuleInit {
     return findOnePipelineResponse;
   }
 
+  // CDC lifecycle operations (Kafka Connect backed).
+  async pause(
+    id: string,
+    metadata,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesClientService - Pause');
+    return lastValueFrom(
+      this.pipelineWriteService.PipelineV2Pause({ id }, metadata),
+    );
+  }
+
+  async unpause(
+    id: string,
+    metadata,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesClientService - Unpause');
+    return lastValueFrom(
+      this.pipelineWriteService.PipelineV2Unpause({ id }, metadata),
+    );
+  }
+
+  async restart(
+    id: string,
+    metadata,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesClientService - Restart');
+    return lastValueFrom(
+      this.pipelineWriteService.PipelineV2Restart({ id }, metadata),
+    );
+  }
+
+  async resetJobState(
+    id: string,
+    job_id: string,
+    schedule_minutes: number | undefined,
+    metadata,
+  ): Promise<Messages.PipelineV2OperationResponse> {
+    this.logger.info('PipelinesClientService - ResetJobState');
+    return lastValueFrom(
+      this.pipelineWriteService.PipelineV2ResetJobState(
+        { id, job_id, schedule_minutes },
+        metadata,
+      ),
+    );
+  }
+
+  async getLiveStatus(
+    id: string,
+    metadata,
+  ): Promise<Messages.PipelineV2GetLiveStatusResponse> {
+    this.logger.info('PipelinesClientService - GetLiveStatus');
+    return lastValueFrom(
+      this.pipelineReadService.PipelineV2GetLiveStatus({ id }, metadata),
+    );
+  }
+
   async update(
     UpdatePipelineRequest: Messages.PipelineV2UpdateRequest,
     metadata,
@@ -377,6 +438,7 @@ export class PipelinesService implements OnModuleInit {
     });
 
     this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
+    const isCdc = isCdcPlugin(oldInput.plugin);
     const pipelineIdFormat = pipelineId.split('-').join('_');
     const rollback: RollbackPromise[] = [];
 
@@ -437,17 +499,21 @@ export class PipelinesService implements OnModuleInit {
       }
     }
 
-    try {
-      await this.updatePlatformJobs(
-        pipelineIdFormat,
-        updateInputResponse.input.type,
-        updateInputDTO,
-        user
-      );
-    } catch (error) {
-      this.logger.error(error);
-      await this.executeRenameRollback(rollback)
-      throw new Error("Error Platform API updating jobs");
+    if (!isCdc) {
+      try {
+        await this.updatePlatformJobs(
+          pipelineIdFormat,
+          updateInputResponse.input.type,
+          updateInputDTO,
+          user
+        );
+      } catch (error) {
+        this.logger.error(error);
+        await this.executeRenameRollback(rollback)
+        throw new Error("Error Platform API updating jobs");
+      }
+    } else {
+      this.logger.info('CDC input: skipping updatePlatformJobs (batch sync_mode/memory do not apply to CDC jobs)');
     }
 
     return updateInputResponse;
@@ -662,6 +728,10 @@ export class PipelinesService implements OnModuleInit {
     return statusPipelineResponse;
   }
 
+  async addCdcJobs(data: AddCdcJobsRequest): Promise<AddCdcJobsResponse> {
+    return lastValueFrom(this.pipelineWriteService.AddCdcJobs(data));
+  }
+
   async runPipeline({ id, info }: IIdRequest) {
     this.logger.info('PipelinesClientService - RunPipeline');
     const statusPipelineResponse = await lastValueFrom(

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

@@ -0,0 +1,85 @@
+// 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. Each mock severs an entire import subtree and still
+// provides a class usable as a DI token.
+jest.mock('../customers/customers.service', () => ({ CustomersService: class {} }));
+jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} }));
+jest.mock('../inputs/inputs.service', () => ({ InputsService: 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';
+
+const logger = { info: jest.fn(), error: jest.fn() };
+const mockUser = { customer_id: 'c1', customer_name: 'cust', user_id: 'u1' } as never;
+
+// 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 tables: { removeTable: jest.Mock; removeTables: jest.Mock; addTable: jest.Mock };
+
+  beforeEach(async () => {
+    tables = { removeTable: jest.fn(), removeTables: jest.fn(), addTable: jest.fn() };
+
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [PlatformApiController],
+      providers: [
+        { provide: PlatformApiService, useValue: {} },
+        { provide: ElasticsearchService, useValue: {} },
+        { provide: DynamoDBService, useValue: {} },
+        { provide: CustomersService, useValue: {} },
+        { provide: CatalogService, useValue: {} },
+        { provide: InputsService, useValue: {} },
+        { provide: PipelineTablesService, useValue: tables },
+        { provide: DadosferaLogger, useValue: { logger } },
+      ],
+    }).compile();
+
+    controller = module.get<PlatformApiController>(PlatformApiController);
+  });
+
+  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('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('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 delegates', async () => {
+    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' },
+      },
+    };
+    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';
@@ -20,18 +19,20 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import {
   Authenticated,
   RequireAllPermissions,
+  RequireModule,
 } from '../../decorators/authentication.decorator';
 import { User, RequestUser } from '../../decorators/user.decorator';
 import { PlatformApiService } from './platform-api.service';
-import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
+import { DADOSFERA_MODULES_KEYS, PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
 import { ElasticsearchService } from '../../services/elasticsearch';
 import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
 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 { PipelineTablesService } from './pipeline-tables.service';
 import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
 
 
@@ -49,6 +50,7 @@ type RenameTablesBody = {
 
 @ApiTags('Platform API')
 @Controller('platform')
+@RequireModule(DADOSFERA_MODULES_KEYS.COLLECT)
 export class PlatformApiController {
   private logger: any;
 
@@ -59,6 +61,7 @@ export class PlatformApiController {
     private readonly customersService: CustomersService,
     private readonly catalogService: CatalogService,
     private readonly inputsService: InputsService,
+    private readonly pipelineTablesService: PipelineTablesService,
     @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
   ) {
     this.logger = dadosferaLogger.logger;
@@ -543,6 +546,20 @@ export class PlatformApiController {
     return this.platformApiService.proxy('GET', `/pipeline/${normalizedId}`, user);
   }
 
+  @Get('iceberg/namespaces')
+  @ApiOperation({ summary: 'List existing Polaris Iceberg namespaces (CDC destination dropdown)' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
+  async getIcebergNamespaces(@User() user: RequestUser) {
+    return this.platformApiService.proxy('GET', '/iceberg/namespaces', user);
+  }
+
+  @Post('iceberg/tables/validate')
+  @ApiOperation({ summary: 'Validate CDC Iceberg raw table names against Polaris' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
+  async validateIcebergTables(@Body() body: any, @User() user: RequestUser) {
+    return this.platformApiService.proxy('POST', '/iceberg/tables/validate', user, body);
+  }
+
   @Patch('pipelines/:pipelineId')
   @ApiOperation({ summary: 'Update pipeline by ID' })
   @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
@@ -970,49 +987,46 @@ 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,
-    };
+    return this.pipelineTablesService.removeTable(pipelineId, inputId, body.table_name, user);
+  }
 
-    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 });
-      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;
+  @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: DeleteTablesBody,
+    @User() user: RequestUser,
+  ) {
+    const tableNames = body.table_names ?? [];
+    if (tableNames.length === 0) {
+      throw new BadRequestException('table_names must be a non-empty array');
     }
+    return this.pipelineTablesService.removeTables(pipelineId, inputId, tableNames, user);
+  }
+
+  @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)
+  @UseGuards(PipelineExecutionGuard)
+  async addTable(
+    @Param('pipelineId') pipelineId: string,
+    @Param('inputId') inputId: string,
+    @Body() body: AddCdcTableBody,
+    @User() user: RequestUser,
+  ) {
+    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

@@ -1,19 +1,28 @@
-import { Module } from '@nestjs/common';
+import { Module, forwardRef } from '@nestjs/common';
 
 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';
 import { CatalogModule } from '../catalog/catalog.module';
 import { InputsModule } from '../inputs/inputs.module';
+import { PipelinesV2Module } from '../pipelinesV2/pipelines.module';
 
 @Module({
-  imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, InputsModule],
+  imports: [
+    ElasticsearchModule,
+    DynamoDBModule,
+    CustomersModule,
+    CatalogModule,
+    InputsModule,
+    forwardRef(() => PipelinesV2Module),
+  ],
   controllers: [PlatformApiController],
-  providers: [PlatformApiService, DadosferaLogger],
+  providers: [PlatformApiService, PipelineTablesService, DadosferaLogger],
   exports: [PlatformApiService],
 })
 export class PlatformApiModule {}

src/modules/release_note/release_note.controller.spec.ts

@@ -1,14 +1,26 @@
 import { Test, TestingModule } from '@nestjs/testing';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { ReleaseNoteController } from './release_note.controller';
 import { ReleaseNoteService } from './release_note.service';
 
+const logger = {
+  info: (...args) => args,
+  error: (...args) => args,
+};
+
 describe('ReleaseNoteController', () => {
   let controller: ReleaseNoteController;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
       controllers: [ReleaseNoteController],
-      providers: [ReleaseNoteService],
+      providers: [
+        ReleaseNoteService,
+        {
+          provide: DadosferaLogger,
+          useValue: { logger },
+        },
+      ],
     }).compile();
 
     controller = module.get<ReleaseNoteController>(ReleaseNoteController);

src/modules/release_note/release_note.service.spec.ts

@@ -1,12 +1,24 @@
 import { Test, TestingModule } from '@nestjs/testing';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { ReleaseNoteService } from './release_note.service';
 
+const logger = {
+  info: (...args) => args,
+  error: (...args) => args,
+};
+
 describe('ReleaseNoteService', () => {
   let service: ReleaseNoteService;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
-      providers: [ReleaseNoteService],
+      providers: [
+        ReleaseNoteService,
+        {
+          provide: DadosferaLogger,
+          useValue: { logger },
+        },
+      ],
     }).compile();
 
     service = module.get<ReleaseNoteService>(ReleaseNoteService);

src/utils/ErrorBuilder.ts

@@ -315,6 +315,14 @@ export function EnrichErrorCode(code: string) {
           'Tente realizar a ação novamente. Caso o erro persista, entre em contato com o suporte',
         code,
       };
+    case ErrorCodes.CATALOG.COLUMN_NOT_FOUND:
+      return {
+        statusCode: HttpStatus.NOT_FOUND,
+        error: 'Coluna não encontrada',
+        message:
+          'Uma ou mais colunas informadas não existem neste ativo. Verifique os nomes e tente novamente.',
+        code,
+      };
     case ErrorCodes.CATALOG.PREVIEW_TOO_BIG:
       return {
         statusCode: HttpStatus.INTERNAL_SERVER_ERROR,

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));
+}

src/utils/errorCodes.ts

@@ -74,6 +74,7 @@ const CATALOG = {
   DATA_ASSET_NOT_FOUND: 'CATALOG.DATA_ASSET_NOT_FOUND',
   PREVIEW_TOO_BIG: 'CATALOG.PREVIEW_TOO_BIG',
   METADATA_TOO_BIG: 'CATALOG.METADATA_TOO_BIG',
+  COLUMN_NOT_FOUND: 'CATALOG.COLUMN_NOT_FOUND',
 };
 const IDENTITY_PROVIDER = {
   INVALID_RESPONSE: 'IDENTITY_PROVIDER.INVALID_RESPONSE',