1import { PgBoss } from 'pg-boss';2import { env, logger } from '@rareindex/shared';34/**5 * Queue abstraction (§140). Implemented with pg-boss (PostgreSQL-backed, schema `pgboss`).6 * Only this file knows pg-boss; handlers receive plain payloads so Redis Streams / Kafka can7 * replace it later.8 */9export const JOBS = {10 crawlRun: 'crawl.run',11 normalizeBatch: 'normalize.batch',12 resolveBatch: 'resolve.batch',13 valuationAsset: 'valuation.asset',14 valuationRebuild: 'valuation.rebuild',15 indicesDaily: 'indices.daily',16 snapshotsDaily: 'snapshots.daily',17 radarScan: 'radar.scan',18 healthCompute: 'health.compute',19 fxSync: 'fx.sync',20 benchmarksSync: 'benchmarks.sync',21 listingsExpire: 'listings.expire',22 accountJobs: 'account.jobs',23} as const;24export type JobName = (typeof JOBS)[keyof typeof JOBS];2526export interface SendOptions {27 /** de-duplicate: only one queued/active job with this key */28 singletonKey?: string;29 /** collapse sends within this many seconds */30 singletonSeconds?: number;31 priority?: number;32 startAfterSeconds?: number;33 retryLimit?: number;34 expireInSeconds?: number;35}3637export interface WorkOptions {38 batchSize?: number;39 concurrency?: number;40 pollingIntervalSeconds?: number;41}4243export interface Queue {44 start(): Promise<void>;45 stop(): Promise<void>;46 send<T extends object>(name: JobName, data: T, opts?: SendOptions): Promise<string | null>;47 schedule(name: JobName, cron: string, data?: object, opts?: { tz?: string }): Promise<void>;48 work<T extends object>(name: JobName, opts: WorkOptions, handler: (data: T, job: { id: string; signal: AbortSignal }) => Promise<unknown>): Promise<void>;49 stats(): Promise<Array<{ name: string; queued: number; active: number; failed: number }>>;50}5152export function createQueue(): Queue {53 const boss = new PgBoss({54 connectionString: env().DATABASE_URL,55 schema: 'pgboss',56 max: 6,57 monitorIntervalSeconds: 60,58 });59 boss.on('error', (err) => logger.error({ err }, 'pg-boss error'));60 const log = logger.child({ component: 'queue' });61 const created = new Set<string>();6263 async function ensure(name: string, opts: { retryLimit?: number; expireInSeconds?: number } = {}) {64 if (created.has(name)) return;65 try {66 await boss.createQueue(name, { retryLimit: opts.retryLimit ?? 3, retryDelay: 30, retryBackoff: true, expireInSeconds: opts.expireInSeconds ?? 3600, deleteAfterSeconds: 3 * 86_400 });67 } catch (err) {68 const msg = err instanceof Error ? err.message : String(err);69 if (!/already exists|duplicate key/i.test(msg)) throw err;70 }71 created.add(name);72 }7374 return {75 async start() {76 await boss.start();77 for (const name of Object.values(JOBS)) await ensure(name);78 log.info('queue started');79 },80 async stop() {81 await boss.stop({ graceful: true, timeout: 20_000 });82 },83 async send(name, data, opts = {}) {84 await ensure(name);85 // pg-boss asserts on explicitly-undefined option keys (e.g. "priority must be an integer"): only pass defined ones.86 const sendOpts: Record<string, unknown> = {};87 if (opts.singletonKey !== undefined) sendOpts.singletonKey = opts.singletonKey;88 if (opts.singletonSeconds !== undefined) sendOpts.singletonSeconds = opts.singletonSeconds;89 if (opts.priority !== undefined) sendOpts.priority = Math.trunc(opts.priority);90 if (opts.startAfterSeconds !== undefined) sendOpts.startAfter = opts.startAfterSeconds;91 if (opts.retryLimit !== undefined) sendOpts.retryLimit = opts.retryLimit;92 if (opts.expireInSeconds !== undefined) sendOpts.expireInSeconds = opts.expireInSeconds;93 return boss.send(name, data, sendOpts);94 },95 async schedule(name, cron, data = {}, opts = {}) {96 await ensure(name);97 await boss.schedule(name, cron, data, { tz: opts.tz ?? 'UTC' });98 },99 async work(name, opts, handler) {100 await ensure(name);101 await boss.work(102 name,103 { batchSize: opts.batchSize ?? 1, localConcurrency: opts.concurrency ?? 1, pollingIntervalSeconds: opts.pollingIntervalSeconds ?? 2 },104 async (jobs) => {105 for (const job of jobs) {106 try {107 await handler(job.data as never, { id: job.id, signal: job.signal });108 } catch (err) {109 log.error({ err, job: name, id: job.id }, 'job failed');110 throw err;111 }112 }113 },114 );115 },116 async stats() {117 const out: Array<{ name: string; queued: number; active: number; failed: number }> = [];118 for (const name of Object.values(JOBS)) {119 const q = await boss.getQueue(name);120 if (q) out.push({ name, queued: Number((q as { queuedCount?: number }).queuedCount ?? 0), active: Number((q as { activeCount?: number }).activeCount ?? 0), failed: Number((q as { failedCount?: number }).failedCount ?? 0) });121 }122 return out;123 },124 };125}126