fix(events): address all audit findings — security, prod-readiness, tests, docs

giovaborgogno · claude · giovaborgogno · commit 7493b863e88e · 2026-03-04T20:27:35.000-08:00
HIGH fixes:
- Change changeset bump levels from patch to minor (new feature)
- EventLogWriter: use logger.error instead of logger.warn for ClickHouse failures
- Log warning when InMemory rate limiter used in production (no RATE_LIMIT_REDIS_HOST)
- Fix sendEmail naming collision in SKILL.md docs

MEDIUM fixes:
- Document consumerRateLimit, metrics endpoint, and DLQ config in events.md
- Add z.enum() validation on DLQ status query param (was unvalidated cast)
- Add limit param validation (Math.max/min) on DLQ list route
- Add fail-open try/catch around rate limiter checks (Redis down → allow publish)
- Add defensive safety comment on ClickHouse interval interpolation in stats route

Tests added:
- DLQ retry: nonexistent ID, already-discarded, retryAll empty
- SchemaRegistry DB: registerSchema, upsert, getSchema latest/versioned/null, listSchemas
- PublishEvent: payload size limit (512KB → 413), per-subscriber rate limit skipping

LOW fixes:
- Remove dead EventPatternSource type from core
- Make CleanupStaleSubscriptionsService extend BaseService
- Drop redundant 2-col EventSubscription index (3-col supersedes it)
- Parallelize fan-out with Promise.allSettled for better throughput

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/.changeset/event-dead-letter-queue.md b/.changeset/event-dead-letter-queue.md
@@ -1,6 +1,6 @@
 ---
-"@trigger.dev/core": patch
-"trigger.dev": patch
+"@trigger.dev/core": minor
+"trigger.dev": minor
 ---
 
 Add dead letter queue for failed event-triggered task runs
diff --git a/.changeset/event-docs-cli-dx.md b/.changeset/event-docs-cli-dx.md
@@ -1,6 +1,6 @@
 ---
-"@trigger.dev/sdk": patch
-"trigger.dev": patch
+"@trigger.dev/sdk": minor
+"trigger.dev": minor
 ---
 
 Add event system documentation, CLI commands, and developer experience improvements.
diff --git a/.changeset/event-observability-dx.md b/.changeset/event-observability-dx.md
@@ -1,8 +1,8 @@
 ---
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
-"@internal/clickhouse": patch
-"apps-webapp": patch
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"@internal/clickhouse": minor
+"apps-webapp": minor
 ---
 
 Add observability and developer experience improvements to the event system.
diff --git a/.changeset/event-ordering-consumer-groups.md b/.changeset/event-ordering-consumer-groups.md
@@ -1,7 +1,7 @@
 ---
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
-"trigger.dev": patch
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"trigger.dev": minor
 ---
 
 Add ordering keys and consumer groups for event subscriptions
diff --git a/.changeset/event-persistence-replay.md b/.changeset/event-persistence-replay.md
@@ -1,6 +1,6 @@
 ---
-"@trigger.dev/core": patch
-"trigger.dev": patch
+"@trigger.dev/core": minor
+"trigger.dev": minor
 ---
 
 Add event persistence in ClickHouse and replay API for pub/sub events
diff --git a/.changeset/event-publish-and-wait.md b/.changeset/event-publish-and-wait.md
@@ -1,7 +1,7 @@
 ---
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
-"apps-webapp": patch
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"apps-webapp": minor
 ---
 
 Add publishAndWait support to the event system. Events can now be published
diff --git a/.changeset/event-rate-limiting.md b/.changeset/event-rate-limiting.md
@@ -1,8 +1,8 @@
 ---
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
-"@trigger.dev/database": patch
-"apps-webapp": patch
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"@trigger.dev/database": minor
+"apps-webapp": minor
 ---
 
 Add per-event rate limiting to the pub/sub system. Events can now be configured
diff --git a/.changeset/event-schema-registry.md b/.changeset/event-schema-registry.md
@@ -1,7 +1,7 @@
 ---
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
-"trigger.dev": patch
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"trigger.dev": minor
 ---
 
 Add event schema registry with versioning, validation, and discovery API endpoints
diff --git a/.changeset/event-smart-routing.md b/.changeset/event-smart-routing.md
@@ -1,7 +1,7 @@
 ---
-"@trigger.dev/core": patch
-"@trigger.dev/sdk": patch
-"trigger.dev": patch
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"trigger.dev": minor
 ---
 
 Add smart routing for events: content-based filters and wildcard pattern subscriptions
diff --git a/.claude/skills/trigger-dev-tasks/SKILL.md b/.claude/skills/trigger-dev-tasks/SKILL.md
@@ -199,7 +199,7 @@ export const orderCreated = event({
 });
 
 // Subscribe task — payload typed from schema
-export const sendEmail = task({
+export const sendOrderEmail = task({
   id: "send-order-email",
   on: orderCreated,
   run: async (payload) => {
diff --git a/apps/webapp/app/routes/api.v1.events.$eventId.stats.ts b/apps/webapp/app/routes/api.v1.events.$eventId.stats.ts
@@ -22,7 +22,9 @@ export const loader = createLoaderApiRoute(
     const url = new URL(request.url);
     const period = url.searchParams.get("period") ?? "24h";
 
-    // Parse period to a ClickHouse interval
+    // SAFETY: interval is NOT user input — it comes from a closed allowlist below.
+    // Invalid periods are rejected with 400 before the value is used in the query.
+    // This is safe from SQL injection because only hardcoded strings can reach the query.
     const intervalMap: Record<string, string> = {
       "1h": "1 HOUR",
       "6h": "6 HOUR",
diff --git a/apps/webapp/app/routes/api.v1.events.dlq.ts b/apps/webapp/app/routes/api.v1.events.dlq.ts
@@ -1,7 +1,10 @@
 import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { DeadLetterManagementService } from "~/v3/services/events/deadLetterManagement.server";
 
+const DeadLetterStatusEnum = z.enum(["PENDING", "RETRIED", "DISCARDED"]);
+
 export const loader = createLoaderApiRoute(
   {
     corsStrategy: "all",
@@ -15,14 +18,17 @@ export const loader = createLoaderApiRoute(
   async ({ authentication, request }) => {
     const url = new URL(request.url);
     const eventType = url.searchParams.get("eventType") ?? undefined;
-    const status = url.searchParams.get("status") as
-      | "PENDING"
-      | "RETRIED"
-      | "DISCARDED"
-      | undefined;
-    const limit = url.searchParams.get("limit")
-      ? parseInt(url.searchParams.get("limit")!, 10)
-      : undefined;
+    const rawStatus = url.searchParams.get("status");
+    const statusParse = rawStatus ? DeadLetterStatusEnum.safeParse(rawStatus) : undefined;
+    if (rawStatus && (!statusParse || !statusParse.success)) {
+      return json(
+        { error: `Invalid status "${rawStatus}". Use: PENDING, RETRIED, DISCARDED` },
+        { status: 400 }
+      );
+    }
+    const status = statusParse?.success ? statusParse.data : undefined;
+    const rawLimit = url.searchParams.get("limit");
+    const limit = rawLimit ? Math.max(1, Math.min(parseInt(rawLimit, 10) || 20, 200)) : undefined;
     const cursor = url.searchParams.get("cursor") ?? undefined;
 
     const service = new DeadLetterManagementService();
diff --git a/apps/webapp/app/v3/services/events/cleanupStaleSubscriptions.server.ts b/apps/webapp/app/v3/services/events/cleanupStaleSubscriptions.server.ts
@@ -1,12 +1,11 @@
-import { PrismaClientOrTransaction } from "~/db.server";
 import { logger } from "~/services/logger.server";
+import { BaseService } from "../baseService.server";
 
 /**
  * Cleans up stale EventSubscriptions — disabled subscriptions whose associated
  * task no longer exists in any active worker for that environment.
  */
-export class CleanupStaleSubscriptionsService {
-  constructor(private readonly _prisma: PrismaClientOrTransaction) {}
+export class CleanupStaleSubscriptionsService extends BaseService {
 
   async call(): Promise<{ deletedCount: number; scannedCount: number }> {
     // Find all disabled subscriptions
diff --git a/apps/webapp/app/v3/services/events/eventLogWriter.server.ts b/apps/webapp/app/v3/services/events/eventLogWriter.server.ts
@@ -27,15 +27,15 @@ export function writeEventLog(entry: EventLogEntry): void {
   insertFn(row).then(
     ([error]) => {
       if (error) {
-        logger.warn("Failed to insert event into ClickHouse event log", {
+        logger.error("Failed to insert event into ClickHouse event log", {
           eventId: entry.eventId,
           eventType: entry.eventType,
           error: error.message,
         });
       }
     },
     (err) => {
-      logger.warn("Failed to insert event into ClickHouse event log", {
+      logger.error("Failed to insert event into ClickHouse event log", {
         eventId: entry.eventId,
         eventType: entry.eventType,
         error: err instanceof Error ? err.message : String(err),
diff --git a/apps/webapp/app/v3/services/events/eventRateLimiterGlobal.server.ts b/apps/webapp/app/v3/services/events/eventRateLimiterGlobal.server.ts
@@ -35,6 +35,12 @@ function initializeRateLimitChecker(): EventRateLimitChecker {
     return new RedisEventRateLimitChecker(redisClient);
   }
 
-  logger.info("Event rate limiter: using in-memory implementation (no RATE_LIMIT_REDIS_HOST)");
+  if (env.NODE_ENV === "production") {
+    logger.warn(
+      "Event rate limiter: using in-memory implementation in production (no RATE_LIMIT_REDIS_HOST). Rate limits will be per-process and ineffective across multiple instances."
+    );
+  } else {
+    logger.info("Event rate limiter: using in-memory implementation (no RATE_LIMIT_REDIS_HOST)");
+  }
   return new InMemoryEventRateLimitChecker();
 }
diff --git a/apps/webapp/app/v3/services/events/publishEvent.server.ts b/apps/webapp/app/v3/services/events/publishEvent.server.ts
diff --git a/apps/webapp/test/engine/deadLetterManagement.test.ts b/apps/webapp/test/engine/deadLetterManagement.test.ts
diff --git a/apps/webapp/test/engine/publishEvent.test.ts b/apps/webapp/test/engine/publishEvent.test.ts
diff --git a/apps/webapp/test/engine/schemaRegistryDb.test.ts b/apps/webapp/test/engine/schemaRegistryDb.test.ts
diff --git a/internal-packages/database/prisma/migrations/20260305000000_drop_redundant_event_subscription_idx/migration.sql b/internal-packages/database/prisma/migrations/20260305000000_drop_redundant_event_subscription_idx/migration.sql
diff --git a/packages/core/src/v3/types/tasks.ts b/packages/core/src/v3/types/tasks.ts
diff --git a/rules/4.4.0/events.md b/rules/4.4.0/events.md
