RI:7799: Stream bulk delete report (#5278)

Author: pawelangelow

redisinsight/api/src/modules/bulk-actions/bulk-actions.controller.ts

@@ -0,0 +1,37 @@
+import { Response } from 'express';
+import {
+  Controller,
+  Get,
+  Param,
+  Res,
+  UsePipes,
+  ValidationPipe,
+} from '@nestjs/common';
+import { ApiParam, ApiTags } from '@nestjs/swagger';
+import { ApiEndpoint } from 'src/decorators/api-endpoint.decorator';
+import { BulkActionsService } from 'src/modules/bulk-actions/bulk-actions.service';
+import { BulkActionIdDto } from 'src/modules/bulk-actions/dto/bulk-action-id.dto';
+
+@UsePipes(new ValidationPipe({ transform: true }))
+@ApiTags('Bulk Actions')
+@Controller('bulk-actions')
+export class BulkActionsController {
+  constructor(private readonly service: BulkActionsService) {}
+
+  @ApiEndpoint({
+    description: 'Stream bulk action report as downloadable file',
+    statusCode: 200,
+  })
+  @ApiParam({
+    name: 'id',
+    description: 'Bulk action id',
+    type: String,
+  })
+  @Get(':id/report/download')
+  async downloadReport(
+    @Param() { id }: BulkActionIdDto,
+    @Res() res: Response,
+  ): Promise<void> {
+    await this.service.streamReport(id, res);
+  }
+}

redisinsight/api/src/modules/bulk-actions/bulk-actions.module.ts

@@ -3,11 +3,12 @@ import { BulkActionsService } from 'src/modules/bulk-actions/bulk-actions.servic
 import { BulkActionsProvider } from 'src/modules/bulk-actions/providers/bulk-actions.provider';
 import { BulkActionsGateway } from 'src/modules/bulk-actions/bulk-actions.gateway';
 import { BulkActionsAnalytics } from 'src/modules/bulk-actions/bulk-actions.analytics';
+import { BulkActionsController } from 'src/modules/bulk-actions/bulk-actions.controller';
 import { BulkImportController } from 'src/modules/bulk-actions/bulk-import.controller';
 import { BulkImportService } from 'src/modules/bulk-actions/bulk-import.service';
 
 @Module({
-  controllers: [BulkImportController],
+  controllers: [BulkActionsController, BulkImportController],
   providers: [
     BulkActionsGateway,
     BulkActionsService,

redisinsight/api/src/modules/bulk-actions/bulk-actions.service.spec.ts

@@ -110,4 +110,70 @@ describe('BulkActionsService', () => {
       expect(bulkActionProvider.abortUsersBulkActions).toHaveBeenCalledTimes(1);
     });
   });
+
+  describe('streamReport', () => {
+    let mockResponse: any;
+    let mockBulkActionWithReport: any;
+
+    beforeEach(() => {
+      mockResponse = {
+        setHeader: jest.fn(),
+        write: jest.fn(),
+        end: jest.fn(),
+      };
+
+      mockBulkActionWithReport = {
+        setStreamingResponse: jest.fn(),
+        isReportEnabled: jest.fn().mockReturnValue(true),
+      };
+    });
+
+    it('should throw NotFoundException when bulk action not found', async () => {
+      bulkActionProvider.get = jest.fn().mockReturnValue(null);
+
+      await expect(
+        service.streamReport('non-existent-id', mockResponse),
+      ).rejects.toThrow('Bulk action not found');
+    });
+
+    it('should throw BadRequestException when report not enabled', async () => {
+      mockBulkActionWithReport.isReportEnabled.mockReturnValue(false);
+      bulkActionProvider.get = jest
+        .fn()
+        .mockReturnValue(mockBulkActionWithReport);
+
+      await expect(
+        service.streamReport('bulk-action-id', mockResponse),
+      ).rejects.toThrow(
+        'Report generation was not enabled for this bulk action',
+      );
+    });
+
+    it('should set headers and attach stream to bulk action', async () => {
+      bulkActionProvider.get = jest
+        .fn()
+        .mockReturnValue(mockBulkActionWithReport);
+      const mockTimestamp = '1733047200000'; // 2024-12-01T10:00:00.000Z
+      const expectedFilename =
+        'bulk-delete-report-2024-12-01T10-00-00-000Z.txt';
+
+      await service.streamReport(mockTimestamp, mockResponse);
+
+      expect(mockResponse.setHeader).toHaveBeenCalledWith(
+        'Content-Type',
+        'text/plain',
+      );
+      expect(mockResponse.setHeader).toHaveBeenCalledWith(
+        'Content-Disposition',
+        `attachment; filename="${expectedFilename}"`,
+      );
+      expect(mockResponse.setHeader).toHaveBeenCalledWith(
+        'Transfer-Encoding',
+        'chunked',
+      );
+      expect(
+        mockBulkActionWithReport.setStreamingResponse,
+      ).toHaveBeenCalledWith(mockResponse);
+    });
+  });
 });

redisinsight/api/src/modules/bulk-actions/bulk-actions.service.ts

@@ -1,5 +1,10 @@
 import { Socket } from 'socket.io';
-import { Injectable } from '@nestjs/common';
+import { Response } from 'express';
+import {
+  BadRequestException,
+  Injectable,
+  NotFoundException,
+} from '@nestjs/common';
 import { BulkActionsProvider } from 'src/modules/bulk-actions/providers/bulk-actions.provider';
 import { CreateBulkActionDto } from 'src/modules/bulk-actions/dto/create-bulk-action.dto';
 import { BulkActionIdDto } from 'src/modules/bulk-actions/dto/bulk-action-id.dto';
@@ -44,4 +49,36 @@ export class BulkActionsService {
   disconnect(socketId: string) {
     this.bulkActionsProvider.abortUsersBulkActions(socketId);
   }
+
+  /**
+   * Stream bulk action report as downloadable file
+   * @param id Bulk action id
+   * @param res Express response object
+   */
+  async streamReport(id: string, res: Response): Promise<void> {
+    const bulkAction = this.bulkActionsProvider.get(id);
+
+    if (!bulkAction) {
+      throw new NotFoundException('Bulk action not found');
+    }
+
+    if (!bulkAction.isReportEnabled()) {
+      throw new BadRequestException(
+        'Report generation was not enabled for this bulk action',
+      );
+    }
+
+    // Set headers for file download
+    const timestamp = new Date(Number(id)).toISOString().replace(/[:.]/g, '-');
+    res.setHeader('Content-Type', 'text/plain');
+    res.setHeader(
+      'Content-Disposition',
+      `attachment; filename="bulk-delete-report-${timestamp}.txt"`,
+    );
+    res.setHeader('Transfer-Encoding', 'chunked');
+
+    // Attach the response stream to the bulk action
+    // This will trigger the bulk action to start processing
+    bulkAction.setStreamingResponse(res);
+  }
 }

redisinsight/api/src/modules/bulk-actions/dto/create-bulk-action.dto.ts

@@ -1,6 +1,7 @@
 import { BulkActionFilter } from 'src/modules/bulk-actions/models/bulk-action-filter';
 import { BulkActionType } from 'src/modules/bulk-actions/constants';
 import {
+  IsBoolean,
   IsEnum,
   IsNotEmpty,
   IsNumber,
@@ -35,4 +36,9 @@ export class CreateBulkActionDto extends BulkActionIdDto {
   @Min(0)
   @Max(2147483647)
   db?: number;
+
+  @IsOptional()
+  @IsBoolean()
+  @Type(() => Boolean)
+  generateReport?: boolean;
 }

redisinsight/api/src/modules/bulk-actions/interfaces/bulk-action-overview.interface.ts

@@ -15,4 +15,6 @@ export interface IBulkActionOverview {
   filter: IBulkActionFilterOverview; // Note: This can be null, according to the API response
   progress: IBulkActionProgressOverview;
   summary: IBulkActionSummaryOverview;
+  downloadUrl?: string;
+  error?: string;
 }

redisinsight/api/src/modules/bulk-actions/interfaces/bulk-action.interface.ts

@@ -7,4 +7,5 @@ export interface IBulkAction {
   getFilter(): BulkActionFilter;
   changeState(): void;
   getSocket(): Socket;
+  writeToReport(keyName: Buffer, success: boolean, error?: string): void;
 }

redisinsight/api/src/modules/bulk-actions/models/bulk-action-summary.spec.ts

@@ -81,12 +81,30 @@ describe('BulkActionSummary', () => {
       expect(summary['errors']).toEqual(generateErrors(500));
     });
   });
+  describe('addKeys', () => {
+    it('should add keys', async () => {
+      expect(summary['keys']).toEqual([]);
+
+      summary.addKeys([Buffer.from('key1')]);
+
+      expect(summary['keys']).toEqual([Buffer.from('key1')]);
+
+      summary.addKeys([Buffer.from('key2'), Buffer.from('key3')]);
+
+      expect(summary['keys']).toEqual([
+        Buffer.from('key1'),
+        Buffer.from('key2'),
+        Buffer.from('key3'),
+      ]);
+    });
+  });
   describe('getOverview', () => {
     it('should get overview and clear errors', async () => {
       expect(summary['processed']).toEqual(0);
       expect(summary['succeed']).toEqual(0);
       expect(summary['failed']).toEqual(0);
       expect(summary['errors']).toEqual([]);
+      expect(summary['keys']).toEqual([]);
 
       summary.addProcessed(1500);
       summary.addSuccess(500);