fix(events): phase 11 — MEDIUM audit fixes

- 11.1: Fix N+1 in DLQ retryAll — inline retry logic, share TriggerTaskService
- 11.2: Add 512KB payload size check before fan-out (returns 413)
- 11.3: Add try/catch with ServiceValidationError handling to events routes
- 11.4: Add --delay, --tags, --idempotency-key, --ordering-key to CLI publish

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

Author: giovaborgogno

apps/webapp/app/routes/api.v1.events.dlq.retry-all.ts

@@ -1,6 +1,7 @@
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { DeadLetterManagementService } from "~/v3/services/events/deadLetterManagement.server";
 
 const BodySchema = z
@@ -22,14 +23,24 @@ const { action, loader } = createActionApiRoute(
   async ({ body, authentication }) => {
     const service = new DeadLetterManagementService();
 
-    const result = await service.retryAll({
-      projectId: authentication.environment.projectId,
-      environmentId: authentication.environment.id,
-      eventType: body?.eventType,
-      environment: authentication.environment,
-    });
+    try {
+      const result = await service.retryAll({
+        projectId: authentication.environment.projectId,
+        environmentId: authentication.environment.id,
+        eventType: body?.eventType,
+        environment: authentication.environment,
+      });
 
-    return json(result, { status: 200 });
+      return json(result, { status: 200 });
+    } catch (error) {
+      if (error instanceof ServiceValidationError) {
+        return json({ error: error.message }, { status: error.status ?? 422 });
+      }
+      return json(
+        { error: error instanceof Error ? error.message : "Something went wrong" },
+        { status: 500 }
+      );
+    }
   }
 );
 

apps/webapp/app/routes/api.v1.events.ts

@@ -1,5 +1,6 @@
 import { json } from "@remix-run/server-runtime";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { SchemaRegistryService } from "~/v3/services/events/schemaRegistry.server";
 
 export const loader = createLoaderApiRoute(
@@ -15,23 +16,33 @@ export const loader = createLoaderApiRoute(
   async ({ authentication }) => {
     const service = new SchemaRegistryService();
 
-    const events = await service.listSchemas({
-      projectId: authentication.environment.projectId,
-      environmentId: authentication.environment.id,
-    });
+    try {
+      const events = await service.listSchemas({
+        projectId: authentication.environment.projectId,
+        environmentId: authentication.environment.id,
+      });
 
-    return json({
-      data: events.map((e) => ({
-        id: e.id,
-        slug: e.slug,
-        version: e.version,
-        description: e.description,
-        hasSchema: e.schema !== null,
-        deprecatedAt: e.deprecatedAt,
-        subscriberCount: e.subscriberCount,
-        createdAt: e.createdAt,
-        updatedAt: e.updatedAt,
-      })),
-    });
+      return json({
+        data: events.map((e) => ({
+          id: e.id,
+          slug: e.slug,
+          version: e.version,
+          description: e.description,
+          hasSchema: e.schema !== null,
+          deprecatedAt: e.deprecatedAt,
+          subscriberCount: e.subscriberCount,
+          createdAt: e.createdAt,
+          updatedAt: e.updatedAt,
+        })),
+      });
+    } catch (error) {
+      if (error instanceof ServiceValidationError) {
+        return json({ error: error.message }, { status: error.status ?? 422 });
+      }
+      return json(
+        { error: error instanceof Error ? error.message : "Something went wrong" },
+        { status: 500 }
+      );
+    }
   }
 );

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

@@ -136,9 +136,26 @@ export class DeadLetterManagementService extends BaseService {
     let retriedCount = 0;
     let failedCount = 0;
 
+    const triggerService = new TriggerTaskService();
+
     for (const dle of pendingItems) {
       try {
-        await this.retry(dle.id, params.environment);
+        const body: TriggerTaskRequestBody = {
+          payload: dle.payload,
+          options: {
+            idempotencyKey: `dlq-retry:${dle.id}`,
+          },
+        };
+
+        await triggerService.call(dle.taskSlug, params.environment, body, {
+          idempotencyKey: `dlq-retry:${dle.id}`,
+        });
+
+        await this._prisma.deadLetterEvent.update({
+          where: { id: dle.id },
+          data: { status: "RETRIED", processedAt: new Date() },
+        });
+
         retriedCount++;
       } catch {
         failedCount++;

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

@@ -171,6 +171,16 @@ export class PublishEventService extends BaseService {
         }
       }
 
+      // 2b. Check payload size (512KB limit — larger payloads require object store)
+      const payloadBytes = Buffer.byteLength(JSON.stringify(payload), "utf-8");
+      const MAX_PAYLOAD_BYTES = 512 * 1024; // 512KB
+      if (payloadBytes > MAX_PAYLOAD_BYTES) {
+        throw new ServiceValidationError(
+          `Payload size ${payloadBytes} bytes exceeds the 512KB limit. Use smaller payloads or configure the object store for large payloads.`,
+          413
+        );
+      }
+
       // 3. Find all active subscriptions: exact match + pattern-based
       const [exactSubscriptions, patternSubscriptions] = await Promise.all([
         // Exact subscriptions: tied to this specific EventDefinition

packages/cli-v3/src/apiClient.ts

@@ -563,7 +563,17 @@ export class CliApiClient {
     });
   }
 
-  async publishEvent(projectRef: string, eventId: string, payload: unknown) {
+  async publishEvent(
+    projectRef: string,
+    eventId: string,
+    payload: unknown,
+    options?: {
+      idempotencyKey?: string;
+      delay?: string;
+      tags?: string[];
+      orderingKey?: string;
+    }
+  ) {
     if (!this.accessToken) {
       throw new Error("publishEvent: No access token");
     }
@@ -579,7 +589,17 @@ export class CliApiClient {
           ...this.getHeaders(),
           "x-trigger-project-ref": projectRef,
         },
-        body: JSON.stringify({ payload }),
+        body: JSON.stringify({
+          payload,
+          options: options
+            ? {
+                idempotencyKey: options.idempotencyKey,
+                delay: options.delay,
+                tags: options.tags,
+                orderingKey: options.orderingKey,
+              }
+            : undefined,
+        }),
       }
     );
   }

packages/cli-v3/src/commands/events/publish.ts

@@ -21,6 +21,10 @@ const EventsPublishOptions = CommonCommandOptions.extend({
   projectRef: z.string().optional(),
   envFile: z.string().optional(),
   payload: z.string(),
+  delay: z.string().optional(),
+  tags: z.string().optional(),
+  idempotencyKey: z.string().optional(),
+  orderingKey: z.string().optional(),
 });
 
 type EventsPublishOptions = z.infer<typeof EventsPublishOptions>;
@@ -34,6 +38,10 @@ export function configureEventsPublishCommand(program: Command) {
       .option("-c, --config <config file>", "The name of the config file")
       .option("-p, --project-ref <project ref>", "The project ref")
       .option("--env-file <env file>", "Path to the .env file")
+      .option("--delay <delay>", "Delay before execution (e.g. '30s', '5m', ISO date)")
+      .option("--tags <tags>", "Comma-separated tags to attach")
+      .option("--idempotency-key <key>", "Idempotency key for deduplication")
+      .option("--ordering-key <key>", "Ordering key for sequential processing")
   ).action(async (eventId: string, options) => {
     await handleTelemetry(async () => {
       await printInitialBanner(false, options.profile);
@@ -96,7 +104,19 @@ async function _eventsPublishCommand(options: EventsPublishCommandInput) {
   loadingSpinner.start("Publishing event...");
 
   const apiClient = new CliApiClient(authentication.auth.apiUrl, authentication.auth.accessToken);
-  const result = await apiClient.publishEvent(resolvedConfig.project, options.eventId, payload);
+  const publishOptions = {
+    idempotencyKey: options.idempotencyKey,
+    delay: options.delay,
+    tags: options.tags ? options.tags.split(",").map((t: string) => t.trim()) : undefined,
+    orderingKey: options.orderingKey,
+  };
+  const hasOptions = Object.values(publishOptions).some((v) => v !== undefined);
+  const result = await apiClient.publishEvent(
+    resolvedConfig.project,
+    options.eventId,
+    payload,
+    hasOptions ? publishOptions : undefined
+  );
 
   if (!result.success) {
     loadingSpinner.stop("Failed to publish event");