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