Implement Batch Processing

Creates high-performance batch API processing infrastructure for handling bulk operations efficiently. Implements job queues with Bull/BullMQ, real-time progress tracking, transaction management, and intelligent error recovery. Supports millions of records with optimal resource utilization.

When to Use

Use this command when:

• Processing thousands or millions of records in bulk operations
• Import/export functionality requires progress feedback
• Long-running operations exceed HTTP timeout limits
• Partial failures need graceful handling and retry logic
• Resource-intensive operations require rate limiting
• Background processing needs monitoring and management
• Data migration or synchronization between systems

Do NOT use this command for:

• Simple CRUD operations on single records
• Real-time operations requiring immediate responses
• Operations that must be synchronous by nature
• Small datasets that fit in memory (<1000 records)

Prerequisites

Before running this command, ensure:

• Redis is available for job queue management
• Database supports transactions or bulk operations
• API rate limits and quotas are understood
• Error handling strategy is defined
• Monitoring infrastructure is in place

Process

Step 1: Analyze Batch Requirements

The command examines your data processing needs:

• Identifies optimal batch sizes based on memory and performance
• Determines transaction boundaries for consistency
• Maps data validation requirements
• Calculates processing time estimates
• Defines retry and failure strategies

Step 2: Implement Job Queue System

Sets up Bull/BullMQ for reliable job processing:

• Queue configuration with concurrency limits
• Worker processes for parallel execution
• Dead letter queues for failed jobs
• Priority queues for urgent operations
• Rate limiting to prevent overload

Step 3: Create Batch API Endpoints

Implements RESTful endpoints for batch operations:

• Job submission with validation
• Status checking and progress monitoring
• Result retrieval with pagination
• Job cancellation and cleanup
• Error log access

Step 4: Implement Processing Logic

Creates efficient batch processing workflows:

• Chunked processing for memory efficiency
• Transaction management for data integrity
• Progress reporting at configurable intervals
• Error aggregation and reporting
• Result caching for retrieval

Step 5: Add Monitoring & Observability

Integrates comprehensive monitoring:

• Job metrics and performance tracking
• Error rate monitoring and alerting
• Queue depth and processing rate
• Resource utilization metrics
• Business-level success metrics

Output Format

The command generates a complete batch processing system:

batch-processing/
├── src/
│   ├── queues/
│   │   ├── batch-queue.js
│   │   ├── workers/
│   │   │   ├── batch-processor.js
│   │   │   └── chunk-worker.js
│   │   └── jobs/
│   │       ├── import-job.js
│   │       └── export-job.js
│   ├── api/
│   │   ├── batch-controller.js
│   │   └── batch-routes.js
│   ├── services/
│   │   ├── batch-service.js
│   │   ├── validation-service.js
│   │   └── transaction-manager.js
│   └── utils/
…

Examples

Example 1: User Import with Validation and Progress

Scenario: Import 100,000 users from CSV with validation and deduplication

Generated Implementation:

// queues/batch-queue.js
import Queue from 'bull';
import Redis from 'ioredis';

const batchQueue = new Queue('batch-processing', {
  redis: {
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT
  },
  defaultJobOptions: {
    removeOnComplete: 100,
    removeOnFail: 500,
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000
    }
  }
…

Example 2: Export with Streaming and Compression

Scenario: Export millions of records with streaming and compression

Generated Streaming Export:

// services/export-service.js
import { Transform } from 'stream';
import zlib from 'zlib';

class ExportService {
  async createExportJob(query, format, options) {
    const job = await batchQueue.add('data-export', {
      query,
      format,
      options
    });

    return job;
  }

  async processExportJob(job) {
    const { query, format, options } = job.data;
…