`src/ingest/parcels/wake_nc.ts`

178 lines
/**
 * Wake County NC (Raleigh, FIPS 37183) parcel ingest — dedicated full-universe
 * adapter (applies the Miami-Dade DTD verdict B precedent: rich full-universe
 * ArcGIS county gets a dedicated adapter, not the sale-cursor generic one, so the
 * real OWNER field is preserved).
 *
 * Source: the county's authoritative keyless ArcGIS MapServer (437,234 parcels,
 * polygon geom, supportsPagination, 2k/page, $0):
 *   https://maps.wakegov.com/arcgis/rest/services/Property/Parcels/MapServer/0
 * (The county's www.wake.gov *data-download page* 404s, but this REST service is
 * live — that URL drift is what stalled the earlier pass.)
 *
 * Richest county in the system: OWNER, SITE_ADDRESS/CITY/ZIP, LAND_VAL +
 * BLDG_VAL + TOTAL_VALUE_ASSD (assessed values), HEATEDAREA (sqft), YEAR_BUILT,
 * TYPE_USE_DECODE (use), UNITS, TOTSALPRICE + SALE_DATE (epoch-ms, current to
 * 2024+), DEED_BOOK/PAGE. Only beds/baths + zoning are absent. Bulk guard:
 * a deed (DEED_BOOK-DEED_PAGE) spanning >1 REID is an assemblage → price nulled
 * + flagged in-DB (Miami/Franklin lesson).
 *
 * Run: NODE_OPTIONS=--max-old-space-size=4096 npm run ingest:parcels wake
 */
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';

const FIPS = '37183';
const SOURCE_KEY = 'wake_nc';
const LAYER = 'https://maps.wakegov.com/arcgis/rest/services/Property/Parcels/MapServer/0';
const PAGE = 2000;
// TK-50 field-level provenance: our parcel field → the exact source attribute it
// was mapped from in the loop below. Registered into source_field_map at ingest.
const FIELD_MAP: Record<string, string> = {
  source_id: 'REID', address: 'SITE_ADDRESS', city: 'CITY_DECODE', zip: 'ZIPNUM',
  year_built: 'YEAR_BUILT', sqft: 'HEATEDAREA', units: 'UNITS', use_desc: 'TYPE_USE_DECODE',
  land_value: 'LAND_VAL', improvement_value: 'BLDG_VAL', total_value: 'TOTAL_VALUE_ASSD',
  owner_name: 'OWNER', last_sale_price: 'TOTSALPRICE', last_sale_date: 'SALE_DATE',
};

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; };
// reject <=0 AND sub-1971 (31_536_000_000ms) — Wake's oldest real digital deed is
// 1971; anything earlier is a mainframe near-epoch precision artifact, not a sale.
const epochToISO = (v: unknown): string | null => { const x = Number(v); if (!Number.isFinite(x) || x < 31_536_000_000) return null; return sanitizeEventDate(new Date(x).toISOString().slice(0, 10)); };
/** plain vertex-average of ring[0] — NOT a true polygon centroid (fine for a map
 *  pin; multipart parcels only get the first ring). */
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 OUT = ['REID', 'PIN_NUM', 'OWNER', 'SITE_ADDRESS', 'CITY_DECODE', 'ZIPNUM',
  'LAND_VAL', 'BLDG_VAL', 'TOTAL_VALUE_ASSD', 'HEATEDAREA', 'YEAR_BUILT', 'UNITS',
  'TYPE_USE_DECODE', 'DESIGN_STYLE_DECODE', 'TOTSALPRICE', 'SALE_DATE', 'DEED_BOOK', 'DEED_PAGE', 'DEED_DATE'].join(',');

async function fetchPage(offset: number): Promise<any[]> {
  const u = new URL(LAYER + '/query');
  u.searchParams.set('where', '1=1');
  u.searchParams.set('outFields', OUT);
  u.searchParams.set('orderByFields', 'OBJECTID ASC');
  u.searchParams.set('resultOffset', String(offset));
  u.searchParams.set('resultRecordCount', String(PAGE));
  u.searchParams.set('returnGeometry', 'true');
  u.searchParams.set('outSR', '4326');
  u.searchParams.set('f', 'json');
  let lastErr: any;
  for (let a = 0; a < 3; a++) {
    try {
      const res = await fetch(u, { signal: AbortSignal.timeout(120_000) });
      if (!res.ok) throw new Error(`wake ${res.status}: ${(await res.text()).slice(0, 140)}`);
      const j: any = await res.json();
      if (j.error) throw new Error(`wake error: ${JSON.stringify(j.error).slice(0, 140)}`);
      return j.features || [];
    } catch (e) { lastErr = e; if (a < 2) await new Promise(r => setTimeout(r, 2000 * (a + 1))); }
  }
  throw lastErr;
}

export async function ingestWake(): Promise<{ upserted: number }> {
  const runId = await openRun('parcel_wake_nc', LAYER);
  // TK-50: run start time — the fetched_at stamped on every row this run touches.
  const fetchedAt = new Date().toISOString();
  try {
    await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'Wake County NC Assessor (authoritative ArcGIS MapServer)');
    let offset = 0, seen = 0, upserted = 0, sales = 0;
    for (;;) {
      const feats = await fetchPage(offset);
      if (!feats.length) break;
      const batch: ParcelUpsertRow[] = [];
      const events: any[] = [];
      for (const f of feats) {
        const a = f.attributes || {};
        const reid = s(a.REID); if (!reid) continue;
        seen++;
        const [lng, lat] = centroid(f.geometry);
        const addr = s(a.SITE_ADDRESS);
        const norm = addr ? normAddress(addr) : null;
        const saleDate = epochToISO(a.SALE_DATE);
        const salePrice = num(a.TOTSALPRICE);
        const deed = `${s(a.DEED_BOOK) ?? ''}-${s(a.DEED_PAGE) ?? ''}`;
        batch.push({
          county_fips: FIPS, source_id: reid,
          address: norm, norm_address: norm,
          city: s(a.CITY_DECODE), zip: s(a.ZIPNUM),
          lat, lng,
          year_built: num(a.YEAR_BUILT), sqft: num(a.HEATEDAREA), beds: null, baths: null, units: num(a.UNITS),
          use_desc: s(a.TYPE_USE_DECODE),
          land_value: num(a.LAND_VAL), improvement_value: num(a.BLDG_VAL), total_value: num(a.TOTAL_VALUE_ASSD),
          tax_year: null, owner_name: s(a.OWNER), zoning: null,
          last_sale_date: saleDate, last_sale_price: salePrice,
          extra: JSON.stringify({
            pin: s(a.PIN_NUM) || undefined, design_style: s(a.DESIGN_STYLE_DECODE) || undefined,
            // exclude all-zero/blank deed sentinels ('-', '000000-00000') — the county
          // writes those for govt/right-of-way parcels with no deed on record.
          latest_deed: /^[-0]*$/.test(deed) ? undefined : deed,
            note: 'Wake County NC Assessor — assessed value; no beds/baths or zoning in this layer',
          }),
          // ── TK-50 field-level provenance (record-level: one feature → this row) ──
          sourceKey: SOURCE_KEY,
          sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`REID='${reid}'`)}&outFields=*&f=html`,
          fetchedAt,
          rawSource: JSON.stringify(a),
        });
        if (salePrice && saleDate) events.push({ reid, price: salePrice, date: saleDate, deed });
      }
      upserted += await upsertParcels(batch);
      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 => { const realDeed = e.deed && !/^[-0]*$/.test(e.deed) ? e.deed : null;
            return [FIPS, e.reid, e.date, e.price, 'Deed', realDeed ?? e.date,
              'parcel_wake_nc', `https://services.wake.gov/realestate/Account.asp?id=${e.reid}`, JSON.stringify({ deed: realDeed })]; }),
        );
        sales += chunk.length;
      }
      offset += feats.length;
      if (seen % 100000 < feats.length) console.log(`[wake] ${upserted} upserted, ${sales} sales`);
      if (feats.length < PAGE) break;
    }
    if (upserted < 380000) throw new Error(`only ${upserted} Wake parcels (expected ~437k) — layer/paging drift?`);

    // in-DB bulk flag: a deed spanning >1 REID is an assemblage; null the un-allocatable
    // price + flag. Resolve bulk parcels to their source_id in a materialized CTE, then
    // UPDATE joining on the PKEY (county_fips, source_id) — NOT on the JSONB expression
    // extra->>'latest_deed', which can't use an index and seq-scanned for ~53min on the
    // 5M-row prod parcel table. The pkey join makes the write index-driven.
    const b = await query(
      `WITH deeds AS MATERIALIZED (
         SELECT extra->>'latest_deed' AS deed, COUNT(*)::int AS cnt
           FROM parcel WHERE county_fips=$1 AND extra->>'latest_deed' IS NOT NULL AND extra->>'latest_deed' !~ '^[-0]*$'
          GROUP BY 1 HAVING COUNT(*) > 1),
       targets AS MATERIALIZED (
         SELECT p.source_id, d.cnt FROM parcel p JOIN deeds d ON p.extra->>'latest_deed' = d.deed
          WHERE p.county_fips=$1)
       UPDATE parcel p SET last_sale_price = NULL,
         extra = p.extra || jsonb_build_object('bulk_sale', true, 'bulk_deed_folio_count', t.cnt)
         FROM targets t WHERE p.county_fips=$1 AND p.source_id = t.source_id RETURNING 1`, [FIPS]);
    const bulk = b.rowCount ?? 0;

    const n = await registerParcelSource(FIPS, LAYER,
      `Wake County NC Assessor (authoritative ArcGIS MapServer) — ${upserted} parcels: owner, site address, assessed LAND+BLDG+TOTAL value, sqft (heated area), year built, use, units, latest sale (price+date+deed) → ${sales} parcel_event sale rows. No beds/baths or zoning in this layer. ${bulk} multi-parcel-deed (assemblage) sales flagged bulk + price nulled.`);
    await closeRun(runId, 'ok', { upserted, notes: `Wake NC: ${upserted} parcels, ${sales} sales, ${bulk} bulk-flagged` });
    console.log(`[wake] ok: ${upserted} parcels, ${sales} sales, ${bulk} bulk (registry ${n})`);
    return { upserted };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e).slice(0, 500) });
    throw e;
  }
}

if (import.meta.url === `file://${process.argv[1]}`) {
  ingestWake().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}