Added rate limiting to IngestSendEvent

Author: matt-aitken

apps/webapp/app/env.server.ts

@@ -100,6 +100,10 @@ const EnvironmentSchema = z.object({
   API_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
   API_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
 
+  //Ingesting event rate limit
+  INGEST_EVENT_RATE_LIMIT_WINDOW: z.string().default("60s"),
+  INGEST_EVENT_RATE_LIMIT_MAX: z.coerce.number().int().default(1_000),
+
   //v3
   V3_ENABLED: z.string().default("false"),
   PROVIDER_SECRET: z.string().default("provider-secret"),

apps/webapp/app/services/events/ingestSendEvent.server.ts

@@ -4,6 +4,9 @@ import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { workerQueue } from "~/services/worker.server";
 import { logger } from "../logger.server";
 import { EventRecord, ExternalAccount } from "@trigger.dev/database";
+import { Duration, RateLimiter } from "../rateLimiter.server";
+import { Ratelimit } from "@upstash/ratelimit";
+import { env } from "~/env.server";
 
 type UpdateEventInput = {
   tx: PrismaClientOrTransaction;
@@ -31,9 +34,17 @@ const EVENT_UPDATE_THRESHOLD_WINDOW_IN_MSECS = 5 * 1000; // 5 seconds
 
 export class IngestSendEvent {
   #prismaClient: PrismaClientOrTransaction;
+  #rateLimiter: RateLimiter;
 
   constructor(prismaClient: PrismaClientOrTransaction = prisma, private deliverEvents = true) {
     this.#prismaClient = prismaClient;
+    this.#rateLimiter = new RateLimiter({
+      keyPrefix: "ingestsendevent",
+      limiter: Ratelimit.slidingWindow(
+        env.INGEST_EVENT_RATE_LIMIT_MAX,
+        env.INGEST_EVENT_RATE_LIMIT_WINDOW as Duration
+      ),
+    });
   }
 
   #calculateDeliverAt(options?: SendEventOptions) {
@@ -104,6 +115,20 @@ export class IngestSendEvent {
               eventSource,
             }));
 
+        //rate limit
+        const { success, reset, limit } = await this.#rateLimiter.limit(environment.organizationId);
+
+        if (success) {
+          await this.enqueueWorkerEvent(tx, eventLog);
+        } else {
+          logger.info("IngestSendEvent: Rate limit exceeded", {
+            organizationId: environment.organizationId,
+            reset,
+            limit,
+          });
+          return;
+        }
+
         return eventLog;
       });
     } catch (error) {
@@ -151,8 +176,6 @@ export class IngestSendEvent {
       },
     });
 
-    await this.enqueueWorkerEvent(tx, eventLog);
-
     return eventLog;
   }
 
@@ -177,8 +200,6 @@ export class IngestSendEvent {
       },
     });
 
-    await this.enqueueWorkerEvent(tx, updatedEventLog);
-
     return updatedEventLog;
   }