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

WIP trigger variants

Browse Source
This commit is contained in:
Eric Allam
2023-05-04 10:06:06 +01:00
parent eb83f941d2
commit 3e87e832c7
33 changed files with 1521 additions and 733 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/models/jobConnection.server.ts
+46 -20
View File
@@ -13,30 +13,56 @@ export async function resolveJobConnections(
const result: Record<string, ConnectionAuth> = {};
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<ConnectionAuth | undefined> {
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<ConnectionAuth | undefined> {
if (!connection) {
return;
}
const response = await apiConnectionRepository.getCredentials(connection);
if (!response) {
return;
}
return {
type: "oauth2",
scopes: response.scopes,
accessToken: response.accessToken,
};
}
apps/webapp/app/routes/api/v3/$endpointSlug/jobs/$jobId.$jobVersion.variants.ts
+73
View File
@@ -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 });
}
}
apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts
+1
View File
@@ -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({
apps/webapp/app/services/apiAuth.server.ts
-2
View File
@@ -12,8 +12,6 @@ export async function authenticateApiRequest(
): Promise<AuthenticatedEnvironment | null | undefined> {
const rawAuthorization = request.headers.get("Authorization");
console.log(rawAuthorization);
const authorization = AuthorizationHeaderSchema.safeParse(rawAuthorization);
if (!authorization.success) {
apps/webapp/app/services/clientApi.server.ts
+3 -3
View File
@@ -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),
});
apps/webapp/app/services/endpoints/endpointRegistered.server.ts
+4 -475
View File
@@ -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<void> {
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<JobConnection> = [];
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<JobConnection> {
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<JobConnection> {
// 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";
}
apps/webapp/app/services/endpoints/prepareForJobExecution.server.ts
-34
View File
@@ -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}` }
);
}
}
}
apps/webapp/app/services/endpoints/prepareJobInstance.server.ts
+28 -3
View File
@@ -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}`,
}
);
}
}
}
apps/webapp/app/services/endpoints/prepareTriggerVariant.server.ts
+79
View File
@@ -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,
},
});
}
}
apps/webapp/app/services/events/deliverEvent.server.ts
+3 -3
View File
@@ -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,
apps/webapp/app/services/jobs/registerJob.server.ts
+635
View File
@@ -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<JobInstance> {
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<string>();
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<string, ApiConnection>();
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<string>();
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<string, ApiConnection>,
previousVariants?: Array<JobTriggerVariant>
): Promise<JobConnection | undefined> {
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<string, ApiConnection>,
overrideKey?: string
): Promise<JobConnection> {
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<JobConnection> {
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,
},
});
}
}
apps/webapp/app/services/jobs/registerJobVariant.server.ts
+111
View File
@@ -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<TriggerVariantResponseBody> {
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),
};
}
}
apps/webapp/app/services/jobs/utils.server.ts
-21
View File
@@ -1,21 +0,0 @@
import { JobConnection } from ".prisma/client";
export async function allConnectionsReady(
connections: Array<JobConnection>
): Promise<boolean> {
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;
});
}
apps/webapp/app/services/runs/createRun.server.ts
+1 -1
View File
@@ -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;
apps/webapp/app/services/worker.server.ts
+19 -6
View File
@@ -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<typeof workerCatalog>;
@@ -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);
},
apps/webapp/prisma/migrations/20230503094127_add_trigger_variants/migration.sql
+21
View File
@@ -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;
apps/webapp/prisma/migrations/20230503113400_add_slug_to_trigger_variants/migration.sql
+12
View File
@@ -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");
apps/webapp/prisma/migrations/20230503172156_rename_create_execution_to_create_run/migration.sql
+19
View File
@@ -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';
apps/webapp/prisma/schema.prisma
+64 -43
View File
@@ -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
}
examples/nextjs-example/src/pages/api/trigger.ts
+48 -11
View File
@@ -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]) => {
integrations/github/src/index.ts
+1 -1
View File
@@ -92,7 +92,7 @@ function buildRepoWebhookTrigger<TEventType>(
client: ClientOptions,
id?: string,
filter?: EventFilter
): (params: { repo: string }) => Trigger<TEventType> {
): (params: { repo: string }) => ExternalSourceEventTrigger<TEventType> {
return (params: { repo: string }) =>
new ExternalSourceEventTrigger<TEventType>({
title,
integrations/slack/src/client.ts
-2
View File
@@ -4,7 +4,5 @@ import { WebClient } from "@slack/web-api";
export const clientFactory: ClientFactory<InstanceType<typeof WebClient>> = (
auth
) => {
console.log("Creating slack client", auth);
return new WebClient(auth.accessToken);
};
packages/internal/src/index.ts
+1
View File
@@ -1,2 +1,3 @@
export * from "./logger";
export * from "./schemas";
export * from "./types";
packages/internal/src/schemas/api.ts
+31 -20
View File
@@ -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<typeof JobSchema>;
export type JobMetadata = z.infer<typeof JobSchema>;
export const GetJobResponseSchema = z.object({
job: JobSchema,
metadata: JobSchema,
triggerVariants: z.array(
z.object({
id: z.string(),
trigger: TriggerMetadataSchema,
})
),
});
export type GetJobResponse = z.infer<typeof GetJobResponseSchema>;
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<typeof PrepareJobTriggerBodySchema>;
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
>;
packages/internal/src/schemas/connections.ts
+27
View File
@@ -16,3 +16,30 @@ export const ConnectionAuthSchema = z.object({
});
export type ConnectionAuth = z.infer<typeof ConnectionAuthSchema>;
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<typeof ConnectionConfigSchema>;
export type LocalAuthConnectionConfig = z.infer<
typeof LocalAuthConnectionConfigSchema
>;
export type HostedAuthConnectionConfig = z.infer<
typeof HostedAuthConnectionConfigSchema
>;
packages/internal/src/schemas/triggers.ts
+10 -8
View File
@@ -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<typeof TriggerMetadataSchema>;
export const TriggerVariantConfigSchema = z.object({
id: z.string(),
trigger: TriggerMetadataSchema,
});
export type TriggerVariantConfig = z.infer<typeof TriggerVariantConfigSchema>;
packages/internal/src/types.ts
+4
View File
@@ -0,0 +1,4 @@
// See this for more: https://twitter.com/mattpocockuk/status/1653403198885904387?s=20
export type Prettify<T> = {
[K in keyof T]: T[K];
} & {};
packages/trigger-sdk/src/apiClient.ts
+55 -1
View File
@@ -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<TriggerVariantResponseBody> {
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
packages/trigger-sdk/src/externalSource.ts
+2 -5
View File
@@ -75,10 +75,7 @@ export interface AnyExternalSource {
auth?: ConnectionAuth
) => Promise<{ response: NormalizedResponse; events: SendEvent[] }>;
eventElements: (event: ApiEventLog) => DisplayElement[];
prepareForExecution: (
client: TriggerClient,
auth?: ConnectionAuth
) => Promise<void>;
prepare: (client: TriggerClient, auth?: ConnectionAuth) => Promise<void>;
}
export class ExternalSource<TChannel extends ChannelNames>
@@ -108,7 +105,7 @@ export class ExternalSource<TChannel extends ChannelNames>
return this.options.usesLocalAuth;
}
async prepareForExecution(client: TriggerClient, auth?: ConnectionAuth) {
async prepare(client: TriggerClient, auth?: ConnectionAuth) {
return this.options.register(client, auth);
}
packages/trigger-sdk/src/io.ts
+71
View File
@@ -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<CachedTask>;
@@ -29,6 +32,7 @@ export type IOOptions = {
export class IO {
#id: string;
#apiClient: ApiClient;
#client: TriggerClient;
#logger: Logger;
#cachedTasks: Map<string, CachedTask>;
#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<TTrigger extends Trigger<any>>(
job: Job<TTrigger, any>,
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<T extends SerializableJson | void = void>(
key: string | any[],
options: RunTaskOptions,
packages/trigger-sdk/src/job.ts
+55 -19
View File
@@ -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<any>,
TConnections extends Record<string, Connection<any, any>> = {}
> = {
id: string;
name: string;
version: string;
trigger: Trigger<TEventType>;
trigger: TTrigger;
logLevel?: LogLevel;
connections?: TConnections;
run: (
event: TEventType,
event: TriggerEventType<TTrigger>,
io: IOWithConnections<TConnections>,
ctx: TriggerContext
) => Promise<any>;
};
export class Job<
TEventType extends object,
TTrigger extends Trigger<any>,
TConnections extends Record<string, Connection<any, any>>
> {
readonly options: JobOptions<TEventType, TConnections>;
readonly options: JobOptions<TTrigger, TConnections>;
constructor(options: JobOptions<TEventType, TConnections>) {
client?: TriggerClient;
constructor(options: JobOptions<TTrigger, TConnections>) {
this.options = options;
this.#validate();
}
@@ -49,31 +56,60 @@ export class Job<
return this.options.version;
}
get connections() {
get connections(): Array<ConnectionConfig> {
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<string, ConnectionAuth>
) {
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)
packages/trigger-sdk/src/triggerClient.ts
+62 -38
View File
@@ -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<string, Job<{}, any>> = {};
#registeredJobs: Record<string, Job<Trigger<any>, any>> = {};
#registeredTriggerVariants: Record<
string,
Array<{ trigger: Trigger<any>; id: string }>
> = {};
#registeredSources = new Map<string, AnyExternalSource>();
#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<Trigger<any>, any>): void;
register(thing: Job<Trigger<any>, any> | AnyExternalSource): void {
if (thing instanceof Job) {
this.#registeredJobs[thing.id] = thing;
@@ -265,6 +281,18 @@ export class TriggerClient {
}
}
addTriggerVariant<TTrigger extends Trigger<any>>(
job: Job<TTrigger, any>,
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<Trigger<any>, any>,
preparationData: {
id: string;
version: string;
connections: Record<string, ConnectionAuth>;
connection?: ConnectionAuth;
variantId?: string;
}
): Promise<void> {
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<Trigger<any>, 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<Trigger<any>, TConnections>
): IOWithConnections<TConnections> {
const jobConnections = job.options.connections;
packages/trigger-sdk/src/triggers.ts
+35 -17
View File
@@ -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<any>> =
TTrigger extends Trigger<infer TEventType> ? TEventType : never;
export interface Trigger<TEventType = any> {
eventElements(event: ApiEventLog): DisplayElement[];
toJSON(): TriggerMetadata;
registerWith(client: TriggerClient): void;
prepareForExecution(
client: TriggerClient,
auth?: ConnectionAuth
): Promise<void>;
supportsPreparation: boolean;
prepare(client: TriggerClient, auth?: ConnectionAuth): Promise<void>;
parsePayload(payload: unknown): TEventType;
}
export type CustomEventTriggerOptions<TSchema extends z.ZodTypeAny> = {
@@ -54,15 +55,20 @@ export class CustomEventTrigger<TSchema extends z.ZodTypeAny>
source: this.#options.source ?? "trigger.dev",
payload: this.#options.filter ?? {},
},
supportsPreparation: false,
};
}
get supportsPreparation() {
return false;
parsePayload(payload: unknown): z.infer<TSchema> {
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<TSchema extends z.ZodTypeAny>(
@@ -85,16 +91,17 @@ export class ExternalSourceEventTrigger<TEvent> implements Trigger<TEvent> {
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<TEvent> implements Trigger<TEvent> {
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!,
};
}
}
}