fix(events): phase 10 — CRITICAL + HIGH audit fixes

- 10.1: Fix expireTtlRuns Lua global concurrency slot leak (CRITICAL)
  Add SREM for globalCurrentConcurrency in TTL expiration script
- 10.2: Fix clearMessageFromConcurrencySets bare queue name (HIGH)
  Add queueGlobalCurrentConcurrencyKey(env, queue) to build correct key
- 10.3: Add .max(100) to batch publish items array (HIGH)
- 10.4: Fix publishAndWait schema — move parentRunId to top-level (HIGH)
- 10.5: ClickHouse interval already safe (whitelist map, not interpolation)
- 10.6: Add @@index([projectId, environmentId, enabled]) to EventSubscription
- 10.7: Fix batch publish partial failure — per-item error handling with 207

Also tightens Zod schemas: z.any() → z.unknown(), idempotencyKey .max(256),
metadata → z.record(z.unknown())

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

Author: giovaborgogno

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

@@ -34,10 +34,13 @@ const { action, loader } = createActionApiRoute(
       eventPublishRateLimitChecker
     );
 
-    try {
-      const results: PublishEventResult[] = [];
+    const results: Array<
+      | { ok: true; eventId: string; runs: PublishEventResult["runs"] }
+      | { ok: false; error: string }
+    > = [];
 
-      for (const item of body.items) {
+    for (const item of body.items) {
+      try {
         const result = await service.call(
           params.eventId,
           authentication.environment,
@@ -52,31 +55,23 @@ const { action, loader } = createActionApiRoute(
           }
         );
 
-        results.push(result);
+        results.push({ ok: true, eventId: result.eventId, runs: result.runs });
+      } catch (error) {
+        if (error instanceof EventPublishRateLimitError) {
+          results.push({ ok: false, error: error.message });
+        } else if (error instanceof ServiceValidationError) {
+          results.push({ ok: false, error: error.message });
+        } else {
+          results.push({
+            ok: false,
+            error: error instanceof Error ? error.message : "Unknown error",
+          });
+        }
       }
-
-      return json({ results }, { status: 200 });
-    } catch (error) {
-      if (error instanceof EventPublishRateLimitError) {
-        return json(
-          { error: error.message },
-          {
-            status: 429,
-            headers: {
-              "x-ratelimit-limit": String(error.limit),
-              "x-ratelimit-remaining": String(error.remaining),
-              "retry-after": String(Math.ceil(error.retryAfterMs / 1000)),
-            },
-          }
-        );
-      } else if (error instanceof ServiceValidationError) {
-        return json({ error: error.message }, { status: error.status ?? 422 });
-      } else if (error instanceof Error) {
-        return json({ error: error.message }, { status: 500 });
-      }
-
-      return json({ error: "Something went wrong" }, { status: 500 });
     }
+
+    const hasErrors = results.some((r) => !r.ok);
+    return json({ results }, { status: hasErrors ? 207 : 200 });
   }
 );
 

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

@@ -26,13 +26,7 @@ const { action, loader } = createActionApiRoute(
     },
   },
   async ({ body, params, authentication }) => {
-    const parentRunId = body.options?.parentRunId;
-    if (!parentRunId) {
-      return json(
-        { error: "parentRunId is required for publishAndWait" },
-        { status: 400 }
-      );
-    }
+    const parentRunId = body.parentRunId;
 
     const service = new PublishEventService(
       undefined,

internal-packages/database/prisma/migrations/20260303061123_add_event_subscription_pattern_idx/migration.sql

@@ -0,0 +1,2 @@
+-- CreateIndex
+CREATE INDEX CONCURRENTLY "EventSubscription_projectId_environmentId_enabled_idx" ON "public"."EventSubscription"("projectId", "environmentId", "enabled");

internal-packages/database/prisma/schema.prisma

@@ -644,7 +644,7 @@ model EventSubscription {
 
   @@unique([eventDefinitionId, taskSlug, environmentId])
   @@index([eventDefinitionId, environmentId, enabled])
-  @@index([projectId, environmentId])
+  @@index([projectId, environmentId, enabled])
 }
 
 enum DeadLetterStatus {

internal-packages/run-engine/src/run-queue/index.ts

@@ -2106,7 +2106,7 @@ export class RunQueue {
       envCurrentConcurrencyKey,
       queueCurrentDequeuedKey,
       envCurrentDequeuedKey,
-      this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(queue),
+      this.keys.queueGlobalCurrentConcurrencyKey(env, queue),
       messageId
     );
   }
@@ -2698,6 +2698,11 @@ for i, member in ipairs(expiredMembers) do
       redis.call('SREM', concurrencyKey, runId)
       redis.call('SREM', dequeuedKey, runId)
 
+      -- Remove from global concurrency set (strip :ck:* suffix to get base queue key)
+      local globalQueueKey = string.gsub(rawQueueKey, ":ck:.+$", "")
+      local globalCurrentConcurrencyKey = keyPrefix .. globalQueueKey .. ":globalCurrentConcurrency"
+      redis.call('SREM', globalCurrentConcurrencyKey, runId)
+
       -- Env concurrency (derive from rawQueueKey; must match RunQueueKeyProducer: org + proj + env)
       -- rawQueueKey format: {org:X}:proj:Y:env:Z:queue:Q[:ck:C]
       local projMatch = string.match(rawQueueKey, ":proj:([^:]+):env:")

internal-packages/run-engine/src/run-queue/keyProducer.ts

@@ -154,6 +154,10 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
     return [this.queueKey(env, queue), constants.GLOBAL_CONCURRENCY_LIMIT_PART].join(":");
   }
 
+  queueGlobalCurrentConcurrencyKey(env: RunQueueKeyProducerEnvironment, queue: string) {
+    return [this.queueKey(env, queue), constants.GLOBAL_CURRENT_CONCURRENCY_PART].join(":");
+  }
+
   queueCurrentConcurrencyKeyFromQueue(queue: string) {
     return `${queue}:${constants.CURRENT_CONCURRENCY_PART}`;
   }

internal-packages/run-engine/src/run-queue/types.ts

@@ -78,6 +78,7 @@ export interface RunQueueKeyProducer {
   queueGlobalConcurrencyLimitKeyFromQueue(queue: string): string;
   queueGlobalCurrentConcurrencyKeyFromQueue(queue: string): string;
   queueGlobalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string;
+  queueGlobalCurrentConcurrencyKey(env: RunQueueKeyProducerEnvironment, queue: string): string;
   queueCurrentConcurrencyKeyFromQueue(queue: string): string;
   queueCurrentConcurrencyKey(
     env: RunQueueKeyProducerEnvironment,

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

@@ -1602,14 +1602,14 @@ export type AppendToStreamResponseBody = z.infer<typeof AppendToStreamResponseBo
 // ---- Event publish schemas ----
 
 export const PublishEventRequestBody = z.object({
-  payload: z.any(),
+  payload: z.unknown(),
   options: z
     .object({
-      idempotencyKey: z.string().optional(),
+      idempotencyKey: z.string().max(256).optional(),
       delay: z.string().or(z.coerce.date()).optional(),
       tags: RunTags.optional(),
-      metadata: z.any().optional(),
-      context: z.any().optional(),
+      metadata: z.record(z.unknown()).optional(),
+      context: z.unknown().optional(),
       orderingKey: z.string().optional(),
     })
     .optional(),
@@ -1630,21 +1630,23 @@ export const PublishEventResponseBody = z.object({
 export type PublishEventResponseBody = z.infer<typeof PublishEventResponseBody>;
 
 export const BatchPublishEventRequestBody = z.object({
-  items: z.array(
-    z.object({
-      payload: z.any(),
-      options: z
-        .object({
-          idempotencyKey: z.string().optional(),
-          delay: z.string().or(z.coerce.date()).optional(),
-          tags: RunTags.optional(),
-          metadata: z.any().optional(),
-          context: z.any().optional(),
-          orderingKey: z.string().optional(),
-        })
-        .optional(),
-    })
-  ),
+  items: z
+    .array(
+      z.object({
+        payload: z.unknown(),
+        options: z
+          .object({
+            idempotencyKey: z.string().max(256).optional(),
+            delay: z.string().or(z.coerce.date()).optional(),
+            tags: RunTags.optional(),
+            metadata: z.record(z.unknown()).optional(),
+            context: z.unknown().optional(),
+            orderingKey: z.string().optional(),
+          })
+          .optional(),
+      })
+    )
+    .max(100),
 });
 
 export type BatchPublishEventRequestBody = z.infer<typeof BatchPublishEventRequestBody>;
@@ -1656,16 +1658,16 @@ export const BatchPublishEventResponseBody = z.object({
 export type BatchPublishEventResponseBody = z.infer<typeof BatchPublishEventResponseBody>;
 
 export const PublishAndWaitEventRequestBody = z.object({
-  payload: z.any(),
+  payload: z.unknown(),
+  parentRunId: z.string(),
   options: z
     .object({
-      idempotencyKey: z.string().optional(),
+      idempotencyKey: z.string().max(256).optional(),
       delay: z.string().or(z.coerce.date()).optional(),
       tags: RunTags.optional(),
-      metadata: z.any().optional(),
-      context: z.any().optional(),
+      metadata: z.record(z.unknown()).optional(),
+      context: z.unknown().optional(),
       orderingKey: z.string().optional(),
-      parentRunId: z.string(),
     })
     .optional(),
 });

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

@@ -239,18 +239,16 @@ export function createEvent<TId extends string, TSchema extends Schema | undefin
 
       const response = await apiClient.publishAndWaitEvent(id, {
         payload: validatedPayload,
+        parentRunId: ctx.run.id,
         options: options
           ? {
               idempotencyKey: options.idempotencyKey,
               delay: options.delay instanceof Date ? options.delay.toISOString() : options.delay,
               tags: options.tags,
               metadata: options.metadata,
               orderingKey: options.orderingKey,
-              parentRunId: ctx.run.id,
             }
-          : {
-              parentRunId: ctx.run.id,
-            },
+          : undefined,
       });
 
       if (response.runs.length === 0) {