fix: address unicode issue in context.call

Author: CahidArda

.github/workflows/test.yaml

@@ -224,7 +224,7 @@ jobs:
       - name: Run example
         run: npm run dev &
         working-directory: examples/image-gen-with-workflow
-  
+
   agents-researcher-local-build:
     needs:
       - local-tests

examples/ci/app/test-routes/call/constants.ts

@@ -5,4 +5,6 @@ export const FAILING_HEADER_VALUE = "fail-header-value-BAR"
 export const GET_HEADER = "Get-Header"
 export const GET_HEADER_VALUE = "get-header-value-FOO"
 
-export const PATCH_RESULT = 99999999
+export const PATCH_RESULT = 99999999
+
+export const DELETE_RESULT = { foo: "bar", zed: 2, unicode: "`“X” - 𐐷𐐶𐐹`" }

examples/ci/app/test-routes/call/third-party/route.ts

@@ -1,4 +1,4 @@
-import { FAILING_HEADER_VALUE, FAILING_HEADER, GET_HEADER, GET_HEADER_VALUE, PATCH_RESULT } from "../constants";
+import { FAILING_HEADER_VALUE, FAILING_HEADER, GET_HEADER, GET_HEADER_VALUE, PATCH_RESULT, DELETE_RESULT } from "../constants";
 
 const thirdPartyResult = "third-party-result";
 
@@ -8,7 +8,7 @@ export const GET = async (request: Request) => {
     {
       status: 200,
       headers: {
-        [ GET_HEADER ]: GET_HEADER_VALUE
+        [GET_HEADER]: GET_HEADER_VALUE
       }
     }
   )
@@ -28,15 +28,15 @@ export const PATCH = async () => {
     {
       status: 401,
       headers: {
-        [ FAILING_HEADER ]: FAILING_HEADER_VALUE
+        [FAILING_HEADER]: FAILING_HEADER_VALUE
       }
     }
   )
 }
 
 export const DELETE = async () => {
   return new Response(
-    JSON.stringify({ foo: "bar", zed: 2 }),
+    JSON.stringify(DELETE_RESULT),
     {
       status: 400
     }

examples/ci/app/test-routes/call/workflow/route.ts

@@ -2,7 +2,7 @@ import { serve } from "@upstash/workflow/nextjs";
 import { BASE_URL, TEST_ROUTE_PREFIX } from "app/ci/constants";
 import { testServe, expect } from "app/ci/utils";
 import { saveResult } from "app/ci/upstash/redis"
-import { FAILING_HEADER, FAILING_HEADER_VALUE, GET_HEADER, GET_HEADER_VALUE, PATCH_RESULT } from "../constants";
+import { DELETE_RESULT, FAILING_HEADER, FAILING_HEADER_VALUE, GET_HEADER, GET_HEADER_VALUE, PATCH_RESULT } from "../constants";
 
 const testHeader = `test-header-foo`
 const headerValue = `header-foo`
@@ -38,18 +38,18 @@ export const { POST, GET } = testServe(
       // check payload after first step because we can't check above
       expect(input, payload);
       expect(postStatus, 201)
-      
-      expect(postResult as string, 
+
+      expect(postResult as string,
         "called POST 'third-party-result' 'post-header-value-x' '\"post-payload\"'"
       );
-      
+
       await context.sleep("sleep 1", 2);
-      
+
       const { body: getResult, header: getHeaders, status: getStatus } = await context.call<string>("get call", {
         url: thirdPartyEndpoint,
         headers: getHeader,
       });
-      
+
       expect(getStatus, 200)
       expect(getHeaders[GET_HEADER][0], GET_HEADER_VALUE)
       expect(getResult, "called GET 'third-party-result' 'get-header-value-x'");
@@ -84,7 +84,7 @@ export const { POST, GET } = testServe(
 
       expect(deleteStatus, 400)
       expect(typeof deleteBody, "object");
-      expect(JSON.stringify(deleteBody), '{"foo":"bar","zed":2}');
+      expect(JSON.stringify(deleteBody), JSON.stringify(DELETE_RESULT));
 
       await saveResult(
         context,

examples/ci/app/test-routes/invoke/workflows/[...]/route.ts

@@ -1,6 +1,6 @@
 import { WorkflowContext } from "@upstash/workflow";
 import { createWorkflow, serveMany } from "@upstash/workflow/nextjs";
-import { CI_RANDOM_ID_HEADER, CI_ROUTE_HEADER, TEST_ROUTE_PREFIX } from "app/ci/constants";
+import { BASE_URL, CI_RANDOM_ID_HEADER, CI_ROUTE_HEADER, TEST_ROUTE_PREFIX } from "app/ci/constants";
 import { saveResult } from "app/ci/upstash/redis";
 import { expect, nanoid, testServe } from "app/ci/utils";
 import { z } from "zod";
@@ -13,6 +13,9 @@ const payload = 123
 const invokePayload = "invoke-payload"
 const invokeResult = "invoke-result"
 
+const BRANCH_ONE_RESULT = { "branch-one": "result" }
+const BRANCH_TWO_RESULT = "branch-two-result"
+
 const invokeHeader = "invoke-header"
 const invokeHeaderValue = "invoke-header-value"
 
@@ -64,7 +67,7 @@ const workflowOne = createWorkflow(async (context: WorkflowContext<number>) => {
     "done invoke"
   )
 }, {
-  schema: z.number()
+  schema: z.number(),
 })
 
 const workflowTwo = createWorkflow(async (context: WorkflowContext<string>) => {
@@ -105,8 +108,9 @@ const workflowTwo = createWorkflow(async (context: WorkflowContext<string>) => {
     })
   ])
 
-  expect(result[0].body, "branch-one-result")
-  expect(result[1].body, "branch-two-result")
+  expect(typeof result[0].body, "object")
+  expect(JSON.stringify(result[0].body), JSON.stringify(BRANCH_ONE_RESULT))
+  expect(result[1].body, BRANCH_TWO_RESULT)
 
   return invokeResult
 })
@@ -143,7 +147,7 @@ const branchOne = createWorkflow(async (context: WorkflowContext<number>) => {
 
   await context.sleep("check", 1)
 
-  return "branch-one-result"
+  return BRANCH_ONE_RESULT
 })
 
 /**
@@ -176,7 +180,7 @@ const branchTwo = createWorkflow(async (context: WorkflowContext<number>) => {
 
   await context.sleep("check", 1)
 
-  return "branch-two-result"
+  return BRANCH_TWO_RESULT
 })
 
 export const { POST, GET } = testServe(
@@ -186,6 +190,8 @@ export const { POST, GET } = testServe(
     workflowThree,
     branchOne,
     branchTwo,
+  }, {
+    baseUrl: BASE_URL
   }),
   {
     expectedCallCount: 26,

examples/ci/app/test-routes/path/route.ts

@@ -31,6 +31,11 @@ export const { POST, GET } = testServe(
       });
 
       expect(result2, "processed 'processed '“unicode-quotes”''");
+
+      const result3 = await context.run("step 3", () => true)
+      expect(result3, true)
+      expect(typeof result3, "boolean")
+      
       await saveResult(
         context,
         result2
@@ -40,7 +45,7 @@ export const { POST, GET } = testServe(
       retries: 0
     }
   ), {
-    expectedCallCount: 4,
+    expectedCallCount: 5,
     expectedResult: "processed 'processed '“unicode-quotes”''",
     payload,
     headers: {

src/context/auto-executor.test.ts

@@ -46,7 +46,7 @@ describe("auto-executor", () => {
     stepId: 1,
     stepName: "attemptCharge",
     stepType: "Run",
-    out: { input: initialPayload, success: false },
+    out: JSON.stringify({ input: initialPayload, success: false }),
     concurrent: 1,
   };
 
@@ -134,7 +134,6 @@ describe("auto-executor", () => {
               },
               body: JSON.stringify({
                 ...singleStep,
-                out: JSON.stringify(singleStep.out),
               }),
             },
           ],

src/context/auto-executor.ts

@@ -135,7 +135,7 @@ export class AutoExecutor {
         step,
         stepCount: this.stepCount,
       });
-      return step.out as TResult;
+      return lazyStep.parseOut(step.out);
     }
 
     const resultStep = await lazyStep.getResultStep(NO_CONCURRENCY, this.stepCount);
@@ -268,7 +268,9 @@ export class AutoExecutor {
 
         validateParallelSteps(parallelSteps, parallelResultSteps);
 
-        return parallelResultSteps.map((step) => step.out) as TResults;
+        return parallelResultSteps.map((step, index) =>
+          parallelSteps[index].parseOut(step.out)
+        ) as TResults;
       }
     }
     const fillValue = undefined;

src/context/context.ts

@@ -237,7 +237,7 @@ export class WorkflowContext<TInitialPayload = unknown> {
   ): Promise<TResult> {
     const wrappedStepFunction = (() =>
       this.executor.wrapStep(stepName, stepFunction)) as StepFunction<TResult>;
-    return this.addStep<TResult>(new LazyFunctionStep(stepName, wrappedStepFunction));
+    return await this.addStep<TResult>(new LazyFunctionStep(stepName, wrappedStepFunction));
   }
 
   /**
@@ -316,51 +316,28 @@ export class WorkflowContext<TInitialPayload = unknown> {
     stepName: string,
     settings: CallSettings<TBody>
   ): Promise<CallResponse<TResult>> {
-    const { url, method = "GET", body, headers = {}, retries = 0, timeout, flowControl } = settings;
+    const {
+      url,
+      method = "GET",
+      body: requestBody,
+      headers = {},
+      retries = 0,
+      timeout,
+      flowControl,
+    } = settings;
 
-    const result = await this.addStep(
-      new LazyCallStep<CallResponse<string> | string>(
+    return await this.addStep(
+      new LazyCallStep<TResult>(
         stepName,
         url,
         method,
-        body,
+        requestBody,
         headers,
         retries,
         timeout,
         flowControl
       )
     );
-
-    // <for backwards compatibity>
-    // if you transition to upstash/workflow from upstash/qstash,
-    // the out field in the steps will be the body of the response.
-    // we need to handle them explicitly here
-    if (typeof result === "string") {
-      try {
-        const body = JSON.parse(result);
-        return {
-          status: 200,
-          header: {},
-          body,
-        };
-      } catch {
-        return {
-          status: 200,
-          header: {},
-          body: result as TResult,
-        };
-      }
-    }
-    // </for backwards compatibity>
-
-    try {
-      return {
-        ...result,
-        body: JSON.parse(result.body as string),
-      };
-    } catch {
-      return result as CallResponse<TResult>;
-    }
   }
 
   /**
@@ -406,16 +383,7 @@ export class WorkflowContext<TInitialPayload = unknown> {
 
     const timeoutStr = typeof timeout === "string" ? timeout : `${timeout}s`;
 
-    const result = await this.addStep(new LazyWaitForEventStep(stepName, eventId, timeoutStr));
-
-    try {
-      return {
-        ...result,
-        eventData: JSON.parse(result.eventData as string),
-      };
-    } catch {
-      return result;
-    }
+    return await this.addStep(new LazyWaitForEventStep(stepName, eventId, timeoutStr));
   }
 
   /**
@@ -444,30 +412,16 @@ export class WorkflowContext<TInitialPayload = unknown> {
     eventId: string,
     eventData: unknown
   ): Promise<NotifyStepResponse> {
-    const result = await this.addStep(
+    return await this.addStep(
       new LazyNotifyStep(stepName, eventId, eventData, this.qstashClient.http)
     );
-
-    try {
-      return {
-        ...result,
-        eventData: JSON.parse(result.eventData as string),
-      };
-    } catch {
-      return result;
-    }
   }
 
   public async invoke<TInitialPayload, TResult>(
     stepName: string,
     settings: LazyInvokeStepParams<TInitialPayload, TResult>
   ) {
-    const result = await this.addStep(new LazyInvokeStep(stepName, settings));
-
-    return {
-      ...result,
-      body: (result.body ? JSON.parse(result.body as string) : undefined) as TResult,
-    };
+    return await this.addStep(new LazyInvokeStep(stepName, settings));
   }
 
   /**