`src/server/broker-of-record.ts`

177 lines
/**
 * TK-10139: broker-of-record recorder.
 *
 * Append-only change log — every time a broker/firm of record is written for an
 * address (or parcel), we record it IF it differs from the most-recent observation
 * for that key. This is what turns the table into a history of every UPDATE.
 *
 * Keyed by a normalized address_key ("address, city" lowercased/trimmed/collapsed),
 * plus an optional parcel_key ("county_fips:ain") when a stable parcel identity exists.
 *
 * Callable both from server request handlers and from ingest/backfill scripts —
 * takes a db handle (the pool's `query` fn) so it doesn't hard-import a pool.
 */

type QueryFn = <T extends Record<string, any> = any>(
  text: string,
  params?: unknown[],
) => Promise<{ rows: T[] }>;

export interface BrokerOfRecordInput {
  address?: string | null;
  city?: string | null;
  state_code?: string | null;
  county_fips?: string | null;
  ain?: string | null;
  broker_id?: number | null;
  broker_name?: string | null;
  firm_id?: number | null;
  firm_name?: string | null;
  source?: string | null;
  source_url?: string | null;
}

/**
 * CANONICAL address_key normalization — the SINGLE source of truth, imported by the
 * recorder, the backfill, the listings-ingest engine, and the commercial + listings API
 * joins so they can never drift apart.
 *
 * Goal: a listing's full "street, city, ST, ZIP" string and a commercial_parcel's separate
 * (address="street", city="city") fields for the SAME street both reduce to the SAME key,
 * e.g. both "3601 Roosevelt Ave, San Antonio, TX, 78225" and ("3601 Roosevelt Ave","San Antonio")
 * → "3601 roosevelt ave, san antonio".
 *
 * Steps:
 *   1. combine address + optional city as "address, city"
 *   2. lowercase, trim, collapse internal whitespace
 *   3. strip trailing state/zip tokens from the END — a 2-letter US state and/or a 5(-4) ZIP,
 *      whether comma-separated (", tx, 78225") or space-separated ("tx 78225"), repeatedly.
 *
 * IMPORTANT: the matching SQL in `addressKeySql()` MUST stay behaviorally identical.
 */
export function computeAddressKey(address?: string | null, city?: string | null): string {
  const a = (address || '').trim();
  const c = (city || '').trim();
  const combined = c ? `${a}, ${c}` : a;
  let k = combined.toLowerCase().replace(/\s+/g, ' ').trim();
  // strip trailing ZIP (5 or 5-4), then trailing 2-letter state, comma- OR space-separated,
  // repeatedly so ", tx, 78225" / " tx 78225" / ", tx" / ", 78225" all fall off.
  let prev: string;
  do {
    prev = k;
    k = k.replace(/[\s,]+\d{5}(?:-\d{4})?$/, '');   // trailing zip
    k = k.replace(/[\s,]+[a-z]{2}$/, '');           // trailing 2-letter state
  } while (k !== prev);
  return k.trim().replace(/[\s,]+$/, '');
}

/**
 * SQL expression producing the SAME canonical key as computeAddressKey(), given a raw
 * "address, city" text expression `expr`. Callers build `expr` themselves (e.g. the listing's
 * `address` column alone, or a parcel's `address || ', ' || city`) then wrap it here so the
 * join key matches the recorder/backfill exactly.
 *
 * Mirrors computeAddressKey: lower(collapse(trim(expr))) then strip trailing zip/state tokens.
 */
export function addressKeySql(expr: string): string {
  // 1. lowercase + collapse whitespace + trim
  const base = `btrim(regexp_replace(lower(${expr}), '\\s+', ' ', 'g'))`;
  // 2. strip trailing zip then trailing state, applied twice to peel ", ST, ZIP".
  //    Each regexp_replace peels one trailing token; two rounds cover the ST+ZIP pair
  //    (and any single leftover), matching the JS loop's fixed point for our data.
  const stripZip = (e: string) => `regexp_replace(${e}, '[[:space:],]+[0-9]{5}(-[0-9]{4})?$', '')`;
  const stripState = (e: string) => `regexp_replace(${e}, '[[:space:],]+[a-z]{2}$', '')`;
  let e = base;
  for (let i = 0; i < 2; i++) { e = stripZip(e); e = stripState(e); }
  // final trailing punctuation/space trim
  return `regexp_replace(${e}, '[[:space:],]+$', '')`;
}

/** stable parcel key when both county_fips + ain are present, else null */
export function computeParcelKey(county_fips?: string | null, ain?: string | null): string | null {
  const f = (county_fips || '').trim();
  const n = (ain || '').trim();
  return f && n ? `${f}:${n}` : null;
}

const norm = (v: unknown) => (v == null || v === '' ? null : v);

/**
 * Record a broker/firm-of-record observation. Inserts a new history row when there is
 * no prior observation for the address_key, OR when broker_id/broker_name/firm_id/firm_name
 * differ from the latest. No-ops when unchanged. Skips entirely when both broker AND firm
 * are empty (nothing to record).
 *
 * Returns the inserted row id, or null if it was a no-op / skipped.
 */
export async function recordBrokerOfRecord(
  db: QueryFn,
  input: BrokerOfRecordInput,
): Promise<number | null> {
  const brokerId = norm(input.broker_id) as number | null;
  const firmId = norm(input.firm_id) as number | null;
  const brokerName = norm(input.broker_name) as string | null;
  const firmName = norm(input.firm_name) as string | null;

  // nothing to record if there's no broker AND no firm of record
  if (brokerId == null && firmId == null && brokerName == null && firmName == null) return null;

  const addressKey = computeAddressKey(input.address, input.city);
  if (!addressKey) return null; // can't key an empty address

  const parcelKey = computeParcelKey(input.county_fips, input.ain);

  // latest observation for this address_key
  const prev = await db<{
    broker_id: number | null;
    broker_name: string | null;
    firm_id: number | null;
    firm_name: string | null;
  }>(
    `SELECT broker_id, broker_name, firm_id, firm_name
       FROM broker_of_record_history
      WHERE address_key = $1
      ORDER BY observed_at DESC
      LIMIT 1`,
    [addressKey],
  );

  if (prev.rows.length) {
    const p = prev.rows[0];
    // NOTE: pg returns BIGINT columns (broker_id, firm_id) as STRINGS, while callers pass
    // numbers — so compare id fields as strings to avoid a spurious "changed" on every run.
    const idEq = (a: unknown, b: unknown) =>
      (a == null ? null : String(a)) === (b == null ? null : String(b));
    const same =
      idEq(p.broker_id, brokerId) &&
      idEq(p.firm_id, firmId) &&
      (p.broker_name ?? null) === (brokerName ?? null) &&
      (p.firm_name ?? null) === (firmName ?? null);
    if (same) return null; // unchanged — no-op
  }

  const ins = await db<{ id: number }>(
    `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)
     VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
     RETURNING id`,
    [
      addressKey,
      parcelKey,
      norm(input.county_fips),
      norm(input.ain),
      norm(input.address),
      norm(input.city),
      norm(input.state_code),
      brokerId,
      brokerName,
      firmId,
      firmName,
      norm(input.source),
      norm(input.source_url),
    ],
  );
  return ins.rows[0]?.id ?? null;
}