diff --git a/docsfera.json b/docsfera.json index f6130d2..f6515c0 100644 --- a/docsfera.json +++ b/docsfera.json @@ -3343,7 +3343,14 @@ ], "responses": { "200": { - "description": "" + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } } }, "tags": [ @@ -4570,50 +4577,6 @@ ] } }, - "/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": { - "get": { - "operationId": "PlatformApiController_getPipelineRunJobs", - "summary": "Get pipeline run jobs", - "parameters": [ - { - "name": "pipelineId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - }, - { - "name": "runId", - "required": true, - "in": "path", - "schema": { - "type": "string" - } - } - ], - "responses": { - "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } - } - }, - "tags": [ - "Platform API" - ], - "security": [ - { - "access-token": [] - } - ] - } - }, "/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": { "get": { "operationId": "PlatformApiController_getPipelineRunJobs", @@ -8647,10 +8610,33 @@ "Health" ] } + }, + "/release_note": { + "get": { + "operationId": "ReleaseNoteController_getLatestReleaseNote", + "parameters": [], + "responses": { + "200": { + "description": "", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + } + }, + "security": [ + { + "access-token": [] + } + ] + } } }, "info": { - "title": "Maestro", + "title": "Maestro - feat/pipeline-run-jobs", "description": "This is the Maestro API", "version": "1.0.0", "contact": {} diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index 70fab15..8e9a0e4 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -49,9 +49,23 @@ import { Language } from 'src/decorators/language.decorator'; import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard'; +import { PlatformApiService } from '../platform-api/platform-api.service'; type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any }; type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] }; +type PlatformPipelineJob = { job_id?: string; input?: Record }; +type PlatformPipeline = { + pipeline_id?: string; + created_at?: string; + description?: string; + name?: string; + last_status?: string; + status?: string; + cron?: string; + jobs?: PlatformPipelineJob[]; + properties?: Record; + user_id?: string; +}; @ApiTags('PipelinesV2') @ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }]) @@ -63,10 +77,70 @@ export class PipelinesController { @Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger, private pipelinesClientService: PipelinesService, + private platformApiService: PlatformApiService, ) { this.logger = dadosferaLogger.logger; } + private normalizePipelineId(id: string): string { + return id?.replace(/-/g, '_') || ''; + } + + private buildPlatformPipelineFallback( + id: string, + platformPipeline: PlatformPipeline, + ): Messages.PipelineV2FindOneResponse { + const jobs = platformPipeline.jobs || []; + const firstInput = jobs.find((job) => job.input)?.input || {}; + const plugin = firstInput.plugin || firstInput.connector; + + const tables = jobs.map((job) => ({ + ...job.input, + name: + job.input?.table_name || + job.input?.source_prefix || + job.input?.stream || + job.job_id || + '', + job_id: job.job_id, + })); + + return { + pipeline: { + id, + created_at: platformPipeline.created_at, + description: platformPipeline.description, + name: platformPipeline.name, + transformations: [], + status: platformPipeline.status || platformPipeline.last_status || '', + input: { + ...firstInput, + plugin, + category: firstInput.category || (firstInput.connector === 's3' ? 'file' : 'database'), + cron: platformPipeline.cron, + tables, + input_id: firstInput.input_id || null, + }, + properties: platformPipeline.properties || {}, + username: platformPipeline.user_id, + } as any, + }; + } + + private async getPipelineFromPlatformApi( + id: string, + user: RequestUser, + ): Promise { + const normalizedId = this.normalizePipelineId(id); + const platformPipeline = await this.platformApiService.proxy( + 'GET', + `/pipeline/${normalizedId}`, + user, + ); + + return this.buildPlatformPipelineFallback(id, platformPipeline); + } + @Get('monitoring-dashboard') @RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getMonitoringDashboard(@User() user: RequestUser) { @@ -189,18 +263,38 @@ export class PipelinesController { @Get(':id/status') @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET) - async getPipelineStatus(@Body() body, @Param('id') id: string) { - - body.id = id; + async getPipelineStatus( + @Param('id') id: string, + @User() user: RequestUser, + ) { + const body = { + id, + info: { + customer_id: user.customer_id, + user_id: user.user_id, + customer: user.customer_name, + }, + }; this.logger.info(`/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`, { user: body.info.user_id, customer: body.info.customer, }); - const response = await this.pipelinesClientService.getPipelineStatus(body); + try { + return await this.pipelinesClientService.getPipelineStatus(body); + } catch (error) { + this.logger.warn('PipelinesController - getPipelineStatus fallback to platform-api', { + id, + error: error.message, + }); - return response; + return this.platformApiService.proxy( + 'GET', + `/pipeline/${this.normalizePipelineId(id)}/pipeline_run`, + user, + ); + } } @Get('/:id') @@ -222,7 +316,17 @@ export class PipelinesController { language, }); - const pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata); + let pipelineRes: Messages.PipelineV2FindOneResponse; + try { + pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata); + } catch (error) { + this.logger.warn('PipelinesController - findOne fallback to platform-api', { + id, + error: error.message, + }); + + return this.getPipelineFromPlatformApi(id, user); + } const parsed: PipelineTablesConfig = JSON.parse(pipelineRes.pipeline.config.tables); const input_id = parsed.input_id; diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 4263842..ba6ddd1 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -72,6 +72,10 @@ export class PlatformApiController { return id?.replace(/-/g, '_') || ''; } + private decodePathParam(value: string): string { + return value ? decodeURIComponent(value) : ''; + } + /** * Denormalize ID back to UUID format (replace _ with -). * Used when we receive a normalized ID but need the original UUID. @@ -778,10 +782,10 @@ export class PlatformApiController { @User() user: RequestUser, ) { const normalizedPipelineId = this.normalizePipelineId(pipelineId); - const normalizedRunId = this.normalizePipelineId(runId); + const decodedRunId = this.decodePathParam(runId); return this.platformApiService.proxy( 'GET', - `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`, + `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}`, user, ); } @@ -794,10 +798,10 @@ export class PlatformApiController { @User() user: RequestUser, @Query() query: Record, ) { - const normalizedRunId = this.normalizePipelineId(runId); + const decodedRunId = this.decodePathParam(runId); return this.platformApiService.proxy( 'GET', - `/pipeline/pipeline_run/${normalizedRunId}/logs`, + `/pipeline/pipeline_run/${decodedRunId}/logs`, user, undefined, query, @@ -813,7 +817,7 @@ export class PlatformApiController { @User() user: RequestUser, ) { const normalizedPipelineId = this.normalizePipelineId(pipelineId); - const normalizedRunId = this.normalizePipelineId(runId); + const decodedRunId = this.decodePathParam(runId); const status = await this.platformApiService.proxy( 'GET', @@ -827,7 +831,7 @@ export class PlatformApiController { return this.platformApiService.proxy( 'POST', - `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`, + `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/cancel`, user, ); } @@ -857,13 +861,25 @@ export class PlatformApiController { @User() user: RequestUser, ) { const normalizedPipelineId = this.normalizePipelineId(pipelineId); - const normalizedRunId = this.normalizePipelineId(runId); + const decodedRunId = this.decodePathParam(runId); - return this.platformApiService.proxy( - 'GET', - `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/jobs`, - user, - ); + try { + return await this.platformApiService.proxy( + 'GET', + `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/jobs`, + user, + ); + } catch (error) { + if (error instanceof HttpException && error.getStatus() === 404) { + this.logger.warn('Platform API pipeline run jobs not found; returning empty jobs list', { + pipelineId: normalizedPipelineId, + runId: decodedRunId, + }); + return { jobs: [] }; + } + + throw error; + } } // ==================== JOBS - COLUMN EDITING ROUTES ====================