fix(wh): add a global throttle queue

src/v1/merger.ts

@@ -10,7 +10,17 @@ type Pending = {
   count: number;
 };
 
+type ForwardJob = {
+  id: string;
+  url: string;
+  data: unknown;
+};
+
 const pending = new Map<string, Pending>();
+const forwardQueue: ForwardJob[] = [];
+const recentForwardTimestamps: number[] = [];
+
+let processingForwardQueue = false;
 
 const log = createLogger("WH");
 const routes = parseRouteMap(env.WH_ROUTE_MAP);
@@ -46,9 +56,9 @@ export function webhookMerger(app: App) {
         if (JSON.stringify(merged).length > 2000) {
           existing.data = body;
           existing.count = 1;
-          forward(id, url, oldData);
+          queueForward(id, url, oldData);
           existing.timer = setTimeout(
-            () => forward(id, url, existing.data),
+            () => flushPending(id, url, existing),
             bufferTime,
           );
           return status(202);
@@ -56,9 +66,9 @@ export function webhookMerger(app: App) {
 
         if (flushed.size > 0) {
           existing.count = 1;
-          forward(id, url, oldData);
+          queueForward(id, url, oldData);
           existing.timer = setTimeout(
-            () => forward(id, url, existing.data),
+            () => flushPending(id, url, existing),
             bufferTime,
           );
           return status(202);
@@ -67,7 +77,7 @@ export function webhookMerger(app: App) {
         log`collected ${existing}`;
 
         existing.timer = setTimeout(
-          () => forward(id, url, existing.data),
+          () => flushPending(id, url, existing),
           bufferTime,
         );
 
@@ -82,7 +92,7 @@ export function webhookMerger(app: App) {
 
       log`collected ${state}`;
 
-      state.timer = setTimeout(() => forward(id, url, state.data), bufferTime);
+      state.timer = setTimeout(() => flushPending(id, url, state), bufferTime);
 
       pending.set(id, state);
 
@@ -130,6 +140,54 @@ function parseRouteMap(raw: string): RouteMap {
 }
 
 const MAX_FIELD_LENGTH = Number(env.MAX_FIELD_LENGTH) || 4000;
+const MAX_FORWARDS_PER_WINDOW = 5;
+const FORWARD_WINDOW_MS = 2000;
+
+function flushPending(id: string, url: string, state: Pending) {
+  if (pending.get(id) === state) pending.delete(id);
+
+  queueForward(id, url, state.data);
+}
+
+function queueForward(id: string, url: string, data: unknown) {
+  forwardQueue.push({ id, url, data });
+  void processForwardQueue();
+}
+
+async function processForwardQueue() {
+  if (processingForwardQueue) return;
+
+  processingForwardQueue = true;
+
+  try {
+    while (forwardQueue.length > 0) {
+      const now = Date.now();
+
+      while (
+        recentForwardTimestamps.length > 0 &&
+        now - recentForwardTimestamps[0] >= FORWARD_WINDOW_MS
+      ) {
+        recentForwardTimestamps.shift();
+      }
+
+      if (recentForwardTimestamps.length >= MAX_FORWARDS_PER_WINDOW) {
+        await Bun.sleep(
+          FORWARD_WINDOW_MS - (now - recentForwardTimestamps[0]),
+        );
+        continue;
+      }
+
+      const job = forwardQueue.shift();
+
+      if (!job) continue;
+
+      recentForwardTimestamps.push(Date.now());
+      await forward(job.id, job.url, job.data);
+    }
+  } finally {
+    processingForwardQueue = false;
+  }
+}
 
 async function forward(id: string, url: string, data: unknown) {
   log`running ${id}`;
@@ -141,8 +199,8 @@ async function forward(id: string, url: string, data: unknown) {
     });
     if (!res.ok)
       log.warn`Could not forward webhook: ${res.status}\n${await res.json().catch(() => res.text())}`;
-  } finally {
-    pending.delete(id);
+  } catch (error) {
+    log.warn`Could not forward webhook: ${error}`;
   }
 }