`src/ingest/parcels/miami_dade.ts`

206 lines
/**
 * Miami-Dade County FL (FIPS 12086) parcel ingest — DTD verdict B (2026-07-26,
 * unanimous 5/5): a dedicated full-universe adapter, NOT a config on the generic
 * sale-cursor arcgis_sales adapter, because this source is far richer.
 *
 * Source: the Property Appraiser GIS layer `MDC.PaGis` on the county's keyless
 * ArcGIS MapServer (943,019 parcels, point geom, supportsPagination, 20k/page, $0):
 *   https://gisweb.miamidade.gov/arcgis/rest/services/MD_ComparableSales/MapServer/5
 *
 * POPULATED per parcel: FOLIO, site address/city/zip, TRUE_OWNER1 (current owner),
 * DOR_DESC (use), PRIMARY_ZONE (zoning), lat/lng, and up to THREE recorded sales
 * (DOS_n=YYYYMMDD, PRICE_n, GRANTOR_n, GRANTEE_n, OR_BK_n/OR_PG_n deed ref,
 * QU_FLG_n Q=qualified/arms-length). Assessed value / beds / baths / sqft /
 * year-built columns exist in the schema but are globally NULL in this public
 * layer (verified: count(TOTAL_VAL_CUR>0)=0) — recorded null, noted.
 *
 * Bulk-sale guard (Franklin lesson): assemblage deeds stamp the same total price on
 * every FOLIO they cover. There is no ParcelCount field, but a deed (OR_BK|OR_PG)
 * that spans >1 FOLIO is a bundle — we detect that, null the un-allocatable
 * last_sale_price, and flag bulk_sale in extra. Every sale still lands in
 * parcel_event (with a bulk marker) so nothing is lost.
 *
 * Run: NODE_OPTIONS=--max-old-space-size=6144 npm run ingest:parcels miami
 */
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 = '12086';
const SOURCE_KEY = 'miami_dade';
const LAYER = 'https://gisweb.miamidade.gov/arcgis/rest/services/MD_ComparableSales/MapServer/5';
const PAGE = 20000;
// TK-50 field-level provenance: our parcel field → the exact source attribute it
// maps from. last_sale_* come from the NEWEST of the up-to-3 recorded sales
// (DOS_n/PRICE_n slots, n=1 the county's newest) — declared as the slot family.
const FIELD_MAP: Record<string, string> = {
  source_id: 'FOLIO', address: 'TRUE_SITE_ADDR', city: 'TRUE_SITE_CITY', zip: 'TRUE_SITE_ZIP_CODE',
  use_desc: 'DOR_DESC', owner_name: 'TRUE_OWNER1', zoning: 'PRIMARY_ZONE',
  last_sale_date: 'DOS_1|DOS_2|DOS_3 (newest recorded sale)',
  last_sale_price: 'PRICE_1|PRICE_2|PRICE_3 (newest recorded sale)',
};

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; };
/** "20210623" -> "2021-06-23"; future/absurd/non-calendar -> null (shared guard) */
const toDate = (v: unknown): string | null => {
  const t = String(v ?? '').trim();
  const iso = /^\d{8}$/.test(t) ? `${t.slice(0, 4)}-${t.slice(4, 6)}-${t.slice(6, 8)}` : null;
  return sanitizeEventDate(iso);
};

const OUT = [
  'FOLIO', 'TRUE_SITE_ADDR', 'TRUE_SITE_CITY', 'TRUE_SITE_ZIP_CODE', 'TRUE_OWNER1', 'TRUE_OWNER2',
  'DOR_DESC', 'DOR_CODE_CUR', 'PRIMARY_ZONE',
  'DOS_1', 'PRICE_1', 'GRANTOR_1', 'GRANTEE_1', 'OR_BK_1', 'OR_PG_1', 'QU_FLG_1',
  'DOS_2', 'PRICE_2', 'GRANTOR_2', 'GRANTEE_2', 'OR_BK_2', 'OR_PG_2', 'QU_FLG_2',
  'DOS_3', 'PRICE_3', 'GRANTOR_3', 'GRANTEE_3', 'OR_BK_3', 'OR_PG_3', 'QU_FLG_3',
].join(',');

interface Sale { n: number; date: string; price: number | null; grantor: string | null; grantee: string | null; deed: string; qualified: boolean; }

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(`miami ${res.status}: ${(await res.text()).slice(0, 140)}`);
      const j: any = await res.json();
      if (j.error) throw new Error(`miami 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;
}

/** pull the up-to-3 sales off a feature's attributes, newest first */
function salesOf(a: any): Sale[] {
  const out: Sale[] = [];
  for (const n of [1, 2, 3]) {
    const date = toDate(a['DOS_' + n]); const price = num(a['PRICE_' + n]);
    if (!date && !price) continue;
    out.push({
      n, date: date ?? '', price, grantor: s(a['GRANTOR_' + n]), grantee: s(a['GRANTEE_' + n]),
      deed: `${s(a['OR_BK_' + n]) ?? ''}-${s(a['OR_PG_' + n]) ?? ''}`,
      qualified: String(a['QU_FLG_' + n] ?? '').trim().toUpperCase() === 'Q',
    });
  }
  // newest first; deterministic tiebreak by slot (_1 is the county's newest) on date ties
  return out.sort((x, y) => y.date.localeCompare(x.date) || (x.n - y.n));
}

export async function ingestMiami(): Promise<{ upserted: number }> {
  const runId = await openRun('parcel_miami_dade', 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, 'Miami-Dade County FL Property Appraiser GIS (MDC.PaGis, keyless ArcGIS)');
    // Stream: fetch a page -> upsert its parcels + insert its sale events -> release.
    // Memory-bounded (never holds the whole county); partial progress persists.
    // Bulk-sale detection is deferred to an in-DB post-pass on the stored latest_deed.
    let offset = 0, seen = 0, upserted = 0, saleCount = 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 folio = s(a.FOLIO); if (!folio) continue;
        seen++;
        const sales = salesOf(a);
        const addr = s(a.TRUE_SITE_ADDR);
        const norm = addr ? normAddress(addr) : null;
        const latest = sales[0] || null;
        batch.push({
          county_fips: FIPS, source_id: folio,
          address: norm, norm_address: norm,
          city: s(a.TRUE_SITE_CITY), zip: s(a.TRUE_SITE_ZIP_CODE),
          lat: num(f.geometry?.y), lng: num(f.geometry?.x),
          year_built: null, sqft: null, beds: null, baths: null, units: null,
          use_desc: s(a.DOR_DESC), land_value: null, improvement_value: null, total_value: null, tax_year: null,
          owner_name: s(a.TRUE_OWNER1), zoning: s(a.PRIMARY_ZONE),
          last_sale_date: latest?.date || null, last_sale_price: latest?.price ?? null,
          extra: JSON.stringify({
            owner2: s(a.TRUE_OWNER2) || undefined, dor_code: s(a.DOR_CODE_CUR) || undefined,
            // exclude the PA's all-zero "no deed on record" placeholder so it never
            // groups thousands of un-recorded parcels into a fake assemblage.
            latest_deed: latest?.deed && latest.deed !== '-' && !/^0+-0+$/.test(latest.deed) ? latest.deed : undefined,
            latest_sale_qualified: latest ? latest.qualified : undefined,
            note: 'no assessed value / beds / baths / sqft / year-built in the public PaGis layer',
          }),
          // ── TK-50 field-level provenance (record-level: one feature → this row) ──
          sourceKey: SOURCE_KEY,
          sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`FOLIO='${folio}'`)}&outFields=*&f=html`,
          fetchedAt,
          rawSource: JSON.stringify(a),
        });
        // every dated transfer becomes an event (incl. $0 non-arms-length) so the
        // parcel's last_sale_date never points to a missing event row; price/doc_type
        // distinguishes them. Only truly undated rows are skipped.
        for (const sale of sales) if (sale.date) events.push({ folio, ...sale });
      }
      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 => [FIPS, e.folio, e.date, e.price, e.qualified ? 'Qualified' : (e.price ? 'Unqualified' : 'Non-Arms-Length'), e.deed && e.deed !== '-' ? e.deed : `${e.date}-${e.n}`,
            'parcel_miami_dade', `https://www.miamidade.gov/Apps/PA/propertysearch/#/?folio=${e.folio}`,
            JSON.stringify({ grantor: e.grantor, grantee: e.grantee, qualified: e.qualified })]),
        );
        saleCount += chunk.length;
      }
      offset += feats.length;
      if (seen % 100000 < feats.length) console.log(`[miami] ${upserted} upserted, ${saleCount} sales`);
      if (feats.length < PAGE) break;
    }
    if (upserted < 800000) throw new Error(`only ${upserted} Miami-Dade parcels (expected ~943k) — layer/paging drift?`);

    // ── in-DB bulk-sale flag: a deed (latest_deed) spanning >1 folio is an assemblage;
    // its price is the bundle total, not per-parcel. Null the price + flag, in one pass.
    // Resolve bulk parcels to source_id first, then UPDATE joining on the PKEY
    // (county_fips, source_id) — NOT the unindexable JSONB expr extra->>'latest_deed',
    // which seq-scanned the 5M-row prod table for ~53min on a sibling county (TK-00037).
    const b = await query<{ n: string }>(
      `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+-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 AS n`, [FIPS]);
    const bulk = b.rowCount ?? 0;

    const n = await registerParcelSource(FIPS, LAYER,
      `Miami-Dade County FL Property Appraiser GIS (MDC.PaGis, keyless ArcGIS) — ${upserted} parcels: site address, current owner (TRUE_OWNER1), use (DOR), zoning, lat/lng, and up to 3 recorded sales each (date/price/grantor/grantee/deed/qualified) fanned into parcel_event. Assessed value/beds/baths/sqft/year-built are NULL in this public layer. ${bulk} parcels had a bulk (multi-folio deed) latest sale — price nulled + flagged.`);
    await closeRun(runId, 'ok', { upserted, notes: `Miami-Dade FL: ${upserted} parcels, ${saleCount} sales, ${bulk} bulk-flagged` });
    console.log(`[miami] ok: ${upserted} parcels, ${saleCount} 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]}`) {
  ingestMiami().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}