fix(redis): detect an existing connection by shape instead of instanceof, and support ioredis 6

package.json

@@ -49,6 +49,7 @@
     "del-cli": "^7.0.0",
     "eslint": "^10.6.0",
     "ioredis": "^5.11.1",
+    "ioredis-v6": "npm:ioredis@^6.0.0",
     "mqtt": "^5.15.2",
     "prettier": "^3.9.4",
     "release-it": "^20.2.1",
@@ -62,7 +63,7 @@
     "object-hash": "^3.0.0"
   },
   "peerDependencies": {
-    "ioredis": "^5.0.0"
+    "ioredis": "^5.0.0 || ^6.0.0"
   },
   "peerDependenciesMeta": {
     "ioredis": {

src/transports/redis.ts

@@ -5,8 +5,10 @@
  * @copyright BoringNode
  */
 
-import { Redis, Cluster } from 'ioredis'
+import { Redis } from 'ioredis'
+import type { Cluster } from 'ioredis'
 import { assert } from '@poppinss/utils/assert'
+import { InvalidArgumentsException } from '@poppinss/utils/exception'
 
 import debug from '../debug.js'
 import { JsonEncoder } from '../encoders/json_encoder.js'
@@ -24,6 +26,58 @@ export function redis(config: RedisTransportConfig, encoder?: TransportEncoder)
   return () => new RedisTransport(config, encoder)
 }
 
+/**
+ * Detect an existing `ioredis` client by its shape rather than with `instanceof`.
+ *
+ * `instanceof` is evaluated against the copy of `ioredis` this package resolved.
+ * When the host application resolves a different copy - a different major, or
+ * simply a duplicated one in the dependency tree - the check returns `false` for
+ * a perfectly valid client and we would fall through to `new Redis(client)`,
+ * which silently connects to `127.0.0.1:6379`.
+ *
+ * `duplicate` and `sendCommand` are on the prototype of both `Redis` and
+ * `Cluster` in every major, and neither name exists in `RedisOptions` or
+ * `ClusterOptions`, so an options object can never be mistaken for a client.
+ */
+function isRedisClient(value: unknown): value is Redis | Cluster {
+  if (typeof value !== 'object' || value === null) return false
+
+  const candidate = value as Partial<Redis>
+
+  return typeof candidate.duplicate === 'function' && typeof candidate.sendCommand === 'function'
+}
+
+/**
+ * `Redis#duplicate` and `Cluster#duplicate` are both parameterless-callable at
+ * runtime, but TypeScript cannot resolve the call against a `Redis | Cluster`
+ * union: from ioredis 6 on, both signatures are generic over the reply mapping
+ * and none of them is compatible with the other. Going through a narrow local
+ * signature keeps the source compiling against ioredis 5 and 6 alike.
+ */
+function duplicateClient(client: Redis | Cluster): Redis | Cluster {
+  return (client.duplicate as unknown as () => Redis | Cluster)()
+}
+
+/**
+ * Values that look like a client - they carry a `status`, an `options` bag and
+ * an `emit` method - but that we could not recognize as one. None of those
+ * names are valid `ioredis` options, so a legitimate configuration object never
+ * lands here. Rather than quietly building a connection to `127.0.0.1:6379`
+ * out of it, we fail loudly.
+ */
+function looksLikeUnsupportedRedisClient(value: unknown): boolean {
+  if (typeof value !== 'object' || value === null) return false
+
+  const candidate = value as Record<string, unknown>
+
+  return (
+    typeof candidate.status === 'string' &&
+    typeof candidate.options === 'object' &&
+    candidate.options !== null &&
+    typeof candidate.emit === 'function'
+  )
+}
+
 export class RedisTransport implements Transport {
   readonly #publisher: Redis | Cluster
   readonly #subscriber: Redis | Cluster
@@ -51,11 +105,18 @@ export class RedisTransport implements Transport {
      * If an existing Redis or Cluster instance is passed, we duplicate it
      * to have separate connections for publisher and subscriber
      */
-    if (options instanceof Redis || options instanceof Cluster) {
-      this.#publisher = options.duplicate()
-      this.#subscriber = options.duplicate()
+    if (isRedisClient(options)) {
+      this.#publisher = duplicateClient(options)
+      this.#subscriber = duplicateClient(options)
       this.#useMessageBuffer = transportOptions?.useMessageBuffer ?? false
     } else {
+      if (looksLikeUnsupportedRedisClient(options)) {
+        throw new InvalidArgumentsException(
+          'Cannot use the given Redis connection with "RedisTransport". Expected an "ioredis" ' +
+            'client exposing "duplicate()" and "sendCommand()", or a connection options object.'
+        )
+      }
+
       // @ts-expect-error - merged definitions of overloaded constructor is not public
       this.#publisher = new Redis(options)
       // @ts-expect-error - merged definitions of overloaded constructor is not public

tests/drivers/redis_transport.spec.ts

@@ -8,11 +8,52 @@
 import { setImmediate, setTimeout } from 'node:timers/promises'
 import { test } from '@japa/runner'
 import { Redis, Cluster } from 'ioredis'
+import { Redis as ForeignRedis, Cluster as ForeignCluster } from 'ioredis-v6'
 import { RedisContainer, type StartedRedisContainer } from '@testcontainers/redis'
 import { RedisTransport } from '../../src/transports/redis.js'
 import { JsonEncoder } from '../../src/encoders/json_encoder.js'
 import { type TransportEncoder, type TransportMessage } from '../../src/types/main.js'
 
+/**
+ * `ioredis-v6` is a second, real copy of `ioredis` installed side by side with
+ * the one this package resolves, through the npm alias
+ * `"ioredis-v6": "npm:ioredis@^6.0.0"`. A client built from it is exactly what
+ * an application running `@adonisjs/redis@11` hands to a bus resolved against
+ * `ioredis` 5: same API, different classes, so `instanceof` says `false`.
+ *
+ * The cast is part of the scenario too - the two copies ship two unrelated sets
+ * of types - but only the runtime behaviour is under test here.
+ */
+function asForeignClient(client: ForeignRedis): Redis
+function asForeignClient(client: ForeignCluster): Cluster
+function asForeignClient(client: ForeignRedis | ForeignCluster) {
+  return client as unknown as Redis | Cluster
+}
+
+/**
+ * Records what `duplicate()` returns, so a test can tell "the transport reused
+ * the connection" from "the transport built a brand new one".
+ *
+ * The spy is an own property shadowing the prototype method, so the client
+ * remains an instance of its own `ioredis` copy - which is the very thing these
+ * tests are about.
+ */
+function spyOnDuplicate<T extends ForeignRedis | ForeignCluster>(client: T): T[] {
+  const duplicates: T[] = []
+  const duplicate = (client.duplicate as unknown as () => T).bind(client)
+
+  Object.defineProperty(client, 'duplicate', {
+    configurable: true,
+    value: () => {
+      const duplicated = duplicate()
+      duplicates.push(duplicated)
+      return duplicated
+    },
+  })
+
+  return duplicates
+}
+
 test.group('Redis Transport', (group) => {
   let container: StartedRedisContainer
 
@@ -360,4 +401,222 @@ test.group('Redis Transport', (group) => {
     await setTimeout(200)
     await transport2.publish('testing-channel', data)
   }).waitForDone()
+
+  test('should reuse a client coming from another copy of ioredis', async ({ assert, cleanup }) => {
+    const foreignClient = new ForeignRedis({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    })
+
+    cleanup(async () => {
+      await foreignClient.quit()
+    })
+
+    /**
+     * The premise of the test: a genuine `ioredis` client, built from a genuine
+     * second copy of the package, which is therefore not an instance of the
+     * classes this package resolved.
+     */
+    assert.isFalse((ForeignRedis as unknown) === (Redis as unknown))
+    assert.instanceOf(foreignClient, ForeignRedis)
+    assert.isFalse(foreignClient instanceof Redis)
+    assert.isFalse(foreignClient instanceof Cluster)
+
+    const duplicates = spyOnDuplicate(foreignClient)
+    const transport = new RedisTransport(asForeignClient(foreignClient)).setId('bus1')
+    cleanup(() => transport.disconnect())
+
+    /**
+     * The connection must have been duplicated (publisher + subscriber) rather
+     * than handed to `new Redis()` as if it were an options object, which would
+     * silently dial 127.0.0.1:6379.
+     */
+    assert.lengthOf(duplicates, 2)
+    assert.instanceOf(duplicates[0], ForeignRedis)
+
+    /**
+     * And the messages must really land on the configured server, not on the
+     * default one. The witness connects to the container explicitly, so it only
+     * ever sees what was published there.
+     */
+    const witness = new Redis({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    })
+    cleanup(async () => {
+      await witness.quit()
+    })
+
+    const received = new Promise<string>((resolve) => {
+      witness.on('message', (_channel, message) => resolve(message))
+    })
+    await witness.subscribe('foreign-client-channel')
+
+    await transport.publish('foreign-client-channel', 'test')
+
+    const message = await Promise.race([received, setTimeout(2000).then(() => 'timed-out')])
+
+    assert.deepEqual(JSON.parse(message), { payload: 'test', busId: 'bus1' })
+  })
+
+  test('should reuse a cluster client coming from another copy of ioredis', async ({
+    assert,
+    cleanup,
+  }) => {
+    const foreignCluster = new ForeignCluster([{ host: '127.0.0.1', port: 7000 }], {
+      lazyConnect: true,
+    })
+
+    cleanup(() => {
+      foreignCluster.disconnect()
+    })
+
+    assert.isFalse((ForeignCluster as unknown) === (Cluster as unknown))
+    assert.instanceOf(foreignCluster, ForeignCluster)
+    assert.isFalse(foreignCluster instanceof Redis)
+    assert.isFalse(foreignCluster instanceof Cluster)
+
+    const duplicates = spyOnDuplicate(foreignCluster)
+    const transport = new RedisTransport(asForeignClient(foreignCluster))
+    cleanup(() => {
+      for (const duplicated of duplicates) duplicated.disconnect()
+    })
+
+    assert.lengthOf(duplicates, 2)
+    assert.instanceOf(duplicates[0], ForeignCluster)
+    assert.isFalse(duplicates[0] instanceof Cluster)
+
+    transport.onReconnect(() => {})
+    assert.equal(duplicates[1].listenerCount('reconnecting'), 1)
+  })
+
+  test('should keep useMessageBuffer when given a client from another copy of ioredis', async ({
+    assert,
+    cleanup,
+  }) => {
+    class BinaryEncoder implements TransportEncoder {
+      encode(message: TransportMessage<any>) {
+        return Buffer.from(JSON.stringify(message))
+      }
+
+      decode(data: string | Buffer) {
+        const buffer = Buffer.isBuffer(data) ? data : Buffer.from(data, 'binary')
+        return JSON.parse(buffer.toString())
+      }
+    }
+
+    const subscriberClient = new ForeignRedis({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    })
+
+    const publisherClient = new ForeignRedis({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    })
+
+    cleanup(async () => {
+      await subscriberClient.quit()
+      await publisherClient.quit()
+    })
+
+    assert.instanceOf(subscriberClient, ForeignRedis)
+    assert.isFalse(subscriberClient instanceof Redis)
+
+    const duplicates = spyOnDuplicate(subscriberClient)
+
+    const transport1 = new RedisTransport(asForeignClient(subscriberClient), new BinaryEncoder(), {
+      useMessageBuffer: true,
+    }).setId('bus1')
+
+    const transport2 = new RedisTransport(asForeignClient(publisherClient), new BinaryEncoder(), {
+      useMessageBuffer: true,
+    }).setId('bus2')
+
+    cleanup(async () => {
+      await transport1.disconnect()
+      await transport2.disconnect()
+    })
+
+    /**
+     * `useMessageBuffer` only reaches the transport through the third argument,
+     * which is ignored on the options-object branch. If the client is not
+     * recognized, it silently degrades to `false` and the subscriber listens on
+     * "message" instead of "messageBuffer" - decoding binary payloads through a
+     * lossy string.
+     */
+    assert.lengthOf(duplicates, 2)
+    assert.equal(duplicates[1].listenerCount('messageBuffer'), 1)
+    assert.equal(duplicates[1].listenerCount('message'), 0)
+
+    const data = ['foo', '👍']
+
+    /**
+     * The exact bytes the publisher writes. Listening alongside the transport on
+     * its own subscriber lets us assert the payload survived the trip untouched,
+     * rather than through a UTF-8 round trip that happens to be lossless.
+     */
+    const expectedBytes = new BinaryEncoder().encode({ payload: data, busId: 'bus2' })
+    let receivedBytes: Buffer | undefined
+    duplicates[1].on('messageBuffer', (_channel: Buffer, message: Buffer) => {
+      receivedBytes = message
+    })
+
+    let resolvePayload!: (payload: unknown) => void
+    const payloadReceived = new Promise<unknown>((resolve) => (resolvePayload = resolve))
+    await transport1.subscribe('testing-channel', resolvePayload)
+
+    await setTimeout(200)
+    await transport2.publish('testing-channel', data)
+
+    const payload = await Promise.race([payloadReceived, setTimeout(2000).then(() => 'timed-out')])
+
+    assert.deepEqual(payload, data)
+    assert.isTrue(Buffer.isBuffer(receivedBytes))
+    assert.isTrue(receivedBytes!.equals(expectedBytes as Buffer))
+  })
+
+  test('should throw instead of dialing localhost for an unusable client', async ({ assert }) => {
+    const unusable = {
+      status: 'ready',
+      options: { host: 'redis.internal', port: 6380 },
+      emit: () => true,
+    }
+
+    assert.throws(
+      // @ts-expect-error - deliberately not a valid argument
+      () => new RedisTransport(unusable),
+      /Cannot use the given Redis connection with "RedisTransport"/
+    )
+  })
+
+  test('should still build a new connection from a plain options object', async ({
+    assert,
+    cleanup,
+  }, done) => {
+    assert.plan(1)
+
+    const transport1 = new RedisTransport({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    }).setId('bus1')
+
+    const transport2 = new RedisTransport({
+      host: container.getHost(),
+      port: container.getMappedPort(6379),
+    }).setId('bus2')
+
+    cleanup(async () => {
+      await transport1.disconnect()
+      await transport2.disconnect()
+    })
+
+    await transport1.subscribe('options-object-channel', (payload) => {
+      assert.equal(payload, 'test')
+      done()
+    })
+
+    await setTimeout(200)
+    await transport2.publish('options-object-channel', 'test')
+  }).waitForDone()
 })