chore: add shared redis client

Author: giovannibaratta

app/external/src/persistence.module.ts

@@ -36,6 +36,7 @@ import {PKCE_SESSION_REPOSITORY_TOKEN, REFRESH_TOKEN_REPOSITORY_TOKEN} from "@se
 import {ConfigModule} from "./config.module"
 import {QueueModule} from "./queue/queue.module"
 import {BullQueueProvider} from "./queue/queue.provider"
+import {RedisClient} from "./redis"
 
 const agentRepository = {
   provide: AGENT_REPOSITORY_TOKEN,
@@ -137,7 +138,7 @@ const repositories = [
 
 @Module({
   imports: [ConfigModule, QueueModule],
-  providers: [DatabaseClient, ...repositories, queueProvider],
+  providers: [DatabaseClient, ...repositories, queueProvider, RedisClient],
   exports: [...repositories, queueProvider]
 })
 export class PersistenceModule {}

app/external/src/rate-limiter/rate-limiter.provider.ts

@@ -6,6 +6,7 @@ import {ConsumePointsError, RateLimiterProvider} from "@services"
 import * as TE from "fp-ts/lib/TaskEither"
 import {pipe} from "fp-ts/lib/function"
 import {TaskEither} from "fp-ts/lib/TaskEither"
+import {waitForRedisConnection} from "../redis/redis.utils"
 
 export {RateLimiterRes} from "rate-limiter-flexible"
 
@@ -26,19 +27,7 @@ export class RedisRateLimiterProvider implements RateLimiterProvider, OnModuleIn
       enableOfflineQueue: false
     })
 
-    // Wait for the Redis TCP connection to be established before accepting requests.
-    // This is required because enableOfflineQueue is false, which means commands sent
-    // before the connection is ready will fail immediately instead of being queued.
-    // The "error" listener ensures that connection failures (e.g. wrong host/port)
-    // surface at startup rather than silently falling through to the fail-open path.
-    try {
-      await new Promise<void>((resolve, reject) => {
-        this.getRedisClient().once("ready", resolve)
-        this.getRedisClient().once("error", reject)
-      })
-    } catch {
-      throw new Error(`Unable to connect to Redis at ${config.host}:${config.port}`)
-    }
+    await waitForRedisConnection(this.getRedisClient(), "rate-limiter-redis-client")
 
     this.rateLimiter = new RateLimiterRedis({
       storeClient: this.redisClient,
@@ -71,9 +60,7 @@ export class RedisRateLimiterProvider implements RateLimiterProvider, OnModuleIn
           this.getRateLimiter()
             .consume(key, points)
             .catch(error => {
-              if (error instanceof RateLimiterRes) {
-                return error
-              }
+              if (error instanceof RateLimiterRes) return error
               throw error
             })
         ),

app/external/src/redis/index.ts

@@ -0,0 +1,2 @@
+export * from "./redis-client"
+export * from "./redis.utils"

app/external/src/redis/redis-client.ts

@@ -0,0 +1,25 @@
+import {Injectable, OnModuleDestroy, OnModuleInit} from "@nestjs/common"
+import Redis from "ioredis"
+import {ConfigProvider} from "../config"
+import {waitForRedisConnection} from "./redis.utils"
+
+@Injectable()
+export class RedisClient extends Redis implements OnModuleInit, OnModuleDestroy {
+  constructor(readonly configProvider: ConfigProvider) {
+    const config = configProvider.redisConfig
+    super({
+      host: config.host,
+      port: config.port,
+      db: config.db,
+      enableOfflineQueue: false
+    })
+  }
+
+  async onModuleInit() {
+    await waitForRedisConnection(this, "shared-redis-client")
+  }
+
+  onModuleDestroy() {
+    this.disconnect()
+  }
+}

app/external/src/redis/redis.utils.ts

@@ -0,0 +1,32 @@
+import {Logger} from "@nestjs/common"
+import Redis from "ioredis"
+
+/**
+ * Ensures a Redis connection is fully established before continuing.
+ *
+ * By default, ioredis connects asynchronously. We wait for the `"ready"` event here to guarantee the connection is usable.
+ *
+ * We also listen for `"error"` during this startup phase to eagerly throw an error and abort
+ * startup if the Redis server is unreachable (e.g. wrong host or port).
+ *
+ * A check for `client.status === "ready"` is included to avoid a race condition where the
+ * connection is established instantly and the `"ready"` event fires
+ * *before* the promise listener is attached, which would cause the startup to hang indefinitely.
+ */
+export async function waitForRedisConnection(client: Redis, clientName: string): Promise<void> {
+  if (client.status === "ready") {
+    Logger.log(`Redis connection ${clientName} already ready`, "RedisClient")
+    return
+  }
+
+  try {
+    await new Promise<void>((resolve, reject) => {
+      client.once("ready", resolve)
+      client.once("error", reject)
+    })
+    Logger.log(`Redis connection ${clientName} ready`, "RedisClient")
+  } catch {
+    const options = client.options
+    throw new Error(`Unable to connect to Redis at ${options.host}:${options.port}`)
+  }
+}