`scripts/broker-of-record-recorder.js`

104 lines
// broker-of-record-recorder.js — append-only recorder for the broker/firm of record per address.
//
// Steve's ask (TK-10139): "start recording every update to any address" for the broker + firm of record,
// and surface the "Last Broker of Record" / "Last Firm of Record" per property. This module owns the
// history store (data/broker-of-record.sqlite, table broker_of_record_history) and the read helpers.
//
// Behaviour of recordBrokerOfRecord(obs):
//   1. compute address_key = normalize(address + ", " + city)
//   2. look up the MOST RECENT row for that address_key
//   3. if NONE exists, OR broker_firm/broker_agent differs from that latest row -> INSERT a new row
//      (observed_at = now). Otherwise NO-OP (unchanged observations are not recorded).
//   4. observations with BOTH broker_firm and broker_agent empty/null are skipped.
// This is what makes the log record every *update* (a change), not every observation.
//
// $0, local, additive. better-sqlite3 (already a dep, used by serve.js for the assessor roll).
'use strict';
const path = require('path');
const fs = require('fs');
const ROOT = path.join(__dirname, '..');
const DB_PATH = path.join(ROOT, 'data', 'broker-of-record.sqlite');
const SCHEMA_PATH = path.join(__dirname, 'db', 'broker-of-record-history.sql');

// Lazily-opened writable handle (read-write, created on first use). Shared across calls in a process.
let _db = null;
function db() {
  if (_db) return _db;
  const Database = require('better-sqlite3');
  _db = new Database(DB_PATH);                 // read-write; file created if absent
  _db.pragma('journal_mode = WAL');
  const schema = fs.readFileSync(SCHEMA_PATH, 'utf8');
  _db.exec(schema);                            // idempotent (CREATE TABLE/INDEX IF NOT EXISTS)
  return _db;
}

// Normalize an address into a stable key: "<address>, <city>" lowercased, trimmed, whitespace collapsed.
function addressKey(address, city) {
  const a = String(address == null ? '' : address).trim();
  const c = String(city == null ? '' : city).trim();
  const joined = c ? `${a}, ${c}` : a;
  return joined.toLowerCase().replace(/\s+/g, ' ').trim();
}

const _clean = (v) => { const s = v == null ? '' : String(v).trim(); return s === '' ? null : s; };
const _eq = (a, b) => (_clean(a) || '') === (_clean(b) || '');

// Explicit column list (no SELECT *) — satisfies the deploy PII-leak lint gate.
const COLS = 'id, address_key, address, city, ain, broker_firm, broker_agent, source_of_truth, source_host, observed_at';

// Newest row for an address_key, or null.
function latestRow(address_key) {
  return db().prepare(
    `SELECT ${COLS} FROM broker_of_record_history WHERE address_key = ? ORDER BY observed_at DESC, id DESC LIMIT 1`
  ).get(address_key) || null;
}

// Record an observation IFF it's new or a change. Returns { recorded:bool, reason, id?, address_key }.
function recordBrokerOfRecord(obs) {
  const o = obs || {};
  const broker_firm = _clean(o.broker_firm);
  const broker_agent = _clean(o.broker_agent);
  // Skip empty observations — no broker of record to record.
  if (!broker_firm && !broker_agent) return { recorded: false, reason: 'empty', address_key: null };

  const address_key = addressKey(o.address, o.city);
  if (!address_key) return { recorded: false, reason: 'no-address', address_key: null };

  const prev = latestRow(address_key);
  if (prev && _eq(prev.broker_firm, broker_firm) && _eq(prev.broker_agent, broker_agent)) {
    return { recorded: false, reason: 'unchanged', address_key };   // no-op: same broker/firm as latest
  }

  const observed_at = _clean(o.observed_at) || new Date().toISOString();
  const info = db().prepare(
    `INSERT INTO broker_of_record_history
       (address_key, address, city, ain, broker_firm, broker_agent, source_of_truth, source_host, observed_at)
     VALUES (@address_key, @address, @city, @ain, @broker_firm, @broker_agent, @source_of_truth, @source_host, @observed_at)`
  ).run({
    address_key,
    address: _clean(o.address),
    city: _clean(o.city),
    ain: _clean(o.ain),
    broker_firm,
    broker_agent,
    source_of_truth: _clean(o.source_of_truth),
    source_host: _clean(o.source_host),
    observed_at,
  });
  return { recorded: true, reason: prev ? 'changed' : 'new', id: info.lastInsertRowid, address_key };
}

// Newest history row for an address_key (the "Last" broker/firm of record). Null if none.
function lastBrokerOfRecord(address_key) {
  return latestRow(address_key);
}

// All history rows for an address_key, newest first.
function historyFor(address_key) {
  return db().prepare(
    `SELECT ${COLS} FROM broker_of_record_history WHERE address_key = ? ORDER BY observed_at DESC, id DESC`
  ).all(address_key);
}

module.exports = { recordBrokerOfRecord, lastBrokerOfRecord, historyFor, addressKey, DB_PATH };