← back to Vendor Recrawl Dispatcher
dispatcher.js
253 lines
#!/usr/bin/env node
'use strict';
/*
* vendor-recrawl-dispatcher — registry-driven nightly recrawl of ~463 active DW vendors.
*
* DTD verdict (committed): ONE nightly launchd job that:
* 1. Reads dw_unified.vendor_registry, selects the vendors DUE tonight.
* 2. Orders by crawl_priority DESC (HIGH>MEDIUM>LOW), then last_product_update
* ASC NULLS FIRST (stalest first).
* 3. Runs each vendor's canonical refresh pipeline (scrape -> write dw_unified
* staging -> price-finder -> CNCP win). NO PUBLISH.
* 4. Stamps last_product_update + next_scheduled_crawl per vendor after each run.
* 5. Batch = 10 real scrapes/night, 6-8h wall-clock cap, per-vendor politeness jitter.
*
* Enrichment (if any) is LOCAL-MODELS-ONLY via exo -> Mac1 Ollama -> skip. No paid APIs.
* READ-ONLY DATA REFRESH into staging. Publish stays gated.
*
* Usage:
* node dispatcher.js --dry-run # print the wave ordering, scrape NOTHING
* node dispatcher.js # run tonight's batch
* node dispatcher.js --batch=12 # override batch size
* node dispatcher.js --max-hours=6 # override wall-clock cap
*/
const { spawn } = require('child_process');
const fs = require('fs');
const cfg = require('./lib/config');
const { pool } = require('./lib/db');
const pipeline = require('./lib/resolve-pipeline');
const enrich = require('./lib/enrich-router');
// ---------- args ----------
const ARGS = process.argv.slice(2);
const DRY = ARGS.includes('--dry-run');
const argVal = (k, d) => { const a = ARGS.find(x => x.startsWith(`--${k}=`)); return a ? a.split('=')[1] : d; };
const BATCH = parseInt(argVal('batch', cfg.BATCH), 10);
const MAX_WALL_MS = argVal('max-hours', null) ? parseFloat(argVal('max-hours')) * 3600e3 : cfg.MAX_WALL_MS;
const PRIORITY_RANK = { HIGH: 0, MEDIUM: 1, LOW: 2 };
const nowISO = () => new Date().toISOString();
const sleep = (ms) => new Promise(r => setTimeout(r, ms));
const jitterMs = () => (cfg.JITTER_MIN_S + Math.random() * (cfg.JITTER_MAX_S - cfg.JITTER_MIN_S)) * 1000;
function log(...a) { console.log(`[${nowISO()}]`, ...a); }
function appendJsonl(file, obj) { try { fs.appendFileSync(file, JSON.stringify(obj) + '\n'); } catch (_) {} }
async function cncp(payload) {
if (DRY) return;
try {
await fetch(cfg.CNCP_URL, {
method: 'POST', headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
} catch (_) { /* CNCP down is non-fatal */ }
}
// ---------- selection ----------
// DUE = active, not excluded, and (never crawled OR next_scheduled_crawl in the past).
// Ordered by priority then stalest-first. Returns the FULL due set; the loop applies
// the resolved-scrape budget + wall-clock cap.
async function selectDue() {
const excl = cfg.EXCLUDE_VENDOR_CODES;
const { rows } = await pool.query(
`SELECT id, vendor_name, vendor_code, crawl_priority, catalog_table, pm2_name,
has_credentials, is_private_label, total_products,
last_product_update, next_scheduled_crawl
FROM vendor_registry
WHERE is_active
AND ($1::text[] IS NULL OR vendor_code <> ALL($1))
AND (next_scheduled_crawl IS NULL OR next_scheduled_crawl <= now())
ORDER BY CASE crawl_priority
WHEN 'HIGH' THEN 0 WHEN 'MEDIUM' THEN 1 WHEN 'LOW' THEN 2 ELSE 3 END,
last_product_update ASC NULLS FIRST,
total_products DESC`,
[excl.length ? excl : null]
);
return rows;
}
// ---------- stamping (dispatcher runtime scheduling write — allowed) ----------
function cadenceDays(priority) { return cfg.CADENCE_DAYS[priority] || cfg.CADENCE_DAYS.DEFAULT; }
async function stampSuccess(v) {
const days = cadenceDays(v.crawl_priority);
await pool.query(
`UPDATE vendor_registry
SET last_product_update = now(),
next_scheduled_crawl = now() + ($2 || ' days')::interval,
updated_at = now()
WHERE id = $1`,
[v.id, String(days)]
);
}
// Unresolved / failed vendors: defer their next check so they don't clog every wave,
// WITHOUT stamping last_product_update (no data was refreshed). Short defer so a newly
// wired scraper gets picked up soon.
async function deferVendor(v, days) {
await pool.query(
`UPDATE vendor_registry
SET next_scheduled_crawl = now() + ($2 || ' days')::interval, updated_at = now()
WHERE id = $1`,
[v.id, String(days)]
);
}
// ---------- run one refresh pipeline ----------
function runScript(cmd, cwd, timeoutMs) {
return new Promise((resolve) => {
const [bin, ...rest] = cmd;
const child = spawn(bin, rest, {
cwd, env: { ...process.env }, stdio: ['ignore', 'pipe', 'pipe'],
});
let out = '', err = '', done = false;
const timer = setTimeout(() => {
if (done) return;
done = true;
try { child.kill('SIGKILL'); } catch (_) {}
resolve({ ok: false, code: -1, timedOut: true, out, err });
}, timeoutMs);
child.stdout.on('data', d => { out += d; });
child.stderr.on('data', d => { err += d; });
child.on('close', (code) => {
if (done) return;
done = true; clearTimeout(timer);
resolve({ ok: code === 0, code, timedOut: false, out, err });
});
child.on('error', (e) => {
if (done) return;
done = true; clearTimeout(timer);
resolve({ ok: false, code: -1, timedOut: false, out, err: String(e) });
});
});
}
// Optional local-only enrichment touch after a successful refresh. If both exo and
// Ollama are down, this SKIPS (never blocks the data refresh, never a paid API).
async function enrichTouch(v) {
const provider = await enrich.resolveProvider();
if (provider === 'none') {
log(` enrich: SKIP (exo + Ollama both down) — structured scrape already wrote. $0 (local)`);
return { provider: 'none', skipped: true };
}
const res = await enrich.chat([
{ role: 'system', content: 'You are a wallcovering catalog assistant. Reply in one short line.' },
{ role: 'user', content: `Refresh sanity-check: vendor "${v.vendor_name}" catalog just recrawled. Reply "OK".` },
]);
log(` enrich: ${res.provider} ${res.skipped ? 'SKIPPED' : 'ok'} — ${res.cost}`);
return res;
}
// ---------- DRY RUN ----------
async function dryRun(due) {
const resolved = [];
const unresolved = [];
for (const v of due) {
const r = pipeline.resolve(v);
(r.kind === 'refresh-script' ? resolved : unresolved).push({ v, r });
}
console.log('\n================ RECRAWL WAVE (DRY RUN — nothing scraped) ================');
console.log(`due vendors: ${due.length} | resolvable pipelines: ${resolved.length} | unresolved (skip+log): ${unresolved.length}`);
console.log(`batch (real scrapes tonight): ${BATCH} wall-clock cap: ${(MAX_WALL_MS/3600e3).toFixed(1)}h excluded: [${cfg.EXCLUDE_VENDOR_CODES.join(', ')}]`);
console.log(`enrichment route: exo(${cfg.EXO_BASE}) -> ollama(${cfg.OLLAMA_MODEL}) -> skip cost: $0 (local)`);
console.log('\n rank pri last_product_update prod pipeline vendor');
console.log(' ---- ------ ------------------------- ------ ----------- ------------------------------');
const show = due.slice(0, 30);
show.forEach((v, i) => {
const r = pipeline.resolve(v);
const lpu = v.last_product_update ? new Date(v.last_product_update).toISOString().slice(0, 19) : 'NULL (never) ';
const tag = r.kind === 'refresh-script' ? 'SCRAPE ' : 'unresolved';
console.log(
` ${String(i + 1).padStart(4)} ${(v.crawl_priority || '').padEnd(6)} ${lpu.padEnd(25)} ${String(v.total_products || 0).padStart(6)} ${tag.padEnd(11)} ${v.vendor_name} [${v.vendor_code}]`
);
});
if (due.length > 30) console.log(` … +${due.length - 30} more due vendors (ordered same way)`);
// Which vendors would actually be scraped tonight (first BATCH resolvable, in order):
const tonight = [];
for (const v of due) {
if (tonight.length >= BATCH) break;
if (pipeline.resolve(v).kind === 'refresh-script') tonight.push(v);
}
console.log(`\n >>> Tonight's ${BATCH}-vendor scrape batch would be (first ${BATCH} resolvable, in wave order):`);
if (!tonight.length) {
console.log(' (none — no resolvable <code>-refresh/ pipelines exist yet besides excluded WallQuest)');
} else {
tonight.forEach((v, i) => console.log(` ${i + 1}. ${v.vendor_name} [${v.vendor_code}] pri=${v.crawl_priority}`));
}
console.log('==========================================================================\n');
}
// ---------- REAL RUN ----------
async function realRun(due) {
const startedAt = Date.now();
await cncp({ project: 'vendor-recrawl', title: 'Nightly recrawl wave started',
summary: `due=${due.length}, batch=${BATCH}, cap=${(MAX_WALL_MS/3600e3).toFixed(1)}h` });
let scraped = 0, ok = 0, failed = 0, skipped = 0;
for (const v of due) {
if (scraped >= BATCH) { log(`batch limit ${BATCH} reached — stopping`); break; }
if (Date.now() - startedAt >= MAX_WALL_MS) { log('wall-clock cap reached — stopping'); break; }
const r = pipeline.resolve(v);
if (r.kind !== 'refresh-script') {
skipped++;
appendJsonl(cfg.SKIP_LOG, { ts: nowISO(), vendor_code: v.vendor_code, vendor: v.vendor_name, reason: r.reason });
await deferVendor(v, 2); // re-check in 2 days once a scraper is wired
continue;
}
scraped++;
log(`[${scraped}/${BATCH}] SCRAPE ${v.vendor_name} [${v.vendor_code}] pri=${v.crawl_priority} via ${r.reason}`);
const res = await runScript(r.cmd, r.cwd, cfg.PER_VENDOR_TIMEOUT_MS);
if (res.ok) {
ok++;
await enrichTouch(v).catch(() => {});
await stampSuccess(v);
log(` DONE ${v.vendor_name} — stamped last_product_update + next crawl (+${cadenceDays(v.crawl_priority)}d)`);
} else {
failed++;
appendJsonl(cfg.RUN_LOG, { ts: nowISO(), vendor_code: v.vendor_code, status: res.timedOut ? 'timeout' : 'error', code: res.code, tail: (res.err || res.out).slice(-500) });
await deferVendor(v, 1); // retry tomorrow
log(` FAIL ${v.vendor_name} (${res.timedOut ? 'timeout' : 'exit ' + res.code}) — deferred 1 day`);
}
if (scraped < BATCH && Date.now() - startedAt < MAX_WALL_MS) {
const j = jitterMs();
log(` politeness sleep ${(j/1000).toFixed(0)}s`);
await sleep(j);
}
}
const mins = ((Date.now() - startedAt) / 60000).toFixed(1);
appendJsonl(cfg.RUN_LOG, { ts: nowISO(), event: 'wave-complete', scraped, ok, failed, skipped, minutes: Number(mins) });
await cncp({ project: 'vendor-recrawl', title: 'Nightly recrawl wave complete',
summary: `scraped=${scraped} ok=${ok} failed=${failed} unresolved-skipped=${skipped} in ${mins}m`,
valueToday: 'fresh vendor data in dw_unified staging (no publish)' });
log(`WAVE COMPLETE — scraped=${scraped} ok=${ok} failed=${failed} unresolved=${skipped} (${mins}m). cost: $0 (local)`);
}
// ---------- main ----------
(async () => {
try {
const due = await selectDue();
if (DRY) await dryRun(due);
else await realRun(due);
} catch (e) {
console.error('FATAL', e);
process.exitCode = 1;
} finally {
await pool.end().catch(() => {});
}
})();