#RI-2931 BE implementation for create new consumer group(s)

#RI-2933 BE implementation for consumer groups list

Author: Artem

redisinsight/api/src/constants/redis-error-codes.ts

@@ -10,6 +10,7 @@ export enum RedisErrorCodes {
   ConnectionReset = 'ECONNRESET',
   Timeout = 'ETIMEDOUT',
   CommandSyntaxError = 'syntax error',
+  BusyGroup = 'BUSYGROUP',
   UnknownCommand = 'unknown command',
 }
 

redisinsight/api/src/modules/browser/browser.module.ts

@@ -4,6 +4,8 @@ import { SharedModule } from 'src/modules/shared/shared.module';
 import { RedisConnectionMiddleware } from 'src/middleware/redis-connection.middleware';
 import { StreamController } from 'src/modules/browser/controllers/stream/stream.controller';
 import { StreamService } from 'src/modules/browser/services/stream/stream.service';
+import { ConsumerGroupController } from 'src/modules/browser/controllers/stream/consumer-group.controller';
+import { ConsumerGroupService } from 'src/modules/browser/services/stream/consumer-group.service';
 import { HashController } from './controllers/hash/hash.controller';
 import { KeysController } from './controllers/keys/keys.controller';
 import { KeysBusinessService } from './services/keys-business/keys-business.service';
@@ -32,6 +34,7 @@ import { BrowserToolClusterService } from './services/browser-tool-cluster/brows
     RejsonRlController,
     HashController,
     StreamController,
+    ConsumerGroupController,
   ],
   providers: [
     KeysBusinessService,
@@ -42,6 +45,7 @@ import { BrowserToolClusterService } from './services/browser-tool-cluster/brows
     RejsonRlBusinessService,
     HashBusinessService,
     StreamService,
+    ConsumerGroupService,
     BrowserToolService,
     BrowserToolClusterService,
   ],

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

@@ -80,6 +80,9 @@ export enum BrowserToolStreamCommands {
   XRevRange = 'xrevrange',
   XAdd = 'xadd',
   XDel = 'xdel',
+  XInfoGroups = 'xinfo groups',
+  XPending = 'xpending',
+  XGroupCreate = 'xgroup create',
 }
 
 export enum BrowserToolTSCommands {

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

@@ -0,0 +1,54 @@
+import {
+  Body,
+  Controller,
+  Param,
+  Post,
+  UsePipes,
+  ValidationPipe,
+} from '@nestjs/common';
+import { ApiTags } from '@nestjs/swagger';
+import { ApiRedisInstanceOperation } from 'src/decorators/api-redis-instance-operation.decorator';
+import {
+  ConsumerGroupDto, CreateConsumerGroupsDto,
+  GetStreamEntriesResponse,
+} from 'src/modules/browser/dto/stream.dto';
+import { ConsumerGroupService } from 'src/modules/browser/services/stream/consumer-group.service';
+import { KeyDto } from 'src/modules/browser/dto';
+
+@ApiTags('Streams')
+@Controller('streams/consumer-groups')
+@UsePipes(new ValidationPipe({ transform: true }))
+export class ConsumerGroupController {
+  constructor(private service: ConsumerGroupService) {}
+
+  @Post('/get')
+  @ApiRedisInstanceOperation({
+    description: 'Get stream entries',
+    statusCode: 200,
+    responses: [
+      {
+        status: 200,
+        description: 'Returns ordered stream entries in defined range.',
+        type: GetStreamEntriesResponse,
+      },
+    ],
+  })
+  async getGroups(
+    @Param('dbInstance') instanceId: string,
+      @Body() dto: KeyDto,
+  ): Promise<ConsumerGroupDto[]> {
+    return this.service.getGroups({ instanceId }, dto);
+  }
+
+  @Post('')
+  @ApiRedisInstanceOperation({
+    description: 'Create stream consumer group',
+    statusCode: 201,
+  })
+  async createGroups(
+    @Param('dbInstance') instanceId: string,
+      @Body() dto: CreateConsumerGroupsDto,
+  ): Promise<void> {
+    return this.service.createGroups({ instanceId }, dto);
+  }
+}

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

@@ -1,4 +1,6 @@
-import { ApiProperty, ApiPropertyOptional, IntersectionType } from '@nestjs/swagger';
+import {
+  ApiProperty, ApiPropertyOptional, IntersectionType
+} from '@nestjs/swagger';
 import {
   ArrayNotEmpty,
   IsArray,
@@ -183,3 +185,79 @@ export class CreateStreamDto extends IntersectionType(
   AddStreamEntriesDto,
   KeyWithExpireDto,
 ) {}
+
+export class ConsumerGroupDto {
+  @ApiProperty({
+    type: String,
+    description: 'Consumer group name',
+    example: 'group',
+  })
+  name: string;
+
+  @ApiProperty({
+    type: Number,
+    description: 'Number of consumers',
+    example: 2,
+  })
+  consumers: number = 0;
+
+  @ApiProperty({
+    type: Number,
+    description: 'Number of pending messages',
+    example: 2,
+  })
+  pending: number = 0;
+
+  @ApiProperty({
+    type: String,
+    description: 'Smallest Id of the message that is pending in the group',
+    example: '1657892649-0',
+  })
+  smallestPendingId: string;
+
+  @ApiProperty({
+    type: String,
+    description: 'Greatest Id of the message that is pending in the group',
+    example: '1657892680-0',
+  })
+  greatestPendingId: string;
+
+  @ApiProperty({
+    type: String,
+    description: 'Id of last delivered message',
+    example: '1657892648-0',
+  })
+  lastDeliveredId: string;
+}
+
+export class CreateConsumerGroupDto {
+  @ApiProperty({
+    type: String,
+    description: 'Consumer group name',
+    example: 'group',
+  })
+  @IsNotEmpty()
+  @IsString()
+  name: string;
+
+  @ApiProperty({
+    type: String,
+    description: 'Id of last delivered message',
+    example: '1657892648-0',
+  })
+  @IsNotEmpty()
+  @IsString()
+  lastDeliveredId: string;
+}
+
+export class CreateConsumerGroupsDto extends KeyDto {
+  @ApiProperty({
+    type: () => CreateConsumerGroupDto,
+    isArray: true,
+    description: 'List of consumer groups to create',
+  })
+  @ValidateNested()
+  @IsArray()
+  @Type(() => CreateConsumerGroupDto)
+  consumerGroups: CreateConsumerGroupDto[];
+}

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

@@ -0,0 +1,205 @@
+import {
+  BadRequestException, ConflictException, Injectable, Logger, NotFoundException,
+} from '@nestjs/common';
+import { IFindRedisClientInstanceByOptions } from 'src/modules/core/services/redis/redis.service';
+import { RedisErrorCodes } from 'src/constants';
+import { catchAclError, catchTransactionError, convertStringsArrayToObject } from 'src/utils';
+import {
+  BrowserToolCommands,
+  BrowserToolKeysCommands, BrowserToolStreamCommands,
+} from 'src/modules/browser/constants/browser-tool-commands';
+import { BrowserToolService } from 'src/modules/browser/services/browser-tool/browser-tool.service';
+import { KeyDto } from 'src/modules/browser/dto';
+import ERROR_MESSAGES from 'src/constants/error-messages';
+import { ConsumerGroupDto, CreateConsumerGroupsDto } from 'src/modules/browser/dto/stream.dto';
+
+@Injectable()
+export class ConsumerGroupService {
+  private logger = new Logger('ConsumerGroupService');
+
+  constructor(private browserTool: BrowserToolService) {}
+
+  /**
+   * Get consumer groups list for particular stream
+   * In addition fetch pending messages info for each group
+   * !May be slow on huge streams as 'XPENDING' command tagged with as @slow
+   * @param clientOptions
+   * @param dto
+   */
+  async getGroups(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: KeyDto,
+  ): Promise<ConsumerGroupDto[]> {
+    try {
+      this.logger.log('Getting consumer groups list.');
+
+      const exists = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolKeysCommands.Exists,
+        [dto.keyName],
+      );
+
+      if (!exists) {
+        return Promise.reject(new NotFoundException(ERROR_MESSAGES.KEY_NOT_EXIST));
+      }
+
+      const groups = ConsumerGroupService.formatReplyToDto(await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolStreamCommands.XInfoGroups,
+        [dto.keyName],
+      ));
+
+      return await Promise.all(groups.map((group) => this.getGroupInfo(
+        clientOptions,
+        dto,
+        group,
+      )));
+    } catch (error) {
+      if (error instanceof NotFoundException) {
+        throw error;
+      }
+
+      if (error?.message.includes(RedisErrorCodes.WrongType)) {
+        throw new BadRequestException(error.message);
+      }
+
+      throw catchAclError(error);
+    }
+  }
+
+  /**
+   * Get consumer group pending info using 'XPENDING' command
+   * @param clientOptions
+   * @param dto
+   * @param group
+   */
+  async getGroupInfo(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: KeyDto,
+    group: ConsumerGroupDto,
+  ): Promise<ConsumerGroupDto> {
+    const info = await this.browserTool.execCommand(
+      clientOptions,
+      BrowserToolStreamCommands.XPending,
+      [dto.keyName, group.name],
+    );
+
+    return {
+      ...group,
+      smallestPendingId: info?.[1] || null,
+      greatestPendingId: info?.[2] || null,
+    };
+  }
+
+  /**
+   * Create consumer group(s)
+   * @param clientOptions
+   * @param dto
+   */
+  async createGroups(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: CreateConsumerGroupsDto,
+  ): Promise<void> {
+    try {
+      this.logger.log('Creating consumer groups.');
+      const { keyName, consumerGroups } = dto;
+
+      const exists = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolKeysCommands.Exists,
+        [keyName],
+      );
+
+      if (!exists) {
+        return Promise.reject(new NotFoundException(ERROR_MESSAGES.KEY_NOT_EXIST));
+      }
+
+      const toolCommands: Array<[
+        toolCommand: BrowserToolCommands,
+        ...args: Array<string | number>,
+      ]> = consumerGroups.map((group) => (
+        [
+          BrowserToolStreamCommands.XGroupCreate,
+          keyName,
+          group.name,
+          group.lastDeliveredId,
+        ]
+      ));
+
+      const [
+        transactionError,
+        transactionResults,
+      ] = await this.browserTool.execMulti(clientOptions, toolCommands);
+      catchTransactionError(transactionError, transactionResults);
+
+      this.logger.log('Stream consumer group(s) created.');
+
+      return undefined;
+    } catch (error) {
+      if (error instanceof NotFoundException) {
+        throw error;
+      }
+
+      if (error?.message.includes(RedisErrorCodes.WrongType)) {
+        throw new BadRequestException(error.message);
+      }
+
+      if (error?.message.includes(RedisErrorCodes.BusyGroup)) {
+        throw new ConflictException(error.message);
+      }
+
+      throw catchAclError(error);
+    }
+  }
+
+  /**
+   * Converts RESP response from Redis
+   * [
+   *  ['name', 'g1', 'consumers', 0, 'pending', 0, 'last-delivered-id', '1653034260278-0'],
+   *  ['name', 'g2', 'consumers', 0, 'pending', 0, 'last-delivered-id', '1653034260278-0'],
+   *  ...
+   * ]
+   *
+   * to DTO
+   *
+   * [
+   *  {
+   *    name: 'g1',
+   *    consumers: 0,
+   *    pending: 0,
+   *    lastDeliveredId: '1653034260278-0'
+   *  },
+   *  {
+   *    name: 'g2',
+   *    consumers: 0,
+   *    pending: 0,
+   *    lastDeliveredId: '1653034260278-0'
+   *  },
+   *   ...
+   * ]
+   * @param reply
+   */
+  static formatReplyToDto(reply: Array<Array<string | number>>): ConsumerGroupDto[] {
+    return reply.map(ConsumerGroupService.formatArrayToDto);
+  }
+
+  /**
+   * Format single reply entry to DTO
+   * @param entry
+   */
+  static formatArrayToDto(entry: Array<string | number>): ConsumerGroupDto {
+    if (!entry?.length) {
+      return null;
+    }
+    const entryObj = convertStringsArrayToObject(entry as string[]);
+
+    return {
+      name: entryObj['name'],
+      consumers: entryObj['consumers'],
+      pending: entryObj['pending'],
+      lastDeliveredId: entryObj['last-delivered-id'],
+      smallestPendingId: null,
+      greatestPendingId: null,
+    };
+  }
+}