Initial commit

Author: sarfarazahmedklpw

.gitignore

@@ -0,0 +1,5 @@
+/.idea
+/.vscode
+.env
+/coverage/
+/node_modules/

README.md

@@ -0,0 +1,124 @@
+# Redis Cluster Pipeline Library
+
+## Overview
+
+This is a custom wrapper library around the popular ioredis client that adds a new method called `clusterPipeline` to handle Redis Cluster pipelines more efficiently. This library was developed by the PhysicsWallah Private Limited tech team to solve issues with executing multiple Redis commands across Redis Cluster nodes.
+
+## Problem Statement
+
+While using Redis Cluster with the ioredis client, handling multiple commands in a way that correctly distributes them across nodes while maintaining the order of execution was challenging. To address this, we developed this library that adds the `clusterPipeline` method, making it easy to manage Redis commands in a Redis Cluster environment.
+
+This library is simply a wrapper around ioredis, extending it with an additional method that enables Redis Cluster commands to be executed in parallel, with the results returned in the correct order.
+
+## Features
+
+- **Efficient Redis Cluster Pipeline**: Adds support for executing commands across multiple Redis Cluster nodes using `clusterPipeline`.
+- **Seamless Integration**: Works with the existing ioredis client without requiring any changes to your existing Redis configuration.
+- **Cluster-Aware**: Automatically handles Redis commands based on slots, distributing them across the appropriate Redis Cluster node.
+- **Backward Compatible**: All existing ioredis functionality is preserved; the `clusterPipeline` method is simply added to the Cluster class.
+
+## Installation
+
+To install the library, use npm or yarn to add it to your project.
+
+```bash
+npm install @pw-tech/ioredis
+```
+
+Or with yarn:
+
+```bash
+yarn add @pw-tech/ioredis
+```
+
+## Usage
+
+After installing, you can use the Cluster class just as you would with ioredis, but with the added functionality of the `clusterPipeline` method for Redis Cluster commands.
+
+### Example
+
+1. Basic Redis Usage (Without Clusters)
+
+```javascript
+import { Redis } from '@pw-tech/ioredis';
+
+const redis = new Redis({
+  host: 'localhost',
+  port: 6379,
+});
+
+redis.set('key1', 'value1').then(() => redis.get('key1')).then(console.log);
+```
+
+2. Redis Cluster Usage (With `clusterPipeline`)
+
+In case you are using a Redis Cluster, you can now use the `clusterPipeline` method to handle pipelines across multiple nodes.
+
+```javascript
+import { Cluster } from '@pw-tech/ioredis';
+
+const clusterRedis = new Cluster({
+  redisOptions: [{ host: 'localhost', port: 6379 }],
+});
+
+const commands = [
+  ['set', 'key1', 'value1'],
+  ['get', 'key1'],
+];
+
+clusterRedis.clusterPipeline(commands).then(results => {
+  console.log(results); // Output: [ ['OK'], ['value1'] ]
+});
+```
+
+## Explanation of `clusterPipeline`
+
+`clusterPipeline` takes an array of Redis commands (e.g., [['set', 'key', 'value'], ['get', 'key']]).
+It automatically divides the commands across the Redis Cluster nodes based on the slot of the key.
+It executes the commands in parallel and ensures the results are returned in the same order as the original commands.
+
+## How it Works
+
+- **Slot Calculation**: Redis Cluster splits keys across multiple nodes using slots. This library automatically calculates which node each key belongs to and groups the commands accordingly.
+- **Pipeline Execution**: Once the commands are grouped by node, they are executed in parallel on each node, ensuring better performance when dealing with large pipelines.
+- **Result Handling**: Results from each node are merged and returned in the same order as the original commands.
+
+## API Documentation
+
+`clusterPipeline(commands: [string, ...any][]): Promise<any[]>`
+
+### Parameters:
+
+- `commands`: An array of Redis commands, where each command is an array starting with the command name and followed by its arguments.
+  - Example: [['set', 'key1', 'value1'], ['get', 'key1']].
+
+### Returns:
+
+- A Promise that resolves to an array of command results in the same order as they were provided.
+  - Example: [ ['OK'], ['value1'] ].
+
+## Redis and Cluster
+
+All the normal ioredis Redis and Cluster functionality remains available. This library only adds the `clusterPipeline` method to Cluster.
+
+## Why This Library?
+
+### Problem at PhysicsWallah Private Limited
+
+At PhysicsWallah Private Limited, we needed an efficient way to handle Redis Cluster commands when multiple Redis nodes were involved. The standard Redis pipeline approach didn’t take slot distribution into account, leading to inefficiencies and complexity. We developed this library to:
+
+- Simplify the execution of multiple Redis commands across a Redis Cluster.
+- Maintain the order of results while executing commands in parallel.
+- Make our Redis Cluster usage more efficient and less error-prone.
+
+### How We Solve It
+
+By adding the `clusterPipeline` method, we ensure commands are distributed to the appropriate Redis node, executed in parallel, and results are returned in the correct order. This approach eliminates the need to manually manage slot calculations and node selection.
+
+## License
+
+This library is open-source and licensed under the MIT License.
+
+## Contributing
+
+If you have any improvements or bug fixes, feel free to submit a pull request or open an issue.

dist/clusterPipeline.d.ts

@@ -0,0 +1,8 @@
+import Redis, { Cluster } from 'ioredis';
+declare class ExtendedClusterRedis extends Cluster {
+    clusterPipeline(commands: [string, ...any][]): Promise<any[]>;
+    private calculateSlot;
+    private hashCRC16;
+    private findNodeForSlot;
+}
+export { Redis, ExtendedClusterRedis as Cluster };

dist/clusterPipeline.js

@@ -0,0 +1,86 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });
+exports.Cluster = exports.Redis = void 0;
+const ioredis_1 = require("ioredis");
+exports.Redis = ioredis_1.default;
+class ExtendedClusterRedis extends ioredis_1.Cluster {
+    async clusterPipeline(commands) {
+        const pipelinesByNode = {};
+        const clusterSlots = await this.cluster('SHARDS');
+        const nodeSlotRanges = clusterSlots.flatMap(([, slotRanges, , [node]]) => {
+            // slotRanges can contain multiple start-end pairs
+            const ranges = [];
+            for (let i = 0; i < slotRanges.length; i += 2) {
+                ranges.push({
+                    node: node[1],
+                    startSlot: slotRanges[i],
+                    endSlot: slotRanges[i + 1],
+                });
+            }
+            return ranges;
+        });
+        // Group commands by node
+        for (let i = 0; i < commands.length; i++) {
+            const command = commands[i];
+            const key = command[1];
+            const slot = this.calculateSlot(key);
+            const node = this.findNodeForSlot(nodeSlotRanges, slot);
+            if (!pipelinesByNode[node]) {
+                pipelinesByNode[node] = { commands: [], originalIndices: [] };
+            }
+            pipelinesByNode[node].commands.push(command);
+            pipelinesByNode[node].originalIndices.push(i);
+        }
+        // Execute pipelines per node
+        const results = [];
+        for (const node in pipelinesByNode) {
+            const { commands, originalIndices } = pipelinesByNode[node];
+            const pipeline = this.pipeline();
+            commands.forEach(cmd => pipeline[cmd[0]](...cmd.slice(1)));
+            const nodeResult = await pipeline.exec();
+            nodeResult.forEach((result, localIndex) => {
+                const originalIndex = originalIndices[localIndex];
+                results[originalIndex] = result;
+            });
+        }
+        return results;
+    }
+    calculateSlot(key) {
+        if (key == null) {
+            return 0;
+        }
+        key = String(key);
+        // Handle hash tag (keys inside {})
+        const hashTagMatch = key.match(/\{(.+?)\}/);
+        if (hashTagMatch) {
+            key = hashTagMatch[1];
+        }
+        if (key.trim().length === 0) {
+            return 0;
+        }
+        return this.hashCRC16(key) % 16384;
+    }
+    hashCRC16(str) {
+        let crc = 0;
+        const polynomial = 0x1021;
+        const buffer = Buffer.from(str, 'utf8');
+        for (let byte of buffer) {
+            crc ^= byte << 8;
+            for (let i = 0; i < 8; i++) {
+                if (crc & 0x8000) {
+                    crc = (crc << 1) ^ polynomial;
+                }
+                else {
+                    crc <<= 1;
+                }
+                crc &= 0xFFFF;
+            }
+        }
+        return crc;
+    }
+    findNodeForSlot(nodeSlotRanges, slot) {
+        const matchingRange = nodeSlotRanges.find(range => slot >= range.startSlot && slot <= range.endSlot);
+        return matchingRange ? matchingRange.node : null;
+    }
+}
+exports.Cluster = ExtendedClusterRedis;

dist/index.d.ts

@@ -0,0 +1 @@
+export * from './clusterPipeline';

dist/index.js

@@ -0,0 +1,17 @@
+"use strict";
+var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    var desc = Object.getOwnPropertyDescriptor(m, k);
+    if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
+      desc = { enumerable: true, get: function() { return m[k]; } };
+    }
+    Object.defineProperty(o, k2, desc);
+}) : (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    o[k2] = m[k];
+}));
+var __exportStar = (this && this.__exportStar) || function(m, exports) {
+    for (var p in m) if (p !== "default" && !Object.prototype.hasOwnProperty.call(exports, p)) __createBinding(exports, m, p);
+};
+Object.defineProperty(exports, "__esModule", { value: true });
+__exportStar(require("./clusterPipeline"), exports);