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 <contact@withwoz.com>

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 ?? [],
+  };
+}

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).

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

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<Record<'raw' | 'qualify', PipelineTableDestination>>;
+  [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;

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[] = [];
 

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' },

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