diff --git a/docsfera.json b/docsfera.json index a99d60e..fc5e27b 100644 --- a/docsfera.json +++ b/docsfera.json @@ -4705,6 +4705,43 @@ ] } }, + "/platform/pipelines/{pipelineId}/inputs/{inputId}": { + "delete": { + "operationId": "PlatformApiController_deleteTable", + "summary": "Mark a table as deleted and delete its associated job via platform-api", + "parameters": [ + { + "name": "pipelineId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "inputId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "" + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + } + ] + } + }, "/platform/jobs/jdbc/{jobId}": { "get": { "operationId": "PlatformApiController_getJdbcJob", diff --git a/package-lock.json b/package-lock.json index 4d19410..8fd7864 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,7 +17,7 @@ "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", "@dadosfera/protospack": "2.5.3", - "@dadosfera/protospack-v2": "3.40.0-beta.2", + "@dadosfera/protospack-v2": "^3.40.0-beta.5", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", @@ -1745,10 +1745,9 @@ } }, "node_modules/@dadosfera/protospack-v2": { - "version": "3.40.0-beta.2", - "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.2.tgz", - "integrity": "sha512-WMoL9OhKJ05qDojd7cPcP7x45bHHnnQttfVoTWrc+lOkcVHygu0ob52ArfD8RRmpAXezBh9vDSbxz45n/lxbKg==", - "license": "ISC", + "version": "3.40.0-beta.5", + "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.5.tgz", + "integrity": "sha512-iocKv/XXp2jKAasO5ONgm31cKfLgNsU4pEKZMr6YnR7nQaH11WcW7rnuagNxWoik++wLUqbYyf0bZWRDzMlCPA==", "dependencies": { "@grpc/grpc-js": "^1.9.3", "rxjs": "^7.5.5" diff --git a/package.json b/package.json index 86ab824..b750c0e 100644 --- a/package.json +++ b/package.json @@ -35,7 +35,7 @@ "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", "@dadosfera/protospack": "2.5.3", - "@dadosfera/protospack-v2": "3.40.0-beta.2", + "@dadosfera/protospack-v2": "^3.40.0-beta.5", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 46ba2a4..96df891 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -288,4 +288,12 @@ export class InputsService { }; return formatedPayload; } + + async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) { + 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)); + } } diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index 1c00bcd..e91d217 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -53,6 +53,9 @@ import { TableColumns } from '../inputs/dtos/input.model'; import { UpdateInputRequest } from '../inputs/dtos/old_interfaces'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; +type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any }; +type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] }; + @ApiTags('PipelinesV2') @ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }]) @UseFilters(new GrpcToHttpExceptionFilter()) @@ -62,6 +65,7 @@ export class PipelinesController { constructor( @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, + private pipelinesClientService: PipelinesService, private oldPipelinesService: OldPipelineService, ) { @@ -222,30 +226,27 @@ export class PipelinesController { language, }); - const result = await this.pipelinesClientService - .findOne({ id }, metadata) - .then((res) => { - //{pipeline:{tables: {tables: [], input_id: ''}}} - let tables = JSON.parse(res.pipeline.config.tables); - const input_id = tables?.input_id; - if (tables?.tables) tables = tables.tables; - Object.assign(res.pipeline, { - transformations: res.pipeline.transformations - ? JSON.parse(res.pipeline.transformations) - : [], - config: { - cron: res.pipeline.config.cron, - tables, - input_id - }, - properties: res.pipeline.properties - ? JSON.parse(res.pipeline.properties) - : {}, - }); - return res; - }); + const pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata); - return result; + const parsed: PipelineTablesConfig = JSON.parse(pipelineRes.pipeline.config.tables); + const input_id = parsed.input_id; + const tables: PipelineTable[] = parsed.tables ?? []; + + Object.assign(pipelineRes.pipeline, { + transformations: pipelineRes.pipeline.transformations + ? JSON.parse(pipelineRes.pipeline.transformations) + : [], + config: { + cron: pipelineRes.pipeline.config.cron, + tables, + input_id, + }, + properties: pipelineRes.pipeline.properties + ? JSON.parse(pipelineRes.pipeline.properties) + : {}, + }); + + return pipelineRes; } @Patch('/:id') diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 8c42102..5fd0b94 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -11,6 +11,7 @@ import { Inject, BadRequestException, HttpException, + NotFoundException, } from '@nestjs/common'; import { ApiTags, ApiOperation } from '@nestjs/swagger'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; @@ -29,6 +30,7 @@ import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation'; import { CatalogService } from '../catalog/catalog.service'; import { PackTheMetadata } from '../../utils/PackTheMetadata'; import { ValidationTableDTO } from './platform-api.dto'; +import { InputsService } from '../inputs/inputs.service'; type ValidateTablesDTO = { @@ -54,6 +56,7 @@ export class PlatformApiController { private readonly dynamoDBService: DynamoDBService, private readonly customersService: CustomersService, private readonly catalogService: CatalogService, + private readonly inputsService: InputsService, @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, ) { this.logger = dadosferaLogger.logger; @@ -962,6 +965,51 @@ export class PlatformApiController { ); } + @Delete('pipelines/:pipelineId/inputs/:inputId') + @ApiOperation({ summary: 'Mark a table as deleted and delete its associated job via platform-api' }) + @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE) + async deleteTable( + @Param('pipelineId') pipelineId: string, + @Param('inputId') inputId: string, + @Body() body: { table_name: string }, + @User() user: RequestUser, + ) { + const tableName = body.table_name; + const info = { + customer_id: user.customer_id, + customer: user.customer_name, + user_id: user.user_id, + }; + + this.logger.info('deleteTable: marking table as deleted', { inputId, tableName }); + const updatedInput: any = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info }); + this.logger.info('deleteTable: table marked as deleted', { inputId, tableName }); + + try { + const normalizedPipelineId = this.normalizePipelineId(pipelineId); + this.logger.info('deleteTable: fetching pipeline from platform-api', { pipelineId, normalizedPipelineId }); + const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user); + this.logger.info('deleteTable: pipeline fetched', { jobCount: platformPipeline?.jobs?.length }); + + const job = platformPipeline?.jobs?.find((j: any) => j.input?.table_name === tableName); + if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`); + + this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id }); + await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user); + this.logger.info('deleteTable: job deleted', { jobId: job.job_id }); + + return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at }; + } catch (error) { + this.logger.error('deleteTable: platform-api delete failed, attempting rollback', { tableName, error: error.message }); + try { + await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: tableName, info }); + } catch (rollbackError) { + this.logger.error('deleteTable: rollback failed', { tableName, error: rollbackError.message }); + } + throw error; + } + } + // ==================== JOBS - JDBC SYNC MODE ROUTES ==================== @Get('jobs/jdbc/:jobId') diff --git a/src/modules/platform-api/platform-api.module.ts b/src/modules/platform-api/platform-api.module.ts index b78e680..b2b0726 100644 --- a/src/modules/platform-api/platform-api.module.ts +++ b/src/modules/platform-api/platform-api.module.ts @@ -8,9 +8,10 @@ import { ElasticsearchModule } from '../../services/elasticsearch'; import { DynamoDBModule } from '../../services/dynamodb'; import { CustomersModule } from '../customers/customers.module'; import { CatalogModule } from '../catalog/catalog.module'; +import { InputsModule } from '../inputs/inputs.module'; @Module({ - imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule], + imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, InputsModule], controllers: [PlatformApiController], providers: [PlatformApiService, DadosferaLogger], exports: [PlatformApiService], diff --git a/src/services/dynamodb/dynamodb.service.ts b/src/services/dynamodb/dynamodb.service.ts index 15dde41..0022b35 100644 --- a/src/services/dynamodb/dynamodb.service.ts +++ b/src/services/dynamodb/dynamodb.service.ts @@ -125,9 +125,13 @@ export class DynamoDBService { }, }); - const { Item } = await this.documentClient.send(getCommand); - - return Item as InputDocument | null; + try { + const { Item } = await this.documentClient.send(getCommand); + return Item as InputDocument | null; + } catch (error) { + this.logger.error('DynamoDB: findInput failed', { inputId, clientId, error: error.message }); + throw error; + } } async deleteInput(clientId: string, inputId: string): Promise { @@ -203,7 +207,6 @@ export class DynamoDBService { updatedTable.reference_column = changes.reference_column; } } - tables[tableIndex] = updatedTable; // Save updated document @@ -231,4 +234,5 @@ export class DynamoDBService { throw error; } } + }