fix(core-flows): capture before order created (#9980)

What:
 When `autocapture` is enabled, the webhook is processed before the order was created.
 The payment processing workflows were merged into a single one
 
FIXES: SUP-118, SUP-9

https://github.com/medusajs/medusa/issues/9998

.changeset/thirty-lamps-collect.md

@@ -0,0 +1,6 @@
+---
+"@medusajs/core-flows": patch
+"@medusajs/medusa": patch
+---
+
+Create Order before payment capture

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

@@ -77,7 +77,7 @@ export const completeCartWorkflow = createWorkflow(
 
       const paymentSessions = validateCartPaymentsStep({ cart })
 
-      authorizePaymentSessionStep({
+      const payment = authorizePaymentSessionStep({
         // We choose the first payment session, as there will only be one active payment session
         // This might change in the future.
         id: paymentSessions[0].id,
@@ -103,7 +103,17 @@ export const completeCartWorkflow = createWorkflow(
         }
       })
 
-      const cartToOrder = transform({ cart }, ({ cart }) => {
+      const cartToOrder = transform({ cart, payment }, ({ cart, payment }) => {
+        const transactions =
+          payment?.captures?.map((capture) => {
+            return {
+              amount: capture.raw_amount ?? capture.amount,
+              currency_code: payment.currency_code,
+              reference: "capture",
+              reference_id: capture.id,
+            }
+          }) ?? []
+
         const allItems = (cart.items ?? []).map((item) => {
           return prepareLineItemData({
             item,
@@ -158,6 +168,7 @@ export const completeCartWorkflow = createWorkflow(
           shipping_methods: shippingMethods,
           metadata: cart.metadata,
           promo_codes: promoCodes,
+          transactions,
         }
       })
 

packages/core/core-flows/src/order/steps/add-order-transaction.ts

@@ -4,19 +4,44 @@ import { StepResponse, createStep } from "@medusajs/framework/workflows-sdk"
 
 export const addOrderTransactionStepId = "add-order-transaction"
 /**
- * This step creates an order transaction.
+ * This step creates order transactions.
  */
 export const addOrderTransactionStep = createStep(
   addOrderTransactionStepId,
-  async (data: CreateOrderTransactionDTO, { container }) => {
+  async (
+    data: CreateOrderTransactionDTO | CreateOrderTransactionDTO[],
+    { container }
+  ) => {
     const service = container.resolve(Modules.ORDER)
 
-    const created = await service.addOrderTransactions(data)
+    const trxsData = Array.isArray(data) ? data : [data]
 
-    return new StepResponse(created, created.id)
+    for (const trx of trxsData) {
+      const existing = await service.listOrderTransactions(
+        {
+          order_id: trx.order_id,
+          reference: trx.reference,
+          reference_id: trx.reference_id,
+        },
+        {
+          select: ["id"],
+        }
+      )
+
+      if (existing.length) {
+        return new StepResponse(null)
+      }
+    }
+
+    const created = await service.addOrderTransactions(trxsData)
+
+    return new StepResponse(
+      Array.isArray(data) ? created : created[0],
+      created.map((c) => c.id)
+    )
   },
   async (id, { container }) => {
-    if (!id) {
+    if (!id?.length) {
       return
     }
 

packages/core/core-flows/src/payment/steps/authorize-payment-session.ts

@@ -40,7 +40,12 @@ export const authorizePaymentSessionStep = createStep(
       )
     }
 
-    const paymentSession = await paymentModule.retrievePaymentSession(input.id)
+    const paymentSession = await paymentModule.retrievePaymentSession(
+      input.id,
+      {
+        relations: ["payment", "payment.captures"],
+      }
+    )
 
     // Throw a special error type when the status is requires_more as it requires a specific further action
     // from the consumer

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

@@ -39,13 +39,15 @@ export const capturePaymentWorkflow = createWorkflow(
       const orderTransactionData = transform(
         { input, payment, orderPayment },
         ({ input, payment, orderPayment }) => {
-          return {
-            order_id: orderPayment.order.id,
-            amount: input.amount ?? payment.raw_amount ?? payment.amount,
-            currency_code: payment.currency_code,
-            reference_id: payment.id,
-            reference: "capture",
-          }
+          return payment.captures?.map((capture) => {
+            return {
+              order_id: orderPayment.order.id,
+              amount: input.amount ?? capture.raw_amount ?? capture.amount,
+              currency_code: payment.currency_code,
+              reference_id: capture.id,
+              reference: "capture",
+            }
+          })
         }
       )
 

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

@@ -1,5 +1,3 @@
 export * from "./capture-payment"
-export * from "./on-payment-processed"
 export * from "./process-payment"
 export * from "./refund-payment"
-

packages/core/core-flows/src/payment/workflows/on-payment-processed.ts

@@ -1,38 +0,0 @@
-import { WebhookActionResult } from "@medusajs/types"
-import { createWorkflow, when } from "@medusajs/workflows-sdk"
-import { completeCartWorkflow } from "../../cart"
-import { useRemoteQueryStep } from "../../common"
-import { useQueryStep } from "../../common/steps/use-query"
-
-export const onPaymentProcessedWorkflowId = "on-payment-processed-workflow"
-export const onPaymentProcessedWorkflow = createWorkflow(
-  onPaymentProcessedWorkflowId,
-  (input: WebhookActionResult) => {
-    const paymentSessionResult = useRemoteQueryStep({
-      entry_point: "payment_session",
-      fields: ["payment_collection_id"],
-      variables: { filters: { id: input.data?.session_id } },
-      list: false,
-    })
-
-    const cartPaymentCollection = useQueryStep({
-      entity: "cart_payment_collection",
-      fields: ["cart_id"],
-      filters: {
-        payment_collection_id: paymentSessionResult.payment_collection_id,
-      },
-    })
-
-    when({ cartPaymentCollection }, ({ cartPaymentCollection }) => {
-      return !!cartPaymentCollection.data.length
-    }).then(() => {
-      completeCartWorkflow.runAsStep({
-        input: {
-          id: cartPaymentCollection.data[0].cart_id,
-        },
-      })
-    })
-
-    // TODO: Add more cases down the line, e.g. order payments
-  }
-)

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

@@ -1,6 +1,7 @@
 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 { useQueryStep } from "../../common/steps/use-query"
 import { authorizePaymentSessionStep } from "../steps"
 import { capturePaymentWorkflow } from "./capture-payment"
@@ -15,6 +16,41 @@ export const processPaymentWorkflow = createWorkflow(
       entity: "payment",
       fields: ["id"],
       filters: { payment_session_id: input.data?.session_id },
+    }).config({
+      name: "payment-query",
+    })
+
+    const paymentSessionResult = useQueryStep({
+      entity: "payment_session",
+      fields: ["payment_collection_id"],
+      filters: { id: input.data?.session_id },
+    }).config({
+      name: "payment-session-query",
+    })
+
+    const cartPaymentCollection = useQueryStep({
+      entity: "cart_payment_collection",
+      fields: ["cart_id"],
+      filters: {
+        payment_collection_id:
+          paymentSessionResult.data[0].payment_collection_id,
+      },
+    }).config({
+      name: "cart-payment-query",
+    })
+
+    when({ cartPaymentCollection }, ({ cartPaymentCollection }) => {
+      return !!cartPaymentCollection.data.length
+    }).then(() => {
+      completeCartWorkflow
+        .runAsStep({
+          input: {
+            id: cartPaymentCollection.data[0].cart_id,
+          },
+        })
+        .config({
+          continueOnPermanentFailure: true, // Continue payment processing even if cart completion fails
+        })
     })
 
     when({ input }, ({ input }) => {
@@ -31,8 +67,12 @@ export const processPaymentWorkflow = createWorkflow(
     })
 
     when({ input }, ({ input }) => {
+      // Authorize payment session if no Cart is linked to the payment
+      // When associated with a Cart, the complete cart workflow will handle the authorization
       return (
-        input.action === PaymentActions.AUTHORIZED && !!input.data?.session_id
+        !cartPaymentCollection.data.length &&
+        input.action === PaymentActions.AUTHORIZED &&
+        !!input.data?.session_id
       )
     }).then(() => {
       authorizePaymentSessionStep({

packages/core/types/src/payment/common.ts

@@ -471,6 +471,11 @@ export interface CaptureDTO {
    */
   amount: BigNumberValue
 
+  /**
+   * The raw captured amount.
+   */
+  raw_amount?: BigNumberValue
+
   /**
    * The creation date of the capture.
    */
@@ -502,6 +507,11 @@ export interface RefundDTO {
    */
   amount: BigNumberValue
 
+  /**
+   * The raw refunded amount.
+   */
+  raw_amount?: BigNumberValue
+
   /**
    * The id of the refund_reason that is associated with the refund
    */

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

@@ -1,7 +1,4 @@
-import {
-  onPaymentProcessedWorkflow,
-  processPaymentWorkflow,
-} from "@medusajs/core-flows"
+import { processPaymentWorkflow } from "@medusajs/core-flows"
 import {
   IPaymentModuleService,
   ProviderWebhookPayload,
@@ -29,7 +26,7 @@ export default async function paymentWebhookhandler({
   const input = event.data
 
   if (
-    (input.payload.rawData as unknown as SerializedBuffer).type === "Buffer"
+    (input.payload?.rawData as unknown as SerializedBuffer)?.type === "Buffer"
   ) {
     input.payload.rawData = Buffer.from(
       (input.payload.rawData as unknown as SerializedBuffer).data
@@ -49,11 +46,6 @@ export default async function paymentWebhookhandler({
   await processPaymentWorkflow(container).run({
     input: processedEvent,
   })
-
-  // We process the intended side effects of payment processing separately.
-  await onPaymentProcessedWorkflow(container).run({
-    input: processedEvent,
-  })
 }
 
 export const config: SubscriberConfig = {

packages/modules/event-bus-local/src/services/__tests__/event-bus-local.ts

@@ -123,6 +123,7 @@ describe("LocalEventBusService", () => {
           data: { test: "1234" },
           metadata: { eventGroupId: "test" },
           name: "test-event",
+          options: {},
         })
 
         jest.clearAllMocks()

packages/modules/event-bus-local/src/services/event-bus-local.ts

@@ -9,6 +9,7 @@ import {
 } from "@medusajs/framework/types"
 import { AbstractEventBusModuleService } from "@medusajs/framework/utils"
 import { EventEmitter } from "events"
+import { setTimeout } from "timers/promises"
 import { ulid } from "ulid"
 
 type InjectedDependencies = {
@@ -69,7 +70,10 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
         )
       }
 
-      await this.groupOrEmitEvent(eventData)
+      await this.groupOrEmitEvent({
+        ...eventData,
+        options,
+      })
     }
   }
 
@@ -86,7 +90,13 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
       await this.groupEvent(eventGroupId, eventData)
     } else {
       const { options, ...eventBody } = eventData
-      this.eventEmitter_.emit(eventData.name, eventBody)
+
+      const options_ = options as { delay: number }
+      const delay = options?.delay ? setTimeout : async () => {}
+
+      delay(options_?.delay).then(() =>
+        this.eventEmitter_.emit(eventData.name, eventBody)
+      )
     }
   }
 
@@ -108,7 +118,12 @@ export default class LocalEventBusService extends AbstractEventBusModuleService
     for (const event of groupedEvents) {
       const { options, ...eventBody } = event
 
-      this.eventEmitter_.emit(event.name, eventBody)
+      const options_ = options as { delay: number }
+      const delay = options?.delay ? setTimeout : async () => {}
+
+      delay(options_?.delay).then(() =>
+        this.eventEmitter_.emit(event.name, eventBody)
+      )
     }
 
     await this.clearGroupedEvents(eventGroupId)