Adds branching (#216)

* Adds branching

* branches with step order

Author: elliotBraem

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

@@ -80,34 +80,68 @@ export class ProcessingService implements IBaseService {
       "Processing job started.",
     );
 
-    let currentContent: unknown = content;
     const executedSteps: SelectProcessingStep[] = [];
-
     try {
-      for (let i = 0; i < plan.length; i++) {
-        const planStep = plan[i];
-        const stepOrder = i + 1;
-
-        if (planStep.type === "transformation") {
-          const result = await this.executeTransform(
-            job.id,
-            currentContent,
-            planStep,
-            stepOrder,
-          );
-          executedSteps.push(result.step);
-          currentContent = result.output;
-        } else if (planStep.type === "distribution") {
-          const step = await this.executeDistribution(
-            job.id,
-            currentContent,
-            planStep,
-            stepOrder,
-          );
-          executedSteps.push(step);
+      let globalOutput: unknown = content;
+
+      const planWithOrder = plan.map((step, index) => ({
+        ...step,
+        order: index + 1,
+      }));
+
+      const globalStepsWithOrder: typeof planWithOrder = [];
+      const branchesWithOrder = new Map<string, typeof planWithOrder>();
+
+      for (const step of planWithOrder) {
+        if (step.branchId) {
+          if (!branchesWithOrder.has(step.branchId)) {
+            branchesWithOrder.set(step.branchId, []);
+          }
+          branchesWithOrder.get(step.branchId)!.push(step);
+        } else {
+          globalStepsWithOrder.push(step);
         }
       }
 
+      for (const planStep of globalStepsWithOrder) {
+        const result = await this.executeTransform(
+          job.id,
+          globalOutput,
+          planStep,
+          planStep.order,
+        );
+        executedSteps.push(result.step);
+        globalOutput = result.output;
+      }
+
+      const branchPromises = Array.from(branchesWithOrder.values()).map(
+        async (branch) => {
+          let branchContent = globalOutput;
+          for (const planStep of branch) {
+            if (planStep.type === "transformation") {
+              const result = await this.executeTransform(
+                job.id,
+                branchContent,
+                planStep,
+                planStep.order,
+              );
+              executedSteps.push(result.step);
+              branchContent = result.output;
+            } else if (planStep.type === "distribution") {
+              const step = await this.executeDistribution(
+                job.id,
+                branchContent,
+                planStep,
+                planStep.order,
+              );
+              executedSteps.push(step);
+            }
+          }
+        },
+      );
+
+      await Promise.all(branchPromises);
+
       const hasFailures = executedSteps.some((s) => s.status === "failed");
       const finalStatus = hasFailures ? "completed_with_errors" : "completed";
 
@@ -293,7 +327,17 @@ export class ProcessingService implements IBaseService {
       );
     }
 
-    const partialPlan = plan.slice(stepIndex);
+    const stepToRetryDetails = plan[stepIndex];
+    let partialPlan: ProcessingPlan;
+
+    if (stepToRetryDetails.branchId) {
+      partialPlan = plan
+        .slice(stepIndex)
+        .filter((step) => step.branchId === stepToRetryDetails.branchId);
+    } else {
+      // Fallback for old plans without branchId
+      partialPlan = plan.slice(stepIndex);
+    }
     const content = stepToRetry.input;
 
     return this.process(
@@ -323,20 +367,23 @@ export class ProcessingService implements IBaseService {
 
     // Add distributor transforms and distribution steps
     config.distribute?.forEach((d) => {
+      const branchId = randomUUID();
       d.transform?.forEach((t) => {
         plan.push({
           type: "transformation",
           stage: "distributor",
           plugin: t.plugin,
           config: t.config,
           distributorPlugin: d.plugin,
+          branchId,
         });
       });
       plan.push({
         type: "distribution",
         stage: "distributor",
         plugin: d.plugin,
         config: d.config,
+        branchId,
       });
     });
 
@@ -405,7 +452,25 @@ export class ProcessingService implements IBaseService {
 
     const plan = originalJob.plan as ProcessingPlan;
     const stepIndex = stepToTweak.stepOrder - 1;
-    const partialPlan = plan.slice(stepIndex);
+
+    if (stepIndex < 0 || stepIndex >= plan.length) {
+      throw new ProcessorError(
+        "invalid_step_index",
+        "Step to tweak is out of bounds of the processing plan.",
+      );
+    }
+
+    const stepToTweakDetails = plan[stepIndex];
+    let partialPlan: ProcessingPlan;
+
+    if (stepToTweakDetails.branchId) {
+      partialPlan = plan
+        .slice(stepIndex)
+        .filter((step) => step.branchId === stepToTweakDetails.branchId);
+    } else {
+      // Fallback for old plans without branchId
+      partialPlan = plan.slice(stepIndex);
+    }
 
     return this.process(
       parsedInput,

packages/shared-db/src/schema/processing.ts

@@ -57,6 +57,8 @@ export const ProcessingPlanStepSchema = z.object({
   config: z.record(z.unknown()).optional(),
   // For distributor transforms, this links back to the parent distributor plugin
   distributorPlugin: z.string().optional(),
+  // A unique identifier for a branch of distributor steps
+  branchId: z.string().optional(),
 });
 export type ProcessingPlanStep = z.infer<typeof ProcessingPlanStepSchema>;