← back to Nationalrealestate
Add hourly self-balancing ingest loop + Google Places discovery seed (free-tier capped)
8b3108396ead3dc8b39296359c6678492a512a3c · 2026-07-25 13:27:30 -0700 · Steve Abrams
Files touched
A db/migrations/008_places_seed.sqlM package.jsonA src/ingest/places/seed.tsA src/jobs/hourly_loop.ts
Diff
commit 8b3108396ead3dc8b39296359c6678492a512a3c
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Jul 25 13:27:30 2026 -0700
Add hourly self-balancing ingest loop + Google Places discovery seed (free-tier capped)
---
db/migrations/008_places_seed.sql | 29 ++++++
package.json | 6 +-
src/ingest/places/seed.ts | 195 ++++++++++++++++++++++++++++++++++++++
src/jobs/hourly_loop.ts | 90 ++++++++++++++++++
4 files changed, 318 insertions(+), 2 deletions(-)
diff --git a/db/migrations/008_places_seed.sql b/db/migrations/008_places_seed.sql
new file mode 100644
index 0000000..4c7f4c3
--- /dev/null
+++ b/db/migrations/008_places_seed.sql
@@ -0,0 +1,29 @@
+-- M-P1: Google Places DISCOVERY seed + free-tier hard-cap counter.
+-- No BEGIN/COMMIT here — migrate.ts wraps each file in a transaction.
+--
+-- ToS posture (Google Maps Platform §3.2.3): place_id is the ONLY Places field
+-- we persist. Everything durable (firm name/website/phone) is (re)derived from
+-- the firm's OWN public site by the existing crawl pipeline, so Places is a
+-- pointer ("where to look next"), never our stored dataset. place_id doubles as
+-- the idempotent dedupe key so a seeded office is never re-seeded.
+
+CREATE TABLE places_seed (
+ place_id TEXT PRIMARY KEY, -- Google's stable id — the only cacheable field
+ query TEXT, -- the search text that surfaced it (audit)
+ region_id INT REFERENCES region(id), -- metro/county we were seeding
+ website_host TEXT, -- host only, for dedupe vs firm_site (not "Places Content")
+ firm_id INT REFERENCES firm(id), -- the firm row this seed produced/linked (nullable until crawl)
+ status TEXT NOT NULL DEFAULT 'new', -- new | linked | no_website | crawled | dead
+ first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+ last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE INDEX idx_places_seed_region ON places_seed (region_id);
+CREATE INDEX idx_places_seed_status ON places_seed (status);
+
+-- Monthly billing guard. One row per YYYY-MM; seed job HARD-STOPS when
+-- calls_used >= cap so we never cross the free-tier line into real spend.
+CREATE TABLE places_quota (
+ year_month TEXT PRIMARY KEY, -- 'YYYY-MM'
+ calls_used INT NOT NULL DEFAULT 0,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
diff --git a/package.json b/package.json
index 260e7a1..f01e9c1 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
{
"name": "nationalrealestate",
- "version": "0.2.0",
+ "version": "0.3.0",
"private": true,
"type": "module",
"description": "USRealEstate — national U.S. residential market explorer. Free-data v1 (Redfin Data Center, Zillow Research, Census ACS, FHFA). Internal analyst tool behind basic auth.",
@@ -24,7 +24,9 @@
"ingest:fmr": "tsx src/ingest/hud_fmr.ts",
"ingest:parcels": "tsx src/ingest/parcels/engine.ts",
"ingest:commercial": "tsx src/ingest/commercial/engine.ts",
- "ingest:commercial-briefs": "tsx src/ingest/commercial/briefs.ts"
+ "ingest:commercial-briefs": "tsx src/ingest/commercial/briefs.ts",
+ "ingest:places": "tsx src/ingest/places/seed.ts",
+ "loop": "tsx src/jobs/hourly_loop.ts"
},
"dependencies": {
"better-sqlite3": "^13.0.1",
diff --git a/src/ingest/places/seed.ts b/src/ingest/places/seed.ts
new file mode 100644
index 0000000..df5fd1e
--- /dev/null
+++ b/src/ingest/places/seed.ts
@@ -0,0 +1,195 @@
+/**
+ * M-P1 — Google Places DISCOVERY seed (feeds the free firm pipeline).
+ *
+ * Places is used as a SEED/pointer, not a stored dataset:
+ * pick stalest metro → official Places API (v1 searchText) → for each hit,
+ * persist ONLY place_id (the one cacheable field) + create/link a `firm` row
+ * (source='google_places', source_id=place_id) carrying its website → the
+ * existing discover:firms / crawl:firms jobs then enrich firm_site from the
+ * firm's OWN public site (first-party, storable). Idempotent: place_id dedupe.
+ *
+ * COST GUARD — this is the only metered source in the app. A monthly hard cap
+ * (places_quota) STOPS all calls before we cross the Google free-tier line, so
+ * steady-state spend is $0. Every call is logged to the cost-tracker ledger.
+ * Runs DRY by default; set PLACES_SEED_LIVE=1 to make real calls.
+ *
+ * npm run ingest:places # stalest metro, dry-run unless PLACES_SEED_LIVE=1
+ * npm run ingest:places -- --metro=31080 --live
+ *
+ * Env: GOOGLE_PLACES_API_KEY (required for live), PLACES_MONTHLY_CAP=1500,
+ * PLACES_SEED_LIVE=0, PLACES_QUERIES=<comma templates>.
+ */
+import 'dotenv/config';
+import { appendFileSync, mkdirSync } from 'node:fs';
+import { join } from 'node:path';
+import { homedir } from 'node:os';
+import { pool, query } from '../../../db/pool.ts';
+import { openRun, closeRun } from '../run.ts';
+
+const SOURCE = 'google_places';
+const API = 'https://places.googleapis.com/v1/places:searchText';
+const KEY = process.env.GOOGLE_PLACES_API_KEY || '';
+const LIVE = process.env.PLACES_SEED_LIVE === '1' || process.argv.includes('--live');
+const CAP = Number(process.env.PLACES_MONTHLY_CAP || 1500);
+// Text-Search (Pro SKU) list price if we ever exceeded the free tier — for the ledger only.
+const RATE_PER_CALL = 0.032;
+
+const QUERY_TEMPLATES = (process.env.PLACES_QUERIES ||
+ 'real estate agency in {m};commercial real estate brokerage in {m};apartment leasing office in {m}')
+ .split(';').map(s => s.trim()).filter(Boolean);
+
+const FIELD_MASK = 'places.id,places.displayName,places.websiteUri,places.formattedAddress,places.location';
+
+function ym(): string {
+ // Date.now/new Date() are fine at runtime here (not a workflow script).
+ const d = new Date();
+ return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, '0')}`;
+}
+
+function logCost(calls: number, note: string): void {
+ try {
+ const dir = join(homedir(), '.claude');
+ mkdirSync(dir, { recursive: true });
+ const entry = {
+ ts: new Date().toISOString(), skill: 'usre-places-seed', provider: 'google_places',
+ units: calls, unit: 'searchText_call', rate: RATE_PER_CALL,
+ cost: 0, list_cost_if_billed: +(calls * RATE_PER_CALL).toFixed(4),
+ note: `${note} (free-tier, capped ${CAP}/mo)`,
+ };
+ appendFileSync(join(dir, 'cost-ledger.jsonl'), JSON.stringify(entry) + '\n');
+ } catch { /* ledger is best-effort */ }
+}
+
+async function callsUsed(): Promise<number> {
+ const r = await query<{ calls_used: number }>(
+ `SELECT calls_used FROM places_quota WHERE year_month = $1`, [ym()]);
+ return r.rows[0]?.calls_used ?? 0;
+}
+async function bumpQuota(n: number): Promise<void> {
+ await query(
+ `INSERT INTO places_quota (year_month, calls_used) VALUES ($1, $2)
+ ON CONFLICT (year_month) DO UPDATE
+ SET calls_used = places_quota.calls_used + $2, updated_at = NOW()`,
+ [ym(), n]);
+}
+
+/** Stalest metro: fewest seeds first, then least-recently seeded. */
+async function pickMetro(): Promise<{ id: number; name: string; key: string } | null> {
+ const argIdx = process.argv.findIndex(a => a.startsWith('--metro='));
+ if (argIdx > -1) {
+ const cbsa = process.argv[argIdx].split('=')[1];
+ const r = await query<{ id: number; name: string; canonical_key: string }>(
+ `SELECT id, name, canonical_key FROM region WHERE region_type='metro' AND cbsa_code=$1 LIMIT 1`, [cbsa]);
+ if (r.rows.length) return { id: r.rows[0].id, name: r.rows[0].name, key: r.rows[0].canonical_key };
+ }
+ const r = await query<{ id: number; name: string; canonical_key: string }>(
+ `SELECT r.id, r.name, r.canonical_key
+ FROM region r
+ LEFT JOIN LATERAL (SELECT COUNT(*) c, MAX(last_seen) ls FROM places_seed s WHERE s.region_id = r.id) agg ON true
+ WHERE r.region_type = 'metro'
+ ORDER BY COALESCE(agg.c,0) ASC, agg.ls ASC NULLS FIRST, r.population DESC NULLS LAST
+ LIMIT 1`);
+ return r.rows.length ? { id: r.rows[0].id, name: r.rows[0].name, key: r.rows[0].canonical_key } : null;
+}
+
+/** "Asheville, NC Metro Area" -> "Asheville, NC" for natural Places queries. */
+function placeLabel(name: string): string {
+ return name.replace(/\s+(Metro(politan)?( Statistical)?( Area)?|Micro(politan)?( Area)?)\s*$/i, '').trim();
+}
+
+function hostOf(url: string): string | null {
+ try { return new URL(url).host.replace(/^www\./, '').toLowerCase(); } catch { return null; }
+}
+
+async function searchText(q: string): Promise<Array<{ id: string; name: string; website?: string; addr?: string }>> {
+ const res = await fetch(API, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', 'X-Goog-Api-Key': KEY, 'X-Goog-FieldMask': FIELD_MASK },
+ body: JSON.stringify({ textQuery: q, maxResultCount: 20 }),
+ });
+ if (!res.ok) throw new Error(`places searchText ${res.status}: ${(await res.text()).slice(0, 200)}`);
+ const j: any = await res.json();
+ return (j.places || []).map((p: any) => ({
+ id: p.id, name: p.displayName?.text || p.id, website: p.websiteUri, addr: p.formattedAddress,
+ }));
+}
+
+/** Persist place_id + create/link a firm (source-of-record data comes later from the crawl). */
+async function linkSeed(placeId: string, q: string, regionId: number, name: string, website?: string): Promise<'new' | 'seen'> {
+ const existed = await query(`SELECT 1 FROM places_seed WHERE place_id=$1`, [placeId]);
+ const host = website ? hostOf(website) : null;
+ if (existed.rows.length) {
+ await query(`UPDATE places_seed SET last_seen=NOW() WHERE place_id=$1`, [placeId]);
+ return 'seen';
+ }
+ let firmId: number | null = null;
+ if (website) {
+ const f = await query<{ id: number }>(
+ `INSERT INTO firm (name, normalized_name, website, source, source_id, region_id)
+ VALUES ($1,$2,$3,$4,$5,$6)
+ ON CONFLICT (source, source_id) DO UPDATE SET website = COALESCE(EXCLUDED.website, firm.website)
+ RETURNING id`,
+ [name, name.toLowerCase().replace(/[^a-z0-9]+/g, ' ').trim(), website, SOURCE, placeId, regionId]);
+ firmId = f.rows[0].id;
+ // Seed firm_site directly so crawl:firms enriches from the firm's OWN site
+ // (first-party, storable) — skips the redundant discover:firms web search.
+ // crawl_status left NULL: that IS the crawler's work-queue signal.
+ await query(
+ `INSERT INTO firm_site (firm_id, url, discovery_method)
+ VALUES ($1,$2,'google_places')
+ ON CONFLICT (firm_id) DO NOTHING`,
+ [firmId, website]);
+ }
+ await query(
+ `INSERT INTO places_seed (place_id, query, region_id, website_host, firm_id, status)
+ VALUES ($1,$2,$3,$4,$5,$6)`,
+ [placeId, q, regionId, host, firmId, website ? 'linked' : 'no_website']);
+ return 'new';
+}
+
+async function main() {
+ const metro = await pickMetro();
+ if (!metro) { console.log('[places] no metro regions seeded yet — run seed:regions first'); await pool.end(); return; }
+
+ const used = await callsUsed();
+ const budget = Math.max(0, CAP - used);
+ console.log(`[places] metro=${metro.name} (${metro.key}) quota ${used}/${CAP} used, ${budget} left, live=${LIVE}`);
+
+ if (!LIVE) {
+ console.log('[places] DRY-RUN (set PLACES_SEED_LIVE=1 to make real calls). Would query:');
+ for (const t of QUERY_TEMPLATES) console.log(` • "${t.replace('{m}', placeLabel(metro.name))}"`);
+ console.log(`[places] est cost: $0 (free-tier; ${QUERY_TEMPLATES.length} calls, would be $${(QUERY_TEMPLATES.length * RATE_PER_CALL).toFixed(3)} if billed)`);
+ await pool.end();
+ return;
+ }
+ if (!KEY) { console.error('[places] PLACES_SEED_LIVE=1 but GOOGLE_PLACES_API_KEY unset'); process.exit(2); }
+ if (budget <= 0) { console.log(`[places] monthly cap ${CAP} reached — HARD STOP, $0 spent`); await pool.end(); return; }
+
+ const runId = await openRun(SOURCE, `${metro.key}`);
+ let calls = 0, fresh = 0, seen = 0, linked = 0;
+ try {
+ for (const t of QUERY_TEMPLATES) {
+ if (calls >= budget) { console.log('[places] cap reached mid-run — stopping'); break; }
+ const q = t.replace('{m}', placeLabel(metro.name));
+ const hits = await searchText(q);
+ calls++; await bumpQuota(1); logCost(1, `searchText "${q}"`);
+ for (const h of hits) {
+ const r = await linkSeed(h.id, q, metro.id, h.name, h.website);
+ if (r === 'new') { fresh++; if (h.website) linked++; } else seen++;
+ }
+ console.log(`[places] "${q}" -> ${hits.length} hits`);
+ await new Promise(r => setTimeout(r, 400)); // polite gap
+ }
+ await closeRun(runId, 'ok', { upserted: fresh, skipped: seen,
+ notes: `${calls} calls, ${fresh} new (${linked} w/ website), ${seen} seen; metro=${metro.name}` });
+ console.log(`[places] done: ${calls} calls, ${fresh} new firms (${linked} w/ site), ${seen} already seen. $0 (free-tier).`);
+ } catch (e: any) {
+ await closeRun(runId, 'failed', { notes: e.message });
+ console.error('[places] FAILED:', e.message);
+ await pool.end();
+ process.exit(1);
+ }
+ await pool.end();
+}
+
+main().catch(e => { console.error('[places] FAILED:', e); process.exit(1); });
diff --git a/src/jobs/hourly_loop.ts b/src/jobs/hourly_loop.ts
new file mode 100644
index 0000000..bd58710
--- /dev/null
+++ b/src/jobs/hourly_loop.ts
@@ -0,0 +1,90 @@
+/**
+ * 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';
+
+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[];
+};
+
+// 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'] },
+ { name: 'places-seed',script: 'src/ingest/places/seed.ts', args: [], minIntervalHours: 6, 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'] },
+];
+
+/** 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 scored = await Promise.all(JOBS.map(async j => {
+ const age = await ageHours(j);
+ return { job: j, age, overdueBy: age - j.minIntervalHours };
+ }));
+
+ const eligible = scored
+ .filter(s => s.overdueBy >= 0)
+ .sort((a, b) => (b.overdueBy - a.overdueBy) || (JOBS.indexOf(a.job) - JOBS.indexOf(b.job)));
+
+ console.log(`[loop] ${new Date().toISOString()} eligible=${eligible.length}/${JOBS.length} cap=${MAX_JOBS_PER_TICK}`);
+ for (const s of scored.sort((a, b) => b.overdueBy - a.overdueBy)) {
+ const a = s.age === Infinity ? 'never' : `${s.age.toFixed(1)}h`;
+ console.log(` ${s.overdueBy >= 0 ? '▶' : '·'} ${s.job.name.padEnd(12)} age=${a.padStart(6)} every ${s.job.minIntervalHours}h`);
+ }
+
+ const pick = eligible.slice(0, MAX_JOBS_PER_TICK);
+ if (!pick.length) { console.log('[loop] nothing overdue — idle tick'); 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); });
← 289894d fix(sublease): reject non-finite numerics + malformed dates
·
back to Nationalrealestate
·
auto-save: 2026-07-25T13:36:34 (2 files) — db/migrations/009 2b38f61 →