DX-1811: workflow endpoint feature set

examples/ci/app/ci/constants.ts

@@ -99,6 +99,10 @@ export const TEST_ROUTES: Pick<TestConfig, "route" | "waitForSeconds">[] = [
   {
     route: "invoke/workflows/workflowOne",
     waitForSeconds: 10
+  },
+  {
+    route: "trigger-non-workflow/workflow",
+    waitForSeconds: 10
   }
 
   /**

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

@@ -210,7 +210,7 @@ export const { POST, GET } = testServe(
     baseUrl: BASE_URL
   }),
   {
-    expectedCallCount: 28,
+    expectedCallCount: 29,
     expectedResult: "done invoke",
     payload,
     headers: {

examples/ci/app/test-routes/trigger-non-workflow/constants.ts

@@ -0,0 +1 @@
+export const NON_WORKFLOW_ROUTE_RESPONSE = "super-secret-response-foo";

examples/ci/app/test-routes/trigger-non-workflow/non-workflow/route.ts

@@ -0,0 +1,5 @@
+import { NON_WORKFLOW_ROUTE_RESPONSE } from "../constants";
+
+export const POST = async (request: Request) => {
+  return new Response(NON_WORKFLOW_ROUTE_RESPONSE, { status: 200 });
+}

examples/ci/app/test-routes/trigger-non-workflow/workflow/route.ts

@@ -0,0 +1,68 @@
+import { serve } from "@upstash/workflow/nextjs";
+import { Client, StepError } from "@upstash/workflow";
+import { BASE_URL } from "app/ci/constants";
+import { saveResult } from "app/ci/upstash/redis";
+import { expect, testServe } from "app/ci/utils";
+import { NON_WORKFLOW_ROUTE_RESPONSE } from "../constants";
+
+const header = `test-header-foo`
+const headerValue = `header-bar`
+
+const workflowClient = new Client({ baseUrl: process.env.QSTASH_URL, token: process.env.QSTASH_TOKEN! })
+
+export const { POST, GET } = testServe(
+  serve(async (context) => {
+
+    const { workflowRunId } = await context.run("trigger non-workflow", async () =>
+      workflowClient.trigger({
+        url: `${BASE_URL}/test-routes/trigger-non-workflow/non-workflow`,
+      })
+    )
+
+    await context.sleep("wait before checking logs", 5)
+
+    const errorBody = await context.run("check run logs", async () => {
+      for (let counter = 0; counter < 5; counter++) {
+        const { runs } = await workflowClient.logs({ workflowRunId })        
+        if (runs.length === 1) {
+          const run = runs[0];
+          expect(run.workflowState, "RUN_FAILED")
+          expect(run.steps.length, 2)
+
+          const secondStep = run.steps[1];
+          expect(secondStep.type, "next")
+          expect(secondStep.steps.length, 1)
+          expect(secondStep.steps[0].state, "STEP_FAILED")
+
+          const errors = (secondStep.steps[0] as { errors: StepError[] }).errors;
+
+          expect(errors.length, 1)
+          expect(errors[0].error, "detected a non-workflow destination for trigger/invoke. make sure you are sending the request to the correct endpoint")
+          return errors[0].body
+        } else {
+          await new Promise(r => setTimeout(r, 2000));
+        }
+      }
+      return false
+    })
+
+    if (!errorBody) {
+      throw new Error(`Workflow run with ID ${workflowRunId} did not complete successfully`)
+    }
+
+    await saveResult(
+      context,
+      errorBody
+    )
+
+  }, {
+    baseUrl: BASE_URL,
+    retries: 0
+  }), {
+  expectedCallCount: 5,
+  expectedResult: NON_WORKFLOW_ROUTE_RESPONSE,
+  payload: NON_WORKFLOW_ROUTE_RESPONSE,
+  headers: {
+    [header]: headerValue
+  }
+})

examples/express/api/index.ts

@@ -26,7 +26,7 @@ app.use('/workflow', serve<{ message: string }>(async (context) => {
 
   const { body } = await context.call("get-data", {
     url: `${process.env.UPSTASH_WORKFLOW_URL ?? "http://localhost:3001"}/get-data`,
-    method: "POST",
+    method: "GET",
     body: { message: result1 }
   })
 
@@ -38,6 +38,11 @@ app.use('/workflow', serve<{ message: string }>(async (context) => {
   })
 }));
 
+app.get('/get-data', (req, res) => {
+  const message = req.body.message as string;
+  res.json({ message: `Received: ${message}` });
+});
+
 /**
  * ServeMany
  */
@@ -122,6 +127,6 @@ app.post("/ci", serve(
   }
 ))
 
-app.listen(3000, () => {
-  console.log('Server running on port 3000');
+app.listen(3001, () => {
+  console.log('Server running on port 3001');
 });

platforms/express.ts

@@ -65,6 +65,11 @@ function createExpressHandler<TInitialPayload = unknown, TResult = unknown>(
 
     const response = await serveHandler(webRequest);
 
+    // set headers
+    for (const [key, value] of response.headers.entries()) {
+      res.setHeader(key, value);
+    }
+
     res.status(response.status).json(await response.json());
   };
 }
@@ -77,7 +82,7 @@ export function serve<TInitialPayload = unknown, TResult = unknown>(
 
   const handler: RequestHandler = createExpressHandler([routeFunction, options]);
 
-  router.all("*", handler);
+  router.all(/(.*)/, handler);
 
   return router;
 }
@@ -108,6 +113,6 @@ export const serveMany = (
     options,
   });
 
-  router.all("*", handler);
+  router.all(/(.*)/, handler);
   return router;
 };

platforms/nextjs.ts

@@ -105,6 +105,12 @@ export const servePagesRouter = <TInitialPayload = unknown, TResult = unknown>(
       method: "POST",
     });
     const response = await serveHandler(request);
+
+    // set headers
+    for (const [key, value] of response.headers.entries()) {
+      res.setHeader(key, value);
+    }
+
     res.status(response.status).json(await response.json());
   };
 

src/agents/adapters.test.ts

@@ -81,7 +81,7 @@ describe("wrapTools", () => {
             headers: {
               "upstash-workflow-sdk-version": "1",
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-forward-upstash-workflow-sdk-version": "1",
               "upstash-method": "POST",
               "upstash-workflow-init": "false",
@@ -129,7 +129,7 @@ describe("wrapTools", () => {
             headers: {
               "upstash-workflow-sdk-version": "1",
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-forward-upstash-workflow-sdk-version": "1",
               "upstash-method": "POST",
               "upstash-workflow-init": "false",
@@ -198,7 +198,7 @@ describe("wrapTools", () => {
             headers: {
               "content-type": "application/json",
               "upstash-delay": "1000s",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-forward-upstash-workflow-sdk-version": "1",
               "upstash-method": "POST",
               "upstash-workflow-init": "false",
@@ -246,7 +246,7 @@ describe("wrapTools", () => {
             destination: WORKFLOW_ENDPOINT,
             headers: {
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-forward-upstash-workflow-sdk-version": "1",
               "upstash-method": "POST",
               "upstash-workflow-init": "false",

src/agents/agent.test.ts

@@ -98,7 +98,7 @@ describe("agents", () => {
                 "upstash-workflow-sdk-version": "1",
                 "content-type": "application/json",
                 "upstash-callback": "https://requestcatcher.com/api",
-                "upstash-callback-feature-set": "LazyFetch,InitialBody",
+                "upstash-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
                 "upstash-callback-forward-upstash-workflow-callback": "true",
                 "upstash-callback-forward-upstash-workflow-concurrent": "1",
                 "upstash-callback-forward-upstash-workflow-contenttype": "application/json",
@@ -165,7 +165,7 @@ describe("agents", () => {
                 "upstash-workflow-sdk-version": "1",
                 "content-type": "application/json",
                 "upstash-callback": "https://requestcatcher.com/api",
-                "upstash-callback-feature-set": "LazyFetch,InitialBody",
+                "upstash-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
                 "upstash-callback-forward-upstash-workflow-callback": "true",
                 "upstash-callback-forward-upstash-workflow-concurrent": "1",
                 "upstash-callback-forward-upstash-workflow-contenttype": "application/json",
@@ -231,7 +231,7 @@ describe("agents", () => {
               "upstash-workflow-sdk-version": "1",
               "content-type": "application/json",
               "upstash-callback": "https://requestcatcher.com/api",
-              "upstash-callback-feature-set": "LazyFetch,InitialBody",
+              "upstash-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-callback-forward-upstash-workflow-callback": "true",
               "upstash-callback-forward-upstash-workflow-concurrent": "1",
               "upstash-callback-forward-upstash-workflow-contenttype": "application/json",

src/agents/task.test.ts

@@ -115,7 +115,7 @@ describe("tasks", () => {
               "upstash-workflow-sdk-version": "1",
               "content-type": "application/json",
               "upstash-callback": "https://requestcatcher.com/api",
-              "upstash-callback-feature-set": "LazyFetch,InitialBody",
+              "upstash-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-callback-forward-upstash-workflow-callback": "true",
               "upstash-callback-forward-upstash-workflow-concurrent": "1",
               "upstash-callback-forward-upstash-workflow-contenttype": "application/json",
@@ -198,7 +198,7 @@ describe("tasks", () => {
               "upstash-workflow-sdk-version": "1",
               "content-type": "application/json",
               "upstash-callback": "https://requestcatcher.com/api",
-              "upstash-callback-feature-set": "LazyFetch,InitialBody",
+              "upstash-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-callback-forward-upstash-workflow-callback": "true",
               "upstash-callback-forward-upstash-workflow-concurrent": "1",
               "upstash-callback-forward-upstash-workflow-contenttype": "application/json",
@@ -273,7 +273,7 @@ describe("tasks", () => {
               "upstash-workflow-sdk-version": "1",
               "content-type": "application/json",
               "upstash-callback": "https://requestcatcher.com/api",
-              "upstash-callback-feature-set": "LazyFetch,InitialBody",
+              "upstash-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-callback-forward-upstash-workflow-callback": "true",
               "upstash-callback-forward-upstash-workflow-concurrent": "1",
               "upstash-callback-forward-upstash-workflow-contenttype": "application/json",

src/client/index.test.ts

@@ -234,7 +234,7 @@ describe("workflow client", () => {
               "upstash-workflow-url": "https://requestcatcher.com/api",
               "upstash-delay": "1s",
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-telemetry-framework": "unknown",
               "upstash-telemetry-runtime": expect.stringMatching(/bun@/),
               "upstash-telemetry-sdk": expect.stringContaining("@upstash/workflow"),
@@ -299,7 +299,7 @@ describe("workflow client", () => {
               "upstash-workflow-url": "https://requestcatcher.com/api",
               "upstash-delay": "1s",
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-telemetry-framework": "unknown",
               "upstash-telemetry-runtime": expect.stringMatching(/bun@/),
               "upstash-telemetry-sdk": expect.stringContaining("@upstash/workflow"),
@@ -319,13 +319,13 @@ describe("workflow client", () => {
               "upstash-workflow-url": "https://requestcatcher.com/api",
               "upstash-delay": "1s",
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-telemetry-framework": "unknown",
               "upstash-telemetry-runtime": expect.stringMatching(/bun@/),
               "upstash-telemetry-sdk": expect.stringContaining("@upstash/workflow"),
               "upstash-workflow-sdk-version": "1",
               "upstash-failure-callback": "https://requestcatcher.com/api",
-              "upstash-failure-callback-feature-set": "LazyFetch,InitialBody",
+              "upstash-failure-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-failure-callback-forward-upstash-workflow-failure-callback": "true",
               "upstash-failure-callback-forward-upstash-workflow-is-failure": "true",
               "upstash-failure-callback-forward-user-header": "user-header-value",
@@ -379,8 +379,8 @@ describe("workflow client", () => {
               "upstash-delay": "1s",
               "upstash-failure-callback": "https://requestcatcher.com/some-failure-callback",
               "content-type": "application/json",
-              "upstash-feature-set": "LazyFetch,InitialBody",
-              "upstash-failure-callback-feature-set": "LazyFetch,InitialBody",
+              "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
+              "upstash-failure-callback-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger",
               "upstash-failure-callback-forward-upstash-workflow-failure-callback": "true",
               "upstash-failure-callback-forward-upstash-workflow-is-failure": "true",
               "upstash-failure-callback-forward-user-header": "user-header-value",
@@ -668,6 +668,7 @@ describe("workflow client", () => {
                       errors: [
                         {
                           error: "400 Bad Request",
+                          body: expect.any(String),
                           headers: expect.any(Object),
                           status: 400,
                           time: expect.any(Number),

src/client/types.ts

@@ -145,8 +145,21 @@ export type StepLog = BaseStepLog &
   AsOptional<{ sleepUntil: number }> &
   AsOptional<WaitEventGroup>;
 
-type StepError = {
+export type StepError = {
+  /**
+   * error message associated with the request
+   *
+   * example:
+   * ```
+   * detected a non-workflow destination for trigger/invoke.
+   * make sure you are sending the request to the correct endpoint
+   * ```
+   */
   error: string;
+  /**
+   * response body returned in the request which resulted in an error
+   */
+  body: string;
   headers: Record<string, string[]>;
   status: number;
   time: number;