FIX: update connectionTestListSchemas to map cached columns and derive references from the engine allowlist

FIX: enhance getTableMetadata to filter columns by allowed data types
Merge pull request #518 from dadosfera/feature/cdc-connector
2026-11-15 13:59:09 +00:00 · 2026-09-04 08:19:42 -03:00 · 2026-09-03 12:19:08 -03: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
41 changed files with 2077 additions and 620 deletions
@@ -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.
@@ -29,6 +29,7 @@ COPY . .
 # unit test specific build
 FROM ci_image AS test
 ENV DUC_URL=0.0.0.0:50051
+ENV INFACTORY_URL=0.0.0.0:50052
 ENTRYPOINT ["npm", "run", "test"]


@@ -111,6 +111,8 @@ 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
@@ -9,6 +9,7 @@ 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

@@ -5071,6 +5071,65 @@
          ]
        }
      },
+      "/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": {
+        "get": {
+          "operationId": "PlatformApiController_getPipelineRunJobs",
+          "summary": "Get pipeline run jobs",
+          "description": "Proxies platform-api DB-backed job runs and returns `{ jobs: [...] }`.",
+          "parameters": [
+            {
+              "name": "pipelineId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            },
+            {
+              "name": "runId",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "DB-backed job runs for the selected pipeline run.",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object",
+                    "properties": {
+                      "jobs": {
+                        "type": "array",
+                        "items": {
+                          "type": "object"
+                        }
+                      }
+                    },
+                    "required": [
+                      "jobs"
+                    ]
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Platform API"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
      "/platform/jobs/{jobId}/input": {
        "put": {
          "operationId": "PlatformApiController_updateJobInput",
@@ -6104,6 +6163,48 @@
          ]
        }
      },
+      "/catalog/custom-properties": {
+        "get": {
+          "operationId": "CatalogController_getCustomPropertyDefinitions",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "type": "object"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Catalog"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
      "/catalog/data-asset/{id}": {
        "get": {
          "operationId": "CatalogController_getDataAsset",
@@ -6364,6 +6465,57 @@
              "access-token": []
            }
          ]
+        },
+        "patch": {
+          "operationId": "CatalogController_updateColumnsMetadata",
+          "parameters": [
+            {
+              "name": "dadosfera-lang",
+              "in": "header",
+              "required": false,
+              "schema": {
+                "enum": [
+                  "pt-br",
+                  "en-us"
+                ],
+                "type": "string"
+              }
+            },
+            {
+              "name": "id",
+              "required": true,
+              "in": "path",
+              "schema": {
+                "type": "string"
+              }
+            }
+          ],
+          "requestBody": {
+            "required": true,
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/UpdateColumnsMetadataRequest"
+                }
+              }
+            }
+          },
+          "responses": {
+            "200": {
+              "description": ""
+            }
+          },
+          "tags": [
+            "Catalog"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
        }
      },
      "/catalog/data-asset/{id}/preview": {
@@ -7842,6 +7994,94 @@
          ]
        }
      },
+      "/connection-test/refresh-catalog": {
+        "post": {
+          "operationId": "ConnectionTestController_refreshCatalog",
+          "parameters": [],
+          "requestBody": {
+            "required": true,
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/RefreshCatalogReq"
+                }
+              }
+            }
+          },
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "$ref": "#/components/schemas/RefreshCatalogRes"
+                  }
+                }
+              }
+            },
+            "202": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "$ref": "#/components/schemas/RefreshCatalogRes"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Connection Test"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
+      "/connection-test/refresh-catalog/status": {
+        "post": {
+          "operationId": "ConnectionTestController_refreshCatalogStatus",
+          "parameters": [],
+          "requestBody": {
+            "required": true,
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/RefreshCatalogStatusReq"
+                }
+              }
+            }
+          },
+          "responses": {
+            "200": {
+              "description": "",
+              "content": {
+                "application/json": {
+                  "schema": {
+                    "$ref": "#/components/schemas/RefreshCatalogRes"
+                  }
+                }
+              }
+            }
+          },
+          "tags": [
+            "Connection Test"
+          ],
+          "security": [
+            {
+              "access-token": []
+            },
+            {
+              "access-token": []
+            }
+          ]
+        }
+      },
      "/oauth/hubspot": {
        "get": {
          "operationId": "OauthController_oauthHubspot",
@@ -11715,6 +11955,35 @@
            "columns_metadata"
          ]
        },
+        "ColumnDescriptionDto": {
+          "type": "object",
+          "properties": {
+            "column_name": {
+              "type": "string"
+            },
+            "description": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "column_name",
+            "description"
+          ]
+        },
+        "UpdateColumnsMetadataRequest": {
+          "type": "object",
+          "properties": {
+            "columns": {
+              "type": "array",
+              "items": {
+                "$ref": "#/components/schemas/ColumnDescriptionDto"
+              }
+            }
+          },
+          "required": [
+            "columns"
+          ]
+        },
        "IData": {
          "type": "object",
          "properties": {
@@ -11817,6 +12086,37 @@
            "docs"
          ]
        },
+        "CustomPropertyDto": {
+          "type": "object",
+          "properties": {
+            "key": {
+              "type": "string"
+            },
+            "value": {
+              "type": "string"
+            },
+            "type": {
+              "type": "string",
+              "enum": [
+                "text",
+                "number",
+                "date",
+                "boolean"
+              ]
+            },
+            "color": {
+              "type": "string"
+            },
+            "emoji": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "key",
+            "value",
+            "type"
+          ]
+        },
        "IUpdateDataRequest": {
          "type": "object",
          "properties": {
@@ -11845,6 +12145,12 @@
            },
            "docs": {
              "type": "string"
+            },
+            "custom_properties": {
+              "type": "array",
+              "items": {
+                "$ref": "#/components/schemas/CustomPropertyDto"
+              }
            }
          },
          "required": [
@@ -12299,11 +12605,15 @@
            },
            "type": {
              "type": "string"
+            },
+            "is_primary_key": {
+              "type": "boolean"
            }
          },
          "required": [
            "name",
-            "type"
+            "type",
+            "is_primary_key"
          ]
        },
        "TableMetadataDto": {
@@ -12405,6 +12715,85 @@
            "checks"
          ]
        },
+        "RefreshCatalogReq": {
+          "type": "object",
+          "properties": {
+            "connection_id": {
+              "type": "string"
+            },
+            "plugin": {
+              "type": "string",
+              "enum": [
+                "oracle",
+                "mysql",
+                "postgresql",
+                "sqlserver",
+                "mysql_cdc",
+                "postgresql_cdc",
+                "oracle_cdc"
+              ]
+            }
+          },
+          "required": [
+            "connection_id",
+            "plugin"
+          ]
+        },
+        "RefreshCatalogRes": {
+          "type": "object",
+          "properties": {
+            "operation_result": {
+              "type": "boolean"
+            },
+            "status": {
+              "type": "string"
+            },
+            "session_id": {
+              "type": "string"
+            },
+            "date": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "operation_result",
+            "status",
+            "session_id",
+            "date"
+          ]
+        },
+        "RefreshCatalogStatusReq": {
+          "type": "object",
+          "properties": {
+            "connection_id": {
+              "type": "string"
+            },
+            "plugin": {
+              "type": "string",
+              "enum": [
+                "oracle",
+                "mysql",
+                "postgresql",
+                "sqlserver",
+                "mysql_cdc",
+                "postgresql_cdc",
+                "oracle_cdc"
+              ]
+            },
+            "session_id": {
+              "type": "string"
+            },
+            "date": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "connection_id",
+            "plugin",
+            "session_id",
+            "date"
+          ]
+        },
        "INote": {
          "type": "object",
          "properties": {
@@ -16,7 +16,7 @@
        "@aws-sdk/lib-dynamodb": "^3.414.0",
        "@aws-sdk/signature-v4": "^3.370.0",
        "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
-        "@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.4.tgz",
+        "@dadosfera/protospack-v2": "3.40.0-beta.21",
        "@grpc/grpc-js": "^1.9.3",
        "@grpc/proto-loader": "^0.7.9",
        "@nestjs/cli": "^9.5.0",
@@ -1735,9 +1735,9 @@
      }
    },
    "node_modules/@dadosfera/protospack-v2": {
-      "version": "3.41.0-cdc-iceberg.4",
-      "resolved": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.4.tgz",
-      "integrity": "sha512-QcRhM2wHgoRgXbHfpu+ki+Iozbz7qunKD544v2DCdD1fbeuI6kEit/S4napquSzmzP2qcMjANobyAsWabC/f2w==",
+      "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",
@@ -34,7 +34,7 @@
    "@aws-sdk/lib-dynamodb": "^3.414.0",
    "@aws-sdk/signature-v4": "^3.370.0",
    "@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
-    "@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.4.tgz",
+    "@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",
@@ -3,12 +3,23 @@ import { PipelineExecutionGuard } from './pipeline-execution.guard';

 const logger = { info: jest.fn(), error: jest.fn() };

-function buildGuard(proxyImpl: jest.Mock) {
-  const platformApiService: any = { proxy: proxyImpl };
-  return new PipelineExecutionGuard(
-    { logger } as any,
-    platformApiService,
+type Routes = {
+  pipeline?: unknown | Error;
+  runs?: unknown | Error;
+};
+
+// The guard makes up to two platform-api reads: GET /pipeline/{id} (type)
+// and, for batch only, GET /pipeline/{id}/pipeline_run (last run).
+function buildGuard(routes: Routes) {
+  const proxy = jest.fn((method: string, path: string) => {
+    const answer = path.endsWith('/pipeline_run') ? routes.runs : routes.pipeline;
+    return answer instanceof Error ? Promise.reject(answer) : Promise.resolve(answer);
+  });
+  const guard = new PipelineExecutionGuard(
+    { logger } as unknown as ConstructorParameters<typeof PipelineExecutionGuard>[0],
+    { proxy } as unknown as ConstructorParameters<typeof PipelineExecutionGuard>[1],
  );
+  return { guard, proxy };
 }

 function contextWith(pipelineId = 'abc-123') {
@@ -16,56 +27,96 @@ function contextWith(pipelineId = 'abc-123') {
    switchToHttp: () => ({
      getRequest: () => ({ params: { pipelineId }, user: {} }),
    }),
-  } as any;
+  } as unknown as Parameters<PipelineExecutionGuard['canActivate']>[0];
 }

+const cdcPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 't' } }] };
+const batchPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 't' } }] };
+
 describe('PipelineExecutionGuard', () => {
  afterEach(() => jest.clearAllMocks());

-  it('allows the edit when the pipeline has no run history (empty array)', async () => {
-    const guard = buildGuard(jest.fn().mockResolvedValue([]));
-    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  describe('CDC pipelines (explicit connector type)', () => {
+    it('allows the edit without consulting run history', async () => {
+      const { guard, proxy } = buildGuard({ pipeline: cdcPipeline });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+      expect(proxy).toHaveBeenCalledTimes(1);
+      expect(proxy).toHaveBeenCalledWith('GET', '/pipeline/abc_123', {});
+    });
+
+    it('is CDC when any job is a CDC job', async () => {
+      const mixed = { jobs: [...batchPipeline.jobs, ...cdcPipeline.jobs] };
+      const { guard, proxy } = buildGuard({ pipeline: mixed });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+      expect(proxy).toHaveBeenCalledTimes(1);
+    });
  });

-  it('allows the edit when the last run has no last_status', async () => {
-    const guard = buildGuard(jest.fn().mockResolvedValue([{}]));
-    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+  describe('batch pipelines', () => {
+    it('allows the edit when the pipeline never ran (empty run history)', async () => {
+      const { guard } = buildGuard({ pipeline: batchPipeline, runs: [] });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+    });
+
+    it('allows the edit when the last run has no last_status', async () => {
+      const { guard } = buildGuard({ pipeline: batchPipeline, runs: [{}] });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+    });
+
+    it('allows the edit when the pipeline is not running', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: [{ last_status: 'SUCCEEDED' }],
+      });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+    });
+
+    it('blocks with the is-running message when the pipeline is running', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: [{ last_status: 'RUNNING' }],
+      });
+      await expect(guard.canActivate(contextWith())).rejects.toThrow(
+        'Pipeline is running, cannot update input now',
+      );
+    });
+
+    it('treats a pipeline with no jobs as batch and checks its runs', async () => {
+      const { guard, proxy } = buildGuard({ pipeline: { jobs: [] }, runs: [] });
+      await expect(guard.canActivate(contextWith())).resolves.toBe(true);
+      expect(proxy).toHaveBeenCalledTimes(2);
+    });
+
+    it('does not double-wrap the is-running BadRequestException', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: [{ last_status: 'running' }],
+      });
+      await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf(
+        BadRequestException,
+      );
+      await expect(guard.canActivate(contextWith())).rejects.not.toThrow(
+        /Error checking pipeline status/,
+      );
+    });
  });

-  it('allows the edit when the pipeline is not running', async () => {
-    const guard = buildGuard(
-      jest.fn().mockResolvedValue([{ last_status: 'SUCCEEDED' }]),
-    );
-    await expect(guard.canActivate(contextWith())).resolves.toBe(true);
-  });
+  describe('failures', () => {
+    it('fails closed when the pipeline read fails', async () => {
+      const { guard } = buildGuard({ pipeline: new Error('platform down') });
+      await expect(guard.canActivate(contextWith())).rejects.toThrow(
+        'Error checking pipeline status: platform down',
+      );
+    });

-  it('blocks with the is-running message when the pipeline is running', async () => {
-    const guard = buildGuard(
-      jest.fn().mockResolvedValue([{ last_status: 'RUNNING' }]),
-    );
-    await expect(guard.canActivate(contextWith())).rejects.toThrow(
-      'Pipeline is running, cannot update input now',
-    );
-  });
-
-  it('wraps a genuine status-check failure (fail closed)', async () => {
-    const guard = buildGuard(
-      jest.fn().mockRejectedValue(new Error('platform down')),
-    );
-    await expect(guard.canActivate(contextWith())).rejects.toThrow(
-      'Error checking pipeline status: platform down',
-    );
-  });
-
-  it('does not double-wrap the is-running BadRequestException', async () => {
-    const guard = buildGuard(
-      jest.fn().mockResolvedValue([{ last_status: 'running' }]),
-    );
-    await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf(
-      BadRequestException,
-    );
-    await expect(guard.canActivate(contextWith())).rejects.not.toThrow(
-      /Error checking pipeline status/,
-    );
+    it('fails closed when the run-history read fails', async () => {
+      const { guard } = buildGuard({
+        pipeline: batchPipeline,
+        runs: new Error('runs down'),
+      });
+      await expect(guard.canActivate(contextWith())).rejects.toThrow(
+        'Error checking pipeline status: runs down',
+      );
+    });
  });
 });
@@ -16,6 +16,7 @@ import {
 import { PipelinesClientConfiguration } from 'src/modules/pipelinesV2/pipelines-client';
 import { PlatformApiService } from 'src/modules/platform-api/platform-api.service';
 import DadosferaLogger from '@dadosfera/dadosfera-logs';
+import { isCdcPipeline } from 'src/utils/cdc';

@Injectable()
 export class PipelineExecutionGuard implements CanActivate {
@@ -39,7 +40,22 @@ export class PipelineExecutionGuard implements CanActivate {
      const user = request.user;
      const idRegex = /[^0-9a-zA-Z_$]+/g;
      const convertedId = pipelineId.replace(idRegex, '_');
-  
+
+      // Decide by the pipeline's explicit type (platform job
+      // `input.connector`), never by the absence of run history.
+      const pipeline = await this.platformApiService.proxy(
+        'GET',
+        `/pipeline/${convertedId}`,
+        user,
+      );
+
+      if (isCdcPipeline(pipeline)) {
+        // CDC pipelines have no batch runs; the platform-api gates connector
+        // edits itself (require_pipeline_running on add/remove tables).
+        this.logger.info('PipelineExecutionGuard: CDC pipeline, no run to block on');
+        return true;
+      }
+
      const status = await this.platformApiService.proxy(
        'GET',
        `/pipeline/${convertedId}/pipeline_run`,
@@ -50,8 +66,7 @@ export class PipelineExecutionGuard implements CanActivate {

      this.logger.info('Pipeline current status response:' +  JSON.stringify(currentStatus));

-      // No run history (e.g. CDC pipelines never record batch runs) means
-      // nothing is executing — allow the edit rather than crash on .last_status.
+      // Batch pipeline that never ran yet: nothing can be executing.
      if (!currentStatus?.last_status) {
        return true;
      }
@@ -59,9 +74,9 @@ export class PipelineExecutionGuard implements CanActivate {
      if (currentStatus.last_status.toLowerCase() === 'running') {
        this.logger.error('Pipeline is running, cannot update input now');
        throw new BadRequestException('Pipeline is running, cannot update input now');
-      } else {
-        return true;
      }
+
+      return true;
    } catch (error) {
      // Preserve the deliberate is-running rejection; only wrap genuine
      // status-check failures (fail closed on those for a destructive gate).
@@ -71,6 +86,5 @@ export class PipelineExecutionGuard implements CanActivate {
      this.logger.error('Error in PipelineExecutionGuard: ' + error.message);
      throw new BadRequestException('Error checking pipeline status: ' + error.message);
    }
-
  }
 }
@@ -37,6 +37,7 @@ import {
  RequireAllPermissions,
 } from 'src/decorators/authentication.decorator';
 import { AuthClientService } from './auth.service';
+import { UserDTO } from './dtos/login';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { GrpcToHttpExceptionFilter } from '../../error/grpc-to-http-exception.filter';
 import { RequestUser, User } from 'src/decorators/user.decorator';
@@ -486,7 +487,7 @@ export class AuthController {
      this.logger.info('Authenticating via X-Api-key header');
      const { api_key } = await this.apiKeyService.get(apiKey);

-      const userDto = {
+      const userDto: UserDTO = {
        id: api_key.user_id,
        name: api_key.username,
        email: api_key.username,
@@ -494,7 +495,8 @@ export class AuthController {
          id: api_key.customer_id,
          name: api_key.customer_name,
          tier: api_key.customer_tier,
-        }
+        },
+        permissions: [],
      };

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

    return userDto;
@@ -152,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,
@@ -51,6 +52,7 @@ import {
  IUpdateDataRequest,
  TriggerCatalogReq,
  TriggerCatalogRes,
+  UpdateColumnsMetadataRequest,
 } from './dtos';
 import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
 import { Language } from 'src/decorators/language.decorator';
@@ -283,6 +285,23 @@ export class CatalogController {

  }

+  @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,
@@ -431,6 +450,42 @@ 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,
@@ -1056,4 +1111,4 @@ export class CatalogController {
      this.logger.error(error.message);
    }
  }
-}
+}
@@ -120,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;
@@ -463,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);
@@ -828,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,5 +1,13 @@
 import { ApiProperty, ApiPropertyOptional, PickType } from '@nestjs/swagger';
-import { IsEnum } from 'class-validator';
+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 {
@@ -198,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;
@@ -211,6 +240,8 @@ export class IUpdateDataRequest {
  share_type?: DataAssetShareType;
  @ApiPropertyOptional()
  docs?: string;
+  @ApiPropertyOptional({ type: [CustomPropertyDto] })
+  custom_properties?: CustomPropertyDto[];
 }

 export class IUpdateCertificationStatusRequest {
@@ -219,6 +250,26 @@ export class IUpdateCertificationStatusRequest {
  certification_status: CertificationStatus;
 }

+export class ColumnDescriptionDto {
+  @ApiProperty()
+  @IsString()
+  @IsNotEmpty()
+  column_name: string;
+
+  @ApiProperty()
+  @IsString()
+  description: string;
+}
+
+export class UpdateColumnsMetadataRequest {
+  @ApiProperty({ type: [ColumnDescriptionDto] })
+  @IsArray()
+  @ArrayNotEmpty()
+  @ValidateNested({ each: true })
+  @Type(() => ColumnDescriptionDto)
+  columns: ColumnDescriptionDto[];
+}
+
 export class ICreateDataAsset implements CreateDataAssetRequest {
  @ApiProperty()
  display_name: string;
@@ -372,4 +423,4 @@ export type CreateDataDocsDTO = {
  docs: string;
  asset_type: string;

-}
+}
@@ -24,6 +24,9 @@ import {
  GetTableMetadataReq,
  ValidateCdcPrerequisitesReq,
  ValidateCdcPrerequisitesRes,
+  RefreshCatalogReq,
+  RefreshCatalogRes,
+  RefreshCatalogStatusReq,
 } from './dto/connection-test';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { Authenticated, RequireModule } from 'src/decorators/authentication.decorator';
@@ -90,7 +93,7 @@ export class ConnectionTestController {
    });
    return this.connectionTestService.connectionTestListSchemas(
      body,
-      user.customer_name,
+      user,
    );
  }

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

@@ -124,7 +127,7 @@ export class ConnectionTestController {
    });
    return this.connectionTestService.getTableMetadata(
      body,
-      user.customer_name,
+      user,
    );
  }

@@ -144,4 +147,35 @@ export class ConnectionTestController {
      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,263 @@
+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 and derives references from the engine allowlist', async () => {
+    connectionsApiService.proxy.mockResolvedValue({
+      columns: [
+        {
+          column_name: 'id',
+          data_type: 'bigint',
+          is_primary_key: true,
+        },
+        {
+          column_name: 'name',
+          data_type: 'varchar',
+          is_primary_key: false,
+        },
+      ],
+    });
+    platformApiService.proxy.mockResolvedValue({
+      allowed_datatypes: [
+        { engine: 'postgresql', allowed_datatypes: ['bigint', 'timestamp'] },
+      ],
+    });
+
+    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 },
+            { name: 'name', type: 'varchar', is_primary_key: false },
+          ],
+          references: [{ name: 'id', type: 'bigint', is_primary_key: true }],
+        },
+      ],
+    });
+    expect(connectionsApiService.proxy).toHaveBeenCalledWith(
+      'GET',
+      '/connection_catalog/config-id/schemas/public/tables/customers/columns',
+      user,
+    );
+    expect(platformApiService.proxy).toHaveBeenCalledWith(
+      'GET',
+      '/jobs/jdbc/configs/allowed_datatypes',
+      user,
+    );
+  });
+
+  it('returns empty references when the allowlist call fails', async () => {
+    connectionsApiService.proxy.mockResolvedValue({
+      columns: [{ column_name: 'id', data_type: 'bigint', is_primary_key: true }],
+    });
+    platformApiService.proxy.mockRejectedValue(new Error('platform down'));
+
+    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: [],
+        },
+      ],
+    });
+  });
+
+  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';
@@ -15,6 +15,9 @@ import {
  GetTableMetadataRes,
  ValidateCdcPrerequisitesReq,
  ValidateCdcPrerequisitesRes,
+  RefreshCatalogReq,
+  RefreshCatalogRes,
+  RefreshCatalogStatusReq,
 } from './dto/connection-test';
 import { ConnectionClientService } from '../connection/client.service';
 import {
@@ -23,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 {
@@ -30,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>(
@@ -149,46 +156,180 @@ 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;
-    return lastValueFrom(
-      this.connectionTestReadClient.GetTableMetadata({
-        connection_id,
-        customer_name,
-        plugin,
-        schema,
-        table_list,
+    // Columns eligible as the incremental reference field are the ones whose
+    // data type is allowed for this engine (e.g. int/date/timestamp). The
+    // allowlist is owned by the platform API, keyed by engine === plugin.
+    const allowedByEngine = await this.platformApiService
+      .proxy('GET', '/jobs/jdbc/configs/allowed_datatypes', user)
+      .catch(() => null);
+    const allowedDataTypes: string[] =
+      allowedByEngine?.allowed_datatypes?.find(
+        (datatypes) => datatypes.engine === body.plugin,
+      )?.allowed_datatypes ?? [];
+    const allowedSet = new Set(
+      allowedDataTypes.map((type) => type.toLowerCase()),
+    );
+
+    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,
+        );
+        const columns = result.columns.map((column) => ({
+          name: column.column_name,
+          type: column.data_type,
+          is_primary_key: column.is_primary_key,
+        }));
+        const references = columns.filter((column) =>
+          allowedSet.has(String(column.type).toLowerCase()),
+        );
+        return {
+          table_name,
+          columns,
+          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(
@@ -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()
@@ -168,3 +170,55 @@ export class ValidateCdcPrerequisitesRes {
  @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;
+}
@@ -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);
+    }
+  }
+}
@@ -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 ?? [],
+  };
+}
@@ -78,8 +78,10 @@ export class CdcColumnReq {
 export class CdcTableReq {
  @ApiProperty()
  name: string;
-  @ApiPropertyOptional()
-  table_schema?: string;
+  // Source database/schema. Required: Debezium addresses tables as
+  // `schema.table`, a CDC table without it cannot be replicated.
+  @ApiProperty()
+  table_schema: string;
  @ApiPropertyOptional({ type: [String] })
  primary_keys?: string[];
  // Per-table raw Iceberg table name override (iceberg destination only).
@@ -13,17 +13,21 @@ import { objectCamelToSnake } from 'src/utils/CaseConverter';
 import { IIdRequest, UpdateInputRequest } from './dtos/old_interfaces';
 import { Input } from '@dadosfera/protospack-v2';
 import {
+  AddCdcTableRequest,
  GetAvailableEntitiesRequest,
  InputCreateGenericRequest,
  InputCreateCdcRequest,
  InputCreateS3Request,
  InputNewCreateRequest,
  InputUpdateResponse,
+  MarkTableDeletedRequest,
+  RemoveCdcTableRequest,
  RollbackInputRequest,
  TestConnectionRequest,
 } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
 import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
 import { CreateCdcInputReq, CreateInputReq } from './dtos/input.model';
+import { toCdcTable } from './cdc-table.mapper';
 import { Metadata } from '@grpc/grpc-js';

@Injectable()
@@ -192,16 +196,7 @@ export class InputsService {
        name: body.name,
        plugin: body.plugin,
        read_only: body.read_only ?? true,
-        tables: body.tables.map((t) => ({
-          table_schema: t.table_schema,
-          table_name: t.name,
-          name: t.name, // canonical identity == table_name (in-factory also backfills)
-          primary_keys: t.primary_keys ?? [],
-          iceberg_table_name: t.iceberg_table_name,
-          iceberg_qualify_table_name: t.iceberg_qualify_table_name,
-          columns: t.columns ?? [],
-          column_exclude_list: t.column_exclude_list ?? [],
-        })),
+        tables: body.tables.map(toCdcTable),
        destination: body.destination,
      },
      info,
@@ -320,19 +315,19 @@ export class InputsService {
    return formatedPayload;
  }

-  async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
+  async markTableDeleted(data: MarkTableDeletedRequest) {
    return lastValueFrom(this.inputWriteService.MarkTableDeleted(data));
  }

-  async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
-    return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data));
+  async unmarkTableDeleted(data: MarkTableDeletedRequest) {
+    return lastValueFrom(this.inputWriteService.UnmarkTableDeleted(data));
  }

-  async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) {
-    return lastValueFrom(this.inputWriteService.AddCdcTable(data as any));
+  async addCdcTable(data: AddCdcTableRequest) {
+    return lastValueFrom(this.inputWriteService.AddCdcTable(data));
  }

-  async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) {
-    return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data));
+  async removeCdcTable(data: RemoveCdcTableRequest) {
+    return lastValueFrom(this.inputWriteService.RemoveCdcTable(data));
  }
 }
@@ -9,6 +9,33 @@ export class PipelineInputsDTO {
  }>
 }

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

  @ApiProperty()
  connector_name: string;
@@ -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,
      },
@@ -1,3 +1,4 @@
+import { isCdcPlugin } from 'src/utils/cdc';
 /* eslint-disable no-async-promise-executor */
 import {
  BadRequestException,
@@ -437,7 +438,7 @@ export class PipelinesService implements OnModuleInit {
    });

    this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput));
-    const isCdc = !!oldInput.plugin?.endsWith('_cdc');
+    const isCdc = isCdcPlugin(oldInput.plugin);
    const pipelineIdFormat = pipelineId.split('-').join('_');
    const rollback: RollbackPromise[] = [];

@@ -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 });
+      }
+    }
+  }
+}
@@ -5,53 +5,43 @@
 jest.mock('../customers/customers.service', () => ({ CustomersService: class {} }));
 jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} }));
 jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
-jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));
+jest.mock('./pipeline-tables.service', () => ({ PipelineTablesService: class {} }));

+import { BadRequestException } from '@nestjs/common';
 import { Test, TestingModule } from '@nestjs/testing';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';

 import { PlatformApiController } from './platform-api.controller';
 import { PlatformApiService } from './platform-api.service';
+import { PipelineTablesService } from './pipeline-tables.service';
 import { ElasticsearchService } from '../../services/elasticsearch';
 import { DynamoDBService } from '../../services/dynamodb';
 import { CustomersService } from '../customers/customers.service';
 import { CatalogService } from '../catalog/catalog.service';
 import { InputsService } from '../inputs/inputs.service';
-import { PipelinesService } from '../pipelinesV2/pipelines.service';

-const logger = {
-  info: (...args) => args,
-  error: (...args) => args,
-};
+const logger = { info: jest.fn(), error: jest.fn() };
+const mockUser = { customer_id: 'c1', customer_name: 'cust', user_id: 'u1' } as never;

-const mockUser: any = {
-  customer_id: 'c1',
-  customer_name: 'cust',
-  user_id: 'u1',
-};
-
-describe('PlatformApiController - deleteTable', () => {
+// The table routes are thin: validate the body, delegate to
+// PipelineTablesService (covered in pipeline-tables.service.spec.ts).
+describe('PlatformApiController - table routes', () => {
  let controller: PlatformApiController;
-  let platformApiService: { proxy: jest.Mock };
-  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };
+  let tables: { removeTable: jest.Mock; removeTables: jest.Mock; addTable: jest.Mock };

  beforeEach(async () => {
-    platformApiService = { proxy: jest.fn() };
-    inputsService = {
-      markTableDeleted: jest.fn(),
-      unmarkTableDeleted: jest.fn(),
-    };
+    tables = { removeTable: jest.fn(), removeTables: jest.fn(), addTable: jest.fn() };

    const module: TestingModule = await Test.createTestingModule({
      controllers: [PlatformApiController],
      providers: [
-        { provide: PlatformApiService, useValue: platformApiService },
+        { provide: PlatformApiService, useValue: {} },
        { provide: ElasticsearchService, useValue: {} },
        { provide: DynamoDBService, useValue: {} },
        { provide: CustomersService, useValue: {} },
        { provide: CatalogService, useValue: {} },
-        { provide: InputsService, useValue: inputsService },
-        { provide: PipelinesService, useValue: {} },
+        { provide: InputsService, useValue: {} },
+        { provide: PipelineTablesService, useValue: tables },
        { provide: DadosferaLogger, useValue: { logger } },
      ],
    }).compile();
@@ -59,306 +49,37 @@ describe('PlatformApiController - deleteTable', () => {
    controller = module.get<PlatformApiController>(PlatformApiController);
  });

-  it('CDC removal reconfigures the connector', async () => {
-    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
-    platformApiService.proxy.mockImplementation((method: string, path: string) => {
-      if (method === 'GET') {
-        return Promise.resolve({
-          jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } }],
-        });
-      }
-      return Promise.resolve({});
-    });
-
-    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);
-
-    expect(platformApiService.proxy).toHaveBeenCalledWith(
-      'DELETE',
-      '/pipeline/pid/jobs',
-      mockUser,
-      { job_ids: ['p_0'], delete_snowflake_tables: false },
-    );
-    const deletedViaJobsRoute = platformApiService.proxy.mock.calls.some(
-      ([method, path]: any[]) => method === 'DELETE' && path === '/jobs/p_0',
-    );
-    expect(deletedViaJobsRoute).toBe(false);
+  it('deleteTable delegates', async () => {
+    tables.removeTable.mockResolvedValue({ name: 'pedidos', is_deleted: true });
+    await expect(controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser))
+      .resolves.toEqual({ name: 'pedidos', is_deleted: true });
+    expect(tables.removeTable).toHaveBeenCalledWith('pid', 'iid', 'pedidos', mockUser);
  });

-  it('batch removal unchanged', async () => {
-    inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' });
-    platformApiService.proxy.mockImplementation((method: string, path: string) => {
-      if (method === 'GET') {
-        return Promise.resolve({
-          jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 'pedidos' } }],
-        });
-      }
-      return Promise.resolve({});
-    });
-
-    await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser);
-
-    expect(platformApiService.proxy).toHaveBeenCalledWith(
-      'DELETE',
-      '/jobs/p_0',
-      mockUser,
-    );
-    const reconfiguredConnector = platformApiService.proxy.mock.calls.some(
-      ([method, path]: any[]) => method === 'DELETE' && path === '/pipeline/pid/jobs',
-    );
-    expect(reconfiguredConnector).toBe(false);
-  });
-});
-
-describe('PlatformApiController - addTable', () => {
-  let controller: PlatformApiController;
-  let inputsService: { addCdcTable: jest.Mock; removeCdcTable: jest.Mock };
-  let pipelinesClientService: { addCdcJobs: jest.Mock };
-
-  const body = {
-    table_name: 'orders',
-    table_schema: 'public',
-    primary_keys: ['id'],
-    destinations: {
-      raw: { table_schema: 'raw', table_name: 'orders' },
-      qualify: { table_schema: 'qualify', table_name: 'orders' },
-    },
-  };
-
-  beforeEach(async () => {
-    inputsService = {
-      addCdcTable: jest.fn(),
-      removeCdcTable: jest.fn(),
-    };
-    pipelinesClientService = {
-      addCdcJobs: jest.fn(),
-    };
-
-    const module: TestingModule = await Test.createTestingModule({
-      controllers: [PlatformApiController],
-      providers: [
-        { provide: PlatformApiService, useValue: { proxy: jest.fn() } },
-        { provide: ElasticsearchService, useValue: {} },
-        { provide: DynamoDBService, useValue: {} },
-        { provide: CustomersService, useValue: {} },
-        { provide: CatalogService, useValue: {} },
-        { provide: InputsService, useValue: inputsService },
-        { provide: PipelinesService, useValue: pipelinesClientService },
-        { provide: DadosferaLogger, useValue: { logger } },
-      ],
-    }).compile();
-
-    controller = module.get<PlatformApiController>(PlatformApiController);
+  it('deleteTables delegates', async () => {
+    tables.removeTables.mockResolvedValue({ deleted: true });
+    await controller.deleteTables('pid', 'iid', { table_names: ['a', 'b'] }, mockUser);
+    expect(tables.removeTables).toHaveBeenCalledWith('pid', 'iid', ['a', 'b'], mockUser);
  });

-  it('addTable: DynamoDB append then platform AddJobs, returns job_ids', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
-
-    const result = await controller.addTable('pid', 'iid', body, mockUser);
-
-    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
-      table: {
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        name: 'orders',
-        columns: [],
-      },
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-    expect(pipelinesClientService.addCdcJobs).toHaveBeenCalledWith({
-      pipeline_id: 'pid',
-      input_id: 'iid',
-      tables: [{
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        destinations: body.destinations,
-      }],
-      info: { customer_id: 'c1', user_id: 'u1', customer: 'cust' },
-    });
-    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
-    expect(inputsService.removeCdcTable).not.toHaveBeenCalled();
-
-    const addCdcTableOrder = inputsService.addCdcTable.mock.invocationCallOrder[0];
-    const addCdcJobsOrder = pipelinesClientService.addCdcJobs.mock.invocationCallOrder[0];
-    expect(addCdcTableOrder).toBeLessThan(addCdcJobsOrder);
+  it('deleteTables rejects an empty table_names before touching anything', async () => {
+    await expect(controller.deleteTables('pid', 'iid', { table_names: [] }, mockUser))
+      .rejects.toBeInstanceOf(BadRequestException);
+    expect(tables.removeTables).not.toHaveBeenCalled();
  });

-  it('addTable: carries iceberg_table_name on the added table through to AddCdcTable', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] });
-
-    const icebergBody = { ...body, iceberg_table_name: 'cdc_raw.public__orders' };
-
-    await controller.addTable('pid', 'iid', icebergBody, mockUser);
-
-    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
-      table: {
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        name: 'orders',
-        iceberg_table_name: 'cdc_raw.public__orders',
-        columns: [],
-      },
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-  });
-
-  it('addTable: carries columns on the added table through to AddCdcTable', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_4'], skipped: [] });
-
-    const columnsBody = {
-      ...body,
-      columns: [
-        { name: 'id', type: 'int', is_primary_key: true },
-        { name: 'descr', type: 'varchar(255)', is_primary_key: false },
-      ],
-    };
-
-    await controller.addTable('pid', 'iid', columnsBody, mockUser);
-
-    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
-      table: {
-        table_schema: 'public',
-        table_name: 'orders',
-        primary_keys: ['id'],
-        name: 'orders',
-        columns: [
-          { name: 'id', type: 'int', is_primary_key: true },
-          { name: 'descr', type: 'varchar(255)', is_primary_key: false },
-        ],
-      },
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-  });
-
-  it('addTable: rolls back the DynamoDB row when AddJobs fails', async () => {
-    inputsService.addCdcTable.mockResolvedValue({ input: {} });
-    pipelinesClientService.addCdcJobs.mockRejectedValue(new Error('platform down'));
-    inputsService.removeCdcTable.mockResolvedValue({});
-
-    await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down');
-
-    expect(inputsService.removeCdcTable).toHaveBeenCalledWith({
-      id: 'iid',
+  it('addTable delegates', async () => {
+    const body = {
      table_name: 'orders',
-      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
-    });
-  });
-});
-
-
-describe('PlatformApiController - deleteTables (batch)', () => {
-  let controller: PlatformApiController;
-  let platformApiService: { proxy: jest.Mock };
-  let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock };
-
-  beforeEach(async () => {
-    platformApiService = { proxy: jest.fn() };
-    inputsService = {
-      markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }),
-      unmarkTableDeleted: jest.fn().mockResolvedValue({}),
+      table_schema: 'public',
+      primary_keys: ['id'],
+      destinations: {
+        raw: { table_schema: 'raw', table_name: 'orders' },
+        qualify: { table_schema: 'qualify', table_name: 'orders' },
+      },
    };
-
-    const module: TestingModule = await Test.createTestingModule({
-      controllers: [PlatformApiController],
-      providers: [
-        { provide: PlatformApiService, useValue: platformApiService },
-        { provide: ElasticsearchService, useValue: {} },
-        { provide: DynamoDBService, useValue: {} },
-        { provide: CustomersService, useValue: {} },
-        { provide: CatalogService, useValue: {} },
-        { provide: InputsService, useValue: inputsService },
-        { provide: PipelinesService, useValue: {} },
-        { provide: DadosferaLogger, useValue: { logger } },
-      ],
-    }).compile();
-
-    controller = module.get<PlatformApiController>(PlatformApiController);
-  });
-
-  const pipelineWithJobs = () => ({
-    jobs: [
-      { job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } },
-      { job_id: 'p_1', input: { connector: 'cdc', table_name: 'clientes' } },
-      { job_id: 'p_2', input: { connector: 'cdc', table_name: 'produtos' } },
-    ],
-  });
-
-  it('removes N tables in ONE platform call (connectors reconfigured once)', async () => {
-    platformApiService.proxy.mockImplementation((method: string) =>
-      Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}),
-    );
-
-    await controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'produtos'] }, mockUser);
-
-    // one mark per table
-    expect(inputsService.markTableDeleted).toHaveBeenCalledTimes(2);
-
-    // exactly one DELETE to the batch endpoint, with BOTH job_ids
-    const deleteCalls = platformApiService.proxy.mock.calls.filter(
-      ([m, p]: any[]) => m === 'DELETE' && p === '/pipeline/pid/jobs',
-    );
-    expect(deleteCalls).toHaveLength(1);
-    expect(deleteCalls[0][3]).toEqual({
-      job_ids: ['p_0', 'p_2'],
-      delete_snowflake_tables: false,
-    });
-    // never the per-job route
-    const perJob = platformApiService.proxy.mock.calls.some(
-      ([m, p]: any[]) => m === 'DELETE' && String(p).startsWith('/jobs/'),
-    );
-    expect(perJob).toBe(false);
-  });
-
-  it('rolls back only this call\'s marks when the platform delete fails', async () => {
-    platformApiService.proxy.mockImplementation((method: string) => {
-      if (method === 'GET') return Promise.resolve(pipelineWithJobs());
-      return Promise.reject(new Error('platform boom'));
-    });
-
-    await expect(
-      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser),
-    ).rejects.toThrow();
-
-    // both marks rolled back, nothing else
-    expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(2);
-    const unmarked = inputsService.unmarkTableDeleted.mock.calls.map((c: any[]) => c[0].table_name).sort();
-    expect(unmarked).toEqual(['clientes', 'pedidos']);
-  });
-
-  it('rolls back the marks made so far if a later mark fails (atomic)', async () => {
-    // second mark fails → first must be rolled back, no platform delete attempted
-    inputsService.markTableDeleted
-      .mockResolvedValueOnce({ is_deleted: true, deleted_at: 't' })
-      .mockRejectedValueOnce(new Error('dynamo boom'));
-
-    await expect(
-      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'clientes'] }, mockUser),
-    ).rejects.toThrow();
-
-    expect(inputsService.unmarkTableDeleted).toHaveBeenCalledTimes(1);
-    expect(inputsService.unmarkTableDeleted.mock.calls[0][0].table_name).toBe('pedidos');
-    // never reached the platform delete
-    const attemptedDelete = platformApiService.proxy.mock.calls.some(([m]: any[]) => m === 'DELETE');
-    expect(attemptedDelete).toBe(false);
-  });
-
-  it('404s when a requested table has no matching job', async () => {
-    platformApiService.proxy.mockImplementation((method: string) =>
-      Promise.resolve(method === 'GET' ? pipelineWithJobs() : {}),
-    );
-
-    await expect(
-      controller.deleteTables('pid', 'iid', { table_names: ['pedidos', 'ghost'] }, mockUser),
-    ).rejects.toThrow();
-    // the successful mark (pedidos) must be rolled back
-    expect(inputsService.unmarkTableDeleted).toHaveBeenCalled();
+    tables.addTable.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
+    await expect(controller.addTable('pid', 'iid', body, mockUser)).resolves.toEqual({ job_ids: ['p_2'], skipped: [] });
+    expect(tables.addTable).toHaveBeenCalledWith('pid', 'iid', body, mockUser);
  });
 });
@@ -11,10 +11,9 @@ import {
  Inject,
  BadRequestException,
  HttpException,
-  NotFoundException,
  UseGuards,
 } from '@nestjs/common';
-import { ApiTags, ApiOperation } from '@nestjs/swagger';
+import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';

 import {
@@ -31,9 +30,9 @@ import { CustomersService } from '../customers/customers.service';
 import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
 import { CatalogService } from '../catalog/catalog.service';
 import { PackTheMetadata } from '../../utils/PackTheMetadata';
-import { ValidationTableDTO } from './platform-api.dto';
+import { AddCdcTableBody, DeleteTableBody, DeleteTablesBody, ValidationTableDTO } from './platform-api.dto';
 import { InputsService } from '../inputs/inputs.service';
-import { PipelinesService } from '../pipelinesV2/pipelines.service';
+import { PipelineTablesService } from './pipeline-tables.service';
 import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';


@@ -62,7 +61,7 @@ export class PlatformApiController {
    private readonly customersService: CustomersService,
    private readonly catalogService: CatalogService,
    private readonly inputsService: InputsService,
-    private readonly pipelinesClientService: PipelinesService,
+    private readonly pipelineTablesService: PipelineTablesService,
    @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
  ) {
    this.logger = dadosferaLogger.logger;
@@ -76,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.
@@ -850,6 +853,40 @@ export class PlatformApiController {
    );
  }

+  @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')
@@ -950,60 +987,16 @@ export class PlatformApiController {
  }

  @Delete('pipelines/:pipelineId/inputs/:inputId')
-  @ApiOperation({ summary: 'Mark a table as deleted and delete its associated job via platform-api' })
+  @ApiOperation({ summary: 'Mark a table as deleted and remove its job via platform-api' })
  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
  @UseGuards(PipelineExecutionGuard)
  async deleteTable(
    @Param('pipelineId') pipelineId: string,
    @Param('inputId') inputId: string,
-    @Body() body: { table_name: string },
+    @Body() body: DeleteTableBody,
    @User() user: RequestUser,
  ) {
-    const tableName = body.table_name;
-    const info = {
-      customer_id: user.customer_id,
-      customer: user.customer_name,
-      user_id: user.user_id,
-    };
-
-    this.logger.info('deleteTable: marking table as deleted', { inputId, tableName });
-    const updatedInput: any = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info });
-    this.logger.info('deleteTable: table marked as deleted', { inputId, tableName });
-
-    try {
-      const normalizedPipelineId = this.normalizePipelineId(pipelineId);
-      this.logger.info('deleteTable: fetching pipeline from platform-api', { pipelineId, normalizedPipelineId });
-      const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
-      this.logger.info('deleteTable: pipeline fetched', { jobCount: platformPipeline?.jobs?.length });
-
-      const job = platformPipeline?.jobs?.find((j: any) => j.input?.table_name === tableName);
-      if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`);
-
-      this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id });
-      if (job.input?.connector === 'cdc') {
-        // CDC: reconfigure the Debezium/Kafka-Connect connector (stop
-        // replicating this table); keep the landed Snowflake data.
-        await this.platformApiService.proxy(
-          'DELETE',
-          `/pipeline/${normalizedPipelineId}/jobs`,
-          user,
-          { job_ids: [job.job_id], delete_snowflake_tables: false },
-        );
-      } else {
-        await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user);
-      }
-      this.logger.info('deleteTable: job deleted', { jobId: job.job_id });
-
-      return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at };
-    } catch (error) {
-      this.logger.error('deleteTable: platform-api delete failed, attempting rollback', { tableName, error: error.message });
-      try {
-        await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: tableName, info });
-      } catch (rollbackError) {
-        this.logger.error('deleteTable: rollback failed', { tableName, error: rollbackError.message });
-      }
-      throw error;
-    }
+    return this.pipelineTablesService.removeTable(pipelineId, inputId, body.table_name, user);
  }

  @Delete('pipelines/:pipelineId/inputs/:inputId/tables')
@@ -1013,66 +1006,14 @@ export class PlatformApiController {
  async deleteTables(
    @Param('pipelineId') pipelineId: string,
    @Param('inputId') inputId: string,
-    @Body() body: { table_names: string[] },
+    @Body() body: DeleteTablesBody,
    @User() user: RequestUser,
  ) {
    const tableNames = body.table_names ?? [];
    if (tableNames.length === 0) {
      throw new BadRequestException('table_names must be a non-empty array');
    }
-    const info = {
-      customer_id: user.customer_id,
-      customer: user.customer_name,
-      user_id: user.user_id,
-    };
-
-    // 1) Soft-delete each table in DynamoDB, tracking which succeeded so a later
-    //    failure only rolls back the marks made in THIS call.
-    const marked: string[] = [];
-    const rollback = async () => {
-      for (const name of marked) {
-        try {
-          await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info });
-        } catch (rollbackError) {
-          this.logger.error('deleteTables: rollback failed', { tableName: name, error: rollbackError.message });
-        }
-      }
-    };
-
-    try {
-      for (const name of tableNames) {
-        await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info });
-        marked.push(name);
-      }
-
-      // 2) Resolve all table_names -> job_ids from the platform pipeline (one GET).
-      const normalizedPipelineId = this.normalizePipelineId(pipelineId);
-      const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
-      const jobs = platformPipeline?.jobs ?? [];
-
-      const jobIds: string[] = [];
-      for (const name of tableNames) {
-        const job = jobs.find((j: any) => j.input?.table_name === name);
-        if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`);
-        jobIds.push(job.job_id);
-      }
-
-      // 3) Remove them all in ONE platform call so the Debezium source + Snowflake
-      //    sink connectors are reconfigured a single time, not once per table.
-      await this.platformApiService.proxy(
-        'DELETE',
-        `/pipeline/${normalizedPipelineId}/jobs`,
-        user,
-        { job_ids: jobIds, delete_snowflake_tables: false },
-      );
-      this.logger.info('deleteTables: jobs deleted', { jobIds });
-
-      return { table_names: tableNames, deleted: true };
-    } catch (error) {
-      this.logger.error('deleteTables: failed, rolling back this call\'s marks', { tableNames, error: error.message });
-      await rollback();
-      throw error;
-    }
+    return this.pipelineTablesService.removeTables(pipelineId, inputId, tableNames, user);
  }

  @Post('pipelines/:pipelineId/inputs/:inputId/tables')
@@ -1082,77 +1023,10 @@ export class PlatformApiController {
  async addTable(
    @Param('pipelineId') pipelineId: string,
    @Param('inputId') inputId: string,
-    @Body() body: {
-      table_name: string;
-      table_schema: string;
-      primary_keys: string[];
-      destinations: {
-        raw: { table_schema: string; table_name: string };
-        qualify: { table_schema: string; table_name: string };
-      };
-      // Iceberg destination only (protospack CdcTable.iceberg_table_name);
-      // absent for snowflake, back-compat.
-      iceberg_table_name?: string;
-      // Per-table deduped (qualify) Iceberg table name (protospack
-      // CdcTable.iceberg_qualify_table_name); absent => same as the raw name.
-      iceberg_qualify_table_name?: string;
-      // Source column schema for iceberg deduped table pre-create (protospack CdcTable.columns).
-      columns?: { name: string; type: string; is_primary_key: boolean }[];
-      // Columns the user chose to ignore -> Debezium column.exclude.list.
-      column_exclude_list?: string[];
-    },
+    @Body() body: AddCdcTableBody,
    @User() user: RequestUser,
  ) {
-    const info = {
-      customer_id: user.customer_id,
-      customer: user.customer_name,
-      user_id: user.user_id,
-    };
-
-    const cdcTable = {
-      table_schema: body.table_schema,
-      table_name: body.table_name,
-      primary_keys: body.primary_keys,
-      name: body.table_name,
-      iceberg_table_name: body.iceberg_table_name,
-      iceberg_qualify_table_name: body.iceberg_qualify_table_name,
-      columns: body.columns ?? [],
-      column_exclude_list: body.column_exclude_list ?? [],
-    };
-
-    this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name });
-    await this.inputsService.addCdcTable({ id: inputId, table: cdcTable, info });
-    this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name });
-
-    try {
-      const customInfo = {
-        customer_id: user.customer_id,
-        user_id: user.user_id,
-        customer: user.customer_name,
-      };
-
-      const res = await this.pipelinesClientService.addCdcJobs({
-        pipeline_id: pipelineId,
-        input_id: inputId,
-        tables: [{
-          table_schema: body.table_schema,
-          table_name: body.table_name,
-          primary_keys: body.primary_keys,
-          destinations: body.destinations,
-        }],
-        info: customInfo,
-      });
-
-      return res;
-    } catch (error) {
-      this.logger.error('addTable: platform AddJobs failed, rolling back the DynamoDB row', { tableName: body.table_name, error: error.message });
-      try {
-        await this.inputsService.removeCdcTable({ id: inputId, table_name: body.table_name, info });
-      } catch (rbErr) {
-        this.logger.error('addTable: rollback failed', { error: rbErr.message });
-      }
-      throw error;
-    }
+    return this.pipelineTablesService.addTable(pipelineId, inputId, body, user);
  }

  // ==================== JOBS - JDBC SYNC MODE ROUTES ====================
@@ -1,9 +1,66 @@
-import { ApiProperty } from "@nestjs/swagger";
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';

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

 import { PlatformApiController } from './platform-api.controller';
 import { PlatformApiService } from './platform-api.service';
+import { PipelineTablesService } from './pipeline-tables.service';
 import { ElasticsearchModule } from '../../services/elasticsearch';
 import { DynamoDBModule } from '../../services/dynamodb';
 import { CustomersModule } from '../customers/customers.module';
@@ -21,7 +22,7 @@ import { PipelinesV2Module } from '../pipelinesV2/pipelines.module';
    forwardRef(() => PipelinesV2Module),
  ],
  controllers: [PlatformApiController],
-  providers: [PlatformApiService, DadosferaLogger],
+  providers: [PlatformApiService, PipelineTablesService, DadosferaLogger],
  exports: [PlatformApiService],
 })
 export class PlatformApiModule {}
@@ -1,7 +1,12 @@
 import { Test, TestingModule } from '@nestjs/testing';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { ReleaseNoteController } from './release_note.controller';
 import { ReleaseNoteService } from './release_note.service';
-import DadosferaLogger from '@dadosfera/dadosfera-logs';
+
+const logger = {
+  info: (...args) => args,
+  error: (...args) => args,
+};

 describe('ReleaseNoteController', () => {
  let controller: ReleaseNoteController;
@@ -9,7 +14,13 @@ describe('ReleaseNoteController', () => {
  beforeEach(async () => {
    const module: TestingModule = await Test.createTestingModule({
      controllers: [ReleaseNoteController],
-      providers: [ReleaseNoteService, DadosferaLogger],
+      providers: [
+        ReleaseNoteService,
+        {
+          provide: DadosferaLogger,
+          useValue: { logger },
+        },
+      ],
    }).compile();

    controller = module.get<ReleaseNoteController>(ReleaseNoteController);
@@ -1,13 +1,24 @@
 import { Test, TestingModule } from '@nestjs/testing';
+import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { ReleaseNoteService } from './release_note.service';
-import DadosferaLogger from '@dadosfera/dadosfera-logs';
+
+const logger = {
+  info: (...args) => args,
+  error: (...args) => args,
+};

 describe('ReleaseNoteService', () => {
  let service: ReleaseNoteService;

  beforeEach(async () => {
    const module: TestingModule = await Test.createTestingModule({
-      providers: [ReleaseNoteService, DadosferaLogger],
+      providers: [
+        ReleaseNoteService,
+        {
+          provide: DadosferaLogger,
+          useValue: { logger },
+        },
+      ],
    }).compile();

    service = module.get<ReleaseNoteService>(ReleaseNoteService);
@@ -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,
@@ -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',
Author	SHA1	Message	Date
viniciusgadea	b0ec305763	FIX: update connectionTestListSchemas to map cached columns and derive references from the engine allowlist	2026-09-04 08:19:42 -03:00
viniciusgadea	4543c1ad05	FIX: enhance getTableMetadata to filter columns by allowed data types	2026-09-03 12:19:08 -03:00
Rafael Santana	ec39b82843	Merge pull request #518 from dadosfera/feature/cdc-connector FIX: CDC live status decodes int64 offsets as numbers	2026-08-31 17:07:20 -03:00
RafaelandWOZCODE	20d3532c0f	FIX: address review round 2 on the CDC refactor - toCdcTable accepts the source table named either table_name (platform bodies) or name (create DTO), so call sites pass it point-free: body.tables.map(toCdcTable). The identity is still derived in one place. - PipelineTablesService declares its logger like every other maestro service (logger: DadosferaLogger assigned from the injected instance). Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-31 16:09:25 -03:00
Rafael Santana	9457fcc85e	Merge pull request #519 from dadosfera/refactor/cdc-review UPDATE: apply the CDC code-review policy (maestro #510)	2026-08-31 12:52:51 -03:00
RafaelandWOZCODE	73aa3504a6	UPDATE: add CLAUDE.md with the development policy from the CDC review Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-29 13:01:35 -03:00
RafaelandWOZCODE	2347f65b78	UPDATE: one platform route for table removal; table orchestration moves to PipelineTablesService Review (maestro #510): batch and CDC tables are both removed through DELETE /pipeline/{id}/jobs (the platform dispatches by type: Airflow refresh vs Kafka Connect reconfigure), so the connector 'if' in the controller is gone. deleteTable/deleteTables/addTable are now thin controller methods over PipelineTablesService (mark -> resolve jobs -> platform -> rollback), with typed request bodies. Behavior change for batch: the platform refuses to remove the LAST table of a pipeline (400 'Cannot remove all jobs'), where DELETE /jobs/{id} allowed it. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-29 12:57:49 -03:00
RafaelandWOZCODE	23f0554311	UPDATE: type CDC input calls against the protospack gRPC contracts Review (maestro #510): drop every 'as any' on the Input write client (AddCdcTable/RemoveCdcTable/UnmarkTableDeleted take the generated request types), type IPipelineV2.config, and build the CdcTable payload in one mapper so 'name' mirrors table_name in a single place. isCdc via helper, no '!!'. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-29 12:56:00 -03:00
RafaelandWOZCODE	27eed59a1f	UPDATE: PipelineExecutionGuard decides by the pipeline's explicit connector type Review (maestro #510): the guard inferred 'editable' from the absence of run history. It now reads GET /pipeline/{id} (DB only) and short-circuits on job.input.connector === 'cdc'; batch keeps the pipeline_run check. Adds src/utils/cdc.ts as the single isCdc* helper. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-29 12:48:28 -03:00
RafaelandWOZCODE	688615dbee	UPDATE: bump protospack-v2 to 3.40.0-beta.21 (live-status sink offsets) Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-29 12:16:18 -03:00
RafaelandWOZCODE	ce48e69162	FIX: decode int64 as Number in the PipelineV2 gRPC client PipelineV2SinkTableOffset.committed_offset is int64. Without a `longs` option proto-loader decodes it as a Long.js object ({low,high,unsigned}), which then serialized to the frontend as an object and blew up the card's DecimalPipe (NG02100) on every 10s poll — the visible symptom was the raw/deduped layers flickering. Kafka offsets fit in 2^53, so `longs: Number` is exact and keeps the JSON a plain number. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-28 19:45:06 -03:00
Rafael Santana	1400df0ab9	Merge pull request #510 from dadosfera/feature/cdc-connector Feature/cdc connector	2026-08-28 13:36:54 -03:00
RafaelandWOZCODE	b0fa8d29fc	test(maestro): set INFACTORY_URL in the test Docker target Beta's cache-first work added connection-test.service.spec.ts, whose import graph (connection/client.config.ts) reads process.env.INFACTORY_URL at load time. The Dockerfile test target only set DUC_URL, so that suite crashed at import ("Cannot read properties of undefined (reading 'startsWith')") in CI and in any bare `npm test` run. Add ENV INFACTORY_URL=0.0.0.0:50052 alongside the existing DUC_URL, matching the local-connection convention (0.0.0.0 => no SSL). Full suite: 52 passed, 5 skipped. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-28 10:32:56 -03:00
RafaelandWOZCODE	5c609a8df8	test(maestro): fix connection-test merge test failures - platform-api.controller.spec: addCdcTable expectations now include the CDC fields the controller threads (iceberg_table_name, iceberg_qualify_table_name, column_exclude_list) which were added by the CDC-iceberg work. - release_note specs: provide DadosferaLogger mock — ReleaseNoteService gained an @Inject(DadosferaLogger) dependency (from beta) without its specs being updated, so they failed DI resolution on merge. Full suite: 52 passed, 5 skipped, 0 failed. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-28 09:38:20 -03:00
RafaelandWOZCODE	28a11c961a	fix(cdc): allow CDC plugins in RefreshCatalogReq validator Beta's refresh-catalog RefreshCatalogReq DTO restricted plugin to oracle/mysql/postgresql/sqlserver. Under the cache-first catalog model (adopted for CDC in the beta merge), the CDC schema-fetch flow posts /connection-test/refresh-catalog with plugin=mysql_cdc, which the @IsIn rejected ("plugin must be one of: oracle, mysql, postgresql, sqlserver"). Add mysql_cdc/postgresql_cdc/oracle_cdc to the @IsIn and @ApiProperty enum, matching the platform connection-test SQS plugin set. docsfera.json regenerated. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-27 20:13:00 -03:00
RafaelandWOZCODE	107938aa19	Merge origin/beta into feature/cdc-connector Resolves conflicts for PR #510 (base: beta): - package.json: keep protospack ^3.40.0-beta.20 (carries CDC→Iceberg fields); package-lock.json reconciled (protospack was the only dep delta vs beta). - connection-test controller/service/dto: keep BOTH feature sets — our CDC prerequisites validation AND beta's refresh-catalog endpoints. - Adopt beta's cache-first catalog reads (connections-api proxy) over our gRPC path. connectionTestListTables now enriches each table with primary_keys derived from the cached columns endpoint (is_primary_key), preserving the CDC create flow's need for PKs under the new architecture; spec updated. - docsfera.json resolved to ours; regenerated on next app bootstrap. connection-test spec: 6 passed. tsc --noEmit: clean. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-27 19:37:31 -03:00
RafaelandWOZCODE	23a633badf	UPDATE: switch protospack-v2 to published @3.40.0-beta.20 Replaces the local file:../protospack-v2/...cdc-iceberg.4.tgz tarball reference with the published CodeArtifact version ^3.40.0-beta.20 (carries the CDC→Iceberg qualify_namespace / iceberg_qualify_table_name / column_exclude_list proto fields). Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-27 19:14:26 -03:00
vinicius gadea	c1f5c013ee	Merge pull request #514 from dadosfera/feat/migration-nimbus-data-beta Feat/migration nimbus data beta	2026-08-25 17:27:45 -03:00
viniciusgadea	ba609a9494	FEAT: add data_preview field to data asset creation in CatalogService	2026-08-25 17:14:22 -03:00
viniciusgadea	8916dfc46c	FEAT: update @dadosfera/protospack-v2 dependency to version 3.40.0-beta.19	2026-08-25 17:05:34 -03:00
viniciusgadea	cf4e0aea22	FEAT: update protospack-v2 dependency to version 3.40.0-beta.18 and refactor UpdateDataAsset method in CatalogService	2026-08-25 14:06:25 -03:00
viniciusgadea	9d57e69e74	Merge remote-tracking branch 'origin/beta' into feat/migration-nimbus-data-beta # Conflicts: # docsfera.json # src/modules/catalog/dtos/index.ts	2026-08-25 12:08:47 -03:00
viniciusgadea	4db15fceab	FEAT: add request body and DTOs for updating columns metadata in catalog	2026-08-25 09:36:02 -03:00
viniciusgadea	a4f4335e43	FEAT: update columns metadata endpoint to accept batch updates and add error handling for missing columns	2026-08-25 09:05:46 -03:00
RafaelandWOZCODE	bf0314f5b1	FIX: trigger release for /auth/me permission seqids (PR #513 ) PR #513 merged to beta but no semantic-release ran: its commits used conventional-commits prefixes (feat(auth):, fix(auth):) which the .releaserc.json eslint preset does not recognise, so commit-analyzer found no release-worthy change. This empty FIX: commit matches the eslint preset's releaseRules (tag FIX -> patch) to cut a beta release that includes the /auth/me permission-seqids change, so stg can deploy it. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-24 19:18:59 -03:00
Rafael Santana	c97180ac95	Merge pull request #513 from dadosfera/feat/auth-me-orchest-identity feat(auth): return the user's permission seqids from /auth/me	2026-08-24 17:00:15 -03:00
RafaelandWOZCODE	cac36f2c60	refactor(auth): /auth/me returns raw permission seqids Return payload.permissions verbatim (numeric seqids) instead of translating them to claim strings. Consumers own the seqid->meaning mapping. Drops permission-claims.ts entirely; UserDTO.permissions is now number[]. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-24 16:56:23 -03:00
RafaelandWOZCODE	cd21fd0b7b	refactor(auth): /auth/me returns permissions only (drop roles/modules) Keep Maestro a pure identity provider: /auth/me exposes the user's permission claim strings and nothing consumer-specific. Consumers derive whatever meaning they need (roles, module access, groups) from the claim vocabulary — claims are already namespaced group:action. - UserDTO: drop roles[]/modules[], keep permissions[]. - Helper shrinks to a generic seqid->claim translation (orchest-identity.ts -> permission-claims.ts, translateSeqidsToClaims). - api-key branch: permissions: [] only. The roles/modules derivation moves entirely to the consumer (Orchest's auth-server adapter). Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-24 16:30:33 -03:00
RafaelandWOZCODE	a5a685ee3f	fix(auth): derive Orchest identity from numeric seqids (JWT carries seqids not claim strings) The JWT `permissions` claim is an array of numeric seqids at runtime (see authentication.guard.ts / authentication.decorator.ts), not claim strings. deriveOrchestIdentity previously matched claim strings against this numeric array, so roles[]/modules[] were always empty for every real user. - deriveOrchestIdentity now takes number[] \| undefined and matches seqids sourced from PERMISSIONS_GROUPS (permissions.enum.ts) instead of hand-copied literals. - permissions is translated back to claim strings via a full seqid->claim catalog built once from PERMISSIONS_GROUPS; unknown seqids are dropped (auth-server ignores permissions[] in v1). - auth.controller.ts's api-key branch literal is now annotated `: UserDTO` so tsc enforces the three fields there. - Both spec files re-fixtured with numeric seqid inputs, including a mixed admin+module case and an exact claim-string translation assertion. Co-Authored-By: WOZCODE <contact@withwoz.com>	2026-08-24 16:03:20 -03:00
Rafael	a16fefe691	feat(auth): return permissions/roles/modules from /auth/me (all branches)	2026-08-24 15:45:09 -03:00
Rafael	9c57485031	feat(auth): pure helper deriving Orchest identity from permissions	2026-08-24 15:39:40 -03:00
viniciusgadea	6efc25ad5e	FEAT: add endpoint to update column description in catalog	2026-08-20 08:26:56 -03:00
marcos.rodrigues	e1b0e88bd8	Merge pull request #507 from dadosfera/bugfix/catalog-module-main Bugfix/catalog module main	2026-08-12 09:59:08 -03:00
marcos.rodrigues	cef1184908	Merge branch 'beta' into bugfix/catalog-module-main	2026-08-12 09:58:58 -03:00
marcos.rodrigues	31dda867d1	FIX: require collect module in endpoints	2026-08-12 09:57:23 -03:00
marcos.rodrigues	2bb280e8de	Merge pull request #505 from dadosfera/bugfix/catalog-module Bugfix/catalog module	2026-08-07 12:26:21 -03:00
marcos.rodrigues	00cbadbb45	Merge branch 'beta' into bugfix/catalog-module	2026-08-07 12:26:11 -03:00
marcos.rodrigues	47ad527d38	FIX: skip nimbus update when customer haven't catalog module	2026-08-07 12:18:37 -03:00
iruy-fr	51044a23b3	FIX: trigger cache connections rollout	2026-08-03 09:19:27 -03:00
yuri.rodrigues	bb29d126c1	Merge pull request #503 from dadosfera/feat/cache-connections-rollout feat(connection-test): refresh connection catalog cache	2026-07-31 20:56:11 -03:00
iruy-fr	9d0f449eeb	feat(connection-test): refresh connection catalog cache	2026-07-31 16:33:15 -03:00
iruy-fr	0eafa67e6f	FIX: trigger cache connections deployment	2026-07-31 09:56:05 -03:00
yuri.rodrigues	c3937472ec	Merge pull request #502 from dadosfera/feat/cache-connections-rollout feat: read connection metadata from catalog cache	2026-07-31 09:46:57 -03:00
iruy-fr	84d64424ca	feat: read connection metadata from catalog cache	2026-07-30 10:17:28 -03:00
vinicius gadea	f0bfc5c94b	Merge pull request #500 from dadosfera/feat/custom-properties FEAT: add color and emoji properties to CustomPropertyDto	2026-07-28 12:21:19 -03:00
viniciusgadea	ad86a6a698	FEAT: simplify color and emoji property definitions in docsfera.json	2026-07-28 12:01:50 -03:00
viniciusgadea	af3b11ad54	FEAT: add color and emoji properties to CustomPropertyDto	2026-07-28 07:42:18 -03:00
vinicius gadea	15860519e1	Merge pull request #498 from dadosfera/feat/custom-properties FEAT: add custom properties endpoints and DTOs for catalog management	2026-07-22 11:57:51 -03:00
viniciusgadea	cffda86eab	FEAT: add CustomPropertyDto and update IUpdateDataRequest to use custom properties array	2026-07-22 10:51:53 -03:00
viniciusgadea	a93fbfbbd9	FEAT: remove custom properties endpoints and DTOs from catalog management	2026-07-21 16:32:38 -03:00
viniciusgadea	e1cfc6e1a8	FEAT: add custom properties endpoints and DTOs for catalog management	2026-07-21 11:21:22 -03:00
vinicius gadea	1e63df6536	Merge pull request #497 from dadosfera/feat/documentation-status FEAT: add endpoint and logic to update data asset certification status	2026-07-20 13:44:43 -03:00
viniciusgadea	8eb7fd0169	FEAT: add endpoint and logic to update data asset certification status	2026-07-17 09:55:17 -03:00
vinicius gadea	f8c6a8b747	Merge pull request #496 from dadosfera/feat/documentation-status FIX: pin npm version to 10.8.2 in Dockerfile for consistency	2026-07-16 17:06:04 -03:00
viniciusgadea	8a9d6c2f9c	FIX: pin npm version to 10.8.2 in Dockerfile for consistency	2026-07-16 17:05:10 -03:00
vinicius gadea	bcb2a0b7cb	Merge pull request #495 from dadosfera/feat/documentation-status Feat/documentation status	2026-07-16 16:47:52 -03:00
viniciusgadea	2e181e70af	FEAT: rename documentation_status to certification_status in docs and update package.json for protospack versioning	2026-07-16 09:08:11 -03:00
vinicius gadea	ef615adb8e	Merge branch 'beta' into feat/documentation-status	2026-07-13 08:42:37 -03:00
viniciusgadea	e03900811b	FEAT: add documentation status enum and property to data asset	2026-07-13 08:33:35 -03:00
yuri.rodrigues	5bc5fb0977	Merge pull request #494 from dadosfera/feat/pipeline-run-jobs feat: add endpoint to retrieve pipeline run jobs	2026-06-23 13:38:55 -03:00
iruy-fr	011032e3d4	FEAT: update API title in docsfera.json to reflect project name	2026-06-23 11:58:28 -03:00
iruy-fr	17363e74f4	FEAT: simplify pipeline run jobs handling and normalize run ID usage	2026-06-23 11:54:59 -03:00
iruy-fr	63efff6adf	FEAT: enhance pipeline run jobs endpoint with error handling and response structure	2026-06-22 20:25:16 -03:00
iruy-fr	b5f569e522	Merge branch 'beta' into feat/pipeline-run-jobs # Conflicts: # docsfera.json	2026-06-19 17:08:34 -03:00
iruy-fr	5c77577992	feat: add endpoint to retrieve pipeline run jobs	2026-06-19 17:04:42 -03:00
marcos.rodrigues	4b9e113185	Merge pull request #493 from dadosfera/chore/remove-deprecated-lib FEAT: remove deprecated protospack lib	2026-06-14 15:10:22 -03:00
marcos.rodrigues	06d505c50a	FEAT: remove deprecated protospack lib	2026-06-13 21:07:14 -03:00
Marcos Rodrigues Silva	2177f6725c	Merge pull request #491 from dadosfera/hotfix/release-notes Hotfix/release notes	2026-05-25 14:10:46 -03:00
marcos-silva-rodrigues	65ab16236f	FEAT: update deployment to include firebase base url	2026-05-25 14:09:16 -03:00
marcos-silva-rodrigues	b4cc8151d7	FIX: release note endpoint	2026-05-25 14:01:00 -03:00
yuri.rodrigues	e8b982998f	Merge pull request #490 from dadosfera/feat/platform-job-rout Feat/platform job rout	2026-05-21 16:48:38 -03:00
iruy-fr	ccd4159c59	fix: validate workflow	2026-05-21 16:43:37 -03:00
iruy-fr	9e49abb40d	fix: validate workflow	2026-05-21 16:37:47 -03:00
iruy-fr	21a82b64f3	fix: validate workflow	2026-05-21 16:31:45 -03:00
iruy-fr	2414fcf21e	fix: validate workflow	2026-05-21 16:23:45 -03:00
iruy-fr	94fbdb2226	fix: validate workflow	2026-05-21 16:21:30 -03:00
iruy-fr	985170d7ae	chore: exposure from route pipeline run jobs to maestro	2026-05-21 16:12:35 -03:00
iruy-fr	b38b8f51e2	FEAT: Add endpoint to fetch pipeline run jobs	2026-05-20 15:34:49 -03:00
Marcos Rodrigues Silva	ea16e62d6b	Merge pull request #489 from dadosfera/feature/pipeline-upgrade FEAT: pipeline upgrade route	2026-05-11 12:58:06 -03:00
marcos-silva-rodrigues	6182705410	FEAT: pipeline upgrade route	2026-05-11 12:52:07 -03:00