Merge branch 'main' into upstream/ontario-511-key

server/providers/cctv.js

@@ -143,7 +143,7 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
 
         if (url.pathname === '/sources') {
           const body = {
-            trimmedPacks: getCctvSources.trimmedPacks?.() ?? [],
+            trimmedPacks: getCctvSources.trimmedPacks?.(sources) ?? [],
             sources: sources.map((source) => ({
               id: source.id,
               pack: source.pack,

server/providers/cctv/catalog.js

@@ -204,16 +204,19 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
   /** @type {Promise<Array<object>>|null} In-flight refresh, shared by concurrent
    * callers so a post-TTL burst launches ONE refetch, not one per request. */
   let _cctvSourceInflight = null;
-  /** @type {Array<{pack: string, available: number, served: number}>} Packs
-   * that serve fewer cameras than they offer. */
-  let _cctvTrimmedPacks = [];
+  /** @type {WeakMap<Array<object>, Array<{pack: string, available: number, served: number}>>}
+   * Packs that serve fewer cameras than they offer, keyed by the source list
+   * they describe, so a request served the old list during a background
+   * refresh never pairs it with the new list's trim notes. */
+  const _cctvTrimmedPacks = new WeakMap();
 
   /**
    * Assemble and cache the merged CCTV source list.
    *
    * Merges every source pack (live open-data packs, local file, env
    * variable), deduplicates by ID, shares the catalog cap fairly across
-   * packs, and caches for CCTV_SOURCE_CACHE_MS.
+   * packs, and caches for CCTV_SOURCE_CACHE_MS. Past that, the cached list is
+   * returned while a background refresh replaces it; only a first load waits.
    *
    * @returns {Promise<Array<object>>} Deduplicated, capped source list.
    */
@@ -228,11 +231,31 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
     // Single-flight: a burst of requests arriving past the TTL shares ONE refresh
     // instead of each launching the full multi-provider refetch. The `.finally`
     // clears the ref so the next post-TTL cycle starts fresh.
-    if (_cctvSourceInflight) return _cctvSourceInflight;
-    _cctvSourceInflight = refreshCctvSources().finally(() => {
-      _cctvSourceInflight = null;
-    });
-    return _cctvSourceInflight;
+    if (!_cctvSourceInflight) {
+      _cctvSourceInflight = refreshCctvSources().finally(() => {
+        _cctvSourceInflight = null;
+      });
+      // A background refresh has no caller waiting on it: report its failure
+      // here, once per refresh rather than once per request it serves. Keep
+      // the old catalog for another TTL, as an empty refresh does, so a refresh
+      // that keeps failing is retried once per TTL, not back to back.
+      if (_cctvSourceCache.length) {
+        _cctvSourceInflight.catch((error) => {
+          _cctvSourceCacheAt = Date.now();
+          console.warn(
+            '[CCTV] background catalog refresh failed; serving',
+            _cctvSourceCache.length,
+            'cameras from the last catalog:',
+            error?.message || error,
+          );
+        });
+      }
+    }
+    // With a catalog in hand, answer with it while the refresh runs: the
+    // refresh lasts as long as the slowest pack (Tarktee's DATEX feeds take up
+    // to its 30 s timeout), and every /api/cctv request would otherwise wait
+    // for it. Only the first load, with nothing to serve yet, waits.
+    return _cctvSourceCache.length ? _cctvSourceCache : _cctvSourceInflight;
   }
 
   /**
@@ -322,7 +345,7 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
     }
     if (capped.length > 0 || _cctvSourceCache.length === 0) {
       _cctvSourceCache = capped;
-      _cctvTrimmedPacks = trimmedPacks;
+      _cctvTrimmedPacks.set(capped, trimmedPacks);
     } else {
       // Every source came back empty (all live packs timed out / upstream outage)
       // but a good catalog is already cached — serve it stale rather than blanking
@@ -336,7 +359,10 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
     return _cctvSourceCache;
   }
 
-  /** Packs serving fewer cameras than they offer, as of the last refresh. */
-  getCctvSources.trimmedPacks = () => _cctvTrimmedPacks;
+  /** Packs serving fewer cameras than they offer in `sources` (by default the
+   * current catalog). Pass the list a request was served to keep the two in
+   * step while a background refresh swaps the catalog. */
+  getCctvSources.trimmedPacks = (sources = _cctvSourceCache) =>
+    _cctvTrimmedPacks.get(sources) ?? [];
   return getCctvSources;
 }

server/providers/cctv/constants.js

@@ -209,6 +209,8 @@ export const TARKTEE_IMAGES_URL =
   'https://tarktee.transpordiamet.ee/api/v1/datex/roadCameraImages';
 export const TARKTEE_IMAGE_ORIGIN = 'https://tarktee.transpordiamet.ee/images/';
 export const DEFAULT_TARKTEE_MAX_SOURCES = 179;
+/** Hard ceiling per Tarktee DATEX XML document (~120–165 KB live). */
+export const TARKTEE_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
 export const TARKTEE_ANCHORS = [
   { lat: 59.437, lon: 24.753 }, // Tallinn
   { lat: 58.378, lon: 26.729 }, // Tartu
@@ -338,6 +340,49 @@ export const NORWAY_ANCHORS = [
   { lat: 69.6492, lon: 18.9553 }, // Tromsø
 ];
 
+/** Kartverket's keyless point-height API (CC BY 4.0): terrain height above
+ * sea level (NN2000), used as the Vegvesen cameras' ground prior because the
+ * camera feed's points are 2D. */
+export const KARTVERKET_HEIGHT_URL =
+  'https://ws.geonorge.no/hoydedata/v1/punkt';
+/** Points per request. The API takes up to 50, but above ~20 it answers the
+ * extra points from contour lines (`hoydekurver`, metres to ~20 m off) or not
+ * at all; at 20 nearly every camera resolves from the 1 m terrain model. */
+export const KARTVERKET_MAX_POINTS = 20;
+/** Requests in flight at once. The whole country is ~36 requests of
+ * 0.6-9.5 s each, so a cold lookup takes ~6-13 s. */
+export const KARTVERKET_CONCURRENCY = 8;
+/** Deadline for one complete lookup, shared by every request. A lookup cut
+ * short by it is resumed on the next refresh, without a backoff. */
+export const KARTVERKET_LOOKUP_TIMEOUT_MS = 30 * 1000;
+/** One request that takes longer than this counts as a failing service. */
+export const KARTVERKET_REQUEST_TIMEOUT_MS = 10 * 1000;
+/** The Vegvesen lane's budget, measured from the start of the loader (catalog
+ * fetch included), for waiting on heights. Under the 15 s per-provider fetch
+ * timeout so this pack is never the catalog's slowest lane. A lookup still
+ * running after it keeps filling the cache for the next refresh. */
+export const KARTVERKET_CATALOG_WAIT_MS = 12 * 1000;
+/** The lane only waits when more than this share of its cameras has no
+ * answer yet (a cold cache). A few new cameras or expired misses are looked
+ * up in the background instead of holding the refresh. */
+export const KARTVERKET_WAIT_MISSING_SHARE = 0.1;
+/** A point Kartverket has no height for (some ferry quays and bridges) is not
+ * asked about again for this long. */
+export const KARTVERKET_NULL_TTL_MS = 6 * 60 * 60 * 1000;
+/** After a failing request (408, 429, 5xx, a network error, a request
+ * timeout or an unreadable body), no lookup starts for this long, or for the
+ * Retry-After the service sent, bounded to [1 min, 1 h]. Longer than the
+ * 15-minute catalog refresh (CCTV_SOURCE_CACHE_MS), so a failing service
+ * costs one attempt per backoff, not one per refresh. */
+export const KARTVERKET_FAILURE_BACKOFF_MS = 30 * 60 * 1000;
+export const KARTVERKET_MAX_BACKOFF_MS = 60 * 60 * 1000;
+export const KARTVERKET_MAX_RESPONSE_BYTES = 256 * 1024;
+/** Plausible Norwegian terrain: below sea level only at the shore, and no road
+ * camera sits above ~1,500 m. Anything outside is treated as no height. */
+export const KARTVERKET_HEIGHT_RANGE_M = Object.freeze({ min: -20, max: 2500 });
+/** Same identity the other CCTV upstream requests send. */
+export const KARTVERKET_USER_AGENT = 'gods-eye-view-cctv-proxy/1.0';
+
 /** NYC DOT Traffic Management Center cameras: one keyless JSON list for the
  * five boroughs (~980 cameras). Frames are stills synthesized from each
  * camera UUID on the city's webcam host — never taken from the payload. */
@@ -460,6 +505,10 @@ export const CCTV_SOURCE_CACHE_MS = 15 * 60 * 1000;
  * pending forever — a hung fetch aborts, the loader returns [], and
  * serve-stale/other packs take over. */
 export const CCTV_SOURCE_FETCH_TIMEOUT_MS = 15 * 1000;
+/** Tarktee's two DATEX feeds (120 and 160 KB) answered in 2 to 16.6 s across
+ * ten requests on 2026-10-18, so the shared 15 s timeout dropped the Estonian
+ * pack on about a third of loads. Both must arrive for any camera to load. */
+export const TARKTEE_FETCH_TIMEOUT_MS = 30 * 1000;
 /** Individual CCTV image fetches must settle before the active 10-second
  * client refresh cadence. A bounded miss can fall through to Street View or
  * the synthetic frame instead of leaving the browser preview pending. */

server/providers/cctv/sources.js

@@ -42,6 +42,7 @@ import {
   TARKTEE_IMAGES_URL,
   TARKTEE_IMAGE_ORIGIN,
   DEFAULT_TARKTEE_MAX_SOURCES,
+  TARKTEE_MAX_CATALOG_BYTES,
   TARKTEE_ANCHORS,
   DEFAULT_WARENDORF_SOURCE_FILE,
   WARENDORF_IMAGE_ORIGINS,
@@ -76,6 +77,7 @@ import {
   TORONTO_GROUND_ELEVATION_M,
   TORONTO_MAX_CATALOG_BYTES,
   CCTV_SOURCE_FETCH_TIMEOUT_MS,
+  TARKTEE_FETCH_TIMEOUT_MS,
   DELDOT_CCTV_URL,
   DEFAULT_DELDOT_MAX_SOURCES,
   DELDOT_ANCHORS,
@@ -95,6 +97,19 @@ import {
   DEFAULT_VEGVESEN_MAX_SOURCES,
   VEGVESEN_MAX_CATALOG_BYTES,
   NORWAY_ANCHORS,
+  KARTVERKET_HEIGHT_URL,
+  KARTVERKET_MAX_POINTS,
+  KARTVERKET_CONCURRENCY,
+  KARTVERKET_LOOKUP_TIMEOUT_MS,
+  KARTVERKET_REQUEST_TIMEOUT_MS,
+  KARTVERKET_WAIT_MISSING_SHARE,
+  KARTVERKET_CATALOG_WAIT_MS,
+  KARTVERKET_NULL_TTL_MS,
+  KARTVERKET_FAILURE_BACKOFF_MS,
+  KARTVERKET_MAX_BACKOFF_MS,
+  KARTVERKET_USER_AGENT,
+  KARTVERKET_MAX_RESPONSE_BYTES,
+  KARTVERKET_HEIGHT_RANGE_M,
   NYC_CAMERAS_URL,
   NYC_IMAGE_URL,
   NYC_CAMERA_ID_RE,
@@ -1232,28 +1247,73 @@ export function parseTarkteeDatexImages(xml) {
  */
 export async function loadTarkteeSourcesFromDatex() {
   try {
-    const [locResp, imgResp] = await Promise.all([
+    const responses = await Promise.allSettled([
       fetch(TARKTEE_LOCATIONS_URL, {
         headers: { Accept: 'application/xml,text/xml,*/*' },
-        signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
+        redirect: 'manual',
+        signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS),
       }),
       fetch(TARKTEE_IMAGES_URL, {
         headers: { Accept: 'application/xml,text/xml,*/*' },
-        signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
+        redirect: 'manual',
+        signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS),
       }),
     ]);
+    const [locResp, imgResp] = responses.map((result) =>
+      result.status === 'fulfilled' ? result.value : undefined,
+    );
+    // Both responses are fetched in parallel; if either is rejected or overflows
+    // its byte cap, both bodies must be released so neither transport leaks.
+    const discardBoth = async () => {
+      await Promise.allSettled([
+        locResp?.body?.cancel(),
+        imgResp?.body?.cancel(),
+      ]);
+      return [];
+    };
+    const rejected = responses.find((result) => result.status === 'rejected');
+    if (rejected) {
+      await discardBoth();
+      throw rejected.reason;
+    }
+    if (
+      (locResp.status >= 300 && locResp.status < 400) ||
+      (imgResp.status >= 300 && imgResp.status < 400)
+    ) {
+      console.warn(
+        '[CCTV] Tarktee DATEX catalog redirected; redirects are not followed',
+      );
+      return discardBoth();
+    }
     if (!locResp.ok) {
       console.warn('[CCTV] Tarktee locations download failed:', locResp.status);
-      return [];
+      return discardBoth();
     }
     if (!imgResp.ok) {
       console.warn('[CCTV] Tarktee images download failed:', imgResp.status);
-      return [];
+      return discardBoth();
+    }
+    let locXml;
+    let imgXml;
+    const readAbort = new AbortController();
+    try {
+      [locXml, imgXml] = await Promise.all([
+        readResponseTextCapped(
+          locResp,
+          TARKTEE_MAX_CATALOG_BYTES,
+          readAbort.signal,
+        ),
+        readResponseTextCapped(
+          imgResp,
+          TARKTEE_MAX_CATALOG_BYTES,
+          readAbort.signal,
+        ),
+      ]);
+    } catch (error) {
+      readAbort.abort();
+      await discardBoth();
+      throw error;
     }
-    const [locXml, imgXml] = await Promise.all([
-      locResp.text(),
-      imgResp.text(),
-    ]);
     const locations = parseTarkteeDatexLocations(locXml);
     const images = parseTarkteeDatexImages(imgXml);
     if (!locations.size || !images.size) {
@@ -1265,6 +1325,7 @@ export async function loadTarkteeSourcesFromDatex() {
     }
 
     const cameras = [];
+    const seen = new Set();
     for (const [locationId, loc] of locations.entries()) {
       const imageUrl = images.get(locationId);
       if (!imageUrl) continue;
@@ -1276,6 +1337,8 @@ export async function loadTarkteeSourcesFromDatex() {
       const cameraId = numMatch
         ? `ee-tarktee-${numMatch[1]}`
         : `ee-tarktee-${locationId}`;
+      if (seen.has(cameraId)) continue;
+      seen.add(cameraId);
       cameras.push({
         id: cameraId,
         name: loc.name,
@@ -2427,14 +2490,11 @@ export function vegvesenCameraToSource(feature) {
     fovDeg: 44,
     rangeM: 145,
     mountHeightM: 8,
-    // KNOWN LIMITATION: the feed's coordinates are 2D, so this is one flat
-    // prior for the whole country, while cameras run from sea level to
-    // mountain passes near 1,000 m (Haukelifjell, Sjonfjellet, Rugeldalen).
-    // The point-height prior and the client's one-shot ground snap correct it
-    // where they resolve; on a stack where neither does (no 3D tiles), a pass
-    // camera stays hundreds of metres below the terrain, the same risk the
-    // Caltrans pack documents. A per-camera height (e.g. Kartverket's keyless
-    // point-height API) would remove it.
+    // The feed's coordinates are 2D. The loader replaces this flat prior with
+    // Kartverket's terrain height for each camera (applyVegvesenGroundHeights);
+    // it only survives when that lookup fails, and then a mountain-pass camera
+    // can sit below the terrain on a stack without 3D tiles, the same risk the
+    // Caltrans pack documents.
     groundElevationM: 150,
     feedType: videoUrl ? 'hls' : 'image',
     url: videoUrl || imageUrl,
@@ -2454,6 +2514,7 @@ export function vegvesenCameraToSource(feature) {
  * @returns {Promise<Array<object>>} Normalized camera source objects.
  */
 export async function loadVegvesenSourcesFromOpenData() {
+  const startedAt = Date.now();
   try {
     const endpoint = process.env.CCTV_VEGVESEN_URL || DEFAULT_VEGVESEN_CCTV_URL;
     const resp = await fetch(endpoint, {
@@ -2499,6 +2560,12 @@ export async function loadVegvesenSourcesFromOpenData() {
       ? Math.max(8, Math.min(1000, Math.floor(maxRaw)))
       : DEFAULT_VEGVESEN_MAX_SOURCES;
     const prioritized = prioritizeSources(cameras, maxCount, NORWAY_ANCHORS);
+    await applyVegvesenGroundHeights(prioritized, {
+      waitMs: Math.max(
+        0,
+        KARTVERKET_CATALOG_WAIT_MS - (Date.now() - startedAt),
+      ),
+    });
     console.log(
       `[CCTV] Loaded Vegvesen camera sources: ${cameras.length} (using nearest ${prioritized.length})`,
     );
@@ -2512,6 +2579,322 @@ export async function loadVegvesenSourcesFromOpenData() {
   }
 }
 
+/**
+ * Kartverket answers by rounded position: `{z, expires}`, where z is metres or
+ * null for "no height there". Heights never expire (camera positions barely
+ * change); nulls expire after KARTVERKET_NULL_TTL_MS. The 15-minute catalog
+ * refresh therefore only asks about new or moved cameras. Bounded well above
+ * the ~900 cameras the pack can load.
+ */
+const kartverketHeightCache = new Map();
+const KARTVERKET_CACHE_LIMIT = 5000;
+/** The lookup in flight, so overlapping catalog loads never start a second. */
+let kartverketPending = null;
+/** No lookup starts before this time (epoch ms) after a transport failure. */
+let kartverketBackoffUntil = 0;
+/** Bumped by clearKartverketHeightCache, so a lookup started before a clear
+ * can no longer write to the cache, the backoff or the pending slot. */
+let kartverketGeneration = 0;
+
+/** Cache key for one position, at the ~0.1 m precision the request sends. */
+const kartverketKey = (lon, lat) => `${lon.toFixed(6)},${lat.toFixed(6)}`;
+
+/** Statuses that say the service, not the request, is failing. */
+const isKartverketTransportStatus = (status) =>
+  status === 408 || status === 429 || status >= 500;
+
+/** Forget cached heights, the backoff and any lookup in flight. Tests only. */
+export function clearKartverketHeightCache() {
+  kartverketGeneration += 1;
+  kartverketHeightCache.clear();
+  kartverketPending = null;
+  kartverketBackoffUntil = 0;
+}
+
+/**
+ * How long to back off after a failed request: the Retry-After the service
+ * sent (seconds or an HTTP date), else the default, bounded to [1 min, 1 h].
+ *
+ * @param {?string} retryAfter - The Retry-After header, if any.
+ * @param {number} now - Epoch ms.
+ * @returns {number} Milliseconds.
+ */
+export function kartverketBackoffMs(retryAfter, now = Date.now()) {
+  let ms = KARTVERKET_FAILURE_BACKOFF_MS;
+  const text = String(retryAfter ?? '').trim();
+  if (/^\d+$/.test(text)) ms = Number(text) * 1000;
+  else if (text && Number.isFinite(Date.parse(text))) {
+    ms = Date.parse(text) - now;
+  }
+  return Math.max(60 * 1000, Math.min(KARTVERKET_MAX_BACKOFF_MS, ms));
+}
+
+/**
+ * Terrain heights above sea level (NN2000) from Kartverket's keyless
+ * point-height API, KARTVERKET_MAX_POINTS points per request. Each answer is
+ * matched back by the coordinates the API echoes, and only for points that
+ * request asked about, then reported through `into.set(key, z)` as its
+ * request lands, so a caller that stops waiting keeps what has arrived. `z` is
+ * metres, or null when the API has no usable height there. A point a 200
+ * answer leaves out is reported as null too, so it is not asked again on
+ * every refresh.
+ *
+ * Over water `/punkt` returns the depth of the sea floor (`datakilde:
+ * dybdekurver`, `terreng: Havflate`; −547 m under Nordhordlandsbrua). Either
+ * marker counts as 0 m, the water surface. For the ~14 bridge cameras that is
+ * below the deck, but far closer than the sea floor or a flat prior.
+ *
+ * The first transport failure (a 408, 429 or 5xx, a network error, a request
+ * past KARTVERKET_REQUEST_TIMEOUT_MS, an unreadable or oversized body, or the
+ * shared deadline) stops the lookup: no further requests start, and the
+ * failure is returned so the caller can back off. Points in failed requests
+ * are not reported, so they are asked again later. Any other 4xx rejects that
+ * request's points, not the service: they are reported as null and the lookup
+ * carries on.
+ *
+ * @param {Array<{lon:number, lat:number}>} points
+ * @param {{into?: {set: Function}, timeoutMs?: number, requestTimeoutMs?: number}} [options]
+ * @returns {Promise<{heights: Map<string, ?number>, failure: ?{status: ?number, retryAfter: ?string, deadline: boolean}}>}
+ *   `heights` is `into` when it is a Map, else a Map of what was reported.
+ *   `failure.deadline` is true when only the shared deadline ran out.
+ */
+export async function fetchKartverketHeights(
+  points,
+  {
+    into = new Map(),
+    timeoutMs = KARTVERKET_LOOKUP_TIMEOUT_MS,
+    requestTimeoutMs = KARTVERKET_REQUEST_TIMEOUT_MS,
+  } = {},
+) {
+  const heights = into instanceof Map ? into : new Map();
+  const report = (key, z) => {
+    if (heights !== into) heights.set(key, z);
+    into.set(key, z);
+  };
+  const unique = Array.from(
+    new Map(
+      (points || [])
+        .filter((p) => isLikelyNorwayCoordinate(p?.lat, p?.lon))
+        .map((p) => [kartverketKey(p.lon, p.lat), p]),
+    ).values(),
+  );
+  let failure = null;
+  if (!unique.length) return { heights, failure };
+  const batches = [];
+  for (let i = 0; i < unique.length; i += KARTVERKET_MAX_POINTS) {
+    batches.push(unique.slice(i, i + KARTVERKET_MAX_POINTS));
+  }
+  const deadline = AbortSignal.timeout(timeoutMs);
+  const fetchBatch = async (batch) => {
+    const asked = new Set(batch.map((p) => kartverketKey(p.lon, p.lat)));
+    const url = new URL(KARTVERKET_HEIGHT_URL);
+    url.searchParams.set(
+      'punkter',
+      JSON.stringify(
+        batch.map((p) => [Number(p.lon.toFixed(6)), Number(p.lat.toFixed(6))]),
+      ),
+    );
+    url.searchParams.set('koordsys', '4258');
+    const resp = await fetch(url, {
+      headers: {
+        Accept: 'application/json',
+        'User-Agent': KARTVERKET_USER_AGENT,
+      },
+      redirect: 'error',
+      signal: AbortSignal.any([
+        deadline,
+        AbortSignal.timeout(requestTimeoutMs),
+      ]),
+    });
+    if (!resp.ok) {
+      try {
+        await resp.body?.cancel();
+      } catch {
+        /* no-op */
+      }
+      if (!isKartverketTransportStatus(resp.status)) {
+        console.warn(
+          `[CCTV] Kartverket rejected ${asked.size} points: HTTP ${resp.status}`,
+        );
+        for (const key of asked) report(key, null);
+        return;
+      }
+      const error = new Error(`HTTP ${resp.status}`);
+      error.status = resp.status;
+      error.retryAfter = resp.headers.get('retry-after');
+      throw error;
+    }
+    const payload = await readResponseJsonCapped(
+      resp,
+      KARTVERKET_MAX_RESPONSE_BYTES,
+    );
+    for (const point of Array.isArray(payload?.punkter)
+      ? payload.punkter
+      : []) {
+      const x = toFiniteNumber(point?.x);
+      const y = toFiniteNumber(point?.y);
+      if (!Number.isFinite(x) || !Number.isFinite(y)) continue;
+      const key = kartverketKey(x, y);
+      // Only the points this request asked about: an odd answer can neither
+      // grow the cache nor overwrite another camera.
+      if (!asked.has(key)) continue;
+      asked.delete(key);
+      // `z: null` means no height there; Number(null) would read 0.
+      const raw = point?.z;
+      let z =
+        raw === null || raw === undefined || raw === ''
+          ? NaN
+          : toFiniteNumber(raw);
+      if (
+        Number.isFinite(z) &&
+        (point?.datakilde === 'dybdekurver' || point?.terreng === 'Havflate')
+      ) {
+        z = 0;
+      }
+      const usable =
+        Number.isFinite(z) &&
+        z >= KARTVERKET_HEIGHT_RANGE_M.min &&
+        z <= KARTVERKET_HEIGHT_RANGE_M.max;
+      report(key, usable ? z : null);
+    }
+    // A 200 that leaves a point out has no height for it either.
+    for (const key of asked) report(key, null);
+  };
+  let next = 0;
+  const worker = async () => {
+    while (next < batches.length && !failure) {
+      const batch = batches[next++];
+      try {
+        await fetchBatch(batch);
+      } catch (error) {
+        if (!failure) {
+          failure = {
+            status: Number.isFinite(error?.status) ? error.status : null,
+            retryAfter: error?.retryAfter ?? null,
+            deadline: deadline.aborted,
+          };
+          console.warn(
+            '[CCTV] Kartverket height lookup failed:',
+            error?.message || error,
+          );
+        }
+      }
+    }
+  };
+  await Promise.all(
+    Array.from(
+      { length: Math.min(KARTVERKET_CONCURRENCY, batches.length) },
+      worker,
+    ),
+  );
+  return { heights, failure };
+}
+
+/**
+ * Give each Vegvesen camera its own terrain height as the ground prior, in
+ * place. Waits at most `waitMs` for positions it has no answer for, and only
+ * when more than KARTVERKET_WAIT_MISSING_SHARE of the cameras are missing
+ * (a cold cache); a few new cameras or expired misses are looked up in the
+ * background. A lookup still running when the wait ends keeps filling the
+ * cache for the next catalog refresh. Cameras without a height keep the flat
+ * prior. After a transport failure no lookup starts until the backoff ends,
+ * which outlasts the 15-minute catalog refresh, so a failing Kartverket costs
+ * one attempt per backoff, not one per refresh. A lookup cut short only by its
+ * shared deadline is resumed on the next refresh without a backoff. Cameras
+ * without finite coordinates are skipped. Never throws.
+ * CCTV_VEGVESEN_HEIGHTS=0 turns it off.
+ *
+ * @param {Array<object>} cameras - Sources from vegvesenCameraToSource().
+ * @param {{waitMs?: number, now?: () => number, timeoutMs?: number}} [options]
+ *   `timeoutMs` is the lookup's shared deadline.
+ * @returns {Promise<number>} How many cameras got a terrain height.
+ */
+export async function applyVegvesenGroundHeights(
+  cameras,
+  {
+    waitMs = KARTVERKET_CATALOG_WAIT_MS,
+    now = Date.now,
+    timeoutMs = KARTVERKET_LOOKUP_TIMEOUT_MS,
+  } = {},
+) {
+  if (String(process.env.CCTV_VEGVESEN_HEIGHTS || '1').trim() === '0') return 0;
+  const list = (Array.isArray(cameras) ? cameras : []).filter(
+    (c) => Number.isFinite(c?.lon) && Number.isFinite(c?.lat),
+  );
+  const startedAt = now();
+  const missing = list.filter((c) => {
+    const entry = kartverketHeightCache.get(kartverketKey(c.lon, c.lat));
+    return !entry || entry.expires <= startedAt;
+  });
+  if (
+    missing.length &&
+    !kartverketPending &&
+    startedAt >= kartverketBackoffUntil
+  ) {
+    if (kartverketHeightCache.size + missing.length > KARTVERKET_CACHE_LIMIT) {
+      kartverketHeightCache.clear();
+    }
+    const generation = kartverketGeneration;
+    const current = () => generation === kartverketGeneration;
+    const sink = {
+      set: (key, z) => {
+        if (!current()) return;
+        kartverketHeightCache.set(key, {
+          z,
+          expires: z === null ? now() + KARTVERKET_NULL_TTL_MS : Infinity,
+        });
+      },
+    };
+    const pending = fetchKartverketHeights(missing, { into: sink, timeoutMs })
+      .then(({ failure }) => {
+        if (failure && !failure.deadline && current()) {
+          kartverketBackoffUntil =
+            now() + kartverketBackoffMs(failure.retryAfter, now());
+        }
+      })
+      .catch((error) => {
+        if (current()) {
+          kartverketBackoffUntil = now() + KARTVERKET_FAILURE_BACKOFF_MS;
+        }
+        console.warn(
+          '[CCTV] Kartverket height lookup error:',
+          error?.message || error,
+        );
+      })
+      .finally(() => {
+        if (kartverketPending === pending) kartverketPending = null;
+      });
+    kartverketPending = pending;
+  }
+  const cold = missing.length > list.length * KARTVERKET_WAIT_MISSING_SHARE;
+  if (cold && kartverketPending && waitMs > 0) {
+    let timer;
+    await Promise.race([
+      kartverketPending,
+      new Promise((resolve) => {
+        timer = setTimeout(resolve, waitMs);
+        timer.unref?.();
+      }),
+    ]);
+    clearTimeout(timer);
+  }
+  let applied = 0;
+  for (const camera of list) {
+    const z = kartverketHeightCache.get(
+      kartverketKey(camera.lon, camera.lat),
+    )?.z;
+    if (!Number.isFinite(z)) continue;
+    camera.groundElevationM = Math.round(z * 10) / 10;
+    applied += 1;
+  }
+  if (list.length) {
+    console.log(
+      `[CCTV] Kartverket ground heights: ${applied}/${list.length} Vegvesen cameras`,
+    );
+  }
+  return applied;
+}
+
 const NYC_BOROUGHS = new Set([
   'Manhattan',
   'Brooklyn',

server/providers/common/request.js

@@ -14,16 +14,34 @@ async function readRequestBodyCapped(req, maxBytes) {
   return Buffer.concat(chunks);
 }
 
-function readRequestBody(req, maxBytes = 1024 * 1024) {
+/** Optional cancellation discards buffered data while the request drains. */
+function readRequestBody(req, maxBytes = 1024 * 1024, signal) {
   return new Promise((resolve, reject) => {
     let total = 0;
     let bodyTooLarge = false;
+    let settled = false;
     const chunks = [];
-    req.on('data', (chunk) => {
+    const stopReading = () => {
+      if (!signal) return;
+      req.off('data', data);
+      req.off('end', end);
+      signal.removeEventListener('abort', abort);
+    };
+    const fail = (error) => {
+      if (signal) {
+        if (settled) return;
+        settled = true;
+        chunks.length = 0;
+        signal.removeEventListener('abort', abort);
+      }
+      reject(error);
+    };
+    const abort = () => fail(signal.reason);
+    const data = (chunk) => {
       // Keep draining after the cap without retaining more bytes. Destroying
       // the request also destroys its socket, so the handler cannot deliver
       // the fixed error response it constructs for BODY_TOO_LARGE.
-      if (bodyTooLarge) return;
+      if (bodyTooLarge || settled) return;
       total += chunk.length;
       if (total > maxBytes) {
         // Same marker readRequestBodyCapped sets, so a caller can tell an
@@ -32,15 +50,42 @@ function readRequestBody(req, maxBytes = 1024 * 1024) {
         chunks.length = 0;
         const error = new Error(`Request body exceeds ${maxBytes} bytes`);
         error.code = 'BODY_TOO_LARGE';
-        reject(error);
+        fail(error);
         return;
       }
       chunks.push(chunk);
-    });
-    req.on('end', () => {
+    };
+    const end = () => {
+      if (settled) {
+        stopReading();
+        return;
+      }
       if (!bodyTooLarge) resolve(Buffer.concat(chunks).toString('utf8'));
-    });
-    req.on('error', reject);
+      if (signal) {
+        settled = true;
+        chunks.length = 0;
+        stopReading();
+      }
+    };
+    const close = () => {
+      fail(
+        new DOMException('Request closed before its body ended', 'AbortError'),
+      );
+      stopReading();
+      req.off('error', fail);
+      req.off('close', close);
+    };
+    if (signal) {
+      // Abort/overflow release retained bytes but keep draining. Retain an
+      // error handler until close, including a late error after end/abort.
+      req.on('close', close);
+      signal.addEventListener('abort', abort, { once: true });
+      if (signal.aborted) abort();
+    }
+    req.on('data', data);
+    req.on('end', end);
+    req.on('error', fail);
+    if (signal && req.destroyed) close();
   });
 }
 

server/providers/common/request.test.mjs

@@ -0,0 +1,137 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { EventEmitter, getEventListeners } from 'node:events';
+import { readRequestBody } from './request.js';
+
+function fixture() {
+  const req = new EventEmitter();
+  req.pause = () => assert.fail('The reader must keep draining');
+  req.destroy = () =>
+    assert.fail('The reader must preserve the response socket');
+  const owner = new AbortController();
+  return { req, owner };
+}
+
+function finish(req, owner) {
+  assert.doesNotThrow(() => req.emit('error', new Error('late stream error')));
+  req.emit('end');
+  assert.doesNotThrow(() => req.emit('error', new Error('error before close')));
+  req.emit('close');
+  for (const event of ['data', 'end', 'error', 'close'])
+    assert.equal(req.listenerCount(event), 0, event);
+  assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
+}
+
+test('default and cancellable readers preserve byte caps and UTF-8 across chunks', async () => {
+  for (const useSignal of [false, true]) {
+    for (const text of ['', 'hello', 'é']) {
+      const { req, owner } = fixture();
+      const bytes = Buffer.from(text);
+      const body = readRequestBody(
+        req,
+        bytes.length,
+        useSignal ? owner.signal : undefined,
+      );
+      for (const byte of bytes) req.emit('data', Buffer.from([byte]));
+      req.emit('end');
+      assert.equal(await body, text);
+      if (useSignal) finish(req, owner);
+    }
+  }
+});
+
+test('default overflow rejects before end and drains without pausing or destroying', async () => {
+  const { req } = fixture();
+  const rejection = assert.rejects(readRequestBody(req, 2), {
+    code: 'BODY_TOO_LARGE',
+  });
+  req.emit('data', Buffer.from('é'));
+  req.emit('data', Buffer.from('!'));
+  await rejection;
+  req.emit('data', Buffer.alloc(4096));
+  assert.doesNotThrow(() => req.emit('error', new Error('late')));
+  req.emit('end');
+});
+
+test('partial abort settles before end, discards late data and guards errors until close', async () => {
+  const { req, owner } = fixture();
+  const reason = new Error('request deadline');
+  const rejection = assert.rejects(
+    readRequestBody(req, 1024, owner.signal),
+    (error) => error === reason,
+  );
+  req.emit('data', Buffer.from('partial'));
+  owner.abort(reason);
+  await rejection;
+  assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
+  req.emit('data', Buffer.alloc(4096));
+  finish(req, owner);
+});
+
+test('already-aborted and already-closed requests settle and release their listeners', async () => {
+  for (const closed of [false, true]) {
+    const { req, owner } = fixture();
+    if (closed) req.destroyed = true;
+    else owner.abort();
+    await assert.rejects(readRequestBody(req, 1024, owner.signal), {
+      name: 'AbortError',
+    });
+    if (!closed) finish(req, owner);
+    for (const event of ['data', 'end', 'error', 'close'])
+      assert.equal(req.listenerCount(event), 0);
+    assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
+  }
+});
+
+test('overflow keeps its first failure through later abort and stream errors', async () => {
+  const { req, owner } = fixture();
+  const rejection = assert.rejects(readRequestBody(req, 2, owner.signal), {
+    code: 'BODY_TOO_LARGE',
+  });
+  req.emit('data', Buffer.from('é!'));
+  await rejection;
+  owner.abort(new Error('later deadline'));
+  req.emit('data', Buffer.alloc(4096));
+  finish(req, owner);
+});
+
+test('early close and request error reject once; successful end wins over later abort', async () => {
+  for (const event of ['close', 'error', 'end']) {
+    const { req, owner } = fixture();
+    const reason = new Error('request failed');
+    const body = readRequestBody(req, 32, owner.signal);
+    const observed =
+      event === 'end'
+        ? body.then((value) => assert.equal(value, 'ok'))
+        : assert.rejects(
+            body,
+            event === 'close'
+              ? { name: 'AbortError' }
+              : (error) => error === reason,
+          );
+    req.emit('data', Buffer.from('ok'));
+    req.emit(event, reason);
+    owner.abort(new Error('later abort'));
+    await observed;
+    if (event !== 'close') finish(req, owner);
+    assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
+  }
+});
+
+test('a prior data observer can abort synchronously without retaining or resolving that chunk', async () => {
+  const { req, owner } = fixture();
+  const reason = new Error('observer stopped request');
+  const external = () => owner.abort(reason);
+  req.on('data', external);
+  const rejection = assert.rejects(
+    readRequestBody(req, 1024, owner.signal),
+    (error) => error === reason,
+  );
+  req.emit('data', Buffer.from('must be discarded'));
+  await rejection;
+  req.emit('end');
+  req.emit('close');
+  assert.deepEqual(req.listeners('data'), [external]);
+  assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
+  req.off('data', external);
+});

server/providers/gemini.js

@@ -0,0 +1,22 @@
+import { createGeminiTokenHandler } from './gemini/realtime.js';
+
+/** Mint constrained Live API credentials; the permanent key stays on Node. */
+function geminiLiveProxy({ annotationGuidance, realtime = {} } = {}) {
+  function install(middlewares) {
+    middlewares.use(
+      '/api/gemini/token',
+      createGeminiTokenHandler({ ...realtime, annotationGuidance }),
+    );
+  }
+  return {
+    name: 'gemini-live-proxy',
+    configureServer(server) {
+      install(server.middlewares);
+    },
+    configurePreviewServer(server) {
+      install(server.middlewares);
+    },
+  };
+}
+
+export { geminiLiveProxy };

server/providers/gemini/config.js

@@ -0,0 +1,53 @@
+import { realtimeInstructions } from '../openai/instructions.js';
+import { GEV_REALTIME_TOOLS } from '../openai/tools.js';
+
+export const GEMINI_LIVE_MODEL_DEFAULT = 'gemini-3.8-live';
+export const GEMINI_TOKEN_ENDPOINT =
+  'https://generativelanguage.googleapis.com/v1beta/auth_tokens';
+export const GEMINI_LIVE_WEBSOCKET_URL =
+  'wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContentConstrained';
+
+export function normalizeGeminiModel(value) {
+  const model = String(value ?? GEMINI_LIVE_MODEL_DEFAULT)
+    .trim()
+    .replace(/^models\//, '');
+  return /^gemini-[a-z0-9][a-z0-9.-]{0,119}$/.test(model) ? model : null;
+}
+
+/** REST BidiGenerateContentSetup fields, excluding its model resource name. */
+export function createGeminiLiveConfig(
+  annotationGuidance,
+  inputMode = 'open-mic',
+  tools = GEV_REALTIME_TOOLS,
+) {
+  if (!['open-mic', 'push-to-talk'].includes(inputMode))
+    throw new Error('Invalid Gemini input mode');
+  const config = {
+    generationConfig: { responseModalities: ['AUDIO'] },
+    systemInstruction: {
+      parts: [{ text: realtimeInstructions(annotationGuidance) }],
+    },
+    tools: [
+      {
+        functionDeclarations: tools.map(
+          ({ name, description, parameters }) => ({
+            name,
+            description,
+            // GEV's canonical schemas are JSON Schema, not Google's Schema.
+            parametersJsonSchema: structuredClone(parameters),
+            // App actions must finish before the model confirms their result.
+            behavior: 'BLOCKING',
+          }),
+        ),
+      },
+    ],
+    inputAudioTranscription: {},
+    outputAudioTranscription: {},
+    contextWindowCompression: { slidingWindow: {} },
+  };
+  if (inputMode === 'push-to-talk')
+    config.realtimeInputConfig = {
+      automaticActivityDetection: { disabled: true },
+    };
+  return config;
+}

server/providers/gemini/realtime.js

@@ -0,0 +1,208 @@
+import { waitForSignal } from '../../../src/sources/requestDeadline.js';
+import { clientKey, makeCostRateLimiter } from '../common/rate-limit.js';
+import { readResponseJsonCapped } from '../../../src/sources/httpBody.js';
+import {
+  GEMINI_TOKEN_ENDPOINT,
+  GEMINI_LIVE_WEBSOCKET_URL,
+  GEMINI_LIVE_MODEL_DEFAULT,
+  createGeminiLiveConfig,
+  normalizeGeminiModel,
+} from './config.js';
+import {
+  isGeminiRequestOriginAllowed,
+  readGeminiTokenRequest,
+} from './request.js';
+
+const UPSTREAM_RESPONSE_BYTES = 32 * 1024;
+const MINT_FAILURE = 'Gemini voice could not connect. Try again.';
+
+export function createGeminiTokenHandler({
+  annotationGuidance,
+  tools,
+  fetchImpl = (...args) => fetch(...args),
+  resolveApiKey = () => process.env.GEMINI_API_KEY,
+  resolveModel = () =>
+    process.env.GEMINI_LIVE_MODEL || GEMINI_LIVE_MODEL_DEFAULT,
+  resolveHost = () => process.env.HOST,
+  resolveAllowedHosts = () => process.env.GEV_ALLOWED_HOSTS,
+  resolveRateLimit = () => process.env.GEV_RATELIMIT_GEMINI_PER_MIN ?? '30',
+  now = Date.now,
+  timeoutMs = 15_000,
+} = {}) {
+  let limiter;
+  return async (req, res) => {
+    if (req.aborted || res.destroyed || res.writableEnded) return;
+    res.setHeader('Cache-Control', 'no-store');
+    res.setHeader('Content-Type', 'application/json; charset=utf-8');
+    res.setHeader('X-Content-Type-Options', 'nosniff');
+    const reply = (status, errorOrPayload) => {
+      if (res.destroyed || res.writableEnded) return;
+      res.statusCode = status;
+      if (status >= 400) res.setHeader('Connection', 'close');
+      res.end(
+        JSON.stringify(
+          typeof errorOrPayload === 'string'
+            ? { error: errorOrPayload }
+            : errorOrPayload,
+        ),
+      );
+    };
+    if (req.method !== 'POST') {
+      res.setHeader('Allow', 'POST');
+      return reply(405, 'Method not allowed');
+    }
+    let apiKey;
+    let model;
+    try {
+      if (
+        !isGeminiRequestOriginAllowed(req, resolveHost(), resolveAllowedHosts())
+      )
+        return reply(403, 'Gemini voice requires a same-origin app request');
+      apiKey = String(resolveApiKey() || '').trim();
+      if (!apiKey)
+        return reply(503, 'Add a Gemini API key in Provider Settings');
+      model = normalizeGeminiModel(resolveModel());
+      if (!model)
+        return reply(503, 'The Gemini Live model configuration is invalid');
+      if (limiter === undefined)
+        limiter = makeCostRateLimiter(resolveRateLimit(), 30);
+      if (limiter && !limiter(clientKey(req))) {
+        res.setHeader('Retry-After', '60');
+        return reply(
+          429,
+          'Gemini voice is starting too often. Try again shortly.',
+        );
+      }
+    } catch {
+      // Connect does not observe the handler's returned promise. A synchronous
+      // configuration resolver must fail with a bounded, fixed response.
+      return reply(503, MINT_FAILURE);
+    }
+
+    const lifetime = new AbortController();
+    let disconnected = false;
+    const disconnect = () => {
+      disconnected = true;
+      lifetime.abort(new DOMException('Request closed', 'AbortError'));
+    };
+    const responseClosed = () => {
+      if (!res.writableEnded) disconnect();
+    };
+    req.once('aborted', disconnect);
+    res.once('close', responseClosed);
+    const deadline = setTimeout(() => {
+      lifetime.abort(
+        new DOMException('Token request timed out', 'TimeoutError'),
+      );
+    }, timeoutMs);
+    deadline.unref?.();
+    let acquiredResponse;
+    try {
+      const inputMode = await readGeminiTokenRequest(req, lifetime.signal);
+      lifetime.signal.throwIfAborted();
+      const config = createGeminiLiveConfig(
+        annotationGuidance,
+        inputMode,
+        tools,
+      );
+      const issuedAt = now();
+      // REST uses bidiGenerateContentSetup. liveConnectConstraints is the SDK
+      // input shape, converted before transport. With no fieldMask the complete
+      // setup is locked to this server-owned configuration.
+      const request = {
+        uses: 1,
+        expireTime: new Date(issuedAt + 30 * 60_000).toISOString(),
+        newSessionExpireTime: new Date(issuedAt + 60_000).toISOString(),
+        bidiGenerateContentSetup: { model: `models/${model}`, ...config },
+      };
+      const acquisition = Promise.resolve(
+        fetchImpl(GEMINI_TOKEN_ENDPOINT, {
+          method: 'POST',
+          redirect: 'error',
+          signal: lifetime.signal,
+          headers: {
+            'x-goog-api-key': apiKey,
+            'Content-Type': 'application/json',
+          },
+          body: JSON.stringify(request),
+        }),
+      );
+      // A fetch implementation can ignore abort. Bound this caller's wait and
+      // discard a response that arrives after timeout/disconnection.
+      void acquisition.then(
+        (late) => {
+          acquiredResponse = late;
+          if (lifetime.signal.aborted) void late.body?.cancel().catch(() => {});
+        },
+        () => {},
+      );
+      const response = await waitForSignal(acquisition, lifetime.signal);
+      lifetime.signal.throwIfAborted();
+      if (
+        response.redirected ||
+        (response.url && response.url !== GEMINI_TOKEN_ENDPOINT)
+      ) {
+        void response.body?.cancel().catch(() => {});
+        throw new Error('Unexpected token response destination');
+      }
+      if (!response.ok) {
+        // Never forward upstream bodies, headers or exception messages.
+        void response.body?.cancel().catch(() => {});
+        if (response.status === 429) {
+          res.setHeader('Retry-After', '60');
+          return reply(429, 'Gemini quota is unavailable. Try again later.');
+        }
+        if (response.status === 401 || response.status === 403)
+          return reply(
+            503,
+            'Gemini access was refused. Check the API key and model access.',
+          );
+        if (response.status === 400 || response.status === 404)
+          return reply(
+            502,
+            'Gemini Live is unavailable for the configured model.',
+          );
+        return reply(502, MINT_FAILURE);
+      }
+      const data = await readResponseJsonCapped(
+        response,
+        UPSTREAM_RESPONSE_BYTES,
+        lifetime.signal,
+      );
+      lifetime.signal.throwIfAborted();
+      if (
+        !data ||
+        Array.isArray(data) ||
+        typeof data !== 'object' ||
+        typeof data.name !== 'string' ||
+        data.name.length > 8192 ||
+        !/^auth_tokens\/[A-Za-z0-9._~+/=-]+$/.test(data.name) ||
+        data.name.includes(apiKey)
+      )
+        throw new Error('Invalid ephemeral token');
+      reply(200, {
+        token: data.name,
+        model,
+        config,
+        websocketUrl: GEMINI_LIVE_WEBSOCKET_URL,
+      });
+    } catch (error) {
+      if (disconnected || req.aborted) return;
+      if (error?.code === 'BODY_TOO_LARGE')
+        return reply(413, 'Token request is too large');
+      if (error?.code === 'BAD_CONTENT_TYPE')
+        return reply(415, 'Use an empty JSON token request');
+      if (error?.code === 'BAD_BODY')
+        return reply(400, 'Use an empty JSON token request');
+      reply(lifetime.signal.aborted ? 504 : 502, MINT_FAILURE);
+    } finally {
+      // Abort can happen after acquisition resolves but before this handler
+      // resumes and passes ownership to the body reader.
+      if (lifetime.signal.aborted)
+        void acquiredResponse?.body?.cancel().catch(() => {});
+      clearTimeout(deadline);
+      req.off('aborted', disconnect);
+      res.off('close', responseClosed);
+    }
+  };
+}

server/providers/gemini/request.js

@@ -0,0 +1,79 @@
+import { readRequestBody } from '../common/request.js';
+import { admitSameSiteRequest } from '../../../src/localRequestGate.mjs';
+import {
+  isAllowedHost,
+  resolveAllowedHosts,
+} from '../../../build/allowedHosts.js';
+
+/** Use the app's Host and request policies, with a required browser Origin. */
+export function isGeminiRequestOriginAllowed(
+  req,
+  configuredHost = '',
+  configuredAllowedHosts,
+) {
+  const host = req.headers?.host;
+  const origin = req.headers?.origin;
+  if (
+    typeof host !== 'string' ||
+    typeof origin !== 'string' ||
+    !host ||
+    !origin ||
+    /[\s/?#@\\]/.test(host)
+  )
+    return false;
+  const protocol = req.socket?.encrypted ? 'https:' : 'http:';
+  if (
+    !admitSameSiteRequest({
+      hostHeader: host,
+      protocol,
+      origin,
+      secFetchSite: req.headers?.['sec-fetch-site'],
+      proxyHeaders: req.headers || {},
+    }).ok
+  )
+    return false;
+  // The shared gate compares a present Origin exactly with this authority;
+  // this route also requires the Origin and an allowed Host.
+  const boundHost = String(configuredHost || '')
+    .trim()
+    .toLowerCase();
+  return isAllowedHost(host, resolveAllowedHosts(configuredAllowedHosts), [
+    boundHost,
+  ]);
+}
+
+/** Admit only the existing input gesture; callers cannot supply model/setup. */
+export async function readGeminiTokenRequest(req, signal, maxBytes = 1024) {
+  const declared = Number(req.headers?.['content-length']);
+  const fail = (code) =>
+    Object.assign(new Error('Invalid token request'), { code });
+  const body = await readRequestBody(
+    req,
+    maxBytes,
+    Number.isFinite(declared) && declared > maxBytes
+      ? AbortSignal.abort(fail('BODY_TOO_LARGE'))
+      : signal,
+  );
+  if (!body.trim()) return 'open-mic';
+  if (
+    !/^application\/json(?:\s*;|$)/i.test(req.headers?.['content-type'] || '')
+  )
+    throw fail('BAD_CONTENT_TYPE');
+  let parsed;
+  try {
+    parsed = JSON.parse(body);
+  } catch {
+    throw fail('BAD_BODY');
+  }
+  if (!parsed || Array.isArray(parsed) || typeof parsed !== 'object')
+    throw fail('BAD_BODY');
+  const keys = Object.keys(parsed);
+  if (!keys.length) return 'open-mic';
+  if (
+    keys.length === 1 &&
+    keys[0] === 'inputMode' &&
+    parsed.inputMode === 'push-to-talk'
+  )
+    return 'push-to-talk';
+  throw fail('BAD_BODY');
+}

server/providers/local.js

@@ -19,6 +19,7 @@ import { adsbLolProxy } from './aircraft/adsb-lol.js';
 import { aisLiveProxy } from './vessels/ais-live.js';
 import { trackBackfillProxies } from './aircraft/tracks.js';
 import { openAiRealtimeProxy } from './openai.js';
+import { geminiLiveProxy } from './gemini.js';
 import { googlePlacesContextProxy } from './places.js';
 import { mapillaryProxy } from './mapillary.js';
 import { keySetupEndpoint } from '../standalone/key-setup.js';
@@ -30,9 +31,9 @@ import { windProxy } from './wind.js';
 
 /**
  * Construct the local provider plugins in their established order.
- * `realtime` configures the voice session token endpoint.
+ * `realtime` configures OpenAI; `gemini` explicitly opts into Gemini Live.
  */
-function localProviderPlugins({ realtime } = {}) {
+function localProviderPlugins({ realtime, gemini } = {}) {
   return [
     openSkyProxy(),
     celestrakProxy(),
@@ -55,6 +56,7 @@ function localProviderPlugins({ realtime } = {}) {
     aisLiveProxy(),
     trackBackfillProxies(),
     openAiRealtimeProxy({ realtime }),
+    ...(gemini ? [geminiLiveProxy({ realtime: gemini })] : []),
     googlePlacesContextProxy(),
     windProxy(),
     weatherProxy(),

server/providers/mapillary.js

@@ -1,19 +1,106 @@
-import { installMapillaryRoutes } from './mapillary/routes.js';
+import { sameSiteGated } from './common/same-site.js';
+import { makeRateLimiter, clientKey } from './common/rate-limit.js';
+import {
+  fetchTile,
+  mapillaryToken,
+  parseTilePath,
+  retryAfterWait,
+  TileUpstreamError,
+} from './mapillary/tiles.js';
 
 /**
- * Vite plugin: cached Mapillary coverage tiles (token added server-side) and a
- * status route that says whether a token is configured.
+ * Tile requests per client IP per minute, cache hits included: over 60 new
+ * 9-tile views a minute, more than a person flying the camera reaches.
  */
-function mapillaryProxy() {
-  return {
-    name: 'mapillary-proxy',
-    configureServer(server) {
-      installMapillaryRoutes(server.middlewares);
-    },
-    configurePreviewServer(server) {
-      installMapillaryRoutes(server.middlewares);
-    },
-  };
+export const TILE_ROUTE_MAX_PER_MIN = 600;
+
+function sendJson(res, status, payload) {
+  res.statusCode = status;
+  res.setHeader('Content-Type', 'application/json; charset=utf-8');
+  res.setHeader('Cache-Control', 'no-store');
+  res.end(JSON.stringify(payload));
 }
 
-export { mapillaryProxy };
+/** GET /api/mapillary/status — whether a token is configured, never its value. */
+function handleStatus(req, res) {
+  if (req.method !== 'GET')
+    return sendJson(res, 405, { error: 'Method not allowed' });
+  sendJson(res, 200, { configured: Boolean(mapillaryToken()) });
+}
+
+/** GET /api/mapillary/tiles/coverage/{z}/{x}/{y} — trimmed protobuf tile. */
+async function handleTile(req, res, allow) {
+  if (req.method !== 'GET')
+    return sendJson(res, 405, { error: 'Method not allowed' });
+  if (!allow(clientKey(req))) {
+    res.setHeader('Retry-After', '5');
+    return sendJson(res, 429, {
+      error: 'Too many tile requests',
+      retryAfter: 5,
+    });
+  }
+  const address = parseTilePath((req.url || '').split('?')[0]);
+  if (!address)
+    return sendJson(res, 400, {
+      error: 'Tile path must be /coverage/{z}/{x}/{y} with z 11–14',
+    });
+  if (!mapillaryToken())
+    return sendJson(res, 503, { error: 'no_key', keyRequired: true });
+  // A client that leaves (a tile the camera moved past) cancels its fetch.
+  const left = new AbortController();
+  res.on?.('close', () => {
+    if (!res.writableEnded) left.abort();
+  });
+  try {
+    const { bytes, source } = await fetchTile(address, { signal: left.signal });
+    res.statusCode = bytes.length ? 200 : 204;
+    res.setHeader('Content-Type', 'application/x-protobuf');
+    res.setHeader('Cache-Control', 'public, max-age=3600');
+    res.setHeader('X-Gev-Cache', source);
+    res.end(bytes.length ? bytes : undefined);
+  } catch (error) {
+    if (left.signal.aborted) return;
+    const status = error instanceof TileUpstreamError ? error.status : 0;
+    // A rejected token is a key problem the panel can name, not a fault.
+    if (status === 401 || status === 403)
+      return sendJson(res, 403, {
+        error: 'Mapillary rejected the access token',
+        keyRejected: true,
+      });
+    if (status === 429) {
+      const retryAfter = retryAfterWait(error.retryAfter);
+      res.setHeader('Retry-After', String(retryAfter));
+      return sendJson(res, 429, {
+        error: 'Mapillary is rate-limiting tile requests',
+        retryAfter,
+      });
+    }
+    // An upstream 400 must not read as a malformed request to this proxy.
+    sendJson(res, 502, { error: error?.message || 'Tile fetch failed' });
+  }
+}
+
+/**
+ * Vite plugin: Mapillary coverage tiles with the token added server-side, and
+ * a status route. Both refuse cross-site requests so another page cannot
+ * spend the token.
+ */
+export function mapillaryProxy() {
+  const install = (middlewares) => {
+    const allow = makeRateLimiter({
+      windowMs: 60_000,
+      max: TILE_ROUTE_MAX_PER_MIN,
+      globalMax: TILE_ROUTE_MAX_PER_MIN * 4,
+    });
+    middlewares.use('/api/mapillary/status', sameSiteGated(handleStatus));
+    middlewares.use(
+      '/api/mapillary/tiles',
+      sameSiteGated((req, res) => handleTile(req, res, allow)),
+    );
+  };
+  return {
+    name: 'mapillary-proxy',
+    configureServer: (server) => install(server.middlewares),
+    configurePreviewServer: (server) => install(server.middlewares),
+  };
+}

server/providers/mapillary/constants.js

@@ -1,72 +0,0 @@
-import path from 'node:path';
-
-/** Mapillary vector tile root; the path template is /{layer}/2/{z}/{x}/{y}. */
-export const MAPILLARY_TILE_HOST = 'https://tiles.mapillary.com/maps/vtp';
-
-/** Public tile layer names this proxy exposes, mapped to Mapillary's ids. */
-export const TILE_LAYERS = Object.freeze({
-  // Region-sized z6–10 tiles are refused; the app never asks for them. The
-  // z14 `image` layer is ~98% of a tile and unused (positions come from the
-  // graph API), so it is dropped in transit.
-  coverage: Object.freeze({
-    upstream: 'mly1_public',
-    zoomRanges: Object.freeze([Object.freeze([0, 5]), Object.freeze([11, 14])]),
-    dropLayers: Object.freeze(['image']),
-  }),
-});
-
-/** Disk cache root, alongside the other providers' caches. */
-export const MAPILLARY_CACHE_DIR = path.join(
-  process.cwd(),
-  '.gev-cache',
-  'mapillary',
-);
-export const TILE_DISK_DIR = path.join(MAPILLARY_CACHE_DIR, 'tiles');
-
-/**
- * Longer than the 15-minute upstream Cache-Control: tiles change only when new
- * imagery is processed, and a day avoids re-downloading a city per camera move.
- */
-export const TILE_DISK_TTL_MS = 24 * 60 * 60 * 1000;
-
-/**
- * A tile write starts a background sweep at most this often: expired files go
- * first, then the oldest until the cache fits in TILE_DISK_MAX_BYTES.
- */
-export const TILE_DISK_SWEEP_INTERVAL_MS = 15 * 60 * 1000;
-export const TILE_DISK_MAX_BYTES = 1024 * 1024 * 1024;
-/** Per-file cost on top of its size, so empty tiles still count. */
-export const TILE_DISK_FILE_OVERHEAD_BYTES = 4096;
-
-/** A z14 image tile over a dense city is ~11 MB; anything past this is wrong. */
-export const TILE_MAX_BYTES = 48 * 1024 * 1024;
-
-export const TILE_MEMORY_BUDGET_BYTES = 96 * 1024 * 1024;
-/** Per-entry cost on top of its bytes, so empty tiles still count. */
-export const TILE_MEMORY_ENTRY_OVERHEAD_BYTES = 1024;
-export const TILE_FETCH_TIMEOUT_MS = 60_000;
-/**
- * Upstream fetches at once; further misses queue. A view needs at most 25
- * tiles, and the cap stops a burst of misses opening thousands of requests.
- */
-export const TILE_UPSTREAM_CONCURRENCY = 6;
-/**
- * Tile requests per client IP per minute, cache hits included. 600 is 24 new
- * 25-tile views a minute, more than a person flying the camera reaches, but it
- * stops one page draining the shared token into a Mapillary rate limit.
- */
-export const TILE_ROUTE_MAX_PER_MIN = 600;
-/**
- * How long a 401/403 answers tile misses before asking Mapillary again. A
- * changed token is tried at once.
- */
-export const TILE_KEY_REJECTED_HOLD_MS = 5 * 60 * 1000;
-/** After a 429 without a usable Retry-After, hold tile misses this long. */
-export const TILE_RATE_LIMIT_HOLD_MS = 60_000;
-/** The longest Retry-After honoured, so a bad header cannot stall coverage. */
-export const TILE_RATE_LIMIT_MAX_HOLD_MS = 10 * 60 * 1000;
-
-/** The client token lives in the browser by design; the server adds it to tile URLs too. */
-export function mapillaryToken() {
-  return String(process.env.MAPILLARY_CLIENT_TOKEN || '').trim();
-}

server/providers/mapillary/routes.js

@@ -1,95 +0,0 @@
-import { sameSiteGated } from '../common/same-site.js';
-import { makeRateLimiter, clientKey } from '../common/rate-limit.js';
-import { fetchTile, TileRequestError, TileUpstreamError } from './tiles.js';
-import { mapillaryToken, TILE_ROUTE_MAX_PER_MIN } from './constants.js';
-
-function sendJson(res, status, payload) {
-  res.statusCode = status;
-  res.setHeader('Content-Type', 'application/json; charset=utf-8');
-  res.setHeader('Cache-Control', 'no-store');
-  res.end(JSON.stringify(payload));
-}
-
-/** GET /api/mapillary/status — whether a token is configured, never its value. */
-function handleStatus(req, res) {
-  if (req.method !== 'GET')
-    return sendJson(res, 405, { error: 'Method not allowed' });
-  sendJson(res, 200, { configured: Boolean(mapillaryToken()) });
-}
-
-/** GET /api/mapillary/tiles/coverage/{z}/{x}/{y} — cached protobuf tile. */
-async function handleTile(req, res, allow) {
-  if (req.method !== 'GET')
-    return sendJson(res, 405, { error: 'Method not allowed' });
-  if (!allow(clientKey(req))) {
-    res.setHeader('Retry-After', '5');
-    return sendJson(res, 429, {
-      error: 'Too many tile requests',
-      retryAfter: 5,
-    });
-  }
-  const match = /^\/(coverage)\/(\d{1,2})\/(\d{1,6})\/(\d{1,6})$/.exec(
-    (req.url || '').split('?')[0],
-  );
-  if (!match)
-    return sendJson(res, 400, {
-      error: 'Tile path must be /coverage/{z}/{x}/{y}',
-    });
-  if (!mapillaryToken())
-    return sendJson(res, 503, { error: 'no_key', keyRequired: true });
-  const abandoned = new AbortController();
-  req.on?.('aborted', () => abandoned.abort());
-  res.on?.('close', () => abandoned.abort());
-  try {
-    const { bytes, source } = await fetchTile(
-      { layer: match[1], z: match[2], x: match[3], y: match[4] },
-      { signal: abandoned.signal },
-    );
-    if (res.writableEnded) return;
-    res.statusCode = bytes.length ? 200 : 204;
-    res.setHeader('Content-Type', 'application/x-protobuf');
-    res.setHeader('Cache-Control', 'public, max-age=3600');
-    res.setHeader('X-Gev-Cache', source);
-    res.end(bytes.length ? bytes : undefined);
-  } catch (error) {
-    if (res.writableEnded || abandoned.signal.aborted) return;
-    if (error instanceof TileRequestError)
-      return sendJson(res, error.status, { error: error.message });
-    if (error instanceof TileUpstreamError) {
-      // A rejected token is a key problem the panel can name, not a fault.
-      if (error.keyRejected)
-        return sendJson(res, 403, {
-          error: 'Mapillary rejected the access token',
-          keyRejected: true,
-        });
-      if (error.status === 429) {
-        const retryAfter = error.retryAfterSec || 60;
-        res.setHeader('Retry-After', String(retryAfter));
-        return sendJson(res, 429, {
-          error: 'Mapillary is rate-limiting tile requests',
-          retryAfter,
-        });
-      }
-      // An upstream 400 must not read as a malformed request to this proxy.
-      return sendJson(res, 502, { error: error.message });
-    }
-    sendJson(res, 502, { error: error?.message || 'Tile fetch failed' });
-  }
-}
-
-/**
- * Attach the Mapillary routes. Both refuse cross-site requests so another page
- * cannot spend the token or provoke a 429 that holds every tile miss.
- */
-export function installMapillaryRoutes(middlewares) {
-  const allow = makeRateLimiter({
-    windowMs: 60_000,
-    max: TILE_ROUTE_MAX_PER_MIN,
-    globalMax: TILE_ROUTE_MAX_PER_MIN * 4,
-  });
-  middlewares.use('/api/mapillary/status', sameSiteGated(handleStatus));
-  middlewares.use(
-    '/api/mapillary/tiles',
-    sameSiteGated((req, res) => handleTile(req, res, allow)),
-  );
-}

server/providers/mapillary/tiles.js

@@ -1,518 +1,265 @@
-import path from 'node:path';
-import { promises as fsp } from 'node:fs';
 import { readResponseBytesCapped } from '../common/http.js';
 import { stripTileLayers } from './trim.js';
-import {
-  MAPILLARY_TILE_HOST,
-  TILE_LAYERS,
-  TILE_DISK_DIR,
-  TILE_DISK_TTL_MS,
-  TILE_DISK_SWEEP_INTERVAL_MS,
-  TILE_DISK_MAX_BYTES,
-  TILE_DISK_FILE_OVERHEAD_BYTES,
-  TILE_MAX_BYTES,
-  TILE_MEMORY_BUDGET_BYTES,
-  TILE_MEMORY_ENTRY_OVERHEAD_BYTES,
-  TILE_FETCH_TIMEOUT_MS,
-  TILE_UPSTREAM_CONCURRENCY,
-  TILE_KEY_REJECTED_HOLD_MS,
-  TILE_RATE_LIMIT_HOLD_MS,
-  TILE_RATE_LIMIT_MAX_HOLD_MS,
-  mapillaryToken,
-} from './constants.js';
 
-/** Log without the stack; the token is never in scope here. */
-function warn(what, error) {
-  console.warn(`[Mapillary Proxy] ${what}:`, error?.message || error);
+/** Mapillary vector tiles, /{layer}/2/{z}/{x}/{y}: the only host the token goes to. */
+const TILE_HOST = 'https://tiles.mapillary.com/maps/vtp';
+/** Only the street zooms the app draws are proxied. */
+const MIN_ZOOM = 11;
+const MAX_ZOOM = 14;
+/**
+ * The z14 `image` layer is ~98% of a tile (12 MB over a dense city) and unused
+ * (positions come from the Graph API), so it is dropped in transit.
+ */
+const DROP_LAYERS = ['image'];
+/** A z14 tile with its image layer is ~12 MB; anything past this is wrong. */
+export const TILE_MAX_BYTES = 48 * 1024 * 1024;
+const FETCH_TIMEOUT_MS = 60_000;
+/** Upstream fetches at once; more wait their turn. */
+export const TILE_UPSTREAM_CONCURRENCY = 6;
+/** Tiles change only when new imagery is processed. */
+export const TILE_TTL_MS = 24 * 60 * 60 * 1000;
+/** Trimmed tiles are ~80 KB: a dozen 9-tile views, well under 100 MB. */
+export const TILE_MEMORY_MAX_TILES = 120;
+/** A trimmed tile is ~80 KB; a body this big failed to trim and is not kept. */
+export const TILE_MEMORY_MAX_TILE_BYTES = 4 * 1024 * 1024;
+/** The longest upstream Retry-After honoured, in seconds. */
+export const MAX_RETRY_AFTER_SEC = 600;
+
+/** The client token lives in the browser by design; the server adds it to tile URLs too. */
+export function mapillaryToken() {
+  return String(process.env.MAPILLARY_CLIENT_TOKEN || '').trim();
 }
 
-/** Thrown for a request this proxy refuses before contacting Mapillary. */
-export class TileRequestError extends Error {
-  constructor(message, status = 400) {
-    super(message);
-    this.name = 'TileRequestError';
-    this.status = status;
-  }
-}
-
-/** Mapillary answered with an error status; 401/403 set `keyRejected`. */
+/** Mapillary answered with an error status (or a size this proxy refuses). */
 export class TileUpstreamError extends Error {
-  constructor(status, message, { retryAfterSec = null } = {}) {
+  constructor(status, message, retryAfter = null) {
     super(message || `Mapillary tiles HTTP ${status}`);
     this.name = 'TileUpstreamError';
     this.status = status;
-    this.keyRejected = status === 401 || status === 403;
-    this.retryAfterSec = retryAfterSec;
+    /** The upstream Retry-After header, as sent. */
+    this.retryAfter = retryAfter;
   }
 }
 
-/**
- * A rejected token or rate limit that answers every tile miss until it expires
- * (or the token changes). Cached tiles are still served meanwhile.
- * @type {{status: number, until: number, token: string}|null}
- */
-let _upstreamHold = null;
-
-/** Milliseconds to wait from a Retry-After header (seconds or an HTTP date). */
-function retryAfterMs(header, now = Date.now()) {
+/** Seconds a Retry-After header asks for (an integer or an HTTP date), else null. */
+export function retryAfterSeconds(header, now = Date.now()) {
   const text = String(header ?? '').trim();
-  let ms = NaN;
-  if (/^\d+$/.test(text)) ms = Number(text) * 1000;
-  else if (text) ms = Date.parse(text) - now;
-  if (!Number.isFinite(ms) || ms <= 0) return TILE_RATE_LIMIT_HOLD_MS;
-  return Math.min(ms, TILE_RATE_LIMIT_MAX_HOLD_MS);
+  if (!text) return null;
+  const seconds = /^\d+$/.test(text)
+    ? Number(text)
+    : (Date.parse(text) - now) / 1000;
+  return Number.isFinite(seconds) && seconds > 0 ? Math.ceil(seconds) : null;
 }
 
-function heldRefusal(token) {
-  const hold = _upstreamHold;
-  if (!hold) return null;
-  const left = hold.until - Date.now();
-  if (left <= 0 || hold.token !== token) {
-    _upstreamHold = null;
+/** The wait a 429 asks for, bounded; a missing or malformed header means a minute. */
+export function retryAfterWait(header, now = Date.now()) {
+  return Math.min(retryAfterSeconds(header, now) ?? 60, MAX_RETRY_AFTER_SEC);
+}
+
+/**
+ * While Mapillary rate-limits, tile misses are refused here until the wait is
+ * over, so the rest of a view (and the fetches queued behind the first 429)
+ * do not go on asking. Scoped to the token that was refused.
+ * @type {{until: number, token: string}|null}
+ */
+let refusal = null;
+
+/** The refusal still in force for `token`, as the error to answer with. */
+function refusedNow(token) {
+  if (!refusal) return null;
+  const left = refusal.until - Date.now();
+  if (left <= 0 || refusal.token !== token) {
+    refusal = null;
     return null;
   }
-  return new TileUpstreamError(hold.status, undefined, {
-    retryAfterSec: Math.ceil(left / 1000),
-  });
+  return new TileUpstreamError(429, undefined, String(Math.ceil(left / 1000)));
 }
 
 /**
- * Validate a tile address against the layer's zoom ranges.
- * @returns {{layer:string, upstream:string, z:number, x:number, y:number, key:string}}
+ * Parse `/coverage/{z}/{x}/{y}`; null unless z is 11–14 and x/y are on the grid.
+ * @returns {{z: number, x: number, y: number, key: string}|null}
  */
-export function normalizeTileAddress({ layer, z, x, y }) {
-  const spec = Object.hasOwn(TILE_LAYERS, layer) ? TILE_LAYERS[layer] : null;
-  if (!spec) throw new TileRequestError(`Unknown tile layer: ${layer}`);
-  const zi = Number(z);
-  const xi = Number(x);
-  const yi = Number(y);
-  if (
-    ![zi, xi, yi].every((v) => Number.isInteger(v) && v >= 0) ||
-    !spec.zoomRanges.some(([min, max]) => zi >= min && zi <= max)
-  )
-    throw new TileRequestError(
-      `Tile zoom for ${layer} must be ${spec.zoomRanges
-        .map(([min, max]) => `${min}–${max}`)
-        .join(' or ')}`,
-    );
-  const n = 2 ** zi;
-  if (xi >= n || yi >= n)
-    throw new TileRequestError('Tile address out of range');
-  return {
-    layer,
-    upstream: spec.upstream,
-    dropLayers: spec.dropLayers || [],
-    z: zi,
-    x: xi,
-    y: yi,
-    key: `${layer}/${zi}/${xi}/${yi}`,
-  };
+export function parseTilePath(pathname) {
+  const match = /^\/coverage\/(\d{1,2})\/(\d{1,6})\/(\d{1,6})$/.exec(pathname);
+  if (!match) return null;
+  const [z, x, y] = match.slice(1).map(Number);
+  if (z < MIN_ZOOM || z > MAX_ZOOM || x >= 2 ** z || y >= 2 ** z) return null;
+  return { z, x, y, key: `${z}/${x}/${y}` };
 }
 
-/** @type {Map<string, {bytes: Buffer, at: number}>} insertion-ordered LRU */
-const _memory = new Map();
-let _memoryBytes = 0;
-const memoryCost = (bytes) => bytes.length + TILE_MEMORY_ENTRY_OVERHEAD_BYTES;
-/**
- * Upstream fetches shared by every request for the same tile; the last waiter
- * to leave cancels it (see `joinFlight`).
- * @type {Map<string, {controller: AbortController, promise: Promise<Buffer>, waiters: number}>}
- */
-const _inFlight = new Map();
-
-function memoryGet(key) {
-  const hit = _memory.get(key);
-  if (!hit) return null;
-  // Same life as the disk copy, so a long-running server never serves stale
-  // coverage.
-  if (Date.now() - hit.at > TILE_DISK_TTL_MS) {
-    _memory.delete(key);
-    _memoryBytes -= memoryCost(hit.bytes);
-    return null;
-  }
-  _memory.delete(key);
-  _memory.set(key, hit);
-  return hit.bytes;
-}
-
-/** `at` is when Mapillary served the bytes (a disk tile's mtime). */
-function memoryPut(key, bytes, at = Date.now()) {
-  if (memoryCost(bytes) > TILE_MEMORY_BUDGET_BYTES / 2) return;
-  const existing = _memory.get(key);
-  if (existing) _memoryBytes -= memoryCost(existing.bytes);
-  _memory.set(key, { bytes, at });
-  _memoryBytes += memoryCost(bytes);
-  while (_memoryBytes > TILE_MEMORY_BUDGET_BYTES && _memory.size) {
-    const [oldest, entry] = _memory.entries().next().value;
-    _memory.delete(oldest);
-    _memoryBytes -= memoryCost(entry.bytes);
-  }
-}
-
-/** Disk cache root; tests point it at a temporary directory. */
-let _diskDir = TILE_DISK_DIR;
-
-function diskPath({ layer, z, x, y }, root = _diskDir) {
-  return path.join(root, layer, String(z), `${x}-${y}.pbf`);
-}
-
-async function readDisk(address) {
-  const file = diskPath(address);
+function trim(bytes) {
   try {
-    const stat = await fsp.stat(file);
-    if (Date.now() - stat.mtimeMs > TILE_DISK_TTL_MS) return null;
-    return { bytes: await fsp.readFile(file), at: stat.mtimeMs };
-  } catch {
-    return null;
-  }
-}
-
-/** Unsettled background writes and sweeps; requests never wait on them. */
-const _background = new Set();
-
-function track(task) {
-  const settled = task
-    .catch(() => {})
-    .finally(() => _background.delete(settled));
-  _background.add(settled);
-  return settled;
-}
-
-/** @type {Map<string, number>} cache file to its newest write's sequence */
-const _diskWrites = new Map();
-let _diskWriteSeq = 0;
-
-function diskWritePending(address) {
-  return _diskWrites.has(diskPath(address));
-}
-
-/**
- * Write a tile in the background. `at` keeps a rewrite's original fetch time
- * so it does not extend the tile's life. An overtaken write steps aside.
- */
-function writeDisk(address, bytes, at = null) {
-  // Sweep this write's root even if the cache dir has moved since.
-  const root = _diskDir;
-  const file = diskPath(address, root);
-  const seq = ++_diskWriteSeq;
-  const tmp = `${file}.${process.pid}-${seq}.tmp`;
-  const newest = () => _diskWrites.get(file) === seq;
-  _diskWrites.set(file, seq);
-  return track(
-    fsp
-      .mkdir(path.dirname(file), { recursive: true })
-      .then(() => fsp.writeFile(tmp, bytes))
-      .then(() => at != null && fsp.utimes(tmp, new Date(), new Date(at)))
-      .then(() => {
-        if (!newest()) return fsp.rm(tmp, { force: true });
-        return fsp.rename(tmp, file).then(() => scheduleSweep(root));
-      })
-      .catch(async (error) => {
-        await fsp.rm(tmp, { force: true }).catch(() => {});
-        warn('tile cache write failed', error);
-      })
-      .finally(() => {
-        if (newest()) _diskWrites.delete(file);
-      }),
-  );
-}
-
-let _lastSweepAt = 0;
-let _sweeping = null;
-let _sweepWarned = false;
-
-function scheduleSweep(root) {
-  if (_sweeping || Date.now() - _lastSweepAt < TILE_DISK_SWEEP_INTERVAL_MS)
-    return;
-  _lastSweepAt = Date.now();
-  const sweep = sweepTileDisk({ root })
-    .catch((error) => {
-      if (_sweepWarned) return;
-      _sweepWarned = true;
-      warn('tile cache sweep failed', error);
-    })
-    .finally(() => {
-      if (_sweeping === tracked) _sweeping = null;
-    });
-  const tracked = track(sweep);
-  _sweeping = tracked;
-}
-
-/** Every file under `dir` with size and mtime; a missing dir is empty. */
-async function listCacheFiles(dir) {
-  let entries;
-  try {
-    entries = await fsp.readdir(dir, { withFileTypes: true });
+    return stripTileLayers(bytes, DROP_LAYERS);
   } catch (error) {
-    if (error?.code === 'ENOENT') return [];
-    throw error;
-  }
-  const files = [];
-  for (const entry of entries) {
-    const file = path.join(dir, entry.name);
-    if (entry.isDirectory()) files.push(...(await listCacheFiles(file)));
-    else if (entry.isFile()) {
-      const stat = await fsp.stat(file).catch(() => null);
-      if (stat) files.push({ file, size: stat.size, at: stat.mtimeMs });
-    }
-  }
-  return files;
-}
-
-/**
- * Delete expired tiles, then the oldest until the rest fit in `maxBytes`.
- * @returns {Promise<{removed: number, bytes: number}>} files deleted, charged bytes kept
- */
-export async function sweepTileDisk({
-  root = _diskDir,
-  maxBytes = TILE_DISK_MAX_BYTES,
-} = {}) {
-  const now = Date.now();
-  const files = (await listCacheFiles(root)).sort((a, b) => a.at - b.at);
-  const cost = (size) => size + TILE_DISK_FILE_OVERHEAD_BYTES;
-  let bytes = files.reduce((sum, { size }) => sum + cost(size), 0);
-  let removed = 0;
-  for (const { file, size, at } of files) {
-    if (now - at <= TILE_DISK_TTL_MS && bytes <= maxBytes) break;
-    await fsp.rm(file, { force: true });
-    bytes -= cost(size);
-    removed++;
-  }
-  return { removed, bytes };
-}
-
-/** The one origin the tile token may be sent to, at every redirect hop. */
-const TILE_ORIGIN = new URL(MAPILLARY_TILE_HOST).origin;
-export const TILE_MAX_REDIRECTS = 3;
-const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);
-
-/**
- * Follow redirects by hand so every hop stays HTTPS on the tile origin and the
- * token in the query never reaches another host. A bad redirect is a 502.
- */
-async function fetchPinned(url, init) {
-  let next = url;
-  for (let hop = 0; ; hop++) {
-    const response = await fetch(next, { ...init, redirect: 'manual' });
-    if (!REDIRECT_STATUSES.has(response.status)) return response;
-    await response.body?.cancel().catch(() => {});
-    if (hop >= TILE_MAX_REDIRECTS)
-      throw new TileUpstreamError(502, 'Mapillary tile redirected too often');
-    let target = null;
-    try {
-      target = new URL(response.headers.get('location') ?? '', next);
-    } catch {
-      /* unparseable Location */
-    }
-    if (
-      !response.headers.get('location') ||
-      target?.protocol !== 'https:' ||
-      target.origin !== TILE_ORIGIN ||
-      target.username ||
-      target.password
-    )
-      throw new TileUpstreamError(
-        502,
-        'Mapillary tile redirect left the tile origin',
-      );
-    next = target.href;
-  }
-}
-
-async function fetchUpstream(address, signal) {
-  const token = mapillaryToken();
-  if (!token) throw new TileRequestError('Mapillary token not configured', 503);
-  const held = heldRefusal(token);
-  if (held) throw held;
-  const url = `${MAPILLARY_TILE_HOST}/${address.upstream}/2/${address.z}/${address.x}/${address.y}?access_token=${encodeURIComponent(token)}`;
-  const timeout = AbortSignal.timeout(TILE_FETCH_TIMEOUT_MS);
-  const oversize = new AbortController();
-  const response = await fetchPinned(url, {
-    signal: AbortSignal.any([signal, timeout, oversize.signal].filter(Boolean)),
-    headers: { Accept: 'application/x-protobuf' },
-  });
-  if (response.status === 404 || response.status === 204) {
-    await response.body?.cancel().catch(() => {});
-    return Buffer.alloc(0);
-  }
-  if (!response.ok) {
-    await response.body?.cancel().catch(() => {});
-    const error = new TileUpstreamError(response.status);
-    if (error.keyRejected)
-      _upstreamHold = {
-        status: response.status,
-        until: Date.now() + TILE_KEY_REJECTED_HOLD_MS,
-        token,
-      };
-    else if (response.status === 429) {
-      const wait = retryAfterMs(response.headers.get('retry-after'));
-      _upstreamHold = { status: 429, until: Date.now() + wait, token };
-      error.retryAfterSec = Math.ceil(wait / 1000);
-    }
-    throw error;
-  }
-  // Chunked or compressed bodies have no Content-Length: the cap applies as
-  // the body streams, and an oversized one also aborts the request.
-  try {
-    const bytes = await readResponseBytesCapped(
-      response,
-      TILE_MAX_BYTES,
-      signal,
-    );
-    return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength);
-  } catch (error) {
-    if (error?.code !== 'RESPONSE_TOO_LARGE') throw error;
-    oversize.abort();
-    throw new TileUpstreamError(502, 'Mapillary tile exceeds size cap');
-  }
-}
-
-/** @type {{active: number, queue: Array<() => void>}} upstream slots */
-let _slots = { active: 0, queue: [] };
-
-/**
- * Resolve with a release function once a slot is free. An abort while
- * queued leaves the queue.
- * @returns {Promise<() => void>}
- */
-function acquireUpstreamSlot(signal) {
-  const slots = _slots;
-  return new Promise((resolve, reject) => {
-    if (signal.aborted) return reject(signal.reason);
-    const onAbort = () => {
-      const index = slots.queue.indexOf(grant);
-      if (index !== -1) slots.queue.splice(index, 1);
-      reject(signal.reason);
-    };
-    function grant() {
-      signal.removeEventListener('abort', onAbort);
-      slots.active++;
-      let released = false;
-      resolve(() => {
-        if (released) return;
-        released = true;
-        slots.active--;
-        slots.queue.shift()?.();
-      });
-    }
-    if (slots.active < TILE_UPSTREAM_CONCURRENCY) return grant();
-    signal.addEventListener('abort', onAbort, { once: true });
-    slots.queue.push(grant);
-  });
-}
-
-function trim(address, bytes) {
-  if (!address.dropLayers.length || !bytes.length) return bytes;
-  try {
-    return stripTileLayers(bytes, address.dropLayers);
-  } catch (error) {
-    warn('tile trim failed, serving raw', error);
+    console.warn('[Mapillary Proxy] tile trim failed:', error?.message);
     return bytes;
   }
 }
 
-/**
- * Fetch one tile from memory, disk, a shared flight, then Mapillary. Empty
- * bytes mean no coverage.
- * @param {{layer:string,z:number|string,x:number|string,y:number|string}} request
- * @param {{signal?: AbortSignal}} [options]
- * @returns {Promise<{bytes: Buffer, source: 'memory'|'disk'|'inflight'|'upstream', address: object}>}
- */
-export async function fetchTile(request, { signal } = {}) {
-  const address = normalizeTileAddress(request);
-  const memory = memoryGet(address.key);
-  if (memory) return { bytes: memory, source: 'memory', address };
-  const disk = await readDisk(address);
-  if (disk) {
-    // Older cache files may be untrimmed: rewrite once, keeping the fetch time.
-    const bytes = trim(address, disk.bytes);
-    if (bytes !== disk.bytes && !diskWritePending(address))
-      writeDisk(address, bytes, disk.at);
-    memoryPut(address.key, bytes, disk.at);
-    return { bytes, source: 'disk', address };
+/** One tile from Mapillary, trimmed; empty bytes mean no coverage. */
+async function fetchUpstream({ z, x, y }, signal) {
+  const token = mapillaryToken();
+  const refused = refusedNow(token);
+  if (refused) throw refused;
+  const response = await fetch(
+    `${TILE_HOST}/mly1_public/2/${z}/${x}/${y}?access_token=${encodeURIComponent(token)}`,
+    {
+      signal: AbortSignal.any([signal, AbortSignal.timeout(FETCH_TIMEOUT_MS)]),
+      // The token is in the query: a redirect is never followed (a 3xx is a 502).
+      redirect: 'manual',
+      headers: { Accept: 'application/x-protobuf' },
+    },
+  );
+  if (!response.ok || response.status === 204) {
+    await response.body?.cancel().catch(() => {});
+    if (response.status === 204 || response.status === 404)
+      return Buffer.alloc(0);
+    const retryAfter = response.headers.get('retry-after');
+    if (response.status === 429)
+      refusal = {
+        token,
+        until: Date.now() + retryAfterWait(retryAfter) * 1000,
+      };
+    throw new TileUpstreamError(response.status, undefined, retryAfter);
+  }
+  try {
+    const bytes = await readResponseBytesCapped(response, TILE_MAX_BYTES);
+    return trim(Buffer.from(bytes.buffer, bytes.byteOffset, bytes.length));
+  } catch (error) {
+    if (error?.code !== 'RESPONSE_TOO_LARGE') throw error;
+    throw new TileUpstreamError(502, 'Mapillary tile exceeds size cap');
   }
-  // A caller gone while the disk was read must not start a fetch nobody joins.
-  signal?.throwIfAborted();
-  let flight = _inFlight.get(address.key);
-  const joined = Boolean(flight) && !flight.controller.signal.aborted;
-  if (!joined) flight = startFlight(address);
-  const bytes = await joinFlight(flight, signal);
-  return { bytes, source: joined ? 'inflight' : 'upstream', address };
 }
 
-/** Start the shared upstream fetch; it holds a slot until the body is read. */
+let upstreamActive = 0;
+/** @type {Array<() => void>} fetches waiting for a slot, oldest first */
+const upstreamWaiting = [];
+
+/**
+ * Run `task` in an upstream slot, waiting for one unless `signal` aborts
+ * first. A freed slot passes straight to the oldest waiter, still counted, so
+ * a request arriving meanwhile cannot take it first.
+ */
+async function withUpstreamSlot(signal, task) {
+  if (upstreamActive < TILE_UPSTREAM_CONCURRENCY) upstreamActive++;
+  else
+    await new Promise((resolve, reject) => {
+      const wake = () => {
+        signal.removeEventListener('abort', leave);
+        resolve();
+      };
+      const leave = () => {
+        const at = upstreamWaiting.indexOf(wake);
+        if (at !== -1) upstreamWaiting.splice(at, 1);
+        reject(signal.reason);
+      };
+      upstreamWaiting.push(wake);
+      signal.addEventListener('abort', leave, { once: true });
+    });
+  try {
+    // Handed a slot after leaving: it goes on to the next waiter.
+    signal.throwIfAborted();
+    return await task();
+  } finally {
+    const next = upstreamWaiting.shift();
+    if (next) next();
+    else upstreamActive--;
+  }
+}
+
+/** `promise`, or a rejection as soon as `signal` aborts. */
+function unlessAborted(promise, signal) {
+  if (!signal) return promise;
+  signal.throwIfAborted();
+  return new Promise((resolve, reject) => {
+    const leave = () => reject(signal.reason);
+    signal.addEventListener('abort', leave, { once: true });
+    promise
+      .then(resolve, reject)
+      .finally(() => signal.removeEventListener('abort', leave));
+  });
+}
+
+/** @type {Map<string, {bytes: Buffer, at: number}>} least recently used first */
+const memory = new Map();
+/**
+ * Upstream fetches shared per tile, with the requests still waiting on each.
+ * @type {Map<string, {promise: Promise<Buffer>, controller: AbortController, waiters: number}>}
+ */
+const inFlight = new Map();
+
+function remember(key, bytes, at = Date.now()) {
+  if (bytes.length > TILE_MEMORY_MAX_TILE_BYTES) return;
+  memory.delete(key);
+  memory.set(key, { bytes, at });
+  for (const oldest of memory.keys()) {
+    if (memory.size <= TILE_MEMORY_MAX_TILES) break;
+    memory.delete(oldest);
+  }
+}
+
+function recall(key) {
+  const hit = memory.get(key);
+  if (!hit) return null;
+  if (Date.now() - hit.at > TILE_TTL_MS) {
+    memory.delete(key);
+    return null;
+  }
+  remember(key, hit.bytes, hit.at); // most recently used again
+  return hit.bytes;
+}
+
+/** Start the upstream fetch for a tile, shared until its last request leaves. */
 function startFlight(address) {
   const controller = new AbortController();
-  const flight = { controller, waiters: 0, promise: null };
-  flight.promise = acquireUpstreamSlot(controller.signal)
-    .then((release) =>
-      fetchUpstream(address, controller.signal).finally(release),
-    )
-    .then((raw) => trim(address, raw))
+  const flight = { controller, waiters: 0 };
+  flight.promise = withUpstreamSlot(controller.signal, () =>
+    fetchUpstream(address, controller.signal),
+  )
     .then((bytes) => {
-      memoryPut(address.key, bytes);
-      writeDisk(address, bytes);
+      remember(address.key, bytes);
       return bytes;
     })
     .finally(() => {
-      if (_inFlight.get(address.key) === flight) _inFlight.delete(address.key);
+      if (inFlight.get(address.key) === flight) inFlight.delete(address.key);
     });
-  // Every waiter may have left before it settles; nobody awaits it then.
+  // Every request may have left; the rejection is then nobody's.
   flight.promise.catch(() => {});
-  _inFlight.set(address.key, flight);
+  inFlight.set(address.key, flight);
   return flight;
 }
 
 /**
- * Wait on a shared flight. A caller's abort rejects only that caller; the
- * fetch is cancelled when the last waiter leaves.
+ * One tile from memory, a shared in-flight fetch, or Mapillary. `signal`
+ * aborts when the request's client leaves; the fetch stops once all have.
+ * @param {{z: number, x: number, y: number, key: string}} address from parseTilePath
+ * @param {{signal?: AbortSignal}} [options]
+ * @returns {Promise<{bytes: Buffer, source: 'memory'|'inflight'|'upstream'}>}
  */
-function joinFlight(flight, signal) {
-  signal?.throwIfAborted();
+export async function fetchTile(address, { signal } = {}) {
+  const cached = recall(address.key);
+  if (cached) return { bytes: cached, source: 'memory' };
+  let flight = inFlight.get(address.key);
+  const shared = Boolean(flight);
+  flight ||= startFlight(address);
   flight.waiters++;
-  return new Promise((resolve, reject) => {
-    let done = false;
-    const leave = () => {
-      if (done) return false;
-      done = true;
-      signal?.removeEventListener('abort', onAbort);
-      flight.waiters--;
-      return true;
-    };
-    const onAbort = () => {
-      if (!leave()) return;
-      if (flight.waiters === 0) flight.controller.abort();
-      reject(signal.reason);
-    };
-    signal?.addEventListener('abort', onAbort, { once: true });
-    flight.promise.then(
-      (bytes) => leave() && resolve(bytes),
-      (error) => leave() && reject(error),
-    );
-  });
+  try {
+    const bytes = await unlessAborted(flight.promise, signal);
+    return { bytes, source: shared ? 'inflight' : 'upstream' };
+  } finally {
+    if (--flight.waiters === 0) {
+      // A later request for the tile must not join a cancelled fetch.
+      if (inFlight.get(address.key) === flight) inFlight.delete(address.key);
+      flight.controller.abort();
+    }
+  }
 }
 
-/** Test seam: forget every cached tile held in memory. */
-export function _resetTileMemoryForTest() {
-  _memory.clear();
-  _memoryBytes = 0;
-  _inFlight.clear();
-  _upstreamHold = null;
-  _slots = { active: 0, queue: [] };
-}
-
-/** Test seam: how many tiles memory holds and the bytes charged for them. */
-export function _tileMemoryForTest() {
-  return { entries: _memory.size, bytes: _memoryBytes };
-}
-
-/** Test seam: wait for every background write and the sweeps they start. */
-export async function _settleTileWritesForTest() {
-  while (_background.size) await Promise.allSettled([..._background]);
-}
-
-/** Test seam: keep the disk cache in `dir` (null restores the default). */
-export function _setTileCacheDirForTest(dir) {
-  _diskDir = dir || TILE_DISK_DIR;
-  _lastSweepAt = 0;
+/** Test seam: forget every cached and in-flight tile. */
+export function _resetTileCacheForTest() {
+  memory.clear();
+  inFlight.clear();
+  refusal = null;
+  upstreamActive = 0;
+  upstreamWaiting.length = 0;
 }

server/providers/mapillary/trim.js

@@ -41,16 +41,3 @@ export function stripTileLayers(bytes, dropNames) {
   }, null);
   return dropped ? Buffer.from(writer.finish()) : bytes;
 }
-
-/** Names of the layers inside a vector tile, in order. */
-export function listTileLayers(bytes) {
-  const names = [];
-  if (!bytes?.length) return names;
-  new PbfReader(bytes).readFields((tag, _result, pbf) => {
-    if (tag !== TILE_LAYER_FIELD) return;
-    const end = pbf.readVarint() + pbf.pos;
-    names.push(layerName(bytes.subarray(pbf.pos, end)));
-    pbf.pos = end;
-  }, null);
-  return names;
-}