diff --git a/apps/webapp/app/routes/api.v3.$endpointSlug.auth.$id.ts b/apps/webapp/app/routes/api.v3.$endpointSlug.auth.$id.ts new file mode 100644 index 000000000..60b145417 --- /dev/null +++ b/apps/webapp/app/routes/api.v3.$endpointSlug.auth.$id.ts @@ -0,0 +1,68 @@ +import type { LoaderArgs } from "@remix-run/server-runtime"; +import { json } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { prisma } from "~/db.server"; +import { resolveApiConnection } from "~/models/runConnection.server"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; +import { logger } from "~/services/logger"; + +const ParamsSchema = z.object({ + endpointSlug: z.string(), + id: z.string(), +}); + +export async function loader({ request, params }: LoaderArgs) { + logger.info("Fetching auth", { url: request.url }); + + 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 }); + } + + const apiClient = await prisma.apiConnectionClient.findUnique({ + where: { + organizationId_slug: { + organizationId: authenticatedEnv.organizationId, + slug: parsedParams.data.id, + }, + }, + }); + + if (!apiClient) { + return json({ error: "API Client not found" }, { status: 404 }); + } + + const connection = await prisma.apiConnection.findFirst({ + where: { + clientId: apiClient.id, + connectionType: "DEVELOPER", + }, + include: { + dataReference: true, + }, + }); + + if (!connection) { + return json({ error: "API Connection not found" }, { status: 404 }); + } + + const connectionAuth = await resolveApiConnection(connection); + + if (!connectionAuth) { + return json({ error: "Access token not found" }, { status: 404 }); + } + + return json(connectionAuth); +} diff --git a/apps/webapp/app/routes/api.v3.$endpointSlug.triggers.$id.ts b/apps/webapp/app/routes/api.v3.$endpointSlug.triggers.$id.ts new file mode 100644 index 000000000..cac0e8715 --- /dev/null +++ b/apps/webapp/app/routes/api.v3.$endpointSlug.triggers.$id.ts @@ -0,0 +1,75 @@ +import type { ActionArgs } from "@remix-run/server-runtime"; +import { json } from "@remix-run/server-runtime"; +import { + RegisterTriggerBodySchema, + UpdateTriggerSourceBodySchema, +} from "@trigger.dev/internal"; +import { z } from "zod"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; +import { logger } from "~/services/logger"; +import { UpdateSourceService } from "~/services/sources/updateSource.server"; +import { RegisterTriggerService } from "~/services/triggers/registerTrigger.server"; + +const ParamsSchema = z.object({ + endpointSlug: z.string(), + id: z.string(), +}); + +export async function action({ request, params }: ActionArgs) { + logger.info("Registering trigger", { url: request.url }); + + // Ensure this is a POST request + if (request.method.toUpperCase() !== "PUT") { + 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 = RegisterTriggerBodySchema.safeParse(anyBody); + + if (!body.success) { + return json({ error: "Invalid request body" }, { status: 400 }); + } + + const service = new RegisterTriggerService(); + + try { + const source = await service.call({ + environment: authenticatedEnv, + payload: body.data, + endpointSlug: parsedParams.data.endpointSlug, + id: parsedParams.data.id, + }); + + return json(source); + } catch (error) { + if (error instanceof Error) { + logger.error("Error registering trigger", { + url: request.url, + error: error.message, + }); + + return json({ error: error.message }, { status: 400 }); + } + + return json({ error: "Something went wrong" }, { status: 500 }); + } +} diff --git a/apps/webapp/app/routes/api.v3.runs.$runId.tasks.ts b/apps/webapp/app/routes/api.v3.runs.$runId.tasks.ts index 9cfb3433e..98c24fb8a 100644 --- a/apps/webapp/app/routes/api.v3.runs.$runId.tasks.ts +++ b/apps/webapp/app/routes/api.v3.runs.$runId.tasks.ts @@ -1,6 +1,6 @@ import type { ActionArgs, LoaderArgs } from "@remix-run/server-runtime"; import { json } from "@remix-run/server-runtime"; -import type { RunTaskBodyOutput } from "@trigger.dev/internal"; +import { RunTaskBodyOutput, ServerTaskSchema } from "@trigger.dev/internal"; import { RunTaskBodyOutputSchema } from "@trigger.dev/internal"; import { z } from "zod"; import type { PrismaClient } from "~/db.server"; @@ -145,7 +145,7 @@ export async function action({ request, params }: ActionArgs) { task, }); - return json(task); + return json(ServerTaskSchema.parse(task)); } catch (error) { if (error instanceof Error) { return json({ error: error.message }, { status: 400 }); diff --git a/apps/webapp/app/services/clientApi.server.ts b/apps/webapp/app/services/clientApi.server.ts index 468d8ac1b..04dc7cac5 100644 --- a/apps/webapp/app/services/clientApi.server.ts +++ b/apps/webapp/app/services/clientApi.server.ts @@ -210,6 +210,7 @@ export class ClientApi { async deliverHttpSourceRequest(options: { key: string; + dynamicId?: string; secret: string; params: any; data: any; @@ -228,6 +229,7 @@ export class ClientApi { "x-ts-http-url": options.request.url, "x-ts-http-method": options.request.method, "x-ts-http-headers": JSON.stringify(options.request.headers), + ...(options.dynamicId && { "x-ts-dynamic-id": options.dynamicId }), }, body: options.request.rawBody, }); diff --git a/apps/webapp/app/services/events/deliverEvent.server.ts b/apps/webapp/app/services/events/deliverEvent.server.ts index f26bec97c..3445fdd0d 100644 --- a/apps/webapp/app/services/events/deliverEvent.server.ts +++ b/apps/webapp/app/services/events/deliverEvent.server.ts @@ -3,9 +3,9 @@ import type { EventFilter } from "@trigger.dev/internal"; import { EventFilterSchema } from "@trigger.dev/internal"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import { CreateRunService } from "../runs/createRun.server"; -import { ResumeTaskService } from "../runs/resumeTask.server"; -import { logger } from "../logger"; +import { logger } from "~/services/logger"; +import { CreateRunService } from "~/services/runs/createRun.server"; +import { ResumeTaskService } from "~/services/runs/resumeTask.server"; export class DeliverEventService { #prismaClient: PrismaClient; diff --git a/apps/webapp/app/services/runs/resumeTask.server.ts b/apps/webapp/app/services/runs/resumeTask.server.ts index b457d94a8..ad14b3150 100644 --- a/apps/webapp/app/services/runs/resumeTask.server.ts +++ b/apps/webapp/app/services/runs/resumeTask.server.ts @@ -88,14 +88,21 @@ export class ResumeTaskService { id: run.version.job.slug, version: run.version.version, }, - context: { + run: { id: run.id, - environment: run.environment.slug, - organization: run.organization.slug, isTest: run.isTest, - version: run.version.version, startedAt: run.startedAt ?? new Date(), }, + environment: { + id: run.environment.id, + slug: run.environment.slug, + type: run.environment.type, + }, + organization: { + id: run.organization.id, + slug: run.organization.slug, + title: run.organization.title, + }, tasks: [run.tasks, updatedTask] .flat() .map((t) => CachedTaskSchema.parse(t)), diff --git a/apps/webapp/app/services/runs/startRun.server.ts b/apps/webapp/app/services/runs/startRun.server.ts index 9863bf953..1a33e960b 100644 --- a/apps/webapp/app/services/runs/startRun.server.ts +++ b/apps/webapp/app/services/runs/startRun.server.ts @@ -16,7 +16,7 @@ const RUN_INCLUDES = { job: true, environment: true, organization: true, - connections: { + integrations: { include: { apiConnectionClient: { include: { @@ -71,7 +71,7 @@ export class StartRunService { return { run: updatedRun }; } else { // If any of the connections are missing, we can't start the execution - const runConnectionsByKey = run.version.connections.reduce( + const runConnectionsByKey = run.version.integrations.reduce( (acc: Record, connection) => { if (connection.apiConnectionClient.connections.length === 0) { return acc; @@ -179,14 +179,21 @@ export class StartRunService { id: run.version.job.slug, version: run.version.version, }, - context: { + run: { id: run.id, - environment: run.version.environment.slug, - organization: run.version.organization.slug, isTest: run.isTest, - version: run.version.version, startedAt, }, + environment: { + id: run.version.environment.id, + slug: run.version.environment.slug, + type: run.version.environment.type, + }, + organization: { + id: run.version.organization.id, + slug: run.version.organization.slug, + title: run.version.organization.title, + }, connections, }); diff --git a/apps/webapp/app/services/secrets/secretStore.server.ts b/apps/webapp/app/services/secrets/secretStore.server.ts index 9dfe5a4fc..a43b9b93b 100644 --- a/apps/webapp/app/services/secrets/secretStore.server.ts +++ b/apps/webapp/app/services/secrets/secretStore.server.ts @@ -2,10 +2,7 @@ import { prisma } from "~/db.server"; import { z } from "zod"; export interface ASecretStore { - getSecret( - schema: TSchema, - key: string - ): Promise | undefined>; + getSecret(schema: z.Schema, key: string): Promise; setSecret(key: string, value: T): Promise; } @@ -31,13 +28,20 @@ export class SecretStore implements ASecretStore { } } - getSecret( - schema: TSchema, - key: string - ): Promise | undefined> { + getSecret(schema: z.Schema, key: string): Promise { return this.#provider.getSecret(schema, key); } + async getSecretOrThrow(schema: z.Schema, key: string): Promise { + const value = await this.getSecret(schema, key); + + if (!value) { + throw new Error(`Unable to find secret ${key} in ${this.provider}`); + } + + return value; + } + setSecret(key: string, value: T): Promise { return this.#provider.setSecret(key, value); } @@ -45,10 +49,7 @@ export class SecretStore implements ASecretStore { /** This stores secrets in the Postgres Database, in plain text. NOT recommended outside of localhost. */ class DatabaseSecretStore implements ASecretStore { - async getSecret( - schema: TSchema, - key: string - ): Promise | undefined> { + async getSecret(schema: z.Schema, key: string): Promise { const secret = await prisma.secretStore.findUnique({ where: { key, diff --git a/apps/webapp/app/services/sources/activateSource.server.ts b/apps/webapp/app/services/sources/activateSource.server.ts index 0af623f22..882979c54 100644 --- a/apps/webapp/app/services/sources/activateSource.server.ts +++ b/apps/webapp/app/services/sources/activateSource.server.ts @@ -5,13 +5,13 @@ import type { SecretReference, TriggerSourceEvent, } from ".prisma/client"; -import { IngestSendEvent } from "../events/ingestSendEvent.server"; import type { AuthenticatedEnvironment } from "../apiAuth.server"; import { env } from "~/env.server"; import type { RegisterTriggerSource } from "@trigger.dev/internal"; import type { SecretStoreProvider } from "../secrets/secretStore.server"; import { SecretStore } from "../secrets/secretStore.server"; import { z } from "zod"; +import { IngestSendEvent } from "../events/ingestSendEvent.server"; export class ActivateSourceService { #prismaClient: PrismaClient; diff --git a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts index ce593ac3c..ba16fac39 100644 --- a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts +++ b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts @@ -28,6 +28,7 @@ export class DeliverHttpSourceRequestService { source: { include: { secretReference: true, + dynamicTrigger: true, }, }, }, @@ -61,6 +62,7 @@ export class DeliverHttpSourceRequestService { const { response, events } = await clientApi.deliverHttpSourceRequest({ key: httpSourceRequest.source.key, + dynamicId: httpSourceRequest.source.dynamicTrigger?.slug, secret: secret.secret, params: httpSourceRequest.source.params, data: httpSourceRequest.source.channelData, diff --git a/apps/webapp/app/services/sources/registerSource.server.ts b/apps/webapp/app/services/sources/registerSource.server.ts index c9a4a55db..9e496489f 100644 --- a/apps/webapp/app/services/sources/registerSource.server.ts +++ b/apps/webapp/app/services/sources/registerSource.server.ts @@ -14,7 +14,11 @@ export class RegisterSourceService { this.#prismaClient = prismaClient; } - public async call(endpointId: string, metadata: SourceMetadata) { + public async call( + endpointId: string, + metadata: SourceMetadata, + dynamicTriggerId?: string + ) { const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ where: { id: endpointId, @@ -29,13 +33,19 @@ export class RegisterSourceService { }, }); - await this.#upsertSource(endpoint, endpoint.environment, metadata); + return this.#upsertSource( + endpoint, + endpoint.environment, + metadata, + dynamicTriggerId + ); } async #upsertSource( endpoint: Endpoint, environment: AuthenticatedEnvironment, - metadata: SourceMetadata + metadata: SourceMetadata, + dynamicTriggerId?: string ) { logger.debug("Upserting source", { endpoint, @@ -43,18 +53,33 @@ export class RegisterSourceService { metadata, }); + const key = dynamicTriggerId + ? `${dynamicTriggerId}:${metadata.key}` + : metadata.key; + const { id, orphanedEvents } = await this.#prismaClient.$transaction( async (tx) => { + const apiClient = metadata.clientId + ? await tx.apiConnectionClient.findUnique({ + where: { + organizationId_slug: { + organizationId: environment.organizationId, + slug: metadata.clientId, + }, + }, + }) + : undefined; + const triggerSource = await tx.triggerSource.upsert({ where: { key_endpointId: { endpointId: endpoint.id, - key: metadata.key, + key, }, }, create: { params: metadata.params, - key: metadata.key, + key, channel: metadata.channel, organization: { connect: { @@ -76,6 +101,16 @@ export class RegisterSourceService { id: environment.id, }, }, + apiClient: apiClient + ? { connect: { id: apiClient.id } } + : undefined, + dynamicTrigger: dynamicTriggerId + ? { + connect: { + id: dynamicTriggerId, + }, + } + : undefined, events: { create: metadata.events.map((event) => ({ name: event, @@ -84,10 +119,10 @@ export class RegisterSourceService { secretReference: { connectOrCreate: { where: { - key: `${endpoint.id}:${metadata.key}`, + key: `${endpoint.id}:${key}`, }, create: { - key: `${endpoint.id}:${metadata.key}`, + key: `${endpoint.id}:${key}`, provider: "database", }, }, @@ -164,9 +199,15 @@ export class RegisterSourceService { }, include: { events: true, + secretReference: true, + apiClient: true, }, }); + if (dynamicTriggerId) { + return triggerSource; + } + const triggerIsActive = triggerSource.active; const triggerHasOrphanedEvents = orphanedEvents.length > 0; const triggerHasUnregisteredEvents = triggerSource.events.some( @@ -184,5 +225,7 @@ export class RegisterSourceService { orphanedEvents: orphanedEvents, }); } + + return triggerSource; } } diff --git a/apps/webapp/app/services/triggers/registerTrigger.server.ts b/apps/webapp/app/services/triggers/registerTrigger.server.ts new file mode 100644 index 000000000..8ab2a15d7 --- /dev/null +++ b/apps/webapp/app/services/triggers/registerTrigger.server.ts @@ -0,0 +1,141 @@ +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import { AuthenticatedEnvironment } from "../apiAuth.server"; +import { + RegisterSourceEvent, + RegisterTriggerBody, +} from "@trigger.dev/internal"; +import { RegisterSourceService } from "../sources/registerSource.server"; +import { + SecretStore, + SecretStoreProvider, +} from "../secrets/secretStore.server"; +import { z } from "zod"; +import { env } from "~/env.server"; + +export class RegisterTriggerService { + #prismaClient: PrismaClient; + #registerSource = new RegisterSourceService(); + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call({ + environment, + payload, + endpointSlug, + id, + }: { + environment: AuthenticatedEnvironment; + payload: RegisterTriggerBody; + id: string; + endpointSlug: string; + }): Promise { + const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ + where: { + environmentId_slug: { + environmentId: environment.id, + slug: endpointSlug, + }, + }, + }); + + const dynamicTrigger = + await this.#prismaClient.dynamicTrigger.findUniqueOrThrow({ + where: { + endpointId_slug: { + endpointId: endpoint.id, + slug: id, + }, + }, + include: { + jobs: { + include: { + aliases: { + where: { + name: "latest", + }, + include: { + version: true, + }, + take: 1, + }, + }, + }, + }, + }); + + const triggerSource = await this.#registerSource.call( + endpoint.id, + payload.source, + dynamicTrigger.id + ); + + // For each job, we need to create a JobTrigger + for (const job of dynamicTrigger.jobs) { + const version = job.aliases[0] ? job.aliases[0].version : undefined; + + if (!version) { + continue; + } + + await this.#prismaClient.jobTrigger.upsert({ + where: { + versionId_actionIdentifier: { + versionId: version.id, + actionIdentifier: triggerSource.id, + }, + }, + create: { + event: payload.rule.event, + source: payload.rule.source, + payloadFilter: payload.rule.payload, + contextFilter: payload.rule.context, + jobId: job.id, + versionId: version.id, + environmentId: environment.id, + organizationId: environment.organizationId, + projectId: environment.projectId, + dynamicTriggerId: dynamicTrigger.id, + enabled: true, + actionIdentifier: triggerSource.id, + }, + update: { + event: payload.rule.event, + source: payload.rule.source, + payloadFilter: payload.rule.payload, + contextFilter: payload.rule.context, + }, + }); + } + + const secretStore = new SecretStore( + triggerSource.secretReference.provider as SecretStoreProvider + ); + + const { secret } = await secretStore.getSecretOrThrow( + z.object({ + secret: z.string(), + }), + triggerSource.secretReference.key + ); + + return { + source: { + key: triggerSource.key, + active: triggerSource.active, + secret, + data: triggerSource.channelData as any, + channel: { + type: "HTTP", + url: `${env.APP_ORIGIN}/api/v3/sources/http/${triggerSource.id}`, + }, + clientId: triggerSource.apiClient?.slug, + }, + events: triggerSource.events.map((e) => e.name), + missingEvents: [], + orphanedEvents: [], + }; + } +} diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index 22416cebc..31c5ff728 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -2,7 +2,6 @@ import { z } from "zod"; import { env } from "~/env.server"; import { ZodWorker } from "~/platform/zodWorker.server"; import { EndpointRegisteredService } from "./endpoints/endpointRegistered.server"; -import { DeliverEventService } from "./events/deliverEvent.server"; import { apiAuthenticationRepository } from "./externalApis/apiAuthenticationRepository.server"; import { RegisterJobService } from "./jobs/registerJob.server"; import { ResumeTaskService } from "./runs/resumeTask.server"; @@ -13,6 +12,7 @@ import { RunFinishedService } from "./runs/runFinished.server"; import { JobMetadataSchema, SourceMetadataSchema } from "@trigger.dev/internal"; import { RegisterSourceService } from "./sources/registerSource.server"; import { ActivateSourceService } from "./sources/activateSource.server"; +import { DeliverEventService } from "./events/deliverEvent.server"; const workerCatalog = { organizationCreated: z.object({ id: z.string() }), diff --git a/apps/webapp/prisma/migrations/20230517153909_add_dynamic_trigger_to_trigger_sources/migration.sql b/apps/webapp/prisma/migrations/20230517153909_add_dynamic_trigger_to_trigger_sources/migration.sql new file mode 100644 index 000000000..4755789ae --- /dev/null +++ b/apps/webapp/prisma/migrations/20230517153909_add_dynamic_trigger_to_trigger_sources/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "TriggerSource" ADD COLUMN "dynamicTriggerId" TEXT; + +-- AddForeignKey +ALTER TABLE "TriggerSource" ADD CONSTRAINT "TriggerSource_dynamicTriggerId_fkey" FOREIGN KEY ("dynamicTriggerId") REFERENCES "DynamicTrigger"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20230517154918_add_dynamic_trigger_to_job_triggers/migration.sql b/apps/webapp/prisma/migrations/20230517154918_add_dynamic_trigger_to_job_triggers/migration.sql new file mode 100644 index 000000000..7110be59e --- /dev/null +++ b/apps/webapp/prisma/migrations/20230517154918_add_dynamic_trigger_to_job_triggers/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "JobTrigger" ADD COLUMN "dynamicTriggerId" TEXT; + +-- AddForeignKey +ALTER TABLE "JobTrigger" ADD CONSTRAINT "JobTrigger_dynamicTriggerId_fkey" FOREIGN KEY ("dynamicTriggerId") REFERENCES "DynamicTrigger"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index 410319527..bf659d69d 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -408,7 +408,9 @@ model DynamicTrigger { endpoint Endpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade, onUpdate: Cascade) endpointId String - jobs Job[] + jobs Job[] + sources TriggerSource[] + triggers JobTrigger[] @@unique([endpointId, slug]) } @@ -446,6 +448,9 @@ model JobTrigger { externalAccount ExternalAccount? @relation(fields: [externalAccountId], references: [id], onDelete: Cascade, onUpdate: Cascade) externalAccountId String? + dynamicTrigger DynamicTrigger? @relation(fields: [dynamicTriggerId], references: [id], onDelete: Cascade, onUpdate: Cascade) + dynamicTriggerId String? + @@unique([versionId, actionIdentifier]) } @@ -611,8 +616,7 @@ model SecretStore { model TriggerSource { id String @id @default(cuid()) - key String - + key String params Json? channel TriggerChannel @default(HTTP) @@ -638,6 +642,9 @@ model TriggerSource { apiClient ApiConnectionClient? @relation(fields: [apiClientId], references: [id], onDelete: Cascade, onUpdate: Cascade) apiClientId String? + dynamicTrigger DynamicTrigger? @relation(fields: [dynamicTriggerId], references: [id], onDelete: Cascade, onUpdate: Cascade) + dynamicTriggerId String? + active Boolean @default(false) interactive Boolean @default(false) diff --git a/examples/nextjs-example/.vscode/launch.json b/examples/nextjs-example/.vscode/launch.json new file mode 100644 index 000000000..44c0131a3 --- /dev/null +++ b/examples/nextjs-example/.vscode/launch.json @@ -0,0 +1,11 @@ +{ + "version": "0.2.0", + "configurations": [ + { + "name": "Next.js: debug server-side", + "type": "node-terminal", + "request": "launch", + "command": "pnpm run dev" + } + ] +} diff --git a/examples/nextjs-example/src/pages/api/trigger.ts b/examples/nextjs-example/src/pages/api/trigger.ts index 3960e2352..00f192318 100644 --- a/examples/nextjs-example/src/pages/api/trigger.ts +++ b/examples/nextjs-example/src/pages/api/trigger.ts @@ -35,6 +35,46 @@ const dynamicOnIssueOpenedTrigger = new DynamicTrigger(client, { source: github.sources.repo, }); +new Job(client, { + id: "register-dynamic-trigger-on-new-repo", + name: "Register dynamic trigger on new repo", + version: "0.1.1", + trigger: customTrigger({ + name: "new.repo", + event: customEvent({ + payload: z.object({ repo: z.string() }), + }), + }), + run: async (payload, io, ctx) => { + return await io.registerTrigger( + "register-repo", + dynamicOnIssueOpenedTrigger, + payload.repo, + { + repo: payload.repo, + } + ); + }, +}); + +new Job(client, { + id: "listen-for-dynamic-trigger", + name: "Listen for dynamic trigger", + version: "0.1.1", + trigger: dynamicOnIssueOpenedTrigger, + integrations: { + slack, + }, + run: async (payload, io, ctx) => { + await io.slack.postMessage("Slack 📝", { + text: `New Issue opened on dynamically triggered repo: ${ + payload.issue.html_url + }. \n\n${JSON.stringify(ctx)}`, + channel: "C04GWUTDC3W", + }); + }, +}); + new Job(client, { id: "alert-on-new-github-issues", name: "Alert on new GitHub issues", @@ -46,99 +86,99 @@ new Job(client, { event: events.onIssueOpened, repo: "ericallam/basic-starter-100k", }), - run: async (event, io, ctx) => { + run: async (payload, io, ctx) => { await io.slack.postMessage("Slack 📝", { - text: `New Issue opened: ${event.issue.html_url}`, + text: `New Issue opened: ${payload.issue.html_url}`, channel: "C04GWUTDC3W", }); }, }); -new Job(client, { - id: "comment-on-new-github-issues", - name: "Comment on new GitHub issues", - version: "0.1.1", - integrations: { - githubLocal, - }, - trigger: githubLocal.triggers.repo({ - event: events.onIssueOpened, - repo: "ericallam/basic-starter-100k", - }), - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "comment-on-new-github-issues", +// name: "Comment on new GitHub issues", +// version: "0.1.1", +// integrations: { +// githubLocal, +// }, +// trigger: githubLocal.triggers.repo({ +// event: events.onIssueOpened, +// repo: "ericallam/basic-starter-100k", +// }), +// run: async (payload, io, ctx) => {}, +// }); -new Job(client, { - id: "alert-on-new-github-issues-dynamic", - name: "Alert on new GitHub issues Dynamic", - version: "0.1.1", - trigger: dynamicOnIssueOpenedTrigger, - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "alert-on-new-github-issues-dynamic", +// name: "Alert on new GitHub issues Dynamic", +// version: "0.1.1", +// trigger: dynamicOnIssueOpenedTrigger, +// run: async (payload, io, ctx) => {}, +// }); -new Job(client, { - id: "alert-on-new-github-stars", - name: "Alert on new GitHub stars", - version: "0.1.1", - trigger: github.triggers.repo({ - event: events.onNewStar, - repo: "ericallam/basic-starter-100k", - }), - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "alert-on-new-github-stars", +// name: "Alert on new GitHub stars", +// version: "0.1.1", +// trigger: github.triggers.repo({ +// event: events.onNewStar, +// repo: "ericallam/basic-starter-100k", +// }), +// run: async (payload, io, ctx) => {}, +// }); -new Job(client, { - id: "alert-on-new-issue-comments", - name: "Alert on new github issue comments", - version: "0.1.1", - trigger: github.triggers.repo({ - event: events.onIssueComment, - repo: "ericallam/basic-starter-100k", - }), - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "alert-on-new-issue-comments", +// name: "Alert on new github issue comments", +// version: "0.1.1", +// trigger: github.triggers.repo({ +// event: events.onIssueComment, +// repo: "ericallam/basic-starter-100k", +// }), +// run: async (payload, io, ctx) => {}, +// }); -new Job(client, { - id: "alert-on-new-github-stars-in-org", - name: "Alert on new GitHub stars in Org", - version: "0.1.1", - trigger: comboTrigger({ - event: events.onNewStar, - triggers: [ - github.triggers.repo({ - event: events.onNewStar, - repo: "ericallam/stripe-to-email", - }), - github.triggers.repo({ - event: events.onNewStar, - repo: "ericallam/supabase-to-loops-cloud", - }), - ], - }), - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "alert-on-new-github-stars-in-org", +// name: "Alert on new GitHub stars in Org", +// version: "0.1.1", +// trigger: comboTrigger({ +// event: events.onNewStar, +// triggers: [ +// github.triggers.repo({ +// event: events.onNewStar, +// repo: "ericallam/stripe-to-email", +// }), +// github.triggers.repo({ +// event: events.onNewStar, +// repo: "ericallam/supabase-to-loops-cloud", +// }), +// ], +// }), +// run: async (payload, io, ctx) => {}, +// }); -new Job(client, { - id: "custom-event-example", - name: "Custom Event Example", - version: "0.1.1", - trigger: customTrigger({ - name: "my.custom.trigger", - event: customEvent({ schema: z.object({ id: z.string() }) }), - }), - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "custom-event-example", +// name: "Custom Event Example", +// version: "0.1.1", +// trigger: customTrigger({ +// name: "my.custom.trigger", +// event: customEvent({ payload: z.object({ id: z.string() }) }), +// }), +// run: async (payload, io, ctx) => {}, +// }); -new Job(client, { - id: "custom-github-event-example", - name: "Custom Github Event Example", - version: "0.1.1", - trigger: customTrigger({ - name: "my.custom.trigger", - event: events.onNewStar, - }), - run: async (event, io, ctx) => {}, -}); +// new Job(client, { +// id: "custom-github-event-example", +// name: "Custom Github Event Example", +// version: "0.1.1", +// trigger: customTrigger({ +// name: "my.custom.trigger", +// event: events.onNewStar, +// }), +// run: async (payload, io, ctx) => {}, +// }); export default async function handler( req: NextApiRequest, diff --git a/integrations/github/src/index.ts b/integrations/github/src/index.ts index d0c1f43c3..58ca3d169 100644 --- a/integrations/github/src/index.ts +++ b/integrations/github/src/index.ts @@ -2,6 +2,7 @@ import { IssueCommentEvent, IssuesEvent, IssuesOpenedEvent, + RepositoryCreatedEvent, StarCreatedEvent, StarEvent, } from "@octokit/webhooks-types"; @@ -124,12 +125,23 @@ const onNewStar: EventSpecification = { parsePayload: (payload) => payload as StarCreatedEvent, }; +const onNewRepository: EventSpecification = { + name: "repository", + title: "On new repository", + source: "github.com", + filter: { + action: ["created"], + }, + parsePayload: (payload) => payload as RepositoryCreatedEvent, +}; + export const events = { onIssueOpened, onIssue, onIssueComment, onStar, onNewStar, + onNewRepository, }; // params.event has to be a union of all the values of the exports events object @@ -147,11 +159,6 @@ function createRepoTrigger(source: ReturnType) { event, params: { repo }, source, - filter: { - repository: { - full_name: [repo], - }, - }, }); }; } @@ -168,11 +175,6 @@ function createOrgTrigger(source: ReturnType) { event, params: { org }, source, - filter: { - organization: { - login: [org], - }, - }, }); }; } diff --git a/integrations/github/src/sources.ts b/integrations/github/src/sources.ts index e6abcae7b..d9f61d552 100644 --- a/integrations/github/src/sources.ts +++ b/integrations/github/src/sources.ts @@ -35,6 +35,11 @@ export function createRepoEventSource( schema: z.object({ repo: z.string() }), integration, key: (params) => params.repo, + filter: (params) => ({ + repository: { + full_name: [params.repo], + }, + }), handler: async (event, logger) => { logger.debug("[inside github integration] Handling github repo event"); @@ -185,6 +190,11 @@ export function createOrgEventSource( integration, schema: z.object({ org: z.string() }), key: (params) => params.org, + filter: (params) => ({ + organization: { + login: [params.org], + }, + }), handler: async (event) => {}, register: async (event, io, ctx) => { const { params, source: httpSource, events, missingEvents } = event; diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index f12a55868..b042b8afb 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -9,6 +9,7 @@ import { EventSpecificationSchema, TriggerMetadataSchema, } from "./triggers"; +import { EventRuleSchema } from "./eventFilter"; export const UpdateTriggerSourceBodySchema = z.object({ registeredEvents: z.array(z.string()), @@ -51,6 +52,7 @@ export const RegisterTriggerSourceSchema = z.object({ secret: z.string(), data: DeserializedJsonSchema.optional(), channel: RegisterSourceChannelBodySchema, + clientId: z.string().optional(), }); export type RegisterTriggerSource = z.infer; @@ -91,6 +93,7 @@ export type HttpSourceRequest = z.infer; export const HttpSourceRequestHeadersSchema = z.object({ "x-ts-key": z.string(), + "x-ts-dynamic-id": z.string().optional(), "x-ts-secret": z.string(), "x-ts-data": z.string().transform((s) => JSON.parse(s)), "x-ts-params": z.string().transform((s) => JSON.parse(s)), @@ -135,6 +138,7 @@ export const SourceMetadataSchema = z.object({ key: z.string(), params: z.any(), events: z.array(z.string()), + clientId: z.string().optional(), }); export type SourceMetadata = z.infer; @@ -192,20 +196,38 @@ export const DeliverEventResponseSchema = z.object({ export type DeliverEventResponse = z.infer; +export const RuntimeEnvironmentTypeSchema = z.enum([ + "PRODUCTION", + "STAGING", + "DEVELOPMENT", + "PREVIEW", +]); + +export type RuntimeEnvironmentType = z.infer< + typeof RuntimeEnvironmentTypeSchema +>; + export const RunJobBodySchema = z.object({ event: ApiEventLogSchema, job: z.object({ id: z.string(), version: z.string(), }), - context: z.object({ + run: z.object({ id: z.string(), - environment: z.string(), - organization: z.string(), isTest: z.boolean(), - version: z.string(), startedAt: z.coerce.date(), }), + environment: z.object({ + id: z.string(), + slug: z.string(), + type: RuntimeEnvironmentTypeSchema, + }), + organization: z.object({ + id: z.string(), + title: z.string(), + slug: z.string(), + }), tasks: z.array(CachedTaskSchema).optional(), connections: z.record(ConnectionAuthSchema).optional(), }); @@ -324,7 +346,7 @@ export const CompleteTaskBodyInputSchema = RunTaskBodyInputSchema.pick({ params: true, }).extend({ output: SerializableJsonSchema.optional().transform((v) => - DeserializedJsonSchema.parse(JSON.parse(JSON.stringify(v))) + v ? DeserializedJsonSchema.parse(JSON.parse(JSON.stringify(v))) : {} ), }); @@ -367,3 +389,10 @@ export const TriggerVariantResponseBodySchema = z.object({ export type TriggerVariantResponseBody = z.infer< typeof TriggerVariantResponseBodySchema >; + +export const RegisterTriggerBodySchema = z.object({ + rule: EventRuleSchema, + source: SourceMetadataSchema, +}); + +export type RegisterTriggerBody = z.infer; diff --git a/packages/trigger-sdk/src/apiClient.ts b/packages/trigger-sdk/src/apiClient.ts index f81a4850f..95c9fcd06 100644 --- a/packages/trigger-sdk/src/apiClient.ts +++ b/packages/trigger-sdk/src/apiClient.ts @@ -2,10 +2,14 @@ import { ApiEventLog, ApiEventLogSchema, CompleteTaskBodyInput, + ConnectionAuthSchema, CreateRunBody, CreateRunResponseBodySchema, LogLevel, Logger, + RegisterSourceEvent, + RegisterSourceEventSchema, + RegisterTriggerBody, RunTaskBodyInput, SendEvent, SendEventOptions, @@ -206,6 +210,59 @@ export class ApiClient { return response; } + async registerTrigger( + client: string, + id: string, + payload: RegisterTriggerBody + ): Promise { + const apiKey = await this.#apiKey(); + + this.#logger.debug("registering trigger", { + id, + payload, + }); + + const response = await zodfetch( + RegisterSourceEventSchema, + `${this.#apiUrl}/api/v3/${client}/triggers/${id}`, + { + method: "PUT", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${apiKey}`, + }, + body: JSON.stringify(payload), + } + ); + + return response; + } + + async getAuth(client: string, id: string) { + const apiKey = await this.#apiKey(); + + this.#logger.debug("getting auth", { + id, + }); + + const response = await zodfetch( + ConnectionAuthSchema, + `${this.#apiUrl}/api/v3/${client}/auth/${id}`, + { + method: "GET", + headers: { + Accept: "application/json", + Authorization: `Bearer ${apiKey}`, + }, + }, + { + optional: true, + } + ); + + return response; + } + async #apiKey() { const apiKey = getApiKey(this.#options.apiKey); @@ -269,16 +326,29 @@ function getApiKey(key?: string) { return { status: "valid" as const, apiKey }; } -async function zodfetch( +async function zodfetch< + TResponseBody extends any, + TOptional extends boolean = false +>( schema: z.Schema, url: string, requestInit?: RequestInit, options?: { errorMessage?: string; + optional?: TOptional; } -): Promise { +): Promise { const response = await fetch(url, requestInit); + if ( + (!requestInit || requestInit.method === "GET") && + response.status === 404 && + options?.optional + ) { + // @ts-ignore + return; + } + if (response.status >= 400 && response.status < 500) { const body = await response.json(); diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts index b4719411c..2cc1362e6 100644 --- a/packages/trigger-sdk/src/io.ts +++ b/packages/trigger-sdk/src/io.ts @@ -1,5 +1,6 @@ import { CachedTask, + ConnectionAuth, LogLevel, Logger, RunTaskOptions, @@ -11,6 +12,13 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { webcrypto } from "node:crypto"; import { ApiClient } from "./apiClient"; import { TriggerClient } from "./triggerClient"; +import { DynamicTrigger } from "./triggers/dynamic"; +import { + ExternalSource, + ExternalSourceParams, +} from "./triggers/externalSource"; +import { EventSpecification, TriggerContext } from "./types"; +import { createIOWithIntegrations } from "./ioWithIntegrations"; export class ResumeWithTask { constructor(public task: ServerTask) {} @@ -22,6 +30,7 @@ export type IOOptions = { id: string; apiClient: ApiClient; client: TriggerClient; + context: TriggerContext; logger?: Logger; logLevel?: LogLevel; cachedTasks?: Array; @@ -34,6 +43,7 @@ export class IO { #logger: Logger; #cachedTasks: Map; #taskStorage: AsyncLocalStorage<{ taskId: string }>; + #context: TriggerContext; constructor(options: IOOptions) { this.#id = options.id; @@ -50,6 +60,7 @@ export class IO { } this.#taskStorage = new AsyncLocalStorage(); + this.#context = options.context; } async updateSource( @@ -81,92 +92,86 @@ export class IO { ); } - // TODO: use internal job system for this - // async on( - // key: string | any[], - // trigger: Trigger - // ) { - // const metadata = trigger.toJSON(); + async registerTrigger< + TTrigger extends DynamicTrigger< + EventSpecification, + ExternalSource + > + >( + key: string | any[], + trigger: TTrigger, + id: string, + params: ExternalSourceParams + ): Promise<{ id: string; key: string } | undefined> { + return await this.runTask( + key, + { + name: "register-trigger", + elements: [ + { label: "trigger", text: trigger.id }, + { label: "id", text: id }, + ], + params: params as any, + }, + async (task) => { + const registration = await this.runTask( + "register-source", + { + name: "register-source", + }, + async (subtask1) => { + return trigger.register(params); + } + ); - // return this.runTask( - // key, - // { - // name: metadata.title, - // elements: metadata.elements, - // trigger: trigger.toJSON(), - // }, - // async (task) => { - // return task.output as T; - // } - // ); - // } + const connection = await this.getAuth( + "get-auth", + registration.source.clientId + ); - // async addTriggerVariant>( - // job: Job, - // id: string, - // trigger: TTrigger - // ) { - // const metadata = trigger.toJSON(); + const io = createIOWithIntegrations( + // @ts-ignore + this, + { + integration: connection, + }, + { + integration: trigger.source.integration, + } + ); - // 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, - // } - // ); - // } - // ); + const updates = await trigger.source.register( + params, + registration, + io, + this.#context + ); - // if (subResponse1.ready) { - // return subResponse1; - // } + if (!updates) { + // TODO: do something here? + return; + } - // 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 - // // 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); - // } - // ); + return await this.updateSource("update-source", { + key: registration.source.key, + ...updates, + }); + } + ); + } - // return { ok: true }; - // } - // ); - // } + async getAuth( + key: string | any[], + clientId?: string + ): Promise { + if (!clientId) { + return; + } + + return this.runTask(key, { name: "get-auth" }, async (task) => { + return await this.#client.getAuth(clientId); + }); + } // TODO: investigate why errors from Github tasks are not being caught here async runTask( diff --git a/packages/trigger-sdk/src/ioWithIntegrations.ts b/packages/trigger-sdk/src/ioWithIntegrations.ts new file mode 100644 index 000000000..4eed29d81 --- /dev/null +++ b/packages/trigger-sdk/src/ioWithIntegrations.ts @@ -0,0 +1,88 @@ +import { ConnectionAuth } from "@trigger.dev/internal"; +import { + TriggerIntegration, + IntegrationClient, + IOWithIntegrations, + AuthenticatedTask, +} from "./integrations"; +import { IO } from "./io"; + +export function createIOWithIntegrations< + TIntegrations extends Record< + string, + TriggerIntegration> + > +>( + io: IO, + auths?: Record, + integrations?: TIntegrations +): IOWithIntegrations { + if (!integrations) { + return io as IOWithIntegrations; + } + + const connections = Object.entries(integrations).reduce( + (acc, [key, integration]) => { + const connection = auths?.[key]; + const client = + "client" in integration.client + ? integration.client.client + : connection + ? integration.client.clientFactory?.(connection) + : undefined; + + if (!client) { + return acc; + } + + const ioConnection = { + client, + } as any; + + if (integration.client.tasks) { + const tasks: Record< + string, + AuthenticatedTask + > = integration.client.tasks; + + Object.keys(tasks).forEach((taskName) => { + const authenticatedTask = tasks[taskName]; + + ioConnection[taskName] = async ( + key: string | string[], + params: any + ) => { + return await io.runTask( + key, + authenticatedTask.init(params), + async (ioTask) => { + return authenticatedTask.run(params, client, ioTask, io); + } + ); + }; + }); + } + + acc[key] = ioConnection; + + return acc; + }, + {} as any + ); + + return new Proxy(io, { + get(target, prop, receiver) { + // We can return the original io back if the prop is __io + if (prop === "__io") { + return io; + } + + if (prop in connections) { + return connections[prop]; + } + + const value = Reflect.get(target, prop, receiver); + return typeof value == "function" ? value.bind(target) : value; + }, + }) as IOWithIntegrations; +} diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index f36967dd9..9b4402f08 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -11,24 +11,20 @@ import { NormalizedResponse, RegisterSourceEvent, RegisterSourceEventSchema, + RegisterTriggerBody, RunJobBody, RunJobBodySchema, SendEvent, SourceMetadata, } from "@trigger.dev/internal"; import { ApiClient } from "./apiClient"; -import { - AuthenticatedTask, - IntegrationClient, - IOWithIntegrations, - TriggerIntegration, -} from "./integrations"; import { IO, ResumeWithTask } from "./io"; +import { createIOWithIntegrations } from "./ioWithIntegrations"; import { Job } from "./job"; -import { ContextLogger } from "./logger"; -import type { EventSpecification, Trigger, TriggerContext } from "./types"; -import { ExternalSource, HttpSourceEvent } from "./triggers/externalSource"; import { CustomTrigger } from "./triggers/customTrigger"; +import { ExternalSource, HttpSourceEvent } from "./triggers/externalSource"; +import type { EventSpecification, Trigger, TriggerContext } from "./types"; +import { DynamicTrigger } from "./triggers/dynamic"; export type TriggerClientOptions = { apiKey?: string; @@ -57,6 +53,10 @@ export class TriggerClient { response?: NormalizedResponse; } | void> > = {}; + #registeredDynamicTriggers: Record< + string, + DynamicTrigger, ExternalSource> + > = {}; #client: ApiClient; #logger: Logger; name: string; @@ -180,7 +180,7 @@ export class TriggerClient { body: { completed: results.completed, output: results.output, - executionId: execution.data.context.id, + executionId: execution.data.run.id, task: results.task, }, }; @@ -207,12 +207,14 @@ export class TriggerClient { }; const key = headers.data["x-ts-key"]; + const dynamicId = headers.data["x-ts-dynamic-id"]; const secret = headers.data["x-ts-secret"]; const params = headers.data["x-ts-params"]; const data = headers.data["x-ts-data"]; const source = { key, + dynamicId, secret, params, data, @@ -248,6 +250,10 @@ export class TriggerClient { job.trigger.attachToJob(this, job); } + attachDynamicTrigger(trigger: DynamicTrigger): void { + this.#registeredDynamicTriggers[trigger.id] = trigger; + } + attachSource(options: { key: string; source: ExternalSource; @@ -266,6 +272,9 @@ export class TriggerClient { key: options.key, params: options.params, events: [], + clientId: !options.source.integration.usesLocalAuth + ? options.source.integration.id + : undefined, }; } @@ -321,6 +330,14 @@ export class TriggerClient { }); } + async registerTrigger(id: string, options: RegisterTriggerBody) { + return this.#client.registerTrigger(this.name, id, options); + } + + async getAuth(id: string) { + return this.#client.getAuth(this.name, id); + } + authorized(apiKey: string) { const localApiKey = this.#options.apiKey ?? process.env.TRIGGER_API_KEY; @@ -346,27 +363,28 @@ export class TriggerClient { async #executeJob(execution: RunJobBody, job: Job, any>) { this.#logger.debug("executing job", { execution, job: job.toJSON() }); - const abortController = new AbortController(); + const context = this.#createRunContext(execution); const io = new IO({ - id: execution.context.id, + id: execution.run.id, cachedTasks: execution.tasks, apiClient: this.#client, logger: this.#logger, client: this, + context, }); - const ioWithConnections = this.#createIOWithIntegrations( + const ioWithConnections = createIOWithIntegrations( io, - execution, - job + execution.connections, + job.options.integrations ); try { const output = await job.options.run( job.trigger.event.parsePayload(execution.event.payload ?? {}), ioWithConnections, - this.#createJobContext(execution, io, abortController.signal) + context ); return { completed: true, output }; @@ -394,160 +412,85 @@ export class TriggerClient { } } - #createIOWithIntegrations< - TIntegrations extends Record< - string, - TriggerIntegration> - > - >( - io: IO, - run: RunJobBody, - job: Job, TIntegrations> - ): IOWithIntegrations { - const jobIntegrations = job.options.integrations; + #createRunContext(execution: RunJobBody): TriggerContext { + const { event, organization, environment, job, run } = execution; - if (!jobIntegrations) { - return io as IOWithIntegrations; - } - - const runConnections = run.connections ?? {}; - - const connections = Object.entries(jobIntegrations).reduce( - (acc, [key, jobConnection]) => { - const connection = runConnections[key]; - const client = - "client" in jobConnection.client - ? jobConnection.client.client - : jobConnection.client.clientFactory?.(connection); - - if (!client) { - return acc; - } - - const ioConnection = { - client, - } as any; - - if (jobConnection.client.tasks) { - const tasks: Record< - string, - AuthenticatedTask - > = jobConnection.client.tasks; - - Object.keys(tasks).forEach((taskName) => { - const authenticatedTask = tasks[taskName]; - - ioConnection[taskName] = async ( - key: string | string[], - params: any - ) => { - return await io.runTask( - key, - authenticatedTask.init(params), - async (ioTask) => { - return authenticatedTask.run(params, client, ioTask, io); - } - ); - }; - }); - } - - acc[key] = ioConnection; - - return acc; - }, - {} as any - ); - - return new Proxy(io, { - get(target, prop, receiver) { - if (prop in connections) { - return connections[prop]; - } - - const value = Reflect.get(target, prop, receiver); - return typeof value == "function" ? value.bind(target) : value; - }, - }) as IOWithIntegrations; - } - - #createJobContext( - execution: RunJobBody, - io: IO, - signal: AbortSignal - ): TriggerContext { return { - ...execution.context, - signal, - logger: new ContextLogger(async (level, message, data) => { - switch (level) { - case "DEBUG": { - this.#logger.debug(message, data); - break; - } - case "INFO": { - this.#logger.info(message, data); - break; - } - case "WARN": { - this.#logger.warn(message, data); - break; - } - case "ERROR": { - this.#logger.error(message, data); - break; - } - } - - await io.runTask( - [message, level], - { - name: "log", - icon: "log", - description: message, - params: data, - elements: [{ label: "Level", text: level }], - noop: true, - }, - async (task) => {} - ); - }), - wait: async (id, seconds) => { - await io.runTask( - id, - { - name: "wait", - icon: "clock", - params: { seconds }, - noop: true, - delayUntil: new Date(Date.now() + seconds * 1000), - }, - async (task) => {} - ); - }, - sendEvent: async (key, event, options) => { - return await io.runTask( - key, - { - name: "sendEvent", - params: { event, options }, - }, - async (task) => { - return await this.#client.sendEvent(event, options); - } - ); + event: { + id: event.id, + name: event.name, + context: event.context, + timestamp: event.timestamp, }, + organization, + environment, + job, + run, }; } async #handleHttpSourceRequest( - source: { key: string; secret: string; data: any; params: any }, + source: { + key: string; + dynamicId?: string; + secret: string; + data: any; + params: any; + }, sourceRequest: HttpSourceRequest ): Promise<{ response: NormalizedResponse; events: SendEvent[] }> { this.#logger.debug("Handling HTTP source request", { source, }); + if (source.dynamicId) { + const dynamicTrigger = this.#registeredDynamicTriggers[source.dynamicId]; + + if (!dynamicTrigger) { + this.#logger.debug("No dynamic trigger registered for HTTP source", { + source, + }); + + return { + response: { + status: 200, + body: { + ok: true, + }, + }, + events: [], + }; + } + + const results = await dynamicTrigger.source.handle( + source, + sourceRequest, + this.#logger + ); + + if (!results) { + return { + events: [], + response: { + status: 200, + body: { + ok: true, + }, + }, + }; + } + + return { + events: results.events, + response: results.response ?? { + status: 200, + body: { + ok: true, + }, + }, + }; + } + const handler = this.#registeredHttpSourceHandlers[source.key]; if (!handler) { diff --git a/packages/trigger-sdk/src/triggers/customTrigger.ts b/packages/trigger-sdk/src/triggers/customTrigger.ts index d67ea351b..78b72763e 100644 --- a/packages/trigger-sdk/src/triggers/customTrigger.ts +++ b/packages/trigger-sdk/src/triggers/customTrigger.ts @@ -65,18 +65,18 @@ export function customTrigger< } export function customEvent({ - schema, + payload, source, }: { - schema: z.Schema; + payload: z.Schema; source?: string; }): EventSpecification { return { name: "custom", title: "Custom Event", source: source ?? "trigger.dev", - parsePayload: (payload: any) => { - return schema.parse(payload); + parsePayload: (rawPayload: any) => { + return payload.parse(rawPayload); }, }; } diff --git a/packages/trigger-sdk/src/triggers/dynamic.ts b/packages/trigger-sdk/src/triggers/dynamic.ts index 33adb1fce..2ae21099f 100644 --- a/packages/trigger-sdk/src/triggers/dynamic.ts +++ b/packages/trigger-sdk/src/triggers/dynamic.ts @@ -1,8 +1,13 @@ -import { TriggerMetadata } from "@trigger.dev/internal"; +import { + RegisterSourceEvent, + TriggerMetadata, + deepMergeFilters, +} from "@trigger.dev/internal"; import { Job } from "../job"; import { TriggerClient } from "../triggerClient"; import { EventSpecification, Trigger } from "../types"; import { ExternalSource, ExternalSourceParams } from "./externalSource"; +import { slugifyId } from "../utils"; export type DynamicTriggerOptions< TEventSpec extends EventSpecification, @@ -10,7 +15,7 @@ export type DynamicTriggerOptions< > = { id: string; event: TEventSpec; - source?: TExternalSource; + source: TExternalSource; }; export class DynamicTrigger< @@ -20,6 +25,7 @@ export class DynamicTrigger< { #client: TriggerClient; #options: DynamicTriggerOptions; + source: TExternalSource; constructor( client: TriggerClient, @@ -27,6 +33,9 @@ export class DynamicTrigger< ) { this.#client = client; this.#options = options; + this.source = options.source; + + client.attachDynamicTrigger(this); } toJSON(): Array { @@ -38,6 +47,10 @@ export class DynamicTrigger< ]; } + get id() { + return this.#options.id; + } + get event() { return this.#options.event; } @@ -46,8 +59,29 @@ export class DynamicTrigger< return false; } - // Just an example for the types - register(params: ExternalSourceParams): void {} + async register( + params: ExternalSourceParams + ): Promise { + return this.#client.registerTrigger(this.id, { + rule: { + event: this.event.name, + source: this.event.source, + payload: deepMergeFilters( + this.source.filter(params), + this.event.filter ?? {} + ), + }, + source: { + key: slugifyId(this.source.key(params)), + channel: this.source.channel, + params, + events: [this.event.name], + clientId: !this.source.integration.usesLocalAuth + ? this.source.integration.id + : undefined, + }, + }); + } attachToJob( triggerClient: TriggerClient, diff --git a/packages/trigger-sdk/src/triggers/externalSource.ts b/packages/trigger-sdk/src/triggers/externalSource.ts index 9d5b840bb..5e71bbf7f 100644 --- a/packages/trigger-sdk/src/triggers/externalSource.ts +++ b/packages/trigger-sdk/src/triggers/externalSource.ts @@ -93,6 +93,7 @@ type HandlerFunction = ( ) => Promise<{ events: SendEvent[]; response?: NormalizedResponse } | void>; type KeyFunction = (params: TParams) => string; +type FilterFunction = (params: TParams) => EventFilter; type ExternalSourceOptions< TChannel extends ChannelNames, @@ -104,6 +105,7 @@ type ExternalSourceOptions< schema: z.Schema; integration: TIntegration; register: RegisterFunction; + filter: FilterFunction; handler: HandlerFunction; key: KeyFunction; }; @@ -136,6 +138,10 @@ export class ExternalSource< ); } + filter(params: TParams): EventFilter { + return this.options.filter(params); + } + async register( params: TParams, registerEvent: RegisterSourceEvent, @@ -204,7 +210,6 @@ export type ExternalSourceTriggerOptions< event: TEventSpecification; source: TEventSource; params: ExternalSourceParams; - filter?: EventFilter; }; export class ExternalSourceTrigger< @@ -235,7 +240,7 @@ export class ExternalSourceTrigger< rule: { event: this.event.name, payload: deepMergeFilters( - this.options.filter ?? {}, + this.options.source.filter(this.options.params), this.event.filter ?? {} ), source: this.event.source, diff --git a/packages/trigger-sdk/src/types.ts b/packages/trigger-sdk/src/types.ts index 4a20a4ceb..cf14e6c3b 100644 --- a/packages/trigger-sdk/src/types.ts +++ b/packages/trigger-sdk/src/types.ts @@ -1,6 +1,7 @@ import type { ApiEventLog, EventFilter, + RuntimeEnvironmentType, SecureString, SendEvent, SendEventOptions, @@ -13,23 +14,11 @@ import { TriggerClient } from "./triggerClient"; export type { SecureString }; export interface TriggerContext { - id: string; - version: string; - environment: string; - organization: string; - startedAt: Date; - isTest: boolean; - 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; + job: { id: string; version: string }; + environment: { slug: string; id: string; type: RuntimeEnvironmentType }; + organization: { slug: string; id: string; title: string }; + run: { id: string; isTest: boolean; startedAt: Date }; + event: { id: string; name: string; context: any; timestamp: Date }; } export interface TaskLogger {