#RI-2943 BE for pending messages list

Author: Artem

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

@@ -11,7 +11,7 @@ import { ApiRedisInstanceOperation } from 'src/decorators/api-redis-instance-ope
 import {
   ConsumerDto,
   ConsumerGroupDto,
-  GetConsumersDto,
+  GetConsumersDto, GetPendingMessagesDto, PendingMessageDto,
 } from 'src/modules/browser/dto/stream.dto';
 import { ConsumerService } from 'src/modules/browser/services/stream/consumer.service';
 
@@ -23,12 +23,11 @@ export class ConsumerController {
 
   @Post('/get')
   @ApiRedisInstanceOperation({
-    description: 'Get stream entries',
+    description: 'Get group consumers',
     statusCode: 200,
     responses: [
       {
         status: 200,
-        description: 'Returns stream consumer groups.',
         type: ConsumerGroupDto,
         isArray: true,
       },
@@ -40,4 +39,23 @@ export class ConsumerController {
   ): Promise<ConsumerDto[]> {
     return this.service.getConsumers({ instanceId }, dto);
   }
+
+  @Post('/pending-messages/get')
+  @ApiRedisInstanceOperation({
+    description: 'Get pending messages list',
+    statusCode: 200,
+    responses: [
+      {
+        status: 200,
+        type: PendingMessageDto,
+        isArray: true,
+      },
+    ],
+  })
+  async getPendingMessages(
+    @Param('dbInstance') instanceId: string,
+      @Body() dto: GetPendingMessagesDto,
+  ): Promise<PendingMessageDto[]> {
+    return this.service.getPendingMessages({ instanceId }, dto);
+  }
 }

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

@@ -324,3 +324,75 @@ export class GetConsumersDto extends KeyDto {
   @IsString()
   groupName: string;
 }
+
+export class PendingMessageDto {
+  @ApiProperty({
+    type: String,
+    description: 'Entry ID',
+    example: '*',
+  })
+  id: string;
+
+  @ApiProperty({
+    type: String,
+    description: 'Consumer name',
+    example: 'consumer-1',
+  })
+  consumerName: string;
+
+  @ApiProperty({
+    type: Number,
+    description: 'The number of milliseconds that elapsed since the last time '
+      + 'this message was delivered to this consumer',
+    example: 22442,
+  })
+  idle: number = 0;
+
+  @ApiProperty({
+    type: Number,
+    description: 'The number of times this message was delivered',
+    example: 2,
+  })
+  delivered: number = 0;
+}
+
+export class GetPendingMessagesDto extends IntersectionType(
+  KeyDto,
+  GetConsumersDto,
+) {
+  @ApiProperty({
+    type: String,
+    description: 'Consumer name',
+    example: 'consumer-1',
+  })
+  @IsNotEmpty()
+  @IsString()
+  consumerName: string;
+
+  @ApiPropertyOptional({
+    description: 'Specifying the start id',
+    type: String,
+    default: '-',
+  })
+  @IsString()
+  start?: string = '-';
+
+  @ApiPropertyOptional({
+    description: 'Specifying the end id',
+    type: String,
+    default: '+',
+  })
+  @IsString()
+  end?: string = '+';
+
+  @ApiPropertyOptional({
+    description:
+      'Specifying the number of pending messages to return.',
+    type: Number,
+    minimum: 1,
+    default: 500,
+  })
+  @IsInt()
+  @Min(1)
+  count?: number = 500;
+}

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

@@ -11,7 +11,7 @@ import { BrowserToolService } from 'src/modules/browser/services/browser-tool/br
 import ERROR_MESSAGES from 'src/constants/error-messages';
 import {
   ConsumerDto,
-  GetConsumersDto,
+  GetConsumersDto, GetPendingMessagesDto, PendingMessageDto,
 } from 'src/modules/browser/dto/stream.dto';
 
 @Injectable()
@@ -60,6 +60,46 @@ export class ConsumerService {
     }
   }
 
+  /**
+   * Get list of pending messages info for particular consumer
+   * @param clientOptions
+   * @param dto
+   */
+  async getPendingMessages(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: GetPendingMessagesDto,
+  ): Promise<PendingMessageDto[]> {
+    try {
+      this.logger.log('Getting pending messages list.');
+
+      const exists = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolKeysCommands.Exists,
+        [dto.keyName],
+      );
+
+      if (!exists) {
+        return Promise.reject(new NotFoundException(ERROR_MESSAGES.KEY_NOT_EXIST));
+      }
+
+      return ConsumerService.formatReplyToPendingMessagesDto(await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolStreamCommands.XPending,
+        [dto.keyName, dto.groupName, dto.start, dto.end, dto.count, dto.consumerName],
+      ));
+    } 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
    * [
@@ -106,4 +146,45 @@ export class ConsumerService {
       idle: entryObj['idle'],
     };
   }
+
+  /**
+   * Converts RESP response from Redis
+   * [
+   *  ['1567352639-0', 'consumer-1', 258741, 2],
+   *  ...
+   * ]
+   *
+   * to DTO
+   *
+   * [
+   *  {
+   *    id: '1567352639-0',
+   *    name: 'consumer-1',
+   *    idle: 258741,
+   *    delivered: 2,
+   *  },
+   *   ...
+   * ]
+   * @param reply
+   */
+  static formatReplyToPendingMessagesDto(reply: Array<Array<string | number>>): PendingMessageDto[] {
+    return reply.map(ConsumerService.formatArrayToPendingMessageDto);
+  }
+
+  /**
+   * Format single reply entry to DTO
+   * @param entry
+   */
+  static formatArrayToPendingMessageDto(entry: Array<string | number>): PendingMessageDto {
+    if (!entry?.length) {
+      return null;
+    }
+
+    return {
+      id: `${entry[0]}`,
+      consumerName: `${entry[1]}`,
+      idle: +entry[2],
+      delivered: +entry[3],
+    };
+  }
 }