diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 977c0fb..d293ed9 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -2,9 +2,21 @@ name: Test on: pull_request: branches: + - beta - main jobs: + # Blocks a local (file:/tarball/overlay) protospack-v2 dependency from + # reaching staging (beta) or prod (main). + protospack-dep-guard: + if: github.base_ref == 'beta' || github.base_ref == 'main' + runs-on: [self-hosted, prd] + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Check protospack-v2 is consumed from the registry + run: node scripts/check-protospack-dep.js + test: runs-on: [self-hosted, prd] env: diff --git a/Dockerfile b/Dockerfile index 97b68de..e6d1888 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,6 +29,7 @@ COPY . . # unit test specific build FROM ci_image AS test ENV DUC_URL=0.0.0.0:50051 +ENV INFACTORY_URL=0.0.0.0:50052 ENTRYPOINT ["npm", "run", "test"] diff --git a/docsfera.json b/docsfera.json index 1b0f444..eb4af4e 100644 --- a/docsfera.json +++ b/docsfera.json @@ -3001,6 +3001,45 @@ ] } }, + "/inputs/cdc": { + "post": { + "operationId": "InputsController_createCdc", + "parameters": [], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreateCdcInputReq" + } + } + } + }, + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Input" + } + } + } + }, + "201": { + "description": "" + } + }, + "tags": [ + "Inputs" + ], + "security": [ + { + "access-token": [] + } + ] + } + }, "/inputs/{id}": { "get": { "operationId": "InputsController_findOne", @@ -3960,6 +3999,264 @@ ] } }, + "/pipelinesV2/{id}/live-status": { + "get": { + "operationId": "PipelinesController_getLiveStatus", + "parameters": [ + { + "name": "dadosfera-lang", + "in": "header", + "required": false, + "schema": { + "enum": [ + "pt-br", + "en-us" + ], + "type": "string" + } + }, + { + "name": "id", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "PipelinesV2" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, + "/pipelinesV2/{id}/pause": { + "post": { + "operationId": "PipelinesController_pause", + "parameters": [ + { + "name": "dadosfera-lang", + "in": "header", + "required": false, + "schema": { + "enum": [ + "pt-br", + "en-us" + ], + "type": "string" + } + }, + { + "name": "id", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "PipelinesV2" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, + "/pipelinesV2/{id}/unpause": { + "post": { + "operationId": "PipelinesController_unpause", + "parameters": [ + { + "name": "dadosfera-lang", + "in": "header", + "required": false, + "schema": { + "enum": [ + "pt-br", + "en-us" + ], + "type": "string" + } + }, + { + "name": "id", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "PipelinesV2" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, + "/pipelinesV2/{id}/restart": { + "post": { + "operationId": "PipelinesController_restart", + "parameters": [ + { + "name": "dadosfera-lang", + "in": "header", + "required": false, + "schema": { + "enum": [ + "pt-br", + "en-us" + ], + "type": "string" + } + }, + { + "name": "id", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "PipelinesV2" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, + "/pipelinesV2/{id}/jobs/{jobId}/reset-state": { + "post": { + "operationId": "PipelinesController_resetJobState", + "parameters": [ + { + "name": "dadosfera-lang", + "in": "header", + "required": false, + "schema": { + "enum": [ + "pt-br", + "en-us" + ], + "type": "string" + } + }, + { + "name": "id", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "jobId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "PipelinesV2" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, "/transformations": { "post": { "operationId": "TransformationsController_create", @@ -4274,6 +4571,66 @@ ] } }, + "/platform/iceberg/namespaces": { + "get": { + "operationId": "PlatformApiController_getIcebergNamespaces", + "summary": "List existing Polaris Iceberg namespaces (CDC destination dropdown)", + "parameters": [], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, + "/platform/iceberg/tables/validate": { + "post": { + "operationId": "PlatformApiController_validateIcebergTables", + "summary": "Validate CDC Iceberg raw table names against Polaris", + "parameters": [], + "responses": { + "201": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, "/platform/pipelines/execute": { "post": { "operationId": "PlatformApiController_executePipeline", @@ -4967,6 +5324,91 @@ ] } }, + "/platform/pipelines/{pipelineId}/inputs/{inputId}/tables": { + "delete": { + "operationId": "PlatformApiController_deleteTables", + "summary": "Batch-remove tables from an input; reconfigures the CDC connectors once", + "parameters": [ + { + "name": "pipelineId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "inputId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "" + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + }, + "post": { + "operationId": "PlatformApiController_addTable", + "summary": "Add a CDC table to an input and dispatch its platform jobs", + "parameters": [ + { + "name": "pipelineId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "inputId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "201": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, "/platform/jobs/jdbc/configs/allowed_datatypes": { "get": { "operationId": "PlatformApiController_getJdbcAllowedDatatypes", @@ -7513,6 +7955,45 @@ ] } }, + "/connection-test/cdc-prerequisites": { + "post": { + "operationId": "ConnectionTestController_validateCdcPrerequisites", + "parameters": [], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ValidateCdcPrerequisitesReq" + } + } + } + }, + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ValidateCdcPrerequisitesRes" + } + } + } + } + }, + "tags": [ + "Connection Test" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, "/connection-test/refresh-catalog": { "post": { "operationId": "ConnectionTestController_refreshCatalog", @@ -10744,6 +11225,113 @@ "updated_at" ] }, + "CdcColumnReq": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "type": { + "type": "string" + }, + "is_primary_key": { + "type": "boolean" + } + }, + "required": [ + "name", + "type", + "is_primary_key" + ] + }, + "CdcTableReq": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "table_schema": { + "type": "string" + }, + "primary_keys": { + "type": "array", + "items": { + "type": "string" + } + }, + "iceberg_table_name": { + "type": "string" + }, + "iceberg_qualify_table_name": { + "type": "string" + }, + "columns": { + "type": "array", + "items": { + "$ref": "#/components/schemas/CdcColumnReq" + } + }, + "column_exclude_list": { + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": [ + "name" + ] + }, + "IcebergDestinationReq": { + "type": "object", + "properties": { + "namespace": { + "type": "string" + }, + "qualify_namespace": { + "type": "string" + } + }, + "required": [ + "namespace" + ] + }, + "CdcDestinationReq": { + "type": "object", + "properties": { + "iceberg": { + "$ref": "#/components/schemas/IcebergDestinationReq" + } + } + }, + "CreateCdcInputReq": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "plugin": { + "type": "string" + }, + "tables": { + "type": "array", + "items": { + "$ref": "#/components/schemas/CdcTableReq" + } + }, + "read_only": { + "type": "boolean" + }, + "destination": { + "$ref": "#/components/schemas/CdcDestinationReq" + } + }, + "required": [ + "name", + "plugin", + "tables" + ] + }, "ICreatePipelineV2Req": { "type": "object", "properties": { @@ -10780,6 +11368,9 @@ "properties": { "type": "object" }, + "config": { + "type": "object" + }, "connector_name": { "type": "string" }, @@ -10844,6 +11435,9 @@ "properties": { "type": "object" }, + "config": { + "type": "object" + }, "connector_name": { "type": "string" }, @@ -11934,6 +12528,24 @@ "schema" ] }, + "ConnectionTestListTablesEntry": { + "type": "object", + "properties": { + "table_name": { + "type": "string" + }, + "primary_keys": { + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": [ + "table_name", + "primary_keys" + ] + }, "ConnectionTestListTablesRes": { "type": "object", "properties": { @@ -11945,11 +12557,18 @@ "items": { "type": "string" } + }, + "tables": { + "type": "array", + "items": { + "$ref": "#/components/schemas/ConnectionTestListTablesEntry" + } } }, "required": [ "operation_result", - "table_list" + "table_list", + "tables" ] }, "GetTableMetadataReq": { @@ -12040,6 +12659,62 @@ "tables_metadata" ] }, + "ValidateCdcPrerequisitesReq": { + "type": "object", + "properties": { + "plugin": { + "type": "string" + }, + "connection_id": { + "type": "string" + } + }, + "required": [ + "plugin", + "connection_id" + ] + }, + "CdcCheckDto": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "expected": { + "type": "string" + }, + "actual": { + "type": "string" + }, + "passed": { + "type": "boolean" + } + }, + "required": [ + "name", + "expected", + "actual", + "passed" + ] + }, + "ValidateCdcPrerequisitesRes": { + "type": "object", + "properties": { + "operation_result": { + "type": "boolean" + }, + "checks": { + "type": "array", + "items": { + "$ref": "#/components/schemas/CdcCheckDto" + } + } + }, + "required": [ + "operation_result", + "checks" + ] + }, "RefreshCatalogReq": { "type": "object", "properties": { @@ -12052,7 +12727,10 @@ "oracle", "mysql", "postgresql", - "sqlserver" + "sqlserver", + "mysql_cdc", + "postgresql_cdc", + "oracle_cdc" ] } }, @@ -12096,7 +12774,10 @@ "oracle", "mysql", "postgresql", - "sqlserver" + "sqlserver", + "mysql_cdc", + "postgresql_cdc", + "oracle_cdc" ] }, "session_id": { diff --git a/package-lock.json b/package-lock.json index 5506401..30d0276 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,7 +16,7 @@ "@aws-sdk/lib-dynamodb": "^3.414.0", "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", - "@dadosfera/protospack-v2": "^3.40.0-beta.19", + "@dadosfera/protospack-v2": "^3.40.0-beta.20", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", @@ -1735,9 +1735,9 @@ } }, "node_modules/@dadosfera/protospack-v2": { - "version": "3.40.0-beta.19", - "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.19.tgz", - "integrity": "sha512-RT7BYWHD2aD945FXzDXHa/xM+sD2B2Ypepy/W/WS+RwKYTSMCfuUXGzK2lXcd9HP9+U+Zjfh90Lkktl++8enjg==", + "version": "3.40.0-beta.20", + "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.20.tgz", + "integrity": "sha512-A12jgcVMCylfXZyXZYLuZNFJBuEBV1ZYmo3w01qhemKrFAzXRjOnPTBGHngdWNHahIaEYdocieBoIzl0n6OvFw==", "license": "ISC", "dependencies": { "@grpc/grpc-js": "^1.9.3", diff --git a/package.json b/package.json index fd16bd9..02633dc 100644 --- a/package.json +++ b/package.json @@ -34,7 +34,7 @@ "@aws-sdk/lib-dynamodb": "^3.414.0", "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", - "@dadosfera/protospack-v2": "^3.40.0-beta.19", + "@dadosfera/protospack-v2": "^3.40.0-beta.20", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", diff --git a/scripts/check-protospack-dep.js b/scripts/check-protospack-dep.js new file mode 100644 index 0000000..1d44c0c --- /dev/null +++ b/scripts/check-protospack-dep.js @@ -0,0 +1,76 @@ +#!/usr/bin/env node +/* + * CI guard: fail if @dadosfera/protospack-v2 is consumed from a LOCAL ref + * (file:/link:/git/relative path/bare tarball) instead of the CodeArtifact + * registry. + * + * Only local consumption is blocked. Versions published to CodeArtifact — + * including alpha/beta/rc prereleases produced by the alpha/beta branches — + * are fine; those resolve to a registry URL in the lockfile. The thing that + * must NOT reach beta (staging) or main (prod) is a dependency wired to a + * local `npm pack` tarball / overlay. Runs in the PR test workflow for PRs + * targeting beta/main and exits non-zero on any local ref. + */ +const fs = require('fs'); +const path = require('path'); + +const PKG = '@dadosfera/protospack-v2'; +const root = path.resolve(__dirname, '..'); +const pkg = JSON.parse(fs.readFileSync(path.join(root, 'package.json'), 'utf8')); + +const problems = []; + +// A dependency SPEC is local if it's a filesystem path, symlink, git ref, or a +// bare tarball path. A plain semver (incl. prereleases like 3.35.0-beta.1) +// resolves from the registry and is allowed. +function isLocalSpec(spec) { + return /^(file:|link:|git[:+]|\.\.?\/|\/|~\/)/.test(spec) || spec.endsWith('.tgz'); +} + +const spec = + (pkg.dependencies && pkg.dependencies[PKG]) || + (pkg.devDependencies && pkg.devDependencies[PKG]); + +if (!spec) { + problems.push(`${PKG} is not listed as a dependency at all.`); +} else if (isLocalSpec(spec)) { + problems.push(`${PKG} points at a local path/tarball/git ref: "${spec}".`); +} + +// Also catch a lockfile resolved to a LOCAL ref even if package.json looks +// clean. A registry URL (https://.../-/*.tgz) is the normal published +// resolution and is fine — only file: refs and bare local tarball paths +// (no http host) are blocked. Prerelease VERSIONS are not flagged: an +// alpha/beta/rc published to CodeArtifact resolves to a registry URL. +const lockPath = path.join(root, 'package-lock.json'); +if (fs.existsSync(lockPath)) { + const lock = JSON.parse(fs.readFileSync(lockPath, 'utf8')); + const nodes = { ...(lock.packages || {}), ...(lock.dependencies || {}) }; + for (const [name, node] of Object.entries(nodes)) { + if (!name.includes('protospack-v2') || !node) continue; + const resolved = node.resolved || ''; + const isLocal = + resolved.startsWith('file:') || + (resolved.endsWith('.tgz') && !/^https?:\/\//.test(resolved)); + if (isLocal) { + problems.push( + `package-lock.json resolves ${PKG} to a local ref: "${resolved}".`, + ); + } + } +} + +if (problems.length) { + console.error('✗ protospack-v2 dependency guard FAILED:'); + for (const p of problems) console.error(' - ' + p); + console.error( + '\nMerging to beta/main requires ' + + PKG + + ' to come from CodeArtifact, not a local tarball/overlay. Publish ' + + 'protospack-v2 (a beta prerelease is fine for the beta branch) and ' + + 'repoint this dependency before merging.', + ); + process.exit(1); +} + +console.log(`✓ ${PKG} is consumed from the registry: "${spec}"`); diff --git a/src/guards/pipeline-execution.guard.spec.ts b/src/guards/pipeline-execution.guard.spec.ts new file mode 100644 index 0000000..0e2dbf9 --- /dev/null +++ b/src/guards/pipeline-execution.guard.spec.ts @@ -0,0 +1,71 @@ +import { BadRequestException } from '@nestjs/common'; +import { PipelineExecutionGuard } from './pipeline-execution.guard'; + +const logger = { info: jest.fn(), error: jest.fn() }; + +function buildGuard(proxyImpl: jest.Mock) { + const platformApiService: any = { proxy: proxyImpl }; + return new PipelineExecutionGuard( + { logger } as any, + platformApiService, + ); +} + +function contextWith(pipelineId = 'abc-123') { + return { + switchToHttp: () => ({ + getRequest: () => ({ params: { pipelineId }, user: {} }), + }), + } as any; +} + +describe('PipelineExecutionGuard', () => { + afterEach(() => jest.clearAllMocks()); + + it('allows the edit when the pipeline has no run history (empty array)', async () => { + const guard = buildGuard(jest.fn().mockResolvedValue([])); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('allows the edit when the last run has no last_status', async () => { + const guard = buildGuard(jest.fn().mockResolvedValue([{}])); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('allows the edit when the pipeline is not running', async () => { + const guard = buildGuard( + jest.fn().mockResolvedValue([{ last_status: 'SUCCEEDED' }]), + ); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('blocks with the is-running message when the pipeline is running', async () => { + const guard = buildGuard( + jest.fn().mockResolvedValue([{ last_status: 'RUNNING' }]), + ); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Pipeline is running, cannot update input now', + ); + }); + + it('wraps a genuine status-check failure (fail closed)', async () => { + const guard = buildGuard( + jest.fn().mockRejectedValue(new Error('platform down')), + ); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Error checking pipeline status: platform down', + ); + }); + + it('does not double-wrap the is-running BadRequestException', async () => { + const guard = buildGuard( + jest.fn().mockResolvedValue([{ last_status: 'running' }]), + ); + await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf( + BadRequestException, + ); + await expect(guard.canActivate(contextWith())).rejects.not.toThrow( + /Error checking pipeline status/, + ); + }); +}); diff --git a/src/guards/pipeline-execution.guard.ts b/src/guards/pipeline-execution.guard.ts index ffd651b..1b6ebcd 100644 --- a/src/guards/pipeline-execution.guard.ts +++ b/src/guards/pipeline-execution.guard.ts @@ -46,10 +46,16 @@ export class PipelineExecutionGuard implements CanActivate { user, ); - const currentStatus = status[status.length - 1] + const currentStatus = status?.[status.length - 1]; this.logger.info('Pipeline current status response:' + JSON.stringify(currentStatus)); - + + // No run history (e.g. CDC pipelines never record batch runs) means + // nothing is executing — allow the edit rather than crash on .last_status. + if (!currentStatus?.last_status) { + return true; + } + if (currentStatus.last_status.toLowerCase() === 'running') { this.logger.error('Pipeline is running, cannot update input now'); throw new BadRequestException('Pipeline is running, cannot update input now'); @@ -57,6 +63,11 @@ export class PipelineExecutionGuard implements CanActivate { return true; } } catch (error) { + // Preserve the deliberate is-running rejection; only wrap genuine + // status-check failures (fail closed on those for a destructive gate). + if (error instanceof BadRequestException) { + throw error; + } this.logger.error('Error in PipelineExecutionGuard: ' + error.message); throw new BadRequestException('Error checking pipeline status: ' + error.message); } diff --git a/src/modules/connection-test/connection-test.controller.ts b/src/modules/connection-test/connection-test.controller.ts index 043ab29..bf6d383 100644 --- a/src/modules/connection-test/connection-test.controller.ts +++ b/src/modules/connection-test/connection-test.controller.ts @@ -22,6 +22,8 @@ import { ConnectionTestListTablesRes, GetTableMetadataRes, GetTableMetadataReq, + ValidateCdcPrerequisitesReq, + ValidateCdcPrerequisitesRes, RefreshCatalogReq, RefreshCatalogRes, RefreshCatalogStatusReq, @@ -129,6 +131,23 @@ export class ConnectionTestController { ); } + @Post('cdc-prerequisites') + @ApiOkResponse({ type: ValidateCdcPrerequisitesRes }) + @HttpCode(HttpStatus.OK) + async validateCdcPrerequisites( + @User() user: RequestUser, + @Body(new ValidationPipe()) body: ValidateCdcPrerequisitesReq, + ) { + this.logger.info('/connection-test/cdc-prerequisites', { + user: user.user_id, + customer: user.customer_name, + }); + return this.connectionTestService.validateCdcPrerequisites( + body, + user.customer_name, + ); + } + @Post('refresh-catalog') @ApiOkResponse({ type: RefreshCatalogRes }) @HttpCode(HttpStatus.ACCEPTED) diff --git a/src/modules/connection-test/connection-test.service.spec.ts b/src/modules/connection-test/connection-test.service.spec.ts index aed103f..8398d48 100644 --- a/src/modules/connection-test/connection-test.service.spec.ts +++ b/src/modules/connection-test/connection-test.service.spec.ts @@ -45,10 +45,24 @@ describe('ConnectionTestService catalog cache', () => { }); }); - it('keeps the existing tables response contract', async () => { - connectionsApiService.proxy.mockResolvedValue({ - tables: [{ table_name: 'customers' }, { table_name: 'orders' }], - }); + it('lists tables and enriches each with its cached primary keys', async () => { + connectionsApiService.proxy + // list-tables call (names only from the catalog cache) + .mockResolvedValueOnce({ + tables: [{ table_name: 'customers' }, { table_name: 'orders' }], + }) + // per-table columns calls: customers has a PK, orders has none + .mockResolvedValueOnce({ + columns: [ + { column_name: 'id', data_type: 'bigint', is_primary_key: true }, + { column_name: 'name', data_type: 'text', is_primary_key: false }, + ], + }) + .mockResolvedValueOnce({ + columns: [ + { column_name: 'total', data_type: 'numeric', is_primary_key: false }, + ], + }); await expect( service.connectionTestListTables( @@ -62,6 +76,10 @@ describe('ConnectionTestService catalog cache', () => { ).resolves.toEqual({ operation_result: true, table_list: ['customers', 'orders'], + tables: [ + { table_name: 'customers', primary_keys: ['id'] }, + { table_name: 'orders', primary_keys: [] }, + ], }); }); diff --git a/src/modules/connection-test/connection-test.service.ts b/src/modules/connection-test/connection-test.service.ts index fe4930d..f14efc1 100644 --- a/src/modules/connection-test/connection-test.service.ts +++ b/src/modules/connection-test/connection-test.service.ts @@ -13,6 +13,8 @@ import { ConnectionTestPingRes, GetTableMetadataReq, GetTableMetadataRes, + ValidateCdcPrerequisitesReq, + ValidateCdcPrerequisitesRes, RefreshCatalogReq, RefreshCatalogRes, RefreshCatalogStatusReq, @@ -177,9 +179,33 @@ export class ConnectionTestService { `/schemas/${encodeURIComponent(body.schema)}/tables`, user, ); + const table_names: string[] = result.tables.map((table) => table.table_name); + // CDC create needs the primary keys per table (used to build the deduped + // Iceberg identifier-fields). The catalog-cache list-tables endpoint returns + // only names, so fetch each table's columns from the cache and keep the ones + // flagged is_primary_key. Reads hit the stored catalog snapshot (populated by + // refresh-catalog), never the live connection. + const tables = await Promise.all( + table_names.map(async (table_name) => { + const columns = await this.connectionsApiService.proxy( + 'GET', + `/connection_catalog/${encodeURIComponent(body.connection_id)}` + + `/schemas/${encodeURIComponent(body.schema)}` + + `/tables/${encodeURIComponent(table_name)}/columns`, + user, + ); + return { + table_name, + primary_keys: columns.columns + .filter((column) => column.is_primary_key) + .map((column) => column.column_name), + }; + }), + ); return { operation_result: true, - table_list: result.tables.map((table) => table.table_name), + table_list: table_names, + tables, }; } @@ -287,4 +313,18 @@ export class ConnectionTestService { date: body.date, }; } + + async validateCdcPrerequisites( + body: ValidateCdcPrerequisitesReq, + customer_name: string, + ): Promise { + const { plugin, connection_id } = body; + return lastValueFrom( + this.connectionTestReadClient.ValidateCdcPrerequisites({ + connection_id, + customer_name, + plugin, + }), + ); + } } diff --git a/src/modules/connection-test/dto/connection-test.ts b/src/modules/connection-test/dto/connection-test.ts index 3496d4c..20447b9 100644 --- a/src/modules/connection-test/dto/connection-test.ts +++ b/src/modules/connection-test/dto/connection-test.ts @@ -102,11 +102,20 @@ export class ConnectionTestListTablesReq { schema: string; } +export class ConnectionTestListTablesEntry { + @ApiProperty() + table_name: string; + @ApiProperty({ type: [String] }) + primary_keys: string[]; +} + export class ConnectionTestListTablesRes { @ApiProperty() operation_result: boolean; @ApiProperty() table_list: string[]; + @ApiProperty({ type: [ConnectionTestListTablesEntry] }) + tables: ConnectionTestListTablesEntry[]; } export class GetTableMetadataReq { @@ -134,13 +143,59 @@ export class GetTableMetadataRes { tables_metadata: TableMetadataDto[]; } +export class CdcCheckDto { + @ApiProperty() + name: string; + @ApiProperty() + expected: string; + @ApiProperty() + actual: string; + @ApiProperty() + passed: boolean; +} + +export class ValidateCdcPrerequisitesReq { + @ApiProperty() + @IsString() + plugin: string; + + @ApiProperty() + @IsString() + connection_id: string; +} + +export class ValidateCdcPrerequisitesRes { + @ApiProperty() + operation_result: boolean; + @ApiProperty({ type: [CdcCheckDto] }) + checks: CdcCheckDto[]; +} + export class RefreshCatalogReq { @ApiProperty() @IsString() connection_id: string; - @ApiProperty({ enum: ['oracle', 'mysql', 'postgresql', 'sqlserver'] }) - @IsIn(['oracle', 'mysql', 'postgresql', 'sqlserver']) + @ApiProperty({ + enum: [ + 'oracle', + 'mysql', + 'postgresql', + 'sqlserver', + 'mysql_cdc', + 'postgresql_cdc', + 'oracle_cdc', + ], + }) + @IsIn([ + 'oracle', + 'mysql', + 'postgresql', + 'sqlserver', + 'mysql_cdc', + 'postgresql_cdc', + 'oracle_cdc', + ]) plugin: string; } diff --git a/src/modules/inputs/dtos/input.model.ts b/src/modules/inputs/dtos/input.model.ts index c026f8d..f7d3aa4 100644 --- a/src/modules/inputs/dtos/input.model.ts +++ b/src/modules/inputs/dtos/input.model.ts @@ -65,3 +65,62 @@ export class CreateInputReq extends OmitType(Input, [ 'created_at', 'updated_at', ]) {} + +export class CdcColumnReq { + @ApiProperty() + name: string; + @ApiProperty() + type: string; + @ApiProperty() + is_primary_key: boolean; +} + +export class CdcTableReq { + @ApiProperty() + name: string; + @ApiPropertyOptional() + table_schema?: string; + @ApiPropertyOptional({ type: [String] }) + primary_keys?: string[]; + // Per-table raw Iceberg table name override (iceberg destination only). + // Honored on the create/add path: the platform lowercases + sanitizes it + // authoritatively; empty/absent => the platform derives tb____. + @ApiPropertyOptional() + iceberg_table_name?: string; + // Per-table deduped (qualify) Iceberg table name override (iceberg dest only). + // Empty/absent => the deduped table takes the same name as the raw table. + @ApiPropertyOptional() + iceberg_qualify_table_name?: string; + @ApiPropertyOptional({ type: [CdcColumnReq] }) + columns?: CdcColumnReq[]; + // Columns the user chose to ignore -> Debezium column.exclude.list. + @ApiPropertyOptional({ type: [String] }) + column_exclude_list?: string[]; +} + +export class IcebergDestinationReq { + @ApiProperty() + namespace: string; + // Pipeline-wide deduped (qualify) namespace. Absent => the platform derives + // the sibling of `namespace` (cdc_raw -> cdc_dedup). + @ApiPropertyOptional() + qualify_namespace?: string; +} + +export class CdcDestinationReq { + @ApiPropertyOptional({ type: IcebergDestinationReq }) + iceberg?: IcebergDestinationReq; +} + +export class CreateCdcInputReq { + @ApiProperty() + name: string; + @ApiProperty() + plugin: string; // mysql_cdc (v1) + @ApiProperty({ type: [CdcTableReq] }) + tables: CdcTableReq[]; + @ApiPropertyOptional() + read_only?: boolean; + @ApiPropertyOptional({ type: CdcDestinationReq }) + destination?: CdcDestinationReq; +} diff --git a/src/modules/inputs/inputs.controller.spec.ts b/src/modules/inputs/inputs.controller.spec.ts new file mode 100644 index 0000000..241918f --- /dev/null +++ b/src/modules/inputs/inputs.controller.spec.ts @@ -0,0 +1,75 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { InputsController } from './inputs.controller'; +import { InputsService } from './inputs.service'; +import DadosferaLogger from '@dadosfera/dadosfera-logs'; +import { CreateCdcInputReq } from './dtos/input.model'; +import { RequestUser } from 'src/decorators/user.decorator'; + +describe('InputsController', () => { + let controller: InputsController; + let inputsService: { createCdc: jest.Mock }; + + beforeEach(async () => { + inputsService = { + createCdc: jest.fn().mockResolvedValue({ input: {} }), + }; + + const module: TestingModule = await Test.createTestingModule({ + controllers: [InputsController], + providers: [ + { + provide: DadosferaLogger, + useValue: { logger: { info: jest.fn() } }, + }, + { + provide: InputsService, + useValue: inputsService, + }, + ], + }).compile(); + + controller = module.get(InputsController); + }); + + it('should be defined', () => { + expect(controller).toBeDefined(); + }); + + it('forwards destination.iceberg.namespace to InputsService.createCdc', async () => { + const body: CreateCdcInputReq = { + name: 'my-cdc-input', + plugin: 'mysql_cdc', + tables: [ + { + name: 'orders', + table_schema: 'public', + iceberg_table_name: 'orders_iceberg', + }, + ], + destination: { + iceberg: { + namespace: 'my_namespace', + }, + }, + }; + const user: RequestUser = { + user_id: 'user-1', + customer_id: 'customer-1', + customer_name: 'customer', + } as RequestUser; + + await controller.createCdc(body, user); + + expect(inputsService.createCdc).toHaveBeenCalledWith( + expect.objectContaining({ + body: expect.objectContaining({ + destination: { + iceberg: { + namespace: 'my_namespace', + }, + }, + }), + }), + ); + }); +}); diff --git a/src/modules/inputs/inputs.controller.ts b/src/modules/inputs/inputs.controller.ts index a8cc143..3d753fd 100644 --- a/src/modules/inputs/inputs.controller.ts +++ b/src/modules/inputs/inputs.controller.ts @@ -15,6 +15,7 @@ import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum'; import { AuthenticateCondition } from 'src/decorators/authentication.decorator'; import { ApiOkResponse, ApiTags } from '@nestjs/swagger'; import { + CreateCdcInputReq, CreateInputReq, GetAvailableEntitiesReq, GetAvailableEntitiesRes, @@ -105,6 +106,26 @@ export class InputsController { return response; } + @Post('cdc') + @ApiInternalOnlyEndpoint() + @ApiOkResponse({ type: Input }) + async createCdc( + @Body() body: CreateCdcInputReq, + @User() user: RequestUser, + ) { + const info: Info = { + user_id: user.user_id, + customer: user.customer_name, + customer_id: user.customer_id, + }; + this.logger.info(`/inputs/cdc - ON CREATE CDC INPUT ROUTE`, { + user: info.user_id, + customer: info.customer, + }); + + return this.inputService.createCdc({ body, info }); + } + @ApiInternalOnlyEndpoint() @Get() async findAll(@User() user: RequestUser) { diff --git a/src/modules/inputs/inputs.service.spec.ts b/src/modules/inputs/inputs.service.spec.ts new file mode 100644 index 0000000..46d9e38 --- /dev/null +++ b/src/modules/inputs/inputs.service.spec.ts @@ -0,0 +1,113 @@ +import { of } from 'rxjs'; +import { InputsService } from './inputs.service'; +import DadosferaLogger from '@dadosfera/dadosfera-logs/dist'; +import { CreateCdcInputReq } from './dtos/input.model'; +import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; + +const info = { customer_id: 'cid', user_id: 'u' } as unknown as Info; + +describe('InputsService.createCdc', () => { + let service: InputsService; + let inputCreateCdcMock: jest.Mock; + + beforeEach(async () => { + inputCreateCdcMock = jest + .fn() + .mockImplementation((req) => of({ input: req.input })); + + const grpcClient: any = { + getService: jest.fn().mockReturnValue({ + InputCreateCdc: inputCreateCdcMock, + }), + }; + + service = new InputsService(new DadosferaLogger(), grpcClient); + await service.onModuleInit(); + }); + + it('forwards destination and per-table iceberg_table_name to the gRPC request', async () => { + const body: CreateCdcInputReq = { + name: 'CDC Iceberg Test', + plugin: 'mysql_cdc', + read_only: true, + destination: { iceberg: { namespace: 'cdc_raw' } }, + tables: [ + { + name: 'orders', + table_schema: 'mydb', + primary_keys: ['id'], + iceberg_table_name: 'cdc_raw.mydb__orders', + }, + ], + }; + + await service.createCdc({ body, info }); + + expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); + const sentRequest = inputCreateCdcMock.mock.calls[0][0]; + + expect(sentRequest.input).toEqual( + expect.objectContaining({ + destination: { iceberg: { namespace: 'cdc_raw' } }, + }), + ); + expect(sentRequest.input.tables[0]).toEqual( + expect.objectContaining({ + iceberg_table_name: 'cdc_raw.mydb__orders', + }), + ); + }); + + it('forwards per-table columns to the gRPC request', async () => { + const body: CreateCdcInputReq = { + name: 'CDC Columns Test', + plugin: 'mysql_cdc', + read_only: true, + tables: [ + { + name: 'orders', + table_schema: 'mydb', + primary_keys: ['id'], + columns: [ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ], + }, + ], + }; + + await service.createCdc({ body, info }); + + expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); + const sentRequest = inputCreateCdcMock.mock.calls[0][0]; + + expect(sentRequest.input.tables[0].columns).toEqual([ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ]); + }); + + it('back-compat: a body with no destination sends destination undefined, not an error', async () => { + const body: CreateCdcInputReq = { + name: 'CDC Legacy Test', + plugin: 'mysql_cdc', + read_only: true, + tables: [ + { + name: 'pedidos', + table_schema: 'cadastros', + primary_keys: ['id'], + }, + ], + }; + + const result = await service.createCdc({ body, info }); + + expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); + const sentRequest = inputCreateCdcMock.mock.calls[0][0]; + + expect(sentRequest.input.destination).toBeUndefined(); + expect(sentRequest.input.tables[0].iceberg_table_name).toBeUndefined(); + expect(result.input).toBeDefined(); + }); +}); diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 99a6753..80eef5b 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -15,6 +15,7 @@ import { Input } from '@dadosfera/protospack-v2'; import { GetAvailableEntitiesRequest, InputCreateGenericRequest, + InputCreateCdcRequest, InputCreateS3Request, InputNewCreateRequest, InputUpdateResponse, @@ -22,7 +23,7 @@ import { TestConnectionRequest, } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; -import { CreateInputReq } from './dtos/input.model'; +import { CreateCdcInputReq, CreateInputReq } from './dtos/input.model'; import { Metadata } from '@grpc/grpc-js'; @Injectable() @@ -183,6 +184,35 @@ export class InputsService { return { input: adjustedInput }; } + async createCdc(data: { body: CreateCdcInputReq; info: Info }) { + const { body, info } = data; + + const inputCreateCdcRequest: InputCreateCdcRequest = { + input: { + name: body.name, + plugin: body.plugin, + read_only: body.read_only ?? true, + tables: body.tables.map((t) => ({ + table_schema: t.table_schema, + table_name: t.name, + name: t.name, // canonical identity == table_name (in-factory also backfills) + primary_keys: t.primary_keys ?? [], + iceberg_table_name: t.iceberg_table_name, + iceberg_qualify_table_name: t.iceberg_qualify_table_name, + columns: t.columns ?? [], + column_exclude_list: t.column_exclude_list ?? [], + })), + destination: body.destination, + }, + info, + }; + + const { input } = await lastValueFrom( + this.inputWriteService.InputCreateCdc(inputCreateCdcRequest), + ); + return { input }; + } + async getAvailableEntities(data: GetAvailableEntitiesRequest) { return lastValueFrom(this.inputReadService.GetAvailableEntities(data)); } @@ -297,4 +327,12 @@ export class InputsService { async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) { return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data)); } + + async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) { + return lastValueFrom(this.inputWriteService.AddCdcTable(data as any)); + } + + async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) { + return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data)); + } } diff --git a/src/modules/pipelinesV2/interfaces.ts b/src/modules/pipelinesV2/interfaces.ts index 635ae80..c54e74a 100644 --- a/src/modules/pipelinesV2/interfaces.ts +++ b/src/modules/pipelinesV2/interfaces.ts @@ -31,6 +31,8 @@ export class IPipelineV2 { tags?: string[]; @ApiPropertyOptional() properties?: any; + @ApiPropertyOptional() + config?: any; @ApiProperty() connector_name: string; diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index c24c090..6c529c5 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -510,4 +510,78 @@ export class PipelinesController { return response; } + + // ---- CDC pipeline operations (Kafka Connect backed) ---- + // These operate on an existing pipeline, so they require UPDATE (not CREATE). + + @Get(':id/live-status') + @ApiInternalOnlyEndpoint() + @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) + async getLiveStatus( + @Param('id') id: string, + @User() user: RequestUser, + ): Promise { + this.logger.info('PipelinesController - getLiveStatus', { id }); + const metadata = PackTheMetadata(user); + return this.pipelinesClientService.getLiveStatus(id, metadata); + } + + @Post(':id/pause') + @HttpCode(HttpStatus.OK) + @ApiInternalOnlyEndpoint() + @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + async pause( + @Param('id') id: string, + @User() user: RequestUser, + ): Promise { + this.logger.info('PipelinesController - pause', { id }); + const metadata = PackTheMetadata(user); + return this.pipelinesClientService.pause(id, metadata); + } + + @Post(':id/unpause') + @HttpCode(HttpStatus.OK) + @ApiInternalOnlyEndpoint() + @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + async unpause( + @Param('id') id: string, + @User() user: RequestUser, + ): Promise { + this.logger.info('PipelinesController - unpause', { id }); + const metadata = PackTheMetadata(user); + return this.pipelinesClientService.unpause(id, metadata); + } + + @Post(':id/restart') + @HttpCode(HttpStatus.OK) + @ApiInternalOnlyEndpoint() + @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + async restart( + @Param('id') id: string, + @User() user: RequestUser, + ): Promise { + this.logger.info('PipelinesController - restart', { id }); + const metadata = PackTheMetadata(user); + return this.pipelinesClientService.restart(id, metadata); + } + + @Post(':id/jobs/:jobId/reset-state') + @HttpCode(HttpStatus.OK) + @ApiInternalOnlyEndpoint() + @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + async resetJobState( + @Param('id') id: string, + @Param('jobId') jobId: string, + @Body() body: { schedule_minutes?: number }, + @User() user: RequestUser, + ): Promise { + this.logger.info('PipelinesController - resetJobState', { id, jobId }); + const metadata = PackTheMetadata(user); + return this.pipelinesClientService.resetJobState( + id, + jobId, + body?.schedule_minutes, + metadata, + ); + } } diff --git a/src/modules/pipelinesV2/pipelines.module.ts b/src/modules/pipelinesV2/pipelines.module.ts index b4f24af..efd3c74 100644 --- a/src/modules/pipelinesV2/pipelines.module.ts +++ b/src/modules/pipelinesV2/pipelines.module.ts @@ -1,4 +1,4 @@ -import { Module } from '@nestjs/common'; +import { Module, forwardRef } from '@nestjs/common'; import { ClientsModule } from '@nestjs/microservices'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; @@ -23,7 +23,7 @@ const client = new PipelinesClientConfiguration(); ConnectorModule, InputsModule, TransformationsModule, - PlatformApiModule, + forwardRef(() => PlatformApiModule), NimbusServicesModule, CatalogModule ], diff --git a/src/modules/pipelinesV2/pipelines.service.spec.ts b/src/modules/pipelinesV2/pipelines.service.spec.ts new file mode 100644 index 0000000..279f567 --- /dev/null +++ b/src/modules/pipelinesV2/pipelines.service.spec.ts @@ -0,0 +1,142 @@ +// These imported modules pull in gRPC client-config / service modules that read +// process.env at load time; mock them (hoisted before imports) so the spec needs +// no runtime env. Each mock severs an entire import subtree while still providing +// a class usable as a value/DI token. Mirrors platform-api.controller.spec.ts. +jest.mock('./pipelines-client', () => ({ PipelinesClientConfiguration: class {} })); +jest.mock('../connector/client.service', () => ({ ConnectorClientService: class {} })); +jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} })); +jest.mock('../transformations/transformations.service', () => ({ TransformationsService: class {} })); +jest.mock('../platform-api/platform-api.service', () => ({ PlatformApiService: class {} })); +jest.mock('src/services/nimbus/nimbus.service', () => ({ NimbusService: class {} })); +jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} })); + +import { of } from 'rxjs'; +import { PipelinesService } from './pipelines.service'; + +const logger = { + info: jest.fn(), + error: jest.fn(), +}; + +const cdcOldInput = { input: { plugin: 'mysql_cdc', tables: [] } }; +const batchOldInput = { input: { plugin: 'mysql', type: 'database', tables: [] } }; + +const updateResponse = { + input: { type: 'database' }, + tablesUpdate: [], + dataAssetUpdate: [], +}; + +const user: any = { customer_modules: [] }; +const updateInputDTO: any = { tables: [] }; +const info: any = { customer: 'cust' }; +const metadata: any = {}; + +function buildService(oldInput: any) { + const inputsService: any = { + findOne: jest.fn().mockResolvedValue(oldInput), + update: jest.fn().mockResolvedValue(updateResponse), + rollbackUpdate: jest.fn().mockResolvedValue({}), + }; + const nimbusService: any = { renameTable: jest.fn().mockResolvedValue({}) }; + + const service = new PipelinesService( + { logger } as any, // dadosferaLogger + {} as any, // grpcClient + {} as any, // connectorService + inputsService, // inputsService + {} as any, // transformationsService + {} as any, // platformAPI + nimbusService, // nimbusService + {} as any, // catalogService + ); + + const updatePlatformJobsSpy = jest + .spyOn(service, 'updatePlatformJobs') + .mockResolvedValue(undefined as any); + + return { service, inputsService, updatePlatformJobsSpy }; +} + +describe('PipelinesService - updatePipelineInput', () => { + afterEach(() => jest.clearAllMocks()); + + it('CDC input skips updatePlatformJobs', async () => { + const { service, inputsService, updatePlatformJobsSpy } = buildService(cdcOldInput); + + const result = await service.updatePipelineInput( + 'pipeline-id', + 'input-id', + updateInputDTO, + info, + user, + metadata, + ); + + expect(updatePlatformJobsSpy).not.toHaveBeenCalled(); + expect(inputsService.update).toHaveBeenCalled(); + expect(result).toBe(updateResponse); + }); + + it('batch input calls updatePlatformJobs', async () => { + const { service, inputsService, updatePlatformJobsSpy } = buildService(batchOldInput); + + await service.updatePipelineInput( + 'pipeline-id', + 'input-id', + updateInputDTO, + info, + user, + metadata, + ); + + expect(updatePlatformJobsSpy).toHaveBeenCalled(); + expect(inputsService.update).toHaveBeenCalled(); + }); +}); + +describe('PipelinesService - create', () => { + afterEach(() => jest.clearAllMocks()); + + // The body's `config` (carrying CDC destinations) must reach pi-factory as a + // JSON string — the gRPC proto field is a string, so an object would be + // stripped on the wire. Mirrors how `properties` is serialized. + it('serializes the body config into the gRPC create request', async () => { + const { service } = buildService(cdcOldInput); + + let captured: any; + (service as any).pipelineWriteService = { + PipelineV2Create: (req: any) => { + captured = req; + // The service does `lastValueFrom(...)`; return a real Observable. + return of({ pipeline: {} }); + }, + }; + + const body: any = { + name: 'p', + input_id: 'i', + transformations_ids: [], + tags: [], + properties: { schema: 'cadastros' }, + config: { + cron: '@once', + tables: [ + { + name: 'pedidos', + destinations: { + raw: { table_schema: 'PUBLIC', table_name: 'pedidos_001' }, + }, + }, + ], + }, + }; + + await service.create(body, metadata); + + expect(typeof captured.config).toBe('string'); + expect(JSON.parse(captured.config).tables[0].destinations.raw.table_name).toBe( + 'pedidos_001', + ); + }); +}); diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index 41d9354..77c8059 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -19,7 +19,7 @@ import { lastValueFrom } from 'rxjs'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { PipelinesClientConfiguration } from './pipelines-client'; import { ICreatePipelineV2Req, IIdRequest, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces'; -import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages'; +import { PipelineV2CreateRequest, AddCdcJobsRequest, AddCdcJobsResponse } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages'; import { Metadata } from '@grpc/grpc-js'; import { ConnectorClientService } from '../connector/client.service'; import { InputsService } from '../inputs/inputs.service'; @@ -105,6 +105,10 @@ export class PipelinesService implements OnModuleInit { transformations_ids: body.transformations_ids, tags: body.tags, properties: body.properties && JSON.stringify(body.properties), + // JSON-serialize the create body config so it survives the gRPC wire + // (the proto field is a string). pi-factory's CDC path reads + // config.tables[].destinations to honor a user-supplied raw table name. + config: body.config && JSON.stringify(body.config), }; const createPipelineResponse = await lastValueFrom( @@ -160,6 +164,62 @@ export class PipelinesService implements OnModuleInit { return findOnePipelineResponse; } + // CDC lifecycle operations (Kafka Connect backed). + async pause( + id: string, + metadata, + ): Promise { + this.logger.info('PipelinesClientService - Pause'); + return lastValueFrom( + this.pipelineWriteService.PipelineV2Pause({ id }, metadata), + ); + } + + async unpause( + id: string, + metadata, + ): Promise { + this.logger.info('PipelinesClientService - Unpause'); + return lastValueFrom( + this.pipelineWriteService.PipelineV2Unpause({ id }, metadata), + ); + } + + async restart( + id: string, + metadata, + ): Promise { + this.logger.info('PipelinesClientService - Restart'); + return lastValueFrom( + this.pipelineWriteService.PipelineV2Restart({ id }, metadata), + ); + } + + async resetJobState( + id: string, + job_id: string, + schedule_minutes: number | undefined, + metadata, + ): Promise { + this.logger.info('PipelinesClientService - ResetJobState'); + return lastValueFrom( + this.pipelineWriteService.PipelineV2ResetJobState( + { id, job_id, schedule_minutes }, + metadata, + ), + ); + } + + async getLiveStatus( + id: string, + metadata, + ): Promise { + this.logger.info('PipelinesClientService - GetLiveStatus'); + return lastValueFrom( + this.pipelineReadService.PipelineV2GetLiveStatus({ id }, metadata), + ); + } + async update( UpdatePipelineRequest: Messages.PipelineV2UpdateRequest, metadata, @@ -377,6 +437,7 @@ export class PipelinesService implements OnModuleInit { }); this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput)); + const isCdc = !!oldInput.plugin?.endsWith('_cdc'); const pipelineIdFormat = pipelineId.split('-').join('_'); const rollback: RollbackPromise[] = []; @@ -437,17 +498,21 @@ export class PipelinesService implements OnModuleInit { } } - try { - await this.updatePlatformJobs( - pipelineIdFormat, - updateInputResponse.input.type, - updateInputDTO, - user - ); - } catch (error) { - this.logger.error(error); - await this.executeRenameRollback(rollback) - throw new Error("Error Platform API updating jobs"); + if (!isCdc) { + try { + await this.updatePlatformJobs( + pipelineIdFormat, + updateInputResponse.input.type, + updateInputDTO, + user + ); + } catch (error) { + this.logger.error(error); + await this.executeRenameRollback(rollback) + throw new Error("Error Platform API updating jobs"); + } + } else { + this.logger.info('CDC input: skipping updatePlatformJobs (batch sync_mode/memory do not apply to CDC jobs)'); } return updateInputResponse; @@ -662,6 +727,10 @@ export class PipelinesService implements OnModuleInit { return statusPipelineResponse; } + async addCdcJobs(data: AddCdcJobsRequest): Promise { + return lastValueFrom(this.pipelineWriteService.AddCdcJobs(data)); + } + async runPipeline({ id, info }: IIdRequest) { this.logger.info('PipelinesClientService - RunPipeline'); const statusPipelineResponse = await lastValueFrom( diff --git a/src/modules/platform-api/platform-api.controller.spec.ts b/src/modules/platform-api/platform-api.controller.spec.ts new file mode 100644 index 0000000..f3d58a3 --- /dev/null +++ b/src/modules/platform-api/platform-api.controller.spec.ts @@ -0,0 +1,372 @@ +// These service modules pull in gRPC client-config modules that read +// process.env at load time; mock them (hoisted before imports) so the spec +// needs no runtime env. Each mock severs an entire import subtree and still +// provides a class usable as a DI token. +jest.mock('../customers/customers.service', () => ({ CustomersService: class {} })); +jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} })); +jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} })); +jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} })); + +import { Test, TestingModule } from '@nestjs/testing'; +import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; + +import { PlatformApiController } from './platform-api.controller'; +import { PlatformApiService } from './platform-api.service'; +import { ElasticsearchService } from '../../services/elasticsearch'; +import { DynamoDBService } from '../../services/dynamodb'; +import { CustomersService } from '../customers/customers.service'; +import { CatalogService } from '../catalog/catalog.service'; +import { InputsService } from '../inputs/inputs.service'; +import { PipelinesService } from '../pipelinesV2/pipelines.service'; + +const logger = { + info: (...args) => args, + error: (...args) => args, +}; + +const mockUser: any = { + customer_id: 'c1', + customer_name: 'cust', + user_id: 'u1', +}; + +describe('PlatformApiController - deleteTable', () => { + let controller: PlatformApiController; + let platformApiService: { proxy: jest.Mock }; + let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock }; + + beforeEach(async () => { + platformApiService = { proxy: jest.fn() }; + inputsService = { + markTableDeleted: jest.fn(), + unmarkTableDeleted: jest.fn(), + }; + + const module: TestingModule = await Test.createTestingModule({ + controllers: [PlatformApiController], + providers: [ + { provide: PlatformApiService, useValue: platformApiService }, + { provide: ElasticsearchService, useValue: {} }, + { provide: DynamoDBService, useValue: {} }, + { provide: CustomersService, useValue: {} }, + { provide: CatalogService, useValue: {} }, + { provide: InputsService, useValue: inputsService }, + { provide: PipelinesService, useValue: {} }, + { provide: DadosferaLogger, useValue: { logger } }, + ], + }).compile(); + + controller = module.get(PlatformApiController); + }); + + it('CDC removal reconfigures the connector', async () => { + inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' }); + platformApiService.proxy.mockImplementation((method: string, path: string) => { + if (method === 'GET') { + return Promise.resolve({ + jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 'pedidos' } }], + }); + } + return Promise.resolve({}); + }); + + await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser); + + expect(platformApiService.proxy).toHaveBeenCalledWith( + 'DELETE', + '/pipeline/pid/jobs', + mockUser, + { job_ids: ['p_0'], delete_snowflake_tables: false }, + ); + const deletedViaJobsRoute = platformApiService.proxy.mock.calls.some( + ([method, path]: any[]) => method === 'DELETE' && path === '/jobs/p_0', + ); + expect(deletedViaJobsRoute).toBe(false); + }); + + it('batch removal unchanged', async () => { + inputsService.markTableDeleted.mockResolvedValue({ is_deleted: true, deleted_at: 't' }); + platformApiService.proxy.mockImplementation((method: string, path: string) => { + if (method === 'GET') { + return Promise.resolve({ + jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 'pedidos' } }], + }); + } + return Promise.resolve({}); + }); + + await controller.deleteTable('pid', 'iid', { table_name: 'pedidos' }, mockUser); + + expect(platformApiService.proxy).toHaveBeenCalledWith( + 'DELETE', + '/jobs/p_0', + mockUser, + ); + const reconfiguredConnector = platformApiService.proxy.mock.calls.some( + ([method, path]: any[]) => method === 'DELETE' && path === '/pipeline/pid/jobs', + ); + expect(reconfiguredConnector).toBe(false); + }); +}); + +describe('PlatformApiController - addTable', () => { + let controller: PlatformApiController; + let inputsService: { addCdcTable: jest.Mock; removeCdcTable: jest.Mock }; + let pipelinesClientService: { addCdcJobs: jest.Mock }; + + const body = { + table_name: 'orders', + table_schema: 'public', + primary_keys: ['id'], + destinations: { + raw: { table_schema: 'raw', table_name: 'orders' }, + qualify: { table_schema: 'qualify', table_name: 'orders' }, + }, + }; + + beforeEach(async () => { + inputsService = { + addCdcTable: jest.fn(), + removeCdcTable: jest.fn(), + }; + pipelinesClientService = { + addCdcJobs: jest.fn(), + }; + + const module: TestingModule = await Test.createTestingModule({ + controllers: [PlatformApiController], + providers: [ + { provide: PlatformApiService, useValue: { proxy: jest.fn() } }, + { provide: ElasticsearchService, useValue: {} }, + { provide: DynamoDBService, useValue: {} }, + { provide: CustomersService, useValue: {} }, + { provide: CatalogService, useValue: {} }, + { provide: InputsService, useValue: inputsService }, + { provide: PipelinesService, useValue: pipelinesClientService }, + { provide: DadosferaLogger, useValue: { logger } }, + ], + }).compile(); + + controller = module.get(PlatformApiController); + }); + + it('addTable: DynamoDB append then platform AddJobs, returns job_ids', async () => { + inputsService.addCdcTable.mockResolvedValue({ input: {} }); + pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] }); + + const result = await controller.addTable('pid', 'iid', body, mockUser); + + expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + id: 'iid', + table: { + table_schema: 'public', + table_name: 'orders', + primary_keys: ['id'], + name: 'orders', + iceberg_table_name: undefined, + iceberg_qualify_table_name: undefined, + columns: [], + column_exclude_list: [], + }, + info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, + }); + expect(pipelinesClientService.addCdcJobs).toHaveBeenCalledWith({ + pipeline_id: 'pid', + input_id: 'iid', + tables: [{ + table_schema: 'public', + table_name: 'orders', + primary_keys: ['id'], + destinations: body.destinations, + }], + info: { customer_id: 'c1', user_id: 'u1', customer: 'cust' }, + }); + expect(result).toEqual({ job_ids: ['p_2'], skipped: [] }); + expect(inputsService.removeCdcTable).not.toHaveBeenCalled(); + + const addCdcTableOrder = inputsService.addCdcTable.mock.invocationCallOrder[0]; + const addCdcJobsOrder = pipelinesClientService.addCdcJobs.mock.invocationCallOrder[0]; + expect(addCdcTableOrder).toBeLessThan(addCdcJobsOrder); + }); + + it('addTable: carries iceberg_table_name on the added table through to AddCdcTable', async () => { + inputsService.addCdcTable.mockResolvedValue({ input: {} }); + pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_3'], skipped: [] }); + + const icebergBody = { ...body, iceberg_table_name: 'cdc_raw.public__orders' }; + + await controller.addTable('pid', 'iid', icebergBody, mockUser); + + expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + id: 'iid', + table: { + table_schema: 'public', + table_name: 'orders', + primary_keys: ['id'], + name: 'orders', + iceberg_table_name: 'cdc_raw.public__orders', + iceberg_qualify_table_name: undefined, + columns: [], + column_exclude_list: [], + }, + info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, + }); + }); + + it('addTable: carries columns on the added table through to AddCdcTable', async () => { + inputsService.addCdcTable.mockResolvedValue({ input: {} }); + pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_4'], skipped: [] }); + + const columnsBody = { + ...body, + columns: [ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ], + }; + + await controller.addTable('pid', 'iid', columnsBody, mockUser); + + expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + id: 'iid', + table: { + table_schema: 'public', + table_name: 'orders', + primary_keys: ['id'], + name: 'orders', + iceberg_table_name: undefined, + iceberg_qualify_table_name: undefined, + columns: [ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ], + column_exclude_list: [], + }, + info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, + }); + }); + + it('addTable: rolls back the DynamoDB row when AddJobs fails', async () => { + inputsService.addCdcTable.mockResolvedValue({ input: {} }); + pipelinesClientService.addCdcJobs.mockRejectedValue(new Error('platform down')); + inputsService.removeCdcTable.mockResolvedValue({}); + + await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down'); + + expect(inputsService.removeCdcTable).toHaveBeenCalledWith({ + id: 'iid', + table_name: 'orders', + info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, + }); + }); +}); + + +describe('PlatformApiController - deleteTables (batch)', () => { + let controller: PlatformApiController; + let platformApiService: { proxy: jest.Mock }; + let inputsService: { markTableDeleted: jest.Mock; unmarkTableDeleted: jest.Mock }; + + beforeEach(async () => { + platformApiService = { proxy: jest.fn() }; + inputsService = { + markTableDeleted: jest.fn().mockResolvedValue({ is_deleted: true, deleted_at: 't' }), + unmarkTableDeleted: jest.fn().mockResolvedValue({}), + }; + + const module: TestingModule = await Test.createTestingModule({ + controllers: [PlatformApiController], + providers: [ + { provide: PlatformApiService, useValue: platformApiService }, + { provide: ElasticsearchService, useValue: {} }, + { provide: DynamoDBService, useValue: {} }, + { provide: CustomersService, useValue: {} }, + { provide: CatalogService, useValue: {} }, + { provide: InputsService, useValue: inputsService }, + { provide: PipelinesService, useValue: {} }, + { provide: DadosferaLogger, useValue: { logger } }, + ], + }).compile(); + + controller = module.get(PlatformApiController); + }); + + 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(); + }); +}); diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 8121242..44e1ac9 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -33,6 +33,7 @@ import { CatalogService } from '../catalog/catalog.service'; import { PackTheMetadata } from '../../utils/PackTheMetadata'; import { ValidationTableDTO } from './platform-api.dto'; import { InputsService } from '../inputs/inputs.service'; +import { PipelinesService } from '../pipelinesV2/pipelines.service'; import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; @@ -61,6 +62,7 @@ export class PlatformApiController { private readonly customersService: CustomersService, private readonly catalogService: CatalogService, private readonly inputsService: InputsService, + private readonly pipelinesClientService: PipelinesService, @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, ) { this.logger = dadosferaLogger.logger; @@ -545,6 +547,20 @@ export class PlatformApiController { return this.platformApiService.proxy('GET', `/pipeline/${normalizedId}`, user); } + @Get('iceberg/namespaces') + @ApiOperation({ summary: 'List existing Polaris Iceberg namespaces (CDC destination dropdown)' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) + async getIcebergNamespaces(@User() user: RequestUser) { + return this.platformApiService.proxy('GET', '/iceberg/namespaces', user); + } + + @Post('iceberg/tables/validate') + @ApiOperation({ summary: 'Validate CDC Iceberg raw table names against Polaris' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) + async validateIcebergTables(@Body() body: any, @User() user: RequestUser) { + return this.platformApiService.proxy('POST', '/iceberg/tables/validate', user, body); + } + @Patch('pipelines/:pipelineId') @ApiOperation({ summary: 'Update pipeline by ID' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) @@ -1002,7 +1018,18 @@ export class PlatformApiController { if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`); this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id }); - await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user); + if (job.input?.connector === 'cdc') { + // CDC: reconfigure the Debezium/Kafka-Connect connector (stop + // replicating this table); keep the landed Snowflake data. + await this.platformApiService.proxy( + 'DELETE', + `/pipeline/${normalizedPipelineId}/jobs`, + user, + { job_ids: [job.job_id], delete_snowflake_tables: false }, + ); + } else { + await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user); + } this.logger.info('deleteTable: job deleted', { jobId: job.job_id }); return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at }; @@ -1017,6 +1044,155 @@ export class PlatformApiController { } } + @Delete('pipelines/:pipelineId/inputs/:inputId/tables') + @ApiOperation({ summary: 'Batch-remove tables from an input; reconfigures the CDC connectors once' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE) + @UseGuards(PipelineExecutionGuard) + async deleteTables( + @Param('pipelineId') pipelineId: string, + @Param('inputId') inputId: string, + @Body() body: { table_names: string[] }, + @User() user: RequestUser, + ) { + const tableNames = body.table_names ?? []; + if (tableNames.length === 0) { + throw new BadRequestException('table_names must be a non-empty array'); + } + const info = { + customer_id: user.customer_id, + customer: user.customer_name, + user_id: user.user_id, + }; + + // 1) Soft-delete each table in DynamoDB, tracking which succeeded so a later + // failure only rolls back the marks made in THIS call. + const marked: string[] = []; + const rollback = async () => { + for (const name of marked) { + try { + await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: name, info }); + } catch (rollbackError) { + this.logger.error('deleteTables: rollback failed', { tableName: name, error: rollbackError.message }); + } + } + }; + + try { + for (const name of tableNames) { + await this.inputsService.markTableDeleted({ input_id: inputId, table_name: name, info }); + marked.push(name); + } + + // 2) Resolve all table_names -> job_ids from the platform pipeline (one GET). + const normalizedPipelineId = this.normalizePipelineId(pipelineId); + const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user); + const jobs = platformPipeline?.jobs ?? []; + + const jobIds: string[] = []; + for (const name of tableNames) { + const job = jobs.find((j: any) => j.input?.table_name === name); + if (!job) throw new NotFoundException(`Job for table '${name}' not found in pipeline`); + jobIds.push(job.job_id); + } + + // 3) Remove them all in ONE platform call so the Debezium source + Snowflake + // sink connectors are reconfigured a single time, not once per table. + await this.platformApiService.proxy( + 'DELETE', + `/pipeline/${normalizedPipelineId}/jobs`, + user, + { job_ids: jobIds, delete_snowflake_tables: false }, + ); + this.logger.info('deleteTables: jobs deleted', { jobIds }); + + return { table_names: tableNames, deleted: true }; + } catch (error) { + this.logger.error('deleteTables: failed, rolling back this call\'s marks', { tableNames, error: error.message }); + await rollback(); + throw error; + } + } + + @Post('pipelines/:pipelineId/inputs/:inputId/tables') + @ApiOperation({ summary: 'Add a CDC table to an input and dispatch its platform jobs' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + @UseGuards(PipelineExecutionGuard) + async addTable( + @Param('pipelineId') pipelineId: string, + @Param('inputId') inputId: string, + @Body() body: { + table_name: string; + table_schema: string; + primary_keys: string[]; + destinations: { + raw: { table_schema: string; table_name: string }; + qualify: { table_schema: string; table_name: string }; + }; + // Iceberg destination only (protospack CdcTable.iceberg_table_name); + // absent for snowflake, back-compat. + iceberg_table_name?: string; + // Per-table deduped (qualify) Iceberg table name (protospack + // CdcTable.iceberg_qualify_table_name); absent => same as the raw name. + iceberg_qualify_table_name?: string; + // Source column schema for iceberg deduped table pre-create (protospack CdcTable.columns). + columns?: { name: string; type: string; is_primary_key: boolean }[]; + // Columns the user chose to ignore -> Debezium column.exclude.list. + column_exclude_list?: string[]; + }, + @User() user: RequestUser, + ) { + const info = { + customer_id: user.customer_id, + customer: user.customer_name, + user_id: user.user_id, + }; + + const cdcTable = { + table_schema: body.table_schema, + table_name: body.table_name, + primary_keys: body.primary_keys, + name: body.table_name, + iceberg_table_name: body.iceberg_table_name, + iceberg_qualify_table_name: body.iceberg_qualify_table_name, + columns: body.columns ?? [], + column_exclude_list: body.column_exclude_list ?? [], + }; + + this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name }); + await this.inputsService.addCdcTable({ id: inputId, table: cdcTable, info }); + this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name }); + + try { + const customInfo = { + customer_id: user.customer_id, + user_id: user.user_id, + customer: user.customer_name, + }; + + const res = await this.pipelinesClientService.addCdcJobs({ + pipeline_id: pipelineId, + input_id: inputId, + tables: [{ + table_schema: body.table_schema, + table_name: body.table_name, + primary_keys: body.primary_keys, + destinations: body.destinations, + }], + info: customInfo, + }); + + return res; + } catch (error) { + this.logger.error('addTable: platform AddJobs failed, rolling back the DynamoDB row', { tableName: body.table_name, error: error.message }); + try { + await this.inputsService.removeCdcTable({ id: inputId, table_name: body.table_name, info }); + } catch (rbErr) { + this.logger.error('addTable: rollback failed', { error: rbErr.message }); + } + throw error; + } + } + // ==================== JOBS - JDBC SYNC MODE ROUTES ==================== @Get('jobs/jdbc/configs/allowed_datatypes') diff --git a/src/modules/platform-api/platform-api.module.ts b/src/modules/platform-api/platform-api.module.ts index b2b0726..726e743 100644 --- a/src/modules/platform-api/platform-api.module.ts +++ b/src/modules/platform-api/platform-api.module.ts @@ -1,4 +1,4 @@ -import { Module } from '@nestjs/common'; +import { Module, forwardRef } from '@nestjs/common'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; @@ -9,9 +9,17 @@ import { DynamoDBModule } from '../../services/dynamodb'; import { CustomersModule } from '../customers/customers.module'; import { CatalogModule } from '../catalog/catalog.module'; import { InputsModule } from '../inputs/inputs.module'; +import { PipelinesV2Module } from '../pipelinesV2/pipelines.module'; @Module({ - imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, InputsModule], + imports: [ + ElasticsearchModule, + DynamoDBModule, + CustomersModule, + CatalogModule, + InputsModule, + forwardRef(() => PipelinesV2Module), + ], controllers: [PlatformApiController], providers: [PlatformApiService, DadosferaLogger], exports: [PlatformApiService], diff --git a/src/modules/release_note/release_note.controller.spec.ts b/src/modules/release_note/release_note.controller.spec.ts index 921eca1..1fd7eba 100644 --- a/src/modules/release_note/release_note.controller.spec.ts +++ b/src/modules/release_note/release_note.controller.spec.ts @@ -1,14 +1,26 @@ import { Test, TestingModule } from '@nestjs/testing'; +import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { ReleaseNoteController } from './release_note.controller'; import { ReleaseNoteService } from './release_note.service'; +const logger = { + info: (...args) => args, + error: (...args) => args, +}; + describe('ReleaseNoteController', () => { let controller: ReleaseNoteController; beforeEach(async () => { const module: TestingModule = await Test.createTestingModule({ controllers: [ReleaseNoteController], - providers: [ReleaseNoteService], + providers: [ + ReleaseNoteService, + { + provide: DadosferaLogger, + useValue: { logger }, + }, + ], }).compile(); controller = module.get(ReleaseNoteController); diff --git a/src/modules/release_note/release_note.service.spec.ts b/src/modules/release_note/release_note.service.spec.ts index 72add9d..70d2024 100644 --- a/src/modules/release_note/release_note.service.spec.ts +++ b/src/modules/release_note/release_note.service.spec.ts @@ -1,12 +1,24 @@ import { Test, TestingModule } from '@nestjs/testing'; +import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { ReleaseNoteService } from './release_note.service'; +const logger = { + info: (...args) => args, + error: (...args) => args, +}; + describe('ReleaseNoteService', () => { let service: ReleaseNoteService; beforeEach(async () => { const module: TestingModule = await Test.createTestingModule({ - providers: [ReleaseNoteService], + providers: [ + ReleaseNoteService, + { + provide: DadosferaLogger, + useValue: { logger }, + }, + ], }).compile(); service = module.get(ReleaseNoteService);