`scripts/db/brokers-db.js`

129 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 };
}

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;
  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, topBrokers, brokerContact, enrichStats };