zod

.eslintrc.cjs

@@ -44,6 +44,9 @@ const config = {
     "@typescript-eslint/consistent-indexed-object-style": "off",
     "@typescript-eslint/prefer-nullish-coalescing": "off",
     "prefer-const": "off",
+    "@typescript-eslint/no-unnecessary-type-assertion": "warn",
+    "@typescript-eslint/consistent-generic-constructors": "warn",
+    "@typescript-eslint/no-namespace": "warn",
   },
 };
 

package.json

@@ -86,6 +86,7 @@
   },
   "devDependencies": {
     "@biomejs/biome": "1.7.1",
+    "@faker-js/faker": "^8.4.1",
     "@next/eslint-plugin-next": "^14.2.2",
     "@types/bun": "^1.1.0",
     "@types/eslint": "^8.56.9",

src/app/api/user/[uid]/workflows/route.ts

@@ -35,7 +35,7 @@ export async function GET(
       createdAt: createdAt?.getTime(),
       lastRunAt: workflowRuns[0]?.startedAt?.getTime(),
     }),
-  ) as WorkflowResponse[];
+  ) as Workflow.WorkflowResponse[];
 
   log.info(`Returning workflows for user ${session.user.id}`);
   return NextResponse.json(res);

src/app/api/workflow/[id]/run/route.ts

@@ -6,6 +6,7 @@ import { getAccessTokenFromUserId } from "@/server/db/helper";
 import { getServerSession } from "next-auth";
 import { type NextRequest, NextResponse } from "next/server";
 import { createWorkflowQueue } from "~/lib/workflow/utils/workflowQueue";
+import { WorkflowObjectSchema } from "~/schemas";
 
 const log = new Logger("/api/workflow/[id]/run");
 export async function POST(
@@ -51,18 +52,28 @@ export async function POST(
     slug: session.user.id,
     access_token: accessToken,
   });
-  const workflowObj = JSON.parse(workflow.workflow) as WorkflowObject;
+  const workflowParsed = WorkflowObjectSchema.safeParse(
+    (await JSON.parse(workflow.workflow)) as any,
+  );
 
-  workflowObj.operations = runner.sortOperations(workflowObj);
-  workflowObj.dryrun = dryrun;
+  if (!workflowParsed.success) {
+    log.error("Invalid workflow", {
+      errors: workflowParsed.error,
+      workflowId: workflow.id,
+      userId: session.user.id,
+    } as any);
+    return NextResponse.json({ error: "Invalid workflow" }, { status: 400 });
+  }
+
+  workflowParsed.data.operations = runner.sortOperations(workflowParsed.data);
+  workflowParsed.data.dryrun = dryrun;
   if (dryrun) {
     log.info("Dryrun mode enabled");
   }
-  runner.validateWorkflow(workflowObj);
 
   try {
     const job = await createWorkflowQueue(
-      workflowObj,
+      workflowParsed.data,
       session.user.id,
       workflow.id,
     );

src/app/api/workflow/route.ts

@@ -8,8 +8,7 @@ import {
 import { getServerSession } from "next-auth";
 import { type NextRequest, NextResponse } from "next/server";
 import { v4 as uuidv4 } from "uuid";
-import { Runner } from "~/lib/workflow/Workflow";
-import { getAccessTokenFromUserId } from "~/server/db/helper";
+import { WorkflowObjectSchema } from "~/schemas";
 
 const log = new Logger("/api/workflow");
 
@@ -23,54 +22,19 @@ export async function POST(request: NextRequest) {
       { status: 401 },
     );
   }
-  const accessToken = await getAccessTokenFromUserId(session.user.id);
-  if (!accessToken) {
-    return NextResponse.json(
-      {
-        error: "Something went wrong, unable to get access token",
-      },
-      { status: 500 },
-    );
-  }
-
-  log.info("Received workflow from user", session.user.id);
-  let workflowRes: WorkflowResponse;
-  try {
-    workflowRes = (await request.json()) as WorkflowResponse;
-  } catch (err) {
-    log.error("Error parsing workflow", err);
-    return NextResponse.json(
-      { error: "Error parsing workflow: " + (err as Error).message },
-      { status: 400 },
-    );
-  }
-  const runner = new Runner({
-    slug: session.user.id,
-    access_token: accessToken,
-  });
-
-  const workflow = workflowRes.workflow;
-
-  if (!workflow) {
-    return NextResponse.json(
-      { error: "No workflow provided" },
-      { status: 400 },
-    );
-  }
 
-  const operations = runner.sortOperations(workflow);
-  workflow.operations = operations;
-  const [valid, errors] = await runner.validateWorkflow(workflow);
+  const workflow = (await request.json()).workflow;
+  const workflowParsed = WorkflowObjectSchema.safeParse(workflow);
 
-  if (!valid) {
+  if (!workflowParsed.success) {
     return NextResponse.json(
-      { error: "Invalid workflow", errors },
+      { error: "Invalid workflow", errors: workflow.error },
       { status: 400 },
     );
   }
 
   const job = {
-    id: workflow.id ?? uuidv4(),
+    id: workflowParsed.data.id ?? uuidv4(),
     data: {
       workflow,
     },

src/app/api/workflow/validate/route.ts

@@ -1,29 +1,12 @@
 import { Logger } from "@/lib/log";
 import { authOptions } from "@/server/auth";
+import { WorkflowObjectSchema } from "@schema";
 import { getServerSession } from "next-auth";
 import { type NextRequest, NextResponse } from "next/server";
-import { Runner } from "~/lib/workflow/Workflow";
 import { getAccessTokenFromUserId } from "~/server/db/helper";
 
 const log = new Logger("/api/workflow/validate");
 
-function parseErrors(errorMessage: string) {
-  const errorType = errorMessage.substring(0, errorMessage.indexOf(":"));
-  const operation = errorMessage
-    .substring(errorMessage.indexOf(":") + 1)
-    .trim();
-  let operationObj = {};
-  try {
-    operationObj = operation ? JSON.parse(operation) : undefined;
-  } catch (_e) {
-    operationObj = operation;
-  }
-  return {
-    errorType: errorType.replace("Invalid ", ""),
-    operation: operationObj,
-  };
-}
-
 export async function POST(request: NextRequest) {
   const session = await getServerSession({ req: request, ...authOptions });
   if (!session) {
@@ -36,32 +19,9 @@ export async function POST(request: NextRequest) {
 
   log.info("Received workflow from user", session.user.id);
 
-  let workflow: WorkflowObject;
-  try {
-    workflow = (await request.json()) as WorkflowObject;
-  } catch (err) {
-    log.error("Error parsing workflow", err);
-    return NextResponse.json(
-      { error: "Error parsing workflow: " + (err as Error).message },
-      { status: 400 },
-    );
-  }
-  const runner = new Runner({
-    slug: session.user.id,
-    access_token: accessToken,
-  });
-
-  let res: any;
-  try {
-    const [valid, errors] = await runner.validateWorkflow(workflow);
-    res = { valid, errors: errors?.map(parseErrors) };
-  } catch (err) {
-    const errorMessage = (err as Error).message;
-    const prettyErrors = parseErrors(errorMessage);
+  // workflow = (await request.json()) as Workflow.WorkflowObject;
 
-    res = { valid: false, errors: prettyErrors };
-    return NextResponse.json(res, { status: 500 });
-  }
+  const workflow = WorkflowObjectSchema.safeParse(await request.json());
 
-  return NextResponse.json(res);
+  return NextResponse.json(workflow);
 }

src/app/states/store.ts

@@ -27,7 +27,7 @@ type RFState = {
     id?: string | null;
     name: string;
     description: string;
-    workflow?: WorkflowObject;
+    workflow?: Workflow.WorkflowObject;
     cron?: string;
     dryrun?: boolean;
   };
@@ -75,7 +75,7 @@ type RFState = {
     id?: string | null;
     name: string;
     description: string;
-    workflow?: WorkflowObject;
+    workflow?: Workflow.WorkflowObject;
     cron?: string;
     dryrun?: boolean;
   }) => void;
@@ -275,8 +275,8 @@ const useStore = create<RFState>((set, get) => ({
 }));
 
 type WorkflowRunState = {
-  workflowRun: QueueResponse | null;
-  setWorkflowRun: (workflowRun: QueueResponse) => void;
+  workflowRun: Workflow.QueueResponse | null;
+  setWorkflowRun: (workflowRun: Workflow.QueueResponse) => void;
   resetWorkflowRun: () => void;
 };
 
@@ -294,5 +294,5 @@ export const workflowRunStore = create<WorkflowRunState>((set, get) => ({
     });
   },
 }));
-  
+
 export default useStore;

src/app/utils/reactFlowToWorkflow.ts

@@ -58,7 +58,7 @@ export default async function reactFlowToWorkflow({
   nodes,
   edges,
 }: ReactFlowToWorkflowInput): Promise<{
-  workflowResponse: WorkflowResponse;
+  workflowResponse: Workflow.WorkflowResponse;
   errors: any;
 }> {
   const flowState = useStore.getState().flowState;
@@ -93,12 +93,12 @@ export default async function reactFlowToWorkflow({
 
   console.info("workflow", workflowObject);
 
-  const workflowResponse: WorkflowResponse = {
+  const workflowResponse: Workflow.WorkflowResponse = {
     id: workflowObject.id,
     name: workflowObject.name,
     workflow: workflowObject,
     cron: flowState.cron,
-  } as WorkflowResponse;
+  } as Workflow.WorkflowResponse;
 
   return { workflowResponse, errors };
 }

src/app/utils/runWorkflow.ts

@@ -1,7 +1,7 @@
 import { toast } from "sonner";
 import useStore, { workflowRunStore } from "../states/store";
 
-export async function runWorkflow(workflow: WorkflowResponse) {
+export async function runWorkflow(workflow: Workflow.WorkflowResponse) {
   if (!workflow.id) {
     throw new Error("Workflow ID is undefined");
   }
@@ -44,7 +44,7 @@ export async function runWorkflow(workflow: WorkflowResponse) {
         isRequesting = true;
         fetch(`/api/workflow/queue/${id}`)
           .then((res) => res.json())
-          .then((data: QueueResponse) => {
+          .then((data: Workflow.QueueResponse) => {
             isRequesting = false;
             if (data.error) {
               clearInterval(interval);

src/app/utils/validateWorkflow.ts

@@ -1,14 +1,10 @@
 import { toast } from "sonner";
-
-type validationResponse = {
-  valid: boolean;
-  errors: {
-    errorType: string;
-    operation: string;
-  }[];
-};
+import { ZodError } from "zod";
+type validationResponse =
+  | { success: true; data: Workflow.WorkflowObject }
+  | { success: false; error: ZodError };
 export async function requestValidateWorkflow(
-  workflow: WorkflowObject,
+  workflow: Workflow.WorkflowObject,
 ): Promise<validationResponse> {
   const data = await fetch("/api/workflow/validate", {
     method: "POST",
@@ -17,21 +13,23 @@ export async function requestValidateWorkflow(
     },
     body: JSON.stringify(workflow),
   });
-  const json = await data.json();
+  const json = (await data.json()) as validationResponse;
   return json;
 }
 
-export async function validateWorkflow(workflow: WorkflowObject) {
-  const validatePromise = requestValidateWorkflow(workflow).then((result) => {
-    if (result.errors.length > 0 || !result.valid) {
-      result.errors.forEach((error) => {
-        toast.error(`Error Type: ${error.errorType}`, {
-          description: `Operation: ${JSON.stringify(error.operation, null, 2)}`,
-        });
-      });
-      throw new Error(result.errors.map((error) => error.errorType).join(", "));
+export async function validateWorkflow(workflow: Workflow.WorkflowObject) {
+  const validatePromise = requestValidateWorkflow(workflow).then((response) => {
+    if (response.success) {
+      return { valid: true, errors: [] };
+    } else {
+      return {
+        valid: false,
+        errors: response.error.errors.map((error) => ({
+          errorType: error.message,
+          operation: error.path.join("."),
+        })),
+      };
     }
-    return result;
   });
 
   toast.promise(validatePromise, {