feat: better concurrency with reddit

Author: nevo-david

libraries/helpers/src/utils/concurrency.service.ts

@@ -1,36 +1,28 @@
 import { ioRedis } from '@gitroom/nestjs-libraries/redis/redis.service';
-import { timer } from '@gitroom/helpers/utils/timer';
+import Bottleneck from 'bottleneck';
+
+const connection = new Bottleneck.IORedisConnection({
+  client: ioRedis,
+});
+
+const bottleneck = new Bottleneck.Group({
+  maxConcurrent: 1,
+  datastore: 'ioredis',
+  connection,
+});
 
 export async function concurrencyService<T>(
   identifier: string,
   func: (...args: any[]) => Promise<T>
 ): Promise<T> {
-  const key = `throttle:${identifier.split('-')[0]}`;
-  const expirationSeconds = 180;
-
-  while (true) {
-    const setLock = await ioRedis.set(
-      key,
-      'locked',
-      'EX',
-      expirationSeconds,
-      'NX'
-    );
-
-    if (setLock) {
-      break;
-    }
-
-    // Wait before trying again
-    await timer(1000);
-  }
-
   let load: T;
   try {
-    load = await func();
+    load = await bottleneck
+      .key(identifier.split('-')[0])
+      .schedule<T>({ expiration: 60_000 }, async () => {
+        return await func();
+      });
   } catch (err) {}
-  await timer(2000);
-  await ioRedis.del(key);
 
   return load;
 }

package.json

@@ -127,6 +127,7 @@
     "array-move": "^4.0.0",
     "axios": "^1.7.7",
     "bcrypt": "^5.1.1",
+    "bottleneck": "^2.19.5",
     "bs58": "^6.0.0",
     "bufferutil": "^4.0.8",
     "bullmq": "^5.12.12",