`src/ingest/parcels/upsert.ts`

124 lines
/**
 * Shared chunked upsert into `parcel` — full column set (incl. beds/baths/sales,
 * unlike the NYC-era subset). PG caps a statement at 65535 bind params;
 * 23 cols -> chunks of 2000 rows = 46k params.
 */
import { query } from '../../../db/pool.ts';
import { sanitizeOwnerName } from './owner_guard.ts';

export interface ParcelUpsertRow {
  county_fips: string; source_id: string;
  address: string | null; norm_address: string | null; city: string | null; zip: string | null;
  lat: number | null; lng: number | null;
  year_built: number | null; sqft: number | null; beds: number | null; baths: number | null;
  units: number | null; use_desc: string | null;
  land_value: number | null; improvement_value: number | null; total_value: number | null;
  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
}

// 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 * 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 [dbCol, key] of COL_MAP) {
        params.push(dbCol === 'owner_name' ? sanitizeOwnerName(r.owner_name) : (r[key] ?? null));
      }
      return '(' + DB_COLS.map((_, k) => `$${b + k + 1}`).join(',') + ')';
    });
    await query(
      `INSERT INTO parcel (${DB_COLS.join(',')}) VALUES ${values.join(',')}
       ON CONFLICT (county_fips, source_id) DO UPDATE SET
         ${DB_COLS.filter(c => c !== 'county_fips' && c !== 'source_id').map(c => `${c}=EXCLUDED.${c}`).join(', ')}`,
      params,
    );
    done += chunk.length;
  }
  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. */
export function normAddress(addr: string): string {
  return addr.trim().toUpperCase()
    .replace(/[.,#]/g, ' ')
    .replace(/\b(\d+)(?:ST|ND|RD|TH)\b/g, '$1')
    .replace(/\s+/g, ' ')
    .trim();
}

export async function registerParcelSource(
  countyFips: string, pathOrUrl: string, notes: string,
): Promise<number> {
  const cnt = await query<{ n: string }>(`SELECT COUNT(*)::text AS n FROM parcel WHERE county_fips=$1`, [countyFips]);
  const n = Number(cnt.rows[0].n);
  await query(
    `INSERT INTO parcel_source (county_fips, source_kind, path_or_url, parcel_count, notes, loaded_at)
     VALUES ($1,'pg',$2,$3,$4,NOW())
     ON CONFLICT (county_fips) DO UPDATE SET source_kind='pg', path_or_url=EXCLUDED.path_or_url,
       parcel_count=EXCLUDED.parcel_count, notes=EXCLUDED.notes, loaded_at=NOW()`,
    [countyFips, pathOrUrl, n, notes],
  );
  return n;
}