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

Feature/cdc connector

.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:

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"]
 
 

docsfera.json

@@ -3001,6 +3001,45 @@
           ]
         }
       },
+      "/inputs/cdc": {
+        "post": {
+          "operationId": "InputsController_createCdc",
+          "parameters": [],
+          "requestBody": {
+            "required": true,
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/CreateCdcInputReq"
+                }
+              }
+            }
+          },
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "$ref": "#/components/schemas/Input"
+                  }
+                }
+              }
+            },
+            "201": {
+              "description": ""
+            }
+          },
+          "tags": [
+            "Inputs"
+          ],
+          "security": [
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
       "/inputs/{id}": {
         "get": {
           "operationId": "InputsController_findOne",
@@ -3960,6 +3999,264 @@
           ]
         }
       },
+      "/pipelinesV2/{id}/live-status": {
+        "get": {
+          "operationId": "PipelinesController_getLiveStatus",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            },
+            {
+              "name": "id",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "PipelinesV2"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
+      "/pipelinesV2/{id}/pause": {
+        "post": {
+          "operationId": "PipelinesController_pause",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            },
+            {
+              "name": "id",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "PipelinesV2"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
+      "/pipelinesV2/{id}/unpause": {
+        "post": {
+          "operationId": "PipelinesController_unpause",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            },
+            {
+              "name": "id",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "PipelinesV2"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
+      "/pipelinesV2/{id}/restart": {
+        "post": {
+          "operationId": "PipelinesController_restart",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            },
+            {
+              "name": "id",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "PipelinesV2"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
+      "/pipelinesV2/{id}/jobs/{jobId}/reset-state": {
+        "post": {
+          "operationId": "PipelinesController_resetJobState",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            },
+            {
+              "name": "id",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            },
+            {
+              "name": "jobId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "PipelinesV2"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
       "/transformations": {
         "post": {
           "operationId": "TransformationsController_create",
@@ -4274,6 +4571,66 @@
           ]
         }
       },
+      "/platform/iceberg/namespaces": {
+        "get": {
+          "operationId": "PlatformApiController_getIcebergNamespaces",
+          "summary": "List existing Polaris Iceberg namespaces (CDC destination dropdown)",
+          "parameters": [],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Platform API"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
+      "/platform/iceberg/tables/validate": {
+        "post": {
+          "operationId": "PlatformApiController_validateIcebergTables",
+          "summary": "Validate CDC Iceberg raw table names against Polaris",
+          "parameters": [],
+          "responses": {
+            "201": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Platform API"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
       "/platform/pipelines/execute": {
         "post": {
           "operationId": "PlatformApiController_executePipeline",
@@ -4967,6 +5324,91 @@
           ]
         }
       },
+      "/platform/pipelines/{pipelineId}/inputs/{inputId}/tables": {
+        "delete": {
+          "operationId": "PlatformApiController_deleteTables",
+          "summary": "Batch-remove tables from an input; reconfigures the CDC connectors once",
+          "parameters": [
+            {
+              "name": "pipelineId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            },
+            {
+              "name": "inputId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": ""
+            }
+          },
+          "tags": [
+            "Platform API"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        },
+        "post": {
+          "operationId": "PlatformApiController_addTable",
+          "summary": "Add a CDC table to an input and dispatch its platform jobs",
+          "parameters": [
+            {
+              "name": "pipelineId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            },
+            {
+              "name": "inputId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "201": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Platform API"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
       "/platform/jobs/jdbc/configs/allowed_datatypes": {
         "get": {
           "operationId": "PlatformApiController_getJdbcAllowedDatatypes",
@@ -7513,6 +7955,45 @@
           ]
         }
       },
+      "/connection-test/cdc-prerequisites": {
+        "post": {
+          "operationId": "ConnectionTestController_validateCdcPrerequisites",
+          "parameters": [],
+          "requestBody": {
+            "required": true,
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ValidateCdcPrerequisitesReq"
+                }
+              }
+            }
+          },
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "$ref": "#/components/schemas/ValidateCdcPrerequisitesRes"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Connection Test"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
       "/connection-test/refresh-catalog": {
         "post": {
           "operationId": "ConnectionTestController_refreshCatalog",
@@ -10744,6 +11225,113 @@
             "updated_at"
           ]
         },
+        "CdcColumnReq": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "type": {
+              "type": "string"
+            },
+            "is_primary_key": {
+              "type": "boolean"
+            }
+          },
+          "required": [
+            "name",
+            "type",
+            "is_primary_key"
+          ]
+        },
+        "CdcTableReq": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "table_schema": {
+              "type": "string"
+            },
+            "primary_keys": {
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            },
+            "iceberg_table_name": {
+              "type": "string"
+            },
+            "iceberg_qualify_table_name": {
+              "type": "string"
+            },
+            "columns": {
+              "type": "array",
+              "items": {
+                "$ref": "#/components/schemas/CdcColumnReq"
+              }
+            },
+            "column_exclude_list": {
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            }
+          },
+          "required": [
+            "name"
+          ]
+        },
+        "IcebergDestinationReq": {
+          "type": "object",
+          "properties": {
+            "namespace": {
+              "type": "string"
+            },
+            "qualify_namespace": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "namespace"
+          ]
+        },
+        "CdcDestinationReq": {
+          "type": "object",
+          "properties": {
+            "iceberg": {
+              "$ref": "#/components/schemas/IcebergDestinationReq"
+            }
+          }
+        },
+        "CreateCdcInputReq": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "plugin": {
+              "type": "string"
+            },
+            "tables": {
+              "type": "array",
+              "items": {
+                "$ref": "#/components/schemas/CdcTableReq"
+              }
+            },
+            "read_only": {
+              "type": "boolean"
+            },
+            "destination": {
+              "$ref": "#/components/schemas/CdcDestinationReq"
+            }
+          },
+          "required": [
+            "name",
+            "plugin",
+            "tables"
+          ]
+        },
         "ICreatePipelineV2Req": {
           "type": "object",
           "properties": {
@@ -10780,6 +11368,9 @@
             "properties": {
               "type": "object"
             },
+            "config": {
+              "type": "object"
+            },
             "connector_name": {
               "type": "string"
             },
@@ -10844,6 +11435,9 @@
             "properties": {
               "type": "object"
             },
+            "config": {
+              "type": "object"
+            },
             "connector_name": {
               "type": "string"
             },
@@ -11934,6 +12528,24 @@
             "schema"
           ]
         },
+        "ConnectionTestListTablesEntry": {
+          "type": "object",
+          "properties": {
+            "table_name": {
+              "type": "string"
+            },
+            "primary_keys": {
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            }
+          },
+          "required": [
+            "table_name",
+            "primary_keys"
+          ]
+        },
         "ConnectionTestListTablesRes": {
           "type": "object",
           "properties": {
@@ -11945,11 +12557,18 @@
               "items": {
                 "type": "string"
               }
+            },
+            "tables": {
+              "type": "array",
+              "items": {
+                "$ref": "#/components/schemas/ConnectionTestListTablesEntry"
+              }
             }
           },
           "required": [
             "operation_result",
-            "table_list"
+            "table_list",
+            "tables"
           ]
         },
         "GetTableMetadataReq": {
@@ -12040,6 +12659,62 @@
             "tables_metadata"
           ]
         },
+        "ValidateCdcPrerequisitesReq": {
+          "type": "object",
+          "properties": {
+            "plugin": {
+              "type": "string"
+            },
+            "connection_id": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "plugin",
+            "connection_id"
+          ]
+        },
+        "CdcCheckDto": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "expected": {
+              "type": "string"
+            },
+            "actual": {
+              "type": "string"
+            },
+            "passed": {
+              "type": "boolean"
+            }
+          },
+          "required": [
+            "name",
+            "expected",
+            "actual",
+            "passed"
+          ]
+        },
+        "ValidateCdcPrerequisitesRes": {
+          "type": "object",
+          "properties": {
+            "operation_result": {
+              "type": "boolean"
+            },
+            "checks": {
+              "type": "array",
+              "items": {
+                "$ref": "#/components/schemas/CdcCheckDto"
+              }
+            }
+          },
+          "required": [
+            "operation_result",
+            "checks"
+          ]
+        },
         "RefreshCatalogReq": {
           "type": "object",
           "properties": {
@@ -12052,7 +12727,10 @@
                 "oracle",
                 "mysql",
                 "postgresql",
-                "sqlserver"
+                "sqlserver",
+                "mysql_cdc",
+                "postgresql_cdc",
+                "oracle_cdc"
               ]
             }
           },
@@ -12096,7 +12774,10 @@
                 "oracle",
                 "mysql",
                 "postgresql",
-                "sqlserver"
+                "sqlserver",
+                "mysql_cdc",
+                "postgresql_cdc",
+                "oracle_cdc"
               ]
             },
             "session_id": {

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.19",
+        "@dadosfera/protospack-v2": "^3.40.0-beta.20",
         "@grpc/grpc-js": "^1.9.3",
         "@grpc/proto-loader": "^0.7.9",
         "@nestjs/cli": "^9.5.0",
@@ -1735,9 +1735,9 @@
       }
     },
     "node_modules/@dadosfera/protospack-v2": {
-      "version": "3.40.0-beta.19",
-      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.19.tgz",
-      "integrity": "sha512-RT7BYWHD2aD945FXzDXHa/xM+sD2B2Ypepy/W/WS+RwKYTSMCfuUXGzK2lXcd9HP9+U+Zjfh90Lkktl++8enjg==",
+      "version": "3.40.0-beta.20",
+      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.20.tgz",
+      "integrity": "sha512-A12jgcVMCylfXZyXZYLuZNFJBuEBV1ZYmo3w01qhemKrFAzXRjOnPTBGHngdWNHahIaEYdocieBoIzl0n6OvFw==",
       "license": "ISC",
       "dependencies": {
         "@grpc/grpc-js": "^1.9.3",

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.19",
+    "@dadosfera/protospack-v2": "^3.40.0-beta.20",
     "@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/guards/pipeline-execution.guard.spec.ts

@@ -0,0 +1,71 @@
+import { BadRequestException } from '@nestjs/common';
+import { PipelineExecutionGuard } from './pipeline-execution.guard';
+
+const logger = { info: jest.fn(), error: jest.fn() };
+
+function buildGuard(proxyImpl: jest.Mock) {
+  const platformApiService: any = { proxy: proxyImpl };
+  return new PipelineExecutionGuard(
+    { logger } as any,
+    platformApiService,
+  );
+}
+
+function contextWith(pipelineId = 'abc-123') {
+  return {
+    switchToHttp: () => ({
+      getRequest: () => ({ params: { pipelineId }, user: {} }),
+    }),
+  } as any;
+}
+
+describe('PipelineExecutionGuard', () => {
+  afterEach(() => jest.clearAllMocks());
+
+  it('allows the edit when the pipeline has no run history (empty array)', async () => {
+    const guard = buildGuard(jest.fn().mockResolvedValue([]));
+    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  });
+
+  it('allows the edit when the last run has no last_status', async () => {
+    const guard = buildGuard(jest.fn().mockResolvedValue([{}]));
+    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  });
+
+  it('allows the edit when the pipeline is not running', async () => {
+    const guard = buildGuard(
+      jest.fn().mockResolvedValue([{ last_status: 'SUCCEEDED' }]),
+    );
+    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  });
+
+  it('blocks with the is-running message when the pipeline is running', async () => {
+    const guard = buildGuard(
+      jest.fn().mockResolvedValue([{ last_status: 'RUNNING' }]),
+    );
+    await expect(guard.canActivate(contextWith())).rejects.toThrow(
+      'Pipeline is running, cannot update input now',
+    );
+  });
+
+  it('wraps a genuine status-check failure (fail closed)', async () => {
+    const guard = buildGuard(
+      jest.fn().mockRejectedValue(new Error('platform down')),
+    );
+    await expect(guard.canActivate(contextWith())).rejects.toThrow(
+      'Error checking pipeline status: platform down',
+    );
+  });
+
+  it('does not double-wrap the is-running BadRequestException', async () => {
+    const guard = buildGuard(
+      jest.fn().mockResolvedValue([{ last_status: 'running' }]),
+    );
+    await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf(
+      BadRequestException,
+    );
+    await expect(guard.canActivate(contextWith())).rejects.not.toThrow(
+      /Error checking pipeline status/,
+    );
+  });
+});

src/guards/pipeline-execution.guard.ts

@@ -46,10 +46,16 @@ export class PipelineExecutionGuard implements CanActivate {
         user,
       );
 
-      const currentStatus = status[status.length - 1]
+      const currentStatus = status?.[status.length - 1];
 
       this.logger.info('Pipeline current status response:' +  JSON.stringify(currentStatus));
-  
+
+      // No run history (e.g. CDC pipelines never record batch runs) means
+      // nothing is executing — allow the edit rather than crash on .last_status.
+      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');
@@ -57,6 +63,11 @@ export class PipelineExecutionGuard implements CanActivate {
         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/connection-test/connection-test.controller.ts

@@ -22,6 +22,8 @@ import {
   ConnectionTestListTablesRes,
   GetTableMetadataRes,
   GetTableMetadataReq,
+  ValidateCdcPrerequisitesReq,
+  ValidateCdcPrerequisitesRes,
   RefreshCatalogReq,
   RefreshCatalogRes,
   RefreshCatalogStatusReq,
@@ -129,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/inputs/dtos/input.model.ts

@@ -65,3 +65,62 @@ 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;
+  @ApiPropertyOptional()
+  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

@@ -15,6 +15,7 @@ import { Input } from '@dadosfera/protospack-v2';
 import {
   GetAvailableEntitiesRequest,
   InputCreateGenericRequest,
+  InputCreateCdcRequest,
   InputCreateS3Request,
   InputNewCreateRequest,
   InputUpdateResponse,
@@ -22,7 +23,7 @@ import {
   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 { Metadata } from '@grpc/grpc-js';
 
 @Injectable()
@@ -183,6 +184,35 @@ 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((t) => ({
+          table_schema: t.table_schema,
+          table_name: t.name,
+          name: t.name, // canonical identity == table_name (in-factory also backfills)
+          primary_keys: t.primary_keys ?? [],
+          iceberg_table_name: t.iceberg_table_name,
+          iceberg_qualify_table_name: t.iceberg_qualify_table_name,
+          columns: t.columns ?? [],
+          column_exclude_list: t.column_exclude_list ?? [],
+        })),
+        destination: body.destination,
+      },
+      info,
+    };
+
+    const { input } = await lastValueFrom(
+      this.inputWriteService.InputCreateCdc(inputCreateCdcRequest),
+    );
+    return { input };
+  }
+
   async getAvailableEntities(data: GetAvailableEntitiesRequest) {
     return lastValueFrom(this.inputReadService.GetAvailableEntities(data));
   }
@@ -297,4 +327,12 @@ export class InputsService {
   async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
     return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data));
   }
+
+  async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) {
+    return lastValueFrom(this.inputWriteService.AddCdcTable(data as any));
+  }
+
+  async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) {
+    return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data));
+  }
 }

src/modules/pipelinesV2/interfaces.ts

@@ -31,6 +31,8 @@ export class IPipelineV2 {
   tags?: string[];
   @ApiPropertyOptional()
   properties?: any;
+  @ApiPropertyOptional()
+  config?: any;
 
   @ApiProperty()
   connector_name: string;

src/modules/pipelinesV2/pipelines.controller.ts

@@ -510,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

@@ -19,7 +19,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 +105,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 +164,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 +437,7 @@ export class PipelinesService implements OnModuleInit {
     });
 
     this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
+    const isCdc = !!oldInput.plugin?.endsWith('_cdc');
     const pipelineIdFormat = pipelineId.split('-').join('_');
     const rollback: RollbackPromise[] = [];
 
@@ -437,17 +498,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 +727,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/platform-api.controller.spec.ts

@@ -0,0 +1,372 @@
+// 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('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));
+
+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 { ElasticsearchService } from '../../services/elasticsearch';
+import { DynamoDBService } from '../../services/dynamodb';
+import { CustomersService } from '../customers/customers.service';
+import { CatalogService } from '../catalog/catalog.service';
+import { InputsService } from '../inputs/inputs.service';
+import { PipelinesService } from '../pipelinesV2/pipelines.service';
+
+const logger = {
+  info: (...args) => args,
+  error: (...args) => args,
+};
+
+const mockUser: any = {
+  customer_id: 'c1',
+  customer_name: 'cust',
+  user_id: 'u1',
+};
+
+describe('PlatformApiController - deleteTable', () => {
+  let controller: PlatformApiController;
+  let platformApiService: { proxy: jest.Mock };
+  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };
+
+  beforeEach(async () => {
+    platformApiService = { proxy: jest.fn() };
+    inputsService = {
+      markTableDeleted: jest.fn(),
+      unmarkTableDeleted: jest.fn(),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [PlatformApiController],
+      providers: [
+        { provide: PlatformApiService, useValue: platformApiService },
+        { provide: ElasticsearchService, useValue: {} },
+        { provide: DynamoDBService, useValue: {} },
+        { provide: CustomersService, useValue: {} },
+        { provide: CatalogService, useValue: {} },
+        { provide: InputsService, useValue: inputsService },
+        { provide: PipelinesService, useValue: {} },
+        { provide: DadosferaLogger, useValue: { logger } },
+      ],
+    }).compile();
+
+    controller = module.get<PlatformApiController>(PlatformApiController);
+  });
+
+  it('CDC removal reconfigures the connector', async () => {
+    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
+    platformApiService.proxy.mockImplementation((method: string, path: string) => {
+      if (method === 'GET') {
+        return Promise.resolve({
+          jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } }],
+        });
+      }
+      return Promise.resolve({});
+    });
+
+    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);
+
+    expect(platformApiService.proxy).toHaveBeenCalledWith(
+      'DELETE',
+      '/pipeline/pid/jobs',
+      mockUser,
+      { job_ids: ['p_0'], delete_snowflake_tables: false },
+    );
+    const deletedViaJobsRoute = platformApiService.proxy.mock.calls.some(
+      ([method, path]: any[]) => method === 'DELETE' && path === '/jobs/p_0',
+    );
+    expect(deletedViaJobsRoute).toBe(false);
+  });
+
+  it('batch removal unchanged', async () => {
+    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
+    platformApiService.proxy.mockImplementation((method: string, path: string) => {
+      if (method === 'GET') {
+        return Promise.resolve({
+          jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 'pedidos' } }],
+        });
+      }
+      return Promise.resolve({});
+    });
+
+    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);
+
+    expect(platformApiService.proxy).toHaveBeenCalledWith(
+      'DELETE',
+      '/jobs/p_0',
+      mockUser,
+    );
+    const reconfiguredConnector = platformApiService.proxy.mock.calls.some(
+      ([method, path]: any[]) => method === 'DELETE' && path === '/pipeline/pid/jobs',
+    );
+    expect(reconfiguredConnector).toBe(false);
+  });
+});
+
+describe('PlatformApiController - addTable', () => {
+  let controller: PlatformApiController;
+  let inputsService: { addCdcTable: jest.Mock; removeCdcTable: jest.Mock };
+  let pipelinesClientService: { addCdcJobs: jest.Mock };
+
+  const body = {
+    table_name: 'orders',
+    table_schema: 'public',
+    primary_keys: ['id'],
+    destinations: {
+      raw: { table_schema: 'raw', table_name: 'orders' },
+      qualify: { table_schema: 'qualify', table_name: 'orders' },
+    },
+  };
+
+  beforeEach(async () => {
+    inputsService = {
+      addCdcTable: jest.fn(),
+      removeCdcTable: jest.fn(),
+    };
+    pipelinesClientService = {
+      addCdcJobs: jest.fn(),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [PlatformApiController],
+      providers: [
+        { provide: PlatformApiService, useValue: { proxy: jest.fn() } },
+        { provide: ElasticsearchService, useValue: {} },
+        { provide: DynamoDBService, useValue: {} },
+        { provide: CustomersService, useValue: {} },
+        { provide: CatalogService, useValue: {} },
+        { provide: InputsService, useValue: inputsService },
+        { provide: PipelinesService, useValue: pipelinesClientService },
+        { provide: DadosferaLogger, useValue: { logger } },
+      ],
+    }).compile();
+
+    controller = module.get<PlatformApiController>(PlatformApiController);
+  });
+
+  it('addTable: DynamoDB append then platform AddJobs, returns job_ids', async () => {
+    inputsService.addCdcTable.mockResolvedValue({ input: {} });
+    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
+
+    const result = await controller.addTable('pid', 'iid', body, mockUser);
+
+    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
+      id: 'iid',
+      table: {
+        table_schema: 'public',
+        table_name: 'orders',
+        primary_keys: ['id'],
+        name: 'orders',
+        iceberg_table_name: undefined,
+        iceberg_qualify_table_name: undefined,
+        columns: [],
+        column_exclude_list: [],
+      },
+      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
+    });
+    expect(pipelinesClientService.addCdcJobs).toHaveBeenCalledWith({
+      pipeline_id: 'pid',
+      input_id: 'iid',
+      tables: [{
+        table_schema: 'public',
+        table_name: 'orders',
+        primary_keys: ['id'],
+        destinations: body.destinations,
+      }],
+      info: { customer_id: 'c1', user_id: 'u1', customer: 'cust' },
+    });
+    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
+    expect(inputsService.removeCdcTable).not.toHaveBeenCalled();
+
+    const addCdcTableOrder = inputsService.addCdcTable.mock.invocationCallOrder[0];
+    const addCdcJobsOrder = pipelinesClientService.addCdcJobs.mock.invocationCallOrder[0];
+    expect(addCdcTableOrder).toBeLessThan(addCdcJobsOrder);
+  });
+
+  it('addTable: carries iceberg_table_name on the added table through to AddCdcTable', async () => {
+    inputsService.addCdcTable.mockResolvedValue({ input: {} });
+    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] });
+
+    const icebergBody = { ...body, iceberg_table_name: 'cdc_raw.public__orders' };
+
+    await controller.addTable('pid', 'iid', icebergBody, mockUser);
+
+    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
+      id: 'iid',
+      table: {
+        table_schema: 'public',
+        table_name: 'orders',
+        primary_keys: ['id'],
+        name: 'orders',
+        iceberg_table_name: 'cdc_raw.public__orders',
+        iceberg_qualify_table_name: undefined,
+        columns: [],
+        column_exclude_list: [],
+      },
+      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
+    });
+  });
+
+  it('addTable: carries columns on the added table through to AddCdcTable', async () => {
+    inputsService.addCdcTable.mockResolvedValue({ input: {} });
+    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_4'], skipped: [] });
+
+    const columnsBody = {
+      ...body,
+      columns: [
+        { name: 'id', type: 'int', is_primary_key: true },
+        { name: 'descr', type: 'varchar(255)', is_primary_key: false },
+      ],
+    };
+
+    await controller.addTable('pid', 'iid', columnsBody, mockUser);
+
+    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
+      id: 'iid',
+      table: {
+        table_schema: 'public',
+        table_name: 'orders',
+        primary_keys: ['id'],
+        name: 'orders',
+        iceberg_table_name: undefined,
+        iceberg_qualify_table_name: undefined,
+        columns: [
+          { name: 'id', type: 'int', is_primary_key: true },
+          { name: 'descr', type: 'varchar(255)', is_primary_key: false },
+        ],
+        column_exclude_list: [],
+      },
+      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
+    });
+  });
+
+  it('addTable: rolls back the DynamoDB row when AddJobs fails', async () => {
+    inputsService.addCdcTable.mockResolvedValue({ input: {} });
+    pipelinesClientService.addCdcJobs.mockRejectedValue(new Error('platform down'));
+    inputsService.removeCdcTable.mockResolvedValue({});
+
+    await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down');
+
+    expect(inputsService.removeCdcTable).toHaveBeenCalledWith({
+      id: 'iid',
+      table_name: 'orders',
+      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
+    });
+  });
+});
+
+
+describe('PlatformApiController - deleteTables (batch)', () => {
+  let controller: PlatformApiController;
+  let platformApiService: { proxy: jest.Mock };
+  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };
+
+  beforeEach(async () => {
+    platformApiService = { proxy: jest.fn() };
+    inputsService = {
+      markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }),
+      unmarkTableDeleted: jest.fn().mockResolvedValue({}),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [PlatformApiController],
+      providers: [
+        { provide: PlatformApiService, useValue: platformApiService },
+        { provide: ElasticsearchService, useValue: {} },
+        { provide: DynamoDBService, useValue: {} },
+        { provide: CustomersService, useValue: {} },
+        { provide: CatalogService, useValue: {} },
+        { provide: InputsService, useValue: inputsService },
+        { provide: PipelinesService, useValue: {} },
+        { provide: DadosferaLogger, useValue: { logger } },
+      ],
+    }).compile();
+
+    controller = module.get<PlatformApiController>(PlatformApiController);
+  });
+
+  const pipelineWithJobs = () => ({
+    jobs: [
+      { job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } },
+      { job_id: 'p_1', input: { connector: 'cdc', table_name: 'clientes' } },
+      { job_id: 'p_2', input: { connector: 'cdc', table_name: 'produtos' } },
+    ],
+  });
+
+  it('removes N tables in ONE platform call (connectors reconfigured once)', async () => {
+    platformApiService.proxy.mockImplementation((method: string) =>
+      Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}),
+    );
+
+    await controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'produtos'] }, mockUser);
+
+    // one mark per table
+    expect(inputsService.markTableDeleted).toHaveBeenCalledTimes(2);
+
+    // exactly one DELETE to the batch endpoint, with BOTH job_ids
+    const deleteCalls = platformApiService.proxy.mock.calls.filter(
+      ([m, p]: any[]) => m === 'DELETE' && p === '/pipeline/pid/jobs',
+    );
+    expect(deleteCalls).toHaveLength(1);
+    expect(deleteCalls[0][3]).toEqual({
+      job_ids: ['p_0', 'p_2'],
+      delete_snowflake_tables: false,
+    });
+    // never the per-job route
+    const perJob = platformApiService.proxy.mock.calls.some(
+      ([m, p]: any[]) => m === 'DELETE' && String(p).startsWith('/jobs/'),
+    );
+    expect(perJob).toBe(false);
+  });
+
+  it('rolls back only this call\'s marks when the platform delete fails', async () => {
+    platformApiService.proxy.mockImplementation((method: string) => {
+      if (method === 'GET') return Promise.resolve(pipelineWithJobs());
+      return Promise.reject(new Error('platform boom'));
+    });
+
+    await expect(
+      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser),
+    ).rejects.toThrow();
+
+    // both marks rolled back, nothing else
+    expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(2);
+    const unmarked = inputsService.unmarkTableDeleted.mock.calls.map((c: any[]) => c[0].table_name).sort();
+    expect(unmarked).toEqual(['clientes', 'pedidos']);
+  });
+
+  it('rolls back the marks made so far if a later mark fails (atomic)', async () => {
+    // second mark fails → first must be rolled back, no platform delete attempted
+    inputsService.markTableDeleted
+      .mockResolvedValueOnce({ is_deleted: true, deleted_at: 't' })
+      .mockRejectedValueOnce(new Error('dynamo boom'));
+
+    await expect(
+      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser),
+    ).rejects.toThrow();
+
+    expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(1);
+    expect(inputsService.unmarkTableDeleted.mock.calls[0][0].table_name).toBe('pedidos');
+    // never reached the platform delete
+    const attemptedDelete = platformApiService.proxy.mock.calls.some(([m]: any[]) => m === 'DELETE');
+    expect(attemptedDelete).toBe(false);
+  });
+
+  it('404s when a requested table has no matching job', async () => {
+    platformApiService.proxy.mockImplementation((method: string) =>
+      Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}),
+    );
+
+    await expect(
+      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'ghost'] }, mockUser),
+    ).rejects.toThrow();
+    // the successful mark (pedidos) must be rolled back
+    expect(inputsService.unmarkTableDeleted).toHaveBeenCalled();
+  });
+});

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

@@ -33,6 +33,7 @@ import { CatalogService } from '../catalog/catalog.service';
 import { PackTheMetadata } from '../../utils/PackTheMetadata';
 import { ValidationTableDTO } from './platform-api.dto';
 import { InputsService } from '../inputs/inputs.service';
+import { PipelinesService } from '../pipelinesV2/pipelines.service';
 import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
 
 
@@ -61,6 +62,7 @@ export class PlatformApiController {
     private readonly customersService: CustomersService,
     private readonly catalogService: CatalogService,
     private readonly inputsService: InputsService,
+    private readonly pipelinesClientService: PipelinesService,
     @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
   ) {
     this.logger = dadosferaLogger.logger;
@@ -545,6 +547,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)
@@ -1002,7 +1018,18 @@ export class PlatformApiController {
       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);
+      if (job.input?.connector === 'cdc') {
+        // CDC: reconfigure the Debezium/Kafka-Connect connector (stop
+        // replicating this table); keep the landed Snowflake data.
+        await this.platformApiService.proxy(
+          'DELETE',
+          `/pipeline/${normalizedPipelineId}/jobs`,
+          user,
+          { job_ids: [job.job_id], delete_snowflake_tables: false },
+        );
+      } else {
+        await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user);
+      }
       this.logger.info('deleteTable: job deleted', { jobId: job.job_id });
 
       return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at };
@@ -1017,6 +1044,155 @@ export class PlatformApiController {
     }
   }
 
+  @Delete('pipelines/:pipelineId/inputs/:inputId/tables')
+  @ApiOperation({ summary: 'Batch-remove tables from an input; reconfigures the CDC connectors once' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
+  @UseGuards(PipelineExecutionGuard)
+  async deleteTables(
+    @Param('pipelineId') pipelineId: string,
+    @Param('inputId') inputId: string,
+    @Body() body: { table_names: string[] },
+    @User() user: RequestUser,
+  ) {
+    const tableNames = body.table_names ?? [];
+    if (tableNames.length === 0) {
+      throw new BadRequestException('table_names must be a non-empty array');
+    }
+    const info = {
+      customer_id: user.customer_id,
+      customer: user.customer_name,
+      user_id: user.user_id,
+    };
+
+    // 1) Soft-delete each table in DynamoDB, tracking which succeeded so a later
+    //    failure only rolls back the marks made in THIS call.
+    const marked: string[] = [];
+    const rollback = async () => {
+      for (const name of marked) {
+        try {
+          await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info });
+        } catch (rollbackError) {
+          this.logger.error('deleteTables: rollback failed', { tableName: name, error: rollbackError.message });
+        }
+      }
+    };
+
+    try {
+      for (const name of tableNames) {
+        await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info });
+        marked.push(name);
+      }
+
+      // 2) Resolve all table_names -> job_ids from the platform pipeline (one GET).
+      const normalizedPipelineId = this.normalizePipelineId(pipelineId);
+      const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
+      const jobs = platformPipeline?.jobs ?? [];
+
+      const jobIds: string[] = [];
+      for (const name of tableNames) {
+        const job = jobs.find((j: any) => j.input?.table_name === name);
+        if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`);
+        jobIds.push(job.job_id);
+      }
+
+      // 3) Remove them all in ONE platform call so the Debezium source + Snowflake
+      //    sink connectors are reconfigured a single time, not once per table.
+      await this.platformApiService.proxy(
+        'DELETE',
+        `/pipeline/${normalizedPipelineId}/jobs`,
+        user,
+        { job_ids: jobIds, delete_snowflake_tables: false },
+      );
+      this.logger.info('deleteTables: jobs deleted', { jobIds });
+
+      return { table_names: tableNames, deleted: true };
+    } catch (error) {
+      this.logger.error('deleteTables: failed, rolling back this call\'s marks', { tableNames, error: error.message });
+      await rollback();
+      throw error;
+    }
+  }
+
+  @Post('pipelines/:pipelineId/inputs/:inputId/tables')
+  @ApiOperation({ summary: 'Add a CDC table to an input and dispatch its platform jobs' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  @UseGuards(PipelineExecutionGuard)
+  async addTable(
+    @Param('pipelineId') pipelineId: string,
+    @Param('inputId') inputId: string,
+    @Body() body: {
+      table_name: string;
+      table_schema: string;
+      primary_keys: string[];
+      destinations: {
+        raw: { table_schema: string; table_name: string };
+        qualify: { table_schema: string; table_name: string };
+      };
+      // Iceberg destination only (protospack CdcTable.iceberg_table_name);
+      // absent for snowflake, back-compat.
+      iceberg_table_name?: string;
+      // Per-table deduped (qualify) Iceberg table name (protospack
+      // CdcTable.iceberg_qualify_table_name); absent => same as the raw name.
+      iceberg_qualify_table_name?: string;
+      // Source column schema for iceberg deduped table pre-create (protospack CdcTable.columns).
+      columns?: { name: string; type: string; is_primary_key: boolean }[];
+      // Columns the user chose to ignore -> Debezium column.exclude.list.
+      column_exclude_list?: string[];
+    },
+    @User() user: RequestUser,
+  ) {
+    const info = {
+      customer_id: user.customer_id,
+      customer: user.customer_name,
+      user_id: user.user_id,
+    };
+
+    const cdcTable = {
+      table_schema: body.table_schema,
+      table_name: body.table_name,
+      primary_keys: body.primary_keys,
+      name: body.table_name,
+      iceberg_table_name: body.iceberg_table_name,
+      iceberg_qualify_table_name: body.iceberg_qualify_table_name,
+      columns: body.columns ?? [],
+      column_exclude_list: body.column_exclude_list ?? [],
+    };
+
+    this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name });
+    await this.inputsService.addCdcTable({ id: inputId, table: cdcTable, info });
+    this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name });
+
+    try {
+      const customInfo = {
+        customer_id: user.customer_id,
+        user_id: user.user_id,
+        customer: user.customer_name,
+      };
+
+      const res = await this.pipelinesClientService.addCdcJobs({
+        pipeline_id: pipelineId,
+        input_id: inputId,
+        tables: [{
+          table_schema: body.table_schema,
+          table_name: body.table_name,
+          primary_keys: body.primary_keys,
+          destinations: body.destinations,
+        }],
+        info: customInfo,
+      });
+
+      return res;
+    } catch (error) {
+      this.logger.error('addTable: platform AddJobs failed, rolling back the DynamoDB row', { tableName: body.table_name, error: error.message });
+      try {
+        await this.inputsService.removeCdcTable({ id: inputId, table_name: body.table_name, info });
+      } catch (rbErr) {
+        this.logger.error('addTable: rollback failed', { error: rbErr.message });
+      }
+      throw error;
+    }
+  }
+
   // ==================== JOBS - JDBC SYNC MODE ROUTES ====================
 
   @Get('jobs/jdbc/configs/allowed_datatypes')

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

@@ -1,4 +1,4 @@
-import { Module } from '@nestjs/common';
+import { Module, forwardRef } from '@nestjs/common';
 
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 
@@ -9,9 +9,17 @@ 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],
   exports: [PlatformApiService],

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