`src/jobs/hourly_loop.ts`

121 lines
/**
 * Hourly self-balancing ingest loop (launchd com.steve.usre-hourly, StartInterval 3600).
 *
 * Generalizes broker_rotate's "stalest source wins" idea across every fast-moving
 * stream. Each JOB declares a minIntervalHours; each tick we read the last OK run
 * per job (from ingest_runs) and run the most-overdue eligible job(s) — so no
 * source is ever hammered, coverage self-balances, and a new adapter joins the
 * rotation the moment it registers a run. Jobs spawn as child processes (each
 * script owns its own pool.end() lifecycle), mirroring refresh_all/broker_rotate.
 *
 * Politeness > freshness: MAX_JOBS_PER_TICK (default 2) caps outbound bursts.
 * The slow macro-data (zillow/redfin/acs/fhfa) stays on the MONTHLY refresh_all
 * orchestrator — it does not belong here (those feeds only change monthly).
 *
 *   npm run loop                 # one tick
 *   npm run loop -- --dry        # show the plan, spawn nothing
 */
import { execFileSync } from 'node:child_process';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { query, pool } from '../../db/pool.ts';
import { assess } from './self_correct.ts';

const __dirname = dirname(fileURLToPath(import.meta.url));
const ROOT = join(__dirname, '..', '..');
const DRY = process.argv.includes('--dry');
const MAX_JOBS_PER_TICK = Number(process.env.USRE_MAX_JOBS_PER_TICK || 2);

type Job = {
  name: string;
  script: string;
  args: string[];
  minIntervalHours: number;
  // ingest_runs.source value(s) that mark this job as having run; freshest wins.
  runSources: string[];
  // true only for FIXED-workload jobs — enables the silent-break (degrade) check.
  // Paging/rotating jobs have variable per-run counts, so leave false for them.
  stableVolume?: boolean;
};

// Order = tie-break priority when equally overdue (listings freshest first).
const JOBS: Job[] = [
  { name: 'listings',   script: 'src/ingest/listings/engine.ts',    args: ['all'], minIntervalHours: 2,  runSources: ['coldwellbanker', 'realtytexas'], stableVolume: true },
  { name: 'places-seed',script: 'src/ingest/places/seed.ts',        args: [],      minIntervalHours: 2,  runSources: ['google_places'] },
  { name: 'firm-crawl', script: 'src/crawl/firm_front_page.ts',     args: [],      minIntervalHours: 4,  runSources: ['firm_crawl'] },
  { name: 'discover',   script: 'src/enrich/firm_website_discovery.ts', args: [],  minIntervalHours: 4,  runSources: ['firm_discovery'] },
  { name: 'brokers',    script: 'src/jobs/broker_rotate.ts',        args: [],      minIntervalHours: 6,  runSources: ['%_dre', '%_trec', '%_dos', '%_dbpr', '%_idfpr', '%_dcp', '%_dpr'] },
  { name: 'commercial', script: 'src/ingest/commercial/engine.ts',  args: ['la'], minIntervalHours: 24, runSources: ['la_assessor'], stableVolume: true },
  { name: 'parcels-sd', script: 'src/ingest/parcels/engine.ts',     args: ['sandiego-county'], minIntervalHours: 1, runSources: ['sandiego_ca'] },
  // Free priced-DEED feeds (west-coast) — each paginates its county set over time.
  { name: 'deeds-or',   script: 'src/ingest/parcels/engine.ts',     args: ['oregon-rlis'], minIntervalHours: 2, runSources: ['oregon_rlis'] },
  { name: 'deeds-spok', script: 'src/ingest/parcels/engine.ts',     args: ['spokane'],     minIntervalHours: 3, runSources: ['spokane'] },
  { name: 'deeds-alam', script: 'src/ingest/parcels/engine.ts',     args: ['alameda'],     minIntervalHours: 2, runSources: ['alameda'] },
  { name: 'deeds-desc', script: 'src/ingest/parcels/engine.ts',     args: ['deschutes'],   minIntervalHours: 3, runSources: ['deschutes'] },
  // King County WA — 2.4M full-history priced deeds (grantor+grantee). Heavy 150MB
  // bulk CSV, so weekly cadence (not hourly).
  { name: 'deeds-king', script: 'src/ingest/parcels/engine.ts',     args: ['king'],        minIntervalHours: 168, runSources: ['parcel_king_wa'] },
  // Scout round 2 — more free priced-deed feeds (OR/CA/CO/AZ/WA).
  { name: 'deeds-crook',script: 'src/ingest/parcels/engine.ts',     args: ['crook'],       minIntervalHours: 4, runSources: ['crook'] },
  { name: 'deeds-sonoma',script:'src/ingest/parcels/engine.ts',     args: ['sonoma'],      minIntervalHours: 3, runSources: ['sonoma'] },
  { name: 'deeds-co',   script: 'src/ingest/parcels/engine.ts',     args: ['colorado'],    minIntervalHours: 2, runSources: ['colorado'] },
  { name: 'deeds-maricopa',script:'src/ingest/parcels/engine.ts',   args: ['maricopa'],    minIntervalHours: 2, runSources: ['maricopa'] },
  { name: 'deeds-thur', script: 'src/ingest/parcels/engine.ts',     args: ['thurston'],    minIntervalHours: 4, runSources: ['thurston'] },
  { name: 'deeds-pb',   script: 'src/ingest/parcels/engine.ts',     args: ['palm-beach'],  minIntervalHours: 4, runSources: ['palm_beach'] },
  { name: 'deeds-klam', script: 'src/ingest/parcels/engine.ts',     args: ['klamath'],     minIntervalHours: 4, runSources: ['klamath'] },
  { name: 'deeds-marion',script:'src/ingest/parcels/engine.ts',     args: ['marion'],      minIntervalHours: 3, runSources: ['marion'] },
  { name: 'deeds-yakima',script:'src/ingest/parcels/engine.ts',     args: ['yakima'],      minIntervalHours: 3, runSources: ['yakima'] },
];

/** hours since this job's freshest OK run across its runSources (∞ if never). */
async function ageHours(job: Job): Promise<number> {
  const likeClauses = job.runSources.map((_, i) => `source LIKE $${i + 1}`).join(' OR ');
  const r = await query<{ age: number | null }>(
    `SELECT EXTRACT(EPOCH FROM (NOW() - MAX(started_at)))/3600 AS age
       FROM ingest_runs WHERE status='ok' AND (${likeClauses})`, job.runSources);
  return r.rows[0]?.age == null ? Infinity : Number(r.rows[0].age);
}

async function main() {
  const nowMs = Date.now();
  // Assess health + staleness together. The engine can BLOCK a job (backoff /
  // quarantine cooldown) regardless of how stale it is, and BOOST a broken job
  // so it gets corrected before routine refresh.
  const scored = await Promise.all(JOBS.map(async j => {
    const age = await ageHours(j);
    const health = await assess(j.name, j.runSources, nowMs, j.stableVolume);
    const overdueBy = age - j.minIntervalHours;
    const blocked = health.blockedUntilMs != null && nowMs < health.blockedUntilMs;
    const eligible = !blocked && (overdueBy >= 0 || health.isProbe);
    return { job: j, age, overdueBy, health, blocked, eligible, sortKey: health.priorityBoost + overdueBy };
  }));

  const plan = [...scored].sort((a, b) => b.sortKey - a.sortKey);
  const nEligible = plan.filter(s => s.eligible).length;
  console.log(`[loop] ${new Date().toISOString()}  eligible=${nEligible}/${JOBS.length}  cap=${MAX_JOBS_PER_TICK}`);
  for (const s of plan) {
    const a = s.age === Infinity ? 'never' : `${s.age.toFixed(1)}h`;
    const mark = s.eligible ? (s.health.isProbe ? '◆' : '▶') : (s.blocked ? '⏸' : '·');
    console.log(`   ${mark} ${s.job.name.padEnd(12)} [${s.health.state.padEnd(11)}] age=${a.padStart(6)}  ${s.health.reason}`);
  }

  const pick = plan.filter(s => s.eligible).slice(0, MAX_JOBS_PER_TICK);
  if (!pick.length) { console.log('[loop] nothing eligible — idle tick (all fresh, blocked, or quarantined)'); await pool.end(); return; }

  await pool.end(); // children own their own pools

  for (const { job } of pick) {
    console.log(`\n[loop] ── ${job.name} ──`);
    if (DRY) { console.log(`   (dry) would run: tsx ${job.script} ${job.args.join(' ')}`); continue; }
    try {
      execFileSync('npx', ['tsx', job.script, ...job.args], { cwd: ROOT, stdio: 'inherit', timeout: 50 * 60_000 });
    } catch (e: any) {
      // one dead source must not starve the rest of the tick
      console.error(`[loop] ${job.name} FAILED: ${e.message}`);
    }
  }
  console.log(`\n[loop] tick complete — ran ${pick.length} job(s)`);
}

main().catch(e => { console.error('[loop] FATAL:', e); process.exit(1); });