/** * BullMQ Redis connection options for the WEBHOOK queue's Worker. * * The webhook PRODUCER (the Queue) is configured on the shared `BullModule.forRootAsync` connection * with `enableOfflineQueue: false`, so `queue.add()` fails fast when Redis is unreachable and the * dispatch path can fall back to direct delivery instead of buffering forever. * * The WORKER must NOT inherit that producer-only fast-fail. Its blocking/internal commands * (moveToActive, lock renewal) need to tolerate a brief reconnect during a Redis blip/failover; * BullMQ explicitly recommends leaving the offline queue enabled for Worker connections. Because * @nestjs/bullmq otherwise builds the Worker from the same shared connection, the WebhookProcessor * overrides its connection with these options — host/port/username/password/timeout identical to the producer * (same env vars, same defaults as configuration.ts), but with the offline queue left at ioredis's * default of `true` so a transient outage no longer throws "Stream isn't writeable" and stalls jobs. */ export interface WorkerConnectionOptions { host: string; port: number; username?: string; password?: string; connectTimeout: number; tls?: RedisOptions['tls']; } export function workerConnectionOptions(): WorkerConnectionOptions { const connection = resolveRedisConnectionOptions(); return { host: connection.host || 'localhost', port: connection.port || 6379, username: connection.username, password: connection.password, connectTimeout: connection.connectTimeout || 5000, ...(connection.tls ? { tls: connection.tls } : {}), }; } /** Default number of webhook deliveries the Worker processes in parallel. */ const DEFAULT_WEBHOOK_WORKER_CONCURRENCY = 10; /** * Webhook Worker concurrency. BullMQ defaults a Worker to 1, which serializes ALL webhook deliveries * process-wide: one slow or timing-out receiver head-of-line-blocks every other session's webhooks * until it finishes (up to WEBHOOK_TIMEOUT + retries). Running several in parallel decouples healthy * receivers from a stuck one. Override via WEBHOOK_WORKER_CONCURRENCY; a non-positive/garbage value * falls back to the default. (Read at module import like workerConnectionOptions above.) */ export function webhookWorkerConcurrency(): number { const parsed = parseInt(process.env.WEBHOOK_WORKER_CONCURRENCY || '', 10); return Number.isInteger(parsed) && parsed > 0 ? parsed : DEFAULT_WEBHOOK_WORKER_CONCURRENCY; } /** Default number of ingress events the Worker processes in parallel. */ const DEFAULT_INGRESS_WORKER_CONCURRENCY = 10; /** * Ingress Worker concurrency. Ordering within a conversation is now guaranteed by the * per-conversation KeyedAsyncLock in the processor, not by a single-worker queue, so raising * concurrency here parallelizes unrelated conversations instead of head-of-line-blocking every * inbound event behind the slowest one. Override via INGRESS_WORKER_CONCURRENCY; a * non-positive/garbage value falls back to the default. */ export function ingressWorkerConcurrency(): number { const parsed = parseInt(process.env.INGRESS_WORKER_CONCURRENCY || '', 10); return Number.isInteger(parsed) && parsed > 0 ? parsed : DEFAULT_INGRESS_WORKER_CONCURRENCY; } import type { RedisOptions } from 'ioredis'; import { resolveRedisConnectionOptions } from '../../config/managed-service-connections';