feat(events): pub/sub event system

[giovaborgogno]

Summary

Adds a complete pub/sub event system to Trigger.dev, enabling event-driven task orchestration:

• Event definitions — createEvent() SDK function to define typed events with optional JSON Schema validation
• Event subscriptions — on: { event } option on tasks to subscribe with filters, patterns, and consumer groups
• Publish & fan-out — events.publish() triggers all matching subscribers with content-based filtering, pattern matching, and idempotency
• Publish-and-wait — events.publishAndWait() blocks the parent run until all subscriber runs complete
• Ordering keys — Per-key serialized execution via concurrency keys
• Rate limiting — Event-level and per-subscriber rate limits (Redis-backed with in-memory fallback)
• Consumer groups — FNV-1a hash-based deterministic routing within groups
• Dead letter queue — Failed event deliveries tracked, with retry/discard management via API
• Schema registry — Versioned payload schemas with validation on publish
• Event log — ClickHouse-backed event history with stats/metrics endpoints
• Replay — Re-publish historical events by time range via API
• CLI commands — trigger events history, trigger events replay, trigger events dlq list/retry
• Stale subscription cleanup — Automated daily cleanup of disabled subscriptions for removed tasks
• Documentation — SDK rules (rules/4.4.0/events.md), Claude Code skill, reference project

Database changes

• New models: EventDefinition, EventSubscription, DeadLetterEntry
• New migration to drop redundant composite index (CONCURRENTLY)
• ClickHouse: raw_event_log_v1 table with materialized view for event metrics

Test plan

• publishEvent.test.ts — 26 tests (fan-out, filters, patterns, consumer groups, ordering, idempotency, rate limits, payload size limit, publish-and-wait)
• deadLetterManagement.test.ts — 8 tests (list, discard, retry, retry nonexistent/discarded, retryAll empty)
• schemaRegistryDb.test.ts — 6 tests (register, upsert, getSchema latest/versioned/nonexistent, listSchemas with counts)
• eventRateLimiter.test.ts — 12 tests (in-memory + Redis sliding window)
• replayEvents.test.ts — replay service tests
• All builds pass (core, sdk, webapp)
• Typecheck passes (42/42)

🤖 Generated with Claude Code

[changeset-bot]

🦋 Changeset detected

Latest commit: 7493b86

The changes in this PR will be included in the next version bump.

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[github-actions]

Hi @giovaborgogno, thanks for your interest in contributing!

This project requires that pull request authors are vouched, and you are not in the list of vouched users.

This PR will be closed automatically. See https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md for more details.

[coderabbitai]

Caution

Review failed

The pull request is closed.

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: cd5e2460-24ed-4f52-acc5-d183433281c7

📥 Commits

Reviewing files that changed from the base of the PR and between c013322 and 7493b86.

⛔ Files ignored due to path filters (14)

• pnpm-lock.yaml is excluded by !**/pnpm-lock.yaml
• references/event-system/.gitignore is excluded by !references/**
• references/event-system/package.json is excluded by !references/**
• references/event-system/src/index.ts is excluded by !references/**
• references/event-system/src/trigger/basic-subscribers.ts is excluded by !references/**
• references/event-system/src/trigger/consumer-groups.ts is excluded by !references/**
• references/event-system/src/trigger/events.ts is excluded by !references/**
• references/event-system/src/trigger/filtered-subscribers.ts is excluded by !references/**
• references/event-system/src/trigger/ordering.ts is excluded by !references/**
• references/event-system/src/trigger/pattern-subscribers.ts is excluded by !references/**
• references/event-system/src/trigger/publish-and-wait.ts is excluded by !references/**
• references/event-system/trigger.config.ts is excluded by !references/**
• references/event-system/tsconfig.json is excluded by !references/**
• references/hello-world/src/trigger/events-test.ts is excluded by !references/**

📒 Files selected for processing (107)

• .changeset/event-dead-letter-queue.md
• .changeset/event-docs-cli-dx.md
• .changeset/event-observability-dx.md
• .changeset/event-ordering-consumer-groups.md
• .changeset/event-persistence-replay.md
• .changeset/event-publish-and-wait.md
• .changeset/event-rate-limiting.md
• .changeset/event-schema-registry.md
• .changeset/event-smart-routing.md
• .claude/projects/-Users-terac-repos-trigger-dev/memory/MEMORY.md
• .claude/projects/-Users-terac-repos-trigger-dev/memory/pubsub-audit-fixes.md
• .claude/projects/-Users-terac-repos-trigger-dev/memory/pubsub-pending.md
• .claude/projects/-Users-terac-repos-trigger-dev/memory/pubsub-progress.md
• .claude/projects/-Users-terac-repos-trigger-dev/memory/pubsub-roadmap.md
• .claude/projects/-Users-terac-repos-trigger-dev/memory/repo-conventions.md
• .claude/skills/trigger-dev-tasks/SKILL.md
• apps/webapp/app/routes/api.v1.events.$eventId.batchPublish.ts
• apps/webapp/app/routes/api.v1.events.$eventId.history.ts
• apps/webapp/app/routes/api.v1.events.$eventId.metrics.ts
• apps/webapp/app/routes/api.v1.events.$eventId.publish.ts
• apps/webapp/app/routes/api.v1.events.$eventId.publishAndWait.ts
• apps/webapp/app/routes/api.v1.events.$eventId.replay.ts
• apps/webapp/app/routes/api.v1.events.$eventId.schema.ts
• apps/webapp/app/routes/api.v1.events.$eventId.stats.ts
• apps/webapp/app/routes/api.v1.events.$eventId.ts
• apps/webapp/app/routes/api.v1.events.dlq.$id.discard.ts
• apps/webapp/app/routes/api.v1.events.dlq.$id.retry.ts
• apps/webapp/app/routes/api.v1.events.dlq.retry-all.ts
• apps/webapp/app/routes/api.v1.events.dlq.ts
• apps/webapp/app/routes/api.v1.events.ts
• apps/webapp/app/v3/runQueue.server.ts
• apps/webapp/app/v3/services/adminWorker.server.ts
• apps/webapp/app/v3/services/createBackgroundWorker.server.ts
• apps/webapp/app/v3/services/events/cleanupStaleSubscriptions.server.ts
• apps/webapp/app/v3/services/events/deadLetterManagement.server.ts
• apps/webapp/app/v3/services/events/deadLetterService.server.ts
• apps/webapp/app/v3/services/events/eventLogWriter.server.ts
• apps/webapp/app/v3/services/events/eventRateLimiter.server.ts
• apps/webapp/app/v3/services/events/eventRateLimiterGlobal.server.ts
• apps/webapp/app/v3/services/events/publishEvent.server.ts
• apps/webapp/app/v3/services/events/replayEvents.server.ts
• apps/webapp/app/v3/services/events/schemaRegistry.server.ts
• apps/webapp/app/v3/services/finalizeTaskRun.server.ts
• apps/webapp/package.json
• apps/webapp/test/engine/deadLetterManagement.test.ts
• apps/webapp/test/engine/deadLetterService.test.ts
• apps/webapp/test/engine/eventRateLimiter.test.ts
• apps/webapp/test/engine/publishEvent.test.ts
• apps/webapp/test/engine/replayEvents.test.ts
• apps/webapp/test/engine/schemaRegistryDb.test.ts
• apps/webapp/test/services/schemaRegistry.test.ts
• internal-packages/clickhouse/schema/021_event_log_v1.sql
• internal-packages/clickhouse/schema/022_event_counts_mv_v1.sql
• internal-packages/clickhouse/src/eventCounts.ts
• internal-packages/clickhouse/src/eventLog.ts
• internal-packages/clickhouse/src/index.ts
• internal-packages/database/prisma/migrations/20260227081612_add_event_definition_and_subscription/migration.sql
• internal-packages/database/prisma/migrations/20260228054059_add_event_schema_versioning/migration.sql
• internal-packages/database/prisma/migrations/20260228065743_add_dead_letter_event/migration.sql
• internal-packages/database/prisma/migrations/20260228074544_add_rate_limit_to_event_definition/migration.sql
• internal-packages/database/prisma/migrations/20260303061123_add_event_subscription_pattern_idx/migration.sql
• internal-packages/database/prisma/migrations/20260304064319_add_dlq_config_to_event_definition/migration.sql
• internal-packages/database/prisma/migrations/20260304065828_add_rate_limit_to_event_subscription/migration.sql
• internal-packages/database/prisma/migrations/20260305000000_drop_redundant_event_subscription_idx/migration.sql
• internal-packages/database/prisma/schema.prisma
• internal-packages/run-engine/src/run-queue/index.ts
• internal-packages/run-engine/src/run-queue/keyProducer.ts
• internal-packages/run-engine/src/run-queue/types.ts
• packages/cli-v3/src/apiClient.ts
• packages/cli-v3/src/cli/index.ts
• packages/cli-v3/src/commands/events/dlq.ts
• packages/cli-v3/src/commands/events/history.ts
• packages/cli-v3/src/commands/events/index.ts
• packages/cli-v3/src/commands/events/list.ts
• packages/cli-v3/src/commands/events/publish.ts
• packages/cli-v3/src/commands/events/replay.ts
• packages/cli-v3/src/dev/devSupervisor.ts
• packages/cli-v3/src/entryPoints/dev-index-worker.ts
• packages/cli-v3/src/entryPoints/managed-index-controller.ts
• packages/cli-v3/src/entryPoints/managed-index-worker.ts
• packages/core/src/v3/apiClient/index.ts
• packages/core/src/v3/events/filterEvaluator.ts
• packages/core/src/v3/events/index.ts
• packages/core/src/v3/events/patternMatcher.ts
• packages/core/src/v3/index.ts
• packages/core/src/v3/resource-catalog/catalog.ts
• packages/core/src/v3/resource-catalog/index.ts
• packages/core/src/v3/resource-catalog/noopResourceCatalog.ts
• packages/core/src/v3/resource-catalog/standardResourceCatalog.ts
• packages/core/src/v3/runtime/index.ts
• packages/core/src/v3/runtime/manager.ts
• packages/core/src/v3/runtime/noopRuntimeManager.ts
• packages/core/src/v3/runtime/sharedRuntimeManager.ts
• packages/core/src/v3/schemas/api.ts
• packages/core/src/v3/schemas/build.ts
• packages/core/src/v3/schemas/resources.ts
• packages/core/src/v3/schemas/schemas.ts
• packages/core/src/v3/types/tasks.ts
• packages/core/src/v3/workers/index.ts
• packages/core/test/v3/events/filterEvaluator.test.ts
• packages/core/test/v3/events/patternMatcher.test.ts
• packages/trigger-sdk/src/v3/events.ts
• packages/trigger-sdk/src/v3/index.ts
• packages/trigger-sdk/src/v3/shared.ts
• packages/trigger-sdk/src/v3/tasks.ts
• rules/4.4.0/events.md
• rules/manifest.json

---

Walkthrough

This pull request introduces a comprehensive Pub/Sub event system for Trigger.dev. The implementation spans database schema changes (EventDefinition, EventSubscription, DeadLetterEvent tables), ClickHouse event logging, new API routes for event operations, service layers for publishing/replaying/managing events, rate limiting capabilities (Redis and in-memory), event filtering and pattern matching utilities, schema validation, dead-letter queue handling, and CLI commands. It extends the core SDK with event creation APIs, task event subscriptions, and runtime wait semantics. Multiple changesets document feature releases including rate limiting, ordering keys, consumer groups, DLQ, schema registry, observability, and replay functionality.

Estimated code review effort

🎯 5 (Critical) | ⏱️ ~120 minutes

Rationale: This diff introduces substantial, intricate functionality across multiple architectural layers with high logic density. The PublishEventService alone contains complex state management, consumer group logic (FNV-1a hashing), rate limiting integration, pattern matching, filter evaluation, and idempotency handling. The system spans new database models with cascading relationships, ClickHouse materialized views, diverse rate-limiting backends (Redis-backed with Upstash, in-memory), service orchestration (replay, dead-letter management, schema validation), and heterogeneous API routes with authorization and error handling. Supporting utilities like pattern matching with wildcard semantics and filter compilation caching add further complexity. The breadth (100+ files across packages/webapp, packages/core, packages/trigger-sdk, packages/cli-v3, internal-packages) combined with semantic interdependencies (event subscriptions, ordering queues, global concurrency tracking) and comprehensive test coverage demand sustained, context-aware review across multiple domains (database design, distributed systems, caching, filtering logic).

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment

[devin-ai-integration]

Devin Review found 5 potential issues.

View 5 additional findings in Devin Review.

[devin-ai-integration]

🔴 Batch publish route response shape doesn't match the Zod schema used by API clients

The api.v1.events.$eventId.batchPublish.ts route returns results with shape { ok: true, eventId, runs } | { ok: false, error }, but the BatchPublishEventResponseBody schema at packages/core/src/v3/schemas/api.ts:1654 expects { results: z.array(PublishEventResponseBody) } where PublishEventResponseBody requires { eventId: string, runs: [...] }. When any item in the batch fails, the error result { ok: false, error: '...' } lacks the required eventId and runs fields, causing zodfetch in the API client (packages/core/src/v3/apiClient/index.ts:1488) to throw a Zod validation error. This means any batch publish with partial failures will crash the client instead of returning the partial results. The SDK's batchPublish method at packages/trigger-sdk/src/v3/events.ts:320 also relies on result.results having eventId and runs on every item.

Prompt for agents

The batch publish route at apps/webapp/app/routes/api.v1.events.$eventId.batchPublish.ts (lines 37-74) returns a response with shape { results: Array<{ ok: true, eventId, runs } | { ok: false, error }> }, but the BatchPublishEventResponseBody Zod schema at packages/core/src/v3/schemas/api.ts:1654 expects { results: z.array(PublishEventResponseBody) } where PublishEventResponseBody is { eventId: string, runs: Array<...> }.

Two options to fix:

Option A (recommended): Update BatchPublishEventResponseBody in packages/core/src/v3/schemas/api.ts to match the actual route response shape. Change it to support both success and error results:
  results: z.array(z.union([
    z.object({ ok: z.literal(true), eventId: z.string(), runs: z.array(...) }),
    z.object({ ok: z.literal(false), error: z.string() })
  ]))
Then update the SDK batchPublish method in packages/trigger-sdk/src/v3/events.ts to handle error items.

Option B: Change the route to match the existing schema by validating all items upfront and either succeeding entirely or failing entirely, so the response always contains valid PublishEventResponseBody items.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🟡 DLQ list pagination uses non-unique createdAt as cursor, causing skipped items

The list method in deadLetterManagement.server.ts uses createdAt as the pagination cursor with { lt: new Date(params.cursor) }. Since createdAt has only millisecond precision and multiple DLQ entries can share the same timestamp (e.g., when a fan-out event triggers multiple subscriber failures simultaneously), entries with the same createdAt as the cursor boundary will be skipped on the next page. The same issue exists in the event history endpoint at apps/webapp/app/routes/api.v1.events.$eventId.history.ts:57 which uses published_at as cursor.

Prompt for agents

In apps/webapp/app/v3/services/events/deadLetterManagement.server.ts, the list() method (line 27-40) uses createdAt as the pagination cursor. Since createdAt is not unique, items sharing the same timestamp as the cursor boundary can be skipped.

Fix by switching to cursor-based pagination using the unique id field instead of createdAt. Change the cursor filter from { createdAt: { lt: new Date(params.cursor) } } to use id-based ordering with a compound cursor (createdAt + id) or simply use the id as cursor with { id: { lt: params.cursor } } and order by id DESC. The same fix should be applied to the event history endpoint at apps/webapp/app/routes/api.v1.events.$eventId.history.ts:57.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🚩 syncWorkerEvents disables all non-current subscriptions in the environment

At apps/webapp/app/v3/services/createBackgroundWorker.server.ts:480-491, syncWorkerEvents disables ALL subscriptions for the project+environment that aren't in the current deploy's activeSubscriptionIds. This follows the same "reconcile to desired state" pattern used elsewhere (e.g., schedule syncing). However, during rolling deploys or if multiple workers coexist briefly for the same environment, this could transiently disable subscriptions from a still-active worker. Worth confirming this aligns with the platform's worker lifecycle model (new deploys fully replace old workers).

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🚩 Replay service filters tasks AFTER publishing, not before

In replayEvents.server.ts:143-146, the tasks filter is applied AFTER publishService.call(), meaning all subscribers are triggered for every replayed event, and only the runs matching the requested tasks are included in the response. The actual filtering happens on the response side, not the trigger side. This means replay with a tasks filter still creates runs for ALL subscribers, which may be unexpected. Users might expect tasks to limit which subscribers are triggered, not just which runs are reported.

---

Was this helpful? React with 👍 or 👎 to provide feedback.