Batch Processing Pipeline

The exchange rate sync is orchestrated through a multi-step BullMQ flow within the batch service. This pipeline fetches, transforms, and persists exchange rates as part of the daily batch processing cycle.

Flow Architecture

The exchange rate update uses a three-tier parent-child BullMQ flow:

┌────────────────────────────┐
│   SaveExchangeRates        │  ← Parent job (Step 3)
│   Persists to database     │
└────────────┬───────────────┘
             │
┌────────────▼───────────────┐
│   ProcessExchangeRates     │  ← Child job (Step 2)
│   Transforms API response  │
└────────────┬───────────────┘
             │
┌────────────▼───────────────┐
│   FetchExchangeRates       │  ← Leaf job (Step 1)
│   Calls OpenExchangeRates  │
└────────────────────────────┘

Jobs execute bottom-up: FetchExchangeRates runs first, passes its result to ProcessExchangeRates, which passes its result to SaveExchangeRates.

Step-by-Step Execution

Step 1: FetchExchangeRates

Trigger: Scheduled cron or manual trigger
Action:  HTTP GET to OpenExchangeRates API
Output:  Raw JSON with base currency and rates

The job calls ExchangeRateService.getExchangeRates() with the configured API key. The raw response containing exchange rates is returned as the job result.

Step 2: ProcessExchangeRates

Input:   Raw API response from Step 1
Action:  Transform rates into database-ready records
Output:  Array of { baseCurrency, targetCurrency, exchangeRate }

The job calls ExchangeRateService.transformExchangeRates() to iterate through each rate in the API response and create individual conversion records with 4-decimal precision.

Configuration passed to this step:

Step 3: SaveExchangeRates

Input:   Transformed exchange rate records from Step 2
Action:  POST each record to subscription-service
Output:  Success confirmation

The job loops through each transformed record and calls SubscriptionService.addExchangeRate() to persist it via the subscription-service’s /currency-conversions REST endpoint.

Flow Definition

The flow is defined in batch-service/src/flows/update-exchange-rates/exchange-rate-flow.ts:

{
  name: 'SaveExchangeRates',
  queueName: 'dailyBatchQueue',
  data: { databaseTable: 'ExchangeRates' },
  opts: { attempts: 1, backoff: { type: 'fixed', delay: 5000 } },
  children: [
    {
      name: 'ProcessExchangeRates',
      data: {
        sourceCurrency: 'USD',
        targetCurrencies: ['EUR', 'INR', 'AUD', 'CAD']
      },
      children: [
        {
          name: 'FetchExchangeRates',
          data: {}
        }
      ]
    }
  ]
}

Batch Orchestration

The DailyBatchService manages the overall batch lifecycle:

Infrastructure

Completion Tracking

After the exchange rate flow completes, an UpdateExchangeRateCompleted job records the batch slot number, allowing operators to verify that each scheduled run completed successfully.