Merge pull request #483 from dadosfera/beta

Beta

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",

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<boolean> {
+    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);
+    }
+
+  }
+}

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,

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<void> {
-    // 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<void> {
-    // 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,7 +804,7 @@ 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)
   async cancelPipelineRun(
@@ -859,6 +815,16 @@ export class PlatformApiController {
     const normalizedPipelineId = this.normalizePipelineId(pipelineId);
     const normalizedRunId = this.normalizePipelineId(runId);
 
+    const status = await this.platformApiService.proxy(
+      'GET',
+      `/pipeline/${normalizedPipelineId}/pipeline_run`,
+      user,
+    );
+
+    if (status.length === 1) {
+      throw new BadRequestException('The first pipeline cannot be canceled');
+    }
+
     return this.platformApiService.proxy(
       'POST',
       `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`,
@@ -968,6 +934,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,