github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Re-architect pre-processing and executing jobs, and gather run elements

Browse Source
This commit is contained in:
Eric Allam
2023-06-07 13:11:58 +01:00
parent a03912b804
commit 2ef99d8383
36 changed files with 1275 additions and 799 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/consts.ts
+2
View File
@@ -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;
apps/webapp/app/services/clientApi.server.ts → apps/webapp/app/services/endpointApi.ts
+17 -67
View File
@@ -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(
apps/webapp/app/services/endpoints/createEndpoint.server.ts
+2 -2
View File
@@ -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) => {
apps/webapp/app/services/endpoints/endpointRegistered.server.ts
+2 -2
View File
@@ -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();
apps/webapp/app/services/events/invokeDispatcher.server.ts
+1 -2
View File
@@ -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"),
apps/webapp/app/services/jobs/registerJob.server.ts
+2
View File
@@ -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,
apps/webapp/app/services/runs/createRun.server.ts
+1
View File
@@ -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 } },
apps/webapp/app/services/runs/performRunExecution.ts
+549
View File
@@ -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<ReturnType<typeof findRunExecution>>
>;
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<string, any>
): Promise<void> {
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<string, any>,
status: "FAILURE" | "ABORTED" = "FAILURE"
): Promise<void> {
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,
},
},
},
},
},
});
}
apps/webapp/app/services/runs/resumeTask.server.ts
-214
View File
@@ -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 }
);
});
}
}
}
apps/webapp/app/services/runs/startRun.server.ts
+238 -346
View File
@@ -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<Awaited<ReturnType<typeof findRun>>>;
type RunConnectionsByKey = Awaited<ReturnType<typeof createRunConnections>>;
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"
);
}
apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts
+2 -2
View File
@@ -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
);
apps/webapp/app/services/triggers/initializeTrigger.server.ts
+2 -2
View File
@@ -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,
apps/webapp/app/services/worker.server.ts
+8 -6
View File
@@ -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);
},
apps/webapp/app/utils/json.ts
+15
View File
@@ -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<T>(
schema: z.Schema<T>,
json: string
): z.SafeParseReturnType<unknown, T> | undefined {
const parsed = safeJsonParse(json);
if (parsed === null) {
return;
}
return schema.safeParse(parsed);
}
apps/webapp/prisma/migrations/20230606081054_add_job_run_execution/migration.sql
+29
View File
@@ -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;
apps/webapp/prisma/migrations/20230606081441_add_preprocess_job_execution_reason/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "JobRunExecutionReason" ADD VALUE 'PREPROCESS';
apps/webapp/prisma/migrations/20230606081946_add_preprocess_runs_to_job_version/migration.sql
+15
View File
@@ -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;
apps/webapp/prisma/migrations/20230606082119_add_preprocess_to_runs/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "JobRun" ADD COLUMN "preprocess" BOOLEAN NOT NULL DEFAULT false;
apps/webapp/prisma/migrations/20230606082719_add_preprocess_status_to_runs/migration.sql
+10
View File
@@ -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';
apps/webapp/prisma/migrations/20230606120402_add_task_to_run_execution/migration.sql
+5
View File
@@ -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;
apps/webapp/prisma/migrations/20230606122534_improve_run_execution_model/migration.sql
+28
View File
@@ -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';
apps/webapp/prisma/migrations/20230606132031_removed_http_responses_from_executions/migration.sql
+12
View File
@@ -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";
apps/webapp/prisma/schema.prisma
+45 -8
View File
@@ -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])
}
examples/nextjs-example/src/pages/api/trigger.ts
+26 -35
View File
@@ -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;
},
});
integrations/github/src/index.ts
+36
View File
@@ -93,6 +93,18 @@ const onIssueOpened: EventSpecification<IssuesOpenedEvent> = {
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<IssuesEvent> = {
@@ -101,6 +113,18 @@ const onIssue: EventSpecification<IssuesEvent> = {
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<IssueCommentEvent> = {
@@ -109,6 +133,18 @@ const onIssueComment: EventSpecification<IssueCommentEvent> = {
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<StarEvent> = {
packages/internal/src/schemas/api.ts
+38 -29
View File
@@ -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<typeof JobMetadataSchema>;
@@ -273,6 +274,43 @@ export const RunJobResponseSchema = z.object({
export type RunJobResponse = z.infer<typeof RunJobResponseSchema>;
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<typeof PreprocessRunBodySchema>;
export const PreprocessRunResponseSchema = z.object({
abort: z.boolean(),
elements: z.array(DisplayElementSchema).optional(),
});
export type PreprocessRunResponse = z.infer<typeof PreprocessRunResponseSchema>;
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<typeof PrepareJobTriggerBodySchema>;
export const PrepareForJobExecutionResponseSchema = z.object({
ok: z.boolean(),
});
export type PrepareForJobExecutionResponse = z.infer<
typeof PrepareForJobExecutionResponseSchema
>;
export type SecureString = z.infer<typeof SecureStringSchema>;
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,
packages/trigger-sdk/src/index.ts
+1 -1
View File
@@ -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";
packages/trigger-sdk/src/io.ts
+2 -2
View File
@@ -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) => {
packages/trigger-sdk/src/job.ts
+2
View File
@@ -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,
};
}
packages/trigger-sdk/src/triggerClient.ts
+89 -11
View File
@@ -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<RegisterSourceEvent> = {
@@ -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<Trigger<any>, any>) {
this.#logger.debug("executing job", { execution, job: job.toJSON() });
async #preprocessRun(
body: PreprocessRunBody,
job: Job<Trigger<EventSpecification<any>>, 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<Trigger<any>, 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;
packages/trigger-sdk/src/triggers/dynamic.ts
+10 -5
View File
@@ -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<TExternalSource>
): RegisterTriggerBody {
@@ -99,4 +100,8 @@ export class DynamicTrigger<
): void {
triggerClient.attachJobToDynamicTrigger(job, this);
}
get preprocessRuns() {
return false;
}
}
packages/trigger-sdk/src/triggers/customTrigger.ts → packages/trigger-sdk/src/triggers/eventTrigger.ts
+37 -34
View File
@@ -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<TEventSpecification extends EventSpecification<any>> =
type EventTriggerOptions<TEventSpecification extends EventSpecification<any>> =
{
event: TEventSpecification;
name?: string;
@@ -16,12 +16,12 @@ type CustomTriggerOptions<TEventSpecification extends EventSpecification<any>> =
filter?: EventFilter;
};
export class CustomTrigger<TEventSpecification extends EventSpecification<any>>
export class EventTrigger<TEventSpecification extends EventSpecification<any>>
implements Trigger<TEventSpecification>
{
#options: CustomTriggerOptions<TEventSpecification>;
#options: EventTriggerOptions<TEventSpecification>;
constructor(options: CustomTriggerOptions<TEventSpecification>) {
constructor(options: EventTriggerOptions<TEventSpecification>) {
this.#options = options;
}
@@ -40,10 +40,6 @@ export class CustomTrigger<TEventSpecification extends EventSpecification<any>>
};
}
get requiresPreparaton(): boolean {
return false;
}
get event() {
return this.#options.event;
}
@@ -52,30 +48,37 @@ export class CustomTrigger<TEventSpecification extends EventSpecification<any>>
triggerClient: TriggerClient,
job: Job<Trigger<TEventSpecification>, any>
): void {}
get preprocessRuns() {
return false;
}
}
export function customTrigger<
TEventSpecification extends EventSpecification<any>
>(
options: CustomTriggerOptions<TEventSpecification>
): Trigger<TEventSpecification> {
return new CustomTrigger(options);
}
export function customEvent<TEvent>({
payload,
source,
}: {
payload: z.Schema<TEvent>;
type TriggerOptions<TEvent> = {
name: string;
schema?: z.Schema<TEvent>;
source?: string;
}): EventSpecification<TEvent> {
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<TEvent extends any = any>(
options: TriggerOptions<TEvent>
): Trigger<EventSpecification<TEvent>> {
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;
},
},
};
});
}
packages/trigger-sdk/src/triggers/externalSource.ts
+5 -5
View File
@@ -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<T extends Record<string, unknown>, K extends keyof T>(
packages/trigger-sdk/src/triggers/notifications.ts
+8 -8
View File
@@ -57,6 +57,10 @@ export class MissingConnectionNotification
job: Job<Trigger<MissingConnectionNotificationSpecification>, 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<Trigger<MissingConnectionResolvedNotificationSpecification>, any>
): void {}
get preprocessRuns() {
return false;
}
toJSON(): TriggerMetadata {
return {
type: "static",
@@ -122,8 +126,4 @@ export class MissingConnectionResolvedNotification
},
};
}
get requiresPreparaton(): boolean {
return false;
}
}
packages/trigger-sdk/src/triggers/scheduled.ts
+15 -17
View File
@@ -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<ScheduledPayload>;
@@ -38,6 +36,10 @@ export class IntervalTrigger implements Trigger<ScheduledEventSpecification> {
job: Job<Trigger<ScheduledEventSpecification>, any>
): void {}
get preprocessRuns() {
return false;
}
toJSON(): TriggerMetadata {
return {
type: "scheduled",
@@ -49,10 +51,6 @@ export class IntervalTrigger implements Trigger<ScheduledEventSpecification> {
},
};
}
get requiresPreparaton(): boolean {
return false;
}
}
export function intervalTrigger(options: IntervalOptions) {
@@ -83,6 +81,10 @@ export class CronTrigger implements Trigger<ScheduledEventSpecification> {
job: Job<Trigger<ScheduledEventSpecification>, any>
): void {}
get preprocessRuns() {
return false;
}
toJSON(): TriggerMetadata {
return {
type: "scheduled",
@@ -94,10 +96,6 @@ export class CronTrigger implements Trigger<ScheduledEventSpecification> {
},
};
}
get requiresPreparaton(): boolean {
return false;
}
}
export function cronTrigger(options: CronOptions) {
@@ -141,14 +139,14 @@ export class DynamicSchedule implements Trigger<ScheduledEventSpecification> {
triggerClient.attachDynamicSchedule(this.options.id, job);
}
get preprocessRuns() {
return false;
}
toJSON(): TriggerMetadata {
return {
type: "dynamic",
id: this.options.id,
};
}
get requiresPreparaton(): boolean {
return false;
}
}
packages/trigger-sdk/src/types.ts
+17 -1
View File
@@ -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<string, any>): Promise<void>;
info(message: string, properties?: Record<string, any>): Promise<void>;
@@ -26,6 +35,11 @@ export interface TaskLogger {
error(message: string, properties?: Record<string, any>): Promise<void>;
}
export type PreprocessResults = {
abort: boolean;
elements: DisplayElement[];
};
export type TriggerEventType<TTrigger extends Trigger<any>> =
TTrigger extends Trigger<infer TEventSpec>
? ReturnType<TEventSpec["parsePayload"]>
@@ -40,7 +54,8 @@ export interface Trigger<TEventSpec extends EventSpecification<any>> {
triggerClient: TriggerClient,
job: Job<Trigger<TEventSpec>, any>
): void;
requiresPreparaton: boolean;
preprocessRuns: boolean;
}
export interface EventSpecification<TEvent extends any> {
@@ -53,6 +68,7 @@ export interface EventSpecification<TEvent extends any> {
examples?: Array<TEvent>;
filter?: EventFilter;
parsePayload: (payload: unknown) => TEvent;
runElements?: (payload: TEvent) => DisplayElement[];
}
export type EventTypeFromSpecification<