#RI-2945 BE Ack Pending entries

Author: Artem

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

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

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

@@ -9,6 +9,7 @@ import {
 import { ApiTags } from '@nestjs/swagger';
 import { ApiRedisInstanceOperation } from 'src/decorators/api-redis-instance-operation.decorator';
 import {
+  AckPendingEntriesDto, AckPendingEntriesResponse,
   ConsumerDto,
   ConsumerGroupDto,
   GetConsumersDto, GetPendingEntriesDto, PendingEntryDto,
@@ -52,10 +53,29 @@ export class ConsumerController {
       },
     ],
   })
-  async getPendingMessages(
+  async getPendingEntries(
     @Param('dbInstance') instanceId: string,
       @Body() dto: GetPendingEntriesDto,
   ): Promise<PendingEntryDto[]> {
-    return this.service.getPendingMessages({ instanceId }, dto);
+    return this.service.getPendingEntries({ instanceId }, dto);
+  }
+
+  @Post('/pending-messages/ack')
+  @ApiRedisInstanceOperation({
+    description: 'Get pending messages list',
+    statusCode: 200,
+    responses: [
+      {
+        status: 200,
+        type: PendingEntryDto,
+        isArray: true,
+      },
+    ],
+  })
+  async ackPendingEntriers(
+    @Param('dbInstance') instanceId: string,
+      @Body() dto: AckPendingEntriesDto,
+  ): Promise<AckPendingEntriesResponse> {
+    return this.service.ackPendingEntries({ instanceId }, dto);
   }
 }

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

@@ -189,7 +189,7 @@ export class CreateStreamDto extends IntersectionType(
 export class ConsumerGroupDto {
   @ApiProperty({
     type: String,
-    description: 'Consumer group name',
+    description: 'Consumer Group name',
     example: 'group',
   })
   name: string;
@@ -294,7 +294,7 @@ export class ConsumerDto {
   @ApiProperty({
     type: String,
     description: 'The consumer\'s name',
-    example: 'consumer-1',
+    example: 'consumer-2',
   })
   name: string;
 
@@ -396,3 +396,25 @@ export class GetPendingEntriesDto extends IntersectionType(
   @Min(1)
   count?: number = 500;
 }
+
+export class AckPendingEntriesDto extends GetConsumersDto {
+  @ApiProperty({
+    description: 'Entries IDs',
+    type: String,
+    isArray: true,
+    example: ['1650985323741-0', '1650985323770-0'],
+  })
+  @IsDefined()
+  @IsArray()
+  @ArrayNotEmpty()
+  @Type(() => String)
+  entries: string[];
+}
+
+export class AckPendingEntriesResponse {
+  @ApiProperty({
+    description: 'Number of affected entries',
+    type: Number,
+  })
+  affected: number;
+}

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

@@ -10,6 +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,
   ConsumerDto,
   GetConsumersDto, GetPendingEntriesDto, PendingEntryDto,
 } from 'src/modules/browser/dto/stream.dto';
@@ -61,16 +62,16 @@ export class ConsumerService {
   }
 
   /**
-   * Get list of pending messages info for particular consumer
+   * Get list of pending entries info for particular consumer
    * @param clientOptions
    * @param dto
    */
-  async getPendingMessages(
+  async getPendingEntries(
     clientOptions: IFindRedisClientInstanceByOptions,
     dto: GetPendingEntriesDto,
   ): Promise<PendingEntryDto[]> {
     try {
-      this.logger.log('Getting pending messages list.');
+      this.logger.log('Getting pending entries list.');
 
       const exists = await this.browserTool.execCommand(
         clientOptions,
@@ -82,7 +83,7 @@ export class ConsumerService {
         return Promise.reject(new NotFoundException(ERROR_MESSAGES.KEY_NOT_EXIST));
       }
 
-      return ConsumerService.formatReplyToPendingMessagesDto(await this.browserTool.execCommand(
+      return ConsumerService.formatReplyToPendingEntriesDto(await this.browserTool.execCommand(
         clientOptions,
         BrowserToolStreamCommands.XPending,
         [dto.keyName, dto.groupName, dto.start, dto.end, dto.count, dto.consumerName],
@@ -100,6 +101,52 @@ export class ConsumerService {
     }
   }
 
+  /**
+   * Get list of pending entries info for particular consumer
+   * @param clientOptions
+   * @param dto
+   */
+  async ackPendingEntries(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: AckPendingEntriesDto,
+  ): Promise<AckPendingEntriesResponse> {
+    try {
+      this.logger.log('Acknowledging 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 affected = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolStreamCommands.XAck,
+        [dto.keyName, dto.groupName, ...dto.entries],
+      );
+
+      this.logger.log('Successfully acknowledged 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
    * [
@@ -167,15 +214,15 @@ export class ConsumerService {
    * ]
    * @param reply
    */
-  static formatReplyToPendingMessagesDto(reply: Array<Array<string | number>>): PendingEntryDto[] {
-    return reply.map(ConsumerService.formatArrayToPendingMessageDto);
+  static formatReplyToPendingEntriesDto(reply: Array<Array<string | number>>): PendingEntryDto[] {
+    return reply.map(ConsumerService.formatArrayToPendingEntryDto);
   }
 
   /**
    * Format single reply entry to DTO
    * @param entry
    */
-  static formatArrayToPendingMessageDto(entry: Array<string | number>): PendingEntryDto {
+  static formatArrayToPendingEntryDto(entry: Array<string | number>): PendingEntryDto {
     if (!entry?.length) {
       return null;
     }