test(workflow-engine): Ensure failed subscriber does not compensate workflows (#12427)

* test(workflow-engine): Ensure failed subscriber does not compensate workflows

* test(workflow-engine): Ensure failed subscriber does not compensate workflows

integration-tests/modules/__tests__/workflow-engine/workflow-engine.ts

@@ -1,4 +1,4 @@
-import { Modules } from "@medusajs/framework/utils"
+import { Modules, TransactionState } from "@medusajs/framework/utils"
 import {
   createStep,
   createWorkflow,
@@ -11,8 +11,10 @@ import {
   adminHeaders,
   createAdminUser,
 } from "../../../helpers/create-admin-user"
+import { emitEventStep } from "@medusajs/core-flows"
+import { IEventBusModuleService } from "@medusajs/types"
 
-jest.setTimeout(50000)
+jest.setTimeout(300000)
 
 medusaIntegrationTestRunner({
   testSuite: ({ dbConnection, getContainer, api }) => {
@@ -88,6 +90,81 @@ medusaIntegrationTestRunner({
           )
         })
       })
+
+      describe("Workflows event", () => {
+        const failingEventName = "failing-event"
+
+        it("should not compensate the workflow if the event subscriber fails", async () => {
+          const step1 = createStep(
+            {
+              name: "my-step",
+            },
+            async (_) => {
+              return new StepResponse({ result: "success" })
+            }
+          )
+
+          createWorkflow(
+            {
+              name: "my-workflow-name",
+              retentionTime: 50,
+            },
+            function (input: WorkflowData<{ initial: string }>) {
+              const stepRes = step1()
+
+              emitEventStep({
+                eventName: failingEventName,
+                data: {
+                  input: stepRes,
+                },
+              })
+
+              return new WorkflowResponse(stepRes)
+            }
+          )
+
+          const container = getContainer()
+          const eventBus = container.resolve(
+            Modules.EVENT_BUS
+          ) as IEventBusModuleService
+
+          const eventSpy = jest.fn()
+          eventBus.subscribe(failingEventName, async (event) => {
+            eventSpy(event)
+            throw new Error("Failed to emit event")
+          })
+
+          const engine = container.resolve(Modules.WORKFLOW_ENGINE)
+
+          const transactionId = "trx-id-failing-event"
+          const res = await engine.run("my-workflow-name", {
+            transactionId,
+            input: {
+              initial: "abc",
+            },
+          })
+
+          expect(res.result).toEqual({ result: "success" })
+
+          const executions = await engine.listWorkflowExecutions({
+            transaction_id: transactionId,
+          })
+
+          expect(executions.length).toBe(1)
+          expect(executions[0].state).toBe(TransactionState.DONE)
+
+          expect(eventSpy).toHaveBeenCalledTimes(1)
+          expect(eventSpy).toHaveBeenCalledWith(
+            expect.objectContaining({
+              data: {
+                input: {
+                  result: "success",
+                },
+              },
+            })
+          )
+        })
+      })
     })
   },
 })