`dispatcher.js`

58 lines
#!/usr/bin/env node
// Daily dispatcher — the "don't overload" heart. Runs ONLY the Tier-A vendors
// whose assigned day-of-month == today, at a hard concurrency cap. Installed as a
// single daily launchd job; because each vendor carries a distinct day, only ~1-2
// vendors run on any given day. Local-models-only: feed adapters need no model;
// Gemini enrich-ai-tags is NOT invoked here.
//
// Flags: --day N (override today) | --vendor CODE (one) | --all (ignore day) | --dry-run | --cap N
const fs = require('fs');
const path = require('path');
const { runVendor } = require('./run-vendor');
const { pool } = require('./lib/db');

const args = process.argv.slice(2);
const flag = (n, d) => { const i = args.indexOf(n); return i >= 0 ? (args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : true) : d; };
const CAP = Number(flag('--cap', 2));
const DRY = args.includes('--dry-run');
const ALL = args.includes('--all');
const ONE = flag('--vendor', null);
const DAY = Number(flag('--day', new Date().getDate()));

async function runPool(items, cap, fn) {
  const out = []; let i = 0;
  const workers = Array.from({ length: Math.min(cap, items.length) }, async () => {
    while (i < items.length) { const idx = i++; out[idx] = await fn(items[idx]); }
  });
  await Promise.all(workers);
  return out;
}

(async () => {
  const manifest = JSON.parse(fs.readFileSync(path.join(__dirname, 'data', 'manifest.json'), 'utf8'));
  let todo = manifest.tier_a.filter(e => e.enabled);
  if (ONE) todo = todo.filter(e => e.vendor_code === ONE);
  else if (!ALL) todo = todo.filter(e => e.day === DAY);

  const stamp = new Date().toISOString().slice(0, 10);
  const log = path.join(__dirname, 'logs', `dispatch-${stamp}.log`);
  fs.mkdirSync(path.dirname(log), { recursive: true });
  const line = s => { fs.appendFileSync(log, s + '\n'); console.log(s); };

  line(`[${new Date().toISOString()}] dispatch day=${DAY} cap=${CAP} selected=${todo.length}${DRY ? ' (DRY)' : ''}${ALL ? ' (ALL)' : ''}`);
  if (todo.length === 0) { line('  nothing scheduled today — done.'); return; }
  if (DRY) { todo.forEach(e => line(`  would scrape: ${e.vendor} [${e.adapter}] ${e.base_url}`)); return; }

  const results = await runPool(todo, CAP, runVendor);
  let ok = 0, fail = 0;
  for (const r of results) {
    if (r.ok) {
      ok++;
      const s = r.staged && !r.staged.skipped ? ` [+${r.staged.inserted} new, ~${r.staged.updated} upd, ${r.staged.gone} gone]` : (r.staged && r.staged.skipped ? ` [db-skip: ${r.staged.skipped}]` : '');
      line(`  ✓ ${r.vendor}: ${r.count} products (${r.ms}ms)${s}`);
    } else { fail++; line(`  ✗ ${r.vendor}: ${r.skipped ? 'SKIP ' : 'FAIL '}${r.reason}`); }
  }
  line(`[${new Date().toISOString()}] done: ${ok} ok, ${fail} fail/skip`);
  await pool.end();
})().catch(async e => { console.error('FATAL', e); try { await pool.end(); } catch {} process.exit(1); });