feat(client): add XNACK command with options and tests (#3238)

Author: nkaradzhov

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

@@ -0,0 +1,105 @@
+import { strict as assert } from 'node:assert';
+import XNACK from './XNACK';
+import { parseArgs } from './generic-transformers';
+import testUtils, { GLOBAL } from '../test-utils';
+
+describe('XNACK', () => {
+  describe('transformArguments', () => {
+    it('string - SILENT', () => {
+      assert.deepEqual(
+        parseArgs(XNACK, 'key', 'group', 'SILENT', '0-0'),
+        ['XNACK', 'key', 'group', 'SILENT', 'IDS', '1', '0-0']
+      );
+    });
+
+    it('array - FAIL', () => {
+      assert.deepEqual(
+        parseArgs(XNACK, 'key', 'group', 'FAIL', ['0-0', '1-0']),
+        ['XNACK', 'key', 'group', 'FAIL', 'IDS', '2', '0-0', '1-0']
+      );
+    });
+
+    it('array - FATAL', () => {
+      assert.deepEqual(
+        parseArgs(XNACK, 'key', 'group', 'FATAL', ['0-0', '1-0', '2-0']),
+        ['XNACK', 'key', 'group', 'FATAL', 'IDS', '3', '0-0', '1-0', '2-0']
+      );
+    });
+
+    it('with RETRYCOUNT', () => {
+      assert.deepEqual(
+        parseArgs(XNACK, 'key', 'group', 'FAIL', '0-0', {
+          RETRYCOUNT: 7
+        }),
+        ['XNACK', 'key', 'group', 'FAIL', 'IDS', '1', '0-0', 'RETRYCOUNT', '7']
+      );
+    });
+
+    it('with FORCE', () => {
+      assert.deepEqual(
+        parseArgs(XNACK, 'key', 'group', 'FAIL', ['0-0', '1-0'], {
+          FORCE: true
+        }),
+        ['XNACK', 'key', 'group', 'FAIL', 'IDS', '2', '0-0', '1-0', 'FORCE']
+      );
+    });
+
+    it('with RETRYCOUNT and FORCE', () => {
+      assert.deepEqual(
+        parseArgs(XNACK, 'key', 'group', 'FAIL', ['0-0', '1-0'], {
+          RETRYCOUNT: 3,
+          FORCE: true
+        }),
+        ['XNACK', 'key', 'group', 'FAIL', 'IDS', '2', '0-0', '1-0', 'RETRYCOUNT', '3', 'FORCE']
+      );
+    });
+  });
+
+  testUtils.testWithClient('xNack', async client => {
+
+    const key = `xnack:tmp:${Date.now()}`;
+    const group = 'group';
+    const consumer = 'consumer-1';
+
+    await client.del(key);
+    await client.xGroupCreate(key, group, '0', { MKSTREAM: true });
+
+    const id1 = await client.xAdd(key, '*', { field: '1' });
+    const id2 = await client.xAdd(key, '*', { field: '2' });
+    const id3 = await client.xAdd(key, '*', { field: '3' });
+    const id4 = await client.xAdd(key, '*', { field: '4' });
+    const id5 = await client.xAdd(key, '*', { field: '5' });
+    const id6 = await client.xAdd(key, '*', { field: '6' });
+
+    await client.xReadGroup(group, consumer, {
+      key,
+      id: '>'
+    });
+
+    const reply = await client.xNack(key, group, 'FAIL', [id1, id2]);
+    const replyWithRetryCount = await client.xNack(key, group, 'FAIL', id3, {
+      RETRYCOUNT: 7
+    });
+    const replyWithForce = await client.xNack(key, group, 'FAIL', id4, {
+      FORCE: true
+    });
+    const replyWithRetryCountAndForce = await client.xNack(key, group, 'FAIL', [id5, id6], {
+      RETRYCOUNT: 3,
+      FORCE: true
+    });
+
+    assert.equal(reply, 2);
+    assert.equal(replyWithRetryCount, 1);
+    assert.equal(replyWithForce, 1);
+    assert.equal(replyWithRetryCountAndForce, 2);
+
+    assert.equal(typeof reply, 'number');
+    assert.equal(typeof replyWithRetryCount, 'number');
+    assert.equal(typeof replyWithForce, 'number');
+    assert.equal(typeof replyWithRetryCountAndForce, 'number');
+  }, {
+    ...GLOBAL.SERVERS.OPEN,
+    minimumDockerVersion: [8, 8]
+  });
+
+});

packages/client/lib/commands/XNACK.ts

@@ -0,0 +1,47 @@
+import { CommandParser } from '../client/parser';
+import { Command, NumberReply, RedisArgument } from '../RESP/types';
+import { RedisVariadicArgument } from './generic-transformers';
+
+export type XNackMode = 'SILENT' | 'FAIL' | 'FATAL';
+export interface XNackOptions {
+  RETRYCOUNT?: number;
+  FORCE?: boolean;
+}
+
+export default {
+  IS_READ_ONLY: false,
+  /**
+   * Constructs the XNACK command to negatively acknowledge one or more pending stream entries.
+   * Added since Redis 8.8.
+   *
+   * @param parser - The command parser
+   * @param key - The stream key
+   * @param group - The consumer group name
+   * @param mode - NACK mode: SILENT, FAIL, or FATAL
+   * @param id - One or more message IDs to nack
+   * @param options - Additional options for retry count and force handling
+   * @see https://redis.io/commands/xnack/
+   */
+  parseCommand(
+    parser: CommandParser,
+    key: RedisArgument,
+    group: RedisArgument,
+    mode: XNackMode,
+    id: RedisVariadicArgument,
+    options?: XNackOptions
+  ) {
+    parser.push('XNACK');
+    parser.pushKey(key);
+    parser.push(group, mode, 'IDS');
+    parser.pushVariadicWithLength(id);
+
+    if (options?.RETRYCOUNT !== undefined) {
+      parser.push('RETRYCOUNT', options.RETRYCOUNT.toString());
+    }
+
+    if (options?.FORCE) {
+      parser.push('FORCE');
+    }
+  },
+  transformReply: undefined as unknown as () => NumberReply
+} as const satisfies Command;

packages/client/lib/commands/index.ts

@@ -308,6 +308,7 @@ import XINFO_CONSUMERS from './XINFO_CONSUMERS';
 import XINFO_GROUPS from './XINFO_GROUPS';
 import XINFO_STREAM from './XINFO_STREAM';
 import XLEN from './XLEN';
+import XNACK from './XNACK';
 import XPENDING_RANGE from './XPENDING_RANGE';
 import XPENDING from './XPENDING';
 import XRANGE from './XRANGE';
@@ -1006,6 +1007,8 @@ export default {
   xInfoStream: XINFO_STREAM,
   XLEN,
   xLen: XLEN,
+  XNACK,
+  xNack: XNACK,
   XPENDING_RANGE,
   xPendingRange: XPENDING_RANGE,
   XPENDING,