← back to La Socrata Ingester

src/sources.js

313 lines

// Source registry. Every endpoint here was live-probed by the la-research-agent
// on 2026-08-10. `map(row)` returns the typed columns; the full source record is
// stored separately as `raw` jsonb by ingest.js, so upstream schema drift never
// loses data.

// --- small typed-coercion helpers ---
const str = (v) => (v === undefined || v === null || v === '' ? null : String(v));
const num = (v) => {
  if (v === undefined || v === null || v === '') return null;
  const n = Number(v);
  return Number.isFinite(n) ? n : null;
};
const iso = (v) => str(v); // Socrata/ArcGIS return ISO strings; let pg cast to date/timestamptz

// LA City code-enforcement addresses are split across 6 columns with no combined field.
const codeAddress = (r) =>
  [r.stno, r.predir, r.stname, r.suffix, r.postdir]
    .map((x) => (x == null ? '' : String(x).trim()))
    .filter(Boolean)
    .join(' ') || null;

// Business dataset carries lat/lon inside a GeoJSON `location_1` point ([lon,lat]).
const bizLatLon = (r) => {
  const c = r.location_1 && r.location_1.coordinates;
  return Array.isArray(c) && c.length === 2 ? { lon: num(c[0]), lat: num(c[1]) } : { lon: null, lat: null };
};

export const SOURCES = {
  // ========================= BUILDING PERMITS (Socrata) =====================
  building_permits: {
    platform: 'socrata',
    domain: 'data.lacity.org',
    datasetId: 'pi9x-tg5x', // Issued 2020–present, ~405k, daily
    table: 'la_building_permits_raw',
    conflict: ['dataset_id', 'permit_nbr'],
    cursorField: 'issue_date',
    map: (r) => ({
      permit_nbr: str(r.permit_nbr),
      primary_address: str(r.primary_address),
      apn: str(r.apn),
      zip_code: str(r.zip_code),
      council_district: str(r.cd),
      permit_group: str(r.permit_group),
      permit_type: str(r.permit_type),
      permit_sub_type: str(r.permit_sub_type),
      use_desc: str(r.use_desc),
      issue_date: iso(r.issue_date),
      status_desc: str(r.status_desc),
      valuation: num(r.valuation),
      lat: num(r.lat),
      lon: num(r.lon),
      work_desc: str(r.work_desc),
    }),
  },
  // Static historical + related-trade permits share the same table/mapper.
  building_permits_2010_2019: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: 'dyxf-7hc4',
    table: 'la_building_permits_raw', conflict: ['dataset_id', 'permit_nbr'],
    static: true, mapFrom: 'building_permits',
  },
  building_permits_pre2010: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: 'e67z-kt2n',
    table: 'la_building_permits_raw', conflict: ['dataset_id', 'permit_nbr'],
    static: true, mapFrom: 'building_permits',
  },
  building_permits_electrical: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: 'ysqd-apz7',
    table: 'la_building_permits_raw', conflict: ['dataset_id', 'permit_nbr'],
    cursorField: 'issue_date', mapFrom: 'building_permits',
  },
  building_permits_mech_plumb: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: '67is-svtd',
    table: 'la_building_permits_raw', conflict: ['dataset_id', 'permit_nbr'],
    cursorField: 'issue_date', mapFrom: 'building_permits',
  },

  // ========================= ASSESSOR PARCELS (ArcGIS) =====================
  assessor_parcels: {
    platform: 'arcgis',
    endpoint:
      'https://services.arcgis.com/RmCCgQtiZLDCtblq/arcgis/rest/services/Parcel_Data_2021_Table/FeatureServer/0',
    table: 'la_assessor_parcels_raw',
    conflict: ['ain', 'roll_year'],
    orderBy: 'OBJECTID',
    oidField: 'OBJECTID',
    paginate: 'oid', // OBJECTID-cursor — robust vs deep-offset 400s
    cursorField: 'RollYear',
    // Default (non-full) refresh: newest roll year only (~2.4M vs 12.1M total).
    defaultWhere: "RollYear = '2025'",
    // Audited exception (TK-10955): this FeatureServer has twice returned this exact
    // generic 400 on page 1 — before any row is fetched — always ~55s after the request
    // (2026-09-10, 2026-09-12). Both times a manual replay of the byte-identical query
    // succeeded seconds to minutes later, so this is upstream load, not a bad query.
    // Table is upsert-only (never deleted) and the OID-cursor rescans the whole current
    // roll year every run, so a skipped day just leaves yesterday's snapshot in place one
    // day longer — no row loss, no corruption. Scoped to the EXACT audited message so a
    // genuinely different assessor_parcels failure (a real bad query) still fails the
    // aggregate run loud, same discipline as gis_zoning's exemption above.
    allowFailure: {
      since: '2026-09-10',
      scope: 'aggregate-all-only',
      reason: 'FeatureServer intermittently 400s on page 1 under load (~55s in); resolves on manual retry; upsert-only table so a skipped day is not data loss',
      match: ['Cannot perform query. Invalid query parameters.'],
    },
    map: (r) => ({
      ain: str(r.AIN),
      roll_year: str(r.RollYear),
      assessor_id: str(r.AssessorID),
      property_location: str(r.PropertyLocation),
      situs_house_no: str(r.SitusHouseNo),
      situs_street: str(r.SitusStreet),
      situs_city: str(r.SitusCity),
      situs_zip5: str(r.SitusZIP5),
      use_type: str(r.UseType),
      use_code: str(r.UseCode),
      year_built: str(r.YearBuilt),
      sqft_main: num(r.SQFTmain),
      bedrooms: str(r.Bedrooms),
      bathrooms: str(r.Bathrooms),
      units: str(r.Units),
      land_value: num(r.Roll_LandValue),
      imp_value: num(r.Roll_ImpValue),
      total_value: num(r.Roll_TotalValue),
      recording_date: num(r.RecordingDate),
      center_lat: num(r.CENTER_LAT),
      center_lon: num(r.CENTER_LON),
    }),
  },

  // ========================= CODE ENFORCEMENT (Socrata) ====================
  code_enforcement_open: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: 'u82d-eh7z',
    table: 'la_code_enforcement_raw', conflict: ['apno'], cursorField: 'adddttm',
    map: (r) => ({
      apno: str(r.apno), apname: str(r.apname), address: codeAddress(r),
      zip: str(r.zip), add_dttm: iso(r.adddttm), res_dttm: iso(r.resdttm),
      prclid: str(r.prclid), ap_type: str(r.aptype), apc: str(r.apc), stat: str(r.stat) || 'O',
    }),
  },
  code_enforcement_closed: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: 'rken-a55j',
    table: 'la_code_enforcement_raw', conflict: ['apno'], cursorField: 'resdttm',
    mapFrom: 'code_enforcement_open',
    mapOverride: (r) => ({ stat: 'C' }),
  },

  // ========================= BUSINESS REGISTRATIONS (Socrata) ==============
  business_registrations: {
    platform: 'socrata', domain: 'data.lacity.org', datasetId: '6rrh-rzua',
    table: 'la_business_registrations_raw', conflict: ['location_account'],
    cursorField: 'location_start_date',
    map: (r) => {
      const { lat, lon } = bizLatLon(r);
      return {
        location_account: str(r.location_account),
        business_name: str(r.business_name),
        dba_name: str(r.dba_name),
        street_address: str(r.street_address),
        city: str(r.city),
        zip_code: str(r.zip_code),
        naics: str(r.naics),
        primary_naics_description: str(r.primary_naics_description),
        council_district: str(r.council_district),
        location_start_date: iso(r.location_start_date),
        location_end_date: iso(r.location_end_date),
        lat, lon,
      };
    },
  },

  // ========================= FILM PERMITS (WeHo Socrata) ===================
  // LA City proper has no live public dataset (both deprecated; FilmLA = CPRA only).
  // WeHo is the live free feed. Schema fields probed defensively.
  film_permits_weho: {
    platform: 'socrata', domain: 'data.weho.org', datasetId: 'xd5y-dzvf',
    table: 'la_film_permits_raw', conflict: ['permit_id'],
    // WeHo schema (probed): permit_no, production, production_company, category, address, date.
    map: (r) => ({
      permit_id: str(r.permit_no || r.permit_number || r.id),
      title: str(r.production || r.production_company),
      permit_type: str(r.category),
      address: str(r.address || r.location),
      start_date: iso(r.date),
      end_date: iso(r.date),
    }),
  },
};

// ========================= GIS LAYERS (LA City NavigateLA ArcGIS) ==========
// NavigateLA layer ids are POSITIONAL: they are an index into a list LA edits, so
// inserting ONE layer renumbers every layer below it. This has now happened twice.
//
//  2026-09-05  ids shifted +1 (71->72, 75->76, 119->120, 124->125, 254->255,
//              358->359, 414->415, 418->419, 439->440). Sources pinned bare ids, so
//              all 9 silently re-pointed: fault_zones + flood 400'd loudly, the other
//              7 ingested a DIFFERENT dataset under our labels for 6 days with no
//              error (council_district 15->3, neighborhood_council 99->26, hpoz
//              35->7843, community_plan_area 36->7, liquefaction 474->14181,
//              fire_vhfhsz 14->247). Fixed by repointing +1 and adding expectName.
//  2026-09-11  ids shifted +1 AGAIN — LA inserted the Feature Layer 'Special Event
//              Permits' at position 52 (layer count 452 -> 453), shifting everything
//              at id >= 52. This time the expectName guard caught it: all 8 pinned
//              NavigateLA sources failed loud at preflight and ingested NOTHING, so
//              there was no corruption to clean up. Repointed +1 again (see ids below).
//
// POLICY since 2026-09-11 (Steve's call): the pinned ids below are a HINT, not the
// contract. On a name mismatch the adapter FINDS the layer by exact name, PROVES it on the
// authoritative per-layer endpoint (name + type + field fingerprint), logs an AUTO-REPOINT
// warning and continues — and still fails loud on a rename, an ambiguous match, or a match
// it cannot prove. Fail-loud-only was correct on 09-10 but became a roughly weekly outage
// once the drift recurred. Kill switch: LA_NO_AUTO_REPOINT=1. See src/adapters/arcgis.js.
//
// TRAP, learned 2026-09-11 — do NOT resolve an id by name from the SERVICE CATALOG
// (/MapServer?f=json) alone. That response is cacheable and was observed serving a
// STALE body (452 layers, old index) while the per-layer endpoints already served the
// new index; resolving 'Council Districts' off it would have returned 419, which by
// then was 'Contract Administration Inspection Districts'. The authoritative plane is
// the per-layer endpoint /MapServer/<id>?f=json, because that is what /query follows —
// and that is the plane assertLayerIdentity reads.
const NAV = 'https://maps.lacity.org/arcgis/rest/services/Mapping/NavigateLA/MapServer';
// generic feature -> la_gis_features row
const gisFeat = (layer, nameFields = []) => (r) => ({
  layer, oid: r.OBJECTID,
  name: nameFields.map((f) => r[f]).find((v) => v != null && v !== '') ?? str(r.TOOLTIP),
  geom: r.__geometry,
});
// NavigateLA layer ids are POSITIONAL and DO drift (see the 2026-09-05 +1 re-index
// below), so every gisSrc pins the expected upstream layer NAME. The arcgis adapter
// verifies it before paging and fails loud rather than ingesting the wrong dataset.
const gisSrc = (layer, layerId, nameFields, expectName, extra = {}) => ({
  platform: 'arcgis', endpoint: `${NAV}/${layerId}`, geometry: true,
  table: 'la_gis_features', conflict: ['layer', 'oid'],
  expectName,
  // Surfaced (not just closed over by gisFeat) so the adapter's schema fingerprint can
  // check them: a same-named REPLACEMENT dataset that lacks every one of our nameFields
  // is not ours, and must never be auto-repointed onto.
  nameFields,
  map: gisFeat(layer, nameFields), ...extra,
});

Object.assign(SOURCES, {
  // Parcel polygons — the NavigateLA 397 layer (2.4M) 502s the gateway on any
  // feature query; the hosted Assessor_Parcels_Land_2026 FeatureServer (reliable
  // geometry) is used instead. APN -> ain for the permit join. Repointed
  // 2025->2026 (2026-08-25): the _2025 item was unshared (ArcGIS 400 "Item does
  // not exist"), failing the whole nightly run via pipefail; _2026_gdb is live
  // with the same OBJECTID/APN/APN9 fields the mapper needs.
  gis_parcels: {
    platform: 'arcgis',
    endpoint: 'https://services5.arcgis.com/Y8jwjGUWbRjuqpG5/arcgis/rest/services/Assessor_Parcels_Land_2026_gdb/FeatureServer/0',
    geometry: true, paginate: 'oid', oidField: 'OBJECTID',
    table: 'la_parcel_geom', conflict: ['oid'],
    map: (r) => ({ oid: r.OBJECTID, ain: str(r.APN || r.APN9 || r.AIN), geom: r.__geometry }),
  },
  // Zoning + land use (~50–59k). NavigateLA gateway 502s on sorted geometry -> use
  // offset paging with NO orderByFields (noOrder) + small pages.
  gis_zoning:  gisSrc('zoning',  73, ['ZONE_CMPLT', 'ZONE_CLASS'], 'Generalized Zoning', {
    pageSize: 2000,
    noOrder: true,
    // Audited exception: keep attempting this source so an upstream recovery is
    // detected automatically, but do not fail the aggregate `all` run for its
    // documented deep-offset gateway failure. Direct gis_zoning runs stay strict.
    allowFailure: {
      since: '2026-08-11',
      scope: 'aggregate-all-only',
      reason: 'NavigateLA gateway consistently HTTP 502s on geometry pagination near offset 42000',
      // Scoped to the AUDITED failure only. Without this, the exemption would also swallow
      // an 'ArcGIS layer identity drift' throw — silencing, for the one source that carries
      // an exemption, precisely the alarm the identity guard exists to raise (TK-10955).
      // Both error shapes: the HTTP-level form from fetchJson ("HTTP 502 after 5 tries")
      // and the ArcGIS body-level form ("ArcGIS error: {\"code\":502,...}").
      match: ['HTTP 502', 'HTTP 503', 'HTTP 504', '"code":502', '"code":503', '"code":504'],
    },
  }),
  // Land use: hosted LADCP FeatureServer (reliable) instead of the flaky NavigateLA
  // gateway. Fields are lowercase here (objectid/gplu/gplu_desc).
  gis_landuse: {
    platform: 'arcgis',
    endpoint: 'https://services1.arcgis.com/tzwalEyxl2rpamKs/arcgis/rest/services/GeneralPlanLandUse_6a318/FeatureServer/0',
    geometry: true, paginate: 'oid', oidField: 'objectid',
    table: 'la_gis_features', conflict: ['layer', 'oid'],
    map: (r) => ({ layer: 'landuse', oid: r.objectid, name: str(r.gplu_desc || r.gplu), geom: r.__geometry }),
  },
  // Boundaries (small).
  gis_council_districts:     gisSrc('council_district',    420, ['District_Name', 'NAME', 'District'], 'Council Districts',                          { pageSize: 20000 }),
  gis_neighborhood_councils: gisSrc('neighborhood_council', 441, ['NAME'],                                'Neighborhood Councils (Certified)',          { pageSize: 20000 }),
  gis_hpoz:                  gisSrc('hpoz',                 77,  ['NAME'],                                'Historic Preservation Overlay Zone District',{ pageSize: 20000 }),
  gis_community_plan_areas:  gisSrc('community_plan_area',  416, ['NAME'],                                'Community Plan Areas',                       { pageSize: 20000 }),
  // Hazards (small).
  gis_fault_zones:  gisSrc('fault_zone',   121, ['HAZ_TYPE'], 'Alquist Priolo Earthquake Fault Zones', { pageSize: 20000 }),
  gis_liquefaction: gisSrc('liquefaction', 126, [],           'Liquefaction',                          { pageSize: 20000 }),
  gis_flood:        gisSrc('flood',        256, ['FLD_ZONE'], 'Special Flood Hazard Areas (S_FLD_HAZ_AR) (OPCS) (Eff. 4/21/2021)', { pageSize: 20000, oidField: 'FLD_AR_ID' }),
  gis_fire_vhfhsz:  gisSrc('fire_vhfhsz',  360, [],           'Very High Fire Hazard Severity Zones',  { pageSize: 20000 }),
  // NOTE: methane (layer 354) has NO pagination support — needs spatial tiling; deferred.
});

// Resolve a source's effective mapper (supports mapFrom inheritance + mapOverride).
export function resolveMap(name) {
  const src = SOURCES[name];
  const base = src.mapFrom ? SOURCES[src.mapFrom].map : src.map;
  if (!base) throw new Error(`source ${name} has no map()`);
  if (src.mapOverride) return (r) => ({ ...base(r), ...src.mapOverride(r) });
  return base;
}

// Grouped for --list output.
export const GROUPS = {
  'building permits': ['building_permits', 'building_permits_2010_2019', 'building_permits_pre2010', 'building_permits_electrical', 'building_permits_mech_plumb'],
  'assessor parcels': ['assessor_parcels'],
  'code enforcement': ['code_enforcement_open', 'code_enforcement_closed'],
  'business / film': ['business_registrations', 'film_permits_weho'],
};