From 292cd610f3bda22a671f06b69ac5aa33e1a5c6c2 Mon Sep 17 00:00:00 2001 From: Bilawal Sidhu Date: Tue, 15 Sep 2026 10:52:30 -0500 Subject: [PATCH] fix(cctv): bound HLS sessions and isolate consumer playback leases --- CHANGELOG.md | 8 + DATA_SOURCES.md | 4 +- README.md | 34 +- SECURITY.md | 2 +- TESTING.md | 8 +- docs/CURRENT-STATE.md | 16 + scripts/format-scope.json | 8 +- server/providers/cctv.js | 163 ++----- server/providers/cctv/sources.js | 17 +- server/providers/cctv/stream.js | 645 ++++++++----------------- src/data/cctvHlsStream.test.mjs | 128 +++++ src/data/dataCredits.js | 4 + src/layers/cctv/presentation.js | 2 +- src/layers/cctv/projection.js | 126 ++--- src/layers/cctv/rendering.js | 2 +- src/layers/cctv/videoPlayback.js | 81 ++++ src/layers/cctv/videoPlayback.test.mjs | 45 ++ src/ui/cctvControls.js | 7 + src/ui/cctvPresentation.js | 41 +- src/ui/cctvVideo.js | 31 ++ src/ui/cctvVideo.test.mjs | 21 + src/ui/styles/cctv.css | 2 - 22 files changed, 665 insertions(+), 730 deletions(-) create mode 100644 src/data/cctvHlsStream.test.mjs create mode 100644 src/layers/cctv/videoPlayback.js create mode 100644 src/layers/cctv/videoPlayback.test.mjs create mode 100644 src/ui/cctvVideo.js create mode 100644 src/ui/cctvVideo.test.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index 515c20845..7f84f97b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -893,3 +893,11 @@ represent previously published GitHub Releases. ## [0.1.0] — 2026-02-09 - Initial project version. + + +### Live CCTV integration candidate + +- Live HLS video shares one decoder between the camera panel and projection, + with a DelDOT HTTPS source pack. Credit: Daniel Slay (@Danielslay86), PR #489. +- Maintainer adjustments bound sessions and downloads, remove disk/subprocess + remuxing, reject redirects, and clean up playback on switching or disabling. diff --git a/DATA_SOURCES.md b/DATA_SOURCES.md index 6e8c2edd8..2bdb39d76 100644 --- a/DATA_SOURCES.md +++ b/DATA_SOURCES.md @@ -40,7 +40,7 @@ How to read this: | **Stadt Warendorf webcam** | Webcam frames, Marktplatz / Historisches Rathaus (`config/cctv_sources.warendorf.json`) | Public municipal webcam data | "Stadt Warendorf — webcam.warendorf.de" (courtesy); mount and heading derived from OSM geometry, "© OpenStreetMap contributors" (ODbL) | | **Live Traffic NSW (Transport for NSW)** | CCTV camera catalog + frames, New South Wales | [CC BY 4.0](https://opendata.transport.nsw.gov.au/) — attribution REQUIRED | "Live Traffic NSW — Transport for NSW" | | **Caltrans (cwwp2.dot.ca.gov)** | CCTV camera catalogs + frames, California districts | Public Caltrans traffic camera data | "Caltrans — cwwp2.dot.ca.gov" (courtesy) | -| **DelDOT (video.deldot.gov)** | CCTV camera catalog + live RTMP/HLS video, Delaware | Public DelDOT traffic camera streams via the [DelDOT traffic map](https://deldot.gov/map/) |"DelDOT — Delaware Department of Transportation" (courtesy) | +| **DelDOT (video.deldot.gov)** | CCTV camera catalog + live HTTPS HLS video, Delaware | Public DelDOT traffic camera streams via the [DelDOT traffic map](https://deldot.gov/map/) |"DelDOT — Delaware Department of Transportation" (courtesy) | | **TfL Open Data (JamCams)** | CCTV camera catalog + frames, London | [TfL Open Data terms](https://tfl.gov.uk/info-for/open-data-users/) — attribution REQUIRED | "Powered by TfL Open Data. Contains OS data © Crown copyright and database rights" | | **Ontario 511** | CCTV camera catalog + frames, Ontario highways including Kitchener-area routes | [Open Government Licence – Ontario](https://www.ontario.ca/page/open-government-licence-ontario) | "Ontario 511" plus Open Government Licence - Ontario | | **Fintraffic / Digitraffic (weathercams)** | CCTV camera catalog + frames, Finland | [CC BY 4.0](https://www.digitraffic.fi/en/terms-of-service/) — attribution REQUIRED | "Fintraffic / digitraffic.fi, license CC BY 4.0" | @@ -59,7 +59,7 @@ How to read this: - **Launch Library 2.** `/api/launches` makes a server-side rolling-30-day query against the supported v2.3 detailed launch endpoint, caches successful responses for 15 minutes in memory and on disk, and serves the last successful response during a throttle or transient outage. Anonymous access is limited to 15 calls/hour; deployments can provide `LL2_API_TOKEN` for authenticated access. The Space Devs' published terms permit using and sharing the API data in any form, ask users not to forward it without adding value, disclaim complete accuracy, and encourage—but do not require—attribution. This app keeps a courtesy credit. Payload and stage/recovery records are shown only when supplied. Failed launches expose their source status and never receive fallback orbit geometry or a live/estimated marker. LL2 supplies launch context and event timing, not continuous ascent telemetry or live orbital state. - **TfL JamCams.** The camera list comes from the keyless `api.tfl.gov.uk` endpoint (an optional `TFL_APP_KEY` raises its rate limit); frames come from TfL's public S3 bucket. The "Powered by TfL Open Data" attribution is required by TfL's terms and is registered in the Data attribution popover. - **Ontario 511 CCTV.** The camera list comes from the keyless `511on.ca/api/v2/get/cameras` endpoint; frames are server-registered still images under `511on.ca/map/Cctv/`. The source is live-fetched at runtime and attributed under the Open Government Licence - Ontario. -- **DelDOT live video cameras.** The camera list is the keyless `tmc.deldot.gov/json/videocamera.json` catalog; only cameras reporting `Active` are registered, and each camera's `rtmpt://video.deldot.gov:80` stream URL is preferred (the identical stream to `rtmp://…:1935`, tunneled over HTTP). Live video is remuxed transiently on the server — segments live under the OS temp dir in a short rolling window, pruned and reaped when idle, never stored beyond that window or redistributed. Disable with `CCTV_DELDOT_ENABLED=0`. +- **DelDOT live video cameras.** The keyless `tmc.deldot.gov/json/videocamera.json` catalog supplies active cameras and official `https://video.deldot.gov/live/…/playlist.m3u8` links. The local server transiently buffers bounded MPEG-TS segments in memory and deletes them on session release or idle expiry. No recordings are retained. `CCTV_DELDOT_ENABLED=0` disables the pack. Public availability is not a blanket redistribution license; deployment beyond personal local viewing needs confirmation of applicable DelDOT terms. - **Fintraffic weathercams.** The camera list is one keyless GeoJSON request to `tie.digitraffic.fi/api/weathercam/v1/stations` (809 stations / 2,275 presets), sent with the `Digitraffic-User: gods-eye-view` header the service asks clients to identify themselves with and with gzip; frames are stills on `weathercam.digitraffic.fi`, refreshed on each station's 600 s collection interval, which is also the client's ambient-still refresh cadence for this pack (the active camera keeps the usual 10-second refresh). One preset (a fixed camera view) is one camera; stations not `GATHERING` and presets not `inCollection` are dropped. Attribution is required by the CC BY 4.0 licence and is registered in the Data attribution popover. `CCTV_FINTRAFFIC_ENABLED=0` is the kill switch. - **DriveBC highway cameras.** The camera list comes from the keyless `https://www.drivebc.ca/api/webcams/` endpoint served by the DriveBC.ca site ([bcgov/DriveBC.ca](https://github.com/bcgov/DriveBC.ca)); frames come from `https://www.drivebc.ca/images/.jpg`, built from the numeric camera id. DataBC's [DriveBC HighwayCams](https://catalogue.data.gov.bc.ca/dataset/bc-highwaycams) catalogue entry publishes the same cameras under the Open Government Licence – British Columbia, but its CSV still lists the retired `images.drivebc.ca` frame URLs, which now return a placeholder. The licence's attribution statement is required and is registered in the Data attribution popover. By default the 250 cameras nearest Vancouver and Victoria load; `CCTV_DRIVEBC_MAX_SOURCES` changes the cap and `CCTV_DRIVEBC_ENABLED=0` turns the pack off. Some cameras in the feed are supplied by partners (TransLink, the City of Vancouver, the City of Surrey, Parks Canada / Alberta Motor Association and others); the feed's per-camera credit is carried onto the camera record and shown beside the provider in the CCTV panel. - **TxDOT ITS cameras.** The catalog comes from the keyless `its.txdot.gov/its/DistrictIts/GetCctvStatusListByDistrict` endpoint, one call per district. Frames come from the per-camera `GetCctvSnapshotByIcdId` endpoint, which answers JSON carrying a base64 JPEG rather than an image body; the proxy decodes it only for responses from the official TxDOT origin and validates the JPEG header before serving. Only cameras reporting `Device Online` are registered. Districts default to Austin and San Antonio (`CCTV_TXDOT_DISTRICTS`; 25 codes exist statewide); `CCTV_TXDOT_MAX_SOURCES` caps the pack and `CCTV_TXDOT_ENABLED=0` turns it off. Frames are fetched live at request time and are never stored or redistributed. diff --git a/README.md b/README.md index e5d3299b7..93628b799 100644 --- a/README.md +++ b/README.md @@ -434,28 +434,22 @@ OpenSky can run fully anonymous (`OPENSKY_AUTH_MODE=anon`), or import OAuth cred
-Live video cameras (HLS / RTMP) +Live video cameras (HLS) -CCTV sources with `"feedType": "hls"` play as continuous video instead of -refreshed stills. The server owns the stream: it keeps one upstream session -per active camera, serves the browser a locally generated playlist, and the -client plays it through hls.js. Two upstream strategies, chosen by URL: +CCTV sources with `"feedType": "hls"` and a registered HTTP(S) `.m3u8` +URL play through a lazily loaded hls.js decoder shared by the monitor plane +and panel. DelDOT uses its official HTTPS HLS catalog links. Disable that +pack with `CCTV_DELDOT_ENABLED=0`. -- **`.m3u8` upstreams** are pulled directly in Node — no extra dependencies. -- **RTMP and other stream URLs** go through **ffmpeg** (`-c copy`, no - re-encode), which is optional: without it those cameras fall back to the - stills path and everything else is unchanged. If you self-host in Docker, - add `ffmpeg` to your image. - -The DelDOT Delaware pack is the working example — every `Active` camera in -DelDOT's public catalog registers with its `rtmpt://video.deldot.gov:80/…` -stream (the same stream as `rtmp://…:1935`, tunneled over HTTP, so it works -where port 1935 is blocked). `CCTV_DELDOT_ENABLED=0` turns the pack off. - -Some agencies restart their streams on a timer or run encoders whose clock -lags real time (DelDOT does both). The pipeline absorbs that automatically — -a brief hitch at a restart, and a playback-rate governor that holds the -stream a few seconds behind live so it never starves. Nothing to configure. +The server allows two concurrent sessions. Each retains at most 12 segments +and 24 MiB in memory; individual downloads are capped at 4 MiB with a ten +second deadline. There are no segment files or ffmpeg processes. Redirects, +off-origin references, encrypted playlists and non-MPEG-TS segments are refused. +Each decoder has its own client lease (at most eight per session), including +native HLS. Closing it releases only that lease; abandoned leases expire after +15 seconds without access. The last release stops upstream work. Failed live video +uses the existing still/Street View/synthetic fallback, which is not live video. +RTMP-only sources are not supported by this integration.
diff --git a/SECURITY.md b/SECURITY.md index ee0a1625d..934f51032 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -62,7 +62,7 @@ browser-key fallback for existing single-key setups. The data proxies under `server/providers/` are written so the browser cannot turn the server into an open relay: - **No arbitrary-URL fetching.** The CCTV frame proxy fetches only server-registered camera/frame URLs — clients cannot pass an upstream URL to fetch (SSRF mitigation). Other proxies target fixed upstream hosts. -- **Live video stays origin-pinned.** `/api/cctv/media//…` sub-paths resolve only against the registered camera's origin; a playlist reference or segment that would resolve elsewhere is refused. Segment stores live under the OS temp dir, keyed by a sanitized camera id, and are pruned to a fixed window and reaped when idle. When ffmpeg is present it is spawned only with a registered camera URL as input — never anything supplied by a client. +- **Live HLS is bounded and origin-pinned.** Registered HTTP(S) HLS sources use a Node puller that rejects redirects and off-origin playlist references, bounds playlists and segment bodies while streaming, and aborts timed-out or released sessions. At most two sessions retain 12 segments/24 MiB each, plus bounded download buffers. Stores are memory-only and idle sessions expire after 15 seconds. RTMP/ffmpeg execution is not enabled. Locally configured source URLs remain an operator trust boundary. - **Radio is not an audio relay.** `/api/radio/stations` contacts only allowlisted Radio Browser HTTPS hosts and paths, rejects redirects, rejects any hostname with a loopback/private/link-local/metadata/non-public A or AAAA result, and pins each TLS connection to a validated address. It returns normalized public HTTPS stream URLs; `/api/radio/click/:uuid` applies the same destination policy and accepts only station IDs from the current bounded catalog. The browser then connects directly to the broadcaster after an explicit playback action, so the broadcaster sees the listener's IP address. GEV never proxies, caches, records, or redistributes audio. - **No verbatim client headers upstream.** The CCTV media route is the one route that relays a request header (`Range`, for video seeking). It is parsed and canonicalized before it is forwarded: one `bytes=` range only, with every accepted form — explicit span, open-ended and suffix — bounded to the same 64 MiB ceiling the relay applies to a declared response body. Multi-range, malformed, inverted and non-`bytes` values are dropped and the request proceeds without a `Range`, as RFC 7233 §3.1 prescribes. Bounding the request bounds what is asked for: a response that declares no length — live streamed media, or a chunked body from an upstream that ignores the `Range` — has no ceiling. An upstream request the browser has stopped waiting for is cancelled rather than left running, whether the viewer leaves before the headers arrive or during the body. - **Response-size caps and timeouts** on proxied responses. diff --git a/TESTING.md b/TESTING.md index 1c2e907af..88daf1a32 100644 --- a/TESTING.md +++ b/TESTING.md @@ -212,15 +212,15 @@ While recording, call out anything in these areas — this is the feedback I mos --- -## 6. Live video CCTV (HLS / RTMP) — soak, no mic +## 6. Live video CCTV (HTTPS HLS) — soak, no mic -> Requires the DelDOT pack (loads by default; `CCTV_DELDOT_ENABLED=0` disables it) and, for RTMP cameras, `ffmpeg` on the server's PATH. +> Requires the DelDOT pack (loads by default; `CCTV_DELDOT_ENABLED=0` disables it). -1. Turn on **CCTV**, fly to Dover, DE, and activate **DE 1 @ Main Toll Plaza**. ✅ Within ~10 s the monitor plane and the panel preview both show moving video, and the panel widens for it. +1. Turn on **CCTV**, fly to Dover, DE, and activate **DE 1 @ Main Toll Plaza**. ✅ Within ~30 s the monitor plane and the panel preview both show moving video,. 2. Leave it for **five minutes**. ✅ Continuous playback; at most a short hitch about once a minute (the agency restarts its stream on a timer). ❌ A freeze that does not recover, or the panel picture stopping while the plane keeps moving. 3. Toggle **PROJECTION** off. ✅ The panel keeps playing; the plane and the active camera's cone hide. 4. Switch to **DE 8 @ Saulsbury Rd**, then back. ✅ Each switch resumes within ~10 s with no stale frame from the previous camera. -5. Optional: DevTools → Network, filter `media/`. ✅ Playlist responses carry `X-CCTV-Source: ffmpeg-remux` (RTMP) or `hls-pull` (HLS); segments return `video/mp2t`. +5. Optional: DevTools → Network, filter `media/`. ✅ Playlist responses carry `X-CCTV-Source: hls-pull`; segments return `video/mp2t`. ## If something looks off diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 5e533f4e5..bb788939b 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -3621,3 +3621,19 @@ releases imagery layers/listeners on replacement. Destroy invalidates pending work and releases owned resources, including late factory results. Supplied 3D tilesets remain owned by the caller; tilesets created through the controller's factory are added to its viewer and removed on destruction. + + +## Live CCTV integration candidate (#489) + +Daniel Slay's shared-decoder panel/projection feature is adapted to bounded +HTTP(S) HLS. DelDOT uses official HTTPS HLS URLs; RTMP/ffmpeg remux is excluded +because subprocess transport does not inherit the Node redirect policy. Server +sessions reserve synchronously, cap memory and response bytes, reject redirects, +and abort on release/shutdown. No disk store or background sweep exists while +inactive. The panel paints the same video at at most 640px/15 fps; collapse stops +its paint loop. Camera deactivation destroys the decoder, and fatal playback +errors revert to the labeled frame fallback. Camera changes clear the old panel +frame. hls.js is imported only when an HLS camera starts. Two live sessions may +coexist with at most eight independent consumer leases each; no-access lease +expiry is 15 seconds. Closing one consumer leaves other consumers running. Encrypted/fMP4/byte-range playlists fail +closed. Performance and live agency reliability require recorded validation. diff --git a/scripts/format-scope.json b/scripts/format-scope.json index 30b035a0e..7c0e172ae 100644 --- a/scripts/format-scope.json +++ b/scripts/format-scope.json @@ -675,5 +675,11 @@ "src/search/nominatim.test.mjs", "src/sources/httpBody.js", "src/sources/overpass.js", - "src/sources/nominatim.js" + "src/sources/nominatim.js", + "server/providers/cctv/stream.js", + "src/data/cctvHlsStream.test.mjs", + "src/layers/cctv/videoPlayback.js", + "src/layers/cctv/videoPlayback.test.mjs", + "src/ui/cctvVideo.js", + "src/ui/cctvVideo.test.mjs" ] diff --git a/server/providers/cctv.js b/server/providers/cctv.js index 44be68bf3..1e7a1a938 100644 --- a/server/providers/cctv.js +++ b/server/providers/cctv.js @@ -1,5 +1,3 @@ -import path from 'node:path'; -import { promises as fsp } from 'node:fs'; import { createCctvCatalog } from './cctv/catalog.js'; import { normalizeFeedType, @@ -19,7 +17,7 @@ import { CCTV_MAX_SOURCES_CEILING, } from './cctv/constants.js'; import { sanitizeCctvRangeHeader } from './cctv/range.js'; -import { createHlsPuller, createHlsRemuxer } from './cctv/stream.js'; +import { createHlsPuller } from './cctv/stream.js'; import { googleServerApiKey } from './places/google-key.js'; export { CCTV_FRAME_FETCH_TIMEOUT_MS, fetchCctvImageFromUpstream }; /** @@ -45,7 +43,6 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { const HEALTH_MAX_ENTRIES = CCTV_MAX_SOURCES_CEILING; /** Live HLS strategies (see ./cctv/stream.js). Shared across dev and preview. */ const puller = createHlsPuller(); - const remuxer = createHlsRemuxer(); /** Update the health entry for a camera, evicting the oldest entry if at capacity. */ const setHealth = (cameraId, patch) => { @@ -136,7 +133,6 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { server.httpServer?.on('close', () => { puller.shutdown(); - remuxer.shutdown(); }); server.middlewares.use('/api/cctv', async (req, res) => { try { @@ -205,133 +201,56 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { } if (url.pathname.startsWith('/media/')) { - // /media/ is the playlist (or the whole media for non-HLS feeds); - - // /media// is a playlist- or segment-relative sub-path. - - const mediaPath = url.pathname.replace('/media/', ''); - - const slash = mediaPath.indexOf('/'); - - const cameraId = - decodeURIComponent( - (slash === -1 ? mediaPath : mediaPath.slice(0, slash)).trim(), - ) || 'camera'; - - const rel = slash === -1 ? '' : mediaPath.slice(slash + 1); - + const match = /^\/media\/([^/]+)(?:\/(seg_(\d+)\.ts))?$/.exec(url.pathname); + if (!match) { res.writeHead(404); res.end(); return; } + const cameraId = decodeURIComponent(match[1]); const source = sourceById.get(cameraId); - const mediaUrl = source?.url || ''; - const feedType = normalizeFeedType(source?.feedType || 'image'); - - // HLS feeds are served from a proxy-owned segment store with a - - // proxy-generated playlist, so upstream session churn never reaches - - // the browser. http(s) .m3u8 upstreams use the Node puller; RTMP and - - // other stream kinds use ffmpeg when it is available. Anything else - - // falls through to the direct proxy below. - - const hlsStrategy = - feedType === 'hls' && - mediaUrl && - /^(https?|rtmps?|rtmpt):\/\//i.test(mediaUrl) - ? /\.m3u8(\?|$)/i.test(mediaUrl) - ? puller - : remuxer.isAvailable() - ? remuxer - : null - : null; - - if (hlsStrategy) { - const entry = await hlsStrategy.ensure(cameraId, mediaUrl); - - const safe = path.basename(rel || 'out.m3u8'); - - const isPlaylist = safe.endsWith('.m3u8'); - - const starting = () => { - res.writeHead(503, { - 'Content-Type': 'application/json', - - 'Cache-Control': 'no-store', - - 'Retry-After': '2', - }); - - res.end(JSON.stringify({ error: 'Stream starting' })); - }; - - if (isPlaylist) { - if (!(await hlsStrategy.waitReady(entry))) { - starting(); - - return; - } - - const playlist = await hlsStrategy.buildPlaylist(entry, cameraId); - - if (!playlist) { - starting(); - - return; - } - - setHealth(cameraId, { - status: 'ok', - - sourceKind: 'live', - - label: source?.provider || 'Configured source', - - message: 'Live stream connected', - }); - - res.writeHead(200, { - 'Content-Type': 'application/vnd.apple.mpegurl', - - 'Cache-Control': 'no-store', - - 'X-CCTV-Source': - hlsStrategy === puller ? 'hls-pull' : 'ffmpeg-remux', - }); - - res.end(playlist); - + const leaseId = url.searchParams.get('lease'); + if (feedType === 'hls' && !/^[a-f0-9-]{36}$/i.test(leaseId || '')) { res.writeHead(400); res.end(); return; } + if (req.method === 'DELETE') { + if (leaseId) puller.release(cameraId, leaseId); + res.writeHead(204); res.end(); return; + } + if (req.method !== 'GET') { res.writeHead(405); res.end(); return; } + if (feedType === 'hls') { + if (!/^https?:\/\//i.test(mediaUrl) || !/\.m3u8(?:\?|$)/i.test(mediaUrl)) { + res.writeHead(503, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: 'This stream requires an unsupported transport; use the frame fallback' })); return; } - + if (match[2]) { + const body = puller.getSegment(cameraId, url.searchParams.get('session'), Number(match[3]), leaseId); + res.writeHead(body ? 200 : 404, { 'Content-Type': 'video/mp2t', 'Cache-Control': 'no-store' }); + res.end(body || undefined); return; + } + const downstream = watchDownstreamClose(res); + let entry; + const cancelPending = () => { if (entry) puller.release(cameraId, leaseId); }; try { - entry.lastAccess = Date.now(); - - const buf = await fsp.readFile(path.join(entry.dir, safe)); - - res.writeHead(200, { - 'Content-Type': 'video/mp2t', - - 'Cache-Control': 'no-store', - - 'X-CCTV-Source': - hlsStrategy === puller ? 'hls-pull' : 'ffmpeg-remux', - }); - - res.end(buf); + entry = await puller.ensure(cameraId, mediaUrl, leaseId); + if (downstream.closed) { cancelPending(); return; } + downstream.signal.addEventListener('abort', cancelPending, { once: true }); + if (!(await puller.waitReady(entry, downstream.signal))) throw new Error('Stream unavailable'); + const playlist = await puller.buildPlaylist(entry, cameraId, leaseId); + if (downstream.closed) return; + if (!playlist) throw new Error('Stream unavailable'); + setHealth(cameraId, { status: 'ok', sourceKind: 'live', label: source?.provider || 'Configured source', message: 'Live HLS connected' }); + res.writeHead(200, { 'Content-Type': 'application/vnd.apple.mpegurl', 'Cache-Control': 'no-store', 'X-CCTV-Source': 'hls-pull', 'X-CCTV-Session': entry.token }); + res.end(playlist); } catch { - res.writeHead(404, { - 'Content-Type': 'application/json', - - 'Cache-Control': 'no-store', - }); - - res.end(JSON.stringify({ error: 'Segment not found' })); + setHealth(cameraId, { status: 'degraded', sourceKind: 'fallback', label: source?.provider || 'Configured source', message: 'Live HLS unavailable' }); + if (!downstream.closed) { + res.writeHead(503, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'Retry-After': '2' }); + res.end(JSON.stringify({ error: 'Live stream unavailable' })); + } + } finally { + downstream.signal.removeEventListener('abort', cancelPending); } - return; } + if (match[2] || req.method !== 'GET') { res.writeHead(404); res.end(); return; } if (!mediaUrl || !/^https?:\/\//i.test(mediaUrl)) { setHealth(cameraId, { diff --git a/server/providers/cctv/sources.js b/server/providers/cctv/sources.js index 9cb9910a5..49b7f806e 100644 --- a/server/providers/cctv/sources.js +++ b/server/providers/cctv/sources.js @@ -1603,12 +1603,13 @@ export async function loadDelDOTSourcesFromOpenData() { 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 resp.json(); + const payload = await readResponseJsonCapped(resp, 2 * 1024 * 1024); const rows = Array.isArray(payload?.videoCameras) ? payload.videoCameras : []; @@ -1623,12 +1624,14 @@ export async function loadDelDOTSourcesFromOpenData() { // 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; - // Official-host pin: RTMP-over-HTTP (rtmpt:80) is the steady transport. - const streamUrl = String(row?.urls?.rtmp || ''); - if (!streamUrl.startsWith('rtmpt://video.deldot.gov:80/live/')) 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 (!id) continue; + if (!/^[A-Za-z0-9_-]{1,80}$/.test(id)) continue; const cameraId = `deldot-${id.toLowerCase()}`; const title = String(row?.title || '').trim(); @@ -1657,10 +1660,10 @@ export async function loadDelDOTSourcesFromOpenData() { fovDeg: hasHeading ? 56 : 44, rangeM: hasHeading ? 210 : 145, mountHeightM: hasHeading ? 10 : 8, - groundElevationM: 10, // Delaware is near sea level and flat. + groundElevationM: 10, // Estimated prior; client ground resolution owns placement. feedType: 'hls', url: streamUrl, - snapshotUrl: streamUrl, + snapshotUrl: '', sourceKind: 'deldot-open-data', license: 'Public DelDOT traffic camera', }); diff --git a/server/providers/cctv/stream.js b/server/providers/cctv/stream.js index 9310090b3..e3bb85ab6 100644 --- a/server/providers/cctv/stream.js +++ b/server/providers/cctv/stream.js @@ -1,469 +1,238 @@ -/** - * Live HLS delivery strategies for the CCTV media proxy. - * - * Both strategies own a per-camera segment store on disk (seg_.ts in a - * temp dir) and generate the playlist the browser sees, so upstream session - * behaviour never reaches the client: - * - * createHlsPuller — for http(s) .m3u8 upstreams. Pure Node, no dependencies. - * createHlsRemuxer — for RTMP and other stream kinds, via ffmpeg (-c copy). - * Optional: absent, those cameras fall back to stills. - * - * Both expose the same surface: ensure(cameraId, url) -> entry, - * waitReady(entry), buildPlaylist(entry, cameraId), shutdown(). The media - * route in ../cctv.js picks a strategy by URL and serves files from entry.dir. - */ -import os from 'node:os'; -import path from 'node:path'; -import { promises as fsp } from 'node:fs'; -import { spawn } from 'node:child_process'; -import { CCTV_MEDIA_FETCH_TIMEOUT_MS } from './constants.js'; +/** Bounded, demand-owned HLS sessions. No subprocess, disk cache or idle sweep. */ +import { randomUUID } from 'node:crypto'; -/** - * Read a text response with a byte cap, cancelling the body on overflow. - * media.js keeps its byte-buffer reader private; playlists need text. - * @returns {Promise<{tooLarge:boolean,text:string}>} - */ -export async function readCappedResponseText(upstream, maxBytes) { - const declared = Number(upstream.headers.get('content-length')); - if (Number.isFinite(declared) && declared > maxBytes) { - try { await upstream.body?.cancel(); } catch { /* no-op */ } - return { tooLarge: true, text: '' }; - } - if (!upstream.body || typeof upstream.body[Symbol.asyncIterator] !== 'function') { - const text = await upstream.text(); - return text.length > maxBytes ? { tooLarge: true, text: '' } : { tooLarge: false, text }; - } - const decoder = new TextDecoder(); - let text = ''; - let total = 0; - for await (const chunk of upstream.body) { - total += chunk.length; - if (total > maxBytes) { - try { await upstream.body.cancel(); } catch { /* no-op */ } - return { tooLarge: true, text: '' }; +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)); } - text += decoder.decode(chunk, { stream: true }); + 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); } - text += decoder.decode(); - return { tooLarge: false, text }; } -/** - * Per-camera HLS puller with a proxy-owned segment store (no ffmpeg). - * - * Wowza-style upstreams expire sessions fast, but their segment numbering is - * stream-global (EXT-X-MEDIA-SEQUENCE carries across sessions). So the proxy - * polls the chunklist every ~2 s (which also keeps the session warm), re-reads - * the master the moment a poll is refused, and downloads every segment it has - * not yet seen, keyed by that global sequence. Zero overlap, zero loss across - * session rotations, exact EXTINF durations from upstream, PTS continuous - * because it is the same encoder. Serves the same file layout as - * createHlsRemuxer (seg_.ts in a per-camera dir) so the media route is - * shared. Idle cameras are reaped after IDLE_MS. - */ -export function createHlsPuller() { - const OUT_ROOT = path.join(os.tmpdir(), 'gev-hls-pull'); - const IDLE_MS = 60_000; - const READY_TIMEOUT_MS = 20_000; - const POLL_MS = 2_000; - const PLAYLIST_WINDOW = 15; - const KEEP_ON_DISK = 24; - const MIN_READY_SEGMENTS = 3; - const FETCH_TIMEOUT_MS = CCTV_MEDIA_FETCH_TIMEOUT_MS; - const MAX_PLAYLIST_BYTES = 256 * 1024; - const UA = 'gods-eye-view-cctv-proxy/1.0'; - /** @type {Map} */ +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(); - const headers = { 'User-Agent': UA }; - const fetchText = async (url) => { - const r = await fetch(url, { headers, signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) }); - if (!r.ok) { const e = new Error(`HTTP ${r.status}`); e.status = r.status; throw e; } - const { tooLarge, text } = await readCappedResponseText(r, MAX_PLAYLIST_BYTES); - if (tooLarge) throw new Error('playlist too large'); - return text; + 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; }; - /** Follow a master playlist to its first media playlist (or return it if already one). */ - const resolveChunklist = async (masterUrl) => { - const text = await fetchText(masterUrl); - if (text.includes('#EXTINF')) return { url: masterUrl, text }; - const first = text.split('\n').map((l) => l.trim()).find((l) => l && !l.startsWith('#')); - if (!first) throw new Error('master playlist has no variants'); - const url = new URL(first, masterUrl); - if (url.origin !== new URL(masterUrl).origin) throw new Error('variant escapes origin'); - return { url: url.toString(), text: await fetchText(url.toString()) }; + 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); }; - /** Parse a media playlist into [{seq, duration, uri}] using MEDIA-SEQUENCE + index. */ - const parseMedia = (text, baseUrl) => { - let mediaSeq = 0; - let target = 0; - let dur = null; - const out = []; - for (const raw of text.split('\n')) { - const line = raw.trim(); - if (!line) continue; - if (line.startsWith('#EXT-X-MEDIA-SEQUENCE:')) mediaSeq = Number(line.slice(22)) || 0; - else if (line.startsWith('#EXT-X-TARGETDURATION:')) target = Number(line.slice(22)) || 0; - else if (line.startsWith('#EXTINF:')) dur = parseFloat(line.slice(8)) || null; - else if (!line.startsWith('#')) { - let uri; - try { uri = new URL(line, baseUrl); } catch { continue; } - if (uri.origin !== new URL(baseUrl).origin) continue; - const m = /_(\d+)\.ts(\?|$)/i.exec(uri.pathname); - const seq = m ? Number(m[1]) : mediaSeq + out.length; - out.push({ seq, duration: dur || target || 2, uri: uri.toString() }); - dur = null; - } + 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); } - return { segments: out, target }; + clearTimeout(lease.timer); + lease.timer = setTimeout(() => release(entry.cameraId, leaseId), limits.idleMs); + lease.timer.unref?.(); }; - const downloadSegment = async (entry, seg) => { - const r = await fetch(seg.uri, { headers, signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) }); - if (!r.ok) throw new Error(`segment ${seg.seq} HTTP ${r.status}`); - const buf = Buffer.from(await r.arrayBuffer()); - const file = path.join(entry.dir, `seg_${seg.seq}.ts`); - await fsp.writeFile(`${file}.part`, buf); - await fsp.rename(`${file}.part`, file); - entry.segments.set(seg.seq, { duration: seg.duration }); - }; - const prune = (entry) => { - const seqs = [...entry.segments.keys()].sort((a, b) => a - b); - while (seqs.length > KEEP_ON_DISK) { - const n = seqs.shift(); - entry.segments.delete(n); - fsp.rm(path.join(entry.dir, `seg_${n}.ts`), { force: true }).catch(() => {}); - } - }; - const poll = async (cameraId, entry) => { - if (entry.stopping || entry.polling) return; - entry.polling = true; + const read = (entry, url, maxBytes) => fetchHlsBytes(url, { + fetchImpl, signal: entry.controller.signal, maxBytes, timeoutMs: limits.timeoutMs, + }); + const poll = async (entry) => { try { - let text; - if (!entry.chunklistUrl) { - const r = await resolveChunklist(entry.masterUrl); - entry.chunklistUrl = r.url; - text = r.text; - } else { - try { - text = await fetchText(entry.chunklistUrl); - } catch (err) { - if (err.status === 403 || err.status === 404) { - // Session rotated: re-read the master for the live chunklist. - const r = await resolveChunklist(entry.masterUrl); - if (r.url !== entry.chunklistUrl) console.info(`[CCTV:pull ${cameraId}] session rotated`); - entry.chunklistUrl = r.url; - text = r.text; - } else { - throw err; - } - } + 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'); } - const { segments } = parseMedia(text, entry.chunklistUrl); - for (const seg of segments) { - if (entry.stopping) break; - if (entry.segments.has(seg.seq) || entry.inFlight.has(seg.seq)) continue; - if (entry.lastSeq !== null && seg.seq <= entry.lastSeq) continue; // older than what we hold - entry.inFlight.add(seg.seq); - try { - await downloadSegment(entry, seg); - entry.lastSeq = Math.max(entry.lastSeq ?? -1, seg.seq); - } catch (err) { - console.warn(`[CCTV:pull ${cameraId}]`, err.message); - } finally { - entry.inFlight.delete(seg.seq); + 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; } - prune(entry); entry.failures = 0; - } catch (err) { - entry.failures += 1; - if (entry.failures === 1 || entry.failures % 10 === 0) { - console.warn(`[CCTV:pull ${cameraId}] poll failed (${entry.failures}):`, err.message); - } - if (entry.failures >= 3) entry.chunklistUrl = null; // force master re-resolve - } finally { - entry.polling = false; - } - }; - const ensure = async (cameraId, masterUrl) => { - let entry = active.get(cameraId); - if (entry) { - entry.lastAccess = Date.now(); - return entry; - } - const dir = path.join(OUT_ROOT, cameraId.replace(/[^a-zA-Z0-9_-]/g, '_')); - await fsp.rm(dir, { recursive: true, force: true }).catch(() => {}); - await fsp.mkdir(dir, { recursive: true }); - entry = { - dir, - masterUrl, - chunklistUrl: null, - segments: new Map(), - inFlight: new Set(), - lastSeq: null, - lastAccess: Date.now(), - failures: 0, - polling: false, - stopping: false, - timer: null, - }; - active.set(cameraId, entry); - entry.timer = setInterval(() => { poll(cameraId, entry); }, POLL_MS); - entry.timer.unref?.(); - poll(cameraId, entry); - return entry; - }; - const waitReady = async (entry) => { - const deadline = Date.now() + READY_TIMEOUT_MS; - while (Date.now() < deadline) { - if (entry.segments.size >= MIN_READY_SEGMENTS) return true; - await new Promise((r) => setTimeout(r, 250)); - } - return false; - }; - const buildPlaylist = async (entry, cameraId) => { - const seqs = [...entry.segments.keys()].sort((a, b) => a - b); - if (seqs.length < 2) return null; - const window = seqs.slice(-PLAYLIST_WINDOW); - const prefix = `/api/cctv/media/${encodeURIComponent(cameraId)}/`; - const maxDur = Math.max(...window.map((n) => entry.segments.get(n).duration)); - const lines = [ - '#EXTM3U', - '#EXT-X-VERSION:3', - `#EXT-X-TARGETDURATION:${Math.ceil(maxDur)}`, - `#EXT-X-MEDIA-SEQUENCE:${window[0]}`, - ]; - let prev = null; - for (const n of window) { - if (prev !== null && n !== prev + 1) lines.push('#EXT-X-DISCONTINUITY'); - lines.push(`#EXTINF:${entry.segments.get(n).duration.toFixed(3)},`, `${prefix}seg_${n}.ts`); - prev = n; - } - return lines.join('\n') + '\n'; - }; - const stopEntry = (entry) => { - entry.stopping = true; - if (entry.timer) { clearInterval(entry.timer); entry.timer = null; } - fsp.rm(entry.dir, { recursive: true, force: true }).catch(() => {}); - }; - const sweep = setInterval(() => { - const now = Date.now(); - for (const [id, entry] of active) { - if (now - entry.lastAccess > IDLE_MS) { stopEntry(entry); active.delete(id); } - } - }, 30_000); - sweep.unref?.(); - const shutdown = () => { - clearInterval(sweep); - for (const [, entry] of active) stopEntry(entry); - active.clear(); - }; - return { isAvailable: () => true, ensure, waitReady, buildPlaylist, shutdown }; -} -/** - * Per-camera ffmpeg segment producer with a proxy-owned playlist. - * - * Wowza-style upstreams expire HLS sessions fast and irregularly; no single - * long-lived reader survives them (ffmpeg's HLS demuxer exits on a master - * playlist 403 above the -reconnect layer). So ffmpeg is treated as - * disposable: it only writes numbered MPEG-TS segments (-f segment) into a - * per-camera dir, and when it dies it is respawned with -segment_start_number - * continuing from the last file on disk. The playlist is generated here from - * whatever segments exist, with a monotonic media sequence, so respawns are - * invisible to hls.js. -c copy: no re-encode, no quality loss. Processes are - * reaped after IDLE_MS with no client access; dirs are pruned to a window. - */ -export function createHlsRemuxer() { - const OUT_ROOT = path.join(os.tmpdir(), 'gev-hls'); - const IDLE_MS = 60_000; - const READY_TIMEOUT_MS = 20_000; - const SEGMENT_SECONDS = 4; - const STALL_MS = 15_000; - const PLAYLIST_WINDOW = 15; // segments listed - const KEEP_ON_DISK = 24; // segments retained w/ in-flight fetches - const MIN_READY_SEGMENTS = 3; - const RESPAWN_DELAY_MS = 500; - const SEG_RE = /^seg_(\d+)\.ts$/; - /** @type {Map} */ - const active = new Map(); - let available = null; - const isAvailable = () => { - if (available !== null) return available; - try { - const probe = spawn('ffmpeg', ['-version'], { stdio: 'ignore' }); - probe.on('error', () => { available = false; }); - available = true; } catch { - available = false; - } - return available; - }; - /** Sorted ascending list of segment numbers currently on disk */ - const listSegments = async (dir) => { - let names = []; - try { names = await fsp.readdir(dir); } catch { return []; } - return names - .map((n) => SEG_RE.exec(n)) - .filter(Boolean) - .map((m) => Number(m[1])) - .sort((a, b) => a - b); - }; - /** seg number -> duration, from every list_*.csv in the dir. */ - const readDurations = async (dir) => { - const out = new Map(); - let names = []; - try { names = await fsp.readdir(dir); } catch { return out; } - for (const n of names) { - if (!/^list_\d+\.csv$/.test(n)) continue; - const txt = await fsp.readFile(path.join(dir, n), 'utf8').catch(() => ''); - for (const line of txt.split('\n')) { - const [file, start, end] = line.split(','); - const m = file && SEG_RE.exec(file.trim()); - if (m && start && end) out.set(Number(m[1]), Math.max(0.1, Number(end) - Number(start))); + 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?.(); } } - return out; }; - const spawnProducer = (cameraId, entry) => { - if (entry.stopping) return; - entry.boundaries.push(entry.nextSeg); - const proc = spawn('ffmpeg', [ - '-nostdin', '-loglevel', 'warning', - ...(/^rtmp/i.test(entry.upstreamUrl) - ? ['-rtmp_live', 'live', '-rw_timeout', '10000000', - '-fflags', 'nobuffer', '-analyzeduration', '1000000', '-probesize', '500000'] - : ['-user_agent', 'gods-eye-view-cctv-proxy/1.0', - '-reconnect', '1', '-reconnect_on_network_error', '1', '-reconnect_delay_max', '1']), - '-i', entry.upstreamUrl, - '-c', 'copy', - '-f', 'segment', - '-segment_format', 'mpegts', - '-segment_time', String(SEGMENT_SECONDS), - '-segment_start_number', String(entry.nextSeg), - '-segment_list_type', 'csv', - '-segment_list_size', '0', - '-segment_list', path.join(entry.dir, `list_${entry.nextSeg}.csv`), - '-reset_timestamps', '0', - path.join(entry.dir, 'seg_%d.ts'), - ], { stdio: ['ignore', 'ignore', 'pipe'] }); - entry.proc = proc; - proc.stderr?.on('data', (d) => { - const line = String(d).trim(); - if (line) console.warn(`[CCTV:ffmpeg ${cameraId}]`, line); - }); - proc.on('exit', async (code) => { - if (entry.proc !== proc) return; - entry.proc = null; - const segs = await listSegments(entry.dir); - entry.nextSeg = segs.length ? segs[segs.length - 1] + 1 : entry.nextSeg; - if (entry.stopping || active.get(cameraId) !== entry) return; - const idle = Date.now() - entry.lastAccess > IDLE_MS; - console.warn(`[CCTV:ffmpeg ${cameraId}] exited (${code}); next seg ${entry.nextSeg}${idle ? ' - idle, not respawning' : ' - respawning'}`); - if (idle) return; - entry.respawnTimer = setTimeout(() => { - entry.respawnTimer = null; - spawnProducer(cameraId, entry); - }, RESPAWN_DELAY_MS); - }); - }; - const ensure = async (cameraId, upstreamUrl) => { + 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.lastAccess = Date.now(); - if (!entry.proc && !entry.respawnTimer && !entry.stopping) spawnProducer(cameraId, entry); - return entry; - } - const dir = path.join(OUT_ROOT, cameraId.replace(/[^a-zA-Z0-9_-]/g, '_')); - await fsp.mkdir(dir, { recursive: true }); - const existing = await listSegments(dir); - entry = { - proc: null, - dir, - upstreamUrl, - lastAccess: Date.now(), - nextSeg: existing.length ? existing[existing.length - 1] + 1 : 0, - respawnTimer: null, - stopping: false, - boundaries: [], - }; + 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); - spawnProducer(cameraId, entry); + touch(entry, leaseId, true); + entry.polling = poll(entry); return entry; }; - const waitReady = async (entry) => { - const deadline = Date.now() + READY_TIMEOUT_MS; - while (Date.now() < deadline) { - const segs = await listSegments(entry.dir); - if (segs.length >= MIN_READY_SEGMENTS) return true; - await new Promise((r) => setTimeout(r, 250)); + 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; }; - /** - * Build the playlist from the segments on disk and prune old files. - * Newest segment may still be mid-write so it is excluded. - * @returns {Promise} playlist text, or null if not enough segments. - */ - const buildPlaylist = async (entry, cameraId) => { - const segs = await listSegments(entry.dir); - if (segs.length < 2) return null; - const complete = segs.slice(0, -1); - const window = complete.slice(-PLAYLIST_WINDOW); - const keepFrom = segs.length > KEEP_ON_DISK ? segs[segs.length - KEEP_ON_DISK] : -1; - for (const n of segs) { - if (n < keepFrom) fsp.rm(path.join(entry.dir, `seg_${n}.ts`), { force: true }).catch(() => {}); - } - const prefix = `/api/cctv/media/${encodeURIComponent(cameraId)}/`; - const first = window[0]; - const priorBoundaries = entry.boundaries.filter((b) => b > 0 && b <= first).length; - const durations = await readDurations(entry.dir); - const maxDur = Math.max(SEGMENT_SECONDS, ...window.map((n) => durations.get(n) || SEGMENT_SECONDS)); - const lines = [ - '#EXTM3U', - '#EXT-X-VERSION:3', - `#EXT-X-TARGETDURATION:${Math.ceil(maxDur)}`, - `#EXT-X-MEDIA-SEQUENCE:${first}`, - `#EXT-X-DISCONTINUITY-SEQUENCE:${priorBoundaries}`, - ]; - let prev = null; - for (const n of window) { - const isBoundary = n !== first && entry.boundaries.includes(n); - const isGap = prev !== null && n !== prev + 1; - const dur = durations.get(n) || SEGMENT_SECONDS; - if (isBoundary || isGap) lines.push('#EXT-X-DISCONTINUITY'); - lines.push(`#EXTINF:${dur.toFixed(3)},`, `${prefix}seg_${n}.ts`); - prev = n; + 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 stopEntry = (entry) => { - entry.stopping = true; - if (entry.respawnTimer) { clearTimeout(entry.respawnTimer); entry.respawnTimer = null; } - if (entry.proc) entry.proc.kill('SIGTERM'); - fsp.rm(entry.dir, { recursive: true, force: true }).catch(() => {}); + 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 sweep = setInterval(async () => { - const now = Date.now(); - for (const [id, entry] of active) { - if (now - entry.lastAccess > IDLE_MS) { stopEntry(entry); active.delete(id); continue; } - if (entry.proc) { - const segs = await listSegments(entry.dir); - const newest = segs.length ? segs[segs.length - 1] : -1; - if (newest !== entry.watchSeg) { entry.watchSeg = newest; entry.watchAt = now; } - else if (now - (entry.watchAt || now) > STALL_MS) { - console.warn(`[CCTV:ffmpeg ${id}] no new segment for ${STALL_MS / 1000}s — killing producer`); - entry.watchAt = now; - entry.proc.kill('SIGKILL'); - } - } - } - }, 5_000); - sweep.unref?.(); - const shutdown = () => { - clearInterval(sweep); - for (const [, entry] of active) stopEntry(entry); - active.clear(); + 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 { isAvailable, ensure, waitReady, buildPlaylist, shutdown }; -} \ No newline at end of file + return { ensure, waitReady, buildPlaylist, getSegment, stop, release, shutdown, + stats: () => ({ sessions: active.size, bytes: [...active.values()].reduce((total, e) => total + e.bytes, 0) }) }; +} diff --git a/src/data/cctvHlsStream.test.mjs b/src/data/cctvHlsStream.test.mjs new file mode 100644 index 000000000..6d9e08d5d --- /dev/null +++ b/src/data/cctvHlsStream.test.mjs @@ -0,0 +1,128 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { createHlsPuller, fetchHlsBytes, parseHlsMedia, HLS_LIMITS } from '../../server/providers/cctv/stream.js'; +const playlist = '#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:12\n#EXTINF:2,\na.ts\n#EXTINF:2,\nb.ts\n#EXTINF:2,\nc.ts\n'; +const base = 'https://camera.example/live/list.m3u8'; + +test('playlist parser uses sequence, refuses escaping and unsupported references', () => { + assert.deepEqual(parseHlsMedia(playlist, base).map((s) => s.seq), [12, 13, 14]); + for (const bad of [playlist.replace('a.ts', 'https://evil.example/a.ts'), playlist.replace('a.ts', '//evil.example/a.ts'), '#EXTM3U\n#EXT-X-KEY:METHOD=AES-128\n' + playlist, playlist.replace('12', '-1'), playlist.replace('2,', 'Infinity,')]) assert.throws(() => parseHlsMedia(bad, base)); +}); + +test('downloads reject redirect responses, oversized declared and chunked bodies', async () => { + let init; + await assert.rejects(fetchHlsBytes(base, { maxBytes: 5, fetchImpl: async (_url, options) => { init = options; return new Response('', { status: 302, headers: { Location: 'https://evil.example' } }); } })); + assert.equal(init.redirect, 'error'); + await assert.rejects(fetchHlsBytes(base, { maxBytes: 5, fetchImpl: async () => new Response('abcdef') })); + await assert.rejects(fetchHlsBytes(base, { maxBytes: 5, fetchImpl: async () => new Response('a', { headers: { 'content-length': '100' } }) })); +}); + +test('one session is reserved before await; disk-free cache and capacity remain bounded', async () => { + const manager = createHlsPuller({ limits: { ...HLS_LIMITS, sessions: 1, segmentBytes: 4, sessionBytes: 6, segments: 2, pollMs: 100000 }, fetchImpl: async (url) => new Response(url.endsWith('.m3u8') ? playlist : 'abc') }); + const [a, b] = await Promise.all([manager.ensure('a', base), manager.ensure('a', base)]); + assert.equal(a, b); + await assert.rejects(manager.ensure('b', base)); + assert.equal(await manager.waitReady(a), true); + assert.deepEqual(manager.stats(), { sessions: 1, bytes: 6 }); + const text = await manager.buildPlaylist(a, 'a'); + assert.match(text, /seg_0\.ts\?session=/); + assert.equal(manager.getSegment('a', 'stale-token', 0), null); + assert.equal(manager.getSegment('a', a.token, 0).length, 3); + manager.stop('a', 'stale-token'); + assert.equal(manager.stats().sessions, 1); + manager.stop('a', a.token); + assert.deepEqual(manager.stats(), { sessions: 0, bytes: 0 }); + await manager.shutdown(); +}); + +test('shutdown cancels in-flight downloads and late responses cannot refill cache', async () => { + let observed; + const manager = createHlsPuller({ fetchImpl: async (_url, { signal }) => { observed = signal; return new Promise((resolve, reject) => signal.addEventListener('abort', () => reject(new Error('cancelled')), { once: true })); } }); + const entry = await manager.ensure('a', base); + await manager.shutdown(); + assert.equal(observed.aborted, true); + assert.equal(entry.stopping, true); + assert.deepEqual(manager.stats(), { sessions: 0, bytes: 0 }); + await assert.rejects(manager.ensure('a', base)); +}); + +test('idle cleanup stops all polling without a background sweep', async () => { + let calls = 0; + const manager = createHlsPuller({ limits: { ...HLS_LIMITS, idleMs: 15, pollMs: 100000 }, fetchImpl: async (url) => { calls++; return new Response(url.endsWith('.m3u8') ? playlist : 'abc'); } }); + await manager.ensure('a', base); + await new Promise((r) => setTimeout(r, 50)); + assert.deepEqual(manager.stats(), { sessions: 0, bytes: 0 }); + const stoppedCalls = calls; + await new Promise((r) => setTimeout(r, 30)); + assert.equal(calls, stoppedCalls); + await manager.shutdown(); +}); + +test('agency sequence rollback creates a monotonic local discontinuity', async () => { + let current = playlist; + const manager = createHlsPuller({ limits: { ...HLS_LIMITS, pollMs: 5 }, fetchImpl: async (url) => new Response(url.endsWith('.m3u8') ? current : 'abc') }); + const entry = await manager.ensure('a', base); + await manager.waitReady(entry); + const before = await manager.buildPlaylist(entry, 'a'); + assert.match(before, /seg_0\.ts/); + current = playlist.replace('SEQUENCE:12', 'SEQUENCE:0'); + await new Promise((r) => setTimeout(r, 30)); + const after = await manager.buildPlaylist(entry, 'a'); + assert.match(after, /#EXT-X-DISCONTINUITY\n#EXTINF:2\.000,\n\/api\/cctv\/media\/a\/seg_3\.ts/); + await manager.shutdown(); +}); + +test('upstream discontinuity tags survive the media parser', () => { + const parsed = parseHlsMedia(playlist.replace('a.ts', 'a.ts\n#EXT-X-DISCONTINUITY'), base); + assert.equal(parsed[0].discontinuity, false); + assert.equal(parsed[1].discontinuity, true); +}); + +test('a reused agency sequence with changed segment URI cannot remain stale', async () => { + let current = playlist; + const manager = createHlsPuller({ limits: { ...HLS_LIMITS, pollMs: 5 }, fetchImpl: async (url) => new Response(url.endsWith('.m3u8') ? current : 'abc') }); + const entry = await manager.ensure('a', base); + await manager.waitReady(entry); + current = playlist.replaceAll('.ts', '.ts?generation=2'); + await new Promise((r) => setTimeout(r, 30)); + assert.match(await manager.buildPlaylist(entry, 'a'), /#EXT-X-DISCONTINUITY\n#EXTINF:2\.000,\n\/api\/cctv\/media\/a\/seg_3\.ts/); + await manager.shutdown(); +}); + +test('two consumers share downloads but release and abandoned expiry are independent', async () => { + let downloads = 0; + const manager = createHlsPuller({ limits: { ...HLS_LIMITS, pollMs: 100000, leasesPerSession: 2 }, fetchImpl: async (url) => { + downloads++; + return new Response(url.endsWith('.m3u8') ? playlist : 'abc'); + } }); + const [a, b] = await Promise.all([manager.ensure('camera', base, 'viewer-a'), manager.ensure('camera', base, 'viewer-b')]); + assert.equal(a, b); + await manager.waitReady(a); + assert.equal(downloads, 4); // One manifest and three segments, not per consumer. + assert.equal(a.leases.size, 2); + await assert.rejects(manager.ensure('camera', base, 'viewer-c')); + manager.release('camera', 'viewer-a'); + assert.equal(manager.stats().sessions, 1); + assert.equal(manager.getSegment('camera', a.token, 0, 'viewer-a'), null); + assert.equal(manager.getSegment('camera', a.token, 0, 'viewer-b').length, 3); + assert.match(await manager.buildPlaylist(b, 'camera', 'viewer-b'), /lease=viewer-b/); + manager.release('camera', 'viewer-a'); // Duplicate/late release cannot stop B. + assert.equal(b.controller.signal.aborted, false); + manager.release('camera', 'viewer-b'); + assert.equal(b.controller.signal.aborted, true); + assert.deepEqual(manager.stats(), { sessions: 0, bytes: 0 }); + await manager.shutdown(); +}); + +test('an abandoned consumer expires while a renewed consumer keeps the session', async () => { + const manager = createHlsPuller({ limits: { ...HLS_LIMITS, idleMs: 80, pollMs: 100000 }, fetchImpl: async (url) => new Response(url.endsWith('.m3u8') ? playlist : 'abc') }); + const entry = await manager.ensure('camera', base, 'abandoned'); + await manager.ensure('camera', base, 'active'); + await new Promise((r) => setTimeout(r, 50)); + await manager.ensure('camera', base, 'active'); + await new Promise((r) => setTimeout(r, 50)); + assert.equal(entry.leases.has('abandoned'), false); + assert.equal(entry.leases.has('active'), true); + assert.equal(entry.controller.signal.aborted, false); + await manager.shutdown(); +}); diff --git a/src/data/dataCredits.js b/src/data/dataCredits.js index 29a76a15e..b29bafc9d 100644 --- a/src/data/dataCredits.js +++ b/src/data/dataCredits.js @@ -136,6 +136,10 @@ export const DATA_CREDITS = [ 'CCTV cameras & frames (Texas): ' + 'Texas Department of Transportation (courtesy)', }, + { + key: 'deldot-cctv', + html: 'CCTV live video (Delaware): DelDOT — Delaware Department of Transportation', + }, { key: 'caltrans-cctv', html: diff --git a/src/layers/cctv/presentation.js b/src/layers/cctv/presentation.js index 13fa751bd..7daa8c1e5 100644 --- a/src/layers/cctv/presentation.js +++ b/src/layers/cctv/presentation.js @@ -83,7 +83,7 @@ export function createPresentation({ mountHeightM: camera.mountHeightM, active: isActive, feedType: camera.feedType, - isVideo: parts.model.isVideoFeedType(camera.feedType), + isVideo: parts.model.isVideoFeedType(camera.feedType) && record.projection?.mode !== 'image', sourceKind: health?.sourceKind || camera.sourceKind || diff --git a/src/layers/cctv/projection.js b/src/layers/cctv/projection.js index 1d9b54fc7..edac3a66a 100644 --- a/src/layers/cctv/projection.js +++ b/src/layers/cctv/projection.js @@ -1,3 +1,4 @@ +import { attachCctvVideo } from './videoPlayback.js'; import * as Cesium from 'cesium'; import { CCTV_PROJECTION_OVERLAY_SOURCE_ID, @@ -168,79 +169,6 @@ export function createProjection({ * @returns {Object|null} Projection runtime, or null if no viewer. */ - async function attachVideoSource(runtime, url, feedType) { - const video = runtime?.video; - if (!video) return; - if (feedType !== 'hls') { - video.src = url; - return; - } - try { - const { default: Hls } = await import('hls.js'); - if (runtime.disposed || runtime.video !== video) return; - if (!Hls.isSupported()) { - video.src = url; - return; - } - const hls = new Hls({ - lowLatencyMode: false, - enableWorker: true, - liveSyncDurationCount: 5, - liveMaxLatencyDurationCount: 6, - maxBufferLength: 60, - backBufferLength: 30, - }); - hls.on(Hls.Events.ERROR, (_event, data) => { - if (!data?.fatal) return; - if (data.type === Hls.ErrorTypes.NETWORK_ERROR) { - // reread master for a fresh session - console.warn( - '[Data:CCTV] hls network error, reloading source:', - data.details, - ); - setTimeout(() => { - if (runtime.hls === hls) hls.loadSource(url); - }, 2000); - return; - } - if (data.type === Hls.ErrorTypes.MEDIA_ERROR) { - hls.recoverMediaError(); - return; - } - console.warn('[Data:CCTV] hls fatal:', data.type, data.details); - hls.destroy(); - if (runtime.hls === hls) runtime.hls = null; - }); - hls.loadSource(url); - hls.attachMedia(video); - runtime.hls = hls; - - // Rate governor: DelDOT cameras deliver ~52 s of content per 60 s wall - // time. Hold playbackRate by buffer depth so the stream never starves; - // cameras with honest clocks sit at 1.0 and never leave it. - const governor = setInterval(() => { - if (runtime.video !== video || !runtime.hls) { - clearInterval(governor); - return; - } - const b = video.buffered; - if (!b.length) return; - const ahead = b.end(b.length - 1) - video.currentTime; - let rate = 1.0; - if (ahead < 15) rate = 0.75; - else if (ahead < 20) rate = 0.85; - else if (ahead < 30) rate = 0.93; - else if (ahead > 40) rate = 1.05; - if (Math.abs(video.playbackRate - rate) > 0.01) - video.playbackRate = rate; - }, 1000); - runtime.rateGovernor = governor; - } catch (error) { - console.warn('[Data:CCTV] hls.js unavailable:', error?.message || error); - video.src = url; - } - } - function createProjectionRuntime(record) { if (!layerState._viewer) return null; const canvas = document.createElement('canvas'); @@ -291,9 +219,6 @@ export function createProjection({ video.playsInline = true; video.crossOrigin = 'anonymous'; video.preload = 'auto'; - video.addEventListener('canplay', () => { - video.play().catch(() => {}); - }); // Cesium sizes the video texture from the element's width/height // attributes at first upload. Set them from the real stream dimensions // and rebind on any resolution change (camera switch, adaptive source). @@ -308,12 +233,20 @@ export function createProjection({ video.addEventListener('loadedmetadata', bindVideoTexture); video.addEventListener('resize', bindVideoTexture); runtime.video = video; - runtime.hls = null; - attachVideoSource( - runtime, - parts.frames.mediaUrlFor(record.camera), - feedType, - ); + runtime.playback = attachCctvVideo(video, parts.frames.mediaUrlFor(record.camera), feedType, { + onFailure: () => { + if (runtime.disposed) return; + runtime.video = null; + runtime.mode = 'image'; + runtime.image = new Image(); + runtime.image.decoding = 'async'; + runtime.image.onload = () => { runtime.imageLoading = false; runtime.imageReady = true; runtime.imageStamp = Date.now(); }; + runtime.image.onerror = () => { runtime.imageLoading = false; runtime.imageReady = false; }; + runtime.planeMaterial.image = runtime.canvas; + parts.frames.refreshProjectionImage(record, true); + parts.presentation.notifyListeners(); + }, + }); } else { const img = new Image(); img.decoding = 'async'; @@ -361,7 +294,7 @@ export function createProjection({ function ensureProjectionRuntime(record) { if (!record) return null; - if (record.projection) return record.projection; + if (record.projection && !record.projection.disposed) return record.projection; const runtime = createProjectionRuntime(record); record.projection = runtime; if (runtime) { @@ -377,16 +310,11 @@ export function createProjection({ */ function destroyProjectionRuntime(runtime) { - if (!runtime) return; + if (!runtime || runtime.disposed) return; runtime.disposed = true; - if (runtime.hls) { - runtime.hls.destroy(); - runtime.hls = null; - } - if (runtime.rateGovernor) { - clearInterval(runtime.rateGovernor); - runtime.rateGovernor = null; - } + runtime.playback?.dispose(); + runtime.playback = null; + if (runtime.image) { runtime.image.onload = null; runtime.image.onerror = null; runtime.image.src = ''; } if (runtime.video) { runtime.video.pause(); runtime.video.removeAttribute('src'); @@ -423,10 +351,7 @@ export function createProjection({ const active = parts.selection.getActiveRecord(); if (layerState._enabled && layerState._showProjection && active) { ensureProjectionRuntime(active); - if (active.projection?.video) { - active.projection.video.play().catch(() => {}); - } - if (active.projection) { + if (active.projection && !active.projection.video) { parts.frames.drawProjectionFrame(active); parts.frames.refreshProjectionTextures(active); } @@ -460,6 +385,7 @@ export function createProjection({ * @returns {HTMLVideoElement|null} */ function getActiveVideoElement() { + if (!layerState._enabled) return null; return parts.selection.getActiveRecord()?.projection?.video || null; } @@ -468,12 +394,14 @@ export function createProjection({ if (!record.projection?.video) continue; if ( record.camera.id === activeId && - layerState._enabled && - layerState._showProjection + layerState._enabled ) { record.projection.video.play().catch(() => {}); } else { - record.projection.video.pause(); + const runtime = record.projection; + destroyProjectionRuntime(runtime); + record.projection = null; + layerState._projectionEntities = layerState._projectionEntities.filter((entry) => entry !== runtime); } } } diff --git a/src/layers/cctv/rendering.js b/src/layers/cctv/rendering.js index 8c4dbb9be..13bb2de22 100644 --- a/src/layers/cctv/rendering.js +++ b/src/layers/cctv/rendering.js @@ -215,7 +215,7 @@ export function createRendering({ // (set at creation) — see the field-test far-zoom submerge fix there. } - if (layerState._enabled && layerState._showProjection && isActive) { + if (layerState._enabled && isActive && (layerState._showProjection || parts.model.isVideoFeedType(record.camera.feedType))) { parts.projection.ensureProjectionRuntime(record); } // One live plane in the world at a time (§2c): only the active camera's diff --git a/src/layers/cctv/videoPlayback.js b/src/layers/cctv/videoPlayback.js new file mode 100644 index 000000000..a43f2dc2e --- /dev/null +++ b/src/layers/cctv/videoPlayback.js @@ -0,0 +1,81 @@ +/** One decoder per active camera; both surfaces consume this video element. */ +export function attachCctvVideo(video, url, feedType, { + loadHls = () => import('hls.js'), + onFailure = () => {}, + fetchImpl = globalThis.fetch, +} = {}) { + let disposed = false; + let hls = null; + const leaseId = feedType === 'hls' ? globalThis.crypto.randomUUID() : null; + const mediaUrl = feedType === 'hls' ? `${url}${url.includes('?') ? '&' : '?'}lease=${leaseId}` : url; + let timer = null; + let startup = null; + let governor = null; + let retries = 0; + const release = () => { + if (feedType === 'hls') { + void fetchImpl?.(mediaUrl, { method: 'DELETE', keepalive: true }).catch(() => {}); + } + }; + const dispose = () => { + if (disposed) return; + disposed = true; + clearTimeout(timer); + clearTimeout(startup); + clearInterval(governor); + hls?.destroy(); + hls = null; + video.removeEventListener('canplay', play); + video.removeEventListener('error', fail); + video.pause(); + video.removeAttribute('src'); + video.load(); + release(); + }; + const fail = () => { + if (disposed) return; + dispose(); + onFailure(); + }; + const play = () => { + if (disposed) return; + clearTimeout(startup); + video.play().catch(() => {}); + }; + video.addEventListener('canplay', play); + video.addEventListener('error', fail); + startup = setTimeout(fail, 30000); + const ready = (async () => { + if (feedType !== 'hls') { video.src = mediaUrl; return; } + try { + const { default: Hls } = await loadHls(); + if (disposed) return; + if (!Hls.isSupported()) { + if (video.canPlayType('application/vnd.apple.mpegurl')) video.src = mediaUrl; + else fail(); + return; + } + hls = new Hls({ enableWorker: true, maxBufferLength: 24, maxMaxBufferLength: 30, backBufferLength: 0, maxBufferSize: 16 * 1024 * 1024, liveSyncDurationCount: 3 }); + hls.on(Hls.Events.ERROR, (_event, data) => { + if (disposed || !data?.fatal) return; + if (++retries > 2) { fail(); return; } + clearTimeout(timer); + timer = setTimeout(() => { + if (disposed) return; + if (data.type === Hls.ErrorTypes.MEDIA_ERROR) hls.recoverMediaError(); + else hls.loadSource(mediaUrl); + }, 2000); + }); + governor = setInterval(() => { + if (disposed || !video.buffered?.length) return; + const ahead = video.buffered.end(video.buffered.length - 1) - video.currentTime; + // Some agency encoders publish less media time than wall time. Bound + // correction rather than repeatedly draining the live buffer at 1x. + video.playbackRate = ahead < 6 ? 0.8 : ahead < 12 ? 0.9 : ahead > 24 ? 1.05 : 1; + }, 1000); + hls.attachMedia(video); + hls.loadSource(mediaUrl); + } catch { fail(); } + })(); + return { dispose, ready }; +} diff --git a/src/layers/cctv/videoPlayback.test.mjs b/src/layers/cctv/videoPlayback.test.mjs new file mode 100644 index 000000000..6a8d41c4a --- /dev/null +++ b/src/layers/cctv/videoPlayback.test.mjs @@ -0,0 +1,45 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { attachCctvVideo } from './videoPlayback.js'; +function video() { + const v = new EventTarget(); + Object.assign(v, { src: '', pause() {}, load() {}, removeAttribute() {}, play: () => Promise.resolve() }); + return v; +} +test('switch during lazy import never starts the stale decoder', async () => { + let resolve; + let constructed = 0; + class Hls { constructor() { constructed++; } static isSupported() { return true; } } + const source = video(); + const playback = attachCctvVideo(source, '/api/cctv/media/a', 'hls', { loadHls: () => new Promise((r) => { resolve = r; }) }); + playback.dispose(); + resolve({ default: Hls }); + await playback.ready; + assert.equal(constructed, 0); + assert.equal(source.src, ''); +}); +test('missing decoder uses honest failure fallback once', async () => { + let failures = 0; + const playback = attachCctvVideo(video(), '/api/cctv/media/a', 'hls', { loadHls: async () => { throw new Error('unavailable'); }, onFailure: () => failures++ }); + await playback.ready; + playback.dispose(); + assert.equal(failures, 1); +}); + +test('native HLS releases its client lease without response-header access', async () => { + const source = video(); + source.canPlayType = () => 'probably'; + const releases = []; + const playback = attachCctvVideo(source, '/api/cctv/media/a', 'hls', { + loadHls: async () => ({ default: { isSupported: () => false } }), + fetchImpl: async (url, init) => { releases.push({ url, init }); }, + }); + await playback.ready; + assert.match(source.src, /\/api\/cctv\/media\/a\?lease=[a-f0-9-]{36}$/); + const requested = source.src; + playback.dispose(); + playback.dispose(); + assert.equal(releases.length, 1); + assert.equal(releases[0].url, requested); + assert.equal(releases[0].init.method, 'DELETE'); +}); diff --git a/src/ui/cctvControls.js b/src/ui/cctvControls.js index 75ab0be98..10594ea76 100644 --- a/src/ui/cctvControls.js +++ b/src/ui/cctvControls.js @@ -38,6 +38,11 @@ export class CctvControls { this._calibrationEdit = null; this._actionGeneration = 0; this._initCctvPanel(); + if (this._cctvVideo && typeof MutationObserver !== 'undefined') { + this._videoVisibilityObserver = new MutationObserver(() => this._renderCctvState(this._cctvState)); + if (this._cctvPanel) this._videoVisibilityObserver.observe(this._cctvPanel, { attributes: true, attributeFilter: ['class', 'hidden'] }); + this.listen(document, 'visibilitychange', () => this._renderCctvState(this._cctvState)); + } } listen(target, type, handler, options = {}) { target?.addEventListener(type, handler, { @@ -99,6 +104,8 @@ export class CctvControls { destroy() { if (this.destroyed) return; this.destroyed = true; + this._cctvVideoSurface?.stop(); + this._videoVisibilityObserver?.disconnect(); this._actionGeneration++; this.listeners.abort(); this._cctvUnsubscribe?.(); diff --git a/src/ui/cctvPresentation.js b/src/ui/cctvPresentation.js index 6ea2aa318..01fc3b1f6 100644 --- a/src/ui/cctvPresentation.js +++ b/src/ui/cctvPresentation.js @@ -1,3 +1,4 @@ +import { createCctvVideoSurface } from './cctvVideo.js'; export function _calBadgeLabel(badge) { switch (badge) { case 'calibrated': @@ -164,42 +165,18 @@ export function _renderCctvState(state) { } } - // Live video: for HLS cameras, paint the active camera's decoded