← back to Nationalrealestate
src/ingest/parcels/arcgis_sales.ts
406 lines
/**
* Config-driven ArcGIS priced-DEED ingest — free/keyless county sales feeds that
* publish real recorded SALE PRICE (unlike SD, which AB-1785-gated it). Each
* source maps a page of ArcGIS features to normalized sale records → parcel row
* + a parcel_event(event_type='sale', amount=price) carrying grantor/grantee/doc
* where available, with a source_url back to the record.
*
* Verified 2026-07-26 (west-coast first):
* oregon-rlis — Portland metro tri-county (Multnomah/Washington/Clackamas),
* 428,872 priced (SALEPRICE + SALEDATE YYYYMM).
* spokane — Spokane County WA, 146,691 priced (gross_sale_price, buyer, deed type).
* alameda — Alameda County CA deed-transfer list, ~51,800 priced
* (value_from_trans_tax + grantor/grantee split across rows).
*
* npm run ingest:parcels -- oregon-rlis | spokane | alameda
*/
import { query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { upsertParcels, normAddress, registerParcelSource, registerSourceFieldMap, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
import { sanitizeSalePrice } from './price_guard.ts';
const PAGE = 2000;
const MAX_PER_RUN = Number(process.env.SALES_MAX_PER_RUN || 6000);
interface Rec {
fips: string; apn: string;
address?: string | null; city?: string | null; zip?: string | null;
lat?: number | null; lng?: number | null;
beds?: number | null; baths?: number | null; sqft?: number | null; year?: number | null;
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; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
const epochToISO = (v: unknown): string | null => { const x = Number(v); if (!Number.isFinite(x) || x <= 0) return null; const d = new Date(x); return d.toISOString().slice(0, 10); };
const yyyymm = (v: unknown): string | null => { const t = String(v ?? '').trim(); return /^\d{6}$/.test(t) ? `${t.slice(0, 4)}-${t.slice(4, 6)}-01` : null; };
function centroid(geom: any): [number | null, number | null] {
const ring = geom?.rings?.[0]; if (!Array.isArray(ring) || !ring.length) return [null, null];
let sx = 0, sy = 0; for (const [x, y] of ring) { sx += x; sy += y; }
return [+(sx / ring.length).toFixed(6), +(sy / ring.length).toFixed(6)];
}
const clean = (v: unknown): string | null => { const t = s(v); return t ? t.replace(/\s+/g, ' ') : null; };
// price may be a real number OR a string ('117500', ' 000001060000.'); the shared
// guard strips currency formatting and rejects negative/zero/non-finite amounts.
// (Delegating here also fixes a latent bug: the old inline strip removed the
// minus sign, so a corrupt "-400" would have become 400 instead of being dropped.)
const anyPrice = (v: unknown): number | null => sanitizeSalePrice(v as string | number | null | undefined);
// date may be epoch-ms, 'YYYY-MM-DD', or 'M/D/YYYY'.
const anyDate = (v: unknown): string | null => {
if (v == null || v === '') return null;
if (typeof v === 'number' || /^\d{11,}$/.test(String(v).trim())) return epochToISO(Number(v));
const t = String(v).trim();
let m = t.match(/^(\d{4})-(\d{2})-(\d{2})/); if (m) return `${m[1]}-${m[2]}-${m[3]}`;
m = t.match(/^(\d{1,2})\/(\d{1,2})\/(\d{4})/); if (m) return `${m[3]}-${m[1].padStart(2, '0')}-${m[2].padStart(2, '0')}`; // M/D/YYYY -> YYYY-MM-DD
return null;
};
// ── source configs ────────────────────────────────────────────────────────
const OR_FIPS: Record<string, string> = { M: '41051', W: '41067', C: '41005' };
const SOURCES: Record<string, Source> = {
'oregon-rlis': {
key: 'oregon-rlis', label: 'Oregon Metro RLIS (Portland tri-county)',
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), 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), 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>();
for (const f of fs) {
const a = f.attributes; const apn = s(a.apn); const doc = (s(a.doc_prefix) || '') + (s(a.doc_series) || '');
if (!apn || !doc) continue;
const price = anyPrice(a.value_from_trans_tax); // zero-padded dollars e.g. '000001060000.'
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, 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';
if (nm === 'TRANSFEREE') r.grantee = r.grantee || 'recorded';
byDoc.set(key, r);
}
return [...byDoc.values()];
},
},
deschutes: {
key: 'deschutes', label: 'Deschutes County OR (Bend)', cursorKey: 'deschutes', geometry: false, orderBy: 'Taxlot', pageSize: 1000, // server caps pages at 1000
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[] = [];
for (const f of fs) {
const a = f.attributes; const apn = s(a.Taxlot); if (!apn) continue;
for (const nx of ['1', '2']) {
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}`, raw: a } as Rec);
}
}
return out;
},
},
crook: {
key: 'crook', label: 'Crook County OR (Prineville)', cursorKey: 'crook', geometry: false, orderBy: 'OBJECTID', pageSize: 2000,
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), 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), 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), raw: a } as Rec);
} return out; },
},
colorado: {
key: 'colorado', label: 'Colorado (statewide composite)', cursorKey: 'colorado', geometry: false, orderBy: 'OBJECTID', pageSize: 2000,
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), 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), 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), 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), 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), 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), 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,
// rich Yakima assessor parcel layer — attribute keys come back FULLY-QUALIFIED (DBO.Parcels.*).
// GROSS_SALE_PRICE(double)+SALE_DATE(string M/D/YYYY)+EXCISE_NUMBER(doc)+GRANTOR_NAME+owner(LAST/FIRST/ORG)+situs+use+yrblt+sqft+beds/baths+market land/impvt.
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']);
const person = [clean(a['DBO.Parcels.FIRST_NAME']), clean(a['DBO.Parcels.LAST_NAME'])].filter(Boolean).join(' ').trim() || null;
const owner = org || person;
const land = num(a['DBO.Parcels.MKT_LAND']) || 0, impvt = num(a['DBO.Parcels.MKT_IMPVT']) || 0;
return { fips: '53077', apn, price: anyPrice(a['DBO.Parcels.GROSS_SALE_PRICE']), saleDate: anyDate(a['DBO.Parcels.SALE_DATE']),
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']), raw: a } as Rec; }).filter(Boolean) as Rec[],
},
};
async function fetchPage(src: Source, offset: number): Promise<any[]> {
const u = new URL(src.layer + '/query');
u.searchParams.set('where', src.where);
u.searchParams.set('outFields', src.outFields);
u.searchParams.set('orderByFields', src.orderBy + ' ASC');
u.searchParams.set('resultOffset', String(offset));
u.searchParams.set('resultRecordCount', String(src.pageSize || PAGE));
u.searchParams.set('returnGeometry', src.geometry ? 'true' : 'false');
if (src.geometry) u.searchParams.set('outSR', '4326');
u.searchParams.set('f', 'json');
// some county servers (Spokane MapServer) are slow — retry transient timeouts.
let lastErr: any;
for (let attempt = 0; attempt < 3; attempt++) {
try {
const res = await fetch(u, { signal: AbortSignal.timeout(90_000) });
if (!res.ok) throw new Error(`${src.key} ${res.status}: ${(await res.text()).slice(0, 140)}`);
const j: any = await res.json();
if (j.error) throw new Error(`${src.key} error: ${JSON.stringify(j.error).slice(0, 140)}`);
return j.features || [];
} catch (e: any) { lastErr = e; if (attempt < 2) await new Promise(r => setTimeout(r, 2000 * (attempt + 1))); }
}
throw lastErr;
}
async function countTotal(src: Source): Promise<number> {
const u = new URL(src.layer + '/query');
u.searchParams.set('where', src.where); u.searchParams.set('returnCountOnly', 'true'); u.searchParams.set('f', 'json');
const j: any = await (await fetch(u, { signal: AbortSignal.timeout(30_000) })).json();
return Number(j.count || 0);
}
export function ingestArcgisSales(key: string) {
const src = SOURCES[key];
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;
if (offset >= total) offset = 0;
console.log(`[${src.key}] ${src.label} total=${total}, resume offset=${offset}`);
const rows: ParcelUpsertRow[] = []; const events: Rec[] = []; const seenFips = new Set<string>();
let fetched = 0;
while (fetched < MAX_PER_RUN) {
const feats = await fetchPage(src, offset);
if (!feats.length) break;
for (const r of src.toRecords(feats)) {
if (!r.apn) continue;
r.saleDate = sanitizeEventDate(r.saleDate); // drop future/absurd bad-entry dates (shared guard)
seenFips.add(r.fips);
const gisUrl = `${src.layer}/query?where=${encodeURIComponent(src.outFields.split(',')[0] + "='" + r.apn + "'")}&outFields=*&f=html`;
rows.push({
county_fips: r.fips, source_id: r.apn,
address: r.address ?? null, norm_address: r.address ? normAddress(r.address) : null,
city: r.city ?? null, zip: r.zip ?? null, lat: r.lat ?? null, lng: r.lng ?? null,
year_built: r.year ?? null, sqft: r.sqft ?? null, beds: r.beds ?? null, baths: r.baths ?? null,
units: null, use_desc: r.use ?? null, land_value: null, improvement_value: null, total_value: r.totalValue ?? null,
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
}
fetched += feats.length; offset += feats.length;
if (feats.length < (src.pageSize || PAGE)) break;
}
// A parcel can have MANY sales in one batch (Alameda) — the parcel row must be
// unique per (fips,source_id) or ON CONFLICT DO UPDATE errors "cannot affect row
// a second time". Keep the most-recent-sale row per parcel; every sale still lands
// in parcel_event (its UNIQUE key includes doc_number).
const byParcel = new Map<string, ParcelUpsertRow>();
for (const r of rows) {
const k = r.county_fips + '|' + r.source_id; const prev = byParcel.get(k);
if (!prev || (r.last_sale_date || '') > (prev.last_sale_date || '')) byParcel.set(k, r);
}
const dedupRows = [...byParcel.values()];
const up = await upsertParcels(dedupRows);
// sale events with price + parties + public-record link
for (let i = 0; i < events.length; i += 500) {
const chunk = events.slice(i, i + 500);
await query(
`INSERT INTO parcel_event (county_fips, source_id, event_type, event_date, amount, doc_type, doc_number, source, source_url, detail)
VALUES ${chunk.map((_, j) => { const b = j * 9; return `($${b+1},$${b+2},'sale',$${b+3}::date,$${b+4},$${b+5},$${b+6},$${b+7},$${b+8},$${b+9}::jsonb)`; }).join(',')}
ON CONFLICT (county_fips, source_id, event_type, event_date, doc_number) DO NOTHING`,
chunk.flatMap(e => [e.fips, e.apn, e.saleDate, e.price, e.docType ?? null, e.docNum ?? '', src.cursorKey,
e.address /* gisUrl */, JSON.stringify({ grantor: e.grantor ?? null, grantee: e.grantee ?? null })]),
);
}
// per-parcel GIS deep link (deduped — one link per parcel)
for (let i = 0; i < dedupRows.length; i += 400) {
const chunk = dedupRows.slice(i, i + 400);
await query(
`INSERT INTO parcel_links (county_fips, source_id, kind, url, label) VALUES ${
chunk.map((_, j) => { const b = j * 5; return `($${b+1},$${b+2},$${b+3},$${b+4},$${b+5})`; }).join(',')
} ON CONFLICT (county_fips, source_id, kind) DO UPDATE SET url=EXCLUDED.url, label=EXCLUDED.label`,
chunk.flatMap(r => [r.county_fips, r.source_id, 'gis',
`${src.layer}/query?where=${encodeURIComponent(src.outFields.split(',')[0] + "='" + r.source_id + "'")}&outFields=*&f=html`,
`${src.label} record`]),
);
}
for (const fips of seenFips) await registerParcelSource(fips, src.layer, `${src.label} — free priced deeds (ArcGIS REST)`);
const nextOffset = offset >= total ? 0 : offset;
await query(`INSERT INTO ingest_cursor (source, next_offset, total) VALUES ($1,$2,$3)
ON CONFLICT (source) DO UPDATE SET next_offset=$2, total=$3, updated_at=NOW()`, [src.cursorKey, nextOffset, total]);
await closeRun(runId, 'ok', { upserted: up, notes: `${src.label}: ${up} parcels, ${events.length} sales (offset ${offset}/${total})` });
console.log(`[${src.key}] done: ${up} parcels, ${events.length} priced sales`);
return { upserted: up };
} catch (e: any) {
await closeRun(runId, 'failed', { notes: e.message });
throw e;
}
};
}