fix: handle string values in push message reply comparison

Buffer.equals() was failing when reply[0] was a string instead of
a Buffer, causing hangs on push notifications. Now converts strings
to Buffers before comparison in PubSub and commands-queue handlers.

Changes:
 - PubSub.isStatusReply: convert reply[0] to Buffer if string
 - PubSub.isShardedUnsubscribe: convert reply[0] to Buffer if string
 - PubSub.handleMessageReply: convert reply[0] to Buffer if string
 - commands-queue PONG handler: convert reply[0] to Buffer if string

Author: nkaradzhov

packages/client/lib/client/commands-queue.ts

@@ -290,7 +290,8 @@ export default class RedisCommandsQueue {
       if (Array.isArray(reply)) {
         if (this.#onPush(reply)) return;
 
-        if (PONG.equals(reply[0] as Buffer)) {
+        const firstElement = typeof reply[0] === 'string' ? Buffer.from(reply[0]) : reply[0];
+        if (PONG.equals(firstElement as Buffer)) {
           const { resolve, typeMapping } = this.#waitingForReply.shift()!,
             buffer = ((reply[1] as Buffer).length === 0 ? reply[0] : reply[1]) as Buffer;
           resolve(typeMapping?.[RESP_TYPES.SIMPLE_STRING] === Buffer ? buffer : buffer.toString());

packages/client/lib/client/pub-sub.ts

@@ -52,17 +52,19 @@ export type PubSubCommand = (
 
 export class PubSub {
   static isStatusReply(reply: Array<Buffer>): boolean {
+    const firstElement = typeof reply[0] === 'string' ? Buffer.from(reply[0]) : reply[0];
     return (
-      COMMANDS[PUBSUB_TYPE.CHANNELS].subscribe.equals(reply[0]) ||
-      COMMANDS[PUBSUB_TYPE.CHANNELS].unsubscribe.equals(reply[0]) ||
-      COMMANDS[PUBSUB_TYPE.PATTERNS].subscribe.equals(reply[0]) ||
-      COMMANDS[PUBSUB_TYPE.PATTERNS].unsubscribe.equals(reply[0]) ||
-      COMMANDS[PUBSUB_TYPE.SHARDED].subscribe.equals(reply[0])
+      COMMANDS[PUBSUB_TYPE.CHANNELS].subscribe.equals(firstElement) ||
+      COMMANDS[PUBSUB_TYPE.CHANNELS].unsubscribe.equals(firstElement) ||
+      COMMANDS[PUBSUB_TYPE.PATTERNS].subscribe.equals(firstElement) ||
+      COMMANDS[PUBSUB_TYPE.PATTERNS].unsubscribe.equals(firstElement) ||
+      COMMANDS[PUBSUB_TYPE.SHARDED].subscribe.equals(firstElement)
     );
   }
 
   static isShardedUnsubscribe(reply: Array<Buffer>): boolean {
-    return COMMANDS[PUBSUB_TYPE.SHARDED].unsubscribe.equals(reply[0]);
+    const firstElement = typeof reply[0] === 'string' ? Buffer.from(reply[0]) : reply[0];
+    return COMMANDS[PUBSUB_TYPE.SHARDED].unsubscribe.equals(firstElement);
   }
 
   static #channelsArray(channels: string | Array<string>) {
@@ -371,22 +373,23 @@ export class PubSub {
   }
 
   handleMessageReply(reply: Array<Buffer>): boolean {
-    if (COMMANDS[PUBSUB_TYPE.CHANNELS].message.equals(reply[0])) {
+    const firstElement = typeof reply[0] === 'string' ? Buffer.from(reply[0]) : reply[0];
+    if (COMMANDS[PUBSUB_TYPE.CHANNELS].message.equals(firstElement)) {
       this.#emitPubSubMessage(
         PUBSUB_TYPE.CHANNELS,
         reply[2],
         reply[1]
       );
       return true;
-    } else if (COMMANDS[PUBSUB_TYPE.PATTERNS].message.equals(reply[0])) {
+    } else if (COMMANDS[PUBSUB_TYPE.PATTERNS].message.equals(firstElement)) {
       this.#emitPubSubMessage(
         PUBSUB_TYPE.PATTERNS,
         reply[3],
         reply[2],
         reply[1]
       );
       return true;
-    } else if (COMMANDS[PUBSUB_TYPE.SHARDED].message.equals(reply[0])) {
+    } else if (COMMANDS[PUBSUB_TYPE.SHARDED].message.equals(firstElement)) {
       this.#emitPubSubMessage(
         PUBSUB_TYPE.SHARDED,
         reply[2],