#RI-2947 BE Claim Pending entries

Author: Artem

redisinsight/api/src/modules/browser/constants/browser-tool-commands.ts

@@ -84,6 +84,7 @@ export enum BrowserToolStreamCommands {
   XInfoConsumers = 'xinfo consumers',
   XPending = 'xpending',
   XAck = 'xack',
+  XClaim = 'xclaim',
   XGroupCreate = 'xgroup create',
   XGroupSetId = 'xgroup setid',
   XGroupDestroy = 'xgroup destroy',

redisinsight/api/src/modules/browser/controllers/stream/consumer.controller.ts

@@ -9,7 +9,7 @@ import {
 import { ApiTags } from '@nestjs/swagger';
 import { ApiRedisInstanceOperation } from 'src/decorators/api-redis-instance-operation.decorator';
 import {
-  AckPendingEntriesDto, AckPendingEntriesResponse,
+  AckPendingEntriesDto, AckPendingEntriesResponse, ClaimPendingEntriesResponse, ClaimPendingEntryDto,
   ConsumerDto,
   ConsumerGroupDto,
   GetConsumersDto, GetPendingEntriesDto, PendingEntryDto,
@@ -24,7 +24,7 @@ export class ConsumerController {
 
   @Post('/get')
   @ApiRedisInstanceOperation({
-    description: 'Get group consumers',
+    description: 'Get consumers list in the group',
     statusCode: 200,
     responses: [
       {
@@ -43,7 +43,7 @@ export class ConsumerController {
 
   @Post('/pending-messages/get')
   @ApiRedisInstanceOperation({
-    description: 'Get pending messages list',
+    description: 'Get pending entries list',
     statusCode: 200,
     responses: [
       {
@@ -62,20 +62,37 @@ export class ConsumerController {
 
   @Post('/pending-messages/ack')
   @ApiRedisInstanceOperation({
-    description: 'Get pending messages list',
+    description: 'Ack pending entries',
     statusCode: 200,
     responses: [
       {
         status: 200,
-        type: PendingEntryDto,
-        isArray: true,
+        type: AckPendingEntriesResponse,
       },
     ],
   })
-  async ackPendingEntriers(
+  async ackPendingEntries(
     @Param('dbInstance') instanceId: string,
       @Body() dto: AckPendingEntriesDto,
   ): Promise<AckPendingEntriesResponse> {
     return this.service.ackPendingEntries({ instanceId }, dto);
   }
+
+  @Post('/pending-messages/claim')
+  @ApiRedisInstanceOperation({
+    description: 'Claim pending entries',
+    statusCode: 200,
+    responses: [
+      {
+        status: 200,
+        type: ClaimPendingEntriesResponse,
+      },
+    ],
+  })
+  async claimPendingEntries(
+    @Param('dbInstance') instanceId: string,
+      @Body() dto: ClaimPendingEntryDto,
+  ): Promise<ClaimPendingEntriesResponse> {
+    return this.service.claimPendingEntries({ instanceId }, dto);
+  }
 }

redisinsight/api/src/modules/browser/dto/stream.dto.ts

@@ -5,7 +5,7 @@ import {
   ArrayNotEmpty,
   IsArray,
   IsDefined,
-  IsEnum, IsInt, IsNotEmpty, IsString, Min, ValidateNested, isString,
+  IsEnum, IsInt, IsNotEmpty, IsString, Min, ValidateNested, isString, IsOptional, NotEquals, ValidateIf, IsBoolean,
 } from 'class-validator';
 import { KeyDto, KeyWithExpireDto } from 'src/modules/browser/dto/keys.dto';
 import { SortOrder } from 'src/constants';
@@ -418,3 +418,105 @@ export class AckPendingEntriesResponse {
   })
   affected: number;
 }
+
+export class ClaimPendingEntryDto extends KeyDto {
+  @ApiProperty({
+    type: String,
+    description: 'Consumer group name',
+    example: 'group-1',
+  })
+  @IsNotEmpty()
+  @IsString()
+  groupName: string;
+
+  @ApiProperty({
+    type: String,
+    description: 'Consumer name',
+    example: 'consumer-1',
+  })
+  @IsNotEmpty()
+  @IsString()
+  consumerName: string;
+
+  @ApiProperty({
+    description: 'Claim only if its idle time is greater the minimum idle time ',
+    type: Number,
+    minimum: 0,
+    default: 0,
+  })
+  @IsInt()
+  @Min(0)
+  minIdleTime: number = 0;
+
+  @ApiProperty({
+    description: 'Entries IDs',
+    type: String,
+    isArray: true,
+    example: ['1650985323741-0', '1650985323770-0'],
+  })
+  @IsDefined()
+  @IsArray()
+  @ArrayNotEmpty()
+  @Type(() => String)
+  entries: string[];
+
+  @ApiPropertyOptional({
+    description: 'Set the idle time (last time it was delivered) of the message',
+    type: Number,
+    minimum: 0,
+  })
+  @NotEquals(null)
+  @ValidateIf((object, value) => value !== undefined)
+  @IsInt()
+  @Min(0)
+  idle?: number;
+
+  @ApiPropertyOptional({
+    description: 'This is the same as IDLE but instead of a relative amount of milliseconds, '
+      + 'it sets the idle time to a specific Unix time (in milliseconds)',
+    type: Number,
+  })
+  @NotEquals(null)
+  @ValidateIf((object, value) => value !== undefined)
+  @IsInt()
+  time?: number;
+
+  @ApiPropertyOptional({
+    description: 'Set the retry counter to the specified value. '
+      + 'This counter is incremented every time a message is delivered again. '
+      + 'Normally XCLAIM does not alter this counter, which is just served to clients when the XPENDING command '
+      + 'is called: this way clients can detect anomalies, like messages that are never processed '
+      + 'for some reason after a big number of delivery attempts',
+    type: Number,
+    minimum: 0,
+  })
+  @NotEquals(null)
+  @ValidateIf((object, value) => value !== undefined)
+  @IsInt()
+  @Min(0)
+  retryCount?: number;
+
+  @ApiPropertyOptional({
+    description: 'Creates the pending message entry in the PEL even if certain specified IDs are not already '
+      + 'in the PEL assigned to a different client',
+    type: Boolean,
+  })
+  @NotEquals(null)
+  @ValidateIf((object, value) => value !== undefined)
+  @IsBoolean()
+  force?: boolean;
+}
+
+export class ClaimPendingEntriesResponse {
+  @ApiProperty({
+    description: 'Entries IDs were affected by claim command',
+    type: String,
+    isArray: true,
+    example: ['1650985323741-0', '1650985323770-0'],
+  })
+  @IsDefined()
+  @IsArray()
+  @ArrayNotEmpty()
+  @Type(() => String)
+  affected: string[];
+}

redisinsight/api/src/modules/browser/services/stream/consumer.service.ts

@@ -10,7 +10,7 @@ import {
 import { BrowserToolService } from 'src/modules/browser/services/browser-tool/browser-tool.service';
 import ERROR_MESSAGES from 'src/constants/error-messages';
 import {
-  AckPendingEntriesDto, AckPendingEntriesResponse,
+  AckPendingEntriesDto, AckPendingEntriesResponse, ClaimPendingEntriesResponse, ClaimPendingEntryDto,
   ConsumerDto,
   GetConsumersDto, GetPendingEntriesDto, PendingEntryDto,
 } from 'src/modules/browser/dto/stream.dto';
@@ -102,7 +102,7 @@ export class ConsumerService {
   }
 
   /**
-   * Get list of pending entries info for particular consumer
+   * Acknowledge pending entries
    * @param clientOptions
    * @param dto
    */
@@ -147,6 +147,71 @@ export class ConsumerService {
     }
   }
 
+  /**
+   * Claim pending entries with additional parameters
+   * @param clientOptions
+   * @param dto
+   */
+  async claimPendingEntries(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: ClaimPendingEntryDto,
+  ): Promise<ClaimPendingEntriesResponse> {
+    try {
+      this.logger.log('Claiming pending entries.');
+
+      const exists = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolKeysCommands.Exists,
+        [dto.keyName],
+      );
+
+      if (!exists) {
+        return Promise.reject(new NotFoundException(ERROR_MESSAGES.KEY_NOT_EXIST));
+      }
+
+      const args = [dto.keyName, dto.groupName, dto.consumerName, dto.minIdleTime, ...dto.entries];
+
+      if (dto.idle !== undefined) {
+        args.push('idle', dto.idle);
+      } else if (dto.time !== undefined) {
+        args.push('time', dto.time);
+      }
+
+      if (dto.retryCount !== undefined) {
+        args.push('retrycount', dto.retryCount);
+      }
+
+      if (dto.force) {
+        args.push('force');
+      }
+
+      // Return just an array of IDs of messages successfully claimed, without returning the actual message.
+      args.push('justid');
+
+      const affected = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolStreamCommands.XClaim,
+        args,
+      );
+
+      this.logger.log('Successfully claimed pending entries.');
+
+      return {
+        affected,
+      };
+    } catch (error) {
+      if (error instanceof NotFoundException) {
+        throw error;
+      }
+
+      if (error?.message.includes(RedisErrorCodes.WrongType)) {
+        throw new BadRequestException(error.message);
+      }
+
+      throw catchAclError(error);
+    }
+  }
+
   /**
    * Converts RESP response from Redis
    * [