Merge branch 'main' into feat/lithuania-cctv

server/providers/cctv/catalog.js

@@ -3,7 +3,7 @@ import path from 'node:path';
 import { DEFAULT_CCTV_SOURCE_FILE, CCTV_SOURCE_CACHE_MS } from './constants.js';
 import { allocateSourceCap, resolveCatalogCap } from './cap.js';
 import { loadGroundHeights, joinGroundHeights } from './groundHeights.js';
-import { normalizeSourceItem } from './normalize.js';
+import { cameraRegion, normalizeSourceItem } from './normalize.js';
 import { loadLithuaniaSources } from './lithuania.js';
 import {
   loadAustinSourcesFromOpenData,
@@ -18,24 +18,25 @@ import {
   loadWarendorfSourcesFromCatalog,
   loadNswSourcesFromOpenData,
   loadCalgarySourcesFromOpenData,
+  loadDelDOTSourcesFromOpenData,
+  loadVegvesenSourcesFromOpenData,
+  loadNycSourcesFromOpenData,
 } from './sources.js';
 
 /** Env kill switch: unset or anything but "0" means enabled. */
 const envEnabled = (name) => String(process.env[name] || '1').trim() !== '0';
+/** Env opt-in: only "1" enables. For packs whose feed access needs an
+ * operator agreement, so an unset flag never reaches the upstream. */
+const envOptIn = (name) => String(process.env[name] || '').trim() === '1';
 
 /**
  * Live open-data packs, in merge order. Adding a region is one entry here
  * plus its loader in sources.js; the catalog cap is shared across entries
  * round-robin (cap.js), so a new pack never silently evicts an older one.
  * Each pack fails independently (allSettled) and is gated by its own env
- * kill switch.
+ * kill switch, or by an opt-in flag where access needs an agreement.
  */
 const LIVE_PACKS = [
-  {
-    name: 'lithuania',
-    enabled: () => envEnabled('CCTV_LITHUANIA_ENABLED'),
-    load: loadLithuaniaSources,
-  },
   { name: 'austin', enabled: () => true, load: loadAustinSourcesFromOpenData },
   {
     name: 'caltrans',
@@ -92,6 +93,28 @@ const LIVE_PACKS = [
     enabled: () => envEnabled('CCTV_CALGARY_ENABLED'),
     load: loadCalgarySourcesFromOpenData,
   },
+  {
+    name: 'deldot',
+    enabled: () => envEnabled('CCTV_DELDOT_ENABLED'),
+    load: loadDelDOTSourcesFromOpenData,
+  },
+  {
+    name: 'vegvesen',
+    enabled: () => envEnabled('CCTV_VEGVESEN_ENABLED'),
+    load: loadVegvesenSourcesFromOpenData,
+  },
+  {
+    // NYC DOT requires a data-sharing agreement for this feed, so the pack is
+    // off unless the operator opts in (see DATA_SOURCES.md).
+    name: 'nyc',
+    enabled: () => envOptIn('CCTV_NYC_ENABLED'),
+    load: loadNycSourcesFromOpenData,
+  },
+  {
+    name: 'lithuania',
+    enabled: () => envEnabled('CCTV_LITHUANIA_ENABLED'),
+    load: loadLithuaniaSources,
+  },
 ];
 /**
  * Load CCTV sources from a local JSON file (CCTV_SOURCES_FILE env or default).
@@ -143,6 +166,9 @@ 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 = [];
 
   /**
    * Assemble and cache the merged CCTV source list.
@@ -209,8 +235,12 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
       name,
       sources: items
         .filter((item) => item && typeof item === 'object')
-        .map((item) => normalizeSourceItem(item))
+        .map((item) => ({ ...normalizeSourceItem(item), pack: name }))
         .filter((item) => item.id),
+      available: Number.isFinite(items.available)
+        ? items.available
+        : items.length,
+      region: items.region ?? null,
     });
     const packs = [
       ...LIVE_PACKS.map((pack, index) =>
@@ -234,6 +264,16 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
       loadGroundHeights(sourceRoot),
     );
     const trimmed = allocation.packs.filter((pack) => pack.kept < pack.offered);
+    // Each pack's own nearest-first cap and the shared catalog cap both trim.
+    const trimmedPacks = packs
+      .map((pack, index) => ({
+        pack: pack.name,
+        available: Math.max(pack.available, allocation.packs[index].offered),
+        served: allocation.packs[index].kept,
+        // Where the pack's cameras are, including those not served.
+        region: pack.region ?? cameraRegion(pack.sources),
+      }))
+      .filter((pack) => pack.served < pack.available);
     if (trimmed.length) {
       const detail = trimmed
         .map((pack) => `${pack.name} ${pack.kept}/${pack.offered}`)
@@ -244,6 +284,7 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
     }
     if (capped.length > 0 || _cctvSourceCache.length === 0) {
       _cctvSourceCache = capped;
+      _cctvTrimmedPacks = 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
@@ -257,5 +298,7 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
     return _cctvSourceCache;
   }
 
+  /** Packs serving fewer cameras than they offer, as of the last refresh. */
+  getCctvSources.trimmedPacks = () => _cctvTrimmedPacks;
   return getCctvSources;
 }

server/providers/cctv/constants.js

@@ -236,6 +236,75 @@ export const CALGARY_DOWNTOWN = { lat: 51.0461, lon: -114.0626 };
  * body cannot be buffered without limit. */
 export const CALGARY_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
 
+/** Statens vegvesen (Norway) road cameras: the keyless OGC API Features view
+ * of the official DATEX 3.1 CCTV site table, one GeoJSON request for the whole
+ * country (~900 cameras). The raw DATEX node needs an account; this view does
+ * not. Frames are stills on the agency's camera host. */
+export const DEFAULT_VEGVESEN_CCTV_URL =
+  'https://ogckart-sn1.atlas.vegvesen.no/ogc/features/v1/collections/datex_3_1:CctvSimple/items?f=application/geo%2Bjson&limit=5000';
+/** The only origin Vegvesen camera frames may come from. */
+export const VEGVESEN_IMAGE_ORIGIN =
+  'https://kamera.atlas.vegvesen.no/api/images/';
+/** Live HLS for the ~140 cameras that publish video; only this exact
+ * per-camera manifest path is registered. */
+export const VEGVESEN_VIDEO_URL = (cameraId) =>
+  `https://kamera.vegvesen.no/public/${cameraId}/manifest.m3u8`;
+/** Above the ~850 working cameras so a default install keeps them all, while
+ * staying close to them: the per-pack defaults are meant to sum to no more
+ * than DEFAULT_CCTV_MAX_SOURCES. */
+export const DEFAULT_VEGVESEN_MAX_SOURCES = 900;
+/** Hard ceiling on the catalog body. The whole country is ~800 KB. */
+export const VEGVESEN_MAX_CATALOG_BYTES = 8 * 1024 * 1024;
+/** Prioritization anchors: the largest cities from south to north, so a cap
+ * keeps national coverage rather than just the capital region. */
+export const NORWAY_ANCHORS = [
+  { lat: 59.9139, lon: 10.7522 }, // Oslo
+  { lat: 60.3913, lon: 5.3221 }, // Bergen
+  { lat: 58.97, lon: 5.7331 }, // Stavanger
+  { lat: 63.4305, lon: 10.3951 }, // Trondheim
+  { lat: 58.1467, lon: 7.9956 }, // Kristiansand
+  { lat: 67.2804, lon: 14.4049 }, // Bodø
+  { lat: 69.6492, lon: 18.9553 }, // Tromsø
+];
+
+/** 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. */
+export const NYC_CAMERAS_URL = 'https://webcams.nyctmc.org/api/cameras';
+/** The only origin NYC camera frames may come from. The catalog's `imageUrl`
+ * is ignored; the registered path is always this origin plus a validated
+ * UUID plus `/image`. */
+export const NYC_IMAGE_ORIGIN = 'https://webcams.nyctmc.org/api/cameras/';
+export const NYC_IMAGE_URL = (id) => `${NYC_IMAGE_ORIGIN}${id}/image`;
+export const NYC_CAMERA_ID_RE =
+  /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+export const DEFAULT_NYC_MAX_SOURCES = 250;
+/** Hard ceiling on the NYC catalog body. The live list is ~980 rows / ~250 KB;
+ * this only exists so an upstream that streams an unbounded body cannot be
+ * buffered without limit. */
+export const NYC_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
+/** Ground-elevation prior in metres. Matches the New York CITY_POIS prior;
+ * the client's one-shot ground snap corrects it wherever 3D tiles load. */
+export const NYC_GROUND_ELEVATION_M = 10;
+/** Prioritization anchors: one per borough so a cap keeps citywide coverage
+ * rather than Midtown alone. */
+export const NYC_ANCHORS = [
+  { lat: 40.758, lon: -73.9855 }, // Manhattan (Times Square)
+  { lat: 40.7128, lon: -74.006 }, // Lower Manhattan
+  { lat: 40.6782, lon: -73.9442 }, // Brooklyn
+  { lat: 40.7282, lon: -73.7949 }, // Queens
+  { lat: 40.8448, lon: -73.8648 }, // Bronx
+  { lat: 40.5795, lon: -74.1502 }, // Staten Island
+];
+
+/** DelDOT CCTV: one keyless statewide JSON catalog; live video via RTMP-over-HTTP (rtmpt:80). */
+export const DELDOT_CCTV_URL = 'https://tmc.deldot.gov/json/videocamera.json';
+export const DEFAULT_DELDOT_MAX_SOURCES = 300;
+export const DELDOT_ANCHORS = [
+  { lat: 39.7459, lon: -75.5466 }, // Wilmington (New Castle)
+  { lat: 39.1582, lon: -75.5244 }, // Dover (Kent)
+  { lat: 38.6906, lon: -75.3877 }, // Georgetown (Sussex)
+];
 /** Camera CATALOGS change rarely; 15 min keeps multi-megabyte upstream list refetches (Austin rows.json + 4 Caltrans districts + TfL + Ontario 511) infrequent. Frames are fetched per-request and are unaffected. */
 export const CCTV_SOURCE_CACHE_MS = 15 * 60 * 1000;
 /** Per-provider catalog-fetch timeout. Bounds the worst-case refresh so one
@@ -253,5 +322,10 @@ export const CCTV_FRAME_MAX_BODY_BYTES = 16 * 1024 * 1024;
 
 /** Deadline for upstream response headers; live bodies keep streaming afterward. */
 export const CCTV_MEDIA_FETCH_TIMEOUT_MS = 15 * 1000;
+/** Silence a live body may carry before the relay gives up on it. Twice the
+ * header deadline, because a camera that is merely slow between frames is far
+ * more common than one that has died mid-stream, and a viewer would rather
+ * wait than be dropped. */
+export const CCTV_MEDIA_IDLE_TIMEOUT_MS = 30 * 1000;
 /** Declared size ceiling for fixed media responses. */
 export const CCTV_MEDIA_MAX_BODY_BYTES = 64 * 1024 * 1024;

server/providers/cctv/media.js

@@ -4,6 +4,7 @@ import {
   CCTV_FRAME_FETCH_TIMEOUT_MS,
   CCTV_FRAME_MAX_BODY_BYTES,
   CCTV_MEDIA_FETCH_TIMEOUT_MS,
+  CCTV_MEDIA_IDLE_TIMEOUT_MS,
   CCTV_MEDIA_MAX_BODY_BYTES,
   NSW_IMAGE_ORIGIN,
   NSW_IMAGE_USER_AGENT,
@@ -103,11 +104,17 @@ export function toReadable(body) {
  * @param {Response} upstream - fetch() Response object.
  * @param {object} [opts]
  * @param {string} [opts.sourceHeader='upstream'] - Value for X-CCTV-Source header.
+ * @param {number} [opts.idleTimeoutMs=CCTV_MEDIA_IDLE_TIMEOUT_MS] - Silence
+ *   allowed between upstream chunks before the stream is released. Injectable
+ *   only to keep the deadline unit-testable.
  */
 export async function proxyMediaResponse(
   res,
   upstream,
-  { sourceHeader = 'upstream' } = {},
+  {
+    sourceHeader = 'upstream',
+    idleTimeoutMs = CCTV_MEDIA_IDLE_TIMEOUT_MS,
+  } = {},
 ) {
   const contentType =
     upstream.headers.get('content-type') || 'application/octet-stream';
@@ -153,7 +160,20 @@ export async function proxyMediaResponse(
     return;
   }
 
+  // The header deadline only covers the wait for a response line. Past that an
+  // upstream can hold the connection open and send nothing at all, and the
+  // relay would wait on it for as long as the camera host cared to. The
+  // deadline below measures the gap between upstream chunks rather than the
+  // life of the stream, so a feed that keeps delivering keeps its connection.
+  let idleTimer = null;
+  const clearIdleDeadline = () => {
+    if (!idleTimer) return;
+    clearTimeout(idleTimer);
+    idleTimer = null;
+  };
+
   stream.on('error', () => {
+    clearIdleDeadline();
     if (!res.writableEnded) res.end();
   });
 
@@ -164,6 +184,7 @@ export async function proxyMediaResponse(
   const releaseUpstream = () => {
     if (released) return;
     released = true;
+    clearIdleDeadline();
     stream.unpipe(res);
     // Destroying the Node stream cancels the web body it wraps; the direct
     // cancel covers a body that was never wrapped, and rejects harmlessly when
@@ -176,14 +197,37 @@ export async function proxyMediaResponse(
       /* already closed */
     }
   };
+  const armIdleDeadline = () => {
+    clearIdleDeadline();
+    idleTimer = setTimeout(onIdleDeadline, idleTimeoutMs);
+    idleTimer.unref?.();
+  };
+  const onIdleDeadline = () => {
+    // A viewer who cannot keep up pauses the pipe, and no upstream bytes arrive
+    // while it is paused. That is a slow client rather than a dead camera, so
+    // it gets the deadline again instead of a teardown.
+    if (res.writableNeedDrain) {
+      armIdleDeadline();
+      return;
+    }
+    releaseUpstream();
+    if (!res.writableEnded) res.end();
+  };
   res.once('close', () => {
+    clearIdleDeadline();
     if (!res.writableEnded) releaseUpstream();
   });
   res.once('error', releaseUpstream);
   stream.once('end', () => {
+    clearIdleDeadline();
     released = true;
   });
+  armIdleDeadline();
   stream.pipe(res);
+  // Attached after the pipe because a data listener resumes the stream, and
+  // flowing before the destination is attached would spill chunks nobody
+  // forwards. Only bytes from upstream renew the deadline.
+  stream.on('data', armIdleDeadline);
 }
 
 /**

server/providers/cctv/normalize.js

@@ -367,6 +367,28 @@ export function isLikelyCalgaryCoordinate(lat, lon) {
   );
 }
 
+/** New York City's five boroughs, with slack for the harbor and city line. */
+export function isLikelyNycCoordinate(lat, lon) {
+  return (
+    isPlausibleLatLon(lat, lon) &&
+    lat >= 40.45 &&
+    lat <= 40.95 &&
+    lon >= -74.3 &&
+    lon <= -73.65
+  );
+}
+
+/** Mainland Norway, with slack for the coast and the Finnmark border. */
+export function isLikelyNorwayCoordinate(lat, lon) {
+  return (
+    isPlausibleLatLon(lat, lon) &&
+    lat >= 57.9 &&
+    lat <= 71.3 &&
+    lon >= 4.4 &&
+    lon <= 31.3
+  );
+}
+
 export function isLikelyFinlandCoordinate(lat, lon) {
   if (!Number.isFinite(lat) || !Number.isFinite(lon)) return false;
   return lat >= 59.5 && lat <= 70.5 && lon >= 19 && lon <= 32;
@@ -470,7 +492,31 @@ export function prioritizeSources(cameras, maxCount, anchors) {
     return a.idx - b.idx;
   });
 
-  return scored.slice(0, cap).map((entry) => entry.camera);
+  const kept = scored.slice(0, cap).map((entry) => entry.camera);
+  // How many cameras the pack offered before trimming, and where, for
+  // coverage reports.
+  Object.defineProperty(kept, 'available', { value: list.length });
+  Object.defineProperty(kept, 'region', { value: cameraRegion(list) });
+  return kept;
+}
+
+/** The bounding box of cameras with coordinates, or null. */
+export function cameraRegion(cameras) {
+  let region = null;
+  for (const camera of cameras || []) {
+    const lat = Number(camera?.lat);
+    const lon = Number(camera?.lon);
+    if (!Number.isFinite(lat) || !Number.isFinite(lon)) continue;
+    region = region
+      ? {
+          west: Math.min(region.west, lon),
+          south: Math.min(region.south, lat),
+          east: Math.max(region.east, lon),
+          north: Math.max(region.north, lat),
+        }
+      : { west: lon, south: lat, east: lon, north: lat };
+  }
+  return region;
 }
 
 /**

server/providers/cctv/sources.js

@@ -56,6 +56,22 @@ import {
   CALGARY_DOWNTOWN,
   CALGARY_MAX_CATALOG_BYTES,
   CCTV_SOURCE_FETCH_TIMEOUT_MS,
+  DELDOT_CCTV_URL,
+  DEFAULT_DELDOT_MAX_SOURCES,
+  DELDOT_ANCHORS,
+  DEFAULT_VEGVESEN_CCTV_URL,
+  VEGVESEN_IMAGE_ORIGIN,
+  VEGVESEN_VIDEO_URL,
+  DEFAULT_VEGVESEN_MAX_SOURCES,
+  VEGVESEN_MAX_CATALOG_BYTES,
+  NORWAY_ANCHORS,
+  NYC_CAMERAS_URL,
+  NYC_IMAGE_URL,
+  NYC_CAMERA_ID_RE,
+  DEFAULT_NYC_MAX_SOURCES,
+  NYC_MAX_CATALOG_BYTES,
+  NYC_GROUND_ELEVATION_M,
+  NYC_ANCHORS,
 } from './constants.js';
 import {
   toFiniteNumber,
@@ -73,6 +89,8 @@ import {
   isLikelyTexasCoordinate,
   isLikelyNswCoordinate,
   isLikelyCalgaryCoordinate,
+  isLikelyNycCoordinate,
+  isLikelyNorwayCoordinate,
   cameraDisplayCode,
   rowArrayToObject,
   prioritizeSources,
@@ -1594,3 +1612,407 @@ export async function loadCalgarySourcesFromOpenData() {
     return [];
   }
 }
+
+export async function loadDelDOTSourcesFromOpenData() {
+  try {
+    const resp = await fetch(DELDOT_CCTV_URL, {
+      headers: { Accept: 'application/json' },
+      signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
+      redirect: 'error',
+    });
+    if (!resp.ok) {
+      console.warn('[CCTV] DelDOT source download failed:', resp.status);
+      return [];
+    }
+    const payload = await readResponseJsonCapped(resp, 2 * 1024 * 1024);
+    const rows = Array.isArray(payload?.videoCameras)
+      ? payload.videoCameras
+      : [];
+    if (!rows.length) return [];
+
+    const cameras = [];
+    for (const row of rows) {
+      if (String(row?.status).toLowerCase() !== 'active') continue;
+      const lat = toFiniteNumber(row?.lat);
+      const lon = toFiniteNumber(row?.lon);
+      if (!Number.isFinite(lat) || !Number.isFinite(lon)) continue;
+      // Delaware bounding box — a bad upstream coord can't place a camera out of state.
+      if (lat < 38.4 || lat > 39.9 || lon < -75.85 || lon > -75.0) continue;
+
+      // Prefer the catalog's HTTPS HLS transport; no subprocess is required.
+      let stream;
+      try {
+        stream = new URL(String(row?.urls?.m3u8s || ''));
+      } catch {
+        continue;
+      }
+      if (
+        stream.origin !== 'https://video.deldot.gov' ||
+        stream.username ||
+        stream.password ||
+        !/^\/live\/[A-Za-z0-9_.-]+\/playlist\.m3u8$/.test(stream.pathname)
+      )
+        continue;
+      const streamUrl = stream.href;
+
+      const id = String(row?.id || '').trim();
+      if (!/^[A-Za-z0-9_-]{1,80}$/.test(id)) continue;
+      const cameraId = `deldot-${id.toLowerCase()}`;
+      const title = String(row?.title || '').trim();
+
+      // Title carries an unambiguous travel token for ~1 in 5 cameras
+      // ("US 13 SB @ ..."). directionToHeading (allowBare=false) matches only
+      // NB/SB/EB/WB, never a bare "W"/"N" street prefix — "W NORTH ST" yields
+      // no heading and falls back.
+      const heading = directionToHeading(title);
+      const hasHeading = Number.isFinite(heading);
+
+      cameras.push({
+        id: cameraId,
+        name: title || `DelDOT ${id}`,
+        city: String(row?.county ? `${row.county} County` : 'Delaware'),
+        cityId: `deldot-${String(row?.county || 'de')
+          .toLowerCase()
+          .replace(/\s+/g, '-')}`,
+        provider: 'DelDOT',
+        lat,
+        lon,
+        headingDeg: hasHeading ? heading : fallbackHeadingFromId(cameraId),
+        headingConfidence: hasHeading ? 'high' : 'low',
+        // Fabricated RAW PRIOR poses (same personalities as Austin/Caltrans);
+        // the client ground-snap + manual calibration own the truth.
+        pitchDeg: hasHeading ? -24 : -18,
+        fovDeg: hasHeading ? 56 : 44,
+        rangeM: hasHeading ? 210 : 145,
+        mountHeightM: hasHeading ? 10 : 8,
+        groundElevationM: 10, // Estimated prior; client ground resolution owns placement.
+        feedType: 'hls',
+        url: streamUrl,
+        snapshotUrl: '',
+        sourceKind: 'deldot-open-data',
+        license: 'Public DelDOT traffic camera',
+      });
+    }
+
+    const unique = Array.from(
+      new Map(cameras.map((camera) => [camera.id, camera])).values(),
+    );
+    const maxRaw = Number(
+      process.env.CCTV_DELDOT_MAX_SOURCES || DEFAULT_DELDOT_MAX_SOURCES,
+    );
+    const maxCount = Number.isFinite(maxRaw)
+      ? Math.max(8, Math.min(400, Math.floor(maxRaw)))
+      : DEFAULT_DELDOT_MAX_SOURCES;
+    const prioritized = prioritizeSources(unique, maxCount, DELDOT_ANCHORS);
+    console.log(
+      `[CCTV] Loaded DelDOT camera sources: ${unique.length} Active (using nearest ${prioritized.length})`,
+    );
+    return prioritized;
+  } catch (error) {
+    console.warn(
+      '[CCTV] DelDOT source download error:',
+      error?.message || error,
+    );
+    return [];
+  }
+}
+
+/**
+ * One Statens vegvesen `CctvSimple` feature -> one catalog source, or null.
+ *
+ * `orientationDescription` names the place the camera looks towards
+ * ("Svelgen"), not a bearing, so it goes into the label and the heading uses
+ * the shared id-hash fallback at low confidence, like Calgary and Fintraffic.
+ * Cameras the feed reports as faulty are dropped. Cameras that publish HLS
+ * get live video, with the still as the snapshot fallback.
+ *
+ * @param {object} feature - GeoJSON feature from the OGC items response.
+ * @returns {?object}
+ */
+export function vegvesenCameraToSource(feature) {
+  if (!feature || typeof feature !== 'object') return null;
+  const props = feature.properties;
+  if (!props || typeof props !== 'object') return null;
+  const availability = String(
+    props['status.stillImageAvailability'] ?? '',
+  ).trim();
+  if (availability && availability !== 'videoOrImagesAvailable') return null;
+
+  const coordinates = feature?.geometry?.coordinates;
+  if (!Array.isArray(coordinates) || coordinates.length < 2) return null;
+  const lon = toFiniteNumber(coordinates[0]);
+  const lat = toFiniteNumber(coordinates[1]);
+  if (!isLikelyNorwayCoordinate(lat, lon)) return null;
+
+  const rawId = String(props.cameraId ?? '').trim();
+  if (!/^[A-Za-z0-9_-]{1,40}$/.test(rawId)) return null;
+  let image;
+  try {
+    image = new URL(String(props.stillImageUrl ?? ''));
+  } catch {
+    return null;
+  }
+  if (image.username || image.password) return null;
+  const imageUrl = image.href;
+  if (imageUrl !== `${VEGVESEN_IMAGE_ORIGIN}${rawId}`) return null;
+
+  const cameraId = `no-vegvesen-${rawId.toLowerCase()}`;
+  const place = String(props.description ?? '').trim() || `Kamera ${rawId}`;
+  const towards = String(props.orientationDescription ?? '').trim();
+  const road = String(props.roadNumber ?? '').trim();
+  const name = towards && towards !== place ? `${place} → ${towards}` : place;
+  // Live video when the camera publishes HLS on the agency's own host; the
+  // still stays the snapshot fallback. CCTV_VEGVESEN_VIDEO=0 keeps stills only.
+  const videoUrl =
+    String(process.env.CCTV_VEGVESEN_VIDEO || '1').trim() !== '0' &&
+    Number(props.videoServiceLevel) > 0 &&
+    String(props.videoEncodingStandard ?? '').toLowerCase() === 'hls' &&
+    String(props.videoUrl ?? '').trim() === VEGVESEN_VIDEO_URL(rawId)
+      ? VEGVESEN_VIDEO_URL(rawId)
+      : '';
+
+  return {
+    id: cameraId,
+    name: road ? `${road} ${name}` : name,
+    // One country-wide category in the camera picker, like Finland; the
+    // place itself is already in the name.
+    city: 'Norway',
+    cityId: 'norway',
+    provider: 'Statens vegvesen',
+    lat,
+    lon,
+    headingDeg: fallbackHeadingFromId(cameraId),
+    headingConfidence: 'low',
+    pitchDeg: -18,
+    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.
+    groundElevationM: 150,
+    feedType: videoUrl ? 'hls' : 'image',
+    url: videoUrl || imageUrl,
+    snapshotUrl: imageUrl,
+    sourceKind: 'vegvesen-datex',
+    license:
+      'Contains data under the Norwegian licence for Open Government data (NLOD) distributed by Statens vegvesen',
+    code: cameraDisplayCode(place.toUpperCase()),
+  };
+}
+
+/**
+ * Fetch Statens vegvesen (Norway) road cameras from the keyless OGC API
+ * Features view of the DATEX 3.1 CCTV table. Frames are stills on
+ * kamera.atlas.vegvesen.no.
+ *
+ * @returns {Promise<Array<object>>} Normalized camera source objects.
+ */
+export async function loadVegvesenSourcesFromOpenData() {
+  try {
+    const endpoint = process.env.CCTV_VEGVESEN_URL || DEFAULT_VEGVESEN_CCTV_URL;
+    const resp = await fetch(endpoint, {
+      headers: { Accept: 'application/geo+json,application/json' },
+      redirect: 'manual',
+      signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
+    });
+    const discard = async () => {
+      try {
+        await resp.body?.cancel();
+      } catch {
+        /* no-op */
+      }
+      return [];
+    };
+    if (resp.status >= 300 && resp.status < 400) {
+      console.warn(
+        '[CCTV] Vegvesen catalog redirected; redirects are not followed',
+      );
+      return discard();
+    }
+    if (!resp.ok) {
+      console.warn('[CCTV] Vegvesen camera download failed:', resp.status);
+      return discard();
+    }
+    const payload = await readResponseJsonCapped(
+      resp,
+      VEGVESEN_MAX_CATALOG_BYTES,
+    );
+    const features = Array.isArray(payload?.features) ? payload.features : [];
+    const cameras = [];
+    const seen = new Set();
+    for (const feature of features) {
+      const camera = vegvesenCameraToSource(feature);
+      if (!camera || seen.has(camera.id)) continue;
+      seen.add(camera.id);
+      cameras.push(camera);
+    }
+    const maxRaw = Number(
+      process.env.CCTV_VEGVESEN_MAX_SOURCES || DEFAULT_VEGVESEN_MAX_SOURCES,
+    );
+    const maxCount = Number.isFinite(maxRaw)
+      ? Math.max(8, Math.min(1000, Math.floor(maxRaw)))
+      : DEFAULT_VEGVESEN_MAX_SOURCES;
+    const prioritized = prioritizeSources(cameras, maxCount, NORWAY_ANCHORS);
+    console.log(
+      `[CCTV] Loaded Vegvesen camera sources: ${cameras.length} (using nearest ${prioritized.length})`,
+    );
+    return prioritized;
+  } catch (error) {
+    console.warn(
+      '[CCTV] Vegvesen camera download error:',
+      error?.message || error,
+    );
+    return [];
+  }
+}
+
+const NYC_BOROUGHS = new Set([
+  'Manhattan',
+  'Brooklyn',
+  'Queens',
+  'Bronx',
+  'Staten Island',
+]);
+
+/** NYC TMC publishes `isOnline` as the strings "true"/"false". */
+export function nycCameraIsOnline(value) {
+  return (
+    value === true ||
+    String(value ?? '')
+      .trim()
+      .toLowerCase() === 'true'
+  );
+}
+
+/**
+ * Label for one NYC camera: the intersection an operator recognises
+ * ("Central Park West @ 86 St"). The catalog's `area` is the borough, not
+ * a facing, and is not part of the name.
+ *
+ * @param {object} record - Raw TMC camera row.
+ * @param {string} cameraId - Derived stable id.
+ * @returns {string}
+ */
+export function nycCameraName(record, cameraId) {
+  const name = String(record?.name ?? '').trim();
+  if (name) return name;
+  const short = String(cameraId || '')
+    .replace(/^nyc-/, '')
+    .slice(0, 8);
+  return short ? `NYC Camera ${short}` : 'NYC Camera';
+}
+
+/**
+ * One NYC DOT TMC camera row -> one catalog source, or null.
+ *
+ * Frame URLs are synthesized from a validated UUID on the pinned TMC image
+ * origin. The payload's `imageUrl` is ignored so a catalog edit cannot steer
+ * the frame proxy off-host. There is no heading in the feed; headings use
+ * the shared id-hash fallback at low confidence, like Calgary and Fintraffic.
+ *
+ * @param {object} record - Raw TMC camera row.
+ * @returns {?object}
+ */
+export function nycCameraToSource(record) {
+  if (!record || typeof record !== 'object') return null;
+  if (!nycCameraIsOnline(record.isOnline)) return null;
+
+  const lat = toFiniteNumber(record.latitude);
+  const lon = toFiniteNumber(record.longitude);
+  if (!isLikelyNycCoordinate(lat, lon)) return null;
+
+  const uuid = String(record.id ?? '').trim();
+  if (!NYC_CAMERA_ID_RE.test(uuid)) return null;
+  const cameraId = `nyc-${uuid.toLowerCase()}`;
+  const imageUrl = NYC_IMAGE_URL(uuid);
+  const name = nycCameraName(record, cameraId);
+  const area = String(record.area ?? '').trim();
+  const city = NYC_BOROUGHS.has(area) ? area : 'New York';
+
+  return {
+    id: cameraId,
+    name,
+    city,
+    cityId: 'nyc',
+    provider: 'NYC DOT Traffic Management Center',
+    lat,
+    lon,
+    headingDeg: fallbackHeadingFromId(cameraId),
+    headingConfidence: 'low',
+    pitchDeg: -18,
+    fovDeg: 44,
+    rangeM: 145,
+    mountHeightM: 8,
+    groundElevationM: NYC_GROUND_ELEVATION_M,
+    feedType: 'image',
+    url: imageUrl,
+    snapshotUrl: imageUrl,
+    sourceKind: 'nyc-dot-tmc',
+    license: 'Public NYC DOT Traffic Management Center traffic camera',
+    code: cameraDisplayCode(name.toUpperCase()),
+  };
+}
+
+/**
+ * Fetch NYC DOT Traffic Management Center cameras from the keyless
+ * `webcams.nyctmc.org` catalog. Frames are stills on that same host,
+ * registered from a validated UUID rather than from the payload URL.
+ *
+ * @returns {Promise<Array<object>>} Normalized camera source objects.
+ */
+export async function loadNycSourcesFromOpenData() {
+  try {
+    const endpoint = process.env.CCTV_NYC_CAMERAS_URL || NYC_CAMERAS_URL;
+    const resp = await fetch(endpoint, {
+      headers: { Accept: 'application/json' },
+      redirect: 'manual',
+      signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
+    });
+    const discard = async () => {
+      try {
+        await resp.body?.cancel();
+      } catch {
+        /* no-op */
+      }
+      return [];
+    };
+    if (resp.status >= 300 && resp.status < 400) {
+      console.warn('[CCTV] NYC catalog redirected; redirects are not followed');
+      return discard();
+    }
+    if (!resp.ok) {
+      console.warn('[CCTV] NYC camera download failed:', resp.status);
+      return discard();
+    }
+    const rows = await readResponseJsonCapped(resp, NYC_MAX_CATALOG_BYTES);
+    if (!Array.isArray(rows)) return [];
+    const cameras = [];
+    const seen = new Set();
+    for (const record of rows) {
+      const camera = nycCameraToSource(record);
+      if (!camera || seen.has(camera.id)) continue;
+      seen.add(camera.id);
+      cameras.push(camera);
+    }
+    const maxRaw = Number(
+      process.env.CCTV_NYC_MAX_SOURCES || DEFAULT_NYC_MAX_SOURCES,
+    );
+    const maxCount = Number.isFinite(maxRaw)
+      ? Math.max(8, Math.min(400, Math.floor(maxRaw)))
+      : DEFAULT_NYC_MAX_SOURCES;
+    const prioritized = prioritizeSources(cameras, maxCount, NYC_ANCHORS);
+    console.log(
+      `[CCTV] Loaded NYC camera sources: ${cameras.length} (using nearest ${prioritized.length})`,
+    );
+    return prioritized;
+  } catch (error) {
+    console.warn('[CCTV] NYC camera download error:', error?.message || error);
+    return [];
+  }
+}

server/providers/cctv/stream.js

@@ -0,0 +1,348 @@
+/** Bounded, demand-owned HLS sessions. No subprocess, disk cache or idle sweep. */
+import { randomUUID } from 'node:crypto';
+
+export const HLS_LIMITS = Object.freeze({
+  sessions: 2,
+  leasesPerSession: 8,
+  playlistBytes: 256 * 1024,
+  segmentBytes: 4 * 1024 * 1024,
+  sessionBytes: 24 * 1024 * 1024,
+  segments: 12,
+  pollMs: 2000,
+  idleMs: 15000,
+  timeoutMs: 10000,
+  readyMs: 12000,
+});
+
+/** Refuse redirects; count actual streamed bytes, including chunked responses. */
+export async function fetchHlsBytes(
+  url,
+  {
+    fetchImpl = fetch,
+    signal,
+    maxBytes,
+    timeoutMs = HLS_LIMITS.timeoutMs,
+  } = {},
+) {
+  const controller = new AbortController();
+  const abort = () => controller.abort();
+  if (signal?.aborted) controller.abort();
+  else signal?.addEventListener('abort', abort, { once: true });
+  const timer = setTimeout(abort, timeoutMs);
+  let reader;
+  let response;
+  try {
+    controller.signal.throwIfAborted();
+    response = await fetchImpl(url, {
+      redirect: 'error',
+      signal: controller.signal,
+      headers: { 'User-Agent': 'gods-eye-view-cctv-proxy/1.0' },
+    });
+    if (
+      !response.ok ||
+      response.redirected ||
+      (response.url && new URL(response.url).href !== new URL(url).href)
+    )
+      throw new Error('HLS upstream refused');
+    if (Number(response.headers.get('content-length')) > maxBytes)
+      throw new Error('HLS response too large');
+    reader = response.body?.getReader?.();
+    if (!reader) throw new Error('HLS response is not streamable');
+    const chunks = [];
+    let length = 0;
+    for (;;) {
+      controller.signal.throwIfAborted();
+      const { done, value } = await reader.read();
+      controller.signal.throwIfAborted();
+      if (done) break;
+      length += value.byteLength;
+      if (length > maxBytes) throw new Error('HLS response too large');
+      chunks.push(Buffer.from(value));
+    }
+    return Buffer.concat(chunks, length);
+  } finally {
+    controller.abort();
+    if (reader) await reader.cancel().catch(() => {});
+    else await response?.body?.cancel().catch(() => {});
+    reader?.releaseLock();
+    clearTimeout(timer);
+    signal?.removeEventListener('abort', abort);
+  }
+}
+
+export function sameOriginHlsUrl(value, base) {
+  const url = new URL(value, base);
+  if (
+    !['https:', 'http:'].includes(url.protocol) ||
+    url.origin !== new URL(base).origin ||
+    url.username ||
+    url.password
+  )
+    throw new Error('HLS reference escapes registered origin');
+  return url.href;
+}
+
+/** Only clear MPEG-TS media playlists: encrypted, fMP4 and byte ranges fail closed. */
+export function parseHlsMedia(text, base, limit = HLS_LIMITS.segments) {
+  if (
+    !text.startsWith('#EXTM3U') ||
+    /#EXT-X-(?:KEY|MAP|BYTERANGE|I-FRAMES-ONLY)/.test(text)
+  )
+    throw new Error('Unsupported HLS playlist');
+  let seq = 0;
+  let duration = null;
+  let discontinuity = false;
+  const segments = [];
+  for (const raw of text.split('\n')) {
+    const line = raw.trim();
+    if (line.startsWith('#EXT-X-MEDIA-SEQUENCE:')) {
+      seq = Number(line.slice(22));
+      if (!Number.isSafeInteger(seq) || seq < 0)
+        throw new Error('Invalid media sequence');
+    } else if (line === '#EXT-X-DISCONTINUITY') {
+      discontinuity = true;
+    } else if (line.startsWith('#EXTINF:')) {
+      duration = Number.parseFloat(line.slice(8));
+      if (!Number.isFinite(duration) || duration <= 0 || duration > 60)
+        throw new Error('Invalid segment duration');
+    } else if (line && !line.startsWith('#')) {
+      if (duration === null || !Number.isSafeInteger(seq))
+        throw new Error('Invalid HLS segment');
+      const uri = sameOriginHlsUrl(line, base);
+      if (!new URL(uri).pathname.endsWith('.ts'))
+        throw new Error('Only MPEG-TS segments are supported');
+      segments.push({ seq: seq++, duration, uri, discontinuity });
+      discontinuity = false;
+      duration = null;
+    }
+  }
+  return segments.slice(-limit);
+}
+
+export function createHlsPuller({
+  fetchImpl = fetch,
+  limits = HLS_LIMITS,
+} = {}) {
+  const active = new Map();
+  let closed = false;
+  const stop = (cameraId, token) => {
+    const entry = active.get(cameraId);
+    if (!entry || (token && token !== entry.token)) return;
+    active.delete(cameraId);
+    entry.stopping = true;
+    entry.controller.abort();
+    clearTimeout(entry.timer);
+    for (const lease of entry.leases.values()) clearTimeout(lease.timer);
+    entry.leases.clear();
+    entry.segments.clear();
+    entry.upstream.clear();
+    entry.bytes = 0;
+  };
+  const release = (cameraId, leaseId) => {
+    const entry = active.get(cameraId);
+    const lease = entry?.leases.get(leaseId);
+    if (!lease) return;
+    clearTimeout(lease.timer);
+    entry.leases.delete(leaseId);
+    if (!entry.leases.size) stop(cameraId, entry.token);
+  };
+  const touch = (entry, leaseId, create = false) => {
+    let lease = entry.leases.get(leaseId);
+    if (!lease) {
+      if (!create || entry.leases.size >= limits.leasesPerSession)
+        throw new Error('HLS lease capacity reached');
+      lease = {};
+      entry.leases.set(leaseId, lease);
+    }
+    clearTimeout(lease.timer);
+    lease.timer = setTimeout(
+      () => release(entry.cameraId, leaseId),
+      limits.idleMs,
+    );
+    lease.timer.unref?.();
+  };
+  const read = (entry, url, maxBytes) =>
+    fetchHlsBytes(url, {
+      fetchImpl,
+      signal: entry.controller.signal,
+      maxBytes,
+      timeoutMs: limits.timeoutMs,
+    });
+  const poll = async (entry) => {
+    try {
+      let base = entry.chunklistUrl || entry.url;
+      let text = (await read(entry, base, limits.playlistBytes)).toString(
+        'utf8',
+      );
+      if (!text.includes('#EXTINF:')) {
+        const variant = text
+          .split('\n')
+          .map((line) => line.trim())
+          .find((line) => line && !line.startsWith('#'));
+        if (!variant || !text.includes('#EXT-X-STREAM-INF:'))
+          throw new Error('Missing HLS variant');
+        base = sameOriginHlsUrl(variant, entry.url);
+        text = (await read(entry, base, limits.playlistBytes)).toString('utf8');
+      }
+      entry.chunklistUrl = base;
+      const segments = parseHlsMedia(text, base, limits.segments);
+      const newest = segments.at(-1)?.seq ?? -1;
+      const restarted =
+        newest < entry.upstreamNewest ||
+        segments.some((segment) => {
+          const prior = entry.upstream.get(segment.seq);
+          return prior && prior !== segment.uri;
+        });
+      if (restarted) {
+        entry.upstream.clear();
+        entry.pendingDiscontinuity = true;
+      }
+      entry.upstreamNewest = newest;
+      for (const segment of segments) {
+        if (entry.stopping) return;
+        if (entry.upstream.has(segment.seq)) continue;
+        const body = await read(entry, segment.uri, limits.segmentBytes);
+        if (entry.stopping) return;
+        // Evict BEFORE retaining the incoming segment. One bounded download may
+        // exist in addition to the cache; no unbounded disk output is possible.
+        while (
+          entry.segments.size &&
+          (entry.bytes + body.length > limits.sessionBytes ||
+            entry.segments.size >= limits.segments)
+        ) {
+          const oldest = entry.segments.keys().next().value;
+          const removed = entry.segments.get(oldest);
+          entry.bytes -= removed.body.length;
+          if (removed.discontinuity) entry.discontinuitiesRemoved++;
+          entry.segments.delete(oldest);
+        }
+        if (body.length > limits.sessionBytes)
+          throw new Error('Segment exceeds session budget');
+        const seq = entry.nextSeq++;
+        entry.segments.set(seq, {
+          ...segment,
+          seq,
+          body,
+          discontinuity: segment.discontinuity || entry.pendingDiscontinuity,
+        });
+        entry.pendingDiscontinuity = false;
+        entry.upstream.set(segment.seq, segment.uri);
+        while (entry.upstream.size > limits.segments * 2)
+          entry.upstream.delete(entry.upstream.keys().next().value);
+        entry.bytes += body.length;
+      }
+      entry.failures = 0;
+    } catch {
+      entry.chunklistUrl = null;
+      entry.failures++;
+      if (entry.failures >= 3) {
+        stop(entry.cameraId, entry.token);
+        return;
+      }
+    } finally {
+      if (!entry.stopping) {
+        entry.timer = setTimeout(() => {
+          entry.polling = poll(entry);
+        }, limits.pollMs);
+        entry.timer.unref?.();
+      }
+    }
+  };
+  const ensure = async (cameraId, url, leaseId = 'legacy') => {
+    if (closed) throw new Error('HLS service closed');
+    sameOriginHlsUrl(url, url);
+    let entry = active.get(cameraId);
+    if (entry && entry.url !== url) {
+      stop(cameraId);
+      entry = null;
+    }
+    if (entry) {
+      touch(entry, leaseId, true);
+      return entry;
+    }
+    // Reserve synchronously, before any await: concurrent creation is singleflight.
+    if (active.size >= limits.sessions)
+      throw new Error('HLS session capacity reached');
+    entry = {
+      cameraId,
+      url,
+      token: randomUUID(),
+      controller: new AbortController(),
+      segments: new Map(),
+      bytes: 0,
+      stopping: false,
+      failures: 0,
+      nextSeq: 0,
+      upstream: new Map(),
+      leases: new Map(),
+      upstreamNewest: -1,
+      pendingDiscontinuity: false,
+      discontinuitiesRemoved: 0,
+    };
+    active.set(cameraId, entry);
+    touch(entry, leaseId, true);
+    entry.polling = poll(entry);
+    return entry;
+  };
+  const waitReady = async (entry, signal) => {
+    const until = Date.now() + limits.readyMs;
+    while (!entry.stopping && !signal?.aborted && Date.now() < until) {
+      if (entry.segments.size >= 2) return true;
+      await new Promise((resolve) => setTimeout(resolve, 50));
+    }
+    return false;
+  };
+  const buildPlaylist = async (entry, cameraId, leaseId = 'legacy') => {
+    if (entry.stopping || entry.segments.size < 2) return null;
+    touch(entry, leaseId);
+    const segments = [...entry.segments.values()].sort((a, b) => a.seq - b.seq);
+    const lines = [
+      '#EXTM3U',
+      '#EXT-X-VERSION:3',
+      `#EXT-X-TARGETDURATION:${Math.ceil(Math.max(...segments.map((s) => s.duration)))}`,
+      `#EXT-X-MEDIA-SEQUENCE:${segments[0].seq}`,
+      `#EXT-X-DISCONTINUITY-SEQUENCE:${entry.discontinuitiesRemoved}`,
+    ];
+    let previous;
+    for (const segment of segments) {
+      if (
+        segment.discontinuity ||
+        (previous !== undefined && segment.seq !== previous + 1)
+      )
+        lines.push('#EXT-X-DISCONTINUITY');
+      lines.push(
+        `#EXTINF:${segment.duration.toFixed(3)},`,
+        `/api/cctv/media/${encodeURIComponent(cameraId)}/seg_${segment.seq}.ts?session=${entry.token}&lease=${encodeURIComponent(leaseId)}`,
+      );
+      previous = segment.seq;
+    }
+    return lines.join('\n') + '\n';
+  };
+  const getSegment = (cameraId, token, seq, leaseId = 'legacy') => {
+    const entry = active.get(cameraId);
+    if (!entry || entry.token !== token || !entry.leases.has(leaseId))
+      return null;
+    const body = entry.segments.get(seq)?.body;
+    if (body) touch(entry, leaseId);
+    return body || null;
+  };
+  const shutdown = async () => {
+    closed = true;
+    const entries = [...active.values()];
+    for (const entry of entries) stop(entry.cameraId);
+    await Promise.allSettled(entries.map((entry) => entry.polling));
+  };
+  return {
+    ensure,
+    waitReady,
+    buildPlaylist,
+    getSegment,
+    stop,
+    release,
+    shutdown,
+    stats: () => ({
+      sessions: active.size,
+      bytes: [...active.values()].reduce((total, e) => total + e.bytes, 0),
+    }),
+  };
+}