`src/adapters/arcgis.js`

357 lines
import { fetchJson, sleep } from './http.js';

// ArcGIS tunnels errors as HTTP 200 with an {error:{code,messageCode}} BODY, so the
// HTTP-level retry in fetchJson never sees them. Throttle/5xx codes are transient and
// retried; anything else (e.g. 400 bad query) is permanent and thrown immediately so we
// fail fast instead of backing off 5x. NOTE: bare 499 is NOT transient — canonically it's
// "Token Required" (a permanent auth failure on these public sources); only 499 carrying
// messageCode CONT_0001 ("item momentarily inaccessible") is the transient we retry.
const TRANSIENT_ARCGIS = new Set([429, 500, 502, 503, 504]);
const isTransientArcgisError = (e) =>
  !!e && (TRANSIENT_ARCGIS.has(e.code) || e.messageCode === 'CONT_0001');

// fetchJson + ArcGIS-payload-aware retry with capped exponential backoff.
// httpTries is the HTTP-level retry budget inside fetchJson (default 5). The adaptive
// page-size backoff (below) passes a SMALL httpTries while it still has room to shrink a
// page, so an oversized-geometry band does not pay the full 5x/~30s backoff storm on every
// halving step — and it restores the full budget at the floor, so a genuine gateway outage
// (a page that 502s even at the floor) still surfaces loudly rather than being cut short.
async function fetchArcgis(url, { tries = 5, httpTries = 5 } = {}) {
  let attempt = 0;
  for (;;) {
    attempt++;
    const data = await fetchJson(url, { tries: httpTries }); // HTTP-level (429/5xx/network) retry already inside
    if (!data.error) return data;
    if (!isTransientArcgisError(data.error) || attempt >= tries) {
      throw new Error(`ArcGIS error: ${JSON.stringify(data.error).slice(0, 200)}`);
    }
    const wait = Math.min(30000, 500 * 2 ** attempt);
    console.error(
      `  ⟳ ArcGIS transient ${data.error.code ?? ''} ${data.error.messageCode ?? ''} — retry ${attempt}/${tries} in ${wait}ms`
    );
    await sleep(wait);
  }
}

// ── Adaptive page-size backoff (TK-11500) ───────────────────────────────────
// The NavigateLA gateway 502s on a GEOMETRY page whose response takes >30s to serialize.
// MEASURED on the Generalized Zoning layer: the OBJECTID 42000..56000 band carries ~22KB of
// geometry per feature (vs ~0.2KB elsewhere), so a 2000-row page there is ~44MB and the
// gateway times out. Because it is a fixed OBJECTID band and not a fixed depth, NEITHER
// offset- nor oid-cursor paging avoids it (both were live-tested and both 502 at the same
// band) — the real constraint is response SIZE. So on a gateway size-failure, HALVE the page
// and retry the SAME cursor position until the response fits (down to a floor), then grow the
// size back so the rest of the layer still pages fast (AIMD: multiplicative-decrease on
// failure, x2 grow-back on success). General — any oversized-geometry source benefits, not
// zoning-specific. A page that still fails at the floor is a real outage, not a size problem,
// so it is rethrown (fail-loud; the aggregate run's allowFailure exemption still catches a
// genuinely stuck source, so this degrades exactly as before rather than looping forever).
// Read at CALL time, not module-load time — same lesson the repoint kill switches learned
// (a module-level const bakes the value in at import and ignores a later env override).
const adaptiveMinPage = () => Math.max(1, Number(process.env.LA_ADAPTIVE_MIN_PAGE ?? 50));
// A gateway 502/503/504 in either shape fetchArcgis can throw it: the HTTP-level form from
// fetchJson ("HTTP 502 after N tries") and the ArcGIS body-level form ('ArcGIS error:
// {"code":502,...}'). This is the class we shrink for.
const isGatewaySizeFailure = (err) =>
  /HTTP 50[234]\b|"code":\s*50[234]\b/.test(String(err?.message || ''));
// After a shrunk page succeeds, grow back toward pageSize (x2) so throughput recovers once
// past a heavy band; a page that ran at full size stays there (no-op).
const nextPageSize = (usedCount, pageSize) =>
  usedCount < pageSize ? Math.min(pageSize, usedCount * 2) : pageSize;

// Fetch one page at `count`, shrinking on a gateway size-failure and retrying the SAME
// cursor. buildUrl(count) produces the query URL for a given resultRecordCount. Only shrinks
// for GEOMETRY pages — a non-geometry page is already tiny, so a transient there is a real
// upstream outage, not a size problem, and must not be masked. Returns { data, usedCount }
// where usedCount is the size that actually succeeded (feeds the end-of-layer test + grow-back).
async function fetchPageAdaptive(buildUrl, count, geom) {
  const floor = adaptiveMinPage();
  for (;;) {
    const atFloor = count <= floor;
    try {
      // Fail fast (httpTries:2) while there is still room to shrink; spend the full retry
      // budget only at the floor, so a real outage (fails even at the floor) still throws.
      const data = await fetchArcgis(buildUrl(count), { httpTries: atFloor ? 5 : 2 });
      return { data, usedCount: count };
    } catch (err) {
      if (geom && !atFloor && isGatewaySizeFailure(err)) {
        const next = Math.max(floor, Math.floor(count / 2));
        console.error(
          `  ⤵ ArcGIS geometry page too large — halving resultRecordCount ${count} → ${next} at the same cursor ` +
          `(${String(err.message).slice(0, 70)})`
        );
        count = next;
        continue;
      }
      throw err;
    }
  }
}

// ── Layer-identity preflight ────────────────────────────────────────────────
// An ArcGIS layer id is a POSITIONAL INDEX, not a stable identifier. On 2026-09-05
// LA re-indexed the NavigateLA MapServer and EVERY layer id shifted +1, silently
// re-pointing all 9 of our layers at a different dataset. Only 2 failed loudly (a
// Group Layer and a layer missing our oidField); the other 7 landed on queryable
// layers and ingested a stranger's rows under our labels for 6 days with no error.
// So: when a source declares expectName, verify the LIVE layer name before paging
// and fail LOUD on drift. Ingesting the wrong layer is strictly worse than not
// ingesting at all — one metadata GET per source per run ($0, public API).
const normName = (s) => String(s ?? '').trim().toLowerCase().replace(/\s+/g, ' ');

// ── Verified auto-repoint (TK-10955, Steve 2026-09-11) ──────────────────────
// Positional ids drifted twice in six days (2026-09-05, 2026-09-11), so fail-loud-only
// meant a weekly outage. Policy is now: on a name mismatch, FIND the layer by exact name,
// PROVE it on the authoritative plane, then ingest — and still fail loud on anything we
// cannot prove (a rename, no match, or an ambiguous match).
//
// Two discovery paths, in order:
//   1. the service catalog (…/MapServer?f=json) — one cached GET, covers any drift distance;
//   2. a bounded ±N window scan of per-layer endpoints — the fallback for when the catalog
//      is STALE, which was MEASURED on 2026-09-11: it served a 452-layer body (old index)
//      while per-layer endpoints already served 453. Resolving off the catalog ALONE would
//      have returned the wrong layer, so a catalog hit is never trusted on its own.
// Either way the candidate is re-proved on /MapServer/<id>?f=json — the plane /query
// follows — before a single row is read.
const LAYER_ENDPOINT_RE = /^(.*)\/(\d+)$/;
// Read at CALL time, not module-load time. As module-level consts these were baked in at
// import and the kill switch silently did nothing — caught by its own negative test.
const repointWindow = () => Math.max(0, Number(process.env.LA_REPOINT_WINDOW ?? 10));
const autoRepointDisabled = () => process.env.LA_NO_AUTO_REPOINT === '1';

// One catalog GET per process — but a FAILED fetch is deliberately NOT cached. Caching the
// failure would let a single transient blip on the first drifted source silently downgrade
// every later source in the same run to the window-scan path (still safe, since the window
// path also demands full proof, but needlessly narrower). Re-try it instead.
const catalogCache = new Map(); // baseUrl -> Promise<catalog>, successful fetches only
function fetchCatalog(baseUrl, fetcher) {
  if (!catalogCache.has(baseUrl)) {
    const p = Promise.resolve(fetcher(`${baseUrl}?f=json`));
    catalogCache.set(baseUrl, p);
    p.catch(() => catalogCache.delete(baseUrl)); // forget failures so the next source retries
  }
  return catalogCache.get(baseUrl).catch(() => null);
}
export function _resetCatalogCache() { catalogCache.clear(); } // tests only

const nameMatches = (meta, src) => normName(meta?.name) === normName(src.expectName);

// Field/type fingerprint. Closes the "a replacement dataset reuses the exact name" risk:
// a same-named layer that lacks our oidField (or every one of our nameFields) is NOT ours.
// Returns 'ok' | 'unverifiable' | 'mismatch'. 'unverifiable' means the metadata carried no
// field list, so we could not measure — deliberately NOT folded into 'ok'.
function fingerprint(meta, src) {
  if (normName(meta?.type) !== 'feature layer') return 'mismatch';
  const list = Array.isArray(meta?.fields) ? meta.fields : null;
  if (!list || list.length === 0) return 'unverifiable';
  const fields = new Set(list.map((f) => f?.name));
  if (!fields.has(src.oidField || 'OBJECTID')) return 'mismatch';
  const nameFields = Array.isArray(src.nameFields) ? src.nameFields : [];
  if (nameFields.length && !nameFields.some((f) => fields.has(f))) return 'mismatch';
  return 'ok';
}

// Probe failures are counted, not just swallowed. A probe that ERRORED is not evidence the
// layer is absent, and the loud failure message must not imply it was: fetcher already
// retries 429/5xx five times, so a surviving error is real, and reporting how many probes
// errored is what tells a human "upstream was flaky" apart from "it genuinely moved away".
// Either way we fail loud rather than ingest, so this is a message-honesty fix, not a gate.
async function probeLayer(baseUrl, id, fetcher, stats = null) {
  if (!Number.isInteger(id) || id < 0) return null;
  try {
    return await fetcher(`${baseUrl}/${id}?f=json`);
  } catch {
    if (stats) stats.errors++;
    return null;
  }
}

// A candidate is accepted ONLY on full proof: exact name AND a verified fingerprint.
// 'unverifiable' is never enough to MOVE onto a new layer (asymmetry is intentional —
// see resolveLayerIdentity for why the pinned id is allowed to stay on 'unverifiable').
const candidateProved = (meta, src) => !!meta && nameMatches(meta, src) && fingerprint(meta, src) === 'ok';

async function findLayerByName(baseUrl, pinnedId, src, fetcher) {
  const tried = [];
  const stats = { errors: 0 };
  const cat = await fetchCatalog(baseUrl, fetcher);
  const catHits = (cat?.layers ?? []).filter(
    (l) => normName(l?.name) === normName(src.expectName) && normName(l?.type) === 'feature layer'
  );
  if (catHits.length === 1) {
    const id = catHits[0].id;
    tried.push(`catalog->${id}`);
    if (candidateProved(await probeLayer(baseUrl, id, fetcher, stats), src)) return { id, via: 'catalog', tried };
    tried.push(`catalog->${id} FAILED re-proof (stale catalog?)`);
  } else if (catHits.length > 1) {
    tried.push(`catalog ambiguous (${catHits.length} layers named "${src.expectName}")`);
  } else {
    tried.push('catalog: no exact-name Feature Layer');
  }

  // Fallback: scan a bounded window on the authoritative per-layer plane. Scans the WHOLE
  // window (does not stop at the first hit) so an ambiguous match is detected, not masked.
  const hits = [];
  const win = repointWindow();
  for (let d = 1; d <= win; d++) {
    for (const id of [pinnedId + d, pinnedId - d]) {
      if (candidateProved(await probeLayer(baseUrl, id, fetcher, stats), src)) hits.push(id);
    }
  }
  tried.push(
    `window +/-${win}: ${hits.length} proved match(es)${hits.length ? ` [${hits.join(', ')}]` : ''}` +
    (stats.errors ? ` (${stats.errors} probe(s) ERRORED — absence here is not proof of absence)` : '')
  );
  if (hits.length === 1) return { id: hits[0], via: `window+/-${win}`, tried };
  return { id: null, ambiguous: hits.length > 1, tried };
}

// Resolve the endpoint a source should actually page. Returns { endpoint, repointed, from,
// to, via }. Throws the loud drift error when the layer cannot be PROVED anywhere.
export async function resolveLayerIdentity(src, fetcher = fetchArcgis) {
  if (!src.expectName) return { endpoint: src.endpoint, repointed: false }; // opt-in

  const pinned = await fetcher(`${src.endpoint}?f=json`);
  if (nameMatches(pinned, src)) {
    const fp = fingerprint(pinned, src);
    if (fp === 'mismatch') {
      // Right name, wrong shape: the dataset behind our label changed. Never auto-move on
      // this — a human must look, because "same name, different schema" is how a
      // replacement dataset would impersonate ours.
      throw new Error(
        `ArcGIS layer schema drift: ${src.endpoint} still reports "${pinned?.name}" but no longer ` +
        `matches this source's shape (type=${pinned?.type}, oidField=${src.oidField || 'OBJECTID'}, ` +
        `nameFields=${JSON.stringify(src.nameFields ?? [])}). Upstream replaced the dataset — ` +
        `do NOT ingest; review the source definition.`
      );
    }
    // fp === 'unverifiable' (metadata carried no field list) keeps the PINNED id, matching
    // pre-existing behaviour. We decline to break a working pin on missing metadata, but we
    // still refuse to MOVE to a new layer on the same weak evidence.
    return { endpoint: src.endpoint, repointed: false };
  }

  const m = LAYER_ENDPOINT_RE.exec(src.endpoint);
  const driftMsg =
    `ArcGIS layer identity drift: ${src.endpoint} is now "${pinned?.name ?? '(unnamed)'}" ` +
    `but this source expects "${src.expectName}".`;
  if (autoRepointDisabled() || !m) {
    throw new Error(`${driftMsg} Upstream re-indexed its layer ids — repoint the layer id (verify by NAME), do NOT ingest.`);
  }

  const [, baseUrl, pinnedIdStr] = m;
  const found = await findLayerByName(baseUrl, Number(pinnedIdStr), src, fetcher);
  if (found.id == null) {
    throw new Error(
      `${driftMsg} Could not PROVE where it moved, so nothing was ingested. Tried: ${found.tried.join('; ')}. ` +
      (found.ambiguous
        ? 'Multiple layers matched — ambiguous, a human must choose.'
        : 'A rename, a removal, or a shift beyond the search window — repoint by hand (verify by NAME).')
    );
  }

  const endpoint = `${baseUrl}/${found.id}`;
  console.error(
    `  ⚠ AUTO-REPOINT ${pinnedIdStr} -> ${found.id} for "${src.expectName}" (via ${found.via}; ` +
    `name + schema re-proved on the per-layer endpoint). Upstream shifted its positional ids; ` +
    `update the pinned id in src/sources.js.`
  );
  return { endpoint, repointed: true, from: Number(pinnedIdStr), to: found.id, via: found.via };
}

// Back-compat wrapper: throws on drift, never repoints. Kept because it is the narrow
// "is this layer what we think it is" assertion the tests exercise directly.
export async function assertLayerIdentity(src, fetcher = fetchArcgis) {
  if (!src.expectName) return;
  const meta = await fetcher(`${src.endpoint}?f=json`);
  if (normName(meta?.name) !== normName(src.expectName)) {
    throw new Error(
      `ArcGIS layer identity drift: ${src.endpoint} is now "${meta?.name ?? '(unnamed)'}" ` +
      `but this source expects "${src.expectName}". Upstream re-indexed its layer ids — ` +
      `repoint the layer id (verify by NAME), do NOT ingest.`
    );
  }
}

// ArcGIS FeatureServer/MapServer paginator. Yields { rows, url } per page, where
// each row is the feature's attributes (with __geometry attached when src.geometry).
//
// Two pagination modes:
//   • 'oid'  (src.paginate==='oid') — OBJECTID-cursor: where <oid> > lastMax, ordered
//     by oid, no resultOffset. Robust for large/deep extracts (avoids the deep-offset
//     400 "Invalid query parameters" that offset paging hits past ~800 pages).
//   • 'offset' (default) — resultOffset/resultRecordCount. Fine for small layers.
//
// src fields: endpoint (…/{n}), orderBy?, oidField? (default OBJECTID), cursorField?,
//   defaultWhere?, paginate? ('oid'|'offset'), geometry? (bool)
// opts: since, pageSize, maxRows, fullScan, whereOverride
export async function* arcgisPages(src, opts = {}) {
  const { since = null, pageSize = 2000, maxRows = Infinity, fullScan = false, whereOverride = null } = opts;
  // Resolve BEFORE building the query url: a verified auto-repoint changes the endpoint.
  const resolved = await resolveLayerIdentity(src);
  const base = `${resolved.endpoint}/query`;
  const oidField = src.oidField || 'OBJECTID';
  const geom = !!src.geometry;

  // base filters (everything except the oid cursor)
  const filters = [];
  if (whereOverride) filters.push(whereOverride);
  else if (fullScan) { /* all rows */ }
  else if (since && src.cursorField) filters.push(`${src.cursorField} >= '${since}'`);
  else if (src.defaultWhere) filters.push(src.defaultWhere);

  const common = {
    outFields: '*',
    returnGeometry: geom ? 'true' : 'false',
    f: 'json',
  };
  if (geom) common.outSR = '4326'; // WGS84 lat/lon

  const mapRows = (feats) =>
    feats.map((ft) => (geom ? { ...ft.attributes, __geometry: ft.geometry } : ft.attributes));

  if (src.paginate === 'oid') {
    let lastOid = -1, fetched = 0, curCount = pageSize;
    for (;;) {
      if (fetched >= maxRows) break;
      const where = [...filters, `${oidField} > ${lastOid}`].join(' AND ');
      const remaining = maxRows - fetched;
      const buildUrl = (n) =>
        `${base}?${new URLSearchParams({ ...common, where, orderByFields: oidField, resultRecordCount: String(n) })}`;
      // Adaptive page size: shrink on an oversized-geometry gateway 502, grow back on success.
      const { data, usedCount } = await fetchPageAdaptive(buildUrl, Math.min(curCount, remaining), geom);
      const feats = data.features || [];
      if (!feats.length) break;
      yield { rows: mapRows(feats), url: buildUrl(usedCount) };
      lastOid = Math.max(...feats.map((ft) => ft.attributes[oidField]));
      fetched += feats.length;
      curCount = nextPageSize(usedCount, pageSize);
      if (feats.length < usedCount) break;
    }
    return;
  }

  // offset mode
  let offset = 0, curCount = pageSize;
  const where = filters.length ? filters.join(' AND ') : '1=1';
  const buildUrl = (n) => {
    const params = { ...common, where, resultOffset: String(offset), resultRecordCount: String(n) };
    // src.noOrder: omit orderByFields — the NavigateLA gateway 502s on sorted
    // geometry queries; MapServer returns natural OBJECTID order anyway.
    if (!src.noOrder) params.orderByFields = src.orderBy || oidField;
    return `${base}?${new URLSearchParams(params)}`;
  };
  for (;;) {
    const remaining = maxRows - offset;
    if (remaining <= 0) break;
    // Adaptive page size: shrink on an oversized-geometry gateway 502, grow back on success.
    const { data, usedCount } = await fetchPageAdaptive(buildUrl, Math.min(curCount, remaining), geom);
    const feats = data.features || [];
    if (!feats.length) break;
    yield { rows: mapRows(feats), url: buildUrl(usedCount) };
    offset += feats.length;
    curCount = nextPageSize(usedCount, pageSize);
    if (!data.exceededTransferLimit && feats.length < usedCount) break;
  }
}