refactor: create shared stream deletion types

  for Redis 8.2 commands

Author: PavelPashov

packages/client/lib/commands/XACKDEL.spec.ts

@@ -1,7 +1,11 @@
 import { strict as assert } from "node:assert";
-import XACKDEL, { XACKDEL_REPLY_CODES, XAckDelPolicy } from "./XACKDEL";
+import XACKDEL from "./XACKDEL";
 import { parseArgs } from "./generic-transformers";
 import testUtils, { GLOBAL } from "../test-utils";
+import {
+  STREAM_DELETION_POLICY,
+  STREAM_DELETION_REPLY_CODES,
+} from "./common-stream.types";
 
 describe("XACKDEL", () => {
   describe("transformArguments", () => {
@@ -18,7 +22,13 @@ describe("XACKDEL", () => {
 
     it("string - with policy", () => {
       assert.deepEqual(
-        parseArgs(XACKDEL, "key", "group", "0-0", XAckDelPolicy.KEEPREF),
+        parseArgs(
+          XACKDEL,
+          "key",
+          "group",
+          "0-0",
+          STREAM_DELETION_POLICY.KEEPREF
+        ),
         ["XACKDEL", "key", "group", "KEEPREF", "IDS", "1", "0-0"]
       );
     });
@@ -42,7 +52,7 @@ describe("XACKDEL", () => {
           "key",
           "group",
           ["0-0", "1-0"],
-          XAckDelPolicy.DELREF
+          STREAM_DELETION_POLICY.DELREF
         ),
         ["XACKDEL", "key", "group", "DELREF", "IDS", "2", "0-0", "1-0"]
       );
@@ -53,7 +63,7 @@ describe("XACKDEL", () => {
     `XACKDEL non-existing key - without policy`,
     async (client) => {
       const reply = await client.xAckDel("{tag}stream-key", "testgroup", "0-0");
-      assert.deepEqual(reply, [XACKDEL_REPLY_CODES.NOT_FOUND]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.NOT_FOUND]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -78,7 +88,7 @@ describe("XACKDEL", () => {
       });
 
       const reply = await client.xAckDel(streamKey, groupName, messageId);
-      assert.deepEqual(reply, [XACKDEL_REPLY_CODES.ACKNOWLEDGED_AND_DELETED]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.DELETED]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -106,9 +116,9 @@ describe("XACKDEL", () => {
         streamKey,
         groupName,
         messageId,
-        XAckDelPolicy.DELREF
+        STREAM_DELETION_POLICY.DELREF
       );
-      assert.deepEqual(reply, [XACKDEL_REPLY_CODES.ACKNOWLEDGED_AND_DELETED]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.DELETED]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -137,9 +147,9 @@ describe("XACKDEL", () => {
         streamKey,
         groupName,
         messageId,
-        XAckDelPolicy.ACKED
+        STREAM_DELETION_POLICY.ACKED
       );
-      assert.deepEqual(reply, [XACKDEL_REPLY_CODES.ACKNOWLEDGED_DANGLING_REFS]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.DANGLING_REFS]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -170,12 +180,12 @@ describe("XACKDEL", () => {
         streamKey,
         groupName,
         [...messageIds, "0-0"],
-        XAckDelPolicy.DELREF
+        STREAM_DELETION_POLICY.DELREF
       );
       assert.deepEqual(reply, [
-        XACKDEL_REPLY_CODES.ACKNOWLEDGED_AND_DELETED,
-        XACKDEL_REPLY_CODES.ACKNOWLEDGED_AND_DELETED,
-        XACKDEL_REPLY_CODES.NOT_FOUND,
+        STREAM_DELETION_REPLY_CODES.DELETED,
+        STREAM_DELETION_REPLY_CODES.DELETED,
+        STREAM_DELETION_REPLY_CODES.NOT_FOUND,
       ]);
     },
     {

packages/client/lib/commands/XACKDEL.ts

@@ -1,27 +1,11 @@
 import { CommandParser } from "../client/parser";
 import { RedisArgument, ArrayReply, Command } from "../RESP/types";
+import {
+  StreamDeletionReplyCode,
+  StreamDeletionPolicy,
+} from "./common-stream.types";
 import { RedisVariadicArgument } from "./generic-transformers";
 
-/** XACKDEL deletion policies */
-export const XAckDelPolicy = {
-  /** Preserve references (default) */
-  KEEPREF: "KEEPREF",
-  /** Delete all references */
-  DELREF: "DELREF",
-  /** Only acknowledged entries */
-  ACKED: "ACKED",
-} as const;
-
-/** XACKDEL reply codes */
-export const XACKDEL_REPLY_CODES = {
-  /** ID not found */
-  NOT_FOUND: -1,
-  /** Entry acknowledged and deleted */
-  ACKNOWLEDGED_AND_DELETED: 1,
-  /** Entry acknowledged but dangling references remain */
-  ACKNOWLEDGED_DANGLING_REFS: 2,
-} as const;
-
 /**
  * Acknowledges and deletes one or multiple messages for a stream consumer group
  */
@@ -43,7 +27,7 @@ export default {
     key: RedisArgument,
     group: RedisArgument,
     id: RedisVariadicArgument,
-    policy?: (typeof XAckDelPolicy)[keyof typeof XAckDelPolicy],
+    policy?: StreamDeletionPolicy
   ) {
     parser.push("XACKDEL");
     parser.pushKey(key);
@@ -56,7 +40,6 @@ export default {
     parser.push("IDS");
     parser.pushVariadicWithLength(id);
   },
-  transformReply: undefined as unknown as () => ArrayReply<
-    (typeof XACKDEL_REPLY_CODES)[keyof typeof XACKDEL_REPLY_CODES]
-  >,
+  transformReply:
+    undefined as unknown as () => ArrayReply<StreamDeletionReplyCode>,
 } as const satisfies Command;

packages/client/lib/commands/XDELEX.spec.ts

@@ -1,7 +1,11 @@
 import { strict as assert } from "node:assert";
-import XDELEX, { XDELEX_REPLY_CODES, XDelexPolicy } from "./XDELEX";
+import XDELEX from "./XDELEX";
 import { parseArgs } from "./generic-transformers";
 import testUtils, { GLOBAL } from "../test-utils";
+import {
+  STREAM_DELETION_POLICY,
+  STREAM_DELETION_REPLY_CODES,
+} from "./common-stream.types";
 
 describe("XDELEX", () => {
   describe("transformArguments", () => {
@@ -16,14 +20,10 @@ describe("XDELEX", () => {
     });
 
     it("string - with policy", () => {
-      assert.deepEqual(parseArgs(XDELEX, "key", "0-0", XDelexPolicy.KEEPREF), [
-        "XDELEX",
-        "key",
-        "KEEPREF",
-        "IDS",
-        "1",
-        "0-0",
-      ]);
+      assert.deepEqual(
+        parseArgs(XDELEX, "key", "0-0", STREAM_DELETION_POLICY.KEEPREF),
+        ["XDELEX", "key", "KEEPREF", "IDS", "1", "0-0"]
+      );
     });
 
     it("array - without policy", () => {
@@ -39,7 +39,7 @@ describe("XDELEX", () => {
 
     it("array - with policy", () => {
       assert.deepEqual(
-        parseArgs(XDELEX, "key", ["0-0", "1-0"], XDelexPolicy.DELREF),
+        parseArgs(XDELEX, "key", ["0-0", "1-0"], STREAM_DELETION_POLICY.DELREF),
         ["XDELEX", "key", "DELREF", "IDS", "2", "0-0", "1-0"]
       );
     });
@@ -49,7 +49,7 @@ describe("XDELEX", () => {
     `XDELEX non-existing key - without policy`,
     async (client) => {
       const reply = await client.xDelex("{tag}stream-key", "0-0");
-      assert.deepEqual(reply, [XDELEX_REPLY_CODES.NOT_FOUND]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.NOT_FOUND]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -68,9 +68,9 @@ describe("XDELEX", () => {
       const reply = await client.xDelex(
         streamKey,
         messageId,
-        XDelexPolicy.KEEPREF
+        STREAM_DELETION_POLICY.KEEPREF
       );
-      assert.deepEqual(reply, [XDELEX_REPLY_CODES.DELETED]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.DELETED]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -89,9 +89,9 @@ describe("XDELEX", () => {
       const reply = await client.xDelex(
         streamKey,
         messageId,
-        XDelexPolicy.DELREF
+        STREAM_DELETION_POLICY.DELREF
       );
-      assert.deepEqual(reply, [XDELEX_REPLY_CODES.DELETED]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.DELETED]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -115,9 +115,9 @@ describe("XDELEX", () => {
       const reply = await client.xDelex(
         streamKey,
         messageId,
-        XDelexPolicy.ACKED
+        STREAM_DELETION_POLICY.ACKED
       );
-      assert.deepEqual(reply, [XDELEX_REPLY_CODES.DANGLING_REFS]);
+      assert.deepEqual(reply, [STREAM_DELETION_REPLY_CODES.DANGLING_REFS]);
     },
     {
       client: { ...GLOBAL.SERVERS.OPEN, minimumDockerVersion: [8, 2] },
@@ -141,12 +141,12 @@ describe("XDELEX", () => {
       const reply = await client.xDelex(
         streamKey,
         [...messageIds, "0-0"],
-        XDelexPolicy.DELREF
+        STREAM_DELETION_POLICY.DELREF
       );
       assert.deepEqual(reply, [
-        XDELEX_REPLY_CODES.DELETED,
-        XDELEX_REPLY_CODES.DELETED,
-        XDELEX_REPLY_CODES.NOT_FOUND,
+        STREAM_DELETION_REPLY_CODES.DELETED,
+        STREAM_DELETION_REPLY_CODES.DELETED,
+        STREAM_DELETION_REPLY_CODES.NOT_FOUND,
       ]);
     },
     {

packages/client/lib/commands/XDELEX.ts

@@ -1,27 +1,11 @@
 import { CommandParser } from "../client/parser";
 import { RedisArgument, ArrayReply, Command } from "../RESP/types";
+import {
+  StreamDeletionPolicy,
+  StreamDeletionReplyCode,
+} from "./common-stream.types";
 import { RedisVariadicArgument } from "./generic-transformers";
 
-/** XDELEX deletion policies */
-export const XDelexPolicy = {
-  /** Preserve references (default) */
-  KEEPREF: "KEEPREF",
-  /** Delete all references */
-  DELREF: "DELREF",
-  /** Only acknowledged entries */
-  ACKED: "ACKED",
-} as const;
-
-/** XDELEX reply codes */
-export const XDELEX_REPLY_CODES = {
-  /** ID not found */
-  NOT_FOUND: -1,
-  /** Entry deleted */
-  DELETED: 1,
-  /** Dangling references */
-  DANGLING_REFS: 2,
-} as const;
-
 /**
  * Deletes one or multiple entries from the stream
  */
@@ -41,7 +25,7 @@ export default {
     parser: CommandParser,
     key: RedisArgument,
     id: RedisVariadicArgument,
-    policy?: (typeof XDelexPolicy)[keyof typeof XDelexPolicy],
+    policy?: StreamDeletionPolicy
   ) {
     parser.push("XDELEX");
     parser.pushKey(key);
@@ -53,7 +37,6 @@ export default {
     parser.push("IDS");
     parser.pushVariadicWithLength(id);
   },
-  transformReply: undefined as unknown as () => ArrayReply<
-    (typeof XDELEX_REPLY_CODES)[keyof typeof XDELEX_REPLY_CODES]
-  >,
+  transformReply:
+    undefined as unknown as () => ArrayReply<StreamDeletionReplyCode>,
 } as const satisfies Command;

packages/client/lib/commands/common-stream.types.ts

@@ -0,0 +1,28 @@
+/** Common stream deletion policies
+ *
+ *  Added in Redis 8.2
+ */
+export const STREAM_DELETION_POLICY = {
+  /** Preserve references (default) */
+  KEEPREF: "KEEPREF",
+  /** Delete all references */
+  DELREF: "DELREF",
+  /** Only acknowledged entries */
+  ACKED: "ACKED",
+} as const;
+
+export type StreamDeletionPolicy =
+  (typeof STREAM_DELETION_POLICY)[keyof typeof STREAM_DELETION_POLICY];
+
+/** Common reply codes for stream deletion operations */
+export const STREAM_DELETION_REPLY_CODES = {
+  /** ID not found */
+  NOT_FOUND: -1,
+  /** Entry deleted */
+  DELETED: 1,
+  /** Dangling references */
+  DANGLING_REFS: 2,
+} as const;
+
+export type StreamDeletionReplyCode =
+  (typeof STREAM_DELETION_REPLY_CODES)[keyof typeof STREAM_DELETION_REPLY_CODES];