gitea_admin/maestro
1
0
Fork 0
mirror of https://github.com/dadosfera/maestro.git synced 2026-08-31 19:58:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

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