fix processing flows

Author: elliotBraem

apps/app/src/lib/api/processing.ts

@@ -16,20 +16,6 @@ export function useProcessingJobs(
   feedId: string | null,
   options?: { enabled?: boolean },
 ) {
-  const query = useApiQuery<ProcessingJobsResponse, Error, ProcessingJob[]>(
-    ["processingJobs", submissionId, feedId],
-    `/processing/jobs/${submissionId}?feedId=${feedId}`,
-    {
-      enabled:
-        options?.enabled !== undefined
-          ? options.enabled && !!submissionId && !!feedId
-          : !!submissionId && !!feedId,
-      select: (data) => data.data.jobs,
-    },
-  );
-
-  const isProcessing = query.data?.some((job) => job.status === "processing");
-
   return useApiQuery<ProcessingJobsResponse, Error, ProcessingJob[]>(
     ["processingJobs", submissionId, feedId],
     `/processing/jobs/${submissionId}?feedId=${feedId}`,
@@ -39,7 +25,12 @@ export function useProcessingJobs(
           ? options.enabled && !!submissionId && !!feedId
           : !!submissionId && !!feedId,
       select: (data) => data.data.jobs,
-      refetchInterval: isProcessing ? 5000 : false,
+      refetchInterval: (query) =>
+        query.state.data?.data.jobs.some(
+          (job: ProcessingJob) => job.status === "processing",
+        )
+          ? 5000
+          : false,
     },
   );
 }
@@ -51,26 +42,19 @@ export function useProcessingSteps(
   jobId: string | null,
   options?: { enabled?: boolean },
 ) {
-  const query = useApiQuery<ProcessingStepsResponse, Error, ProcessingStep[]>(
-    ["processingSteps", jobId],
-    `/processing/jobs/${jobId}/steps`,
-    {
-      enabled:
-        options?.enabled !== undefined ? options.enabled && !!jobId : !!jobId,
-      select: (data) => data.data.steps,
-    },
-  );
-
-  const isProcessing = query.data?.some((step) => step.status === "processing");
-
   return useApiQuery<ProcessingStepsResponse, Error, ProcessingStep[]>(
     ["processingSteps", jobId],
     `/processing/jobs/${jobId}/steps`,
     {
       enabled:
         options?.enabled !== undefined ? options.enabled && !!jobId : !!jobId,
       select: (data) => data.data.steps,
-      refetchInterval: isProcessing ? 5000 : false,
+      refetchInterval: (query) =>
+        query.state.data?.data.steps.some(
+          (step: ProcessingStep) => step.status === "processing",
+        )
+          ? 5000
+          : false,
     },
   );
 }

packages/core-services/src/services/processing.service.ts

@@ -104,18 +104,29 @@ export class ProcessingService implements IBaseService {
       }
 
       for (const planStep of globalStepsWithOrder) {
-        const result = await this.executeTransform(
-          job.id,
-          globalOutput,
-          planStep,
-          planStep.order,
-        );
-        executedSteps.push(result.step);
-        globalOutput = result.output;
+        if (planStep.type === "transformation") {
+          const result = await this.executeTransform(
+            job.id,
+            globalOutput,
+            planStep,
+            planStep.order,
+          );
+          executedSteps.push(result.step);
+          globalOutput = result.output;
+        } else if (planStep.type === "distribution") {
+          const step = await this.executeDistribution(
+            job.id,
+            globalOutput,
+            planStep,
+            planStep.order,
+          );
+          executedSteps.push(step);
+        }
       }
 
       const branchPromises = Array.from(branchesWithOrder.values()).map(
         async (branch) => {
+          const branchExecutedSteps: SelectProcessingStep[] = [];
           let branchContent = globalOutput;
           for (const planStep of branch) {
             if (planStep.type === "transformation") {
@@ -125,7 +136,7 @@ export class ProcessingService implements IBaseService {
                 planStep,
                 planStep.order,
               );
-              executedSteps.push(result.step);
+              branchExecutedSteps.push(result.step);
               branchContent = result.output;
             } else if (planStep.type === "distribution") {
               const step = await this.executeDistribution(
@@ -134,13 +145,17 @@ export class ProcessingService implements IBaseService {
                 planStep,
                 planStep.order,
               );
-              executedSteps.push(step);
+              branchExecutedSteps.push(step);
             }
           }
+          return branchExecutedSteps;
         },
       );
 
-      await Promise.all(branchPromises);
+      const branchesExecutedSteps = await Promise.all(branchPromises);
+      branchesExecutedSteps.forEach((branchSteps) => {
+        executedSteps.push(...branchSteps);
+      });
 
       const hasFailures = executedSteps.some((s) => s.status === "failed");
       const finalStatus = hasFailures ? "completed_with_errors" : "completed";