diff --git a/package-lock.json b/package-lock.json index 2cc5927..bef9f40 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": "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", diff --git a/package.json b/package.json index 882ad39..53ec490 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": "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", 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.service.spec.ts b/src/modules/pipelinesV2/pipelines.service.spec.ts index 564639e..279f567 100644 --- a/src/modules/pipelinesV2/pipelines.service.spec.ts +++ b/src/modules/pipelinesV2/pipelines.service.spec.ts @@ -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', + ); + }); +}); diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index 992077f..b3d641b 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -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(