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 return boss.send(name, data, {86 singletonKey: opts.singletonKey,87 singletonSeconds: opts.singletonSeconds,88 priority: opts.priority,89 startAfter: opts.startAfterSeconds,90 retryLimit: opts.retryLimit,91 expireInSeconds: opts.expireInSeconds,92 });93 },94 async schedule(name, cron, data = {}, opts = {}) {95 await ensure(name);96 await boss.schedule(name, cron, data, { tz: opts.tz ?? 'UTC' });97 },98 async work(name, opts, handler) {99 await ensure(name);100 await boss.work(101 name,102 { batchSize: opts.batchSize ?? 1, localConcurrency: opts.concurrency ?? 1, pollingIntervalSeconds: opts.pollingIntervalSeconds ?? 2 },103 async (jobs) => {104 for (const job of jobs) {105 try {106 await handler(job.data as never, { id: job.id, signal: job.signal });107 } catch (err) {108 log.error({ err, job: name, id: job.id }, 'job failed');109 throw err;110 }111 }112 },113 );114 },115 async stats() {116 const out: Array<{ name: string; queued: number; active: number; failed: number }> = [];117 for (const name of Object.values(JOBS)) {118 const q = await boss.getQueue(name);119 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) });120 }121 return out;122 },123 };124}125