feat(cdc): addTable route — DynamoDB-first + rollback

src/modules/platform-api/platform-api.controller.spec.ts

@@ -5,6 +5,7 @@
 jest.mock('../customers/customers.service', () => ({ CustomersService: class {} }));
 jest.mock('../catalog/catalog.service', () => ({ CatalogService: class {} }));
 jest.mock('../inputs/inputs.service', () => ({ InputsService: class {} }));
+jest.mock('../pipelinesV2/pipelines.service', () => ({ PipelinesService: class {} }));
 
 import { Test, TestingModule } from '@nestjs/testing';
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
@@ -16,6 +17,7 @@ import { DynamoDBService } from '../../services/dynamodb';
 import { CustomersService } from '../customers/customers.service';
 import { CatalogService } from '../catalog/catalog.service';
 import { InputsService } from '../inputs/inputs.service';
+import { PipelinesService } from '../pipelinesV2/pipelines.service';
 
 const logger = {
   info: (...args) => args,
@@ -49,6 +51,7 @@ describe('PlatformApiController - deleteTable', () => {
         { provide: CustomersService, useValue: {} },
         { provide: CatalogService, useValue: {} },
         { provide: InputsService, useValue: inputsService },
+        { provide: PipelinesService, useValue: {} },
         { provide: DadosferaLogger, useValue: { logger } },
       ],
     }).compile();
@@ -105,3 +108,94 @@ describe('PlatformApiController - deleteTable', () => {
     expect(reconfiguredConnector).toBe(false);
   });
 });
+
+describe('PlatformApiController - addTable', () => {
+  let controller: PlatformApiController;
+  let inputsService: { addCdcTable: jest.Mock; removeCdcTable: jest.Mock };
+  let pipelinesClientService: { addCdcJobs: jest.Mock };
+
+  const body = {
+    table_name: 'orders',
+    table_schema: 'public',
+    primary_keys: ['id'],
+    destinations: {
+      raw: { table_schema: 'raw', table_name: 'orders' },
+      qualify: { table_schema: 'qualify', table_name: 'orders' },
+    },
+  };
+
+  beforeEach(async () => {
+    inputsService = {
+      addCdcTable: jest.fn(),
+      removeCdcTable: jest.fn(),
+    };
+    pipelinesClientService = {
+      addCdcJobs: jest.fn(),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [PlatformApiController],
+      providers: [
+        { provide: PlatformApiService, useValue: { proxy: jest.fn() } },
+        { provide: ElasticsearchService, useValue: {} },
+        { provide: DynamoDBService, useValue: {} },
+        { provide: CustomersService, useValue: {} },
+        { provide: CatalogService, useValue: {} },
+        { provide: InputsService, useValue: inputsService },
+        { provide: PipelinesService, useValue: pipelinesClientService },
+        { provide: DadosferaLogger, useValue: { logger } },
+      ],
+    }).compile();
+
+    controller = module.get<PlatformApiController>(PlatformApiController);
+  });
+
+  it('addTable: DynamoDB append then platform AddJobs, returns job_ids', async () => {
+    inputsService.addCdcTable.mockResolvedValue({ input: {} });
+    pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_2'], skipped: [] });
+
+    const result = await controller.addTable('pid', 'iid', body, mockUser);
+
+    expect(inputsService.addCdcTable).toHaveBeenCalledWith({
+      id: 'iid',
+      table: {
+        table_schema: 'public',
+        table_name: 'orders',
+        primary_keys: ['id'],
+        name: 'orders',
+      },
+      info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' },
+    });
+    expect(pipelinesClientService.addCdcJobs).toHaveBeenCalledWith({
+      pipeline_id: 'pid',
+      input_id: 'iid',
+      tables: [{
+        table_schema: 'public',
+        table_name: 'orders',
+        primary_keys: ['id'],
+        destinations: body.destinations,
+      }],
+      info: { customer_id: 'c1', user_id: 'u1', customer: 'cust' },
+    });
+    expect(result).toEqual({ job_ids: ['p_2'], skipped: [] });
+    expect(inputsService.removeCdcTable).not.toHaveBeenCalled();
+
+    const addCdcTableOrder = inputsService.addCdcTable.mock.invocationCallOrder[0];
+    const addCdcJobsOrder = pipelinesClientService.addCdcJobs.mock.invocationCallOrder[0];
+    expect(addCdcTableOrder).toBeLessThan(addCdcJobsOrder);
+  });
+
+  it('addTable: rolls back the DynamoDB row when AddJobs fails', async () => {
+    inputsService.addCdcTable.mockResolvedValue({ input: {} });
+    pipelinesClientService.addCdcJobs.mockRejectedValue(new Error('platform down'));
+    inputsService.removeCdcTable.mockResolvedValue({});
+
+    await expect(controller.addTable('pid', 'iid', body, mockUser)).rejects.toThrow('platform down');
+
+    expect(inputsService.removeCdcTable).toHaveBeenCalledWith({
+      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 { PipelinesService } from '../pipelinesV2/pipelines.service';
 import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
 
 
@@ -61,6 +62,7 @@ export class PlatformApiController {
     private readonly customersService: CustomersService,
     private readonly catalogService: CatalogService,
     private readonly inputsService: InputsService,
+    private readonly pipelinesClientService: PipelinesService,
     @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
   ) {
     this.logger = dadosferaLogger.logger;
@@ -990,6 +992,72 @@ export class PlatformApiController {
     }
   }
 
+  @Post('pipelines/:pipelineId/inputs/:inputId/tables')
+  @ApiOperation({ summary: 'Add a CDC table to an input and dispatch its platform jobs' })
+  @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
+  @UseGuards(PipelineExecutionGuard)
+  async addTable(
+    @Param('pipelineId') pipelineId: string,
+    @Param('inputId') inputId: string,
+    @Body() body: {
+      table_name: string;
+      table_schema: string;
+      primary_keys: string[];
+      destinations: {
+        raw: { table_schema: string; table_name: string };
+        qualify: { table_schema: string; table_name: string };
+      };
+    },
+    @User() user: RequestUser,
+  ) {
+    const info = {
+      customer_id: user.customer_id,
+      customer: user.customer_name,
+      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,
+    };
+
+    this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name });
+    await this.inputsService.addCdcTable({ id: inputId, table: cdcTable, info });
+    this.logger.info('addTable: DynamoDB row appended', { inputId, tableName: body.table_name });
+
+    try {
+      const customInfo = {
+        customer_id: user.customer_id,
+        user_id: user.user_id,
+        customer: user.customer_name,
+      };
+
+      const res = await this.pipelinesClientService.addCdcJobs({
+        pipeline_id: pipelineId,
+        input_id: inputId,
+        tables: [{
+          table_schema: body.table_schema,
+          table_name: body.table_name,
+          primary_keys: body.primary_keys,
+          destinations: body.destinations,
+        }],
+        info: customInfo,
+      });
+
+      return res;
+    } 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 });
+      } catch (rbErr) {
+        this.logger.error('addTable: rollback failed', { error: rbErr.message });
+      }
+      throw error;
+    }
+  }
+
   // ==================== JOBS - JDBC SYNC MODE ROUTES ====================
 
   @Get('jobs/jdbc/configs/allowed_datatypes')

src/modules/platform-api/platform-api.module.ts

@@ -1,4 +1,4 @@
-import { Module } from '@nestjs/common';
+import { Module, forwardRef } from '@nestjs/common';
 
 import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 
@@ -9,9 +9,17 @@ import { DynamoDBModule } from '../../services/dynamodb';
 import { CustomersModule } from '../customers/customers.module';
 import { CatalogModule } from '../catalog/catalog.module';
 import { InputsModule } from '../inputs/inputs.module';
+import { PipelinesV2Module } from '../pipelinesV2/pipelines.module';
 
 @Module({
-  imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, InputsModule],
+  imports: [
+    ElasticsearchModule,
+    DynamoDBModule,
+    CustomersModule,
+    CatalogModule,
+    InputsModule,
+    forwardRef(() => PipelinesV2Module),
+  ],
   controllers: [PlatformApiController],
   providers: [PlatformApiService, DadosferaLogger],
   exports: [PlatformApiService],