feat(events): phase 8.7 — event-system reference project

Demo project showcasing all event system features:
- events.ts: event definitions with schemas, rate limits
- basic-subscribers.ts: fan-out to multiple tasks
- filtered-subscribers.ts: content-based filtering
- pattern-subscribers.ts: wildcard patterns (*, #)
- publish-and-wait.ts: scatter-gather orchestration
- consumer-groups.ts: load-balanced event handling
- ordering.ts: sequential processing per entity

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: giovaborgogno

references/event-system/.gitignore

@@ -0,0 +1,2 @@
+node_modules
+.trigger

references/event-system/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "references-event-system",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "trigger dev",
+    "deploy": "trigger deploy"
+  },
+  "devDependencies": {
+    "trigger.dev": "workspace:*"
+  },
+  "dependencies": {
+    "@trigger.dev/build": "workspace:*",
+    "@trigger.dev/sdk": "workspace:*",
+    "zod": "3.25.76"
+  }
+}

references/event-system/src/index.ts

@@ -0,0 +1 @@
+export {};

references/event-system/src/trigger/basic-subscribers.ts

@@ -0,0 +1,51 @@
+import { task, logger } from "@trigger.dev/sdk";
+import { orderCreated, orderShipped } from "./events";
+
+// ---- Basic Fan-out: Multiple tasks subscribe to the same event ----
+
+/** Send a confirmation email when an order is created */
+export const sendConfirmationEmail = task({
+  id: "send-confirmation-email",
+  on: orderCreated,
+  run: async (payload) => {
+    logger.info("Sending confirmation email", {
+      orderId: payload.orderId,
+      customerId: payload.customerId,
+    });
+
+    // Simulate email sending
+    return { sent: true, to: payload.customerId };
+  },
+});
+
+/** Update inventory when an order is created */
+export const updateInventory = task({
+  id: "update-inventory",
+  on: orderCreated,
+  run: async (payload) => {
+    logger.info("Updating inventory", {
+      orderId: payload.orderId,
+      itemCount: payload.items.length,
+    });
+
+    for (const item of payload.items) {
+      logger.info(`Adjusting stock: ${item.sku} -${item.qty}`);
+    }
+
+    return { adjusted: payload.items.length };
+  },
+});
+
+/** Notify customer when order is shipped */
+export const notifyShipped = task({
+  id: "notify-shipped",
+  on: orderShipped,
+  run: async (payload) => {
+    logger.info("Order shipped notification", {
+      orderId: payload.orderId,
+      tracking: payload.trackingNumber,
+    });
+
+    return { notified: true };
+  },
+});

references/event-system/src/trigger/consumer-groups.ts

@@ -0,0 +1,44 @@
+import { task, logger } from "@trigger.dev/sdk";
+import { userActivity } from "./events";
+
+// ---- Consumer Groups: Load-balanced event handling ----
+// Within a consumer group, only ONE task receives each event.
+
+export const activityProcessorA = task({
+  id: "activity-processor-a",
+  on: userActivity,
+  consumerGroup: "activity-processors",
+  run: async (payload) => {
+    logger.info("Processor A handling activity", {
+      userId: payload.userId,
+      action: payload.action,
+    });
+    return { processor: "A", userId: payload.userId };
+  },
+});
+
+export const activityProcessorB = task({
+  id: "activity-processor-b",
+  on: userActivity,
+  consumerGroup: "activity-processors",
+  run: async (payload) => {
+    logger.info("Processor B handling activity", {
+      userId: payload.userId,
+      action: payload.action,
+    });
+    return { processor: "B", userId: payload.userId };
+  },
+});
+
+// This task is NOT in the consumer group — it receives ALL events
+export const activityAnalytics = task({
+  id: "activity-analytics",
+  on: userActivity,
+  run: async (payload) => {
+    logger.info("Analytics: recording activity", {
+      userId: payload.userId,
+      action: payload.action,
+    });
+    return { recorded: true };
+  },
+});

references/event-system/src/trigger/events.ts

@@ -0,0 +1,38 @@
+import { event } from "@trigger.dev/sdk";
+import { z } from "zod";
+
+// ---- Event Definitions ----
+
+/** Published when an order is placed */
+export const orderCreated = event({
+  id: "order.created",
+  schema: z.object({
+    orderId: z.string(),
+    amount: z.number(),
+    customerId: z.string(),
+    items: z.array(z.object({ sku: z.string(), qty: z.number() })),
+  }),
+});
+
+/** Published when an order is shipped */
+export const orderShipped = event({
+  id: "order.shipped",
+  schema: z.object({
+    orderId: z.string(),
+    trackingNumber: z.string(),
+  }),
+});
+
+/** Published for any user action (rate-limited) */
+export const userActivity = event({
+  id: "user.activity",
+  schema: z.object({
+    userId: z.string(),
+    action: z.string(),
+    timestamp: z.string(),
+  }),
+  rateLimit: {
+    limit: 100,
+    window: "1m",
+  },
+});

references/event-system/src/trigger/filtered-subscribers.ts

@@ -0,0 +1,35 @@
+import { task, logger } from "@trigger.dev/sdk";
+import { orderCreated } from "./events";
+
+// ---- Content-based Filtering: Only receive events that match ----
+
+/** Only handles high-value orders (amount >= 1000) */
+export const highValueOrderHandler = task({
+  id: "high-value-order",
+  on: orderCreated,
+  filter: {
+    amount: [{ $gte: 1000 }],
+  },
+  run: async (payload) => {
+    logger.info("High-value order detected!", {
+      orderId: payload.orderId,
+      amount: payload.amount,
+    });
+
+    // Alert VIP team, apply special handling, etc.
+    return { flagged: true, amount: payload.amount };
+  },
+});
+
+/** Only handles orders from a specific customer */
+export const vipCustomerHandler = task({
+  id: "vip-customer-handler",
+  on: orderCreated,
+  filter: {
+    customerId: ["customer-vip-001", "customer-vip-002"],
+  },
+  run: async (payload) => {
+    logger.info("VIP customer order", { customerId: payload.customerId });
+    return { vip: true };
+  },
+});

references/event-system/src/trigger/ordering.ts

@@ -0,0 +1,39 @@
+import { task, logger } from "@trigger.dev/sdk";
+import { orderCreated } from "./events";
+
+// ---- Ordering Keys: Sequential processing per entity ----
+
+/**
+ * This publisher uses ordering keys to ensure events for the same customer
+ * are processed sequentially (no concurrent runs per customer).
+ */
+export const placeOrder = task({
+  id: "place-order",
+  run: async (payload: {
+    orderId: string;
+    amount: number;
+    customerId: string;
+  }) => {
+    logger.info("Publishing order with ordering key", {
+      orderId: payload.orderId,
+      customerId: payload.customerId,
+    });
+
+    const result = await orderCreated.publish(
+      {
+        orderId: payload.orderId,
+        amount: payload.amount,
+        customerId: payload.customerId,
+        items: [{ sku: "ITEM-001", qty: 1 }],
+      },
+      {
+        // Events for the same customer are processed one at a time
+        orderingKey: payload.customerId,
+        // Prevent duplicate publishes
+        idempotencyKey: `order-${payload.orderId}`,
+      }
+    );
+
+    return { eventId: result.id, runs: result.runs.length };
+  },
+});

references/event-system/src/trigger/pattern-subscribers.ts

@@ -0,0 +1,24 @@
+import { events, task, logger } from "@trigger.dev/sdk";
+
+// ---- Wildcard Pattern Subscriptions ----
+
+/** Catches all order.* events (order.created, order.shipped, etc.) */
+export const orderAuditLog = task({
+  id: "order-audit-log",
+  on: events.match("order.*"),
+  run: async (payload) => {
+    // payload is `unknown` for pattern subscriptions
+    logger.info("Order event received", { payload });
+    return { logged: true };
+  },
+});
+
+/** Catches all user.# events (user.activity, user.profile.updated, etc.) */
+export const userEventTracker = task({
+  id: "user-event-tracker",
+  on: events.match("user.#"),
+  run: async (payload) => {
+    logger.info("User event tracked", { payload });
+    return { tracked: true };
+  },
+});

references/event-system/src/trigger/publish-and-wait.ts

@@ -0,0 +1,45 @@
+import { task, logger } from "@trigger.dev/sdk";
+import { orderCreated, orderShipped } from "./events";
+
+// ---- Publish-and-Wait: Fan-out then collect results ----
+
+/**
+ * Orchestrator task that publishes an event and waits for all subscribers
+ * to finish before proceeding (scatter-gather pattern).
+ */
+export const processOrder = task({
+  id: "process-order",
+  run: async (payload: { orderId: string; amount: number; customerId: string }) => {
+    logger.info("Starting order processing", { orderId: payload.orderId });
+
+    // Publish and wait for ALL subscribers (sendConfirmationEmail, updateInventory, etc.)
+    const result = await orderCreated.publishAndWait({
+      orderId: payload.orderId,
+      amount: payload.amount,
+      customerId: payload.customerId,
+      items: [{ sku: "WIDGET-001", qty: 2 }],
+    });
+
+    logger.info("All subscribers completed", {
+      eventId: result.id,
+      subscriberCount: Object.keys(result.results).length,
+    });
+
+    // Check results from each subscriber
+    for (const [taskSlug, runResult] of Object.entries(result.results)) {
+      if (runResult.ok) {
+        logger.info(`${taskSlug}: success`, { output: runResult.output });
+      } else {
+        logger.error(`${taskSlug}: failed`, { error: runResult.error });
+      }
+    }
+
+    // Continue with next step: publish shipped event
+    await orderShipped.publish({
+      orderId: payload.orderId,
+      trackingNumber: `TRK-${Date.now()}`,
+    });
+
+    return { orderId: payload.orderId, status: "completed" };
+  },
+});