feat(analytics): enhance Redis key migration and health monitoring

- Added a backward compatibility helper for Redis keys to facilitate migration from old to new formats.
- Improved Redis retry strategies with logging for better debugging.
- Implemented health monitoring for Redis connections, including automatic client rebuilding after consecutive errors.
- Updated Redis options for better connection handling and timeouts.

Author: abretonc7s

packages/sdk-socket-server-next/src/analytics-api.ts

@@ -24,6 +24,7 @@ import {
   incrementAnalyticsError,
   incrementAnalyticsEvents,
   incrementRedisCacheOperation,
+  incrementKeyMigration,
 } from './metrics';
 import genericPool from "generic-pool";
 
@@ -72,10 +73,21 @@ export const getRedisOptions = (
     connectTimeout: 30000,
     keepAlive: 369,
     maxRetriesPerRequest: 4,
-    retryStrategy: (times) => Math.min(times * 30, 1000),
+    retryStrategy: (times) => {
+      const delay = Math.min(times * 30, 1000);
+      logger.info(`Redis retry attempt ${times} with delay ${delay}ms`);
+      return delay;
+    },
     reconnectOnError: (error) => {
-      // eslint-disable-next-line require-unicode-regexp
-      const targetErrors = [/MOVED/, /READONLY/, /ETIMEDOUT/];
+      const targetErrors = [
+        /MOVED/,
+        /READONLY/,
+        /ETIMEDOUT/,
+        /ECONNRESET/,
+        /ECONNREFUSED/,
+        /EPIPE/,
+        /ENOTFOUND/,
+      ];
 
       logger.error('Redis reconnect error:', error);
       return targetErrors.some((targetError) =>
@@ -105,9 +117,25 @@ export const buildRedisClient = (usePipelining: boolean = true) => {
       slotsRefreshTimeout: 5000,
       showFriendlyErrorStack: true,
       slotsRefreshInterval: 2000,
-      clusterRetryStrategy: (times) => Math.min(times * 30, 1000),
+      natMap: process.env.REDIS_NAT_MAP ? JSON.parse(process.env.REDIS_NAT_MAP) : undefined,
+      redisOptions: {
+        ...redisOptions,
+        // Queues commands when disconnected from Redis, executing them when connection is restored
+        // This prevents data loss during network issues or cluster topology changes
+        offlineQueue: true,
+        // Default is 10000ms (10s). Increasing this allows more time to establish
+        // connection during network instability while balancing real-time requirements
+        connectTimeout: 10000,
+        // Default is no timeout. Setting to 5000ms prevents hanging commands
+        // while still allowing reasonable time for completion
+        commandTimeout: 5000,
+      },
+      clusterRetryStrategy: (times) => {
+        const delay = Math.min(times * 100, 3000);
+        logger.info(`Redis Cluster retry attempt ${times} with delay ${delay}ms`);
+        return delay;
+      },
       enableAutoPipelining: usePipelining,
-      redisOptions,
     };
 
     logger.debug(
@@ -179,6 +207,11 @@ export const pubClient = getGlobalRedisClient();
 export const pubClientPool = genericPool.createPool(redisFactory, {
   max: 35,
   min: 15,
+  acquireTimeoutMillis: 5000,
+  idleTimeoutMillis: 30000,
+  evictionRunIntervalMillis: 15000,
+  numTestsPerEvictionRun: 3,
+  softIdleTimeoutMillis: 10000,
 });
 
 const app = express();
@@ -260,6 +293,28 @@ app.post('/debug', (req, _res, next) => {
   next(); // Pass control to the next handler (which will be /evt)
 });
 
+// Add Redis key backward compatibility helper
+const getWithBackwardCompatibility = async ({
+  newKey,
+  oldKey,
+}: {
+  newKey: string;
+  oldKey: string;
+}) => {
+  let value = await pubClient.get(newKey);
+  if (!value) {
+    // Try old key format if new key returns nothing
+    value = await pubClient.get(oldKey);
+    if (value) {
+      // If found with old key, migrate to new format
+      await pubClient.set(newKey, value, 'EX', config.channelExpiry.toString());
+      incrementKeyMigration({ migrationType: 'channel-id' });
+      logger.info(`Migrated key from ${oldKey} to ${newKey}`);
+    }
+  }
+  return value;
+}
+
 app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
   try {
     const { body } = _req;
@@ -287,7 +342,7 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
     ];
 
     // Filter: drop RPC events with unallowed methods silently, let all else through
-    if (toCheckEvents.includes(body.event) && 
+    if (toCheckEvents.includes(body.event) &&
         (!body.method || !allowedMethods.includes(body.method))) {
       return res.json({ success: true });
     }
@@ -316,7 +371,10 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
 
     let userIdHash = isAnonUser
       ? crypto.createHash('sha1').update(channelId).digest('hex')
-      : await pubClient.get(channelId);
+      : await getWithBackwardCompatibility({
+          newKey: `{${channelId}}:id`,
+          oldKey: channelId,
+        });
 
     incrementRedisCacheOperation('analytics-get-channel-id', !!userIdHash);
 
@@ -327,8 +385,9 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
       );
 
       if (!isExtensionEvent) {
+        // Always write to the new format
         await pubClient.set(
-          channelId,
+          `{${channelId}}:id`,
           userIdHash,
           'EX',
           config.channelExpiry.toString(),
@@ -338,12 +397,16 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
 
     if (REDIS_DEBUG_LOGS) {
       await inspectRedis(channelId);
+      await inspectRedis(`{${channelId}}:id`);
     }
 
     let channelInfo: ChannelInfo | null;
     const cachedChannelInfo = isAnonUser
       ? null
-      : await pubClient.get(userIdHash);
+      : await getWithBackwardCompatibility({
+          newKey: `{${userIdHash}}:info`,
+          oldKey: userIdHash,
+        });
 
     incrementRedisCacheOperation(
       'analytics-get-channel-info',
@@ -380,8 +443,9 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
       );
 
       if (!isExtensionEvent) {
+        // Always write to the new format
         await pubClient.set(
-          userIdHash,
+          `{${userIdHash}}:info`,
           JSON.stringify(channelInfo),
           'EX',
           config.channelExpiry.toString(),
@@ -391,6 +455,7 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
 
     if (REDIS_DEBUG_LOGS) {
       await inspectRedis(userIdHash);
+      await inspectRedis(`{${userIdHash}}:info`);
     }
 
     const event = {
@@ -464,4 +529,51 @@ app.post('/evt', evtMetricsMiddleware, async (_req, res) => {
   }
 });
 
+// Add Redis health checking and recovery
+let redisHealthCheckInterval: NodeJS.Timeout;
+let consecutiveRedisErrors = 0;
+const MAX_CONSECUTIVE_ERRORS = 10;
+
+// Add Redis health monitoring function after getGlobalRedisClient
+export const monitorRedisHealth = () => {
+  if (redisHealthCheckInterval) {
+    clearInterval(redisHealthCheckInterval);
+  }
+
+  redisHealthCheckInterval = setInterval(async () => {
+    try {
+      // Simple ping to check Redis health
+      await pubClient.ping();
+      // Reset error counter on successful ping
+      if (consecutiveRedisErrors > 0) {
+        logger.info(`Redis health restored after ${consecutiveRedisErrors} consecutive errors`);
+        consecutiveRedisErrors = 0;
+      }
+    } catch (error) {
+      consecutiveRedisErrors++;
+      logger.error(`Redis health check failed (${consecutiveRedisErrors}/${MAX_CONSECUTIVE_ERRORS}):`, error);
+
+      // If too many consecutive errors, attempt to rebuild the Redis client
+      if (consecutiveRedisErrors >= MAX_CONSECUTIVE_ERRORS) {
+        logger.warn(`Rebuilding Redis client after ${consecutiveRedisErrors} consecutive errors`);
+        try {
+          // Disconnect the old client first
+          pubClient.disconnect();
+          // Build a new Redis client and update the global reference
+          redisClient = buildRedisClient();
+          // Update the pubClient reference
+          Object.assign(pubClient, redisClient);
+          consecutiveRedisErrors = 0;
+          logger.info('Redis client rebuilt successfully');
+        } catch (rebuildError) {
+          logger.error('Failed to rebuild Redis client:', rebuildError);
+        }
+      }
+    }
+  }, 30000); // Check every 30 seconds
+};
+
+// Start monitoring when the module is loaded
+monitorRedisHealth();
+
 export { analytics, app };