fix(core-flows): complete cart improvements (#12646)

* fix(core-flows): use cartId as transactionId and acquire lock to complete cart

* fix cart update compensation

.changeset/proud-dancers-film.md

@@ -0,0 +1,7 @@
+---
+"@medusajs/workflows-sdk": patch
+"@medusajs/core-flows": patch
+"@medusajs/utils": patch
+---
+
+fix(core-flows): use cart_id to complete cart on payment webhook and lock cart before completion

integration-tests/modules/__tests__/cart/store/cart.completion.ts

@@ -747,6 +747,135 @@ medusaIntegrationTestRunner({
             paymentSession.id
           )
         })
+        it("should complete cart when payment webhook and storefront are called in simultaneously", async () => {
+          const salesChannel = await scModuleService.createSalesChannels({
+            name: "Webshop",
+          })
+
+          const location = await stockLocationModule.createStockLocations({
+            name: "Warehouse",
+          })
+
+          const [product] = await productModule.createProducts([
+            {
+              title: "Test product",
+              variants: [
+                {
+                  title: "Test variant",
+                  manage_inventory: false,
+                },
+              ],
+            },
+          ])
+
+          const priceSet = await pricingModule.createPriceSets({
+            prices: [
+              {
+                amount: 3000,
+                currency_code: "usd",
+              },
+            ],
+          })
+
+          await pricingModule.createPricePreferences({
+            attribute: "currency_code",
+            value: "usd",
+            has_tax_inclusive: true,
+          })
+
+          await remoteLink.create([
+            {
+              [Modules.PRODUCT]: {
+                variant_id: product.variants[0].id,
+              },
+              [Modules.PRICING]: {
+                price_set_id: priceSet.id,
+              },
+            },
+            {
+              [Modules.SALES_CHANNEL]: {
+                sales_channel_id: salesChannel.id,
+              },
+              [Modules.STOCK_LOCATION]: {
+                stock_location_id: location.id,
+              },
+            },
+          ])
+
+          // create cart
+          const cart = await cartModuleService.createCarts({
+            currency_code: "usd",
+            sales_channel_id: salesChannel.id,
+          })
+
+          await addToCartWorkflow(appContainer).run({
+            input: {
+              items: [
+                {
+                  variant_id: product.variants[0].id,
+                  quantity: 1,
+                  requires_shipping: false,
+                },
+              ],
+              cart_id: cart.id,
+            },
+          })
+
+          await createPaymentCollectionForCartWorkflow(appContainer).run({
+            input: {
+              cart_id: cart.id,
+            },
+          })
+
+          const [payCol] = await remoteQuery(
+            remoteQueryObjectFromString({
+              entryPoint: "cart_payment_collection",
+              variables: { filters: { cart_id: cart.id } },
+              fields: ["payment_collection_id"],
+            })
+          )
+
+          const { result: paymentSession } =
+            await createPaymentSessionsWorkflow(appContainer).run({
+              input: {
+                payment_collection_id: payCol.payment_collection_id,
+                provider_id: "pp_system_default",
+                context: {},
+                data: {},
+              },
+            })
+
+          const [{ result: order }] = await Promise.all([
+            completeCartWorkflow(appContainer).run({
+              input: {
+                id: cart.id,
+              },
+            }),
+            processPaymentWorkflow(appContainer).run({
+              input: {
+                action: "captured",
+                data: {
+                  session_id: paymentSession.id,
+                  amount: 3000,
+                },
+              },
+            }),
+          ])
+
+          const { result: fullOrder } = await getOrderDetailWorkflow(
+            appContainer
+          ).run({
+            input: {
+              fields: ["*"],
+              order_id: order.id,
+            },
+          })
+
+          expect(fullOrder.payment_status).toBe("captured")
+          expect(fullOrder.payment_collections[0].authorized_amount).toBe(3000)
+          expect(fullOrder.payment_collections[0].captured_amount).toBe(3000)
+          expect(fullOrder.payment_collections[0].status).toBe("completed")
+        })
       })
     })
   },

packages/core/core-flows/src/cart/steps/update-carts.ts

@@ -29,7 +29,18 @@ export const updateCartsStep = createStep(
   async (data: UpdateCartsStepInput, { container }) => {
     const cartModule = container.resolve<ICartModuleService>(Modules.CART)
 
-    const { selects, relations } = getSelectsAndRelationsFromObjectArray(data)
+    const { selects, relations } = getSelectsAndRelationsFromObjectArray(data, {
+      requiredFields: [
+        "id",
+        "region_id",
+        "customer_id",
+        "sales_channel_id",
+        "email",
+        "currency_code",
+        "metadata",
+        "completed_at",
+      ],
+    })
     const cartsBeforeUpdate = await cartModule.listCarts(
       { id: data.map((d) => d.id) },
       { select: selects, relations }

packages/core/core-flows/src/cart/workflows/complete-cart.ts

@@ -24,6 +24,8 @@ import {
   useQueryGraphStep,
   useRemoteQueryStep,
 } from "../../common"
+import { acquireLockStep } from "../../locking/acquire-lock"
+import { releaseLockStep } from "../../locking/release-lock"
 import { addOrderTransactionStep } from "../../order/steps/add-order-transaction"
 import { createOrdersStep } from "../../order/steps/create-orders"
 import { authorizePaymentSessionStep } from "../../payment/steps/authorize-payment-session"
@@ -60,7 +62,9 @@ export type CompleteCartWorkflowOutput = {
   id: string
 }
 
-export const THREE_DAYS = 60 * 60 * 24 * 3
+const THREE_DAYS = 60 * 60 * 24 * 3
+const THIRTY_SECONDS = 30
+const TWO_MINUTES = 60 * 2
 
 export const completeCartWorkflowId = "complete-cart"
 /**
@@ -93,6 +97,12 @@ export const completeCartWorkflow = createWorkflow(
     retentionTime: THREE_DAYS,
   },
   (input: WorkflowData<CompleteCartWorkflowInput>) => {
+    acquireLockStep({
+      key: input.id,
+      timeout: THIRTY_SECONDS,
+      ttl: TWO_MINUTES,
+    })
+
     const orderCart = useQueryGraphStep({
       entity: "order_cart",
       fields: ["cart_id", "order_id"],
@@ -381,6 +391,10 @@ export const completeCartWorkflow = createWorkflow(
       return createdOrder
     })
 
+    releaseLockStep({
+      key: input.id,
+    })
+
     const result = transform({ order, orderId }, ({ order, orderId }) => {
       return { id: order?.id ?? orderId } as CompleteCartWorkflowOutput
     })

packages/core/core-flows/src/index.ts

@@ -11,6 +11,7 @@ export * from "./fulfillment"
 export * from "./inventory"
 export * from "./invite"
 export * from "./line-item"
+export * from "./locking"
 export * from "./notification"
 export * from "./order"
 export * from "./payment"

packages/core/core-flows/src/locking/acquire-lock.ts

@@ -0,0 +1,82 @@
+import { isDefined, Modules } from "@medusajs/framework/utils"
+import { createStep, StepResponse } from "@medusajs/framework/workflows-sdk"
+import { setTimeout } from "timers/promises"
+
+/**
+ * The keys to be locked
+ */
+export interface AcquireLockStepInput {
+  key: string | string[]
+  timeout?: number // in seconds. Defaults to 0
+  retryInterval?: number // in seconds. Defaults to 0.3
+  ttl?: number // in seconds
+  ownerId?: string
+  provider?: string
+}
+
+export const acquireLockStepId = "acquire-lock-step"
+/**
+ * This step acquires a lock for a given key.
+ *
+ * @example
+ * const data = acquireLockStep({
+ *   "key": "my-lock-key",
+ *   "ttl": 60
+ * })
+ */
+export const acquireLockStep = createStep(
+  acquireLockStepId,
+  async (data: AcquireLockStepInput, { container }) => {
+    const keys = Array.isArray(data.key)
+      ? data.key
+      : isDefined(data.key)
+      ? [data.key]
+      : []
+
+    if (!keys.length) {
+      return new StepResponse(void 0)
+    }
+
+    const locking = container.resolve(Modules.LOCKING)
+
+    const retryInterval = data.retryInterval ?? 0.3
+    const tryUntil = Date.now() + (data.timeout ?? 0) * 1000
+
+    while (true) {
+      try {
+        await locking.acquire(data.key, {
+          expire: data.ttl,
+          ownerId: data.ownerId,
+          provider: data.provider,
+        })
+        break
+      } catch (e) {
+        if (Date.now() >= tryUntil) {
+          throw e
+        }
+      }
+
+      await setTimeout(retryInterval * 1000)
+    }
+
+    return new StepResponse(void 0, {
+      keys,
+      ownerId: data.ownerId,
+      provider: data.provider,
+    })
+  },
+  async (data, { container }) => {
+    if (!data?.keys?.length) {
+      return
+    }
+
+    const locking = container.resolve(Modules.LOCKING)
+
+    await locking.release(data.keys, {
+      ownerId: data.ownerId,
+      provider: data.provider,
+    })
+
+    return new StepResponse()
+  }
+)

packages/core/core-flows/src/locking/index.ts

@@ -0,0 +1,2 @@
+export * from "./acquire-lock"
+export * from "./release-lock"

packages/core/core-flows/src/locking/release-lock.ts

@@ -0,0 +1,43 @@
+import { isDefined, Modules } from "@medusajs/framework/utils"
+import { createStep, StepResponse } from "@medusajs/framework/workflows-sdk"
+
+/**
+ * The locked keys to be released
+ */
+export interface ReleaseLockStepInput {
+  key: string | string[]
+  ownerId?: string
+  provider?: string
+}
+
+export const releaseLockStepId = "release-lock-step"
+/**
+ * This step releases a lock for a given key.
+ *
+ * @example
+ * const data = releaseLockStep({
+ *   "key": "my-lock-key"
+ * })
+ */
+export const releaseLockStep = createStep(
+  releaseLockStepId,
+  async (data: ReleaseLockStepInput, { container }) => {
+    const keys = Array.isArray(data.key)
+      ? data.key
+      : isDefined(data.key)
+      ? [data.key]
+      : []
+
+    if (!keys.length) {
+      return new StepResponse(true)
+    }
+
+    const locking = container.resolve(Modules.LOCKING)
+    const released = await locking.release(keys, {
+      ownerId: data.ownerId,
+      provider: data.provider,
+    })
+
+    return new StepResponse(released)
+  }
+)

packages/core/core-flows/src/payment/steps/complete-cart-after-payment.ts

@@ -0,0 +1,31 @@
+import { Modules } from "@medusajs/framework/utils"
+import { createStep } from "@medusajs/framework/workflows-sdk"
+import { completeCartWorkflowId } from "../../cart/workflows/complete-cart"
+
+/**
+ * The data to complete a cart after a payment is captured.
+ */
+export type CompleteCartAfterPaymentStepInput = {
+  /**
+   * The ID of the cart to complete.
+   */
+  cart_id: string
+}
+
+export const completeCartAfterPaymentStepId = "complete-cart-after-payment-step"
+/**
+ * This step completes a cart after a payment is captured.
+ */
+export const completeCartAfterPaymentStep = createStep(
+  completeCartAfterPaymentStepId,
+  async (input: CompleteCartAfterPaymentStepInput, { container }) => {
+    const workflowEngine = container.resolve(Modules.WORKFLOW_ENGINE)
+
+    await workflowEngine.run(completeCartWorkflowId, {
+      input: {
+        id: input.cart_id,
+      },
+      transactionId: input.cart_id,
+    })
+  }
+)

packages/core/core-flows/src/payment/workflows/process-payment.ts

@@ -1,9 +1,9 @@
 import { WebhookActionResult } from "@medusajs/types"
 import { PaymentActions } from "@medusajs/utils"
 import { createWorkflow, when } from "@medusajs/workflows-sdk"
-import { completeCartWorkflow } from "../../cart/workflows/complete-cart"
 import { useQueryGraphStep } from "../../common"
 import { authorizePaymentSessionStep } from "../steps"
+import { completeCartAfterPaymentStep } from "../steps/complete-cart-after-payment"
 import { capturePaymentWorkflow } from "./capture-payment"
 
 /**
@@ -131,11 +131,9 @@ export const processPaymentWorkflow = createWorkflow(
     when({ cartPaymentCollection }, ({ cartPaymentCollection }) => {
       return !!cartPaymentCollection.data.length
     }).then(() => {
-      completeCartWorkflow
+      completeCartAfterPaymentStep({
-        .runAsStep({
+        cart_id: cartPaymentCollection.data[0].cart_id,
-          input: { id: cartPaymentCollection.data[0].cart_id },
+      }).config({
-        })
-        .config({
         continueOnPermanentFailure: true, // Continue payment processing even if cart completion fails
       })
     })

packages/core/utils/src/common/__tests__/get-selects-and-relations-from-object-array.spec.ts

@@ -40,7 +40,7 @@ describe("getSelectsAndRelationsFromObjectArray", function () {
           },
         ],
         output: {
-          selects: [
+          selects: expect.arrayContaining([
             "attr_string",
             "attr_boolean",
             "attr_null",
@@ -55,16 +55,19 @@ describe("getSelectsAndRelationsFromObjectArray", function () {
             "attr_array.attr_object.attr_boolean",
             "attr_array.attr_object.attr_null",
             "attr_array.attr_object.attr_undefined",
-          ],
+            "required_attr",
+          ]),
           relations: ["attr_object", "attr_array", "attr_array.attr_object"],
         },
       },
     ]
 
     expectations.forEach((expectation) => {
-      expect(getSelectsAndRelationsFromObjectArray(expectation.input)).toEqual(
+      expect(
-        expectation.output
+        getSelectsAndRelationsFromObjectArray(expectation.input, {
-      )
+          requiredFields: ["required_attr"],
+        })
+      ).toEqual(expectation.output)
     })
   })
 })

packages/core/utils/src/common/get-selects-and-relations-from-object-array.ts

@@ -5,8 +5,9 @@ const KEYS_THAT_ARE_NOT_RELATIONS = ["metadata"]
 
 export function getSelectsAndRelationsFromObjectArray(
   dataArray: object[],
-  options: { objectFields: string[] } = {
+  options?: {
-    objectFields: [],
+    objectFields?: string[]
+    requiredFields?: string[]
   },
   prefix?: string
 ): {
@@ -15,10 +16,11 @@ export function getSelectsAndRelationsFromObjectArray(
 } {
   const selects: string[] = []
   const relations: string[] = []
+  const { objectFields, requiredFields } = options ?? {}
 
   for (const data of dataArray) {
     for (const [key, value] of Object.entries(data)) {
-      if (isObject(value) && !options.objectFields.includes(key)) {
+      if (isObject(value) && !objectFields?.includes(key)) {
         const res = getSelectsAndRelationsFromObjectArray(
           [value],
           options,
@@ -48,7 +50,10 @@ export function getSelectsAndRelationsFromObjectArray(
     }
   }
 
-  const uniqueSelects: string[] = deduplicate(selects)
+  const uniqueSelects: string[] = deduplicate([
+    ...selects,
+    ...(requiredFields ?? []),
+  ])
   const uniqueRelations: string[] = deduplicate(relations)
 
   return {

packages/core/workflows-sdk/src/utils/composer/create-hook.ts

@@ -116,7 +116,7 @@ export function createHook<Name extends string, TInvokeInput, TInvokeOutput>(
         stepName: name,
         input: hookInput,
         invokeFn,
-        compensateFn,
+        compensateFn: compensateFn ?? (() => void 0),
       })
 
       if (this.hooks_.registered.includes(name)) {

packages/core/workflows-sdk/src/utils/composer/helpers/resolve-value.ts

@@ -1,4 +1,9 @@
-import { deepCopy, OrchestrationUtils, promiseAll } from "@medusajs/utils"
+import {
+  deepCopy,
+  isDefined,
+  OrchestrationUtils,
+  promiseAll,
+} from "@medusajs/utils"
 
 async function resolveProperty(property, transactionContext) {
   const { invoke: invokeRes } = transactionContext
@@ -78,5 +83,8 @@ export async function resolveValue(input, transactionContext) {
     ? await resolveProperty(copiedInput, transactionContext)
     : await unwrapInput(copiedInput, {})
 
-  return result && JSON.parse(JSON.stringify(result))
+  const strResult = JSON.stringify(result) // Symbols return undefined
+  if (isDefined(strResult)) {
+    return JSON.parse(strResult)
+  }
 }