refactor: Remove dedupe strategies, use only dedupe list

packages/backend/src/helpers/global-variable.ts

@@ -10,6 +10,7 @@ import {
   ITriggerItem,
   IActionItem,
 } from '@automatisch/types';
+import EarlyExitError from '../errors/early-exit';
 
 type GlobalVariableOptions = {
   connection?: Connection;
@@ -25,9 +26,7 @@ const globalVariable = async (
 ): Promise<IGlobalVariable> => {
   const { connection, app, flow, step, execution, testRun = false } = options;
 
-  const lastInternalId = await flow?.lastInternalId();
-
-  const trigger = await step?.getTriggerCommand();
+  const lastInternalId = testRun ? undefined : await flow?.lastInternalId();
   const nextStep = await step?.getNextStep();
 
   const $: IGlobalVariable = {
@@ -75,6 +74,13 @@ const globalVariable = async (
     },
     pushTriggerItem: (triggerItem: ITriggerItem) => {
       $.triggerOutput.data.push(triggerItem);
+
+      if (
+        $.execution.testRun ||
+        isAlreadyProcessed(triggerItem.meta.internalId)
+      ) {
+        throw new EarlyExitError();
+      }
     },
     setActionItem: (actionItem: IActionItem) => {
       $.actionOutput.data = actionItem;
@@ -87,27 +93,12 @@ const globalVariable = async (
     beforeRequest: app.beforeRequest,
   });
 
-  if (trigger) {
-    if (trigger.dedupeStrategy === 'unique') {
-      const lastInternalIds = testRun ? [] : await flow?.lastInternalIds();
+  const lastInternalIds =
+    testRun || (flow && step.isAction) ? [] : await flow?.lastInternalIds();
 
-      const isAlreadyProcessed = (internalId: string) => {
-        if (testRun) return false;
-
-        return lastInternalIds?.includes(internalId);
-      };
-
-      $.flow.isAlreadyProcessed = isAlreadyProcessed;
-    } else if (trigger.dedupeStrategy === 'greatest') {
-      const isAlreadyProcessed = (internalId: string) => {
-        if (testRun) return false;
-
-        return Number(internalId) <= Number($.flow.lastInternalId);
-      };
-
-      $.flow.isAlreadyProcessed = isAlreadyProcessed;
-    }
-  }
+  const isAlreadyProcessed = (internalId: string) => {
+    return lastInternalIds?.includes(internalId);
+  };
 
   return $;
 };

packages/backend/src/services/flow.ts

@@ -1,5 +1,6 @@
 import Flow from '../models/flow';
 import globalVariable from '../helpers/global-variable';
+import EarlyExitError from '../errors/early-exit';
 
 type ProcessFlowOptions = {
   flowId: string;
@@ -23,13 +24,15 @@ export const processFlow = async (options: ProcessFlowOptions) => {
   try {
     await triggerCommand.run($);
   } catch (error) {
-    if (error?.response?.httpError) {
-      $.triggerOutput.error = error.response.httpError;
-    } else {
-      try {
-        $.triggerOutput.error = JSON.parse(error.message);
-      } catch {
-        $.triggerOutput.error = { error: error.message };
+    if (error instanceof EarlyExitError === false) {
+      if (error?.response?.httpError) {
+        $.triggerOutput.error = error.response.httpError;
+      } else {
+        try {
+          $.triggerOutput.error = JSON.parse(error.message);
+        } catch {
+          $.triggerOutput.error = { error: error.message };
+        }
       }
     }
   }