Caching & Queues

Vezta uses Redis for two purposes: application-level caching of frequently accessed data, and BullMQ job queues for all background processing. Both are backed by the same Redis instance running as a Docker container (vezta-production-redis in production, vezta-staging-redis on staging).

Redis Cache

The RedisCacheService (in src/common/cache/) provides a global caching layer available to all modules. Cache invalidation happens when the underlying data changes -- for example, when a price-sync job writes new prices, the corresponding cache entries are refreshed.

Other Redis-backed concerns:

• Idempotency keys for orders -- idempotency:{key} with 5-min TTL
• Circuit breaker -- circuit-breaker:{exchange} with 5-min auto-reset TTL
• Insider-signals dedup -- alert:dedup:{wallet}:{marketId} with 24h TTL
• Insider-signals daily caps -- alert:count:*
• Endpoint rate limiting -- sliding-window counters via RedisCacheService.increment()
• Geo-block IP cache -- 24h TTL on the DFlow proxy guard

BullMQ Queues

BullMQ handles all background processing through named queues. Each queue is registered per-module via BullModule.registerQueue() in the module's imports array -- not at the application level.

Active Queues

Worker Domain Routing

Each queue is bound to one of six job-bearing worker domains: traders, market, misc, sports, ai, or combos. (The websocket and rtds containers run persistent connectors, not BullMQ queues.) The matching container (e.g. vezta-production-worker-traders) is the only consumer for that domain.

• backgroundOnly([Processor, Scheduler], '<domain>') excludes providers from DI on sibling workers, so BullMQ never registers a competing consumer
• shouldRunInDomain('<domain>') is checked inside onModuleInit for setInterval-based services and WS connectors

The vezta-production-api container runs none of these workers -- it is HTTP + WebSocket only.

Scheduling Pattern

Modules that need repeatable jobs implement OnModuleInit and follow a consistent pattern:

async onModuleInit() {
  // 1. Clean old repeatable jobs to avoid duplicates
  const existing = await this.queue.getRepeatableJobs();
  for (const job of existing) {
    await this.queue.removeRepeatableByKey(job.key);
  }

  // 2. Register new repeatable jobs
  await this.queue.add('full-sync', {}, {
    repeat: { every: 5 * 60 * 1000 },
  });

  await this.queue.add('price-sync', {}, {
    repeat: { every: 60 * 1000 },
  });

  // 3. Optionally trigger an immediate run
  await this.queue.add('full-sync', {}, {
    jobId: 'initial-sync',
  });
}

Some modules (counter-trade, copy-trade) also use native setInterval for simpler polling. Prefer BullMQ for new schedulers -- better reliability, retry handling, and observability.

Processors

Each queue has a corresponding processor class decorated with @Processor('queue-name') that handles job execution:

@Processor('market-sync')
export class MarketSyncProcessor {
  @Process('full-sync')
  async handleFullSync(job: Job) {
    // Fetch markets, normalize, upsert...
  }

  @Process('price-sync')
  async handlePriceSync(job: Job) {
    // Fetch prices, write snapshots...
  }
}

Redis Configuration

Connection Details

Both the cache and BullMQ queues share the same Redis instance. The Redis container has health checks (redis-cli ping, 10s interval, 3 retries); the API container depends on Redis being healthy before starting.