From 23f0554311eae9aec0c3e07bcd0befb906d257eb Mon Sep 17 00:00:00 2001 From: Rafael Date: Sat, 29 Aug 2026 12:56:00 -0300 Subject: [PATCH] UPDATE: type CDC input calls against the protospack gRPC contracts Review (maestro #510): drop every 'as any' on the Input write client (AddCdcTable/RemoveCdcTable/UnmarkTableDeleted take the generated request types), type IPipelineV2.config, and build the CdcTable payload in one mapper so 'name' mirrors table_name in a single place. isCdc via helper, no '!!'. Co-Authored-By: WOZCODE --- src/modules/inputs/cdc-table.mapper.ts | 38 +++++++++++++++++++ src/modules/inputs/dtos/input.model.ts | 6 ++- src/modules/inputs/inputs.service.ts | 31 +++++++-------- src/modules/pipelinesV2/interfaces.ts | 31 ++++++++++++++- src/modules/pipelinesV2/pipelines.service.ts | 3 +- .../platform-api.controller.spec.ts | 4 ++ .../platform-api/platform-api.controller.ts | 16 ++------ 7 files changed, 95 insertions(+), 34 deletions(-) create mode 100644 src/modules/inputs/cdc-table.mapper.ts diff --git a/src/modules/inputs/cdc-table.mapper.ts b/src/modules/inputs/cdc-table.mapper.ts new file mode 100644 index 0000000..491c8b8 --- /dev/null +++ b/src/modules/inputs/cdc-table.mapper.ts @@ -0,0 +1,38 @@ +import { + CdcColumn, + CdcTable, +} from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; + +/** What a caller knows about a CDC table before it is stored. */ +export interface CdcTableInput { + // Optional on the create DTO (CdcTableReq); the platform validates it. + table_schema?: string; + table_name: string; + primary_keys?: string[]; + iceberg_table_name?: string; + iceberg_qualify_table_name?: string; + columns?: CdcColumn[]; + column_exclude_list?: string[]; +} + +/** + * The only place maestro builds a protospack `CdcTable`. + * + * `name` is the identity key shared with batch `NewTable` (in-factory keys + * update/soft-delete on it) and the proto keeps it required, so it mirrors + * `table_name` here and nowhere else. Follow-up (protospack): make `name` + * optional and let in-factory be its sole writer (it already backfills + * `name ?? table_name`). + */ +export function toCdcTable(table: CdcTableInput): CdcTable { + return { + table_schema: table.table_schema, + table_name: table.table_name, + name: table.table_name, + primary_keys: table.primary_keys ?? [], + iceberg_table_name: table.iceberg_table_name, + iceberg_qualify_table_name: table.iceberg_qualify_table_name, + columns: table.columns ?? [], + column_exclude_list: table.column_exclude_list ?? [], + }; +} diff --git a/src/modules/inputs/dtos/input.model.ts b/src/modules/inputs/dtos/input.model.ts index f7d3aa4..0b2490c 100644 --- a/src/modules/inputs/dtos/input.model.ts +++ b/src/modules/inputs/dtos/input.model.ts @@ -78,8 +78,10 @@ export class CdcColumnReq { export class CdcTableReq { @ApiProperty() name: string; - @ApiPropertyOptional() - table_schema?: string; + // Source database/schema. Required: Debezium addresses tables as + // `schema.table`, a CDC table without it cannot be replicated. + @ApiProperty() + table_schema: string; @ApiPropertyOptional({ type: [String] }) primary_keys?: string[]; // Per-table raw Iceberg table name override (iceberg destination only). diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 80eef5b..5e622e7 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -13,17 +13,21 @@ import { objectCamelToSnake } from 'src/utils/CaseConverter'; import { IIdRequest, UpdateInputRequest } from './dtos/old_interfaces'; import { Input } from '@dadosfera/protospack-v2'; import { + AddCdcTableRequest, GetAvailableEntitiesRequest, InputCreateGenericRequest, InputCreateCdcRequest, InputCreateS3Request, InputNewCreateRequest, InputUpdateResponse, + MarkTableDeletedRequest, + RemoveCdcTableRequest, RollbackInputRequest, TestConnectionRequest, } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; import { CreateCdcInputReq, CreateInputReq } from './dtos/input.model'; +import { toCdcTable } from './cdc-table.mapper'; import { Metadata } from '@grpc/grpc-js'; @Injectable() @@ -192,16 +196,9 @@ export class InputsService { name: body.name, plugin: body.plugin, read_only: body.read_only ?? true, - tables: body.tables.map((t) => ({ - table_schema: t.table_schema, - table_name: t.name, - name: t.name, // canonical identity == table_name (in-factory also backfills) - primary_keys: t.primary_keys ?? [], - iceberg_table_name: t.iceberg_table_name, - iceberg_qualify_table_name: t.iceberg_qualify_table_name, - columns: t.columns ?? [], - column_exclude_list: t.column_exclude_list ?? [], - })), + tables: body.tables.map((t) => + toCdcTable({ ...t, table_name: t.name }), + ), destination: body.destination, }, info, @@ -320,19 +317,19 @@ export class InputsService { return formatedPayload; } - async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) { + async markTableDeleted(data: MarkTableDeletedRequest) { return lastValueFrom(this.inputWriteService.MarkTableDeleted(data)); } - async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) { - return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data)); + async unmarkTableDeleted(data: MarkTableDeletedRequest) { + return lastValueFrom(this.inputWriteService.UnmarkTableDeleted(data)); } - async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) { - return lastValueFrom(this.inputWriteService.AddCdcTable(data as any)); + async addCdcTable(data: AddCdcTableRequest) { + return lastValueFrom(this.inputWriteService.AddCdcTable(data)); } - async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) { - return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data)); + async removeCdcTable(data: RemoveCdcTableRequest) { + return lastValueFrom(this.inputWriteService.RemoveCdcTable(data)); } } diff --git a/src/modules/pipelinesV2/interfaces.ts b/src/modules/pipelinesV2/interfaces.ts index c54e74a..90bf076 100644 --- a/src/modules/pipelinesV2/interfaces.ts +++ b/src/modules/pipelinesV2/interfaces.ts @@ -9,6 +9,33 @@ export class PipelineInputsDTO { }> } +export class PipelineTableDestination { + @ApiPropertyOptional() + table_schema?: string; + @ApiPropertyOptional() + table_name?: string; +} + +/** One entry of the create body's `config.tables`. Mirrors the frontend's + * EntityWithColumnsNames; only the fields maestro/pi-factory read are typed, + * the rest travels as-is. */ +export class PipelineTableConfig { + @ApiProperty() + name: string; + @ApiPropertyOptional({ type: () => PipelineTableDestination }) + destinations?: Partial>; + [extra: string]: unknown; +} + +/** `{ cron, tables }` as sent by the frontend on create; JSON-serialized onto + * the gRPC `config` string field. */ +export class PipelineConfig { + @ApiPropertyOptional() + cron?: string; + @ApiPropertyOptional({ type: [PipelineTableConfig] }) + tables?: PipelineTableConfig[]; +} + export class IPipelineV2 { @ApiProperty() id: string; @@ -31,8 +58,8 @@ export class IPipelineV2 { tags?: string[]; @ApiPropertyOptional() properties?: any; - @ApiPropertyOptional() - config?: any; + @ApiPropertyOptional({ type: () => PipelineConfig }) + config?: PipelineConfig; @ApiProperty() connector_name: string; diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index 77c8059..62287c4 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -1,3 +1,4 @@ +import { isCdcPlugin } from 'src/utils/cdc'; /* eslint-disable no-async-promise-executor */ import { BadRequestException, @@ -437,7 +438,7 @@ export class PipelinesService implements OnModuleInit { }); this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput)); - const isCdc = !!oldInput.plugin?.endsWith('_cdc'); + const isCdc = isCdcPlugin(oldInput.plugin); const pipelineIdFormat = pipelineId.split('-').join('_'); const rollback: RollbackPromise[] = []; diff --git a/src/modules/platform-api/platform-api.controller.spec.ts b/src/modules/platform-api/platform-api.controller.spec.ts index f3d58a3..ff1bb1a 100644 --- a/src/modules/platform-api/platform-api.controller.spec.ts +++ b/src/modules/platform-api/platform-api.controller.spec.ts @@ -157,6 +157,7 @@ describe('PlatformApiController - addTable', () => { const result = await controller.addTable('pid', 'iid', body, mockUser); expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + client_id: 'c1', id: 'iid', table: { table_schema: 'public', @@ -198,6 +199,7 @@ describe('PlatformApiController - addTable', () => { await controller.addTable('pid', 'iid', icebergBody, mockUser); expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + client_id: 'c1', id: 'iid', table: { table_schema: 'public', @@ -228,6 +230,7 @@ describe('PlatformApiController - addTable', () => { await controller.addTable('pid', 'iid', columnsBody, mockUser); expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + client_id: 'c1', id: 'iid', table: { table_schema: 'public', @@ -254,6 +257,7 @@ describe('PlatformApiController - addTable', () => { await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down'); expect(inputsService.removeCdcTable).toHaveBeenCalledWith({ + client_id: 'c1', id: 'iid', table_name: 'orders', info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 44e1ac9..a5c0420 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 { toCdcTable } from '../inputs/cdc-table.mapper'; import { PipelinesService } from '../pipelinesV2/pipelines.service'; import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; @@ -1147,19 +1148,10 @@ export class PlatformApiController { 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 ?? [], - }; + const cdcTable = toCdcTable(body); this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name }); - await this.inputsService.addCdcTable({ id: inputId, table: cdcTable, info }); + await this.inputsService.addCdcTable({ client_id: user.customer_id, id: inputId, table: cdcTable, info }); this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name }); try { @@ -1185,7 +1177,7 @@ export class PlatformApiController { } 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 }); + await this.inputsService.removeCdcTable({ client_id: user.customer_id, id: inputId, table_name: body.table_name, info }); } catch (rbErr) { this.logger.error('addTable: rollback failed', { error: rbErr.message }); }