← back to Nationalrealestate
src/server/index.ts
594 lines
/**
* USRealEstate server — Express on :9913, whole-site Basic Auth (CRCP pattern).
* /healthz stays OPEN so deploy smoke-tests + uptime canaries get a 200 without creds.
*/
import 'dotenv/config';
import express from 'express';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { query } from '../../db/pool.ts';
import { mountWatchlistAdmin } from './watchlist_admin.ts';
import { mountProperty } from './property.ts';
import { mountListings } from './listings.ts';
import { mountCommercial } from './commercial.ts';
import { mountSublease } from './sublease.ts';
import { mountParcels } from './parcels.ts';
import { mountDeals } from './deals.ts';
import { mountContractors } from './contractors.ts';
const __dirname = dirname(fileURLToPath(import.meta.url));
const ROOT = join(__dirname, '..', '..');
const PORT = Number(process.env.PORT || 9913);
const app = express();
app.use(express.json({ limit: '256kb' }));
app.get('/healthz', (_req, res) => res.type('text').send('ok'));
// Whole-site Basic Auth — the default admin login, named stakeholder logins, and any
// extra comma-separated user:pass pairs via BASIC_AUTH_EXTRA. Accepts a SET of creds
// (was single-user) so more than one person can be granted access.
const AUTH_PAIRS = [
`${process.env.USRE_USER || 'admin'}:${process.env.USRE_PASS || 'DW2024!'}`,
'Daniel:Dandeana1992',
'Boomer:rentfree911',
'Scott:Russia1980',
].concat((process.env.BASIC_AUTH_EXTRA || '').split(',').map((s) => s.trim()).filter(Boolean));
const ACCEPTED = new Set(AUTH_PAIRS.map((p) => 'Basic ' + Buffer.from(p).toString('base64')));
// Dedicated, SCOPED credential for the shareable Contractors browse page.
// Set CONTRACTORS_USER + CONTRACTORS_PASS in .env to enable. This login grants
// access to /contractors* + /api/contractors + the shared nav chrome ONLY — it
// can never reach the rest of usre admin (deals, parcels, listings, market data).
// If the env vars are unset, behavior is identical to before (no new access).
const CONTRACTOR_TOKEN =
process.env.CONTRACTORS_USER && process.env.CONTRACTORS_PASS
? 'Basic ' +
Buffer.from(`${process.env.CONTRACTORS_USER}:${process.env.CONTRACTORS_PASS}`).toString('base64')
: null;
const CONTRACTOR_ASSETS = new Set(['/contractors', '/contractors.html', '/nav-drawer.css', '/nav-drawer.js']);
const contractorScopeAllows = (p: string) =>
CONTRACTOR_ASSETS.has(p) || p.startsWith('/contractors/') || p.startsWith('/api/contractors');
app.use((req, res, next) => {
const auth = req.headers.authorization || '';
if (ACCEPTED.has(auth)) return next(); // full usre logins → everywhere
if (CONTRACTOR_TOKEN && auth === CONTRACTOR_TOKEN && contractorScopeAllows(req.path)) return next(); // scoped
res.set('WWW-Authenticate', 'Basic realm="USRealEstate", charset="UTF-8"');
return res.status(401).send('Authentication required');
});
app.get('/api/health', async (_req, res) => {
try {
const r = await query<{ region_type: string; n: string }>(
`SELECT region_type, COUNT(*)::text AS n FROM region GROUP BY region_type`,
);
const regions: Record<string, number> = {};
for (const row of r.rows) regions[row.region_type] = Number(row.n);
const fresh = await query<{ d: string | null }>(
`SELECT to_char(MAX(period),'YYYY-MM-DD') AS d FROM metric_series WHERE metric='zhvi'`,
);
res.json({ ok: true, regions, data_through: fresh.rows[0]?.d ?? null });
} catch (e: any) {
res.status(500).json({ ok: false, error: String(e.message || e) });
}
});
// Self-correcting engine status — per-source health + recent events + Places quota.
app.get('/api/ingest-health', async (_req, res) => {
try {
const sources = await query(
`SELECT sh.job, sh.state, sh.consecutive_failures, sh.baseline_upserted, sh.last_upserted,
to_char(sh.last_ok_at,'YYYY-MM-DD HH24:MI') AS last_ok_at,
to_char(sh.last_failure_at,'YYYY-MM-DD HH24:MI') AS last_failure_at,
to_char(sh.quarantined_at,'YYYY-MM-DD HH24:MI') AS quarantined_at,
ROUND(EXTRACT(EPOCH FROM (NOW()-sh.last_ok_at))/3600, 1) AS ok_age_h
FROM source_health sh ORDER BY
CASE sh.state WHEN 'quarantined' THEN 0 WHEN 'degraded' THEN 1
WHEN 'retrying' THEN 2 WHEN 'healthy' THEN 3 ELSE 4 END, sh.job`);
const events = await query(
`SELECT job, event, detail, to_char(at,'MM-DD HH24:MI') AS at
FROM source_events ORDER BY id DESC LIMIT 20`);
const pq = await query<{ year_month: string; calls_used: number }>(
`SELECT year_month, calls_used FROM places_quota ORDER BY year_month DESC LIMIT 1`);
// Per-state broker roster health: latest run status/age + live row count per state source.
const brokerStates = await query(
`SELECT s.source, s.brokers,
(SELECT status FROM ingest_runs i WHERE i.source = s.source ORDER BY started_at DESC LIMIT 1) AS last_status,
to_char((SELECT MAX(started_at) FROM ingest_runs i WHERE i.source = s.source AND status='ok'),'MM-DD HH24:MI') AS last_ok,
ROUND(EXTRACT(EPOCH FROM (NOW()-(SELECT MAX(started_at) FROM ingest_runs i WHERE i.source = s.source AND status='ok')))/86400, 1) AS ok_age_d
FROM (SELECT source, COUNT(*)::int AS brokers FROM broker GROUP BY source) s
ORDER BY s.source`);
// Places geographic coverage: metros seeded per West→East tier.
const coverage = await query(
`WITH m AS (SELECT substring(name from ',\\s+([A-Z]{2}(-[A-Z]{2})*)\\s') AS st,
(SELECT COUNT(*) FROM places_seed s WHERE s.region_id=r.id) AS seeds
FROM region r WHERE region_type='metro')
SELECT
CASE WHEN st ~ '(^|-)(CA|OR|WA)($|-)' THEN '1·West Coast'
WHEN st ~ '(^|-)(NV|AZ|ID|UT)($|-)' THEN '2·Interior West'
WHEN st ~ '(^|-)(MT|WY|CO|NM)($|-)' THEN '3·Mountain'
ELSE '4·Rest of US' END AS tier,
COUNT(*) AS metros, COUNT(*) FILTER (WHERE seeds>0) AS started,
COUNT(*) FILTER (WHERE seeds>=60) AS covered
FROM m GROUP BY tier ORDER BY tier`);
res.json({
ok: true,
generated_at: new Date().toISOString(),
sources: sources.rows,
broker_states: brokerStates.rows,
coverage: coverage.rows,
events: events.rows,
places_quota: { ...(pq.rows[0] || { year_month: null, calls_used: 0 }), cap: Number(process.env.PLACES_MONTHLY_CAP || 1500) },
});
} catch (e: any) {
res.status(500).json({ ok: false, error: String(e.message || e) });
}
});
app.get('/health', (_req, res) => res.sendFile(join(ROOT, 'public', 'health.html')));
// Commercial deals feed + clean per-deal detail paths (TK-10482). The deals.html
// SPA reads the deal id from ?id=; /deal/123 and /deals/123 redirect into it so
// each deal is addressable by a clean URL (href-drill: no dead-end data points).
app.get('/deals', (_req, res) => res.sendFile(join(ROOT, 'public', 'deals.html')));
app.get(['/deal/:id', '/deals/:id'], (req, res) => {
const id = String(req.params.id || '').replace(/[^0-9]/g, '').slice(0, 18);
return res.redirect(302, id ? `/deals.html?id=${id}` : '/deals.html');
});
app.get('/api/regions', async (req, res) => {
try {
const type = String(req.query.type || 'county');
const r = await query(
`SELECT id, region_type, canonical_key, name, state_code, fips, cbsa_code, lat, lng, population
FROM region WHERE region_type = $1 ORDER BY name`,
[type],
);
res.json({ regions: r.rows });
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// parcel-covered counties (parcel_source registry), cached — lets geo-search deep-link
// an address to the property page only where we actually have parcels.
let _coveredCache: { at: number; set: Set<string> } | null = null;
async function coveredCounties(): Promise<Set<string>> {
if (_coveredCache && Date.now() - _coveredCache.at < 10 * 60_000) return _coveredCache.set;
try {
const r = await query<{ county_fips: string }>(`SELECT county_fips FROM parcel_source`);
_coveredCache = { at: Date.now(), set: new Set(r.rows.map(x => x.county_fips)) };
} catch { _coveredCache = { at: Date.now(), set: new Set() }; }
return _coveredCache.set;
}
// universal location search — resolves a typed county / city / metro / state to a
// market page. National coverage via the region table (fast; ~4.2k rows). ZIP + exact
// street address are county-scoped and live on the property page (no national ZIP index).
app.get('/api/geo-search', async (req, res) => {
try {
const q = String(req.query.q || '').trim();
if (q.length < 2) return res.json({ q, results: [] });
// ZIP → primary county (zip_county crosswalk). National.
if (/^\d{5}$/.test(q)) {
const r = await query<{ canonical_key: string; name: string; state_code: string | null }>(
`SELECT rg.canonical_key, rg.name, rg.state_code
FROM zip_county zc JOIN region rg ON rg.fips = zc.county_fips AND rg.region_type = 'county'
WHERE zc.zip = $1`, [q]);
if (r.rows.length) { const x = r.rows[0];
return res.json({ q, results: [{ type: 'zip', name: `${x.name} — ZIP ${q}`, state: x.state_code || null,
key: x.canonical_key, url: '/market.html?key=' + encodeURIComponent(x.canonical_key) }] }); }
return res.json({ q, results: [] });
}
// street address (starts with a house number) → Census geocoder → county. National.
if (/^\d+\s+\S/.test(q)) {
try {
const gu = 'https://geocoding.geo.census.gov/geocoder/geographies/onelineaddress?address='
+ encodeURIComponent(q) + '&benchmark=Public_AR_Current&vintage=Current_Current&format=json';
const gj: any = await (await fetch(gu, { signal: AbortSignal.timeout(6000) })).json();
const m = gj?.result?.addressMatches?.[0];
const co = m?.geographies?.Counties?.[0];
if (co?.GEOID) {
const r = await query<{ canonical_key: string; name: string; state_code: string | null }>(
`SELECT canonical_key, name, state_code FROM region WHERE fips = $1 AND region_type = 'county'`, [co.GEOID]);
if (r.rows.length) { const x = r.rows[0];
// property-level intent: if the county has parcel coverage, deep-link into the
// property page prefilled with the address; else fall back to the county market.
const covered = await coveredCounties();
const hasParcels = covered.has(co.GEOID);
const url = hasParcels
? '/property.html?county=' + encodeURIComponent(co.GEOID) + '&q=' + encodeURIComponent(q)
: '/market.html?key=' + encodeURIComponent(x.canonical_key);
return res.json({ q, results: [{ type: 'address', name: m.matchedAddress || q, state: x.state_code || null,
sub: hasParcels ? x.name + ' · view property →' : x.name, key: x.canonical_key, url }] }); }
}
} catch (e) { /* geocoder timeout/failure → fall through to empty (client shows a hint) */ }
return res.json({ q, results: [] });
}
const esc = q.replace(/[%_\\]/g, '\\$&');
const like = '%' + esc + '%';
const starts = esc + '%';
const r = await query<{ region_type: string; canonical_key: string; name: string; state_code: string | null }>(
`SELECT region_type, canonical_key, name, state_code
FROM region
WHERE name ILIKE $1 ESCAPE '\\'
OR (char_length($2) = 2 AND region_type = 'state' AND state_code = upper($2))
ORDER BY (char_length($2) = 2 AND region_type = 'state' AND state_code = upper($2)) DESC,
(name ILIKE $3 ESCAPE '\\') DESC,
CASE region_type WHEN 'county' THEN 0 WHEN 'metro' THEN 1 ELSE 2 END,
population DESC NULLS LAST
LIMIT 12`,
[like, q, starts],
);
res.json({ q, results: r.rows.map(x => ({
type: x.region_type, name: x.name, state: x.state_code || null,
key: x.canonical_key, url: '/market.html?key=' + encodeURIComponent(x.canonical_key),
})) });
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// choropleth data: {fips: value} for the latest period of a metric (M2 fills metric_series)
app.get('/api/choropleth', async (req, res) => {
try {
const metric = String(req.query.metric || 'zhvi');
const r = await query<{ fips: string; value: string }>(
`SELECT rg.fips, ms.value::text AS value
FROM metric_series ms
JOIN region rg ON rg.id = ms.region_id AND rg.region_type = 'county'
WHERE ms.metric = $1
AND ms.period = (SELECT MAX(period) FROM metric_series WHERE metric = $1)`,
[metric],
);
const out: Record<string, number> = {};
for (const row of r.rows) if (row.fips) out[row.fips] = Number(row.value);
res.json({ metric, count: r.rows.length, values: out });
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// market detail: full 36mo series (all sources) + latest per metric for one region
app.get('/api/market', async (req, res) => {
try {
const key = String(req.query.key || '');
const rg = await query(
`SELECT id, region_type, canonical_key, name, state_code, fips, cbsa_code, lat, lng, population
FROM region WHERE canonical_key = $1`,
[key],
);
if (!rg.rows.length) return res.status(404).json({ error: 'region not found: ' + key });
const region = rg.rows[0];
const r = await query<{ source: string; metric: string; period: string; value: string }>(
`SELECT source, metric, to_char(period, 'YYYY-MM-DD') AS period, value::text AS value
FROM metric_series
WHERE region_id = $1
AND (period >= (CURRENT_DATE - INTERVAL '36 months') OR source = 'acs')
ORDER BY metric, period`,
[region.id],
);
const series: Record<string, [string, number][]> = {};
const latest: Record<string, number> = {};
for (const row of r.rows) {
(series[row.metric] ||= []).push([row.period, Number(row.value)]);
latest[row.metric] = Number(row.value); // rows are period-ASC per metric, so last write wins
}
res.json({ region, series, latest });
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
const MARKETS_METRICS = [
'zhvi', 'zhvi_yoy', 'median_sale_price', 'median_sale_price_yoy', 'inventory', 'dom',
'months_of_supply', 'sale_to_list', 'median_hh_income', 'rent_yield', 'affordability',
];
// ranking dashboard: one row per region with latest score + latest key metrics.
// The latest-per-metric scan covers 1.2M rows (~1.8s) — cache per type; data changes monthly.
const marketsCache = new Map<string, { at: number; body: any }>();
const MARKETS_TTL_MS = 5 * 60_000;
app.get('/api/markets', async (req, res) => {
try {
const type = String(req.query.type || 'county');
if (type !== 'county' && type !== 'metro' && type !== 'state') return res.status(400).json({ error: 'type must be county|metro|state' });
const hit = marketsCache.get(type);
if (hit && Date.now() - hit.at < MARKETS_TTL_MS) return res.json(hit.body);
const metricsQ = query<{ region_id: number; metric: string; value: string }>(
`SELECT DISTINCT ON (ms.region_id, ms.metric) ms.region_id, ms.metric, ms.value::text AS value
FROM metric_series ms
JOIN region rg ON rg.id = ms.region_id AND rg.region_type = $1
WHERE ms.metric = ANY($2)
ORDER BY ms.region_id, ms.metric, ms.period DESC`,
[type, MARKETS_METRICS],
);
const scoresQ = query<{ region_id: number; opportunity_score: string; rank: number; components: any }>(
`SELECT rs.region_id, rs.opportunity_score::text AS opportunity_score, rs.rank, rs.components
FROM region_scores rs
JOIN region rg ON rg.id = rs.region_id AND rg.region_type = $1
WHERE rs.score_date = (SELECT MAX(rs2.score_date) FROM region_scores rs2
JOIN region rg2 ON rg2.id = rs2.region_id AND rg2.region_type = $1)`,
[type],
);
const regionsQ = query(
`SELECT id, canonical_key, name, state_code, fips, cbsa_code, population
FROM region WHERE region_type = $1`,
[type],
);
const [metrics, scores, regions] = await Promise.all([metricsQ, scoresQ, regionsQ]);
const metricMap = new Map<number, Record<string, number>>();
for (const m of metrics.rows) {
let o = metricMap.get(m.region_id);
if (!o) { o = {}; metricMap.set(m.region_id, o); }
o[m.metric] = Number(m.value);
}
const scoreMap = new Map(scores.rows.map(s => [s.region_id, s]));
const rows = regions.rows.map((rg: any) => {
const m = metricMap.get(rg.id) || {};
const s = scoreMap.get(rg.id);
const sub = s?.components?.sub_scores || {};
return {
key: rg.canonical_key, name: rg.name, state: rg.state_code, population: rg.population,
opportunity_score: s ? Number(s.opportunity_score) : null,
rank: s?.rank ?? null,
momentum: sub.momentum ?? null, affordability_score: sub.affordability ?? null,
rent_yield_score: sub.rent_yield ?? null, inventory_trend: sub.inventory_trend ?? null,
velocity: sub.velocity ?? null,
data_completeness: s?.components?.data_completeness ?? null,
thin_market: s?.components?.thin_market ?? false,
zhvi: m.zhvi ?? null, zhvi_yoy: m.zhvi_yoy ?? null,
median_sale_price: m.median_sale_price ?? null, median_sale_price_yoy: m.median_sale_price_yoy ?? null,
inventory: m.inventory ?? null, dom: m.dom ?? null,
months_of_supply: m.months_of_supply ?? null, sale_to_list: m.sale_to_list ?? null,
median_hh_income: m.median_hh_income ?? null, rent_yield: m.rent_yield ?? null,
affordability: m.affordability ?? null,
};
});
const body = { type, count: rows.length, rows };
marketsCache.set(type, { at: Date.now(), body });
res.json(body);
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// ---- M-B1 broker registry ----
// paginated broker directory: ?state=&q=&firm=&limit=&offset=
app.get('/api/brokers', async (req, res) => {
try {
const limit = Math.min(Math.max(Number(req.query.limit) || 500, 1), 1000);
const offset = Math.max(Number(req.query.offset) || 0, 0);
const conds: string[] = [];
const params: unknown[] = [];
if (req.query.state) { params.push(String(req.query.state).toUpperCase()); conds.push(`b.license_state = $${params.length}`); }
if (req.query.city) { params.push(String(req.query.city).toUpperCase()); conds.push(`upper(b.city) = $${params.length}`); }
if (req.query.q) { params.push('%' + String(req.query.q) + '%'); conds.push(`(b.name ILIKE $${params.length} OR f.name ILIKE $${params.length} OR b.license_no ILIKE $${params.length})`); }
if (req.query.firm) { params.push('%' + String(req.query.firm) + '%'); conds.push(`f.name ILIKE $${params.length}`); }
if (req.query.status) {
const s = String(req.query.status).toLowerCase();
// convenience: ?status=active matches ACTIVE + "ACTIVE UNDER REVIEW"; else exact-ish match
if (s === 'active') { conds.push(`b.license_status ILIKE 'active%'`); }
else { params.push(s); conds.push(`lower(b.license_status) = $${params.length}`); }
}
// An agent only appears if it's associated with a broker/firm (has an employing
// firm_id). Drops standalone DRE licensees — incl. "Licensed NBA" (No Broker
// Affiliation) — so this returns real practicing brokers, not the full DRE list.
// Default ON for every consumer (desk, directory, etc.); ?unaffiliated=1 opts back
// into the full registry.
if (String(req.query.unaffiliated || '') !== '1') { conds.push(`b.firm_id IS NOT NULL`); }
// asset_class=commercial|residential — RENTV (CRE) requests commercial; usreal/CRCP request residential.
if (req.query.asset_class) { params.push(String(req.query.asset_class).toLowerCase()); conds.push(`b.asset_class = $${params.length}`); }
const where = conds.length ? 'WHERE ' + conds.join(' AND ') : '';
// whitelisted server-side sort (TK-10590) — covers ALL rows, not just the page. Values are
// fixed literals (never user strings) so no injection surface; unknown sort → default name order.
const BROKER_SORT: Record<string,string> = { name:'b.name, b.id', firm:'f.name NULLS LAST, b.name', city:'b.city NULLS LAST, b.name', license:'b.license_no NULLS LAST, b.name' };
const brokerOrder = BROKER_SORT[String(req.query.sort || '').toLowerCase()] || 'b.name, b.id';
params.push(limit, offset);
const r = await query(
`SELECT b.id, b.name, b.license_no, b.license_state, b.license_type, b.license_status, b.asset_class,
b.city, b.state_code, b.source, f.name AS firm_name, f.id AS firm_id,
b.phone AS broker_phone, b.email AS broker_email, b.website AS broker_website, b.website_status,
COALESCE(fs.url, f.website) AS firm_website,
f.street_address AS firm_address,
-- firm-level contact fallback (the agent's own phone/email is usually blank on the DRE
-- registry). Prefer the registry-grade firm.phone (Google Places resolve, TK-10669);
-- fall back to the firm's crawled contact number.
COALESCE(NULLIF(f.phone,''), (SELECT value FROM firm_contacts WHERE firm_id = f.id AND kind = 'phone' ORDER BY id LIMIT 1)) AS firm_phone,
(SELECT value FROM firm_contacts WHERE firm_id = f.id AND kind = 'email' ORDER BY id LIMIT 1) AS firm_email,
-- TK-10687 "every broker must have a phone": single resolved contact for the desk card.
-- Broker's direct line first, then the firm's registry phone, then its crawled contact
-- number. phone_tier tells the UI whether it's the broker's own line or the firm fallback.
COALESCE(NULLIF(b.phone,''), NULLIF(f.phone,''), (SELECT value FROM firm_contacts WHERE firm_id = f.id AND kind = 'phone' ORDER BY id LIMIT 1)) AS phone,
CASE WHEN NULLIF(b.phone,'') IS NOT NULL THEN 'direct'
WHEN NULLIF(f.phone,'') IS NOT NULL THEN 'firm'
WHEN (SELECT 1 FROM firm_contacts WHERE firm_id = f.id AND kind = 'phone' LIMIT 1) IS NOT NULL THEN 'firm'
ELSE NULL END AS phone_tier,
COALESCE(NULLIF(b.email,''), NULLIF(f.email,''), (SELECT value FROM firm_contacts WHERE firm_id = f.id AND kind = 'email' ORDER BY id LIMIT 1)) AS email,
f.street_address AS address,
COUNT(*) OVER()::int AS total
FROM broker b
LEFT JOIN firm f ON f.id = b.firm_id
LEFT JOIN firm_site fs ON fs.firm_id = f.id
${where}
ORDER BY ${brokerOrder}
LIMIT $${params.length - 1} OFFSET $${params.length}`,
params,
);
const total = r.rows.length ? Number(r.rows[0].total) : 0;
res.json({ total, limit, offset, rows: r.rows.map(({ total: _t, ...row }: any) => row) });
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// firm directory: ?state=&q=&limit=&offset=
app.get('/api/firms', async (req, res) => {
try {
const limit = Math.min(Math.max(Number(req.query.limit) || 500, 1), 1000);
const offset = Math.max(Number(req.query.offset) || 0, 0);
const conds: string[] = [];
const params: unknown[] = [];
if (req.query.state) { params.push(String(req.query.state).toUpperCase()); conds.push(`license_state = $${params.length}`); }
if (req.query.city) { params.push(String(req.query.city).toUpperCase()); conds.push(`upper(hq_city) = $${params.length}`); }
if (req.query.q) { params.push('%' + String(req.query.q) + '%'); conds.push(`(name ILIKE $${params.length} OR license_no ILIKE $${params.length})`); }
// asset_class=commercial|residential — RENTV (CRE) requests commercial; usreal/CRCP request residential.
if (req.query.asset_class) { params.push(String(req.query.asset_class).toLowerCase()); conds.push(`asset_class = $${params.length}`); }
const where = conds.length ? 'WHERE ' + conds.join(' AND ') : '';
// whitelisted server-side sort (TK-10590) — covers ALL rows, not just the page. Fixed literals only.
const FIRM_SORT: Record<string,string> = { name:'name', firm:'name', agents:'agent_count DESC NULLS LAST, name', city:'hq_city NULLS LAST, name' };
const firmOrder = FIRM_SORT[String(req.query.sort || '').toLowerCase()] || 'agent_count DESC NULLS LAST, name';
params.push(limit, offset);
const r = await query(
`SELECT id, name, license_no, license_state, hq_city, hq_state, agent_count, source, asset_class,
website, phone, street_address,
(SELECT value FROM firm_contacts WHERE firm_id = firm.id AND kind = 'email' ORDER BY id LIMIT 1) AS email,
COUNT(*) OVER()::int AS total
FROM firm
${where}
ORDER BY ${firmOrder}
LIMIT $${params.length - 1} OFFSET $${params.length}`,
params,
);
const total = r.rows.length ? Number(r.rows[0].total) : 0;
res.json({ total, limit, offset, rows: r.rows.map(({ total: _t, ...row }: any) => row) });
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// per-state coverage counts for the brokers.html header bar (+ M-B2 site coverage)
app.get('/api/broker-stats', async (_req, res) => {
try {
const [b, f, s] = await Promise.all([
query<{ license_state: string; n: string }>(`SELECT license_state, COUNT(*)::text AS n FROM broker GROUP BY 1 ORDER BY 1`),
query<{ license_state: string; n: string }>(`SELECT license_state, COUNT(*)::text AS n FROM firm GROUP BY 1 ORDER BY 1`),
query<{ discovered: string; with_url: string; crawled: string; idx: string; contacts: string }>(
`SELECT COUNT(*)::text AS discovered,
COUNT(url)::text AS with_url,
COUNT(*) FILTER (WHERE http_status IS NOT NULL)::text AS crawled,
COUNT(*) FILTER (WHERE has_idx_listings)::text AS idx,
(SELECT COUNT(*) FROM firm_contacts)::text AS contacts
FROM firm_site`),
]);
const firms: Record<string, number> = {};
for (const row of f.rows) firms[row.license_state] = Number(row.n);
const states = b.rows.map(row => ({ state: row.license_state, brokers: Number(row.n), firms: firms[row.license_state] || 0 }));
const sc = s.rows[0];
res.json({
states,
total_brokers: states.reduce((s, x) => s + x.brokers, 0),
total_firms: Object.values(firms).reduce((s, x) => s + x, 0),
site_coverage: {
firms_discovered: Number(sc.discovered),
firms_with_url: Number(sc.with_url),
firms_crawled: Number(sc.crawled),
firms_with_idx: Number(sc.idx),
contact_rows: Number(sc.contacts),
},
});
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// M-B2 firm detail: firm + discovered site + contacts + its brokers (limit 50)
app.get('/api/firm/:id', async (req, res) => {
try {
const id = Number(req.params.id);
if (!Number.isInteger(id) || id <= 0) return res.status(400).json({ error: 'bad firm id' });
const [firm, site, contacts, brokers] = await Promise.all([
query(`SELECT id, name, website, phone, street_address, hq_city, hq_state, license_no, license_state,
source, agent_count, asset_class, created_at
FROM firm WHERE id = $1`, [id]),
query(`SELECT url, discovery_method, discovered_at, http_status, title, has_idx_listings,
crawl_status, crawled_at, screenshot_path
FROM firm_site WHERE firm_id = $1`, [id]),
query(`SELECT kind, value, source_url, found_at FROM firm_contacts
WHERE firm_id = $1 ORDER BY kind, value`, [id]),
query(`SELECT id, name, license_no, license_type, license_status, city, state_code, asset_class
FROM broker WHERE firm_id = $1 ORDER BY name LIMIT 50`, [id]),
]);
if (!firm.rows.length) return res.status(404).json({ error: 'firm not found: ' + id });
// Class-gated drill: a commercial-only consumer (RENTV) that passes ?asset_class=commercial
// must not be able to open a residential firm's detail/roster by a hand-crafted id (TK-10535).
if (req.query.asset_class && String(firm.rows[0].asset_class) !== String(req.query.asset_class).toLowerCase()) {
return res.status(404).json({ error: 'firm not found: ' + id });
}
res.json({
firm: firm.rows[0],
site: site.rows[0] || null,
contacts: contacts.rows,
brokers: brokers.rows,
});
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// On-demand agent-website discovery — the $0 lazy path the broker registry fires
// when a specific agent record is opened and website_status is still NULL. One free
// SERP query + local-LLM identity gate; a throttle returns {status:'throttled'} so
// the UI can say "check back" without poisoning the record. Never sweeps.
app.post('/api/broker/:id/discover', async (req, res) => {
try {
const id = Number(req.params.id);
if (!Number.isInteger(id) || id <= 0) return res.status(400).json({ error: 'bad broker id' });
const { discoverOne } = await import('../enrich/broker_website_discovery.ts');
const out = await discoverOne(id);
res.json(out);
} catch (e: any) {
res.status(500).json({ error: String(e.message || e) });
}
});
// CSV export of the ranked markets table (analyst pull) — reuses the same
// /api/markets cache-backed dataset via an internal fetch so the shape never drifts.
app.get('/api/markets.csv', async (req, res) => {
try {
const type = String(req.query.type || 'county');
if (type !== 'county' && type !== 'metro' && type !== 'state') return res.status(400).send('bad type');
const r = await fetch(`http://127.0.0.1:${PORT}/api/markets?type=${type}`, {
headers: { authorization: req.headers.authorization || '' },
});
const { rows } = await r.json() as { rows: Record<string, unknown>[] };
if (!rows?.length) return res.status(404).send('no data');
const cols = ['rank', 'name', 'state', 'opportunity_score', 'data_completeness', 'thin_market',
'zhvi', 'zhvi_yoy', 'median_sale_price', 'median_sale_price_yoy', 'inventory', 'dom',
'months_of_supply', 'sale_to_list', 'median_hh_income', 'rent_yield', 'affordability', 'population'];
const esc = (v: unknown) => {
if (v == null) return '';
const s = String(v);
return /[",\n]/.test(s) ? '"' + s.replace(/"/g, '""') + '"' : s;
};
const csv = [cols.join(',')]
.concat(rows.map(row => cols.map(c => esc(row[c])).join(',')))
.join('\n');
res.setHeader('Content-Type', 'text/csv; charset=utf-8');
res.setHeader('Content-Disposition', `attachment; filename="usrealestate-${type}-markets.csv"`);
res.send(csv);
} catch (e: any) {
res.status(500).send('error: ' + String(e.message || e));
}
});
mountWatchlistAdmin(app);
mountProperty(app);
mountListings(app);
mountCommercial(app);
mountSublease(app);
mountParcels(app);
mountDeals(app);
mountContractors(app);
app.use(express.static(join(ROOT, 'public')));
app.listen(PORT, () => {
console.log(`USRealEstate listening on http://127.0.0.1:${PORT}`);
});