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