diff --git a/factories/redis_manager.ts b/factories/redis_manager.ts index b8710e6..41a9a86 100644 --- a/factories/redis_manager.ts +++ b/factories/redis_manager.ts @@ -9,15 +9,13 @@ import RedisManager from '../src/redis_manager.ts' import { LoggerFactory } from '@adonisjs/core/factories/logger' -import type { RedisClusterConnectionConfig, RedisConnectionConfig } from '../src/types.ts' +import type { RedisConnectionsList } from '../src/types.ts' /** * Redis manager factory is used to create an instance of the redis * manager for testing */ -export class RedisManagerFactory< - ConnectionsList extends Record, -> { +export class RedisManagerFactory { #config: { connection: keyof ConnectionsList connections: ConnectionsList diff --git a/index.ts b/index.ts index 9ba5818..fec9d89 100644 --- a/index.ts +++ b/index.ts @@ -15,5 +15,6 @@ export { RedisCheck } from './src/checks/redis_check.ts' export * as tracingChannels from './src/tracing_channels.ts' export { default as RedisManager } from './src/redis_manager.ts' export { RedisConnection } from './src/connections/redis_connection.ts' +export { NodeRedisConnection } from './src/connections/node_redis_connection.ts' export { RedisMemoryUsageCheck } from './src/checks/redis_memory_usage_check.ts' export { RedisClusterConnection } from './src/connections/redis_cluster_connection.ts' diff --git a/package-lock.json b/package-lock.json index 758f80d..a9c4c97 100644 --- a/package-lock.json +++ b/package-lock.json @@ -33,6 +33,7 @@ "dotenv": "^17.4.2", "eslint": "^10.8.1", "prettier": "^3.9.6", + "redis": "^6.2.1", "release-it": "^21.0.2", "tsdown": "^0.22.14", "typescript": "^5.9.3" @@ -74,7 +75,6 @@ "integrity": "sha512-KWsTfkITDozW4csD4YofDEbbWuORKI056VgEr9WBwjIdrzIC7hqPxLtnszFVmyv14THPi0SRBTN/fIYQ6/SQ4Q==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/hooks": "^7.3.1", "@poppinss/macroable": "^1.1.2", @@ -102,7 +102,6 @@ "integrity": "sha512-/P0SEKMAeiIPlAqct/WPNTw5KdHfxmCDWJtiHDN+GL+Vz5gZezKFU8vVxBTuAwIIMhDNlTqkEFKGyrHCqNnSiQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@adonisjs/env": "^7.0.0", "@antfu/install-pkg": "^2.0.1", @@ -163,7 +162,6 @@ "integrity": "sha512-YVDRL8xHCtM6iMnAefOBaz6iXVpojwBPDQWPKxnVSucycYeNGrGitJiLy+cGaeAU7Gjm8al9SJRJt3rRPr5PKg==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/utils": "^7.0.0" }, @@ -346,7 +344,6 @@ "integrity": "sha512-RnmDPWz2imVp/B74xitxCPqTdoP07bZvfJe1bh9CD9Rmia4jjDvehZF67KFyGNMZ24MuKasqs3jOcM1vGJp0GA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/utils": "^7.0.0", "parse-imports": "^3.0.0" @@ -401,7 +398,6 @@ "integrity": "sha512-yTrnmo1Sn3k6d1/O9J+8mpaMMAflFwYM1P+Y8drAr9thVT55G9ywxRUSXeEp2RnOyxNcGVdjfdRACt6M/aP5Dw==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/macroable": "^1.1.2", "@poppinss/matchit": "^3.2.0", @@ -463,7 +459,6 @@ "integrity": "sha512-MmUlp8xBMT6zZy0+vnQcQjHIlNfU4pUJARlINr7Bqha9BvhIn03QZgJL5QJ+kJe1tl6ZpNAryoRTJUiOk/wINQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/utils": "^7.0.0", "abstract-logging": "^2.0.1", @@ -762,7 +757,6 @@ "integrity": "sha512-+UL9erFUngnxNEDdXSKCnGQA8D62C/x2vxLF5vc9zaX7T2JolV/bGjpZAvFLzlY2gIffHyw+KsqsEWod5+zV0w==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/utils": "^7.0.1" }, @@ -1364,7 +1358,6 @@ "integrity": "sha512-Krgrcee01BN1StlVwK5JQP6LL5t3DE3uFNbfFoDTfW7kQuHB0xh6yfaV0hrgcoiEjsqmm2OOsVWeju9aXK4vIA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/macroable": "^1.1.0", "@types/chai": "^5.2.3", @@ -1452,7 +1445,6 @@ "integrity": "sha512-WCnTd1q2EpbKKa96NzL16kVxJXVLRj1VqbswNAn17hYSuMlNKKhPGNbAosB32QZVFcoe9fv4Ebh1HtjyAT/viw==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@japa/core": "^10.4.0", "@japa/errors-printer": "^4.1.4", @@ -1595,7 +1587,6 @@ "integrity": "sha512-DcB0M3KFgr9ECI328lhBMVsyFT2DnmNucSBTqEN3exyNKUzkkpUSCHmTRcunF41Eou2TIQKW4seewri8ON9bSA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@octokit/auth-token": "^6.0.0", "@octokit/graphql": "^9.0.4", @@ -2035,6 +2026,93 @@ "url": "https://github.com/sponsors/sxzz" } }, + "node_modules/@redis/bloom": { + "version": "6.2.1", + "resolved": "https://registry.npmjs.org/@redis/bloom/-/bloom-6.2.1.tgz", + "integrity": "sha512-huQgNLaCIZfQ9SeLn4q9124uOUd8HbZDYHwwUzNcRgHqCHiHKl2dDxMqJCeWh8cMqZAoWuHR8XnWbDMIf+o7ag==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.0.0" + }, + "peerDependencies": { + "@redis/client": "^6.2.1" + } + }, + "node_modules/@redis/client": { + "version": "6.2.1", + "resolved": "https://registry.npmjs.org/@redis/client/-/client-6.2.1.tgz", + "integrity": "sha512-LzxBY7SIBvvJiyCgcaJZZakE3fJrZZ++i24+EDW9fKpCl68D35uJcKFpZZwCfOoG9WZTbyZlMzMeM0gtOAMU9Q==", + "dev": true, + "license": "MIT", + "dependencies": { + "cluster-key-slot": "1.1.2" + }, + "engines": { + "node": ">= 20.0.0" + }, + "peerDependencies": { + "@node-rs/xxhash": "^1.1.0", + "@opentelemetry/api": ">=1 <2" + }, + "peerDependenciesMeta": { + "@node-rs/xxhash": { + "optional": true + }, + "@opentelemetry/api": { + "optional": true + } + } + }, + "node_modules/@redis/client/node_modules/cluster-key-slot": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz", + "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==", + "dev": true, + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/@redis/json": { + "version": "6.2.1", + "resolved": "https://registry.npmjs.org/@redis/json/-/json-6.2.1.tgz", + "integrity": "sha512-AFIUJ8Gj0DaaSBHYuSt8+O0oYWM+50OK1c0OmodB7XERIA8+BbyV3O4v76f9iccWasd1/7qjfZTpuzexUaZtrQ==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.0.0" + }, + "peerDependencies": { + "@redis/client": "^6.2.1" + } + }, + "node_modules/@redis/search": { + "version": "6.2.1", + "resolved": "https://registry.npmjs.org/@redis/search/-/search-6.2.1.tgz", + "integrity": "sha512-2vfOAOyYFE7UUw3sBBlkqqruBtOUS4HRY5MtW4hp83llrwvtrTE4r22CEqXddlV+54zkLxBE4nmsIJ/dpezQrQ==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.0.0" + }, + "peerDependencies": { + "@redis/client": "^6.2.1" + } + }, + "node_modules/@redis/time-series": { + "version": "6.2.1", + "resolved": "https://registry.npmjs.org/@redis/time-series/-/time-series-6.2.1.tgz", + "integrity": "sha512-kiYniph04dJOole+L359B6C9E+jYS2uDP7hca6Onj0xF38ZIpyxARO0Iq0W4ZRn1e8Q6vqW00QFZVSMRA/2Ijw==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.0.0" + }, + "peerDependencies": { + "@redis/client": "^6.2.1" + } + }, "node_modules/@release-it/conventional-changelog": { "version": "12.0.0", "resolved": "https://registry.npmjs.org/@release-it/conventional-changelog/-/conventional-changelog-12.0.0.tgz", @@ -2804,7 +2882,6 @@ "integrity": "sha512-5IviulTZeRNp2vAJ514cc/HUlY5nZ9fCbq9DMyC52BrhFZACo3nI0R7qBxhQmo/d27NFe96ur/b7Wwxklda+kg==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "undici-types": "~8.3.0" } @@ -2874,7 +2951,6 @@ "integrity": "sha512-fUBfTuuEulWqX6V8+O3PtScV01tzYYRUDTAirHFKoRAt7nOzoGiPt0M/bB47wWNy0coOOcgEwAMUtBpykMxl6w==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@typescript-eslint/scope-manager": "8.67.0", "@typescript-eslint/types": "8.67.0", @@ -3429,7 +3505,6 @@ "integrity": "sha512-lGq+9yr1/GuAWaVYIHRjvvySG5/4VfKIvC8EWxStPdcDh/Ka7FG3twP6v4d5BkravUilhIAsG4Qj83t02LWUPQ==", "dev": true, "license": "MIT", - "peer": true, "bin": { "acorn": "bin/acorn" }, @@ -3688,7 +3763,6 @@ } ], "license": "MIT", - "peer": true, "dependencies": { "baseline-browser-mapping": "^2.11.12", "caniuse-lite": "^1.0.30001809", @@ -5031,7 +5105,6 @@ "integrity": "sha512-wqA7W2jbsC/BnV9Iv1UZpKVFkO1AdNoSmYW8NWG4HNOBbkAMvIqDZ27pI2f07dqn583NcIC44ckjAcOXDL1QbQ==", "dev": true, "license": "MIT", - "peer": true, "workspaces": [ "packages/*" ], @@ -5091,7 +5164,6 @@ "integrity": "sha512-82GZUjRS0p/jganf6q1rEO25VSoHH0hKPCTrgillPjdI/3bgBhAE1QzHrHTizjpRvy6pGAvKjDJtk2pF9NDq8w==", "dev": true, "license": "MIT", - "peer": true, "bin": { "eslint-config-prettier": "bin/cli.js" }, @@ -6540,7 +6612,6 @@ "integrity": "sha512-AC/7JofJvZGrrneWNaEnJeOLUx+JlGt7tNa0wZiRPT4MY1wmfKjt2+6O2p2uz2+skll8OZZmJMNqeke7kKbNgQ==", "dev": true, "license": "MIT", - "peer": true, "bin": { "jiti": "lib/jiti-cli.mjs" } @@ -7660,7 +7731,6 @@ "integrity": "sha512-OpN0zzVdiaiAhxpuuj5efpIS4sY9j7bY6uR5mnj5yPzGkdkjNKSJeUThPb60Jw29QuAZgA4o+/iB49kFiaBX6g==", "dev": true, "license": "MIT", - "peer": true, "bin": { "prettier": "bin/prettier.cjs" }, @@ -7919,8 +7989,7 @@ "resolved": "https://registry.npmjs.org/quickjs-wasi/-/quickjs-wasi-2.2.0.tgz", "integrity": "sha512-zQxXmQMrEoD3S+jQdYsloq4qAuaxKFHZj6hHqOYGwB2iQZH+q9e/lf5zQPXCKOk0WJuAjzRFbO4KwHIp2D05Iw==", "dev": true, - "license": "MIT", - "peer": true + "license": "MIT" }, "node_modules/raw-body": { "version": "3.0.2", @@ -8055,6 +8124,23 @@ "node": ">= 12.13.0" } }, + "node_modules/redis": { + "version": "6.2.1", + "resolved": "https://registry.npmjs.org/redis/-/redis-6.2.1.tgz", + "integrity": "sha512-Z9VHtgYs48PiQC77X9O2Er8Hj4T+5BtFjT91/vi5Is1D04N72cA946ZslM1ImJw8ZctFBZWAVjM7S5wJNeHMpg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@redis/bloom": "6.2.1", + "@redis/client": "6.2.1", + "@redis/json": "6.2.1", + "@redis/search": "6.2.1", + "@redis/time-series": "6.2.1" + }, + "engines": { + "node": ">= 20.0.0" + } + }, "node_modules/redis-errors": { "version": "1.2.0", "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz", @@ -8103,7 +8189,6 @@ } ], "license": "MIT", - "peer": true, "dependencies": { "@inquirer/prompts": "8.5.2", "@octokit/rest": "22.0.1", @@ -8250,7 +8335,6 @@ "integrity": "sha512-VD2IE5PUG4Oj8zz2VGykiYd5wbnjdIiSsNQb8Qu5B+noEp+A78mu2iVvpp27g8es14Tk9rofNs5Tku9iQCS4fA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@oxc-project/types": "=0.146.0", "@rolldown/pluginutils": "^1.0.0" @@ -9219,7 +9303,6 @@ "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "dev": true, "license": "Apache-2.0", - "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -9649,7 +9732,6 @@ "integrity": "sha512-mxW3qiSnl+GRxXsaUMzv2Mbada1Y8CDltET9UxejDQe6DBYlSekghl5U5K0ReAikcHDi0G1vKZEmmo/NWAGKLA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@poppinss/colors": "^4.1.6", "@poppinss/dumper": "^0.7.0", diff --git a/package.json b/package.json index 3396fa1..756843a 100644 --- a/package.json +++ b/package.json @@ -59,6 +59,7 @@ "dotenv": "^17.4.2", "eslint": "^10.8.1", "prettier": "^3.9.6", + "redis": "^6.2.1", "release-it": "^21.0.2", "tsdown": "^0.22.14", "typescript": "^5.9.3" @@ -69,7 +70,13 @@ "ioredis": "^6.0.0" }, "peerDependencies": { - "@adonisjs/core": "^7.0.0-next.0 || ^7.0.0" + "@adonisjs/core": "^7.0.0-next.0 || ^7.0.0", + "redis": "^5.0.0 || ^6.0.0" + }, + "peerDependenciesMeta": { + "redis": { + "optional": true + } }, "keywords": [ "redis", diff --git a/src/connections/node_redis_connection.ts b/src/connections/node_redis_connection.ts new file mode 100644 index 0000000..2bda59f --- /dev/null +++ b/src/connections/node_redis_connection.ts @@ -0,0 +1,484 @@ +/* + * @adonisjs/redis + * + * (c) AdonisJS + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +import Emittery from 'emittery' +import { RuntimeException } from '@poppinss/utils/exception' +import type { RedisClientOptions, RedisClientType } from 'redis' + +import debug from '../debug.ts' +import { nodeRedisMethods } from './node_redis_methods.ts' +import type { + ConnectionEvents, + NodeRedisCommands, + NodeRedisConnectionConfig, + NodeRedisConnectionStatus, +} from '../types.ts' + +/** + * NodeRedisConnection exposes the API to run Redis commands using the + * official "redis" (node-redis) client. + * + * ## Why this class does not extend "AbstractConnection" + * + * "AbstractConnection" is written against the ioredis client. Every + * accessor on it (status, autoPipelineQueueSize, quitIoConnection, and + * so on) reads from "this.ioConnection". Making it generic enough to + * host both clients would mean rewriting the class the existing + * connections rely on, and this prototype must leave the ioredis + * code path untouched. So this connection is a sibling + * implementation that speaks the same public dialect + * (status, isReady, events, quit, disconnect). + * + * ## The async connect problem + * + * node-redis requires an awaited "connect()" call before any command + * can be issued, but "redis.connection()" resolves synchronously + * inside the IoC container, and constructors cannot be async. + * + * The answer used here: the connection object is a synchronously + * created handle. The client itself is created and connected inside + * a single memoized promise, and every command awaits that promise + * before running. The first command pays the connection cost, and + * every later one hits an already resolved promise. + * + * @example + * ```ts + * const connection = new NodeRedisConnection('main', { + * client: 'node-redis', + * socket: { host: 'localhost', port: 6379 }, + * }) + * + * // No connect() call needed, the command awaits it internally + * await connection.set('key', 'value') + * const value = await connection.get('key') + * ``` + */ +export class NodeRedisConnection extends Emittery> { + /** + * Normalized node-redis client options + */ + #config: RedisClientOptions + + /** + * Reference to the underlying node-redis client. It stays undefined + * until the connection has been established at-least once. + */ + #client?: RedisClientType + + /** + * The memoized "connect" promise. This is the crux of this class: it + * is created on the first command (or on an explicit "connect" call) + * and re-used by every subsequent command. + */ + #connection?: Promise + + /** + * Connection status. We track it by hand (from the client events) and + * re-use the ioredis vocabulary, so that code duck-typing on + * "connection.status" keeps working. + */ + #status: NodeRedisConnectionStatus = 'wait' + + /** + * Set to true after an explicit "quit" or "disconnect" call. Reviving + * the connection after that point would silently dial the server + * again, so we throw instead. + */ + #ended: boolean = false + + /** + * Lua scripts registered via "defineCommand" and executed by + * "runCommand" + */ + #scripts: Record = {} + + /** + * The last error emitted by the "error" event. We set it to null after + * the "ready" event + */ + lastError?: any + + /** + * Returns status of the main connection + */ + get status(): NodeRedisConnectionStatus { + return this.#status + } + + /** + * Reference to the underlying node-redis client. This is the + * node-redis counterpart of ".ioConnection". + * + * It is undefined until the connection has been established, since + * the client is created lazily (see the class docblock). Use + * "await connection.connect()" to get a guaranteed client. + */ + get nodeConnection(): RedisClientType | undefined { + return this.#client + } + + /** + * Create a new node-redis connection + * + * @param connectionName - Unique name for this connection + * @param config - node-redis client options + */ + constructor( + public connectionName: string, + config: NodeRedisConnectionConfig + ) { + debug('creating node-redis connection %s: %O', connectionName, config) + super() + this.#config = this.#normalizeConfig(config) + } + + /** + * Normalizes the config to be compatible with node-redis. We strip the + * "client" discriminant and unwrap the AdonisJS "Secret" values, so + * that the config file can look like the ioredis one. + */ + #normalizeConfig(config: NodeRedisConnectionConfig): RedisClientOptions { + const { client, ...rest } = config + const normalized = { ...rest } as RedisClientOptions + + if (normalized.password && typeof normalized.password !== 'string') { + normalized.password = (normalized.password as any).release() + } + + return normalized + } + + /** + * Imports the node-redis client. "redis" is an optional peer dependency, + * so we import it lazily and turn the module resolution error into an + * actionable message. + */ + static async #importClient() { + try { + const { createClient } = await import('redis') + return createClient + } catch (error) { + throw new RuntimeException( + 'Cannot find the "redis" package. It must be installed to use node-redis backed connections', + { cause: error } + ) + } + } + + /** + * Monitoring the redis client via its event emitter to cleanup things + * properly and also notify subscribers of this class. + */ + #monitorConnection(client: RedisClientType) { + client.on('connect', () => { + this.#status = 'connect' + this.emit('connect', { connection: this }) + }) + + client.on('ready', () => { + /** + * We must set the error to null when server is ready to accept + * commands + */ + this.lastError = null + this.#status = 'ready' + this.emit('ready', { connection: this }) + }) + + client.on('error', (error: any) => { + this.lastError = error + this.emit('error', { error, connection: this }) + }) + + /** + * node-redis does not share the retry delay with the "reconnecting" + * event, hence "waitTime" is always zero here. + */ + client.on('reconnecting', () => { + this.#status = 'reconnecting' + this.emit('reconnecting', { connection: this, waitTime: 0 }) + }) + + /** + * On end, we must cleanup the client and self listeners + */ + client.on('end', () => { + this.#status = 'end' + this.#client = undefined + this.#connection = undefined + + client.removeAllListeners() + this.emit('end', { connection: this }).finally(() => { + this.clearListeners(['connect', 'ready', 'error', 'reconnecting', 'end']) + }) + }) + } + + /** + * Establishes the connection (once) and returns the underlying + * node-redis client. + * + * Calling this method multiple times is safe: the promise is memoized + * and shared by every caller. You usually do not have to call it, as + * every command method awaits it internally. + * + * @example + * ```ts + * const client = await redis.connection('main').connect() + * await client.set('key', 'value', { PX: 1000 }) + * ``` + */ + connect(): Promise { + if (this.#ended) { + return Promise.reject( + new RuntimeException( + `Cannot use the "${this.connectionName}" connection after it has been closed` + ) + ) + } + + if (this.#connection) { + return this.#connection + } + + debug('connecting node-redis connection %s', this.connectionName) + this.#status = 'connecting' + + this.#connection = (async () => { + /** + * The import is lazy on purpose. "redis" is an optional peer + * dependency, so the module must never be loaded for apps that + * only use ioredis connections. + */ + const createClient = await NodeRedisConnection.#importClient() + + const client = createClient(this.#config) as unknown as RedisClientType + this.#client = client + this.#monitorConnection(client) + + await client.connect() + return client + })().catch((error) => { + /** + * Reset the memoized promise, so that a later command can retry + * instead of re-throwing the same error forever. + */ + this.#connection = undefined + this.#status = 'close' + + const client = this.#client + this.#client = undefined + if (client) { + client.removeAllListeners() + } + + throw error + }) + + return this.#connection + } + + /** + * Returns a boolean notifying if the connection is in connecting state + */ + isConnecting() { + return this.#status === 'connecting' || this.#status === 'reconnecting' + } + + /** + * Returns a boolean notifying if the connection is in ready state + */ + isReady() { + return this.#status === 'ready' || this.#status === 'connect' + } + + /** + * Returns a boolean notifying if the connection has been closed + */ + isClosed() { + return this.#status === 'end' || this.#status === 'close' + } + + /** + * Gracefully end the redis connection. Pending commands are flushed + * before the socket is closed. + */ + async quit() { + return this.#endConnection(true) + } + + /** + * Forcefully end the redis connection + */ + async disconnect() { + return this.#endConnection(false) + } + + /** + * Shared implementation for "quit" and "disconnect". + */ + async #endConnection(graceful: boolean) { + this.#ended = true + + const client = this.#client + this.#connection = undefined + + /** + * A connection that has never dialed the server has nothing to quit. + * node-redis throws a "ClientClosedError" when closing a client that + * was never opened, so we emit the "end" event by hand to keep the + * manager bookkeeping in sync. + */ + if (!client || !client.isOpen) { + this.#status = 'end' + this.#client = undefined + client?.removeAllListeners() + await this.emit('end', { connection: this }) + return + } + + if (graceful) { + await client.close() + return + } + + client.destroy() + } + + /** + * Publish a pub/sub message. + * + * Note: subscribing is not implemented in this prototype (node-redis + * needs a dedicated client for it), but publishing works on the main + * client just like it does with ioredis. + */ + publish( + channel: string, + message: string, + callback: (error: Error | null | undefined, count: number | undefined) => void + ): void + publish(channel: string, message: string): Promise + publish( + channel: string, + message: string, + callback?: (error: Error | null | undefined, count: number | undefined) => void + ): void | Promise { + const promise = this.connect().then((client) => client.publish(channel, message)) + + if (!callback) { + return promise + } + + promise.then( + (count) => callback(null, count), + (error) => callback(error, undefined) + ) + } + + /** + * Subscriptions are not part of this prototype. node-redis requires a + * dedicated client for pub/sub (like ioredis does), so the shape of + * the solution is known, it is just not implemented yet. + */ + subscribe(): void { + throw new RuntimeException( + 'Pub/Sub is not implemented for node-redis connections. Use "await connection.connect()" and subscribe using a duplicated client' + ) + } + + /** + * See "subscribe" + */ + unsubscribe(): Promise { + return Promise.reject( + new RuntimeException('Pub/Sub is not implemented for node-redis connections') + ) + } + + /** + * See "subscribe" + */ + psubscribe(): void { + throw new RuntimeException( + 'Pub/Sub is not implemented for node-redis connections. Use "await connection.connect()" and subscribe using a duplicated client' + ) + } + + /** + * See "subscribe" + */ + punsubscribe(): Promise { + return Promise.reject( + new RuntimeException('Pub/Sub is not implemented for node-redis connections') + ) + } + + /** + * Define a custom command using a LUA script. You can run the + * registered command using the "runCommand" method. + * + * ioredis defines the script on the client and exposes it as a method. + * node-redis has its own "scripts" option, but it must be given at + * client creation time, hence we keep the scripts around and EVAL + * them from "runCommand". + */ + defineCommand( + name: string, + definition: { lua?: string; numberOfKeys?: number; readOnly?: boolean } + ): this { + this.#scripts[name] = definition + return this + } + + /** + * Run a pre-registered command + * + * @example + * ```ts + * connection.defineCommand('setAndGet', { + * lua: 'redis.call("set", KEYS[1], ARGV[1]) return redis.call("get", KEYS[1])', + * numberOfKeys: 1, + * }) + * + * await connection.runCommand('setAndGet', 'greeting', 'hello') + * ``` + */ + async runCommand(command: string, ...args: any[]): Promise { + const script = this.#scripts[command] + if (!script || !script.lua) { + throw new RuntimeException(`Cannot run undefined command "${command}"`) + } + + const client = await this.connect() + const numberOfKeys = script.numberOfKeys ?? 0 + + return client.eval(script.lua, { + keys: args.slice(0, numberOfKeys).map((key) => String(key)), + arguments: args.slice(numberOfKeys).map((value) => String(value)), + }) + } +} + +export interface NodeRedisConnection extends NodeRedisCommands {} + +/** + * Proxy the supported node-redis commands to the underlying client. The + * "await this.connect()" call is what makes the async connect + * transparent to the end user. + */ +nodeRedisMethods.forEach((method) => { + ;(NodeRedisConnection.prototype as any)[method] = async function nodeRedisConnectionProxyFn( + ...args: any[] + ) { + const client = await this.connect() + return client[method](...args) + } +}) + +export default NodeRedisConnection diff --git a/src/connections/node_redis_methods.ts b/src/connections/node_redis_methods.ts new file mode 100644 index 0000000..c2db0d6 --- /dev/null +++ b/src/connections/node_redis_methods.ts @@ -0,0 +1,132 @@ +/* + * @adonisjs/redis + * + * (c) AdonisJS + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +import type { RedisClientType } from 'redis' + +/** + * The subset of node-redis commands proxied by the "NodeRedisConnection" + * class. Unlike the ioredis "baseMethods" list (which enumerates the + * complete ioredis surface), this list is intentionally partial: it + * exists to prove the design, not to reach feature parity. + * + * Commands are named using node-redis conventions (camelCase), since + * node-redis has no ioredis compatible aliases. + * + * Any command that is not part of this list can still be executed by + * grabbing the underlying client using "await connection.connect()". + * + * @example + * ```ts + * nodeRedisMethods.forEach((method) => { + * ConnectionClass.prototype[method] = async function (...args) { + * const client = await this.connect() + * return client[method](...args) + * } + * }) + * ``` + */ +export const nodeRedisMethods = [ + /** + * Strings + */ + 'append', + 'decr', + 'decrBy', + 'get', + 'getDel', + 'getEx', + 'getSet', + 'incr', + 'incrBy', + 'incrByFloat', + 'mGet', + 'mSet', + 'set', + 'setEx', + 'setNX', + 'strLen', + + /** + * Keys + */ + 'copy', + 'del', + 'exists', + 'expire', + 'expireAt', + 'keys', + 'persist', + 'pTTL', + 'randomKey', + 'rename', + 'scan', + 'ttl', + 'type', + 'unlink', + + /** + * Hashes + */ + 'hDel', + 'hExists', + 'hGet', + 'hGetAll', + 'hIncrBy', + 'hKeys', + 'hLen', + 'hSet', + 'hVals', + + /** + * Lists + */ + 'lLen', + 'lPop', + 'lPush', + 'lRange', + 'lRem', + 'rPop', + 'rPush', + + /** + * Sets + */ + 'sAdd', + 'sCard', + 'sIsMember', + 'sMembers', + 'sRem', + + /** + * Sorted sets + */ + 'zAdd', + 'zCard', + 'zRange', + 'zRem', + 'zScore', + + /** + * Scripting + */ + 'eval', + 'evalSha', + 'scriptLoad', + + /** + * Server + */ + 'dbSize', + 'echo', + 'flushAll', + 'flushDb', + 'info', + 'ping', + 'time', +] satisfies (keyof RedisClientType)[] diff --git a/src/redis_manager.ts b/src/redis_manager.ts index 530e33b..aaf7c85 100644 --- a/src/redis_manager.ts +++ b/src/redis_manager.ts @@ -15,15 +15,24 @@ import { type ClusterOptions, type RedisOptions, Redis } from 'ioredis' import debug from './debug.ts' import { baseMethods } from './connections/io_methods.ts' import RedisConnection from './connections/redis_connection.ts' +import NodeRedisConnection from './connections/node_redis_connection.ts' import RedisClusterConnection from './connections/redis_cluster_connection.ts' import type { GetConnectionType, IORedisBaseCommands, + NodeRedisConnectionConfig, PubSubChannelHandler, PubSubPatternHandler, RedisConnectionsList, } from './types.ts' +/** + * Detects a connection that opted into the node-redis client + */ +function isNodeRedisConfig(config: unknown): config is NodeRedisConnectionConfig { + return (config as NodeRedisConnectionConfig)?.client === 'node-redis' +} + /** * Redis Manager exposes the API to manage multiple redis connections * based upon user defined config. @@ -47,7 +56,7 @@ import type { * ``` */ class RedisManager extends Emittery<{ - connection: RedisConnection | RedisClusterConnection + connection: RedisConnection | RedisClusterConnection | NodeRedisConnection }> { /** * Lua scripts to apply to all the connections @@ -193,14 +202,28 @@ class RedisManager extends Emitter * Instantiate the connection based upon the config */ debug('creating new connection %s', name) - const connection = - 'clusters' in config - ? new RedisClusterConnection( - name as string, - config.clusters, - this.#mergeScripts(config.clusterOptions || {}) - ) - : new RedisConnection(name as string, this.#mergeScripts(config)) + let connection: RedisConnection | RedisClusterConnection | NodeRedisConnection + + if (isNodeRedisConfig(config)) { + connection = new NodeRedisConnection(name as string, config) + + /** + * node-redis has its own "scripts" option and it can only be defined + * when creating the client, hence we register the manager level + * scripts on the connection instead. + */ + Object.entries(this.#scripts).forEach(([script, definition]) => { + ;(connection as NodeRedisConnection).defineCommand(script, definition) + }) + } else if ('clusters' in config) { + connection = new RedisClusterConnection( + name as string, + config.clusters, + this.#mergeScripts(config.clusterOptions || {}) + ) + } else { + connection = new RedisConnection(name as string, this.#mergeScripts(config)) + } /** * Notify about a new connection diff --git a/src/types.ts b/src/types.ts index ae91e7f..793e875 100644 --- a/src/types.ts +++ b/src/types.ts @@ -10,10 +10,13 @@ import { type Secret } from '@adonisjs/core/helpers' import { type AsyncOrSync } from '@poppinss/utils/types' import type { Redis, Cluster, RedisOptions, ClusterOptions, Command } from 'ioredis' +import type { RedisClientOptions, RedisClientType } from 'redis' import type RedisManager from './redis_manager.ts' import type RedisConnection from './connections/redis_connection.ts' import type { baseMethods, redisMethods } from './connections/io_methods.ts' +import type NodeRedisConnection from './connections/node_redis_connection.ts' +import type { nodeRedisMethods } from './connections/node_redis_methods.ts' import type RedisClusterConnection from './connections/redis_cluster_connection.ts' export type RedisCommandData = { @@ -198,6 +201,56 @@ export type RedisClusterConnectionConfig = { clusterOptions?: ClusterOptions } +/** + * The subset of node-redis commands exposed by a "NodeRedisConnection". + * The signatures are the node-redis ones, so commands are camelCased and + * take options objects instead of positional flags. + * + * @example + * ```ts + * // ioredis + * await connection.set('key', 'value', 'PX', 1000) + * + * // node-redis + * await connection.set('key', 'value', { PX: 1000 }) + * ``` + */ +export type NodeRedisCommands = { + [K in (typeof nodeRedisMethods)[number]]: RedisClientType[K] +} + +/** + * Status of a node-redis connection. We re-use the ioredis vocabulary, + * so that code reading "connection.status" behaves the same with both + * clients. + */ +export type NodeRedisConnectionStatus = + 'wait' | 'connecting' | 'connect' | 'ready' | 'reconnecting' | 'close' | 'end' + +/** + * Configuration for a connection backed by the official "redis" + * (node-redis) client. The "client" property is the opt-in: any + * connection without it keeps using ioredis. + * + * @example + * ```ts + * const config: NodeRedisConnectionConfig = { + * client: 'node-redis', + * socket: { + * host: 'localhost', + * port: 6379, + * }, + * password: Secret.create('my-password'), + * } + * ``` + */ +export type NodeRedisConnectionConfig = Omit & { + /** Opt-in to the node-redis backed connection */ + client: 'node-redis' + /** Redis password (can be string or Secret) */ + password?: string | Secret +} + /** * Union type representing either a cluster or standalone Redis connection * @@ -242,7 +295,7 @@ export type Connection = RedisClusterConnection | RedisConnection */ export type RedisConnectionsList = Record< string, - RedisConnectionConfig | RedisClusterConnectionConfig + RedisConnectionConfig | RedisClusterConnectionConfig | NodeRedisConnectionConfig > /** @@ -266,9 +319,11 @@ export type RedisConnectionsList = Record< export type GetConnectionType< ConnectionsList extends RedisConnectionsList, T extends keyof ConnectionsList, -> = ConnectionsList[T] extends RedisClusterConnectionConfig - ? RedisClusterConnection - : RedisConnection +> = ConnectionsList[T] extends NodeRedisConnectionConfig + ? NodeRedisConnection + : ConnectionsList[T] extends RedisClusterConnectionConfig + ? RedisClusterConnection + : RedisConnection /** * Interface for defining available Redis connections. diff --git a/tests/node_redis_connection.spec.ts b/tests/node_redis_connection.spec.ts new file mode 100644 index 0000000..e802514 --- /dev/null +++ b/tests/node_redis_connection.spec.ts @@ -0,0 +1,332 @@ +/* + * @adonisjs/redis + * + * (c) AdonisJS + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +import { test } from '@japa/runner' + +import { pEvent } from '../tests_helpers/main.ts' +import { RedisManagerFactory } from '../factories/redis_manager.ts' +import type RedisConnection from '../src/connections/redis_connection.ts' +import NodeRedisConnection from '../src/connections/node_redis_connection.ts' + +const socket = { + host: process.env.REDIS_HOST, + port: Number(process.env.REDIS_PORT), +} + +test.group('Node redis connection | lifecycle', () => { + test('do not dial the server until the first command', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + assert.equal(connection.status, 'wait') + assert.isUndefined(connection.nodeConnection) + assert.isFalse(connection.isReady()) + + await connection.set('greeting', 'hello world') + + assert.equal(connection.status, 'ready') + assert.isTrue(connection.isReady()) + assert.isDefined(connection.nodeConnection) + }) + + test('emit connect and ready events when connecting', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + const [connect, ready] = await Promise.all([ + pEvent(connection, 'connect'), + pEvent(connection, 'ready'), + connection.connect(), + ]) + + assert.strictEqual(connect?.connection, connection) + assert.strictEqual(ready?.connection, connection) + assert.equal(connection.status, 'ready') + }) + + test('memoize the connect promise and share one client', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + const [client, ...clients] = await Promise.all([ + connection.connect(), + connection.connect(), + connection.connect(), + ]) + + clients.forEach((otherClient) => assert.strictEqual(otherClient, client)) + assert.strictEqual(connection.nodeConnection, client) + }) + + test('run concurrent commands issued before the connection is ready', async ({ + assert, + cleanup, + }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(async () => { + await connection.del('counter') + await connection.quit() + }) + + const results = await Promise.all([ + connection.incr('counter'), + connection.incr('counter'), + connection.incr('counter'), + ]) + + assert.deepEqual(results.sort(), [1, 2, 3]) + }) + + test('emit error and reject when unable to connect', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { + client: 'node-redis', + socket: { host: '127.0.0.1', port: 4444, reconnectStrategy: false }, + }) + cleanup(() => connection.disconnect()) + + const [event, error] = await Promise.all([ + pEvent(connection, 'error'), + connection.connect().catch((e) => e), + ]) + + assert.equal(error.message, 'connect ECONNREFUSED 127.0.0.1:4444') + assert.equal(event?.error.message, 'connect ECONNREFUSED 127.0.0.1:4444') + assert.equal(connection.lastError.message, 'connect ECONNREFUSED 127.0.0.1:4444') + assert.isTrue(connection.isClosed()) + }) + + test('emit end and cleanup listeners on quit', async ({ assert }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + await connection.connect() + + const [event] = await Promise.all([pEvent(connection, 'end'), connection.quit()]) + + assert.isNotNull(event) + assert.equal(connection.status, 'end') + assert.isTrue(connection.isClosed()) + assert.isUndefined(connection.nodeConnection) + }) + + test('do not dial the server when quitting a lazy connection', async ({ assert }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + + const [event] = await Promise.all([pEvent(connection, 'end'), connection.quit()]) + + assert.isNotNull(event) + assert.equal(connection.status, 'end') + assert.isUndefined(connection.nodeConnection) + }) + + test('emit end on disconnect', async ({ assert }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + await connection.connect() + + const [event] = await Promise.all([pEvent(connection, 'end'), connection.disconnect()]) + + assert.isNotNull(event) + assert.equal(connection.status, 'end') + }) + + test('throw when using the connection after it has been closed', async ({ assert }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + await connection.connect() + await connection.quit() + + await assert.rejects( + () => connection.get('greeting'), + 'Cannot use the "main" connection after it has been closed' + ) + }) +}) + +test.group('Node redis connection | commands', (group) => { + group.each.setup(() => { + const connection = new NodeRedisConnection('cleanup', { client: 'node-redis', socket }) + return async () => { + await connection.del(['greeting', 'user', 'items', 'counter', 'ttl-key']) + await connection.quit() + } + }) + + test('run string commands', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + assert.equal(await connection.set('greeting', 'hello world'), 'OK') + assert.equal(await connection.get('greeting'), 'hello world') + assert.equal(await connection.exists('greeting'), 1) + assert.equal(await connection.del('greeting'), 1) + assert.isNull(await connection.get('greeting')) + }) + + test('use an options object for command modifiers', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + /** + * ioredis: connection.set('ttl-key', 'value', 'EX', 60) + */ + await connection.set('ttl-key', 'value', { EX: 60 }) + + const ttl = await connection.ttl('ttl-key') + assert.isAbove(ttl, 0) + assert.isAtMost(ttl, 60) + }) + + test('run hash commands', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + await connection.hSet('user', { username: 'virk', age: '32' }) + + assert.equal(await connection.hGet('user', 'username'), 'virk') + assert.deepEqual(await connection.hGetAll('user'), { username: 'virk', age: '32' }) + const keys = await connection.hKeys('user') + assert.deepEqual(keys.sort(), ['age', 'username']) + }) + + test('run list commands', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + await connection.rPush('items', ['one', 'two']) + + assert.equal(await connection.lLen('items'), 2) + assert.deepEqual(await connection.lRange('items', 0, -1), ['one', 'two']) + }) + + test('publish a message', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + assert.equal(await connection.publish('news', 'hello'), 0) + }) + + test('publish a message using a callback', async ({ assert, cleanup }, done) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + connection.publish('news', 'hello', (error, count) => { + assert.isNull(error) + assert.equal(count, 0) + done() + }) + }).waitForDone() + + test('define and run a lua script', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + connection.defineCommand('setAndGet', { + numberOfKeys: 1, + lua: `redis.call('set', KEYS[1], ARGV[1]) return redis.call('get', KEYS[1])`, + }) + + assert.equal(await connection.runCommand('setAndGet', 'greeting', 'hello world'), 'hello world') + assert.equal(await connection.get('greeting'), 'hello world') + }) + + test('throw when running an undefined command', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + await assert.rejects(() => connection.runCommand('foo'), 'Cannot run undefined command "foo"') + }) + + test('throw when subscribing', async ({ assert, cleanup }) => { + const connection = new NodeRedisConnection('main', { client: 'node-redis', socket }) + cleanup(() => connection.quit()) + + assert.throws( + () => connection.subscribe(), + 'Pub/Sub is not implemented for node-redis connections. Use "await connection.connect()" and subscribe using a duplicated client' + ) + }) +}) + +test.group('Node redis connection | manager', () => { + test('create a node-redis connection from the manager config', async ({ assert }) => { + const redis = new RedisManagerFactory({ + connection: 'primary', + connections: { + primary: { host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }, + modern: { client: 'node-redis' as const, socket }, + }, + }).create() + + const connection = redis.connection('modern') + assert.instanceOf(connection, NodeRedisConnection) + + await connection.set('greeting', 'hello world') + assert.equal(await connection.get('greeting'), 'hello world') + await connection.del('greeting') + + assert.equal(redis.activeConnectionsCount, 1) + await redis.quitAll() + }) + + test('infer the connection type from the config', async ({ expectTypeOf }) => { + const redis = new RedisManagerFactory({ + connection: 'primary', + connections: { + primary: { host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }, + modern: { client: 'node-redis' as const, socket }, + }, + }).create() + + expectTypeOf(redis.connection('primary')).toEqualTypeOf() + expectTypeOf(redis.connection('modern')).toEqualTypeOf() + }) + + test('re-use the same connection instance', async ({ assert }) => { + const redis = new RedisManagerFactory({ + connection: 'modern', + connections: { + modern: { client: 'node-redis' as const, socket }, + }, + }).create() + + assert.strictEqual(redis.connection('modern'), redis.connection()) + await redis.quitAll() + }) + + test('forget the connection once it has been closed', async ({ assert }) => { + const redis = new RedisManagerFactory({ + connection: 'modern', + connections: { + modern: { client: 'node-redis' as const, socket }, + }, + }).create() + + const connection = redis.connection() + await connection.connect() + + await Promise.all([pEvent(connection, 'end'), redis.quitAll()]) + assert.equal(redis.activeConnectionsCount, 0) + }) + + test('share manager level lua scripts with node-redis connections', async ({ assert }) => { + const redis = new RedisManagerFactory({ + connection: 'modern', + connections: { + modern: { client: 'node-redis' as const, socket }, + }, + }).create() + + redis.defineCommand('setAndGet', { + numberOfKeys: 1, + lua: `redis.call('set', KEYS[1], ARGV[1]) return redis.call('get', KEYS[1])`, + }) + + assert.equal(await redis.runCommand('setAndGet', 'greeting', 'hi'), 'hi') + + await redis.connection().del('greeting') + await redis.quitAll() + }) +})