feat(medusa): Improve add line item to cart perf and transaction management + clustering start command (#5701)

.changeset/tasty-ants-visit.md

@@ -0,0 +1,7 @@
+---
+"@medusajs/medusa": patch
+"@medusajs/medusa-cli": patch
+"@medusajs/admin-ui": patch
+---
+
+feat(medusa, medusa-cli): Improve add line item + cluster starting with medusa cli

integration-tests/helpers/draft-order-seeder.js

@@ -145,27 +145,23 @@ module.exports = async (dataSource, data = {}) => {
     name: "Test Region",
     currency_code: "usd",
     tax_rate: 0,
-    payment_providers: [
-      {
-        id: "test-pay",
-        is_installed: true,
-      },
-    ],
   })
 
+  await manager.query(
+    `insert into region_payment_providers values ('test-region', 'test-pay');`
+  )
+
   await manager.insert(Region, {
     id: "test-region-2",
     name: "Test Region 2",
     currency_code: "eur",
     tax_rate: 0,
-    payment_providers: [
-      {
-        id: "test-pay",
-        is_installed: true,
-      },
-    ],
   })
 
+  await manager.query(
+    `insert into region_payment_providers values ('test-region-2', 'test-pay');`
+  )
+
   await manager.insert(DiscountRule, {
     id: "discount_rule_id",
     description: "test description",

integration-tests/plugins/__tests__/inventory/cart/cart.js

@@ -17,7 +17,7 @@ const {
   getContainer,
 } = require("../../../../environment-helpers/use-container")
 
-jest.setTimeout(30000)
+jest.setTimeout(60000)
 
 const adminHeaders = { headers: { "x-medusa-access-token": "test_token" } }
 

packages/admin-ui/src/node/actions/build.ts

@@ -15,7 +15,9 @@ export async function build({
   options,
   reporting = "fancy",
 }: BuildArgs) {
+  if (!process.env.PLUGIN_ADMIN_UI_SKIP_CACHE) {
     await createCacheDir({ appDir, plugins })
+  }
 
   const cacheDir = path.resolve(appDir, ".cache")
   const entry = path.resolve(cacheDir, "admin", "src", "main.tsx")

packages/admin-ui/src/node/utils/create-cache-dir.ts

@@ -5,6 +5,10 @@ import { createEntry } from "./create-entry"
 import { logger } from "./logger"
 
 async function copyAdmin(dest: string) {
+  if (process.env.PLUGIN_ADMIN_UI_SKIP_CACHE) {
+    return true
+  }
+
   const adminDir = path.resolve(__dirname, "..", "ui")
   const destDir = path.resolve(dest, "admin")
 

packages/admin-ui/src/node/utils/create-entry.ts

@@ -13,6 +13,10 @@ import {
 const FILE_EXT_REGEX = /\.[^/.]+$/
 
 async function copyLocalExtensions(src: string, dest: string) {
+  if (process.env.PLUGIN_ADMIN_UI_SKIP_CACHE) {
+    return true
+  }
+
   try {
     await fse.copy(src, dest, {
       filter: copyFilter,

packages/medusa-cli/src/create-cli.ts

@@ -1,13 +1,13 @@
-import path from "path"
-import resolveCwd from "resolve-cwd"
 import { sync as existsSync } from "fs-exists-cached"
 import { setTelemetryEnabled } from "medusa-telemetry"
+import path from "path"
+import resolveCwd from "resolve-cwd"
 
-import { getLocalMedusaVersion } from "./util/version"
 import { didYouMean } from "./did-you-mean"
+import { getLocalMedusaVersion } from "./util/version"
 
-import reporter from "./reporter"
 import { newStarter } from "./commands/new"
+import reporter from "./reporter"
 
 const yargs = require(`yargs`)
 
@@ -233,6 +233,42 @@ function buildLocalCommands(cli, isLocalProject) {
         })
       ),
     })
+    .command({
+      command: `start-cluster`,
+      desc: `Start development server in cluster mode (beta).`,
+      builder: (_) =>
+        _.option(`H`, {
+          alias: `host`,
+          type: `string`,
+          default: defaultHost,
+          describe: `Set host. Defaults to ${defaultHost}`,
+        })
+          .option(`p`, {
+            alias: `port`,
+            type: `string`,
+            default: process.env.PORT || defaultPort,
+            describe: process.env.PORT
+              ? `Set port. Defaults to ${process.env.PORT} (set by env.PORT) (otherwise defaults ${defaultPort})`
+              : `Set port. Defaults to ${defaultPort}`,
+          })
+          .option(`c`, {
+            alias: `cpus`,
+            type: `number`,
+            default: process.env.CPUS,
+            describe:
+              "Set number of cpus to use. Defaults to max number of cpus available on the system (set by env.CPUS)",
+          }),
+      handler: handlerP(
+        getCommandHandler(`start-cluster`, (args, cmd) => {
+          process.env.NODE_ENV = process.env.NODE_ENV || `development`
+          cmd(args)
+          // Return an empty promise to prevent handlerP from exiting early.
+          // The development server shouldn't ever exit until the user directly
+          // kills it so this is fine.
+          return new Promise((resolve) => {})
+        })
+      ),
+    })
     .command({
       command: `user`,
       desc: `Create a user`,

packages/medusa/src/api/routes/store/carts/create-line-item/index.ts

@@ -2,16 +2,18 @@ import { IsInt, IsOptional, IsString } from "class-validator"
 import { EntityManager } from "typeorm"
 import { validator } from "../../../../../utils/validator"
 import {
+  addOrUpdateLineItem,
   CreateLineItemSteps,
-  handleAddOrUpdateLineItem,
+  setPaymentSession,
+  setVariantAvailability,
 } from "./utils/handler-steps"
 import { IdempotencyKey } from "../../../../../models"
-import {
+import { initializeIdempotencyRequest } from "../../../../../utils/idempotency"
-  initializeIdempotencyRequest,
-  runIdempotencyStep,
-  RunIdempotencyStepOptions,
-} from "../../../../../utils/idempotency"
 import { cleanResponseData } from "../../../../../utils/clean-response-data"
+import IdempotencyKeyService from "../../../../../services/idempotency-key"
+import { defaultStoreCartFields, defaultStoreCartRelations } from "../index"
+import { CartService } from "../../../../../services"
+import { promiseAll } from "@medusajs/utils"
 
 /**
  * @oas [post] /store/carts/{id}/line-items
@@ -89,34 +91,82 @@ export default async (req, res) => {
   let inProgress = true
   let err: unknown = false
 
-  const stepOptions: RunIdempotencyStepOptions = {
+  const idempotencyKeyService: IdempotencyKeyService = req.scope.resolve(
-    manager,
+    "idempotencyKeyService"
-    idempotencyKey,
+  )
-    container: req.scope,
-    isolationLevel: "SERIALIZABLE",
-  }
 
   while (inProgress) {
     switch (idempotencyKey.recovery_point) {
       case CreateLineItemSteps.STARTED: {
-        await runIdempotencyStep(async ({ manager }) => {
+        try {
-          return await handleAddOrUpdateLineItem(
+          const cartId = id
-            id,
+          const data = {
-            {
             customer_id: customerId,
             metadata: validated.metadata,
             quantity: validated.quantity,
             variant_id: validated.variant_id,
-            },
-            {
-              manager,
-              container: req.scope,
           }
-          )
+
-        }, stepOptions).catch((e) => {
+          await addOrUpdateLineItem({
+            cartId,
+            container: req.scope,
+            manager,
+            data,
+          })
+
+          idempotencyKey = await idempotencyKeyService
+            .withTransaction(manager)
+            .update(idempotencyKey.idempotency_key, {
+              recovery_point: CreateLineItemSteps.SET_PAYMENT_SESSIONS,
+            })
+        } catch (e) {
           inProgress = false
           err = e
+        }
+
+        break
+      }
+
+      case CreateLineItemSteps.SET_PAYMENT_SESSIONS: {
+        try {
+          const cartService: CartService = req.scope.resolve("cartService")
+
+          const cart = await cartService
+            .withTransaction(manager)
+            .retrieveWithTotals(id, {
+              select: defaultStoreCartFields,
+              relations: [
+                ...defaultStoreCartRelations,
+                "billing_address",
+                "region.payment_providers",
+                "payment_sessions",
+                "customer",
+              ],
             })
+
+          const args = {
+            cart,
+            container: req.scope,
+            manager,
+          }
+
+          await promiseAll([
+            setVariantAvailability(args),
+            setPaymentSession(args),
+          ])
+
+          idempotencyKey = await idempotencyKeyService
+            .withTransaction(manager)
+            .update(idempotencyKey.idempotency_key, {
+              recovery_point: CreateLineItemSteps.FINISHED,
+              response_code: 200,
+              response_body: { cart },
+            })
+        } catch (e) {
+          inProgress = false
+          err = e
+        }
+
         break
       }
 

packages/medusa/src/api/routes/store/carts/create-line-item/utils/handler-steps.ts

@@ -1,6 +1,3 @@
-import { FlagRouter } from "@medusajs/utils"
-import { AwilixContainer } from "awilix"
-import { EntityManager } from "typeorm"
 import { Cart } from "../../../../../../models"
 import {
   CartService,
@@ -8,36 +5,25 @@ import {
   ProductVariantInventoryService,
 } from "../../../../../../services"
 import { WithRequiredProperty } from "../../../../../../types/common"
-import { IdempotencyCallbackResult } from "../../../../../../types/idempotency-key"
-import { defaultStoreCartFields, defaultStoreCartRelations } from "../../index"
 import SalesChannelFeatureFlag from "../../../../../../loaders/feature-flags/sales-channels"
-import { MedusaError } from "medusa-core-utils"
+import { featureFlagRouter } from "../../../../../../loaders/feature-flags"
 
 export const CreateLineItemSteps = {
   STARTED: "started",
+  SET_PAYMENT_SESSIONS: "set-payment-sessions",
   FINISHED: "finished",
 }
 
-export async function handleAddOrUpdateLineItem(
+export async function addOrUpdateLineItem({
-  cartId: string,
+  cartId,
-  data: {
+  container,
-    metadata?: Record<string, unknown>
+  manager,
-    customer_id?: string
+  data,
-    variant_id: string
+}) {
-    quantity: number
-  },
-  { container, manager }: { container: AwilixContainer; manager: EntityManager }
-): Promise<IdempotencyCallbackResult> {
   const cartService: CartService = container.resolve("cartService")
   const lineItemService: LineItemService = container.resolve("lineItemService")
-  const featureFlagRouter: FlagRouter = container.resolve("featureFlagRouter")
 
-  const productVariantInventoryService: ProductVariantInventoryService =
+  const cart = await cartService.retrieve(cartId, {
-    container.resolve("productVariantInventoryService")
-
-  const txCartService = cartService.withTransaction(manager)
-
-  let cart = await txCartService.retrieve(cartId, {
     select: ["id", "region_id", "customer_id"],
   })
 
@@ -48,42 +34,46 @@ export async function handleAddOrUpdateLineItem(
       metadata: data.metadata,
     })
 
+  await manager.transaction(async (transactionManager) => {
+    const txCartService = cartService.withTransaction(transactionManager)
+
     await txCartService.addOrUpdateLineItems(cart.id, line, {
-    validateSalesChannels: featureFlagRouter.isFeatureEnabled("sales_channels"),
+      validateSalesChannels:
+        featureFlagRouter.isFeatureEnabled("sales_channels"),
     })
-
-  const relations = [
-    ...defaultStoreCartRelations,
-    "billing_address",
-    "region.payment_providers",
-    "payment_sessions",
-    "customer",
-  ]
-
-  const shouldSetAvailability =
-    relations?.some((rel) => rel.includes("variant")) &&
-    featureFlagRouter.isFeatureEnabled(SalesChannelFeatureFlag.key)
-
-  cart = await txCartService.retrieveWithTotals(cart.id, {
-    select: defaultStoreCartFields,
-    relations,
   })
-
-  if (shouldSetAvailability) {
-    await productVariantInventoryService.setVariantAvailability(
-      cart.items.map((i) => i.variant),
-      cart.sales_channel_id!
-    )
 }
 
-  if (cart.payment_sessions?.length) {
+export async function setPaymentSession({ cart, container, manager }) {
-    await txCartService.setPaymentSessions(
+  const cartService: CartService = container.resolve("cartService")
+
+  const txCartService = cartService.withTransaction(manager)
+
+  if (!cart.payment_sessions?.length) {
+    return
+  }
+
+  return await txCartService.setPaymentSessions(
     cart as WithRequiredProperty<Cart, "total">
   )
 }
 
-  return {
+export async function setVariantAvailability({ cart, container, manager }) {
-    response_code: 200,
+  const productVariantInventoryService: ProductVariantInventoryService =
-    response_body: { cart },
+    container.resolve("productVariantInventoryService")
+
+  const shouldSetAvailability =
+    cart.items?.some((item) => !!item.variant) &&
+    featureFlagRouter.isFeatureEnabled(SalesChannelFeatureFlag.key)
+
+  if (!shouldSetAvailability) {
+    return
   }
+
+  return await productVariantInventoryService
+    .withTransaction(manager)
+    .setVariantAvailability(
+      cart.items.map((i) => i.variant),
+      cart.sales_channel_id!
+    )
 }

packages/medusa/src/api/routes/store/carts/update-line-item.ts

@@ -8,7 +8,6 @@ import { defaultStoreCartFields, defaultStoreCartRelations } from "."
 import { EntityManager } from "typeorm"
 import { MedusaError } from "medusa-core-utils"
 import { cleanResponseData } from "../../../../utils/clean-response-data"
-import { handleAddOrUpdateLineItem } from "./create-line-item/utils/handler-steps"
 
 /**
  * @oas [post] /store/carts/{id}/line-items/{line_id}

packages/medusa/src/commands/start-cluster.js

@@ -0,0 +1,94 @@
+import "core-js/stable"
+import "regenerator-runtime/runtime"
+
+import cluster from "cluster"
+import express from "express"
+import { GracefulShutdownServer } from "medusa-core-utils"
+import { track } from "medusa-telemetry"
+import { scheduleJob } from "node-schedule"
+import os from "os"
+
+import loaders from "../loaders"
+import Logger from "../loaders/logger"
+
+const EVERY_SIXTH_HOUR = "0 */6 * * *"
+const CRON_SCHEDULE = EVERY_SIXTH_HOUR
+
+let isShuttingDown = false
+export default async function ({ port, cpus, directory }) {
+  if (cluster.isPrimary) {
+    const killMainProccess = () => process.exit(0)
+
+    cpus ??= os.cpus().length
+    const numCPUs = Math.min(os.cpus().length, cpus)
+
+    for (let index = 0; index < numCPUs; index++) {
+      const worker = cluster.fork()
+      worker.send({ index })
+    }
+
+    cluster.on("exit", (worker) => {
+      if (!isShuttingDown) {
+        cluster.fork()
+      } else if (Object.keys(cluster.workers).length === 0) {
+        setTimeout(killMainProccess, 100)
+      }
+    })
+
+    const gracefulShutDown = () => {
+      isShuttingDown = true
+      for (const id of Object.keys(cluster.workers)) {
+        cluster.workers[id].kill("SIGTERM")
+      }
+    }
+
+    scheduleJob(CRON_SCHEDULE, () => {
+      track("PING")
+    })
+
+    process.on("SIGTERM", gracefulShutDown)
+    process.on("SIGINT", gracefulShutDown)
+  } else {
+    const start = async () => {
+      track("CLI_START")
+
+      const app = express()
+
+      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 gracefulShutDown = () => {
+        server
+          .shutdown()
+          .then(() => {
+            process.exit(0)
+          })
+          .catch((e) => {
+            process.exit(1)
+          })
+      }
+
+      process.on("SIGTERM", gracefulShutDown)
+      process.on("SIGINT", gracefulShutDown)
+
+      return { dbConnection, server }
+    }
+
+    process.on("message", async (msg) => {
+      if (msg.index > 0) {
+        process.env.PLUGIN_ADMIN_UI_SKIP_CACHE = true
+      }
+
+      await start()
+    })
+  }
+}

packages/medusa/src/services/__mocks__/idempotency-key.js

@@ -27,6 +27,12 @@ export const IdempotencyKeyService = {
       }
     }
   }),
+  update: jest.fn().mockImplementation(async (key, data) => {
+    return {
+      ...data,
+      idempotency_key: key,
+    }
+  }),
 }
 
 const mock = jest.fn().mockImplementation(() => {

packages/medusa/src/services/product-variant-inventory.ts

@@ -797,8 +797,7 @@ class ProductVariantInventoryService extends TransactionBaseService {
         availabilityContext
       )
 
-    return await promiseAll(
+    return variants.map((variant) => {
-      variants.map(async (variant) => {
       if (!variant.id) {
         return variant
       }
@@ -837,7 +836,6 @@ class ProductVariantInventoryService extends TransactionBaseService {
 
       return variant
     })
-    )
   }
 
   private async getAvailabilityContext(

packages/medusa/src/strategies/cart-completion.ts

@@ -244,9 +244,17 @@ class CartCompletionStrategy extends AbstractCartCompletionStrategy {
       return res
     }
 
-    const cart = await this.cartService_
+    const txCartService = this.cartService_.withTransaction(manager)
-      .withTransaction(manager)
+
-      .authorizePayment(id, {
+    let cart = await txCartService.retrieve(id, {
+      relations: ["payment_sessions"],
+    })
+
+    if (cart.payment_sessions?.length) {
+      await txCartService.setPaymentSessions(id)
+    }
+
+    cart = await txCartService.authorizePayment(id, {
       ...context,
       cart_id: id,
       idempotency_key: idempotencyKey,
@@ -257,7 +265,7 @@ class CartCompletionStrategy extends AbstractCartCompletionStrategy {
         cart.payment_session.status === "requires_more" ||
         cart.payment_session.status === "pending"
       ) {
-        await this.cartService_.withTransaction(manager).deleteTaxLines(id)
+        await txCartService.deleteTaxLines(id)
 
         return {
           response_code: 200,