diff --git a/docsfera.json b/docsfera.json index ff2d395..da23ed4 100644 --- a/docsfera.json +++ b/docsfera.json @@ -5205,6 +5205,53 @@ ] } }, + "/platform/pipelines/{pipelineId}/inputs/{inputId}/tables": { + "post": { + "operationId": "PlatformApiController_addTable", + "summary": "Add a CDC table to an input and dispatch its platform jobs", + "parameters": [ + { + "name": "pipelineId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "inputId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "201": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + } + }, "/platform/jobs/jdbc/configs/allowed_datatypes": { "get": { "operationId": "PlatformApiController_getJdbcAllowedDatatypes", @@ -10921,6 +10968,9 @@ "properties": { "type": "object" }, + "config": { + "type": "object" + }, "connector_name": { "type": "string" }, @@ -10985,6 +11035,9 @@ "properties": { "type": "object" }, + "config": { + "type": "object" + }, "connector_name": { "type": "string" }, diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 29048ed..ad31e1c 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -322,4 +322,12 @@ export class InputsService { async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) { return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data)); } + + async addCdcTable(data: { client_id?: string; id: string; table: any; info: Info }) { + return lastValueFrom(this.inputWriteService.AddCdcTable(data as any)); + } + + async removeCdcTable(data: { client_id?: string; id: string; table_name: string; info: Info }) { + return lastValueFrom((this.inputWriteService as any).RemoveCdcTable(data)); + } } diff --git a/src/modules/pipelinesV2/pipelines.module.ts b/src/modules/pipelinesV2/pipelines.module.ts index b4f24af..efd3c74 100644 --- a/src/modules/pipelinesV2/pipelines.module.ts +++ b/src/modules/pipelinesV2/pipelines.module.ts @@ -1,4 +1,4 @@ -import { Module } from '@nestjs/common'; +import { Module, forwardRef } from '@nestjs/common'; import { ClientsModule } from '@nestjs/microservices'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; @@ -23,7 +23,7 @@ const client = new PipelinesClientConfiguration(); ConnectorModule, InputsModule, TransformationsModule, - PlatformApiModule, + forwardRef(() => PlatformApiModule), NimbusServicesModule, CatalogModule ], diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index b3d641b..77c8059 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -19,7 +19,7 @@ import { lastValueFrom } from 'rxjs'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { PipelinesClientConfiguration } from './pipelines-client'; import { ICreatePipelineV2Req, IIdRequest, UpdatePlatformInputRequest, UpdateTableDTO } from './interfaces'; -import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages'; +import { PipelineV2CreateRequest, AddCdcJobsRequest, AddCdcJobsResponse } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages'; import { Metadata } from '@grpc/grpc-js'; import { ConnectorClientService } from '../connector/client.service'; import { InputsService } from '../inputs/inputs.service'; @@ -727,6 +727,10 @@ export class PipelinesService implements OnModuleInit { return statusPipelineResponse; } + async addCdcJobs(data: AddCdcJobsRequest): Promise { + return lastValueFrom(this.pipelineWriteService.AddCdcJobs(data)); + } + async runPipeline({ id, info }: IIdRequest) { this.logger.info('PipelinesClientService - RunPipeline'); const statusPipelineResponse = await lastValueFrom( diff --git a/src/modules/platform-api/platform-api.controller.spec.ts b/src/modules/platform-api/platform-api.controller.spec.ts index c235dd3..25077a1 100644 --- a/src/modules/platform-api/platform-api.controller.spec.ts +++ b/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); + }); + + 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' }, + }); + }); +}); diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index b10b09e..0a6c6cc 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 { 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') diff --git a/src/modules/platform-api/platform-api.module.ts b/src/modules/platform-api/platform-api.module.ts index b2b0726..726e743 100644 --- a/src/modules/platform-api/platform-api.module.ts +++ b/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],