use p-ratelimit for event limiting

src/subscription.ts

@@ -66,14 +66,6 @@ export class FirehoseSubscription extends FirehoseSubscriptionBase {
 
     if (!ops) return
 
-    const delay = (ms: number) => new Promise((res) => setTimeout(res, ms))
-
-    while (this.runningEvents > 128) {
-      await delay(1000)
-    }
-
-    this.runningEvents++
-
     const postsToDelete = ops.posts.deletes.map((del) => del.uri)
 
     // Transform posts in parallel
@@ -134,7 +126,5 @@ export class FirehoseSubscription extends FirehoseSubscriptionBase {
           await this.db.replaceOneURI('post', to_insert.uri, to_insert)
       })
     }
-
-    this.runningEvents--
   }
 }

src/util/subscription.ts

@@ -15,6 +15,15 @@ import { Database } from '../db'
 
 const includedRecords = new Set(['app.bsky.feed.post'])
 
+import { pRateLimit } from 'p-ratelimit'
+
+const eventLimit = pRateLimit({
+  interval: undefined,
+  rate: undefined,
+  concurrency: 128,
+  maxDelay: undefined,
+})
+
 export abstract class FirehoseSubscriptionBase {
   public sub: Subscription<RepoEvent>
 
@@ -42,7 +51,7 @@ export abstract class FirehoseSubscriptionBase {
 
             if (includedRecords.has(collection)) {
               try {
-                this.handleEvent(evt) // no longer awaiting this
+                await eventLimit(async () => this.handleEvent(evt)) // no longer awaiting this
               } catch (err) {
                 console.error('repo subscription could not handle message', err)
               }