spb/rareindex

1import 'server-only';2import { PgBoss } from 'pg-boss';3import { logger } from '@rareindex/shared';45/**6 * Thin pg-boss producer used by the admin console to enqueue pipeline jobs. Consumers live in7 * workers/ (agent D). Queue names and payloads are the contract:8 *   crawl.run      { connectorId, mode: 'incremental'|'backfill'|'probe', trigger: 'manual'|'retry'|'schedule', limit? }9 *   pipeline.normalize { connectorId? }10 *   valuation.run  { assetId? }11 */12export const QUEUES = {13  crawlRun: 'crawl.run',14  normalize: 'pipeline.normalize',15  valuation: 'valuation.run',16  indices: 'indices.run',17} as const;1819let boss: PgBoss | null = null;20let starting: Promise<PgBoss> | null = null;2122export async function getBoss(): Promise<PgBoss> {23  if (boss) return boss;24  if (!starting) {25    starting = (async () => {26      const b = new PgBoss({ connectionString: process.env.DATABASE_URL ?? 'postgres://localhost:5432/rareindex', application_name: 'rareindex-web-admin', max: 2, supervise: false, schedule: false });27      b.on('error', (err: Error) => logger.warn({ err: err.message }, 'pg-boss error'));28      await b.start();29      boss = b;30      return b;31    })();32  }33  return starting;34}3536export async function enqueue(queue: string, data: Record<string, unknown>, opts: { singletonKey?: string; priority?: number } = {}): Promise<string | null> {37  const b = await getBoss();38  await b.createQueue(queue).catch(() => {});39  return b.send(queue, data, { ...(opts.singletonKey ? { singletonKey: opts.singletonKey } : {}), ...(opts.priority !== undefined ? { priority: opts.priority } : {}), retryLimit: 2, expireInSeconds: 3600 });40}4142export async function queueDepths(): Promise<Array<{ name: string; queued: number; active: number; failed: number }>> {43  try {44    const b = await getBoss();45    const out: Array<{ name: string; queued: number; active: number; failed: number }> = [];46    for (const name of Object.values(QUEUES)) {47      const q = await b.getQueue(name).catch(() => null);48      if (!q) continue;49      const stats = (await (b as unknown as { getQueueStats?: (n: string) => Promise<{ queuedCount?: number; activeCount?: number; failedCount?: number }> }).getQueueStats?.(name).catch(() => null)) ?? null;50      out.push({ name, queued: stats?.queuedCount ?? 0, active: stats?.activeCount ?? 0, failed: stats?.failedCount ?? 0 });51    }52    return out;53  } catch (err) {54    logger.warn({ err: err instanceof Error ? err.message : String(err) }, 'queue depth unavailable');55    return [];56  }57}58