benmichael/gods-eye-view
1
0 Fork 0
mirror of https://github.com/bilawalsidhu/gods-eye-view.git synced 2026-11-05 19:04:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

style(cctv): format live video integration

Browse Source
This commit is contained in:
Bilawal Sidhu committed 2026-09-15 10:54:40 -05:00
1 parent 3571284782
commit 6f40fff09e
13 files changed
+564 -144
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/providers/cctv.js
+88 -21
View File
@@ -132,7 +132,6 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
const installMiddleware = (server) => {
server.httpServer?.on('close', () => {
puller.shutdown();
});
server.middlewares.use('/api/cctv', async (req, res) => {
try {
@@ -201,48 +200,112 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
}
if (url.pathname.startsWith('/media/')) {
const match = /^\/media\/([^/]+)(?:\/(seg_(\d+)\.ts))?$/.exec(url.pathname);
if (!match) { res.writeHead(404); res.end(); return; }
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');
const leaseId = url.searchParams.get('lease');
if (feedType === 'hls' && !/^[a-f0-9-]{36}$/i.test(leaseId || '')) { res.writeHead(400); res.end(); return; }
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;
res.writeHead(204);
res.end();
return;
}
if (req.method !== 'GET') {
res.writeHead(405);
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)) {
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' }));
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 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); };
const cancelPending = () => {
if (entry) puller.release(cameraId, leaseId);
};
try {
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) {
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 });
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 {
setHealth(cameraId, { status: 'degraded', sourceKind: 'fallback', label: source?.provider || 'Configured source', message: 'Live HLS unavailable' });
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.writeHead(503, {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'Retry-After': '2',
});
res.end(JSON.stringify({ error: 'Live stream unavailable' }));
}
} finally {
@@ -250,7 +313,11 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
}
return;
}
if (match[2] || req.method !== 'GET') { res.writeHead(404); res.end(); return; }
if (match[2] || req.method !== 'GET') {
res.writeHead(404);
res.end();
return;
}
if (!mediaUrl || !/^https?:\/\//i.test(mediaUrl)) {
setHealth(cameraId, {
server/providers/cctv/sources.js
+12 -2
View File
@@ -1626,8 +1626,18 @@ export async function loadDelDOTSourcesFromOpenData() {
// 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;
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();
server/providers/cctv/stream.js
+150 -40
View File
@@ -15,7 +15,15 @@ export const HLS_LIMITS = Object.freeze({
});
/** Refuse redirects; count actual streamed bytes, including chunked responses. */
export async function fetchHlsBytes(url, { fetchImpl = fetch, signal, maxBytes, timeoutMs = HLS_LIMITS.timeoutMs } = {}) {
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();
@@ -25,11 +33,19 @@ export async function fetchHlsBytes(url, { fetchImpl = fetch, signal, maxBytes,
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))
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');
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 = [];
@@ -56,14 +72,22 @@ export async function fetchHlsBytes(url, { fetchImpl = fetch, signal, maxBytes,
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)
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))
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;
@@ -73,16 +97,20 @@ export function parseHlsMedia(text, base, limit = HLS_LIMITS.segments) {
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');
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');
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');
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');
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;
@@ -91,7 +119,10 @@ export function parseHlsMedia(text, base, limit = HLS_LIMITS.segments) {
return segments.slice(-limit);
}
export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {}) {
export function createHlsPuller({
fetchImpl = fetch,
limits = HLS_LIMITS,
} = {}) {
const active = new Map();
let closed = false;
const stop = (cameraId, token) => {
@@ -118,34 +149,50 @@ export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {})
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');
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 = 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 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');
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');
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;
});
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;
@@ -158,29 +205,45 @@ export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {})
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)) {
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');
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.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);
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; }
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 = setTimeout(() => {
entry.polling = poll(entry);
}, limits.pollMs);
entry.timer.unref?.();
}
}
@@ -189,11 +252,33 @@ export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {})
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; }
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 };
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);
@@ -211,18 +296,32 @@ export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {})
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}`];
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)}`);
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;
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;
@@ -233,6 +332,17 @@ export function createHlsPuller({ fetchImpl = fetch, limits = HLS_LIMITS } = {})
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) }) };
return {
ensure,
waitReady,
buildPlaylist,
getSegment,
stop,
release,
shutdown,
stats: () => ({
sessions: active.size,
bytes: [...active.values()].reduce((total, e) => total + e.bytes, 0),
}),
};
}