`db/backfill-broker-of-record.ts`

70 lines
/**
 * TK-10139: seed broker_of_record_history with ONE initial observation per address
 * from the current `listing` table (every listing that has a broker_id or firm_id).
 *
 * Dedupe: for each normalized address_key we keep the MOST-RECENT listing (by last_seen,
 * then first_seen, then id) and insert a single history row observed at that listing's
 * last_seen. Re-runnable: skips any address_key already present in the history table, so
 * running it twice does not double-seed.
 *
 *   npx tsx db/backfill-broker-of-record.ts
 */
import 'dotenv/config';
import { pool, query } from './pool.ts';
import { addressKeySql } from '../src/server/broker-of-record.ts';

// CANONICAL address_key — uses the SINGLE shared normalizer (addressKeySql), identical to
// computeAddressKey() in src/server/broker-of-record.ts, so backfilled keys match the recorder,
// the listings-ingest engine, and the commercial + listings API joins (state/zip stripped).
// NOTE: the `listing` table has NO separate city column — its `address` is already the
// full "street, city, state, zip" string — so address_key = normalized full address.
const ADDRESS_KEY_SQL = addressKeySql(`coalesce(l.address,'')`);

async function main() {
  const before = await query<{ n: string }>(`SELECT count(*)::text n FROM broker_of_record_history`);
  console.log(`[backfill] history rows before: ${before.rows[0].n}`);

  const res = await query<{ id: number }>(
    `WITH ranked AS (
       SELECT
         l.*,
         ${ADDRESS_KEY_SQL} AS address_key,
         rg.state_code       AS rg_state,
         f.name              AS firm_name,
         ROW_NUMBER() OVER (
           PARTITION BY ${ADDRESS_KEY_SQL}
           ORDER BY l.last_seen DESC NULLS LAST, l.first_seen DESC NULLS LAST, l.id DESC
         ) AS rn
       FROM listing l
       LEFT JOIN firm f   ON f.id = l.firm_id
       LEFT JOIN region rg ON rg.id = l.region_id
       WHERE (l.broker_id IS NOT NULL OR l.firm_id IS NOT NULL)
         AND btrim(coalesce(l.address,'')) <> ''
     )
     INSERT INTO broker_of_record_history
       (address_key, parcel_key, county_fips, ain, address, city, state_code,
        broker_id, broker_name, firm_id, firm_name, source, source_url, observed_at)
     SELECT
       r.address_key, NULL, NULL, NULL, r.address, NULL, r.rg_state,
       r.broker_id, NULL, r.firm_id, r.firm_name, r.source, r.url,
       COALESCE(r.last_seen, r.first_seen, now())
     FROM ranked r
     WHERE r.rn = 1
       AND NOT EXISTS (
         SELECT 1 FROM broker_of_record_history h WHERE h.address_key = r.address_key
       )
     RETURNING id`,
  );

  const after = await query<{ n: string }>(`SELECT count(*)::text n FROM broker_of_record_history`);
  console.log(`[backfill] inserted: ${res.rows.length}`);
  console.log(`[backfill] history rows after:  ${after.rows[0].n}`);

  await pool.end();
}

main().catch((e) => {
  console.error(e);
  process.exit(1);
});