feat: forward create body config over gRPC (cdc.3) for CDC destinations

The pipeline create body carries `config.tables[].destinations`, which
the CDC path in pi-factory needs to honor a user-supplied raw Snowflake
table name. The gRPC PipelineV2CreateRequest previously had no `config`
field, so `...body` dropped it on the wire.

Bump protospack to cdc.3 (adds optional `config` string). Serialize
`body.config` into the create request the same way `properties` is
handled, and add `config?` to the ICreatePipelineV2Req DTO so it's
typed. Add a spec asserting the gRPC request carries a stringified
config with the destination intact.

Co-Authored-By: WOZCODE <contact@withwoz.com>
Claude-Session: https://claude.ai/code/session_01145m1zZMfx8RSJBxAhySdg
This commit is contained in:
Rafael
2026-08-17 11:33:29 -03:00
co-authored by WOZCODE
parent 0d771d1e4a
commit c85e2ac5da
5 changed files with 58 additions and 5 deletions
+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": "file:../protospack-v2/dadosfera-protospack-v2-3.40.0-beta.15-cdc.2.tgz",
"@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.40.0-beta.15-cdc.3.tgz",
"@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.15-cdc.2",
"resolved": "file:../protospack-v2/dadosfera-protospack-v2-3.40.0-beta.15-cdc.2.tgz",
"integrity": "sha512-FiZXH+bSv2Skxy9YJkczQOPnZxYbca5JlJSNbSzNxuk5e044i6Mp79LAA7SQDGt5yyAEfNHq2bAorHUxf0Q0TA==",
"version": "3.40.0-beta.15-cdc.3",
"resolved": "file:../protospack-v2/dadosfera-protospack-v2-3.40.0-beta.15-cdc.3.tgz",
"integrity": "sha512-/JG35L/4iStITRhtk4GZnzKiF0oa6Hfc5i66rC7yDr2wG/MSEogQjtiU9G4LLJK84iAMpSstm6fdX7nWsjr5hQ==",
"license": "ISC",
"dependencies": {
"@grpc/grpc-js": "^1.9.3",
+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": "file:../protospack-v2/dadosfera-protospack-v2-3.40.0-beta.15-cdc.2.tgz",
"@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.40.0-beta.15-cdc.3.tgz",
"@grpc/grpc-js": "^1.9.3",
"@grpc/proto-loader": "^0.7.9",
"@nestjs/cli": "^9.5.0",
+2
View File
@@ -31,6 +31,8 @@ export class IPipelineV2 {
tags?: string[];
@ApiPropertyOptional()
properties?: any;
@ApiPropertyOptional()
config?: any;
@ApiProperty()
connector_name: string;
@@ -10,6 +10,7 @@ jest.mock('../platform-api/platform-api.service', () => ({ PlatformApiService: c
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 = {
@@ -93,3 +94,49 @@ describe('PipelinesService - updatePipelineInput', () => {
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',
);
});
});
@@ -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(