"Wait/Pause" component for workflows

App/FeatureSet/Workflow/Services/QueueWorkflow.ts

@@ -56,6 +56,7 @@ export default class QueueWorkflow {
   public static async addWorkflowToQueue(
     executeWorkflow: ExecuteWorkflowType,
     scheduleAt?: string,
+    delay?: number,
   ): Promise<void> {
     const workflowId: ObjectID = executeWorkflow.workflowId;
 
@@ -153,7 +154,7 @@ export default class QueueWorkflow {
 
     // Add Workflow Run Log.
     let workflowLog: WorkflowLog | null = null;
-    if (!scheduleAt) {
+    if (!scheduleAt && !delay) {
       // if the workflow is to be run immediately.
       const runLog: WorkflowLog = new WorkflowLog();
       runLog.workflowId = workflowId;
@@ -186,6 +187,7 @@ export default class QueueWorkflow {
       },
       {
         scheduleAt: scheduleAt,
+        delay: delay,
         repeatableKey: workflow.repeatableJobKey || undefined,
       },
     );

App/FeatureSet/Workflow/Services/RunWorkflow.ts

@@ -20,6 +20,7 @@ import WorkflowService from "Common/Server/Services/WorkflowService";
 import WorkflowVariableService from "Common/Server/Services/WorkflowVariableService";
 import QueryHelper from "Common/Server/Types/Database/QueryHelper";
 import ComponentCode, {
+  Interactive,
   RunReturnType,
 } from "Common/Server/Types/Workflow/ComponentCode";
 import Components from "Common/Server/Types/Workflow/Components/Index";
@@ -29,6 +30,8 @@ import VMAPI from "Common/Server/Utils/VM/VMAPI";
 import Workflow from "Common/Models/DatabaseModels/Workflow";
 import WorkflowLog from "Common/Models/DatabaseModels/WorkflowLog";
 import WorkflowVariable from "Common/Models/DatabaseModels/WorkflowVariable";
+import { ExecuteWorkflowType } from "Common/Server/Types/Workflow/TriggerCode";
+import QueueWorkflow from "./QueueWorkflow";
 
 const AllComponents: Dictionary<ComponentMetadata> = loadAllComponentMetadata();
 
@@ -83,6 +86,7 @@ export default class RunWorkflow {
         select: {
           graph: true,
           projectId: true,
+          interactiveData: true,
         },
         props: {
           isRoot: true,
@@ -97,6 +101,10 @@ export default class RunWorkflow {
         throw new BadDataException("Workflow graph not found");
       }
 
+      const interactive: Interactive | undefined =
+        workflow.interactiveData &&
+        (workflow.interactiveData as unknown as Interactive);
+
       this.projectId = workflow.projectId || null;
 
       if (!runProps.workflowLogId) {
@@ -146,7 +154,9 @@ export default class RunWorkflow {
       variables = getVariableResult.variables;
 
       // start execute different components.
-      let executeComponentId: string = runStack.startWithComponentId;
+      let executeComponentId: string =
+        interactive?.componentId ?? runStack.startWithComponentId;
+      // if there is an interactive node set for this workflow, start from that node, all the previous nodes are already executed
 
       const fifoStackOfComponentsPendingExecution: Array<string> = [
         executeComponentId,
@@ -230,6 +240,80 @@ export default class RunWorkflow {
           returnValues: result.returnValues,
         };
 
+        const interactiveRes: Interactive | undefined = result.interactive;
+        if (interactiveRes?.waiting) {
+          this.log(
+            "Interactive component " +
+              executeComponentId +
+              " is in the waiting  state, delaying the workflow execution",
+          );
+
+          let delay: number | undefined =
+            new Date(interactiveRes.nextStateCheck).getTime() -
+            new Date().getTime();
+          if (delay < 0) {
+            delay = undefined;
+          }
+
+          this.log(result.interactive as any);
+          this.log(result.interactive?.lastTimeChecked);
+          this.log(new Date(result.interactive!.lastTimeChecked!).toString());
+          this.log(
+            "Adding the workflow to the queue with the delay " + delay + " ms",
+          );
+
+          // update workflow log.
+          await WorkflowLogService.updateOneById({
+            id: runProps.workflowLogId,
+            data: {
+              workflowStatus: WorkflowStatus.Waiting,
+              logs: this.logs.join("\n"),
+              completedAt: OneUptimeDate.getCurrentDate(),
+            },
+            props: {
+              isRoot: true,
+            },
+          });
+
+          await WorkflowService.updateOneById({
+            data: {
+              interactiveData: interactiveRes as any, // TS-2589
+            },
+            id: runProps.workflowId,
+            props: {
+              isRoot: true,
+            },
+          });
+          const executeWorkflow: ExecuteWorkflowType = {
+            workflowId: workflow.id!,
+            returnValues: {},
+          };
+
+          await QueueWorkflow.addWorkflowToQueue(
+            executeWorkflow,
+            undefined,
+            delay,
+          );
+          this.cleanLogs(variables);
+
+          return;
+        }
+
+        if (interactive?.waiting && interactiveRes && !interactiveRes.waiting) {
+          // The value from the database still has the interactive value in it, need to update it
+          this.log(
+            "Updating the workflow database record, removing the interactive data",
+          );
+          await WorkflowService.updateOneById({
+            data: {
+              interactiveData: interactiveRes as any, //TS-2589
+            },
+            id: runProps.workflowId,
+            props: {
+              isRoot: true,
+            },
+          });
+        }
         const portToBeExecuted: Port | undefined = result.executePort;
 
         if (!portToBeExecuted) {
@@ -397,6 +481,7 @@ export default class RunWorkflow {
         workflowId: this.workflowId!,
         workflowLogId: this.workflowLogId!,
         projectId: this.projectId!,
+        nodeId: node.id,
         onError: (exception: Exception) => {
           this.log(exception);
           onError();

Common/Models/DatabaseModels/Workflow.ts

@@ -521,4 +521,21 @@ export default class Workflow extends BaseModel {
     nullable: true,
   })
   public repeatableJobKey?: string = undefined;
+
+  // Interactive node, all kind of nodes that stop the workflow and schedule it again at some future time
+  @ColumnAccessControl({
+    create: [],
+    read: [],
+    update: [],
+  })
+  @TableColumn({
+    isDefaultValueColumn: false,
+    required: false,
+    type: TableColumnType.JSON,
+  })
+  @Column({
+    type: ColumnType.JSON,
+    nullable: true,
+  })
+  public interactiveData?: JSONObject = undefined;
 }

Common/Server/Infrastructure/Postgres/SchemaMigrations/1725811636668-MigrationName.ts

@@ -0,0 +1,16 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class MigrationName1725811636668 implements MigrationInterface {
+  public name = "MigrationName1725811636668";
+  public async down(queryRunner: QueryRunner): Promise<any> {
+    await queryRunner.query(
+      `ALTER TABLE "Workflow" DROP COLUMN "interactiveData"`,
+    );
+  }
+
+  public async up(queryRunner: QueryRunner): Promise<any> {
+    await queryRunner.query(
+      `ALTER TABLE "Workflow" ADD "interactiveData" jsonb`,
+    );
+  }
+}

Common/Server/Infrastructure/Postgres/SchemaMigrations/Index.ts

@@ -51,6 +51,7 @@ import { MigrationName1725379949648 } from "./1725379949648-MigrationName";
 import { MigrationName1725551629492 } from "./1725551629492-MigrationName";
 import { MigrationName1725556630384 } from "./1725556630384-MigrationName";
 import { MigrationName1725618842598 } from "./1725618842598-MigrationName";
+import { MigrationName1725811636668 } from "./1725811636668-MigrationName";
 
 export default [
   InitialMigration,
@@ -106,4 +107,5 @@ export default [
   MigrationName1725551629492,
   MigrationName1725556630384,
   MigrationName1725618842598,
+  MigrationName1725811636668,
 ];

Common/Server/Infrastructure/Queue.ts

@@ -94,13 +94,15 @@ export default class Queue {
     options?: {
       scheduleAt?: string | undefined;
       repeatableKey?: string | undefined;
+      delay?: number | undefined; // either delay or scheduleAt
     },
   ): Promise<Job> {
     const optionsObject: JobsOptions = {
       jobId: jobId.toString(),
     };
-
-    if (options && options.scheduleAt) {
+    if (options?.delay) {
+      optionsObject.delay = options.delay;
+    } else if (options?.scheduleAt) {
       optionsObject.repeat = {
         pattern: options.scheduleAt,
       };

Common/Server/Types/Workflow/ComponentCode.ts

@@ -11,12 +11,26 @@ export interface RunOptions {
   workflowLogId: ObjectID;
   workflowId: ObjectID;
   projectId: ObjectID;
+  nodeId: string;
   onError: (exception: Exception) => Exception;
 }
 
+export interface Interactive {
+  // If the component returns Interactive property
+  // and waiting is true
+  // the workflow stops the execution, saves the current state
+  // and schedules the new workflow, starting from the id of the node
+  waiting: boolean;
+  componentId: string; // where is waiting in the workflow
+  startedWaiting: Date;
+  lastTimeChecked?: Date;
+  nextStateCheck: Date; // component is responsible to define when the next run date happens
+}
+
 export interface RunReturnType {
   returnValues: JSONObject;
   executePort?: Port | undefined;
+  interactive?: Interactive | undefined;
 }
 
 export default class ComponentCode {

Common/Server/Types/Workflow/Components/Index.ts

@@ -33,6 +33,7 @@ import Dictionary from "Common/Types/Dictionary";
 import Text from "Common/Types/Text";
 import ComponentID from "Common/Types/Workflow/ComponentID";
 import ApiPatch from "./API/Patch";
+import Delay from "./Interactive/Delay";
 
 const Components: Dictionary<ComponentCode> = {
   [ComponentID.Webhook]: new WebhookTrigger(),
@@ -53,6 +54,7 @@ const Components: Dictionary<ComponentCode> = {
   [ComponentID.ApiPut]: new ApiPut(),
   [ComponentID.SendEmail]: new Email(),
   [ComponentID.IfElse]: new IfElse(),
+  [ComponentID.Delay]: new Delay(),
 };
 
 for (const baseModelService of Services) {

Common/Server/Types/Workflow/Components/Interactive/Delay.ts

@@ -0,0 +1,112 @@
+import ComponentCode, {
+  Interactive,
+  RunOptions,
+  RunReturnType,
+} from "../../ComponentCode";
+import ComponentMetadata, {
+  Port,
+} from "../../../../../Types/Workflow/Component";
+import InteractiveComponents from "../../../../../Types/Workflow/Components/Interactive";
+import ComponentID from "../../../../../Types/Workflow/ComponentID";
+import BadDataException from "../../../../../Types/Exception/BadDataException";
+import { JSONObject } from "../../../../../Types/JSON";
+import Workflow from "../../../../../Models/DatabaseModels/Workflow";
+import ObjectID from "../../../../../Types/ObjectID";
+import WorkflowService from "../../../../Services/WorkflowService";
+
+export default class Delay extends ComponentCode {
+  public constructor() {
+    super();
+
+    const DelayComponent: ComponentMetadata | undefined =
+      InteractiveComponents.find((i: ComponentMetadata) => {
+        return i.id === ComponentID.Delay;
+      });
+
+    if (!DelayComponent) {
+      throw new BadDataException("Component not found.");
+    }
+
+    this.setMetadata(DelayComponent);
+  }
+
+  public override async run(
+    args: JSONObject,
+    options: RunOptions,
+  ): Promise<RunReturnType> {
+    const outPort: Port | undefined = this.getMetadata().outPorts.find(
+      (p: Port) => {
+        return p.id === "out";
+      },
+    );
+
+    if (!outPort) {
+      throw options.onError(new BadDataException("Out port not found"));
+    }
+
+    const _delay: string = args["delay"] as string;
+    const delay: number = parseInt(_delay, 10);
+    const workflowId: ObjectID = options.workflowId;
+    const workflow: Workflow | null = await WorkflowService.findOneById({
+      id: workflowId,
+      select: {
+        graph: true,
+        projectId: true,
+        interactiveData: true,
+      },
+      props: {
+        isRoot: true,
+      },
+    });
+    if (!workflow) {
+      throw options.onError(new BadDataException("Workflow not found"));
+    }
+    let interactive: Interactive | undefined = workflow.interactiveData
+      ? (workflow.interactiveData as unknown as Interactive)
+      : undefined; // TODO this is not 100% correct, the Date attributes are still strings, find how to convert
+    if (interactive?.waiting) {
+      options.log(
+        "Interactive data present, workflow already in the waiting state",
+      );
+      options.log(workflow.interactiveData);
+      // already delayed, check is it the same node first
+      const interactive: Interactive =
+        workflow.interactiveData as unknown as Interactive;
+      if (interactive.componentId !== options.nodeId) {
+        throw options.onError(
+          new BadDataException(
+            "Waiting in different interactive component, aborting",
+          ),
+        );
+      }
+      if (
+        new Date(interactive.startedWaiting).getTime() + delay <
+        new Date().getTime()
+      ) {
+        interactive.waiting = false;
+        options.log("Delay expired, continuing to the next node");
+      } else {
+        options.log("Delay not yet expired, continuing with the wait");
+      }
+      interactive.lastTimeChecked = new Date();
+      return Promise.resolve({
+        returnValues: {},
+        executePort: outPort,
+        interactive,
+      });
+    }
+    interactive = {
+      componentId: options.nodeId,
+      lastTimeChecked: new Date(),
+      startedWaiting: new Date(),
+      nextStateCheck: new Date(new Date().getTime() + delay),
+      waiting: true,
+    };
+    options.log("Initiated the waiting node");
+    return Promise.resolve({
+      returnValues: {},
+      executePort: outPort,
+      interactive,
+    });
+  }
+}

Common/Types/Workflow/ComponentID.ts

@@ -16,6 +16,7 @@ enum ComponentID {
   ApiPatch = "api-patch",
   SendEmail = "send-email",
   IfElse = "if-else",
+  Delay = "delay",
 }
 
 export default ComponentID;