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 ====================