From 0d627b04519b9e4380f3b316bee320524bc63511 Mon Sep 17 00:00:00 2001 From: marcos-silva-rodrigues Date: Mon, 13 Apr 2026 14:23:44 -0300 Subject: [PATCH] FEAT: guard to prevent pipeline update when pipeline is running --- docsfera.json | 242 +----------------- src/guards/pipeline-execution.guard.ts | 65 +++++ .../pipelinesV2/pipelines.controller.ts | 7 +- .../platform-api/platform-api.controller.ts | 80 ++---- 4 files changed, 99 insertions(+), 295 deletions(-) create mode 100644 src/guards/pipeline-execution.guard.ts diff --git a/docsfera.json b/docsfera.json index fc5e27b..1b2e1cd 100644 --- a/docsfera.json +++ b/docsfera.json @@ -4001,7 +4001,7 @@ ] } }, - "/platform/pipeline": { + "/platform/pipelines": { "post": { "operationId": "PlatformApiController_createPipeline", "summary": "Create a new pipeline", @@ -4026,9 +4026,7 @@ "access-token": [] } ] - } - }, - "/platform/pipelines": { + }, "get": { "operationId": "PlatformApiController_getPipelines", "summary": "List all pipelines for customer", @@ -4055,7 +4053,7 @@ ] } }, - "/platform/pipeline/{pipelineId}": { + "/platform/pipelines/{pipelineId}": { "get": { "operationId": "PlatformApiController_getPipeline", "summary": "Get pipeline by ID", @@ -4159,7 +4157,7 @@ ] } }, - "/platform/pipeline/execute": { + "/platform/pipelines/execute": { "post": { "operationId": "PlatformApiController_executePipeline", "summary": "Execute a pipeline", @@ -4186,7 +4184,7 @@ ] } }, - "/platform/pipeline/pause": { + "/platform/pipelines/pause": { "post": { "operationId": "PlatformApiController_pausePipeline", "summary": "Pause a pipeline", @@ -4213,7 +4211,7 @@ ] } }, - "/platform/pipeline/unpause": { + "/platform/pipelines/unpause": { "post": { "operationId": "PlatformApiController_unpausePipeline", "summary": "Unpause a pipeline", @@ -4240,7 +4238,7 @@ ] } }, - "/platform/pipeline/{pipelineId}/memory": { + "/platform/pipelines/{pipelineId}/memory": { "put": { "operationId": "PlatformApiController_updatePipelineMemory", "summary": "Update pipeline memory configuration", @@ -4276,7 +4274,7 @@ ] } }, - "/platform/pipeline/{pipelineId}/metadata": { + "/platform/pipelines/{pipelineId}/metadata": { "put": { "operationId": "PlatformApiController_updatePipelineMetadata", "summary": "Update pipeline metadata", @@ -4403,7 +4401,7 @@ ] } }, - "/platform/pipeline/{pipelineId}/pipeline_run": { + "/platform/pipelines/{pipelineId}/pipeline_run": { "get": { "operationId": "PlatformApiController_getPipelineRuns", "summary": "Get pipeline runs for a pipeline", @@ -4439,7 +4437,7 @@ ] } }, - "/platform/pipeline/{pipelineId}/pipeline_run/{runId}": { + "/platform/pipelines/{pipelineId}/pipeline_run/{runId}": { "get": { "operationId": "PlatformApiController_getPipelineRun", "summary": "Get specific pipeline run", @@ -4483,7 +4481,7 @@ ] } }, - "/platform/pipeline/pipeline_run/{runId}/logs": { + "/platform/pipelines/pipeline_run/{runId}/logs": { "get": { "operationId": "PlatformApiController_getPipelineRunLogs", "summary": "Get pipeline run logs", @@ -4519,7 +4517,7 @@ ] } }, - "/platform/pipeline/{pipelineId}/pipeline_run/{runId}/cancel": { + "/platform/pipelines/{pipelineId}/pipeline_run/{runId}/cancel": { "post": { "operationId": "PlatformApiController_cancelPipelineRun", "summary": "Cancel a running pipeline run", @@ -4742,114 +4740,6 @@ ] } }, - "/platform/jobs/jdbc/{jobId}": { - "get": { - "operationId": "PlatformApiController_getJdbcJob", - "summary": "Get JDBC job details", - "parameters": [ - { - "name": "jobId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, - "/platform/jobs/jdbc/{jobId}/sync-mode": { - "post": { - "operationId": "PlatformApiController_updateJdbcSyncMode", - "summary": "Update JDBC job sync mode", - "parameters": [ - { - "name": "jobId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "201": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, - "/platform/jobs/{jobId}/rename-tables": { - "post": { - "operationId": "PlatformApiController_renameJobTables", - "summary": "Rename job output tables and sync to catalog", - "parameters": [ - { - "name": "jobId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "201": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, "/platform/jobs/jdbc/configs/allowed_datatypes": { "get": { "operationId": "PlatformApiController_getJdbcAllowedDatatypes", @@ -4877,114 +4767,6 @@ ] } }, - "/platform/jobs/singer/{jobId}": { - "get": { - "operationId": "PlatformApiController_getSingerJob", - "summary": "Get Singer job details", - "parameters": [ - { - "name": "jobId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, - "/platform/jobs/singer/{jobId}/sync-mode": { - "post": { - "operationId": "PlatformApiController_updateSingerSyncMode", - "summary": "Update Singer job sync mode", - "parameters": [ - { - "name": "jobId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "201": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, - "/platform/jobs/s3/{jobId}": { - "get": { - "operationId": "PlatformApiController_getS3Job", - "summary": "Get S3 job details", - "parameters": [ - { - "name": "jobId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, "/platform/health": { "get": { "operationId": "PlatformApiController_healthCheck", diff --git a/src/guards/pipeline-execution.guard.ts b/src/guards/pipeline-execution.guard.ts new file mode 100644 index 0000000..ffd651b --- /dev/null +++ b/src/guards/pipeline-execution.guard.ts @@ -0,0 +1,65 @@ +import { + BadRequestException, + CanActivate, + ExecutionContext, + Inject, + Injectable, + OnModuleInit, +} from '@nestjs/common'; +import { ClientGrpc } from '@nestjs/microservices'; +import { map, Observable } from 'rxjs'; +import { PackTheMetadata } from 'src/utils/PackTheMetadata'; +import { + ReadService, + ProtoServices, +} from '@dadosfera/protospack-v2/dist/lib/PipelineV2'; +import { PipelinesClientConfiguration } from 'src/modules/pipelinesV2/pipelines-client'; +import { PlatformApiService } from 'src/modules/platform-api/platform-api.service'; +import DadosferaLogger from '@dadosfera/dadosfera-logs'; + +@Injectable() +export class PipelineExecutionGuard implements CanActivate { + logger: DadosferaLogger; + + constructor( + @Inject(DadosferaLogger) + dadosferaLogger: DadosferaLogger, + private readonly platformApiService: PlatformApiService, + ) { + this.logger = dadosferaLogger.logger; + } + + async canActivate(context: ExecutionContext): Promise { + try { + this.logger.info( + 'PipelineExecutionGuard: Checking if pipeline can be executed...', + ); + const request = context.switchToHttp().getRequest(); + const pipelineId = request.params.pipelineId; + const user = request.user; + const idRegex = /[^0-9a-zA-Z_$]+/g; + const convertedId = pipelineId.replace(idRegex, '_'); + + const status = await this.platformApiService.proxy( + 'GET', + `/pipeline/${convertedId}/pipeline_run`, + user, + ); + + const currentStatus = status[status.length - 1] + + this.logger.info('Pipeline current status response:' + JSON.stringify(currentStatus)); + + if (currentStatus.last_status.toLowerCase() === 'running') { + this.logger.error('Pipeline is running, cannot update input now'); + throw new BadRequestException('Pipeline is running, cannot update input now'); + } else { + return true; + } + } catch (error) { + this.logger.error('Error in PipelineExecutionGuard: ' + error.message); + throw new BadRequestException('Error checking pipeline status: ' + error.message); + } + + } +} diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index e91d217..22dcd05 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -14,7 +14,7 @@ import { Patch, HttpException, BadRequestException, - CacheTTL, + UseGuards, } from '@nestjs/common'; import { ApiCreatedResponse, @@ -24,7 +24,6 @@ import { ApiTags, } from '@nestjs/swagger'; import { - AuthenticateCondition, RequireAllPermissions, RequireSomePermission, } from 'src/decorators/authentication.decorator'; @@ -49,9 +48,8 @@ import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filt import { LanguageEnum } from 'src/utils/languages.enum'; import { Language } from 'src/decorators/language.decorator'; import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator'; -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'; +import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any }; type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] }; @@ -286,6 +284,7 @@ export class PipelinesController { @Patch('/:pipelineId/inputs/:id') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + @UseGuards(PipelineExecutionGuard) async updatePipelineInput( @Language() language: LanguageEnum, @Body() pipelineInputDTO: UpdatePlatformInputRequest, diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index ff5539a..0c3cd55 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -12,6 +12,7 @@ import { BadRequestException, HttpException, NotFoundException, + UseGuards, } from '@nestjs/common'; import { ApiTags, ApiOperation } from '@nestjs/swagger'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; @@ -31,6 +32,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 { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; type ValidateTablesDTO = { @@ -132,7 +134,7 @@ export class PlatformApiController { } private readonly VALID_CONNECTORS = ['jdbc', 'singer', 's3']; - private readonly MAX_MEMORY_MB = 12000; // 12GB maximum memory per pipeline/job + private readonly MAX_MEMORY_MB = 12000; // 12GB maximum memory per pipelines/job /** * Validate that connector is provided and is a valid type. @@ -403,55 +405,9 @@ export class PlatformApiController { } } - /** - * Sync sync-mode changes to DynamoDB for JDBC connectors. - * Always passes both target_load_type and incremental_column_name to ensure proper sync. - */ - private async syncJdbcSyncModeToDynamoDB( - jobId: string, - body: any, - user: RequestUser, - ): Promise { - // JDBC sync mode uses target_load_type field - const changes: any = {}; - - if ('target_load_type' in body) { - changes.target_load_type = body.target_load_type; - } - - // Handle incremental_column_name: - // - If provided in body, use that value - // - If changing to full_load, explicitly clear it - if ('incremental_column_name' in body) { - changes.incremental_column_name = body.incremental_column_name; - changes.incremental_column_type = body.incremental_column_type; - } else if (body.target_load_type === 'full_load') { - // Changing to full_load without specifying incremental_column - clear it - changes.incremental_column_name = null; - } - - await this.syncJobInputToDynamoDB(jobId, changes, user, 'jdbc'); - } - - /** - * Sync sync-mode changes to DynamoDB for Singer connectors. - */ - private async syncSingerSyncModeToDynamoDB( - jobId: string, - body: any, - user: RequestUser, - ): Promise { - // Singer sync mode uses replication_method field - // Map to DynamoDB type: FULL_TABLE -> full_load, INCREMENTAL -> incremental - if ('replication_method' in body) { - const type = body.replication_method === 'INCREMENTAL' ? 'incremental' : 'full_load'; - await this.syncJobInputToDynamoDB(jobId, { load_type: type }, user, 'singer'); - } - } - // ==================== PIPELINE ROUTES ==================== - @Post('pipeline') + @Post('pipelines') @ApiOperation({ summary: 'Create a new pipeline' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.CREATE) async createPipeline(@Body() body: any, @User() user: RequestUser) { @@ -572,7 +528,7 @@ export class PlatformApiController { ); } - @Get('pipeline/:pipelineId') + @Get('pipelines/:pipelineId') @ApiOperation({ summary: 'Get pipeline by ID' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getPipeline( @@ -583,7 +539,7 @@ export class PlatformApiController { return this.platformApiService.proxy('GET', `/pipeline/${normalizedId}`, user); } - @Patch('pipeline/:pipelineId') + @Patch('pipelines/:pipelineId') @ApiOperation({ summary: 'Update pipeline by ID' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) async updatePipeline( @@ -634,7 +590,7 @@ export class PlatformApiController { return result; } - @Delete('pipeline/:pipelineId') + @Delete('pipelines/:pipelineId') @ApiOperation({ summary: 'Delete pipeline by ID' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE) async deletePipeline( @@ -664,7 +620,7 @@ export class PlatformApiController { return result; } - @Post('pipeline/execute') + @Post('pipelines/execute') @ApiOperation({ summary: 'Execute a pipeline' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) async executePipeline(@Body() body: any, @User() user: RequestUser) { @@ -674,10 +630,10 @@ export class PlatformApiController { ...body, customer_id: user.customer_name, }; - return this.platformApiService.proxy('POST', '/pipeline/execute', user, enrichedBody); + return this.platformApiService.proxy('POST', '/pipelines/execute', user, enrichedBody); } - @Post('pipeline/pause') + @Post('pipelines/pause') @ApiOperation({ summary: 'Pause a pipeline' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) async pausePipeline(@Body() body: any, @User() user: RequestUser) { @@ -690,7 +646,7 @@ export class PlatformApiController { return this.platformApiService.proxy('POST', '/pipeline/pause', user, enrichedBody); } - @Post('pipeline/unpause') + @Post('pipelines/unpause') @ApiOperation({ summary: 'Unpause a pipeline' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) async unpausePipeline(@Body() body: any, @User() user: RequestUser) { @@ -703,7 +659,7 @@ export class PlatformApiController { return this.platformApiService.proxy('POST', '/pipeline/unpause', user, enrichedBody); } - @Put('pipeline/:pipelineId/memory') + @Put('pipelines/:pipelineId/memory') @ApiOperation({ summary: 'Update pipeline memory configuration' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) async updatePipelineMemory( @@ -726,7 +682,7 @@ export class PlatformApiController { // ==================== PIPELINE METADATA ROUTES ==================== - @Put('pipeline/:pipelineId/metadata') + @Put('pipelines/:pipelineId/metadata') @ApiOperation({ summary: 'Update pipeline metadata' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) async updatePipelineMetadata( @@ -795,7 +751,7 @@ export class PlatformApiController { // ==================== PIPELINE RUN ROUTES ==================== - @Get('pipeline/:pipelineId/pipeline_run') + @Get('pipelines/:pipelineId/pipeline_run') @ApiOperation({ summary: 'Get pipeline runs for a pipeline' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getPipelineRuns( @@ -813,7 +769,7 @@ export class PlatformApiController { ); } - @Get('pipeline/:pipelineId/pipeline_run/:runId') + @Get('pipelines/:pipelineId/pipeline_run/:runId') @ApiOperation({ summary: 'Get specific pipeline run' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getPipelineRun( @@ -830,7 +786,7 @@ export class PlatformApiController { ); } - @Get('pipeline/pipeline_run/:runId/logs') + @Get('pipelines/pipeline_run/:runId/logs') @ApiOperation({ summary: 'Get pipeline run logs' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getPipelineRunLogs( @@ -848,9 +804,10 @@ export class PlatformApiController { ); } - @Post('pipeline/:pipelineId/pipeline_run/:runId/cancel') + @Post('pipelines/:pipelineId/pipeline_run/:runId/cancel') @ApiOperation({ summary: 'Cancel a running pipeline run' }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE) + @UseGuards(PipelineExecutionGuard) async cancelPipelineRun( @Param('pipelineId') pipelineId: string, @Param('runId') runId: string, @@ -968,6 +925,7 @@ 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) + @UseGuards(PipelineExecutionGuard) async deleteTable( @Param('pipelineId') pipelineId: string, @Param('inputId') inputId: string,