import { QUEUE_AI_ANALYSIS, QUEUE_DOCUMENT_IMPORT, QUEUE_NOTIFICATION, QUEUE_DOMAIN_EVENTS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, } from './queue.service'; // ═══════════════════════════════════════════════════════════════════════ // Queue Definition Registry // // Centralizes BullMQ worker options and default job options for every queue. // Environment variables override the defaults defined here. // // lockDuration is NOT a business timeout — it is the BullMQ lock renewal // interval. A processor that runs longer than lockDuration will have its // lock automatically extended by the worker, unless the worker crashes // (then stalledInterval detects the stall and maxStalledCount controls // retry). Business timeouts must be enforced inside the processor logic. // ═══════════════════════════════════════════════════════════════════════ function envInt(key: string, fallback: number, min?: number): number { const raw = process.env[key]; if (raw === undefined || raw === '') return fallback; const n = parseInt(raw, 10); if (isNaN(n) || n < 0) return fallback; if (min !== undefined && n < min) return fallback; return n; } export interface QueueWorkerOptions { concurrency: number; lockDuration: number; stalledInterval: number; maxStalledCount: number; } export interface QueueDefaultJobOptions { attempts: number; backoff: { type: 'exponential'; delay: number }; removeOnComplete: { count: number; age: number }; removeOnFail: { count: number; age: number }; } export interface QueueDefinition { name: string; workerOptions: QueueWorkerOptions; defaultJobOptions: QueueDefaultJobOptions; } const DEFAULT_WORKER: QueueWorkerOptions = { concurrency: 1, lockDuration: 30_000, stalledInterval: 30_000, maxStalledCount: 1, }; const DEFAULT_JOB: QueueDefaultJobOptions = { attempts: 3, backoff: { type: 'exponential', delay: 1000 }, removeOnComplete: { count: 1000, age: 24 * 3600 }, removeOnFail: { count: 5000, age: 7 * 24 * 3600 }, }; function buildDef(name: string): QueueDefinition { const prefix = name.toUpperCase().replace(/-/g, '_'); return { name, workerOptions: { concurrency: envInt(`BULL_${prefix}_CONCURRENCY`, DEFAULT_WORKER.concurrency, 1), lockDuration: envInt(`BULL_${prefix}_LOCK_DURATION`, DEFAULT_WORKER.lockDuration, 1), stalledInterval: envInt(`BULL_${prefix}_STALLED_INTERVAL`, DEFAULT_WORKER.stalledInterval, 1), maxStalledCount: envInt(`BULL_${prefix}_MAX_STALLED`, DEFAULT_WORKER.maxStalledCount, 1), }, defaultJobOptions: { attempts: envInt(`BULL_${prefix}_ATTEMPTS`, DEFAULT_JOB.attempts, 1), backoff: { type: 'exponential', delay: envInt(`BULL_${prefix}_BACKOFF_DELAY`, DEFAULT_JOB.backoff.delay), }, removeOnComplete: { count: envInt('BULL_COMPLETED_MAX_COUNT', DEFAULT_JOB.removeOnComplete.count), age: envInt('BULL_COMPLETED_MAX_AGE_SEC', DEFAULT_JOB.removeOnComplete.age), }, removeOnFail: { count: envInt('BULL_FAILED_MAX_COUNT', DEFAULT_JOB.removeOnFail.count), age: envInt('BULL_FAILED_MAX_AGE_SEC', DEFAULT_JOB.removeOnFail.age), }, }, }; } export const QUEUE_DEFINITIONS: Record = { [QUEUE_AI_ANALYSIS]: buildDef(QUEUE_AI_ANALYSIS), [QUEUE_DOCUMENT_IMPORT]: buildDef(QUEUE_DOCUMENT_IMPORT), [QUEUE_NOTIFICATION]: buildDef(QUEUE_NOTIFICATION), [QUEUE_DOMAIN_EVENTS]: buildDef(QUEUE_DOMAIN_EVENTS), [QUEUE_AUDIT_LOG]: buildDef(QUEUE_AUDIT_LOG), [QUEUE_FILE_CLEANUP]: buildDef(QUEUE_FILE_CLEANUP), }; export function queueDef(name: string): QueueDefinition { const def = QUEUE_DEFINITIONS[name]; if (!def) throw new Error(`Unknown queue: ${name}`); return def; } export function workerOptionsFor(name: string): QueueWorkerOptions { return queueDef(name).workerOptions; } export function defaultJobOptionsFor(name: string): QueueDefaultJobOptions { return queueDef(name).defaultJobOptions; }