From c77dfae65304fff23edc15da55bbfb1960e2227a Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 4 May 2023 21:18:19 +0100 Subject: [PATCH] WIP new internal job system --- apps/webapp/app/routes/api/v3/events.ts | 6 +- .../api/v3/runs.$runId.tasks.$id.complete.ts | 19 +- .../app/routes/api/v3/runs.$runId.tasks.ts | 3 +- apps/webapp/app/services/clientApi.server.ts | 8 +- .../endpoints/prepareJobInstance.server.ts | 90 ++--- .../app/services/jobs/registerJob.server.ts | 136 ++------ .../jobs/registerJobVariant.server.ts | 22 +- .../deliverHttpSourceRequest.server.ts | 58 ++-- apps/webapp/package.json | 1 + .../migration.sql | 2 + .../migration.sql | 9 + .../migration.sql | 2 + apps/webapp/prisma/schema.prisma | 8 +- .../nextjs-example/src/pages/api/trigger.ts | 230 ++++++------- integrations/github/src/index.ts | 70 ++-- integrations/github/src/sources.ts | 123 +++---- integrations/github/src/tasks.ts | 132 ++++++++ integrations/github/src/types.ts | 12 - packages/internal/src/schemas/api.ts | 16 +- packages/internal/src/schemas/connections.ts | 1 - packages/internal/src/schemas/triggers.ts | 3 - packages/trigger-sdk/src/connections.ts | 2 +- packages/trigger-sdk/src/externalSource.ts | 123 ------- packages/trigger-sdk/src/index.ts | 5 +- packages/trigger-sdk/src/io.ts | 73 +++- packages/trigger-sdk/src/job.ts | 68 ++-- packages/trigger-sdk/src/triggerClient.ts | 172 +--------- packages/trigger-sdk/src/triggers.ts | 130 -------- .../trigger-sdk/src/triggers/customEvent.ts | 67 ++++ .../src/triggers/externalSource.ts | 315 ++++++++++++++++++ packages/trigger-sdk/src/types.ts | 39 ++- pnpm-lock.yaml | 48 +-- 32 files changed, 986 insertions(+), 1007 deletions(-) create mode 100644 apps/webapp/prisma/migrations/20230504090710_add_shadow_to_job/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230504152611_rename_shadow_to_internal/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230504200916_add_redact_to_tasks/migration.sql delete mode 100644 integrations/github/src/types.ts delete mode 100644 packages/trigger-sdk/src/externalSource.ts delete mode 100644 packages/trigger-sdk/src/triggers.ts create mode 100644 packages/trigger-sdk/src/triggers/customEvent.ts create mode 100644 packages/trigger-sdk/src/triggers/externalSource.ts diff --git a/apps/webapp/app/routes/api/v3/events.ts b/apps/webapp/app/routes/api/v3/events.ts index 1d4a2ddfa..bf45aee5f 100644 --- a/apps/webapp/app/routes/api/v3/events.ts +++ b/apps/webapp/app/routes/api/v3/events.ts @@ -5,10 +5,8 @@ import { SendEventBodySchema } from "@trigger.dev/internal"; import { generateErrorMessage } from "zod-error"; import type { PrismaClient } from "~/db.server"; import { prisma, PrismaErrorSchema } from "~/db.server"; -import { - authenticateApiRequest, - AuthenticatedEnvironment, -} from "~/services/apiAuth.server"; +import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; import { workerQueue } from "~/services/worker.server"; export async function action({ request }: ActionArgs) { diff --git a/apps/webapp/app/routes/api/v3/runs.$runId.tasks.$id.complete.ts b/apps/webapp/app/routes/api/v3/runs.$runId.tasks.$id.complete.ts index 25f4c29be..e015d2229 100644 --- a/apps/webapp/app/routes/api/v3/runs.$runId.tasks.$id.complete.ts +++ b/apps/webapp/app/routes/api/v3/runs.$runId.tasks.$id.complete.ts @@ -1,17 +1,12 @@ -import { RuntimeEnvironment } from ".prisma/client"; import type { ActionArgs } from "@remix-run/server-runtime"; import { json } from "@remix-run/server-runtime"; -import { - CompleteTaskBodyInputSchema, - CompleteTaskBodyOutput, -} from "@trigger.dev/internal"; +import type { CompleteTaskBodyOutput } from "@trigger.dev/internal"; +import { CompleteTaskBodyInputSchema } from "@trigger.dev/internal"; import { z } from "zod"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import { - authenticateApiRequest, - AuthenticatedEnvironment, -} from "~/services/apiAuth.server"; +import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; import { logger } from "~/services/logger"; const ParamsSchema = z.object({ @@ -37,7 +32,7 @@ export async function action({ request, params }: ActionArgs) { // Now parse the request body const anyBody = await request.json(); - logger.debug("CompleteExecutionTaskService.call() request body", { + logger.debug("CompleteRunTaskService.call() request body", { body: anyBody, runId, id, @@ -115,6 +110,10 @@ export class CompleteRunTaskService { existingTask.status === "COMPLETED" || existingTask.status === "ERRORED" ) { + logger.debug("Task already completed", { + existingTask, + }); + return existingTask; } 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 56c620aa5..de68cabde 100644 --- a/apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts +++ b/apps/webapp/app/routes/api/v3/runs.$runId.tasks.ts @@ -215,13 +215,14 @@ export class RunTaskService { delayUntil: taskBody.delayUntil, params: taskBody.params ?? undefined, elements: taskBody.elements ?? undefined, + redact: taskBody.redact ?? undefined, }, include: { run: true, }, }); - // todo: do this client side instead of adding an option to taskBody + // 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/clientApi.server.ts b/apps/webapp/app/services/clientApi.server.ts index 7111fd594..eef2a7286 100644 --- a/apps/webapp/app/services/clientApi.server.ts +++ b/apps/webapp/app/services/clientApi.server.ts @@ -1,14 +1,14 @@ import type { ApiEventLog, ConnectionAuth, - ExecuteJobBody, + RunJobBody, HttpSourceRequest, PrepareJobTriggerBody, } from "@trigger.dev/internal"; import { DeliverEventResponseSchema, ErrorWithStackSchema, - ExecuteJobResponseSchema, + RunJobResponseSchema, GetJobsResponseSchema, HttpSourceResponseSchema, PongResponseSchema, @@ -126,7 +126,7 @@ export class ClientApi { return DeliverEventResponseSchema.parse(anyBody); } - async executeJob(options: ExecuteJobBody) { + async executeJob(options: RunJobBody) { const response = await safeFetch(this.#url, { method: "POST", headers: { @@ -164,7 +164,7 @@ export class ClientApi { body: anyBody, }); - return ExecuteJobResponseSchema.parse(anyBody); + return RunJobResponseSchema.parse(anyBody); } async prepareJobTrigger(payload: PrepareJobTriggerBody) { diff --git a/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts b/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts index aa54376b7..f06d2badf 100644 --- a/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts +++ b/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts @@ -1,8 +1,7 @@ import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import { resolveJobConnection } from "~/models/jobConnection.server"; -import { ClientApi } from "../clientApi.server"; -import { workerQueue } from "../worker.server"; +import { IngestSendEvent } from "~/routes/api/v3/events"; +import semver from "semver"; export class PrepareJobInstanceService { #prismaClient: PrismaClient; @@ -17,70 +16,55 @@ export class PrepareJobInstanceService { id, }, include: { - connections: { - include: { - apiConnection: { - include: { - dataReference: true, - }, - }, - }, - where: { - key: "__trigger", - }, - }, job: true, - endpoint: { + environment: { include: { - environment: true, + organization: true, + project: true, }, }, triggerVariants: true, }, }); - const client = new ClientApi( - jobInstance.endpoint.environment.apiKey, - jobInstance.endpoint.url - ); + const service = new IngestSendEvent(); - const connection = jobInstance.connections[0]; - - const response = await client.prepareJobTrigger({ - id: jobInstance.job.slug, - version: jobInstance.version, - connection: connection - ? await resolveJobConnection(connection) - : undefined, - }); - - if (!response.ok) { - throw new Error("Something went wrong when preparing a job instance"); - } - - await this.#prismaClient.jobInstance.update({ - where: { - id, - }, - data: { - ready: true, + await service.call(jobInstance.environment, { + id: `${jobInstance.id}:prepare:${versionScopedToMinor( + jobInstance.version + )}`, + name: "internal.trigger.prepare", + source: "trigger.dev", + payload: { + jobId: jobInstance.job.slug, + jobVersion: jobInstance.version, }, }); for (const variant of jobInstance.triggerVariants) { - if (variant.ready) { - continue; - } - - await workerQueue.enqueue( - "prepareTriggerVariant", - { - id: variant.id, + await service.call(jobInstance.environment, { + id: `${jobInstance.id}:prepare:${versionScopedToMinor( + jobInstance.version + )}:${variant.id}`, + name: "internal.trigger.prepare", + source: "trigger.dev", + payload: { + jobId: jobInstance.job.slug, + jobVersion: jobInstance.version, + variantId: variant.slug, }, - { - queueName: `endpoint-${jobInstance.endpoint.id}`, - } - ); + }); } } } + +// Take a version string (e.g. 1.2.3) and return a version string that is scoped to the minor version (e.g. 1.2) +function versionScopedToMinor(version: string) { + const parsed = semver.parse(version); + + if (!parsed) { + throw new Error(`Invalid version: ${version}`); + } + + return `${parsed.major}.${parsed.minor}`; +} diff --git a/apps/webapp/app/services/jobs/registerJob.server.ts b/apps/webapp/app/services/jobs/registerJob.server.ts index 0d1da9c9a..af1224fcb 100644 --- a/apps/webapp/app/services/jobs/registerJob.server.ts +++ b/apps/webapp/app/services/jobs/registerJob.server.ts @@ -4,7 +4,6 @@ import type { Job, JobConnection, JobInstance, - JobTriggerVariant, } from ".prisma/client"; import type { ConnectionConfig, @@ -46,6 +45,8 @@ export class RegisterJobService { jobResponse ); + // TODO: deliver internal event that will prepare the main trigger + await workerQueue.enqueue( "prepareJobInstance", { id: jobInstance.id }, @@ -75,31 +76,13 @@ export class RegisterJobService { const connectionSlugs = new Set(); if (metadata.connections) { - for (const connection of metadata.connections) { + for (const connection of Object.values(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) { @@ -113,7 +96,7 @@ export class RegisterJobService { }); if (!apiConnection) { - // todo: find a better way to handle and message the user about this issue + // TODO: find a better way to handle and message the user about this issue throw new Error( `Could not find ApiConnection with slug ${connectionSlug}` ); @@ -143,6 +126,7 @@ export class RegisterJobService { }, slug: metadata.id, title: metadata.name, + internal: metadata.internal, }, update: { title: metadata.name, @@ -153,29 +137,9 @@ export class RegisterJobService { 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: { @@ -213,7 +177,6 @@ export class RegisterJobService { }, version: metadata.version, trigger: metadata.trigger, - ready, }, update: { trigger: metadata.trigger, @@ -229,25 +192,14 @@ export class RegisterJobService { 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) { + for (const [key, connection] of Object.entries(metadata.connections)) { const jobConnection = await this.#upsertJobConnection( job, jobInstance, connection, - apiConnections + apiConnections, + key ); jobConnections.add(jobConnection.id); @@ -291,20 +243,16 @@ export class RegisterJobService { if (triggerVariants) { for (const triggerVariant of triggerVariants) { - const jobConnection = await this.#upsertTriggerVariant( + await this.#upsertTriggerVariant( job, jobInstance, environment, triggerVariant.id, - triggerVariant.trigger, - apiConnections, - latestInstance?.triggerVariants + triggerVariant.trigger ); - - if (jobConnection) { - jobConnections.add(jobConnection.id); - } } + + // TODO: deliver internal event that will prepare the trigger variants } // Delete any connections that are no longer in the job @@ -352,24 +300,12 @@ export class RegisterJobService { } async #upsertTriggerVariant( - job: Job & { - connections: Array< - JobConnection & { apiConnection: ApiConnection | null } - >; - }, - jobInstance: JobInstance & { - connections: Array< - JobConnection & { apiConnection: ApiConnection | null } - >; - }, + job: Job, + jobInstance: JobInstance, environment: AuthenticatedEnvironment, id: string, - trigger: TriggerMetadata, - apiConnections: Map, - previousVariants?: Array - ): Promise { - const previousVariant = previousVariants?.find((v) => v.id === id); - + trigger: TriggerMetadata + ) { await this.#prismaClient.jobTriggerVariant.upsert({ where: { jobInstanceId_slug: { @@ -385,11 +321,6 @@ export class RegisterJobService { }, slug: id, data: trigger, - ready: trigger.supportsPreparation - ? previousVariant - ? previousVariant.ready - : false - : true, eventRule: { create: { event: trigger.eventRule.event, @@ -410,16 +341,6 @@ export class RegisterJobService { data: trigger, }, }); - - if (trigger.connection) { - return await this.#upsertJobConnection( - job, - jobInstance, - trigger.connection, - apiConnections, - `__trigger_${id}` - ); - } } async #upsertJobConnection( @@ -435,15 +356,10 @@ export class RegisterJobService { }, config: ConnectionConfig, apiConnections: Map, - overrideKey?: string + key: string ): Promise { if (config.auth === "local") { - return this.#upsertLocalAuthConnection( - job, - jobInstance, - config, - overrideKey - ); + return this.#upsertLocalAuthConnection(job, jobInstance, config, key); } const apiConnection = apiConnections.get(config.id); @@ -454,14 +370,6 @@ export class RegisterJobService { ); } - 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 @@ -557,14 +465,8 @@ export class RegisterJobService { >; }, config: LocalAuthConnectionConfig, - overrideKey?: string + key: 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 diff --git a/apps/webapp/app/services/jobs/registerJobVariant.server.ts b/apps/webapp/app/services/jobs/registerJobVariant.server.ts index fdacf9bd7..7a62d00d4 100644 --- a/apps/webapp/app/services/jobs/registerJobVariant.server.ts +++ b/apps/webapp/app/services/jobs/registerJobVariant.server.ts @@ -5,8 +5,6 @@ import type { 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; @@ -47,22 +45,6 @@ export class RegisterJobVariantService { }, }); - 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: { @@ -78,7 +60,6 @@ export class RegisterJobVariantService { }, slug: id, data: trigger, - ready: !trigger.supportsPreparation, eventRule: { create: { event: trigger.eventRule.event, @@ -100,12 +81,13 @@ export class RegisterJobVariantService { }, }); + // TODO: fire event to prepare trigger variant + return { id: triggerVariant.id, slug: triggerVariant.slug, data: trigger, ready: triggerVariant.ready, - auth: await resolveApiConnection(apiConnection), }; } } diff --git a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts index 1361f61d9..ff505dd2b 100644 --- a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts +++ b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts @@ -34,41 +34,41 @@ export class DeliverHttpSourceRequestService { return; } - // TODO: auth + // TODO: deliver the raw source event through the new internal job system // const auth = await getConnectionAuth(httpSourceRequest.source.connection); - const clientApi = new ClientApi( - httpSourceRequest.environment.apiKey, - httpSourceRequest.endpoint.url - ); + // const clientApi = new ClientApi( + // httpSourceRequest.environment.apiKey, + // httpSourceRequest.endpoint.url + // ); - const { response, events } = await clientApi.deliverHttpSourceRequest({ - key: httpSourceRequest.source.key, - secret: httpSourceRequest.source.secret ?? undefined, - auth: undefined, // TODO: auth - request: { - url: httpSourceRequest.url, - method: httpSourceRequest.method, - headers: httpSourceRequest.headers as Record, - rawBody: httpSourceRequest.body, - }, - }); + // const { response, events } = await clientApi.deliverHttpSourceRequest({ + // key: httpSourceRequest.source.key, + // secret: httpSourceRequest.source.secret ?? undefined, + // auth: undefined, // TODO: auth + // request: { + // url: httpSourceRequest.url, + // method: httpSourceRequest.method, + // headers: httpSourceRequest.headers as Record, + // rawBody: httpSourceRequest.body, + // }, + // }); - await this.#prismaClient.httpSourceRequestDelivery.update({ - where: { - id, - }, - data: { - deliveredAt: new Date(), - }, - }); + // await this.#prismaClient.httpSourceRequestDelivery.update({ + // where: { + // id, + // }, + // data: { + // deliveredAt: new Date(), + // }, + // }); - const ingestService = new IngestSendEvent(); + // const ingestService = new IngestSendEvent(); - for (const event of events) { - await ingestService.call(httpSourceRequest.environment, event); - } + // for (const event of events) { + // await ingestService.call(httpSourceRequest.environment, event); + // } - return response; + // return response; } } diff --git a/apps/webapp/package.json b/apps/webapp/package.json index ec1a6b676..3698f551e 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -90,6 +90,7 @@ "date-fns": "2.0.0-alpha.7 || >=2.0.0", "emails": "workspace:*", "express": "^4.18.1", + "fast-redact": "^3.1.2", "graphile-worker": "^0.13.0", "humanize-duration": "^3.27.3", "intl-parse-accept-language": "^1.0.0", diff --git a/apps/webapp/prisma/migrations/20230504090710_add_shadow_to_job/migration.sql b/apps/webapp/prisma/migrations/20230504090710_add_shadow_to_job/migration.sql new file mode 100644 index 000000000..00db79291 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230504090710_add_shadow_to_job/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Job" ADD COLUMN "shadow" BOOLEAN NOT NULL DEFAULT false; diff --git a/apps/webapp/prisma/migrations/20230504152611_rename_shadow_to_internal/migration.sql b/apps/webapp/prisma/migrations/20230504152611_rename_shadow_to_internal/migration.sql new file mode 100644 index 000000000..8aee61943 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230504152611_rename_shadow_to_internal/migration.sql @@ -0,0 +1,9 @@ +/* + Warnings: + + - You are about to drop the column `shadow` on the `Job` table. All the data in the column will be lost. + +*/ +-- AlterTable +ALTER TABLE "Job" DROP COLUMN "shadow", +ADD COLUMN "internal" BOOLEAN NOT NULL DEFAULT false; diff --git a/apps/webapp/prisma/migrations/20230504200916_add_redact_to_tasks/migration.sql b/apps/webapp/prisma/migrations/20230504200916_add_redact_to_tasks/migration.sql new file mode 100644 index 000000000..af13adb90 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230504200916_add_redact_to_tasks/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Task" ADD COLUMN "redact" JSONB; diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index c26f065c2..3b392037d 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -967,9 +967,10 @@ model Endpoint { } model Job { - id String @id @default(cuid()) - slug String - title String + id String @id @default(cuid()) + slug String + title String + internal Boolean @default(false) organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade) organizationId String @@ -1213,6 +1214,7 @@ model Task { params Json? output Json? error String? + redact Json? startedAt DateTime? completedAt DateTime? diff --git a/examples/nextjs-example/src/pages/api/trigger.ts b/examples/nextjs-example/src/pages/api/trigger.ts index bf3c72db6..3ebe3d1fb 100644 --- a/examples/nextjs-example/src/pages/api/trigger.ts +++ b/examples/nextjs-example/src/pages/api/trigger.ts @@ -25,158 +25,114 @@ new Job({ version: "0.1.1", logLevel: "debug", connections: { - gh, sl, }, trigger: gh.triggers.onIssueOpened({ repo: "ericallam/basic-starter-100k", }), run: async (event, io, ctx) => { - const slackMessage = await io.sl.postMessage("Slack 📝", { + await io.sl.postMessage("Slack 📝", { text: `New Issue opened: ${event.issue.html_url}`, channel: "C04GWUTDC3W", }); - - await io.runTask( - "Comment on Issue with a reaction", - { name: "Parent Task" }, - async (task) => { - const comment = await io.runTask( - "Comment on issue", - { name: "Comment on Issue" }, - async (t) => { - return io.gh.client.rest.issues - .createComment({ - owner: event.repository.owner.login, - repo: event.repository.name, - issue_number: event.issue.number, - body: "Hello from Trigger!", - }) - .then((res) => res.data); - } - ); - - await io.runTask( - "Add react to comment", - { name: "Add reaction to comment" }, - async (t) => { - return io.gh.client.rest.reactions.createForIssueComment({ - owner: event.repository.owner.login, - repo: event.repository.name, - comment_id: comment.id, - content: "rocket", - }); - } - ); - - return comment; - } - ); - - await io.gh.createIssueCommentWithReaction("📝", { - repo: `${event.repository.owner.login}/${event.repository.name}`, - issueNumber: event.issue.number, - body: "Hello from Trigger!", - reaction: "rocket", - }); }, -}).registerWith(client); +}).attachTo(client); -const notifySlackONNewCommentsJob = new Job({ - id: "notify-slack-on-new-comments", - name: "Notify Slack on new GitHub comments", - version: "0.1.1", - logLevel: "debug", - connections: { - gh, - sl, - }, - trigger: gh.triggers.onIssueComment({ - repo: "ericallam/basic-starter-100k", - }), - run: async (event, io, ctx) => { - await io.sl.postMessage("Slack 📝", { - text: `New Comment on Issue: ${event.comment.html_url}`, - channel: "C04GWUTDC3W", - }); - }, -}) - .registerWith(client) - .addTriggerVariant( - "ericallam/hello-world", - gh.triggers.onIssueComment({ - repo: "ericallam/hello-world", - }) - ); +// const notifySlackONNewCommentsJob = new Job({ +// id: "notify-slack-on-new-comments", +// name: "Notify Slack on new GitHub comments", +// version: "0.1.1", +// logLevel: "debug", +// connections: { +// gh, +// sl, +// }, +// trigger: gh.triggers.onIssueComment({ +// repo: "ericallam/basic-starter-100k", +// }), +// run: async (event, io, ctx) => { +// await io.sl.postMessage("Slack 📝", { +// text: `New Comment on Issue: ${event.comment.html_url}`, +// channel: "C04GWUTDC3W", +// }); +// }, +// }) +// .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); +// 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", - logLevel: "debug", - trigger: customEvent({ - name: "my-custom-event", - source: "my-source", - filter: { - foo: ["bar"], - }, - schema: z.object({ - foo: z.string(), - }), - }), - run: async (event, io, ctx) => { - const payload = await io.on( - "Wait for another event", - customEvent({ - name: "my-custom-event-2", - source: "my-source", - schema: z.object({ - foo: z.string(), - }), - }) - ); +// const waitForEventInJob = new Job({ +// id: "wait-for-event-in-job", +// name: "Wait for event in job", +// version: "0.1.1", +// logLevel: "debug", +// trigger: customEvent({ +// name: "my-custom-event", +// source: "my-source", +// filter: { +// foo: ["bar"], +// }, +// schema: z.object({ +// foo: z.string(), +// }), +// }), +// run: async (event, io, ctx) => { +// const payload = await io.on( +// "Wait for another event", +// customEvent({ +// name: "my-custom-event-2", +// source: "my-source", +// schema: z.object({ +// foo: z.string(), +// }), +// }) +// ); - return payload; - }, -}).registerWith(client); +// return payload; +// }, +// }).registerWith(client); -client.addTriggerVariant( - waitForEventInJob, - "custom-event-3", - customEvent({ - name: "my-custom-event-3", - source: "my-source", - schema: z.object({ - foo: z.string(), - }), - }) -); +// 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, diff --git a/integrations/github/src/index.ts b/integrations/github/src/index.ts index 5fd779457..eefae662b 100644 --- a/integrations/github/src/index.ts +++ b/integrations/github/src/index.ts @@ -3,26 +3,16 @@ import { IssuesEvent, IssuesOpenedEvent, } from "@octokit/webhooks-types"; -import type { Connection, EventFilter } from "@trigger.dev/sdk"; -import { ExternalSourceEventTrigger, Trigger } from "@trigger.dev/sdk/triggers"; +import { + Connection, + EventFilter, + ExternalSourceEventTrigger, +} from "@trigger.dev/sdk"; import { Octokit } from "octokit"; import { clientFactory } from "./clientFactory"; import { metadata } from "./metadata"; import { repositoryWebhookSource } from "./sources"; -import { - createIssue, - createIssueComment, - createIssueCommentWithReaction, - getRepo, -} from "./tasks"; -import { ClientOptions } from "./types"; - -const tasks = { - createIssue, - createIssueComment, - getRepo, - createIssueCommentWithReaction, -}; +import { tasks } from "./tasks"; export type GitHubConnectionOptions = | { @@ -33,6 +23,17 @@ export type GitHubConnectionOptions = }; export const github = (options: GitHubConnectionOptions) => { + const connection = createConnectionFromOptions(options); + + return { + ...connection, + triggers: createTriggers(connection), + }; +}; + +function createConnectionFromOptions( + options: GitHubConnectionOptions +): Connection { if ("token" in options) { const client = new Octokit({ auth: options.token, @@ -43,8 +44,7 @@ export const github = (options: GitHubConnectionOptions) => { tasks, usesLocalAuth: true, client, - triggers: createTriggers({ usesLocalAuth: true, octokit: client }), - } satisfies Connection; + }; } return { @@ -53,26 +53,20 @@ export const github = (options: GitHubConnectionOptions) => { tasks, usesLocalAuth: false, clientFactory, - triggers: createTriggers( - { usesLocalAuth: false, clientFactory }, - options.id - ), - } satisfies Connection; -}; -0; -function createTriggers(client: ClientOptions, id?: string) { + }; +} + +function createTriggers(connection: Connection) { return { onIssue: buildRepoWebhookTrigger( "On Issue", "issues", - client, - id + connection ), onIssueOpened: buildRepoWebhookTrigger( "On Issue Opened", "issues", - client, - id, + connection, { action: ["opened"], } @@ -80,21 +74,19 @@ function createTriggers(client: ClientOptions, id?: string) { onIssueComment: buildRepoWebhookTrigger( "On Issue Comment", "issue_comment", - client, - id + connection ), }; } -function buildRepoWebhookTrigger( +function buildRepoWebhookTrigger( title: string, event: string, - client: ClientOptions, - id?: string, + connection: Connection, filter?: EventFilter -): (params: { repo: string }) => ExternalSourceEventTrigger { +) { return (params: { repo: string }) => - new ExternalSourceEventTrigger({ + new ExternalSourceEventTrigger({ title, elements: [ { @@ -111,8 +103,8 @@ function buildRepoWebhookTrigger( repo: params.repo, events: [event], }, - client, - id + connection, + (payload) => payload as TEvent ), eventRule: { event, diff --git a/integrations/github/src/sources.ts b/integrations/github/src/sources.ts index 34738c51f..e9d3b1e98 100644 --- a/integrations/github/src/sources.ts +++ b/integrations/github/src/sources.ts @@ -1,7 +1,7 @@ import { Webhooks } from "@octokit/webhooks"; -import { ExternalSource } from "@trigger.dev/sdk/externalSource"; -import { metadata } from "./metadata"; -import { ClientOptions } from "./types"; +import { Connection, ExternalSource } from "@trigger.dev/sdk"; +import { Octokit } from "octokit"; +import { tasks } from "./tasks"; type WebhookData = { id: number; @@ -21,37 +21,26 @@ function webhookData(data: any): data is WebhookData { ); } -export function repositoryWebhookSource( +export function repositoryWebhookSource( params: { repo: string; events: string[]; secret?: string; }, - client: ClientOptions, - id?: string + connection: Connection, + parsePayload: (payload: any) => TEventType ) { // Create a stable key for this source so we only register it once const key = `github.repo.${params.repo}.webhook`; - return new ExternalSource("http", metadata, { - id, - usesLocalAuth: client.usesLocalAuth, - key, - register: async (triggerClient, auth) => { - if (!auth) { - throw new Error("No auth provided"); - } - - const octokit = client.usesLocalAuth - ? client.octokit - : client.clientFactory(auth); - - const httpSource = await triggerClient.registerHttpSource({ + return new ExternalSource("http", { + parsePayload, + connection, + register: async (io, ctx) => { + const httpSource = await io.registerHttpSource("register-http-source", { key, }); - const [owner, repo] = params.repo.split("/"); - if ( httpSource.active && webhookData(httpSource.data) && @@ -70,20 +59,19 @@ export function repositoryWebhookSource( if (missingEvents.length > 0) { // We need to update the webhook to add the new events and then return - const { data: newWebhookData } = - await octokit.rest.repos.updateWebhook({ - owner, - repo, - hook_id: existingData.id, - config: { - content_type: "json", - url: httpSource.url, - secret: httpSource.secret, - }, - add_events: missingEvents, - }); + const newWebhookData = await io.client.updateWebhook( + "update-webhook", + { + repo: params.repo, + hookId: existingData.id, + url: httpSource.url, + secret: httpSource.secret, + addEvents: missingEvents, + } + ); - await triggerClient.updateHttpSource(httpSource.id, { + await io.updateHttpSource("update-http-source", { + id: httpSource.id, data: newWebhookData, }); } @@ -91,9 +79,8 @@ export function repositoryWebhookSource( return; } - const { data: webhooks } = await octokit.rest.repos.listWebhooks({ - owner, - repo, + const webhooks = await io.client.listWebhooks("list-webhooks", { + repo: params.repo, }); const existingWebhook = webhooks.find( @@ -103,18 +90,15 @@ export function repositoryWebhookSource( const secret = params.secret || Math.random().toString(36).slice(2); if (existingWebhook && existingWebhook.active) { - await octokit.rest.repos.updateWebhook({ - owner, - repo, - hook_id: existingWebhook.id, - config: { - content_type: "json", - url: httpSource.url, - secret, - }, + await io.client.updateWebhook("update-webhook", { + repo: params.repo, + hookId: existingWebhook.id, + url: httpSource.url, + secret, }); - await triggerClient.updateHttpSource(httpSource.id, { + await io.updateHttpSource("update-http-source", { + id: httpSource.id, secret, data: existingWebhook, active: true, @@ -123,42 +107,31 @@ export function repositoryWebhookSource( return; } - // Generate secret - - if (!owner || !repo) { - throw new Error( - 'Invalid repo, should be in format "owner/repo". For example: "triggerdotdev/trigger.dev"' - ); - } - - const { data: webhook } = await octokit.rest.repos.createWebhook({ - owner, - repo, + const webhook = await io.client.createWebhook("create-webhook", { + repo: params.repo, events: params.events, - config: { - url: httpSource.url, - content_type: "json", - secret, - }, + url: httpSource.url, + secret, }); - await triggerClient.updateHttpSource(httpSource.id, { + await io.updateHttpSource("update-http-source", { + id: httpSource.id, secret, data: webhook, active: true, }); }, - handler: async (client, source, auth) => { - const deliveryId = source.request.headers["x-github-delivery"]; - const hookId = source.request.headers["x-github-hook-id"]; - const signature = source.request.headers["x-hub-signature-256"]; + handler: async (event, io, ctx) => { + const deliveryId = event.request.headers["x-github-delivery"]; + const hookId = event.request.headers["x-github-hook-id"]; + const signature = event.request.headers["x-hub-signature-256"]; - if (source.secret && signature) { + if (event.secret && signature) { const githubWebhooks = new Webhooks({ - secret: source.secret, + secret: event.secret, }); - if (!githubWebhooks.verify(source.request.body, signature)) { + if (!githubWebhooks.verify(event.request.body, signature)) { return { events: [], response: { @@ -171,9 +144,9 @@ export function repositoryWebhookSource( } } - const name = source.request.headers["x-github-event"]; + const name = event.request.headers["x-github-event"]; - const context = omit(source.request.headers, [ + const context = omit(event.request.headers, [ "x-github-event", "x-github-delivery", "x-hub-signature-256", @@ -185,7 +158,7 @@ export function repositoryWebhookSource( "x-forwarded-proto", ]); - const payload = parseBody(source.request.body); + const payload = parseBody(event.request.body); return { events: [ diff --git a/integrations/github/src/tasks.ts b/integrations/github/src/tasks.ts index 9a3800ead..9177019f3 100644 --- a/integrations/github/src/tasks.ts +++ b/integrations/github/src/tasks.ts @@ -238,3 +238,135 @@ export const createIssueCommentWithReaction = authenticatedTask({ }; }, }); + +export const updateWebhook = authenticatedTask({ + run: async ( + params: { + repo: string; + hookId: number; + url: string; + secret: string; + addEvents?: string[]; + }, + client: InstanceType, + task + ) => { + const [owner, repo] = params.repo.split("/"); + + return client.rest.repos + .updateWebhook({ + owner, + repo, + hook_id: params.hookId, + config: { + content_type: "json", + url: params.url, + secret: params.secret, + }, + add_events: params.addEvents, + }) + .then((response) => response.data); + }, + init: (params) => { + return { + name: "Update Webhook", + params, + elements: [ + { + label: "Repo", + text: params.repo, + }, + { + label: "Hook ID", + text: String(params.hookId), + }, + ], + }; + }, +}); + +export const createWebhook = authenticatedTask({ + run: async ( + params: { + repo: string; + url: string; + secret: string; + events: string[]; + }, + client: InstanceType, + task + ) => { + const [owner, repo] = params.repo.split("/"); + + return client.rest.repos + .createWebhook({ + owner, + repo, + config: { + content_type: "json", + url: params.url, + secret: params.secret, + }, + events: params.events, + }) + .then((response) => response.data); + }, + init: (params) => { + return { + name: "Create Webhook", + params, + elements: [ + { + label: "Repo", + text: params.repo, + }, + { + label: "Events", + text: params.events.join(", "), + }, + ], + }; + }, +}); + +export const listWebhooks = authenticatedTask({ + run: async ( + params: { + repo: string; + }, + client: InstanceType, + task + ) => { + const [owner, repo] = params.repo.split("/"); + + return client.rest.repos + .listWebhooks({ + owner, + repo, + }) + .then((response) => response.data); + }, + init: (params) => { + return { + name: "List Webhooks", + params, + elements: [ + { + label: "Repo", + text: params.repo, + }, + ], + }; + }, +}); + +export const tasks = { + createIssue, + createIssueComment, + getRepo, + createIssueCommentWithReaction, + addIssueCommentReaction, + updateWebhook, + createWebhook, + listWebhooks, +}; diff --git a/integrations/github/src/types.ts b/integrations/github/src/types.ts deleted file mode 100644 index a2fa1663a..000000000 --- a/integrations/github/src/types.ts +++ /dev/null @@ -1,12 +0,0 @@ -import { ClientFactory } from "@trigger.dev/sdk"; -import { Octokit } from "octokit"; - -export type ClientOptions = - | { - usesLocalAuth: true; - octokit: Octokit; - } - | { - usesLocalAuth: false; - clientFactory: ClientFactory; - }; diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index 252ecd18e..a1851cf73 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -76,7 +76,8 @@ export const JobSchema = z.object({ name: z.string(), version: z.string(), trigger: TriggerMetadataSchema, - connections: z.array(ConnectionConfigSchema), + connections: z.record(ConnectionConfigSchema), + internal: z.boolean().default(false), }); export type JobMetadata = z.infer; @@ -140,7 +141,7 @@ export const DeliverEventResponseSchema = z.object({ export type DeliverEventResponse = z.infer; -export const ExecuteJobBodySchema = z.object({ +export const RunJobBodySchema = z.object({ event: ApiEventLogSchema, job: z.object({ id: z.string(), @@ -158,16 +159,16 @@ export const ExecuteJobBodySchema = z.object({ connections: z.record(ConnectionAuthSchema).optional(), }); -export type ExecuteJobBody = z.infer; +export type RunJobBody = z.infer; -export const ExecuteJobResponseSchema = z.object({ +export const RunJobResponseSchema = z.object({ executionId: z.string(), completed: z.boolean(), output: DeserializedJsonSchema.optional(), task: TaskSchema.optional(), }); -export type ExecuteJobResponse = z.infer; +export type RunJobResponse = z.infer; export const CreateRunBodySchema = z.object({ client: z.string(), @@ -234,6 +235,10 @@ export type ClientTask = z.infer; export type ServerTask = z.infer; export type CachedTask = z.infer; +export const RedactSchema = z.object({ + paths: z.array(z.string()), +}); + export const RunTaskOptionsSchema = z.object({ name: z.string(), icon: z.string().optional(), @@ -244,6 +249,7 @@ export const RunTaskOptionsSchema = z.object({ elements: z.array(DisplayElementSchema).optional(), params: SerializableJsonSchema.optional(), trigger: TriggerMetadataSchema.optional(), + redact: RedactSchema.optional(), }); export type RunTaskOptions = z.input; diff --git a/packages/internal/src/schemas/connections.ts b/packages/internal/src/schemas/connections.ts index 81e92ba3e..33ec6b7f4 100644 --- a/packages/internal/src/schemas/connections.ts +++ b/packages/internal/src/schemas/connections.ts @@ -18,7 +18,6 @@ export const ConnectionAuthSchema = z.object({ export type ConnectionAuth = z.infer; const CommonConnectionConfigSchema = z.object({ - key: z.string().optional(), metadata: ConnectionMetadataSchema, }); diff --git a/packages/internal/src/schemas/triggers.ts b/packages/internal/src/schemas/triggers.ts index a5737f836..0024b6e60 100644 --- a/packages/internal/src/schemas/triggers.ts +++ b/packages/internal/src/schemas/triggers.ts @@ -1,7 +1,6 @@ import { z } from "zod"; import { EventRuleSchema } from "./eventFilter"; import { DeserializedJsonSchema } from "./json"; -import { ConnectionAuthSchema, ConnectionConfigSchema } from "./connections"; export const TriggerMetadataSchema = z.object({ title: z.string(), @@ -14,8 +13,6 @@ export const TriggerMetadataSchema = z.object({ ), eventRule: EventRuleSchema, schema: DeserializedJsonSchema.optional(), - connection: ConnectionConfigSchema.optional(), - supportsPreparation: z.boolean(), }); export type TriggerMetadata = z.infer; diff --git a/packages/trigger-sdk/src/connections.ts b/packages/trigger-sdk/src/connections.ts index 13b057da7..e8c20ea02 100644 --- a/packages/trigger-sdk/src/connections.ts +++ b/packages/trigger-sdk/src/connections.ts @@ -6,7 +6,7 @@ import { } from "@trigger.dev/internal"; import { IO } from "./io"; import { TriggerClient } from "./triggerClient"; -import { Trigger } from "./triggers"; +import { Trigger } from "./types"; export type ClientFactory = (auth: ConnectionAuth) => TClientType; diff --git a/packages/trigger-sdk/src/externalSource.ts b/packages/trigger-sdk/src/externalSource.ts deleted file mode 100644 index 2759d6525..000000000 --- a/packages/trigger-sdk/src/externalSource.ts +++ /dev/null @@ -1,123 +0,0 @@ -import { - ApiEventLog, - ConnectionAuth, - ConnectionMetadata, - DisplayElement, - NormalizedRequest, - NormalizedResponse, - SendEvent, -} from "@trigger.dev/internal"; -import { TriggerClient } from "./triggerClient"; - -export type HttpSourceEvent = { - request: NormalizedRequest; - secret?: string; -}; - -export type SmtpSourceEvent = { - from: string; - to: string; - subject: string; - body: string; -}; - -export type SqsSourceEvent = { - body: string; -}; - -type ExternalSourceChannelMap = { - http: { - event: HttpSourceEvent; - }; - smtp: { - event: SmtpSourceEvent; - }; - sqs: { - event: SqsSourceEvent; - }; -}; - -export type ChannelNames = keyof ExternalSourceChannelMap; - -export type HandlerFunction< - TChannel extends ChannelNames, - TSourceEvent extends ExternalSourceChannelMap[TChannel]["event"] -> = ( - triggerClient: TriggerClient, - event: TSourceEvent, - auth?: ConnectionAuth -) => Promise<{ response: NormalizedResponse; events: SendEvent[] }>; - -export type ExternalSourceOptions = { - key: string; - usesLocalAuth: boolean; - id?: string; - register: ( - triggerClient: TriggerClient, - auth?: ConnectionAuth - ) => Promise; - handler: HandlerFunction< - TChannel, - ExternalSourceChannelMap[TChannel]["event"] - >; - eventElements?: (event: ApiEventLog) => DisplayElement[]; -}; - -export interface AnyExternalSource { - key: string; - usesLocalAuth: boolean; - id?: string; - connection: ConnectionMetadata; - channel: ChannelNames; - handler: ( - triggerClient: TriggerClient, - event: any, - auth?: ConnectionAuth - ) => Promise<{ response: NormalizedResponse; events: SendEvent[] }>; - eventElements: (event: ApiEventLog) => DisplayElement[]; - prepare: (client: TriggerClient, auth?: ConnectionAuth) => Promise; -} - -export class ExternalSource - implements AnyExternalSource -{ - channel: TChannel; - connection: ConnectionMetadata; - - constructor( - channel: TChannel, - connection: ConnectionMetadata, - private options: ExternalSourceOptions - ) { - this.channel = channel; - this.connection = connection; - } - - get id() { - return this.options.id; - } - - get key() { - return this.options.key; - } - - get usesLocalAuth() { - return this.options.usesLocalAuth; - } - - async prepare(client: TriggerClient, auth?: ConnectionAuth) { - return this.options.register(client, auth); - } - - async handler( - triggerClient: TriggerClient, - event: ExternalSourceChannelMap[TChannel]["event"], - auth?: ConnectionAuth - ) { - return this.options.handler(triggerClient, event, auth); - } - - eventElements(event: ApiEventLog) { - return this.options.eventElements?.(event) ?? []; - } -} diff --git a/packages/trigger-sdk/src/index.ts b/packages/trigger-sdk/src/index.ts index 0d5c5bc76..12ceca21e 100644 --- a/packages/trigger-sdk/src/index.ts +++ b/packages/trigger-sdk/src/index.ts @@ -1,9 +1,10 @@ -export * from "./triggers"; export * from "./job"; export * from "./triggerClient"; export * from "./connections"; -export * from "./externalSource"; +export * from "./triggers/customEvent"; +export * from "./triggers/externalSource"; export * from "./io"; +export * from "./types"; import { SecureString } from "./types"; diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts index 7b1dfa175..0eb705f9c 100644 --- a/packages/trigger-sdk/src/io.ts +++ b/packages/trigger-sdk/src/io.ts @@ -1,18 +1,19 @@ import { CachedTask, - RunTaskOptions, - Logger, LogLevel, + Logger, + RegisterHttpEventSourceBody, + RunTaskOptions, SerializableJson, ServerTask, - DeserializedJson, + UpdateHttpEventSourceBody, } from "@trigger.dev/internal"; 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"; +import { Trigger } from "./types"; export class ResumeWithTask { constructor(public task: ServerTask) {} @@ -54,7 +55,64 @@ export class IO { this.#taskStorage = new AsyncLocalStorage(); } - // TODO: finish implementing this (needs to support registering and preparing) + async registerHttpSource( + key: string | any[], + options: RegisterHttpEventSourceBody + ) { + return this.runTask( + key, + { + name: "Register HTTP Source", + description: `Register HTTP Source ${options.key}`, + elements: [ + { + label: "Key", + text: options.key, + }, + ], + redact: { + paths: ["secret"], + }, + }, + async (task) => { + return await this.#apiClient.registerHttpSource( + this.#client.name, + options + ); + } + ); + } + + async updateHttpSource( + key: string | any[], + options: { id: string } & UpdateHttpEventSourceBody + ) { + return this.runTask( + key, + { + name: "Update HTTP Source", + description: `Update HTTP Source ${options.id}`, + elements: [ + { + label: "id", + text: options.id, + }, + ], + redact: { + paths: ["secret"], + }, + }, + async (task) => { + return await this.#apiClient.updateHttpSource( + this.#client.name, + options.id, + options + ); + } + ); + } + + // TODO: use internal job system for this async on( key: string | any[], trigger: Trigger @@ -121,7 +179,7 @@ export class IO { elements: metadata.elements, }, async (task) => { - // todo: trigger.prepare should take the io as an argument and everything inside there should happen within subtasks + // 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 @@ -131,7 +189,8 @@ export class IO { // 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); + // We need to make a new "child IO" here that is used for preparing the trigger which does not have access to other connections or auth in the context + // return await trigger.prepare(this.#client, subResponse1.auth); } ); diff --git a/packages/trigger-sdk/src/job.ts b/packages/trigger-sdk/src/job.ts index 0c1f2b4b2..757af20b3 100644 --- a/packages/trigger-sdk/src/job.ts +++ b/packages/trigger-sdk/src/job.ts @@ -1,13 +1,7 @@ -import { - ConnectionAuth, - ConnectionConfig, - JobMetadata, - LogLevel, -} from "@trigger.dev/internal"; +import { ConnectionConfig, JobMetadata, LogLevel } from "@trigger.dev/internal"; import { Connection, IOWithConnections } from "./connections"; import { TriggerClient } from "./triggerClient"; -import { Trigger, TriggerEventType } from "./triggers"; -import type { TriggerContext } from "./types"; +import type { TriggerContext, Trigger, TriggerEventType } from "./types"; export type JobOptions< TTrigger extends Trigger, @@ -56,28 +50,31 @@ export class Job< return this.options.version; } - get connections(): Array { - return Object.keys(this.options.connections ?? {}).map((key) => { - const connection = this.options.connections![key]; + get connections(): Record { + return Object.keys(this.options.connections ?? {}).reduce( + (acc: Record, key) => { + const connection = this.options.connections![key]; - if (connection.usesLocalAuth) { - return { - auth: "local", - key, - metadata: connection.metadata, - }; - } else { - return { - auth: "hosted", - key, - metadata: connection.metadata, - id: connection.id!, - }; - } - }); + if (connection.usesLocalAuth) { + acc[key] = { + auth: "local", + metadata: connection.metadata, + }; + } else { + acc[key] = { + auth: "hosted", + metadata: connection.metadata, + id: connection.id!, + }; + } + + return acc; + }, + {} + ); } - registerWith(client: TriggerClient) { + attachTo(client: TriggerClient) { if (this.client) { throw new Error( `Job "${this.id}" has already been registered with a client.` @@ -86,40 +83,37 @@ export class Job< this.client = client; - client.register(this); + client.attach(this); return this; } - addTriggerVariant(id: string, trigger: TTrigger) { + attachVariant(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); + this.client.attachVariant(this, id, trigger); return this; } toJSON(): JobMetadata { + // @ts-ignore + const internal = this.options.__internal as JobMetadata["internal"]; + return { id: this.id, name: this.name, version: this.version, trigger: this.trigger.toJSON(), connections: this.connections, + internal, }; } - async prepareForExecution( - client: TriggerClient, - connections: Record - ) { - await this.trigger.prepare(client, connections.__trigger); - } - // Make sure the id is valid (must only contain alphanumeric characters and dashes) // Make sure the version is valid (must be a valid semver version) #validate() { diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 144da66f7..96205b72e 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -1,19 +1,13 @@ import { - ApiEventLog, - ConnectionAuth, ErrorWithMessage, ErrorWithStackSchema, - ExecuteJobBody, - ExecuteJobBodySchema, - HttpSourceRequest, - HttpSourceRequestHeadersSchema, - Logger, LogLevel, + Logger, NormalizedRequest, NormalizedResponse, - PrepareJobTriggerBodySchema, RegisterHttpEventSourceBody, - SendEvent, + RunJobBody, + RunJobBodySchema, UpdateHttpEventSourceBody, } from "@trigger.dev/internal"; import { ApiClient } from "./apiClient"; @@ -22,12 +16,10 @@ import { Connection, IOWithConnections, } from "./connections"; -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"; +import type { Trigger, TriggerContext } from "./types"; export type TriggerClientOptions = { apiKey?: string; @@ -48,7 +40,6 @@ export class TriggerClient { string, Array<{ trigger: Trigger; id: string }> > = {}; - #registeredSources = new Map(); #client: ApiClient; #logger: Logger; name: string; @@ -145,7 +136,7 @@ export class TriggerClient { }; } case "EXECUTE_JOB": { - const execution = ExecuteJobBodySchema.safeParse(request.body); + const execution = RunJobBodySchema.safeParse(request.body); if (!execution.success) { return { @@ -186,78 +177,6 @@ export class TriggerClient { }, }; } - case "PREPARE_JOB_TRIGGER": { - const payload = PrepareJobTriggerBodySchema.safeParse(request.body); - - if (!payload.success) { - return { - status: 400, - body: { - message: "Invalid payload", - }, - }; - } - - const registeredJob = this.#registeredJobs[payload.data.id]; - - if (!registeredJob) { - return { - status: 404, - body: { - message: "Job not found", - }, - }; - } - - await this.#prepareJobTrigger(registeredJob, payload.data); - - return { - status: 200, - body: { - ok: true, - }, - }; - } - case "DELIVER_HTTP_SOURCE_REQUEST": { - const headers = HttpSourceRequestHeadersSchema.safeParse( - request.headers - ); - - if (!headers.success) { - return { - status: 400, - body: { - message: "Invalid headers", - }, - }; - } - - const sourceRequest = { - url: headers.data["x-trigger-url"], - method: headers.data["x-trigger-method"], - headers: headers.data["x-trigger-headers"], - body: request.body, - }; - - const auth = headers.data["x-trigger-auth"]; - const key = headers.data["x-trigger-key"]; - const secret = headers.data["x-trigger-secret"]; - - const { response, events } = await this.#handleHttpSourceRequest( - key, - sourceRequest, - secret, - auth - ); - - return { - status: 200, - body: { - events, - response, - }, - }; - } } } @@ -269,19 +188,13 @@ export class TriggerClient { }; } - register(thing: AnyExternalSource): void; - register(thing: Job, any>): void; - register(thing: Job, any> | AnyExternalSource): void { - if (thing instanceof Job) { - this.#registeredJobs[thing.id] = thing; + attach(job: Job, any>): void { + this.#registeredJobs[job.id] = job; - thing.trigger.registerWith(this); - } else { - this.#registeredSources.set(thing.key, thing); - } + job.trigger.attach(this, job); } - addTriggerVariant>( + attachVariant>( job: Job, id: string, trigger: TTrigger @@ -290,7 +203,7 @@ export class TriggerClient { jobTriggerVariants.push({ trigger, id }); this.#registeredTriggerVariants[job.id] = jobTriggerVariants; - trigger.registerWith(this); + trigger.attach(this, job, id); } authorized(apiKey: string) { @@ -323,60 +236,7 @@ export class TriggerClient { return await this.#client.updateHttpSource(this.name, id, source); } - async #prepareJobTrigger( - job: Job, any>, - preparationData: { - id: string; - version: string; - connection?: ConnectionAuth; - variantId?: string; - } - ): Promise { - this.#logger.debug("preparing job trigger", { job: job.toJSON() }); - - if (job.version !== preparationData.version) { - return; - } - - 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( - key: string, - sourceRequest: HttpSourceRequest, - secret?: string, - auth?: ConnectionAuth - ): Promise<{ response: NormalizedResponse; events: SendEvent[] }> { - const source = this.#registeredSources.get(key); - - if (!source) { - return { - response: { - status: 200, - body: { - ok: true, - }, - }, - events: [], - }; - } - - return await source.handler(this, { request: sourceRequest, secret }, auth); - } - - async #executeJob(execution: ExecuteJobBody, job: Job, any>) { + async #executeJob(execution: RunJobBody, job: Job, any>) { this.#logger.debug("executing job", { execution, job: job.toJSON() }); const abortController = new AbortController(); @@ -393,7 +253,7 @@ export class TriggerClient { try { const output = await job.options.run( - job.trigger.parsePayload(execution.event.payload ?? {}), // todo: actually parse the payload through the trigger + job.trigger.parsePayload(execution.event.payload ?? {}), ioWithConnections, this.#createJobContext(execution, io, abortController.signal) ); @@ -427,7 +287,7 @@ export class TriggerClient { TConnections extends Record> >( io: IO, - execution: ExecuteJobBody, + run: RunJobBody, job: Job, TConnections> ): IOWithConnections { const jobConnections = job.options.connections; @@ -436,11 +296,11 @@ export class TriggerClient { return io as IOWithConnections; } - const executionConnections = execution.connections ?? {}; + const runConnections = run.connections ?? {}; const connections = Object.entries(jobConnections).reduce( (acc, [key, jobConnection]) => { - const connection = executionConnections[key]; + const connection = runConnections[key]; const client = jobConnection.client ?? jobConnection.clientFactory?.(connection); @@ -496,7 +356,7 @@ export class TriggerClient { } #createJobContext( - execution: ExecuteJobBody, + execution: RunJobBody, io: IO, signal: AbortSignal ): TriggerContext { diff --git a/packages/trigger-sdk/src/triggers.ts b/packages/trigger-sdk/src/triggers.ts deleted file mode 100644 index e03f2bb57..000000000 --- a/packages/trigger-sdk/src/triggers.ts +++ /dev/null @@ -1,130 +0,0 @@ -import type { - ApiEventLog, - ConnectionAuth, - ConnectionConfig, - EventFilter, - EventRule, - TriggerMetadata, -} from "@trigger.dev/internal"; -import { DisplayElement } from "@trigger.dev/internal"; -import { z } from "zod"; -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; - prepare(client: TriggerClient, auth?: ConnectionAuth): Promise; - parsePayload(payload: unknown): TEventType; -} - -export type CustomEventTriggerOptions = { - name: string; - source?: string; - schema?: TSchema; - filter?: EventFilter; -}; - -export class CustomEventTrigger - implements Trigger> -{ - #options: CustomEventTriggerOptions; - - constructor(options: CustomEventTriggerOptions) { - this.#options = options; - } - - eventElements(event: ApiEventLog): DisplayElement[] { - return []; - } - - toJSON(): TriggerMetadata { - return { - title: "Custom Event", - elements: [{ label: "on", text: this.#options.name }], - schema: this.#options.schema - ? zodToJsonSchema(this.#options.schema) - : undefined, - eventRule: { - event: this.#options.name, - source: this.#options.source ?? "trigger.dev", - payload: this.#options.filter ?? {}, - }, - supportsPreparation: false, - }; - } - - parsePayload(payload: unknown): z.infer { - if (!this.#options.schema) { - return payload; - } - - return this.#options.schema.parse(payload); - } - - registerWith(client: TriggerClient) {} - async prepare(client: TriggerClient, auth?: ConnectionAuth) {} -} - -export function customEvent( - options: CustomEventTriggerOptions -): Trigger> { - return new CustomEventTrigger(options); -} - -export type ExteralSourceEventTriggerOptions = { - title: string; - eventRule: EventRule; - elements: DisplayElement[]; - source: AnyExternalSource; -}; - -export class ExternalSourceEventTrigger implements Trigger { - constructor(private options: ExteralSourceEventTriggerOptions) {} - - eventElements(event: ApiEventLog): DisplayElement[] { - 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: this.connection, - eventRule: this.options.eventRule, - supportsPreparation: true, - }; - } - - registerWith(client: TriggerClient) { - client.register(this.options.source); - } - - async prepare(client: TriggerClient, auth?: ConnectionAuth) { - return this.options.source.prepare(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!, - }; - } - } -} diff --git a/packages/trigger-sdk/src/triggers/customEvent.ts b/packages/trigger-sdk/src/triggers/customEvent.ts new file mode 100644 index 000000000..9be6ef8bc --- /dev/null +++ b/packages/trigger-sdk/src/triggers/customEvent.ts @@ -0,0 +1,67 @@ +import type { + ApiEventLog, + EventFilter, + TriggerMetadata, +} from "@trigger.dev/internal"; +import { DisplayElement } from "@trigger.dev/internal"; +import { z } from "zod"; +import zodToJsonSchema from "zod-to-json-schema"; +import { Trigger } from "../types"; +import { TriggerClient } from "../triggerClient"; +import { Job } from "../job"; + +type CustomEventTriggerOptions = { + name: string; + source?: string; + schema?: TSchema; + filter?: EventFilter; +}; + +class CustomEventTrigger + implements Trigger> +{ + #options: CustomEventTriggerOptions; + + constructor(options: CustomEventTriggerOptions) { + this.#options = options; + } + + eventElements(event: ApiEventLog): DisplayElement[] { + return []; + } + + toJSON(): TriggerMetadata { + return { + title: "Custom Event", + elements: [{ label: "on", text: this.#options.name }], + schema: this.#options.schema + ? zodToJsonSchema(this.#options.schema) + : undefined, + eventRule: { + event: this.#options.name, + source: this.#options.source ?? "trigger.dev", + payload: this.#options.filter ?? {}, + }, + }; + } + + parsePayload(payload: unknown): z.infer { + if (!this.#options.schema) { + return payload; + } + + return this.#options.schema.parse(payload); + } + + attach( + triggerClient: TriggerClient, + job: Job>, any>, + variantId?: string + ): void {} +} + +export function customEvent( + options: CustomEventTriggerOptions +): Trigger> { + return new CustomEventTrigger(options); +} diff --git a/packages/trigger-sdk/src/triggers/externalSource.ts b/packages/trigger-sdk/src/triggers/externalSource.ts new file mode 100644 index 000000000..460940531 --- /dev/null +++ b/packages/trigger-sdk/src/triggers/externalSource.ts @@ -0,0 +1,315 @@ +import type { + ApiEventLog, + EventRule, + TriggerMetadata, +} from "@trigger.dev/internal"; +import { DisplayElement } from "@trigger.dev/internal"; +import { z } from "zod"; + +import { + NormalizedRequest, + NormalizedResponse, + SendEvent, +} from "@trigger.dev/internal"; +import { Connection, IOWithConnections } from "../connections"; +import { Job } from "../job"; +import { TriggerClient } from "../triggerClient"; +import type { Trigger, TriggerContext } from "../types"; + +type HttpSourceEvent = { + request: NormalizedRequest; + secret?: string; +}; + +type SmtpSourceEvent = { + from: string; + to: string; + subject: string; + body: string; +}; + +type SqsSourceEvent = { + body: string; +}; + +type ExternalSourceChannelMap = { + http: { + event: HttpSourceEvent; + }; + smtp: { + event: SmtpSourceEvent; + }; + sqs: { + event: SqsSourceEvent; + }; +}; + +type ChannelNames = keyof ExternalSourceChannelMap; + +type RegisterFunction> = ( + io: IOWithConnections<{ client: TConnection }>, + ctx: TriggerContext +) => Promise; + +type HandlerFunction< + TEvent extends any, + TConnection extends Connection +> = ( + event: TEvent, + io: IOWithConnections<{ client: TConnection }>, + ctx: TriggerContext +) => Promise<{ response: NormalizedResponse; events: SendEvent[] }>; + +type ExternalSourceOptions< + TEvent extends any, + TChannel extends ChannelNames, + TConnection extends Connection +> = { + connection: TConnection; + register: RegisterFunction; + handler: HandlerFunction< + ExternalSourceChannelMap[TChannel]["event"], + TConnection + >; + parsePayload: (payload: unknown) => TEvent; +}; + +export class ExternalSource< + TEvent extends any, + TChannel extends ChannelNames, + TConnection extends Connection +> { + channel: TChannel; + + constructor( + channel: TChannel, + private options: ExternalSourceOptions + ) { + this.channel = channel; + } + + async register( + io: IOWithConnections<{ client: TConnection }>, + ctx: TriggerContext + ) { + await this.options.register(io, ctx); + } + + async handle( + event: ExternalSourceChannelMap[TChannel]["event"], + io: IOWithConnections<{ client: TConnection }>, + ctx: TriggerContext + ) { + return await this.options.handler(event, io, ctx); + } + + get connection() { + return this.options.connection; + } +} + +export type ExternalSourceEventTriggerOptions< + TEvent extends any, + TChannel extends ChannelNames, + TConnection extends Connection +> = { + title: string; + eventRule: EventRule; + elements: DisplayElement[]; + source: ExternalSource; +}; + +export class ExternalSourceEventTrigger< + TEventType extends any, + TChannel extends ChannelNames, + TConnection extends Connection +> implements Trigger +{ + constructor( + private options: ExternalSourceEventTriggerOptions< + TEventType, + TChannel, + TConnection + > + ) {} + + eventElements(event: ApiEventLog): DisplayElement[] { + return []; + } + + parsePayload(payload: unknown): TEventType { + return payload as TEventType; + } + + toJSON(): TriggerMetadata { + return { + title: this.options.title, + elements: this.options.elements, + eventRule: this.options.eventRule, + }; + } + + attach( + triggerClient: TriggerClient, + job: Job, any>, + variantId?: string + ): void { + triggerClient.attach( + new Job({ + id: `${job.id}-prepare-external-trigger${ + variantId ? `-${variantId}` : "" + }`, + name: `Prepare ${this.options.title}`, + version: job.version, + trigger: internalPrepareTrigger(job, variantId), + connections: { + client: this.options.source.connection, + }, + run: async (event, io, ctx) => { + return await this.options.source.register(io, ctx); + }, + // @ts-ignore + __internal: true, + }) + ); + + triggerClient.attach( + new Job({ + id: `${job.id}-handle-external-trigger${ + variantId ? `-${variantId}` : "" + }`, + name: `Handle ${this.options.title}`, + version: job.version, + trigger: rawSourceTrigger(this.options.source.channel, job, variantId), + connections: { + client: this.options.source.connection, + }, + run: async (event, io, ctx) => { + const { response, events } = await this.options.source.handle( + event, + io, + ctx + ); + + return { + response, + events, + }; + }, + // @ts-ignore + __internal: true, + }) + ); + } +} + +function rawSourceTrigger< + TChannel extends ChannelNames, + TEvent extends ExternalSourceChannelMap[TChannel]["event"] +>( + channel: TChannel, + job: Job, any>, + variantId?: string +): Trigger { + return new RawSourceEventTrigger(channel, job, variantId); +} + +class RawSourceEventTrigger< + TChannel extends ChannelNames, + TEvent extends ExternalSourceChannelMap[TChannel]["event"] +> implements Trigger +{ + constructor( + private channel: TChannel, + private job: Job, any>, + private variantId?: string + ) {} + + eventElements(event: ApiEventLog): DisplayElement[] { + return []; + } + + toJSON(): TriggerMetadata { + return { + title: "Handle Raw Source Event", + elements: [ + { label: "id", text: this.job.id }, + { label: "version", text: this.job.version }, + ], + eventRule: { + event: "internal.trigger.handle-raw-source-event", + source: "trigger.dev", + payload: { + jobId: [this.job.id], + jobVersion: [this.job.version], + variantId: this.variantId ? [this.variantId] : [], + }, + }, + }; + } + + parsePayload(payload: unknown): TEvent { + return payload as TEvent; + } + + attach( + triggerClient: TriggerClient, + job: Job, any>, + variantId?: string + ): void {} +} + +const PrepareTriggerEventSchema = z.object({ + jobId: z.string(), + jobVersion: z.string(), + variantId: z.string().optional(), +}); + +type PrepareTriggerEvent = z.infer; + +class PrepareTriggerInternalTrigger implements Trigger { + constructor( + private job: Job, any>, + private variantId?: string + ) {} + + eventElements(event: ApiEventLog): DisplayElement[] { + return []; + } + + toJSON(): TriggerMetadata { + return { + title: "Prepare Trigger", + elements: [ + { label: "id", text: this.job.id }, + { label: "version", text: this.job.version }, + ], + eventRule: { + event: "internal.trigger.prepare", + source: "trigger.dev", + payload: { + jobId: [this.job.id], + jobVersion: [this.job.version], + variantId: this.variantId ? [this.variantId] : [], + }, + }, + }; + } + + parsePayload(payload: unknown): PrepareTriggerEvent { + return PrepareTriggerEventSchema.parse(payload); + } + + attach( + triggerClient: TriggerClient, + job: Job, any>, + variantId?: string + ): void {} +} + +export function internalPrepareTrigger( + job: Job, any>, + variantId?: string +): Trigger { + return new PrepareTriggerInternalTrigger(job, variantId); +} diff --git a/packages/trigger-sdk/src/types.ts b/packages/trigger-sdk/src/types.ts index d394494e4..02b6109f2 100644 --- a/packages/trigger-sdk/src/types.ts +++ b/packages/trigger-sdk/src/types.ts @@ -4,7 +4,11 @@ import type { SecureString, SendEvent, SendEventOptions, + TriggerMetadata, } from "@trigger.dev/internal"; +import { DisplayElement } from "@trigger.dev/internal"; +import { Job } from "./job"; +import { TriggerClient } from "./triggerClient"; export type { SecureString }; @@ -15,27 +19,17 @@ export interface TriggerContext { organization: string; startedAt: Date; isTest: boolean; - logger: TaskLogger; signal: AbortSignal; + // TODO: move this to io + logger: TaskLogger; + // TODO: move this to io wait(key: string | any[], seconds: number): Promise; + // TODO: move this to io sendEvent( key: string | any[], event: SendEvent, options?: SendEventOptions ): Promise; - // waitUntil(key: string, date: Date): Promise; - // runOnce( - // key: string, - // callback: T - // ): Promise>>; - // runOnceLocalOnly( - // key: string, - // callback: T - // ): Promise>>; - // fetch: TriggerFetch; - // kv: TriggerKeyValueStorage; - // globalKv: TriggerKeyValueStorage; - // runKv: TriggerKeyValueStorage; } export interface TaskLogger { @@ -44,3 +38,20 @@ export interface TaskLogger { warn(message: string, properties?: Record): Promise; error(message: string, properties?: Record): Promise; } + +export type TriggerEventType> = + TTrigger extends Trigger ? TEventType : never; + +export interface Trigger { + eventElements(event: ApiEventLog): DisplayElement[]; + toJSON(): TriggerMetadata; + parsePayload(payload: unknown): TEventType; + + // Attach this trigger to the job and the trigger client + // Gives different triggers the ability to do things like register internal jobs + attach( + triggerClient: TriggerClient, + job: Job, any>, + variantId?: string + ): void; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ef9eb1358..b3fdb0375 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -134,6 +134,7 @@ importers: eslint-config-prettier: ^8.5.0 eslint-plugin-cypress: ^2.12.1 express: ^4.18.1 + fast-redact: ^3.1.2 glob: ^8.0.3 graphile-worker: ^0.13.0 happy-dom: ^6.0.4 @@ -211,7 +212,7 @@ importers: '@aws-sdk/s3-request-presigner': 3.245.0 '@cakework/client': 0.0.54 '@cfworker/json-schema': 1.12.5 - '@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde + '@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu '@codemirror/commands': 6.1.3 '@codemirror/lang-javascript': 6.1.2 '@codemirror/lang-json': 6.0.1 @@ -241,7 +242,7 @@ importers: '@trigger.dev/internal': link:../../packages/internal '@typeform/embed-react': 2.14.1_react@18.2.0 '@types/simple-oauth2': 5.0.4 - '@uiw/react-codemirror': 4.19.5_aguurb4bmecpxzejz52amioxne + '@uiw/react-codemirror': 4.19.5_k4ec5g7vuuzzonc3d6xbjnmmle bcryptjs: 2.4.3 class-variance-authority: 0.5.2_typescript@4.9.4 classnames: 2.3.2 @@ -255,6 +256,7 @@ importers: date-fns: 2.29.3 emails: link:../../packages/emails express: 4.18.2 + fast-redact: 3.1.2 graphile-worker: 0.13.0 humanize-duration: 3.27.3 intl-parse-accept-language: 1.0.0 @@ -3203,13 +3205,12 @@ packages: prettier: 2.8.2 dev: false - /@codemirror/autocomplete/6.4.0_eo6pz6bvsllvatnnwfprpuflde: + /@codemirror/autocomplete/6.4.0_czcfkg2f66rxeiodoti7r2gulu: resolution: {integrity: sha512-HLF2PnZAm1s4kGs30EiqKMgD7XsYaQ0XJnMR0rofEWQ5t5D60SfqpDIkIh1ze5tiEbyUWm8+VJ6W1/erVvBMIA==} peerDependencies: '@codemirror/language': ^6.0.0 '@codemirror/state': ^6.0.0 '@codemirror/view': ^6.0.0 - '@lezer/common': ^1.0.0 dependencies: '@codemirror/language': 6.3.2 '@codemirror/state': 6.2.0 @@ -3229,7 +3230,7 @@ packages: /@codemirror/lang-javascript/6.1.2: resolution: {integrity: sha512-OcwLfZXdQ1OHrLiIcKCn7MqZ7nx205CMKlhe+vL88pe2ymhT9+2P+QhwkYGxMICj8TDHyp8HFKVwpiisUT7iEQ==} dependencies: - '@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde + '@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu '@codemirror/language': 6.3.2 '@codemirror/lint': 6.1.0 '@codemirror/state': 6.2.0 @@ -5488,7 +5489,7 @@ packages: eslint: 8.31.0 eslint-import-resolver-node: 0.3.6 eslint-import-resolver-typescript: 3.5.3_vz4tyq5r7fh66imfi352lmrvhq - eslint-plugin-import: 2.27.5_qdjeohovcytra7xto5vgmxssaq + eslint-plugin-import: 2.27.5_2ac3tknkazjoq5fxmuugu665ny eslint-plugin-jest: 26.9.0_y6565ziejixavcuubgd3r7fqr4 eslint-plugin-jest-dom: 4.0.3_eslint@8.31.0 eslint-plugin-jsx-a11y: 6.7.1_eslint@8.31.0 @@ -6762,18 +6763,17 @@ packages: eslint-visitor-keys: 3.3.0 dev: true - /@uiw/codemirror-extensions-basic-setup/4.19.5_tbeldtdcrf45b35pezgkzq2u4e: + /@uiw/codemirror-extensions-basic-setup/4.19.5_wd2tsis3in55bkaiwnc2c46tom: resolution: {integrity: sha512-1zt7ZPJ01xKkSW/KDy0FZNga0bngN1fC594wCVG7FBi60ehfcAucpooQ+JSPScKXopxcb+ugPKZvVLzr9/OfzA==} peerDependencies: '@codemirror/autocomplete': '>=6.0.0' '@codemirror/commands': '>=6.0.0' '@codemirror/language': '>=6.0.0' - '@codemirror/lint': '>=6.0.0' '@codemirror/search': '>=6.0.0' '@codemirror/state': '>=6.0.0' '@codemirror/view': '>=6.0.0' dependencies: - '@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde + '@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu '@codemirror/commands': 6.1.3 '@codemirror/language': 6.3.2 '@codemirror/lint': 6.1.0 @@ -6782,14 +6782,11 @@ packages: '@codemirror/view': 6.7.2 dev: false - /@uiw/react-codemirror/4.19.5_aguurb4bmecpxzejz52amioxne: + /@uiw/react-codemirror/4.19.5_k4ec5g7vuuzzonc3d6xbjnmmle: resolution: {integrity: sha512-ZCHh8d7beXbF8/t7F1+yHht6A9Y6CdKeOkZq4A09lxJEnyTQrj1FMf2zvfaqc7K23KNjkTCtSlbqKKbVDgrWaw==} peerDependencies: - '@babel/runtime': '>=7.11.0' '@codemirror/state': '>=6.0.0' - '@codemirror/theme-one-dark': '>=6.0.0' '@codemirror/view': '>=6.0.0' - codemirror: '>=6.0.0' react: '>=16.8.0' react-dom: '>=16.8.0' dependencies: @@ -6798,14 +6795,13 @@ packages: '@codemirror/state': 6.2.0 '@codemirror/theme-one-dark': 6.1.0 '@codemirror/view': 6.7.2 - '@uiw/codemirror-extensions-basic-setup': 4.19.5_tbeldtdcrf45b35pezgkzq2u4e - codemirror: 6.0.1_@lezer+common@1.0.2 + '@uiw/codemirror-extensions-basic-setup': 4.19.5_wd2tsis3in55bkaiwnc2c46tom + codemirror: 6.0.1 react: 18.2.0 react-dom: 18.2.0_react@18.2.0 transitivePeerDependencies: - '@codemirror/autocomplete' - '@codemirror/language' - - '@codemirror/lint' - '@codemirror/search' dev: false @@ -8374,18 +8370,16 @@ packages: engines: {node: '>=0.10.0'} dev: false - /codemirror/6.0.1_@lezer+common@1.0.2: + /codemirror/6.0.1: resolution: {integrity: sha512-J8j+nZ+CdWmIeFIGXEFbFPtpiYacFMDR8GlHK3IyHQJMCaVRfGx9NT+Hxivv1ckLWPvNdZqndbr/7lVhrf/Svg==} dependencies: - '@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde + '@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu '@codemirror/commands': 6.1.3 '@codemirror/language': 6.3.2 '@codemirror/lint': 6.1.0 '@codemirror/search': 6.2.3 '@codemirror/state': 6.2.0 '@codemirror/view': 6.7.2 - transitivePeerDependencies: - - '@lezer/common' dev: false /collection-visit/1.0.0: @@ -9748,7 +9742,7 @@ packages: debug: 4.3.4 enhanced-resolve: 5.12.0 eslint: 8.31.0 - eslint-plugin-import: 2.27.5_qdjeohovcytra7xto5vgmxssaq + eslint-plugin-import: 2.27.5_2ac3tknkazjoq5fxmuugu665ny get-tsconfig: 4.4.0 globby: 13.1.3 is-core-module: 2.11.0 @@ -9758,7 +9752,7 @@ packages: - supports-color dev: true - /eslint-module-utils/2.7.4_sqt5xxn4ciiurbqrzlaarm6ama: + /eslint-module-utils/2.7.4_v73lhamtbyinynmwa5fn7kpmfq: resolution: {integrity: sha512-j4GT+rqzCoRKHwURX7pddtIPGySnX9Si/cgMI5ztrcqOPtk5dDEeZ34CQVPphnqkJytlc97Vuk05Um2mJ3gEQA==} engines: {node: '>=4'} peerDependencies: @@ -9783,6 +9777,7 @@ packages: debug: 3.2.7 eslint: 8.31.0 eslint-import-resolver-node: 0.3.7 + eslint-import-resolver-typescript: 3.5.3_vz4tyq5r7fh66imfi352lmrvhq transitivePeerDependencies: - supports-color dev: true @@ -9807,7 +9802,7 @@ packages: regexpp: 3.2.0 dev: true - /eslint-plugin-import/2.27.5_qdjeohovcytra7xto5vgmxssaq: + /eslint-plugin-import/2.27.5_2ac3tknkazjoq5fxmuugu665ny: resolution: {integrity: sha512-LmEt3GVofgiGuiE+ORpnvP+kAm3h6MLZJ4Q5HCyHADofsb4VzXFsRiWj3c0OFiV+3DWFh0qg3v9gcPlfc3zRow==} engines: {node: '>=4'} peerDependencies: @@ -9825,7 +9820,7 @@ packages: doctrine: 2.1.0 eslint: 8.31.0 eslint-import-resolver-node: 0.3.7 - eslint-module-utils: 2.7.4_sqt5xxn4ciiurbqrzlaarm6ama + eslint-module-utils: 2.7.4_v73lhamtbyinynmwa5fn7kpmfq has: 1.0.3 is-core-module: 2.11.0 is-glob: 4.0.3 @@ -10483,6 +10478,11 @@ packages: resolution: {integrity: sha512-DCXu6Ifhqcks7TZKY3Hxp3y6qphY5SJZmrWMDrKcERSOXWQdMhU9Ig/PYrzyw/ul9jOIyh0N4M0tbC5hodg8dw==} dev: true + /fast-redact/3.1.2: + resolution: {integrity: sha512-+0em+Iya9fKGfEQGcd62Yv6onjBmmhV1uh86XVfOU8VwAe6kaFdQCWI9s0/Nnugx5Vd9tdbZ7e6gE2tR9dzXdw==} + engines: {node: '>=6'} + dev: false + /fast-xml-parser/4.0.11: resolution: {integrity: sha512-4aUg3aNRR/WjQAcpceODG1C3x3lFANXRo8+1biqfieHmg9pyMt7qB4lQV/Ta6sJCTbA5vfD8fnA8S54JATiFUA==} hasBin: true