New unified event system is creating workflow runs from events

apps/webapp/app/models/workflowRun.server.ts

@@ -5,16 +5,24 @@ import type {
   LogMessageSchema,
   WaitSchema,
 } from "@trigger.dev/common-schemas";
+import { ulid } from "ulid";
 import type { z } from "zod";
 import { prisma } from "~/db.server";
 import { IngestEvent } from "~/services/events/ingest.server";
-import type { Organization } from "./organization.server";
-import type { User } from "./user.server";
-import type { Workflow } from "./workflow.server";
 
 type WorkflowRunStatus = WorkflowRun["status"];
 export type { WorkflowRun, WorkflowRunStep, WorkflowRunStatus };
 
+export async function findWorklowRunById(id: string) {
+  return prisma.workflowRun.findUnique({
+    where: { id },
+    include: {
+      event: true,
+      environment: true,
+    },
+  });
+}
+
 export async function startWorkflowRun(id: string, apiKey: string) {
   const workflowRun = await findWorkflowRunScopedToApiKey(id, apiKey);
 
@@ -94,9 +102,16 @@ export async function triggerEventInRun(
   const ingestService = new IngestEvent();
 
   await ingestService.call(
-    event,
-    workflowRun.environment.organization,
-    workflowRun.environment
+    {
+      id: ulid(),
+      name: event.name,
+      type: "CUSTOM_EVENT",
+      service: "trigger",
+      payload: event.payload,
+      context: event.context,
+      apiKey: workflowRun.environment.apiKey,
+    },
+    workflowRun.environment.organization
   );
 }
 

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

@@ -0,0 +1,176 @@
+import type { EventRule, TriggerEvent } from ".prisma/client";
+import { EventFilterSchema } from "@trigger.dev/common-schemas";
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+import type { RuntimeEnvironment } from "~/models/runtimeEnvironment.server";
+import type { Workflow } from "~/models/workflow.server";
+import { internalPubSub } from "../messageBroker.server";
+
+export class DispatchEvent {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    const event = await this.#prismaClient.triggerEvent.findUnique({
+      where: {
+        id,
+      },
+    });
+
+    if (!event) {
+      throw new Error("Event not found");
+    }
+
+    console.log(
+      `Dispatching event ${event.id}, type = ${event.type}, name = ${event.name}, service = ${event.service}, environment = ${event.environmentId}`
+    );
+
+    const eventRules = await this.#prismaClient.eventRule.findMany({
+      where: {
+        organizationId: event.organizationId ?? undefined,
+        environmentId: event.environmentId ?? undefined,
+        type: event.type,
+      },
+      include: {
+        workflow: true,
+        environment: true,
+      },
+    });
+
+    const matcher = new EventMatcher(event);
+
+    const matchingEventRules = eventRules.filter((eventRule) => {
+      return matcher.matches(eventRule);
+    });
+
+    console.log(
+      `Found ${matchingEventRules.length} matching event rules for event ${event.id}`
+    );
+
+    const dispatchWorkflowRun = new DispatchWorkflowRun();
+
+    await Promise.all(
+      matchingEventRules.map((eventRule) => {
+        return dispatchWorkflowRun.call(
+          eventRule.workflow,
+          eventRule,
+          event,
+          eventRule.environment
+        );
+      })
+    );
+
+    await this.#prismaClient.triggerEvent.update({
+      where: {
+        id,
+      },
+      data: {
+        dispatchedAt: new Date(),
+        status: "DISPATCHED",
+      },
+    });
+  }
+}
+
+class EventMatcher {
+  #json: any;
+
+  constructor(event: TriggerEvent) {
+    this.#json = this.#createEventJsonFromEvent(event);
+  }
+
+  public matches(eventRule: EventRule) {
+    const filter = this.#parseFilter(eventRule);
+
+    if (!filter.success) {
+      return false;
+    }
+
+    return patternMatches(this.#json, filter.data);
+  }
+
+  #parseFilter(eventRule: EventRule) {
+    const filter = EventFilterSchema.safeParse(eventRule.filter);
+
+    return filter;
+  }
+
+  #createEventJsonFromEvent(event: TriggerEvent) {
+    return {
+      event: event.name,
+      service: event.service,
+      payload: event.payload,
+    };
+  }
+}
+
+function patternMatches(payload: any, pattern: any): boolean {
+  for (const [key, value] of Object.entries(pattern)) {
+    if (Array.isArray(value)) {
+      if (!value.includes(payload[key])) {
+        return false;
+      }
+    } else if (typeof value === "object") {
+      if (!patternMatches(payload[key], value)) {
+        return false;
+      }
+    }
+  }
+  return true;
+}
+
+class DispatchWorkflowRun {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    workflow: Workflow,
+    eventRule: EventRule,
+    event: TriggerEvent,
+    environment: RuntimeEnvironment
+  ) {
+    if (!workflow) {
+      throw new Error("Workflow not found");
+    }
+
+    const workflowRun = await this.#prismaClient.workflowRun.create({
+      data: {
+        workflow: {
+          connect: {
+            id: workflow.id,
+          },
+        },
+        environment: {
+          connect: {
+            id: environment.id,
+          },
+        },
+        event: {
+          connect: {
+            id: event.id,
+          },
+        },
+        eventRule: {
+          connect: {
+            id: eventRule.id,
+          },
+        },
+        status: "PENDING",
+      },
+    });
+
+    console.log(
+      `Created workflow run ${workflowRun.id} for event rule ${eventRule.id}`
+    );
+
+    await internalPubSub.publish("WORKFLOW_RUN_CREATED", {
+      id: workflowRun.id,
+    });
+  }
+}

apps/webapp/app/services/messageBroker.server.ts

@@ -1,3 +1,4 @@
+import { JsonSchema } from "@trigger.dev/common-schemas";
 import type { CoordinatorCatalog, PlatformCatalog } from "internal-platform";
 import {
   coordinatorCatalog,
@@ -13,11 +14,13 @@ import { env } from "~/env.server";
 import {
   completeWorkflowRun,
   failWorkflowRun,
+  findWorklowRunById,
   initiateWaitInRun,
   logMessageInRun,
   startWorkflowRun,
   triggerEventInRun,
 } from "~/models/workflowRun.server";
+import { DispatchEvent } from "./events/dispatch.server";
 import { RegisterExternalSource } from "./externalSources/registerExternalSource.server";
 
 let pulsarClient: PulsarClient;
@@ -164,6 +167,10 @@ const InternalCatalog = {
     data: z.object({ id: z.string() }),
     properties: z.object({}),
   },
+  WORKFLOW_RUN_CREATED: {
+    data: z.object({ id: z.string() }),
+    properties: z.object({}),
+  },
   EXTERNAL_SOURCE_UPSERTED: {
     data: z.object({ id: z.string() }),
     properties: z.object({}),
@@ -191,12 +198,38 @@ async function createInternalPubSub() {
         return isRegistered; // Returning true will mean we don't retry
       },
       EVENT_CREATED: async (id, data, properties) => {
-        // TODO: this is where we will need to handle the event and find all the event rules that match it
-        // If the event has an environment associated with it, then query like this:
-        // SELECT * FROM event_rules WHERE org_id = ${event.orgId} AND environmentId = ${event.environmentId} AND type = ${event.type}
-        // If the event does not have an environment, then query like this:
-        // SELECT * FROM event_rules WHERE org_id = ${event.orgId} AND type = ${event.type}
-        // For each matching event rule, we need to create a new workflow run and publish that to the trigger publisher
+        const service = new DispatchEvent();
+
+        try {
+          await service.call(data.id);
+        } catch (error) {
+          console.log(error);
+          return false;
+        }
+
+        return true;
+      },
+      WORKFLOW_RUN_CREATED: async (id, data, properties) => {
+        const run = await findWorklowRunById(data.id);
+
+        if (!run) {
+          return true;
+        }
+
+        await triggerPublisher.publish(
+          "TRIGGER_WORKFLOW",
+          {
+            id: run.id,
+            input: JsonSchema.parse(run.event.payload),
+            context: JsonSchema.parse(run.event.context),
+          },
+          {
+            "x-api-key": run.environment.apiKey,
+            "x-org-id": run.environment.organizationId,
+            "x-workflow-id": run.workflowId,
+            "x-env": run.environment.slug,
+          }
+        );
 
         return true;
       },

apps/webapp/app/services/workflows/registerWorkflow.server.ts

@@ -81,13 +81,13 @@ export class RegisterWorkflow {
         },
       },
       update: {
-        rule: payload.trigger.rule,
+        filter: payload.trigger.filter,
       },
       create: {
         workflowId: workflow.id,
         environmentId: environment.id,
         organizationId: organization.id,
-        rule: payload.trigger.rule,
+        filter: payload.trigger.filter,
         type: payload.trigger.type,
         trigger: payload.trigger,
       },

apps/webapp/prisma/migrations/20221227182358_rename_rule_to_filter/migration.sql

@@ -0,0 +1,10 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `rule` on the `EventRule` table. All the data in the column will be lost.
+  - Added the required column `filter` to the `EventRule` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "EventRule" DROP COLUMN "rule",
+ADD COLUMN     "filter" JSONB NOT NULL;

apps/webapp/prisma/migrations/20221227191746_renamed_processed_to_dispatch_for_events/migration.sql

@@ -0,0 +1,21 @@
+/*
+  Warnings:
+
+  - The values [PROCESSED] on the enum `TriggerEventStatus` will be removed. If these variants are still used in the database, this will fail.
+  - You are about to drop the column `processedAt` on the `TriggerEvent` table. All the data in the column will be lost.
+
+*/
+-- AlterEnum
+BEGIN;
+CREATE TYPE "TriggerEventStatus_new" AS ENUM ('PENDING', 'DISPATCHED');
+ALTER TABLE "TriggerEvent" ALTER COLUMN "status" DROP DEFAULT;
+ALTER TABLE "TriggerEvent" ALTER COLUMN "status" TYPE "TriggerEventStatus_new" USING ("status"::text::"TriggerEventStatus_new");
+ALTER TYPE "TriggerEventStatus" RENAME TO "TriggerEventStatus_old";
+ALTER TYPE "TriggerEventStatus_new" RENAME TO "TriggerEventStatus";
+DROP TYPE "TriggerEventStatus_old";
+ALTER TABLE "TriggerEvent" ALTER COLUMN "status" SET DEFAULT 'PENDING';
+COMMIT;
+
+-- AlterTable
+ALTER TABLE "TriggerEvent" DROP COLUMN "processedAt",
+ADD COLUMN     "dispatchedAt" TIMESTAMP(3);

apps/webapp/prisma/schema.prisma

@@ -137,7 +137,7 @@ model EventRule {
   environment   RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   environmentId String
 
-  rule    Json
+  filter  Json
   trigger Json
 
   createdAt DateTime @default(now())
@@ -215,14 +215,14 @@ model TriggerEvent {
   status      TriggerEventStatus @default(PENDING)
   WorkflowRun WorkflowRun[]
 
-  createdAt   DateTime  @default(now())
-  updatedAt   DateTime  @updatedAt
-  processedAt DateTime?
+  createdAt    DateTime  @default(now())
+  updatedAt    DateTime  @updatedAt
+  dispatchedAt DateTime?
 }
 
 enum TriggerEventStatus {
   PENDING
-  PROCESSED
+  DISPATCHED
 }
 
 model WorkflowRun {

packages/common-schemas/src/events.ts

@@ -22,8 +22,8 @@ const EventMatcherSchema = z.union([
 ]);
 type EventMatcher = z.infer<typeof EventMatcherSchema>;
 
-export type EventRule = { [key: string]: EventMatcher | EventRule };
+export type EventFilter = { [key: string]: EventMatcher | EventFilter };
 
-export const EventRuleSchema: z.ZodType<EventRule> = z.lazy(() =>
-  z.record(z.union([EventMatcherSchema, EventRuleSchema]))
+export const EventFilterSchema: z.ZodType<EventFilter> = z.lazy(() =>
+  z.record(z.union([EventMatcherSchema, EventFilterSchema]))
 );

packages/common-schemas/src/triggers.ts

@@ -1,19 +1,19 @@
 import { z } from "zod";
-import { EventRuleSchema } from "./events";
+import { EventFilterSchema } from "./events";
 import { JsonSchema } from "./json";
 
 export const CustomEventTriggerSchema = z.object({
   type: z.literal("CUSTOM_EVENT"),
   service: z.literal("trigger"),
   name: z.string(),
-  rule: EventRuleSchema,
+  filter: EventFilterSchema,
 });
 
 export const WebhookEventTriggerSchema = z.object({
   type: z.literal("WEBHOOK"),
   service: z.string(),
   name: z.string(),
-  rule: EventRuleSchema,
+  filter: EventFilterSchema,
   source: JsonSchema,
 });
 
@@ -21,14 +21,14 @@ export const HttpEventTriggerSchema = z.object({
   type: z.literal("HTTP_ENDPOINT"),
   service: z.literal("trigger"),
   name: z.string(),
-  rule: EventRuleSchema,
+  filter: EventFilterSchema,
 });
 
 export const ScheduledEventTriggerSchema = z.object({
   type: z.literal("SCHEDULE"),
   service: z.literal("trigger"),
   name: z.string(),
-  rule: EventRuleSchema,
+  filter: EventFilterSchema,
 });
 
 export const TriggerMetadataSchema = z.discriminatedUnion("type", [

packages/trigger-integrations/src/integrations/github/webhooks.ts

@@ -9,7 +9,7 @@ export function repoIssueEvent(params: {
       type: "WEBHOOK",
       service: "github",
       name: "On Issue Event",
-      rule: {
+      filter: {
         service: ["github"],
         payload: {
           repository: {
@@ -37,7 +37,7 @@ export function orgIssueEvent(params: {
       type: "WEBHOOK",
       service: "github",
       name: "On Issue Event",
-      rule: {
+      filter: {
         service: ["github"],
         payload: {
           organizaton: {

packages/trigger-sdk/src/events.ts

@@ -1,10 +1,10 @@
 import {
-  EventRuleSchema,
+  EventFilterSchema,
   TriggerMetadataSchema,
 } from "@trigger.dev/common-schemas";
 import { z } from "zod";
 
-export type EventRule = z.infer<typeof EventRuleSchema>;
+export type EventRule = z.infer<typeof EventFilterSchema>;
 
 export type TriggerEvent<TSchema extends z.ZodTypeAny> = {
   metadata: z.infer<typeof TriggerMetadataSchema>;
@@ -24,7 +24,7 @@ export function customEvent<TSchema extends z.ZodTypeAny>(
       type: "CUSTOM_EVENT",
       service: "trigger",
       name: options.name,
-      rule: { name: [options.name] },
+      filter: { event: [options.name] },
     },
     schema: options.schema,
   };