← back to Nationalrealestate
TK-10139: broker-of-record history — append-only table + last_broker_of_record view + recorder + listing-ingest hook + backfill (584 rows seeded)
cd763bf32bbf42b6300c88b8c86c0662e50ed12a · 2026-08-02 07:53:18 -0700 · Steve
Files touched
A db/backfill-broker-of-record.tsA db/migrations/013_broker_of_record_history.sqlA db/views/last_broker_of_record.sqlM src/ingest/listings/engine.tsA src/server/broker-of-record.ts
Diff
commit cd763bf32bbf42b6300c88b8c86c0662e50ed12a
Author: Steve <steve@designerwallcoverings.com>
Date: Sun Aug 2 07:53:18 2026 -0700
TK-10139: broker-of-record history — append-only table + last_broker_of_record view + recorder + listing-ingest hook + backfill (584 rows seeded)
---
db/backfill-broker-of-record.ts | 75 +++++++++++++++
db/migrations/013_broker_of_record_history.sql | 49 ++++++++++
db/views/last_broker_of_record.sql | 17 ++++
src/ingest/listings/engine.ts | 29 +++++-
src/server/broker-of-record.ts | 124 +++++++++++++++++++++++++
5 files changed, 293 insertions(+), 1 deletion(-)
diff --git a/db/backfill-broker-of-record.ts b/db/backfill-broker-of-record.ts
new file mode 100644
index 0000000..eabdebe
--- /dev/null
+++ b/db/backfill-broker-of-record.ts
@@ -0,0 +1,75 @@
+/**
+ * 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';
+
+// normalized address_key must match computeAddressKey() in src/server/broker-of-record.ts:
+// lower(trim(collapse-whitespace("address, city"))).
+// NOTE: the `listing` table has NO separate city column — its `address` is already the
+// full "street, city, state, zip" string — so here address_key = normalized full address
+// (city is null). computeAddressKey with a null city produces exactly this form.
+const ADDRESS_KEY_SQL = `
+ lower(
+ regexp_replace(
+ btrim(coalesce(l.address,'')),
+ '\\s+', ' ', 'g'
+ )
+ )
+`;
+
+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);
+});
diff --git a/db/migrations/013_broker_of_record_history.sql b/db/migrations/013_broker_of_record_history.sql
new file mode 100644
index 0000000..3250238
--- /dev/null
+++ b/db/migrations/013_broker_of_record_history.sql
@@ -0,0 +1,49 @@
+-- TK-10139: append-only "broker of record history" — every observed change to the
+-- broker/firm of record for an address (or parcel) is recorded as a new row. Nothing
+-- here mutates existing catalog/listing data; this is a NEW table + supporting indexes.
+-- No BEGIN/COMMIT here — migrate.ts wraps each file in a transaction.
+--
+-- Keyed primarily by a NORMALIZED address_key (lowercased, trimmed, whitespace-collapsed
+-- "address, city"), with an OPTIONAL parcel_key ("county_fips:ain") when a stable parcel
+-- identity is known. The recorder inserts a new row only when the broker/firm of record
+-- DIFFERS from the most-recent observation for that key, so the table is the change log.
+
+CREATE TABLE broker_of_record_history (
+ id BIGSERIAL PRIMARY KEY,
+ address_key TEXT NOT NULL, -- normalized "address, city"
+ parcel_key TEXT, -- "county_fips:ain" when known
+ county_fips TEXT,
+ ain TEXT,
+ address TEXT,
+ city TEXT,
+ state_code TEXT,
+ broker_id BIGINT REFERENCES broker(id),
+ broker_name TEXT,
+ firm_id BIGINT REFERENCES firm(id),
+ firm_name TEXT,
+ source TEXT,
+ source_url TEXT,
+ observed_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- "newest observation per key" reads (drives the last_broker_of_record view + API joins).
+CREATE INDEX idx_borh_addrkey_observed ON broker_of_record_history (address_key, observed_at DESC);
+CREATE INDEX idx_borh_parcelkey_observed ON broker_of_record_history (parcel_key, observed_at DESC);
+
+-- Newest broker/firm of record per address_key. Kept in sync with
+-- db/views/last_broker_of_record.sql (standalone copy for reference).
+CREATE OR REPLACE VIEW last_broker_of_record AS
+SELECT DISTINCT ON (address_key)
+ address_key,
+ parcel_key,
+ county_fips,
+ ain,
+ address,
+ city,
+ broker_id,
+ broker_name,
+ firm_id,
+ firm_name,
+ observed_at
+FROM broker_of_record_history
+ORDER BY address_key, observed_at DESC;
diff --git a/db/views/last_broker_of_record.sql b/db/views/last_broker_of_record.sql
new file mode 100644
index 0000000..0f463c0
--- /dev/null
+++ b/db/views/last_broker_of_record.sql
@@ -0,0 +1,17 @@
+-- TK-10139: newest broker/firm of record per address_key (the "Last Broker of Record"
+-- + "Last Firm of Record" surface). DISTINCT ON picks the most-recent observation per key.
+CREATE OR REPLACE VIEW last_broker_of_record AS
+SELECT DISTINCT ON (address_key)
+ address_key,
+ parcel_key,
+ county_fips,
+ ain,
+ address,
+ city,
+ broker_id,
+ broker_name,
+ firm_id,
+ firm_name,
+ observed_at
+FROM broker_of_record_history
+ORDER BY address_key, observed_at DESC;
diff --git a/src/ingest/listings/engine.ts b/src/ingest/listings/engine.ts
index b87f59e..ee6cd87 100644
--- a/src/ingest/listings/engine.ts
+++ b/src/ingest/listings/engine.ts
@@ -20,6 +20,7 @@ import { coldwellbanker } from './coldwellbanker.ts';
import { realtytexas } from './realtytexas.ts';
import { weichert } from './weichert.ts';
import { politeFetch, robotsGate, resolveFirmByHost, resolveRegion, resolveRegionByZip } from './shared.ts';
+import { recordBrokerOfRecord } from '../../server/broker-of-record.ts';
const ADAPTERS: ListingAdapter[] = [coldwellbanker, realtytexas, weichert];
const CAP = Number(process.env.USRE_LISTINGS_CAP || 100); // per-site cap (env-tunable; was pilot 50)
@@ -28,6 +29,7 @@ async function upsertListing(
source: string,
f: import('./types.ts').ListingFacts,
firmId: number | null,
+ firmName: string | null,
regionId: number | null,
): Promise<void> {
await query(
@@ -49,6 +51,25 @@ async function upsertListing(
last_seen= NOW()`,
[source, f.sourceId, firmId, regionId, f.address, f.price, f.beds, f.baths, f.sqft, f.lat, f.lng, f.url, f.status],
);
+
+ // TK-10139: record every broker/firm-of-record observation for this address (append-only,
+ // no-ops when unchanged). Non-fatal — a recorder failure must never break listing ingest.
+ try {
+ // NOTE: the `listing` table stores the full "street, city, state, zip" in `address`
+ // (no separate city column), so we key on the address alone (city: null) to stay
+ // consistent with db/backfill-broker-of-record.ts and avoid a spurious re-observation.
+ await recordBrokerOfRecord(query, {
+ address: f.address,
+ city: null,
+ state_code: f.state,
+ firm_id: firmId,
+ firm_name: firmName,
+ source,
+ source_url: f.url,
+ });
+ } catch (e: any) {
+ console.warn(`[${source}] broker-of-record record skipped: ${e?.message || e}`);
+ }
}
async function runAdapter(a: ListingAdapter): Promise<{ upserted: number; skipped: number }> {
@@ -65,6 +86,12 @@ async function runAdapter(a: ListingAdapter): Promise<{ upserted: number; skippe
const firmId = await resolveFirmByHost(a.host);
console.log(`[${a.source}] firm_id resolved: ${firmId ?? 'NONE'}`);
+ // firm name for the broker-of-record history (resolved once per site)
+ let firmName: string | null = null;
+ if (firmId != null) {
+ const fr = await query<{ name: string }>(`SELECT name FROM firm WHERE id=$1`, [firmId]);
+ firmName = fr.rows[0]?.name ?? null;
+ }
const urls = await a.discover(CAP);
console.log(`[${a.source}] discovered ${urls.length} listing URLs (cap ${CAP})`);
@@ -80,7 +107,7 @@ async function runAdapter(a: ListingAdapter): Promise<{ upserted: number; skippe
if (!facts) { skipped++; continue; }
const regionId = (await resolveRegion(facts.state, facts.lat, facts.lng))
?? (await resolveRegionByZip(facts.zip ?? null));
- await upsertListing(a.source, facts, firmId, regionId);
+ await upsertListing(a.source, facts, firmId, firmName, regionId);
upserted++;
if (upserted % 5 === 0) console.log(`[${a.source}] upserted ${upserted}…`);
}
diff --git a/src/server/broker-of-record.ts b/src/server/broker-of-record.ts
new file mode 100644
index 0000000..83d58e5
--- /dev/null
+++ b/src/server/broker-of-record.ts
@@ -0,0 +1,124 @@
+/**
+ * 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;
+}
+
+/** normalized "address, city": lowercased, trimmed, internal whitespace collapsed */
+export function computeAddressKey(address?: string | null, city?: string | null): string {
+ const a = (address || '').trim();
+ const c = (city || '').trim();
+ const combined = c ? `${a}, ${c}` : a;
+ return combined.toLowerCase().replace(/\s+/g, ' ').trim();
+}
+
+/** 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];
+ const same =
+ (p.broker_id ?? null) === (brokerId ?? null) &&
+ (p.firm_id ?? null) === (firmId ?? null) &&
+ (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;
+}
← 49158f6 TK-5: add san-diego-sandag and santa-clara to hourly loop ro
·
back to Nationalrealestate
·
TK-10139: expose last_broker_of_record + last_firm_of_record 9026471 →