feat(events): consumer-side rate limiting + metrics endpoint

Consumer-side rate limiting:
- Add `rateLimit` JSON field to EventSubscription model
- Add `consumerRateLimit` option to TaskOptionsWithEvent in SDK
- Propagate through manifest pipeline → createBackgroundWorker
- PublishEventService checks per-subscriber rate limits during fan-out
- Rate-limited subscribers are skipped with warning log

Metrics endpoint:
- Add GET /api/v1/events/:eventId/metrics for backpressure monitoring
- Returns subscriber health (active/disabled/filters/rate limits)
- Returns DLQ depth (pending/retried/discarded counts)
- Returns event-level rate limit configuration
- Add GetEventMetricsResponseBody schema

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

Author: giovaborgogno

apps/webapp/app/routes/api.v1.events.$eventId.metrics.ts

@@ -0,0 +1,111 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { parseEventRateLimitConfig } from "~/v3/services/events/eventRateLimiter.server";
+
+const ParamsSchema = z.object({
+  eventId: z.string(),
+});
+
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    corsStrategy: "all",
+    authorization: {
+      action: "read",
+      resource: (_resource, params) => ({ tasks: params.eventId }),
+      superScopes: ["read:runs", "read:all", "admin"],
+    },
+    findResource: async () => 1 as const,
+  },
+  async ({ params, authentication }) => {
+    const environment = authentication.environment;
+
+    // Find event definition
+    const eventDef = await prisma.eventDefinition.findFirst({
+      where: {
+        slug: params.eventId,
+        projectId: environment.projectId,
+      },
+      orderBy: { createdAt: "desc" },
+    });
+
+    if (!eventDef) {
+      return json({ error: `Event "${params.eventId}" not found` }, { status: 404 });
+    }
+
+    // Get subscribers
+    const subscriptions = await prisma.eventSubscription.findMany({
+      where: {
+        eventDefinitionId: eventDef.id,
+        environmentId: environment.id,
+      },
+      select: {
+        taskSlug: true,
+        enabled: true,
+        rateLimit: true,
+        filter: true,
+        consumerGroup: true,
+      },
+    });
+
+    const activeCount = subscriptions.filter((s) => s.enabled).length;
+    const disabledCount = subscriptions.length - activeCount;
+
+    // Get DLQ counts
+    const [pendingCount, retriedCount, discardedCount] = await Promise.all([
+      prisma.deadLetterEvent.count({
+        where: {
+          eventType: params.eventId,
+          projectId: environment.projectId,
+          environmentId: environment.id,
+          status: "PENDING",
+        },
+      }),
+      prisma.deadLetterEvent.count({
+        where: {
+          eventType: params.eventId,
+          projectId: environment.projectId,
+          environmentId: environment.id,
+          status: "RETRIED",
+        },
+      }),
+      prisma.deadLetterEvent.count({
+        where: {
+          eventType: params.eventId,
+          projectId: environment.projectId,
+          environmentId: environment.id,
+          status: "DISCARDED",
+        },
+      }),
+    ]);
+
+    // Parse rate limit config
+    const rateLimitConfig = parseEventRateLimitConfig(eventDef.rateLimit);
+
+    return json({
+      eventType: params.eventId,
+      subscribers: {
+        total: subscriptions.length,
+        active: activeCount,
+        disabled: disabledCount,
+        list: subscriptions.map((s) => ({
+          taskSlug: s.taskSlug,
+          enabled: s.enabled,
+          hasRateLimit: !!s.rateLimit,
+          hasFilter: !!s.filter,
+          consumerGroup: s.consumerGroup,
+        })),
+      },
+      dlq: {
+        pending: pendingCount,
+        retried: retriedCount,
+        discarded: discardedCount,
+      },
+      rateLimit: rateLimitConfig
+        ? { limit: rateLimitConfig.limit, window: rateLimitConfig.window }
+        : null,
+    });
+  }
+);

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -460,13 +460,15 @@ async function syncWorkerEvents(
         filter: (task.onEventFilter as any) ?? undefined,
         pattern: task.onEventPattern ?? undefined,
         consumerGroup: task.onEventConsumerGroup ?? undefined,
+        rateLimit: (task.onEventConsumerRateLimit as any) ?? undefined,
       },
       update: {
         workerId: worker.id,
         enabled: true,
         filter: (task.onEventFilter as any) ?? undefined,
         pattern: task.onEventPattern ?? undefined,
         consumerGroup: task.onEventConsumerGroup ?? undefined,
+        rateLimit: (task.onEventConsumerRateLimit as any) ?? undefined,
       },
     });
 

apps/webapp/app/v3/services/events/publishEvent.server.ts

@@ -278,11 +278,34 @@ export class PublishEventService extends BaseService {
         );
       }
 
-      // 7. Fan out: trigger each matching subscribed task
+      // 7. Check per-subscriber rate limits and fan out
       const runs: PublishEventResult["runs"] = [];
 
       for (const subscription of subscriptionsToTrigger) {
         try {
+          // Check per-subscriber rate limit (if configured)
+          if (this._rateLimitChecker && subscription.rateLimit) {
+            const subRateLimitConfig = parseEventRateLimitConfig(subscription.rateLimit);
+            if (subRateLimitConfig) {
+              const subRateLimitKey = `consumer:${subscription.taskSlug}:${eventSlug}`;
+              const subRateLimitResult = await this._rateLimitChecker.check(
+                subRateLimitKey,
+                subRateLimitConfig
+              );
+              if (!subRateLimitResult.allowed) {
+                logger.warn("Subscriber rate limit exceeded, skipping", {
+                  eventSlug,
+                  eventId,
+                  taskSlug: subscription.taskSlug,
+                  limit: subRateLimitResult.limit,
+                  retryAfter: subRateLimitResult.retryAfter,
+                });
+                span.setAttribute("consumerRateLimited", true);
+                continue;
+              }
+            }
+          }
+
           // Derive per-consumer idempotency key if a global one was provided
           const consumerIdempotencyKey = options.idempotencyKey
             ? `${options.idempotencyKey}:${subscription.taskSlug}`
@@ -396,7 +419,7 @@ export class PublishEventService extends BaseService {
    * Different eventIds distribute evenly across group members.
    */
   private applyConsumerGroups(
-    subscriptions: Array<{ id: string; consumerGroup: string | null; taskSlug: string }>,
+    subscriptions: Array<{ id: string; consumerGroup: string | null; taskSlug: string; rateLimit: unknown }>,
     eventId?: string
   ): typeof subscriptions {
     const ungrouped: typeof subscriptions = [];

internal-packages/database/prisma/migrations/20260304065828_add_rate_limit_to_event_subscription/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "public"."EventSubscription" ADD COLUMN     "rateLimit" JSONB;

internal-packages/database/prisma/schema.prisma

@@ -639,6 +639,7 @@ model EventSubscription {
   filter        Json?    // EventFilter (Phase 2)
   pattern       String?  // Wildcard pattern (Phase 2)
   consumerGroup String?  // Consumer group name (Phase 5)
+  rateLimit     Json?    // Per-subscriber rate limit { limit, window }
   enabled       Boolean  @default(true)
   priority      Int      @default(0)
 

packages/core/src/v3/schemas/api.ts

@@ -1859,3 +1859,38 @@ export const GetEventStatsResponseBody = z.object({
 });
 
 export type GetEventStatsResponseBody = z.infer<typeof GetEventStatsResponseBody>;
+
+// ---- Event Metrics schemas ----
+
+export const EventMetricsSubscriber = z.object({
+  taskSlug: z.string(),
+  enabled: z.boolean(),
+  hasRateLimit: z.boolean(),
+  hasFilter: z.boolean(),
+  consumerGroup: z.string().nullable(),
+});
+
+export type EventMetricsSubscriber = z.infer<typeof EventMetricsSubscriber>;
+
+export const GetEventMetricsResponseBody = z.object({
+  eventType: z.string(),
+  subscribers: z.object({
+    total: z.number().int(),
+    active: z.number().int(),
+    disabled: z.number().int(),
+    list: z.array(EventMetricsSubscriber),
+  }),
+  dlq: z.object({
+    pending: z.number().int(),
+    retried: z.number().int(),
+    discarded: z.number().int(),
+  }),
+  rateLimit: z
+    .object({
+      limit: z.number().int(),
+      window: z.string(),
+    })
+    .nullable(),
+});
+
+export type GetEventMetricsResponseBody = z.infer<typeof GetEventMetricsResponseBody>;

packages/core/src/v3/schemas/resources.ts

@@ -19,6 +19,10 @@ export const TaskResource = z.object({
   onEventFilter: z.unknown().optional(),
   onEventPattern: z.string().optional(),
   onEventConsumerGroup: z.string().optional(),
+  onEventConsumerRateLimit: z.object({
+    limit: z.number().int().positive(),
+    window: z.string(),
+  }).optional(),
 });
 
 export type TaskResource = z.infer<typeof TaskResource>;

packages/core/src/v3/schemas/schemas.ts

@@ -240,6 +240,11 @@ const taskMetadata = {
   onEventPattern: z.string().optional(),
   /** Consumer group name — within a group, only one task receives each event */
   onEventConsumerGroup: z.string().optional(),
+  /** Per-subscriber rate limit — controls how fast this task receives events */
+  onEventConsumerRateLimit: z.object({
+    limit: z.number().int().positive(),
+    window: z.string(),
+  }).optional(),
 };
 
 export const TaskMetadata = z.object(taskMetadata);

packages/core/src/v3/types/tasks.ts

@@ -424,6 +424,8 @@ export type TaskOptionsWithEvent<
   filter?: import("../schemas/eventFilter.js").EventFilter;
   /** Consumer group — within a group, only one task receives each event */
   consumerGroup?: string;
+  /** Per-subscriber rate limit — controls how fast this task receives events */
+  consumerRateLimit?: { limit: number; window: string };
 };
 
 declare const __output: unique symbol;

packages/trigger-sdk/src/v3/shared.ts

@@ -250,6 +250,7 @@ export function createTask<
   const onEventFilter = "filter" in params && params.filter ? params.filter : undefined;
   const onEventPattern = eventSource && "pattern" in eventSource ? eventSource.pattern : undefined;
   const onEventConsumerGroup = "consumerGroup" in params && params.consumerGroup ? params.consumerGroup as string : undefined;
+  const onEventConsumerRateLimit = "consumerRateLimit" in params && params.consumerRateLimit ? params.consumerRateLimit as { limit: number; window: string } : undefined;
 
   resourceCatalog.registerTaskMetadata({
     id: params.id,
@@ -263,6 +264,7 @@ export function createTask<
     onEventFilter,
     onEventPattern,
     onEventConsumerGroup,
+    onEventConsumerRateLimit,
     fns: {
       run: params.run,
     },