fix(medusa): API workflow subscription (#15134)

## Summary

**What** —
1) Change req.query to req.params in appropriate api routes
+ move /api/admin/workflows-executions/[workflow_id]/[transaction_id]/[step_id]/subscribe/route.ts
to /api/admin/workflows-executions/[workflow_id]/[transaction_id]/subscribe/route.ts, because it is not possible to subscribe to individual steps, only to individual workflows.
+ modify API reference in documentation to mirror changes

2) Fix SSE Streaming in JS-SDK

**Why** —
1) Current implementation relies on **req.query** to get the id of workflow/transaction, but it should get them from **req.params** instead. It is also impossible to subscribe to specific steps.

2) When setting up SSE streams with JS-SDK, if an event has yet to be streamed, it is impossible to abort, because nothing has returned from the JS-SDK yet.

**How** —
1) Change req.query to req.params

2) fetchStream now always returns (non-null generator), so that it is possible to abort before receiving any events

**Testing** —
https://docs.medusajs.com/api/admin#workflows-executions_getworkflowsexecutionsworkflow_idsubscribe

---

## Examples
https://docs.medusajs.com/resources/js-sdk#stream-server-sent-events

```ts
const StreamTestPage = () => {
  const [messages, setMessages] = useState<string[]>([])
  const [isStreaming, setIsStreaming] = useState(false)
  const [abortStream, setAbortStream] = useState<(() => void) | null>(null)

  const startStream = async () => {
    setIsStreaming(true)
    setMessages([])
    
    const { stream, abort } = await sdk.client.fetchStream("/admin/stream")

    // Store the abort function for the abort button
    setAbortStream(() => abort)

    try {
      for await (const chunk of stream) {
        // Since the server sends plain text, convert to string
        const message = typeof chunk === "string" ? chunk : (chunk.data || String(chunk))
        setMessages((prev) => [...prev, message.trim()])
      }
    } catch (error) {
      // Don't log abort errors as they're expected when user clicks abort
      if (error instanceof Error && error.name !== "AbortError") {
        console.error("Stream error:", error)

        if (error.name === "HttpError") {
          abort()
        }
      }
    } finally {
      setIsStreaming(false)
      setAbortStream(null)
    }
  }

  const handleAbort = () => {
    if (abortStream) {
      abortStream()
      setIsStreaming(false)
      setAbortStream(null)
    }
  }

  return (
    <Container className="p-6">
      <Heading level="h1" className="mb-6">
        fetchStream Example
      </Heading>
      
      <div className="space-y-4">
        <div className="flex gap-2">
          <Button 
            onClick={startStream} 
            disabled={isStreaming}
            variant="primary"
          >
            {isStreaming ? "Streaming..." : "Start Stream"}
          </Button>
          
          <Button 
            onClick={handleAbort} 
            disabled={!isStreaming}
            variant="secondary"
          >
            Abort Stream
          </Button>
        </div>
        
        <div className="border rounded p-4 h-64 overflow-y-auto bg-ui-bg-subtle">
          {messages.length === 0 ? (
            <Text className="text-ui-fg-muted">No messages yet...</Text>
          ) : (
            messages.map((msg, index) => (
              <div key={index} className="mb-2 text-sm">
                {msg}
              </div>
            ))
          )}
        </div>
      </div>
    </Container>
  )
}
```
---

## Checklist

Please ensure the following before requesting a review:

- [ X] I have added a **changeset** for this PR
    - Every non-breaking change should be marked as a **patch**
    - To add a changeset, run `yarn changeset` and follow the prompts
- [ X] The changes are covered by relevant **tests**
- [ X] I have verified the code works as intended locally
- [ X] I have linked the related issue(s) if applicable

---

## Additional Context

This is a new PR based on [13886](#13886).
Unfortunately I synced the fork and deleted my commits, which then automatically closed the PR.

Closes #15135
Closes #15136

Author: v0eak

.changeset/hungry-numbers-sing.md

@@ -0,0 +1,6 @@
+---
+"@medusajs/medusa": minor
+"@medusajs/js-sdk": minor
+---
+
+fix(medusa,js-sdk): Refactor Workflow Subscription & Fix SSE Stream

integration-tests/http/__tests__/workflow-engine/admin/index.spec.ts

@@ -95,15 +95,16 @@ medusaIntegrationTestRunner({
       })
     })
 
-    describe("POST /admin/workflow-execution/[workflow_id]/steps/failure", function () {
+    describe("POST /admin/workflows-executions/[workflow_id]/steps/failure", function () {
         it("should set step as failed", async () => {
             const stepId = 'test-step'
             const step = createStep({
                 name: stepId,
                 async: true,
             }, () => { })
 
-            const workflowId = 'test-workflow'
+            const workflowId =
+              "workflow-f" + Math.random().toString(36).substring(2, 15)
             createWorkflow({
                 name: workflowId,
                 retentionTime: 60,
@@ -112,7 +113,8 @@ medusaIntegrationTestRunner({
                 return new WorkflowResponse(void 0)
             })
 
-            const transactionId = "test-transaction"
+            const transactionId =
+              "trx_123_f" + Math.random().toString(36).substring(2, 15)
             const engine = container.resolve(Modules.WORKFLOW_ENGINE) as IWorkflowEngineService
             await engine.run(workflowId, {
                 transactionId
@@ -143,95 +145,58 @@ medusaIntegrationTestRunner({
         })
     })
 
-    describe("Workflow Orchestrator module subscribe", function () {
-      it("should subscribe to a workflow and receive the response when it finishes", async () => {
-        const step1 = createStep({ name: "step1" }, async () => {
-          return new StepResponse("step1")
-        })
-        const step2 = createStep({ name: "step2" }, async () => {
-          await setTimeout(1000)
-          return new StepResponse("step2")
-        })
-
-        const workflowId =
-          "workflow" + Math.random().toString(36).substring(2, 15)
-        createWorkflow(workflowId, function (input) {
-          step1()
-          step2().config({
-            async: true,
-          })
-          return new WorkflowResponse("workflow")
-        })
+    describe("POST /admin/workflows-executions/[workflow_id]/steps/success", function () {
+        it("should set step as successful", async () => {
+            const stepId = 'test-step'
+            const step = createStep({
+                name: stepId,
+                async: true,
+            }, () => { })
 
-        const step1_1 = createStep({ name: "step1_1" }, async () => {
-          return new StepResponse("step1_1")
-        })
-        const step2_1 = createStep({ name: "step2_1" }, async () => {
-          await setTimeout(1000)
-          return new StepResponse("step2_1")
-        })
+            const workflowId =
+              "workflow_s" + Math.random().toString(36).substring(2, 15)
+            createWorkflow({
+                name: workflowId,
+                retentionTime: 60,
+            }, () => {
+                step()
+                return new WorkflowResponse(void 0)
+            })
 
-        const workflow2Id =
-          "workflow_2" + Math.random().toString(36).substring(2, 15)
-        createWorkflow(workflow2Id, function (input) {
-          step1_1()
-          step2_1().config({
-            async: true,
-          })
-          return new WorkflowResponse("workflow_2")
-        })
+            const transactionId =
+              "trx_123_s" + Math.random().toString(36).substring(2, 15)
+            const engine = container.resolve(Modules.WORKFLOW_ENGINE) as IWorkflowEngineService
+            await engine.run(workflowId, {
+                transactionId
+            })
+            let workflowDetail = (await api.get(`/admin/workflows-executions/${workflowId}/${transactionId}`, adminHeaders)).data.workflow_execution
 
-        const transactionId =
-          "trx_123" + Math.random().toString(36).substring(2, 15)
-        const transactionId2 =
-          "trx_124" + Math.random().toString(36).substring(2, 15)
+            expect(workflowDetail.state).toBe(TransactionState.INVOKING)
 
-        const onWorkflowFinishSpy = jest.fn()
+            const setSuccessResponse = await api.post(`/admin/workflows-executions/${workflowId}/steps/success`, {
+                transaction_id: transactionId,
+                step_id: stepId
+            }, adminHeaders)
 
-        const onWorkflowFinishPromise = new Promise<void>((resolve) => {
-          void workflowOrcModule.subscribe({
-            workflowId: workflowId,
-            transactionId,
-            subscriber: (event) => {
-              console.log("event", event)
-              if (event.eventType === "onFinish") {
-                onWorkflowFinishSpy()
-                workflowOrcModule.run(workflow2Id, {
-                  transactionId: transactionId2,
+            expect(setSuccessResponse.status).toBe(200)
+            expect(setSuccessResponse.data).toEqual(
+                expect.objectContaining({
+                    success: true,
                 })
-                resolve()
-              }
-            },
-          })
-        })
-
-        const onWorkflow2FinishSpy = jest.fn()
-
-        const workflow2FinishPromise = new Promise<void>((resolve) => {
-          void workflowOrcModule.subscribe({
-            workflowId: workflow2Id,
-            subscriber: (event) => {
-              console.log("event", event)
-              if (event.eventType === "onFinish") {
-                onWorkflow2FinishSpy()
-                resolve()
-              }
-            },
-          })
-        })
+            )
 
-        workflowOrcModule.run(workflowId, {
-          transactionId,
+            workflowDetail = (await api.get(`/admin/workflows-executions/${workflowId}/${transactionId}`, adminHeaders)).data.workflow_execution
+            
+            expect(workflowDetail).toEqual(
+                expect.objectContaining({
+                    state: TransactionState.DONE,
+                })
+            )
         })
+    })
 
-        await onWorkflowFinishPromise
-        await workflow2FinishPromise
-
-        expect(onWorkflowFinishSpy).toHaveBeenCalledTimes(1)
-        expect(onWorkflow2FinishSpy).toHaveBeenCalledTimes(1)
-      })
-
-      it("should subscribe to a workflow and receive the response when it finishes (2)", async () => {
+    describe("Workflow Orchestrator module subscribe", function () {
+      it("should subscribe to a workflow and receive the response when it finishes (1)", async () => {
         const step1 = createStep({ name: "step1" }, async () => {
           return new StepResponse("step1")
         })

packages/core/js-sdk/src/client.ts

@@ -194,17 +194,26 @@ export class Client {
     const abortController = new AbortController()
     const abortFunc = abortController.abort.bind(abortController)
 
-    let res = await this.fetch_(input, {
+    const fetchPromise = this.fetch_(input, {
       ...init,
       signal: abortController.signal,
       headers: { ...init?.headers, accept: "text/event-stream" },
     })
 
-    if (res.ok) {
-      return { stream: events(res, abortController.signal), abort: abortFunc }
-    }
+    return {
+      stream: (async function* () {
+        const res = await fetchPromise
 
-    return { stream: null, abort: abortFunc }
+        if (!res.ok) {
+          const error = new Error(`Stream failed with status ${res.status}`)
+          error.name = "HttpError"
+          throw error
+        }
+
+        yield* events(res, abortController.signal)
+      })(),
+      abort: abortFunc
+    }
   }
 
   async setToken(token: string) {

packages/core/js-sdk/src/types.ts

@@ -78,6 +78,6 @@ export interface ServerSentEventMessage {
 }
 
 export interface FetchStreamResponse {
-  stream: AsyncGenerator<ServerSentEventMessage, void, unknown> | null
+  stream: AsyncGenerator<ServerSentEventMessage, void, unknown>
   abort: () => void
 }

…saction_id]/[step_id]/subscribe/route.ts …_id]/[transaction_id]/subscribe/route.tspackages/medusa/src/api/admin/workflows-executions/[workflow_id]/[transaction_id]/[step_id]/subscribe/route.ts renamed to packages/medusa/src/api/admin/workflows-executions/[workflow_id]/[transaction_id]/subscribe/route.ts packages/medusa/src/api/admin/workflows-executions/[workflow_id]/[transaction_id]/[step_id]/subscribe/route.ts renamed to packages/medusa/src/api/admin/workflows-executions/[workflow_id]/[transaction_id]/subscribe/route.ts

@@ -14,7 +14,7 @@ export const GET = async (
     Modules.WORKFLOW_ENGINE
   )
 
-  const { workflow_id, transaction_id } = req.query as any
+  const { workflow_id, transaction_id } = req.params
 
   const subscriberId = "__sub__" + Math.random().toString(36).substring(2, 9)
   res.writeHead(200, {

packages/medusa/src/api/admin/workflows-executions/[workflow_id]/subscribe/route.ts

@@ -14,7 +14,7 @@ export const GET = async (
     Modules.WORKFLOW_ENGINE
   )
 
-  const { workflow_id } = req.query as any
+  const { workflow_id } = req.params
 
   const subscriberId = "__sub__" + Math.random().toString(36).substring(2, 9)
   res.writeHead(200, {