fix: Resume workflow execution (#7103)

Supercedes #7051 – if OK, I'll move the base of this PR to `develop` and we can run reviews only of this one.

**What**
- Gracefully close BullMQ Worker in Redis Event Bus and Redis Workflow Engine (by @edast, @sradevski)
- Register workflows before MedusaApp is loaded*
- Introduce `onApplicationPrepareShutdown`**
- Refactor plugin resolving for reusability purposes

*We now register workflows before modules are loaded to ensure modules can run workflows as part of bootstrapping. E.g. the Redis Workflow Engine resumes workflows when it starts, which has until this change failed, because the workflows were not registered yet.

**We introduce a new hook to prepare resources for an application shutdown. E.g. closing the BullMQ worker as a preparatory step to closing the BullMQ queue. The worker will continue to process jobs while the queue is still open to receive new jobs (without processing them). 

Co-authored-by: Stevche Radevski <4820812+sradevski@users.noreply.github.com>
Co-authored-by: Darius <618221+edast@users.noreply.github.com>
Co-authored-by: Carlos R. L. Rodrigues <37986729+carlos-r-l-rodrigues@users.noreply.github.com>

packages/workflow-engine-redis/src/services/workflow-orchestrator.ts

@@ -4,8 +4,13 @@ import {
   TransactionHandlerType,
   TransactionStep,
 } from "@medusajs/orchestration"
-import { ContainerLike, Context, MedusaContainer } from "@medusajs/types"
-import { InjectSharedContext, isString, MedusaContext } from "@medusajs/utils"
+import {
+  ContainerLike,
+  Context,
+  Logger,
+  MedusaContainer,
+} from "@medusajs/types"
+import { InjectSharedContext, MedusaContext, isString } from "@medusajs/utils"
 import {
   FlowRunOptions,
   MedusaWorkflow,
@@ -75,6 +80,8 @@ export class WorkflowOrchestratorService {
   protected redisPublisher: Redis
   protected redisSubscriber: Redis
   private subscribers: Subscribers = new Map()
+  private activeStepsCount: number = 0
+  private logger: Logger
 
   protected redisDistributedTransactionStorage_: RedisDistributedTransactionStorage
 
@@ -83,15 +90,18 @@ export class WorkflowOrchestratorService {
     redisDistributedTransactionStorage,
     redisPublisher,
     redisSubscriber,
+    logger,
   }: {
     dataLoaderOnly: boolean
     redisDistributedTransactionStorage: RedisDistributedTransactionStorage
     workflowOrchestratorService: WorkflowOrchestratorService
     redisPublisher: Redis
     redisSubscriber: Redis
+    logger: Logger
   }) {
     this.redisPublisher = redisPublisher
     this.redisSubscriber = redisSubscriber
+    this.logger = logger
 
     redisDistributedTransactionStorage.setWorkflowOrchestratorService(this)
 
@@ -113,6 +123,15 @@ export class WorkflowOrchestratorService {
     await this.redisDistributedTransactionStorage_.onApplicationShutdown()
   }
 
+  async onApplicationPrepareShutdown() {
+    // eslint-disable-next-line max-len
+    await this.redisDistributedTransactionStorage_.onApplicationPrepareShutdown()
+
+    while (this.activeStepsCount > 0) {
+      await new Promise((resolve) => setTimeout(resolve, 1000))
+    }
+  }
+
   @InjectSharedContext()
   async run<T = unknown>(
     workflowIdOrWorkflow: string | ReturnWorkflow<any, any, any>,
@@ -523,6 +542,7 @@ export class WorkflowOrchestratorService {
 
       onStepBegin: async ({ step, transaction }) => {
         customEventHandlers?.onStepBegin?.({ step, transaction })
+        this.activeStepsCount++
 
         await notify({ eventType: "onStepBegin", step })
       },
@@ -533,8 +553,9 @@ export class WorkflowOrchestratorService {
           transaction
         )
         customEventHandlers?.onStepSuccess?.({ step, transaction, response })
-
         await notify({ eventType: "onStepSuccess", step, response })
+
+        this.activeStepsCount--
       },
       onStepFailure: async ({ step, transaction }) => {
         const stepName = step.definition.action!
@@ -543,14 +564,28 @@ export class WorkflowOrchestratorService {
           .filter((err) => err.action === stepName)
 
         customEventHandlers?.onStepFailure?.({ step, transaction, errors })
-
         await notify({ eventType: "onStepFailure", step, errors })
+
+        this.activeStepsCount--
+      },
+      onStepAwaiting: async ({ step, transaction }) => {
+        customEventHandlers?.onStepAwaiting?.({ step, transaction })
+
+        await notify({ eventType: "onStepAwaiting", step })
+
+        if (!step.definition.backgroundExecution) {
+          this.activeStepsCount--
+        }
       },
 
       onCompensateStepSuccess: async ({ step, transaction }) => {
         const stepName = step.definition.action!
         const response = transaction.getContext().compensate[stepName]
-        customEventHandlers?.onStepSuccess?.({ step, transaction, response })
+        customEventHandlers?.onCompensateStepSuccess?.({
+          step,
+          transaction,
+          response,
+        })
 
         await notify({ eventType: "onCompensateStepSuccess", step, response })
       },

packages/workflow-engine-redis/src/services/workflows-module.ts

@@ -59,6 +59,9 @@ export class WorkflowsModuleService implements IWorkflowEngineService {
       await this.workflowOrchestratorService_.onApplicationShutdown()
       await this.redisDisconnectHandler_()
     },
+    onApplicationPrepareShutdown: async () => {
+      await this.workflowOrchestratorService_.onApplicationPrepareShutdown()
+    },
   }
 
   @InjectManager("baseRepository_")

packages/workflow-engine-redis/src/utils/workflow-orchestrator-storage.ts

@@ -64,9 +64,13 @@ export class RedisDistributedTransactionStorage extends DistributedTransactionSt
     )
   }
 
+  async onApplicationPrepareShutdown() {
+    // Close worker gracefully, i.e. wait for the current jobs to finish
+    await this.worker.close()
+  }
+
   async onApplicationShutdown() {
     await this.queue.close()
-    await this.worker.close(true)
   }
 
   setWorkflowOrchestratorService(workflowOrchestratorService) {