Bug 1 prefix

examples/users-microservice/package.json

@@ -36,7 +36,6 @@
     "@nestjs/schematics": "^9.2.0",
     "@nestjs/testing": "^9.4.3",
     "@types/express": "^4.17.21",
-    "@types/ioredis": "^4.28.10",
     "@types/jest": "28.1.8",
     "@types/node": "^16.18.101",
     "@types/supertest": "^2.0.16",

examples/users-microservice/src/main.ts

@@ -11,6 +11,7 @@ async function bootstrap() {
     strategy: new RedisStreamStrategy({
       connection: {
         url: '0.0.0.0:6379',
+        // keyPrefix: 'my-key-prefix:',
       },
       streams: {
         block: 5000,

lib/redis.server.ts

@@ -96,7 +96,15 @@ export class RedisStreamStrategy
     try {
       if (!this.redis) throw new Error('Redis instance not found.');
 
-      await this.redis.xgroup('CREATE', stream, consumerGroup, '$', 'MKSTREAM');
+      const modifiedStreamKey = this.prependPrefix(stream);
+
+      await this.redis.xgroup(
+        'CREATE',
+        modifiedStreamKey,
+        consumerGroup,
+        '$',
+        'MKSTREAM',
+      );
 
       return true;
     } catch (error) {
@@ -106,7 +114,7 @@ export class RedisStreamStrategy
           'Consumer Group "' +
             consumerGroup +
             '" already exists for stream: ' +
-            stream,
+            this.prependPrefix(stream),
         );
         return true;
       } else {
@@ -214,12 +222,13 @@ export class RedisStreamStrategy
 
   private async notifyHandlers(stream: string, messages: any[]) {
     try {
-      const handler = this.streamHandlerMap[stream];
+      const modifiedStream = this.stripPrefix(stream);
+      const handler = this.streamHandlerMap[modifiedStream];
 
       await Promise.all(
         messages.map(async (message) => {
           let ctx = new RedisStreamContext([
-            stream,
+            modifiedStream,
             message[0], // message id needed for ACK.
             this.options?.streams?.consumerGroup,
             this.options?.streams?.consumer,
@@ -269,10 +278,14 @@ export class RedisStreamStrategy
         'BLOCK',
         this.options?.streams?.block || 0,
         'STREAMS',
-        ...(Object.keys(this.streamHandlerMap) as string[]), // streams keys
-        ...(Object.keys(this.streamHandlerMap) as string[]).map(
-          (stream: string) => '>',
-        ), // '>', this is needed for xreadgroup as id.
+        ...(Object.keys(this.streamHandlerMap).map((s) =>
+          this.stripPrefix(s),
+        ) as string[]), // streams keys
+        ...(
+          Object.keys(this.streamHandlerMap).map((s) =>
+            this.stripPrefix(s),
+          ) as string[]
+        ).map((stream: string) => '>'), // '>', this is needed for xreadgroup as id.
       );
 
       // if BLOCK time ended, and results are null, listen again.
@@ -285,10 +298,32 @@ export class RedisStreamStrategy
 
       return this.listenOnStreams();
     } catch (error) {
+      console.log('Error in listenOnStreams: ', error);
       this.logger.error(error);
     }
   }
 
+  // When the stream handler name is stored in streamHandlerMap, its stored WITH the key prefix, so sending additional redis commands when using the prefix with the existing key will cause a duplicate prefix. This ensures to strip the first occurrence of the prefix when binding listeners.
+  private stripPrefix(streamHandlerName: string) {
+    const keyPrefix = this?.redis?.options?.keyPrefix;
+    if (!keyPrefix || !streamHandlerName.startsWith(keyPrefix)) {
+      return streamHandlerName;
+    }
+    // Replace just the first instance of the substring
+    return streamHandlerName.replace(keyPrefix, '');
+  }
+
+  // xgroup CREATE command with ioredis does not automatically prefix the keyPrefix, though many other commands do, such as xreadgroup.
+  // https://github.com/redis/ioredis/issues/1659
+  private prependPrefix(key: string) {
+    const keyPrefix = this?.redis?.options?.keyPrefix;
+    if (keyPrefix && !key.startsWith(keyPrefix)) {
+      return `${keyPrefix}${key}`;
+    } else {
+      return key;
+    }
+  }
+
   // for redis instances. need to add mechanism to try to connect back.
   public handleError(stream: any) {
     stream.on(ERROR_EVENT, (err: any) => {