← back to Nationalrealestate
parcels: field-level data provenance — source every parcel field to (feed + attr + time + raw)
536b40801663e5a96a8bae27e1602ceace508193 · 2026-07-28 07:58:27 -0700 · Steve Abrams
Migration 012_parcel_provenance.sql (additive): parcel.source_url/source_key/
fetched_at/raw_source + source_field_map(source_key,our_field,source_attr).
upsert.ts: ParcelUpsertRow +optional {sourceUrl,sourceKey,fetchedAt,rawSource};
new registerSourceFieldMap(). arcgis_sales.ts: per-source fieldMap + raw feature,
stamps provenance per row + registers the field map at ingest (all 14 sources).
Adds provenance_selftest.ts (asserts persistence + "where did field X come from"
resolution, self-purging). Verified vs local usre dev DB + a bounded oregon-rlis run.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Files touched
A db/migrations/012_parcel_provenance.sqlM package.jsonM src/ingest/parcels/arcgis_sales.tsA src/ingest/parcels/provenance_selftest.tsM src/ingest/parcels/upsert.ts
Diff
commit 536b40801663e5a96a8bae27e1602ceace508193
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Tue Jul 28 07:58:27 2026 -0700
parcels: field-level data provenance — source every parcel field to (feed + attr + time + raw)
Migration 012_parcel_provenance.sql (additive): parcel.source_url/source_key/
fetched_at/raw_source + source_field_map(source_key,our_field,source_attr).
upsert.ts: ParcelUpsertRow +optional {sourceUrl,sourceKey,fetchedAt,rawSource};
new registerSourceFieldMap(). arcgis_sales.ts: per-source fieldMap + raw feature,
stamps provenance per row + registers the field map at ingest (all 14 sources).
Adds provenance_selftest.ts (asserts persistence + "where did field X come from"
resolution, self-purging). Verified vs local usre dev DB + a bounded oregon-rlis run.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
db/migrations/012_parcel_provenance.sql | 38 +++++++++++
package.json | 1 +
src/ingest/parcels/arcgis_sales.ts | 79 ++++++++++++++++++-----
src/ingest/parcels/provenance_selftest.ts | 101 ++++++++++++++++++++++++++++++
src/ingest/parcels/upsert.ts | 67 +++++++++++++++++---
5 files changed, 261 insertions(+), 25 deletions(-)
diff --git a/db/migrations/012_parcel_provenance.sql b/db/migrations/012_parcel_provenance.sql
new file mode 100644
index 0000000..d56becd
--- /dev/null
+++ b/db/migrations/012_parcel_provenance.sql
@@ -0,0 +1,38 @@
+-- TK-50: FIELD-LEVEL DATA PROVENANCE — "source every bit of info."
+-- No BEGIN/COMMIT here — migrate.ts wraps each file in a transaction.
+--
+-- One source record (e.g. one ArcGIS feature) → one parcel row, so ALL of a
+-- parcel's fields share the same source. Provenance is therefore
+-- (record source_url + fetched_at + source_key) + (which source attribute
+-- each of our fields mapped from) + the raw record.
+--
+-- We store the record-level part on the parcel row (source_url/source_key/
+-- fetched_at/raw_source) and the per-field attribute mapping ONCE per source
+-- in source_field_map (our_field → source_attr). "Where did field X come
+-- from?" resolves to:
+-- parcel.source_url (the addressable record link)
+-- + source_field_map[source_key][X].source_attr (which raw attribute)
+-- + parcel.raw_source[source_attr] (the raw value we mapped from)
+--
+-- Additive only (ADD COLUMN / CREATE TABLE) — safe + reversible. Existing
+-- rows get source_url=NULL until they are re-ingested (backfill is a separate,
+-- gated prod step).
+
+ALTER TABLE parcel ADD COLUMN IF NOT EXISTS source_url TEXT; -- per-record addressable link back to the exact source record
+ALTER TABLE parcel ADD COLUMN IF NOT EXISTS source_key TEXT; -- ingest source id, e.g. 'oregon-rlis' — FK-ish into source_field_map
+ALTER TABLE parcel ADD COLUMN IF NOT EXISTS fetched_at TIMESTAMPTZ; -- when the record was pulled (run start time)
+ALTER TABLE parcel ADD COLUMN IF NOT EXISTS raw_source JSONB; -- the raw source-record attributes we mapped from
+
+CREATE INDEX IF NOT EXISTS idx_parcel_source_key ON parcel (source_key);
+
+-- Per-source declaration of which raw source attribute each of our fields maps
+-- from. One row per (source_key, our_field). Registered at ingest time by
+-- registerSourceFieldMap() so provenance is queryable without re-reading code.
+CREATE TABLE IF NOT EXISTS source_field_map (
+ source_key TEXT NOT NULL, -- matches parcel.source_key
+ our_field TEXT NOT NULL, -- our parcel column, e.g. 'owner_name'
+ source_attr TEXT NOT NULL, -- the raw source attribute, e.g. 'OWNER_NAME'
+ notes TEXT, -- free-text (e.g. 'grantee/current owner')
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+ PRIMARY KEY (source_key, our_field)
+);
diff --git a/package.json b/package.json
index 37671d5..6b804b4 100644
--- a/package.json
+++ b/package.json
@@ -23,6 +23,7 @@
"ingest:nri": "tsx src/ingest/fema_nri.ts",
"ingest:fmr": "tsx src/ingest/hud_fmr.ts",
"ingest:parcels": "tsx src/ingest/parcels/engine.ts",
+ "test:provenance": "tsx src/ingest/parcels/provenance_selftest.ts",
"ingest:commercial": "tsx src/ingest/commercial/engine.ts",
"ingest:commercial-briefs": "tsx src/ingest/commercial/briefs.ts",
"ingest:places": "tsx src/ingest/places/seed.ts",
diff --git a/src/ingest/parcels/arcgis_sales.ts b/src/ingest/parcels/arcgis_sales.ts
index 4c144d3..292bb8b 100644
--- a/src/ingest/parcels/arcgis_sales.ts
+++ b/src/ingest/parcels/arcgis_sales.ts
@@ -16,7 +16,7 @@
*/
import { query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
-import { upsertParcels, normAddress, registerParcelSource, type ParcelUpsertRow } from './upsert.ts';
+import { upsertParcels, normAddress, registerParcelSource, registerSourceFieldMap, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
import { sanitizeSalePrice } from './price_guard.ts';
@@ -31,11 +31,16 @@ interface Rec {
use?: string | null; totalValue?: number | null;
price?: number | null; saleDate?: string | null;
docNum?: string | null; docType?: string | null; grantor?: string | null; grantee?: string | null;
+ raw?: Record<string, unknown> | null; // TK-50: the raw source-record attributes this Rec was mapped from
}
interface Source {
key: string; label: string; layer: string; where: string; outFields: string; cursorKey: string;
geometry: boolean; orderBy: string; pageSize?: number; // a valid field for stable resultOffset paging (OID name varies)
toRecords: (features: any[]) => Rec[];
+ // TK-50 field-level provenance: our parcel field → the exact source attribute
+ // it was mapped from in toRecords(). Registered into source_field_map at
+ // ingest time so "where did field X come from?" resolves without reading code.
+ fieldMap: Record<string, string>;
}
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
@@ -71,27 +76,34 @@ const SOURCES: Record<string, Source> = {
layer: 'https://services2.arcgis.com/McQ0OlIABe29rJJy/arcgis/rest/services/Taxlots_(Public)/FeatureServer/3',
where: 'SALEPRICE>1000', cursorKey: 'oregon_rlis', geometry: true, orderBy: 'FID',
outFields: 'TLID,PRIMACCNUM,SITEADDR,SITECITY,SITEZIP,SALEDATE,SALEPRICE,COUNTY,YEARBUILT,BLDGSQFT,LANDUSE',
+ fieldMap: { source_id: 'PRIMACCNUM', address: 'SITEADDR', city: 'SITECITY', zip: 'SITEZIP',
+ sqft: 'BLDGSQFT', year_built: 'YEARBUILT', use_desc: 'LANDUSE',
+ last_sale_price: 'SALEPRICE', last_sale_date: 'SALEDATE', county_fips: 'COUNTY' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.PRIMACCNUM) || s(a.TLID); if (!apn) return null;
const [lng, lat] = centroid(f.geometry);
return { fips: OR_FIPS[s(a.COUNTY) || ''] || '41051', apn, address: clean(a.SITEADDR), city: clean(a.SITECITY), zip: s(a.SITEZIP),
lat, lng, sqft: num(a.BLDGSQFT), year: num(a.YEARBUILT), use: s(a.LANDUSE),
- price: anyPrice(a.SALEPRICE), saleDate: yyyymm(a.SALEDATE) } as Rec; }).filter(Boolean) as Rec[],
+ price: anyPrice(a.SALEPRICE), saleDate: yyyymm(a.SALEDATE), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
spokane: {
key: 'spokane', label: 'Spokane County WA', cursorKey: 'spokane', geometry: false,
layer: 'https://gismo.spokanecounty.org/arcgis/rest/services/SCOUT/PropertyLookup/MapServer/0',
where: 'gross_sale_price>1000', orderBy: 'PID_NUM', pageSize: 500, // slow MapServer (~50ms/row)
outFields: 'PID_NUM,gross_sale_price,document_date,excise_nbr,site_address,owner_name,transfer_type',
+ fieldMap: { source_id: 'PID_NUM', address: 'site_address', last_sale_price: 'gross_sale_price',
+ last_sale_date: 'document_date', owner_name: 'owner_name' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.PID_NUM); if (!apn) return null;
return { fips: '53063', apn, address: clean(a.site_address), city: 'Spokane', zip: null, lat: null, lng: null,
price: anyPrice(a.gross_sale_price), saleDate: epochToISO(a.document_date), docNum: s(a.excise_nbr),
- docType: clean(a.transfer_type), grantee: clean(a.owner_name) } as Rec; }).filter(Boolean) as Rec[],
+ docType: clean(a.transfer_type), grantee: clean(a.owner_name), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
alameda: {
key: 'alameda', label: 'Alameda County CA (deed transfers)', cursorKey: 'alameda', geometry: false,
layer: 'https://services5.arcgis.com/ROBnTHSNjoZ2Wm1P/arcgis/rest/services/Assessor_Office_Ownership_Transfer_List/FeatureServer/0',
where: 'value_from_trans_tax IS NOT NULL', orderBy: 'OBJECTID',
outFields: 'apn,use_name,street_num,street_name,street_suffix,city_name,zip_cd,transfer_dt,doc_prefix,doc_series,value_from_trans_tax,name_type',
+ fieldMap: { source_id: 'apn', address: 'street_num+street_name+street_suffix', city: 'city_name', zip: 'zip_cd',
+ use_desc: 'use_name', last_sale_price: 'value_from_trans_tax', last_sale_date: 'transfer_dt' },
// grantor+grantee are separate rows sharing doc_series — collapse per doc.
toRecords: (fs) => {
const byDoc = new Map<string, Rec>();
@@ -102,7 +114,7 @@ const SOURCES: Record<string, Source> = {
const addr = [s(a.street_num), clean(a.street_name), s(a.street_suffix)].filter(Boolean).join(' ').trim() || null;
const key = apn + '|' + doc;
const r = byDoc.get(key) || { fips: '06001', apn, address: addr, city: clean(a.city_name), zip: s(a.zip_cd),
- lat: null, lng: null, use: clean(a.use_name), price, saleDate: epochToISO(a.transfer_dt), docNum: doc } as Rec;
+ lat: null, lng: null, use: clean(a.use_name), price, saleDate: epochToISO(a.transfer_dt), docNum: doc, raw: a } as Rec;
const nm = clean(a.name_type)?.toUpperCase();
// name_type marks the party; the party NAME itself isn't in this field set, so record role presence.
if (nm === 'TRANSFEROR') r.grantor = r.grantor || 'recorded';
@@ -117,6 +129,8 @@ const SOURCES: Record<string, Source> = {
layer: 'https://maps.deschutes.org/arcgis/rest/services/OpenData/TablesFD/MapServer/9',
where: 'Total_Sales_Price_1>1000',
outFields: 'Taxlot,Total_Sales_Price_1,Sales_Date_1,Seller_1,Buyer_1,Book_Page_1,Total_Sales_Price_2,Sales_Date_2,Seller_2,Buyer_2,Book_Page_2',
+ fieldMap: { source_id: 'Taxlot', last_sale_price: 'Total_Sales_Price_1', last_sale_date: 'Sales_Date_1',
+ owner_name: 'Buyer_1' },
// Two sales per row (_1 newest, _2 prior), each with REAL grantor/grantee NAMES.
toRecords: (fs) => {
const out: Rec[] = [];
@@ -126,7 +140,7 @@ const SOURCES: Record<string, Source> = {
const price = num(a['Total_Sales_Price_' + nx]); const date = epochToISO(a['Sales_Date_' + nx]);
if (!price || !date) continue;
out.push({ fips: '41017', apn, price, saleDate: date, grantor: clean(a['Seller_' + nx]),
- grantee: clean(a['Buyer_' + nx]), docNum: s(a['Book_Page_' + nx]) || `${date}-${nx}` } as Rec);
+ grantee: clean(a['Buyer_' + nx]), docNum: s(a['Book_Page_' + nx]) || `${date}-${nx}`, raw: a } as Rec);
}
}
return out;
@@ -137,23 +151,27 @@ const SOURCES: Record<string, Source> = {
layer: 'https://gis.crookcountyor.gov/server/rest/services/OpenData/TaxlotandTables/MapServer/6',
where: 'SALES_PRICE>0',
outFields: 'MAPTAXLOT,SALES_PRICE,SALES_DATE,GRANTOR_NAME,GRANTEE_NAME,BOOK,NUMBER,SALE_ID',
+ fieldMap: { source_id: 'MAPTAXLOT', last_sale_price: 'SALES_PRICE', last_sale_date: 'SALES_DATE',
+ owner_name: 'GRANTEE_NAME' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.MAPTAXLOT); if (!apn) return null;
return { fips: '41013', apn, price: anyPrice(a.SALES_PRICE), saleDate: anyDate(a.SALES_DATE),
grantor: clean(a.GRANTOR_NAME), grantee: clean(a.GRANTEE_NAME),
- docNum: [s(a.BOOK), s(a.NUMBER)].filter(Boolean).join('-') || s(a.SALE_ID) } as Rec; }).filter(Boolean) as Rec[],
+ docNum: [s(a.BOOK), s(a.NUMBER)].filter(Boolean).join('-') || s(a.SALE_ID), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
sonoma: {
key: 'sonoma', label: 'Sonoma County CA', cursorKey: 'sonoma', geometry: false, orderBy: 'OBJECTID', pageSize: 2000,
layer: 'https://socogis.sonomacounty.ca.gov/map/rest/services/CRAPublic/ParcelsPublic/FeatureServer/0',
where: 'SaleSalesPrice>0',
outFields: 'APN,SaleSalesPrice,SaleRecordingDate,SaleDocNum,SalePriorSalesPrice,SalePriorRecordingDate,SalePriorDocNum,SitusFormatted1,SitusCity',
+ fieldMap: { source_id: 'APN', address: 'SitusFormatted1', city: 'SitusCity',
+ last_sale_price: 'SaleSalesPrice', last_sale_date: 'SaleRecordingDate' },
toRecords: (fs) => { const out: Rec[] = [];
for (const f of fs) { const a = f.attributes; const apn = s(a.APN); if (!apn) continue;
const addr = clean(a.SitusFormatted1), city = clean(a.SitusCity);
const cur = anyPrice(a.SaleSalesPrice), cd = anyDate(a.SaleRecordingDate);
- if (cur && cd) out.push({ fips: '06097', apn, address: addr, city, price: cur, saleDate: cd, docNum: s(a.SaleDocNum) } as Rec);
+ if (cur && cd) out.push({ fips: '06097', apn, address: addr, city, price: cur, saleDate: cd, docNum: s(a.SaleDocNum), raw: a } as Rec);
const pp = anyPrice(a.SalePriorSalesPrice), pd = anyDate(a.SalePriorRecordingDate);
- if (pp && pd) out.push({ fips: '06097', apn, address: addr, city, price: pp, saleDate: pd, docNum: s(a.SalePriorDocNum) } as Rec);
+ if (pp && pd) out.push({ fips: '06097', apn, address: addr, city, price: pp, saleDate: pd, docNum: s(a.SalePriorDocNum), raw: a } as Rec);
} return out; },
},
colorado: {
@@ -161,61 +179,76 @@ const SOURCES: Record<string, Source> = {
layer: 'https://gis.colorado.gov/Public/rest/services/Address_and_Parcel/Colorado_Public_Parcels/MapServer/0',
where: "salePrice>'0'",
outFields: 'parcel_id,salePrice,saleDate,countyFips,countyName',
+ fieldMap: { source_id: 'parcel_id', county_fips: 'countyFips', city: 'countyName',
+ last_sale_price: 'salePrice', last_sale_date: 'saleDate' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.parcel_id); if (!apn) return null;
const cf = (s(a.countyFips) || '').replace(/\D/g, '');
const fips = cf.length >= 5 ? cf.slice(0, 5) : '08' + cf.padStart(3, '0').slice(-3);
- return { fips, apn, city: clean(a.countyName), price: anyPrice(a.salePrice), saleDate: anyDate(a.saleDate) } as Rec; }).filter(Boolean) as Rec[],
+ return { fips, apn, city: clean(a.countyName), price: anyPrice(a.salePrice), saleDate: anyDate(a.saleDate), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
maricopa: {
key: 'maricopa', label: 'Maricopa County AZ (Phoenix)', cursorKey: 'maricopa', geometry: false, orderBy: 'OBJECTID', pageSize: 1000,
layer: 'https://gis.mcassessor.maricopa.gov/arcgis/rest/services/MaricopaDynamicQueryService/MapServer/3',
where: "SALE_PRICE>'0'",
outFields: 'APN,SALE_PRICE,SALE_DATE,DEED_DATE,DEED_NUMBER,PHYSICAL_STREET_NUM,PHYSICAL_STREET_DIR,PHYSICAL_STREET_NAME,PHYSICAL_STREET_TYPE,PHYSICAL_CITY,PHYSICAL_ZIP',
+ fieldMap: { source_id: 'APN', address: 'PHYSICAL_STREET_NUM+PHYSICAL_STREET_DIR+PHYSICAL_STREET_NAME+PHYSICAL_STREET_TYPE',
+ city: 'PHYSICAL_CITY', zip: 'PHYSICAL_ZIP', last_sale_price: 'SALE_PRICE', last_sale_date: 'SALE_DATE' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.APN); if (!apn) return null;
const addr = [s(a.PHYSICAL_STREET_NUM), s(a.PHYSICAL_STREET_DIR), clean(a.PHYSICAL_STREET_NAME), s(a.PHYSICAL_STREET_TYPE)].filter(Boolean).join(' ').trim() || null;
return { fips: '04013', apn, address: addr, city: clean(a.PHYSICAL_CITY), zip: s(a.PHYSICAL_ZIP),
- price: anyPrice(a.SALE_PRICE), saleDate: anyDate(a.SALE_DATE) || anyDate(a.DEED_DATE), docNum: s(a.DEED_NUMBER) } as Rec; }).filter(Boolean) as Rec[],
+ price: anyPrice(a.SALE_PRICE), saleDate: anyDate(a.SALE_DATE) || anyDate(a.DEED_DATE), docNum: s(a.DEED_NUMBER), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
thurston: {
key: 'thurston', label: 'Thurston County WA (Olympia)', cursorKey: 'thurston', geometry: false, orderBy: 'OBJECTID', pageSize: 1000,
layer: 'https://map.co.thurston.wa.us/arcgis/rest/services/Thurston/Thurston_Parcels/FeatureServer/0',
where: 'SALE_PRICE>0',
outFields: 'PARCEL_NO,SALE_PRICE,SALE_DATE,SITUS_STRE,SITUS_CITY,SITUS_ZIP',
+ fieldMap: { source_id: 'PARCEL_NO', address: 'SITUS_STRE', city: 'SITUS_CITY', zip: 'SITUS_ZIP',
+ last_sale_price: 'SALE_PRICE', last_sale_date: 'SALE_DATE' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.PARCEL_NO); if (!apn) return null;
return { fips: '53067', apn, address: clean(a.SITUS_STRE), city: clean(a.SITUS_CITY), zip: s(a.SITUS_ZIP),
- price: anyPrice(a.SALE_PRICE), saleDate: anyDate(a.SALE_DATE) } as Rec; }).filter(Boolean) as Rec[],
+ price: anyPrice(a.SALE_PRICE), saleDate: anyDate(a.SALE_DATE), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
'palm-beach': {
key: 'palm-beach', label: 'Palm Beach County FL', cursorKey: 'palm_beach', geometry: false, orderBy: 'OBJECTID', pageSize: 2000,
layer: 'https://services1.arcgis.com/RTiKiFNGzgAobBzy/arcgis/rest/services/ParcelPropertyDetails/FeatureServer/1',
where: 'PRICE>0',
outFields: 'PARCEL_ID,PRICE,SALE_DATE,SITE_ADDR,CITY,OWNER_NAME1',
+ fieldMap: { source_id: 'PARCEL_ID', address: 'SITE_ADDR', city: 'CITY',
+ last_sale_price: 'PRICE', last_sale_date: 'SALE_DATE', owner_name: 'OWNER_NAME1' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.PARCEL_ID); if (!apn) return null;
return { fips: '12099', apn, price: anyPrice(a.PRICE), saleDate: anyDate(a.SALE_DATE),
- address: clean(a.SITE_ADDR), city: clean(a.CITY), grantee: clean(a.OWNER_NAME1) } as Rec; }).filter(Boolean) as Rec[],
+ address: clean(a.SITE_ADDR), city: clean(a.CITY), grantee: clean(a.OWNER_NAME1), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
klamath: {
key: 'klamath', label: 'Klamath County OR (Klamath Falls)', cursorKey: 'klamath', geometry: false, orderBy: 'OBJECTID', pageSize: 2000,
layer: 'https://services.arcgis.com/H6Mh1bySxR4oHx6x/arcgis/rest/services/KC_ParcelSales/FeatureServer/0',
where: 'SALE_PRICE>1000', // rich Klamath assessor sales layer: SALE_PRICE(int)+SALE_DATE(epoch-ms)+owner+situs+sqft/beds/baths/yrblt+total appraised
outFields: 'FIRST_PROP_ID,SALE_PRICE,SALE_DATE,SALEBK,OWNER_NAME,MIN_SITUS_ADDRESS,FIRST_CITY_NAME,SUM_FINSQFT,SUM_BEDRMS,SUM_BATH,MIN_YRBLT,SUM_Tot_Appr,FIRST_PCLCD',
+ fieldMap: { source_id: 'FIRST_PROP_ID', address: 'MIN_SITUS_ADDRESS', city: 'FIRST_CITY_NAME',
+ sqft: 'SUM_FINSQFT', beds: 'SUM_BEDRMS', baths: 'SUM_BATH', year_built: 'MIN_YRBLT',
+ total_value: 'SUM_Tot_Appr', use_desc: 'FIRST_PCLCD', owner_name: 'OWNER_NAME',
+ last_sale_price: 'SALE_PRICE', last_sale_date: 'SALE_DATE' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.FIRST_PROP_ID); if (!apn) return null;
return { fips: '41035', apn, price: anyPrice(a.SALE_PRICE), saleDate: anyDate(a.SALE_DATE),
address: clean(a.MIN_SITUS_ADDRESS), city: clean(a.FIRST_CITY_NAME),
sqft: num(a.SUM_FINSQFT), beds: num(a.SUM_BEDRMS), baths: num(a.SUM_BATH), year: num(a.MIN_YRBLT),
totalValue: num(a.SUM_Tot_Appr), use: clean(a.FIRST_PCLCD),
- grantee: clean(a.OWNER_NAME), docNum: clean(a.SALEBK) } as Rec; }).filter(Boolean) as Rec[],
+ grantee: clean(a.OWNER_NAME), docNum: clean(a.SALEBK), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
marion: {
key: 'marion', label: 'Marion County OR (Salem)', cursorKey: 'marion', geometry: false, orderBy: 'OBJECTID', pageSize: 2000,
layer: 'https://services1.arcgis.com/sYGZnQPdJ0azuLyn/arcgis/rest/services/TaxParcel_Assessment/FeatureServer/0',
where: 'SALEPRICE>1000', // rich Marion assessor layer: SALEPRICE(int)+INSTDATE(epoch-ms deed date)+deed doc/type+owner+situs+livingarea+yearbuilt+RMV total appraised
outFields: 'TAXLOT,SALEPRICE,INSTDATE,INSTNUM,INSTTYPE,OWNERNAME,SITUS,YEARBUILT,LIVINGAREA,RMVTOTAL,PROPCLASS,STCLSDESC',
+ fieldMap: { source_id: 'TAXLOT', address: 'SITUS', sqft: 'LIVINGAREA', year_built: 'YEARBUILT',
+ total_value: 'RMVTOTAL', use_desc: 'STCLSDESC', owner_name: 'OWNERNAME',
+ last_sale_price: 'SALEPRICE', last_sale_date: 'INSTDATE' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a.TAXLOT); if (!apn) return null;
return { fips: '41047', apn, price: anyPrice(a.SALEPRICE), saleDate: anyDate(a.INSTDATE),
address: clean(a.SITUS), sqft: num(a.LIVINGAREA), year: num(a.YEARBUILT), totalValue: num(a.RMVTOTAL),
use: clean(a.STCLSDESC) || clean(a.PROPCLASS),
- grantee: clean(a.OWNERNAME), docNum: s(a.INSTNUM), docType: clean(a.INSTTYPE) } as Rec; }).filter(Boolean) as Rec[],
+ grantee: clean(a.OWNERNAME), docNum: s(a.INSTNUM), docType: clean(a.INSTTYPE), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
yakima: {
key: 'yakima', label: 'Yakima County WA', cursorKey: 'yakima', geometry: false, orderBy: 'DBO.Parcels.OBJECTID', pageSize: 2000,
@@ -224,6 +257,11 @@ const SOURCES: Record<string, Source> = {
layer: 'https://gis.yakimawa.gov/arcgis/rest/services/Assessor/AssessorParcels/MapServer/0',
where: 'DBO.Parcels.GROSS_SALE_PRICE>1000',
outFields: 'DBO.Parcels.ASSESSOR_NO,DBO.Parcels.GROSS_SALE_PRICE,DBO.Parcels.SALE_DATE,DBO.Parcels.EXCISE_NUMBER,DBO.Parcels.GRANTOR_NAME,DBO.Parcels.SITUS_ADDR,DBO.Parcels.SITUS_CITY,DBO.Parcels.SITUS_ZIP,DBO.Parcels.USE_CODE,DBO.Parcels.YEAR_BLT,DBO.Parcels.MAIN_SQFT,DBO.Parcels.BEDROOMS,DBO.Parcels.FULL_BATH,DBO.Parcels.MKT_LAND,DBO.Parcels.MKT_IMPVT,DBO.Parcels.LAST_NAME,DBO.Parcels.FIRST_NAME,DBO.Parcels.ORG_NAME',
+ fieldMap: { source_id: 'DBO.Parcels.ASSESSOR_NO', address: 'DBO.Parcels.SITUS_ADDR', city: 'DBO.Parcels.SITUS_CITY',
+ zip: 'DBO.Parcels.SITUS_ZIP', sqft: 'DBO.Parcels.MAIN_SQFT', beds: 'DBO.Parcels.BEDROOMS',
+ baths: 'DBO.Parcels.FULL_BATH', year_built: 'DBO.Parcels.YEAR_BLT', use_desc: 'DBO.Parcels.USE_CODE',
+ total_value: 'DBO.Parcels.MKT_LAND+DBO.Parcels.MKT_IMPVT', owner_name: 'DBO.Parcels.ORG_NAME|FIRST_NAME+LAST_NAME',
+ last_sale_price: 'DBO.Parcels.GROSS_SALE_PRICE', last_sale_date: 'DBO.Parcels.SALE_DATE' },
toRecords: (fs) => fs.map(f => { const a = f.attributes; const apn = s(a['DBO.Parcels.ASSESSOR_NO']); if (!apn) return null;
let situs = clean(a['DBO.Parcels.SITUS_ADDR']); if (situs && /^UNASSIGNED$/i.test(situs)) situs = null;
const org = clean(a['DBO.Parcels.ORG_NAME']);
@@ -234,7 +272,7 @@ const SOURCES: Record<string, Source> = {
address: situs, city: clean(a['DBO.Parcels.SITUS_CITY']), zip: s(a['DBO.Parcels.SITUS_ZIP']),
sqft: num(a['DBO.Parcels.MAIN_SQFT']), beds: num(a['DBO.Parcels.BEDROOMS']), baths: num(a['DBO.Parcels.FULL_BATH']),
year: num(a['DBO.Parcels.YEAR_BLT']), totalValue: (land + impvt) || null, use: clean(a['DBO.Parcels.USE_CODE']),
- grantor: clean(a['DBO.Parcels.GRANTOR_NAME']), grantee: owner, docNum: s(a['DBO.Parcels.EXCISE_NUMBER']) } as Rec; }).filter(Boolean) as Rec[],
+ grantor: clean(a['DBO.Parcels.GRANTOR_NAME']), grantee: owner, docNum: s(a['DBO.Parcels.EXCISE_NUMBER']), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
};
@@ -273,7 +311,13 @@ export function ingestArcgisSales(key: string) {
if (!src) throw new Error(`unknown sales source '${key}' — have: ${Object.keys(SOURCES).join(', ')}`);
return async (): Promise<{ upserted: number }> => {
const runId = await openRun(src.cursorKey, src.layer);
+ // TK-50: run start time — the fetched_at stamped on every row this run touches.
+ const fetchedAt = new Date().toISOString();
try {
+ // TK-50: declare this source's our_field → source_attr mapping up front, so
+ // "where did parcel field X come from?" is answerable from source_field_map
+ // without re-reading toRecords().
+ await registerSourceFieldMap(src.key, src.fieldMap, src.label);
const total = await countTotal(src);
const cur = await query<{ next_offset: number }>(`SELECT next_offset FROM ingest_cursor WHERE source=$1`, [src.cursorKey]);
let offset = cur.rows[0]?.next_offset ?? 0;
@@ -299,6 +343,11 @@ export function ingestArcgisSales(key: string) {
tax_year: null, owner_name: r.grantee && r.grantee !== 'recorded' ? r.grantee : null, zoning: null,
last_sale_date: r.saleDate ?? null, last_sale_price: r.price ?? null,
extra: JSON.stringify({ source: src.key, doc: r.docNum, doc_type: r.docType }),
+ // ── TK-50 field-level provenance (record-level: one feature → this row) ──
+ sourceKey: src.key,
+ sourceUrl: gisUrl, // deep-links to the exact source record (layer + where APN='…')
+ fetchedAt, // run start time
+ rawSource: r.raw ? JSON.stringify(r.raw) : null, // the raw source attributes we mapped from
});
if (r.price && r.saleDate) events.push({ ...r, address: gisUrl }); // reuse address slot to carry the link
}
diff --git a/src/ingest/parcels/provenance_selftest.ts b/src/ingest/parcels/provenance_selftest.ts
new file mode 100644
index 0000000..37d897a
--- /dev/null
+++ b/src/ingest/parcels/provenance_selftest.ts
@@ -0,0 +1,101 @@
+/**
+ * TK-50 field-level-provenance self-test (run: `tsx src/ingest/parcels/provenance_selftest.ts`).
+ *
+ * No test framework is installed in this repo, so this ships as a deterministic
+ * self-test that exits non-zero on any failure. It:
+ * (a) upserts a synthetic parcel carrying record-level provenance
+ * (source_url / source_key / fetched_at / raw_source) + registers a
+ * source_field_map for the synthetic source,
+ * (b) asserts all four provenance columns persisted,
+ * (c) asserts a "where did field X come from?" resolution =
+ * parcel.source_url + source_field_map[source_key][X] + raw_source[source_attr],
+ * (d) purges every synthetic row it created.
+ *
+ * Uses a clearly-synthetic county_fips ('99999') + source_key ('selftest-tk50')
+ * so it can never collide with, or leave residue in, real ingested data.
+ */
+import { pool, query } from '../../../db/pool.ts';
+import { upsertParcels, registerSourceFieldMap, type ParcelUpsertRow } from './upsert.ts';
+
+const FIPS = '99999';
+const APN = 'TK50-SELFTEST-APN';
+const KEY = 'selftest-tk50';
+
+let failed = 0;
+function check(ok: boolean, label: string, detail?: unknown) {
+ if (ok) { console.log(` ok ${label}`); return; }
+ failed++;
+ console.error(` FAIL ${label}${detail !== undefined ? `\n ${JSON.stringify(detail)}` : ''}`);
+}
+
+async function purge() {
+ await query(`DELETE FROM parcel WHERE county_fips=$1 AND source_id=$2`, [FIPS, APN]);
+ await query(`DELETE FROM source_field_map WHERE source_key=$1`, [KEY]);
+}
+
+async function main() {
+ await purge(); // clean slate in case a prior run aborted before its own purge
+
+ const fetchedAt = new Date().toISOString();
+ // The raw source record we "obtained" the data from. owner_name maps from
+ // OWNER_NAME1, last_sale_price from PRICE — declared in the field map below.
+ const raw = { PARCEL_ID: APN, OWNER_NAME1: 'JANE Q SYNTHETIC', PRICE: 725000, SITE_ADDR: '1 TEST WAY', CITY: 'TESTVILLE' };
+ const sourceUrl = `https://example.invalid/arcgis/query?where=PARCEL_ID='${APN}'&f=html`;
+
+ const row: ParcelUpsertRow = {
+ county_fips: FIPS, source_id: APN,
+ address: '1 TEST WAY', norm_address: '1 TEST WAY', city: 'TESTVILLE', zip: '00000',
+ lat: null, lng: null, year_built: null, sqft: null, beds: null, baths: null,
+ units: null, use_desc: null, land_value: null, improvement_value: null, total_value: null,
+ tax_year: null, owner_name: 'JANE Q SYNTHETIC', zoning: null,
+ last_sale_date: '2025-01-15', last_sale_price: 725000,
+ extra: JSON.stringify({ source: KEY }),
+ sourceKey: KEY, sourceUrl, fetchedAt, rawSource: JSON.stringify(raw),
+ };
+
+ await upsertParcels([row]);
+ await registerSourceFieldMap(KEY, { owner_name: 'OWNER_NAME1', last_sale_price: 'PRICE', address: 'SITE_ADDR', city: 'CITY' }, 'TK-50 self-test source');
+
+ // (b) provenance columns persisted on the parcel row
+ const p = await query<{ source_url: string; source_key: string; fetched_at: string; raw_source: Record<string, unknown> }>(
+ `SELECT source_url, source_key, fetched_at, raw_source FROM parcel WHERE county_fips=$1 AND source_id=$2`, [FIPS, APN]);
+ check(p.rows.length === 1, 'parcel row exists', p.rows.length);
+ const pr = p.rows[0] || ({} as any);
+ check(pr.source_url === sourceUrl, 'source_url persisted', pr.source_url);
+ check(pr.source_key === KEY, 'source_key persisted', pr.source_key);
+ check(pr.fetched_at != null, 'fetched_at persisted', pr.fetched_at);
+ check(pr.raw_source != null && (pr.raw_source as any).OWNER_NAME1 === 'JANE Q SYNTHETIC', 'raw_source JSONB persisted', pr.raw_source);
+
+ // (c) "where did field X come from?" resolution for X = owner_name.
+ // A field traces to: parcel.source_url (the record link)
+ // + source_field_map[source_key][X].source_attr (which raw attribute)
+ // + raw_source[source_attr] (the raw value we mapped from).
+ const X = 'owner_name';
+ const res = await query<{ source_url: string; source_attr: string; raw_value: string }>(
+ `SELECT p.source_url,
+ m.source_attr,
+ p.raw_source ->> m.source_attr AS raw_value
+ FROM parcel p
+ JOIN source_field_map m
+ ON m.source_key = p.source_key AND m.our_field = $3
+ WHERE p.county_fips = $1 AND p.source_id = $2`,
+ [FIPS, APN, X]);
+ check(res.rows.length === 1, `provenance resolves for field '${X}'`, res.rows.length);
+ const rr = res.rows[0] || ({} as any);
+ check(rr.source_url === sourceUrl, ` ${X} → source_url`, rr.source_url);
+ check(rr.source_attr === 'OWNER_NAME1', ` ${X} → source_attr = OWNER_NAME1`, rr.source_attr);
+ check(rr.raw_value === 'JANE Q SYNTHETIC', ` ${X} → raw value from that attr`, rr.raw_value);
+ console.log(` resolved: ${X} came from ${rr.source_url} :: attr '${rr.source_attr}' = '${rr.raw_value}'`);
+
+ // (d) purge synthetic rows
+ await purge();
+ const gone = await query(`SELECT 1 FROM parcel WHERE county_fips=$1 AND source_id=$2`, [FIPS, APN]);
+ const mapGone = await query(`SELECT 1 FROM source_field_map WHERE source_key=$1`, [KEY]);
+ check(gone.rows.length === 0 && mapGone.rows.length === 0, 'synthetic rows purged', { parcel: gone.rows.length, map: mapGone.rows.length });
+
+ await pool.end();
+ if (failed) { console.error(`\n${failed} provenance self-test check(s) FAILED`); process.exit(1); }
+ console.log('\nprovenance self-test: all checks passed');
+}
+
+main().catch(async (e) => { console.error(e); try { await purge(); await pool.end(); } catch {} process.exit(1); });
diff --git a/src/ingest/parcels/upsert.ts b/src/ingest/parcels/upsert.ts
index 3740c7d..94320ff 100644
--- a/src/ingest/parcels/upsert.ts
+++ b/src/ingest/parcels/upsert.ts
@@ -16,32 +16,49 @@ export interface ParcelUpsertRow {
tax_year: string | null; owner_name: string | null; zoning: string | null;
last_sale_date: string | null; last_sale_price: number | null;
extra: string | null; // JSON string
+ // ── TK-50 field-level provenance (all OPTIONAL — backward compatible) ──
+ // Because one source record → one parcel row, these apply to the WHOLE row:
+ // every field on this parcel was obtained from this source record.
+ sourceUrl?: string | null; // per-record addressable link back to the exact source record
+ sourceKey?: string | null; // ingest source id, e.g. 'oregon-rlis' — joins source_field_map
+ fetchedAt?: string | null; // ISO timestamp the record was pulled (run start time)
+ rawSource?: string | null; // JSON string of the raw source-record attributes we mapped from
}
-const COLS: (keyof ParcelUpsertRow)[] = ['county_fips', 'source_id', 'address', 'norm_address', 'city', 'zip',
- 'lat', 'lng', 'year_built', 'sqft', 'beds', 'baths', 'units', 'use_desc',
- 'land_value', 'improvement_value', 'total_value', 'tax_year', 'owner_name', 'zoning',
- 'last_sale_date', 'last_sale_price', 'extra'];
+// DB columns (snake_case) paired with the ParcelUpsertRow key that feeds each.
+// The provenance columns accept the camelCase optional fields above.
+const COL_MAP: [string, keyof ParcelUpsertRow][] = [
+ ['county_fips', 'county_fips'], ['source_id', 'source_id'], ['address', 'address'],
+ ['norm_address', 'norm_address'], ['city', 'city'], ['zip', 'zip'], ['lat', 'lat'], ['lng', 'lng'],
+ ['year_built', 'year_built'], ['sqft', 'sqft'], ['beds', 'beds'], ['baths', 'baths'],
+ ['units', 'units'], ['use_desc', 'use_desc'], ['land_value', 'land_value'],
+ ['improvement_value', 'improvement_value'], ['total_value', 'total_value'], ['tax_year', 'tax_year'],
+ ['owner_name', 'owner_name'], ['zoning', 'zoning'], ['last_sale_date', 'last_sale_date'],
+ ['last_sale_price', 'last_sale_price'], ['extra', 'extra'],
+ ['source_url', 'sourceUrl'], ['source_key', 'sourceKey'], ['fetched_at', 'fetchedAt'], ['raw_source', 'rawSource'],
+];
+const DB_COLS = COL_MAP.map(([db]) => db);
export async function upsertParcels(rows: ParcelUpsertRow[]): Promise<number> {
+ const N = COL_MAP.length; // 27 cols → 2000 rows = 54k params (< 65535)
let done = 0;
for (let i = 0; i < rows.length; i += 2000) {
const chunk = rows.slice(i, i + 2000);
const params: unknown[] = [];
const values = chunk.map((r, j) => {
- const b = j * COLS.length;
+ const b = j * N;
// Sanitize owner_name at the shared chokepoint so every ingester routing
// through upsertParcels drops placeholder/junk owner values (see
// owner_guard.ts) without each emit point having to remember to.
- for (const c of COLS) {
- params.push(c === 'owner_name' ? sanitizeOwnerName(r.owner_name) : (r[c] ?? null));
+ for (const [dbCol, key] of COL_MAP) {
+ params.push(dbCol === 'owner_name' ? sanitizeOwnerName(r.owner_name) : (r[key] ?? null));
}
- return '(' + COLS.map((_, k) => `$${b + k + 1}`).join(',') + ')';
+ return '(' + DB_COLS.map((_, k) => `$${b + k + 1}`).join(',') + ')';
});
await query(
- `INSERT INTO parcel (${COLS.join(',')}) VALUES ${values.join(',')}
+ `INSERT INTO parcel (${DB_COLS.join(',')}) VALUES ${values.join(',')}
ON CONFLICT (county_fips, source_id) DO UPDATE SET
- ${COLS.filter(c => c !== 'county_fips' && c !== 'source_id').map(c => `${c}=EXCLUDED.${c}`).join(', ')}`,
+ ${DB_COLS.filter(c => c !== 'county_fips' && c !== 'source_id').map(c => `${c}=EXCLUDED.${c}`).join(', ')}`,
params,
);
done += chunk.length;
@@ -49,6 +66,36 @@ export async function upsertParcels(rows: ParcelUpsertRow[]): Promise<number> {
return done;
}
+/**
+ * TK-50: declare, once per source, which raw source attribute each of our
+ * parcel fields maps from — the "which attribute did field X come from" half of
+ * field-level provenance. Idempotent upsert into source_field_map; call at
+ * ingest time so provenance is queryable without re-reading ingester code.
+ *
+ * registerSourceFieldMap('oregon-rlis', { owner_name: 'OWNER', sqft: 'BLDGSQFT' }, 'RLIS taxlots')
+ */
+export async function registerSourceFieldMap(
+ sourceKey: string,
+ mapping: Record<string, string>,
+ notes?: string,
+): Promise<number> {
+ const entries = Object.entries(mapping).filter(([, attr]) => attr != null && attr !== '');
+ if (!entries.length) return 0;
+ const params: unknown[] = [];
+ const values = entries.map(([ourField, sourceAttr], j) => {
+ const b = j * 4;
+ params.push(sourceKey, ourField, sourceAttr, notes ?? null);
+ return `($${b + 1},$${b + 2},$${b + 3},$${b + 4})`;
+ });
+ await query(
+ `INSERT INTO source_field_map (source_key, our_field, source_attr, notes) VALUES ${values.join(',')}
+ ON CONFLICT (source_key, our_field) DO UPDATE SET
+ source_attr=EXCLUDED.source_attr, notes=EXCLUDED.notes, updated_at=NOW()`,
+ params,
+ );
+ return entries.length;
+}
+
/** Uppercase, collapse whitespace, strip ordinal suffixes so queries and stored
* addresses normalize the same way ("61ST AVE S" -> "61 AVE S"). Must mirror
* normAddressQuery in src/server/property.ts. */
← 1c33393 parcels: guard null/placeholder/junk owner names — shared sa
·
back to Nationalrealestate
·
TK-50: field-level provenance across ALL remaining parcel in 6ff7a81 →