Merge branch 'main' into context-call-timeout-param

examples/ci/app/ci/constants.ts

@@ -49,6 +49,11 @@ export const TEST_ROUTES: Pick<TestConfig, "route" | "waitForSeconds">[] = [
     route: "call/workflow",
     waitForSeconds: 24
   },
+  {
+    // check the error when wf early returns
+    route: "returns-before-step",
+    waitForSeconds: 3
+  },
   {
     // checks context.run with async and sync route methods
     route: "async-sync-run",

examples/ci/app/ci/types.ts

@@ -25,6 +25,12 @@ export type TestConfig<TPayload = unknown> = {
    * expected result in the Redis
    */
   expectedResult: string
+  /**
+   * whether the workflow should start
+   * 
+   * @default true
+   */
+  shouldWorkflowStart?: boolean
 }
 
 export type RedisResult = {

examples/ci/app/ci/upstash/redis.ts

@@ -4,7 +4,7 @@ import { expect } from "../utils";
 import { type WorkflowContext } from "@upstash/workflow";
 import { CI_RANDOM_ID_HEADER, CI_ROUTE_HEADER } from "../constants";
 
-const redis = Redis.fromEnv();
+export const redis = Redis.fromEnv();
 const EXPIRE_IN_SECS = 60
 
 const getRedisKey = (

examples/ci/app/ci/utils.ts

@@ -65,14 +65,21 @@ export const getTestConfig = async (route: string) => {
 
 export const initiateTest = async (route: string, waitForSeconds: number) => {
   const randomTestId = nanoid()
-  const { headers, payload, expectedCallCount, expectedResult } = await getTestConfig(route)
+  const { headers, payload, expectedCallCount, expectedResult, shouldWorkflowStart = true } = await getTestConfig(route)
 
   const { messageId } = await qstash.startWorkflow({ route, headers, payload }, randomTestId)
 
   // sleep for 4 secs and check that message is delivered
   await new Promise(r => setTimeout(r, 4000));
 
-  await qstash.checkWorkflowStart(messageId)
+  try {
+    await qstash.checkWorkflowStart(messageId);
+  } catch (error) {
+    console.error(error);
+    if (shouldWorkflowStart) {
+      throw error;
+    };
+  }
 
   await new Promise(r => setTimeout(r, waitForSeconds * 1000));
 

examples/ci/app/test-routes/auth/custom/workflow/route.ts

@@ -15,8 +15,10 @@ const makeCall = async (
   context: WorkflowContext,
   stepName: string,
   method: "GET" | "POST",
+  headers: Record<string, string>,
   expectedStatus: number,
-  expectedBody: unknown
+  expectedBody: unknown,
+  callBody?: unknown
 ) => {
   const randomId = context.headers.get(CI_RANDOM_ID_HEADER)
   const route = context.headers.get(CI_ROUTE_HEADER)
@@ -27,8 +29,7 @@ const makeCall = async (
 
   const { status, body } = await context.call<FailureFunctionPayload>(stepName, {
     url: thirdPartyEndpoint,
-    body: 
-    {
+    body: callBody ?? {
       status: 200,
       header: "",
       body: "",
@@ -45,7 +46,7 @@ const makeCall = async (
     headers: {
       [ CI_RANDOM_ID_HEADER ]: randomId,
       [ CI_ROUTE_HEADER ]: route,
-      "Upstash-Workflow-Is-Failure": "true"
+      ...headers
     }
   })
 
@@ -63,15 +64,65 @@ export const { POST, GET } = testServe(
 
       await makeCall(
         context,
-        "regular call should fail",
+        "failure call should fail",
         "POST",
+        {
+          "Upstash-Workflow-Is-Failure": "true"
+        },
         500,
         {
           error: "WorkflowError",
           message: "Not authorized to run the failure function."
         }
       )
 
+      await makeCall(
+        context,
+        "callback request should fail",
+        "POST",
+        {
+          "Upstash-Workflow-Callback": "true",
+        },
+        400,
+        {
+          message: "Failed to authenticate Workflow request. If this is unexpected, see the caveat https://upstash.com/docs/workflow/basics/caveats#avoid-non-deterministic-code-outside-context-run",
+          workflowRunId: "no-workflow-id"
+        }
+      )
+
+      await makeCall(
+        context,
+        "init call should fail",
+        "POST",
+        {},
+        400,
+        {
+          message: "Failed to authenticate Workflow request. If this is unexpected, see the caveat https://upstash.com/docs/workflow/basics/caveats#avoid-non-deterministic-code-outside-context-run",
+          workflowRunId: "no-workflow-id"
+        }
+      )
+
+      // TODO: enable back
+      // was disabled because Upstash-Workflow-RunId header is being overwritten by backend with the current workflow
+      // await makeCall(
+      //   context,
+      //   "intermediate call should fail",
+      //   "POST",
+      //   {
+      //     "Upstash-Workflow-Sdk-Version": "1",
+      //     "Upstash-Workflow-RunId": customRunId
+      //   },
+      //   400,
+      //   {
+      //     message: "Failed to authenticate Workflow request. If this is unexpected, see the caveat https://upstash.com/docs/workflow/basics/caveats#avoid-non-deterministic-code-outside-context-run",
+      //     workflowRunId: customRunId
+      //   },
+      //   [{
+      //     body: btoa("hello there"),
+      //     callType: "step"
+      //   }]
+      // )
+
       const input = context.requestPayload;
       expect(input, payload);
 
@@ -84,7 +135,7 @@ export const { POST, GET } = testServe(
       retries: 0,
     }
   ), {
-    expectedCallCount: 4,
+    expectedCallCount: 8,
     expectedResult: "not authorized for failure",
     payload,
     headers: {

examples/ci/app/test-routes/auth/fail/route.ts

@@ -40,6 +40,7 @@ export const { POST, GET } = testServe(
     headers: {
       [ header ]: headerValue,
       "authentication": authentication
-    }
+    },
+    shouldWorkflowStart: false
   }
 ) 

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

@@ -3,4 +3,6 @@ export const FAILING_HEADER = "Fail-Header-Foo"
 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 GET_HEADER_VALUE = "get-header-value-FOO"
+
+export const PATCH_RESULT = 99999999

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 } from "../constants";
+import { FAILING_HEADER_VALUE, FAILING_HEADER, GET_HEADER, GET_HEADER_VALUE, PATCH_RESULT } from "../constants";
 
 const thirdPartyResult = "third-party-result";
 
@@ -24,7 +24,7 @@ export const POST = async (request: Request) => {
 
 export const PATCH = async () => {
   return new Response(
-    "failing request",
+    PATCH_RESULT.toString(),
     {
       status: 401,
       headers: {
@@ -34,6 +34,15 @@ export const PATCH = async () => {
   )
 }
 
+export const DELETE = async () => {
+  return new Response(
+    JSON.stringify({ foo: "bar", zed: 2 }),
+    {
+      status: 400
+    }
+  )
+}
+
 export const PUT = async () => {
   return new Response(
     undefined,

examples/ci/app/test-routes/call/workflow-with-failureFunction/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 { fail, saveResult } from "app/ci/upstash/redis"
-import { FAILING_HEADER, FAILING_HEADER_VALUE } from "../constants";
+import { FAILING_HEADER, FAILING_HEADER_VALUE, PATCH_RESULT } from "../constants";
 import { WorkflowContext } from "@upstash/workflow";
 
 const testHeader = `test-header-foo`
@@ -15,19 +15,19 @@ export const { POST, GET } = testServe(
   serve<string>(
     async (context) => {
 
-      const { body: patchResult, status, header } = await context.call("get call", {
+      const { body: patchResult, status, header } = await context.call<number>("get call", {
         url: thirdPartyEndpoint,
         method: "PATCH",
         retries: 0
       });
 
       expect(status, 401)
-      expect(patchResult as string, "failing request");
+      expect(patchResult, PATCH_RESULT);
       expect(header[FAILING_HEADER][0], FAILING_HEADER_VALUE)
 
       await saveResult(
         context,
-        patchResult as string,
+        patchResult.toString(),
       )
     }, {
       baseUrl: BASE_URL,
@@ -38,7 +38,7 @@ export const { POST, GET } = testServe(
     }
   ), {
     expectedCallCount: 4,
-    expectedResult: "failing request",
+    expectedResult: PATCH_RESULT.toString(),
     payload,
     headers: {
       [ testHeader ]: headerValue,

examples/ci/app/test-routes/call/workflow-with-failureUrl/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 } from "../constants";
+import { FAILING_HEADER, FAILING_HEADER_VALUE, PATCH_RESULT } from "../constants";
 
 const testHeader = `test-header-foo`
 const headerValue = `header-foo`
@@ -14,19 +14,19 @@ export const { POST, GET } = testServe(
   serve<string>(
     async (context) => {
 
-      const { body: patchResult, status, header } = await context.call("get call", {
+      const { body: patchResult, status, header } = await context.call<number>("get call", {
         url: thirdPartyEndpoint,
         method: "PATCH",
         retries: 0
       });
 
       expect(status, 401)
-      expect(patchResult as string, "failing request");
+      expect(patchResult, PATCH_RESULT);
       expect(header[FAILING_HEADER][0], FAILING_HEADER_VALUE)
 
       await saveResult(
         context,
-        patchResult as string,
+        patchResult.toString(),
       )
     }, {
       baseUrl: BASE_URL,
@@ -36,7 +36,7 @@ export const { POST, GET } = testServe(
     }
   ), {
     expectedCallCount: 4,
-    expectedResult: "failing request",
+    expectedResult: PATCH_RESULT.toString(),
     payload,
     headers: {
       [ testHeader ]: headerValue,

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 } from "../constants";
+import { FAILING_HEADER, FAILING_HEADER_VALUE, GET_HEADER, GET_HEADER_VALUE, PATCH_RESULT } from "../constants";
 
 const testHeader = `test-header-foo`
 const headerValue = `header-foo`
@@ -54,15 +54,15 @@ export const { POST, GET } = testServe(
       expect(getHeaders[GET_HEADER][0], GET_HEADER_VALUE)
       expect(getResult, "called GET 'third-party-result' 'get-header-value-x'");
 
-      const { body: patchResult, status, header } = await context.call("patch call", {
+      const { body: patchResult, status, header } = await context.call<number>("patch call", {
         url: thirdPartyEndpoint,
         headers: getHeader,
         method: "PATCH",
         retries: 1
       });
 
       expect(status, 401)
-      expect(patchResult as string, "failing request");
+      expect(patchResult, PATCH_RESULT);
       expect(header[FAILING_HEADER][0], FAILING_HEADER_VALUE)
 
       // put will return with an empty body. should return "" as body in that case.
@@ -75,6 +75,17 @@ export const { POST, GET } = testServe(
       expect(putStatus, 300)
       expect(putBody, "");
 
+      // DELETE will return a JSON
+      const { body: deleteBody, status: deleteStatus } = await context.call<{ foo: string, zed: number }>("delete call", {
+        url: thirdPartyEndpoint,
+        method: "DELETE",
+        retries: 0
+      })
+
+      expect(deleteStatus, 400)
+      expect(typeof deleteBody, "object");
+      expect(JSON.stringify(deleteBody), '{"foo":"bar","zed":2}');
+
       await saveResult(
         context,
         getResult
@@ -84,7 +95,7 @@ export const { POST, GET } = testServe(
       retries: 0
     }
   ), {
-    expectedCallCount: 12,
+    expectedCallCount: 14,
     expectedResult: "called GET 'third-party-result' 'get-header-value-x'",
     payload,
     headers: {

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

@@ -9,14 +9,15 @@ const headerValue = `header-bar`
 const authHeaderValue = `Bearer super-secret-token`
 
 const errorMessage = `my-error`
-const payload = "my-payload"
+const payload = undefined
 
 
 export const { POST, GET } = testServe(
   serve<string>(
     async (context) => {
       const input = context.requestPayload;
 
+      expect(typeof input, typeof payload);
       expect(input, payload);
       expect(context.headers.get(header)!, headerValue)
 
@@ -29,6 +30,8 @@ export const { POST, GET } = testServe(
       failureFunction: async ({ context, failStatus, failResponse }) => {
         expect(failStatus, 500);
         expect(failResponse, errorMessage);
+        expect(context.requestPayload, payload);
+        expect(typeof context.requestPayload, typeof payload);
         expect(context.headers.get("authentication")!, authHeaderValue);
 
         await saveResult(