Merge main into pr-688

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

server/providers/cctv/catalog.js

@@ -17,6 +17,7 @@ import {
   loadWarendorfSourcesFromCatalog,
   loadNswSourcesFromOpenData,
   loadCalgarySourcesFromOpenData,
+  loadDelDOTSourcesFromOpenData,
 } from './sources.js';
 
 /** Env kill switch: unset or anything but "0" means enabled. */
@@ -86,6 +87,11 @@ const LIVE_PACKS = [
     enabled: () => envEnabled('CCTV_CALGARY_ENABLED'),
     load: loadCalgarySourcesFromOpenData,
   },
+  {
+    name: 'deldot',
+    enabled: () => envEnabled('CCTV_DELDOT_ENABLED'),
+    load: loadDelDOTSourcesFromOpenData,
+  },
 ];
 /**
  * Load CCTV sources from a local JSON file (CCTV_SOURCES_FILE env or default).

server/providers/cctv/constants.js

@@ -236,6 +236,14 @@ 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;
 
+/** 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

server/providers/cctv/sources.js

@@ -56,6 +56,9 @@ import {
   CALGARY_DOWNTOWN,
   CALGARY_MAX_CATALOG_BYTES,
   CCTV_SOURCE_FETCH_TIMEOUT_MS,
+  DELDOT_CCTV_URL,
+  DEFAULT_DELDOT_MAX_SOURCES,
+  DELDOT_ANCHORS,
 } from './constants.js';
 import {
   toFiniteNumber,
@@ -1594,3 +1597,107 @@ 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 [];
+  }
+}

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),
+    }),
+  };
+}