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 imagesProcess: 'images.process',24 certsVerify: 'certs.verify',25} as const;26export type JobName = (typeof JOBS)[keyof typeof JOBS];2728export interface SendOptions {29 /** de-duplicate: only one queued/active job with this key */30 singletonKey?: string;31 /** collapse sends within this many seconds */32 singletonSeconds?: number;33 priority?: number;34 startAfterSeconds?: number;35 retryLimit?: number;36 expireInSeconds?: number;37}3839export interface WorkOptions {40 batchSize?: number;41 concurrency?: number;42 pollingIntervalSeconds?: number;43}4445export interface Queue {46 start(): Promise<void>;47 stop(): Promise<void>;48 send<T extends object>(name: JobName, data: T, opts?: SendOptions): Promise<string | null>;49 schedule(name: JobName, cron: string, data?: object, opts?: { tz?: string }): Promise<void>;50 work<T extends object>(name: JobName, opts: WorkOptions, handler: (data: T, job: { id: string; signal: AbortSignal }) => Promise<unknown>): Promise<void>;51 stats(): Promise<Array<{ name: string; queued: number; active: number; failed: number }>>;52}5354export function createQueue(): Queue {55 const boss = new PgBoss({56 connectionString: env().DATABASE_URL,57 schema: 'pgboss',58 max: 6,59 monitorIntervalSeconds: 60,60 });61 boss.on('error', (err) => logger.error({ err }, 'pg-boss error'));62 const log = logger.child({ component: 'queue' });63 const created = new Set<string>();6465 async function ensure(name: string, opts: { retryLimit?: number; expireInSeconds?: number } = {}) {66 if (created.has(name)) return;67 try {68 await boss.createQueue(name, { retryLimit: opts.retryLimit ?? 3, retryDelay: 30, retryBackoff: true, expireInSeconds: opts.expireInSeconds ?? 3600, deleteAfterSeconds: 3 * 86_400 });69 } catch (err) {70 const msg = err instanceof Error ? err.message : String(err);71 if (!/already exists|duplicate key/i.test(msg)) throw err;72 }73 created.add(name);74 }7576 return {77 async start() {78 await boss.start();79 for (const name of Object.values(JOBS)) await ensure(name);80 log.info('queue started');81 },82 async stop() {83 await boss.stop({ graceful: true, timeout: 20_000 });84 },85 async send(name, data, opts = {}) {86 await ensure(name);87 // pg-boss asserts on explicitly-undefined option keys (e.g. "priority must be an integer"): only pass defined ones.88 const sendOpts: Record<string, unknown> = {};89 if (opts.singletonKey !== undefined) sendOpts.singletonKey = opts.singletonKey;90 if (opts.singletonSeconds !== undefined) sendOpts.singletonSeconds = opts.singletonSeconds;91 if (opts.priority !== undefined) sendOpts.priority = Math.trunc(opts.priority);92 if (opts.startAfterSeconds !== undefined) sendOpts.startAfter = opts.startAfterSeconds;93 if (opts.retryLimit !== undefined) sendOpts.retryLimit = opts.retryLimit;94 if (opts.expireInSeconds !== undefined) sendOpts.expireInSeconds = opts.expireInSeconds;95 return boss.send(name, data, sendOpts);96 },97 async schedule(name, cron, data = {}, opts = {}) {98 await ensure(name);99 await boss.schedule(name, cron, data, { tz: opts.tz ?? 'UTC' });100 },101 async work(name, opts, handler) {102 await ensure(name);103 await boss.work(104 name,105 { batchSize: opts.batchSize ?? 1, localConcurrency: opts.concurrency ?? 1, pollingIntervalSeconds: opts.pollingIntervalSeconds ?? 2 },106 async (jobs) => {107 for (const job of jobs) {108 try {109 await handler(job.data as never, { id: job.id, signal: job.signal });110 } catch (err) {111 log.error({ err, job: name, id: job.id }, 'job failed');112 throw err;113 }114 }115 },116 );117 },118 async stats() {119 const out: Array<{ name: string; queued: number; active: number; failed: number }> = [];120 for (const name of Object.values(JOBS)) {121 const q = await boss.getQueue(name);122 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) });123 }124 return out;125 },126 };127}128