Node: Add XRANGE command

CHANGELOG.md

@@ -42,6 +42,7 @@
 * Node: Added ZMPOP command ([#1994](https://github.com/valkey-io/valkey-glide/pull/1994))
 * Node: Added ZINCRBY command ([#2009](https://github.com/valkey-io/valkey-glide/pull/2009))
 * Node: Added BZMPOP command ([#2018](https://github.com/valkey-io/valkey-glide/pull/2018))
+* Node: Added XRANGE command ([#2069](https://github.com/valkey-io/valkey-glide/pull/2069))
 * Node: Added PFMERGE command ([#2053](https://github.com/valkey-io/valkey-glide/pull/2053))
 * Node: Added ZLEXCOUNT command ([#2022](https://github.com/valkey-io/valkey-glide/pull/2022))
 * Node: Added ZREMRANGEBYLEX command ([#2025]((https://github.com/valkey-io/valkey-glide/pull/2025))

node/src/BaseClient.ts

@@ -43,6 +43,7 @@ import {
     SearchOrigin,
     SetOptions,
     StreamAddOptions,
+    StreamRangeBound,
     StreamReadOptions,
     StreamTrimOptions,
     ZAddOptions,
@@ -141,6 +142,7 @@ import {
     createUnlink,
     createXAdd,
     createXLen,
+    createXRange,
     createXRead,
     createXTrim,
     createZAdd,
@@ -2560,6 +2562,44 @@ export class BaseClient {
         return this.createWritePromise(scriptInvocation);
     }
 
+    /**
+     * Returns stream entries matching a given range of IDs.
+     *
+     * See https://valkey.io/commands/xrange for more details.
+     *
+     * @param key - The key of the stream.
+     * @param start - The starting stream ID bound for the range.
+     *     - Use `exclusive: "("` to specify an exclusive bounded stream ID.
+     *     - Use `-` to start with the minimum available ID.
+     * @param end - The ending stream ID bound for the range.
+     *     - Use `exclusive: "("` to specify an exclusive bounded stream ID.
+     *     - Use `+` to end with the maximum available ID.
+     * @param count - An optional argument specifying the maximum count of stream entries to return.
+     *     If `count` is not provided, all stream entries in the range will be returned.
+     * @returns A mapping of stream IDs to stream entry data, where entry data is a
+     *     list of pairings with format `[[field, entry], [field, entry], ...]`.
+     *
+     * @example
+     * ```typescript
+     * await client.xadd("mystream", [["field1", "value1"]], {id: "0-1"});
+     * await client.xadd("mystream", [["field2", "value2"], ["field2", "value3"]], {id: "0-2"});
+     * const result = await client.xrange("mystream", "-", "+");
+     * // Output:
+     * // {
+     * //     "0-1": [["field1", "value1"]],
+     * //     "0-2": [["field2", "value2"], ["field2", "value3"]],
+     * // } // Indicates the stream IDs and their associated field-value pairs for all stream entries in "mystream".
+     * ```
+     */
+    public async xrange(
+        key: string,
+        start: StreamRangeBound,
+        end: StreamRangeBound,
+        count?: number,
+    ): Promise<Record<string, string[][]> | null> {
+        return this.createWritePromise(createXRange(key, start, end, count));
+    }
+
     /** Adds members with their scores to the sorted set stored at `key`.
      * If a member is already a part of the sorted set, its score is updated.
      * See https://valkey.io/commands/zadd/ for more details.

node/src/Commands.ts

@@ -1974,6 +1974,61 @@ export function createXTrim(
     return createCommand(RequestType.XTrim, args);
 }
 
+export type StreamRangeBound =
+    /**
+     * Stream ID boundary used to specify the minimum stream entry ID. Can be used in the `XRANGE` or `XREVRANGE` commands
+     * to get the first stream ID.
+     */
+    | "-"
+    /**
+     * Stream ID boundary used to specify the maximum stream entry ID. Can be used in the `XRANGE` or `XREVRANGE` commands
+     * to get the last stream ID.
+     */
+    | "+"
+    /**
+     * Stream ID boundary used to specify a range of IDs to search. Stream ID bounds can be complete with
+     * a timestamp and sequence number separated by a dash ("-"), for example "1526985054069-0". Stream ID bounds can also
+     * be incomplete, with just a timestamp. Can be specified as inclusive or exclusive, where inclusive is the default.
+     */
+    | {
+          exclusive?: "(";
+          id: string | number;
+      };
+
+function addRangeBound(rangeBound: StreamRangeBound, args: string[]) {
+    if (rangeBound === "-" || rangeBound === "+") {
+        args.push(rangeBound);
+        return;
+    }
+
+    if (rangeBound.exclusive) {
+        args.push(rangeBound.exclusive + rangeBound.id.toString());
+    } else {
+        args.push(rangeBound.id.toString());
+    }
+}
+
+/**
+ * @internal
+ */
+export function createXRange(
+    key: string,
+    start: StreamRangeBound,
+    end: StreamRangeBound,
+    count?: number,
+): command_request.Command {
+    const args = [key];
+    addRangeBound(start, args);
+    addRangeBound(end, args);
+
+    if (count !== undefined) {
+        args.push("COUNT");
+        args.push(count.toString());
+    }
+
+    return createCommand(RequestType.XRange, args);
+}
+
 /**
  * @internal
  */

node/src/Transaction.ts

@@ -45,6 +45,7 @@ import {
     SearchOrigin,
     SetOptions,
     StreamAddOptions,
+    StreamRangeBound,
     StreamReadOptions,
     StreamTrimOptions,
     ZAddOptions,
@@ -165,6 +166,7 @@ import {
     createUnlink,
     createXAdd,
     createXLen,
+    createXRange,
     createXRead,
     createXTrim,
     createZAdd,
@@ -2013,6 +2015,33 @@ export class BaseTransaction<T extends BaseTransaction<T>> {
         return this.addAndReturn(createTime());
     }
 
+    /**
+     * Returns stream entries matching a given range of IDs.
+     *
+     * See https://valkey.io/commands/xrange for more details.
+     *
+     * @param key - The key of the stream.
+     * @param start - The starting stream ID bound for the range.
+     *     - Use `exclusive: "("` to specify an exclusive bounded stream ID.
+     *     - Use `-` to start with the minimum available ID.
+     * @param end - The ending stream ID bound for the range.
+     *     - Use `exclusive: "("` to specify an exclusive bounded stream ID.
+     *     - Use `+` to end with the maximum available ID.
+     * @param count - An optional argument specifying the maximum count of stream entries to return.
+     *     If `count` is not provided, all stream entries in the range will be returned.
+     *
+     * Command Response - A mapping of stream IDs to stream entry data, where entry data is a
+     *     list of pairings with format `[[field, entry], [field, entry], ...]`.
+     */
+    public xrange(
+        key: string,
+        start: StreamRangeBound,
+        end: StreamRangeBound,
+        count?: number,
+    ): T {
+        return this.addAndReturn(createXRange(key, start, end, count));
+    }
+
     /**
      * Reads entries from the given streams.
      * See https://valkey.io/commands/xread/ for more details.

node/tests/SharedTests.ts

@@ -4263,6 +4263,88 @@ export function runBaseTests<Context>(config: {
         config.timeout,
     );
 
+    it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
+        `xrange test_%p`,
+        async (protocol) => {
+            await runTest(async (client: BaseClient) => {
+                const key = uuidv4();
+                const nonExistingKey = uuidv4();
+                const stringKey = uuidv4();
+                const streamId1 = "0-1";
+                const streamId2 = "0-2";
+                const streamId3 = "0-3";
+
+                expect(
+                    await client.xadd(key, [["f1", "v1"]], { id: streamId1 }),
+                ).toEqual(streamId1);
+                expect(
+                    await client.xadd(key, [["f2", "v2"]], { id: streamId2 }),
+                ).toEqual(streamId2);
+                expect(await client.xlen(key)).toEqual(2);
+
+                // get everything from the stream
+                expect(await client.xrange(key, "-", "+")).toEqual({
+                    [streamId1]: [["f1", "v1"]],
+                    [streamId2]: [["f2", "v2"]],
+                });
+
+                // returns empty mapping if + before -
+                expect(await client.xrange(key, "+", "-")).toEqual({});
+
+                expect(
+                    await client.xadd(key, [["f3", "v3"]], { id: streamId3 }),
+                ).toEqual(streamId3);
+
+                // get the newest entry
+                expect(
+                    await client.xrange(
+                        key,
+                        { exclusive: "(", id: streamId2 },
+                        { id: 5 },
+                        1,
+                    ),
+                ).toEqual({ [streamId3]: [["f3", "v3"]] });
+
+                // xrange against an emptied stream
+                expect(
+                    await client.customCommand([
+                        "XDEL",
+                        key,
+                        streamId1,
+                        streamId2,
+                        streamId3,
+                    ]),
+                ).toEqual(3);
+                expect(await client.xrange(key, "-", "+", 10)).toEqual({});
+
+                expect(await client.xrange(nonExistingKey, "-", "+")).toEqual(
+                    {},
+                );
+
+                // count value < 1 returns null
+                expect(await client.xrange(key, "-", "+", 0)).toEqual(null);
+                expect(await client.xrange(key, "-", "+", -1)).toEqual(null);
+
+                // key exists, but it is not a stream
+                expect(await client.set(stringKey, "foo"));
+                await expect(
+                    client.xrange(stringKey, "-", "+"),
+                ).rejects.toThrow(RequestError);
+
+                // invalid start bound
+                await expect(
+                    client.xrange(key, { id: "not_a_stream_id" }, "+"),
+                ).rejects.toThrow(RequestError);
+
+                // invalid end bound
+                await expect(
+                    client.xrange(key, "-", { id: "not_a_stream_id" }),
+                ).rejects.toThrow(RequestError);
+            }, protocol);
+        },
+        config.timeout,
+    );
+
     it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
         `zremRangeByLex test_%p`,
         async (protocol) => {

node/tests/TestUtilities.ts

@@ -809,6 +809,8 @@ export async function transactionTest(
     ]);
     baseTransaction.xlen(key9);
     responseData.push(["xlen(key9)", 3]);
+    baseTransaction.xrange(key9, { id: "0-1" }, { id: "0-1" });
+    responseData.push(["xrange(key9)", { "0-1": [["field", "value1"]] }]);
     baseTransaction.xread({ [key9]: "0-1" });
     responseData.push([
         'xread({ [key9]: "0-1" })',