From 5c775779928a4c767f8417ccc84bc2fa1bd5feba Mon Sep 17 00:00:00 2001 From: iruy-fr Date: Fri, 19 Jun 2026 17:04:42 -0300 Subject: [PATCH 1/4] feat: add endpoint to retrieve pipeline run jobs --- docsfera.json | 58 ++++++++++++++++++- .../platform-api/platform-api.controller.ts | 20 ++++++- 2 files changed, 75 insertions(+), 3 deletions(-) diff --git a/docsfera.json b/docsfera.json index 00b43df..aae1a29 100644 --- a/docsfera.json +++ b/docsfera.json @@ -4659,6 +4659,62 @@ ] } }, + "/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": { + "get": { + "operationId": "PlatformApiController_getPipelineRunJobs", + "summary": "Get pipeline run jobs", + "description": "Proxies platform-api DB-backed job runs and returns `{ jobs: [...] }`.", + "parameters": [ + { + "name": "pipelineId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "runId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "DB-backed job runs for the selected pipeline run.", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "jobs": { + "type": "array", + "items": { + "type": "object" + } + } + }, + "required": [ + "jobs" + ] + } + } + } + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + } + ] + } + }, "/platform/jobs/{jobId}/input": { "put": { "operationId": "PlatformApiController_updateJobInput", @@ -11662,4 +11718,4 @@ } } } -} \ No newline at end of file +} diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index f3abaf5..4263842 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -14,7 +14,7 @@ import { NotFoundException, UseGuards, } from '@nestjs/common'; -import { ApiTags, ApiOperation } from '@nestjs/swagger'; +import { ApiTags, ApiOperation, ApiOkResponse } from '@nestjs/swagger'; import { DadosferaLogger } from '@dadosfera/dadosfera-logs'; import { @@ -833,7 +833,23 @@ export class PlatformApiController { } @Get('pipelines/:pipelineId/pipeline_run/:runId/jobs') - @ApiOperation({ summary: 'Get pipeline run jobs' }) + @ApiOperation({ + summary: 'Get pipeline run jobs', + description: 'Proxies platform-api DB-backed job runs and returns `{ jobs: [...] }`.', + }) + @ApiOkResponse({ + description: 'DB-backed job runs for the selected pipeline run.', + schema: { + type: 'object', + properties: { + jobs: { + type: 'array', + items: { type: 'object' }, + }, + }, + required: ['jobs'], + }, + }) @RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET) async getPipelineRunJobs( @Param('pipelineId') pipelineId: string, From 63efff6adf764b010a67422dfe05dabf98e916ef Mon Sep 17 00:00:00 2001 From: iruy-fr Date: Mon, 22 Jun 2026 20:25:16 -0300 Subject: [PATCH 2/4] FEAT: enhance pipeline run jobs endpoint with error handling and response structure --- docsfera.json | 78 +++++------- .../pipelinesV2/pipelines.controller.ts | 116 +++++++++++++++++- .../platform-api/platform-api.controller.ts | 40 ++++-- 3 files changed, 170 insertions(+), 64 deletions(-) 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 ==================== From 17363e74f414fbb9c05bf0298be5bb76fedf4688 Mon Sep 17 00:00:00 2001 From: iruy-fr Date: Tue, 23 Jun 2026 11:54:59 -0300 Subject: [PATCH 3/4] FEAT: simplify pipeline run jobs handling and normalize run ID usage --- docsfera.json | 9 +- .../pipelinesV2/pipelines.controller.ts | 116 +----------------- .../platform-api/platform-api.controller.ts | 34 ++--- 3 files changed, 18 insertions(+), 141 deletions(-) diff --git a/docsfera.json b/docsfera.json index f6515c0..0d70456 100644 --- a/docsfera.json +++ b/docsfera.json @@ -3343,14 +3343,7 @@ ], "responses": { "200": { - "description": "", - "content": { - "application/json": { - "schema": { - "type": "object" - } - } - } + "description": "" } }, "tags": [ diff --git a/src/modules/pipelinesV2/pipelines.controller.ts b/src/modules/pipelinesV2/pipelines.controller.ts index 8e9a0e4..70fab15 100644 --- a/src/modules/pipelinesV2/pipelines.controller.ts +++ b/src/modules/pipelinesV2/pipelines.controller.ts @@ -49,23 +49,9 @@ 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 }]) @@ -77,70 +63,10 @@ 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) { @@ -263,38 +189,18 @@ export class PipelinesController { @Get(':id/status') @RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET) - 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, - }, - }; + async getPipelineStatus(@Body() body, @Param('id') id: string) { + + body.id = id; this.logger.info(`/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`, { user: body.info.user_id, customer: body.info.customer, }); - try { - return await this.pipelinesClientService.getPipelineStatus(body); - } catch (error) { - this.logger.warn('PipelinesController - getPipelineStatus fallback to platform-api', { - id, - error: error.message, - }); + const response = await this.pipelinesClientService.getPipelineStatus(body); - return this.platformApiService.proxy( - 'GET', - `/pipeline/${this.normalizePipelineId(id)}/pipeline_run`, - user, - ); - } + return response; } @Get('/:id') @@ -316,17 +222,7 @@ export class PipelinesController { language, }); - 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 pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata); 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 ba6ddd1..e8de5ca 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -782,10 +782,10 @@ export class PlatformApiController { @User() user: RequestUser, ) { const normalizedPipelineId = this.normalizePipelineId(pipelineId); - const decodedRunId = this.decodePathParam(runId); + const normalizedRunId = this.normalizePipelineId(runId); return this.platformApiService.proxy( 'GET', - `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}`, + `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`, user, ); } @@ -798,10 +798,10 @@ export class PlatformApiController { @User() user: RequestUser, @Query() query: Record, ) { - const decodedRunId = this.decodePathParam(runId); + const normalizedRunId = this.normalizePipelineId(runId); return this.platformApiService.proxy( 'GET', - `/pipeline/pipeline_run/${decodedRunId}/logs`, + `/pipeline/pipeline_run/${normalizedRunId}/logs`, user, undefined, query, @@ -817,7 +817,7 @@ export class PlatformApiController { @User() user: RequestUser, ) { const normalizedPipelineId = this.normalizePipelineId(pipelineId); - const decodedRunId = this.decodePathParam(runId); + const normalizedRunId = this.normalizePipelineId(runId); const status = await this.platformApiService.proxy( 'GET', @@ -831,7 +831,7 @@ export class PlatformApiController { return this.platformApiService.proxy( 'POST', - `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/cancel`, + `/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`, user, ); } @@ -863,23 +863,11 @@ export class PlatformApiController { const normalizedPipelineId = this.normalizePipelineId(pipelineId); const decodedRunId = this.decodePathParam(runId); - 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; - } + return this.platformApiService.proxy( + 'GET', + `/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/jobs`, + user, + ); } // ==================== JOBS - COLUMN EDITING ROUTES ==================== From 011032e3d4c323350409b0a69f18c7976c06b315 Mon Sep 17 00:00:00 2001 From: iruy-fr Date: Tue, 23 Jun 2026 11:58:28 -0300 Subject: [PATCH 4/4] FEAT: update API title in docsfera.json to reflect project name --- docsfera.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docsfera.json b/docsfera.json index 0d70456..ae9f040 100644 --- a/docsfera.json +++ b/docsfera.json @@ -8629,7 +8629,7 @@ } }, "info": { - "title": "Maestro - feat/pipeline-run-jobs", + "title": "Maestro", "description": "This is the Maestro API", "version": "1.0.0", "contact": {}