diff --git a/package.json b/package.json index ba67d3f..a638f15 100644 --- a/package.json +++ b/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": { diff --git a/src/transports/redis.ts b/src/transports/redis.ts index 0900732..491181a 100644 --- a/src/transports/redis.ts +++ b/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 + + 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 + + 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 diff --git a/tests/drivers/redis_transport.spec.ts b/tests/drivers/redis_transport.spec.ts index dc69e9a..90add29 100644 --- a/tests/drivers/redis_transport.spec.ts +++ b/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(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((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) { + 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((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() }) diff --git a/yarn.lock b/yarn.lock index deebef1..d3013b8 100644 --- a/yarn.lock +++ b/yarn.lock @@ -122,6 +122,7 @@ __metadata: del-cli: "npm:^7.0.0" eslint: "npm:^10.6.0" ioredis: "npm:^5.11.1" + ioredis-v6: "npm:ioredis@^6.0.0" mqtt: "npm:^5.15.2" object-hash: "npm:^3.0.0" prettier: "npm:^3.9.4" @@ -130,7 +131,7 @@ __metadata: tsup: "npm:^8.5.1" typescript: "npm:^5.9.3" peerDependencies: - ioredis: ^5.0.0 + ioredis: ^5.0.0 || ^6.0.0 peerDependenciesMeta: ioredis: optional: true @@ -704,6 +705,13 @@ __metadata: languageName: node linkType: hard +"@ioredis/commands@npm:2.0.0": + version: 2.0.0 + resolution: "@ioredis/commands@npm:2.0.0" + checksum: 10c0/2fb5edb9782790c24a375cc78bd69d45ac6c268d7d7ed8e534f2d4ecc28707bde9882a04eecd68be369ae784801bf93fa267b9a76f86418b252a08f341934e35 + languageName: node + linkType: hard + "@isaacs/cliui@npm:^8.0.2": version: 8.0.2 resolution: "@isaacs/cliui@npm:8.0.2" @@ -3922,6 +3930,20 @@ __metadata: languageName: node linkType: hard +"ioredis-v6@npm:ioredis@^6.0.0": + version: 6.0.0 + resolution: "ioredis@npm:6.0.0" + dependencies: + "@ioredis/commands": "npm:2.0.0" + cluster-key-slot: "npm:1.1.1" + debug: "npm:4.4.3" + denque: "npm:2.1.0" + redis-errors: "npm:1.2.0" + standard-as-callback: "npm:2.1.0" + checksum: 10c0/6f74e1d298440c0d814e6bf9b3acdb0991d4c1862d3fa64c3828668bcc306823f7845710abdc5eab7986a35d86ec8c86c4bb9c9c7d727636d385ecb2cd40abd7 + languageName: node + linkType: hard + "ioredis@npm:^5.11.1": version: 5.11.1 resolution: "ioredis@npm:5.11.1"