← back to Vendor Recrawl Dispatcher
vendor-recrawl-dispatcher: registry-driven wave framework + enrich-router serve-check fix (exo daemon-up != serving); dry-run verified 462 due vendors ordered HIGH-first
2c413c7f1b3e4bf6e373b5e70fd0b0bb6da66f3f · 2026-07-06 21:58:09 -0700 · steve
Files touched
A .gitignoreA dispatcher.jsA lib/config.jsA lib/db.jsA lib/enrich-router.jsA lib/resolve-pipeline.jsA package-lock.jsonA package.jsonA run.shA sql/seed-crawl-schedule.sql
Diff
commit 2c413c7f1b3e4bf6e373b5e70fd0b0bb6da66f3f
Author: steve <steve@designerwallcoverings.com>
Date: Mon Jul 6 21:58:09 2026 -0700
vendor-recrawl-dispatcher: registry-driven wave framework + enrich-router serve-check fix (exo daemon-up != serving); dry-run verified 462 due vendors ordered HIGH-first
---
.gitignore | 5 +
dispatcher.js | 252 ++++++++++++++++++++++++++++++++++++++++++++
lib/config.js | 57 ++++++++++
lib/db.js | 27 +++++
lib/enrich-router.js | 109 +++++++++++++++++++
lib/resolve-pipeline.js | 49 +++++++++
package-lock.json | 162 ++++++++++++++++++++++++++++
package.json | 16 +++
run.sh | 20 ++++
sql/seed-crawl-schedule.sql | 54 ++++++++++
10 files changed, 751 insertions(+)
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..8699ee1
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,5 @@
+node_modules/
+.env*
+*.log
+tmp/
+.DS_Store
diff --git a/dispatcher.js b/dispatcher.js
new file mode 100644
index 0000000..d321097
--- /dev/null
+++ b/dispatcher.js
@@ -0,0 +1,252 @@
+#!/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(() => {});
+ }
+})();
diff --git a/lib/config.js b/lib/config.js
new file mode 100644
index 0000000..ec1fdb2
--- /dev/null
+++ b/lib/config.js
@@ -0,0 +1,57 @@
+'use strict';
+// Central tunables for the recrawl dispatcher. Everything overridable via env.
+const path = require('path');
+const os = require('os');
+
+const HOME = os.homedir();
+
+module.exports = {
+ // --- selection / wave ---
+ // Number of REAL scrape attempts (resolved pipelines) per nightly run.
+ // DTD verdict: 10/night, tunable to 12 after a stable week.
+ BATCH: parseInt(process.env.RECRAWL_BATCH || '10', 10),
+
+ // Hard wall-clock cap for the whole nightly run. DTD: 6-8h.
+ MAX_WALL_MS: parseInt(process.env.RECRAWL_MAX_HOURS || '7', 10) * 3600 * 1000,
+
+ // Per-vendor scrape hard timeout (kills a stuck pipeline so it can't eat the cap).
+ PER_VENDOR_TIMEOUT_MS: parseInt(process.env.RECRAWL_VENDOR_TIMEOUT_MIN || '40', 10) * 60 * 1000,
+
+ // Politeness jitter between vendors (seconds). Random in [MIN, MAX].
+ JITTER_MIN_S: parseInt(process.env.RECRAWL_JITTER_MIN_S || '20', 10),
+ JITTER_MAX_S: parseInt(process.env.RECRAWL_JITTER_MAX_S || '90', 10),
+
+ // Re-crawl cadence per priority (days) — how far out next_scheduled_crawl is
+ // pushed AFTER a successful refresh. Stalest-first ordering means a full sweep
+ // of resolvable vendors keeps rolling continuously.
+ CADENCE_DAYS: { HIGH: 7, MEDIUM: 21, LOW: 28, DEFAULT: 21 },
+
+ // Vendors the dispatcher must NOT touch because a dedicated job already owns them.
+ // WallQuest has its own weekly launchd job (com.steve.wallquest-refresh) — skip to
+ // avoid a double-crawl race on the same portal session. (Documented for Steve.)
+ EXCLUDE_VENDOR_CODES: (process.env.RECRAWL_EXCLUDE || 'wallquest').split(',').map(s => s.trim()).filter(Boolean),
+
+ // --- pipeline resolution ---
+ DW_ROOT: path.join(HOME, 'Projects', 'Designer-Wallcoverings'),
+ REFRESH_SCRIPTS_DIR: path.join(HOME, 'Projects', 'Designer-Wallcoverings', 'scripts'),
+
+ // --- enrichment router (LOCAL MODELS ONLY) ---
+ EXO_BASE: process.env.EXO_BASE || 'http://localhost:52415',
+ // exo model id for the serve-check + enrichment. exo's daemon can be UP while unable
+ // to actually run inference (e.g. mlx not yet built — needs full Xcode), so the router
+ // does a REAL completion probe, not just /v1/models. Falls through to Ollama until exo serves.
+ EXO_MODEL: process.env.EXO_MODEL || 'mlx-community/Qwen3-VL-4B-Instruct-4bit',
+ OLLAMA_BASE: process.env.OLLAMA_BASE || 'http://192.168.1.133:11434',
+ OLLAMA_MODEL: process.env.OLLAMA_MODEL || 'qwen3:14b',
+ HEALTH_TIMEOUT_MS: 3500,
+ ENRICH_TIMEOUT_MS: 45000,
+
+ // --- integrations ---
+ CNCP_URL: process.env.CNCP_URL || 'http://127.0.0.1:3333/api/wins',
+ DB_ENV_FILE: process.env.RECRAWL_DB_ENV ||
+ path.join(HOME, 'Projects', 'Designer-Wallcoverings', 'DW-Programming', 'ImportNewSkufromURL', '.env.local'),
+
+ // --- paths ---
+ SKIP_LOG: path.join(__dirname, '..', 'logs', 'unresolved.jsonl'),
+ RUN_LOG: path.join(__dirname, '..', 'logs', 'runs.jsonl'),
+};
diff --git a/lib/db.js b/lib/db.js
new file mode 100644
index 0000000..3d17b22
--- /dev/null
+++ b/lib/db.js
@@ -0,0 +1,27 @@
+'use strict';
+// Postgres pool for the LOCAL dw_unified mirror on mac3.
+// Reads DATABASE_URL from the DW ImportNewSkufromURL/.env.local; falls back to a
+// trust-auth local connection string (dw_admin connects without a password locally).
+const fs = require('fs');
+const { Pool } = require('pg');
+const cfg = require('./config');
+
+function loadDbUrl() {
+ if (process.env.DATABASE_URL) return process.env.DATABASE_URL;
+ try {
+ const txt = fs.readFileSync(cfg.DB_ENV_FILE, 'utf8');
+ const m = txt.match(/^DATABASE_URL=(.+)$/m);
+ if (m) return m[1].trim();
+ } catch (_) { /* fall through */ }
+ // trust-auth fallback (verified: local dw_admin needs no password)
+ return 'postgresql://dw_admin@127.0.0.1:5432/dw_unified';
+}
+
+const pool = new Pool({
+ connectionString: loadDbUrl(),
+ max: 4,
+ connectionTimeoutMillis: 5000,
+ idleTimeoutMillis: 10000,
+});
+
+module.exports = { pool };
diff --git a/lib/enrich-router.js b/lib/enrich-router.js
new file mode 100644
index 0000000..bf3bd98
--- /dev/null
+++ b/lib/enrich-router.js
@@ -0,0 +1,109 @@
+'use strict';
+// LOCAL-MODELS-ONLY enrichment router for the recrawl dispatcher.
+//
+// Order of preference (HARD CONSTRAINT — no paid APIs ever):
+// 1. exo cluster -> http://localhost:52415/v1/chat/completions (OpenAI-compatible)
+// 2. Mac1 Ollama fallback -> http://192.168.1.133:11434/api/chat (qwen3:14b)
+// 3. SKIP enrichment -> if BOTH are down, structured scrape still writes;
+// the data refresh is NEVER blocked on a model being down.
+//
+// Every call reports a "$0 (local)" cost line. Reaching for Gemini/OpenAI/Replicate
+// here is forbidden.
+const cfg = require('./config');
+
+async function fetchWithTimeout(url, opts, ms) {
+ const ctrl = new AbortController();
+ const t = setTimeout(() => ctrl.abort(), ms);
+ try {
+ return await fetch(url, { ...opts, signal: ctrl.signal });
+ } finally {
+ clearTimeout(t);
+ }
+}
+
+// --- health checks -------------------------------------------------------
+// exo's daemon can be UP (/v1/models = 200) while it CANNOT serve inference
+// (mlx not built — needs full Xcode). So "up" means it actually returns non-empty
+// content for a tiny completion, not merely that the API answers. This makes the
+// router correctly fall through to Ollama today and auto-upgrade to exo once it serves.
+async function exoUp() {
+ try {
+ const models = await fetchWithTimeout(`${cfg.EXO_BASE}/v1/models`, {}, cfg.HEALTH_TIMEOUT_MS);
+ if (!models.ok) return false;
+ const r = await fetchWithTimeout(`${cfg.EXO_BASE}/v1/chat/completions`, {
+ method: 'POST', headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ model: cfg.EXO_MODEL, messages: [{ role: 'user', content: 'ok' }], max_tokens: 4, stream: false }),
+ }, cfg.ENRICH_TIMEOUT_MS);
+ if (!r.ok) return false;
+ const j = await r.json();
+ return !!(j?.choices?.[0]?.message?.content || '').trim(); // serves only if non-empty
+ } catch (_) { return false; }
+}
+
+async function ollamaUp() {
+ try {
+ const r = await fetchWithTimeout(`${cfg.OLLAMA_BASE}/api/tags`, {}, cfg.HEALTH_TIMEOUT_MS);
+ if (!r.ok) return false;
+ const j = await r.json();
+ return Array.isArray(j.models) && j.models.some(m => (m.name || '').startsWith(cfg.OLLAMA_MODEL.split(':')[0]));
+ } catch (_) { return false; }
+}
+
+// Resolve which provider to use this run. Cache within a process so we health-check once.
+let _routeCache = null;
+async function resolveProvider() {
+ if (_routeCache) return _routeCache;
+ if (await exoUp()) _routeCache = 'exo';
+ else if (await ollamaUp()) _routeCache = 'ollama';
+ else _routeCache = 'none';
+ return _routeCache;
+}
+
+// --- enrichment call -----------------------------------------------------
+// messages: [{role, content}]. Returns { provider, text, costUSD:0, skipped }.
+async function chat(messages) {
+ const provider = await resolveProvider();
+ const costLine = { costUSD: 0, cost: '$0 (local)' };
+
+ if (provider === 'exo') {
+ try {
+ const r = await fetchWithTimeout(`${cfg.EXO_BASE}/v1/chat/completions`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ model: cfg.EXO_MODEL, messages, stream: false, max_tokens: 512 }),
+ }, cfg.ENRICH_TIMEOUT_MS);
+ const j = await r.json();
+ const text = j?.choices?.[0]?.message?.content || '';
+ if (text.trim()) return { provider: 'exo', text, ...costLine, skipped: false };
+ // exo answered EMPTY (daemon up but not serving) -> fall back to ollama, don't return blank
+ if (await ollamaUp()) return chatOllama(messages, costLine);
+ return { provider: 'none', text: '', ...costLine, skipped: true };
+ } catch (e) {
+ // exo blipped mid-run -> try ollama once before giving up
+ if (await ollamaUp()) return chatOllama(messages, costLine);
+ return { provider: 'none', text: '', ...costLine, skipped: true, error: String(e) };
+ }
+ }
+
+ if (provider === 'ollama') return chatOllama(messages, costLine);
+
+ // both down -> skip (never block the refresh, never reach for a paid API)
+ return { provider: 'none', text: '', ...costLine, skipped: true };
+}
+
+async function chatOllama(messages, costLine) {
+ try {
+ const r = await fetchWithTimeout(`${cfg.OLLAMA_BASE}/api/chat`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ model: cfg.OLLAMA_MODEL, messages, stream: false }),
+ }, cfg.ENRICH_TIMEOUT_MS);
+ const j = await r.json();
+ const text = j?.message?.content || '';
+ return { provider: 'ollama', text, ...costLine, skipped: false };
+ } catch (e) {
+ return { provider: 'none', text: '', ...costLine, skipped: true, error: String(e) };
+ }
+}
+
+module.exports = { exoUp, ollamaUp, resolveProvider, chat };
diff --git a/lib/resolve-pipeline.js b/lib/resolve-pipeline.js
new file mode 100644
index 0000000..82749f0
--- /dev/null
+++ b/lib/resolve-pipeline.js
@@ -0,0 +1,49 @@
+'use strict';
+// Resolve a vendor row -> the canonical refresh pipeline to run.
+//
+// The canonical pattern is scripts/<name>-refresh/run.sh (WallQuest is the
+// reference), which does: scrape -> write dw_unified staging -> price-finder -> CNCP.
+// NO publish.
+//
+// HARD RULE (constraint): if no refresh pipeline is resolvable for a vendor, the
+// dispatcher must SKIP + LOG it, never crash and never publish. As dedicated
+// <code>-refresh/ dirs get built (each vendor has its own *-scraper-manager skill),
+// they are auto-discovered here.
+const fs = require('fs');
+const path = require('path');
+const cfg = require('./config');
+
+// Candidate directory names to probe for a per-vendor refresh script, derived
+// from vendor_code and pm2_name (both dash/underscore variants).
+function candidateNames(v) {
+ const out = new Set();
+ const push = (s) => { if (s) { out.add(s); out.add(s.replace(/_/g, '-')); out.add(s.replace(/-/g, '')); } };
+ push(v.vendor_code);
+ if (v.pm2_name) push(v.pm2_name.replace(/-agent$/, ''));
+ return [...out];
+}
+
+// Returns { kind, cmd, cwd, reason }.
+// kind = 'refresh-script' -> runnable bash run.sh
+// kind = 'unresolved' -> no pipeline; caller SKIPS + logs
+function resolve(v) {
+ for (const name of candidateNames(v)) {
+ const dir = path.join(cfg.REFRESH_SCRIPTS_DIR, `${name}-refresh`);
+ const runSh = path.join(dir, 'run.sh');
+ try {
+ if (fs.existsSync(runSh)) {
+ return { kind: 'refresh-script', cmd: ['/bin/bash', runSh], cwd: dir, reason: `${name}-refresh/run.sh` };
+ }
+ } catch (_) { /* keep probing */ }
+ }
+ return {
+ kind: 'unresolved',
+ cmd: null,
+ cwd: null,
+ reason: `no scripts/<code>-refresh/run.sh for vendor_code=${v.vendor_code}` +
+ (v.pm2_name ? ` (pm2=${v.pm2_name})` : '') +
+ ' — wire this vendor\'s *-scraper-manager into a refresh dir',
+ };
+}
+
+module.exports = { resolve, candidateNames };
diff --git a/package-lock.json b/package-lock.json
new file mode 100644
index 0000000..8ac40ad
--- /dev/null
+++ b/package-lock.json
@@ -0,0 +1,162 @@
+{
+ "name": "vendor-recrawl-dispatcher",
+ "version": "0.1.0",
+ "lockfileVersion": 3,
+ "requires": true,
+ "packages": {
+ "": {
+ "name": "vendor-recrawl-dispatcher",
+ "version": "0.1.0",
+ "license": "UNLICENSED",
+ "dependencies": {
+ "pg": "^8.11.3"
+ }
+ },
+ "node_modules/pg": {
+ "version": "8.22.0",
+ "resolved": "https://registry.npmjs.org/pg/-/pg-8.22.0.tgz",
+ "integrity": "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==",
+ "license": "MIT",
+ "dependencies": {
+ "pg-connection-string": "^2.14.0",
+ "pg-pool": "^3.14.0",
+ "pg-protocol": "^1.15.0",
+ "pg-types": "2.2.0",
+ "pgpass": "1.0.5"
+ },
+ "engines": {
+ "node": ">= 16.0.0"
+ },
+ "optionalDependencies": {
+ "pg-cloudflare": "^1.4.0"
+ },
+ "peerDependencies": {
+ "pg-native": ">=3.0.1"
+ },
+ "peerDependenciesMeta": {
+ "pg-native": {
+ "optional": true
+ }
+ }
+ },
+ "node_modules/pg-cloudflare": {
+ "version": "1.4.0",
+ "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz",
+ "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==",
+ "license": "MIT",
+ "optional": true
+ },
+ "node_modules/pg-connection-string": {
+ "version": "2.14.0",
+ "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz",
+ "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==",
+ "license": "MIT"
+ },
+ "node_modules/pg-int8": {
+ "version": "1.0.1",
+ "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
+ "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==",
+ "license": "ISC",
+ "engines": {
+ "node": ">=4.0.0"
+ }
+ },
+ "node_modules/pg-pool": {
+ "version": "3.14.0",
+ "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz",
+ "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==",
+ "license": "MIT",
+ "peerDependencies": {
+ "pg": ">=8.0"
+ }
+ },
+ "node_modules/pg-protocol": {
+ "version": "1.15.0",
+ "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.15.0.tgz",
+ "integrity": "sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==",
+ "license": "MIT"
+ },
+ "node_modules/pg-types": {
+ "version": "2.2.0",
+ "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",
+ "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==",
+ "license": "MIT",
+ "dependencies": {
+ "pg-int8": "1.0.1",
+ "postgres-array": "~2.0.0",
+ "postgres-bytea": "~1.0.0",
+ "postgres-date": "~1.0.4",
+ "postgres-interval": "^1.1.0"
+ },
+ "engines": {
+ "node": ">=4"
+ }
+ },
+ "node_modules/pgpass": {
+ "version": "1.0.5",
+ "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz",
+ "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==",
+ "license": "MIT",
+ "dependencies": {
+ "split2": "^4.1.0"
+ }
+ },
+ "node_modules/postgres-array": {
+ "version": "2.0.0",
+ "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz",
+ "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==",
+ "license": "MIT",
+ "engines": {
+ "node": ">=4"
+ }
+ },
+ "node_modules/postgres-bytea": {
+ "version": "1.0.1",
+ "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz",
+ "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==",
+ "license": "MIT",
+ "engines": {
+ "node": ">=0.10.0"
+ }
+ },
+ "node_modules/postgres-date": {
+ "version": "1.0.7",
+ "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz",
+ "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==",
+ "license": "MIT",
+ "engines": {
+ "node": ">=0.10.0"
+ }
+ },
+ "node_modules/postgres-interval": {
+ "version": "1.2.0",
+ "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz",
+ "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==",
+ "license": "MIT",
+ "dependencies": {
+ "xtend": "^4.0.0"
+ },
+ "engines": {
+ "node": ">=0.10.0"
+ }
+ },
+ "node_modules/split2": {
+ "version": "4.2.0",
+ "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz",
+ "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==",
+ "license": "ISC",
+ "engines": {
+ "node": ">= 10.x"
+ }
+ },
+ "node_modules/xtend": {
+ "version": "4.0.2",
+ "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz",
+ "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==",
+ "license": "MIT",
+ "engines": {
+ "node": ">=0.4"
+ }
+ }
+ }
+}
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..3ce1a6f
--- /dev/null
+++ b/package.json
@@ -0,0 +1,16 @@
+{
+ "name": "vendor-recrawl-dispatcher",
+ "version": "0.1.0",
+ "private": true,
+ "description": "Registry-driven nightly recrawl dispatcher for ~463 active DW vendors. Read-only data refresh into dw_unified staging. NO publish.",
+ "main": "dispatcher.js",
+ "scripts": {
+ "dry-run": "node dispatcher.js --dry-run",
+ "run": "node dispatcher.js"
+ },
+ "dependencies": {
+ "pg": "^8.11.3"
+ },
+ "author": "steve@designerwallcoverings.com",
+ "license": "UNLICENSED"
+}
diff --git a/run.sh b/run.sh
new file mode 100644
index 0000000..aa13c65
--- /dev/null
+++ b/run.sh
@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+# Nightly recrawl dispatcher wrapper (mirrors wallquest-refresh/run.sh conventions).
+# Registry-driven, read-only DATA refresh into dw_unified staging. NO PUBLISH.
+set -uo pipefail
+DIR="$HOME/Projects/vendor-recrawl-dispatcher"
+LOG="$DIR/logs/dispatcher.log"
+TS(){ date +%FT%T%z; }
+mkdir -p "$DIR/logs"
+
+echo "[$(TS)] === vendor-recrawl dispatcher START ===" >> "$LOG"
+# CNCP: mark running
+curl -s -X POST http://127.0.0.1:3333/api/wins -H 'Content-Type: application/json' \
+ -d '{"project":"vendor-recrawl","title":"Nightly recrawl dispatcher started","summary":"registry-driven wave: scrape -> dw_unified staging -> price-finder. NO publish."}' >/dev/null 2>&1 || true
+
+cd "$DIR"
+node dispatcher.js >> "$LOG" 2>&1
+RC=$?
+
+echo "[$(TS)] === vendor-recrawl dispatcher DONE (rc=$RC) ===" >> "$LOG"
+exit $RC
diff --git a/sql/seed-crawl-schedule.sql b/sql/seed-crawl-schedule.sql
new file mode 100644
index 0000000..188429c
--- /dev/null
+++ b/sql/seed-crawl-schedule.sql
@@ -0,0 +1,54 @@
+-- ============================================================================
+-- DRAFT — ONE-TIME SEED of crawl_cron + staggered next_scheduled_crawl
+-- across all ACTIVE DW vendors, for the rolling nightly recrawl wave.
+--
+-- *** GATED: writes canonical vendor_registry scheduling rows. Steve runs this. ***
+-- psql -d dw_unified -f sql/seed-crawl-schedule.sql
+--
+-- What it does:
+-- * Orders active vendors by priority (HIGH>MEDIUM>LOW) then stalest-first
+-- (last_product_update ASC NULLS FIRST), then largest catalog first.
+-- * Assigns each vendor a "night offset" = floor(rank / :batch), so ~:batch
+-- vendors come DUE per night, spreading the full sweep across the wave.
+-- * Stamps crawl_cron = '0 2 * * *' (participates in the 2am nightly wave; the
+-- dispatcher decides who is actually due via next_scheduled_crawl).
+-- * next_scheduled_crawl = tonight 02:00 + (offset days).
+--
+-- Idempotent: re-running re-stamps the same deterministic schedule. It does NOT
+-- touch last_product_update or any catalog data. Reversible: to undo, set
+-- next_scheduled_crawl = NULL (dispatcher then treats all as due, ordered as above).
+--
+-- Tunable: change the divisor (10) to match the dispatcher batch size.
+-- ============================================================================
+BEGIN;
+
+WITH ordered AS (
+ SELECT id,
+ (row_number() OVER (
+ 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
+ ) - 1) AS rn
+ FROM vendor_registry
+ WHERE is_active
+ AND vendor_code <> 'wallquest' -- owned by com.steve.wallquest-refresh (weekly)
+)
+UPDATE vendor_registry v
+ SET crawl_cron = '0 2 * * *',
+ next_scheduled_crawl = date_trunc('day', now()) + interval '2 hours'
+ + ((o.rn / 10)::int) * interval '1 day',
+ updated_at = now()
+ FROM ordered o
+ WHERE v.id = o.id;
+
+-- WallQuest keeps its own weekly cadence; do not enrol it in the nightly wave.
+UPDATE vendor_registry
+ SET crawl_cron = '0 3 * * 0' -- Sunday 3am, matches its dedicated job
+ WHERE vendor_code = 'wallquest' AND is_active;
+
+-- Report the resulting first week of the wave (sanity view):
+-- SELECT date_trunc('day', next_scheduled_crawl) d, count(*)
+-- FROM vendor_registry WHERE is_active GROUP BY 1 ORDER BY 1 LIMIT 10;
+
+COMMIT;
(oldest)
·
back to Vendor Recrawl Dispatcher
·
add README + launchd plist draft + pending-approval go-live bec8dba →