← back to Nationalrealestate
src/jobs/hourly_loop.ts
172 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);
// Fast-consumer streams (homes + businesses + firm enrichment) that the huge
// paginating parcel/deed backfills would otherwise starve out of every slot.
// Reserve >=1 slot/tick for the stalest eligible consumer so they get a
// guaranteed cadence. Env-tunable (0 = old global-stalest behavior).
const CONSUMER = new Set(['listings', 'places-seed', 'firm-crawl', 'discover']);
const CONSUMER_SLOTS = Number(process.env.USRE_CONSUMER_SLOTS || 1);
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'] },
// West Coast first — LA County (2.37M parcels), paginates a page/run like the deed jobs.
{ name: 'parcels-la', script: 'src/ingest/parcels/engine.ts', args: ['la-county'], minIntervalHours: 1, runSources: ['la-county'] },
{ 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'] },
{ name: 'deeds-lane', script: 'src/ingest/parcels/engine.ts', args: ['lane'], minIntervalHours: 4, runSources: ['lane'] },
{ name: 'deeds-jackson',script:'src/ingest/parcels/engine.ts', args: ['jackson'], minIntervalHours: 3, runSources: ['jackson'] },
{ name: 'deeds-josephine',script:'src/ingest/parcels/engine.ts', args: ['josephine'], minIntervalHours: 3, runSources: ['josephine'] }, // TK-16: OR priced-deed (Grants Pass, 30k priced)
{ name: 'deeds-snohomish',script:'src/ingest/parcels/engine.ts', args: ['snohomish'], minIntervalHours: 3, runSources: ['snohomish'] }, // TK-16: WA priced-deed (Everett, 53061, 17k priced)
{ name: 'parcels-orange',script:'src/ingest/parcels/engine.ts', args: ['orange-county'], minIntervalHours: 2, runSources: ['orange_county'] },
{ name: 'parcels-ventura',script:'src/ingest/parcels/engine.ts', args: ['ventura-county'], minIntervalHours: 2, runSources: ['ventura_county'] },
{ name: 'parcels-sb', script: 'src/ingest/parcels/engine.ts', args: ['santa-barbara'], minIntervalHours: 2, runSources: ['santa_barbara'] },
{ name: 'parcels-imperial',script:'src/ingest/parcels/engine.ts', args: ['imperial'], minIntervalHours: 2, runSources: ['imperial'] },
{ name: 'parcels-sbd', script: 'src/ingest/parcels/engine.ts', args: ['san-bernardino'], minIntervalHours: 2, runSources: ['san_bernardino'] },
{ name: 'parcels-sanjoaquin',script:'src/ingest/parcels/engine.ts', args: ['san-joaquin'], minIntervalHours: 2, runSources: ['san_joaquin'] },
{ name: 'parcels-fresno',script:'src/ingest/parcels/engine.ts', args: ['fresno'], minIntervalHours: 2, runSources: ['fresno'] },
{ name: 'parcels-monterey',script:'src/ingest/parcels/engine.ts', args: ['monterey'], minIntervalHours: 2, runSources: ['monterey'] },
{ name: 'parcels-solano',script:'src/ingest/parcels/engine.ts', args: ['solano'], minIntervalHours: 2, runSources: ['solano'] },
{ name: 'parcels-butte', script: 'src/ingest/parcels/engine.ts', args: ['butte'], minIntervalHours: 2, runSources: ['butte'] },
{ name: 'parcels-sacramento',script:'src/ingest/parcels/engine.ts', args: ['sacramento'], minIntervalHours: 2, runSources: ['sacramento'] },
{ name: 'parcels-stanislaus',script:'src/ingest/parcels/engine.ts', args: ['stanislaus'], minIntervalHours: 2, runSources: ['stanislaus'] },
{ name: 'parcels-napa', script: 'src/ingest/parcels/engine.ts', args: ['napa'], minIntervalHours: 2, runSources: ['napa'] },
{ name: 'parcels-contracosta',script:'src/ingest/parcels/engine.ts',args: ['contra-costa'], minIntervalHours: 2, runSources: ['contra_costa'] },
{ name: 'parcels-eldorado',script:'src/ingest/parcels/engine.ts', args: ['el-dorado'], minIntervalHours: 2, runSources: ['el_dorado'] },
{ name: 'parcels-nevada',script:'src/ingest/parcels/engine.ts', args: ['nevada'], minIntervalHours: 2, runSources: ['nevada'] },
{ name: 'parcels-tulare',script:'src/ingest/parcels/engine.ts', args: ['tulare'], minIntervalHours: 2, runSources: ['tulare'] },
// TK-16 pass 4: Whatcom County WA (Bellingham, 53073) — genuine priced-deed feed (WhatcomCo_PropertySales excise sales).
{ name: 'deeds-whatcom', script:'src/ingest/parcels/engine.ts', args: ['whatcom'], minIntervalHours: 3, runSources: ['whatcom'] },
// TK-10777: Whatcom history layers (2021/2020) — history-safe upsert, run weekly (data is static/historical).
{ name: 'deeds-whatcom-2021',script:'src/ingest/parcels/engine.ts', args: ['whatcom-2021'], minIntervalHours: 168, runSources: ['whatcom_2021'] },
{ name: 'deeds-whatcom-2020',script:'src/ingest/parcels/engine.ts', args: ['whatcom-2020'], minIntervalHours: 168, runSources: ['whatcom_2020'] },
// TK-5: full San Diego County (1.09M) via SANDAG — zoning + assessed value; was in engine but missing from rotation.
{ name: 'parcels-sd-sandag', script: 'src/ingest/parcels/engine.ts', args: ['san-diego-sandag'], minIntervalHours: 2, runSources: ['san_diego_sandag'] },
// TK-5: Santa Clara County CA (San Jose, 493k) — situs address; was in engine but missing from rotation.
{ name: 'parcels-sc', script: 'src/ingest/parcels/engine.ts', args: ['santa-clara'], minIntervalHours: 2, runSources: ['santa_clara'] },
];
/** 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}`);
}
// Pool-aware pick: reserve up to CONSUMER_SLOTS for the stalest eligible
// fast-consumer jobs (homes/businesses/firm) so paginating parcel/deed backfills
// can't monopolize every slot; fill the rest with the global stalest. `plan` is
// already sorted stalest/most-boosted first, so first-match = stalest.
const eligible = plan.filter(s => s.eligible);
const pick: typeof eligible = [];
for (const s of eligible) {
if (pick.length >= CONSUMER_SLOTS) break;
if (CONSUMER.has(s.job.name)) pick.push(s);
}
for (const s of eligible) {
if (pick.length >= MAX_JOBS_PER_TICK) break;
if (!pick.includes(s)) pick.push(s);
}
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); });