diff --git a/src/modules/inputs/dtos/input.model.ts b/src/modules/inputs/dtos/input.model.ts index fa73fd1..259d756 100644 --- a/src/modules/inputs/dtos/input.model.ts +++ b/src/modules/inputs/dtos/input.model.ts @@ -73,6 +73,18 @@ export class CdcTableReq { table_schema?: string; @ApiPropertyOptional({ type: [String] }) primary_keys?: string[]; + @ApiPropertyOptional() + iceberg_table_name?: string; +} + +export class IcebergDestinationReq { + @ApiProperty() + namespace: string; +} + +export class CdcDestinationReq { + @ApiPropertyOptional({ type: IcebergDestinationReq }) + iceberg?: IcebergDestinationReq; } export class CreateCdcInputReq { @@ -84,4 +96,6 @@ export class CreateCdcInputReq { 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.service.ts b/src/modules/inputs/inputs.service.ts index ad31e1c..c221520 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -197,7 +197,9 @@ export class InputsService { 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, })), + destination: body.destination, }, info, };