1import { loadDotenv } from './lib/env.ts';2loadDotenv();3process.env.RI_SERVICE ??= 'rareindex-worker';45import { env, logger } from '@rareindex/shared';6import { createQueue, JOBS } from './lib/queue.ts';7import { closeDb, db } from './lib/db.ts';8import { flushCosts } from './lib/costs.ts';9import { runCrawl } from './crawler/run.ts';10import { scheduleDueCrawls } from './crawler/scheduler.ts';11import { normalizeBatch } from './normalizer/index.ts';12import { pendingCount, resolveBatch } from './entity-resolution/index.ts';13import { assetsNeedingValuation, computePremiums, valueMany } from './valuation/run.ts';14import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';15import { syncFx } from './fx.ts';16import { syncBenchmarks } from './benchmarks.ts';17import { computeHealth } from './health.ts';18import { expireListings } from './listings-expire.ts';1920/**21 * RareIndex worker process (PM2: rareindex-worker). One process runs the queue handlers and the22 * scheduler; scale out by starting more processes against the same database.23 */24export async function startWorker(): Promise<() => Promise<void>> {25 const log = logger.child({ component: 'worker' });26 const queue = createQueue();27 await queue.start();28 const concurrency = env().WORKER_CONCURRENCY;2930 // ---- handlers ----31 await queue.work<{ connectorId: string; mode?: 'incremental' | 'backfill' | 'probe'; limit?: number; trigger?: string }>(JOBS.crawlRun, { concurrency: Math.max(1, Math.floor(concurrency / 2)), pollingIntervalSeconds: 5 }, async (data) => {32 const res = await runCrawl(data.connectorId, { mode: data.mode ?? 'incremental', limit: data.limit, trigger: data.trigger ?? 'schedule' });33 if (res.recordsRaw > 0) await queue.send(JOBS.normalizeBatch, { connectorId: data.connectorId }, { singletonKey: `normalize:${data.connectorId}`, singletonSeconds: 30 });34 });35 await queue.work<{ connectorId?: string }>(JOBS.normalizeBatch, { concurrency: 2, pollingIntervalSeconds: 5 }, async (data) => {36 let total = 0;37 for (let i = 0; i < 40; i++) {38 const r = await normalizeBatch({ connectorId: data.connectorId, limit: 500 });39 total += r.processed;40 if (r.processed < 500) break;41 }42 if (total > 0) await queue.send(JOBS.resolveBatch, {}, { singletonKey: 'resolve', singletonSeconds: 30 });43 });44 await queue.work<Record<string, never>>(JOBS.resolveBatch, { concurrency: 1, pollingIntervalSeconds: 5 }, async () => {45 const touched = new Set<string>();46 for (let i = 0; i < 40; i++) {47 const r = await resolveBatch({ limit: 500 });48 for (const a of r.touchedAssets) touched.add(a);49 if (r.processed < 500) break;50 }51 if ((await pendingCount()) > 0) await queue.send(JOBS.resolveBatch, {}, { singletonKey: 'resolve', singletonSeconds: 30, startAfterSeconds: 10 });52 if (touched.size) await queue.send(JOBS.valuationAsset, { assetIds: [...touched].slice(0, 5000) }, { singletonKey: `value:${Date.now()}` });53 });54 await queue.work<{ assetIds: string[] }>(JOBS.valuationAsset, { concurrency: 1, pollingIntervalSeconds: 5 }, async (data) => {55 await valueMany(data.assetIds, { concurrency: 4 });56 });57 await queue.work<{ all?: boolean; rebuildHistory?: boolean }>(JOBS.valuationRebuild, { concurrency: 1, pollingIntervalSeconds: 30 }, async (data) => {58 await computePremiums();59 const ids = await assetsNeedingValuation({ all: data.all, limit: 200_000 });60 await valueMany(ids, { rebuildHistory: data.rebuildHistory, concurrency: 4 });61 });62 await queue.work(JOBS.indicesDaily, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {63 await runIndices();64 });65 await queue.work(JOBS.snapshotsDaily, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {66 await runCategorySnapshots();67 });68 await queue.work(JOBS.radarScan, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {69 await runRadar();70 });71 await queue.work(JOBS.healthCompute, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {72 await computeHealth();73 });74 await queue.work<{ backfill?: boolean }>(JOBS.fxSync, { concurrency: 1, pollingIntervalSeconds: 30 }, async (data) => {75 await syncFx({ backfill: data.backfill });76 });77 await queue.work(JOBS.benchmarksSync, { concurrency: 1, pollingIntervalSeconds: 60 }, async () => {78 await syncBenchmarks();79 });80 await queue.work(JOBS.listingsExpire, { concurrency: 1, pollingIntervalSeconds: 60 }, async () => {81 await expireListings();82 });83 await queue.work(JOBS.accountJobs, { concurrency: 1, pollingIntervalSeconds: 60 }, async () => {84 try {85 const accountModule = './account/index.ts'; // provided by the account module when present86 const mod = (await import(accountModule)) as { runAccountJobs?: (d: ReturnType<typeof db>) => Promise<unknown> };87 if (mod.runAccountJobs) await mod.runAccountJobs(db());88 } catch (err) {89 if (!/Cannot find module|ERR_MODULE_NOT_FOUND/.test(String(err))) log.error({ err }, 'account jobs failed');90 }91 });9293 // ---- schedules (UTC) ----94 await queue.schedule(JOBS.fxSync, '15 16 * * 1-5', {}); // after ECB 16:00 CET publication95 await queue.schedule(JOBS.benchmarksSync, '30 22 * * *', {});96 await queue.schedule(JOBS.valuationRebuild, '0 3 * * *', { all: false });97 await queue.schedule(JOBS.snapshotsDaily, '20 4 * * *', {});98 await queue.schedule(JOBS.indicesDaily, '40 4 * * *', {});99 await queue.schedule(JOBS.radarScan, '0 5 * * *', {});100 await queue.schedule(JOBS.healthCompute, '*/30 * * * *', {});101 await queue.schedule(JOBS.listingsExpire, '10 * * * *', {});102 await queue.schedule(JOBS.accountJobs, '*/10 * * * *', {});103104 // ---- crawl scheduler loop ----105 const tick = async () => {106 try {107 await scheduleDueCrawls(queue);108 } catch (err) {109 log.error({ err }, 'scheduler tick failed');110 }111 };112 await tick();113 const timer = setInterval(tick, 60_000);114 // hourly incremental valuation for assets with fresh evidence115 const hourly = setInterval(() => void queue.send(JOBS.valuationRebuild, { all: false }, { singletonKey: 'valuation:hourly', singletonSeconds: 3000 }), 3600_000);116 log.info({ concurrency }, 'worker started');117118 return async () => {119 clearInterval(timer);120 clearInterval(hourly);121 await queue.stop();122 await flushCosts();123 await closeDb();124 log.info('worker stopped');125 };126}127128const isMain = process.argv[1] && /workers\/main\.ts$/.test(process.argv[1]);129if (isMain) {130 startWorker()131 .then((stop) => {132 const shutdown = () => void stop().then(() => process.exit(0));133 process.on('SIGINT', shutdown);134 process.on('SIGTERM', shutdown);135 })136 .catch((err) => {137 logger.error({ err }, 'worker failed to start');138 process.exit(1);139 });140}141