From 2ef99d83834aa548c2b1810849d9d842afa9b905 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 7 Jun 2023 13:11:58 +0100 Subject: [PATCH] Re-architect pre-processing and executing jobs, and gather run elements --- apps/webapp/app/consts.ts | 2 + .../{clientApi.server.ts => endpointApi.ts} | 84 +-- .../endpoints/createEndpoint.server.ts | 4 +- .../endpoints/endpointRegistered.server.ts | 4 +- .../events/invokeDispatcher.server.ts | 3 +- .../app/services/jobs/registerJob.server.ts | 2 + .../app/services/runs/createRun.server.ts | 1 + .../app/services/runs/performRunExecution.ts | 549 ++++++++++++++++ .../app/services/runs/resumeTask.server.ts | 214 ------- .../app/services/runs/startRun.server.ts | 584 +++++++----------- .../deliverHttpSourceRequest.server.ts | 4 +- .../triggers/initializeTrigger.server.ts | 4 +- apps/webapp/app/services/worker.server.ts | 14 +- apps/webapp/app/utils/json.ts | 15 + .../migration.sql | 29 + .../migration.sql | 2 + .../migration.sql | 15 + .../migration.sql | 2 + .../migration.sql | 10 + .../migration.sql | 5 + .../migration.sql | 28 + .../migration.sql | 12 + apps/webapp/prisma/schema.prisma | 53 +- .../nextjs-example/src/pages/api/trigger.ts | 61 +- integrations/github/src/index.ts | 36 ++ packages/internal/src/schemas/api.ts | 67 +- packages/trigger-sdk/src/index.ts | 2 +- packages/trigger-sdk/src/io.ts | 4 +- packages/trigger-sdk/src/job.ts | 2 + packages/trigger-sdk/src/triggerClient.ts | 100 ++- packages/trigger-sdk/src/triggers/dynamic.ts | 15 +- .../{customTrigger.ts => eventTrigger.ts} | 71 ++- .../src/triggers/externalSource.ts | 10 +- .../trigger-sdk/src/triggers/notifications.ts | 16 +- .../trigger-sdk/src/triggers/scheduled.ts | 32 +- packages/trigger-sdk/src/types.ts | 18 +- 36 files changed, 1275 insertions(+), 799 deletions(-) rename apps/webapp/app/services/{clientApi.server.ts => endpointApi.ts} (75%) create mode 100644 apps/webapp/app/services/runs/performRunExecution.ts delete mode 100644 apps/webapp/app/services/runs/resumeTask.server.ts create mode 100644 apps/webapp/prisma/migrations/20230606081054_add_job_run_execution/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606081441_add_preprocess_job_execution_reason/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606081946_add_preprocess_runs_to_job_version/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606082119_add_preprocess_to_runs/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606082719_add_preprocess_status_to_runs/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606120402_add_task_to_run_execution/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606122534_improve_run_execution_model/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230606132031_removed_http_responses_from_executions/migration.sql rename packages/trigger-sdk/src/triggers/{customTrigger.ts => eventTrigger.ts} (53%) diff --git a/apps/webapp/app/consts.ts b/apps/webapp/app/consts.ts index f2eafb63c..4e3071a79 100644 --- a/apps/webapp/app/consts.ts +++ b/apps/webapp/app/consts.ts @@ -2,3 +2,5 @@ export const LIVE_ENVIRONMENT = "live"; export const DEV_ENVIRONMENT = "development"; export const MAX_LIVE_PROJECTS = 1; export const DEFAULT_MAX_CONCURRENT_RUNS = 100; +export const PREPROCESS_RETRY_LIMIT = 2; +export const EXECUTE_JOB_RETRY_LIMIT = 10; diff --git a/apps/webapp/app/services/clientApi.server.ts b/apps/webapp/app/services/endpointApi.ts similarity index 75% rename from apps/webapp/app/services/clientApi.server.ts rename to apps/webapp/app/services/endpointApi.ts index b61437823..d6f6a2e11 100644 --- a/apps/webapp/app/services/clientApi.server.ts +++ b/apps/webapp/app/services/endpointApi.ts @@ -1,7 +1,8 @@ import { ApiEventLog, HttpSourceRequest, - PrepareJobTriggerBody, + PreprocessRunBody, + PreprocessRunResponseSchema, RegisterTriggerBody, RegisterTriggerBodySchema, RunJobBody, @@ -12,21 +13,20 @@ import { GetEndpointDataResponseSchema, HttpSourceResponseSchema, PongResponseSchema, - PrepareForJobExecutionResponseSchema, RunJobResponseSchema, } from "@trigger.dev/internal"; import { logger } from "./logger"; -export class ClientApiError extends Error { +export class EndpointApiError extends Error { constructor(message: string, stack?: string) { - super(`ClientApiError: ${message}`); + super(`EndpointApiError: ${message}`); this.stack = stack; - this.name = "ClientApiError"; + this.name = "EndpointApiError"; } } // TODO: this should work with tunnelling -export class ClientApi { +export class EndpointApi { #apiKey: string; #url: string; @@ -128,7 +128,7 @@ export class ClientApi { return DeliverEventResponseSchema.parse(anyBody); } - async executeJob(options: RunJobBody) { + async executeJobRequest(options: RunJobBody) { const response = await safeFetch(this.#url, { method: "POST", headers: { @@ -139,75 +139,25 @@ export class ClientApi { body: JSON.stringify(options), }); - if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); - } - - if (!response.ok) { - // Attempt to parse the error message - const anyBody = await response.json(); - - const error = ErrorWithStackSchema.safeParse(anyBody); - - if (error.success) { - throw new ClientApiError(error.data.message, error.data.stack); - } - - throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` - ); - } - - const anyBody = await response.json(); - - logger.debug("executeJob() response from endpoint", { - body: anyBody, - }); - - return RunJobResponseSchema.parse(anyBody); + return { + response, + parser: RunJobResponseSchema, + errorParser: ErrorWithStackSchema, + }; } - async prepareJobTrigger(payload: PrepareJobTriggerBody) { + async preprocessRunRequest(options: PreprocessRunBody) { const response = await safeFetch(this.#url, { method: "POST", headers: { "Content-Type": "application/json", "x-trigger-api-key": this.#apiKey, - "x-trigger-action": "PREPARE_JOB_TRIGGER", + "x-trigger-action": "PREPROCESS_RUN", }, - body: JSON.stringify(payload), + body: JSON.stringify(options), }); - if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); - } - - if (!response.ok) { - // Attempt to parse the error message - const anyBody = await response.json(); - - const error = ErrorWithStackSchema.safeParse(anyBody); - - if (error.success) { - throw new ClientApiError(error.data.message, error.data.stack); - } - - throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` - ); - } - - const anyBody = await response.json(); - - logger.debug("prepareForJobExecution() response from endpoint", { - body: anyBody, - }); - - return PrepareForJobExecutionResponseSchema.parse(anyBody); + return { response, parser: PreprocessRunResponseSchema }; } async initializeTrigger( @@ -235,7 +185,7 @@ export class ClientApi { const error = ErrorWithStackSchema.safeParse(anyBody); if (error.success) { - throw new ClientApiError(error.data.message, error.data.stack); + throw new EndpointApiError(error.data.message, error.data.stack); } throw new Error( diff --git a/apps/webapp/app/services/endpoints/createEndpoint.server.ts b/apps/webapp/app/services/endpoints/createEndpoint.server.ts index 1644d32e0..f0eee815a 100644 --- a/apps/webapp/app/services/endpoints/createEndpoint.server.ts +++ b/apps/webapp/app/services/endpoints/createEndpoint.server.ts @@ -2,7 +2,7 @@ import type { Organization, RuntimeEnvironment } from ".prisma/client"; import { $transaction, PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; import { AuthenticatedEnvironment } from "../apiAuth.server"; -import { ClientApi } from "../clientApi.server"; +import { EndpointApi } from "../endpointApi"; import { workerQueue } from "../worker.server"; export class CreateEndpointService { @@ -21,7 +21,7 @@ export class CreateEndpointService { url: string; name: string; }) { - const client = new ClientApi(environment.apiKey, url); + const client = new EndpointApi(environment.apiKey, url); await client.ping(); return await $transaction(this.#prismaClient, async (tx) => { diff --git a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts b/apps/webapp/app/services/endpoints/endpointRegistered.server.ts index af3cfe208..973e13f72 100644 --- a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts +++ b/apps/webapp/app/services/endpoints/endpointRegistered.server.ts @@ -1,6 +1,6 @@ import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import { ClientApi } from "../clientApi.server"; +import { EndpointApi } from "../endpointApi"; import { workerQueue } from "../worker.server"; export class EndpointRegisteredService { @@ -21,7 +21,7 @@ export class EndpointRegisteredService { }); // Make a request to the endpoint to fetch a list of jobs - const client = new ClientApi(endpoint.environment.apiKey, endpoint.url); + const client = new EndpointApi(endpoint.environment.apiKey, endpoint.url); const { jobs, sources, dynamicTriggers, dynamicSchedules } = await client.getEndpointData(); diff --git a/apps/webapp/app/services/events/invokeDispatcher.server.ts b/apps/webapp/app/services/events/invokeDispatcher.server.ts index 9f0b17d5a..2c5db93f0 100644 --- a/apps/webapp/app/services/events/invokeDispatcher.server.ts +++ b/apps/webapp/app/services/events/invokeDispatcher.server.ts @@ -1,9 +1,8 @@ import { z } from "zod"; -import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server"; +import type { PrismaClientOrTransaction } from "~/db.server"; import { prisma } from "~/db.server"; import { logger } from "~/services/logger"; import { CreateRunService } from "~/services/runs/createRun.server"; -import { ResumeTaskService } from "~/services/runs/resumeTask.server"; const JobVersionDispatchableSchema = z.object({ type: z.literal("JOB_VERSION"), diff --git a/apps/webapp/app/services/jobs/registerJob.server.ts b/apps/webapp/app/services/jobs/registerJob.server.ts index 8da6d153e..484f55a59 100644 --- a/apps/webapp/app/services/jobs/registerJob.server.ts +++ b/apps/webapp/app/services/jobs/registerJob.server.ts @@ -198,6 +198,7 @@ export class RegisterJobService { }, version: metadata.version, eventSpecification: metadata.event, + preprocessRuns: metadata.preprocessRuns, startPosition: metadata.startPosition === "initial" ? "INITIAL" : "LATEST", }, @@ -205,6 +206,7 @@ export class RegisterJobService { startPosition: metadata.startPosition === "initial" ? "INITIAL" : "LATEST", eventSpecification: metadata.event, + preprocessRuns: metadata.preprocessRuns, queue: { connect: { id: jobQueue.id, diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index 3287e85c7..11155f5f7 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/apps/webapp/app/services/runs/createRun.server.ts @@ -54,6 +54,7 @@ export class CreateRunService { const run = await tx.jobRun.create({ data: { number: newNumber, + preprocess: version.preprocessRuns, job: { connect: { id: job.id } }, version: { connect: { id: version.id } }, event: { connect: { id: eventId } }, diff --git a/apps/webapp/app/services/runs/performRunExecution.ts b/apps/webapp/app/services/runs/performRunExecution.ts new file mode 100644 index 000000000..a37b1521a --- /dev/null +++ b/apps/webapp/app/services/runs/performRunExecution.ts @@ -0,0 +1,549 @@ +import { ApiEventLogSchema, CachedTaskSchema } from "@trigger.dev/internal"; +import { generateErrorMessage } from "zod-error"; +import { + $transaction, + PrismaClient, + PrismaClientOrTransaction, + prisma, +} from "~/db.server"; +import { resolveRunConnections } from "~/models/runConnection.server"; +import { safeJsonZodParse } from "~/utils/json"; +import { EndpointApi } from "../endpointApi"; +import { workerQueue } from "../worker.server"; +import type { Task } from ".prisma/client"; +import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts"; + +type FoundRunExecution = NonNullable< + Awaited> +>; + +export class PerformRunExecutionService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call(id: string) { + const runExecution = await findRunExecution(this.#prismaClient, id); + + if (!runExecution) { + return; + } + + switch (runExecution.reason) { + case "PREPROCESS": { + await this.#executePreprocessing(runExecution); + break; + } + case "EXECUTE_JOB": { + await this.#executeJob(runExecution); + break; + } + } + } + + // Execute the preprocessing step of a run, which will send the payload to the endpoint and give the job + // an opportunity to generate run elements based on the payload. + // If the endpoint is not available, or the response is not ok, + // the run execution will be marked as failed and the run will start + async #executePreprocessing(execution: FoundRunExecution) { + const { run } = execution; + + const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); + const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId }); + const startedAt = new Date(); + + await this.#prismaClient.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + status: "STARTED", + startedAt, + }, + }); + + const { response, parser } = await client.preprocessRunRequest({ + event, + job: { + id: run.version.job.slug, + version: run.version.version, + }, + run: { + id: run.id, + isTest: run.isTest, + }, + environment: { + id: run.environment.id, + slug: run.environment.slug, + type: run.environment.type, + }, + organization: { + id: run.organization.id, + slug: run.organization.slug, + title: run.organization.title, + }, + account: run.externalAccount + ? { + id: run.externalAccount.identifier, + metadata: run.externalAccount.metadata, + } + : undefined, + }); + + if (!response) { + return await this.#failRunExecutionWithRetry(execution, { + message: "Could not connect to the endpoint", + }); + } + + if (!response.ok) { + return await this.#failRunExecutionWithRetry(execution, { + message: `Endpoint responded with ${response.status} status code`, + }); + } + + const rawBody = await response.text(); + const safeBody = safeJsonZodParse(parser, rawBody); + + if (!safeBody) { + return await this.#failRunExecution(this.#prismaClient, execution, { + message: "Endpoint responded with invalid JSON", + }); + } + + if (!safeBody.success) { + return await this.#failRunExecution(this.#prismaClient, execution, { + message: generateErrorMessage(safeBody.error.issues), + }); + } + + if (safeBody.data.abort) { + return this.#failRunExecution( + this.#prismaClient, + execution, + { message: "Endpoint aborted the run" }, + "ABORTED" + ); + } else { + await $transaction(this.#prismaClient, async (tx) => { + await tx.jobRun.update({ + where: { + id: run.id, + }, + data: { + status: "STARTED", + startedAt: new Date(), + elements: safeBody.data.elements, + }, + }); + + await tx.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + status: "SUCCESS", + completedAt: new Date(), + }, + }); + + const runExecution = await tx.jobRunExecution.create({ + data: { + runId: run.id, + reason: "EXECUTE_JOB", + status: "PENDING", + retryLimit: EXECUTE_JOB_RETRY_LIMIT, + }, + }); + + await workerQueue.enqueue( + "performRunExecution", + { + id: runExecution.id, + }, + { tx } + ); + }); + } + } + async #executeJob(execution: FoundRunExecution) { + const { run } = execution; + + const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); + const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId }); + + const startedAt = new Date(); + + await this.#prismaClient.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + status: "STARTED", + startedAt, + }, + }); + + const connections = await resolveRunConnections(run.runConnections); + + if (Object.keys(connections).length < run.runConnections.length) { + return this.#failRunExecutionWithRetry(execution, { + message: `Could not resolve all connections for run ${ + run.id + }, there should be ${run.runConnections.length} connections but only ${ + Object.keys(connections).length + } were resolved.`, + }); + } + + let resumedTask: Task | undefined; + + if (execution.resumeTaskId) { + resumedTask = await this.#prismaClient.task.update({ + where: { + id: execution.resumeTaskId, + }, + data: { + status: "COMPLETED", + completedAt: new Date(), + }, + }); + } + + const { response, parser, errorParser } = await client.executeJobRequest({ + event, + job: { + id: run.version.job.slug, + version: run.version.version, + }, + run: { + id: run.id, + isTest: run.isTest, + startedAt, + }, + environment: { + id: run.environment.id, + slug: run.environment.slug, + type: run.environment.type, + }, + organization: { + id: run.organization.id, + slug: run.organization.slug, + title: run.organization.title, + }, + account: run.externalAccount + ? { + id: run.externalAccount.identifier, + metadata: run.externalAccount.metadata, + } + : undefined, + connections, + tasks: [run.tasks, resumedTask] + .flat() + .filter(Boolean) + .map((t) => CachedTaskSchema.parse(t)), + }); + + if (!response) { + return await this.#failRunExecutionWithRetry(execution, { + message: "Could not connect to the endpoint", + }); + } + + // TODO: handle timeouts + if (!response.ok) { + const rawErrorBody = await response.text(); + const safeErrorBody = safeJsonZodParse(errorParser, rawErrorBody); + + if (!safeErrorBody || !safeErrorBody.success) { + return await this.#failRunExecutionWithRetry(execution, { + message: `Endpoint responded with ${response.status} status code`, + }); + } + + return await this.#failRunExecution( + this.#prismaClient, + execution, + safeErrorBody.data + ); + } + + const rawBody = await response.text(); + const safeBody = safeJsonZodParse(parser, rawBody); + + if (!safeBody) { + return await this.#failRunExecution(this.#prismaClient, execution, { + message: "Endpoint responded with invalid JSON", + }); + } + + if (!safeBody.success) { + return await this.#failRunExecution(this.#prismaClient, execution, { + message: generateErrorMessage(safeBody.error.issues), + }); + } + + if (safeBody.data.completed) { + return await $transaction(this.#prismaClient, async (tx) => { + await tx.jobRun.update({ + where: { id: run.id }, + data: { + completedAt: new Date(), + status: "SUCCESS", + output: safeBody.data.output ?? undefined, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, + }, + }); + + await tx.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + status: "SUCCESS", + completedAt: new Date(), + }, + }); + + await workerQueue.enqueue( + "runFinished", + { + id: run.id, + }, + { tx } + ); + }); + } + + const resumeTask = safeBody.data.task; + + if (resumeTask) { + return await $transaction(this.#prismaClient, async (tx) => { + await tx.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + status: "SUCCESS", + completedAt: new Date(), + }, + }); + + const newJobExecution = await tx.jobRunExecution.create({ + data: { + runId: run.id, + reason: "EXECUTE_JOB", + status: "PENDING", + retryLimit: EXECUTE_JOB_RETRY_LIMIT, + resumeTaskId: resumeTask.id, + }, + }); + + await workerQueue.enqueue( + "performRunExecution", + { + id: newJobExecution.id, + }, + { tx, runAt: resumeTask.delayUntil ?? undefined } + ); + }); + } + } + + async #failRunExecutionWithRetry( + execution: FoundRunExecution, + output: Record + ): Promise { + await $transaction(this.#prismaClient, async (tx) => { + if (execution.retryCount + 1 > execution.retryLimit) { + // We've reached the retry limit, so we need to fail the execution and stop retrying + return await this.#failRunExecution(tx, execution, output); + } + + // We need to retry execution + const retryCount = execution.retryCount + 1; + // Use an exponential backoff strategy with the exponent being 1.5 + // So when retryCount is 1, retryDelayInMs is 500ms + // When retryCount is 2, retryDelayInMs is 750ms + // When retryCount is 3, retryDelayInMs is 1125ms + const retryDelayInMs = Math.round(500 * Math.pow(1.5, retryCount - 1)); + + await tx.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + retryCount, + retryDelayInMs, + error: output, + }, + }); + + const runAt = new Date(Date.now() + retryDelayInMs); + + await workerQueue.enqueue( + "performRunExecution", + { id: execution.id }, + { runAt, tx } + ); + }); + } + + async #failRunExecution( + prisma: PrismaClientOrTransaction, + execution: FoundRunExecution, + output: Record, + status: "FAILURE" | "ABORTED" = "FAILURE" + ): Promise { + const { run } = execution; + + await $transaction(prisma, async (tx) => { + switch (execution.reason) { + case "EXECUTE_JOB": { + // If the execution is an EXECUTE_JOB reason, we need to fail the run + await tx.jobRun.update({ + where: { id: run.id }, + data: { + completedAt: new Date(), + status, + output, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, + }, + }); + + await workerQueue.enqueue( + "runFinished", + { + id: run.id, + }, + { tx } + ); + break; + } + case "PREPROCESS": { + // If the status is ABORTED, we need to fail the run + if (status === "ABORTED") { + await tx.jobRun.update({ + where: { id: run.id }, + data: { + completedAt: new Date(), + status, + output, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, + }, + }); + + await workerQueue.enqueue( + "runFinished", + { + id: run.id, + }, + { tx } + ); + + break; + } + + await tx.jobRun.update({ + where: { + id: run.id, + }, + data: { + status: "STARTED", + startedAt: new Date(), + }, + }); + + const runExecution = await tx.jobRunExecution.create({ + data: { + runId: run.id, + reason: "EXECUTE_JOB", + status: "PENDING", + retryLimit: EXECUTE_JOB_RETRY_LIMIT, + }, + }); + + await workerQueue.enqueue( + "performRunExecution", + { + id: runExecution.id, + }, + { tx } + ); + + break; + } + } + + await tx.jobRunExecution.update({ + where: { + id: execution.id, + }, + data: { + status: "FAILURE", + completedAt: new Date(), + error: JSON.stringify(output), + }, + }); + }); + } +} + +async function findRunExecution(prisma: PrismaClientOrTransaction, id: string) { + return await prisma.jobRunExecution.findUnique({ + where: { id }, + include: { + run: { + include: { + environment: true, + endpoint: true, + organization: true, + externalAccount: true, + runConnections: { + include: { + apiConnection: { + include: { + dataReference: true, + }, + }, + }, + }, + tasks: { + where: { + status: { + in: ["COMPLETED"], + }, + }, + }, + event: true, + version: { + include: { + job: true, + organization: true, + }, + }, + }, + }, + }, + }); +} diff --git a/apps/webapp/app/services/runs/resumeTask.server.ts b/apps/webapp/app/services/runs/resumeTask.server.ts deleted file mode 100644 index fa1b86c0b..000000000 --- a/apps/webapp/app/services/runs/resumeTask.server.ts +++ /dev/null @@ -1,214 +0,0 @@ -import { ApiEventLogSchema, CachedTaskSchema } from "@trigger.dev/internal"; -import type { PrismaClient } from "~/db.server"; -import { prisma } from "~/db.server"; -import { ClientApi, ClientApiError } from "../clientApi.server"; -import { workerQueue } from "../worker.server"; -import { resolveRunConnections } from "~/models/runConnection.server"; - -export class ResumeTaskService { - #prismaClient: PrismaClient; - - constructor(prismaClient: PrismaClient = prisma) { - this.#prismaClient = prismaClient; - } - - public async call(id: string, output?: any) { - const task = await this.#prismaClient.task.findUniqueOrThrow({ - where: { id }, - include: { - run: { - include: { - version: { - include: { - endpoint: true, - job: true, - }, - }, - environment: true, - event: true, - organization: true, - externalAccount: true, - tasks: { - where: { - status: { - in: ["COMPLETED"], - }, - }, - }, - queue: true, - runConnections: { - include: { - apiConnection: { - include: { - dataReference: true, - }, - }, - }, - }, - }, - }, - }, - }); - - const { run } = task; - - const connections = await resolveRunConnections(run.runConnections); - - if (Object.keys(connections).length < run.runConnections.length) { - throw new Error( - `Could not resolve all connections for run ${run.id} and task ${ - task.id - }, there should be ${run.runConnections.length} connections but only ${ - Object.keys(connections).length - } were resolved.` - ); - } - - const updatedTask = await this.#prismaClient.task.update({ - where: { - id: task.id, - }, - data: { - status: task.noop || output ? "COMPLETED" : "RUNNING", - completedAt: task.noop ? new Date() : undefined, - output: task.noop ? undefined : output, - }, - }); - - const client = new ClientApi( - run.environment.apiKey, - run.version.endpoint.url - ); - - const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId }); - - try { - const results = await client.executeJob({ - event, - job: { - id: run.version.job.slug, - version: run.version.version, - }, - run: { - id: run.id, - isTest: run.isTest, - startedAt: run.startedAt ?? new Date(), - }, - environment: { - id: run.environment.id, - slug: run.environment.slug, - type: run.environment.type, - }, - organization: { - id: run.organization.id, - slug: run.organization.slug, - title: run.organization.title, - }, - account: run.externalAccount - ? { - id: run.externalAccount.identifier, - metadata: run.externalAccount.metadata, - } - : undefined, - tasks: [run.tasks, updatedTask] - .flat() - .map((t) => CachedTaskSchema.parse(t)), - connections, - }); - - if (results.completed) { - await this.#prismaClient.$transaction(async (tx) => { - await tx.jobRun.update({ - where: { id: run.id }, - data: { - completedAt: new Date(), - status: "SUCCESS", - output: results.output ?? undefined, - queue: { - update: { - jobCount: { - decrement: 1, - }, - }, - }, - }, - }); - - await tx.jobQueue.update({ - where: { id: run.queueId }, - data: { - jobCount: { - decrement: 1, - }, - }, - }); - - await workerQueue.enqueue( - "runFinished", - { - id: run.id, - }, - { tx } - ); - }); - - return; - } - - if (results.task) { - await workerQueue.enqueue( - "resumeTask", - { - id: results.task.id, - }, - { runAt: results.task.delayUntil ?? undefined } - ); - - return; - } - } catch (error) { - await this.#prismaClient.$transaction(async (tx) => { - if (error instanceof ClientApiError) { - await tx.jobRun.update({ - where: { id: run.id }, - data: { - completedAt: new Date(), - status: "FAILURE", - output: { message: error.message, stack: error.stack }, - }, - }); - } else { - await tx.jobRun.update({ - where: { id: run.id }, - data: { - completedAt: new Date(), - status: "FAILURE", - output: { - message: - error instanceof Error ? error.message : "Unknown Error", - stack: error instanceof Error ? error.stack : undefined, - }, - }, - }); - } - - await tx.jobQueue.update({ - where: { id: run.queueId }, - data: { - jobCount: { - decrement: 1, - }, - }, - }); - - await workerQueue.enqueue( - "runFinished", - { - id: run.id, - }, - { tx } - ); - }); - } - } -} diff --git a/apps/webapp/app/services/runs/startRun.server.ts b/apps/webapp/app/services/runs/startRun.server.ts index 77512b02d..28ba1ab8e 100644 --- a/apps/webapp/app/services/runs/startRun.server.ts +++ b/apps/webapp/app/services/runs/startRun.server.ts @@ -1,30 +1,11 @@ -import { ApiEventLogSchema } from "@trigger.dev/internal"; -import type { PrismaClient } from "~/db.server"; -import { prisma } from "~/db.server"; -import { resolveRunConnections } from "~/models/runConnection.server"; -import { ClientApi, ClientApiError } from "../clientApi.server"; -import { workerQueue } from "../worker.server"; -import { logger } from "../logger"; import type { ApiConnection, ApiConnectionType } from ".prisma/client"; +import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server"; +import { prisma } from "~/db.server"; +import { workerQueue } from "../worker.server"; +import { EXECUTE_JOB_RETRY_LIMIT, PREPROCESS_RETRY_LIMIT } from "~/consts"; -const RUN_INCLUDES = { - queue: true, - event: true, - externalAccount: true, - version: { - include: { - endpoint: true, - job: true, - environment: true, - organization: true, - integrations: { - include: { - apiConnectionClient: true, - }, - }, - }, - }, -} as const; +type FoundRun = NonNullable>>; +type RunConnectionsByKey = Awaited>; export class StartRunService { #prismaClient: PrismaClient; @@ -34,370 +15,281 @@ export class StartRunService { } public async call(id: string) { - const transactionResults = await this.#prismaClient.$transaction( - async (tx) => { - const run = await tx.jobRun.findUnique({ - where: { id }, - include: RUN_INCLUDES, - }); + await this.#prismaClient.$transaction(async (tx) => { + const run = await findRun(tx, id); - if (!run) { - return; - } + if (!run || !this.#runIsStartable(run)) { + return; + } - const startableStatuses = [ - "PENDING", - "QUEUED", - "WAITING_ON_CONNECTIONS", - ] as const; + if (run.queue.jobCount >= run.queue.maxJobs) { + await this.#queueRun(tx, id); + } else { + const runConnectionsByKey = await createRunConnections(tx, run); - if (!startableStatuses.includes(run.status)) { - return; - } - - // Check the JobQueue to make sure we can start the run - if (run.queue.jobCount >= run.queue.maxJobs) { - // Set the run status to QUEUED and return - const updatedRun = await tx.jobRun.update({ - where: { id }, - data: { - status: "QUEUED", - queuedAt: new Date(), - }, - include: RUN_INCLUDES, - }); - - return { run: updatedRun }; + if (hasMissingConnections(runConnectionsByKey)) { + await this.#handleMissingConnections(tx, id, runConnectionsByKey); } else { - // If any of the connections are missing, we can't start the execution - const runConnectionsByKey = await run.version.integrations.reduce( - async ( - accP: Promise< - Record< - string, - | { result: "resolved"; connection: ApiConnection } - | { - result: "missing"; - connectionType: ApiConnectionType; - apiConnectionClientId: string; - externalAccountId?: string; - } - > - >, - integration - ) => { - const acc = await accP; - - const connection = run.externalAccountId - ? await tx.apiConnection.findFirst({ - where: { - clientId: integration.apiConnectionClient.id, - connectionType: "EXTERNAL", - externalAccountId: run.externalAccountId, - }, - }) - : await tx.apiConnection.findFirst({ - where: { - clientId: integration.apiConnectionClient.id, - connectionType: "DEVELOPER", - }, - }); - - if (connection) { - acc[integration.key] = { result: "resolved", connection }; - } else { - acc[integration.key] = { - result: "missing", - connectionType: run.externalAccountId - ? "EXTERNAL" - : "DEVELOPER", - externalAccountId: run.externalAccountId ?? undefined, - apiConnectionClientId: integration.apiConnectionClient.id, - }; - } - - return acc; - }, - Promise.resolve({}) - ); - - // Make sure we have all the connections we need - if ( - Object.values(runConnectionsByKey).some( - (connection) => connection.result === "missing" - ) - ) { - // Create missing connections and update the jobRun to be WAITING_ON_CONNECTIONS - const missingConnections = Object.values(runConnectionsByKey) - .map((runConnection) => - runConnection.result === "missing" ? runConnection : undefined - ) - .filter(Boolean); - - // Start the jobRun and increment the jobCount - // TODO: what happens when there are more than 1 missing connection on a run? - const updatedRun = await tx.jobRun.update({ - where: { id }, - data: { - status: "WAITING_ON_CONNECTIONS", - missingConnections: { - connectOrCreate: missingConnections.map((connection) => ({ - where: { - apiConnectionClientId_connectionType_externalAccountId: { - apiConnectionClientId: connection.apiConnectionClientId, - connectionType: connection.connectionType, - externalAccountId: - connection.externalAccountId ?? "DEVELOPER", - }, - }, - create: { - apiConnectionClientId: connection.apiConnectionClientId, - connectionType: connection.connectionType, - externalAccountId: - connection.externalAccountId ?? "DEVELOPER", - resolved: false, - }, - })), - }, - }, - include: { - missingConnections: { - include: { - _count: { - select: { runs: true }, - }, - }, - }, - ...RUN_INCLUDES, - }, - }); - - for (const missingConnection of updatedRun.missingConnections) { - if (missingConnection._count.runs === 1) { - workerQueue.enqueue( - "missingConnectionCreated", - { - id: missingConnection.id, - }, - { tx } - ); - } - } - - return { run: updatedRun }; - } - - const createRunConnections = Object.entries(runConnectionsByKey) - .map(([key, runConnection]) => - runConnection.result === "resolved" - ? { - key, - apiConnectionId: runConnection.connection.id, - } - : undefined - ) - .filter(Boolean); - - // Start the jobRun and increment the jobCount - const updatedRun = await tx.jobRun.update({ - where: { id }, - data: { - status: "STARTED", - startedAt: new Date(), - queue: { - update: { - jobCount: { - increment: 1, - }, - }, - }, - runConnections: { - create: createRunConnections, - }, - }, - include: { - runConnections: { - include: { - apiConnection: { - include: { - dataReference: true, - }, - }, - }, - }, - ...RUN_INCLUDES, - }, - }); - - const connections = await resolveRunConnections( - updatedRun.runConnections - ); - - if ( - Object.keys(connections).length < updatedRun.runConnections.length - ) { - throw new Error( - `Could not resolve all connections for run ${ - run.id - }, there should be ${ - updatedRun.runConnections.length - } connections but only ${ - Object.keys(connections).length - } were resolved.` - ); - } - - return { run: updatedRun, connections }; + await this.#startRun(tx, id, run, runConnectionsByKey); } } - ); - - if (!transactionResults) { - logger.debug(`Run ${id} not found, aborting start run`, { id }); - - return; - } - - const { run, connections } = transactionResults; - - if (run.status === "QUEUED") { - logger.debug(`Run ${id} queued, aborting start run`, { id }); - - return; - } - - if (run.status === "WAITING_ON_CONNECTIONS") { - logger.debug(`Run ${id} waiting on connections, aborting start run`, { - id, - }); - - return; - } - - await workerQueue.enqueue("startQueuedRuns", { - id: run.queueId, }); + } - const startedAt = run.startedAt ?? new Date(); + #runIsStartable(run: FoundRun) { + const startableStatuses = [ + "PENDING", + "QUEUED", + "WAITING_ON_CONNECTIONS", + ] as const; + return startableStatuses.includes(run.status); + } - const event = ApiEventLogSchema.parse({ - ...run.event, - id: run.event.eventId, + async #queueRun(tx: PrismaClientOrTransaction, id: string) { + await tx.jobRun.update({ + where: { id }, + data: { + status: "QUEUED", + queuedAt: new Date(), + }, }); + } - const client = new ClientApi( - run.version.environment.apiKey, - run.version.endpoint.url - ); - - try { - // TODO: update this to implement retrying - const results = await client.executeJob({ - event, - job: { - id: run.version.job.slug, - version: run.version.version, - }, - run: { - id: run.id, - isTest: run.isTest, - startedAt, - }, - environment: { - id: run.version.environment.id, - slug: run.version.environment.slug, - type: run.version.environment.type, - }, - organization: { - id: run.version.organization.id, - slug: run.version.organization.slug, - title: run.version.organization.title, - }, - account: run.externalAccount + async #startRun( + tx: PrismaClientOrTransaction, + id: string, + run: FoundRun, + runConnectionsByKey: RunConnectionsByKey + ) { + const createRunConnections = Object.entries(runConnectionsByKey) + .map(([key, runConnection]) => + runConnection.result === "resolved" ? { - id: run.externalAccount.identifier, - metadata: run.externalAccount.metadata, + key, + apiConnectionId: runConnection.connection.id, } - : undefined, - connections, - }); + : undefined + ) + .filter(Boolean); - if (results.completed) { - await this.#prismaClient.jobRun.update({ + const updateRunAndCreateExecution = async () => { + if (run.preprocess) { + // Start the jobRun and increment the jobCount + await tx.jobRun.update({ where: { id }, data: { - completedAt: new Date(), - status: "SUCCESS", - output: results.output ?? undefined, + status: "PREPROCESSING", queue: { update: { jobCount: { - decrement: 1, + increment: 1, }, }, }, + runConnections: { + create: createRunConnections, + }, }, }); - await workerQueue.enqueue("runFinished", { - id: run.id, - }); - - return; - } - - if (results.task) { - await workerQueue.enqueue( - "resumeTask", - { - id: results.task.id, - }, - { runAt: results.task.delayUntil ?? undefined } - ); - - return; - } - } catch (error) { - if (error instanceof ClientApiError) { - await this.#prismaClient.jobRun.update({ - where: { id }, + return await tx.jobRunExecution.create({ data: { - completedAt: new Date(), - status: "FAILURE", - output: { message: error.message, stack: error.stack }, - queue: { - update: { - jobCount: { - decrement: 1, - }, + run: { + connect: { + id, }, }, + status: "PENDING", + reason: "PREPROCESS", + retryLimit: PREPROCESS_RETRY_LIMIT, }, }); } else { - await this.#prismaClient.jobRun.update({ + // Start the jobRun and increment the jobCount + await tx.jobRun.update({ where: { id }, data: { - completedAt: new Date(), - status: "FAILURE", - output: { - message: error instanceof Error ? error.message : "Unknown Error", - stack: error instanceof Error ? error.stack : undefined, - }, + status: "STARTED", + startedAt: new Date(), queue: { update: { jobCount: { - decrement: 1, + increment: 1, }, }, }, + runConnections: { + create: createRunConnections, + }, + }, + }); + + return await tx.jobRunExecution.create({ + data: { + run: { + connect: { + id, + }, + }, + status: "PENDING", + reason: "EXECUTE_JOB", + retryLimit: EXECUTE_JOB_RETRY_LIMIT, }, }); } + }; - await workerQueue.enqueue("runFinished", { - id: run.id, - }); + const execution = await updateRunAndCreateExecution(); + + await workerQueue.enqueue( + "performRunExecution", + { + id: execution.id, + }, + { tx } + ); + + await workerQueue.enqueue( + "startQueuedRuns", + { + id: run.queueId, + }, + { tx } + ); + } + + async #handleMissingConnections( + tx: PrismaClientOrTransaction, + id: string, + runConnectionsByKey: RunConnectionsByKey + ) { + const missingConnections = Object.values(runConnectionsByKey) + .map((runConnection) => + runConnection.result === "missing" ? runConnection : undefined + ) + .filter(Boolean); + + const updatedRun = await tx.jobRun.update({ + where: { id }, + data: { + status: "WAITING_ON_CONNECTIONS", + missingConnections: { + connectOrCreate: missingConnections.map((connection) => ({ + where: { + apiConnectionClientId_connectionType_externalAccountId: { + apiConnectionClientId: connection.apiConnectionClientId, + connectionType: connection.connectionType, + externalAccountId: connection.externalAccountId ?? "DEVELOPER", + }, + }, + create: { + apiConnectionClientId: connection.apiConnectionClientId, + connectionType: connection.connectionType, + externalAccountId: connection.externalAccountId ?? "DEVELOPER", + resolved: false, + }, + })), + }, + }, + include: { + missingConnections: { + include: { + _count: { + select: { runs: true }, + }, + }, + }, + }, + }); + + for (const missingConnection of updatedRun.missingConnections) { + if (missingConnection._count.runs === 1) { + workerQueue.enqueue( + "missingConnectionCreated", + { + id: missingConnection.id, + }, + { tx } + ); + } } } } + +async function findRun(tx: PrismaClientOrTransaction, id: string) { + return await tx.jobRun.findUnique({ + where: { id }, + include: { + queue: true, + version: { + include: { + integrations: { + include: { + apiConnectionClient: true, + }, + }, + }, + }, + }, + }); +} + +async function createRunConnections( + tx: PrismaClientOrTransaction, + run: FoundRun +) { + return await run.version.integrations.reduce( + async ( + accP: Promise< + Record< + string, + | { result: "resolved"; connection: ApiConnection } + | { + result: "missing"; + connectionType: ApiConnectionType; + apiConnectionClientId: string; + externalAccountId?: string; + } + > + >, + integration + ) => { + const acc = await accP; + + const connection = run.externalAccountId + ? await tx.apiConnection.findFirst({ + where: { + clientId: integration.apiConnectionClient.id, + connectionType: "EXTERNAL", + externalAccountId: run.externalAccountId, + }, + }) + : await tx.apiConnection.findFirst({ + where: { + clientId: integration.apiConnectionClient.id, + connectionType: "DEVELOPER", + }, + }); + + if (connection) { + acc[integration.key] = { result: "resolved", connection }; + } else { + acc[integration.key] = { + result: "missing", + connectionType: run.externalAccountId ? "EXTERNAL" : "DEVELOPER", + externalAccountId: run.externalAccountId ?? undefined, + apiConnectionClientId: integration.apiConnectionClient.id, + }; + } + + return acc; + }, + Promise.resolve({}) + ); +} + +function hasMissingConnections(runConnectionsByKey: RunConnectionsByKey) { + return Object.values(runConnectionsByKey).some( + (connection) => connection.result === "missing" + ); +} diff --git a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts index b7d39f689..4cf0e1ccd 100644 --- a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts +++ b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts @@ -1,7 +1,7 @@ import { z } from "zod"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import { ClientApi } from "../clientApi.server"; +import { EndpointApi } from "../endpointApi"; import { IngestSendEvent } from "../events/ingestSendEvent.server"; import { getSecretStore } from "../secrets/secretStore.server"; @@ -55,7 +55,7 @@ export class DeliverHttpSourceRequestService { // TODO: implement auth for http source requests - const clientApi = new ClientApi( + const clientApi = new EndpointApi( httpSourceRequest.environment.apiKey, httpSourceRequest.endpoint.url ); diff --git a/apps/webapp/app/services/triggers/initializeTrigger.server.ts b/apps/webapp/app/services/triggers/initializeTrigger.server.ts index 8a8ab6bc9..a4ec076ba 100644 --- a/apps/webapp/app/services/triggers/initializeTrigger.server.ts +++ b/apps/webapp/app/services/triggers/initializeTrigger.server.ts @@ -5,7 +5,7 @@ import { import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; import { AuthenticatedEnvironment } from "../apiAuth.server"; -import { ClientApi } from "../clientApi.server"; +import { EndpointApi } from "../endpointApi"; import { RegisterTriggerSourceService } from "./registerTriggerSource.server"; import { IngestSendEvent } from "../events/ingestSendEvent.server"; @@ -49,7 +49,7 @@ export class InitializeTriggerService { }, }); - const clientApi = new ClientApi(environment.apiKey, endpoint.url); + const clientApi = new EndpointApi(environment.apiKey, endpoint.url); const registerMetadata = await clientApi.initializeTrigger( dynamicTrigger.slug, diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index e9b720128..b953a46a9 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -4,7 +4,6 @@ import { ZodWorker } from "~/platform/zodWorker.server"; import { EndpointRegisteredService } from "./endpoints/endpointRegistered.server"; import { apiAuthenticationRepository } from "./externalApis/apiAuthenticationRepository.server"; import { RegisterJobService } from "./jobs/registerJob.server"; -import { ResumeTaskService } from "./runs/resumeTask.server"; import { StartRunService } from "./runs/startRun.server"; import { DeliverHttpSourceRequestService } from "./sources/deliverHttpSourceRequest.server"; import { StartQueuedRunsService } from "./runs/startQueuedRuns.server"; @@ -28,6 +27,7 @@ import { MissingConnectionCreatedService } from "./runs/missingConnectionCreated import { ApiConnectionCreatedService } from "./externalApis/apiConnectionCreated.server"; import { sendEmail } from "./email.server"; import { DeliverEmailSchema } from "@/../../packages/emails/src"; +import { PerformRunExecutionService } from "./runs/performRunExecution"; const workerCatalog = { organizationCreated: z.object({ id: z.string() }), @@ -42,8 +42,10 @@ const workerCatalog = { stopVM: z.object({ id: z.string() }), startInitialProjectDeployment: z.object({ id: z.string() }), startRun: z.object({ id: z.string() }), + performRunExecution: z.object({ + id: z.string(), + }), runFinished: z.object({ id: z.string() }), - resumeTask: z.object({ id: z.string() }), deliverHttpSourceRequest: z.object({ id: z.string() }), refreshOAuthToken: z.object({ organizationId: z.string(), @@ -226,11 +228,11 @@ function getWorkerQueue() { await service.call(payload.id); }, }, - resumeTask: { - queueName: "executions", - maxAttempts: 13, + performRunExecution: { + queueName: (payload) => `runs:${payload.id}`, + maxAttempts: 1, handler: async (payload, job) => { - const service = new ResumeTaskService(); + const service = new PerformRunExecutionService(); await service.call(payload.id); }, diff --git a/apps/webapp/app/utils/json.ts b/apps/webapp/app/utils/json.ts index 2da50949c..ab4f9d74e 100644 --- a/apps/webapp/app/utils/json.ts +++ b/apps/webapp/app/utils/json.ts @@ -1,3 +1,5 @@ +import { z } from "zod"; + export function safeJsonParse(json: string): unknown { try { return JSON.parse(json); @@ -5,3 +7,16 @@ export function safeJsonParse(json: string): unknown { return null; } } + +export function safeJsonZodParse( + schema: z.Schema, + json: string +): z.SafeParseReturnType | undefined { + const parsed = safeJsonParse(json); + + if (parsed === null) { + return; + } + + return schema.safeParse(parsed); +} diff --git a/apps/webapp/prisma/migrations/20230606081054_add_job_run_execution/migration.sql b/apps/webapp/prisma/migrations/20230606081054_add_job_run_execution/migration.sql new file mode 100644 index 000000000..4bd24e02d --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606081054_add_job_run_execution/migration.sql @@ -0,0 +1,29 @@ +-- CreateEnum +CREATE TYPE "JobRunExecutionReason" AS ENUM ('INITIAL', 'RETRY', 'RESUME'); + +-- CreateEnum +CREATE TYPE "JobRunExecutionStatus" AS ENUM ('PENDING', 'STARTED', 'SUCCESS', 'FAILURE'); + +-- CreateTable +CREATE TABLE "JobRunExecution" ( + "id" TEXT NOT NULL, + "runId" TEXT NOT NULL, + "number" INTEGER NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + "startedAt" TIMESTAMP(3), + "completedAt" TIMESTAMP(3), + "responseStatus" INTEGER, + "responseHeaders" JSONB, + "responseBody" TEXT, + "reason" "JobRunExecutionReason" NOT NULL DEFAULT 'INITIAL', + "status" "JobRunExecutionStatus" NOT NULL DEFAULT 'PENDING', + + CONSTRAINT "JobRunExecution_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "JobRunExecution_runId_number_key" ON "JobRunExecution"("runId", "number"); + +-- AddForeignKey +ALTER TABLE "JobRunExecution" ADD CONSTRAINT "JobRunExecution_runId_fkey" FOREIGN KEY ("runId") REFERENCES "JobRun"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20230606081441_add_preprocess_job_execution_reason/migration.sql b/apps/webapp/prisma/migrations/20230606081441_add_preprocess_job_execution_reason/migration.sql new file mode 100644 index 000000000..29c250dce --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606081441_add_preprocess_job_execution_reason/migration.sql @@ -0,0 +1,2 @@ +-- AlterEnum +ALTER TYPE "JobRunExecutionReason" ADD VALUE 'PREPROCESS'; diff --git a/apps/webapp/prisma/migrations/20230606081946_add_preprocess_runs_to_job_version/migration.sql b/apps/webapp/prisma/migrations/20230606081946_add_preprocess_runs_to_job_version/migration.sql new file mode 100644 index 000000000..50058919c --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606081946_add_preprocess_runs_to_job_version/migration.sql @@ -0,0 +1,15 @@ +/* + Warnings: + + - You are about to drop the column `latest` on the `JobVersion` table. All the data in the column will be lost. + - You are about to drop the column `prepare` on the `JobVersion` table. All the data in the column will be lost. + - You are about to drop the column `prepared` on the `JobVersion` table. All the data in the column will be lost. + - You are about to drop the column `ready` on the `JobVersion` table. All the data in the column will be lost. + +*/ +-- AlterTable +ALTER TABLE "JobVersion" DROP COLUMN "latest", +DROP COLUMN "prepare", +DROP COLUMN "prepared", +DROP COLUMN "ready", +ADD COLUMN "preprocessRuns" BOOLEAN NOT NULL DEFAULT false; diff --git a/apps/webapp/prisma/migrations/20230606082119_add_preprocess_to_runs/migration.sql b/apps/webapp/prisma/migrations/20230606082119_add_preprocess_to_runs/migration.sql new file mode 100644 index 000000000..6e7e8f268 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606082119_add_preprocess_to_runs/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "JobRun" ADD COLUMN "preprocess" BOOLEAN NOT NULL DEFAULT false; diff --git a/apps/webapp/prisma/migrations/20230606082719_add_preprocess_status_to_runs/migration.sql b/apps/webapp/prisma/migrations/20230606082719_add_preprocess_status_to_runs/migration.sql new file mode 100644 index 000000000..d0fc99f07 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606082719_add_preprocess_status_to_runs/migration.sql @@ -0,0 +1,10 @@ +-- AlterEnum +-- This migration adds more than one value to an enum. +-- With PostgreSQL versions 11 and earlier, this is not possible +-- in a single migration. This can be worked around by creating +-- multiple migrations, each migration adding only one value to +-- the enum. + + +ALTER TYPE "JobRunStatus" ADD VALUE 'PREPROCESSING'; +ALTER TYPE "JobRunStatus" ADD VALUE 'ABORTED'; diff --git a/apps/webapp/prisma/migrations/20230606120402_add_task_to_run_execution/migration.sql b/apps/webapp/prisma/migrations/20230606120402_add_task_to_run_execution/migration.sql new file mode 100644 index 000000000..3241a7941 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606120402_add_task_to_run_execution/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "JobRunExecution" ADD COLUMN "resumeTaskId" TEXT; + +-- AddForeignKey +ALTER TABLE "JobRunExecution" ADD CONSTRAINT "JobRunExecution_resumeTaskId_fkey" FOREIGN KEY ("resumeTaskId") REFERENCES "Task"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20230606122534_improve_run_execution_model/migration.sql b/apps/webapp/prisma/migrations/20230606122534_improve_run_execution_model/migration.sql new file mode 100644 index 000000000..723d44b88 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606122534_improve_run_execution_model/migration.sql @@ -0,0 +1,28 @@ +/* + Warnings: + + - The values [INITIAL,RETRY,RESUME] on the enum `JobRunExecutionReason` will be removed. If these variants are still used in the database, this will fail. + - You are about to drop the column `number` on the `JobRunExecution` table. All the data in the column will be lost. + +*/ +-- AlterEnum +BEGIN; +CREATE TYPE "JobRunExecutionReason_new" AS ENUM ('PREPROCESS', 'EXECUTE_JOB'); +ALTER TABLE "JobRunExecution" ALTER COLUMN "reason" DROP DEFAULT; +ALTER TABLE "JobRunExecution" ALTER COLUMN "reason" TYPE "JobRunExecutionReason_new" USING ("reason"::text::"JobRunExecutionReason_new"); +ALTER TYPE "JobRunExecutionReason" RENAME TO "JobRunExecutionReason_old"; +ALTER TYPE "JobRunExecutionReason_new" RENAME TO "JobRunExecutionReason"; +DROP TYPE "JobRunExecutionReason_old"; +ALTER TABLE "JobRunExecution" ALTER COLUMN "reason" SET DEFAULT 'EXECUTE_JOB'; +COMMIT; + +-- DropIndex +DROP INDEX "JobRunExecution_runId_number_key"; + +-- AlterTable +ALTER TABLE "JobRunExecution" DROP COLUMN "number", +ADD COLUMN "error" TEXT, +ADD COLUMN "retryCount" INTEGER NOT NULL DEFAULT 0, +ADD COLUMN "retryDelayInMs" INTEGER NOT NULL DEFAULT 0, +ADD COLUMN "retryLimit" INTEGER NOT NULL DEFAULT 0, +ALTER COLUMN "reason" SET DEFAULT 'EXECUTE_JOB'; diff --git a/apps/webapp/prisma/migrations/20230606132031_removed_http_responses_from_executions/migration.sql b/apps/webapp/prisma/migrations/20230606132031_removed_http_responses_from_executions/migration.sql new file mode 100644 index 000000000..20866adff --- /dev/null +++ b/apps/webapp/prisma/migrations/20230606132031_removed_http_responses_from_executions/migration.sql @@ -0,0 +1,12 @@ +/* + Warnings: + + - You are about to drop the column `responseBody` on the `JobRunExecution` table. All the data in the column will be lost. + - You are about to drop the column `responseHeaders` on the `JobRunExecution` table. All the data in the column will be lost. + - You are about to drop the column `responseStatus` on the `JobRunExecution` table. All the data in the column will be lost. + +*/ +-- AlterTable +ALTER TABLE "JobRunExecution" DROP COLUMN "responseBody", +DROP COLUMN "responseHeaders", +DROP COLUMN "responseStatus"; diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index cf21b30d4..9a3375525 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -346,12 +346,8 @@ model JobVersion { queue JobQueue @relation(fields: [queueId], references: [id]) queueId String - ready Boolean @default(false) - latest Boolean @default(false) - prepare Boolean @default(false) - prepared Boolean @default(false) - - startPosition JobStartPosition @default(INITIAL) + startPosition JobStartPosition @default(INITIAL) + preprocessRuns Boolean @default(false) createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@ -566,21 +562,61 @@ model JobRun { timedOutAt DateTime? timedOutReason String? - isTest Boolean @default(false) + isTest Boolean @default(false) + preprocess Boolean @default(false) tasks Task[] runConnections RunConnection[] missingConnections MissingApiConnection[] + executions JobRunExecution[] } enum JobRunStatus { PENDING QUEUED WAITING_ON_CONNECTIONS + PREPROCESSING STARTED SUCCESS FAILURE TIMED_OUT + ABORTED +} + +model JobRunExecution { + id String @id @default(cuid()) + + run JobRun @relation(fields: [runId], references: [id], onDelete: Cascade, onUpdate: Cascade) + runId String + + retryCount Int @default(0) + retryLimit Int @default(0) + retryDelayInMs Int @default(0) + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + startedAt DateTime? + completedAt DateTime? + + error String? + + reason JobRunExecutionReason @default(EXECUTE_JOB) + status JobRunExecutionStatus @default(PENDING) + + resumeTask Task? @relation(fields: [resumeTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade) + resumeTaskId String? +} + +enum JobRunExecutionReason { + PREPROCESS + EXECUTE_JOB +} + +enum JobRunExecutionStatus { + PENDING + STARTED + SUCCESS + FAILURE } model Task { @@ -617,7 +653,8 @@ model Task { runConnection RunConnection? @relation(fields: [runConnectionId], references: [id], onDelete: Cascade, onUpdate: Cascade) runConnectionId String? - children Task[] @relation("TaskParent") + children Task[] @relation("TaskParent") + executions JobRunExecution[] @@unique([runId, idempotencyKey]) } diff --git a/examples/nextjs-example/src/pages/api/trigger.ts b/examples/nextjs-example/src/pages/api/trigger.ts index 1fd5b369f..c2de2db51 100644 --- a/examples/nextjs-example/src/pages/api/trigger.ts +++ b/examples/nextjs-example/src/pages/api/trigger.ts @@ -1,7 +1,6 @@ import { cronTrigger, - customEvent, - customTrigger, + eventTrigger, DynamicSchedule, DynamicTrigger, intervalTrigger, @@ -132,12 +131,10 @@ new Job(client, { name: "Get User Repo", version: "0.1.1", enabled, - trigger: customTrigger({ + trigger: eventTrigger({ name: "get.repo", - event: customEvent({ - payload: z.object({ - repo: z.string(), - }), + schema: z.object({ + repo: z.string(), }), }), integrations: { @@ -175,13 +172,11 @@ new Job(client, { name: "Register Dynamic Interval", version: "0.1.1", enabled, - trigger: customTrigger({ + trigger: eventTrigger({ name: "dynamic.interval", - event: customEvent({ - payload: z.object({ - id: z.string(), - seconds: z.number().int().positive(), - }), + schema: z.object({ + id: z.string(), + seconds: z.number().int().positive(), }), }), run: async (payload, io, ctx) => { @@ -200,13 +195,11 @@ new Job(client, { name: "Register Dynamic Cron", version: "0.1.1", enabled, - trigger: customTrigger({ + trigger: eventTrigger({ name: "dynamic.cron", - event: customEvent({ - payload: z.object({ - id: z.string(), - cron: z.string(), - }), + schema: z.object({ + id: z.string(), + cron: z.string(), }), }), run: async (payload, io, ctx) => { @@ -231,7 +224,7 @@ new Job(client, { await io.logger.info("This is a log info message", { payload, }); - await io.sendCustomEvent("send-event", { + await io.sendEvent("send-event", { name: "custom.event", payload, context: ctx, @@ -252,7 +245,7 @@ new Job(client, { await io.logger.info("This is a log info message", { payload, }); - await io.sendCustomEvent("send-event", { + await io.sendEvent("send-event", { name: "custom.event", payload, context: ctx, @@ -282,18 +275,15 @@ new Job(client, { name: "Test IO functions", version: "0.1.1", enabled, - trigger: customTrigger({ + trigger: eventTrigger({ name: "test.io", - event: customEvent({ - payload: z.any(), - }), }), run: async (payload, io, ctx) => { await io.wait("wait", 5); // wait for 5 seconds await io.logger.info("This is a log info message", { payload, }); - await io.sendCustomEvent("send-event", { + await io.sendEvent("send-event", { name: "custom.event", payload, context: ctx, @@ -306,11 +296,9 @@ new Job(client, { name: "Register dynamic trigger on new repo", version: "0.1.1", enabled, - trigger: customTrigger({ + trigger: eventTrigger({ name: "new.repo", - event: customEvent({ - payload: z.object({ repo: z.string() }), - }), + schema: z.object({ repo: z.string() }), }), run: async (payload, io, ctx) => { return await io.registerTrigger( @@ -378,7 +366,7 @@ new Job(client, { }); new Job(client, { - id: "alert-on-new-github-issues", + id: "alert-on-new-github-issues-3", name: "Alert on new GitHub issues", version: "0.1.1", enabled, @@ -387,16 +375,19 @@ new Job(client, { }, trigger: github.triggers.repo({ event: events.onIssueOpened, - repo: "ericallam/basic-starter-100k", + repo: "ericallam/basic-starter-12k", }), run: async (payload, io, ctx) => { - //todo logging isn't working - // await io.logger.info("This is a simple log info message"); + await io.wait("wait", 5); // wait for 5 seconds + + await io.logger.info("This is a simple log info message"); + const response = await io.slack.postMessage("Slack 📝", { text: `New Issue opened: ${payload.issue.html_url}`, channel: "C04GWUTDC3W", }); - // await io.logger.warn("You've been warned", response); + + return response; }, }); diff --git a/integrations/github/src/index.ts b/integrations/github/src/index.ts index fa5a5dd06..af28d4c35 100644 --- a/integrations/github/src/index.ts +++ b/integrations/github/src/index.ts @@ -93,6 +93,18 @@ const onIssueOpened: EventSpecification = { action: ["opened"], }, parsePayload: (payload) => payload as IssuesOpenedEvent, + runElements: (payload) => [ + { + label: "Issue", + text: `#${payload.issue.number}: ${payload.issue.title}`, + url: payload.issue.html_url, + }, + { + label: "Author", + text: payload.sender.login, + url: payload.sender.html_url, + }, + ], }; const onIssue: EventSpecification = { @@ -101,6 +113,18 @@ const onIssue: EventSpecification = { source: "github.com", icon: "github", parsePayload: (payload) => payload as IssuesEvent, + runElements: (payload) => [ + { + label: "Issue", + text: `#${payload.issue.number}: ${payload.issue.title}`, + url: payload.issue.html_url, + }, + { + label: "Author", + text: payload.sender.login, + url: payload.sender.html_url, + }, + ], }; const onIssueComment: EventSpecification = { @@ -109,6 +133,18 @@ const onIssueComment: EventSpecification = { source: "github.com", icon: "github", parsePayload: (payload) => payload as IssueCommentEvent, + runElements: (payload) => [ + { + label: "Issue", + text: `#${payload.issue.number}: ${payload.issue.title}`, + url: payload.issue.html_url, + }, + { + label: "Author", + text: payload.sender.login, + url: payload.sender.html_url, + }, + ], }; const onStar: EventSpecification = { diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index 72d00a6e1..dab658fff 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -142,6 +142,7 @@ export const JobMetadataSchema = z.object({ queue: z.union([QueueOptionsSchema, z.string()]).optional(), startPosition: z.enum(["initial", "latest"]), enabled: z.boolean(), + preprocessRuns: z.boolean(), }); export type JobMetadata = z.infer; @@ -273,6 +274,43 @@ export const RunJobResponseSchema = z.object({ export type RunJobResponse = z.infer; +export const PreprocessRunBodySchema = z.object({ + event: ApiEventLogSchema, + job: z.object({ + id: z.string(), + version: z.string(), + }), + run: z.object({ + id: z.string(), + isTest: z.boolean(), + }), + environment: z.object({ + id: z.string(), + slug: z.string(), + type: RuntimeEnvironmentTypeSchema, + }), + organization: z.object({ + id: z.string(), + title: z.string(), + slug: z.string(), + }), + account: z + .object({ + id: z.string(), + metadata: z.any(), + }) + .optional(), +}); + +export type PreprocessRunBody = z.infer; + +export const PreprocessRunResponseSchema = z.object({ + abort: z.boolean(), + elements: z.array(DisplayElementSchema).optional(), +}); + +export type PreprocessRunResponse = z.infer; + export const CreateRunBodySchema = z.object({ client: z.string(), job: JobMetadataSchema, @@ -307,23 +345,6 @@ export const SecureStringSchema = z.object({ interpolations: z.array(z.string()), }); -export const PrepareJobTriggerBodySchema = z.object({ - id: z.string(), - version: z.string(), - connection: ConnectionAuthSchema.optional(), - variantId: z.string().optional(), -}); - -export type PrepareJobTriggerBody = z.infer; - -export const PrepareForJobExecutionResponseSchema = z.object({ - ok: z.boolean(), -}); - -export type PrepareForJobExecutionResponse = z.infer< - typeof PrepareForJobExecutionResponseSchema ->; - export type SecureString = z.infer; export const LogMessageSchema = z.object({ @@ -410,18 +431,6 @@ export const HttpSourceResponseSchema = z.object({ events: z.array(RawEventSchema), }); -export const TriggerVariantResponseBodySchema = z.object({ - id: z.string(), - slug: z.string(), - data: TriggerMetadataSchema, - ready: z.boolean(), - auth: ConnectionAuthSchema.optional(), -}); - -export type TriggerVariantResponseBody = z.infer< - typeof TriggerVariantResponseBodySchema ->; - export const RegisterTriggerBodySchema = z.object({ rule: EventRuleSchema, source: SourceMetadataSchema, diff --git a/packages/trigger-sdk/src/index.ts b/packages/trigger-sdk/src/index.ts index e109dc64e..a61bd9094 100644 --- a/packages/trigger-sdk/src/index.ts +++ b/packages/trigger-sdk/src/index.ts @@ -1,7 +1,7 @@ export * from "./job"; export * from "./triggerClient"; export * from "./integrations"; -export * from "./triggers/customTrigger"; +export * from "./triggers/eventTrigger"; export * from "./triggers/externalSource"; export * from "./triggers/dynamic"; export * from "./triggers/scheduled"; diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts index ff3446083..1107e8011 100644 --- a/packages/trigger-sdk/src/io.ts +++ b/packages/trigger-sdk/src/io.ts @@ -120,7 +120,7 @@ export class IO { ); } - async sendCustomEvent( + async sendEvent( key: string | any[], event: SendEvent, options?: SendEventOptions @@ -128,7 +128,7 @@ export class IO { return await this.runTask( key, { - name: "sendCustomEvent", + name: "sendEvent", params: { event, options }, }, async (task) => { diff --git a/packages/trigger-sdk/src/job.ts b/packages/trigger-sdk/src/job.ts index d5f43f77e..c2a0d5bfd 100644 --- a/packages/trigger-sdk/src/job.ts +++ b/packages/trigger-sdk/src/job.ts @@ -15,6 +15,7 @@ import type { Trigger, TriggerContext, TriggerEventType, + TriggerPreprocessContext, } from "./types"; import { slugifyId } from "./utils"; @@ -119,6 +120,7 @@ export class Job< startPosition: this.options.startPosition ?? "latest", enabled: typeof this.options.enabled === "boolean" ? this.options.enabled : true, + preprocessRuns: this.trigger.preprocessRuns, internal, }; } diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 4e20031ad..372c2db47 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -10,6 +10,8 @@ import { Logger, NormalizedRequest, NormalizedResponse, + PreprocessRunBody, + PreprocessRunBodySchema, REGISTER_SOURCE_EVENT, RegisterSourceEvent, RegisterSourceEventSchema, @@ -25,9 +27,14 @@ import { ApiClient } from "./apiClient"; import { IO, ResumeWithTask } from "./io"; import { createIOWithIntegrations } from "./ioWithIntegrations"; import { Job } from "./job"; -import { CustomTrigger } from "./triggers/customTrigger"; +import { EventTrigger } from "./triggers/eventTrigger"; import { ExternalSource, HttpSourceEvent } from "./triggers/externalSource"; -import type { EventSpecification, Trigger, TriggerContext } from "./types"; +import type { + EventSpecification, + Trigger, + TriggerContext, + TriggerPreprocessContext, +} from "./types"; import { DynamicTrigger } from "./triggers/dynamic"; const registerSourceEvent: EventSpecification = { @@ -243,6 +250,39 @@ export class TriggerClient { }, }; } + case "PREPROCESS_RUN": { + const body = PreprocessRunBodySchema.safeParse(request.body); + + if (!body.success) { + return { + status: 400, + body: { + message: "Invalid body", + }, + }; + } + + const job = this.#registeredJobs[body.data.job.id]; + + if (!job) { + return { + status: 404, + body: { + message: "Job not found", + }, + }; + } + + const results = await this.#preprocessRun(body.data, job); + + return { + status: 200, + body: { + abort: results.abort, + elements: results.elements, + }, + }; + } case "DELIVER_HTTP_SOURCE_REQUEST": { const headers = HttpSourceRequestHeadersSchema.safeParse( request.headers @@ -319,7 +359,7 @@ export class TriggerClient { id: `register-dynamic-trigger-${trigger.id}`, name: `Register dynamic trigger ${trigger.id}`, version: trigger.source.version, - trigger: new CustomTrigger({ + trigger: new EventTrigger({ event: registerSourceEvent, filter: { dynamicTriggerId: [trigger.id] }, }), @@ -394,7 +434,7 @@ export class TriggerClient { id: options.key, name: options.key, version: options.source.version, - trigger: new CustomTrigger({ + trigger: new EventTrigger({ event: registerSourceEvent, filter: { source: { key: [options.key] } }, }), @@ -479,14 +519,32 @@ export class TriggerClient { }); } - async #executeJob(execution: RunJobBody, job: Job, any>) { - this.#logger.debug("executing job", { execution, job: job.toJSON() }); + async #preprocessRun( + body: PreprocessRunBody, + job: Job>, any> + ) { + const context = this.#createPreprocessRunContext(body); - const context = this.#createRunContext(execution); + const parsedPayload = job.trigger.event.parsePayload( + body.event.payload ?? {} + ); + + const elements = job.trigger.event.runElements?.(parsedPayload) ?? []; + + return { + abort: false, + elements, + }; + } + + async #executeJob(body: RunJobBody, job: Job, any>) { + this.#logger.debug("executing job", { execution: body, job: job.toJSON() }); + + const context = this.#createRunContext(body); const io = new IO({ - id: execution.run.id, - cachedTasks: execution.tasks, + id: body.run.id, + cachedTasks: body.tasks, apiClient: this.#client, logger: this.#logger, client: this, @@ -495,13 +553,13 @@ export class TriggerClient { const ioWithConnections = createIOWithIntegrations( io, - execution.connections, + body.connections, job.options.integrations ); try { const output = await job.options.run( - job.trigger.event.parsePayload(execution.event.payload ?? {}), + job.trigger.event.parsePayload(body.event.payload ?? {}), ioWithConnections, context ); @@ -549,6 +607,26 @@ export class TriggerClient { }; } + #createPreprocessRunContext( + body: PreprocessRunBody + ): TriggerPreprocessContext { + const { event, organization, environment, job, run, account } = body; + + return { + event: { + id: event.id, + name: event.name, + context: event.context, + timestamp: event.timestamp, + }, + organization, + environment, + job, + run, + account, + }; + } + async #handleHttpSourceRequest( source: { key: string; diff --git a/packages/trigger-sdk/src/triggers/dynamic.ts b/packages/trigger-sdk/src/triggers/dynamic.ts index f39aa109a..8865e3d2f 100644 --- a/packages/trigger-sdk/src/triggers/dynamic.ts +++ b/packages/trigger-sdk/src/triggers/dynamic.ts @@ -6,7 +6,12 @@ import { } from "@trigger.dev/internal"; import { Job } from "../job"; import { TriggerClient } from "../triggerClient"; -import { EventSpecification, Trigger } from "../types"; +import { + EventSpecification, + PreprocessResults, + Trigger, + TriggerPreprocessContext, +} from "../types"; import { ExternalSource, ExternalSourceParams } from "./externalSource"; import { slugifyId } from "../utils"; @@ -54,10 +59,6 @@ export class DynamicTrigger< return this.#options.event; } - get requiresPreparaton(): boolean { - return false; - } - registeredTriggerForParams( params: ExternalSourceParams ): RegisterTriggerBody { @@ -99,4 +100,8 @@ export class DynamicTrigger< ): void { triggerClient.attachJobToDynamicTrigger(job, this); } + + get preprocessRuns() { + return false; + } } diff --git a/packages/trigger-sdk/src/triggers/customTrigger.ts b/packages/trigger-sdk/src/triggers/eventTrigger.ts similarity index 53% rename from packages/trigger-sdk/src/triggers/customTrigger.ts rename to packages/trigger-sdk/src/triggers/eventTrigger.ts index 06513538f..a3f8bfdd7 100644 --- a/packages/trigger-sdk/src/triggers/customTrigger.ts +++ b/packages/trigger-sdk/src/triggers/eventTrigger.ts @@ -1,14 +1,14 @@ -import { z } from "zod"; -import { Job } from "../job"; -import { TriggerClient } from "../triggerClient"; -import { EventSpecification, Trigger } from "../types"; import { EventFilter, TriggerMetadata, deepMergeFilters, } from "@trigger.dev/internal"; +import { z } from "zod"; +import { Job } from "../job"; +import { TriggerClient } from "../triggerClient"; +import { EventSpecification, Trigger } from "../types"; -type CustomTriggerOptions> = +type EventTriggerOptions> = { event: TEventSpecification; name?: string; @@ -16,12 +16,12 @@ type CustomTriggerOptions> = filter?: EventFilter; }; -export class CustomTrigger> +export class EventTrigger> implements Trigger { - #options: CustomTriggerOptions; + #options: EventTriggerOptions; - constructor(options: CustomTriggerOptions) { + constructor(options: EventTriggerOptions) { this.#options = options; } @@ -40,10 +40,6 @@ export class CustomTrigger> }; } - get requiresPreparaton(): boolean { - return false; - } - get event() { return this.#options.event; } @@ -52,30 +48,37 @@ export class CustomTrigger> triggerClient: TriggerClient, job: Job, any> ): void {} + + get preprocessRuns() { + return false; + } } -export function customTrigger< - TEventSpecification extends EventSpecification ->( - options: CustomTriggerOptions -): Trigger { - return new CustomTrigger(options); -} - -export function customEvent({ - payload, - source, -}: { - payload: z.Schema; +type TriggerOptions = { + name: string; + schema?: z.Schema; source?: string; -}): EventSpecification { - return { - name: "custom", - title: "Custom Event", - source: source ?? "trigger.dev", - icon: "custom-event", - parsePayload: (rawPayload: any) => { - return payload.parse(rawPayload); + filter?: EventFilter; +}; + +export function eventTrigger( + options: TriggerOptions +): Trigger> { + return new EventTrigger({ + name: "Event Trigger", + filter: options.filter, + event: { + name: options.name, + title: "Event", + source: options.source ?? "trigger.dev", + icon: "custom-event", + parsePayload: (rawPayload: any) => { + if (options.schema) { + return options.schema.parse(rawPayload); + } + + return rawPayload as any; + }, }, - }; + }); } diff --git a/packages/trigger-sdk/src/triggers/externalSource.ts b/packages/trigger-sdk/src/triggers/externalSource.ts index 3c738dd05..f61fd2c68 100644 --- a/packages/trigger-sdk/src/triggers/externalSource.ts +++ b/packages/trigger-sdk/src/triggers/externalSource.ts @@ -13,8 +13,8 @@ import { deepMergeFilters, } from "@trigger.dev/internal"; import { - IntegrationClient, IOWithIntegrations, + IntegrationClient, TriggerIntegration, } from "../integrations"; import { IO } from "../io"; @@ -234,10 +234,6 @@ export class ExternalSourceTrigger< return this.options.event; } - get requiresPreparaton(): boolean { - return true; - } - toJSON(): TriggerMetadata { return { type: "static", @@ -265,6 +261,10 @@ export class ExternalSourceTrigger< params: this.options.params, }); } + + get preprocessRuns() { + return true; + } } export function omit, K extends keyof T>( diff --git a/packages/trigger-sdk/src/triggers/notifications.ts b/packages/trigger-sdk/src/triggers/notifications.ts index 2ba6b9b9b..12eb0971a 100644 --- a/packages/trigger-sdk/src/triggers/notifications.ts +++ b/packages/trigger-sdk/src/triggers/notifications.ts @@ -57,6 +57,10 @@ export class MissingConnectionNotification job: Job, any> ): void {} + get preprocessRuns() { + return false; + } + toJSON(): TriggerMetadata { return { type: "static", @@ -72,10 +76,6 @@ export class MissingConnectionNotification }, }; } - - get requiresPreparaton(): boolean { - return false; - } } type MissingConnectionResolvedNotificationSpecification = @@ -107,6 +107,10 @@ export class MissingConnectionResolvedNotification job: Job, any> ): void {} + get preprocessRuns() { + return false; + } + toJSON(): TriggerMetadata { return { type: "static", @@ -122,8 +126,4 @@ export class MissingConnectionResolvedNotification }, }; } - - get requiresPreparaton(): boolean { - return false; - } } diff --git a/packages/trigger-sdk/src/triggers/scheduled.ts b/packages/trigger-sdk/src/triggers/scheduled.ts index 8b8319652..959a7fde2 100644 --- a/packages/trigger-sdk/src/triggers/scheduled.ts +++ b/packages/trigger-sdk/src/triggers/scheduled.ts @@ -1,8 +1,3 @@ -import { z } from "zod"; -import { EventSpecification } from "../types"; -import { Trigger } from "../types"; -import { TriggerClient } from "../triggerClient"; -import { Job } from "../job"; import { CronOptions, IntervalOptions, @@ -11,6 +6,9 @@ import { ScheduledPayloadSchema, TriggerMetadata, } from "@trigger.dev/internal"; +import { Job } from "../job"; +import { TriggerClient } from "../triggerClient"; +import { EventSpecification, Trigger } from "../types"; type ScheduledEventSpecification = EventSpecification; @@ -38,6 +36,10 @@ export class IntervalTrigger implements Trigger { job: Job, any> ): void {} + get preprocessRuns() { + return false; + } + toJSON(): TriggerMetadata { return { type: "scheduled", @@ -49,10 +51,6 @@ export class IntervalTrigger implements Trigger { }, }; } - - get requiresPreparaton(): boolean { - return false; - } } export function intervalTrigger(options: IntervalOptions) { @@ -83,6 +81,10 @@ export class CronTrigger implements Trigger { job: Job, any> ): void {} + get preprocessRuns() { + return false; + } + toJSON(): TriggerMetadata { return { type: "scheduled", @@ -94,10 +96,6 @@ export class CronTrigger implements Trigger { }, }; } - - get requiresPreparaton(): boolean { - return false; - } } export function cronTrigger(options: CronOptions) { @@ -141,14 +139,14 @@ export class DynamicSchedule implements Trigger { triggerClient.attachDynamicSchedule(this.options.id, job); } + get preprocessRuns() { + return false; + } + toJSON(): TriggerMetadata { return { type: "dynamic", id: this.options.id, }; } - - get requiresPreparaton(): boolean { - return false; - } } diff --git a/packages/trigger-sdk/src/types.ts b/packages/trigger-sdk/src/types.ts index 82db829ad..f24ea22e5 100644 --- a/packages/trigger-sdk/src/types.ts +++ b/packages/trigger-sdk/src/types.ts @@ -19,6 +19,15 @@ export interface TriggerContext { account?: { id: string; metadata?: any }; } +export interface TriggerPreprocessContext { + job: { id: string; version: string }; + environment: { slug: string; id: string; type: RuntimeEnvironmentType }; + organization: { slug: string; id: string; title: string }; + run: { id: string; isTest: boolean }; + event: { id: string; name: string; context: any; timestamp: Date }; + account?: { id: string; metadata?: any }; +} + export interface TaskLogger { debug(message: string, properties?: Record): Promise; info(message: string, properties?: Record): Promise; @@ -26,6 +35,11 @@ export interface TaskLogger { error(message: string, properties?: Record): Promise; } +export type PreprocessResults = { + abort: boolean; + elements: DisplayElement[]; +}; + export type TriggerEventType> = TTrigger extends Trigger ? ReturnType @@ -40,7 +54,8 @@ export interface Trigger> { triggerClient: TriggerClient, job: Job, any> ): void; - requiresPreparaton: boolean; + + preprocessRuns: boolean; } export interface EventSpecification { @@ -53,6 +68,7 @@ export interface EventSpecification { examples?: Array; filter?: EventFilter; parsePayload: (payload: unknown) => TEvent; + runElements?: (payload: TEvent) => DisplayElement[]; } export type EventTypeFromSpecification<