← back to Commercialrealestate
scripts/db/brokers-db.js
226 lines
// brokers-db.js — connection + ingest + graph queries for the broker relationship DB (local "cre").
'use strict';
const { Pool } = require('pg');
const pool = new Pool({
host: process.env.CRE_PGHOST || '/tmp', port: +(process.env.CRE_PGPORT || 5432),
// The `cre` DB lives ONLY on the local dev/host box (owner = the mac user). On prod (crcp.agentabrams.com)
// there is NO cre DB, and the process runs as `root` — a role Postgres doesn't have. Guard that: never fall
// through to `root` (the old `process.env.USER` default silently became `root` on prod and 502'd every broker
// endpoint until the snapshot fallback caught it). On prod this connect just fails fast → serve.js serves the
// exported snapshot instead. See export/refresh-broker-snapshot + readBrokerSnap() in serve.js.
database: process.env.CRE_PGDATABASE || 'cre',
user: process.env.CRE_PGUSER || (process.env.USER && process.env.USER !== 'root' ? process.env.USER : 'macstudio3'),
max: 6, idleTimeoutMillis: 10000, connectionTimeoutMillis: 4000
});
pool.on('error', () => {});
async function upsertFirm(name) {
if (!name) return null;
const r = await pool.query(
`INSERT INTO firm(name) VALUES($1) ON CONFLICT(name) DO UPDATE SET name=EXCLUDED.name RETURNING id`, [name.trim()]);
return r.rows[0].id;
}
async function upsertBroker(b) {
const firmId = await upsertFirm(b.firm);
const r = await pool.query(
`INSERT INTO broker(name, firm_id, title, phone, email, crexi_id, source, total_assets)
VALUES($1,$2,$3,$4,$5,$6,$7,$8)
ON CONFLICT(name, firm_id) DO UPDATE SET
phone=COALESCE(EXCLUDED.phone, broker.phone),
email=COALESCE(EXCLUDED.email, broker.email),
crexi_id=COALESCE(EXCLUDED.crexi_id, broker.crexi_id),
title=COALESCE(EXCLUDED.title, broker.title),
total_assets=GREATEST(COALESCE(EXCLUDED.total_assets,0), COALESCE(broker.total_assets,0))
RETURNING id`,
[b.name.trim(), firmId, b.title || null, b.phone || null, b.email || null, b.crexi_id || null, b.source || 'crexi', b.total_assets || null]);
return r.rows[0].id;
}
async function upsertListing(l) {
await pool.query(
`INSERT INTO listing(id, address, city, zip, type, price, cap_rate, units, firm_name, source)
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT(id) DO UPDATE SET firm_name=COALESCE(EXCLUDED.firm_name, listing.firm_name)`,
[l.id, l.address || null, l.city || null, l.zip || null, l.type || null,
l.price || null, l.cap_rate || null, l.units || null, l.firm_name || null, l.source || null]);
return l.id;
}
async function link(brokerId, listingId, role) {
await pool.query(
`INSERT INTO broker_listing(broker_id, listing_id, role) VALUES($1,$2,$3)
ON CONFLICT DO NOTHING`, [brokerId, listingId, role || 'broker']);
}
// The mind-map graph: broker + firm nodes, edges = broker→listing collapsed into broker↔firm and
// broker↔broker (co-listing). Returns {nodes, edges, stats} sized for a force-directed view.
async function graph(limit = 400) {
const brokers = (await pool.query(
`SELECT bn.*, b.website, b.linkedin, b.office_addr, b.total_assets
FROM broker_node bn JOIN broker b ON b.id = bn.id
ORDER BY bn.listings DESC NULLS LAST LIMIT $1`, [limit])).rows;
const ids = new Set(brokers.map(b => b.id));
const co = (await pool.query(`SELECT a, b, shared_listings FROM broker_cobroker`)).rows
.filter(e => ids.has(e.a) && ids.has(e.b));
const firms = {};
brokers.forEach(b => { if (b.firm) firms[b.firm] = (firms[b.firm] || 0) + 1; });
const nodes = [];
Object.entries(firms).forEach(([name, n]) => nodes.push({ id: 'firm:' + name, kind: 'firm', label: name, weight: n }));
brokers.forEach(b => nodes.push({ id: 'broker:' + b.id, kind: 'broker', dbId: b.id, label: b.name, firm: b.firm, listings: +b.listings, total: b.total_assets,
agent_type: b.agent_type, phone: b.phone, email: b.email, website: b.website, linkedin: b.linkedin, office_addr: b.office_addr }));
const edges = [];
brokers.forEach(b => { if (b.firm) edges.push({ source: 'broker:' + b.id, target: 'firm:' + b.firm, kind: 'employs' }); });
co.forEach(e => edges.push({ source: 'broker:' + e.a, target: 'broker:' + e.b, kind: 'colist', weight: e.shared_listings }));
const stats = {
brokers: (await pool.query(`SELECT count(*) c FROM broker`)).rows[0].c,
firms: (await pool.query(`SELECT count(*) c FROM firm`)).rows[0].c,
listings: (await pool.query(`SELECT count(*) c FROM listing`)).rows[0].c,
edges: (await pool.query(`SELECT count(*) c FROM broker_listing`)).rows[0].c
};
return { nodes, edges, stats };
}
// graph() with ?history=1 — additive superset of graph():
// - same nodes (same id scheme, same fields)
// - existing 'employs' edges tagged with current:true
// - 'colist' edges from broker_cobroker (broker↔broker co-listing pairs), capped at 300 pairs
// - 'worked_at' edges from broker_firm_history where is_current=false (empty today — expected)
const COLIST_CAP = 300; // cap co-listing edges to keep graph readable; logged when hit
async function graphHistory(limit = 400) {
// Re-use the same broker + firm node construction as graph()
const brokers = (await pool.query(
`SELECT bn.*, b.website, b.linkedin, b.office_addr, b.total_assets
FROM broker_node bn JOIN broker b ON b.id = bn.id
ORDER BY bn.listings DESC NULLS LAST LIMIT $1`, [limit])).rows;
const ids = new Set(brokers.map(b => b.id));
// ── Colist edges (from broker_cobroker view) ──────────────────────────────
const allCo = (await pool.query(
`SELECT a, b, shared_listings FROM broker_cobroker ORDER BY shared_listings DESC`)).rows
.filter(e => ids.has(e.a) && ids.has(e.b));
const capped = allCo.length > COLIST_CAP;
if (capped) console.log(`[graphHistory] colist edges capped at ${COLIST_CAP} (total available: ${allCo.length})`);
const co = allCo.slice(0, COLIST_CAP);
// ── Past-firm edges (broker_firm_history, is_current=false only) ──────────
const pastRows = (await pool.query(
`SELECT bfh.broker_id, bfh.firm_id, COALESCE(bfh.firm_name, f.name) AS firm_name,
bfh.title, bfh.start_date, bfh.end_date, bfh.source
FROM broker_firm_history bfh
LEFT JOIN firm f ON f.id = bfh.firm_id
WHERE bfh.is_current = false`)).rows;
// ── Build nodes (same shape as graph()) ──────────────────────────────────
const firms = {};
brokers.forEach(b => { if (b.firm) firms[b.firm] = (firms[b.firm] || 0) + 1; });
// Also collect firm names referenced only in history (past firms not in current node set)
const pastFirmNames = new Set();
pastRows.forEach(r => {
if (r.firm_name && ids.has(r.broker_id)) pastFirmNames.add(r.firm_name);
});
pastFirmNames.forEach(name => { if (!firms[name]) firms[name] = 0; });
const nodes = [];
Object.entries(firms).forEach(([name, n]) =>
nodes.push({ id: 'firm:' + name, kind: 'firm', label: name, weight: n }));
brokers.forEach(b =>
nodes.push({ id: 'broker:' + b.id, kind: 'broker', dbId: b.id, label: b.name, firm: b.firm,
listings: +b.listings, total: b.total_assets, agent_type: b.agent_type,
phone: b.phone, email: b.email, website: b.website, linkedin: b.linkedin,
office_addr: b.office_addr }));
// ── Build edges ───────────────────────────────────────────────────────────
const edges = [];
// employs: current firm edges, tagged with kind:'employs' + current:true
brokers.forEach(b => {
if (b.firm) edges.push({ source: 'broker:' + b.id, target: 'firm:' + b.firm, kind: 'employs', current: true });
});
// colist: broker↔broker co-listing pairs (weighted by shared deal count)
co.forEach(e =>
edges.push({ source: 'broker:' + e.a, target: 'broker:' + e.b, kind: 'colist', weight: e.shared_listings }));
// worked_at: past-firm edges for brokers in the current node set
pastRows
.filter(r => ids.has(r.broker_id) && r.firm_name)
.forEach(r =>
edges.push({
source: 'broker:' + r.broker_id,
target: 'firm:' + r.firm_name,
kind: 'worked_at',
current: false,
...(r.title ? { title: r.title } : {}),
...(r.start_date ? { start_date: r.start_date } : {}),
...(r.end_date ? { end_date: r.end_date } : {}),
...(r.source ? { source: r.source } : {})
}));
const stats = {
brokers: (await pool.query(`SELECT count(*) c FROM broker`)).rows[0].c,
firms: (await pool.query(`SELECT count(*) c FROM firm`)).rows[0].c,
listings:(await pool.query(`SELECT count(*) c FROM listing`)).rows[0].c,
edges: (await pool.query(`SELECT count(*) c FROM broker_listing`)).rows[0].c,
colist_total: allCo.length,
colist_capped: capped,
worked_at: pastRows.length
};
return { nodes, edges, stats };
}
async function topBrokers(limit = 50) {
// id + website ride along so snapshot consumers (prod CRCP top-brokers table) can open the
// contact card and drive the "✉ find email" button — broker_node lacks website, hence the join.
return (await pool.query(
`SELECT n.id, n.name, n.firm, n.listings, n.phone, n.email, n.title, n.agent_type, b.website
FROM broker_node n LEFT JOIN broker b ON b.id = n.id
ORDER BY n.listings DESC NULLS LAST LIMIT $1`, [limit])).rows;
}
// Full enriched contact card for one broker (id), with per-field provenance + listing book.
async function brokerContact(id) {
const b = (await pool.query(
`SELECT b.id, b.name, b.title, b.phone, b.email, b.website, b.linkedin, b.office_addr,
b.total_assets, b.enriched_at, b.agent_type, b.license, f.name AS firm
FROM broker b LEFT JOIN firm f ON f.id=b.firm_id WHERE b.id=$1`, [id])).rows[0];
if (!b) return null;
b.provenance = (await pool.query(
`SELECT field, source_url, tier FROM broker_field_source WHERE broker_id=$1`, [id])).rows;
// our_listings spans commercial listings + residential condos (so a residential agent shows their book).
b.our_listings = (await pool.query(
`SELECT l.id, l.address, l.city, l.type, l.price 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, 'Condo' type, c.price FROM broker_condo bc
JOIN condo c ON c.id=bc.condo_id WHERE bc.broker_id=$1
ORDER BY price DESC NULLS LAST LIMIT 50`, [id])).rows;
b.other_listings = (await pool.query(
`SELECT title, address, city, state, price, asset_type, url FROM broker_other_listing
WHERE broker_id=$1 ORDER BY price DESC NULLS LAST LIMIT 50`, [id])).rows;
// closed_listings — the broker's Crexi closed-transaction book (crawl-broker-comps.js). Real deal
// history: address, sold price, sold date, type. Sanity-filter absurd sold_price outliers.
b.closed_listings = (await pool.query(
`SELECT address, city, sold_price, sold_date, type FROM broker_closed_listing
WHERE broker_id=$1 AND (sold_price IS NULL OR sold_price BETWEEN 10000 AND 5000000000)
ORDER BY sold_date DESC NULLS LAST LIMIT 100`, [id])).rows;
b.closed_count = (await pool.query(
`SELECT count(*)::int c FROM broker_closed_listing WHERE broker_id=$1`, [id])).rows[0].c;
return b;
}
// Enrichment coverage rollup for the stats bar.
async function enrichStats() {
return (await pool.query(
`SELECT count(*)::int total,
count(*) FILTER (WHERE phone IS NOT NULL)::int phone,
count(*) FILTER (WHERE email IS NOT NULL)::int email,
count(*) FILTER (WHERE website IS NOT NULL)::int website,
count(*) FILTER (WHERE linkedin IS NOT NULL)::int linkedin,
count(*) FILTER (WHERE phone IS NOT NULL OR email IS NOT NULL)::int contactable
FROM broker`)).rows[0];
}
module.exports = { pool, upsertFirm, upsertBroker, upsertListing, link, graph, graphHistory, topBrokers, brokerContact, enrichStats };