diff --git a/.changeset/strong-seas-impress.md b/.changeset/strong-seas-impress.md new file mode 100644 index 00000000000..2f2aa6f54c8 --- /dev/null +++ b/.changeset/strong-seas-impress.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/core": patch +--- + +Decouple zod diff --git a/apps/webapp/app/components/jobs/JobsTable.tsx b/apps/webapp/app/components/jobs/JobsTable.tsx index 4884e388ee4..8f2e4fb84c2 100644 --- a/apps/webapp/app/components/jobs/JobsTable.tsx +++ b/apps/webapp/app/components/jobs/JobsTable.tsx @@ -198,6 +198,7 @@ function classForJobStatus(status: JobRunStatus) { case "WAITING_ON_CONNECTIONS": case "PENDING": case "UNRESOLVED_AUTH": + case "INVALID_PAYLOAD": return "text-rose-500"; default: return ""; diff --git a/apps/webapp/app/components/run/RunCompletedDetail.tsx b/apps/webapp/app/components/run/RunCompletedDetail.tsx index 5d1641c9e2a..e35396a727c 100644 --- a/apps/webapp/app/components/run/RunCompletedDetail.tsx +++ b/apps/webapp/app/components/run/RunCompletedDetail.tsx @@ -2,7 +2,7 @@ import { CodeBlock } from "~/components/code/CodeBlock"; import { DateTime } from "~/components/primitives/DateTime"; import { Paragraph } from "~/components/primitives/Paragraph"; import { RunStatusIcon, RunStatusLabel } from "~/components/runs/RunStatuses"; -import { MatchedRun, useRun } from "~/hooks/useRun"; +import { MatchedRun } from "~/hooks/useRun"; import { formatDuration } from "~/utils"; import { RunPanel, diff --git a/apps/webapp/app/components/runs/RunStatuses.tsx b/apps/webapp/app/components/runs/RunStatuses.tsx index 4628a1ba446..828fe50b475 100644 --- a/apps/webapp/app/components/runs/RunStatuses.tsx +++ b/apps/webapp/app/components/runs/RunStatuses.tsx @@ -17,7 +17,8 @@ export function hasFinished(status: JobRunStatus): boolean { status === "ABORTED" || status === "TIMED_OUT" || status === "CANCELED" || - status === "UNRESOLVED_AUTH" + status === "UNRESOLVED_AUTH" || + status === "INVALID_PAYLOAD" ); } @@ -49,6 +50,7 @@ export function RunStatusIcon({ status, className }: { status: JobRunStatus; cla case "TIMED_OUT": return ; case "UNRESOLVED_AUTH": + case "INVALID_PAYLOAD": return ; case "WAITING_ON_CONNECTIONS": return ; @@ -77,6 +79,7 @@ export function runBasicStatus(status: JobRunStatus): RunBasicStatus { case "UNRESOLVED_AUTH": case "CANCELED": case "ABORTED": + case "INVALID_PAYLOAD": return "FAILED"; case "SUCCESS": return "COMPLETED"; @@ -111,6 +114,8 @@ export function runStatusTitle(status: JobRunStatus): string { return "Canceled"; case "UNRESOLVED_AUTH": return "Unresolved auth"; + case "INVALID_PAYLOAD": + return "Invalid payload"; default: { const _exhaustiveCheck: never = status; throw new Error(`Non-exhaustive match for value: ${status}`); @@ -130,6 +135,7 @@ export function runStatusClassNameColor(status: JobRunStatus): string { return "text-amber-300"; case "FAILURE": case "UNRESOLVED_AUTH": + case "INVALID_PAYLOAD": return "text-rose-500"; case "TIMED_OUT": return "text-amber-300"; diff --git a/apps/webapp/app/services/runs/performRunExecutionV1.server.ts b/apps/webapp/app/services/runs/performRunExecutionV1.server.ts index 016d71477ff..591f6f00b1b 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV1.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV1.server.ts @@ -1,6 +1,7 @@ import { CachedTaskSchema, RunJobError, + RunJobInvalidPayloadError, RunJobResumeWithTask, RunJobRetryWithTask, RunJobSuccess, @@ -348,6 +349,11 @@ export class PerformRunExecutionV1Service { break; } + case "INVALID_PAYLOAD": { + await this.#failRunWithInvalidPayloadError(execution, safeBody.data); + + break; + } default: { const _exhaustiveCheck: never = status; throw new Error(`Non-exhaustive match for value: ${status}`); @@ -453,6 +459,15 @@ export class PerformRunExecutionV1Service { }); } + async #failRunWithInvalidPayloadError( + execution: FoundRunExecution, + data: RunJobInvalidPayloadError + ) { + return await $transaction(this.#prismaClient, async (tx) => { + await this.#failRunExecution(tx, execution, data.errors, "INVALID_PAYLOAD"); + }); + } + async #retryRunWithTask(execution: FoundRunExecution, data: RunJobRetryWithTask) { const { run } = execution; @@ -572,7 +587,7 @@ export class PerformRunExecutionV1Service { prisma: PrismaClientOrTransaction, execution: FoundRunExecution, output: Record, - status: "FAILURE" | "ABORTED" | "UNRESOLVED_AUTH" = "FAILURE" + status: "FAILURE" | "ABORTED" | "UNRESOLVED_AUTH" | "INVALID_PAYLOAD" = "FAILURE" ): Promise { const { run } = execution; diff --git a/apps/webapp/app/services/runs/performRunExecutionV2.server.ts b/apps/webapp/app/services/runs/performRunExecutionV2.server.ts index 7d5f979c2c0..e6ae5ad30af 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV2.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV2.server.ts @@ -1,6 +1,7 @@ import { CachedTask, RunJobError, + RunJobInvalidPayloadError, RunJobResumeWithTask, RunJobRetryWithTask, RunJobSuccess, @@ -360,6 +361,11 @@ export class PerformRunExecutionV2Service { break; } + case "INVALID_PAYLOAD": { + await this.#failRunWithInvalidPayloadError(run, safeBody.data, durationInMs); + + break; + } default: { const _exhaustiveCheck: never = status; throw new Error(`Non-exhaustive match for value: ${status}`); @@ -455,6 +461,23 @@ export class PerformRunExecutionV2Service { }); } + async #failRunWithInvalidPayloadError( + execution: FoundRun, + data: RunJobInvalidPayloadError, + durationInMs: number + ) { + return await $transaction(this.#prismaClient, async (tx) => { + await this.#failRunExecution( + tx, + "EXECUTE_JOB", + execution, + data.errors, + "INVALID_PAYLOAD", + durationInMs + ); + }); + } + async #retryRunWithTask( run: FoundRun, data: RunJobRetryWithTask, @@ -579,7 +602,7 @@ export class PerformRunExecutionV2Service { reason: "EXECUTE_JOB" | "PREPROCESS", run: FoundRun, output: Record, - status: "FAILURE" | "ABORTED" | "TIMED_OUT" | "UNRESOLVED_AUTH" = "FAILURE", + status: "FAILURE" | "ABORTED" | "TIMED_OUT" | "UNRESOLVED_AUTH" | "INVALID_PAYLOAD" = "FAILURE", durationInMs: number = 0 ): Promise { await $transaction(prisma, async (tx) => { diff --git a/packages/core/src/schemas/api.ts b/packages/core/src/schemas/api.ts index 9f47e9de0ff..b5f1f91610c 100644 --- a/packages/core/src/schemas/api.ts +++ b/packages/core/src/schemas/api.ts @@ -2,7 +2,7 @@ import { ulid } from "ulid"; import { z } from "zod"; import { Prettify } from "../types"; import { addMissingVersionField } from "./addMissingVersionField"; -import { ErrorWithStackSchema } from "./errors"; +import { ErrorWithStackSchema, SchemaErrorSchema } from "./errors"; import { EventRuleSchema } from "./eventFilter"; import { ConnectionAuthSchema, IntegrationConfigSchema } from "./integrations"; import { DeserializedJsonSchema, SerializableJsonSchema } from "./json"; @@ -437,6 +437,13 @@ export const RunJobErrorSchema = z.object({ export type RunJobError = z.infer; +export const RunJobInvalidPayloadErrorSchema = z.object({ + status: z.literal("INVALID_PAYLOAD"), + errors: z.array(SchemaErrorSchema), +}); + +export type RunJobInvalidPayloadError = z.infer; + export const RunJobUnresolvedAuthErrorSchema = z.object({ status: z.literal("UNRESOLVED_AUTH_ERROR"), issues: z.record(z.object({ id: z.string(), error: z.string() })), @@ -477,6 +484,7 @@ export type RunJobSuccess = z.infer; export const RunJobResponseSchema = z.discriminatedUnion("status", [ RunJobErrorSchema, RunJobUnresolvedAuthErrorSchema, + RunJobInvalidPayloadErrorSchema, RunJobResumeWithTaskSchema, RunJobRetryWithTaskSchema, RunJobCanceledWithTaskSchema, diff --git a/packages/core/src/schemas/errors.ts b/packages/core/src/schemas/errors.ts index eab959be8d0..8fb9c186dd5 100644 --- a/packages/core/src/schemas/errors.ts +++ b/packages/core/src/schemas/errors.ts @@ -7,3 +7,10 @@ export const ErrorWithStackSchema = z.object({ }); export type ErrorWithStack = z.infer; + +export const SchemaErrorSchema = z.object({ + path: z.array(z.string()), + message: z.string(), +}); + +export type SchemaError = z.infer; diff --git a/packages/core/src/schemas/runs.ts b/packages/core/src/schemas/runs.ts index 9a4535ff263..cd239a392a8 100644 --- a/packages/core/src/schemas/runs.ts +++ b/packages/core/src/schemas/runs.ts @@ -14,6 +14,7 @@ export const RunStatusSchema = z.union([ z.literal("ABORTED"), z.literal("CANCELED"), z.literal("UNRESOLVED_AUTH"), + z.literal("INVALID_PAYLOAD"), ]); export const RunTaskSchema = z.object({ diff --git a/packages/database/prisma/migrations/20230922205611_add_invalid_payload_run_status/migration.sql b/packages/database/prisma/migrations/20230922205611_add_invalid_payload_run_status/migration.sql new file mode 100644 index 00000000000..c4f931be36b --- /dev/null +++ b/packages/database/prisma/migrations/20230922205611_add_invalid_payload_run_status/migration.sql @@ -0,0 +1,2 @@ +-- AlterEnum +ALTER TYPE "JobRunStatus" ADD VALUE 'INVALID_PAYLOAD'; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 90cf7a55907..86757f86d33 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -732,6 +732,7 @@ enum JobRunStatus { ABORTED CANCELED UNRESOLVED_AUTH + INVALID_PAYLOAD } model JobRunExecution { diff --git a/packages/trigger-sdk/package.json b/packages/trigger-sdk/package.json index 11c13c92d23..f4388a12c08 100644 --- a/packages/trigger-sdk/package.json +++ b/packages/trigger-sdk/package.json @@ -39,8 +39,7 @@ "ulid": "^2.3.0", "uuid": "^9.0.0", "ws": "^8.11.0", - "zod": "3.21.4", - "zod-error": "1.5.0" + "zod": "3.21.4" }, "devDependencies": { "@trigger.dev/tsconfig": "workspace:*", diff --git a/packages/trigger-sdk/src/errors.ts b/packages/trigger-sdk/src/errors.ts index 79a358ab100..5ef02520391 100644 --- a/packages/trigger-sdk/src/errors.ts +++ b/packages/trigger-sdk/src/errors.ts @@ -1,4 +1,4 @@ -import { ErrorWithStack, ServerTask } from "@trigger.dev/core"; +import { ErrorWithStack, SchemaError, ServerTask } from "@trigger.dev/core"; export class ResumeWithTaskError { constructor(public task: ServerTask) {} @@ -16,6 +16,10 @@ export class CanceledWithTaskError { constructor(public task: ServerTask) {} } +export class ParsedPayloadSchemaError { + constructor(public schemaErrors: SchemaError[]) {} +} + /** Use this function if you're using a `try/catch` block to catch errors. * It checks if a thrown error is a special internal error that you should ignore. * If this returns `true` then you must rethrow the error: `throw err;` diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index ed8adaa924d..26e1cfe7f0c 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -31,7 +31,12 @@ import { StatusUpdate, } from "@trigger.dev/core"; import { ApiClient } from "./apiClient"; -import { CanceledWithTaskError, ResumeWithTaskError, RetryWithTaskError } from "./errors"; +import { + CanceledWithTaskError, + ParsedPayloadSchemaError, + ResumeWithTaskError, + RetryWithTaskError, +} from "./errors"; import { TriggerIntegration } from "./integrations"; import { IO } from "./io"; import { createIOWithIntegrations } from "./ioWithIntegrations"; @@ -712,6 +717,10 @@ export class TriggerClient { return { status: "SUCCESS", output }; } catch (error) { + if (error instanceof ParsedPayloadSchemaError) { + return { status: "INVALID_PAYLOAD", errors: error.schemaErrors }; + } + if (error instanceof ResumeWithTaskError) { return { status: "RESUME_WITH_TASK", task: error.task }; } diff --git a/packages/trigger-sdk/src/triggers/eventTrigger.ts b/packages/trigger-sdk/src/triggers/eventTrigger.ts index 3e051bcfd7f..6c2fc68b5cf 100644 --- a/packages/trigger-sdk/src/triggers/eventTrigger.ts +++ b/packages/trigger-sdk/src/triggers/eventTrigger.ts @@ -1,8 +1,9 @@ import { EventFilter, TriggerMetadata, deepMergeFilters } from "@trigger.dev/core"; -import { z } from "zod"; import { Job } from "../job"; import { TriggerClient } from "../triggerClient"; -import { EventSpecification, EventSpecificationExample, Trigger } from "../types"; +import { EventSpecification, EventSpecificationExample, SchemaParser, Trigger } from "../types"; +import { formatSchemaErrors } from "../utils/formatSchemaErrors"; +import { ParsedPayloadSchemaError } from "../errors"; type EventTriggerOptions> = { event: TEventSpecification; @@ -50,7 +51,7 @@ type TriggerOptions = { /** A [Zod](https://trigger.dev/docs/documentation/guides/zod) schema that defines the shape of the event payload. * The default is `z.any()` which is `any`. * */ - schema?: z.Schema; + schema?: SchemaParser; /** You can use this to filter events based on the source. */ source?: string; /** Used to filter which events trigger the Job @@ -94,7 +95,13 @@ export function eventTrigger( examples: options.examples, parsePayload: (rawPayload: any) => { if (options.schema) { - return options.schema.parse(rawPayload); + const results = options.schema.safeParse(rawPayload); + + if (!results.success) { + throw new ParsedPayloadSchemaError(formatSchemaErrors(results.error.issues)); + } + + return results.data; } return rawPayload as any; diff --git a/packages/trigger-sdk/src/triggers/externalSource.ts b/packages/trigger-sdk/src/triggers/externalSource.ts index 8fa50f02470..01006564089 100644 --- a/packages/trigger-sdk/src/triggers/externalSource.ts +++ b/packages/trigger-sdk/src/triggers/externalSource.ts @@ -1,5 +1,3 @@ -import { z } from "zod"; - import { DisplayProperty, EventFilter, @@ -16,7 +14,7 @@ import { IOWithIntegrations, TriggerIntegration } from "../integrations"; import { IO } from "../io"; import { Job } from "../job"; import { TriggerClient } from "../triggerClient"; -import type { EventSpecification, Trigger, TriggerContext } from "../types"; +import type { EventSpecification, SchemaParser, Trigger, TriggerContext } from "../types"; import { slugifyId } from "../utils"; import { SerializableJson } from "@trigger.dev/core"; import { ConnectionAuth } from "@trigger.dev/core"; @@ -154,8 +152,8 @@ type ExternalSourceOptions< > = { id: string; version: string; - schema: z.Schema; - optionSchema?: z.Schema; + schema: SchemaParser; + optionSchema?: SchemaParser; integration: TIntegration; register: RegisterFunction; filter?: FilterFunction; diff --git a/packages/trigger-sdk/src/types.ts b/packages/trigger-sdk/src/types.ts index c28e6ec5a87..e7938b236ea 100644 --- a/packages/trigger-sdk/src/types.ts +++ b/packages/trigger-sdk/src/types.ts @@ -105,3 +105,16 @@ export interface EventSpecification { export type EventTypeFromSpecification> = TEventSpec extends EventSpecification ? TEvent : never; + +export type SchemaParserIssue = { path: PropertyKey[]; message: string }; + +export type SchemaParserResult = + | { + success: true; + data: T; + } + | { success: false; error: { issues: SchemaParserIssue[] } }; + +export type SchemaParser = { + safeParse: (a: unknown) => SchemaParserResult; +}; diff --git a/packages/trigger-sdk/src/utils/formatSchemaErrors.ts b/packages/trigger-sdk/src/utils/formatSchemaErrors.ts new file mode 100644 index 00000000000..344a8ea6b48 --- /dev/null +++ b/packages/trigger-sdk/src/utils/formatSchemaErrors.ts @@ -0,0 +1,9 @@ +import type { SchemaError } from "@trigger.dev/core"; +import { SchemaParserIssue } from "../types"; + +export function formatSchemaErrors(errors: SchemaParserIssue[]): SchemaError[] { + return errors.map((error) => { + const { path, message } = error; + return { path: path.map(String), message }; + }); +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 26a98bf0748..213846bf9a8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -990,7 +990,6 @@ importers: uuid: ^9.0.0 ws: ^8.11.0 zod: 3.21.4 - zod-error: 1.5.0 dependencies: '@trigger.dev/core': link:../core chalk: 5.2.0 @@ -1007,7 +1006,6 @@ importers: uuid: 9.0.0 ws: 8.12.0 zod: 3.21.4 - zod-error: 1.5.0 devDependencies: '@trigger.dev/tsconfig': link:../../config-packages/tsconfig '@types/debug': 4.1.7 diff --git a/references/job-catalog/package.json b/references/job-catalog/package.json index b0f84e07655..5f0384ea910 100644 --- a/references/job-catalog/package.json +++ b/references/job-catalog/package.json @@ -43,7 +43,6 @@ "@types/node": "20.4.2", "typescript": "5.1.6", "zod": "3.21.4", - "@trigger.dev/airtable": "workspace:*", "@trigger.dev/linear": "workspace:*" }, "trigger.dev": { diff --git a/references/job-catalog/src/events.ts b/references/job-catalog/src/events.ts index 327c7ac1360..7bd9d9a0882 100644 --- a/references/job-catalog/src/events.ts +++ b/references/job-catalog/src/events.ts @@ -58,51 +58,18 @@ client.defineJob({ }); client.defineJob({ - id: "example-job", - name: "Example Job: a joke with a delay", + id: "zod-schema", + name: "Job with Zod Schema", version: "0.0.2", trigger: eventTrigger({ - name: "shayan.event", + name: "zod.schema", schema: z.object({ userId: z.string(), delay: z.number(), }), }), run: async (payload, io, ctx) => { - await io.wait("sleeping", payload.delay); - - await io.runTask( - "init", - async () => { - console.log("init function ran", payload.userId); - }, - { name: "init" } - ); - - await io.runTask( - "failable", - async (task) => { - if (task.attempts > 2) { - console.log("task succeeded"); - return { - ok: true, - }; - } - console.log("task failed"); - throw new Error(`Task failed on ${task.attempts} attempt(s)`); - }, - { name: "task-1", retry: { limit: 3 } } - ); - - await io.runTask( - "log", - async () => { - console.log("hello from the job", payload.userId); - }, - { - name: "log", - } - ); + await io.logger.info("Hello World", { ctx, payload }); }, });