Add useJSONContent option for express and sveltekit

examples/express/index.ts

@@ -34,12 +34,94 @@ app.use('/workflow', serve<{ message: string }>(async (context) => {
     console.log('step 2 input', result1, 'output', output)
     return output
   })
+}, {
+  verbose: true,
+  retries: 0
 }));
 
 app.post("/get-data", (req, res) => {
   res.send("hey there");
 });
 
+// here we are
+
+app.use(
+  "/test",
+  serveQstashWorkflow(async (context: Parameters<RouteFunction<unknown>>[0]) => {
+    await context.run("test", async () => {
+      // console.log("test");
+    });
+  }, {
+    useJSONContent: true
+  })
+);
+
+import { IncomingHttpHeaders } from "http";
+import { RouteFunction, serve as basicServe, WorkflowServeOptions } from "@upstash/workflow";
+import { Request as ExpressRequest, Response, Router } from "express";
+
+function transformHeaders(headers: IncomingHttpHeaders): [string, string][] {
+return Object.entries(headers).map(([key, value]) => [
+  key,
+  Array.isArray(value) ? value.join(", ") : value ?? "",
+]);
+}
+
+export function serveQstashWorkflow<TInitialPayload = unknown>(
+routeFunction: RouteFunction<TInitialPayload>,
+options?: Omit<WorkflowServeOptions<globalThis.Response, TInitialPayload>, "onStepFinish">
+): Router {
+const router = express.Router();
+
+router.post("*", async (req: ExpressRequest, res: Response) => {
+  const protocol = (req.headers["x-forwarded-proto"] as string) || req.protocol;
+  const host = req.headers.host;
+  const url = `${protocol}://${host}${req.originalUrl}`;
+  const headers = transformHeaders(req.headers);
+
+  let reqBody: string | undefined;
+
+  if (req.headers["content-type"]?.includes("text/plain")) {
+    reqBody = req.body;
+  } else if (req.headers["content-type"]?.includes("application/json")) {
+    reqBody = JSON.stringify(req.body);
+  }
+
+  const request = new Request(url, {
+    headers: headers,
+    body: reqBody || "{}",
+    method: "POST",
+  });
+
+  const { handler: serveHandler } = basicServe<TInitialPayload>(routeFunction, options);
+
+  try {
+    const response = await serveHandler(request);
+
+    // Set status code
+    res.status(response.status);
+
+    // Set headers
+    response.headers.forEach((value, key) => {
+      res.setHeader(key, value);
+    });
+
+    // Send body
+    if (response.body) {
+      const buffer = await response.arrayBuffer();
+      res.send(Buffer.from(buffer));
+    } else {
+      res.end();
+    }
+  } catch (error) {
+    console.error("Error in workflow handler:", error);
+    res.status(500).json({ error: "Internal Server Error" });
+  }
+});
+
+return router;
+}
+
 app.listen(3001, () => {
   console.log('Server running on port 3001');
 });

examples/nextjs/app/path/route.ts

@@ -1,3 +1,5 @@
+export const runtime = 'nodejs';
+
 import { serve } from '@upstash/workflow/nextjs'
 
 const someWork = (input: string) => {
@@ -7,13 +9,7 @@ const someWork = (input: string) => {
 export const { POST } = serve<string>(async (context) => {
   const input = context.requestPayload
   const result1 = await context.run('step1', async () => {
-    const output = someWork(input)
-    console.log('step 1 input', input, 'output', output)
-    return output
-  })
-
-  await context.run('step2', async () => {
-    const output = someWork(result1)
-    console.log('step 2 input', result1, 'output', output)
+    console.log(someWork(input));
   })
+}, {
 })

platforms/astro.ts

@@ -1,14 +1,14 @@
 import type { APIContext, APIRoute } from "astro";
 
-import type { WorkflowServeOptions, WorkflowContext } from "../src";
+import type { PublicServeOptions, WorkflowContext } from "../src";
 import { serve as serveBase } from "../src";
 
 export function serve<TInitialPayload = unknown>(
   routeFunction: (
     workflowContext: WorkflowContext<TInitialPayload>,
     apiContext: APIContext
   ) => Promise<void>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ) {
   const POST: APIRoute = (apiContext) => {
     const { handler } = serveBase<TInitialPayload>(

platforms/cloudflare.ts

@@ -1,4 +1,4 @@
-import type { RouteFunction, WorkflowServeOptions } from "../src";
+import type { PublicServeOptions, RouteFunction } from "../src";
 import { serve as serveBase } from "../src";
 
 export type WorkflowBindings = {
@@ -58,7 +58,7 @@ const getArgs = (
  */
 export const serve = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ): { fetch: (...args: PagesHandlerArgs | WorkersHandlerArgs) => Promise<Response> } => {
   const fetch = async (...args: PagesHandlerArgs | WorkersHandlerArgs) => {
     const { request, env } = getArgs(args);

platforms/express.ts

@@ -20,12 +20,18 @@ export function serve<TInitialPayload = unknown>(
       return;
     }
 
-    // only allow application/json content type
-    if (request_.headers["content-type"] !== "application/json") {
-      res.status(400).json("Only application/json content type is allowed in express workflows");
-      return;
+    let requestBody: string;
+    if (request_.headers["content-type"]?.includes("text/plain")) {
+      requestBody = request_.body;
+    } else if (request_.headers["content-type"]?.includes("application/json")) {
+      requestBody = JSON.stringify(request_.body);
+    } else {
+      requestBody =
+        typeof request_.body === "string" ? request_.body : JSON.stringify(request_.body);
     }
 
+    console.log(request_.headers, request_.body);
+
     // create Request
     const protocol = request_.protocol;
     const host = request_.get("host") || "localhost";
@@ -34,13 +40,16 @@ export function serve<TInitialPayload = unknown>(
     const webRequest = new Request(url, {
       method: request_.method,
       headers: new Headers(request_.headers as Record<string, string>),
-      body: JSON.stringify(request_.body),
+      body: requestBody,
     });
 
     // create handler
     const { handler: serveHandler } = serveBase<TInitialPayload>(
       (workflowContext) => routeFunction(workflowContext),
-      options
+      {
+        ...options,
+        useJSONContent: true,
+      }
     );
 
     const response = await serveHandler(webRequest);

platforms/h3.ts

@@ -1,6 +1,6 @@
 import { defineEventHandler, readRawBody } from "h3";
 
-import type { RouteFunction, WorkflowServeOptions } from "../src";
+import type { PublicServeOptions, RouteFunction } from "../src";
 import { serve as serveBase } from "../src";
 import type { IncomingHttpHeaders } from "node:http";
 
@@ -14,7 +14,7 @@ function transformHeaders(headers: IncomingHttpHeaders): [string, string][] {
 
 export const serve = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ) => {
   const handler = defineEventHandler(async (event) => {
     const method = event.node.req.method;

platforms/hono.ts

@@ -1,5 +1,5 @@
 import type { Context } from "hono";
-import type { RouteFunction, WorkflowServeOptions } from "../src";
+import type { PublicServeOptions, RouteFunction } from "../src";
 import { serve as serveBase } from "../src";
 import { Variables } from "hono/types";
 
@@ -26,7 +26,7 @@ export const serve = <
   TVariables extends Variables = Variables,
 >(
   routeFunction: RouteFunction<TInitialPayload>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ): ((context: Context<{ Bindings: TBindings; Variables: TVariables }>) => Promise<Response>) => {
   const handler = async (context: Context<{ Bindings: TBindings; Variables: TVariables }>) => {
     const environment = context.env;

platforms/nextjs.ts

@@ -2,7 +2,7 @@
 /* eslint-disable @typescript-eslint/ban-ts-comment */
 import type { NextApiHandler, NextApiRequest, NextApiResponse } from "next";
 
-import type { WorkflowServeOptions, RouteFunction } from "../src";
+import type { RouteFunction, PublicServeOptions } from "../src";
 import { serve as serveBase } from "../src";
 
 /**
@@ -16,7 +16,7 @@ import { serve as serveBase } from "../src";
  */
 export const serve = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ): { POST: (request: Request) => Promise<Response> } => {
   const { handler: serveHandler } = serveBase<TInitialPayload, Request, Response>(
     routeFunction,
@@ -32,7 +32,7 @@ export const serve = <TInitialPayload = unknown>(
 
 export const servePagesRouter = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ): { handler: NextApiHandler } => {
   const { handler: serveHandler } = serveBase(routeFunction, options);
 

platforms/solidjs.ts

@@ -1,6 +1,6 @@
 import type { APIEvent } from "@solidjs/start/server";
 
-import type { RouteFunction, WorkflowServeOptions } from "../src";
+import type { PublicServeOptions, RouteFunction } from "../src";
 import { serve as serveBase } from "../src";
 
 /**
@@ -14,7 +14,7 @@ import { serve as serveBase } from "../src";
  */
 export const serve = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
-  options?: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish">
+  options?: PublicServeOptions<TInitialPayload>
 ) => {
   // Create a handler which receives an event and calls the
   // serveBase method

platforms/svelte.ts

@@ -1,6 +1,6 @@
 import type { RequestHandler } from "@sveltejs/kit";
 
-import type { RouteFunction, WorkflowServeOptions } from "../src";
+import type { PublicServeOptions, RouteFunction } from "../src";
 import { serve as serveBase } from "../src";
 
 /**
@@ -14,12 +14,15 @@ import { serve as serveBase } from "../src";
  */
 export const serve = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
-  options: Omit<WorkflowServeOptions<Response, TInitialPayload>, "onStepFinish"> & {
-    env: WorkflowServeOptions["env"];
+  options: PublicServeOptions<TInitialPayload> & {
+    env: PublicServeOptions["env"]; // make env required
   }
 ): { POST: RequestHandler } => {
   const handler: RequestHandler = async ({ request }) => {
-    const { handler: serveHandler } = serveBase<TInitialPayload>(routeFunction, options);
+    const { handler: serveHandler } = serveBase<TInitialPayload>(routeFunction, {
+      ...options,
+      useJSONContent: true,
+    });
     return await serveHandler(request);
   };
 

src/serve/index.ts

@@ -44,6 +44,7 @@ export const serve = <
     baseUrl,
     env,
     retries,
+    useJSONContent,
   } = processOptions<TResponse, TInitialPayload>(options);
   const debug = WorkflowLogger.getLogger(verbose);
 
@@ -161,7 +162,7 @@ export const serve = <
     } else if (callReturnCheck.value === "continue-workflow") {
       // request is not third party call. Continue workflow as usual
       const result = isFirstInvocation
-        ? await triggerFirstInvocation(workflowContext, retries, debug)
+        ? await triggerFirstInvocation(workflowContext, retries, useJSONContent, debug)
         : await triggerRouteFunction({
             onStep: async () => routeFunction(workflowContext),
             onCleanup: async () => {

src/serve/options.ts

@@ -80,6 +80,7 @@ export const processOptions = <TResponse extends Response = Response, TInitialPa
     baseUrl: environment.UPSTASH_WORKFLOW_URL,
     env: environment,
     retries: DEFAULT_RETRIES,
+    useJSONContent: false,
     ...options,
   };
 };

src/types.ts

@@ -129,6 +129,7 @@ export type FinishCondition =
   | "fromCallback"
   | "auth-fail"
   | "failure-callback";
+
 export type WorkflowServeOptions<
   TResponse extends Response = Response,
   TInitialPayload = unknown,
@@ -218,8 +219,20 @@ export type WorkflowServeOptions<
    * 3 by default
    */
   retries?: number;
+  /**
+   * Whether the framework should use `content-type: application/json`
+   * in `triggerFirstInvocation`.
+   *
+   * Not part of the public API.
+   */
+  useJSONContent?: boolean;
 };
 
+export type PublicServeOptions<
+  TInitialPayload = unknown,
+  TResponse extends Response = Response,
+> = Omit<WorkflowServeOptions<TResponse, TInitialPayload>, "onStepFinish" | "useJSONContent">;
+
 /**
  * Payload passed as body in failureFunction
  */

src/workflow-requests.test.ts

@@ -563,9 +563,9 @@ describe("Workflow Requests", () => {
           url: WORKFLOW_ENDPOINT,
         });
 
-      await triggerFirstInvocation(context, 3);
-      const debug = new WorkflowLogger({ logLevel: "INFO", logOutput: "console" });
-      const spy = spyOn(debug, "log");
+        await triggerFirstInvocation(context, 3);
+        const debug = new WorkflowLogger({ logLevel: "INFO", logOutput: "console" });
+        const spy = spyOn(debug, "log");
 
         const firstDelete = await triggerWorkflowDelete(context, debug);
         expect(firstDelete).toEqual({ deleted: true });
@@ -606,7 +606,7 @@ describe("Workflow Requests", () => {
         const debug = new WorkflowLogger({ logLevel: "INFO", logOutput: "console" });
         const spy = spyOn(debug, "log");
 
-        await triggerFirstInvocation(context, 3, debug);
+        await triggerFirstInvocation(context, 3, false, debug);
         expect(spy).toHaveBeenCalledTimes(1);
 
         await workflowClient.cancel({ ids: [workflowRunId] });
@@ -657,7 +657,7 @@ describe("Workflow Requests", () => {
         const debug = new WorkflowLogger({ logLevel: "INFO", logOutput: "console" });
         const spy = spyOn(debug, "log");
 
-        await triggerFirstInvocation(context, 3, debug);
+        await triggerFirstInvocation(context, 3, false, debug);
         expect(spy).toHaveBeenCalledTimes(1);
 
         await workflowClient.cancel({ ids: [workflowRunId] });
@@ -707,14 +707,14 @@ describe("Workflow Requests", () => {
         const debug = new WorkflowLogger({ logLevel: "INFO", logOutput: "console" });
         const spy = spyOn(debug, "log");
 
-        const resultOne = await triggerFirstInvocation(context, 3, debug);
+        const resultOne = await triggerFirstInvocation(context, 3, false, debug);
         expect(resultOne.isOk()).toBeTrue();
         // @ts-expect-error value will exist because of isOk
         expect(resultOne.value).toBe("success");
 
         expect(spy).toHaveBeenCalledTimes(1);
 
-        const resultTwo = await triggerFirstInvocation(context, 0, debug);
+        const resultTwo = await triggerFirstInvocation(context, 0, false, debug);
         expect(resultTwo.isOk()).toBeTrue();
         // @ts-expect-error value will exist because of isOk
         expect(resultTwo.value).toBe("workflow-run-already-exists");