← back to Commercialrealestate
scripts/serve.js
1231 lines
// serve.js — Express server for the SFV CRE viewer + grounded control-chat.
// Static viewer + /data + the run-notes-chat bridge (free Claude CLI via Max sub).
// Run: node scripts/serve.js -> http://127.0.0.1:9911
const express = require('express');
const path = require('path');
const fs = require('fs');
const mountClaudeChat = require('../claude-chat-bridge.cjs');
const { lookupAddress } = require('./sources/wholivedthere'); // READ-ONLY WhoLivedThere/pastdoor archive
let brokerdb = null; try { brokerdb = require('./db/brokers-db'); } catch (_) {} // broker graph (cre DB)
const ROOT = path.join(__dirname, '..');
// Load the project .env into process.env (Node 20.12+/22+ native, no dotenv dep). Lets prod carry
// GMAIL_BASIC_AUTH_B64/GEORGE_AUTH (for the Send-via-George Gmail draft) + CENSUS_API_KEY without
// baking secrets into pm2. Best-effort: a missing .env just leaves process.env as-is.
try { if (typeof process.loadEnvFile === 'function') process.loadEnvFile(path.join(ROOT, '.env')); } catch (_) {}
// Assessor roll fallback for DB-less prod (Kamatera has no Postgres). The 2.43M-parcel roll is too
// large for a JSON snapshot, so it ships as a read-only SQLite file (data/assessor.sqlite, built by
// scripts/export-assessor-sqlite.js). Opened lazily once — a plain file read, no DB server needed.
// (DTD verdict A, 2026-07-07.) A missing file or missing native module just leaves assessorSqlite null.
let assessorSqlite = null;
try {
const p = path.join(ROOT, 'data', 'assessor.sqlite');
if (fs.existsSync(p)) {
const Database = require('better-sqlite3');
const sdb = new Database(p, { readonly: true, fileMustExist: true });
assessorSqlite = sdb.prepare(
`SELECT property_location, year_built, effective_year_built, sqft_main, bedrooms, bathrooms, units,
use_desc1, use_desc2, roll_total_value, roll_land_value, roll_imp_value, roll_year, recording_date
FROM assessor_parcel WHERE situs_house_no=? AND situs_street LIKE ?
ORDER BY roll_year DESC LIMIT 1`);
}
} catch (_) { assessorSqlite = null; } // native module or file absent -> fallback simply unavailable
const PORT = process.env.PORT || 9911;
const app = express();
app.use(express.json({ limit: '256kb' }));
// ── Whole-site Basic Auth gate (Steve 2026-07-08) ───────────────────────────────────
// crcp.agentabrams.com was fully public; now every page, clean-URL route, /data static dir,
// and /api endpoint sits behind a username/password because this middleware is mounted BEFORE
// all routes. Credentials are env-overridable (CRCP_USER/CRCP_PASS); default is the house DW login.
// /healthz stays OPEN so the deploy smoke-test + uptime canaries get a 200 without credentials.
const AUTH_USER = process.env.CRCP_USER || 'admin';
const AUTH_PASS = process.env.CRCP_PASS || 'DW2024!';
app.get('/healthz', (req, res) => res.type('text').send('ok'));
app.use((req, res, next) => {
const hdr = req.headers.authorization || '';
const [scheme, encoded] = hdr.split(' ');
if (scheme === 'Basic' && encoded) {
const [u, ...rest] = Buffer.from(encoded, 'base64').toString().split(':');
if (u === AUTH_USER && rest.join(':') === AUTH_PASS) return next();
}
res.set('WWW-Authenticate', 'Basic realm="CRCP (crcp.agentabrams.com)", charset="UTF-8"');
return res.status(401).send('Authentication required');
});
// ── Agent-contact CRM (durable, local JSON) ─────────────────────────────────────────
// One record per listing id: editable {name,phone,email}, a `contacted_at` stamp, and an
// append-only `letters[]` log (every letter Steve writes to the agent is SAVED here — sending
// stays a separate Steve-gated action, never fired from this store). Persisted to
// data/agent-contacts.json (same write-to-disk pattern as brokers-snapshot.json). Reversible,
// no prod DB, survives redeploy independent of the scrape.
const CONTACTS_FILE = path.join(ROOT, 'data', 'agent-contacts.json');
function readContacts() {
try { return JSON.parse(fs.readFileSync(CONTACTS_FILE, 'utf8')); } catch (_) { return {}; }
}
function writeContacts(map) {
const tmp = CONTACTS_FILE + '.tmp';
fs.writeFileSync(tmp, JSON.stringify(map, null, 2)); // atomic: write tmp then rename
fs.renameSync(tmp, CONTACTS_FILE);
}
function contactRec(map, key) {
if (!map[key]) map[key] = { name: '', phone: '', email: '', contacted_at: null, letters: [] };
if (!Array.isArray(map[key].letters)) map[key].letters = [];
return map[key];
}
const clip = (s, n) => String(s == null ? '' : s).slice(0, n);
let _letterSeq = 0;
const letterId = () => Date.now().toString(36) + '-' + (++_letterSeq).toString(36); // stable, monotonic
// Serialize every read-modify-write. readContacts→mutate→writeContacts is not atomic across the
// pair, so two interleaved requests would last-writer-win and silently drop data. This promise
// chain makes every mutation run to completion before the next starts. (Contrarian HIGH #1.)
let _cq = Promise.resolve();
function withContacts(mutator) {
const run = _cq.then(() => { const map = readContacts(); const out = mutator(map); writeContacts(map); return out; });
_cq = run.then(() => {}, () => {}); // keep the chain alive even if a mutator throws
return run;
}
const jsonErr = (res) => (e) => res.status(500).json({ error: String(e && e.message || e) });
// Whole store — front-end loads once at boot, keyed by listing id.
app.get('/api/contacts', (req, res) => res.json({ contacts: readContacts() }));
// Save/override the agent's contact info (name/phone/email) for a listing.
app.post('/api/contacts/:key/info', (req, res) => {
const key = clip(req.params.key, 120); if (!key) return res.status(400).json({ error: 'no key' });
const b = req.body || {};
withContacts(map => { const rec = contactRec(map, key);
if (b.name != null) rec.name = clip(b.name, 200);
if (b.phone != null) rec.phone = clip(b.phone, 60);
if (b.email != null) rec.email = clip(b.email, 200);
if (b.firm != null) rec.firm = clip(b.firm, 200);
if (b.address != null) rec.address = clip(b.address, 300);
if (b.notes != null) rec.notes = clip(b.notes, 8000);
rec.updated_at = new Date().toISOString(); return rec;
}).then(rec => res.json({ ok: true, contact: rec })).catch(jsonErr(res));
});
// Stamp (or clear) the contacted date. Body {date} optional (ISO); defaults to now.
app.post('/api/contacts/:key/contacted', (req, res) => {
const key = clip(req.params.key, 120); if (!key) return res.status(400).json({ error: 'no key' });
const b = req.body || {};
withContacts(map => { const rec = contactRec(map, key);
rec.contacted_at = (b.clear ? null : (b.date ? clip(b.date, 40) : new Date().toISOString())); return rec;
}).then(rec => res.json({ ok: true, contact: rec })).catch(jsonErr(res));
});
// Append a saved letter {subject, body, channel}. SAVE only — never sends. Each letter gets a
// stable `id` so it can be addressed without relying on its array position (Contrarian HIGH #2).
app.post('/api/contacts/:key/letter', (req, res) => {
const key = clip(req.params.key, 120); if (!key) return res.status(400).json({ error: 'no key' });
const b = req.body || {}; const body = clip(b.body, 20000);
if (!body.trim() && !clip(b.subject, 400).trim()) return res.status(400).json({ error: 'empty letter' });
withContacts(map => { const rec = contactRec(map, key);
const letter = { id: letterId(), ts: new Date().toISOString(), subject: clip(b.subject, 400), body, channel: clip(b.channel || 'note', 20) };
rec.letters.push(letter);
if (!rec.contacted_at) rec.contacted_at = letter.ts;
return { rec, letter };
}).then(o => res.json({ ok: true, contact: o.rec, letter: o.letter })).catch(jsonErr(res));
});
// Delete a saved letter by stable id (preferred) or legacy array index. Addressing by id removes
// the TOCTOU race where an appended letter shifted positions and the wrong one got deleted.
app.post('/api/contacts/:key/letter/:idx/delete', (req, res) => {
const key = clip(req.params.key, 120); const ref = String(req.params.idx);
withContacts(map => { const rec = contactRec(map, key);
let i = rec.letters.findIndex(l => l.id && l.id === ref);
if (i < 0) { const n = parseInt(ref, 10); if (String(n) === ref && n >= 0 && n < rec.letters.length) i = n; }
if (i >= 0) rec.letters.splice(i, 1);
return rec;
}).then(rec => res.json({ ok: true, contact: rec })).catch(jsonErr(res));
});
// "Send via George" — creates a Gmail DRAFT (never a live send) via the George gmail-agent on
// this same box (127.0.0.1:9850 /api/drafts). GATED THREE WAYS: (1) it only ever creates a DRAFT,
// so nothing leaves until Steve reviews + sends in Gmail; (2) it is FAIL-CLOSED — with no
// GEORGE_AUTH in the CRCP env it returns george_not_configured and does nothing; (3) it never
// calls George's /api/send. From the steve-office account by default.
const GEORGE_URL = process.env.GEORGE_URL || 'http://127.0.0.1:9850';
const GEORGE_AUTH = process.env.GEORGE_AUTH || (process.env.GMAIL_BASIC_AUTH_B64 ? 'Basic ' + process.env.GMAIL_BASIC_AUTH_B64 : '');
app.post('/api/contacts/:key/send-draft', async (req, res) => {
const key = clip(req.params.key, 120); if (!key) return res.status(400).json({ error: 'no key' });
const b = req.body || {};
const to = clip(b.to, 200).trim(); const subject = clip(b.subject, 400); const body = clip(b.body, 20000);
if (!to || !/@/.test(to)) return res.status(400).json({ error: 'no valid agent email — add one first' });
if (!body.trim()) return res.status(400).json({ error: 'empty letter body' });
if (!GEORGE_AUTH) return res.status(503).json({ error: 'george_not_configured', hint: 'set GEORGE_AUTH (or GMAIL_BASIC_AUTH_B64) in the CRCP env to enable Gmail drafting' });
try {
const gr = await fetch(GEORGE_URL + '/api/drafts?account=steve-office', {
method: 'POST', headers: { 'Content-Type': 'application/json', 'Authorization': GEORGE_AUTH },
body: JSON.stringify({ to, subject, body, source: 'crcp-agent-contact' })
});
const gj = await gr.json().catch(() => ({}));
if (!gr.ok || !gj.success) return res.status(502).json({ error: 'george_error', detail: gj.error || ('HTTP ' + gr.status) });
// Record the drafted letter as its own id-stamped entry (no fragile body-matching).
const now = new Date().toISOString();
const rec = await withContacts(map => {
const r = contactRec(map, key);
r.letters.push({ id: letterId(), ts: now, subject, body, channel: 'letter', status: 'gmail_draft', draft_id: gj.draftId, drafted_at: now, to });
if (!r.contacted_at) r.contacted_at = now;
return r;
});
res.json({ ok: true, draftId: gj.draftId, account: gj.account, contact: rec, gmailDrafts: 'https://mail.google.com/mail/u/0/#drafts' });
} catch (e) { res.status(502).json({ error: 'george_unreachable', detail: String(e.message).split('\n')[0] }); }
});
// GATED, cost-surfaced live comps via Browserbase (cloud browser -> LoopNet city/type search).
// ~$0.03/call (a Browserbase session). Shells to scripts/live-comps-bb.js so the cloud-browser
// deps resolve from the browserbase skill's node_modules.
const { execFile } = require('child_process');
app.post('/api/comps', (req, res) => {
const _cb = req.body || {}; const _clip = (v, n, d) => String(v == null ? d : v).slice(0, n); const city = _clip(_cb.city, 80, 'Van Nuys'), type = _clip(_cb.type, 40, 'Multifamily');
const child = execFile('node', [path.join(__dirname, 'live-comps-bb.js')], {
env: { ...process.env, CC_CITY: city, CC_TYPE: type, NODE_PATH: process.env.HOME + '/.claude/skills/browserbase/node_modules' },
timeout: 90000, maxBuffer: 2 * 1024 * 1024
}, (err, stdout) => {
if (err && !stdout) return res.status(502).json({ error: 'live comps failed: ' + String(err.message || err).split('\n')[0] });
let j; try { j = JSON.parse(stdout.trim().split('\n').pop()); } catch { return res.status(502).json({ error: 'bad comps output' }); }
res.json({ comps: j.comps || [], source: j.source, sourceUrl: j.url, error: j.error, cost: '~$0.03 (Browserbase session)' });
});
child.on('error', e => { try { res.status(502).json({ error: String(e.message) }); } catch (_) {} });
});
// WhoLivedThere property-history enrichment (READ-ONLY, local, $0). On-demand per address so we never
// bulk-join the whole grid. Returns assessor parcel + permit history + place events + film history when
// the curated archive has the address; {matched:false} otherwise; {available:false} if the DB is down.
app.get('/api/history', async (req, res) => {
const { address = '', city = '', zip = '' } = req.query || {};
if (!address) return res.status(400).json({ error: 'address required' });
try {
const r = await lookupAddress(String(address), String(city), String(zip));
res.json({ ...r, source: 'WhoLivedThere / pastdoor archive (LA County assessor + permits + film)' });
} catch (e) {
res.status(502).json({ available: false, error: String(e.message).split('\n')[0] });
}
});
// The chat can DRIVE the viewer. This protocol is how it does it.
const SYSTEM_EXTRA = `You are a commercial-real-estate investment analyst embedded in the "San Fernando Valley CRE — $400k Down" ranking viewer. The context block below is the LIVE ranked dataset the user is looking at (every property, its price, units, cap rate, cash-on-cash, DSCR, composite score, rank, recommendation, status, and notes), plus the financing assumptions and the analyst verdict.
GROUNDING RULES:
- Answer ONLY from the data in the context. Never invent properties, prices, cap rates, or returns not present.
- Cap rates marked "projected" are NOT in-place income — always flag that distinction (e.g. the Reseda mixed-use 7302-7304 Canby: ~6% actual vs 7.8% projected).
- Be concise: short paragraphs + bullets. Give numbers.
LIVE CONTROL: the user can ask you to change what the viewer shows. When they want to filter, sort, change the down-payment scenario, cap the price, or jump to a property, append EXACTLY ONE fenced code block at the very END of your reply:
\`\`\`action
{ ...only the keys that should change... }
\`\`\`
Action schema (omit keys that don't change):
- "scenario": "25" | "30" | "cash" (down-payment model)
- "filter": "all" | "multifamily" | "retail" | "mixed" | "affordable" | "active"
- "sort": "composite" | "coc" | "cap" | "dscr" | "priceAsc" | "priceDesc" | "cashNeeded"
- "maxPrice": <number USD> "minPrice": <number USD>
- "highlightId": "<property id from the context>" (scrolls to + flashes that card)
Rules: explain in prose ABOVE the block. Only emit an action block when the user actually wants the view changed — for pure questions, NO action block. Use the exact ids/enum values above.`;
// Broker relationship graph (mind-map) from the cre DB. Read-only.
const snapGraph = () => { const s = readBrokerSnap(); return s && s.graph ? s.graph : { nodes: [], edges: [], stats: {}, unavailable: true }; };
app.get('/api/graph', async (req, res) => {
if (!brokerdb) return res.json(snapGraph());
try { res.json(await brokerdb.graph(+(req.query.limit) || 400)); }
catch (e) { res.json(snapGraph()); }
});
app.get('/api/brokers/top', async (req, res) => {
const snapTop = () => { const s = readBrokerSnap(); return (s && s.top ? s.top : []).slice(0, +(req.query.limit) || 50); };
if (!brokerdb) return res.json(snapTop());
try { res.json(await brokerdb.topBrokers(+(req.query.limit) || 50)); }
catch (e) { res.json(snapTop()); }
});
// Full enriched contact card for one broker (phone/email/website/linkedin/office + per-field
// provenance + listing book). Business-contact research data only (Job B enrichment).
// Snapshot fallback = a basic card from the flat broker row (no listing book / provenance, which
// need the live DB) — enough that the click-through never errors on prod.
const snapContact = (id) => {
const s = readBrokerSnap(); if (!s) return null;
const b = (s.brokers || []).find(x => +x.id === +id); if (!b) return null;
return { id: b.id, name: b.name, title: b.title, phone: b.phone, email: b.email, website: b.website,
linkedin: b.linkedin, office_addr: b.office_addr, total_assets: b.total_assets, agent_type: b.agent_type,
license: b.license, firm: b.firm, provenance: [], our_listings: [], other_listings: [], source: 'snapshot' };
};
app.get('/api/brokers/contact/:id', async (req, res) => {
if (!brokerdb) { const b = snapContact(req.params.id); return b ? res.json(b) : res.status(404).json({ error: 'broker not found' }); }
try {
const b = await brokerdb.brokerContact(+req.params.id);
if (!b) return res.status(404).json({ error: 'broker not found' });
res.json(b);
} catch (e) { const b = snapContact(req.params.id); return b ? res.json(b) : res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Enrichment coverage rollup (how many brokers now have phone/email/website/linkedin).
const snapEnrich = () => { const s = readBrokerSnap(); return s && s.enrichStats ? s.enrichStats : { unavailable: true }; };
app.get('/api/brokers/enrich-stats', async (req, res) => {
if (!brokerdb) return res.json(snapEnrich());
try { res.json(await brokerdb.enrichStats()); }
catch (e) { res.json(snapEnrich()); }
});
// LinkedIn follow-list — every broker/agent with their stored LinkedIn profile, or a
// LinkedIn people-search URL fallback when we don't have one. MANUAL + TOS-compliant by design:
// returns profile/search LINKS only; the "followed" state is tracked client-side (localStorage).
// No automated connecting/following — LinkedIn TOS prohibits 3rd-party automation (see linkedin-agent).
app.get('/api/brokers/linkedin', async (req, res) => {
if (!brokerdb) return res.json({ people: [], unavailable: true });
try {
const term = String(req.query.q || '').trim();
const args = []; let where = '';
if (term) { args.push('%' + term + '%'); where = 'WHERE (b.name ILIKE $1 OR f.name ILIKE $1)'; }
const rows = (await brokerdb.pool.query(
`SELECT b.id, b.name, f.name firm, b.agent_type, b.linkedin,
(SELECT count(*) FROM broker_listing bl WHERE bl.broker_id=b.id)
+ (SELECT count(*) FROM broker_condo bc WHERE bc.broker_id=b.id) AS listings
FROM broker b LEFT JOIN firm f ON f.id=b.firm_id ${where}
ORDER BY listings DESC NULLS LAST, b.name LIMIT $${args.length + 1}`,
[...args, Math.min(+(req.query.limit) || 500, 2000)])).rows;
const people = rows.map(b => ({
id: b.id, name: b.name, firm: b.firm, agent_type: b.agent_type, listings: b.listings,
linkedin: b.linkedin || null,
search: b.linkedin ? null
: 'https://www.linkedin.com/search/results/people/?keywords=' +
encodeURIComponent([b.name, b.firm].filter(Boolean).join(' '))
}));
res.json({
people, total: people.length, withProfile: people.filter(p => p.linkedin).length,
note: 'Manual follow list — open each profile and follow in LinkedIn. No automation (LinkedIn TOS).'
});
} catch (e) { res.json({ people: [], unavailable: true, error: String(e.message).split('\n')[0] }); }
});
// ALL brokers/agents with EVERY field — backs the sortable/searchable broker grid+list (broker-grid.html).
// PROD has no Postgres → serve the exported snapshot (built by export-brokers-snapshot.js),
// applying the same CA scope filter in JS. Mirrors the /api/residential snapshot idiom, so the
// broker grid is never empty on prod just because the DB role/connection is absent.
// Cached full-snapshot reader (re-reads only when the file changes). Backs the mind-map endpoints
// (/api/graph, /api/brokers/top, /api/brokers/enrich-stats, /api/brokers/contact) so prod (no
// Postgres) serves them from data/brokers-snapshot.json exactly like /api/brokers/all does.
let _brokerSnapCache = null, _brokerSnapMtime = 0;
function readBrokerSnap() {
try {
const p = path.join(ROOT, 'data', 'brokers-snapshot.json');
const m = fs.statSync(p).mtimeMs;
if (!_brokerSnapCache || m !== _brokerSnapMtime) { _brokerSnapCache = JSON.parse(fs.readFileSync(p, 'utf8')); _brokerSnapMtime = m; }
return _brokerSnapCache;
} catch { return null; }
}
function serveBrokerSnapshot(res, scope, reason) {
try {
const snap = readBrokerSnap();
if (!snap) throw new Error(reason || 'snapshot unavailable');
let brokers = snap.brokers || [];
if (scope === 'ca') brokers = brokers.filter(b => b.state === 'CA' || b.state == null);
return res.json({ brokers, total: brokers.length, scope, source: 'snapshot',
note: scope === 'ca' ? 'California-scoped (snapshot)' : 'all states (snapshot)' });
} catch (e) {
return res.json({ brokers: [], total: 0, unavailable: true, reason: reason || String(e.message).split('\n')[0] });
}
}
app.get('/api/brokers/all', async (req, res) => {
// Default lists ALL brokers (Steve: "list all, ok if not ca"). Geography is retained as a
// queryable label (b.state / b.dre_match) rather than a hard filter — pass ?scope=ca to get
// the California-only cut (state='CA' or still-unknown; hides the 22 confirmed out-of-state).
const scope = String(req.query.scope || 'all');
if (!brokerdb) return serveBrokerSnapshot(res, scope, 'no db module');
const caFilter = scope === 'ca' ? `WHERE (b.state = 'CA' OR b.state IS NULL)` : '';
try {
const rows = (await brokerdb.pool.query(
`SELECT b.id, b.name, f.name firm, b.agent_type, b.phone, b.email, b.website, b.linkedin,
b.license, b.title, b.total_assets, b.specialties, b.office_addr, b.crexi_id, b.source, b.created_at,
b.state, b.dre_match, b.dre_license,
(SELECT count(*) FROM broker_listing bl WHERE bl.broker_id=b.id)
+ (SELECT count(*) FROM broker_condo bc WHERE bc.broker_id=b.id) AS listings
FROM broker b LEFT JOIN firm f ON f.id=b.firm_id
${caFilter}
ORDER BY listings DESC NULLS LAST, b.name`)).rows;
res.json({ brokers: rows, total: rows.length, scope, note: scope === 'all' ? 'all states' : 'California-scoped (CA + unknown; 22 confirmed out-of-state hidden — use ?scope=all to see them)' });
} catch (e) { return serveBrokerSnapshot(res, scope, String(e.message).split('\n')[0]); }
});
// ── Government-licensed agents (cre.gov_licensed_agent) — multi-metro, loaded from state
// licensing authorities (NY DOS, FL DBPR, NV RED, ...). Metro-scoped with server-side search so
// 135k+ rows never hit the browser at once. DB locally; per-metro snapshot files on prod. ──
const GOV_DIR = path.join(ROOT, 'data', 'gov');
const GOV_LABEL = 'Licensed real-estate agents/brokers pulled from state government licensing authorities (public records, $0).';
function govSnapIndex(res) {
try { return res.json({ ...JSON.parse(fs.readFileSync(path.join(GOV_DIR, 'index.json'), 'utf8')), label: GOV_LABEL, source: 'snapshot' }); }
catch (e) { return res.json({ metros: [], total: 0, unavailable: true }); }
}
// per-metro cache (metro -> {mtime, rows}) so a big file (e.g. TX ~40MB) is read once, not per search
const _govCache = {};
function govMetroRows(metro) {
const p = path.join(GOV_DIR, `${metro}.json`);
const m = fs.statSync(p).mtimeMs;
if (!_govCache[metro] || _govCache[metro].mtime !== m) _govCache[metro] = { mtime: m, rows: JSON.parse(fs.readFileSync(p, 'utf8')) };
return _govCache[metro].rows;
}
function govSnapMetro(res, metro, q, limit) {
try {
let rows = govMetroRows(metro);
if (q) rows = rows.filter(r => (r.name + ' ' + (r.firm || '') + ' ' + (r.license_number || '') + ' ' + (r.city || '')).toLowerCase().includes(q));
return res.json({ metro, total: rows.length, agents: rows.slice(0, limit), source: 'snapshot', label: GOV_LABEL });
} catch (e) { return res.json({ metro, agents: [], total: 0, unavailable: true }); }
}
app.get('/api/gov-agents', async (req, res) => {
const metro = String(req.query.metro || '').trim().toLowerCase();
if (metro && !/^[a-z0-9-]+$/.test(metro)) return res.status(400).json({ error: 'bad metro' });
const q = String(req.query.q || '').trim().toLowerCase();
const limit = Math.min(+req.query.limit || 500, 2000);
// INDEX (metro list) when no metro given
if (!metro) {
if (brokerdb) try {
const metros = (await brokerdb.pool.query(
`SELECT state, metro, source, count(*)::int n, count(*) FILTER (WHERE license_type ILIKE '%broker%')::int brokers
FROM gov_licensed_agent GROUP BY state, metro, source ORDER BY n DESC`)).rows;
return res.json({ metros, total: metros.reduce((a, m) => a + m.n, 0), label: GOV_LABEL, source: 'db' });
} catch (e) { return govSnapIndex(res); }
return govSnapIndex(res);
}
// METRO agents (server-side search + cap)
if (brokerdb) try {
const args = [metro]; let where = 'metro=$1';
if (q) { args.push('%' + q + '%'); where += ` AND (name ILIKE $2 OR firm ILIKE $2 OR license_number ILIKE $2 OR city ILIKE $2)`; }
const total = +(await brokerdb.pool.query(`SELECT count(*) c FROM gov_licensed_agent WHERE ${where}`, args)).rows[0].c;
const agents = (await brokerdb.pool.query(
`SELECT name, firm, license_number, license_type, city, county, zip, to_char(expiration,'YYYY-MM-DD') expiration, state, source
FROM gov_licensed_agent WHERE ${where} ORDER BY name LIMIT ${limit}`, args)).rows;
return res.json({ metro, total, agents, label: GOV_LABEL, source: 'db' });
} catch (e) { return govSnapMetro(res, metro, q, limit); }
return govSnapMetro(res, metro, q, limit);
});
// Top brokers — rank brokers by the quality/mix of their CURRENT listings (mixed-use + multifamily +
// commercial weighted highest). Returns ranked brokers WITH their current listings (MLS-like fields).
// Business listing data only. No owner names.
app.get('/api/leads/top', async (req, res) => {
if (!brokerdb) return res.json({ leads: [], unavailable: true });
try {
const limit = Math.min(+(req.query.limit) || 40, 200);
const leads = (await brokerdb.pool.query(`
WITH ll AS (
SELECT b.id broker_id, b.name, f.name firm, b.agent_type,
l.id lid, l.address, l.city, l.zip, l.type, l.price, l.units, l.cap_rate,
CASE WHEN l.type ILIKE '%mixed%' THEN 3
WHEN l.type ILIKE '%condo%' THEN 3
WHEN l.type ILIKE '%multi%' THEN 2
WHEN l.type ILIKE ANY(ARRAY['%retail%','%office%','%industrial%','%commercial%']) THEN 2
ELSE 1 END AS fit
FROM broker b JOIN firm f ON f.id=b.firm_id
JOIN broker_listing bl ON bl.broker_id=b.id
JOIN listing l ON l.id=bl.listing_id)
SELECT broker_id, name, firm, agent_type,
count(*)::int listings, sum(fit)::int fit_score, coalesce(sum(price),0)::bigint total_value,
json_agg(json_build_object('id',lid,'address',address,'city',city,'zip',zip,'type',type,
'price',price,'units',units,'cap_rate',cap_rate)
ORDER BY price DESC NULLS LAST) listings_json
FROM ll GROUP BY broker_id, name, firm, agent_type
ORDER BY fit_score DESC, total_value DESC LIMIT $1`, [limit])).rows;
res.json({ leads, scoredBy: 'Listing mix quality (mixed-use/condo×3, multifamily/commercial×2). Business data; no owner names.' });
} catch (e) {
// Graceful degrade (prod has no `cre` Postgres → pool.query throws): the front-end iterates
// d.leads, so an empty-array 200 renders "no leads" cleanly instead of a 502 console error.
res.json({ leads: [], unavailable: true, reason: String(e.message).split('\n')[0] });
}
});
// Licensed agents/brokers — every RE agent CRCP harvested, backed by a REAL California DRE
// license pulled from PUBLIC government data (www2.dre.ca.gov). DB-first -> snapshot fallback,
// same idiom as /api/residential so it works locally (Postgres) AND on prod (snapshot JSON,
// produced by scripts/fetch-dre-licenses.js). Public-safe fields only — no mailing addresses.
const LICENSED_LABEL = 'Real-estate agents/brokers verified against the California DRE public license lookup (govt data, $0). Unmatched/ambiguous agents labeled honestly, never fabricated.';
app.get('/api/agents/licensed', (req, res) => {
// DRE license data lives in the snapshot (data/licensed-agents.json, built by
// scripts/fetch-dre-licenses.js). The `broker` DB table has no DRE columns, so there is
// no DB path for this feed — snapshot is canonical (and prod has no Postgres anyway).
// Public-safe fields only; mailing addresses are never included.
const PUBLIC = ['agent', 'brokerage', 'license', 'license_type', 'license_city', 'status',
'status_label', 'expiration', 'issued', 'responsible_broker', 'responsible_broker_id',
'discipline', 'match_confidence', 'dre_match', 'dre_verified', 'dre_url'];
const pubOnly = a => Object.fromEntries(PUBLIC.filter(k => k in a).map(k => [k, a[k]]));
try {
const { meta = {}, agents = [] } = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'licensed-agents.json'), 'utf8'));
return res.json({ agents: agents.map(pubOnly), meta, label: LICENSED_LABEL, source: 'snapshot' });
} catch (e) {
return res.json({ agents: [], note: 'no licensed-agent data yet — run scripts/fetch-dre-licenses.js', error: String(e.message).split('\n')[0] });
}
});
// Broker-level rollup of the SAME snapshot: group the licensed agents by their DRE
// responsible broker. This is the "broker database" half of the original ask — a pure
// in-memory aggregation over data we already captured (no new govt calls, $0). Keyed by
// responsible_broker_id when present (a real DRE license id), else the normalized broker
// name; each broker carries its supervised-agent count + license-verification breakdown.
const BROKERS_LABEL = 'Brokers derived from the CA DRE responsible-broker field of the licensed-agent set — supervising broker + how many verified/probable/unmatched agents report to each ($0, govt data).';
// The DRE responsible_broker field is "NAME <street#> ADDRESS CITY, CA <zip>". Cut from the
// first street-number run through the trailing state — CA addresses reliably end in "…, CA" —
// THEN collapse any remaining double-space tail. Order matters: strip the address first so a
// stray double-space inside the address can't truncate the name before the address is removed.
// Handles numbered streets ("8TH ST") and directionals ("W GLENOAKS") on the current snapshot;
// names without a ", CA" address (e.g. "Century 21 Peak") pass through unchanged.
const cleanBrokerName = s => String(s || '').replace(/\s\d+\s.*,\s*CA\b.*$/, '').replace(/\s{2,}.*$/, '').trim();
app.get('/api/brokers/licensed', (req, res) => {
try {
const { meta = {}, agents = [] } = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'licensed-agents.json'), 'utf8'));
const groups = new Map();
for (const a of agents) {
const raw = a.responsible_broker;
if (!raw || !raw.trim()) continue;
const key = (a.responsible_broker_id || cleanBrokerName(raw)).toUpperCase();
if (!groups.has(key)) {
groups.set(key, { broker: cleanBrokerName(raw), broker_license: a.responsible_broker_id || null,
agent_count: 0, verified: 0, probable: 0, unmatched: 0, cities: {} });
}
const g = groups.get(key);
g.agent_count++;
if (a.dre_verified) g.verified++;
else if (a.dre_match === 'probable') g.probable++;
else g.unmatched++;
const city = (a.license_city || '').trim();
if (city) g.cities[city] = (g.cities[city] || 0) + 1;
}
const brokers = [...groups.values()].map(g => ({
broker: g.broker, broker_license: g.broker_license, agent_count: g.agent_count,
verified: g.verified, probable: g.probable, unmatched: g.unmatched,
top_city: Object.entries(g.cities).sort((a, b) => b[1] - a[1])[0]?.[0] || null,
})).sort((a, b) => b.agent_count - a.agent_count);
return res.json({ brokers, meta: { ...meta, broker_count: brokers.length }, label: BROKERS_LABEL, source: 'snapshot' });
} catch (e) {
return res.json({ brokers: [], note: 'no licensed-agent data yet — run scripts/fetch-dre-licenses.js', error: String(e.message).split('\n')[0] });
}
});
// Property history for ONE address — last ~100 yrs, ADDRESS-ONLY (no owner/resident names, by design).
// Combines: the public assessor parcel roll (year built, beds/baths/sqft, use, valuation — current roll),
// sold transactions from closed_sale (Redfin sold history), and the WhoLivedThere/pastdoor archive
// (permits + curated place events) where available. Degrades gracefully if a source is empty/absent.
app.get('/api/property-history', async (req, res) => {
const address = String(req.query.address || '').trim();
const city = String(req.query.city || '').trim();
if (!address) return res.status(400).json({ error: 'address required' });
const num = (address.match(/^\s*(\d+)/) || [])[1] || '';
const word = (address.replace(/^\s*\d+\s*/, '').match(/[a-z]{3,}/i) || [])[0] || '';
const out = { address, city, assessor: null, sales: [], permits: [], events: [], names: 'omitted by design' };
if (brokerdb && num && word) {
try {
out.assessor = (await brokerdb.pool.query(
`SELECT property_location, year_built, effective_year_built, sqft_main, bedrooms, bathrooms, units,
use_desc1, use_desc2, roll_total_value, roll_land_value, roll_imp_value, roll_year, recording_date
FROM assessor_parcel WHERE situs_house_no=$1 AND situs_street ILIKE $2
ORDER BY roll_year DESC NULLS LAST LIMIT 1`, [+num, '%' + word + '%'])).rows[0] || null;
} catch (_) { /* table not ingested yet — degrade */ }
try {
out.sales = (await brokerdb.pool.query(
`SELECT address, city, sold_price, sold_date, type FROM closed_sale
WHERE address ILIKE $1 AND address ILIKE $2 AND sold_price>0
ORDER BY sold_date DESC NULLS LAST LIMIT 25`, [num + '%', '%' + word + '%'])).rows;
} catch (_) {}
}
// Assessor fallback (prod has no Postgres): if the DB path didn't populate the parcel, read the
// shipped SQLite roll. Same query shape as above; SQLite LIKE is case-insensitive for ASCII.
if (!out.assessor && assessorSqlite && num && word) {
try { out.assessor = assessorSqlite.get(+num, '%' + word + '%') || null; } catch (_) {}
}
try {
const wlt = await lookupAddress(address, city);
if (wlt && wlt.matched) { out.permits = wlt.permits || []; out.events = wlt.events || []; if (!out.assessor && wlt.parcel) out.assessorArchive = wlt.parcel; }
} catch (_) {}
res.json(out);
});
// Firm market-share: listings per firm (the broker graph's listing.firm_name). Powers the share chart.
app.get('/api/firms/share', async (req, res) => {
if (!brokerdb) return res.json([]);
try {
// firm_name on listing is sparse; derive share through broker -> firm -> broker_listing.
const r = await brokerdb.pool.query(
`SELECT f.name AS firm, count(DISTINCT bl.listing_id)::int AS listings,
count(DISTINCT b.id)::int AS brokers
FROM firm f
JOIN broker b ON b.firm_id = f.id
JOIN broker_listing bl ON bl.broker_id = b.id
GROUP BY f.name ORDER BY 2 DESC LIMIT $1`, [+(req.query.limit) || 15]);
res.json(r.rows);
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Warrantability breakdown from the FHA-approved list (the authoritative warrantability source).
// status counts + per-city top approved + expiring-soon list. All from data/fha-approved-condos.json.
app.get('/api/warrantability', (req, res) => {
try {
const file = path.join(ROOT, 'data', 'fha-approved-condos.json');
const { meta, condos } = JSON.parse(fs.readFileSync(file, 'utf8'));
const byStatus = {};
const byCity = {};
const expiring = [];
const now = new Date();
for (const c of condos) {
byStatus[c.warrant_signal] = (byStatus[c.warrant_signal] || 0) + 1;
if (c.warrant_signal === 'fha_approved') {
if (c.city) byCity[c.city] = (byCity[c.city] || 0) + 1;
const m = (c.expiration_date || '').match(/(\d{2})\/(\d{2})\/(\d{4})/);
if (m) {
const exp = new Date(+m[3], +m[1] - 1, +m[2]);
const days = Math.round((exp - now) / 86400000);
if (days >= 0 && days <= 365) expiring.push({ project: c.project_name, city: c.city, zip: c.zip, expiration: c.expiration_date, days });
}
}
}
const cities = Object.entries(byCity).sort((a, b) => b[1] - a[1]).slice(0, 12).map(([city, n]) => ({ city, n }));
expiring.sort((a, b) => a.days - b.days);
res.json({ meta: { label: meta.label, fetched_at: meta.fetched_at, total: condos.length }, byStatus, cities, expiring: expiring.slice(0, 25) });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Condo listings (warrantability-classified). Prefers the live cre DB condo_card view (dev/Mac2);
// falls back to the deployed data/condos-redfin.json snapshot when the DB is unreachable — e.g. on
// Kamatera prod, which has no `cre` Postgres role (the app runs as root, so the pg user resolved to a
// non-existent "root" role and 502'd). The snapshot carries the same rich fields as condo_card.
const CONDO_LABEL = 'FHA/VA-approval-based proxy, NOT lender-verified Fannie/Freddie warrantability.';
app.get('/api/condos', async (req, res) => {
const status = req.query.status;
// 1) live DB first
if (brokerdb) {
try {
const args = []; let where = '';
if (status && status !== 'all') { args.push(status); where = 'WHERE warrantable_status = $1'; }
// Explicit card-safe column projection (condo_card is a curated view; keep it tight, no wildcard).
const r = await brokerdb.pool.query(
`SELECT id, address, city, zip, price, hoa, beds, baths, sqft, year_built, project_name,
warrantable_status, warrant_source, warrant_signals, broker_name, firm_name, source, created_at,
status, off_market_at, disposition, sold_price, sold_date,
days_on_market, listed_date, market_status, prev_market_status, prev_price, price_changed_at
FROM condo_card ${where} ORDER BY (status='active') DESC, created_at DESC LIMIT 5000`, args);
if (r.rows.length) return res.json({ condos: r.rows, label: CONDO_LABEL, source: 'db' });
} catch (_) { /* fall through to snapshot */ }
}
// 2) snapshot fallback (prod)
try {
const file = path.join(ROOT, 'data', 'condos-redfin.json');
const { condos = [] } = JSON.parse(fs.readFileSync(file, 'utf8'));
let rows = condos;
if (status && status !== 'all') rows = rows.filter(c => c.warrantable_status === status);
return res.json({ condos: rows, label: CONDO_LABEL, source: 'snapshot' });
} catch (e) {
return res.json({ condos: [], note: 'no condo data', error: String(e.message).split('\n')[0] });
}
});
// Residential single-family for-sale inventory (cre.sfr, scraped from Redfin gis-csv). Same
// DB-first -> static-snapshot fallback shape as /api/condos so it works locally (Postgres) AND on
// prod (Kamatera has no DB — reads data/sfr-redfin.json produced by export-sfr-snapshot.js).
const SFR_LABEL = 'Single-family for-sale — Redfin LA County (feed-first gis-csv). Listing agent shown where Redfin exposed it.';
app.get('/api/residential', async (req, res) => {
const limit = Math.min(+req.query.limit || 25000, 25000); // must exceed total SFR (active grew >12k) so off-market/sold aren't truncated
// 1) live DB first
if (brokerdb) {
try {
const r = await brokerdb.pool.query(
`SELECT id, address, city, zip, price, beds, baths, sqft, year_built,
firm_name, broker_name, source, lat, lng,
status, off_market_at, disposition, sold_price, sold_date, created_at,
days_on_market, listed_date, market_status, prev_market_status, prev_price, price_changed_at
FROM sfr WHERE price > 0 AND (status='active' OR off_market_at > now() - interval '180 days')
ORDER BY (status='active') DESC, created_at DESC, price DESC LIMIT $1`, [limit]);
if (r.rows.length) return res.json({ sfr: r.rows, label: SFR_LABEL, source: 'db' });
} catch (_) { /* fall through to snapshot */ }
}
// 2) snapshot fallback (prod)
try {
const file = path.join(ROOT, 'data', 'sfr-redfin.json');
const { sfr = [] } = JSON.parse(fs.readFileSync(file, 'utf8'));
return res.json({ sfr: sfr.slice(0, limit), label: SFR_LABEL, source: 'snapshot' });
} catch (e) {
return res.json({ sfr: [], note: 'no sfr data', error: String(e.message).split('\n')[0] });
}
});
// Aggregated stats over the REAL scraped condo listings (cre.condo) — warrantable-vs-unwarranted
// breakdown + by-city, driving the graphics. Distinct from /api/warrantability (which charts the HUD
// FHA reference LIST, not the actual for-sale inventory we scraped).
app.get('/api/condos/stats', async (req, res) => {
if (!brokerdb) return res.json({ byStatus: {}, byCity: [], total: 0, unavailable: true });
try {
const byStatus = {};
(await brokerdb.pool.query(`SELECT warrantable_status s, count(*)::int n FROM condo GROUP BY 1`)).rows
.forEach(r => byStatus[r.s] = r.n);
const byCity = (await brokerdb.pool.query(
`SELECT city, count(*)::int n,
count(*) FILTER (WHERE warrantable_status='fha_approved')::int approved
FROM condo WHERE city <> '' GROUP BY city ORDER BY n DESC LIMIT 14`)).rows;
const price = (await brokerdb.pool.query(
`SELECT warrantable_status s, round(avg(price))::bigint avg_price, count(*)::int n
FROM condo WHERE price>0 GROUP BY 1`)).rows;
const total = (await brokerdb.pool.query(`SELECT count(*)::int n FROM condo`)).rows[0].n;
const [agents] = (await brokerdb.pool.query(`SELECT
(SELECT count(DISTINCT condo_id) FROM broker_condo)::int condos_with_agent,
(SELECT count(*) FROM broker WHERE agent_type='residential')::int residential_agents,
(SELECT count(*) FROM broker WHERE agent_type='residential' AND phone IS NOT NULL)::int agents_phone,
(SELECT count(*) FROM broker WHERE agent_type='residential' AND email IS NOT NULL)::int agents_email`).catch(() => ({ rows: [{}] }))).rows;
res.json({
total, byStatus, byCity, price, agents,
label: 'FHA/VA-approval-based proxy, NOT lender-verified Fannie/Freddie warrantability.',
source: 'Redfin LA County condo listings (feed-first gis-csv) classified against the HUD FHA list',
agentSource: 'Residential listing agents captured per-condo from Redfin detail feed (mainHouseInfoPanelInfo) — business contact only'
});
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// FHA-approved project directory — the real, browsable warrantable (proxy) inventory. Searchable by
// city/zip/name. This is what's authoritative today even before residential listings are scraped.
app.get('/api/fha-condos', (req, res) => {
try {
const file = path.join(ROOT, 'data', 'fha-approved-condos.json');
const { meta, condos } = JSON.parse(fs.readFileSync(file, 'utf8'));
const q = String(req.query.q || '').toLowerCase();
const status = req.query.status || 'fha_approved';
let rows = condos.filter(c => status === 'all' || c.warrant_signal === status);
if (q) rows = rows.filter(c => (c.project_name + ' ' + c.city + ' ' + c.zip + ' ' + c.address).toLowerCase().includes(q));
res.json({ label: meta.label, count: rows.length, condos: rows.slice(0, 400) });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0], condos: [] }); }
});
// CRCP — Commercial Real Estate Control Panel live stats (CNCP-style). One poll → every headline
// number. brokersWithPhone/Email + condos tick UP live while the enrichment/scrape jobs run, so the
// panel visibly "populates". All read-only.
app.get('/api/crcp/stats', async (req, res) => {
const out = { ts: Date.now() };
try {
// listings + market from ranked.json (no DB needed)
try {
const rk = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'ranked.json'), 'utf8'));
out.listings = rk.ranked.length; out.market = rk.meta && rk.meta.market;
const types = {}; rk.ranked.forEach(p => { types[p.type] = (types[p.type] || 0) + 1; });
out.byType = types;
} catch (_) { out.listings = 0; }
// FHA warrantability reference
try {
const fc = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'fha-approved-condos.json'), 'utf8')).condos;
out.fhaApproved = fc.filter(c => c.warrant_signal === 'fha_approved').length;
out.fhaExpired = fc.filter(c => c.warrant_signal === 'fha_expired').length;
} catch (_) {}
if (brokerdb) {
const q = s => brokerdb.pool.query(s).then(r => r.rows);
const [c1] = await q(`SELECT
(SELECT count(*) FROM broker)::int brokers,
(SELECT count(*) FROM firm)::int firms,
(SELECT count(*) FROM broker WHERE phone IS NOT NULL)::int brokers_phone,
(SELECT count(*) FROM broker WHERE email IS NOT NULL)::int brokers_email,
(SELECT count(*) FROM broker_listing)::int edges,
(SELECT count(*) FROM broker_cobroker)::int colist,
(SELECT count(*) FROM broker WHERE agent_type='residential')::int agents_residential,
(SELECT count(*) FROM broker WHERE agent_type='commercial')::int agents_commercial,
(SELECT count(*) FROM broker WHERE agent_type='residential' AND phone IS NOT NULL)::int agents_res_phone,
(SELECT count(*) FROM broker WHERE agent_type='residential' AND email IS NOT NULL)::int agents_res_email,
(SELECT count(DISTINCT condo_id) FROM broker_condo)::int condos_with_agent,
(SELECT count(DISTINCT sfr_id) FROM broker_sfr)::int sfr_with_agent`).catch(() => [{}]);
Object.assign(out, c1);
try { out.brokers_web = (await q(`SELECT count(*)::int n FROM broker WHERE website IS NOT NULL`))[0].n; } catch (_) { out.brokers_web = null; }
try { out.condos = (await q(`SELECT count(*)::int n FROM condo`))[0].n;
out.condosByStatus = {}; (await q(`SELECT warrantable_status s,count(*)::int n FROM condo GROUP BY 1`)).forEach(r => out.condosByStatus[r.s] = r.n); } catch (_) { out.condos = 0; }
try { out.sfr = (await q(`SELECT count(*)::int n FROM sfr`))[0].n; } catch (_) { out.sfr = 0; }
const brokerCols = 'b.id, b.name, f.name firm, b.agent_type, ' +
'(count(DISTINCT bl.listing_id) + count(DISTINCT bc.condo_id))::int listings, b.total_assets, b.phone, b.email' +
(out.brokers_web != null ? ', b.website' : '');
out.topBrokers = await q(`SELECT ${brokerCols}
FROM broker b LEFT JOIN firm f ON f.id=b.firm_id
LEFT JOIN broker_listing bl ON bl.broker_id=b.id
LEFT JOIN broker_condo bc ON bc.broker_id=b.id
GROUP BY b.id, f.name ORDER BY listings DESC NULLS LAST LIMIT 20`).catch(() => []); // prod: no cre DB — fall to snapshot below
out.topFirms = await q(`SELECT f.name firm, count(DISTINCT bl.listing_id)::int listings, count(DISTINCT b.id)::int brokers FROM firm f JOIN broker b ON b.firm_id=f.id JOIN broker_listing bl ON bl.broker_id=b.id GROUP BY 1 ORDER BY 2 DESC LIMIT 12`).catch(() => null);
}
// Prod fallback: brokerdb requires fine on Kamatera but every query fails (no cre DB), which
// used to leave the CRCP broker panels EMPTY on prod. Fill from the snapshot, same as
// /api/brokers/top et al. — top rows carry id+website so the ✉ find-email button works there.
if (!out.topBrokers || !out.topBrokers.length) {
const s = readBrokerSnap();
if (s) {
out.topBrokers = (s.top || []).slice(0, 20);
const all = s.brokers || [];
if (out.brokers == null) out.brokers = all.length;
if (out.brokers_phone == null) out.brokers_phone = (s.enrichStats && s.enrichStats.phone) || all.filter(b => b.phone).length;
if (out.brokers_email == null) out.brokers_email = (s.enrichStats && s.enrichStats.email) || all.filter(b => b.email).length;
if (out.brokers_web == null) out.brokers_web = (s.enrichStats && s.enrichStats.website) || all.filter(b => b.website).length;
if (out.agents_residential == null) out.agents_residential = all.filter(b => b.agent_type === 'residential').length;
if (out.firms == null) out.firms = new Set(all.map(b => b.firm).filter(Boolean)).size;
if (!out.topFirms) {
const fc = {};
all.forEach(b => { if (b.firm) { fc[b.firm] = fc[b.firm] || { firm: b.firm, listings: 0, brokers: 0 }; fc[b.firm].brokers++; fc[b.firm].listings += +b.total_assets || 0; } });
out.topFirms = Object.values(fc).sort((a, b) => b.listings - a.listings).slice(0, 12);
}
out.brokerSource = 'snapshot';
delete out.error;
}
}
res.json(out);
} catch (e) { res.status(502).json({ ...out, error: String(e.message).split('\n')[0] }); }
});
// Broker contact detail — full BUSINESS contact + current listings + public closed listings.
// Legitimate B2B prep only (no personal/family data). Click-through target for the CRCP + mind-map.
app.get('/api/broker', async (req, res) => {
if (!brokerdb) return res.status(503).json({ error: 'broker DB unavailable' });
const id = req.query.id ? +req.query.id : null;
const name = String(req.query.name || '').trim();
if (!id && !name) return res.status(400).json({ error: 'id or name required' });
try {
const q = (s, a) => brokerdb.pool.query(s, a).then(r => r.rows);
// Prefer id (unambiguous — names collide across commercial/residential). For a name lookup,
// resolve deterministically: the row with the most listings (commercial edges + condos) wins.
const sel = `SELECT b.id, b.name, f.name firm, b.phone, b.email, b.title, b.total_assets, b.website, b.agent_type, b.license`;
const brokers = id
? await q(`${sel} FROM broker b LEFT JOIN firm f ON f.id=b.firm_id WHERE b.id=$1`, [id])
: await q(`${sel},
(SELECT count(*) FROM broker_listing bl WHERE bl.broker_id=b.id)
+ (SELECT count(*) FROM broker_condo bc WHERE bc.broker_id=b.id) AS _rank
FROM broker b LEFT JOIN firm f ON f.id=b.firm_id
WHERE b.name ILIKE $1 ORDER BY _rank DESC, b.id LIMIT 1`, [name]);
const broker = brokers[0];
if (!broker) return res.status(404).json({ error: 'broker not found' });
// Current listings = commercial listings (broker_listing) UNION residential condos (broker_condo).
// listed_at = when we captured the listing (best available "listed" date; no MLS list-date feed).
const current = await q(`SELECT l.id, l.address, l.city, l.zip, l.type, l.price, l.units, l.cap_rate, l.created_at AS listed_at, bl.role
FROM broker_listing bl JOIN listing l ON l.id=bl.listing_id WHERE bl.broker_id=$1
UNION ALL
SELECT c.id, c.address, c.city, NULL::text zip, 'Condo' type, c.price, NULL::int units, NULL::numeric cap_rate, NULL::timestamptz listed_at, bc.role
FROM broker_condo bc JOIN condo c ON c.id=bc.condo_id WHERE bc.broker_id=$1
ORDER BY price DESC NULLS LAST`, [broker.id]);
const closed = await q(`SELECT address, city, sold_price, sold_date, type, source FROM broker_closed_listing WHERE broker_id=$1 ORDER BY sold_date DESC NULLS LAST`, [broker.id]).catch(() => []);
// Expired / off-market: best-effort match of THIS broker's listings against the sold feed — a listing
// that now appears as sold is no longer active. Match = same house number + same city + a street-word
// in common. (We have no MLS off-market/expired feed, so this sold-cross-ref is the honest signal.)
const expired = await q(`SELECT DISTINCT l.id, l.address, l.city, l.type, l.price AS list_price,
cs.sold_price, cs.sold_date, cs.mls, cs.url
FROM broker_listing bl JOIN listing l ON l.id=bl.listing_id
JOIN closed_sale cs ON cs.city = l.city
AND split_part(cs.address,' ',1) = split_part(l.address,' ',1)
AND lower(cs.address) LIKE '%'||lower(coalesce((regexp_match(l.address,'[A-Za-z]{3,}'))[1],'~'))||'%'
WHERE bl.broker_id=$1 AND cs.sold_price>0
ORDER BY cs.sold_date DESC NULLS LAST LIMIT 12`, [broker.id]).catch(() => []);
// Sales: recent closed_sale (Redfin sold) in the cities this broker lists in — with sold DATE + MLS# + url.
// Honest area context — NOT attributed to this broker (the sold feed carries no agent).
const cities = [...new Set(current.map(c => c.city).filter(Boolean))].slice(0, 6);
let comps = [];
if (cities.length) comps = await q(
`SELECT address, city, zip, sold_price, sold_date, beds, baths, sqft, mls, url FROM closed_sale
WHERE city = ANY($1) AND sold_price>0 ORDER BY sold_date DESC NULLS LAST LIMIT 12`, [cities]).catch(() => []);
res.json({ broker, current, closed, expired, comps,
compsNote: comps.length ? 'Recent sold comps in this broker’s areas (Redfin) with sold dates + MLS# — area context, not this broker’s own closings.' : null,
expiredNote: expired.length ? 'Listings of this broker that now appear in the sold feed (best-effort address match).' : 'No off-market/expired feed yet — none of this broker’s listings matched a sold record.',
closedNote: closed.length ? null : 'Per-broker closed history needs a gated detail-scrape; area sales shown instead.' });
} catch (e) {
// Graceful degrade (prod has no `cre` Postgres → pool.query throws): SAME {error} body the
// front-end already handles (modal shows d.error; featured card guards !d.broker), but 200 not
// 502 so it isn't a red console error on every internal page load.
res.json({ error: String(e.message).split('\n')[0], unavailable: true });
}
});
// Firm detail — the roster (brokers/agents at the firm) + aggregate listings. Powers the click-through
// on a firm row. Read-only.
app.get('/api/firm', async (req, res) => {
if (!brokerdb) return res.status(503).json({ error: 'broker DB unavailable' });
const name = String(req.query.name || '').trim();
if (!name) return res.status(400).json({ error: 'name required' });
try {
const q = (s, a) => brokerdb.pool.query(s, a).then(r => r.rows);
const firm = (await q(`SELECT id, name FROM firm WHERE name ILIKE $1 LIMIT 1`, [name]))[0];
if (!firm) return res.status(404).json({ error: 'firm not found' });
const roster = await q(
`SELECT b.id, b.name, b.agent_type, b.phone, b.email, b.website,
(SELECT count(*) FROM broker_listing bl WHERE bl.broker_id=b.id)
+ (SELECT count(*) FROM broker_condo bc WHERE bc.broker_id=b.id) AS listings
FROM broker b WHERE b.firm_id=$1 ORDER BY listings DESC, b.name LIMIT 200`, [firm.id]);
const agg = (await q(`SELECT count(*)::int brokers,
count(*) FILTER (WHERE agent_type='residential')::int residential,
count(phone)::int phone, count(email)::int email FROM broker WHERE firm_id=$1`, [firm.id]))[0];
res.json({ firm, roster, agg });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// On-demand email hunt for ONE broker — backs the "✉ find email" button on the CRCP Top-brokers
// table + broker grid + contact modal. Visits the broker's own website (homepage + up to 3
// contact-ish pages), extracts mailto:/text emails, and picks with the SAME guards as the
// broker-email-finder.js batch (shared lib/email-hunt.js): the broker's name in the local-part
// wins; refuses to guess on shared corporate pages — a stranger's email is worse than none.
// $0 (plain local fetch, no paid API). Found email persists: broker.email locally (Postgres),
// or written back into data/brokers-snapshot.json on DB-less prod (sticks until next refresh).
const emailHunt = require('./lib/email-hunt');
// Browser escalation (headless Chrome) when Playwright is available (the Mac); Kamatera prod has
// no Playwright → loadable() false → the button quietly runs fetch-only. Browser is closed after
// each hunt — a button click is rare enough that a ~1s relaunch beats a resident Chrome.
let emailHuntBrowser = null; try { const b = require('./lib/email-hunt-browser'); if (b.loadable()) emailHuntBrowser = b; } catch (_) {}
const _huntInFlight = new Set();
app.post('/api/broker/find-email', async (req, res) => {
const bid = (req.body || {}).id ? +(req.body || {}).id : null;
let name = String((req.body || {}).name || '').trim();
let website = String((req.body || {}).website || '').trim();
try {
// Resolve name/website server-side (DB locally, snapshot on prod) so the hunt + any write
// target the broker ON FILE — the client can't point a broker id at an arbitrary site.
// NOTE prod landmine: brokerdb REQUIRES fine on Kamatera but every query fails (no cre DB
// there) — so a DB error must fall through to the snapshot, same as the other broker routes.
let dbOk = false;
if (bid && brokerdb) {
try {
const b = (await brokerdb.pool.query(`SELECT id, name, website, email FROM broker WHERE id=$1`, [bid])).rows[0];
dbOk = true;
if (!b) return res.status(404).json({ error: 'broker not found' });
if (b.email) return res.json({ email: b.email, basis: 'already-on-file', saved: false });
name = b.name; website = b.website || '';
} catch (_) { dbOk = false; }
}
if (bid && !dbOk) {
const s = readBrokerSnap(); const b = s && (s.brokers || []).find(x => +x.id === bid);
if (b) {
if (b.email) return res.json({ email: b.email, basis: 'already-on-file', saved: false });
name = b.name || name; website = b.website || '';
}
}
if (!website) return res.status(400).json({ error: 'no website on file for this broker' });
if (!/^https?:\/\//i.test(website)) website = 'https://' + website;
const key = bid || website;
if (_huntInFlight.has(key)) return res.status(429).json({ error: 'a hunt is already running for this broker' });
_huntInFlight.add(key);
let r;
try {
const opts = emailHuntBrowser ? { browserHtml: emailHuntBrowser.browserHtml } : {};
const cap = emailHuntBrowser ? 75000 : 45000; // browser tier legitimately needs longer
r = await emailHunt.withTimeout(emailHunt.huntEmails({ id: bid, name, website }, opts), cap,
() => ({ found: null, basis: 'timeout', all: [], tried: [] }));
} finally {
_huntInFlight.delete(key);
if (emailHuntBrowser) emailHuntBrowser.closeBrowser();
}
let saved = false;
if (r.found && bid) {
if (dbOk) {
try {
const u = await brokerdb.pool.query(`UPDATE broker SET email=$1 WHERE id=$2 AND (email IS NULL OR email='')`, [r.found, bid]);
saved = u.rowCount > 0;
} catch (_) {}
} else {
try { // DB-less prod: write into the snapshot so the find survives reloads until the nightly refresh
const s = readBrokerSnap();
if (s) {
for (const arr of [s.brokers, s.top]) (arr || []).forEach(x => { if (+x.id === bid && !x.email) { x.email = r.found; saved = true; } });
if (saved) { const _sp = path.join(ROOT, 'data', 'brokers-snapshot.json'); fs.writeFileSync(_sp + '.tmp', JSON.stringify(s)); fs.renameSync(_sp + '.tmp', _sp); }
}
} catch (_) { saved = false; }
}
}
// "reached" = ANY page fetched OK (a failed sub-page must not read as "site unreachable")
const reachedOk = (r.tried || []).some(t => t.status >= 200 && t.status < 400);
res.json({
email: r.found || null, basis: r.basis || null, saved,
others: (r.all || []).filter(e => e !== r.found).slice(0, 5),
site: emailHunt.siteDomain(website),
reason: r.found ? null
: r.basis === 'ambiguous-shared-page' ? `shared page — ${(r.all || []).length} other agents' emails, none match ${name}; refused to guess`
: r.basis === 'timeout' ? `site too slow (${cap / 1000}s cap)`
: reachedOk ? 'reached the site but no email is published on it'
: 'site unreachable/blocked (' + (r.tried || []).map(t => t.status || t.err || '?').join('/') + ')',
cost: '$0 (local fetch)'
});
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Closed-sales comps (Redfin sold, ~5yr) — browsable market dataset + aggregates. Read-only.
app.get('/api/closed-sales', async (req, res) => {
const city = String(req.query.city || '').trim();
const minP = +req.query.minPrice || 0, maxP = +req.query.maxPrice || 0;
const lim = Math.min(Math.max(+req.query.limit || 600, 1), 8000); // clamped Number — safe to interpolate
// 1) live cre DB (dev/Mac2)
if (brokerdb) {
try {
const q = (s, a) => brokerdb.pool.query(s, a).then(r => r.rows);
const where = ['sold_price>0']; const args = [];
if (city) { args.push(city); where.push(`city ILIKE $${args.length}`); }
if (minP) { args.push(minP); where.push(`sold_price>=$${args.length}`); }
if (maxP) { args.push(maxP); where.push(`sold_price<=$${args.length}`); }
const w = 'WHERE ' + where.join(' AND ');
const rows = await q(`SELECT address, city, zip, sold_price, sold_date, beds, baths, sqft, url FROM closed_sale ${w} ORDER BY sold_date DESC NULLS LAST LIMIT ${lim}`, args);
if (rows.length) {
const byCity = await q(`SELECT city, count(*)::int n, round(avg(sold_price))::bigint avg_price FROM closed_sale ${w} GROUP BY city ORDER BY n DESC LIMIT 16`, args);
const byYear = await q(`SELECT extract(year FROM sold_date)::int yr, count(*)::int n, round(avg(sold_price))::bigint avg_price FROM closed_sale ${w} AND sold_date IS NOT NULL GROUP BY 1 ORDER BY 1`, args);
const tot = (await q(`SELECT count(*)::int n, round(avg(sold_price))::bigint avg_price, min(sold_date) oldest, max(sold_date) newest FROM closed_sale ${w}`, args))[0];
return res.json({ rows, byCity, byYear, total: tot,
source: 'Redfin public sold feed (gis-csv)',
depthNote: 'Redfin public sold feed (~5yr depth, 2021+); full 10-yr history requires a paid records source (ATTOM / county recorder).' });
}
} catch (_) { /* fall through to snapshot */ }
}
// 2) snapshot fallback (prod — no cre DB on Kamatera). Aggregates are precomputed over the full set.
try {
const snap = JSON.parse(fs.readFileSync(path.join(ROOT, 'data', 'closed-sales.json'), 'utf8'));
let rows = snap.rows || [];
if (city) { const cl = city.toLowerCase(); rows = rows.filter(r => String(r.city || '').toLowerCase().includes(cl)); }
if (minP) rows = rows.filter(r => +r.sold_price >= minP);
if (maxP) rows = rows.filter(r => +r.sold_price <= maxP);
rows = rows.slice(0, lim);
return res.json({ rows, byCity: snap.byCity || [], byYear: snap.byYear || [], total: snap.total || {},
source: 'Redfin public sold feed (snapshot)',
depthNote: 'Snapshot of the Redfin sold feed (cre DB not attached on this host).' });
} catch (e) { return res.json({ rows: [], byCity: [], byYear: [], total: {}, unavailable: true, error: String(e.message).split('\n')[0] }); }
});
// Assumable-loan estimates (read-only). HEURISTIC screening signal, NOT title-verified — see
// scripts/assumable-heuristic.js. ?likelihood=high|medium|low filters; ?limit caps (default 200).
app.get('/api/assumable', async (req, res) => {
if (!brokerdb) return res.status(503).json({ error: 'DB unavailable' });
try {
const { likelihood } = req.query || {};
const limit = Math.min(+(req.query.limit) || 200, 1000);
const args = []; let where = '';
if (likelihood && ['high', 'medium', 'low'].includes(likelihood)) { args.push(likelihood); where = 'WHERE assumable_likelihood = $1'; }
const rows = (await brokerdb.pool.query(
`SELECT property_id, src, address, city, zip, price, assumable_likelihood, est_assumable_rate,
est_sale_year, basis, confidence, signals
FROM assumable_card ${where}
ORDER BY CASE assumable_likelihood WHEN 'high' THEN 0 WHEN 'medium' THEN 1 ELSE 2 END, price DESC NULLS LAST
LIMIT ${limit}`, args)).rows;
const counts = (await brokerdb.pool.query(
`SELECT assumable_likelihood s, count(*)::int n FROM assumable_estimate GROUP BY 1`)).rows
.reduce((o, r) => (o[r.s] = r.n, o), {});
res.json({
label: 'HEURISTIC estimate of an assumable FHA/VA loan — screening signal, NOT title-verified. Verify in title records before relying on it.',
counts, estimates: rows
});
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// ── Call Segments — prospect buckets by property/financing category ────────────────────────────────
// Each segment = a set of properties + the AGENTS who list them, with a short category label + note.
// src 'listing' (commercial, has units/type) | 'condo' (residential, warrantability) | 'sfr'
// (none scraped yet) | 'nonqm' (umbrella over the others).
const SEGMENTS = {
'mixed-use': { label: 'Mixed-use', product: 'Commercial / portfolio (non-QM)', hook: 'mixed-use financing conventional lenders avoid', src: 'listing', where: "type ILIKE '%mixed%'" },
'nonwarrantable-condo': { label: 'Non-warrantable condos', product: 'Non-QM / portfolio condo', hook: 'finance the non-warrantable condo (not FHA-approved) — the deal conventional lenders decline', src: 'condo', where: "warrantable_status <> 'fha_approved'" },
'standard-condo': { label: 'Standard condos (FHA-approved)', product: 'Conventional + Fannie Mae CPM check', hook: "offer to run the complex through Fannie Mae Condo Project Manager (CPM) for prior approvals/rejections", src: 'condo', where: "warrantable_status = 'fha_approved'" },
'dscr-1-4': { label: 'DSCR — 1-4 units', product: 'DSCR investor loan (1-4 units)', hook: 'DSCR — qualify on property cash flow, no income docs, 1-4 units', src: 'listing', where: 'units BETWEEN 1 AND 4' },
'dscr-5-9': { label: 'DSCR — 5-9 units', product: 'DSCR small multifamily (5-9 units)', hook: 'DSCR small multifamily, 5-9 units on cash flow', src: 'listing', where: 'units BETWEEN 5 AND 9' },
'sfr-bankstmt': { label: 'SFR — bank-statement / P&L', product: 'Bank Statement / P&L loan', hook: 'self-employed SFR buyers who need bank-statement or P&L income qualification', src: 'sfr', where: 'TRUE' },
// Assumable FHA/VA — agents whose listings LIKELY carry an assumable low-rate FHA/VA loan (heuristic
// estimate, NOT title-verified). src 'assumable' joins cre.assumable_estimate (medium|high) back to its
// condo/sfr property and onto the listing-agent edge. See scripts/assumable-heuristic.js.
'assumable-fha-va': { label: 'Assumable FHA/VA (estimate)', product: 'Assumable FHA/VA (estimate)', hook: 'their listing likely carries an assumable low-rate FHA/VA loan (verify in title records)', src: 'assumable', where: "assumable_likelihood IN ('medium','high')" },
'nonqm': { label: 'Non-QM (all)', product: 'Non-QM umbrella', hook: 'any non-QM scenario — non-warrantable condo, DSCR, bank-statement, asset-based', src: 'nonqm', where: null }
};
function segCallList(seg) {
// every branch returns avg_price + top_price so the call list shows the BUILDING PRICES
// each agent carries (NULLIF avoids /0; prices are the listed/condo/sfr asking prices).
if (seg.src === 'listing') return {
sql: `SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*)::int n, min(l.address) sample,
round(avg(NULLIF(l.price,0)))::bigint avg_price, max(l.price)::bigint top_price
FROM listing l JOIN broker_listing bl ON bl.listing_id=l.id JOIN broker b ON b.id=bl.broker_id
WHERE ${seg.where} GROUP BY b.id ORDER BY n DESC, b.name LIMIT 300`, args: [] };
if (seg.src === 'condo') return {
sql: `SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*)::int n, min(c.address) sample,
round(avg(NULLIF(c.price,0)))::bigint avg_price, max(c.price)::bigint top_price
FROM condo c JOIN broker_condo bc ON bc.condo_id=c.id JOIN broker b ON b.id=bc.broker_id
WHERE ${seg.where} GROUP BY b.id ORDER BY n DESC, b.name LIMIT 300`, args: [] };
if (seg.src === 'nonqm') return {
sql: `SELECT id, name, agent_type, phone, email, sum(n)::int n, min(sample) sample,
round(avg(NULLIF(avg_price,0)))::bigint avg_price, max(top_price)::bigint top_price FROM (
SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*) n, min(c.address) sample,
avg(NULLIF(c.price,0)) avg_price, max(c.price) top_price
FROM condo c JOIN broker_condo bc ON bc.condo_id=c.id JOIN broker b ON b.id=bc.broker_id
WHERE c.warrantable_status <> 'fha_approved' GROUP BY b.id
UNION ALL
SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*) n, min(l.address) sample,
avg(NULLIF(l.price,0)) avg_price, max(l.price) top_price
FROM listing l JOIN broker_listing bl ON bl.listing_id=l.id JOIN broker b ON b.id=bl.broker_id
WHERE l.units BETWEEN 1 AND 9 OR l.type ILIKE '%mixed%' GROUP BY b.id
) u GROUP BY id, name, agent_type, phone, email ORDER BY n DESC LIMIT 300`, args: [] };
if (seg.src === 'sfr') return {
sql: `SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*)::int n, min(s.address) sample,
round(avg(NULLIF(s.price,0)))::bigint avg_price, max(s.price)::bigint top_price
FROM sfr s JOIN broker_sfr bs ON bs.sfr_id=s.id JOIN broker b ON b.id=bs.broker_id
WHERE ${seg.where} GROUP BY b.id ORDER BY n DESC, b.name LIMIT 300`, args: [] };
if (seg.src === 'assumable') return {
// assumable estimate -> property -> listing-agent edge (condo OR sfr). Union both edges so a
// likely-assumable condo OR sfr surfaces its listing agent. Estimate is heuristic (see hook).
sql: `SELECT id, name, agent_type, phone, email, sum(n)::int n, min(sample) sample,
round(avg(NULLIF(avg_price,0)))::bigint avg_price, max(top_price)::bigint top_price FROM (
SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*) n, min(c.address) sample,
avg(NULLIF(c.price,0)) avg_price, max(c.price) top_price
FROM assumable_estimate a JOIN condo c ON c.id=a.property_id AND a.src='condo'
JOIN broker_condo bc ON bc.condo_id=c.id JOIN broker b ON b.id=bc.broker_id
WHERE ${seg.where} GROUP BY b.id
UNION ALL
SELECT b.id, b.name, b.agent_type, b.phone, b.email, count(*) n, min(s.address) sample,
avg(NULLIF(s.price,0)) avg_price, max(s.price) top_price
FROM assumable_estimate a JOIN sfr s ON s.id=a.property_id AND a.src='sfr'
JOIN broker_sfr bs ON bs.sfr_id=s.id JOIN broker b ON b.id=bs.broker_id
WHERE ${seg.where} GROUP BY b.id
) u GROUP BY id, name, agent_type, phone, email ORDER BY n DESC LIMIT 300`, args: [] };
return null;
}
app.get('/api/segments', async (req, res) => {
if (!brokerdb) return res.json({ segments: [] });
try {
const out = [];
for (const [key, seg] of Object.entries(SEGMENTS)) {
let props = 0, agents = 0, priceMed = null, priceMin = null, priceMax = null;
// price band uses p10–p50–p90 so a single outlier ($400M typo) doesn't blow out the range.
const PCT = p => `percentile_disc(${p}) WITHIN GROUP (ORDER BY price)::bigint`;
const tbl = seg.src === 'listing' ? 'listing' : seg.src === 'condo' ? 'condo' : seg.src === 'sfr' ? 'sfr' : null;
let priceFrom = tbl ? `(SELECT price FROM ${tbl} WHERE (${seg.where}) AND price>0)` : null;
if (seg.src === 'nonqm') priceFrom =
`(SELECT price FROM condo WHERE warrantable_status<>'fha_approved' AND price>0
UNION ALL SELECT price FROM listing WHERE (units BETWEEN 1 AND 9 OR type ILIKE '%mixed%') AND price>0)`;
if (seg.src === 'assumable') priceFrom =
`(SELECT c.price FROM assumable_estimate a JOIN condo c ON c.id=a.property_id AND a.src='condo' WHERE (${seg.where}) AND c.price>0
UNION ALL SELECT s.price FROM assumable_estimate a JOIN sfr s ON s.id=a.property_id AND a.src='sfr' WHERE (${seg.where}) AND s.price>0)`;
if (priceFrom) {
const pr = (await brokerdb.pool.query(
`SELECT ${PCT(0.5)} med, ${PCT(0.1)} lo, ${PCT(0.9)} hi FROM ${priceFrom} pp(price)`)).rows[0];
priceMed = pr.med; priceMin = pr.lo; priceMax = pr.hi;
}
if (seg.src === 'listing') {
props = (await brokerdb.pool.query(`SELECT count(*)::int n FROM listing WHERE ${seg.where}`)).rows[0].n;
agents = (await brokerdb.pool.query(`SELECT count(DISTINCT bl.broker_id)::int n FROM listing l JOIN broker_listing bl ON bl.listing_id=l.id WHERE ${seg.where}`)).rows[0].n;
} else if (seg.src === 'condo') {
props = (await brokerdb.pool.query(`SELECT count(*)::int n FROM condo WHERE ${seg.where}`)).rows[0].n;
agents = (await brokerdb.pool.query(`SELECT count(DISTINCT bc.broker_id)::int n FROM condo c JOIN broker_condo bc ON bc.condo_id=c.id WHERE ${seg.where}`)).rows[0].n;
} else if (seg.src === 'sfr') {
props = (await brokerdb.pool.query(`SELECT count(*)::int n FROM sfr WHERE ${seg.where}`)).rows[0].n;
agents = (await brokerdb.pool.query(`SELECT count(DISTINCT bs.broker_id)::int n FROM sfr s JOIN broker_sfr bs ON bs.sfr_id=s.id WHERE ${seg.where}`)).rows[0].n;
} else if (seg.src === 'assumable') {
props = (await brokerdb.pool.query(`SELECT count(*)::int n FROM assumable_estimate WHERE ${seg.where}`)).rows[0].n;
agents = (await brokerdb.pool.query(
`SELECT count(DISTINCT broker_id)::int n FROM (
SELECT bc.broker_id FROM assumable_estimate a JOIN broker_condo bc ON bc.condo_id=a.property_id AND a.src='condo' WHERE ${seg.where}
UNION SELECT bs.broker_id FROM assumable_estimate a JOIN broker_sfr bs ON bs.sfr_id=a.property_id AND a.src='sfr' WHERE ${seg.where}
) u`)).rows[0].n;
} else if (seg.src === 'nonqm') {
props = (await brokerdb.pool.query(`SELECT (SELECT count(*) FROM condo WHERE warrantable_status<>'fha_approved') + (SELECT count(*) FROM listing WHERE units BETWEEN 1 AND 9 OR type ILIKE '%mixed%') n`)).rows[0].n;
// agents across BOTH union legs (was unwired → showed 0): non-FHA condo agents ∪ 1-9-unit/mixed listing agents
agents = (await brokerdb.pool.query(
`SELECT count(DISTINCT broker_id)::int n FROM (
SELECT bc.broker_id FROM condo c JOIN broker_condo bc ON bc.condo_id=c.id WHERE c.warrantable_status<>'fha_approved'
UNION SELECT bl.broker_id FROM listing l JOIN broker_listing bl ON bl.listing_id=l.id WHERE (l.units BETWEEN 1 AND 9 OR l.type ILIKE '%mixed%')
) u`)).rows[0].n;
}
const noAgents = (seg.src === 'sfr' || seg.src === 'assumable') && agents === 0;
out.push({ key, label: seg.label, product: seg.product, hook: seg.hook, props, agents,
priceMed, priceMin, priceMax,
note: noAgents ? 'Listings matched, but no public listing agents captured yet (HONEST: agent edge sparse).'
: seg.src === 'assumable' ? 'HEURISTIC estimate — likely-assumable FHA/VA, NOT title-verified. Verify in title records before relying on it.' : null });
}
res.json({ segments: out });
} catch (e) { res.json({ segments: [], unavailable: true, error: String(e.message).split('\n')[0] }); }
});
app.get('/api/segment', async (req, res) => {
if (!brokerdb) return res.status(503).json({ error: 'DB unavailable' });
const seg = SEGMENTS[req.query.key];
if (!seg) return res.status(404).json({ error: 'unknown segment' });
try {
const cl = segCallList(seg);
const callList = cl ? (await brokerdb.pool.query(cl.sql, cl.args)).rows : [];
const note = (seg.src === 'sfr' || seg.src === 'assumable') && callList.length === 0
? 'Listings matched, but no public listing agents captured yet.'
: seg.src === 'assumable' ? 'HEURISTIC estimate — likely-assumable FHA/VA, NOT title-verified. Verify in title records before relying on it.' : null;
res.json({ key: req.query.key, label: seg.label, product: seg.product, hook: seg.hook, note, callList });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Rent rolls — list the parsed sidecars (data/rent-rolls/*.json) the ingester produces. Read-only.
// Document-grade NOI for the DSCR calculator, vs the Census-proxy rent estimate. No PDFs are served (confidential).
app.get('/api/rent-rolls', (req, res) => {
try {
const dir = path.join(ROOT, 'data', 'rent-rolls');
if (!fs.existsSync(dir)) return res.json({ rentRolls: [] });
const rolls = fs.readdirSync(dir).filter(f => /\.json$/i.test(f)).map(f => {
try { const j = JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8'));
return { file: f, source: j.source || f, isSample: !!j.isSample, property: j.property || null,
parsedAt: j.parsedAt || null, unitCount: j.units ? j.units.length : (j.underwriting?.unitCount || 0),
underwriting: j.underwriting || null, units: j.units || [], note: j.note || null }; }
catch (_) { return null; }
}).filter(Boolean).sort((a, b) => (b.parsedAt || '').localeCompare(a.parsedAt || ''));
res.json({ rentRolls: rolls, count: rolls.length });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// FHA / low-rate loans — HUD/FHA multifamily insured mortgages originally <= 3.5%, benchmarked
// against Freddie Mac PMMS. Real public data only ($0). Built by scripts/fetch-fha-loans.js.
// Clean URL + JSON API; the page itself (public/fha-loans.html) and data/fha-loans.json are also
// served by the static middleware below.
app.get(['/fha-loans', '/fha'], (req, res) => res.sendFile(path.join(ROOT, 'public', 'fha-loans.html')));
app.get('/api/fha-loans', (req, res) => {
try {
const file = path.join(ROOT, 'data', 'fha-loans.json');
if (!fs.existsSync(file)) return res.status(503).json({ error: 'data not built — run scripts/fetch-fha-loans.js' });
const j = JSON.parse(fs.readFileSync(file, 'utf8'));
const red = req.query.red === '1' || req.query.red === 'true';
const q = (req.query.q || '').toString().trim().toLowerCase();
let rows = j.rows || [];
if (red) rows = rows.filter(r => r.red);
if (q) rows = rows.filter(r => ((r.property || '') + ' ' + (r.city || '') + ' ' + (r.state || '') + ' ' + (r.zip || '') + ' ' + (r.holder || '')).toLowerCase().includes(q));
res.json({ ...j, rows, count: rows.length });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Broker-history pilot — transaction history + listing-agent enrichment for the top ranked properties.
// Source: LA County Assessor (recording_date = last deed), public MLS search results (listing agent).
// Broker = CURRENT listing agent only; historical sale brokers are not in any free public record
// (deed records carry no agent name — RR/CC confirmed). Pilot covers top ~30 ranked properties.
// Cost: $0 (assessor DB local, WebSearch free). Scale-up requires ATTOM API ($) or CoStar ($).
app.get('/api/broker-history', (req, res) => {
try {
// Prefer the FULL deed-history set (all 1,407 listings) when built; fall back to the 13-record pilot.
const full = path.join(ROOT, 'data', 'broker-history-full.json');
const pilot = path.join(ROOT, 'data', 'broker-history-pilot.json');
const file = fs.existsSync(full) ? full : pilot;
if (!fs.existsSync(file)) return res.json({ properties: [], meta: { scope: 'not built' } });
const j = JSON.parse(fs.readFileSync(file, 'utf8'));
// Index by the listing id + by address for easy lookup from the UI
const byId = {}, byAddr = {};
(j.properties || []).forEach(p => {
if (p.id) byId[p.id] = p;
if (p.address) byAddr[p.address.toLowerCase().trim()] = p;
});
const id = req.query.id ? String(req.query.id) : null;
const addr = req.query.address ? String(req.query.address).toLowerCase().trim() : null;
if (id && byId[id]) return res.json({ property: byId[id], meta: j.meta });
if (addr && byAddr[addr]) return res.json({ property: byAddr[addr], meta: j.meta });
// Return full set (for the filter badge and bulk UI hydration)
res.json({ properties: j.properties, meta: j.meta, flagged: j.properties.filter(p => p.broker_flag).length });
} catch (e) { res.status(502).json({ error: String(e.message).split('\n')[0] }); }
});
// Clean URLs: serve public/<name>.html at /<name> (extension-less), so /mls, /brokers,
// /map, /condos, etc. work alongside the existing .html links in the nav. GET-only, and
// only when the extension-less path has no dot (skips /api, /data, and real asset files).
app.get(/^\/[a-zA-Z0-9_-]+$/, (req, res, next) => {
const name = req.path.slice(1);
const html = path.join(ROOT, 'public', name + '.html');
if (fs.existsSync(html)) return res.sendFile(html);
next();
});
// Serve col-resize.js with revalidate-every-load caching (ahead of the static
// mount) so edits to it land on a normal refresh — no hard-refresh needed.
app.get('/col-resize.js', (req, res) => {
res.set('Cache-Control', 'no-cache');
res.type('application/javascript');
res.sendFile(path.join(ROOT, 'public', 'col-resize.js'));
});
app.use('/data', express.static(path.join(ROOT, 'data')));
app.use('/', express.static(path.join(ROOT, 'public')));
mountClaudeChat(app, { path: '/api/claude-chat', surface: 'sfv-cre', defaultModel: 'claude-sonnet-4-6', systemExtra: SYSTEM_EXTRA });
const server = app.listen(PORT, '127.0.0.1', () => console.log(`SFV CRE viewer + chat -> http://127.0.0.1:${PORT}`));
server.on('error', (err) => {
if (err && err.code === 'EADDRINUSE') {
// Another instance already owns the port (launchd KeepAlive race / manual start).
// Log ONE concise line and exit cleanly instead of throwing a full stack trace,
// which previously spun launchd into a crash-restart loop (1.18M-line err log).
console.error(`serve.js: port ${PORT} already in use — another instance is serving; exiting quietly.`);
process.exit(0);
}
console.error(`serve.js: listen error: ${err && err.message}`);
process.exit(1);
});