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

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

* fix cart update compensation

Author: carlos-r-l-rodrigues

.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,
+    })
+  }
+)