Adding support for trigger source in the run context

.changeset/bright-horses-bathe.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Adding support for trigger source in the run context, and make sure dynamic trigger runs are preprocessed so they have a chance of populating run properties

apps/webapp/app/routes/api.v1.endpoints.ts

@@ -11,7 +11,7 @@ const BodySchema = z.object({
 });
 
 export async function action({ request }: ActionArgs) {
-  logger.info("Creating endpoint", { url: request.url });
+  logger.info("action", { url: request.url });
 
   // Ensure this is a POST request
   if (request.method.toUpperCase() !== "POST") {

apps/webapp/app/services/events/ingestSendEvent.server.ts

@@ -1,10 +1,10 @@
 import type { RawEvent, SendEventOptions } from "@trigger.dev/internal";
 import {
   $transaction,
-  PrismaClient,
   PrismaClientOrTransaction,
+  PrismaErrorSchema,
+  prisma,
 } from "~/db.server";
-import { PrismaErrorSchema, prisma } from "~/db.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { workerQueue } from "~/services/worker.server";
 
@@ -39,7 +39,8 @@ export class IngestSendEvent {
   public async call(
     environment: AuthenticatedEnvironment,
     event: RawEvent,
-    options?: SendEventOptions
+    options?: SendEventOptions,
+    sourceContext?: { id: string; metadata?: any }
   ) {
     try {
       const deliverAt = this.#calculateDeliverAt(options);
@@ -80,6 +81,7 @@ export class IngestSendEvent {
             payload: event.payload ?? {},
             context: event.context ?? {},
             source: event.source ?? "trigger.dev",
+            sourceContext,
             deliverAt: deliverAt,
             externalAccount: externalAccount
               ? {

apps/webapp/app/services/runs/performRunExecution.ts

@@ -6,6 +6,7 @@ import {
   RunJobResumeWithTask,
   RunJobRetryWithTask,
   RunJobSuccess,
+  RunSourceContextSchema,
 } from "@trigger.dev/internal";
 import { generateErrorMessage } from "zod-error";
 import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts";
@@ -242,6 +243,10 @@ export class PerformRunExecutionService {
       }
     }
 
+    const sourceContext = RunSourceContextSchema.safeParse(
+      run.event.sourceContext
+    );
+
     const { response, parser } = await client.executeJobRequest({
       event,
       job: {
@@ -270,6 +275,7 @@ export class PerformRunExecutionService {
           }
         : undefined,
       connections: connections.auth,
+      source: sourceContext.success ? sourceContext.data : undefined,
       tasks: [run.tasks, resumedTask]
         .flat()
         .filter(Boolean)

apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts

@@ -47,14 +47,12 @@ export class DeliverScheduledEventService {
           id: eventId,
           name: SCHEDULED_EVENT,
           payload,
-          context: {
-            source: {
-              id: scheduleSource.key,
-              metadata: scheduleSource.metadata,
-            },
-          },
         },
-        { accountId: scheduleSource.externalAccount?.identifier }
+        { accountId: scheduleSource.externalAccount?.identifier },
+        {
+          id: scheduleSource.key,
+          metadata: scheduleSource.metadata,
+        }
       );
 
       const invokeDispatcherService = new InvokeDispatcherService(tx);

apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts

@@ -88,9 +88,19 @@ export class DeliverHttpSourceRequestService {
     const ingestService = new IngestSendEvent();
 
     for (const event of events) {
-      await ingestService.call(httpSourceRequest.environment, event, {
-        accountId: httpSourceRequest.source.externalAccount?.identifier,
-      });
+      await ingestService.call(
+        httpSourceRequest.environment,
+        event,
+        {
+          accountId: httpSourceRequest.source.externalAccount?.identifier,
+        },
+        httpSourceRequest.source.dynamicSourceId
+          ? {
+              id: httpSourceRequest.source.dynamicSourceId,
+              metadata: httpSourceRequest.source.dynamicSourceMetadata,
+            }
+          : undefined
+      );
     }
 
     return response;

apps/webapp/app/services/sources/registerSource.server.ts

@@ -18,7 +18,8 @@ export class RegisterSourceService {
     endpointId: string,
     metadata: SourceMetadata,
     dynamicTriggerId?: string,
-    accountId?: string
+    accountId?: string,
+    dynamicSource?: { id: string; metadata: any }
   ) {
     const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({
       where: {
@@ -39,7 +40,8 @@ export class RegisterSourceService {
       endpoint.environment,
       metadata,
       dynamicTriggerId,
-      accountId
+      accountId,
+      dynamicSource
     );
   }
 
@@ -48,7 +50,8 @@ export class RegisterSourceService {
     environment: AuthenticatedEnvironment,
     metadata: SourceMetadata,
     dynamicTriggerId?: string,
-    accountId?: string
+    accountId?: string,
+    dynamicSource?: { id: string; metadata: any }
   ) {
     logger.debug("Upserting source", {
       endpoint,
@@ -57,9 +60,9 @@ export class RegisterSourceService {
       accountId,
     });
 
-    const key = dynamicTriggerId
-      ? `${dynamicTriggerId}:${metadata.key}`
-      : metadata.key;
+    const key = [dynamicTriggerId, dynamicSource?.id, metadata.key]
+      .filter(Boolean)
+      .join(":");
 
     const { id, orphanedEvents } = await $transaction(
       this.#prismaClient,
@@ -143,6 +146,8 @@ export class RegisterSourceService {
                 },
               },
             },
+            dynamicSourceId: dynamicSource?.id,
+            dynamicSourceMetadata: dynamicSource?.metadata,
           },
           update: {
             endpoint: {
@@ -151,6 +156,8 @@ export class RegisterSourceService {
               },
             },
             integration: { connect: { id: integration.id } },
+            dynamicSourceId: dynamicSource?.id,
+            dynamicSourceMetadata: dynamicSource?.metadata,
           },
           include: {
             events: true,

apps/webapp/app/services/triggers/initializeTrigger.server.ts

@@ -71,6 +71,7 @@ export class InitializeTriggerService {
       endpointSlug,
       key: payload.id,
       accountId: payload.accountId,
+      registrationMetadata: payload.metadata,
     });
 
     await this.#sendEvent.call(

apps/webapp/app/services/triggers/registerTriggerSource.server.ts

@@ -23,6 +23,7 @@ export class RegisterTriggerSourceService {
     id,
     key,
     accountId,
+    registrationMetadata,
   }: {
     environment: AuthenticatedEnvironment;
     payload: RegisterTriggerBody;
@@ -30,6 +31,7 @@ export class RegisterTriggerSourceService {
     endpointSlug: string;
     key: string;
     accountId?: string;
+    registrationMetadata?: any;
   }): Promise<RegisterSourceEvent> {
     const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({
       where: {
@@ -58,7 +60,8 @@ export class RegisterTriggerSourceService {
         endpoint.id,
         payload.source,
         dynamicTrigger.id,
-        accountId
+        accountId,
+        { id: key, metadata: registrationMetadata }
       );
 
       const eventDispatcher = await tx.eventDispatcher.upsert({
@@ -104,8 +107,11 @@ export class RegisterTriggerSourceService {
           dynamicTriggerId: dynamicTrigger.id,
           sourceId: triggerSource.id,
           eventDispatcherId: eventDispatcher.id,
+          metadata: registrationMetadata,
+        },
+        update: {
+          metadata: registrationMetadata,
         },
-        update: {},
       });
 
       const secretStore = getSecretStore(

examples/nextjs-example/src/trigger.ts

@@ -245,6 +245,8 @@ new Job(client, {
     github: githubUser,
   },
   run: async (payload, io, ctx) => {
+    await io.logger.info("user-on-issue-opened", { ctx });
+
     return await io.github.getRepo("get.repo", {
       owner: payload.repository.owner.login,
       repo: payload.repository.name,

packages/database/prisma/migrations/20230627212239_add_source_context_to_event_record/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "EventRecord" ADD COLUMN     "sourceContext" JSONB;

packages/database/prisma/migrations/20230628100705_add_metadata_to_dynamic_trigger_registrations/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "DynamicTriggerRegistration" ADD COLUMN     "metadata" JSONB;

packages/database/prisma/migrations/20230628102426_add_dynamic_source_columns_to_trigger_source/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "TriggerSource" ADD COLUMN     "dynamicSourceId" TEXT,
+ADD COLUMN     "dynamicSourceMetadata" JSONB;

packages/database/prisma/schema.prisma

@@ -577,12 +577,13 @@ enum JobStartPosition {
 }
 
 model EventRecord {
-  id        String   @id @default(cuid())
-  eventId   String
-  name      String
-  timestamp DateTime @default(now())
-  payload   Json
-  context   Json?
+  id            String   @id @default(cuid())
+  eventId       String
+  name          String
+  timestamp     DateTime @default(now())
+  payload       Json
+  context       Json?
+  sourceContext Json?
 
   source String @default("trigger.dev")
 
@@ -853,6 +854,9 @@ model TriggerSource {
   externalAccount   ExternalAccount? @relation(fields: [externalAccountId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   externalAccountId String?
 
+  dynamicSourceId       String?
+  dynamicSourceMetadata Json?
+
   active      Boolean @default(false)
   interactive Boolean @default(false)
 
@@ -900,6 +904,8 @@ model DynamicTriggerRegistration {
   source   TriggerSource @relation(fields: [sourceId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   sourceId String
 
+  metadata Json?
+
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
 

packages/internal/src/schemas/api.ts

@@ -238,6 +238,11 @@ export type RuntimeEnvironmentType = z.infer<
   typeof RuntimeEnvironmentTypeSchema
 >;
 
+export const RunSourceContextSchema = z.object({
+  id: z.string(),
+  metadata: z.any(),
+});
+
 export const RunJobBodySchema = z.object({
   event: ApiEventLogSchema,
   job: z.object({
@@ -265,6 +270,7 @@ export const RunJobBodySchema = z.object({
       metadata: z.any(),
     })
     .optional(),
+  source: RunSourceContextSchema.optional(),
   tasks: z.array(CachedTaskSchema).optional(),
   connections: z.record(ConnectionAuthSchema).optional(),
 });
@@ -496,6 +502,7 @@ export const InitializeTriggerBodySchema = z.object({
   id: z.string(),
   params: z.any(),
   accountId: z.string().optional(),
+  metadata: z.any().optional()
 });
 
 export type InitializeTriggerBody = z.infer<typeof InitializeTriggerBodySchema>;

packages/trigger-sdk/src/triggerClient.ts

@@ -638,7 +638,7 @@ export class TriggerClient {
   }
 
   #createRunContext(execution: RunJobBody): TriggerContext {
-    const { event, organization, environment, job, run } = execution;
+    const { event, organization, environment, job, run, source } = execution;
 
     return {
       event: {
@@ -652,6 +652,7 @@ export class TriggerClient {
       job,
       run,
       account: execution.account,
+      source,
     };
   }
 

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

@@ -101,6 +101,6 @@ export class DynamicTrigger<
   }
 
   get preprocessRuns() {
-    return false;
+    return true;
   }
 }

packages/trigger-sdk/src/types.ts

@@ -18,6 +18,7 @@ export interface TriggerContext {
   run: { id: string; isTest: boolean; startedAt: Date };
   event: { id: string; name: string; context: any; timestamp: Date };
   account?: { id: string; metadata?: any };
+  source?: { id: string; metadata?: any };
 }
 
 export interface TriggerPreprocessContext {