feat(transit): live public-transit vehicles from open GTFS-Realtime feeds (#587)

Supersedes #218 by @spcpza, ported onto the current component layout with fixes from review. Verified before merge: format, package boundaries, unit suite, build, and the browser tracking suite; a browser harness for the change where one applies; adversarial review applied. Contributor: @spcpza. His original commit is the base of this branch, and his own re-port onto main (2026-09-15, `5b69011`) was consulted while finishing the height work. That port reads every vehicle's surface from the app's shared ground-floor system (`cachedGroundFloor` / `warmGroundFloor` / `GROUND_FLOOR_LIFT_M`) rather than sampling the scene per layer, keeps a vehicle hidden until its cell answers, warms a bounded batch per poll on a timer instead of per frame, and renders at the ellipsoid above the altitude gate. That is the sounder primitive — one floor system for the app, with the rooftop-versus-street question already settled against a datum-correct DEM prior — and this branch's height work was rebuilt on it, replacing a bespoke cell cache, ground prior and rooftop rule written here. Credit for that direction is his. Future work (not in this PR), both written up in docs/CURRENT-STATE.md: - Readable route names. The realtime feeds carry only `route_id`; `route_short_name` / `route_long_name` live in each operator's static GTFS `routes.txt`, and headsigns in `trips.txt`. Doing it properly wants a server-side metadata loader beside the transit proxy with fixed registered static sources, bounded indexes built once and persisted, a slow conditional refresh, and a versioned route dictionary the browser caches. Shipped here: ids are never rewritten, and a cut one carries an ellipsis instead of reading as a typo. - Below-ground rendering. Subway vehicles are drawn on the surface directly above themselves, because the feeds carry no depth and a train nobody can see is not worth drawing; the card says so plainly. Real treatment wants OSM tunnel geometry per city and a ghosted sprite tethered to the surface. The owner recorded it and left it. Round four and after (2026-09-15). The owner tested round three under FLIR and NVG, and then asked for the design to be taken further: elegant in performance, beautiful on camera, trails with a cache for public transport, Cesium-native, and a core other moving layers can adopt. Codex Astra reviewed the plan (it confirmed the speed fault: a glide started from the drawn position at receipt time but spanned the gap between report times, so with MBTA reporting every ~10 s against a 15 s poll every second glide ran at 1.5x; it rejected a queued-segment patch), then wrote the design, then implemented it in commit-sized passes with adversarial review between them and real-browser harness runs by the integrator. What this PR now contains beyond the contributor's original port: - Delayed playback in a contact-agnostic core (`src/data/contactPlayback.js`): each vehicle keeps a typed, bounded fix history and a display clock that runs at one second per second through it, a feed-derived lag behind real time, holding at its newest fix rather than extrapolating and never sprinting to catch up. Marker, course, motion word, card and trail all read the same displayed sample. The clock has an explicit time-offset and seek API so a rewind is a later feature, not a rewrite; flights and vessels are untouched. - A selected-vehicle trail (`src/data/contactTrailRenderer.js`, `src/layers/transit/trails.js`): the retained history drawn with batched Cesium primitives, clipped at display time so the head is exactly the sprite, faded by age, in the mode colour (white under sensor presets), on the same surface path as the marker. - Bounded proxy-side history (`server/providers/transitHistory.js`): fifteen minutes and at most 128 fixes per vehicle with hard byte ceilings, MBTA only for now (opt-in per feed), served by a small endpoint that never fetches upstream, so a selected trail survives a browser reload while the server keeps running. - Scene work bounded to what is visible and moving, an occluder, cached DETECT candidates, holds on the render loop only while something visible moves. - Sensor presets: white-hot haloed sprites under NVG/FLIR/noir (black-hot inverts), saturated mode colours under CRT, per-mode bracket tiers in every theme, the selected card keeping its mode colour. - A discriminating browser harness (`scripts/qa-transit.mjs`) that samples postRender frames, reads the real card anchor, and measures fleet budgets at 800 and 3,000 vehicles. Not built, recorded as follow-ups: curated route display names (lowest priority per the owner), fleet-wide tails, route snapping, a rewind UI, adoption of the playback core by flights and vessels. Verified at merge (2026-09-15 morning): unit suite 3,896 passing, build clean, format and boundary gates clean, tracking regression 109/109, browser harness keyless 200 pass / 0 fail (13 sensor checks unexercised because the live view offered fewer than six isolated sprites; every judged sprite passed), keyed on Google 3D tiles 198 pass with the per-frame allocation gate as the only failure. Follow-ups recorded: the keyed build's playback sampler still allocates per frame (CPU and frame time stay within budget; the 3,000-vehicle metro view sits at a 27.6 ms frame interval against a 25 ms budget), the shared detection overlay's per-paint allocation, friendlier route names, fleet-wide tails, a rewind UI, and adopting the playback core for vessels and flights. Co-authored-by: spcpza <[email protected]>
2026-11-05 10:53:59 +00:00 · 2026-09-15 05:18:30 -05:00
1 parent 1ad565c8af
commit 76c20be277
@@ -0,0 +1,401 @@
+/**
+ * @module transitProxy
+ * @description Pure server-side mechanics for the `/api/transit` proxy.
+ *
+ * Kept free of Vite/Node middleware state (the terrainHeightsProxy pattern)
+ * so path resolution, snapshot shaping, and the cache/stale policy can be
+ * exercised by the offline node:test suite. The middleware in vite.config.js
+ * only does I/O: fetch the registered upstream, hand the bytes here, send.
+ */
+
+import {
+  GTFS_INCREMENTALITY_FULL_DATASET,
+  decodeVehiclePositions,
+} from './gtfsRealtime.js';
+import { getTransitFeed } from './transitFeeds.js';
+
+/** Fresh window: a snapshot younger than this is served without refetching. */
+export const TRANSIT_PROXY_TTL_MS = 15_000;
+/** Serve-stale window: after an upstream failure, a snapshot this old still ships (marked stale). */
+export const TRANSIT_PROXY_STALE_MAX_MS = 10 * 60_000;
+/** Upstream fetch timeout. National feeds (Entur ≈ 1.4 MB) need headroom. */
+export const TRANSIT_PROXY_TIMEOUT_MS = 15_000;
+/** Hard cap on upstream bytes: the largest registered feed is well under 1 MB. */
+export const TRANSIT_PROXY_MAX_BODY_BYTES = 8 * 1024 * 1024;
+/** Redirect hops followed before a feed is declared unreachable. */
+export const TRANSIT_MAX_REDIRECTS = 3;
+/**
+ * The 3xx codes that actually mean "go somewhere else". 304 is NOT one of them:
+ * it is the successful answer to a conditional request and carries no Location,
+ * so treating the whole 3xx range as a redirect turns every healthy unchanged
+ * feed into a failure and walks it up the backoff ladder.
+ */
+export const TRANSIT_REDIRECT_STATUSES = Object.freeze([
+  301, 302, 303, 307, 308,
+]);
+
+/**
+ * Whether an upstream status is a redirect this proxy should follow.
+ * @param {number} status
+ * @returns {boolean}
+ */
+export function isTransitRedirectStatus(status) {
+  return TRANSIT_REDIRECT_STATUSES.includes(Number(status));
+}
+/**
+ * Admission window for upstream requests. One browser polls a feed every 15 s
+ * (4/min); the allowance leaves room for a second tab and a manual reload, and
+ * the global figure bounds what this process can ask of ALL operators at once.
+ */
+export const TRANSIT_ADMISSION_WINDOW_MS = 60_000;
+export const TRANSIT_ADMISSION_MAX_PER_FEED = 8;
+export const TRANSIT_ADMISSION_MAX_GLOBAL = 40;
+/**
+ * Cooldown ladder after consecutive upstream failures, in ms. A cold feed that
+ * is simply down must not be re-asked once per poll for the whole session: the
+ * first retry is quick, the last rung is five minutes, and the ladder resets on
+ * the first success.
+ */
+export const TRANSIT_BACKOFF_LADDER_MS = Object.freeze([
+  5_000, 15_000, 60_000, 300_000,
+]);
+
+/**
+ * Resolve `/vehicles/<feedId>` (the path after the `/api/transit` mount) to a
+ * registered feed. Anything else — a different route, an unknown id, path
+ * tricks, a query string — resolves to null and the caller 404s.
+ * @param {string} url Request URL relative to the mount point.
+ * @returns {{ route: 'feeds' } | { route: 'vehicles', feed: object } | null}
+ */
+/** Validate the entire decoded identifier, including UTF-8 length. */
+export function validTransitIdentifier(value) {
+  return (
+    typeof value === 'string' &&
+    value.length > 0 &&
+    !/[\u0000-\u001f\u007f/\\]/u.test(value) &&
+    value !== '.' &&
+    value !== '..' &&
+    new TextEncoder().encode(value).length <= 256 &&
+    !/[\uD800-\uDFFF]/u.test(value)
+  );
+}
+export function resolveTransitRoute(url) {
+  const pathname = String(url || '').split('?')[0];
+  if (pathname === '/feeds' || pathname === '/feeds/')
+    return { route: 'feeds' };
+  const match = /^\/(vehicles|trail)\/([^/]+)(?:\/([^/]+))?\/?$/.exec(pathname);
+  if (!match) return null;
+  let id, vehicleId;
+  try {
+    id = decodeURIComponent(match[2]);
+    vehicleId = match[3] === undefined ? null : decodeURIComponent(match[3]);
+  } catch {
+    return null;
+  }
+  const feed = getTransitFeed(id);
+  if (!feed) return null;
+  if (match[1] === 'vehicles')
+    return vehicleId === null ? { route: 'vehicles', feed } : null;
+  return feed.historyRetention === true && validTransitIdentifier(vehicleId)
+    ? { route: 'trail', feed, vehicleId }
+    : null;
+}
+
+/** Selection reads are bounded and cancellable through body parsing. */
+export async function fetchTransitHistory(
+  feedId,
+  vehicleId,
+  signal,
+  fetchImpl = fetch,
+) {
+  if (!validTransitIdentifier(vehicleId))
+    throw new TypeError('Invalid transit vehicle identifier');
+  const response = await fetchImpl(
+    `/api/transit/trail/${encodeURIComponent(feedId)}/${encodeURIComponent(vehicleId)}`,
+    { signal, headers: { Accept: 'application/json' } },
+  );
+  if (!response.ok) throw new Error(`Transit history HTTP ${response.status}`);
+  const reader = response.body?.getReader?.();
+  let text = '';
+  if (reader) {
+    const decoder = new TextDecoder();
+    let bytes = 0;
+    try {
+      while (true) {
+        signal?.throwIfAborted();
+        const chunk = await reader.read();
+        if (chunk.done) break;
+        bytes += chunk.value.byteLength;
+        if (bytes > 32768)
+          throw new RangeError('Transit history exceeds response limit');
+        text += decoder.decode(chunk.value, { stream: true });
+      }
+      text += decoder.decode();
+    } finally {
+      await reader.cancel().catch(() => {});
+    }
+  } else {
+    text = await response.text();
+    if (new TextEncoder().encode(text).length > 32768)
+      throw new RangeError('Transit history exceeds response limit');
+  }
+  signal?.throwIfAborted();
+  const payload = JSON.parse(text);
+  if (
+    payload.version !== 1 ||
+    payload.feedId !== feedId ||
+    payload.vehicleId !== vehicleId ||
+    !Array.isArray(payload.fixes) ||
+    payload.fixes.length > 128 ||
+    !Array.isArray(payload.epochs) ||
+    payload.epochs.length > 8
+  )
+    throw new TypeError('Invalid transit history response');
+  let previous = -Infinity;
+  const epochs = new Set();
+  for (const epoch of payload.epochs) {
+    if (
+      !Number.isInteger(epoch.id) ||
+      epoch.id < 1 ||
+      epoch.id > 0xffffffff ||
+      epochs.has(epoch.id) ||
+      !['trip', 'route', 'mode'].every(
+        (key) =>
+          typeof epoch[key] === 'string' &&
+          new TextEncoder().encode(epoch[key]).length <= 256,
+      )
+    )
+      throw new TypeError('Invalid transit history epoch');
+    epochs.add(epoch.id);
+  }
+  for (const fix of payload.fixes) {
+    if (
+      !Array.isArray(fix) ||
+      fix.length !== 5 ||
+      !fix.every(Number.isFinite) ||
+      fix[0] < 0 ||
+      fix[0] <= previous ||
+      Math.abs(fix[1]) > 90 ||
+      Math.abs(fix[2]) > 180 ||
+      !Number.isInteger(fix[3]) ||
+      !epochs.has(fix[4])
+    )
+      throw new TypeError('Invalid transit history fix');
+    previous = fix[0];
+  }
+  return payload;
+}
+
+/**
+ * Request headers for one upstream fetch.
+ *
+ * Three of these are obligations, not politeness. Feeds that ask consumers to
+ * identify themselves get their header from the registry (Entur requires
+ * `ET-Client-Name`; OVapi asks that the User-Agent say who you are). OVapi also
+ * asks anyone polling faster than once a minute to send conditional-request
+ * validators and to accept gzip, so both are sent to every feed: harmless
+ * where it is not asked for, and it spares each operator a full body whenever
+ * nothing has changed.
+ *
+ * @param {object} feed Registry entry.
+ * @param {{etag?: string|null, lastModified?: string|null}} [validators] From the cached snapshot.
+ * @returns {Record<string, string>}
+ */
+export function transitUpstreamHeaders(feed, validators = null) {
+  return {
+    'User-Agent':
+      'gods-eye-view-transit-proxy/1.0 (+https://github.com/bilawalsidhu/gods-eye-view)',
+    Accept: 'application/x-protobuf, application/octet-stream;q=0.9, */*;q=0.1',
+    'Accept-Encoding': 'gzip',
+    ...(validators?.etag ? { 'If-None-Match': validators.etag } : {}),
+    ...(validators?.lastModified
+      ? { 'If-Modified-Since': validators.lastModified }
+      : {}),
+    ...(feed?.headers || {}),
+  };
+}
+
+/**
+ * Only https upstreams are fetched.
+ * @param {string} url Request or response URL.
+ * @returns {boolean}
+ */
+export function isAcceptableTransitUpstreamUrl(url) {
+  try {
+    return new URL(url).protocol === 'https:';
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * Decide whether one redirect hop may be followed, BEFORE it is requested.
+ *
+ * Checking the FINAL url after `redirect: 'follow'` is too late: by then the
+ * disallowed host has already been contacted, has already seen this server's
+ * request, and has already answered. So every hop is resolved here first and a
+ * hop that leaves the feed's own origin, or drops to plain http, is refused —
+ * the same rule the CCTV frame proxy applies to camera images.
+ *
+ * @param {string} originUrl The registered feed URL (defines the allowed origin).
+ * @param {string} currentUrl The URL that produced this redirect (for relative Location).
+ * @param {string|null} location Raw `Location` header value.
+ * @returns {{ ok: true, url: string } | { ok: false, reason: string }}
+ */
+export function transitRedirectDecision(originUrl, currentUrl, location) {
+  if (!location)
+    return { ok: false, reason: 'redirect without a Location header' };
+  let origin;
+  let next;
+  try {
+    origin = new URL(originUrl).origin;
+    next = new URL(location, currentUrl);
+  } catch {
+    return { ok: false, reason: 'redirect target is not a valid URL' };
+  }
+  if (next.protocol !== 'https:')
+    return { ok: false, reason: 'redirect left https' };
+  if (next.origin !== origin)
+    return { ok: false, reason: `redirect left ${origin}` };
+  return { ok: true, url: next.toString() };
+}
+
+/**
+ * Cooldown before the next upstream attempt after `failures` consecutive
+ * failures. Zero while the feed is healthy.
+ * @param {number} failures Consecutive failures (0 = none).
+ * @returns {number} ms to wait before the next upstream attempt.
+ */
+export function nextTransitBackoffMs(failures) {
+  const count = Number.isFinite(failures) ? Math.floor(failures) : 0;
+  if (count <= 0) return 0;
+  const index = Math.min(count, TRANSIT_BACKOFF_LADDER_MS.length) - 1;
+  return TRANSIT_BACKOFF_LADDER_MS[index];
+}
+
+/**
+ * Error thrown when a feed's shape is unusable rather than merely unavailable.
+ * Carries `transitReason` so the middleware can answer honestly.
+ */
+export class TransitFeedShapeError extends Error {
+  constructor(message, reason) {
+    super(message);
+    this.name = 'TransitFeedShapeError';
+    this.transitReason = reason;
+  }
+}
+
+/**
+ * Give every vehicle a best-available report time, and say WHERE it came from.
+ *
+ * A feed that omits `VehiclePosition.timestamp` is not reporting "now" — it is
+ * reporting nothing, and treating the moment WE fetched as the moment the bus
+ * was there is how a five-minute-old fix gets drawn as live. The ladder is
+ * explicit instead: the vehicle's own timestamp, else the feed header's
+ * timestamp (the operator's own statement about the snapshot), else fetch time,
+ * which is labelled as a guess so the layer can age it conservatively.
+ *
+ * @param {object[]} vehicles Normalized vehicle records.
+ * @param {number|null} headerTimestamp Feed header timestamp (epoch seconds).
+ * @param {number} fetchedAtS Fetch time (epoch seconds).
+ * @returns {object[]} Records with `timestamp` and `timestampSource` set.
+ */
+export function repairVehicleTimestamps(vehicles, headerTimestamp, fetchedAtS) {
+  const header =
+    Number.isFinite(headerTimestamp) && headerTimestamp > 0
+      ? headerTimestamp
+      : null;
+  return (vehicles || []).map((vehicle) => {
+    if (Number.isFinite(vehicle.timestamp) && vehicle.timestamp > 0) {
+      return { ...vehicle, timestampSource: 'vehicle' };
+    }
+    if (header !== null)
+      return { ...vehicle, timestamp: header, timestampSource: 'header' };
+    return { ...vehicle, timestamp: fetchedAtS, timestampSource: 'fetch' };
+  });
+}
+
+/**
+ * Decode upstream bytes into the JSON snapshot the browser consumes.
+ *
+ * A DIFFERENTIAL feed is refused outright. Differential GTFS-Realtime carries
+ * only what changed, so reading one as if it were a full snapshot would make
+ * every unmentioned vehicle look like it had vanished. Deletion semantics are
+ * not implemented, so the honest answer is to decline the feed, not to render
+ * a wrong one.
+ *
+ * @param {object} feed Registry entry.
+ * @param {Uint8Array|ArrayBuffer} bytes Raw GTFS-RT FeedMessage.
+ * @param {number} [now=Date.now()] Fetch time (ms epoch).
+ * @returns {{ feedId: string, name: string, fetchedAt: number, feedTimestamp: number|null,
+ *   version: string|null, entityCount: number, truncated: boolean, count: number, vehicles: object[] }}
+ */
+export function buildTransitSnapshot(feed, bytes, now = Date.now()) {
+  const decoded = decodeVehiclePositions(bytes);
+  if (decoded.incrementality !== GTFS_INCREMENTALITY_FULL_DATASET) {
+    throw new TransitFeedShapeError(
+      `feed is differential (incrementality ${decoded.incrementality})`,
+      'differential',
+    );
+  }
+  return {
+    feedId: feed.id,
+    name: feed.name,
+    fetchedAt: now,
+    feedTimestamp: decoded.timestamp,
+    version: decoded.version,
+    entityCount: decoded.entityCount,
+    truncated: decoded.truncated,
+    count: decoded.vehicles.length,
+    vehicles: repairVehicleTimestamps(
+      decoded.vehicles,
+      decoded.timestamp,
+      Math.floor(now / 1000),
+    ),
+  };
+}
+
+/**
+ * Classify a cache entry for the request policy.
+ * @param {{ at: number }|null|undefined} entry Cached snapshot (`at` = fetch ms).
+ * @param {number} now
+ * @returns {'none'|'fresh'|'stale'|'expired'}
+ */
+export function transitCacheState(entry, now) {
+  if (!entry || !Number.isFinite(entry.at)) return 'none';
+  const age = now - entry.at;
+  if (age < 0) return 'fresh';
+  if (age < TRANSIT_PROXY_TTL_MS) return 'fresh';
+  if (age < TRANSIT_PROXY_STALE_MAX_MS) return 'stale';
+  return 'expired';
+}
+
+/**
+ * Response headers for a snapshot. `X-GEV-Cache` mirrors the other proxies
+ * (HIT / MISS / INFLIGHT / STALE-ERROR) so the layer can surface staleness.
+ * @param {'HIT'|'MISS'|'INFLIGHT'|'STALE-ERROR'} cacheState
+ * @param {string} [upstreamHost]
+ * @returns {Record<string, string>}
+ */
+export function transitResponseHeaders(
+  cacheState,
+  upstreamHost = '',
+  contactedAt = null,
+) {
+  return {
+    'Content-Type': 'application/json; charset=utf-8',
+    'Cache-Control':
+      cacheState === 'STALE-ERROR'
+        ? 'no-store'
+        : `public, max-age=${Math.floor(TRANSIT_PROXY_TTL_MS / 1000)}`,
+    'X-GEV-Cache': cacheState,
+    ...(upstreamHost ? { 'X-Transit-Upstream': upstreamHost } : {}),
+    // When the operator last ANSWERED, which is not when the body was fetched.
+    // A feed whose file has not changed answers 304 forever, and the body we
+    // keep serving carries its original fetch time; without this the browser
+    // would read a healthy revalidated feed as one that had gone silent.
+    ...(Number.isFinite(contactedAt)
+      ? { 'X-Transit-Contact': String(Math.floor(contactedAt)) }
+      : {}),
+  };
+}