fix(core-flows,workflows-sdk): compensate account holders only when its created (#12825)

* fix(core-flows,workflows-sdk): compensate account holders only when its created

* chore: remove only

.changeset/fair-glasses-push.md

@@ -0,0 +1,6 @@
+---
+"@medusajs/workflows-sdk": patch
+"@medusajs/core-flows": patch
+---
+
+fix(core-flows,workflows-sdk): compensate account holders only when its created

integration-tests/modules/__tests__/payment/payment-session.workflows.spec.ts

@@ -2,9 +2,13 @@ import {
   createPaymentSessionsWorkflow,
   createPaymentSessionsWorkflowId,
 } from "@medusajs/core-flows"
-import { IPaymentModuleService, IRegionModuleService } from "@medusajs/types"
-import { Modules } from "@medusajs/utils"
 import { medusaIntegrationTestRunner } from "@medusajs/test-utils"
+import {
+  ICustomerModuleService,
+  IPaymentModuleService,
+  IRegionModuleService,
+} from "@medusajs/types"
+import { ContainerRegistrationKeys, Modules } from "@medusajs/utils"
 
 jest.setTimeout(50000)
 
@@ -17,18 +21,21 @@ medusaIntegrationTestRunner({
       let appContainer
       let paymentModule: IPaymentModuleService
       let regionModule: IRegionModuleService
-      let remoteLink
+      let customerModule: ICustomerModuleService
+      let query
 
       beforeAll(async () => {
         appContainer = getContainer()
         paymentModule = appContainer.resolve(Modules.PAYMENT)
         regionModule = appContainer.resolve(Modules.REGION)
-        remoteLink = appContainer.resolve("remoteLink")
+        customerModule = appContainer.resolve(Modules.CUSTOMER)
+        query = appContainer.resolve(ContainerRegistrationKeys.QUERY)
       })
 
       describe("createPaymentSessionWorkflow", () => {
         let region
         let paymentCollection
+        let customer
 
         beforeEach(async () => {
           region = await regionModule.createRegions({
@@ -40,6 +47,12 @@ medusaIntegrationTestRunner({
             currency_code: "usd",
             amount: 1000,
           })
+
+          customer = await customerModule.createCustomers({
+            email: "test@test.com",
+            first_name: "Test",
+            last_name: "Test",
+          })
         })
 
         it("should create payment sessions", async () => {
@@ -75,6 +88,47 @@ medusaIntegrationTestRunner({
           )
         })
 
+        it("should create payment sessions with customer", async () => {
+          await createPaymentSessionsWorkflow(appContainer).run({
+            input: {
+              payment_collection_id: paymentCollection.id,
+              provider_id: "pp_system_default",
+              customer_id: customer.id,
+            },
+          })
+
+          const {
+            data: [updatedPaymentCollection],
+          } = await query.graph({
+            entity: "payment_collection",
+            filters: {
+              id: paymentCollection.id,
+            },
+            fields: ["id", "currency_code", "amount", "payment_sessions.*"],
+          })
+
+          expect(updatedPaymentCollection.payment_sessions).toHaveLength(1)
+          expect(updatedPaymentCollection).toEqual(
+            expect.objectContaining({
+              id: paymentCollection.id,
+              currency_code: "usd",
+              amount: 1000,
+              payment_sessions: expect.arrayContaining([
+                expect.objectContaining({
+                  context: expect.objectContaining({
+                    customer: expect.objectContaining({
+                      id: customer.id,
+                    }),
+                    account_holder: expect.objectContaining({
+                      email: customer.email,
+                    }),
+                  }),
+                }),
+              ]),
+            })
+          )
+        })
+
         it("should delete existing sessions when create payment sessions", async () => {
           await createPaymentSessionsWorkflow(appContainer).run({
             input: {
@@ -164,6 +218,89 @@ medusaIntegrationTestRunner({
 
             expect(sessions).toHaveLength(0)
           })
+
+          it("should not delete account holder if it exists before creating payment sessions", async () => {
+            await createPaymentSessionsWorkflow(appContainer).run({
+              input: {
+                payment_collection_id: paymentCollection.id,
+                provider_id: "pp_system_default",
+                customer_id: customer.id,
+              },
+            })
+
+            const {
+              data: [updatedCustomer1],
+            } = await query.graph({
+              entity: "customer",
+              filters: {
+                id: customer.id,
+              },
+              fields: ["id", "account_holders.*"],
+            })
+
+            expect(updatedCustomer1.account_holders).toEqual(
+              expect.arrayContaining([
+                expect.objectContaining({
+                  email: customer.email,
+                }),
+              ])
+            )
+
+            const newPaymentCollection =
+              await paymentModule.createPaymentCollections({
+                currency_code: "usd",
+                amount: 2000,
+              })
+
+            const workflow = createPaymentSessionsWorkflow(appContainer)
+
+            workflow.appendAction("throw", createPaymentSessionsWorkflowId, {
+              invoke: async function failStep() {
+                throw new Error(
+                  `Failed to do something after creating payment sessions`
+                )
+              },
+            })
+
+            const { errors } = await workflow.run({
+              input: {
+                payment_collection_id: newPaymentCollection.id,
+                provider_id: "pp_system_default",
+                customer_id: customer.id,
+                context: {},
+                data: {},
+              },
+              throwOnError: false,
+            })
+
+            expect(errors).toEqual([
+              {
+                action: "throw",
+                handlerType: "invoke",
+                error: expect.objectContaining({
+                  message: `Failed to do something after creating payment sessions`,
+                }),
+              },
+            ])
+
+            const {
+              data: [updatedCustomer2],
+            } = await query.graph({
+              entity: "customer",
+              filters: {
+                id: customer.id,
+              },
+              fields: ["id", "account_holders.*"],
+            })
+
+            expect(updatedCustomer2.account_holders).toEqual(
+              expect.arrayContaining([
+                expect.objectContaining({
+                  email: customer.email,
+                }),
+              ])
+            )
+          })
         })
       })
     })

packages/core/core-flows/src/payment-collection/steps/create-payment-account-holder.ts

@@ -1,14 +1,14 @@
 import {
-  IPaymentModuleService,
   CreateAccountHolderDTO,
+  IPaymentModuleService,
 } from "@medusajs/framework/types"
-import { Modules } from "@medusajs/framework/utils"
-import { StepResponse, createStep } from "@medusajs/framework/workflows-sdk"
+import { isPresent, Modules } from "@medusajs/framework/utils"
+import { createStep, StepResponse } from "@medusajs/framework/workflows-sdk"
 
 export const createPaymentAccountHolderStepId = "create-payment-account-holder"
 /**
  * This step creates the account holder in the payment provider.
- * 
+ *
  * @example
  * const accountHolder = createPaymentAccountHolderStep({
  *   provider_id: "pp_stripe_stripe",
@@ -27,7 +27,13 @@ export const createPaymentAccountHolderStep = createStep(
 
     const accountHolder = await service.createAccountHolder(data)
 
-    return new StepResponse(accountHolder, accountHolder)
+    // createAccountHolder is an idempotent operation.
+    // We pass the account holder to the compensation step if it was actually created to avoid deleting the existing account holder.
+    const createdAccountHolder = isPresent(data.context.account_holder)
+      ? null
+      : accountHolder
+
+    return new StepResponse(accountHolder, createdAccountHolder)
   },
   async (createdAccountHolder, { container }) => {
     if (!createdAccountHolder) {

packages/core/core-flows/src/payment-collection/workflows/create-payment-session.ts

@@ -82,7 +82,7 @@ export const createPaymentSessionsWorkflow = createWorkflow(
       list: false,
     }).config({ name: "get-payment-collection" })
 
-    const { paymentCustomer, accountHolder } = when(
+    const { paymentCustomer, accountHolder, existingAccountHolder } = when(
       "customer-id-exists",
       { input },
       (data) => {
@@ -138,20 +138,14 @@ export const createPaymentSessionsWorkflow = createWorkflow(
 
       const accountHolder = createPaymentAccountHolderStep(accountHolderInput)
 
-      return { paymentCustomer, accountHolder }
+      return { paymentCustomer, accountHolder, existingAccountHolder }
     })
 
     when(
       "account-holder-created",
-      { paymentCustomer, accountHolder, input },
-      (data) => {
-        return (
-          !isPresent(
-            data.paymentCustomer?.account_holders.find(
-              (ac) => ac.provider_id === data.input.provider_id
-            )
-          ) && isPresent(data.accountHolder)
-        )
+      { paymentCustomer, accountHolder, input, existingAccountHolder },
+      ({ existingAccountHolder, accountHolder }) => {
+        return !isPresent(existingAccountHolder) && isPresent(accountHolder)
       }
     ).then(() => {
       createRemoteLinkStep([

packages/core/workflows-sdk/src/utils/composer/helpers/step-response.ts

@@ -38,7 +38,9 @@ export class StepResponse<TOutput, TCompensateInput = TOutput> {
     if (isDefined(output)) {
       this.#output = output
     }
-    this.#compensateInput = (compensateInput ?? output) as TCompensateInput
+    this.#compensateInput = (
+      isDefined(compensateInput) ? compensateInput : output
+    ) as TCompensateInput
   }
 
   /**