Getting there with hooks + update both restate sdk and vercel workflow

Author: slinkydeveloper

.gitignore

@@ -5,4 +5,5 @@ dist
 .DS_Store
 .next
 .swc
-.well-known
+.well-known
+.restate

package.json

@@ -34,19 +34,19 @@
   "devDependencies": {
     "@arethetypeswrong/cli": "^0.18.2",
     "@changesets/cli": "^2.29.7",
-    "@eslint/js": "^9.37.0",
+    "@eslint/js": "^9.38.0",
     "@eslint/json": "^0.13.2",
-    "@microsoft/api-extractor": "^7.53.1",
-    "@types/node": "^24.7.2",
-    "eslint": "^9.37.0",
+    "@microsoft/api-extractor": "^7.53.3",
+    "@types/node": "^24.9.1",
+    "eslint": "^9.38.0",
     "globals": "^16.4.0",
     "plop": "^4.0.4",
     "prettier": "^3.6.2",
-    "tsdown": "^0.15.6",
+    "tsdown": "^0.15.10",
     "tsx": "^4.20.6",
     "turbo": "^2.5.8",
     "typescript": "^5.9.3",
-    "typescript-eslint": "^8.46.0",
+    "typescript-eslint": "^8.46.2",
     "vitest": "^3.2.4"
   },
   "packageManager": "[email protected]",

packages/examples/workflow/src/workflows/user-signup.ts

@@ -1,5 +1,5 @@
 /* eslint-disable @typescript-eslint/require-await */
-import { sleep } from "workflow";
+import { createWebhook, sleep } from "workflow";
 import { FatalError } from "workflow";
 
 export async function handleUserSignup(email: string) {
@@ -9,7 +9,8 @@ export async function handleUserSignup(email: string) {
 
   await sendWelcomeEmail(user);
   await sleep("30s");
-  await sendOnboardingEmail(user);
+  const webhook = createWebhook();
+  await sendOnboardingEmail(user, webhook.url);
 
   return { userId: user.id, status: "onboarded" };
 }
@@ -33,7 +34,10 @@ async function sendWelcomeEmail(user: { id: string; email: string }) {
   }
 }
 
-async function sendOnboardingEmail(user: { id: string; email: string }) {
+async function sendOnboardingEmail(
+  user: { id: string; email: string },
+  url: string
+) {
   "use step";
 
   if (!user.email.includes("@")) {
@@ -42,4 +46,5 @@ async function sendOnboardingEmail(user: { id: string; email: string }) {
   }
 
   console.log(`Sending onboarding email to user: ${user.id}`);
+  console.log(`Complete it with webhook: ${url}`);
 }

packages/libs/backend/src/services.ts

@@ -23,40 +23,41 @@ import {
 
 import {
   type CancelWorkflowRunParams,
+  type CreateEventRequest,
+  type CreateHookRequest,
+  type CreateStepRequest,
   type CreateWorkflowRunRequest,
+  type Event,
+  EventSchema,
+  type GetHookParams,
+  type GetStepParams,
   type GetWorkflowRunParams,
+  type Hook,
+  HookSchema,
+  ListEventsByCorrelationIdParams,
+  type ListEventsParams,
   type ListWorkflowRunsParams,
+  ListWorkflowRunStepsParams,
   type PauseWorkflowRunParams,
   type ResumeWorkflowRunParams,
-  type UpdateWorkflowRunRequest,
-  type WorkflowRun,
   type Step,
-  type CreateStepRequest,
-  type GetStepParams,
+  StepSchema,
   type UpdateStepRequest,
-  type Hook,
-  type CreateHookRequest,
-  type GetHookParams,
-  type Event,
-  type CreateEventRequest,
-  type ListEventsParams,
-  ListWorkflowRunStepsParams,
-  ListEventsByCorrelationIdParams,
+  type UpdateWorkflowRunRequest,
+  type WorkflowRun,
   WorkflowRunSchema,
-  StepSchema,
-  EventSchema,
-  HookSchema,
 } from "@workflow/world";
 import {
   DEFAULT_RESOLVE_DATA_OPTION,
+  filterEventData,
+  filterHookData,
   filterRunData,
   filterStepData,
-  filterHookData,
-  filterEventData,
 } from "./utils.js";
 
 import { JsonTransport } from "@vercel/queue";
-import { QueueParamsSchema, serde } from "@restatedev/common";
+import { serde } from "@restatedev/restate-sdk-zod";
+import { schemas } from "@restatedev/common";
 
 // key by runId
 
@@ -591,14 +592,12 @@ export const indexService = service({
       async (ctx: Context, param: { token: string }): Promise<Hook> => {
         const hookTokens = (await ctx
           .objectClient(keyValue, param.token)
-          .get()) as { runId: string; hookId: string }[];
+          .get()) as string[];
 
         if (hookTokens === undefined || hookTokens.length === 0) {
           throw new TerminalError("No hooks found", { errorCode: 404 });
         }
-        const theHook = hookTokens[0];
-        const hook = await ctx.objectClient(hooksApi, theHook!.hookId).get({});
-        return hook;
+        return await ctx.objectClient(hooksApi, hookTokens[0]!).get({});
       }
     ),
 
@@ -626,7 +625,7 @@ export const queue = service({
           onMaxAttempts: "kill",
         },
 
-        input: serde.zod(QueueParamsSchema),
+        input: serde.zod(schemas.QueueParamsSchema),
       },
       async (ctx: Context, params) => {
         let pathname: string;
@@ -650,7 +649,7 @@ export const queue = service({
             headers: {
               "x-vqs-queue-name": params.queueName,
               "x-vqs-message-id": messageId,
-              "x-vqs-message-attempt": String(1), // TODO: fix this.
+              "x-vqs-message-attempt": String(params.attempt),
             },
           }
         );
@@ -661,11 +660,14 @@ export const queue = service({
         if (response.status === 503) {
           const { retryIn } = (await response.json()) as { retryIn: number };
 
+          // Increment attempt count
+          params.attempt += 1;
+
           ctx.serviceSendClient(queue).queue(
             params,
             rpc.sendOpts({
               delay: { seconds: retryIn },
-              input: serde.zod(QueueParamsSchema),
+              input: serde.zod(schemas.QueueParamsSchema),
             })
           );
 

packages/libs/common/package.json

@@ -28,7 +28,6 @@
     "lint": "eslint ."
   },
   "dependencies": {
-    "@restatedev/restate-sdk-core": "catalog:",
     "@workflow/world": "catalog:",
     "zod": "catalog:"
   },

packages/libs/common/src/index.ts

@@ -9,79 +9,24 @@
  * TODO: Add repository URL
  */
 
-import type { Serde } from "@restatedev/restate-sdk-core";
 import { QueuePayloadSchema } from "@workflow/world";
 
 import * as z4 from "zod/v4";
 
-export type { Serde } from "@restatedev/restate-sdk-core";
-
-class ZodSerde<T extends z4.ZodType> implements Serde<z4.infer<T>> {
-  contentType? = "application/json";
-  jsonSchema?: object | undefined;
-
-  constructor(private readonly schema: T) {
-    if ("_zod" in schema) {
-      this.jsonSchema = z4.toJSONSchema(schema, {
-        unrepresentable: "any",
-      });
-    }
-    if (schema instanceof z4.ZodVoid || schema instanceof z4.ZodUndefined) {
-      this.contentType = undefined;
-    }
-  }
-
-  serialize(value: z4.infer<T>): Uint8Array {
-    if (value === undefined) {
-      return new Uint8Array(0);
-    }
-    return new TextEncoder().encode(JSON.stringify(value));
-  }
-
-  deserialize(data: Uint8Array): z4.infer<T> {
-    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
-    const js =
-      data.length === 0
-        ? undefined
-        : JSON.parse(new TextDecoder().decode(data));
-    if (
-      "safeParse" in this.schema &&
-      typeof this.schema.safeParse === "function"
-    ) {
-      const res = this.schema.safeParse(js);
-      if (res.success) {
-        // eslint-disable-next-line @typescript-eslint/no-unsafe-return
-        return res.data;
-      }
-      throw res.error;
-    } else {
-      throw new TypeError("Unsupported data type. Expected 'safeParse'.");
-    }
-  }
+/** Bunch of zod schemas we need **/
+export namespace schemas {
+  export const QueueParamsSchema = z4.object({
+    deliverTo: z4.string(),
+    queueName: z4.string(),
+    attempt: z4.int(),
+    message: QueuePayloadSchema,
+    opts: z4
+      .object({
+        deploymentId: z4.string().optional(),
+        idempotencyKey: z4.string().optional(),
+      })
+      .optional(),
+  });
+
+  export type QueueParams = z4.infer<typeof QueueParamsSchema>;
 }
-
-export namespace serde {
-  /**
-   * A Zod-based serde.
-   *
-   * @param zodType the zod type
-   * @returns a serde that will validate the data with the zod schema
-   */
-  export const zod = <T extends z4.ZodType>(zodType: T): Serde<z4.infer<T>> => {
-    return new ZodSerde(zodType);
-  };
-}
-
-export const QueueParamsSchema = z4.object({
-  deliverTo: z4.string(),
-  queueName: z4.string(),
-  message: QueuePayloadSchema,
-  opts: z4
-    .object({
-      deploymentId: z4.string().optional(),
-      idempotencyKey: z4.string().optional(),
-    })
-    .optional(),
-});
-
-export type QueueParams = z4.infer<typeof QueueParamsSchema>;

packages/libs/world/src/auth.ts

@@ -14,7 +14,6 @@ import { World } from "@workflow/world";
 import { createQueue } from "./queue.js";
 import { createStorage } from "./storage.js";
 import { createStreamer } from "./streamer.js";
-import { auth } from "./auth.js";
 
 export function createWorld(args?: {
   opts: ConnectionOpts;
@@ -28,7 +27,6 @@ export function createWorld(args?: {
     ...createQueue(client, deliverTo),
     ...createStorage(client),
     ...createStreamer(client),
-    ...auth,
     start: async () => {
       // TODO: verify subscription
     },

packages/libs/world/src/index.ts

@@ -2,8 +2,9 @@ import { MessageId, Queue, ValidQueueName } from "@workflow/world";
 import { JsonTransport } from "@vercel/queue";
 import z from "zod/v4";
 import { Ingress, rpc } from "@restatedev/restate-sdk-clients";
+import { serde } from "@restatedev/restate-sdk-zod";
 import { QueueService } from "@restatedev/backend";
-import { QueueParamsSchema, serde } from "@restatedev/common";
+import { schemas } from "@restatedev/common";
 
 const HeaderParser = z.object({
   "x-vqs-queue-name": ValidQueueName,
@@ -22,8 +23,12 @@ export function createQueue(client: Ingress, deliverTo: string): Queue {
           deliverTo,
           queueName: name,
           message: body,
+          attempt: 1,
         },
-        rpc.sendOpts({ idempotencyKey, input: serde.zod(QueueParamsSchema) })
+        rpc.sendOpts({
+          idempotencyKey,
+          input: serde.zod(schemas.QueueParamsSchema),
+        })
       );
 
     const messageId = res.invocationId as MessageId;
@@ -34,7 +39,6 @@ export function createQueue(client: Ingress, deliverTo: string): Queue {
   const createQueueHandler: Queue["createQueueHandler"] = (prefix, handler) => {
     return async (req) => {
       const headers = HeaderParser.safeParse(Object.fromEntries(req.headers));
-
       if (!headers.success || !req.body) {
         return Response.json(
           { error: "Missing required headers" },
@@ -52,16 +56,20 @@ export function createQueue(client: Ingress, deliverTo: string): Queue {
 
       const body = await new JsonTransport().deserialize(req.body);
       try {
+        // Run the queue handler
         const response = await handler(body, { attempt, queueName, messageId });
         const retryIn =
           typeof response === "undefined" ? null : response.timeoutSeconds;
 
+        // 503 is used to notify to retry later
         if (retryIn) {
           return Response.json({ retryIn }, { status: 503 });
         }
 
+        // Processed
         return Response.json({ ok: true });
       } catch (error) {
+        // Some error happened
         return Response.json(String(error), { status: 500 });
       }
     };