diff --git a/apps/webapp/app/models/jobConnection.server.ts b/apps/webapp/app/models/jobConnection.server.ts index 94209c105..ac6877dbb 100644 --- a/apps/webapp/app/models/jobConnection.server.ts +++ b/apps/webapp/app/models/jobConnection.server.ts @@ -13,30 +13,56 @@ export async function resolveJobConnections( const result: Record = {}; for (const connection of connections) { - if (!connection.apiConnection) { + const auth = await resolveJobConnection(connection); + + if (!auth) { continue; } - const response = await apiConnectionRepository.getCredentials( - connection.apiConnection - ); - - if (!response) { - continue; - } - - if (result[connection.key]) { - throw new Error( - `Duplicate connection key ${connection.key} in job instance ${connection.jobInstanceId}` - ); - } - - result[connection.key] = { - type: "oauth2", - scopes: response.scopes, - accessToken: response.accessToken, - }; + result[connection.key] = auth; } return result; } + +export async function resolveJobConnection( + connection: JobConnectionWithApiConnection +): Promise { + if (!connection.apiConnection) { + return; + } + + const response = await apiConnectionRepository.getCredentials( + connection.apiConnection + ); + + if (!response) { + return; + } + + return { + type: "oauth2", + scopes: response.scopes, + accessToken: response.accessToken, + }; +} + +export async function resolveApiConnection( + connection?: ApiConnectionWithSecretReference +): Promise { + if (!connection) { + return; + } + + const response = await apiConnectionRepository.getCredentials(connection); + + if (!response) { + return; + } + + return { + type: "oauth2", + scopes: response.scopes, + accessToken: response.accessToken, + }; +} diff --git a/apps/webapp/app/routes/api/v3/$endpointSlug/jobs/$jobId.$jobVersion.variants.ts b/apps/webapp/app/routes/api/v3/$endpointSlug/jobs/$jobId.$jobVersion.variants.ts new file mode 100644 index 000000000..e41e292c8 --- /dev/null +++ b/apps/webapp/app/routes/api/v3/$endpointSlug/jobs/$jobId.$jobVersion.variants.ts @@ -0,0 +1,73 @@ +import type { ActionArgs } from "@remix-run/server-runtime"; +import { json } from "@remix-run/server-runtime"; +import { TriggerVariantConfigSchema } from "@trigger.dev/internal"; +import { z } from "zod"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; +import { RegisterJobVariantService } from "~/services/jobs/registerJobVariant.server"; +import { logger } from "~/services/logger"; + +const ParamsSchema = z.object({ + endpointSlug: z.string(), + jobId: z.string(), + jobVersion: z.string(), +}); + +export async function action({ request, params }: ActionArgs) { + logger.info("Registering job variant", { url: request.url }); + + // Ensure this is a POST request + if (request.method.toUpperCase() !== "POST") { + return { status: 405, body: "Method Not Allowed" }; + } + + const parsedParams = ParamsSchema.safeParse(params); + + if (!parsedParams.success) { + logger.info("Invalid params", { params }); + + return json({ error: "Invalid params" }, { status: 400 }); + } + + // Next authenticate the request + const authenticatedEnv = await authenticateApiRequest(request); + + if (!authenticatedEnv) { + logger.info("Invalid or missing api key", { url: request.url }); + + return json({ error: "Invalid or Missing API key" }, { status: 401 }); + } + + // Now parse the request body + const anyBody = await request.json(); + + const body = TriggerVariantConfigSchema.safeParse(anyBody); + + if (!body.success) { + return json({ error: "Invalid request body" }, { status: 400 }); + } + + const service = new RegisterJobVariantService(); + + try { + const variant = await service.call({ + environment: authenticatedEnv, + endpointSlug: parsedParams.data.endpointSlug, + jobId: parsedParams.data.jobId, + jobVersion: parsedParams.data.jobVersion, + config: body.data, + }); + + return json(variant); + } catch (error) { + if (error instanceof Error) { + logger.error("Error registering job trigger variant", { + url: request.url, + error: error.message, + }); + + return json({ error: error.message }, { status: 400 }); + } + + return json({ error: "Something went wrong" }, { status: 500 }); + } +} diff --git a/apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts b/apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts index 0621e9127..56c620aa5 100644 --- a/apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts +++ b/apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts @@ -221,6 +221,7 @@ export class RunTaskService { }, }); + // todo: do this client side instead of adding an option to taskBody if (taskBody.trigger) { // Create an eventrule for the task await prisma.jobEventRule.upsert({ diff --git a/apps/webapp/app/services/apiAuth.server.ts b/apps/webapp/app/services/apiAuth.server.ts index b9957e003..f98d8731f 100644 --- a/apps/webapp/app/services/apiAuth.server.ts +++ b/apps/webapp/app/services/apiAuth.server.ts @@ -12,8 +12,6 @@ export async function authenticateApiRequest( ): Promise { const rawAuthorization = request.headers.get("Authorization"); - console.log(rawAuthorization); - const authorization = AuthorizationHeaderSchema.safeParse(rawAuthorization); if (!authorization.success) { diff --git a/apps/webapp/app/services/clientApi.server.ts b/apps/webapp/app/services/clientApi.server.ts index 3413ecf1e..7111fd594 100644 --- a/apps/webapp/app/services/clientApi.server.ts +++ b/apps/webapp/app/services/clientApi.server.ts @@ -3,7 +3,7 @@ import type { ConnectionAuth, ExecuteJobBody, HttpSourceRequest, - PrepareForJobExecutionBody, + PrepareJobTriggerBody, } from "@trigger.dev/internal"; import { DeliverEventResponseSchema, @@ -167,13 +167,13 @@ export class ClientApi { return ExecuteJobResponseSchema.parse(anyBody); } - async prepareForJobExecution(payload: PrepareForJobExecutionBody) { + async prepareJobTrigger(payload: PrepareJobTriggerBody) { const response = await safeFetch(this.#url, { method: "POST", headers: { "Content-Type": "application/json", "x-trigger-api-key": this.#apiKey, - "x-trigger-action": "PREPARE_FOR_JOB_EXECUTION", + "x-trigger-action": "PREPARE_JOB_TRIGGER", }, body: JSON.stringify(payload), }); diff --git a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts b/apps/webapp/app/services/endpoints/endpointRegistered.server.ts index 47313d88d..4e41c316d 100644 --- a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts +++ b/apps/webapp/app/services/endpoints/endpointRegistered.server.ts @@ -1,18 +1,6 @@ -import type { - Endpoint, - Job, - JobConnection, - JobInstance, - ApiConnection, -} from ".prisma/client"; -import type { ApiJob, ConnectionMetadata } from "@trigger.dev/internal"; -import semver from "semver"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import type { AuthenticatedEnvironment } from "../apiAuth.server"; import { ClientApi } from "../clientApi.server"; -import { allConnectionsReady } from "../jobs/utils.server"; -import { logger } from "../logger"; import { workerQueue } from "../worker.server"; export class EndpointRegisteredService { @@ -28,12 +16,7 @@ export class EndpointRegisteredService { id, }, include: { - environment: { - include: { - project: true, - organization: true, - }, - }, + environment: true, }, }); @@ -42,465 +25,11 @@ export class EndpointRegisteredService { const { jobs } = await client.getJobs(); - // Upsert the jobs into the database - await Promise.all( - jobs.map((job) => this.#upsertJob(endpoint, endpoint.environment, job)) - ); - - await workerQueue.enqueue("prepareForJobExecution", { - id: endpoint.id, - }); - } - - async #upsertJob( - endpoint: Endpoint, - environment: AuthenticatedEnvironment, - apiJob: ApiJob - ): Promise { - logger.debug("Upserting job", { - endpoint, - organizationId: environment.organizationId, - apiJob, - }); - - // Upsert the Job - const job = await this.#prismaClient.job.upsert({ - where: { - projectId_slug: { - projectId: environment.projectId, - slug: apiJob.id, - }, - }, - create: { - organization: { - connect: { - id: environment.organizationId, - }, - }, - project: { - connect: { - id: environment.projectId, - }, - }, - slug: apiJob.id, - title: apiJob.name, - }, - update: { - title: apiJob.name, - }, - include: { - connections: { - include: { - apiConnection: true, - }, - }, - instances: { - where: { - endpointId: endpoint.id, - }, - orderBy: { version: "desc" }, - take: 1, - }, - }, - }); - - const latestInstance = job.instances[0]; - - let ready = false; - - if (typeof latestInstance === "undefined") { - ready = !apiJob.supportsPreparation; - } else { - if (latestInstance.ready) { - // Only carry over the ready state if the it's a PATCH or EQUAL update - ready = ["PATCH", "EQUAL"].includes( - getSemverUpdate(latestInstance.version, apiJob.version) - ); - } - } - - // Upsert the JobInstance - const jobInstance = await this.#prismaClient.jobInstance.upsert({ - where: { - jobId_version_endpointId: { - jobId: job.id, - version: apiJob.version, - endpointId: endpoint.id, - }, - }, - create: { - job: { - connect: { - id: job.id, - }, - }, - endpoint: { - connect: { - id: endpoint.id, - }, - }, - environment: { - connect: { - id: environment.id, - }, - }, - organization: { - connect: { - id: environment.organizationId, - }, - }, - project: { - connect: { - id: environment.projectId, - }, - }, - version: apiJob.version, - trigger: apiJob.trigger, - ready, - }, - update: { - trigger: apiJob.trigger, - }, - include: { - connections: { - include: { - apiConnection: true, - }, - }, - }, - }); - - const upsertedConnections: Array = []; - - if (apiJob.trigger.connection) { - upsertedConnections.push( - await this.#upsertJobConnection( - job, - jobInstance, - "__trigger", - apiJob.trigger.connection.metadata, - apiJob.trigger.connection.usesLocalAuth, - apiJob.trigger.connection.id - ) - ); - } - - // Upsert the connections - for (const connection of apiJob.connections) { - upsertedConnections.push( - await this.#upsertJobConnection( - job, - jobInstance, - connection.key, - connection.metadata, - connection.usesLocalAuth, - connection.id - ) - ); - } - - // Delete any connections that are no longer in the job - await this.#prismaClient.jobConnection.deleteMany({ - where: { - jobInstanceId: jobInstance.id, - id: { - notIn: upsertedConnections.map((c) => c.id), - }, - }, - }); - - // Count the number of job instances that have higher version numbers - const laterJobInstanceCount = await this.#prismaClient.jobInstance.count({ - where: { - jobId: job.id, - version: { - gt: apiJob.version, - }, - environmentId: environment.id, - }, - }); - - // If there are no later job instances, then we can upsert the latest jobalias - if (laterJobInstanceCount === 0) { - // upsert the latest jobalias - await this.#prismaClient.jobAlias.upsert({ - where: { - jobId_environmentId_name: { - jobId: job.id, - environmentId: environment.id, - name: "latest", - }, - }, - create: { - jobId: job.id, - jobInstanceId: jobInstance.id, - environmentId: environment.id, - name: "latest", - version: jobInstance.version, - }, - update: { - jobInstanceId: jobInstance.id, - version: jobInstance.version, - }, - }); - } - - const connectionsReady = await allConnectionsReady(upsertedConnections); - - // upsert the eventrule - // The event rule should only be enabled if all the external connections are ready - await this.#prismaClient.jobEventRule.upsert({ - where: { - jobInstanceId_actionIdentifier: { - jobInstanceId: jobInstance.id, - actionIdentifier: "__trigger", - }, - }, - create: { - event: apiJob.trigger.eventRule.event, - source: apiJob.trigger.eventRule.source, - payloadFilter: apiJob.trigger.eventRule.payload, - contextFilter: apiJob.trigger.eventRule.context, - jobId: job.id, - jobInstanceId: jobInstance.id, - environmentId: environment.id, - organizationId: environment.organizationId, - projectId: environment.projectId, - enabled: connectionsReady, - actionIdentifier: "__trigger", - }, - update: { - event: apiJob.trigger.eventRule.event, - source: apiJob.trigger.eventRule.source, - payloadFilter: apiJob.trigger.eventRule.payload, - contextFilter: apiJob.trigger.eventRule.context, - enabled: connectionsReady, - }, - }); - } - - async #upsertJobConnection( - job: Job & { - connections: Array< - JobConnection & { apiConnection: ApiConnection | null } - >; - }, - jobInstance: JobInstance & { - connections: Array< - JobConnection & { apiConnection: ApiConnection | null } - >; - }, - key: string, - metadata: ConnectionMetadata, - usesLocalAuth: boolean, - id?: string - ): Promise { - if (usesLocalAuth) { - return this.#upsertLocalAuthConnection(job, jobInstance, key, metadata); - } - - if (!id) { - logger.debug("Missing connection id", { - key, - metadata, - usesLocalAuth, + for (const job of jobs) { + await workerQueue.enqueue("registerJob", { job, - }); - - throw new Error("Missing connection id"); - } - - const apiConnection = - await this.#prismaClient.apiConnection.findUniqueOrThrow({ - where: { - organizationId_slug: { - organizationId: job.organizationId, - slug: id, - }, - }, - }); - - // Find existing connection in the job instance - const existingInstanceConnection = jobInstance.connections.find( - (connection) => connection.key === key - ); - - if (existingInstanceConnection) { - return await this.#prismaClient.jobConnection.update({ - where: { - id: existingInstanceConnection.id, - }, - data: { - apiConnectionId: apiConnection.id, - usesLocalAuth: false, - }, + endpointId: endpoint.id, }); } - - // Find existing connection in the job - const existingJobConnection = job.connections.find( - (connection) => connection.key === key - ); - - if (existingJobConnection) { - return this.#prismaClient.jobConnection.create({ - data: { - jobInstance: { - connect: { - id: jobInstance.id, - }, - }, - job: { - connect: { - id: job.id, - }, - }, - key, - connectionMetadata: existingJobConnection.connectionMetadata ?? {}, - apiConnection: { - connect: { - id: apiConnection.id, - }, - }, - usesLocalAuth: false, - }, - }); - } - - return this.#prismaClient.jobConnection.create({ - data: { - jobInstance: { - connect: { - id: jobInstance.id, - }, - }, - job: { - connect: { - id: job.id, - }, - }, - key, - connectionMetadata: metadata, - apiConnection: { - connect: { - id: apiConnection.id, - }, - }, - usesLocalAuth, - }, - }); - } - - async #upsertLocalAuthConnection( - job: Job & { - connections: Array< - JobConnection & { apiConnection: ApiConnection | null } - >; - }, - jobInstance: JobInstance & { - connections: Array< - JobConnection & { apiConnection: ApiConnection | null } - >; - }, - key: string, - metadata: ConnectionMetadata - ): Promise { - // Find existing connection in the job instance - const existingInstanceConnection = jobInstance.connections.find( - (connection) => connection.key === key - ); - - if ( - existingInstanceConnection && - existingInstanceConnection.apiConnectionId - ) { - return await this.#prismaClient.jobConnection.update({ - where: { - id: existingInstanceConnection.id, - }, - data: { - apiConnectionId: null, - usesLocalAuth: true, - }, - }); - } - - if (existingInstanceConnection) { - return existingInstanceConnection; - } - - // Find existing connection in the job - const existingJobConnection = job.connections.find( - (connection) => connection.key === key - ); - - if (existingJobConnection) { - return this.#prismaClient.jobConnection.create({ - data: { - jobInstance: { - connect: { - id: jobInstance.id, - }, - }, - job: { - connect: { - id: job.id, - }, - }, - key, - connectionMetadata: existingJobConnection.connectionMetadata ?? {}, - usesLocalAuth: true, - }, - }); - } - - return this.#prismaClient.jobConnection.create({ - data: { - jobInstance: { - connect: { - id: jobInstance.id, - }, - }, - job: { - connect: { - id: job.id, - }, - }, - key, - connectionMetadata: metadata, - usesLocalAuth: true, - }, - }); } } - -// Compares two semver strings and returns the type of update, either EQUAL, PATCH, MINOR, or MAJOR -function getSemverUpdate( - latestVersion: string | undefined, - newVersion: string | undefined -) { - const latest = semver.coerce(latestVersion); - const newV = semver.coerce(newVersion); - - if (!latest || !newV) { - return "EQUAL"; - } - - if (semver.eq(latest, newV)) { - return "EQUAL"; - } - - if (semver.lt(latest, newV)) { - if (semver.major(latest) === semver.major(newV)) { - if (semver.minor(latest) === semver.minor(newV)) { - return "PATCH"; - } - - return "MINOR"; - } - - return "MAJOR"; - } - - return "EQUAL"; -} diff --git a/apps/webapp/app/services/endpoints/prepareForJobExecution.server.ts b/apps/webapp/app/services/endpoints/prepareForJobExecution.server.ts deleted file mode 100644 index 4b45c0185..000000000 --- a/apps/webapp/app/services/endpoints/prepareForJobExecution.server.ts +++ /dev/null @@ -1,34 +0,0 @@ -import type { PrismaClient } from "~/db.server"; -import { prisma } from "~/db.server"; -import { workerQueue } from "../worker.server"; - -export class PrepareForJobExecutionService { - #prismaClient: PrismaClient; - - constructor(prismaClient: PrismaClient = prisma) { - this.#prismaClient = prismaClient; - } - - public async call(id: string) { - const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ - where: { - id, - }, - include: { - jobInstances: true, - }, - }); - - for (const jobInstance of endpoint.jobInstances) { - if (jobInstance.ready) { - continue; - } - - await workerQueue.enqueue( - "prepareJobInstance", - { id: jobInstance.id }, - { queueName: `endpoint-${endpoint.id}` } - ); - } - } -} diff --git a/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts b/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts index 3ea724e06..aa54376b7 100644 --- a/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts +++ b/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts @@ -1,7 +1,8 @@ import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; +import { resolveJobConnection } from "~/models/jobConnection.server"; import { ClientApi } from "../clientApi.server"; -import { resolveJobConnections } from "~/models/jobConnection.server"; +import { workerQueue } from "../worker.server"; export class PrepareJobInstanceService { #prismaClient: PrismaClient; @@ -24,6 +25,9 @@ export class PrepareJobInstanceService { }, }, }, + where: { + key: "__trigger", + }, }, job: true, endpoint: { @@ -31,6 +35,7 @@ export class PrepareJobInstanceService { environment: true, }, }, + triggerVariants: true, }, }); @@ -39,10 +44,14 @@ export class PrepareJobInstanceService { jobInstance.endpoint.url ); - const response = await client.prepareForJobExecution({ + const connection = jobInstance.connections[0]; + + const response = await client.prepareJobTrigger({ id: jobInstance.job.slug, version: jobInstance.version, - connections: await resolveJobConnections(jobInstance.connections), + connection: connection + ? await resolveJobConnection(connection) + : undefined, }); if (!response.ok) { @@ -57,5 +66,21 @@ export class PrepareJobInstanceService { ready: true, }, }); + + for (const variant of jobInstance.triggerVariants) { + if (variant.ready) { + continue; + } + + await workerQueue.enqueue( + "prepareTriggerVariant", + { + id: variant.id, + }, + { + queueName: `endpoint-${jobInstance.endpoint.id}`, + } + ); + } } } diff --git a/apps/webapp/app/services/endpoints/prepareTriggerVariant.server.ts b/apps/webapp/app/services/endpoints/prepareTriggerVariant.server.ts new file mode 100644 index 000000000..ae698fa05 --- /dev/null +++ b/apps/webapp/app/services/endpoints/prepareTriggerVariant.server.ts @@ -0,0 +1,79 @@ +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import { resolveJobConnection } from "~/models/jobConnection.server"; +import { ClientApi } from "../clientApi.server"; + +export class PrepareTriggerVariantService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call(id: string) { + const triggerVariant = + await this.#prismaClient.jobTriggerVariant.findUniqueOrThrow({ + where: { + id, + }, + include: { + jobInstance: { + include: { + job: true, + endpoint: { + include: { + environment: true, + }, + }, + triggerVariants: true, + }, + }, + }, + }); + + const jobInstance = triggerVariant.jobInstance; + + const client = new ClientApi( + jobInstance.endpoint.environment.apiKey, + jobInstance.endpoint.url + ); + + const connection = await this.#prismaClient.jobConnection.findUnique({ + where: { + jobInstanceId_key: { + jobInstanceId: jobInstance.id, + key: `__trigger_${triggerVariant.slug}`, + }, + }, + include: { + apiConnection: { + include: { + dataReference: true, + }, + }, + }, + }); + + const response = await client.prepareJobTrigger({ + id: jobInstance.job.slug, + version: jobInstance.version, + connection: connection + ? await resolveJobConnection(connection) + : undefined, + variantId: triggerVariant.slug, + }); + + if (!response.ok) { + throw new Error("Something went wrong when preparing a trigger variant"); + } + + await this.#prismaClient.jobTriggerVariant.update({ + where: { + id, + }, + data: { + ready: true, + }, + }); + } +} diff --git a/apps/webapp/app/services/events/deliverEvent.server.ts b/apps/webapp/app/services/events/deliverEvent.server.ts index 50fb716ae..4190b33ff 100644 --- a/apps/webapp/app/services/events/deliverEvent.server.ts +++ b/apps/webapp/app/services/events/deliverEvent.server.ts @@ -9,7 +9,7 @@ import { logger } from "../logger"; export class DeliverEventService { #prismaClient: PrismaClient; - #createExecutionService = new CreateRunService(); + #createRunService = new CreateRunService(); #resumeTaskService = new ResumeTaskService(); constructor(prismaClient: PrismaClient = prisma) { @@ -75,8 +75,8 @@ export class DeliverEventService { for (const eventRule of matchingEventRules) { switch (eventRule.action) { - case "CREATE_EXECUTION": { - await this.#createExecutionService.call({ + case "CREATE_RUN": { + await this.#createRunService.call({ eventId: eventLog.id, job: eventRule.job, jobInstance: eventRule.jobInstance, diff --git a/apps/webapp/app/services/jobs/registerJob.server.ts b/apps/webapp/app/services/jobs/registerJob.server.ts new file mode 100644 index 000000000..0d1da9c9a --- /dev/null +++ b/apps/webapp/app/services/jobs/registerJob.server.ts @@ -0,0 +1,635 @@ +import type { + ApiConnection, + Endpoint, + Job, + JobConnection, + JobInstance, + JobTriggerVariant, +} from ".prisma/client"; +import type { + ConnectionConfig, + GetJobResponse, + LocalAuthConnectionConfig, + TriggerMetadata, +} from "@trigger.dev/internal"; +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import type { AuthenticatedEnvironment } from "../apiAuth.server"; +import { logger } from "../logger"; +import { workerQueue } from "../worker.server"; + +export class RegisterJobService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call(endpointId: string, jobResponse: GetJobResponse) { + const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ + where: { + id: endpointId, + }, + include: { + environment: { + include: { + project: true, + organization: true, + }, + }, + }, + }); + + const jobInstance = await this.#upsertJob( + endpoint, + endpoint.environment, + jobResponse + ); + + await workerQueue.enqueue( + "prepareJobInstance", + { id: jobInstance.id }, + { queueName: `endpoint-${endpoint.id}` } + ); + } + + async #upsertJob( + endpoint: Endpoint, + environment: AuthenticatedEnvironment, + jobResponse: GetJobResponse + ): Promise { + const { metadata, triggerVariants } = jobResponse; + + logger.debug("Upserting job", { + endpoint, + organizationId: environment.organizationId, + metadata, + triggerVariants, + }); + + // Make sure all the hosted connections exist before we upsert the job + // Need to check for three places where a connection could be: + // 1. The job.connections + // 2. The job.trigger possible connection + // 3. The job.triggerVariants possible connection + const connectionSlugs = new Set(); + + if (metadata.connections) { + for (const connection of metadata.connections) { + if (connection.auth === "hosted") { + connectionSlugs.add(connection.id); + } + } + } + + if ( + metadata.trigger.connection && + metadata.trigger.connection.auth === "hosted" + ) { + connectionSlugs.add(metadata.trigger.connection.id); + } + + if (triggerVariants) { + for (const triggerVariant of triggerVariants) { + if ( + triggerVariant.trigger.connection && + triggerVariant.trigger.connection.auth === "hosted" + ) { + connectionSlugs.add(triggerVariant.trigger.connection.id); + } + } + } + + const apiConnections = new Map(); + + for (const connectionSlug of connectionSlugs) { + const apiConnection = await this.#prismaClient.apiConnection.findUnique({ + where: { + organizationId_slug: { + organizationId: environment.organizationId, + slug: connectionSlug, + }, + }, + }); + + if (!apiConnection) { + // todo: find a better way to handle and message the user about this issue + throw new Error( + `Could not find ApiConnection with slug ${connectionSlug}` + ); + } + + apiConnections.set(connectionSlug, apiConnection); + } + + // Upsert the Job + const job = await this.#prismaClient.job.upsert({ + where: { + projectId_slug: { + projectId: environment.projectId, + slug: metadata.id, + }, + }, + create: { + organization: { + connect: { + id: environment.organizationId, + }, + }, + project: { + connect: { + id: environment.projectId, + }, + }, + slug: metadata.id, + title: metadata.name, + }, + update: { + title: metadata.name, + }, + include: { + connections: { + include: { + apiConnection: true, + }, + }, + instances: { + where: { + endpointId: endpoint.id, + }, + orderBy: { version: "desc" }, + take: 1, + include: { + triggerVariants: true, + }, + }, + }, + }); + + const latestInstance = job.instances[0]; + + let ready = false; + + if (typeof latestInstance !== "undefined") { + ready = latestInstance.ready; + } else { + ready = !metadata.trigger.supportsPreparation; + } + + // Upsert the JobInstance + const jobInstance = await this.#prismaClient.jobInstance.upsert({ + where: { + jobId_version_endpointId: { + jobId: job.id, + version: metadata.version, + endpointId: endpoint.id, + }, + }, + create: { + job: { + connect: { + id: job.id, + }, + }, + endpoint: { + connect: { + id: endpoint.id, + }, + }, + environment: { + connect: { + id: environment.id, + }, + }, + organization: { + connect: { + id: environment.organizationId, + }, + }, + project: { + connect: { + id: environment.projectId, + }, + }, + version: metadata.version, + trigger: metadata.trigger, + ready, + }, + update: { + trigger: metadata.trigger, + }, + include: { + connections: { + include: { + apiConnection: true, + }, + }, + }, + }); + + const jobConnections = new Set(); + + if (metadata.trigger.connection) { + const triggerConnection = await this.#upsertJobConnection( + job, + jobInstance, + metadata.trigger.connection, + apiConnections, + "__trigger" + ); + + jobConnections.add(triggerConnection.id); + } + + // Upsert the job connections + for (const connection of metadata.connections) { + const jobConnection = await this.#upsertJobConnection( + job, + jobInstance, + connection, + apiConnections + ); + + jobConnections.add(jobConnection.id); + } + + // Count the number of job instances that have higher version numbers + const laterJobInstanceCount = await this.#prismaClient.jobInstance.count({ + where: { + jobId: job.id, + version: { + gt: metadata.version, + }, + environmentId: environment.id, + }, + }); + + // If there are no later job instances, then we can upsert the latest jobalias + if (laterJobInstanceCount === 0) { + // upsert the latest jobalias + await this.#prismaClient.jobAlias.upsert({ + where: { + jobId_environmentId_name: { + jobId: job.id, + environmentId: environment.id, + name: "latest", + }, + }, + create: { + jobId: job.id, + jobInstanceId: jobInstance.id, + environmentId: environment.id, + name: "latest", + version: jobInstance.version, + }, + update: { + jobInstanceId: jobInstance.id, + version: jobInstance.version, + }, + }); + } + + if (triggerVariants) { + for (const triggerVariant of triggerVariants) { + const jobConnection = await this.#upsertTriggerVariant( + job, + jobInstance, + environment, + triggerVariant.id, + triggerVariant.trigger, + apiConnections, + latestInstance?.triggerVariants + ); + + if (jobConnection) { + jobConnections.add(jobConnection.id); + } + } + } + + // Delete any connections that are no longer in the job + // It's import this runs after the trigger variant upserts + await this.#prismaClient.jobConnection.deleteMany({ + where: { + jobInstanceId: jobInstance.id, + id: { + notIn: Array.from(jobConnections), + }, + }, + }); + + // upsert the eventrule + // The event rule should only be enabled if all the external connections are ready + await this.#prismaClient.jobEventRule.upsert({ + where: { + jobInstanceId_actionIdentifier: { + jobInstanceId: jobInstance.id, + actionIdentifier: "__trigger", + }, + }, + create: { + event: metadata.trigger.eventRule.event, + source: metadata.trigger.eventRule.source, + payloadFilter: metadata.trigger.eventRule.payload, + contextFilter: metadata.trigger.eventRule.context, + jobId: job.id, + jobInstanceId: jobInstance.id, + environmentId: environment.id, + organizationId: environment.organizationId, + projectId: environment.projectId, + enabled: true, + actionIdentifier: "__trigger", + }, + update: { + event: metadata.trigger.eventRule.event, + source: metadata.trigger.eventRule.source, + payloadFilter: metadata.trigger.eventRule.payload, + contextFilter: metadata.trigger.eventRule.context, + }, + }); + + return jobInstance; + } + + async #upsertTriggerVariant( + job: Job & { + connections: Array< + JobConnection & { apiConnection: ApiConnection | null } + >; + }, + jobInstance: JobInstance & { + connections: Array< + JobConnection & { apiConnection: ApiConnection | null } + >; + }, + environment: AuthenticatedEnvironment, + id: string, + trigger: TriggerMetadata, + apiConnections: Map, + previousVariants?: Array + ): Promise { + const previousVariant = previousVariants?.find((v) => v.id === id); + + await this.#prismaClient.jobTriggerVariant.upsert({ + where: { + jobInstanceId_slug: { + jobInstanceId: jobInstance.id, + slug: id, + }, + }, + create: { + jobInstance: { + connect: { + id: jobInstance.id, + }, + }, + slug: id, + data: trigger, + ready: trigger.supportsPreparation + ? previousVariant + ? previousVariant.ready + : false + : true, + eventRule: { + create: { + event: trigger.eventRule.event, + source: trigger.eventRule.source, + payloadFilter: trigger.eventRule.payload, + contextFilter: trigger.eventRule.context, + jobId: job.id, + jobInstanceId: jobInstance.id, + environmentId: environment.id, + organizationId: environment.organizationId, + projectId: environment.projectId, + enabled: true, + actionIdentifier: `__trigger_${id}`, + }, + }, + }, + update: { + data: trigger, + }, + }); + + if (trigger.connection) { + return await this.#upsertJobConnection( + job, + jobInstance, + trigger.connection, + apiConnections, + `__trigger_${id}` + ); + } + } + + async #upsertJobConnection( + job: Job & { + connections: Array< + JobConnection & { apiConnection: ApiConnection | null } + >; + }, + jobInstance: JobInstance & { + connections: Array< + JobConnection & { apiConnection: ApiConnection | null } + >; + }, + config: ConnectionConfig, + apiConnections: Map, + overrideKey?: string + ): Promise { + if (config.auth === "local") { + return this.#upsertLocalAuthConnection( + job, + jobInstance, + config, + overrideKey + ); + } + + const apiConnection = apiConnections.get(config.id); + + if (!apiConnection) { + throw new Error( + `Could not find api connection with id ${config.id} for job ${job.id}` + ); + } + + const key = overrideKey ?? config.key; + + if (!key) { + throw new Error( + `Could not find key for connection ${config.id} for job ${job.id}` + ); + } + + // Find existing connection in the job instance + const existingInstanceConnection = jobInstance.connections.find( + (connection) => connection.key === key + ); + + if (existingInstanceConnection) { + return await this.#prismaClient.jobConnection.update({ + where: { + id: existingInstanceConnection.id, + }, + data: { + apiConnectionId: apiConnection.id, + usesLocalAuth: false, + }, + }); + } + + // Find existing connection in the job + const existingJobConnection = job.connections.find( + (connection) => connection.key === key + ); + + if (existingJobConnection) { + logger.debug("Creating new job connection from existing", { + existingJobConnection, + key, + jobInstanceId: jobInstance.id, + }); + + return this.#prismaClient.jobConnection.create({ + data: { + jobInstance: { + connect: { + id: jobInstance.id, + }, + }, + job: { + connect: { + id: job.id, + }, + }, + key, + connectionMetadata: existingJobConnection.connectionMetadata ?? {}, + apiConnection: { + connect: { + id: apiConnection.id, + }, + }, + usesLocalAuth: false, + }, + }); + } + + logger.debug("Creating new job connection", { + key, + jobInstanceId: jobInstance.id, + config, + }); + + return this.#prismaClient.jobConnection.create({ + data: { + jobInstance: { + connect: { + id: jobInstance.id, + }, + }, + job: { + connect: { + id: job.id, + }, + }, + key, + connectionMetadata: config.metadata, + apiConnection: { + connect: { + id: apiConnection.id, + }, + }, + usesLocalAuth: false, + }, + }); + } + + async #upsertLocalAuthConnection( + job: Job & { + connections: Array< + JobConnection & { apiConnection: ApiConnection | null } + >; + }, + jobInstance: JobInstance & { + connections: Array< + JobConnection & { apiConnection: ApiConnection | null } + >; + }, + config: LocalAuthConnectionConfig, + overrideKey?: string + ): Promise { + const key = overrideKey ?? config.key; + + if (!key) { + throw new Error("Missing connection key"); + } + + // Find existing connection in the job instance + const existingInstanceConnection = jobInstance.connections.find( + (connection) => connection.key === key + ); + + if ( + existingInstanceConnection && + existingInstanceConnection.apiConnectionId + ) { + return await this.#prismaClient.jobConnection.update({ + where: { + id: existingInstanceConnection.id, + }, + data: { + apiConnectionId: null, + usesLocalAuth: true, + }, + }); + } + + if (existingInstanceConnection) { + return existingInstanceConnection; + } + + // Find existing connection in the job + const existingJobConnection = job.connections.find( + (connection) => connection.key === key + ); + + if (existingJobConnection) { + return this.#prismaClient.jobConnection.create({ + data: { + jobInstance: { + connect: { + id: jobInstance.id, + }, + }, + job: { + connect: { + id: job.id, + }, + }, + key, + connectionMetadata: existingJobConnection.connectionMetadata ?? {}, + usesLocalAuth: true, + }, + }); + } + + return this.#prismaClient.jobConnection.create({ + data: { + jobInstance: { + connect: { + id: jobInstance.id, + }, + }, + job: { + connect: { + id: job.id, + }, + }, + key, + connectionMetadata: config.metadata, + usesLocalAuth: true, + }, + }); + } +} diff --git a/apps/webapp/app/services/jobs/registerJobVariant.server.ts b/apps/webapp/app/services/jobs/registerJobVariant.server.ts new file mode 100644 index 000000000..fdacf9bd7 --- /dev/null +++ b/apps/webapp/app/services/jobs/registerJobVariant.server.ts @@ -0,0 +1,111 @@ +import type { + TriggerVariantConfig, + TriggerVariantResponseBody, +} from "@trigger.dev/internal"; +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import type { AuthenticatedEnvironment } from "../apiAuth.server"; +import type { ApiConnectionWithSecretReference } from "../externalApis/apiAuthenticationRepository.server"; +import { resolveApiConnection } from "~/models/jobConnection.server"; + +export class RegisterJobVariantService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call({ + endpointSlug, + jobId, + jobVersion, + config: { trigger, id }, + environment, + }: { + environment: AuthenticatedEnvironment; + endpointSlug: string; + jobId: string; + jobVersion: string; + config: TriggerVariantConfig; + }): Promise { + const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ + where: { + environmentId_slug: { + environmentId: environment.id, + slug: endpointSlug, + }, + }, + }); + + const jobInstance = await this.#prismaClient.jobInstance.findUniqueOrThrow({ + where: { + jobId_version_endpointId: { + jobId, + version: jobVersion, + endpointId: endpoint.id, + }, + }, + }); + + let apiConnection: ApiConnectionWithSecretReference | undefined; + + if (trigger.connection && trigger.connection.auth === "hosted") { + apiConnection = await this.#prismaClient.apiConnection.findUniqueOrThrow({ + where: { + organizationId_slug: { + organizationId: endpoint.organizationId, + slug: trigger.connection.id, + }, + }, + include: { + dataReference: true, + }, + }); + } + + const triggerVariant = await this.#prismaClient.jobTriggerVariant.upsert({ + where: { + jobInstanceId_slug: { + jobInstanceId: jobInstance.id, + slug: id, + }, + }, + create: { + jobInstance: { + connect: { + id: jobInstance.id, + }, + }, + slug: id, + data: trigger, + ready: !trigger.supportsPreparation, + eventRule: { + create: { + event: trigger.eventRule.event, + source: trigger.eventRule.source, + payloadFilter: trigger.eventRule.payload, + contextFilter: trigger.eventRule.context, + jobId: jobInstance.jobId, + jobInstanceId: jobInstance.id, + environmentId: environment.id, + organizationId: environment.organizationId, + projectId: environment.projectId, + enabled: true, + actionIdentifier: `__trigger_${id}`, + }, + }, + }, + update: { + data: trigger, + }, + }); + + return { + id: triggerVariant.id, + slug: triggerVariant.slug, + data: trigger, + ready: triggerVariant.ready, + auth: await resolveApiConnection(apiConnection), + }; + } +} diff --git a/apps/webapp/app/services/jobs/utils.server.ts b/apps/webapp/app/services/jobs/utils.server.ts deleted file mode 100644 index 5e998973c..000000000 --- a/apps/webapp/app/services/jobs/utils.server.ts +++ /dev/null @@ -1,21 +0,0 @@ -import { JobConnection } from ".prisma/client"; - -export async function allConnectionsReady( - connections: Array -): Promise { - if (connections.length === 0) { - return true; - } - - const connectionsUsingExternalAuth = connections.filter( - (connection) => !connection.usesLocalAuth - ); - - if (connectionsUsingExternalAuth.length === 0) { - return true; - } - - return connectionsUsingExternalAuth.every((connection) => { - return connection.apiConnectionId; - }); -} diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index 49fc3b9c8..466563e74 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/apps/webapp/app/services/runs/createRun.server.ts @@ -2,7 +2,7 @@ import type { Job, JobInstance } from ".prisma/client"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; import { workerQueue } from "~/services/worker.server"; -import { AuthenticatedEnvironment } from "../apiAuth.server"; +import type { AuthenticatedEnvironment } from "../apiAuth.server"; export class CreateRunService { #prismaClient: PrismaClient; diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index a22b804bf..22f0e8d68 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -1,14 +1,16 @@ +import { GetJobResponseSchema } from "@/../../packages/internal/src"; import { z } from "zod"; import { env } from "~/env.server"; import { ZodWorker } from "~/platform/zodWorker.server"; import { EndpointRegisteredService } from "./endpoints/endpointRegistered.server"; -import { PrepareForJobExecutionService } from "./endpoints/prepareForJobExecution.server"; import { PrepareJobInstanceService } from "./endpoints/prepareJobInstance.server"; import { DeliverEventService } from "./events/deliverEvent.server"; import { apiConnectionRepository } 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 { PrepareTriggerVariantService } from "./endpoints/prepareTriggerVariant.server"; const workerCatalog = { organizationCreated: z.object({ id: z.string() }), @@ -29,13 +31,17 @@ const workerCatalog = { startInitialProjectDeployment: z.object({ id: z.string() }), startRun: z.object({ id: z.string() }), resumeTask: z.object({ id: z.string() }), - prepareForJobExecution: z.object({ id: z.string() }), prepareJobInstance: z.object({ id: z.string() }), + prepareTriggerVariant: z.object({ id: z.string() }), deliverHttpSourceRequest: z.object({ id: z.string() }), refreshOAuthToken: z.object({ organizationId: z.string(), connectionId: z.string(), }), + registerJob: z.object({ + endpointId: z.string(), + job: GetJobResponseSchema, + }), }; let workerQueue: ZodWorker; @@ -71,6 +77,14 @@ function getWorkerQueue() { }, schema: workerCatalog, tasks: { + registerJob: { + maxAttempts: 3, + handler: async (payload, job) => { + const service = new RegisterJobService(); + + await service.call(payload.endpointId, payload.job); + }, + }, deliverHttpSourceRequest: { maxAttempts: 5, handler: async (payload, job) => { @@ -87,11 +101,10 @@ function getWorkerQueue() { await service.call(payload.id); }, }, - prepareForJobExecution: { - queueName: "internal-queue", - maxAttempts: 8, + prepareTriggerVariant: { + maxAttempts: 3, handler: async (payload, job) => { - const service = new PrepareForJobExecutionService(); + const service = new PrepareTriggerVariantService(); await service.call(payload.id); }, diff --git a/apps/webapp/prisma/migrations/20230503094127_add_trigger_variants/migration.sql b/apps/webapp/prisma/migrations/20230503094127_add_trigger_variants/migration.sql new file mode 100644 index 000000000..8fa51b4d1 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230503094127_add_trigger_variants/migration.sql @@ -0,0 +1,21 @@ +-- CreateTable +CREATE TABLE "JobTriggerVariant" ( + "id" TEXT NOT NULL, + "data" JSONB NOT NULL, + "ready" BOOLEAN NOT NULL DEFAULT false, + "jobInstanceId" TEXT NOT NULL, + "eventRuleId" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "JobTriggerVariant_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "JobTriggerVariant_eventRuleId_key" ON "JobTriggerVariant"("eventRuleId"); + +-- AddForeignKey +ALTER TABLE "JobTriggerVariant" ADD CONSTRAINT "JobTriggerVariant_jobInstanceId_fkey" FOREIGN KEY ("jobInstanceId") REFERENCES "JobInstance"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "JobTriggerVariant" ADD CONSTRAINT "JobTriggerVariant_eventRuleId_fkey" FOREIGN KEY ("eventRuleId") REFERENCES "JobEventRule"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20230503113400_add_slug_to_trigger_variants/migration.sql b/apps/webapp/prisma/migrations/20230503113400_add_slug_to_trigger_variants/migration.sql new file mode 100644 index 000000000..9843ab2bc --- /dev/null +++ b/apps/webapp/prisma/migrations/20230503113400_add_slug_to_trigger_variants/migration.sql @@ -0,0 +1,12 @@ +/* + Warnings: + + - A unique constraint covering the columns `[jobInstanceId,slug]` on the table `JobTriggerVariant` will be added. If there are existing duplicate values, this will fail. + - Added the required column `slug` to the `JobTriggerVariant` table without a default value. This is not possible if the table is not empty. + +*/ +-- AlterTable +ALTER TABLE "JobTriggerVariant" ADD COLUMN "slug" TEXT NOT NULL; + +-- CreateIndex +CREATE UNIQUE INDEX "JobTriggerVariant_jobInstanceId_slug_key" ON "JobTriggerVariant"("jobInstanceId", "slug"); diff --git a/apps/webapp/prisma/migrations/20230503172156_rename_create_execution_to_create_run/migration.sql b/apps/webapp/prisma/migrations/20230503172156_rename_create_execution_to_create_run/migration.sql new file mode 100644 index 000000000..90b9eaf97 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230503172156_rename_create_execution_to_create_run/migration.sql @@ -0,0 +1,19 @@ +/* + Warnings: + + - The values [CREATE_EXECUTION] on the enum `JobEventAction` will be removed. If these variants are still used in the database, this will fail. + +*/ +-- AlterEnum +BEGIN; +CREATE TYPE "JobEventAction_new" AS ENUM ('CREATE_RUN', 'RESUME_TASK'); +ALTER TABLE "JobEventRule" ALTER COLUMN "action" DROP DEFAULT; +ALTER TABLE "JobEventRule" ALTER COLUMN "action" TYPE "JobEventAction_new" USING ("action"::text::"JobEventAction_new"); +ALTER TYPE "JobEventAction" RENAME TO "JobEventAction_old"; +ALTER TYPE "JobEventAction_new" RENAME TO "JobEventAction"; +DROP TYPE "JobEventAction_old"; +ALTER TABLE "JobEventRule" ALTER COLUMN "action" SET DEFAULT 'CREATE_RUN'; +COMMIT; + +-- AlterTable +ALTER TABLE "JobEventRule" ALTER COLUMN "action" SET DEFAULT 'CREATE_RUN'; diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index 4874e30de..c26f065c2 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -85,7 +85,7 @@ model ApiConnection { metadata Json dataReference SecretReference @relation(fields: [dataReferenceId], references: [id], onDelete: Cascade, onUpdate: Cascade) - dataReferenceId String + dataReferenceId String createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@ -1012,14 +1012,33 @@ model JobInstance { createdAt DateTime @default(now()) updatedAt DateTime @updatedAt - runs JobRun[] - connections JobConnection[] - eventRules JobEventRule[] - aliases JobAlias[] + runs JobRun[] + connections JobConnection[] + eventRules JobEventRule[] + aliases JobAlias[] + triggerVariants JobTriggerVariant[] @@unique([jobId, version, endpointId]) } +model JobTriggerVariant { + id String @id @default(cuid()) + slug String + data Json + ready Boolean @default(false) + + jobInstance JobInstance @relation(fields: [jobInstanceId], references: [id], onDelete: Cascade, onUpdate: Cascade) + jobInstanceId String + + eventRule JobEventRule @relation(fields: [eventRuleId], references: [id], onDelete: Cascade, onUpdate: Cascade) + eventRuleId String @unique + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([jobInstanceId, slug]) +} + model JobAlias { id String @id @default(cuid()) name String @default("latest") @@ -1060,6 +1079,46 @@ model JobConnection { @@unique([jobInstanceId, key]) } +model JobEventRule { + id String @id @default(cuid()) + event String + source String + payloadFilter Json? + contextFilter Json? + + action JobEventAction @default(CREATE_RUN) + actionIdentifier String + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + enabled Boolean @default(true) + + job Job @relation(fields: [jobId], references: [id], onDelete: Cascade, onUpdate: Cascade) + jobId String + + jobInstance JobInstance @relation(fields: [jobInstanceId], references: [id], onDelete: Cascade, onUpdate: Cascade) + jobInstanceId String + + organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade) + organizationId String + + environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) + environmentId String + + project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) + projectId String + + triggerVariant JobTriggerVariant? + + @@unique([jobInstanceId, actionIdentifier]) +} + +enum JobEventAction { + CREATE_RUN + RESUME_TASK +} + model EventLog { id String @id @default(cuid()) name String @@ -1253,41 +1312,3 @@ model HttpSourceRequestDelivery { updatedAt DateTime @updatedAt deliveredAt DateTime? } - -model JobEventRule { - id String @id @default(cuid()) - event String - source String - payloadFilter Json? - contextFilter Json? - - action JobEventAction @default(CREATE_EXECUTION) - actionIdentifier String - - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - - enabled Boolean @default(true) - - job Job @relation(fields: [jobId], references: [id], onDelete: Cascade, onUpdate: Cascade) - jobId String - - jobInstance JobInstance @relation(fields: [jobInstanceId], references: [id], onDelete: Cascade, onUpdate: Cascade) - jobInstanceId String - - organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade) - organizationId String - - environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) - environmentId String - - project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) - projectId String - - @@unique([jobInstanceId, actionIdentifier]) -} - -enum JobEventAction { - CREATE_EXECUTION - RESUME_TASK -} diff --git a/examples/nextjs-example/src/pages/api/trigger.ts b/examples/nextjs-example/src/pages/api/trigger.ts index ae24b30f7..bf3c72db6 100644 --- a/examples/nextjs-example/src/pages/api/trigger.ts +++ b/examples/nextjs-example/src/pages/api/trigger.ts @@ -82,7 +82,7 @@ new Job({ }, }).registerWith(client); -new Job({ +const notifySlackONNewCommentsJob = new Job({ id: "notify-slack-on-new-comments", name: "Notify Slack on new GitHub comments", version: "0.1.1", @@ -100,16 +100,42 @@ new Job({ channel: "C04GWUTDC3W", }); }, -}).registerWith(client); - -// TODO: Support parameterized jobs -// Example: -// const job = new Job({}); -// await job.registerWith(client, { params: { foo: "bar" } }); -// And registering as a specific user: -// await job.registerWith(client, { params: { foo: "bar" } }, { userId: "..." }); +}) + .registerWith(client) + .addTriggerVariant( + "ericallam/hello-world", + gh.triggers.onIssueComment({ + repo: "ericallam/hello-world", + }) + ); new Job({ + id: "initialize-github-repo", + name: "Initialize GitHub Repo", + version: "0.1.1", + logLevel: "debug", + connections: { + gh, + sl, + }, + trigger: customEvent({ + name: "repo.created", + schema: z.object({ + repo: z.string(), + }), + }), + run: async (event, io, ctx) => { + await io.addTriggerVariant( + notifySlackONNewCommentsJob, + event.repo, + gh.triggers.onIssueComment({ + repo: event.repo, + }) + ); + }, +}).registerWith(client); + +const waitForEventInJob = new Job({ id: "wait-for-event-in-job", name: "Wait for event in job", version: "0.1.1", @@ -140,6 +166,18 @@ new Job({ }, }).registerWith(client); +client.addTriggerVariant( + waitForEventInJob, + "custom-event-3", + customEvent({ + name: "my-custom-event-3", + source: "my-source", + schema: z.object({ + foo: z.string(), + }), + }) +); + export default async function handler( req: NextApiRequest, res: NextApiResponse @@ -150,14 +188,13 @@ export default async function handler( if (!response) { res.status(404).json({ error: "Not found" }); + return; } res.status(response.status).json(response.body); } -client.listen().catch(console.error); - function normalizeRequest(req: NextApiRequest): NormalizedRequest { const normalizedHeaders = Object.entries(req.headers).reduce( (acc, [key, value]) => { diff --git a/integrations/github/src/index.ts b/integrations/github/src/index.ts index d94d1d411..5fd779457 100644 --- a/integrations/github/src/index.ts +++ b/integrations/github/src/index.ts @@ -92,7 +92,7 @@ function buildRepoWebhookTrigger( client: ClientOptions, id?: string, filter?: EventFilter -): (params: { repo: string }) => Trigger { +): (params: { repo: string }) => ExternalSourceEventTrigger { return (params: { repo: string }) => new ExternalSourceEventTrigger({ title, diff --git a/integrations/slack/src/client.ts b/integrations/slack/src/client.ts index 2bb98f85c..fb02717cb 100644 --- a/integrations/slack/src/client.ts +++ b/integrations/slack/src/client.ts @@ -4,7 +4,5 @@ import { WebClient } from "@slack/web-api"; export const clientFactory: ClientFactory> = ( auth ) => { - console.log("Creating slack client", auth); - return new WebClient(auth.accessToken); }; diff --git a/packages/internal/src/index.ts b/packages/internal/src/index.ts index ed65b4e78..3d72e0686 100644 --- a/packages/internal/src/index.ts +++ b/packages/internal/src/index.ts @@ -1,2 +1,3 @@ export * from "./logger"; export * from "./schemas"; +export * from "./types"; diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index c3987c458..252ecd18e 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -1,10 +1,10 @@ +import { ulid } from "ulid"; import { z } from "zod"; +import { ConnectionAuthSchema, ConnectionConfigSchema } from "./connections"; +import { DisplayElementSchema } from "./elements"; import { DeserializedJsonSchema, SerializableJsonSchema } from "./json"; import { CachedTaskSchema, ServerTaskSchema, TaskSchema } from "./tasks"; import { TriggerMetadataSchema } from "./triggers"; -import { ulid } from "ulid"; -import { DisplayElementSchema } from "./elements"; -import { ConnectionAuthSchema, ConnectionMetadataSchema } from "./connections"; export const RegisterHttpEventSourceBodySchema = z.object({ key: z.string(), @@ -76,25 +76,25 @@ export const JobSchema = z.object({ name: z.string(), version: z.string(), trigger: TriggerMetadataSchema, - connections: z.array( - z.object({ - key: z.string(), - metadata: ConnectionMetadataSchema, - usesLocalAuth: z.boolean().default(false), - id: z.string().optional(), - }) - ), - supportsPreparation: z.boolean(), + connections: z.array(ConnectionConfigSchema), }); -export type ApiJob = z.infer; +export type JobMetadata = z.infer; export const GetJobResponseSchema = z.object({ - job: JobSchema, + metadata: JobSchema, + triggerVariants: z.array( + z.object({ + id: z.string(), + trigger: TriggerMetadataSchema, + }) + ), }); +export type GetJobResponse = z.infer; + export const GetJobsResponseSchema = z.object({ - jobs: z.array(JobSchema), + jobs: z.array(GetJobResponseSchema), }); export const RawEventSchema = z.object({ @@ -203,15 +203,14 @@ export const SecureStringSchema = z.object({ interpolations: z.array(z.string()), }); -export const PrepareForJobExecutionBodySchema = z.object({ +export const PrepareJobTriggerBodySchema = z.object({ id: z.string(), version: z.string(), - connections: z.record(ConnectionAuthSchema), + connection: ConnectionAuthSchema.optional(), + variantId: z.string().optional(), }); -export type PrepareForJobExecutionBody = z.infer< - typeof PrepareForJobExecutionBodySchema ->; +export type PrepareJobTriggerBody = z.infer; export const PrepareForJobExecutionResponseSchema = z.object({ ok: z.boolean(), @@ -299,3 +298,15 @@ export const HttpSourceResponseSchema = z.object({ response: NormalizedResponseSchema, 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 +>; diff --git a/packages/internal/src/schemas/connections.ts b/packages/internal/src/schemas/connections.ts index 762cf4ae9..81e92ba3e 100644 --- a/packages/internal/src/schemas/connections.ts +++ b/packages/internal/src/schemas/connections.ts @@ -16,3 +16,30 @@ export const ConnectionAuthSchema = z.object({ }); export type ConnectionAuth = z.infer; + +const CommonConnectionConfigSchema = z.object({ + key: z.string().optional(), + metadata: ConnectionMetadataSchema, +}); + +const LocalAuthConnectionConfigSchema = CommonConnectionConfigSchema.extend({ + auth: z.literal("local"), +}); + +const HostedAuthConnectionConfigSchema = CommonConnectionConfigSchema.extend({ + auth: z.literal("hosted"), + id: z.string(), +}); + +export const ConnectionConfigSchema = z.discriminatedUnion("auth", [ + LocalAuthConnectionConfigSchema, + HostedAuthConnectionConfigSchema, +]); + +export type ConnectionConfig = z.infer; +export type LocalAuthConnectionConfig = z.infer< + typeof LocalAuthConnectionConfigSchema +>; +export type HostedAuthConnectionConfig = z.infer< + typeof HostedAuthConnectionConfigSchema +>; diff --git a/packages/internal/src/schemas/triggers.ts b/packages/internal/src/schemas/triggers.ts index b050f611b..a5737f836 100644 --- a/packages/internal/src/schemas/triggers.ts +++ b/packages/internal/src/schemas/triggers.ts @@ -1,7 +1,7 @@ import { z } from "zod"; -import { ConnectionMetadataSchema } from "./connections"; import { EventRuleSchema } from "./eventFilter"; import { DeserializedJsonSchema } from "./json"; +import { ConnectionAuthSchema, ConnectionConfigSchema } from "./connections"; export const TriggerMetadataSchema = z.object({ title: z.string(), @@ -14,13 +14,15 @@ export const TriggerMetadataSchema = z.object({ ), eventRule: EventRuleSchema, schema: DeserializedJsonSchema.optional(), - connection: z - .object({ - metadata: ConnectionMetadataSchema, - usesLocalAuth: z.boolean(), - id: z.string().optional(), - }) - .optional(), + connection: ConnectionConfigSchema.optional(), + supportsPreparation: z.boolean(), }); export type TriggerMetadata = z.infer; + +export const TriggerVariantConfigSchema = z.object({ + id: z.string(), + trigger: TriggerMetadataSchema, +}); + +export type TriggerVariantConfig = z.infer; diff --git a/packages/internal/src/types.ts b/packages/internal/src/types.ts new file mode 100644 index 000000000..7d0697bf6 --- /dev/null +++ b/packages/internal/src/types.ts @@ -0,0 +1,4 @@ +// See this for more: https://twitter.com/mattpocockuk/status/1653403198885904387?s=20 +export type Prettify = { + [K in keyof T]: T[K]; +} & {}; diff --git a/packages/trigger-sdk/src/apiClient.ts b/packages/trigger-sdk/src/apiClient.ts index 7bfa0b164..d97e3234c 100644 --- a/packages/trigger-sdk/src/apiClient.ts +++ b/packages/trigger-sdk/src/apiClient.ts @@ -4,15 +4,19 @@ import { CreateRunBody, CreateRunResponseBodySchema, HttpEventSource, + LogLevel, LogMessage, + Logger, RegisterHttpEventSourceBody, RunTaskBodyInput, SendEvent, SendEventOptions, ServerTask, + TriggerVariantResponseBody, + TriggerVariantConfig, + TriggerVariantResponseBodySchema, UpdateHttpEventSourceBody, } from "@trigger.dev/internal"; -import { Logger, LogLevel } from "@trigger.dev/internal"; export type ApiClientOptions = { apiKey?: string; @@ -263,6 +267,56 @@ export class ApiClient { return await response.json(); } + async addTriggerVariant( + client: string, + jobId: string, + jobVersion: string, + config: TriggerVariantConfig + ): Promise { + const apiKey = await this.#apiKey(); + + this.#logger.debug("Adding Trigger Variant", { + client, + jobId, + jobVersion, + config, + }); + + const response = await fetch( + `${this.#apiUrl}/api/v3/${client}/jobs/${jobId}/${jobVersion}/variants`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${apiKey}`, + }, + body: JSON.stringify(config), + } + ); + + if (response.status === 404) { + throw new Error( + `Failed to add trigger variant, got status code ${response.status}` + ); + } + + if (response.status >= 400 && response.status < 500) { + const body = await response.json(); + + throw new Error(body.error); + } + + if (response.status !== 200) { + throw new Error( + `Failed to add trigger variant, got status code ${response.status}` + ); + } + + const anyBody = await response.json(); + + return TriggerVariantResponseBodySchema.parse(anyBody); + } + async registerHttpSource( client: string, source: RegisterHttpEventSourceBody diff --git a/packages/trigger-sdk/src/externalSource.ts b/packages/trigger-sdk/src/externalSource.ts index e107c2f4d..2759d6525 100644 --- a/packages/trigger-sdk/src/externalSource.ts +++ b/packages/trigger-sdk/src/externalSource.ts @@ -75,10 +75,7 @@ export interface AnyExternalSource { auth?: ConnectionAuth ) => Promise<{ response: NormalizedResponse; events: SendEvent[] }>; eventElements: (event: ApiEventLog) => DisplayElement[]; - prepareForExecution: ( - client: TriggerClient, - auth?: ConnectionAuth - ) => Promise; + prepare: (client: TriggerClient, auth?: ConnectionAuth) => Promise; } export class ExternalSource @@ -108,7 +105,7 @@ export class ExternalSource return this.options.usesLocalAuth; } - async prepareForExecution(client: TriggerClient, auth?: ConnectionAuth) { + async prepare(client: TriggerClient, auth?: ConnectionAuth) { return this.options.register(client, auth); } diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts index 7148e9360..7b1dfa175 100644 --- a/packages/trigger-sdk/src/io.ts +++ b/packages/trigger-sdk/src/io.ts @@ -11,6 +11,8 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { webcrypto } from "node:crypto"; import { ApiClient } from "./apiClient"; import { Trigger } from "./triggers"; +import { Job } from "./job"; +import { TriggerClient } from "./triggerClient"; export class ResumeWithTask { constructor(public task: ServerTask) {} @@ -21,6 +23,7 @@ export type IOTask = ServerTask; export type IOOptions = { id: string; apiClient: ApiClient; + client: TriggerClient; logger?: Logger; logLevel?: LogLevel; cachedTasks?: Array; @@ -29,6 +32,7 @@ export type IOOptions = { export class IO { #id: string; #apiClient: ApiClient; + #client: TriggerClient; #logger: Logger; #cachedTasks: Map; #taskStorage: AsyncLocalStorage<{ taskId: string }>; @@ -36,6 +40,7 @@ export class IO { constructor(options: IOOptions) { this.#id = options.id; this.#apiClient = options.apiClient; + this.#client = options.client; this.#logger = options.logger ?? new Logger("trigger.dev", options.logLevel); this.#cachedTasks = new Map(); @@ -69,6 +74,72 @@ export class IO { ); } + async addTriggerVariant>( + job: Job, + id: string, + trigger: TTrigger + ) { + const metadata = trigger.toJSON(); + + const response = await this.runTask( + id, + { + name: `Add trigger to job`, + description: `Add trigger ${metadata.title} to job ${job.id}`, + elements: metadata.elements, + }, + async (task) => { + const subResponse1 = await this.runTask( + "register-trigger-variant", + { + name: `Register trigger variant`, + description: `Register trigger variant ${metadata.title} to job ${job.id}`, + elements: metadata.elements, + }, + async (task) => { + return await this.#apiClient.addTriggerVariant( + this.#client.name, + job.id, + job.version, + { + id, + trigger: metadata, + } + ); + } + ); + + if (subResponse1.ready) { + return subResponse1; + } + + await this.runTask( + "prepare-trigger-variant", + { + name: "Prepare trigger variant", + description: `Prepare trigger variant ${metadata.title} to job ${job.id}`, + elements: metadata.elements, + }, + async (task) => { + // todo: trigger.prepare should take the io as an argument and everything inside there should happen within subtasks + // the way we can do this is by reusing the job system when running the trigger.prepare function, using something like "Shadow Jobs" + // that are used internally by the trigger.dev system, but are not exposed to the user + // Each trigger that needs to be prepared will have a shadow job that is run in the background + // so instead of writing custom code for each thing trigger needs to do internally, we can just use the job system + // this will make our internal code much more reliable, and it will also allow us to do stuff like registering a trigger + // both at "static" time and at "runtime", for example when listening for a webhook in the middle of a job + // or registering a trigger variant when a job is running + // This is crucial because if we have a trigger.prepare function that makes many different API calls, we might start running into function timeout issues + // We could also explore showing these to the user, under something like "internal jobs" so we can surface more information to the user about what the system is doing + return await trigger.prepare(this.#client, subResponse1.auth); + } + ); + + return { ok: true }; + } + ); + } + async runTask( key: string | any[], options: RunTaskOptions, diff --git a/packages/trigger-sdk/src/job.ts b/packages/trigger-sdk/src/job.ts index 6bfde7cd0..0c1f2b4b2 100644 --- a/packages/trigger-sdk/src/job.ts +++ b/packages/trigger-sdk/src/job.ts @@ -1,34 +1,41 @@ -import { ConnectionAuth, LogLevel } from "@trigger.dev/internal"; +import { + ConnectionAuth, + ConnectionConfig, + JobMetadata, + LogLevel, +} from "@trigger.dev/internal"; import { Connection, IOWithConnections } from "./connections"; import { TriggerClient } from "./triggerClient"; -import { Trigger } from "./triggers"; +import { Trigger, TriggerEventType } from "./triggers"; import type { TriggerContext } from "./types"; export type JobOptions< - TEventType extends object = {}, + TTrigger extends Trigger, TConnections extends Record> = {} > = { id: string; name: string; version: string; - trigger: Trigger; + trigger: TTrigger; logLevel?: LogLevel; connections?: TConnections; run: ( - event: TEventType, + event: TriggerEventType, io: IOWithConnections, ctx: TriggerContext ) => Promise; }; export class Job< - TEventType extends object, + TTrigger extends Trigger, TConnections extends Record> > { - readonly options: JobOptions; + readonly options: JobOptions; - constructor(options: JobOptions) { + client?: TriggerClient; + + constructor(options: JobOptions) { this.options = options; this.#validate(); } @@ -49,31 +56,60 @@ export class Job< return this.options.version; } - get connections() { + get connections(): Array { return Object.keys(this.options.connections ?? {}).map((key) => { const connection = this.options.connections![key]; - return { - key, - metadata: connection.metadata, - usesLocalAuth: connection.usesLocalAuth, - id: connection.id, - }; + if (connection.usesLocalAuth) { + return { + auth: "local", + key, + metadata: connection.metadata, + }; + } else { + return { + auth: "hosted", + key, + metadata: connection.metadata, + id: connection.id!, + }; + } }); } registerWith(client: TriggerClient) { - client.register(this as unknown as Job<{}, any>); + if (this.client) { + throw new Error( + `Job "${this.id}" has already been registered with a client.` + ); + } + + this.client = client; + + client.register(this); + + return this; } - toJSON() { + addTriggerVariant(id: string, trigger: TTrigger) { + if (!this.client) { + throw new Error( + `Job "${this.id}" has not been registered with a client.` + ); + } + + this.client.addTriggerVariant(this, id, trigger); + + return this; + } + + toJSON(): JobMetadata { return { id: this.id, name: this.name, version: this.version, trigger: this.trigger.toJSON(), connections: this.connections, - supportsPreparation: this.trigger.supportsPreparation, }; } @@ -81,7 +117,7 @@ export class Job< client: TriggerClient, connections: Record ) { - await this.trigger.prepareForExecution(client, connections.__trigger); + await this.trigger.prepare(client, connections.__trigger); } // Make sure the id is valid (must only contain alphanumeric characters and dashes) diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 3b091bad7..144da66f7 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -11,7 +11,7 @@ import { LogLevel, NormalizedRequest, NormalizedResponse, - PrepareForJobExecutionBodySchema, + PrepareJobTriggerBodySchema, RegisterHttpEventSourceBody, SendEvent, UpdateHttpEventSourceBody, @@ -26,6 +26,7 @@ import { AnyExternalSource } from "./externalSource"; import { IO, ResumeWithTask } from "./io"; import { Job } from "./job"; import { ContextLogger } from "./logger"; +import { Trigger } from "./triggers"; import { TriggerContext } from "./types"; export type TriggerClientOptions = { @@ -42,7 +43,11 @@ export type ListenOptions = { export class TriggerClient { #options: TriggerClientOptions; - #registeredJobs: Record> = {}; + #registeredJobs: Record, any>> = {}; + #registeredTriggerVariants: Record< + string, + Array<{ trigger: Trigger; id: string }> + > = {}; #registeredSources = new Map(); #client: ApiClient; #logger: Logger; @@ -96,9 +101,17 @@ export class TriggerClient { }; } + const triggerVariants = this.#registeredTriggerVariants[job.id] ?? []; + return { status: 200, - body: job.toJSON(), + body: { + metadata: job.toJSON(), + triggerVariants: triggerVariants.map(({ trigger, id }) => ({ + id, + trigger: trigger.toJSON(), + })), + }, }; } @@ -106,7 +119,12 @@ export class TriggerClient { return { status: 200, body: { - jobs: Object.values(this.#registeredJobs).map((job) => job.toJSON()), + jobs: Object.values(this.#registeredJobs).map((job) => ({ + metadata: job.toJSON(), + triggerVariants: ( + this.#registeredTriggerVariants[job.id] ?? [] + ).map(({ id, trigger }) => ({ id, trigger: trigger.toJSON() })), + })), }, }; } @@ -168,10 +186,8 @@ export class TriggerClient { }, }; } - case "PREPARE_FOR_JOB_EXECUTION": { - const payload = PrepareForJobExecutionBodySchema.safeParse( - request.body - ); + case "PREPARE_JOB_TRIGGER": { + const payload = PrepareJobTriggerBodySchema.safeParse(request.body); if (!payload.success) { return { @@ -193,7 +209,7 @@ export class TriggerClient { }; } - await this.#prepareJobForExecution(registeredJob, payload.data); + await this.#prepareJobTrigger(registeredJob, payload.data); return { status: 200, @@ -254,8 +270,8 @@ export class TriggerClient { } register(thing: AnyExternalSource): void; - register(thing: Job<{}, any>): void; - register(thing: Job<{}, any> | AnyExternalSource): void { + register(thing: Job, any>): void; + register(thing: Job, any> | AnyExternalSource): void { if (thing instanceof Job) { this.#registeredJobs[thing.id] = thing; @@ -265,6 +281,18 @@ export class TriggerClient { } } + addTriggerVariant>( + job: Job, + id: string, + trigger: TTrigger + ) { + const jobTriggerVariants = this.#registeredTriggerVariants[job.id] ?? []; + jobTriggerVariants.push({ trigger, id }); + this.#registeredTriggerVariants[job.id] = jobTriggerVariants; + + trigger.registerWith(this); + } + authorized(apiKey: string) { const localApiKey = this.#options.apiKey ?? process.env.TRIGGER_API_KEY; @@ -295,21 +323,34 @@ export class TriggerClient { return await this.#client.updateHttpSource(this.name, id, source); } - async #prepareJobForExecution( - job: Job<{}, any>, + async #prepareJobTrigger( + job: Job, any>, preparationData: { id: string; version: string; - connections: Record; + connection?: ConnectionAuth; + variantId?: string; } ): Promise { - this.#logger.debug("preparing job for execution", { job: job.toJSON() }); + this.#logger.debug("preparing job trigger", { job: job.toJSON() }); if (job.version !== preparationData.version) { return; } - await job.prepareForExecution(this, preparationData.connections); + if (preparationData.variantId) { + const variant = this.#registeredTriggerVariants[job.id].find( + (v) => v.id === preparationData.variantId + ); + + if (!variant) { + return; + } + + await variant.trigger.prepare(this, preparationData.connection); + } else { + await job.trigger.prepare(this, preparationData.connection); + } } async #handleHttpSourceRequest( @@ -335,21 +376,7 @@ export class TriggerClient { return await source.handler(this, { request: sourceRequest, secret }, auth); } - async #createExecution(job: Job<{}, any>, event: ApiEventLog) { - this.#logger.debug("creating execution", { event, job: job.toJSON() }); - - // Create a new job execution - const execution = await this.#client.createRun({ - client: this.name, - job: job.toJSON(), - event, - elements: job.trigger.eventElements(event), - }); - - return execution; - } - - async #executeJob(execution: ExecuteJobBody, job: Job<{}, any>) { + async #executeJob(execution: ExecuteJobBody, job: Job, any>) { this.#logger.debug("executing job", { execution, job: job.toJSON() }); const abortController = new AbortController(); @@ -359,17 +386,14 @@ export class TriggerClient { cachedTasks: execution.tasks, apiClient: this.#client, logger: this.#logger, + client: this, }); - const ioWithConnections = await this.#createIOWithConnections( - io, - execution, - job - ); + const ioWithConnections = this.#createIOWithConnections(io, execution, job); try { const output = await job.options.run( - execution.event.payload ?? {}, + job.trigger.parsePayload(execution.event.payload ?? {}), // todo: actually parse the payload through the trigger ioWithConnections, this.#createJobContext(execution, io, abortController.signal) ); @@ -404,7 +428,7 @@ export class TriggerClient { >( io: IO, execution: ExecuteJobBody, - job: Job<{}, TConnections> + job: Job, TConnections> ): IOWithConnections { const jobConnections = job.options.connections; diff --git a/packages/trigger-sdk/src/triggers.ts b/packages/trigger-sdk/src/triggers.ts index c2a626c35..e03f2bb57 100644 --- a/packages/trigger-sdk/src/triggers.ts +++ b/packages/trigger-sdk/src/triggers.ts @@ -1,6 +1,7 @@ import type { ApiEventLog, ConnectionAuth, + ConnectionConfig, EventFilter, EventRule, TriggerMetadata, @@ -11,15 +12,15 @@ import zodToJsonSchema from "zod-to-json-schema"; import { AnyExternalSource } from "./externalSource"; import { TriggerClient } from "./triggerClient"; +export type TriggerEventType> = + TTrigger extends Trigger ? TEventType : never; + export interface Trigger { eventElements(event: ApiEventLog): DisplayElement[]; toJSON(): TriggerMetadata; registerWith(client: TriggerClient): void; - prepareForExecution( - client: TriggerClient, - auth?: ConnectionAuth - ): Promise; - supportsPreparation: boolean; + prepare(client: TriggerClient, auth?: ConnectionAuth): Promise; + parsePayload(payload: unknown): TEventType; } export type CustomEventTriggerOptions = { @@ -54,15 +55,20 @@ export class CustomEventTrigger source: this.#options.source ?? "trigger.dev", payload: this.#options.filter ?? {}, }, + supportsPreparation: false, }; } - get supportsPreparation() { - return false; + parsePayload(payload: unknown): z.infer { + if (!this.#options.schema) { + return payload; + } + + return this.#options.schema.parse(payload); } registerWith(client: TriggerClient) {} - async prepareForExecution(client: TriggerClient, auth?: ConnectionAuth) {} + async prepare(client: TriggerClient, auth?: ConnectionAuth) {} } export function customEvent( @@ -85,16 +91,17 @@ export class ExternalSourceEventTrigger implements Trigger { return this.options.source.eventElements(event); } + parsePayload(payload: unknown): TEvent { + return payload as TEvent; + } + toJSON(): TriggerMetadata { return { title: this.options.title, elements: this.options.elements, - connection: { - metadata: this.options.source.connection, - usesLocalAuth: this.options.source.usesLocalAuth, - id: this.options.source.id, - }, + connection: this.connection, eventRule: this.options.eventRule, + supportsPreparation: true, }; } @@ -102,11 +109,22 @@ export class ExternalSourceEventTrigger implements Trigger { client.register(this.options.source); } - get supportsPreparation() { - return true; + async prepare(client: TriggerClient, auth?: ConnectionAuth) { + return this.options.source.prepare(client, auth); } - async prepareForExecution(client: TriggerClient, auth?: ConnectionAuth) { - return this.options.source.prepareForExecution(client, auth); + get connection(): ConnectionConfig { + if (this.options.source.usesLocalAuth) { + return { + auth: "local", + metadata: this.options.source.connection, + }; + } else { + return { + auth: "hosted", + metadata: this.options.source.connection, + id: this.options.source.id!, + }; + } } }