feat: v2 - add worker mode (#6739)

**What**
- Adds support for starting a Medusa process with a worker mode.
- The worker modes supported are "shared", "worker", "server"
- In "worker" mode, API routes are not registered and modules that need to run workers (e.g., event bus redis) can use the flag to conditionally start workers.
- In "server" mode, API routes are registered and workers are not started.
- In "shared" mode, API routes are registered and workers are started. This is great for development.

packages/medusa/src/commands/start.js

@@ -19,39 +19,52 @@ export default async function ({ port, directory }) {
     const app = express()
 
     try {
-      const { dbConnection } = await loaders({ directory, expressApp: app })
-      const serverActivity = Logger.activity(`Creating server`)
-      const server = GracefulShutdownServer.create(
-        app.listen(port, (err) => {
-          if (err) {
-            return
-          }
-          Logger.success(serverActivity, `Server is ready on port: ${port}`)
-          track("CLI_START_COMPLETED")
-        })
-      )
+      const { dbConnection, configModule, container } = await loaders({
+        directory,
+        expressApp: app,
+      })
 
-      // Handle graceful shutdown
-      const gracefulShutDown = () => {
-        server
-          .shutdown()
-          .then(() => {
-            Logger.info("Gracefully stopping the server.")
-            process.exit(0)
-          })
-          .catch((e) => {
-            Logger.error("Error received when shutting down the server.", e)
-            process.exit(1)
+      const shouldStartServer =
+        configModule.projectConfig.worker_mode !== "worker"
+
+      let server
+      if (shouldStartServer) {
+        const serverActivity = Logger.activity(`Creating server`)
+        server = GracefulShutdownServer.create(
+          app.listen(port, (err) => {
+            if (err) {
+              return
+            }
+            Logger.success(serverActivity, `Server is ready on port: ${port}`)
+            track("CLI_START_COMPLETED")
           })
+        )
+
+        // Handle graceful shutdown
+        const gracefulShutDown = () => {
+          server
+            .shutdown()
+            .then(() => {
+              Logger.info("Gracefully stopping the server.")
+              process.exit(0)
+            })
+            .catch((e) => {
+              Logger.error("Error received when shutting down the server.", e)
+              process.exit(1)
+            })
+        }
+
+        process.on("SIGTERM", gracefulShutDown)
+        process.on("SIGINT", gracefulShutDown)
+      } else {
+        Logger.info("Running in worker mode, server will not be started.")
       }
-      process.on("SIGTERM", gracefulShutDown)
-      process.on("SIGINT", gracefulShutDown)
 
       scheduleJob(CRON_SCHEDULE, () => {
         track("PING")
       })
 
-      return { dbConnection, server }
+      return shouldStartServer ? { dbConnection, server } : { dbConnection }
     } catch (err) {
       Logger.error("Error starting server", err)
       process.exit(1)

packages/medusa/src/loaders/config.ts

@@ -1,4 +1,4 @@
-import { getConfigFile } from "medusa-core-utils"
+import { getConfigFile, isDefined } from "medusa-core-utils"
 import { ConfigModule } from "../types/global"
 import logger from "./logger"
 
@@ -58,11 +58,24 @@ export default (rootDirectory: string): ConfigModule => {
     )
   }
 
+  let worker_mode = configModule?.projectConfig?.worker_mode
+  if (!isDefined(worker_mode)) {
+    const env = process.env.MEDUSA_WORKER_MODE
+    if (isDefined(env)) {
+      if (env === "shared" || env === "worker" || env === "server") {
+        worker_mode = env
+      }
+    } else {
+      worker_mode = "shared"
+    }
+  }
+
   return {
     projectConfig: {
       jwt_secret: jwt_secret ?? "supersecret",
       cookie_secret: cookie_secret ?? "supersecret",
       ...configModule?.projectConfig,
+      worker_mode,
     },
     modules: configModule.modules ?? {},
     featureFlags: configModule?.featureFlags ?? {},

packages/medusa/src/loaders/index.ts

@@ -1,9 +1,9 @@
 import { createDefaultsWorkflow } from "@medusajs/core-flows"
 import {
   InternalModuleDeclaration,
-  ModulesDefinition
+  ModulesDefinition,
 } from "@medusajs/modules-sdk"
-import { MODULE_RESOURCE_TYPE } from "@medusajs/types"
+import { ConfigModule, MODULE_RESOURCE_TYPE } from "@medusajs/types"
 import {
   ContainerRegistrationKeys,
   MedusaV2Flag,
@@ -91,14 +91,7 @@ async function loadMedusaV2({
 }) {
   const container = createMedusaContainer()
 
-  // Add additional information to context of request
-  expressApp.use((req: Request, res: Response, next: NextFunction) => {
-    const ipAddress = requestIp.getClientIp(req) as string
-    ;(req as any).request_context = {
-      ip_address: ipAddress,
-    }
-    next()
-  })
+  const shouldStartAPI = configModule.projectConfig.worker_mode !== "worker"
 
   const pgConnection = await pgConnectionLoader({ container, configModule })
 
@@ -114,22 +107,33 @@ async function loadMedusaV2({
     container,
   })
 
-  await expressLoader({ app: expressApp, configModule })
+  if (shouldStartAPI) {
+    await expressLoader({ app: expressApp, configModule })
 
-  expressApp.use((req: Request, res: Response, next: NextFunction) => {
-    req.scope = container.createScope() as MedusaContainer
-    req.requestId = (req.headers["x-request-id"] as string) ?? v4()
-    next()
-  })
+    expressApp.use((req: Request, res: Response, next: NextFunction) => {
+      req.scope = container.createScope() as MedusaContainer
+      req.requestId = (req.headers["x-request-id"] as string) ?? v4()
+      next()
+    })
 
-  // TODO: Add Subscribers loader
+    // Add additional information to context of request
+    expressApp.use((req: Request, res: Response, next: NextFunction) => {
+      const ipAddress = requestIp.getClientIp(req) as string
+      ;(req as any).request_context = {
+        ip_address: ipAddress,
+      }
+      next()
+    })
 
-  await apiLoader({
-    container,
-    app: expressApp,
-    configModule,
-    featureFlagRouter,
-  })
+    // TODO: Add Subscribers loader
+
+    await apiLoader({
+      container,
+      app: expressApp,
+      configModule,
+      featureFlagRouter,
+    })
+  }
 
   await medusaProjectApisLoader({
     rootDirectory,
@@ -142,6 +146,7 @@ async function loadMedusaV2({
   await createDefaultsWorkflow(container).run()
 
   return {
+    configModule,
     container,
     app: expressApp,
     pgConnection,
@@ -153,6 +158,7 @@ export default async ({
   expressApp,
   isTest,
 }: Options): Promise<{
+  configModule: ConfigModule
   container: MedusaContainer
   dbConnection?: Connection
   app: Express
@@ -319,6 +325,7 @@ export default async ({
   track("SEARCH_ENGINE_INDEXING_COMPLETED", { duration: searchAct.duration })
 
   return {
+    configModule,
     container,
     dbConnection,
     app: expressApp,

packages/medusa/src/loaders/load-medusa-project-apis.ts

@@ -41,9 +41,19 @@ export default async ({
 }: Options): Promise<void> => {
   const resolved = getResolvedPlugins(rootDirectory, configModule) || []
 
+  const shouldStartAPI = configModule.projectConfig.worker_mode !== "worker"
+
   await promiseAll(
     resolved.map(async (pluginDetails) => {
-      await registerApi(pluginDetails, app, container, configModule, activityId)
+      if (shouldStartAPI) {
+        await registerApi(
+          pluginDetails,
+          app,
+          container,
+          configModule,
+          activityId
+        )
+      }
       await registerSubscribers(pluginDetails, container, activityId)
       await registerWorkflows(pluginDetails)
     })

packages/medusa/src/loaders/medusa-app.ts

@@ -168,6 +168,7 @@ export const loadMedusaApp = async (
   }
 
   const medusaApp = await MedusaApp({
+    workerMode: configModule.projectConfig.worker_mode,
     modulesConfig: configModules,
     servicesConfig: joinerConfig,
     remoteFetchData: remoteQueryFetchData(container),