`src/ingest/crosswalk.ts`

201 lines
/**
 * Region-crosswalk helpers shared by ingest jobs.
 * Loads the full region table once, builds lookup maps, consults
 * data/crosswalk-overrides.json before any fuzzy match, and collects
 * unmatched source regions into data/crosswalk-unmatched.json (skipped, never guessed).
 */
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'node:fs';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { query } from '../../db/pool.ts';

const __dirname = dirname(fileURLToPath(import.meta.url));
export const DATA_DIR = join(__dirname, '..', '..', 'data');
const OVERRIDES_PATH = join(DATA_DIR, 'crosswalk-overrides.json');
const UNMATCHED_PATH = join(DATA_DIR, 'crosswalk-unmatched.json');

export interface RegionRow {
  id: number;
  region_type: string;
  canonical_key: string;
  name: string;
  state_code: string | null;
  fips: string | null;
  cbsa_code: string | null;
}

export interface RegionMaps {
  byFips: Map<string, RegionRow>;
  byCbsa: Map<string, RegionRow>;
  byStateCode: Map<string, RegionRow>;
  byCountyNameState: Map<string, RegionRow>; // "los angeles county|CA"
  byCanonicalKey: Map<string, RegionRow>;
  metros: RegionRow[];
}

export async function loadRegionMaps(): Promise<RegionMaps> {
  const r = await query<RegionRow>(
    `SELECT id, region_type, canonical_key, name, state_code, fips, cbsa_code FROM region`,
  );
  const maps: RegionMaps = {
    byFips: new Map(),
    byCbsa: new Map(),
    byStateCode: new Map(),
    byCountyNameState: new Map(),
    byCanonicalKey: new Map(),
    metros: [],
  };
  for (const row of r.rows) {
    maps.byCanonicalKey.set(row.canonical_key, row);
    if (row.region_type === 'county' && row.fips) {
      maps.byFips.set(row.fips, row);
      if (row.state_code) maps.byCountyNameState.set(normName(row.name) + '|' + row.state_code, row);
    }
    if (row.region_type === 'metro') {
      maps.metros.push(row);
      if (row.cbsa_code) maps.byCbsa.set(row.cbsa_code, row);
    }
    if (row.region_type === 'state' && row.state_code) maps.byStateCode.set(row.state_code, row);
  }
  return maps;
}

export function normName(s: string): string {
  // strip diacritics via NFD + combining-mark range (Dona Ana vs Doña Ana)
  return s.toLowerCase().normalize('NFD').replace(/[̀-ͯ]/g, '').replace(/\s+/g, ' ').trim();
}

export function loadOverrides(source: string): Record<string, string> {
  if (!existsSync(OVERRIDES_PATH)) return {};
  const all = JSON.parse(readFileSync(OVERRIDES_PATH, 'utf8'));
  return all[source] || {};
}

/** Metro matcher for Gazetteer CBSA names like "Los Angeles-Long Beach-Anaheim, CA Metro Area". */
export function buildMetroMatcher(maps: RegionMaps) {
  const exact = new Map<string, RegionRow>(); // "los angeles-long beach-anaheim, ca"
  const cityState = new Map<string, RegionRow>(); // "los angeles|CA" (each -segment × each state)
  for (const m of maps.metros) {
    const stripped = m.name.replace(/\s+(Metro|Micro)\s+Area$/i, '');
    exact.set(normName(stripped), m);
    const comma = stripped.lastIndexOf(',');
    if (comma < 0) continue;
    const cityPart = stripped.slice(0, comma);
    const states = stripped.slice(comma + 1).trim().split('-');
    const firstCity = cityPart.split(/[-/]/)[0]; // "Louisville/Jefferson County" -> "Louisville"
    for (const st of states) {
      const key = normName(firstCity) + '|' + st.trim();
      if (!cityState.has(key)) cityState.set(key, m); // first (largest CBSA seeded order) wins; ties skipped by has-check
    }
  }
  /** name = e.g. "Los Angeles, CA" (Zillow style). Returns region or null. */
  return function matchMetro(name: string): { region: RegionRow; method: string } | null {
    const comma = name.lastIndexOf(',');
    const ex = exact.get(normName(name));
    if (ex) return { region: ex, method: 'cbsa_name_exact' };
    if (comma < 0) return null;
    const city = name.slice(0, comma);
    const st = name.slice(comma + 1).trim();
    const cs = cityState.get(normName(city) + '|' + st);
    if (cs) return { region: cs, method: 'cbsa_city_prefix' };
    return null;
  };
}

export class UnmatchedCollector {
  private items: Record<string, unknown[]> = {};
  private seen = new Set<string>();
  add(source: string, item: unknown) {
    const key = source + '|' + JSON.stringify(item);
    if (this.seen.has(key)) return; // same region unmatched by multiple files in one run (zhvi+zori)
    this.seen.add(key);
    (this.items[source] ||= []).push(item);
  }
  count(source: string): number {
    return (this.items[source] || []).length;
  }
  /** Merge this run's unmatched sets into data/crosswalk-unmatched.json (replaces per source key). */
  flush() {
    mkdirSync(DATA_DIR, { recursive: true });
    let existing: Record<string, unknown[]> = {};
    if (existsSync(UNMATCHED_PATH)) {
      try { existing = JSON.parse(readFileSync(UNMATCHED_PATH, 'utf8')); } catch { existing = {}; }
    }
    for (const [k, v] of Object.entries(this.items)) existing[k] = v;
    writeFileSync(UNMATCHED_PATH, JSON.stringify(existing, null, 2));
  }
}

export interface MetricRow {
  regionId: number;
  source: string;
  metric: string;
  period: string; // YYYY-MM-DD
  value: number;
}

/** Batched multi-VALUES upsert into metric_series. Dedupes within the batch (last wins). */
export async function upsertMetricRows(rows: MetricRow[], ingestRunId: number, batchSize = 1000): Promise<number> {
  const deduped = new Map<string, MetricRow>();
  for (const r of rows) deduped.set(`${r.regionId}|${r.source}|${r.metric}|${r.period}`, r);
  const all = [...deduped.values()];
  let n = 0;
  for (let i = 0; i < all.length; i += batchSize) {
    const batch = all.slice(i, i + batchSize);
    const params: unknown[] = [];
    const values = batch.map((r, j) => {
      params.push(r.regionId, r.source, r.metric, r.period, r.value, ingestRunId);
      const b = j * 6;
      return `($${b + 1},$${b + 2},$${b + 3},$${b + 4},$${b + 5},$${b + 6})`;
    });
    await query(
      `INSERT INTO metric_series (region_id, source, metric, period, value, ingest_run_id)
       VALUES ${values.join(',')}
       ON CONFLICT (region_id, source, metric, period)
       DO UPDATE SET value = EXCLUDED.value, ingest_run_id = EXCLUDED.ingest_run_id`,
      params,
    );
    n += batch.length;
  }
  return n;
}

/** Batched upsert into source_region_map. */
export async function upsertSourceRegionMap(
  entries: { source: string; sourceRegionId: string; regionId: number; matchMethod: string }[],
  batchSize = 1000,
): Promise<void> {
  const deduped = new Map<string, (typeof entries)[number]>();
  for (const e of entries) deduped.set(`${e.source}|${e.sourceRegionId}`, e);
  const all = [...deduped.values()];
  for (let i = 0; i < all.length; i += batchSize) {
    const batch = all.slice(i, i + batchSize);
    const params: unknown[] = [];
    const values = batch.map((e, j) => {
      params.push(e.source, e.sourceRegionId, e.regionId, e.matchMethod);
      const b = j * 4;
      return `($${b + 1},$${b + 2},$${b + 3},$${b + 4})`;
    });
    await query(
      `INSERT INTO source_region_map (source, source_region_id, region_id, match_method)
       VALUES ${values.join(',')}
       ON CONFLICT (source, source_region_id)
       DO UPDATE SET region_id = EXCLUDED.region_id, match_method = EXCLUDED.match_method`,
      params,
    );
  }
}

/** First-of-month normalization: '2026-06-30' -> '2026-06-01'. */
export function firstOfMonth(dateStr: string): string {
  return dateStr.slice(0, 7) + '-01';
}

/** First-of-month cutoff N months back from now. */
export function monthsAgoCutoff(months: number): string {
  const d = new Date();
  d.setUTCDate(1);
  d.setUTCMonth(d.getUTCMonth() - (months - 1));
  return d.toISOString().slice(0, 7) + '-01';
}