FEAT(cdc): carry iceberg destination through /inputs/cdc DTOs

Co-Authored-By: WOZCODE <contact@withwoz.com>

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;
 }

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>(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.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,
     };