fix(): Lock process payment to prevent ingesting payment processing t… (#13977)

* fix(): Lock process payment to prevent ingesting payment processing twice

* Create brave-lamps-beg.md

* fix(): Lock process payment to prevent ingesting payment processing twice

* fix changeset

* resolve conflicts

---------

Co-authored-by: Carlos R. L. Rodrigues <37986729+carlos-r-l-rodrigues@users.noreply.github.com>

.changeset/brave-lamps-beg.md

@@ -0,0 +1,6 @@
+---
+"@medusajs/medusa": patch
+"@medusajs/core-flows": patch
+---
+
+fix(): Lock process payment to prevent ingesting payment processing t…

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

@@ -1,10 +1,14 @@
 import type { WebhookActionResult } from "@medusajs/framework/types"
 import { PaymentActions } from "@medusajs/utils"
-import { createWorkflow, when } from "@medusajs/workflows-sdk"
+import { createWorkflow, transform, when } from "@medusajs/workflows-sdk"
 import { useQueryGraphStep } from "../../common"
 import { authorizePaymentSessionStep } from "../steps"
 import { completeCartAfterPaymentStep } from "../steps/complete-cart-after-payment"
 import { capturePaymentWorkflow } from "./capture-payment"
+import { acquireLockStep, releaseLockStep } from "../../locking"
+
+const THIRTY_SECONDS = 30
+const TWO_MINUTES = 60 * 2
 
 /**
  * The data to process a payment from a webhook action.
@@ -66,6 +70,23 @@ export const processPaymentWorkflow = createWorkflow(
       name: "cart-payment-query",
     })
 
+    const cartId = transform(
+      { cartPaymentCollection },
+      ({ cartPaymentCollection }) => {
+        return cartPaymentCollection.data[0].cart_id
+      }
+    )
+
+    when("lock-cart-when-available", { cartId }, ({ cartId }) => {
+      return !!cartId
+    }).then(() => {
+      acquireLockStep({
+        key: cartId,
+        timeout: THIRTY_SECONDS,
+        ttl: TWO_MINUTES,
+      })
+    })
+
     when({ input, paymentData }, ({ input, paymentData }) => {
       return (
         input.action === PaymentActions.SUCCESSFUL && !!paymentData.data.length
@@ -128,6 +149,15 @@ export const processPaymentWorkflow = createWorkflow(
       })
     })
 
+    // We release before the completion to prevent dead locks
+    when("release-lock-cart-when-available", { cartId }, ({ cartId }) => {
+      return !!cartId
+    }).then(() => {
+      releaseLockStep({
+        key: cartId,
+      })
+    })
+
     when({ cartPaymentCollection }, ({ cartPaymentCollection }) => {
       return !!cartPaymentCollection.data.length
     }).then(() => {

packages/medusa/src/subscribers/payment-webhook.ts

@@ -1,4 +1,4 @@
-import { processPaymentWorkflow } from "@medusajs/core-flows"
+import { processPaymentWorkflowId } from "@medusajs/core-flows"
 import {
   IPaymentModuleService,
   ProviderWebhookPayload,
@@ -49,7 +49,8 @@ export default async function paymentWebhookhandler({
     return
   }
 
-  await processPaymentWorkflow(container).run({ input: processedEvent })
+  const wfEngine = container.resolve(Modules.WORKFLOW_ENGINE)
+  await wfEngine.run(processPaymentWorkflowId, { input: processedEvent })
 }
 
 export const config: SubscriberConfig = {