feat: #502 Batch Processing for Portfolio Snapshots

apps/backend/PORTFOLIO_SNAPSHOT_IMPLEMENTATION.md

@@ -26,8 +26,9 @@ Implemented periodic portfolio snapshots to track historical performance over ti
 #### PortfolioService
 - `createSnapshot(userId)`: Creates a snapshot for a specific user
 - `getPortfolioHistory(userId, page, limit)`: Retrieves paginated snapshot history
-- `createSnapshotsForAllUsers()`: Scheduled job that runs daily at midnight
-- `triggerSnapshotCreation()`: Manual trigger for testing/admin use
+- `createSnapshotsForAllUsers()`: Scheduled job that queues a batch for all users
+- `triggerSnapshotCreation()`: Manual trigger that queues a batch and returns progress metadata
+- `getSnapshotBatchStatus(batchId)`: Returns progress for a queued batch
 
 ### 3. API Endpoints
 
@@ -85,17 +86,43 @@ Admin endpoint to trigger snapshot creation for all users.
 **Response:**
 ```json
 {
-  "message": "Snapshot creation triggered",
-  "success": 15,
-  "failed": 0
+  "message": "Snapshot creation queued",
+  "batchId": "uuid",
+  "status": "queued",
+  "total": 0,
+  "completed": 0,
+  "failed": 0,
+  "progressPercent": 0
+}
+```
+
+#### GET /portfolio/snapshots/status
+Query snapshot batch progress for a previously queued job.
+
+**Query Parameters:**
+- `batchId` (required): Batch job identifier
+
+**Response:**
+```json
+{
+  "batchId": "uuid",
+  "status": "running",
+  "total": 1250,
+  "completed": 600,
+  "failed": 12,
+  "progressPercent": 48,
+  "triggeredBy": "manual",
+  "requestedAt": "2026-03-29T10:30:00Z",
+  "startedAt": "2026-03-29T10:31:10Z",
+  "finishedAt": null
 }
 ```
 
 ### 4. Scheduled Jobs
 - **Daily Snapshots**: Runs every day at midnight (00:00)
 - Uses `@nestjs/schedule` with cron expressions
-- Automatically creates snapshots for all users
-- Logs success/failure counts
+- Queues a BullMQ batch job for all users
+- Individual snapshot jobs are processed in parallel with retries
 
 ### 5. Data Flow
 
@@ -127,6 +154,16 @@ Migration file: `1769600000000-CreatePortfolioSnapshot.ts`
 ## Dependencies Added
 
 - `@nestjs/schedule`: For cron job scheduling
+- `bullmq`: Job queue for batch processing
+- `ioredis`: Redis client for BullMQ and progress tracking
+
+## Configuration
+
+Optional environment variables (defaults shown):
+- `PORTFOLIO_SNAPSHOT_CONCURRENCY=25`
+- `PORTFOLIO_SNAPSHOT_BATCH_SIZE=500`
+- `PORTFOLIO_SNAPSHOT_ATTEMPTS=3`
+- `PORTFOLIO_SNAPSHOT_RETRY_DELAY_MS=5000`
 
 ## Testing
 

apps/backend/package.json

@@ -53,8 +53,9 @@
     "class-validator": "^0.14.3",
     "dotenv": "^17.2.3",
     "helmet": "^8.1.0",
-    "multer": "^2.1.1",
+    "ioredis": "^5.6.1",
     "keyv": "^5.6.0",
+    "multer": "^2.1.1",
     "passport": "^0.7.0",
     "passport-jwt": "^4.0.1",
     "pg": "^8.17.2",

apps/backend/src/common/rate-limit/rate-limit.guard.ts

@@ -1,8 +1,4 @@
-import {
-  ExecutionContext,
-  HttpException,
-  Injectable,
-} from '@nestjs/common';
+import { ExecutionContext, HttpException, Injectable } from '@nestjs/common';
 import { ThrottlerGuard, ThrottlerLimitDetail } from '@nestjs/throttler';
 import { ErrorCode } from '../enums/error-code.enum';
 
@@ -15,14 +11,17 @@ export class RateLimitGuard extends ThrottlerGuard {
     void context;
     await Promise.resolve();
 
-    throw new HttpException({
-      code: ErrorCode.SYS_RATE_LIMIT_EXCEEDED,
-      message: 'Too many requests. Please try again later.',
-      details: {
-        limit: throttlerLimitDetail.limit,
-        ttlSeconds: throttlerLimitDetail.ttl / 1000,
-        retryAfterSeconds: throttlerLimitDetail.timeToBlockExpire,
+    throw new HttpException(
+      {
+        code: ErrorCode.SYS_RATE_LIMIT_EXCEEDED,
+        message: 'Too many requests. Please try again later.',
+        details: {
+          limit: throttlerLimitDetail.limit,
+          ttlSeconds: throttlerLimitDetail.ttl / 1000,
+          retryAfterSeconds: throttlerLimitDetail.timeToBlockExpire,
+        },
       },
-    }, 429);
+      429,
+    );
   }
 }

apps/backend/src/common/rate-limit/rate-limit.storage.ts

@@ -94,7 +94,10 @@ export class RateLimitStorageService
     await this.store.set(key, entry, ttlMs);
   }
 
-  private toRecord(entry: RateLimitEntry, now: number): AppRateLimitStorageRecord {
+  private toRecord(
+    entry: RateLimitEntry,
+    now: number,
+  ): AppRateLimitStorageRecord {
     const timeToExpire = Math.max(Math.ceil((entry.expiresAt - now) / 1000), 0);
     const timeToBlockExpire = Math.max(
       Math.ceil((entry.blockedUntil - now) / 1000),

apps/backend/src/filters/global-exception.filter.ts

@@ -131,10 +131,7 @@ export class GlobalExceptionFilter implements ExceptionFilter {
       return responseBody.details as ErrorResponse['details'];
     }
 
-    if (
-      status === 400 &&
-      Array.isArray(responseBody.message)
-    ) {
+    if (status === 400 && Array.isArray(responseBody.message)) {
       return (responseBody.message as string[]).map((message) => ({ message }));
     }
 

apps/backend/src/portfolio/dto/portfolio-snapshot.dto.ts

@@ -119,3 +119,122 @@ export class PortfolioHistoryResponseDto {
   @ApiProperty({ description: 'Total number of pages', example: 15 })
   totalPages: number;
 }
+
+export class PortfolioSnapshotBatchStatusDto {
+  @ApiProperty({
+    description: 'Batch identifier for the snapshot job',
+    example: '9b3b4a07-5b35-4f8c-9f26-8f3ac77e5b41',
+  })
+  batchId: string;
+
+  @ApiProperty({
+    description: 'Current status of the snapshot batch job',
+    example: 'running',
+  })
+  status:
+    | 'queued'
+    | 'running'
+    | 'completed'
+    | 'completed_with_errors'
+    | 'failed';
+
+  @ApiProperty({
+    description: 'Total users scheduled for snapshot generation',
+    example: 1250,
+  })
+  total: number;
+
+  @ApiProperty({
+    description: 'Number of snapshots completed successfully',
+    example: 600,
+  })
+  completed: number;
+
+  @ApiProperty({
+    description: 'Number of snapshots that failed after retries',
+    example: 12,
+  })
+  failed: number;
+
+  @ApiProperty({
+    description: 'Completion percentage for the batch',
+    example: 48,
+  })
+  progressPercent: number;
+
+  @ApiProperty({
+    description: 'Trigger source',
+    example: 'manual',
+  })
+  triggeredBy: 'cron' | 'manual' | 'unknown';
+
+  @ApiProperty({
+    description: 'Timestamp when the batch was requested',
+    example: '2026-03-29T10:30:00Z',
+    nullable: true,
+  })
+  requestedAt: string | null;
+
+  @ApiProperty({
+    description: 'Timestamp when processing started',
+    example: '2026-03-29T10:31:10Z',
+    nullable: true,
+  })
+  startedAt: string | null;
+
+  @ApiProperty({
+    description: 'Timestamp when processing finished',
+    example: '2026-03-29T10:42:10Z',
+    nullable: true,
+  })
+  finishedAt: string | null;
+}
+
+export class TriggerSnapshotBatchResponseDto {
+  @ApiProperty({
+    description: 'Message confirming the batch was queued',
+    example: 'Snapshot creation queued',
+  })
+  message: string;
+
+  @ApiProperty({
+    description: 'Batch identifier for tracking progress',
+    example: '9b3b4a07-5b35-4f8c-9f26-8f3ac77e5b41',
+  })
+  batchId: string;
+
+  @ApiProperty({
+    description: 'Current status of the batch job',
+    example: 'queued',
+  })
+  status:
+    | 'queued'
+    | 'running'
+    | 'completed'
+    | 'completed_with_errors'
+    | 'failed';
+
+  @ApiProperty({
+    description: 'Total users scheduled for snapshot generation',
+    example: 1250,
+  })
+  total: number;
+
+  @ApiProperty({
+    description: 'Number of snapshots completed successfully',
+    example: 0,
+  })
+  completed: number;
+
+  @ApiProperty({
+    description: 'Number of snapshots that failed after retries',
+    example: 0,
+  })
+  failed: number;
+
+  @ApiProperty({
+    description: 'Completion percentage for the batch',
+    example: 0,
+  })
+  progressPercent: number;
+}

apps/backend/src/portfolio/portfolio.controller.ts

@@ -21,6 +21,8 @@ import { JwtAuthGuard } from '../auth/jwt-auth.guard';
 import {
   GetPortfolioHistoryDto,
   PortfolioHistoryResponseDto,
+  PortfolioSnapshotBatchStatusDto,
+  TriggerSnapshotBatchResponseDto,
 } from './dto/portfolio-snapshot.dto';
 import {
   GetPortfolioSummaryQueryDto,
@@ -153,22 +155,56 @@ export class PortfolioController {
   })
   @ApiResponse({
     status: 200,
-    description: 'Snapshot creation triggered',
-    schema: {
-      properties: {
-        message: { type: 'string', example: 'Snapshot creation triggered' },
-        success: { type: 'number', example: 42 },
-        failed: { type: 'number', example: 0 },
-      },
-    },
+    description: 'Snapshot creation queued',
+    type: TriggerSnapshotBatchResponseDto,
   })
   @ApiResponse({ status: 401, description: 'Unauthorized' })
   async triggerSnapshotCreation() {
     const result = await this.portfolioService.triggerSnapshotCreation();
     return {
-      message: 'Snapshot creation triggered',
-      success: result.success,
+      message: 'Snapshot creation queued',
+      batchId: result.batchId,
+      status: result.status,
+      total: result.total,
+      completed: result.completed,
       failed: result.failed,
+      progressPercent: result.progressPercent,
+    };
+  }
+
+  @Get('snapshots/status')
+  @ApiOperation({
+    summary: 'Get snapshot batch status',
+    description:
+      'Returns progress information for a queued snapshot batch job.',
+  })
+  @ApiQuery({
+    name: 'batchId',
+    required: true,
+    type: String,
+    example: '9b3b4a07-5b35-4f8c-9f26-8f3ac77e5b41',
+  })
+  @ApiResponse({
+    status: 200,
+    description: 'Snapshot batch status retrieved',
+    type: PortfolioSnapshotBatchStatusDto,
+  })
+  @ApiResponse({ status: 401, description: 'Unauthorized' })
+  async getSnapshotBatchStatus(
+    @Query('batchId') batchId: string,
+  ): Promise<PortfolioSnapshotBatchStatusDto> {
+    const status = await this.portfolioService.getSnapshotBatchStatus(batchId);
+    return {
+      batchId: status.batchId,
+      status: status.status,
+      total: status.total,
+      completed: status.completed,
+      failed: status.failed,
+      progressPercent: status.progressPercent,
+      requestedAt: status.requestedAt ?? null,
+      startedAt: status.startedAt ?? null,
+      finishedAt: status.finishedAt ?? null,
+      triggeredBy: status.triggeredBy ?? 'unknown',
     };
   }