From 27eed59a1f1ab973c823bed926a79c99da201613 Mon Sep 17 00:00:00 2001 From: Rafael Date: Sat, 29 Aug 2026 12:48:28 -0300 Subject: [PATCH] UPDATE: PipelineExecutionGuard decides by the pipeline's explicit connector type Review (maestro #510): the guard inferred 'editable' from the absence of run history. It now reads GET /pipeline/{id} (DB only) and short-circuits on job.input.connector === 'cdc'; batch keeps the pipeline_run check. Adds src/utils/cdc.ts as the single isCdc* helper. Co-Authored-By: WOZCODE --- src/guards/pipeline-execution.guard.spec.ts | 143 +++++++++++++------- src/guards/pipeline-execution.guard.ts | 26 +++- src/utils/cdc.ts | 34 +++++ 3 files changed, 151 insertions(+), 52 deletions(-) create mode 100644 src/utils/cdc.ts diff --git a/src/guards/pipeline-execution.guard.spec.ts b/src/guards/pipeline-execution.guard.spec.ts index 0e2dbf9..cd063d1 100644 --- a/src/guards/pipeline-execution.guard.spec.ts +++ b/src/guards/pipeline-execution.guard.spec.ts @@ -3,12 +3,23 @@ import { PipelineExecutionGuard } from './pipeline-execution.guard'; const logger = { info: jest.fn(), error: jest.fn() }; -function buildGuard(proxyImpl: jest.Mock) { - const platformApiService: any = { proxy: proxyImpl }; - return new PipelineExecutionGuard( - { logger } as any, - platformApiService, +type Routes = { + pipeline?: unknown | Error; + runs?: unknown | Error; +}; + +// The guard makes up to two platform-api reads: GET /pipeline/{id} (type) +// and, for batch only, GET /pipeline/{id}/pipeline_run (last run). +function buildGuard(routes: Routes) { + const proxy = jest.fn((method: string, path: string) => { + const answer = path.endsWith('/pipeline_run') ? routes.runs : routes.pipeline; + return answer instanceof Error ? Promise.reject(answer) : Promise.resolve(answer); + }); + const guard = new PipelineExecutionGuard( + { logger } as unknown as ConstructorParameters[0], + { proxy } as unknown as ConstructorParameters[1], ); + return { guard, proxy }; } function contextWith(pipelineId = 'abc-123') { @@ -16,56 +27,96 @@ function contextWith(pipelineId = 'abc-123') { switchToHttp: () => ({ getRequest: () => ({ params: { pipelineId }, user: {} }), }), - } as any; + } as unknown as Parameters[0]; } +const cdcPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'cdc', table_name: 't' } }] }; +const batchPipeline = { jobs: [{ job_id: 'p_0', input: { connector: 'jdbc', table_name: 't' } }] }; + describe('PipelineExecutionGuard', () => { afterEach(() => jest.clearAllMocks()); - it('allows the edit when the pipeline has no run history (empty array)', async () => { - const guard = buildGuard(jest.fn().mockResolvedValue([])); - await expect(guard.canActivate(contextWith())).resolves.toBe(true); + describe('CDC pipelines (explicit connector type)', () => { + it('allows the edit without consulting run history', async () => { + const { guard, proxy } = buildGuard({ pipeline: cdcPipeline }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + expect(proxy).toHaveBeenCalledTimes(1); + expect(proxy).toHaveBeenCalledWith('GET', '/pipeline/abc_123', {}); + }); + + it('is CDC when any job is a CDC job', async () => { + const mixed = { jobs: [...batchPipeline.jobs, ...cdcPipeline.jobs] }; + const { guard, proxy } = buildGuard({ pipeline: mixed }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + expect(proxy).toHaveBeenCalledTimes(1); + }); }); - it('allows the edit when the last run has no last_status', async () => { - const guard = buildGuard(jest.fn().mockResolvedValue([{}])); - await expect(guard.canActivate(contextWith())).resolves.toBe(true); + describe('batch pipelines', () => { + it('allows the edit when the pipeline never ran (empty run history)', async () => { + const { guard } = buildGuard({ pipeline: batchPipeline, runs: [] }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('allows the edit when the last run has no last_status', async () => { + const { guard } = buildGuard({ pipeline: batchPipeline, runs: [{}] }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('allows the edit when the pipeline is not running', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: [{ last_status: 'SUCCEEDED' }], + }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + }); + + it('blocks with the is-running message when the pipeline is running', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: [{ last_status: 'RUNNING' }], + }); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Pipeline is running, cannot update input now', + ); + }); + + it('treats a pipeline with no jobs as batch and checks its runs', async () => { + const { guard, proxy } = buildGuard({ pipeline: { jobs: [] }, runs: [] }); + await expect(guard.canActivate(contextWith())).resolves.toBe(true); + expect(proxy).toHaveBeenCalledTimes(2); + }); + + it('does not double-wrap the is-running BadRequestException', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: [{ last_status: 'running' }], + }); + await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf( + BadRequestException, + ); + await expect(guard.canActivate(contextWith())).rejects.not.toThrow( + /Error checking pipeline status/, + ); + }); }); - it('allows the edit when the pipeline is not running', async () => { - const guard = buildGuard( - jest.fn().mockResolvedValue([{ last_status: 'SUCCEEDED' }]), - ); - await expect(guard.canActivate(contextWith())).resolves.toBe(true); - }); + describe('failures', () => { + it('fails closed when the pipeline read fails', async () => { + const { guard } = buildGuard({ pipeline: new Error('platform down') }); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Error checking pipeline status: platform down', + ); + }); - it('blocks with the is-running message when the pipeline is running', async () => { - const guard = buildGuard( - jest.fn().mockResolvedValue([{ last_status: 'RUNNING' }]), - ); - await expect(guard.canActivate(contextWith())).rejects.toThrow( - 'Pipeline is running, cannot update input now', - ); - }); - - it('wraps a genuine status-check failure (fail closed)', async () => { - const guard = buildGuard( - jest.fn().mockRejectedValue(new Error('platform down')), - ); - await expect(guard.canActivate(contextWith())).rejects.toThrow( - 'Error checking pipeline status: platform down', - ); - }); - - it('does not double-wrap the is-running BadRequestException', async () => { - const guard = buildGuard( - jest.fn().mockResolvedValue([{ last_status: 'running' }]), - ); - await expect(guard.canActivate(contextWith())).rejects.toBeInstanceOf( - BadRequestException, - ); - await expect(guard.canActivate(contextWith())).rejects.not.toThrow( - /Error checking pipeline status/, - ); + it('fails closed when the run-history read fails', async () => { + const { guard } = buildGuard({ + pipeline: batchPipeline, + runs: new Error('runs down'), + }); + await expect(guard.canActivate(contextWith())).rejects.toThrow( + 'Error checking pipeline status: runs down', + ); + }); }); }); diff --git a/src/guards/pipeline-execution.guard.ts b/src/guards/pipeline-execution.guard.ts index 1b6ebcd..077d87d 100644 --- a/src/guards/pipeline-execution.guard.ts +++ b/src/guards/pipeline-execution.guard.ts @@ -16,6 +16,7 @@ import { import { PipelinesClientConfiguration } from 'src/modules/pipelinesV2/pipelines-client'; import { PlatformApiService } from 'src/modules/platform-api/platform-api.service'; import DadosferaLogger from '@dadosfera/dadosfera-logs'; +import { isCdcPipeline } from 'src/utils/cdc'; @Injectable() export class PipelineExecutionGuard implements CanActivate { @@ -39,7 +40,22 @@ export class PipelineExecutionGuard implements CanActivate { const user = request.user; const idRegex = /[^0-9a-zA-Z_$]+/g; const convertedId = pipelineId.replace(idRegex, '_'); - + + // Decide by the pipeline's explicit type (platform job + // `input.connector`), never by the absence of run history. + const pipeline = await this.platformApiService.proxy( + 'GET', + `/pipeline/${convertedId}`, + user, + ); + + if (isCdcPipeline(pipeline)) { + // CDC pipelines have no batch runs; the platform-api gates connector + // edits itself (require_pipeline_running on add/remove tables). + this.logger.info('PipelineExecutionGuard: CDC pipeline, no run to block on'); + return true; + } + const status = await this.platformApiService.proxy( 'GET', `/pipeline/${convertedId}/pipeline_run`, @@ -50,8 +66,7 @@ export class PipelineExecutionGuard implements CanActivate { this.logger.info('Pipeline current status response:' + JSON.stringify(currentStatus)); - // No run history (e.g. CDC pipelines never record batch runs) means - // nothing is executing — allow the edit rather than crash on .last_status. + // Batch pipeline that never ran yet: nothing can be executing. if (!currentStatus?.last_status) { return true; } @@ -59,9 +74,9 @@ export class PipelineExecutionGuard implements CanActivate { 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; } + + return true; } catch (error) { // Preserve the deliberate is-running rejection; only wrap genuine // status-check failures (fail closed on those for a destructive gate). @@ -71,6 +86,5 @@ export class PipelineExecutionGuard implements CanActivate { this.logger.error('Error in PipelineExecutionGuard: ' + error.message); throw new BadRequestException('Error checking pipeline status: ' + error.message); } - } } diff --git a/src/utils/cdc.ts b/src/utils/cdc.ts new file mode 100644 index 0000000..e02f232 --- /dev/null +++ b/src/utils/cdc.ts @@ -0,0 +1,34 @@ +/** + * Single source of truth for "is this CDC?" across maestro. + * + * Two shapes carry the discriminator: + * - a connection/input `plugin` (`mysql_cdc`, ...), used before a platform + * pipeline exists (create flows, DynamoDB inputs); + * - a platform job `input.connector === 'cdc'`, the explicit type the + * platform-api stamps on every CDC job. + * + * Keep the plugin list in sync with pi-factory (`CDC_PLUGINS`) and in-factory + * (`cdcPlugins`). + */ +export const CDC_PLUGINS: readonly string[] = ['mysql_cdc', 'postgresql_cdc', 'oracle_cdc']; + +export function isCdcPlugin(plugin?: string | null): boolean { + return plugin != null && CDC_PLUGINS.includes(plugin); +} + +export interface PlatformJobLike { + input?: { connector?: string } | null; +} + +export interface PlatformPipelineLike { + jobs?: PlatformJobLike[] | null; +} + +export function isCdcJob(job?: PlatformJobLike | null): boolean { + return job?.input?.connector === 'cdc'; +} + +/** A pipeline is CDC when any of its platform jobs is a CDC job. */ +export function isCdcPipeline(pipeline?: PlatformPipelineLike | null): boolean { + return (pipeline?.jobs ?? []).some((job) => isCdcJob(job)); +}