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

FIX: CDC live status decodes int64 offsets as numbers
FIX: address review round 2 on the CDC refactor
2026-09-01 04:08:16 +00:00 · 2026-08-31 17:07:20 -03:00 · 2026-08-31 16:09:25 -03:00 · 2026-08-31 12:52:51 -03:00 · 2026-08-29 13:01:35 -03:00 · 2026-08-29 12:57:49 -03:00
82 changed files with 6726 additions and 1941 deletions
@@ -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:
@@ -71,6 +71,11 @@ jobs:
          sudo mv helmfile /usr/local/bin/
          helmfile --version

+      - name: Install Helm Diff plugin
+        run: |
+          helm plugin install https://github.com/databus23/helm-diff --version v3.9.3
+          helm diff version
+
      - name: Debug Helm env
        run: |
          helm env
@@ -102,4 +107,5 @@ jobs:
      - name: Run Helmfile Diff
        env:
          ENV: ${{ needs.extract_environment.outputs.environment }}
+          HELM_PLUGINS: /home/runner/.local/share/helm/plugins
        run: helmfile -f deploy/helmfiles/${ENV}.yaml diff
@@ -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.
@@ -1,5 +1,5 @@
 FROM node:20-alpine AS base_image
-RUN npm install -g npm@latest
+RUN npm install -g npm@10.8.2

 FROM base_image AS build_base
 WORKDIR /app
@@ -22,13 +22,14 @@ ENV PUPPETEER_SKIP_CHROMIUM_DOWNLOAD=true \
 # run aws cli without mounting secret, because CI already has AWS credentials
 FROM build_base AS ci_image
 RUN aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1
-RUN npm ci
+RUN npm ci --ignore-scripts
 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"]


@@ -37,7 +38,7 @@ FROM build_base AS dev
 RUN --mount=type=secret,id=aws,target=/root/.aws/credentials \
  aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1
 # flag --build-from-source is required to force-build sqlite3
-RUN npm ci
+RUN npm ci --ignore-scripts
 COPY . .
 ENTRYPOINT npm run start:dev

@@ -22,7 +22,7 @@ ENV PUPPETEER_SKIP_CHROMIUM_DOWNLOAD=true \
 FROM build_base AS build
 RUN --mount=type=secret,id=aws,target=/root/.aws/credentials \
  aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1
-RUN npm ci
+RUN  npm ci --ignore-scripts
 COPY . .
 RUN npm run build

@@ -4,6 +4,7 @@

 # Maestro

+
 Maestro é a API principal da Dadosfera. É responsável pela comunicação do Frontend com nossos microsserviços.

 ```mermaid
@@ -111,8 +111,12 @@ spec:
              value: "{{ .Values.maestro.redis_tls }}"
            - name: PLATFORM_API_URL
              value: {{ .Values.maestro.platform_api_url }}
+            - name: CONNECTIONS_API_URL
+              value: {{ .Values.maestro.connections_api_url | default "" | quote }}
            - name: STORAGE_EXPLORER_API_URL
              value: {{ .Values.maestro.storage_explorer_api_url | quote }}
+            - name: FIREBASE_BASE_URL
+              value: {{ .Values.maestro.firebase_base_url }}
            - name: JWT_PRIVATE_KEY
              valueFrom:
                secretKeyRef:
@@ -9,7 +9,9 @@ maestro:
  cookie_secret: "ff7bc13823edb2ae50d248e5780bddc9d4b31c36"
  redis_database: "1"
  platform_api_url: https://xs2hkhq07k.execute-api.us-east-1.amazonaws.com
+  connections_api_url: https://iy40eans64.execute-api.us-east-1.amazonaws.com
  storage_explorer_api_url: "http://storage-explorer-{customer}.data-apps.svc.cluster.local:8000/api"
+  firebase_base_url: https://feature-flag-25bf6-default-rtdb.firebaseio.com/stg

 hostname: maestro.stg.dadosfera.ai

@@ -55,6 +55,7 @@ maestro:
  redis_database: "0"
  redis_tls: "true"
  cookie_secret: "13cc5e136d3074bcc05bec8697092ec1f5f376bf"
+  firebase_base_url: https://feature-flag-25bf6-default-rtdb.firebaseio.com/prd
 autoscaling:
  enabled: false
  minReplicas: 1
@@ -16,8 +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": "2.5.3",
-        "@dadosfera/protospack-v2": "^3.38.0-beta.26",
+        "@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",
@@ -31,7 +30,7 @@
        "@nestjs/schematics": "^9.2.0",
        "@nestjs/swagger": "^6.3.0",
        "@nestjs/testing": "^9.4.3",
-        "axios": "^0.30.2",
+        "axios": "0.30.3",
        "cache-manager": "^5.1.4",
        "cache-manager-ioredis-yet": "^1.1.0",
        "class-transformer": "^0.5.1",
@@ -47,6 +46,7 @@
        "jwk-to-pem": "^2.0.5",
        "mixpanel": "^0.17.0",
        "ms": "^3.0.0-canary.1",
+        "multer": "^2.0.2",
        "openid-client": "^5.7.1",
        "passport": "^0.6.0",
        "passport-facebook": "^3.0.0",
@@ -1734,19 +1734,11 @@
        "winston-log2gelf": "^2.4.0"
      }
    },
-    "node_modules/@dadosfera/protospack": {
-      "version": "2.5.3",
-      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack/-/protospack-2.5.3.tgz",
-      "integrity": "sha512-yOLnd+s6n9VkPpZXO8HnUY27CQPHj/qs+ecddviA4Ldn0Gx4KGRgbVdsSSP45nPm0GHhCd2bHg4ap+la7xtRmA==",
-      "license": "ISC",
-      "dependencies": {
-        "rxjs": "^7.5.5"
-      }
-    },
    "node_modules/@dadosfera/protospack-v2": {
-      "version": "3.38.0-beta.26",
-      "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.26.tgz",
-      "integrity": "sha512-N8NS7+djLGy0wJXk00+4oupqd/wBIQ1f+YBBhK2y9x4guFXYK1KWIrPZhPj+gaU8g2KNkqKoT7SnE9PXNxlLSQ==",
+      "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"
@@ -2927,6 +2919,20 @@
        "node": ">= 0.6"
      }
    },
+    "node_modules/@nestjs/platform-express/node_modules/concat-stream": {
+      "version": "1.6.2",
+      "resolved": "https://registry.npmjs.org/concat-stream/-/concat-stream-1.6.2.tgz",
+      "integrity": "sha512-27HBghJxjiZtIk3Ycvn/4kbJk/1uZuJFfuPEns6LaEvpvG1f0hTea8lilrouyo9mVc2GWdcEZ8OLoGmSADlrCw==",
+      "engines": [
+        "node >= 0.8"
+      ],
+      "dependencies": {
+        "buffer-from": "^1.0.0",
+        "inherits": "^2.0.3",
+        "readable-stream": "^2.2.2",
+        "typedarray": "^0.0.6"
+      }
+    },
    "node_modules/@nestjs/platform-express/node_modules/content-disposition": {
      "version": "0.5.4",
      "resolved": "https://registry.npmjs.org/content-disposition/-/content-disposition-0.5.4.tgz",
@@ -3096,6 +3102,24 @@
      "integrity": "sha512-Tpp60P6IUJDTuOq/5Z8cdskzJujfwqfOTkrwIwj7IRISpnkJnT6SyJ4PCPnGMoFjC9ddhal5KVIYtAt97ix05A==",
      "license": "MIT"
    },
+    "node_modules/@nestjs/platform-express/node_modules/multer": {
+      "version": "1.4.4-lts.1",
+      "resolved": "https://registry.npmjs.org/multer/-/multer-1.4.4-lts.1.tgz",
+      "integrity": "sha512-WeSGziVj6+Z2/MwQo3GvqzgR+9Uc+qt8SwHKh3gvNPiISKfsMfG4SvCOFYlxxgkXt7yIV2i1yczehm0EOKIxIg==",
+      "deprecated": "Multer 1.x is impacted by a number of vulnerabilities, which have been patched in 2.x. You should upgrade to the latest 2.x version.",
+      "dependencies": {
+        "append-field": "^1.0.0",
+        "busboy": "^1.0.0",
+        "concat-stream": "^1.5.2",
+        "mkdirp": "^0.5.4",
+        "object-assign": "^4.1.1",
+        "type-is": "^1.6.4",
+        "xtend": "^4.0.0"
+      },
+      "engines": {
+        "node": ">= 6.0.0"
+      }
+    },
    "node_modules/@nestjs/platform-express/node_modules/negotiator": {
      "version": "0.6.3",
      "resolved": "https://registry.npmjs.org/negotiator/-/negotiator-0.6.3.tgz",
@@ -3120,6 +3144,25 @@
        "url": "https://github.com/sponsors/ljharb"
      }
    },
+    "node_modules/@nestjs/platform-express/node_modules/readable-stream": {
+      "version": "2.3.8",
+      "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-2.3.8.tgz",
+      "integrity": "sha512-8p0AUk4XODgIewSi0l8Epjs+EVnWiK7NoDIEGU0HhE7+ZyY8D1IMY7odu5lRrFXGg71L15KG8QrPmum45RTtdA==",
+      "dependencies": {
+        "core-util-is": "~1.0.0",
+        "inherits": "~2.0.3",
+        "isarray": "~1.0.0",
+        "process-nextick-args": "~2.0.0",
+        "safe-buffer": "~5.1.1",
+        "string_decoder": "~1.1.1",
+        "util-deprecate": "~1.0.1"
+      }
+    },
+    "node_modules/@nestjs/platform-express/node_modules/readable-stream/node_modules/safe-buffer": {
+      "version": "5.1.2",
+      "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",
+      "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g=="
+    },
    "node_modules/@nestjs/platform-express/node_modules/safe-buffer": {
      "version": "5.2.1",
      "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz",
@@ -3194,6 +3237,19 @@
        "node": ">= 0.8"
      }
    },
+    "node_modules/@nestjs/platform-express/node_modules/string_decoder": {
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.1.1.tgz",
+      "integrity": "sha512-n/ShnvDi6FHbbVfviro+WojiFzv+s8MPMHBczVePfUpDJLwoLT0ht1l4YwBCbi8pJAveEEdnkHyPyTP/mzRfwg==",
+      "dependencies": {
+        "safe-buffer": "~5.1.0"
+      }
+    },
+    "node_modules/@nestjs/platform-express/node_modules/string_decoder/node_modules/safe-buffer": {
+      "version": "5.1.2",
+      "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",
+      "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g=="
+    },
    "node_modules/@nestjs/platform-express/node_modules/tslib": {
      "version": "2.5.3",
      "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.5.3.tgz",
@@ -5533,9 +5589,9 @@
      }
    },
    "node_modules/axios": {
-      "version": "0.30.2",
-      "resolved": "https://registry.npmjs.org/axios/-/axios-0.30.2.tgz",
-      "integrity": "sha512-0pE4RQ4UQi1jKY6p7u6i1Tkzqmu+d+/tHS7Q7rKunWLB9WyilBTpHHpXzPNMDj5hTbK0B0PTLSz07yqMBiF6xg==",
+      "version": "0.30.3",
+      "resolved": "https://registry.npmjs.org/axios/-/axios-0.30.3.tgz",
+      "integrity": "sha512-5/tmEb6TmE/ax3mdXBc/Mi6YdPGxQsv+0p5YlciXWt3PHIn0VamqCXhRMtScnwY3lbgSXLneOuXAKUhgmSRpwg==",
      "license": "MIT",
      "dependencies": {
        "follow-redirects": "^1.15.4",
@@ -6508,7 +6564,6 @@
      "engines": [
        "node >= 6.0"
      ],
-      "license": "MIT",
      "dependencies": {
        "buffer-from": "^1.0.0",
        "inherits": "^2.0.3",
@@ -9127,6 +9182,11 @@
        "url": "https://github.com/sponsors/sindresorhus"
      }
    },
+    "node_modules/isarray": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/isarray/-/isarray-1.0.0.tgz",
+      "integrity": "sha512-VLghIWNM6ELQzo7zwmcg0NmTVyWKYjvIeM83yjp0wRDTmUnrM678fQbcKBo6n2CJEF0szoG//ytg+TKla89ALQ=="
+    },
    "node_modules/isexe": {
      "version": "2.0.0",
      "resolved": "https://registry.npmjs.org/isexe/-/isexe-2.0.0.tgz",
@@ -10724,7 +10784,6 @@
      "version": "2.0.2",
      "resolved": "https://registry.npmjs.org/multer/-/multer-2.0.2.tgz",
      "integrity": "sha512-u7f2xaZ/UG8oLXHvtF/oWTRvT44p9ecwBBqTwgJVq0+4BW1g8OW01TyMEGWBHbyMOYVHXslaut7qEQ1meATXgw==",
-      "license": "MIT",
      "dependencies": {
        "append-field": "^1.0.0",
        "busboy": "^1.6.0",
@@ -11727,6 +11786,11 @@
        "url": "https://github.com/chalk/ansi-styles?sponsor=1"
      }
    },
+    "node_modules/process-nextick-args": {
+      "version": "2.0.1",
+      "resolved": "https://registry.npmjs.org/process-nextick-args/-/process-nextick-args-2.0.1.tgz",
+      "integrity": "sha512-3ouUOpQhtgrbOa17J7+uxOTpITYWaGP7/AhoR3+A+/1e9skrzelGi/dXzEYyvbxubEF6Wn2ypscTKiKJFFn1ag=="
+    },
    "node_modules/process-warning": {
      "version": "1.0.0",
      "resolved": "https://registry.npmjs.org/process-warning/-/process-warning-1.0.0.tgz",
@@ -10,7 +10,7 @@
  },
  "scripts": {
    "co:login": "aws codeartifact login --tool npm --namespace @dadosfera --repository dadosfera-npm --domain dadosfera --domain-owner 611330257153 --region us-east-1",
-    "proto-update": "npm i @dadosfera/protospack-v2@latest --save-exact",
+    "proto-update": "npm i @dadosfera/protospack-v2@v3.40.0-beta.1 --save-exact",
    "prebuild": "rimraf dist",
    "build": "nest build",
    "format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"",
@@ -34,8 +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": "2.5.3",
-    "@dadosfera/protospack-v2": "^3.38.0-beta.26",
+    "@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",
@@ -49,7 +48,7 @@
    "@nestjs/schematics": "^9.2.0",
    "@nestjs/swagger": "^6.3.0",
    "@nestjs/testing": "^9.4.3",
-    "axios": "^0.30.2",
+    "axios": "0.30.3",
    "cache-manager": "^5.1.4",
    "cache-manager-ioredis-yet": "^1.1.0",
    "class-transformer": "^0.5.1",
@@ -65,6 +64,7 @@
    "jwk-to-pem": "^2.0.5",
    "mixpanel": "^0.17.0",
    "ms": "^3.0.0-canary.1",
+    "multer": "^2.0.2",
    "openid-client": "^5.7.1",
    "passport": "^0.6.0",
    "passport-facebook": "^3.0.0",
@@ -80,7 +80,7 @@
    "swagger-ui-express": "^4.6.3"
  },
  "overrides": {
-    "multer": "2.0.2",
+    "axios": "0.30.3",
    "form-data": "^4.0.4",
    "body-parser": "^1.20.3",
    "cross-spawn": "^7.0.5",
@@ -117,5 +117,8 @@
    "ts-node": "^10.9.1",
    "tsconfig-paths": "^3.14.2",
    "typescript": "^4.9.5"
+  },
+  "resolutions": {
+    "axios": "0.30.3"
  }
 }
@@ -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}"`);
@@ -17,7 +17,6 @@ import { ConnectionTestModule } from './modules/connection-test/connection-test.
 import { NetworkConfigModule } from './modules/network-config/network-config.module';
 import { InputsModule } from './modules/inputs/inputs.module';
 import { OauthModule } from './modules/oauth/oauth.module';
-import { PipelinesModule } from './modules/pipelines/pipelines.module';
 import { TransformationsModule } from './modules/transformations/transformations.module';
 import { HealthModule } from './modules/health/health.module';
 import { CatalogModule } from './modules/catalog/catalog.module';
@@ -35,6 +34,8 @@ import { ShareMetadataModule } from './modules/share-metadata/share-metadata.mod
 import { ApiKeyModule } from './modules/api-key/api-key.module';
 import { PlatformApiModule } from './modules/platform-api/platform-api.module';
 import { StorageExplorerModule } from './modules/storage-explorer/storage-explorer.module';
+import { ReleaseNoteModule } from './modules/release_note/release_note.module';
+

@Module({
  providers: [
@@ -58,7 +59,6 @@ import { StorageExplorerModule } from './modules/storage-explorer/storage-explor
    PermissionsModule,
    TermsOfUseModule,
    ConnectionTestModule,
-    PipelinesModule,
    TransformationsModule,
    UsersModule,
    RolesModule,
@@ -79,6 +79,7 @@ import { StorageExplorerModule } from './modules/storage-explorer/storage-explor
    StorageExplorerModule,
    //Always leave HealthModule last, so it is on the bottom of swagger
    HealthModule,
+    ReleaseNoteModule,
  ],
 })
 export class AppModule {}
@@ -153,6 +153,7 @@ export class AuthenticationGuard
      user_id: accessTokenPayload.user_id,
      username: accessTokenPayload.username,
      permissions: accessTokenPayload.permissions,
+      roles: accessTokenPayload.roles,
      customer_id: accessTokenPayload.customer_id,
      customer_name: accessTokenPayload.customer_name,
      customer_tier: accessTokenPayload.customer_tier,
@@ -11,6 +11,7 @@ export function extractUserFrom(aRawJwt: string) {
    user_id: payload.user_id,
    username: payload.username,
    permissions: payload.permissions,
+    roles: payload.roles,
    customer_id: payload.customer_id,
    customer_name: payload.customer_name,
    customer_tier: payload.customer_tier,
@@ -357,6 +357,16 @@ export const PERMISSIONS_GROUPS = {
          'es-es': 'Crear y editar atributos en el catálogo',
        },
      },
+      CERTIFY: {
+        seqid: 53,
+        claim: 'catalog:certify',
+        usage: PermissionUsages.PUBLIC,
+        name: {
+          'pt-br': 'Alterar o status de certificação dos Ativos',
+          'en-us': "Change Assets' certification status",
+          'es-es': 'Cambiar el estado de certificación de los Activos',
+        },
+      },
      DELETE: {
        seqid: 1,
        claim: 'catalog:delete',
@@ -712,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> = [
@@ -12,6 +12,7 @@ export interface RequestUser {
  customer_tier: string;
  access_token: string;
  customer_modules: string[];
+  roles: string[];
 }

 export const User: (options?: { required?: boolean }) => ParameterDecorator =
@@ -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',
+      );
+    });
+  });
+});
@@ -0,0 +1,90 @@
+import {
+  BadRequestException,
+  CanActivate,
+  ExecutionContext,
+  Inject,
+  Injectable,
+  OnModuleInit,
+} from '@nestjs/common';
+import { ClientGrpc } from '@nestjs/microservices';
+import { map, Observable } from 'rxjs';
+import { PackTheMetadata } from 'src/utils/PackTheMetadata';
+import {
+  ReadService,
+  ProtoServices,
+} from '@dadosfera/protospack-v2/dist/lib/PipelineV2';
+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 {
+  logger: DadosferaLogger;
+
+  constructor(
+    @Inject(DadosferaLogger)
+    dadosferaLogger: DadosferaLogger,
+    private readonly platformApiService: PlatformApiService,
+  ) {
+    this.logger = dadosferaLogger.logger;
+  }
+
+  async canActivate(context: ExecutionContext): Promise<boolean> {
+    try {
+      this.logger.info(
+        'PipelineExecutionGuard: Checking if pipeline can be executed...',
+      );
+      const request = context.switchToHttp().getRequest();
+      const pipelineId = request.params.pipelineId;
+      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];
+
+      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');
+      }
+
+      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);
+    }
+  }
+}
@@ -111,3 +111,4 @@ function configureSwagger(app: INestApplication) {
  );
 }
 bootstrap();
+
@@ -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';
@@ -478,6 +479,7 @@ export class AuthController {
  @Get('me')
  async getMe(@Req() req: Request, @Res() res: Response) {
    this.logger.info('GET /auth/me ')
+    this.logger.info(JSON.stringify(req.headers));

    // Check for API key header first
    const apiKey = req.get('X-Api-key');
@@ -485,14 +487,16 @@ 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,
        customer: {
          id: api_key.customer_id,
          name: api_key.customer_name,
          tier: api_key.customer_tier,
-        }
+        },
+        permissions: [],
      };

      return res.status(200).json(userDto);
@@ -437,7 +437,8 @@ export class AuthClientService implements OnModuleInit {

    const userDto: UserDTO = {
      id: user.id,
-      name: user.username,
+      name: user.name,
+      email: user.email,
      jobTitle: user?.jobTitle || null,
      department: user?.department || null,
      hierarchy: user?.hierarchy || null,
@@ -446,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;
@@ -144,6 +144,7 @@ export interface BulkEditResponse {
 export type UserDTO = {
  id: string,
  name: string,
+  email: string,
  jobTitle?: string,
  department?: string,
  hierarchy?: string,
@@ -151,5 +152,6 @@ export type UserDTO = {
    id: string,
    name: string,
    tier: string,
-  }
+  },
+  permissions: number[],
 }
@@ -9,6 +9,7 @@ import {
  Inject,
  NotFoundException,
  Param,
+  Patch,
  Post,
  Put,
  Query,
@@ -17,6 +18,7 @@ import {
  HttpStatus,
  Res,
 } from '@nestjs/common';
+import { ValidationPipe } from '../../pipes/object-validation.pipe';
 import {
  ApiCreatedResponse,
  ApiHeaders,
@@ -46,9 +48,11 @@ import {
  IMakeAComment,
  IOneDataAsset,
  IPreviewResponse,
+  IUpdateCertificationStatusRequest,
  IUpdateDataRequest,
  TriggerCatalogReq,
  TriggerCatalogRes,
+  UpdateColumnsMetadataRequest,
 } from './dtos';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
 import { Language } from 'src/decorators/language.decorator';
@@ -83,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,
@@ -122,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,
@@ -165,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;
@@ -223,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,
@@ -241,11 +257,59 @@ export class CatalogController {
    return res;
  }

+  @Get('schemas')
+  @RequireSomePermission(
+    PERMISSIONS_GROUPS.CATALOG.permissions.GET,
+    PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
+  )
+  async findSchemas(@User() user: RequestUser) {
+    const { username, user_id, customer_id, customer_name } = user;
+    this.logger.info(`/catalog - ON FIND SCHEMAS ROUTE`, {
+      username,
+      customer_name,
+    });
+
+    const metadata = PackTheMetadata({
+      username,
+      user_id,
+      customer_id,
+      customer_name,
+    });
+
+    try {
+      const res = await this.catalogService.findSchemas(metadata);
+      return res;
+    } catch (error) {
+      throw new HttpException(error.message, HttpStatus.NOT_FOUND);
+    }
+
+  }
+
+  @Get('custom-properties')
+  @RequireSomePermission(
+    PERMISSIONS_GROUPS.CATALOG.permissions.GET,
+    PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
+  )
+  async getCustomPropertyDefinitions(@User() user: RequestUser) {
+    const { customer_id, customer_name, user_id, username } = user;
+    const metadata = PackTheMetadata({
+      customer_id,
+      customer_name,
+      user_id,
+      username,
+    });
+
+    return this.catalogService.getCustomPropertyDefinitions(metadata);
+  }
+
  @Get('data-asset/:id')
  @RequireSomePermission(
    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,
@@ -357,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,
@@ -383,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,
@@ -419,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,
@@ -450,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,
@@ -465,6 +577,8 @@ export class CatalogController {
      language,
    });

+    delete (body as any).certification_status;
+
    const result = await this.catalogService.updateOneDataAsset({
      body,
      data_asset_id,
@@ -478,11 +592,44 @@ export class CatalogController {
    return result;
  }

+  @Put('data-asset/:id/certification-status')
+  @RequireSomePermission(
+    PERMISSIONS_GROUPS.CATALOG.permissions.CERTIFY,
+    PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
+  )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
+  async updateDataAssetCertificationStatus(
+    @User() user: RequestUser,
+    @Language() language: LanguageEnum,
+    @Param('id') data_asset_id: string,
+    @Body(new ValidationPipe()) body: IUpdateCertificationStatusRequest,
+  ): Promise<IUpdateCertificationStatusRequest> {
+    const { customer_id, customer_name, user_id, username } = user;
+    const metadata = PackTheMetadata({
+      customer_id,
+      customer_name,
+      user_id,
+      username,
+      language,
+    });
+
+    return this.catalogService.updateCertificationStatus({
+      body,
+      data_asset_id,
+      metadata,
+    });
+  }
+
  @Post('data-asset/:id/docs')
  @RequireSomePermission(
    PERMISSIONS_GROUPS.CATALOG.permissions.UPDATE,
    PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
  )
+  @RequireModule(
+    DADOSFERA_MODULES_KEYS.CATALOG
+  )
  async manageDataAssetDocs(
    @User() user: RequestUser,
    @Headers() headers,
@@ -520,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,
@@ -542,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,
@@ -567,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,
@@ -591,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,
@@ -617,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({
@@ -638,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,
@@ -791,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,
@@ -943,4 +1111,4 @@ export class CatalogController {
      this.logger.error(error.message);
    }
  }
-}
+}
@@ -5,20 +5,17 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { CatalogController } from './catalog.controller';
 import { CatalogClientConfiguration } from './catalog-client';
 import { ClientsModule } from '@nestjs/microservices';
-import { PipelinesModule as OldPipelineModule } from 'src/modules/pipelines/pipelines.module';
 import { UsersModule } from '../users/users.module';
 import { RolesModule } from '../roles/roles.module';
 import { CustomersModule } from '../customers/customers.module';
 import { ShareModule } from './share/share.module';
 import { CatalogService } from './catalog.service';
-import { MixpanelModule } from '../mixpanel/mixpanel.module';

 const client = new CatalogClientConfiguration();

@Module({
  imports: [
    ClientsModule.register([client.providerOptions]),
-    OldPipelineModule,
    UsersModule,
    RolesModule,
    CustomersModule,
@@ -29,6 +29,7 @@ import {
  AssetReporter,
  BatchRemoveRlsRulesRequest,
  CreateDataDocsDTO,
+  IUpdateCertificationStatusRequest,
  IUpdateDataRequest,
  TriggerCatalogReq,
 } from './dtos';
@@ -119,6 +120,10 @@ class CatalogService implements OnModuleInit {
    }
  }

+  async getCustomPropertyDefinitions(metadata: Metadata) {
+    return lastValueFrom(this.catalogReadService.GetCustomPropertyDefinitions({}, metadata));
+  }
+
  async createDataAsset(data: Messages.CreateDataAssetRequest, metadata) {
    this.logger.info('CatalogService - Manage Data assets permissions');
    if (!data.embed) data.embed = undefined;
@@ -214,15 +219,6 @@ class CatalogService implements OnModuleInit {

    this.logger.debug('Extracted filters:', { filters });

-    console.log('MAESTRO VAI CHAMAR PI-FACTORY COM (ANTES AJUSTE):', {
-      search,
-      page,
-      size,
-      sort_by,
-      order,
-      filters,
-    });
-
    if (
      filters.manually !== undefined &&
      filters.manually !== null &&
@@ -233,15 +229,6 @@ class CatalogService implements OnModuleInit {
      delete filters.manually;
    }

-    console.log('MAESTRO VAI CHAMAR PI-FACTORY COM (DEPOIS AJUSTE):', {
-      search,
-      page,
-      size,
-      sort_by,
-      order,
-      filters,
-    });
-
    if (filters.owner) {
      const { users: customer_users } =
        await this.userService.findAllUsersByCustomerId(customer_id);
@@ -402,6 +389,28 @@ class CatalogService implements OnModuleInit {
    return { data_asset: asset[0] };
  }

+  async updateCertificationStatus(data: {
+    data_asset_id: string;
+    body: IUpdateCertificationStatusRequest;
+    metadata: Metadata;
+  }) {
+    const { body, data_asset_id, metadata } = data;
+
+    await lastValueFrom(
+      this.catalogWriteService.UpdateDataAsset(
+        {
+          id: data_asset_id,
+          changes: JSON.stringify({
+            certification_status: body.certification_status,
+          }),
+        },
+        metadata,
+      ),
+    );
+
+    return { certification_status: body.certification_status };
+  }
+
  async updateOneDataAsset(data: {
    data_asset_id: string;
    customer_id: string;
@@ -458,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);
@@ -517,6 +536,22 @@ class CatalogService implements OnModuleInit {
    return response;
  }

+  async findSchemas(metadata: Metadata) {
+    this.logger.info('CatalogService - findSchemas');
+
+    try {
+      const response = await lastValueFrom(
+        this.catalogReadService.GetSchemas({}, metadata),
+      );
+
+      return response;
+
+    } catch (error) {
+      this.logger.error('Error fetching schemas:', error);
+      throw error;
+    }
+  }
+
  async getAssetsUsersAndRoles(data_assets: Array<any>, customer_id: string) {
    const { users: customer_users } =
      await this.userService.findAllUsersByCustomerId(customer_id);
@@ -734,6 +769,64 @@ class CatalogService implements OnModuleInit {
    }
  }

+  async renameTableOnNimbus(
+    nimbusUrl: string,
+    nimbusId: number,
+    changes: { table_name?: string; table_schema?: string; display_name?: string },
+  ): Promise<void> {
+    const endpoint = `${nimbusUrl}/api/catalog/table-metadata/${nimbusId}`;
+    this.logger.info(`Renaming table-metadata ${nimbusId} on Nimbus`, { endpoint, changes });
+    await axios.patch(endpoint, changes);
+  }
+
+  async renameColumnMetadataOnNimbus(
+    nimbusUrl: string,
+    databaseName: string,
+    oldTableName: string,
+    oldTableSchema: string,
+    newTableName: string,
+    newTableSchema: string,
+  ): Promise<void> {
+    const listEndpoint = `${nimbusUrl}/api/catalog/column-metadata/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
+    this.logger.info(`Fetching column-metadata records to rename`, { listEndpoint });
+    const { data: columns } = await axios.get(listEndpoint);
+
+    const filtered = Array.isArray(columns) ? columns : [];
+
+    for (const column of filtered) {
+      const patchEndpoint = `${nimbusUrl}/api/catalog/column-metadata/${column.id}`;
+      await axios.patch(patchEndpoint, {
+        table_name: newTableName,
+        table_schema: newTableSchema,
+      });
+    }
+    this.logger.info(`Renamed ${filtered.length} column-metadata records on Nimbus`);
+  }
+
+  async renameDataPreviewOnNimbus(
+    nimbusUrl: string,
+    databaseName: string,
+    oldTableName: string,
+    oldTableSchema: string,
+    newTableName: string,
+    newTableSchema: string,
+  ): Promise<void> {
+    const listEndpoint = `${nimbusUrl}/api/catalog/data-preview/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
+    this.logger.info(`Fetching data-preview records to rename`, { listEndpoint });
+    const { data: previews } = await axios.get(listEndpoint);
+
+    const filtered = Array.isArray(previews) ? previews : [];
+
+    for (const preview of filtered) {
+      const patchEndpoint = `${nimbusUrl}/api/catalog/data-preview/${preview.id}`;
+      await axios.patch(patchEndpoint, {
+        table_name: newTableName,
+        table_schema: newTableSchema,
+      });
+    }
+    this.logger.info(`Renamed ${filtered.length} data-preview records on Nimbus`);
+  }
+
  async catalogDatasetItem(table_metadata_id: number, metadata: Metadata) {
    const customer_name_raw = metadata.get('customer_name');

@@ -749,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: '',
            },
          ],
        },
@@ -1,4 +1,13 @@
 import { ApiProperty, ApiPropertyOptional, PickType } from '@nestjs/swagger';
+import {
+  ArrayNotEmpty,
+  IsArray,
+  IsEnum,
+  IsNotEmpty,
+  IsString,
+  ValidateNested,
+} from 'class-validator';
+import { Type } from 'class-transformer';
 import { CreateDataAssetRequest } from '@dadosfera/protospack-v2/dist/lib/Catalog/interfaces/messages';

 export enum DataAssetShareType {
@@ -6,6 +15,12 @@ export enum DataAssetShareType {
  public = 'public',
  private = 'private',
 }
+export enum CertificationStatus {
+  draft = 'draft',
+  in_review = 'in_review',
+  approved = 'approved',
+  deprecated = 'deprecated',
+}
 export enum OrderEnum {
  asc = 'asc',
  desc = 'desc',
@@ -191,6 +206,27 @@ export class IData {
  day_opening: number;
 }

+
+export enum CustomPropertyType {
+  TEXT = 'text',
+  NUMBER = 'number',
+  DATE = 'date',
+  BOOLEAN = 'boolean',
+}
+
+export class CustomPropertyDto {
+  @ApiProperty()
+  key: string;
+  @ApiProperty()
+  value: string;
+  @ApiProperty({ enum: CustomPropertyType })
+  type: CustomPropertyType;
+  @ApiPropertyOptional()
+  color?: string;
+  @ApiPropertyOptional()
+  emoji?: string;
+}
+
 export class IUpdateDataRequest {
  @ApiProperty()
  name: string;
@@ -204,7 +240,36 @@ export class IUpdateDataRequest {
  share_type?: DataAssetShareType;
  @ApiPropertyOptional()
  docs?: string;
+  @ApiPropertyOptional({ type: [CustomPropertyDto] })
+  custom_properties?: CustomPropertyDto[];
 }
+
+export class IUpdateCertificationStatusRequest {
+  @ApiProperty({ enum: CertificationStatus })
+  @IsEnum(CertificationStatus)
+  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;
@@ -358,4 +423,4 @@ export type CreateDataDocsDTO = {
  docs: string;
  asset_type: string;

-}
+}
@@ -262,6 +262,7 @@ export class ShareService implements OnModuleInit {
        user_id: accessTokenPayload.user_id,
        username: accessTokenPayload.username,
        permissions: accessTokenPayload.permissions,
+        roles: accessTokenPayload.roles,
        customer_id: accessTokenPayload.customer_id,
        customer_name: accessTokenPayload.customer_name,
        customer_tier: accessTokenPayload.customer_tier,
@@ -22,17 +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(
@@ -84,7 +93,7 @@ export class ConnectionTestController {
    });
    return this.connectionTestService.connectionTestListSchemas(
      body,
-      user.customer_name,
+      user,
    );
  }

@@ -101,7 +110,7 @@ export class ConnectionTestController {
    });
    return this.connectionTestService.connectionTestListTables(
      body,
-      user.customer_name,
+      user,
    );
  }

@@ -117,8 +126,56 @@ export class ConnectionTestController {
      customer: user.customer_name,
    });
    return this.connectionTestService.getTableMetadata(
+      body,
+      user,
+    );
+  }
+
+  @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)
+  async refreshCatalog(
+    @User() user: RequestUser,
+    @Body(new ValidationPipe()) body: RefreshCatalogReq,
+  ) {
+    this.logger.info('/connection-test/refresh-catalog', {
+      user: user.user_id,
+      customer: user.customer_name,
+      connection: body.connection_id,
+    });
+    return this.connectionTestService.refreshCatalog(body, user);
+  }
+
+  @Post('refresh-catalog/status')
+  @ApiOkResponse({ type: RefreshCatalogRes })
+  @HttpCode(HttpStatus.OK)
+  async refreshCatalogStatus(
+    @User() user: RequestUser,
+    @Body(new ValidationPipe()) body: RefreshCatalogStatusReq,
+  ) {
+    this.logger.info('/connection-test/refresh-catalog/status', {
+      user: user.user_id,
+      customer: user.customer_name,
+      connection: body.connection_id,
+      session: body.session_id,
+    });
+    return this.connectionTestService.refreshCatalogStatus(body, user);
+  }
 }
@@ -5,10 +5,17 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { ClientsModule } from '@nestjs/microservices';
 import { ConnectionTestClientConfiguration } from './connection-test-client.config';
 import { ConnectionModule } from '../connection/connection.module';
+import { ConnectionsApiModule } from '../connections-api/connections-api.module';
+import { PlatformApiModule } from '../platform-api/platform-api.module';
 const client = new ConnectionTestClientConfiguration();
@Module({
  controllers: [ConnectionTestController],
  providers: [ConnectionTestService, DadosferaLogger],
-  imports: [ClientsModule.register([client.providerOptions]), ConnectionModule],
+  imports: [
+    ClientsModule.register([client.providerOptions]),
+    ConnectionModule,
+    ConnectionsApiModule,
+    PlatformApiModule,
+  ],
 })
 export class ConnectionTestModule {}
@@ -0,0 +1,223 @@
+import { ConnectionTestService } from './connection-test.service';
+import { RequestUser } from 'src/decorators/user.decorator';
+
+describe('ConnectionTestService catalog cache', () => {
+  const user: RequestUser = {
+    user_id: 'user-id',
+    username: 'user@example.com',
+    permissions: [],
+    customer_id: 'customer-id',
+    customer_name: 'customer-name',
+    customer_tier: 'standard',
+    access_token: 'token',
+    customer_modules: [],
+    roles: [],
+  };
+  const grpcClient = { getService: jest.fn().mockReturnValue({}) };
+  const connectionsService = {};
+  const connectionsApiService = { proxy: jest.fn() };
+  const platformApiService = { proxy: jest.fn() };
+  let service: ConnectionTestService;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    service = new ConnectionTestService(
+      grpcClient as any,
+      connectionsService as any,
+      connectionsApiService as any,
+      platformApiService as any,
+    );
+  });
+
+  it('keeps the existing schemas response contract', async () => {
+    connectionsApiService.proxy.mockResolvedValue({
+      schemas: [{ schema_name: 'analytics' }, { schema_name: 'public' }],
+    });
+
+    await expect(
+      service.connectionTestListSchemas(
+        { connection_id: 'config-id', plugin: 'postgresql' },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: true,
+      schema_list: ['analytics', 'public'],
+    });
+  });
+
+  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(
+        {
+          connection_id: 'config-id',
+          plugin: 'postgresql',
+          schema: 'public',
+        },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: true,
+      table_list: ['customers', 'orders'],
+      tables: [
+        { table_name: 'customers', primary_keys: ['id'] },
+        { table_name: 'orders', primary_keys: [] },
+      ],
+    });
+  });
+
+  it('maps cached columns to the existing table metadata contract', async () => {
+    connectionsApiService.proxy.mockResolvedValue({
+      columns: [
+        {
+          column_name: 'id',
+          data_type: 'bigint',
+          is_primary_key: true,
+        },
+      ],
+    });
+
+    await expect(
+      service.getTableMetadata(
+        {
+          connection_id: 'config-id',
+          plugin: 'postgresql',
+          schema: 'public',
+          table_list: ['customers'],
+        },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: true,
+      tables_metadata: [
+        {
+          table_name: 'customers',
+          columns: [
+            {
+              name: 'id',
+              type: 'bigint',
+              is_primary_key: true,
+            },
+          ],
+          references: [],
+        },
+      ],
+    });
+    expect(connectionsApiService.proxy).toHaveBeenCalledWith(
+      'GET',
+      '/connection_catalog/config-id/schemas/public/tables/customers/columns',
+      user,
+    );
+  });
+
+  it('submits a catalog refresh without holding the request open', async () => {
+    platformApiService.proxy.mockResolvedValue({
+      session_id: 'session-id',
+      date: '20260731',
+    });
+
+    await expect(
+      service.refreshCatalog(
+        { connection_id: 'config-id', plugin: 'postgresql' },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: true,
+      status: 'PENDING',
+      session_id: 'session-id',
+      date: '20260731',
+    });
+
+    expect(platformApiService.proxy).toHaveBeenCalledWith(
+      'POST',
+      '/connection_test',
+      user,
+      {
+        customer_id: user.customer_name,
+        plugin: 'postgresql',
+        task: {
+          task_type: 'refresh_catalog',
+          connection: {
+            provider: 'connection_manager',
+            config_id: 'config-id',
+          },
+        },
+      },
+    );
+  });
+
+  it('keeps polling without changing the catalog pointer while pending', async () => {
+    platformApiService.proxy.mockResolvedValue({ status: 'PENDING' });
+
+    await expect(
+      service.refreshCatalogStatus(
+        {
+          connection_id: 'config-id',
+          plugin: 'postgresql',
+          session_id: 'session-id',
+          date: '20260731',
+        },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: false,
+      status: 'PENDING',
+      session_id: 'session-id',
+      date: '20260731',
+    });
+
+    expect(connectionsApiService.proxy).not.toHaveBeenCalled();
+  });
+
+  it('publishes the catalog pointer after the refresh finishes', async () => {
+    platformApiService.proxy.mockResolvedValue({ status: 'DONE' });
+    connectionsApiService.proxy.mockResolvedValue({
+      last_catalog_refresh_status: 'SUCCESS',
+    });
+
+    await expect(
+      service.refreshCatalogStatus(
+        {
+          connection_id: 'config/id',
+          plugin: 'postgresql',
+          session_id: 'session-id',
+          date: '20260731',
+        },
+        user,
+      ),
+    ).resolves.toEqual({
+      operation_result: true,
+      status: 'DONE',
+      session_id: 'session-id',
+      date: '20260731',
+    });
+
+    expect(connectionsApiService.proxy).toHaveBeenCalledWith(
+      'PUT',
+      '/connection_config/config%2Fid/catalog_metadata',
+      user,
+      {
+        last_catalog_refresh_status: 'SUCCESS',
+        last_catalog_connection_test_date: '20260731',
+        last_catalog_connection_test_session_id: 'session-id',
+      },
+    );
+  });
+});
@@ -1,4 +1,4 @@
-import { Inject, Injectable } from '@nestjs/common';
+import { HttpException, HttpStatus, Inject, Injectable } from '@nestjs/common';
 import { ClientGrpc } from '@nestjs/microservices';
 import { ConnectionTest } from '@dadosfera/protospack-v2';
 import { lastValueFrom } from 'rxjs';
@@ -13,6 +13,11 @@ import {
  ConnectionTestPingRes,
  GetTableMetadataReq,
  GetTableMetadataRes,
+  ValidateCdcPrerequisitesReq,
+  ValidateCdcPrerequisitesRes,
+  RefreshCatalogReq,
+  RefreshCatalogRes,
+  RefreshCatalogStatusReq,
 } from './dto/connection-test';
 import { ConnectionClientService } from '../connection/client.service';
 import {
@@ -21,6 +26,8 @@ import {
 } from '../connection/dtos/connection';
 import { RequestUser } from 'src/decorators/user.decorator';
 import { PackTheMetadata } from 'src/utils/PackTheMetadata';
+import { ConnectionsApiService } from '../connections-api/connections-api.service';
+import { PlatformApiService } from '../platform-api/platform-api.service';

@Injectable()
 export class ConnectionTestService {
@@ -28,6 +35,8 @@ export class ConnectionTestService {
  constructor(
    @Inject('ConnectionTestGrpcClient') private readonly grpcClient: ClientGrpc,
    private connectionsService: ConnectionClientService,
+    private connectionsApiService: ConnectionsApiService,
+    private platformApiService: PlatformApiService,
  ) {
    this.connectionTestReadClient =
      grpcClient.getService<ConnectionTest.ReadService.ConnectionTestReadServices>(
@@ -147,44 +156,174 @@ export class ConnectionTestService {
  }
  async connectionTestListSchemas(
    body: ConnectionTestListSchemasReq,
-    customer_name: string,
+    user: RequestUser,
  ): Promise<ConnectionTestListSchemasRes> {
-    const { connection_id, plugin } = body;
-    return lastValueFrom(
-      this.connectionTestReadClient.ListSchemas({
-        connection_id,
-        customer_name,
-        plugin,
-      }),
+    const result = await this.connectionsApiService.proxy(
+      'GET',
+      `/connection_catalog/${encodeURIComponent(body.connection_id)}/schemas`,
+      user,
    );
+    return {
+      operation_result: true,
+      schema_list: result.schemas.map((schema) => schema.schema_name),
+    };
  }
+
  async connectionTestListTables(
    body: ConnectionTestListTablesReq,
-    customer_name: string,
+    user: RequestUser,
  ): Promise<ConnectionTestListTablesRes> {
-    const { connection_id, plugin, schema } = body;
-    return lastValueFrom(
-      this.connectionTestReadClient.ListTables({
-        connection_id,
-        customer_name,
-        plugin,
-        schema,
+    const result = await this.connectionsApiService.proxy(
+      'GET',
+      `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
+        `/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: table_names,
+      tables,
+    };
  }

  async getTableMetadata(
    body: GetTableMetadataReq,
-    customer_name: string,
+    user: RequestUser,
  ): Promise<GetTableMetadataRes> {
-    const { schema, plugin, table_list, connection_id } = body;
+    const tables_metadata = await Promise.all(
+      body.table_list.map(async (table_name) => {
+        const result = await this.connectionsApiService.proxy(
+          'GET',
+          `/connection_catalog/${encodeURIComponent(body.connection_id)}` +
+            `/schemas/${encodeURIComponent(body.schema)}` +
+            `/tables/${encodeURIComponent(table_name)}/columns`,
+          user,
+        );
+        return {
+          table_name,
+          columns: result.columns.map((column) => ({
+            name: column.column_name,
+            type: column.data_type,
+            is_primary_key: column.is_primary_key,
+          })),
+          references: [],
+        };
+      }),
+    );
+    return { operation_result: true, tables_metadata };
+  }
+
+  async refreshCatalog(
+    body: RefreshCatalogReq,
+    user: RequestUser,
+  ): Promise<RefreshCatalogRes> {
+    const task = await this.platformApiService.proxy(
+      'POST',
+      '/connection_test',
+      user,
+      {
+        customer_id: user.customer_name,
+        plugin: body.plugin,
+        task: {
+          task_type: 'refresh_catalog',
+          connection: {
+            provider: 'connection_manager',
+            config_id: body.connection_id,
+          },
+        },
+      },
+    );
+
+    if (!task.session_id || !task.date) {
+      throw new HttpException(
+        'Platform API did not return a catalog refresh task identifier',
+        HttpStatus.BAD_GATEWAY,
+      );
+    }
+
+    return {
+      operation_result: true,
+      status: 'PENDING',
+      session_id: task.session_id,
+      date: task.date,
+    };
+  }
+
+  async refreshCatalogStatus(
+    body: RefreshCatalogStatusReq,
+    user: RequestUser,
+  ): Promise<RefreshCatalogRes> {
+    const result = await this.platformApiService.proxy(
+      'POST',
+      '/connection_test/status',
+      user,
+      {
+        session_id: body.session_id,
+        date: body.date,
+      },
+    );
+
+    if (result.status === 'DONE') {
+      await this.connectionsApiService.proxy(
+        'PUT',
+        `/connection_config/${encodeURIComponent(
+          body.connection_id,
+        )}/catalog_metadata`,
+        user,
+        {
+          last_catalog_refresh_status: 'SUCCESS',
+          last_catalog_connection_test_date: body.date,
+          last_catalog_connection_test_session_id: body.session_id,
+        },
+      );
+    } else if (result.status === 'ERROR' || result.status === 'EXPIRED') {
+      throw new HttpException(
+        `Catalog refresh finished with status ${result.status}`,
+        HttpStatus.BAD_GATEWAY,
+      );
+    }
+
+    return {
+      operation_result: result.status === 'DONE',
+      status: result.status,
+      session_id: body.session_id,
+      date: body.date,
+    };
+  }
+
+  async validateCdcPrerequisites(
+    body: ValidateCdcPrerequisitesReq,
+    customer_name: string,
+  ): Promise<ValidateCdcPrerequisitesRes> {
+    const { plugin, connection_id } = body;
    return lastValueFrom(
-      this.connectionTestReadClient.GetTableMetadata({
+      this.connectionTestReadClient.ValidateCdcPrerequisites({
        connection_id,
        customer_name,
        plugin,
-        schema,
-        table_list,
      }),
    );
  }
@@ -1,5 +1,5 @@
 import { ApiProperty, ApiPropertyOptional, OmitType } from '@nestjs/swagger';
-import { IsString, IsOptional } from 'class-validator';
+import { IsIn, IsString, IsOptional } from 'class-validator';
 import { DatabaseConnectionPropertiesDto } from 'src/modules/connection/dtos/connection';
 import { CreateConnectionDto } from 'src/modules/connection/dtos/connection';
 export class ColumnDto {
@@ -7,6 +7,8 @@ export class ColumnDto {
  name: string;
  @ApiProperty()
  type: string;
+  @ApiProperty()
+  is_primary_key: boolean;
 }
 export class TableMetadataDto {
  @ApiProperty()
@@ -100,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 {
@@ -131,3 +142,83 @@ export class GetTableMetadataRes {
  @ApiProperty({ type: [TableMetadataDto] })
  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',
+      'mysql_cdc',
+      'postgresql_cdc',
+      'oracle_cdc',
+    ],
+  })
+  @IsIn([
+    'oracle',
+    'mysql',
+    'postgresql',
+    'sqlserver',
+    'mysql_cdc',
+    'postgresql_cdc',
+    'oracle_cdc',
+  ])
+  plugin: string;
+}
+
+export class RefreshCatalogStatusReq extends RefreshCatalogReq {
+  @ApiProperty()
+  @IsString()
+  session_id: string;
+
+  @ApiProperty()
+  @IsString()
+  date: string;
+}
+
+export class RefreshCatalogRes {
+  @ApiProperty()
+  operation_result: boolean;
+
+  @ApiProperty()
+  status: string;
+
+  @ApiProperty()
+  session_id: string;
+
+  @ApiProperty()
+  date: string;
+}
@@ -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(
@@ -0,0 +1,11 @@
+export const CONNECTIONS_API_CONFIG = {
+  getUrl: (): string => {
+    const url = process.env.CONNECTIONS_API_URL;
+    if (!url) {
+      throw new Error('CONNECTIONS_API_URL environment variable is not set');
+    }
+    return url;
+  },
+  region: process.env.AWS_REGION || 'us-east-1',
+  timeout: parseInt(process.env.CONNECTIONS_API_TIMEOUT || '30000', 10),
+};
@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+
+import { ConnectionsApiService } from './connections-api.service';
+
+@Module({
+  providers: [ConnectionsApiService, DadosferaLogger],
+  exports: [ConnectionsApiService],
+})
+export class ConnectionsApiModule {}
@@ -0,0 +1,99 @@
+import { Injectable, Inject, HttpException } from '@nestjs/common';
+import { SignatureV4 } from '@aws-sdk/signature-v4';
+import { Sha256 } from '@aws-crypto/sha256-js';
+import { defaultProvider } from '@aws-sdk/credential-provider-node';
+import axios, { AxiosResponse, Method } from 'axios';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
+
+import { RequestUser } from '../../decorators/user.decorator';
+import { CONNECTIONS_API_CONFIG } from './connections-api.config';
+
+@Injectable()
+export class ConnectionsApiService {
+  private signer: SignatureV4;
+  private logger: any;
+
+  constructor(@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger) {
+    this.logger = dadosferaLogger.logger;
+    this.signer = new SignatureV4({
+      service: 'execute-api',
+      region: CONNECTIONS_API_CONFIG.region,
+      credentials: defaultProvider(),
+      sha256: Sha256,
+    });
+  }
+
+  async proxy(
+    method: string,
+    path: string,
+    user: RequestUser,
+    body?: any,
+    query?: Record<string, string>,
+  ): Promise<any> {
+    const baseUrl = CONNECTIONS_API_CONFIG.getUrl();
+    const url = new URL(`${baseUrl}${path}`);
+
+    if (query) {
+      Object.entries(query).forEach(([key, value]) => {
+        if (value !== undefined && value !== null) {
+          url.searchParams.set(key, String(value));
+        }
+      });
+    }
+    const headers: Record<string, string> = {
+      host: url.hostname,
+      'content-type': 'application/json',
+      customer_name: user.customer_name || '',
+      customer_id: user.customer_id || '',
+      'x-user-id': user.user_id || '',
+      'x-username': user.username || '',
+      'x-customer-tier': user.customer_tier || '',
+      'x-customer-id': user.customer_id || '',
+    };
+    const requestToSign = {
+      method: method.toUpperCase(),
+      protocol: url.protocol,
+      hostname: url.hostname,
+      port: url.port ? parseInt(url.port, 10) : undefined,
+      path: url.pathname + url.search,
+      headers,
+      body: body ? JSON.stringify(body) : undefined,
+    };
+
+    try {
+      const signedRequest = await this.signer.sign(requestToSign);
+      const response: AxiosResponse = await axios({
+        method: method as Method,
+        url: url.href,
+        headers: signedRequest.headers as Record<string, string>,
+        data: body,
+        timeout: CONNECTIONS_API_CONFIG.timeout,
+        validateStatus: () => true,
+      });
+
+      if (response.status >= 400) {
+        throw new HttpException(response.data, response.status);
+      }
+      return response.data;
+    } catch (error) {
+      this.logger.error('Connections API proxy error', {
+        error: error.message,
+        path,
+        method: method.toUpperCase(),
+      });
+      if (error instanceof HttpException) {
+        throw error;
+      }
+      if (error.response) {
+        throw new HttpException(error.response.data, error.response.status);
+      }
+      if (error.code === 'ECONNREFUSED') {
+        throw new HttpException('Connections API service unavailable', 503);
+      }
+      if (error.code === 'ETIMEDOUT' || error.code === 'ECONNABORTED') {
+        throw new HttpException('Connections API request timeout', 504);
+      }
+      throw new HttpException('Internal server error', 500);
+    }
+  }
+}
@@ -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,
@@ -9,12 +9,12 @@ import {
 } from '@nestjs/common';

 import { firstValueFrom, lastValueFrom } from 'rxjs';
-import { Link } from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/entities';
 import { DucClient } from '../duc/client.config';
 import { ClientGrpc } from '@nestjs/microservices';
 import { ProtoServices } from '@dadosfera/protospack-v2/dist/lib/Duc';
-import { CustomerUpdateRequest } from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/messages';
+import { CustomerSetLinksRequest } from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/messages';
 import { CustomersProtoService } from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/write-service';
+import { CustomerLinksConfig } from './dtos/customers';
 import ErrorCodes from 'src/utils/errorCodes';
 import jwt from 'jsonwebtoken';
 import {
@@ -67,12 +67,12 @@ export class CustomersService implements OnModuleInit {
    )
  }

-  async getLinks(customerId: string) {
+  async getLinks(customerId: string): Promise<CustomerLinksConfig | null> {
    try {
      const result = await lastValueFrom(
-        this.customerService.CustomerFindOneById({ id: customerId }),
+        this.customerService.CustomerGetLinks({ customerId }),
      );
-      return result.customer?.links || [];
+      return (result.links as CustomerLinksConfig) || null;
    } catch (err) {
      if (err.details === ErrorCodes.CUSTOMER.NOT_FOUND)
        throw new HttpException(err.details, HttpStatus.NOT_FOUND);
@@ -80,17 +80,17 @@ export class CustomersService implements OnModuleInit {
    }
  }

-  async setLinks(customerId: string, links: Link[]) {
+  async setLinks(customerId: string, links: CustomerLinksConfig) {
    if (!customerId || !links) {
      throw new HttpException(null, HttpStatus.BAD_REQUEST);
    }

    try {
      return await firstValueFrom(
-        this.customerService.CustomerUpdate({
-          id: customerId,
-          links,
-        } as CustomerUpdateRequest),
+        this.customerService.CustomerSetLinks({
+          customerId,
+          links: links as CustomerSetLinksRequest['links'],
+        }),
      );
    } catch (err) {
      if (err.details === ErrorCodes.CUSTOMER.NOT_FOUND)
@@ -1,7 +1,6 @@
-import { Link } from '@dadosfera/protospack-v2/dist/lib/Duc/interfaces/entities';
 import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';

-export class CustomerLink implements Link {
+export class CustomerLinkItem {
  @ApiProperty()
  href: string;
  @ApiProperty()
@@ -9,15 +8,59 @@ export class CustomerLink implements Link {
  @ApiProperty()
  description: string;
  @ApiPropertyOptional()
-  iconSrc: string;
+  iconSrc?: string;
 }
+
+export class CustomerSidebarLinkItem {
+  @ApiProperty()
+  type: 'link';
+  @ApiProperty({ type: Object })
+  title: Record<string, string>;
+  @ApiProperty()
+  link: string;
+  @ApiPropertyOptional()
+  icon?: string;
+}
+
+export class CustomerSidebarMenuItem {
+  @ApiProperty()
+  type: 'menu';
+  @ApiProperty({ type: Object })
+  title: Record<string, string>;
+  @ApiPropertyOptional()
+  icon?: string;
+  @ApiProperty({ type: [CustomerSidebarLinkItem] })
+  items: CustomerSidebarLinkItem[];
+}
+
+export class CustomerSidebarSection {
+  @ApiProperty({ type: Object })
+  title: Record<string, string>;
+  @ApiProperty({
+    type: 'array',
+    items: {
+      oneOf: [
+        { $ref: '#/components/schemas/CustomerSidebarMenuItem' },
+        { $ref: '#/components/schemas/CustomerSidebarLinkItem' },
+      ],
+    },
+  })
+  items: (CustomerSidebarMenuItem | CustomerSidebarLinkItem)[];
+}
+
+export class CustomerLinksConfig {
+  @ApiPropertyOptional({ type: [CustomerLinkItem] })
+  home?: CustomerLinkItem[];
+  @ApiPropertyOptional({ type: [CustomerSidebarSection] })
+  sidebar?: CustomerSidebarSection[];
+}
+
 export class CustomerLinkRequest {
-  @ApiProperty({ type: [CustomerLink] })
-  links: CustomerLink[];
+  @ApiProperty({ type: CustomerLinksConfig })
+  links: CustomerLinksConfig;
 }

 export class CustomerLinksResponse {
-  @ApiProperty({ type: [CustomerLink] })
-  links: CustomerLink[];
-}
-
+  @ApiPropertyOptional({ type: CustomerLinksConfig })
+  links?: CustomerLinksConfig;
+}
@@ -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 ?? [],
+  };
+}
@@ -11,10 +11,19 @@ export class TableColumns {
  name: string;
  @ApiProperty()
  columns: string[];
-  @ApiProperty()
+  @ApiPropertyOptional({ type: [Column] })
  references: Column[];
  @ApiProperty()
+  destination: Record<'raw' | 'qualify', {
+    table_name: string;
+    table_schema: string;
+  }> | null;
+  @ApiProperty()
  type: string;
+  @ApiPropertyOptional({ type: [String] })
+  identifier_columns?: string[];
+  @ApiPropertyOptional({ type: Column })
+  reference_column?: Column;
 }
 export class AvailableEntity {
  @ApiProperty()
@@ -56,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;
+}
@@ -1,4 +1,8 @@
-import { Info } from '@dadosfera/protospack/dist/lib/interfaces';
+export interface Info {
+  user_id: string;
+  customer_id: string;
+  customer: string;
+}

 interface Values {
  jdbc_user: string;
@@ -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',
+            },
+          },
+        }),
+      }),
+    );
+  });
+});
@@ -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,
@@ -99,11 +100,32 @@ export class InputsController {
      customer: info.customer,
    });

+    this.logger.info(JSON.stringify(body))
    const response = await this.inputService.create({ body, info });

    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) {
@@ -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();
+  });
+});
@@ -13,14 +13,22 @@ 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()

@@ -71,10 +79,10 @@ export class InputsService {
      objectCamelToSnake(createInputResponse);
      return createInputResponse;
    },
-    update: async (updateInputDTO: UpdateInputRequest) => {
-      this.logger.info('InputClientService - Update');
+    update: async (updateInputDTO: UpdateInputRequest, metadata: Metadata): Promise<InputUpdateResponse> => {
+      this.logger.info('InputClientService - Update' + JSON.stringify(updateInputDTO));
      const updateInputResponse = await lastValueFrom(
-        this.inputWriteService.InputUpdate(updateInputDTO),
+        this.inputWriteService.InputUpdate(updateInputDTO, metadata),
      );

      return updateInputResponse;
@@ -164,6 +172,11 @@ export class InputsService {
    const inputCreateGenericRequest: InputCreateGenericRequest = {
      input: {
        ...body,
+        tables: (body.tables || []).map((table) => ({
+          ...table,
+          identifier_columns: table.identifier_columns || [],
+          reference_column: table.reference_column || table.references?.[0],
+        })),
      },
      info,
    };
@@ -175,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));
  }
@@ -199,24 +232,47 @@ export class InputsService {
    return findOneInputResponse;
  }

-  async update(id: string, data, info: Info) {
-    this.validateCron({ ...data, info });
+  async update(id: string, data, info: Info, metadata?: Metadata) {
+    // this.validateCron({ ...data, info });
    try {
-      const updateInputResponse: any = await this.OLD_inputClient.update({
+      const {
+        tablesUpdate,
+        dataAssetUpdate,
+        input
+      } = await this.OLD_inputClient.update({
        id,
-        info,
        ...data,
-      });
+        info,
+      }, metadata);

-      updateInputResponse.input = this.adjustInputPayload(
-        updateInputResponse?.input,
+      const updateInputResponse = this.adjustInputPayload(
+        input,
      );
-      return updateInputResponse;
+      return {
+        input: updateInputResponse,
+        tablesUpdate,
+        dataAssetUpdate
+      };
    } catch (err) {
      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
    }
  }

+  async rollbackUpdate(
+    data: RollbackInputRequest
+  ) {
+    this.logger.info('PipelinesClientService - rollbackUpdate');
+    this.logger.info('Rolling back input update with data: ' + JSON.stringify(data));
+    const updatePipelineResponse = await lastValueFrom(
+      this.inputWriteService.RollbackInputUpdate(
+        data
+      ),
+    );
+    this.logger.info('Done');
+
+    return updatePipelineResponse;
+  }
+
  async remove(idRequest: IIdRequest) {
    return lastValueFrom(this.inputWriteService.InputRemove(idRequest));
  }
@@ -258,4 +314,20 @@ export class InputsService {
    };
    return formatedPayload;
  }
+
+  async markTableDeleted(data: MarkTableDeletedRequest) {
+    return lastValueFrom(this.inputWriteService.MarkTableDeleted(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));
+  }
 }
@@ -1,78 +0,0 @@
-import { ConflictException, Inject, OnModuleInit } from '@nestjs/common';
-import { ClientGrpc } from '@nestjs/microservices';
-import {
-  PipelineServicesNames,
-  PipelinesServiceInterface,
-} from '@dadosfera/protospack';
-import { lastValueFrom } from 'rxjs';
-
-import { IIdRequest } from './interfaces';
-
-import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
-import { PipelinesClientConfiguration } from './pipelines-client';
-
-export class PipelinesClientService implements OnModuleInit {
-  private pipelineService: PipelinesServiceInterface;
-  logger: DadosferaLogger;
-
-  constructor(
-    @Inject(DadosferaLogger)
-      dadosferaLogger: DadosferaLogger,
-    @Inject(PipelinesClientConfiguration.name)
-    private readonly grpcClient: ClientGrpc,
-  ) {
-    this.logger = dadosferaLogger.logger;
-  }
-
-  onModuleInit() {
-    this.pipelineService =
-      this.grpcClient.getService<PipelinesServiceInterface>(
-        PipelineServicesNames.PipelineService,
-      );
-  }
-
-  async getPipelineStatus(data) {
-    this.logger.info('PipelinesClientService - GetPipelineStatus');
-
-    const statusPipelineResponse = await lastValueFrom(
-      this.pipelineService.getPipelineStatus(data),
-    )
-      .then((res) => {
-        const statusArray =
-          res.status?.sort((a, b) => {
-            if (a.id < b.id) {
-              return 1;
-            } else {
-              return -1;
-            }
-          }) || [];
-        return { status: statusArray };
-      })
-      .catch((err) => {
-        this.logger.error(err.message);
-        throw new Error(err);
-      });
-    this.logger.info('Done');
-
-    return statusPipelineResponse;
-  }
-
-  async runPipeline({ id, info }: IIdRequest) {
-    this.logger.info('PipelinesClientService - RunPipeline');
-    const statusPipelineResponse = await lastValueFrom(
-      this.pipelineService.triggerPipeline({ id, info }),
-    ).catch((err) => {
-      this.logger.error(err.message);
-      throw new Error(err);
-    });
-
-    if (statusPipelineResponse.status == false) {
-      throw new ConflictException(
-        'This pipeline is not ready yet to execute, Try again later!',
-      );
-    }
-
-    this.logger.info('Done');
-    return statusPipelineResponse;
-  }
-}
@@ -1,36 +0,0 @@
-import { Info } from '@dadosfera/protospack/dist/lib/interfaces';
-
-export interface ICreatePipelineDto {
-  input: IdRequest;
-  transformations: IdRequest[];
-  output: IdRequest;
-  tags: string[];
-  name: string;
-  description: string;
-  info: Info;
-}
-
-export interface IdRequest {
-  id: string;
-}
-
-export interface IIdRequest {
-  id: string;
-  info: Info;
-}
-
-export interface IUpdatePipelineRequest {
-  input: IdRequest;
-  transformations: IdRequest[];
-  output: IdRequest;
-  tags: string[];
-  name: string;
-  description: string;
-  id: string;
-  info: Info;
-}
-
-export interface IGetPipelineLogsRequest {
-  id: string;
-  details: string;
-}
@@ -1,33 +0,0 @@
-import {
-  ClientsProviderAsyncOptions,
-  GrpcOptions,
-  Transport,
-} from '@nestjs/microservices';
-import { PipelinePackages, PipelineProtoFilePath } from '@dadosfera/protospack';
-import { credentials } from '@grpc/grpc-js';
-
-const isLocalConnection =
-  process.env.PIFACTORY_URL.startsWith('pi-factory:') ||
-  process.env.PIFACTORY_URL.includes('0.0.0.0');
-
-export class PipelinesClientConfiguration {
-  public name = 'PipelinesClientConfiguration';
-  private config: GrpcOptions = {
-    transport: Transport.GRPC,
-    options: {
-      url: process.env.PIFACTORY_URL,
-      package: PipelinePackages,
-      credentials: isLocalConnection ? undefined : credentials.createSsl(),
-      protoPath: PipelineProtoFilePath,
-      loader: {
-        keepCase: true,
-        enums: String,
-        defaults: false,
-      },
-    },
-  };
-  providerOptions: ClientsProviderAsyncOptions = {
-    name: this.name,
-    ...this.config,
-  };
-}
@@ -1,72 +0,0 @@
-import { Body, Controller, Get, Inject, Param, Post } from '@nestjs/common';
-import { ApiOperation, ApiTags } from '@nestjs/swagger';
-import {
-  AuthenticateCondition,
-  Authenticated,
-  RequireSomePermission,
-} from 'src/decorators/authentication.decorator';
-import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
-import { PipelinesService } from './pipelines.service';
-import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
-import { ApiInternalOnlyController } from 'src/decorators/swagger.decorator';
-
-@ApiInternalOnlyController()
-@ApiTags('Pipelines')
-@Controller('pipelines')
-@Authenticated()
-export class PipelinesController {
-  logger: DadosferaLogger;
-  constructor(
-    @Inject(DadosferaLogger)
-    dadosferaLogger: DadosferaLogger,
-    private pipelineService: PipelinesService,
-  ) {
-    this.logger = dadosferaLogger.logger;
-  }
-
-  @Post('start/:id')
-  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
-  @ApiOperation({
-    deprecated: true,
-    description:
-      'This method is deprecated. Please use route /pipelinesV2/start/:id instead',
-  })
-  async activate(@Param('id') id: string, @Body() body) {
-    const { info } = body;
-
-    this.logger.info(
-      process.env.DEV_URL + `/pipeline/start/${id} - ON START PIPELINE ROUTE`,
-      {
-        user: body.info.user_id,
-        customer: body.info.customer,
-      },
-    );
-
-    const response = await this.pipelineService.runPipeline({ id, info });
-
-    return response;
-  }
-
-  @Get(':id/status')
-  @ApiOperation({
-    deprecated: true,
-    description:
-      'This method is deprecated. Please use route /pipelinesV2/:id/status instead',
-  })
-  @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
-  async getPipelineStatus(@Body() body, @Param('id') id: string) {
-    body.id = id;
-
-    this.logger.info(
-      process.env.DEV_URL + `/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`,
-      {
-        user: body.info.user_id,
-        customer: body.info.customer,
-      },
-    );
-
-    const response = await this.pipelineService.getPipelineStatus(body);
-
-    return response;
-  }
-}
@@ -1,19 +0,0 @@
-import { Module } from '@nestjs/common';
-import { ClientsModule } from '@nestjs/microservices';
-import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
-
-import { PipelinesController } from './pipelines.controller';
-import { PipelinesService } from './pipelines.service';
-
-import { PipelinesClientConfiguration } from './pipelines-client';
-import { PipelinesClientService } from './client.service';
-
-const client = new PipelinesClientConfiguration();
-
-@Module({
-  imports: [ClientsModule.register([client.providerOptions])],
-  controllers: [PipelinesController],
-  providers: [PipelinesService, PipelinesClientService, DadosferaLogger],
-  exports: [PipelinesService],
-})
-export class PipelinesModule {}
@@ -1,33 +0,0 @@
-import { HttpException, HttpStatus, Injectable } from '@nestjs/common';
-import { PipelinesClientService } from './client.service';
-import { IIdRequest } from './interfaces';
-import { objectCamelToSnake } from 'src/utils/CaseConverter';
-
-@Injectable()
-export class PipelinesService {
-  constructor(private pipelineClient: PipelinesClientService) {}
-
-  async getPipelineStatus(data: IIdRequest) {
-    try {
-      const pipelineStatusResponse =
-        await this.pipelineClient.getPipelineStatus(data);
-
-      return objectCamelToSnake(pipelineStatusResponse);
-    } catch (err) {
-      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
-    }
-  }
-
-  async runPipeline({ id, info }: IIdRequest) {
-    try {
-      const triggerPipelineResponse = await this.pipelineClient.runPipeline({
-        id,
-        info,
-      });
-
-      return objectCamelToSnake(triggerPipelineResponse);
-    } catch (err) {
-      throw new HttpException(err.message, HttpStatus.NOT_FOUND);
-    }
-  }
-}
@@ -1,5 +1,40 @@
 import { ApiProperty, ApiPropertyOptional, OmitType } from '@nestjs/swagger';
-import { Info } from '@dadosfera/protospack/dist/lib/interfaces';
+
+export class PipelineInputsDTO {
+  @ApiProperty()
+  tables: Array<{
+    name: string,
+    type: string,
+    
+  }>
+}
+
+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()
@@ -23,6 +58,8 @@ export class IPipelineV2 {
  tags?: string[];
  @ApiPropertyOptional()
  properties?: any;
+  @ApiPropertyOptional({ type: () => PipelineConfig })
+  config?: PipelineConfig;

  @ApiProperty()
  connector_name: string;
@@ -53,6 +90,12 @@ export interface IIdRequest {
  info: Info;
 }

+export interface Info {
+  user_id: string;
+  customer_id: string;
+  customer: string;
+}
+
 export interface IUpdatePipelineRequest {
  input: IdRequest;
  transformations: IdRequest[];
@@ -123,3 +166,30 @@ export class PipelineFindAllReq {
  @ApiPropertyOptional()
  type?: string | undefined;
 }
+
+export interface UpdateTableDTO {
+  name: string;
+  type: string;
+  columns: string[];
+  destinations: {
+    raw: {
+      table_schema: string;
+      table_name: string;
+    };
+    qualify: {
+      table_schema: string;
+      table_name: string;
+    };
+  };
+  identifier_columns: string[];
+  reference_column: {
+    name: string;
+    type: string;
+  };
+  memory: number;
+}
+
+export interface UpdatePlatformInputRequest {
+  cron: string;
+  tables: Array<UpdateTableDTO>;
+}
@@ -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,
      },
@@ -14,7 +14,7 @@ import {
  Patch,
  HttpException,
  BadRequestException,
-  CacheTTL,
+  UseGuards,
 } from '@nestjs/common';
 import {
  ApiCreatedResponse,
@@ -24,18 +24,17 @@ import {
  ApiTags,
 } from '@nestjs/swagger';
 import {
-  AuthenticateCondition,
  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';
 import { RequestUser, User } from 'src/decorators/user.decorator';
 import { PackTheMetadata } from 'src/utils/PackTheMetadata';

-import { PipelinesService as OldPipelineService } from 'src/modules/pipelines/pipelines.service';
 import {
  ICompleteUploadCSVFile,
  ICreatePipelineCSVFile,
@@ -43,23 +42,31 @@ import {
  IPipelineV2,
  IInitUploadCSVFile,
  PipelineFindAllReq,
+  UpdatePlatformInputRequest,
 } from './interfaces';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
 import { LanguageEnum } from 'src/utils/languages.enum';
 import { Language } from 'src/decorators/language.decorator';
 import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator';
+import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
+import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
+
+type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any };
+type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] };

@ApiTags('PipelinesV2')
@ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
@UseFilters(new GrpcToHttpExceptionFilter())
@Controller('pipelinesV2')
+@RequireModule(
+  DADOSFERA_MODULES_KEYS.COLLECT
+)
 export class PipelinesController {
  logger: DadosferaLogger;
  constructor(
    @Inject(DadosferaLogger)
    dadosferaLogger: DadosferaLogger,
    private pipelinesClientService: PipelinesService,
-    private oldPipelinesService: OldPipelineService,
  ) {
    this.logger = dadosferaLogger.logger;
  }
@@ -187,6 +194,7 @@ export class PipelinesController {
  @Get(':id/status')
  @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getPipelineStatus(@Body() body, @Param('id') id: string) {
+
    body.id = id;

    this.logger.info(`/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`, {
@@ -194,7 +202,7 @@ export class PipelinesController {
      customer: body.info.customer,
    });

-    const response = await this.oldPipelinesService.getPipelineStatus(body);
+    const response = await this.pipelinesClientService.getPipelineStatus(body);

    return response;
  }
@@ -218,26 +226,50 @@ export class PipelinesController {
      language,
    });

-    const result = await this.pipelinesClientService
-      .findOne({ id }, metadata)
-      .then((res) => {
-        //{pipeline:{tables: {tables: [], input_id: ''}}}
-        let tables = JSON.parse(res.pipeline.config.tables);
-        if (tables?.tables) tables = tables.tables;
-        Object.assign(res.pipeline, {
-          transformations: res.pipeline.transformations
-            ? JSON.parse(res.pipeline.transformations)
-            : [],
-          config: {
-            cron: res.pipeline.config.cron,
-            tables,
-          },
-          properties: res.pipeline.properties
-            ? JSON.parse(res.pipeline.properties)
-            : {},
-        });
-        return res;
-      });
+    const pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata);
+
+    const parsed: PipelineTablesConfig = JSON.parse(pipelineRes.pipeline.config.tables);
+    const input_id = parsed.input_id;
+    const tables: PipelineTable[] = parsed.tables ?? [];
+
+    Object.assign(pipelineRes.pipeline, {
+      transformations: pipelineRes.pipeline.transformations
+        ? JSON.parse(pipelineRes.pipeline.transformations)
+        : [],
+      config: {
+        cron: pipelineRes.pipeline.config.cron,
+        tables,
+        input_id,
+      },
+      properties: pipelineRes.pipeline.properties
+        ? JSON.parse(pipelineRes.pipeline.properties)
+        : {},
+    });
+
+    return pipelineRes;
+  }
+
+  @Get("/:id/data-assets")
+  @RequireSomePermission(
+    PERMISSIONS_GROUPS.PIPELINE.permissions.GET,
+    PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER,
+    PERMISSIONS_GROUPS.CATALOG.permissions.GET
+  )
+  async findAllDataAssetByPipeline(
+    @Language() language: LanguageEnum,
+    @Param('id') id: string,
+    @User() user: RequestUser,
+    @Query('object') object: string
+  ) {
+    const payload = {
+      pipeline: id,
+      object: object, 
+    };
+
+    this.logger.info(`GET pipelinesV2/:id/data-assets` + JSON.stringify(payload));
+    
+    const result =
+      await this.pipelinesClientService.findAllDataAssetByPipeline(payload, user);

    return result;
  }
@@ -277,6 +309,40 @@ export class PipelinesController {
    return response;
  }

+  @Patch('/:pipelineId/inputs/:id')
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  @UseGuards(PipelineExecutionGuard)
+  async updatePipelineInput(
+    @Language() language: LanguageEnum,
+    @Body() pipelineInputDTO: UpdatePlatformInputRequest,
+    @Param('id') inputId: string,
+    @Param('pipelineId') pipelineId: string,
+    @User() user: RequestUser,
+  ) {
+    this.logger.info('PipelinesController - update', { user });
+
+    const info: Info = {
+      user_id: user.user_id,
+      customer: user.customer_name,
+      customer_id: user.customer_id,
+      pipeline_id: pipelineId
+    };
+
+    const metadata = PackTheMetadata(user);
+
+    const response = await this.pipelinesClientService.updatePipelineInput(
+      pipelineId,
+      inputId,
+      pipelineInputDTO,
+      info,
+      user,
+      metadata,
+    );
+
+    this.logger.info('PipelinesController - update: OK', { user });
+    return response;
+  }
+
  @ApiInternalOnlyEndpoint()
  @Put('/:id')
  @ApiOperation({
@@ -296,6 +362,21 @@ export class PipelinesController {
    return response;
  }

+  @Patch('/:id/upgrade')
+  @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  @HttpCode(HttpStatus.NO_CONTENT)
+  async upgradeConnector(
+    @Language() language: LanguageEnum,
+    @Param('id') id: string,
+    @User() user: RequestUser
+  ) {
+    this.logger.info('PipelinesController - upgrade connector');
+
+    const metadata = PackTheMetadata(user);
+
+    await this.pipelinesClientService.upgrade(id, metadata);
+  }
+
  @Delete(':id')
  @ApiNoContentResponse()
  @HttpCode(HttpStatus.NO_CONTENT)
@@ -425,8 +506,82 @@ export class PipelinesController {
      },
    );

-    const response = await this.oldPipelinesService.runPipeline({ id, info });
+    const response = await this.pipelinesClientService.runPipeline({ id, info });

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

@@ -7,23 +7,28 @@ import { PipelinesService } from './pipelines.service';

 import { PipelinesClientConfiguration } from './pipelines-client';

-import { PipelinesModule as OldPipelineModule } from 'src/modules/pipelines/pipelines.module';
 import { ConnectorModule } from '../connector/connector.module';
 import { InputsModule } from '../inputs/inputs.module';
 import { TransformationsModule } from '../transformations/transformations.module';
+import { PlatformApiModule } from '../platform-api/platform-api.module';
+import { NimbusServicesModule } from 'src/services/nimbus/nimbus.module';
+import { NimbusService } from 'src/services/nimbus/nimbus.service';
+import { CatalogModule } from '../catalog/catalog.module';

 const client = new PipelinesClientConfiguration();

@Module({
  imports: [
    ClientsModule.register([client.providerOptions]),
-    OldPipelineModule,
    ConnectorModule,
    InputsModule,
    TransformationsModule,
+    forwardRef(() => PlatformApiModule),
+    NimbusServicesModule,
+    CatalogModule
  ],
  controllers: [PipelinesController],
-  providers: [PipelinesService, DadosferaLogger],
+  providers: [PipelinesService, DadosferaLogger, NimbusService],
  exports: [PipelinesService],
 })
 export class PipelinesV2Module {}
@@ -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',
+    );
+  });
+});
@@ -1,5 +1,8 @@
+import { isCdcPlugin } from 'src/utils/cdc';
+/* eslint-disable no-async-promise-executor */
 import {
  BadRequestException,
+  ConflictException,
  HttpException,
  HttpStatus,
  Inject,
@@ -16,8 +19,8 @@ import { lastValueFrom } from 'rxjs';

 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { PipelinesClientConfiguration } from './pipelines-client';
-import { ICreatePipelineV2Req } from './interfaces';
-import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
+import { ICreatePipelineV2Req, IIdRequest, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces';
+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';
@@ -26,6 +29,17 @@ import { TransformationsService } from '../transformations/transformations.servi
 import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath';
 import ErrorCodes from 'src/utils/errorCodes';
 import ErrorBuilder from 'src/utils/ErrorBuilder';
+import { PlatformApiService } from '../platform-api/platform-api.service';
+import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
+import { TableUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
+import { AxiosError } from 'axios';
+import { NimbusService } from 'src/services/nimbus/nimbus.service';
+import { PERMISSIONS_GROUPS } from 'src/authentication/permissions.enum';
+import { PackTheMetadata } from 'src/utils/PackTheMetadata';
+import { IDataAsset } from '../catalog/dtos';
+import { CatalogService } from '../catalog/catalog.service';
+
+type RollbackPromise = () => Promise<any>;

 export class PipelinesService implements OnModuleInit {
  logger: DadosferaLogger;
@@ -39,6 +53,9 @@ export class PipelinesService implements OnModuleInit {
    private readonly connectorService: ConnectorClientService,
    private readonly inputsService: InputsService,
    private readonly transformationsService: TransformationsService,
+    private readonly platformAPI: PlatformApiService,
+    private readonly nimbusService: NimbusService,
+    private readonly catalogService: CatalogService
  ) {
    this.logger = dadosferaLogger.logger;
  }
@@ -89,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(
@@ -138,11 +159,68 @@ export class PipelinesService implements OnModuleInit {
    const findOnePipelineResponse = await lastValueFrom(
      this.pipelineReadService.PipelineV2FindOne(data, metadata),
    );
+    console.log('pipeline find one response', findOnePipelineResponse);
    this.logger.info('Done');

    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,
@@ -160,6 +238,15 @@ export class PipelinesService implements OnModuleInit {
    return updatePipelineResponse;
  }

+  async upgrade(id: string, metadata: Metadata) {
+    await lastValueFrom(
+      this.pipelineWriteService.Upgrade(
+        { id },
+        metadata,
+      ),
+    );
+  }
+
  async remove(data: { id: string; metadata: Metadata; user: RequestUser }) {
    const { id, metadata, user } = data;
    const info = {
@@ -339,4 +426,329 @@ export class PipelinesService implements OnModuleInit {

    return res;
  }
+
+  async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) {
+    this.logger.info('InputClientService - Update');
+
+    const {
+      input: oldInput
+    } = await this.inputsService.findOne({
+      id: inputId,
+      info: info
+    });
+
+    this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
+    const isCdc = isCdcPlugin(oldInput.plugin);
+    const pipelineIdFormat = pipelineId.split('-').join('_');
+    const rollback: RollbackPromise[] = [];
+
+    const updateInputResponse = await this.inputsService.update(
+      inputId,
+      updateInputDTO,
+      info,
+      metadata
+    );
+
+    const inputRollback = () => {
+      this.logger.info("exec rollback to input: " + JSON.stringify(oldInput));
+      return this.inputsService.rollbackUpdate(
+        {
+          id: inputId,
+          dataAssetUpdate: updateInputResponse.dataAssetUpdate,
+          tables: oldInput.tables,
+          info
+        }
+      ) as Promise<any>;
+    }
+
+    rollback.push(inputRollback);
+
+    this.logger.info("Input Update Response: " + JSON.stringify(updateInputResponse))
+
+    const nimbusUpdates = updateInputResponse?.tablesUpdate || [];
+
+    if (user.customer_modules.includes('catalog')) {
+      nimbusUpdates.forEach(update => {
+        const nimbusRollback = () => {
+          return this.nimbusService.renameTable(
+            info.customer,
+            update.database,
+            {
+              table_name: update.table_name,
+              table_schema: update.table_schema
+            },
+            {
+              table_name: update.old_table_name,
+              table_schema: update.old_table_schema
+            }
+          );
+        }
+        rollback.push(nimbusRollback);
+      });
+  
+      try {
+        await this.updateNimbus(info.customer, nimbusUpdates);
+      } catch (error) {
+        this.logger.error(error);
+        if (error instanceof AxiosError) {
+          this.logger.error(JSON.stringify(error.response.data));
+        }
+        await this.executeRenameRollback(rollback);
+  
+        throw new Error("Error Nimbus updating tables");
+      }
+    }
+
+    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;
+  }
+
+  private async executeRenameRollback(request: RollbackPromise[]) {
+    this.logger.info('rollback steps: ' + request.length)
+    const result = await Promise.allSettled(request.map(func => func()));
+    result.forEach(promise => {
+      this.logger.info("Promise finish with status: " + promise.status)
+
+      if (promise.status === "rejected") {
+        this.logger.error("reject with: " + JSON.stringify(promise.reason || {}))
+      }
+
+      if (promise.status === "fulfilled") {
+        this.logger.info("success with: " + JSON.stringify(promise.value || {}))
+      }
+    });
+
+  }
+
+  private async updateNimbus(customer: string, changes: TableUpdate[]) {
+    // throw new Error("teste error nimbus");
+    this.logger.info('Nimbus Changes: ' + JSON.stringify(changes));
+    if(!changes || changes.length === 0) return;
+
+    const requests = changes.map(change => {
+      return this.nimbusService.renameTable(customer, change.database, {
+        table_name: change.old_table_name,
+        table_schema: change.old_table_schema
+      }, {
+        table_name: change.table_name,
+        table_schema: change.table_schema
+      });
+    })
+
+    const values = await Promise.allSettled(requests);
+
+    const success = values.map(request => request.status === "fulfilled")
+
+    this.logger.info("Updates with succes: " + success.length);
+
+    values.forEach(promise => {
+      this.logger.info("Promise finish with status: " + promise.status)
+
+      if (promise.status === "rejected") {
+        this.logger.error("reject with: " + JSON.stringify(promise.reason || {}));
+        throw new Error(promise.reason );
+      }
+
+      if (promise.status === "fulfilled") {
+        this.logger.info("success with: " + JSON.stringify(promise.value || {}));
+      }
+    });
+
+  }
+
+  async updatePlatformJobs(pipelineId: string, pipelineType: string, updateInputDTO: UpdatePlatformInputRequest, user: RequestUser) {
+    const jobsUpdated = [];
+
+    for (const [index, table] of updateInputDTO.tables.entries()) {
+      const jobUpdate = {
+        job_id: `${pipelineId}_${index}`,
+      }
+
+      if (table.type !== "incremental_with_qualify") {
+        delete table.destinations?.qualify;
+      }
+
+      if (table.memory) {
+        jobUpdate["memory"] = {
+          amount: table.memory * 1000
+        }
+      }
+      
+      this.logger.info('Updating input reference for table: ' + table.name);
+      let hasUpdateSyncMode = false;
+
+      const jobSyncMode = {}
+
+      if (table.columns) {
+        hasUpdateSyncMode = true;
+        jobSyncMode['column_include_list'] = table.columns;
+      }
+
+      if (table.reference_column) {
+        hasUpdateSyncMode = true;
+        jobSyncMode['incremental_column_name'] = table.reference_column.name;
+        jobSyncMode['incremental_column_type'] = table.reference_column.type;
+      }
+
+      if (table.identifier_columns) {
+        hasUpdateSyncMode = true;
+        jobSyncMode['primary_keys'] = table.identifier_columns;
+      }
+
+      if (table.type) {
+        hasUpdateSyncMode = true;
+        
+        jobSyncMode['target_load_type'] = table.type;
+      }
+
+      if(hasUpdateSyncMode) {
+        jobUpdate["sync_mode"] = jobSyncMode;
+      }
+
+      if (Object.keys(table.destinations).length > 1) {
+        let hasChanges = false
+        const jobRenameTables = {
+            raw: {},
+            qualify: {}
+        }
+
+        if (Object.keys(table.destinations.raw).length > 1) {
+          hasChanges = true;
+          jobRenameTables.raw = table.destinations.raw;
+        }
+
+        if (Object.keys(table.destinations.qualify).length > 1) {
+          hasChanges = true;
+          jobRenameTables.qualify = table.destinations.qualify;
+        }
+
+        if (hasChanges) {
+          jobUpdate['rename_tables'] = jobRenameTables;
+        }
+      }
+
+      jobsUpdated.push(jobUpdate);
+    }
+
+    this.logger.info('Request body:' + JSON.stringify({
+      jobs_updated: jobsUpdated
+    }));
+
+    const response = await this.platformAPI.proxy(
+      'PUT',
+        `/pipeline/${pipelineId}/jobs`,
+      user,
+      {
+        job_updates: jobsUpdated
+      }
+    )
+    this.logger.info('Platform api response: ' + JSON.stringify(response));
+  }
+
+  async findAllDataAssetByPipeline(data: {
+    pipeline: string,
+    object?: string
+  }, user: RequestUser) {
+    const metadata = PackTheMetadata(user);
+
+    const isDataAdmin = user.permissions.includes(
+      PERMISSIONS_GROUPS.CATALOG.permissions.DATA_MANAGER.seqid,
+    );
+
+    let has_permission = false;
+
+    const {
+      data_assets: resultString
+    } = await lastValueFrom(
+      this.pipelineReadService.FindAllDataAssetByPipeline(data, metadata)
+    );
+
+    const result = JSON.parse(resultString) as any;
+    const data_assets: IDataAsset[] = []
+    result.forEach(data_asset => {
+      if (data_asset?.owner === user.username) has_permission = true;
+
+      for (const role of user.roles) {
+        if (data_asset.roles.includes(role)) has_permission = true;
+      }
+  
+      if (data_asset.users.includes(user.user_id)) has_permission = true;
+  
+      if (isDataAdmin || has_permission) {
+        delete data_asset.p_roles;
+        delete data_asset.p_users;
+        data_assets.push(data_asset as IDataAsset);
+      }
+    });
+
+    const assets = await this.catalogService.getAssetsUsersAndRoles(data_assets, user.customer_id);
+
+    return assets;
+  }
+
+  async getPipelineStatus(data) {
+    this.logger.info('PipelinesClientService - GetPipelineStatus');
+
+    const statusPipelineResponse = await lastValueFrom(
+      this.pipelineReadService.PipelineV2GetPipelineV2Status(data),
+    )
+      .then((res) => {
+        const statusArray =
+          res.status?.sort((a, b) => {
+            if (a.id < b.id) {
+              return 1;
+            } else {
+              return -1;
+            }
+          }) || [];
+        return { status: statusArray };
+      })
+      .catch((err) => {
+        this.logger.error(err.message);
+        throw new Error(err);
+      });
+    this.logger.info('Done');
+
+    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(
+      this.pipelineWriteService.PipelineV2TriggerPipelineV2({ id, info }),
+    ).catch((err) => {
+      this.logger.error(err.message);
+      throw new Error(err);
+    });
+
+    if (statusPipelineResponse.status == false) {
+      throw new ConflictException(
+        'This pipeline is not ready yet to execute, Try again later!',
+      );
+    }
+
+    this.logger.info('Done');
+    return statusPipelineResponse;
+  }
+
 }
@@ -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 });
+  });
+});
@@ -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 });
+      }
+    }
+  }
+}
@@ -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);
+  });
+});
@@ -10,21 +10,30 @@ import {
  Query,
  Inject,
  BadRequestException,
+  HttpException,
+  UseGuards,
 } from '@nestjs/common';
-import { ApiTags, ApiOperation } from '@nestjs/swagger';
+import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger';
 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 { 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';


 type ValidateTablesDTO = {
@@ -34,8 +43,14 @@ type ValidateTablesDTO = {
  }>
 }

+type RenameTablesBody = {
+  raw?: { table_name: string; table_schema: string };
+  qualify?: { table_name: string; table_schema: string };
+}
+
@ApiTags('Platform API')
@Controller('platform')
+@RequireModule(DADOSFERA_MODULES_KEYS.COLLECT)
 export class PlatformApiController {
  private logger: any;

@@ -44,6 +59,9 @@ export class PlatformApiController {
    private readonly elasticsearchService: ElasticsearchService,
    private readonly dynamoDBService: DynamoDBService,
    private readonly customersService: CustomersService,
+    private readonly catalogService: CatalogService,
+    private readonly inputsService: InputsService,
+    private readonly pipelineTablesService: PipelineTablesService,
    @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
  ) {
    this.logger = dadosferaLogger.logger;
@@ -57,6 +75,10 @@ export class PlatformApiController {
    return id?.replace(/-/g, '_') || '';
  }

+  private decodePathParam(value: string): string {
+    return value ? decodeURIComponent(value) : '';
+  }
+
  /**
   * Denormalize ID back to UUID format (replace _ with -).
   * Used when we receive a normalized ID but need the original UUID.
@@ -75,6 +97,23 @@ export class PlatformApiController {
    return jobId?.replace(/-/g, '_') || '';
  }

+  private async getJobByAnyConnectorType(normalizedJobId: string, user: RequestUser): Promise<any> {
+    const connectorTypes = ['jdbc', 'singer', 's3'];
+    for (const type of connectorTypes) {
+      try {
+        const job = await this.platformApiService.proxy(
+          'GET',
+          `/jobs/${type}/${normalizedJobId}`,
+          user,
+        );
+        return job;
+      } catch (error) {
+        // Continue to next connector type
+      }
+    }
+    throw new HttpException(`Job ${normalizedJobId} not found in any connector type (jdbc, singer, s3)`, 404);
+  }
+
  /**
   * Extract the pipeline ID (base UUID) from a job ID.
   * Job IDs have format "uuid-suffix" where suffix is the job index (e.g., "0", "1").
@@ -102,7 +141,7 @@ export class PlatformApiController {
  }

  private readonly VALID_CONNECTORS = ['jdbc', 'singer', 's3'];
-  private readonly MAX_MEMORY_MB = 12000; // 12GB maximum memory per pipeline/job
+  private readonly MAX_MEMORY_MB = 12000; // 12GB maximum memory per pipelines/job

  /**
   * Validate that connector is provided and is a valid type.
@@ -373,55 +412,9 @@ export class PlatformApiController {
    }
  }

-  /**
-   * Sync sync-mode changes to DynamoDB for JDBC connectors.
-   * Always passes both target_load_type and incremental_column_name to ensure proper sync.
-   */
-  private async syncJdbcSyncModeToDynamoDB(
-    jobId: string,
-    body: any,
-    user: RequestUser,
-  ): Promise<void> {
-    // JDBC sync mode uses target_load_type field
-    const changes: any = {};
-
-    if ('target_load_type' in body) {
-      changes.target_load_type = body.target_load_type;
-    }
-
-    // Handle incremental_column_name:
-    // - If provided in body, use that value
-    // - If changing to full_load, explicitly clear it
-    if ('incremental_column_name' in body) {
-      changes.incremental_column_name = body.incremental_column_name;
-      changes.incremental_column_type = body.incremental_column_type;
-    } else if (body.target_load_type === 'full_load') {
-      // Changing to full_load without specifying incremental_column - clear it
-      changes.incremental_column_name = null;
-    }
-
-    await this.syncJobInputToDynamoDB(jobId, changes, user, 'jdbc');
-  }
-
-  /**
-   * Sync sync-mode changes to DynamoDB for Singer connectors.
-   */
-  private async syncSingerSyncModeToDynamoDB(
-    jobId: string,
-    body: any,
-    user: RequestUser,
-  ): Promise<void> {
-    // Singer sync mode uses replication_method field
-    // Map to DynamoDB type: FULL_TABLE -> full_load, INCREMENTAL -> incremental
-    if ('replication_method' in body) {
-      const type = body.replication_method === 'INCREMENTAL' ? 'incremental' : 'full_load';
-      await this.syncJobInputToDynamoDB(jobId, { load_type: type }, user, 'singer');
-    }
-  }
-
  // ==================== PIPELINE ROUTES ====================

-  @Post('pipeline')
+  @Post('pipelines')
  @ApiOperation({ summary: 'Create a new pipeline' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE)
  async createPipeline(@Body() body: any, @User() user: RequestUser) {
@@ -542,7 +535,7 @@ export class PlatformApiController {
    );
  }

-  @Get('pipeline/:pipelineId')
+  @Get('pipelines/:pipelineId')
  @ApiOperation({ summary: 'Get pipeline by ID' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getPipeline(
@@ -553,7 +546,21 @@ export class PlatformApiController {
    return this.platformApiService.proxy('GET', `/pipeline/${normalizedId}`, user);
  }

-  @Patch('pipeline/:pipelineId')
+  @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)
  async updatePipeline(
@@ -604,7 +611,7 @@ export class PlatformApiController {
    return result;
  }

-  @Delete('pipeline/:pipelineId')
+  @Delete('pipelines/:pipelineId')
  @ApiOperation({ summary: 'Delete pipeline by ID' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
  async deletePipeline(
@@ -634,7 +641,7 @@ export class PlatformApiController {
    return result;
  }

-  @Post('pipeline/execute')
+  @Post('pipelines/execute')
  @ApiOperation({ summary: 'Execute a pipeline' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async executePipeline(@Body() body: any, @User() user: RequestUser) {
@@ -644,10 +651,10 @@ export class PlatformApiController {
      ...body,
      customer_id: user.customer_name,
    };
-    return this.platformApiService.proxy('POST', '/pipeline/execute', user, enrichedBody);
+    return this.platformApiService.proxy('POST', '/pipelines/execute', user, enrichedBody);
  }

-  @Post('pipeline/pause')
+  @Post('pipelines/pause')
  @ApiOperation({ summary: 'Pause a pipeline' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async pausePipeline(@Body() body: any, @User() user: RequestUser) {
@@ -660,7 +667,7 @@ export class PlatformApiController {
    return this.platformApiService.proxy('POST', '/pipeline/pause', user, enrichedBody);
  }

-  @Post('pipeline/unpause')
+  @Post('pipelines/unpause')
  @ApiOperation({ summary: 'Unpause a pipeline' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async unpausePipeline(@Body() body: any, @User() user: RequestUser) {
@@ -673,7 +680,7 @@ export class PlatformApiController {
    return this.platformApiService.proxy('POST', '/pipeline/unpause', user, enrichedBody);
  }

-  @Put('pipeline/:pipelineId/memory')
+  @Put('pipelines/:pipelineId/memory')
  @ApiOperation({ summary: 'Update pipeline memory configuration' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async updatePipelineMemory(
@@ -696,7 +703,7 @@ export class PlatformApiController {

  // ==================== PIPELINE METADATA ROUTES ====================

-  @Put('pipeline/:pipelineId/metadata')
+  @Put('pipelines/:pipelineId/metadata')
  @ApiOperation({ summary: 'Update pipeline metadata' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
  async updatePipelineMetadata(
@@ -728,26 +735,10 @@ export class PlatformApiController {
    );
  }

-  // ==================== Catalog ROUTES ====================
-
-  @Get('pipelines/catalog/tables')
-  @ApiOperation({ summary: 'Get all tables available' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
-  async getAvailableTables(
-    @User() user: RequestUser,
-    @Query() query: Record<string, string>,
-  ) {
-    return this.platformApiService.proxy(
-      'GET',
-      '/catalog/tables',
-      user,
-      undefined,
-      query,
-    );
-  }
+  // ==================== PIPELINE VALIDATION ====================

  @Get('pipelines/catalog/schemas')
-  @ApiOperation({ summary: 'Get all schemas available' })
+  @ApiOperation({ summary: 'Get available schemas' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getAvailableSchemas(
    @User() user: RequestUser,
@@ -755,7 +746,7 @@ export class PlatformApiController {
  ) {
    return this.platformApiService.proxy(
      'GET',
-      '/catalog/schemas',
+      `/catalog/schemas`,
      user,
      undefined,
      query,
@@ -763,25 +754,25 @@ export class PlatformApiController {
  }

  @Post('pipelines/catalog/tables/validate')
-  @ApiOperation({ summary: 'Validate tables and schemas' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  @ApiOperation({ summary: 'Validate Table and Schema' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async validateTableAndSchema(
+    @Body() payload: ValidationTableDTO,
    @User() user: RequestUser,
    @Query() query: Record<string, string>,
-    @Body() validateTablesDto: ValidateTablesDTO[]
  ) {
    return this.platformApiService.proxy(
      'POST',
-      '/catalog/tables/validate',
+      `/catalog/tables/validate`,
      user,
-      validateTablesDto,
+      payload,
      query,
    );
  }

  // ==================== PIPELINE RUN ROUTES ====================

-  @Get('pipeline/:pipelineId/pipeline_run')
+  @Get('pipelines/:pipelineId/pipeline_run')
  @ApiOperation({ summary: 'Get pipeline runs for a pipeline' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getPipelineRuns(
@@ -799,7 +790,7 @@ export class PlatformApiController {
    );
  }

-  @Get('pipeline/:pipelineId/pipeline_run/:runId')
+  @Get('pipelines/:pipelineId/pipeline_run/:runId')
  @ApiOperation({ summary: 'Get specific pipeline run' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getPipelineRun(
@@ -816,7 +807,7 @@ export class PlatformApiController {
    );
  }

-  @Get('pipeline/pipeline_run/:runId/logs')
+  @Get('pipelines/pipeline_run/:runId/logs')
  @ApiOperation({ summary: 'Get pipeline run logs' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
  async getPipelineRunLogs(
@@ -834,6 +825,68 @@ export class PlatformApiController {
    );
  }

+  @Post('pipelines/:pipelineId/pipeline_run/:runId/cancel')
+  @ApiOperation({ summary: 'Cancel a running pipeline run' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  async cancelPipelineRun(
+    @Param('pipelineId') pipelineId: string,
+    @Param('runId') runId: string,
+    @User() user: RequestUser,
+  ) {
+    const normalizedPipelineId = this.normalizePipelineId(pipelineId);
+    const normalizedRunId = this.normalizePipelineId(runId);
+
+    const status = await this.platformApiService.proxy(
+      'GET',
+      `/pipeline/${normalizedPipelineId}/pipeline_run`,
+      user,
+    );
+
+    if (status.length === 1) {
+      throw new BadRequestException('The first pipeline cannot be canceled');
+    }
+
+    return this.platformApiService.proxy(
+      'POST',
+      `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`,
+      user,
+    );
+  }
+
+  @Get('pipelines/:pipelineId/pipeline_run/:runId/jobs')
+  @ApiOperation({
+    summary: 'Get pipeline run jobs',
+    description: 'Proxies platform-api DB-backed job runs and returns `{ jobs: [...] }`.',
+  })
+  @ApiOkResponse({
+    description: 'DB-backed job runs for the selected pipeline run.',
+    schema: {
+      type: 'object',
+      properties: {
+        jobs: {
+          type: 'array',
+          items: { type: 'object' },
+        },
+      },
+      required: ['jobs'],
+    },
+  })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
+  async getPipelineRunJobs(
+    @Param('pipelineId') pipelineId: string,
+    @Param('runId') runId: string,
+    @User() user: RequestUser,
+  ) {
+    const normalizedPipelineId = this.normalizePipelineId(pipelineId);
+    const decodedRunId = this.decodePathParam(runId);
+
+    return this.platformApiService.proxy(
+      'GET',
+      `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/jobs`,
+      user,
+    );
+  }
+
  // ==================== JOBS - COLUMN EDITING ROUTES ====================

  @Put('jobs/:jobId/input')
@@ -933,41 +986,51 @@ export class PlatformApiController {
    );
  }

-  // ==================== JOBS - JDBC SYNC MODE ROUTES ====================
-
-  @Get('jobs/jdbc/:jobId')
-  @ApiOperation({ summary: 'Get JDBC job details' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
-  async getJdbcJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
-    // Normalize job ID for Platform API (replace - with _)
-    const normalizedJobId = this.normalizeJobId(jobId);
-    return this.platformApiService.proxy('GET', `/jobs/jdbc/${normalizedJobId}`, user);
-  }
-
-  @Post('jobs/jdbc/:jobId/sync-mode')
-  @ApiOperation({ summary: 'Update JDBC job sync mode' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
-  async updateJdbcSyncMode(
-    @Param('jobId') jobId: string,
-    @Body() body: any,
+  @Delete('pipelines/:pipelineId/inputs/:inputId')
+  @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: DeleteTableBody,
    @User() user: RequestUser,
  ) {
-    // Normalize job ID for Platform API (replace - with _)
-    const normalizedJobId = this.normalizeJobId(jobId);
-
-    const result = await this.platformApiService.proxy(
-      'POST',
-      `/jobs/jdbc/${normalizedJobId}/sync-mode`,
-      user,
-      body,
-    );
-
-    // Sync to DynamoDB (pass raw jobId for pipeline extraction)
-    await this.syncJdbcSyncModeToDynamoDB(jobId, body, user);
-
-    return result;
+    return this.pipelineTablesService.removeTable(pipelineId, inputId, body.table_name, user);
  }

+  @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 ====================
+
  @Get('jobs/jdbc/configs/allowed_datatypes')
  @ApiOperation({ summary: 'Get allowed datatypes for JDBC' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
@@ -979,52 +1042,6 @@ export class PlatformApiController {
    );
  }

-  // ==================== JOBS - SINGER REPLICATION ROUTES ====================
-
-  @Get('jobs/singer/:jobId')
-  @ApiOperation({ summary: 'Get Singer job details' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
-  async getSingerJob(@Param('jobId') jobId: string, @User() user: RequestUser) {
-    // Normalize job ID for Platform API (replace - with _)
-    const normalizedJobId = this.normalizeJobId(jobId);
-    return this.platformApiService.proxy('GET', `/jobs/singer/${normalizedJobId}`, user);
-  }
-
-  @Post('jobs/singer/:jobId/sync-mode')
-  @ApiOperation({ summary: 'Update Singer job sync mode' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
-  async updateSingerSyncMode(
-    @Param('jobId') jobId: string,
-    @Body() body: any,
-    @User() user: RequestUser,
-  ) {
-    // Normalize job ID for Platform API (replace - with _)
-    const normalizedJobId = this.normalizeJobId(jobId);
-
-    const result = await this.platformApiService.proxy(
-      'POST',
-      `/jobs/singer/${normalizedJobId}/sync-mode`,
-      user,
-      body,
-    );
-
-    // Sync to DynamoDB (pass raw jobId for pipeline extraction)
-    await this.syncSingerSyncModeToDynamoDB(jobId, body, user);
-
-    return result;
-  }
-
-  // ==================== JOBS - S3 ROUTES ====================
-
-  @Get('jobs/s3/:jobId')
-  @ApiOperation({ summary: 'Get S3 job details' })
-  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
-  async getS3Job(@Param('jobId') jobId: string, @User() user: RequestUser) {
-    // Normalize job ID for Platform API (replace - with _)
-    const normalizedJobId = this.normalizeJobId(jobId);
-    return this.platformApiService.proxy('GET', `/jobs/s3/${normalizedJobId}`, user);
-  }
-
  // ==================== HEALTH ROUTE ====================

  @Get('health')
@@ -0,0 +1,66 @@
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+
+export class ValidationTableDTO {
+  @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[];
+}
@@ -1,17 +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],
+  imports: [
+    ElasticsearchModule,
+    DynamoDBModule,
+    CustomersModule,
+    CatalogModule,
+    InputsModule,
+    forwardRef(() => PipelinesV2Module),
+  ],
  controllers: [PlatformApiController],
-  providers: [PlatformApiService, DadosferaLogger],
+  providers: [PlatformApiService, PipelineTablesService, DadosferaLogger],
  exports: [PlatformApiService],
 })
 export class PlatformApiModule {}
@@ -89,6 +89,12 @@ export class PlatformApiService {

      // Propagate non-2xx responses as HttpExceptions
      if (response.status >= 400) {
+        this.logger.error('Platform API upstream error' + JSON.stringify({
+          status: response.status,
+          data: response.data,
+          path,
+          method: method.toUpperCase(),
+        }));
        throw new HttpException(response.data, response.status);
      }

@@ -101,6 +107,8 @@ export class PlatformApiService {
        method: method.toUpperCase(),
      });

+      this.logger.error(error)
+
      if (error instanceof HttpException) {
        throw error;
      }
@@ -0,0 +1,14 @@
+export type ReleaseNoteDTO = {
+    id: string;
+    date: string;
+    tag: string;
+    title: string;
+    visible: boolean;
+    expiryDate: string;
+    content: string;
+    showEmojis: boolean;
+    image?: string;
+    link?: string;
+    linkText?: string;
+  };
+  
@@ -0,0 +1,32 @@
+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,
+        {
+          provide: DadosferaLogger,
+          useValue: { logger },
+        },
+      ],
+    }).compile();
+
+    controller = module.get<ReleaseNoteController>(ReleaseNoteController);
+  });
+
+  it('should be defined', () => {
+    expect(controller).toBeDefined();
+  });
+});
@@ -0,0 +1,26 @@
+import { Controller, Get, Inject } from '@nestjs/common';
+import { ReleaseNoteService } from './release_note.service';
+import { Authenticated } from 'src/decorators/authentication.decorator';
+import { Language } from 'src/decorators/language.decorator';
+import { LanguageEnum } from 'src/utils/languages.enum';
+import DadosferaLogger from '@dadosfera/dadosfera-logs';
+
+@Controller('release_note')
+@Authenticated()
+export class ReleaseNoteController {
+  logger: DadosferaLogger;
+
+  constructor(
+    @Inject(DadosferaLogger)
+    dadosferaLogger: DadosferaLogger,
+    private readonly releaseNoteService: ReleaseNoteService,
+  ) {
+    this.logger = dadosferaLogger.logger;
+  }
+
+  @Get()
+  async getLatestReleaseNote(@Language() language: LanguageEnum) {
+    this.logger.info(`Fetching latest release note for language: ${language}`);
+    return await this.releaseNoteService.getLatestReleaseNote(language);
+  }
+}
@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { ReleaseNoteService } from './release_note.service';
+import { ReleaseNoteController } from './release_note.controller';
+import DadosferaLogger from '@dadosfera/dadosfera-logs';
+
+@Module({
+  controllers: [ReleaseNoteController],
+  providers: [ReleaseNoteService, DadosferaLogger]
+})
+export class ReleaseNoteModule {}
@@ -0,0 +1,30 @@
+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,
+        {
+          provide: DadosferaLogger,
+          useValue: { logger },
+        },
+      ],
+    }).compile();
+
+    service = module.get<ReleaseNoteService>(ReleaseNoteService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});
@@ -0,0 +1,46 @@
+import { Inject, Injectable } from '@nestjs/common';
+import axios, { AxiosInstance } from 'axios';
+import { LanguageEnum } from 'src/utils/languages.enum';
+import { ReleaseNoteDTO } from './dto/release_note.dto';
+import DadosferaLogger from '@dadosfera/dadosfera-logs';
+
+@Injectable()
+export class ReleaseNoteService {
+  client: AxiosInstance;
+    logger: DadosferaLogger;
+
+  constructor(
+        @Inject(DadosferaLogger)
+        dadosferaLogger: DadosferaLogger,
+  ) {
+    this.logger = dadosferaLogger.logger;
+      this.client = axios.create({
+          baseURL: process.env.FIREBASE_BASE_URL,
+      });
+  }
+
+  async getLatestReleaseNote(lang: LanguageEnum) {
+    try {
+      const lng = lang.split('-');
+      const language = lng[0] + '-' + lng[1].toUpperCase();
+
+      const endpoint = `/release_note/${language}.json`;
+      const {
+        data,
+        status,
+        config
+      } = await this.client.get<ReleaseNoteDTO>(endpoint)
+      this.logger.info(`Fetched release note for language: ${lang} with status: ${status}`);
+      this.logger.info(`Request URL: ${config.baseURL}/${config.url}`);
+
+      return data;
+    } catch (error) {
+      this.logger.error(`Error fetching release note: ${error.message}`);
+
+      if (axios.isAxiosError(error)) {
+        this.logger.error(`Axios error details: ${error.toJSON()}`);
+      }
+    }
+
+  }
+}
@@ -1,5 +1,8 @@
-import { Info } from '@dadosfera/protospack/dist/lib/interfaces';
-
+export interface Info {
+  user_id: string;
+  customer_id: string;
+  customer: string;
+}
 export interface ICreateTransformationsRequest {
  transformations: Transformation[];
  info: Info;
@@ -125,9 +125,13 @@ export class DynamoDBService {
      },
    });

-    const { Item } = await this.documentClient.send(getCommand);
-
-    return Item as InputDocument | null;
+    try {
+      const { Item } = await this.documentClient.send(getCommand);
+      return Item as InputDocument | null;
+    } catch (error) {
+      this.logger.error('DynamoDB: findInput failed', { inputId, clientId, error: error.message });
+      throw error;
+    }
  }

  async deleteInput(clientId: string, inputId: string): Promise<void> {
@@ -203,7 +207,6 @@ export class DynamoDBService {
        updatedTable.reference_column = changes.reference_column;
      }
    }
-
    tables[tableIndex] = updatedTable;

    // Save updated document
@@ -231,4 +234,5 @@ export class DynamoDBService {
      throw error;
    }
  }
+
 }
@@ -358,6 +358,81 @@ export class ElasticsearchService {
    }
  }

+  private getDataAssetIndex(customerName: string): string {
+    return `${customerName}_data_assets_catalog`;
+  }
+
+  async findDataAssetByTable(
+    customerName: string,
+    tableName: string,
+    tableSchema: string,
+  ): Promise<{ id: string; nimbus_id: number | null; [key: string]: any } | null> {
+    const index = this.getDataAssetIndex(customerName);
+
+    this.logger.info('Elasticsearch: Searching data asset', {
+      index,
+      tableName,
+      tableSchema,
+    });
+
+    try {
+      const response = await this.client.post(`/${index}/_search`, {
+        query: {
+          bool: {
+            must: [
+              { term: { 'table_name.keyword': tableName.toUpperCase() } },
+              { term: { 'table_schema.keyword': tableSchema.toUpperCase() } },
+            ],
+          },
+        },
+        size: 1,
+      });
+
+      const hits = response.data.hits?.hits || [];
+      if (hits.length === 0) {
+        this.logger.warn('Elasticsearch: Data asset not found', { tableName, tableSchema, index });
+        return null;
+      }
+
+      return { ...hits[0]._source, _es_id: hits[0]._id };
+    } catch (error) {
+      this.handleError('findDataAssetByTable', error, { tableName, tableSchema, index });
+      throw error;
+    }
+  }
+
+  async updateDataAsset(
+    customerName: string,
+    assetId: string,
+    updates: Record<string, any>,
+  ): Promise<any> {
+    const index = this.getDataAssetIndex(customerName);
+
+    this.logger.info('Elasticsearch: Updating data asset', {
+      index,
+      assetId,
+      fields: Object.keys(updates),
+    });
+
+    try {
+      const response = await this.client.post(
+        `/${index}/_update/${assetId}`,
+        { doc: updates },
+        { params: { refresh: 'wait_for' } },
+      );
+
+      this.logger.info('Elasticsearch: Data asset updated', {
+        assetId,
+        result: response.data.result,
+      });
+
+      return response.data;
+    } catch (error) {
+      this.handleError('updateDataAsset', error, { assetId, index });
+      throw error;
+    }
+  }
+
  private handleError(
    operation: string,
    error: any,
@@ -0,0 +1,9 @@
+import { Module } from "@nestjs/common";
+import { NimbusService } from "./nimbus.service";
+import DadosferaLogger from "@dadosfera/dadosfera-logs";
+
+@Module({
+  providers: [NimbusService, DadosferaLogger],
+  exports: [NimbusService],
+})
+export class NimbusServicesModule {}
@@ -0,0 +1,56 @@
+import DadosferaLogger from "@dadosfera/dadosfera-logs";
+import { Inject, Injectable } from "@nestjs/common";
+import axios from "axios";
+
+type TableUpdate = {
+    table_schema: string;
+    table_name: string;
+}
+
+@Injectable()
+export class NimbusService {
+    private logger: DadosferaLogger;
+  
+    constructor(
+      @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
+    ) {
+      this.logger = dadosferaLogger.logger;
+    }
+
+    private buildUrl(customerName: string) {
+        if (process.env.ENV === 'prd') {
+            return `https://nimbus-${customerName}.dadosfera.ai`;
+          }
+      
+          return `https://nimbus-${customerName}.${process.env.ENV.replace(
+            'local',
+            'stg',
+          )}.dadosfera.ai`;
+    }
+
+    async renameTable(customerName: string, database: string, old: TableUpdate, update: TableUpdate) {
+        const nimbusUrl = this.buildUrl(customerName);
+
+        const path = `/api/catalog/rename-tables/?database_name=${encodeURIComponent(database)}&table_name=${encodeURIComponent(old.table_name)}&table_schema=${encodeURIComponent(old.table_schema)}`;
+
+        try {
+            this.logger.info("Request for PATCH " + nimbusUrl + path);
+            this.logger.info("Payload: " + JSON.stringify(update));
+            const { data } = await axios.patch(nimbusUrl + path, {
+                table_name: update.table_name,
+                table_schema: update.table_schema
+            })
+
+            return data;
+        } catch (error) {
+            this.logger.error(error);
+            return {
+                message: error.message,
+                database,
+                old, 
+                update
+            }
+        }
+
+    }
+}
@@ -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,
@@ -9,6 +9,7 @@ interface IMetadata {
  details?: string;
  sensitive?: string;
  roles?: string[];
+  customer_modules?: string[];
  is_data_manager?: boolean;
  access_token?: string;
  host?: string;
@@ -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));
+}
@@ -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',