Dynamic triggers now can be registered at runtime and trigger jobs

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);
+}

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 });
+  }
+}

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 });

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,
     });

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;

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)),

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<string, ApiConnection>, 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,
       });
 

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<TSchema extends z.ZodFirstPartySchemaTypes>(
-    schema: TSchema,
-    key: string
-  ): Promise<z.infer<TSchema> | undefined>;
+  getSecret<T>(schema: z.Schema<T>, key: string): Promise<T | undefined>;
   setSecret<T extends object>(key: string, value: T): Promise<void>;
 }
 
@@ -31,13 +28,20 @@ export class SecretStore implements ASecretStore {
     }
   }
 
-  getSecret<TSchema extends z.ZodFirstPartySchemaTypes>(
-    schema: TSchema,
-    key: string
-  ): Promise<z.TypeOf<TSchema> | undefined> {
+  getSecret<T>(schema: z.Schema<T>, key: string): Promise<T | undefined> {
     return this.#provider.getSecret(schema, key);
   }
 
+  async getSecretOrThrow<T>(schema: z.Schema<T>, key: string): Promise<T> {
+    const value = await this.getSecret(schema, key);
+
+    if (!value) {
+      throw new Error(`Unable to find secret ${key} in ${this.provider}`);
+    }
+
+    return value;
+  }
+
   setSecret<T extends object>(key: string, value: T): Promise<void> {
     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<TSchema extends z.ZodFirstPartySchemaTypes>(
-    schema: TSchema,
-    key: string
-  ): Promise<z.TypeOf<TSchema> | undefined> {
+  async getSecret<T>(schema: z.Schema<T>, key: string): Promise<T | undefined> {
     const secret = await prisma.secretStore.findUnique({
       where: {
         key,

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;

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,

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;
   }
 }

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<RegisterSourceEvent> {
+    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: [],
+    };
+  }
+}

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() }),

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;

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;

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)
 

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"
+    }
+  ]
+}

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,

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<StarCreatedEvent> = {
   parsePayload: (payload) => payload as StarCreatedEvent,
 };
 
+const onNewRepository: EventSpecification<RepositoryCreatedEvent> = {
+  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<typeof createRepoEventSource>) {
       event,
       params: { repo },
       source,
-      filter: {
-        repository: {
-          full_name: [repo],
-        },
-      },
     });
   };
 }
@@ -168,11 +175,6 @@ function createOrgTrigger(source: ReturnType<typeof createOrgEventSource>) {
       event,
       params: { org },
       source,
-      filter: {
-        organization: {
-          login: [org],
-        },
-      },
     });
   };
 }

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;

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<typeof RegisterTriggerSourceSchema>;
@@ -91,6 +93,7 @@ export type HttpSourceRequest = z.infer<typeof HttpSourceRequestSchema>;
 
 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<typeof SourceMetadataSchema>;
@@ -192,20 +196,38 @@ export const DeliverEventResponseSchema = z.object({
 
 export type DeliverEventResponse = z.infer<typeof DeliverEventResponseSchema>;
 
+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<typeof RegisterTriggerBodySchema>;

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<RegisterSourceEvent> {
+    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<TResponseBody extends any>(
+async function zodfetch<
+  TResponseBody extends any,
+  TOptional extends boolean = false
+>(
   schema: z.Schema<TResponseBody>,
   url: string,
   requestInit?: RequestInit,
   options?: {
     errorMessage?: string;
+    optional?: TOptional;
   }
-): Promise<TResponseBody> {
+): Promise<TOptional extends true ? TResponseBody | undefined : TResponseBody> {
   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();
 

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<CachedTask>;
@@ -34,6 +43,7 @@ export class IO {
   #logger: Logger;
   #cachedTasks: Map<string, CachedTask>;
   #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<T extends SerializableJson | void = void>(
-  //   key: string | any[],
-  //   trigger: Trigger<T>
-  // ) {
-  //   const metadata = trigger.toJSON();
+  async registerTrigger<
+    TTrigger extends DynamicTrigger<
+      EventSpecification<any>,
+      ExternalSource<any, any, any>
+    >
+  >(
+    key: string | any[],
+    trigger: TTrigger,
+    id: string,
+    params: ExternalSourceParams<TTrigger["source"]>
+  ): 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<T>(
-  //     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<TTrigger extends Trigger<any>>(
-  //   job: Job<TTrigger, any>,
-  //   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<ConnectionAuth | undefined> {
+    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<T extends SerializableJson | void = void>(

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<IntegrationClient<any, any>>
+  >
+>(
+  io: IO,
+  auths?: Record<string, ConnectionAuth | undefined>,
+  integrations?: TIntegrations
+): IOWithIntegrations<TIntegrations> {
+  if (!integrations) {
+    return io as IOWithIntegrations<TIntegrations>;
+  }
+
+  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<any, any, any>
+        > = 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<TIntegrations>;
+}

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<EventSpecification<any>, ExternalSource<any, any, any>>
+  > = {};
   #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<any, any>): void {
+    this.#registeredDynamicTriggers[trigger.id] = trigger;
+  }
+
   attachSource(options: {
     key: string;
     source: ExternalSource<any, any>;
@@ -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<Trigger<any>, 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<IntegrationClient<any, any>>
-    >
-  >(
-    io: IO,
-    run: RunJobBody,
-    job: Job<Trigger<any>, TIntegrations>
-  ): IOWithIntegrations<TIntegrations> {
-    const jobIntegrations = job.options.integrations;
+  #createRunContext(execution: RunJobBody): TriggerContext {
+    const { event, organization, environment, job, run } = execution;
 
-    if (!jobIntegrations) {
-      return io as IOWithIntegrations<TIntegrations>;
-    }
-
-    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<any, any, any>
-          > = 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<TIntegrations>;
-  }
-
-  #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) {

packages/trigger-sdk/src/triggers/customTrigger.ts

@@ -65,18 +65,18 @@ export function customTrigger<
 }
 
 export function customEvent<TEvent>({
-  schema,
+  payload,
   source,
 }: {
-  schema: z.Schema<TEvent>;
+  payload: z.Schema<TEvent>;
   source?: string;
 }): EventSpecification<TEvent> {
   return {
     name: "custom",
     title: "Custom Event",
     source: source ?? "trigger.dev",
-    parsePayload: (payload: any) => {
-      return schema.parse(payload);
+    parsePayload: (rawPayload: any) => {
+      return payload.parse(rawPayload);
     },
   };
 }

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<any>,
@@ -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<TEventSpec, TExternalSource>;
+  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<TriggerMetadata> {
@@ -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<TExternalSource>): void {}
+  async register(
+    params: ExternalSourceParams<TExternalSource>
+  ): Promise<RegisterSourceEvent> {
+    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,

packages/trigger-sdk/src/triggers/externalSource.ts

@@ -93,6 +93,7 @@ type HandlerFunction<TChannel extends ChannelNames, TParams extends any> = (
 ) => Promise<{ events: SendEvent[]; response?: NormalizedResponse } | void>;
 
 type KeyFunction<TParams extends any> = (params: TParams) => string;
+type FilterFunction<TParams extends any> = (params: TParams) => EventFilter;
 
 type ExternalSourceOptions<
   TChannel extends ChannelNames,
@@ -104,6 +105,7 @@ type ExternalSourceOptions<
   schema: z.Schema<TParams>;
   integration: TIntegration;
   register: RegisterFunction<TIntegration, TParams, TChannel>;
+  filter: FilterFunction<TParams>;
   handler: HandlerFunction<TChannel, TParams>;
   key: KeyFunction<TParams>;
 };
@@ -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<TEventSource>;
-  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,

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<void>;
-  // TODO: move this to io
-  sendEvent(
-    key: string | any[],
-    event: SendEvent,
-    options?: SendEventOptions
-  ): Promise<ApiEventLog>;
+  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 {