gods-eye-view/server/providers/cctv/media.js

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,
  NSW_IMAGE_ORIGIN,
  NSW_IMAGE_USER_AGENT,
} 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 `
<svg xmlns="http://www.w3.org/2000/svg" width="960" height="540" viewBox="0 0 960 540">
  <defs>
    <linearGradient id="bg" x1="0" y1="0" x2="1" y2="1">
      <stop offset="0%" stop-color="hsl(${hue}, 35%, 10%)" />
      <stop offset="60%" stop-color="hsl(${hue2}, 42%, 6%)" />
      <stop offset="100%" stop-color="#020509" />
    </linearGradient>
    <radialGradient id="flare" cx="0.22" cy="0.24" r="0.78">
      <stop offset="0%" stop-color="hsla(${hue2}, 100%, 65%, 0.35)" />
      <stop offset="100%" stop-color="hsla(${hue2}, 100%, 40%, 0)" />
    </radialGradient>
    <pattern id="scan" width="8" height="8" patternUnits="userSpaceOnUse">
      <rect width="8" height="8" fill="transparent" />
      <rect y="0" width="8" height="1" fill="rgba(255,255,255,0.08)" />
      <rect y="4" width="8" height="1" fill="rgba(255,255,255,0.05)" />
    </pattern>
  </defs>
  <rect width="960" height="540" fill="url(#bg)" />
  <rect width="960" height="540" fill="url(#flare)" />
  <rect width="960" height="540" fill="url(#scan)" />
  <g stroke="rgba(123,233,255,0.25)" stroke-width="1" fill="none">
    <path d="M60 460 Q300 300 520 420 T900 320" />
    <path d="M100 160 Q340 40 620 130 T920 90" />
    <path d="M20 280 Q220 230 390 270 T760 250" />
  </g>
  <g fill="none" stroke="rgba(180,248,255,0.2)" stroke-width="1">
    <rect x="70" y="80" width="820" height="380" rx="8" />
    <line x1="70" y1="270" x2="890" y2="270" />
    <line x1="480" y1="80" x2="480" y2="460" />
  </g>
  <g fill="#9cefff" font-family="JetBrains Mono, monospace" text-transform="uppercase">
    <text x="74" y="54" font-size="16" letter-spacing="2">CCTV FEED PLACEHOLDER</text>
    <text x="74" y="512" font-size="14" letter-spacing="1.5">${safeLabel} · ${safeCity}</text>
    <text x="646" y="512" font-size="13" letter-spacing="1.2">${safeId}</text>
    <text x="704" y="54" font-size="15" letter-spacing="2">${escapeXml(ts)}</text>
    <text x="74" y="486" font-size="13" letter-spacing="1.3">${safeStatus}</text>
  </g>
</svg>`.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();
  });

  // A live camera feed has no end of its own. When the viewer goes away the
  // upstream connection must go with it, or every abandoned view leaves a
  // stream open against the camera host for as long as that host will hold it.
  let released = false;
  const releaseUpstream = () => {
    if (released) return;
    released = true;
    stream.unpipe(res);
    // Destroying the Node stream cancels the web body it wraps; the direct
    // cancel covers a body that was never wrapped, and rejects harmlessly when
    // the reader is already held.
    stream.destroy();
    try {
      const cancelled = upstream.body?.cancel?.();
      if (typeof cancelled?.catch === 'function') cancelled.catch(() => {});
    } catch {
      /* already closed */
    }
  };
  res.once('close', () => {
    if (!res.writableEnded) releaseUpstream();
  });
  res.once('error', releaseUpstream);
  stream.once('end', () => {
    released = true;
  });
  stream.pipe(res);
}

/**
 * Watch a client response for an early goodbye.
 *
 * Bound BEFORE the upstream request goes out, because most of the waiting
 * happens before any header comes back: a viewer who closes the tab while a
 * slow camera is still thinking would otherwise leave that request running with
 * nobody to receive it.
 *
 * @param {import('http').ServerResponse} res - The client response.
 * @returns {{signal: AbortSignal, closed: boolean}} `signal` cancels the
 *   upstream request; `closed` says the client left before the response ended.
 */
export function watchDownstreamClose(res) {
  const controller = new AbortController();
  const state = {
    signal: controller.signal,
    closed: false,
  };
  const onClose = () => {
    // A response that ended normally also emits close; only an early one counts.
    if (res.writableEnded) return;
    state.closed = true;
    controller.abort();
  };
  res.once?.('close', onClose);
  res.once?.('error', onClose);
  return state;
}

/** 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,
    signal: downstream = null,
  } = {},
) {
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
  // The client going away cancels the upstream request, not just the response
  // to it.
  const onDownstreamAbort = () => controller.abort();
  if (downstream?.aborted) controller.abort();
  else downstream?.addEventListener?.('abort', onDownstreamAbort);
  try {
    return await fetchImpl(url, { headers, signal: controller.signal });
  } finally {
    clearTimeout(timeoutId);
    downstream?.removeEventListener?.('abort', onDownstreamAbort);
  }
}

/**
 * Fetch and decode a TxDOT ITS / TransGuide snapshot.
 *
 * TxDOT returns JSON with a base64-encoded JPEG in `snippet`, rather than
 * returning image/jpeg directly.
 */
export async function fetchTxdotSnapshot(
  url,
  {
    fetchImpl = fetch,
    timeoutMs = CCTV_FRAME_FETCH_TIMEOUT_MS,
    maxBytes = CCTV_FRAME_MAX_BODY_BYTES,
  } = {},
) {
  if (!url) return null;
  let parsed;
  try {
    parsed = new URL(url);
  } catch {
    return null;
  }
  if (
    parsed.origin !== 'https://its.txdot.gov' ||
    parsed.pathname !== '/its/DistrictIts/GetCctvSnapshotByIcdId'
  ) {
    return null;
  }
  // Base64 inflates by 4/3; the JSON envelope adds a few bytes of framing.
  const maxEnvelopeBytes = Math.ceil((maxBytes * 4) / 3) + 4096;
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
  try {
    // The snapshot endpoint answers directly; a redirect is not followed, so
    // the origin/path pin above holds for the request that is actually made.
    const upstream = await fetchImpl(parsed.toString(), {
      headers: {
        Accept: 'application/json',
        'User-Agent': 'gods-eye-view-cctv-proxy/1.0',
      },
      signal: controller.signal,
      redirect: 'manual',
    });
    if (!upstream.ok) return null;
    const envelope = await readCappedResponseBytes(upstream, maxEnvelopeBytes);
    if (!envelope) return null;
    let payload;
    try {
      payload = JSON.parse(envelope.toString('utf8'));
    } catch {
      return null;
    }
    let snippet =
      typeof payload?.snippet === 'string' ? payload.snippet.trim() : '';
    if (!snippet) return null;
    snippet = snippet.replace(/^data:image\/jpeg;base64,/i, '');
    // Canonical base64 only (4-char groups, padding only at the end):
    // Buffer.from() silently skips junk, which would let a non-image body
    // decode into "something".
    if (
      snippet.length > maxEnvelopeBytes ||
      !/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/.test(
        snippet,
      )
    ) {
      return null;
    }
    const body = Buffer.from(snippet, 'base64');
    if (body.length < 4 || body.length > maxBytes) return null;
    if (body[0] !== 0xff || body[1] !== 0xd8 || body[2] !== 0xff) return null;
    return { ok: true, body, contentType: 'image/jpeg' };
  } catch {
    return null;
  } finally {
    clearTimeout(timeoutId);
    controller.abort();
  }
}

/** Redirect hops the frame path will follow, and only within the same host. */
const MAX_SAME_HOST_REDIRECTS = 2;

/**
 * Fetch a registered frame URL following redirects ONLY within the original
 * origin (scheme, host and port; at most MAX_SAME_HOST_REDIRECTS hops).
 * Default redirect-following would let an upstream steer a host-pinned
 * request, and its host-specific headers, to any origin, another port, or a
 * plaintext downgrade.
 *
 * @param {string} url
 * @param {object} init - fetch init (headers, signal).
 * @param {typeof fetch} fetchImpl
 * @returns {Promise<Response|null>} Final response, or null on an off-host or
 *   over-long redirect chain.
 */
export async function fetchWithinHost(url, init, fetchImpl = fetch) {
  let current;
  try {
    current = new URL(url);
  } catch {
    return null;
  }
  const origin = current.origin;
  for (let hop = 0; hop <= MAX_SAME_HOST_REDIRECTS; hop++) {
    const upstream = await fetchImpl(current.toString(), {
      ...init,
      redirect: 'manual',
    });
    // Anything that is not a 3xx (including a test double with no status) is
    // the final answer.
    const status = Number(upstream?.status);
    if (!(status >= 300 && status < 400)) return upstream;
    const location = upstream.headers.get('location');
    try {
      await upstream.body?.cancel();
    } catch {
      /* no-op */
    }
    if (!location || hop === MAX_SAME_HOST_REDIRECTS) return null;
    let next;
    try {
      next = new URL(location, current);
    } catch {
      return null;
    }
    if (next.origin !== origin) return null;
    current = next;
  }
  return null;
}

/**
 * Image hosts that only serve frames to browser-identified clients, keyed by
 * exact hostname. Every other upstream sees the proxy's own identifying
 * User-Agent. Keyed on host, not on a URL substring, so a look-alike host or a
 * path that merely mentions the host never inherits the header.
 */
const CCTV_IMAGE_USER_AGENT_BY_HOST = Object.freeze({
  [new URL(NSW_IMAGE_ORIGIN).hostname]: NSW_IMAGE_USER_AGENT,
});

/**
 * User-Agent for one upstream frame request.
 *
 * @param {string} url
 * @returns {string}
 */
export function cctvUpstreamUserAgent(url) {
  try {
    return (
      CCTV_IMAGE_USER_AGENT_BY_HOST[new URL(url).hostname] ||
      'gods-eye-view-cctv-proxy/1.0'
    );
  } catch {
    return 'gods-eye-view-cctv-proxy/1.0';
  }
}

/**
 * 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 fetchWithinHost(
      url,
      {
        headers: { 'User-Agent': cctvUpstreamUserAgent(url) },
        signal: controller.signal,
      },
      fetchImpl,
    );
    if (!upstream) return null;
    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();
  }
}