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 <contact@withwoz.com>

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<typeof PipelineExecutionGuard>[0],
+    { proxy } as unknown as ConstructorParameters<typeof PipelineExecutionGuard>[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<PipelineExecutionGuard['canActivate']>[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',
+      );
+    });
   });
 });

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);
     }
-
   }
 }

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