import { Readable } from 'node:stream'; import { hashSeed, escapeXml } from './normalize.js'; import { CCTV_FRAME_FETCH_TIMEOUT_MS, CCTV_FRAME_MAX_BODY_BYTES, CCTV_MEDIA_FETCH_TIMEOUT_MS, CCTV_MEDIA_MAX_BODY_BYTES, } from './constants.js'; /** * Generate a synthetic SVG billboard image for a CCTV camera placeholder. * * Produces a 960x540 SVG with a deterministic gradient (hue derived from * camera ID hash), scanline overlay, HUD-style grid, and text labels * showing camera name, city, ID, status, and current timestamp. Used * when no upstream image or Street View fallback is available. * * @param {object} opts * @param {string} opts.cameraId * @param {string} opts.label * @param {string} [opts.city] * @param {string} [opts.status] * @returns {string} SVG markup string. */ export function buildSyntheticCctvSvg({ cameraId, label, city, status }) { const seed = hashSeed(`${cameraId}:${label}:${city}`); const hue = seed % 360; const hue2 = (hue + 46) % 360; const now = new Date(); const ts = now.toISOString().replace('T', ' ').replace('Z', 'Z').slice(0, 20); const safeLabel = escapeXml(label); const safeCity = escapeXml(city || 'GLOBAL GRID'); const safeId = escapeXml(cameraId); const safeStatus = escapeXml(status || 'SYNTHETIC'); return ` CCTV FEED PLACEHOLDER ${safeLabel} · ${safeCity} ${safeId} ${escapeXml(ts)} ${safeStatus} `.trim(); } /** * Coerce a fetch() response body to a Node.js Readable stream. * * Handles both Node-native streams (.pipe) and web ReadableStreams (.getReader). * * @param {ReadableStream|NodeJS.ReadableStream|null} body * @returns {import('stream').Readable|null} */ export function toReadable(body) { if (!body) return null; if (typeof body.pipe === 'function') return body; if (typeof body.getReader === 'function') { return Readable.fromWeb(body); } return null; } /** * Pipe an upstream fetch Response (image or video) to the client HTTP response. * * Forwards Content-Type, Content-Length, Content-Range, Accept-Ranges, and * Cache-Control headers from the upstream. Falls back to buffered arrayBuffer * if the body is not streamable. * * @param {import('http').ServerResponse} res * @param {Response} upstream - fetch() Response object. * @param {object} [opts] * @param {string} [opts.sourceHeader='upstream'] - Value for X-CCTV-Source header. */ export async function proxyMediaResponse( res, upstream, { sourceHeader = 'upstream' } = {}, ) { const contentType = upstream.headers.get('content-type') || 'application/octet-stream'; const cacheControl = upstream.headers.get('cache-control') || 'no-store'; const contentLength = upstream.headers.get('content-length'); const contentRange = upstream.headers.get('content-range'); const acceptRanges = upstream.headers.get('accept-ranges'); const headers = { 'Content-Type': contentType, 'Cache-Control': cacheControl, 'X-CCTV-Source': sourceHeader, }; if (contentLength) headers['Content-Length'] = contentLength; if (contentRange) headers['Content-Range'] = contentRange; if (acceptRanges) headers['Accept-Ranges'] = acceptRanges; // Cheap defense: reject an upstream that DECLARES an oversized fixed body. // Live MJPEG/HLS streams are unbounded by design and send no content-length, // so they pipe normally (piping streams to the client, never buffering). if ( Number.isFinite(Number(contentLength)) && Number(contentLength) > CCTV_MEDIA_MAX_BODY_BYTES ) { res.writeHead(502, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', }); res.end(JSON.stringify({ error: 'Upstream media exceeds size cap' })); try { await upstream.body?.cancel(); } catch { /* no-op */ } return; } res.writeHead(upstream.status, headers); const stream = toReadable(upstream.body); if (!stream) { const buf = Buffer.from(await upstream.arrayBuffer()); res.end(buf); return; } stream.on('error', () => { if (!res.writableEnded) res.end(); }); stream.pipe(res); } /** Read a snapshot incrementally, retaining at most maxBytes of owned chunks. */ async function readCappedResponseBytes(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 null; } if (!upstream.body) return null; const chunks = []; let total = 0; // Every retained chunk is an OWNED copy: a chunk can be a small view over a // much larger backing ArrayBuffer, and keeping the view would retain that // whole allocation while the byte accounting only counted the view. const keep = (chunk) => { total += chunk.byteLength; if (total > maxBytes) return false; chunks.push(Buffer.from(chunk)); return true; }; if (typeof upstream.body[Symbol.asyncIterator] === 'function') { for await (const chunk of upstream.body) { if (!keep(chunk)) { try { await upstream.body.cancel(); } catch { /* no-op */ } return null; } } return Buffer.concat(chunks, total); } // No async iterator: stream through a reader so the cap still applies while // reading. A body that cannot be streamed at all is refused rather than // buffered uncapped — the helper's whole contract is the cap. const reader = typeof upstream.body.getReader === 'function' ? upstream.body.getReader() : null; if (!reader) return null; try { for (;;) { const { done, value } = await reader.read(); if (done) break; if (!keep(value)) { try { await reader.cancel(); } catch { /* already closed */ } return null; } } return Buffer.concat(chunks, total); } finally { reader.releaseLock(); } } /** Open registered media within a header deadline; leave timely live bodies running. */ export async function fetchCctvMediaUpstream( url, { headers = {}, fetchImpl = fetch, timeoutMs = CCTV_MEDIA_FETCH_TIMEOUT_MS, } = {}, ) { const controller = new AbortController(); const timeoutId = setTimeout(() => controller.abort(), timeoutMs); try { return await fetchImpl(url, { headers, signal: controller.signal }); } finally { clearTimeout(timeoutId); } } /** * Fetch one upstream CCTV image within the frame-refresh budget. * * A timeout is treated like every other upstream miss so the caller can * continue through the Street View and synthetic fallback chain. `fetchImpl` * and `timeoutMs` are injectable only to keep the timeout contract unit-testable. * * @param {string} url - Server-registered upstream image URL. * @param {object} [options] * @param {typeof fetch} [options.fetchImpl=fetch] - Fetch implementation. * @param {number} [options.timeoutMs=CCTV_FRAME_FETCH_TIMEOUT_MS] - Abort timeout. * @param {number} [options.maxBytes=CCTV_FRAME_MAX_BODY_BYTES] - Snapshot byte cap. * @returns {Promise<{ok:true,body:Buffer,contentType:string}|null>} */ export async function fetchCctvImageFromUpstream( url, { fetchImpl = fetch, timeoutMs = CCTV_FRAME_FETCH_TIMEOUT_MS, maxBytes = CCTV_FRAME_MAX_BODY_BYTES, } = {}, ) { if (!url || !/^https?:\/\//i.test(url)) return null; const controller = new AbortController(); const timeoutId = setTimeout(() => { controller.abort( new DOMException('CCTV upstream frame fetch timed out', 'TimeoutError'), ); }, timeoutMs); try { const upstream = await fetchImpl(url, { headers: { 'User-Agent': 'gods-eye-view-cctv-proxy/1.0' }, signal: controller.signal, }); const contentType = upstream.headers.get('content-type') || ''; if (!upstream.ok || !contentType.startsWith('image/')) { controller.abort(); return null; } const body = await readCappedResponseBytes(upstream, maxBytes); if (!body) return null; return { ok: true, body, contentType }; } catch { return null; } finally { clearTimeout(timeoutId); controller.abort(); } }