mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-04 10:24:02 +00:00
Merge upstream main into fix/tarktee-datex-catalog-hardening
Resolve conflicts with #1234: keep redirect: 'manual' on both DATEX fetches and keep main's 30 s TARKTEE_FETCH_TIMEOUT_MS. Drop this branch's 15 s constant and its import, which auto-merged into a duplicate export. Keep both sets of Estonia tests and drop the 15 s claim from the changelog fragment. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
commit
59646e52f9
120 files changed
+19932
-469
No files matched your search
@@ -143,7 +143,7 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
|
||||
|
||||
if (url.pathname === '/sources') {
|
||||
const body = {
|
||||
trimmedPacks: getCctvSources.trimmedPacks?.() ?? [],
|
||||
trimmedPacks: getCctvSources.trimmedPacks?.(sources) ?? [],
|
||||
sources: sources.map((source) => ({
|
||||
id: source.id,
|
||||
pack: source.pack,
|
||||
|
||||
@@ -25,6 +25,7 @@ import {
|
||||
loadVegvesenSourcesFromOpenData,
|
||||
loadNycSourcesFromOpenData,
|
||||
loadTorontoSourcesFromOpenData,
|
||||
loadFloridaSourcesFromOpenData,
|
||||
} from './sources.js';
|
||||
|
||||
/** Env kill switch: unset or anything but "0" means enabled. */
|
||||
@@ -144,6 +145,14 @@ const LIVE_PACKS = [
|
||||
enabled: () => envOptIn('CCTV_LITHUANIA_ENABLED'),
|
||||
load: loadLithuaniaSources,
|
||||
},
|
||||
{
|
||||
// FL511's terms limit its content to individual, personal use (other use
|
||||
// needs FDOT's written consent), so the pack is off unless the operator
|
||||
// opts in (see DATA_SOURCES.md).
|
||||
name: 'florida',
|
||||
enabled: () => envOptIn('CCTV_FLORIDA_ENABLED'),
|
||||
load: loadFloridaSourcesFromOpenData,
|
||||
},
|
||||
];
|
||||
/**
|
||||
* Load CCTV sources from a local JSON file (CCTV_SOURCES_FILE env or default).
|
||||
@@ -195,16 +204,19 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
|
||||
/** @type {Promise<Array<object>>|null} In-flight refresh, shared by concurrent
|
||||
* callers so a post-TTL burst launches ONE refetch, not one per request. */
|
||||
let _cctvSourceInflight = null;
|
||||
/** @type {Array<{pack: string, available: number, served: number}>} Packs
|
||||
* that serve fewer cameras than they offer. */
|
||||
let _cctvTrimmedPacks = [];
|
||||
/** @type {WeakMap<Array<object>, Array<{pack: string, available: number, served: number}>>}
|
||||
* Packs that serve fewer cameras than they offer, keyed by the source list
|
||||
* they describe, so a request served the old list during a background
|
||||
* refresh never pairs it with the new list's trim notes. */
|
||||
const _cctvTrimmedPacks = new WeakMap();
|
||||
|
||||
/**
|
||||
* Assemble and cache the merged CCTV source list.
|
||||
*
|
||||
* Merges every source pack (live open-data packs, local file, env
|
||||
* variable), deduplicates by ID, shares the catalog cap fairly across
|
||||
* packs, and caches for CCTV_SOURCE_CACHE_MS.
|
||||
* packs, and caches for CCTV_SOURCE_CACHE_MS. Past that, the cached list is
|
||||
* returned while a background refresh replaces it; only a first load waits.
|
||||
*
|
||||
* @returns {Promise<Array<object>>} Deduplicated, capped source list.
|
||||
*/
|
||||
@@ -219,11 +231,31 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
|
||||
// Single-flight: a burst of requests arriving past the TTL shares ONE refresh
|
||||
// instead of each launching the full multi-provider refetch. The `.finally`
|
||||
// clears the ref so the next post-TTL cycle starts fresh.
|
||||
if (_cctvSourceInflight) return _cctvSourceInflight;
|
||||
_cctvSourceInflight = refreshCctvSources().finally(() => {
|
||||
_cctvSourceInflight = null;
|
||||
});
|
||||
return _cctvSourceInflight;
|
||||
if (!_cctvSourceInflight) {
|
||||
_cctvSourceInflight = refreshCctvSources().finally(() => {
|
||||
_cctvSourceInflight = null;
|
||||
});
|
||||
// A background refresh has no caller waiting on it: report its failure
|
||||
// here, once per refresh rather than once per request it serves. Keep
|
||||
// the old catalog for another TTL, as an empty refresh does, so a refresh
|
||||
// that keeps failing is retried once per TTL, not back to back.
|
||||
if (_cctvSourceCache.length) {
|
||||
_cctvSourceInflight.catch((error) => {
|
||||
_cctvSourceCacheAt = Date.now();
|
||||
console.warn(
|
||||
'[CCTV] background catalog refresh failed; serving',
|
||||
_cctvSourceCache.length,
|
||||
'cameras from the last catalog:',
|
||||
error?.message || error,
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
// With a catalog in hand, answer with it while the refresh runs: the
|
||||
// refresh lasts as long as the slowest pack (Tarktee's DATEX feeds take up
|
||||
// to its 30 s timeout), and every /api/cctv request would otherwise wait
|
||||
// for it. Only the first load, with nothing to serve yet, waits.
|
||||
return _cctvSourceCache.length ? _cctvSourceCache : _cctvSourceInflight;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -313,7 +345,7 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
|
||||
}
|
||||
if (capped.length > 0 || _cctvSourceCache.length === 0) {
|
||||
_cctvSourceCache = capped;
|
||||
_cctvTrimmedPacks = trimmedPacks;
|
||||
_cctvTrimmedPacks.set(capped, trimmedPacks);
|
||||
} else {
|
||||
// Every source came back empty (all live packs timed out / upstream outage)
|
||||
// but a good catalog is already cached — serve it stale rather than blanking
|
||||
@@ -327,7 +359,10 @@ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) {
|
||||
return _cctvSourceCache;
|
||||
}
|
||||
|
||||
/** Packs serving fewer cameras than they offer, as of the last refresh. */
|
||||
getCctvSources.trimmedPacks = () => _cctvTrimmedPacks;
|
||||
/** Packs serving fewer cameras than they offer in `sources` (by default the
|
||||
* current catalog). Pass the list a request was served to keep the two in
|
||||
* step while a background refresh swaps the catalog. */
|
||||
getCctvSources.trimmedPacks = (sources = _cctvSourceCache) =>
|
||||
_cctvTrimmedPacks.get(sources) ?? [];
|
||||
return getCctvSources;
|
||||
}
|
||||
@@ -113,6 +113,27 @@ export const CALTRANS_ANCHORS = [
|
||||
{ lat: 32.7157, lon: -117.1611 }, // San Diego
|
||||
{ lat: 38.5816, lon: -121.4944 }, // Sacramento
|
||||
];
|
||||
export const FLORIDA_CCTV_URL =
|
||||
'https://services.arcgis.com/3wFbqsFPLeKqOlIK/arcgis/rest/services/FL511_Traffic_Cameras/FeatureServer/0/query';
|
||||
export const FLORIDA_IMAGE_HOSTS = [
|
||||
'images-dis.divas.cloud',
|
||||
'snapshots.divas.cloud',
|
||||
];
|
||||
export const DEFAULT_FLORIDA_MAX_SOURCES = 500;
|
||||
/** Byte cap per catalog page (2,000 rows, about 650 KiB measured). */
|
||||
export const FLORIDA_MAX_CATALOG_BYTES = 2 * 1024 * 1024;
|
||||
/** Catalog page ceiling: 40 pages covers 4,000+ cameras even at 100 rows a page. */
|
||||
export const FLORIDA_MAX_CATALOG_PAGES = 40;
|
||||
export const FLORIDA_ANCHORS = [
|
||||
{ lat: 25.7617, lon: -80.1918 }, // Miami
|
||||
{ lat: 28.5384, lon: -81.3789 }, // Orlando
|
||||
{ lat: 27.9506, lon: -82.4572 }, // Tampa
|
||||
{ lat: 30.3322, lon: -81.6557 }, // Jacksonville
|
||||
{ lat: 30.4383, lon: -84.2807 }, // Tallahassee
|
||||
{ lat: 26.1224, lon: -80.1373 }, // Fort Lauderdale
|
||||
{ lat: 26.7153, lon: -80.0534 }, // West Palm Beach
|
||||
{ lat: 30.4213, lon: -87.2169 }, // Pensacola (panhandle)
|
||||
];
|
||||
/** TfL JamCams: one keyless list endpoint; frames live on a public S3 bucket. */
|
||||
export const TFL_JAMCAM_URL = 'https://api.tfl.gov.uk/Place/Type/JamCam';
|
||||
export const TFL_IMAGE_ORIGIN =
|
||||
@@ -190,9 +211,6 @@ export const TARKTEE_IMAGE_ORIGIN = 'https://tarktee.transpordiamet.ee/images/';
|
||||
export const DEFAULT_TARKTEE_MAX_SOURCES = 179;
|
||||
/** Hard ceiling per Tarktee DATEX XML document (~120–165 KB live). */
|
||||
export const TARKTEE_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
|
||||
/** Two DATEX XML documents download concurrently from the same origin (~5.3 s
|
||||
* baseline) alongside the other packs, so the 8 s default can abort mid-stream. */
|
||||
export const TARKTEE_FETCH_TIMEOUT_MS = 15_000;
|
||||
export const TARKTEE_ANCHORS = [
|
||||
{ lat: 59.437, lon: 24.753 }, // Tallinn
|
||||
{ lat: 58.378, lon: 26.729 }, // Tartu
|
||||
@@ -444,6 +462,10 @@ export const CCTV_SOURCE_CACHE_MS = 15 * 60 * 1000;
|
||||
* pending forever — a hung fetch aborts, the loader returns [], and
|
||||
* serve-stale/other packs take over. */
|
||||
export const CCTV_SOURCE_FETCH_TIMEOUT_MS = 15 * 1000;
|
||||
/** Tarktee's two DATEX feeds (120 and 160 KB) answered in 2 to 16.6 s across
|
||||
* ten requests on 2026-10-18, so the shared 15 s timeout dropped the Estonian
|
||||
* pack on about a third of loads. Both must arrive for any camera to load. */
|
||||
export const TARKTEE_FETCH_TIMEOUT_MS = 30 * 1000;
|
||||
/** Individual CCTV image fetches must settle before the active 10-second
|
||||
* client refresh cadence. A bounded miss can fall through to Street View or
|
||||
* the synthetic frame instead of leaving the browser preview pending. */
|
||||
|
||||
@@ -377,6 +377,17 @@ export function isLikelyCalgaryCoordinate(lat, lon) {
|
||||
);
|
||||
}
|
||||
|
||||
/** Florida statewide, Key West to Georgia state line, with slack at each edge */
|
||||
export function isLikelyFloridaCoordinate(lat, lon) {
|
||||
return (
|
||||
isPlausibleLatLon(lat, lon) &&
|
||||
lat >= 24.3 &&
|
||||
lat <= 31.1 &&
|
||||
lon >= -87.7 &&
|
||||
lon <= -79.8
|
||||
);
|
||||
}
|
||||
|
||||
/** New York City's five boroughs, with slack for the harbor and city line. */
|
||||
export function isLikelyNycCoordinate(lat, lon) {
|
||||
return (
|
||||
|
||||
@@ -43,7 +43,6 @@ import {
|
||||
TARKTEE_IMAGE_ORIGIN,
|
||||
DEFAULT_TARKTEE_MAX_SOURCES,
|
||||
TARKTEE_MAX_CATALOG_BYTES,
|
||||
TARKTEE_FETCH_TIMEOUT_MS,
|
||||
TARKTEE_ANCHORS,
|
||||
DEFAULT_WARENDORF_SOURCE_FILE,
|
||||
WARENDORF_IMAGE_ORIGINS,
|
||||
@@ -57,6 +56,12 @@ import {
|
||||
DEFAULT_CALGARY_MAX_SOURCES,
|
||||
CALGARY_DOWNTOWN,
|
||||
CALGARY_MAX_CATALOG_BYTES,
|
||||
FLORIDA_CCTV_URL,
|
||||
FLORIDA_IMAGE_HOSTS,
|
||||
DEFAULT_FLORIDA_MAX_SOURCES,
|
||||
FLORIDA_ANCHORS,
|
||||
FLORIDA_MAX_CATALOG_BYTES,
|
||||
FLORIDA_MAX_CATALOG_PAGES,
|
||||
CATALONIA_CAMERAS_URL,
|
||||
CATALONIA_MAX_CATALOG_BYTES,
|
||||
CATALONIA_IMAGE_HOSTS,
|
||||
@@ -72,6 +77,7 @@ import {
|
||||
TORONTO_GROUND_ELEVATION_M,
|
||||
TORONTO_MAX_CATALOG_BYTES,
|
||||
CCTV_SOURCE_FETCH_TIMEOUT_MS,
|
||||
TARKTEE_FETCH_TIMEOUT_MS,
|
||||
DELDOT_CCTV_URL,
|
||||
DEFAULT_DELDOT_MAX_SOURCES,
|
||||
DELDOT_ANCHORS,
|
||||
@@ -115,6 +121,7 @@ import {
|
||||
isLikelyTexasCoordinate,
|
||||
isLikelyNswCoordinate,
|
||||
isLikelyCalgaryCoordinate,
|
||||
isLikelyFloridaCoordinate,
|
||||
isLikelyCataloniaCoordinate,
|
||||
decodeNumericEntities,
|
||||
isLikelyNycCoordinate,
|
||||
@@ -2826,3 +2833,203 @@ export async function loadTorontoSourcesFromOpenData() {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/** FL511 DIRECTION codes as headings; "NOT DIRECTIONAL" has none. */
|
||||
const FLORIDA_DIRECTION_HEADINGS = Object.freeze({
|
||||
N: 0,
|
||||
E: 90,
|
||||
S: 180,
|
||||
W: 270,
|
||||
});
|
||||
|
||||
/**
|
||||
* Stable camera id from a frame URL. FL511's ID column repeats across regions
|
||||
* (ID 1220 is both an I-75 and an I-95 camera), so the image channel is the
|
||||
* key, as Calgary keys on its frame filename. Falls back to a path slug.
|
||||
*
|
||||
* @param {string} imageUrl - A host-pinned FL511 frame URL.
|
||||
* @returns {?string} Provider-stable id, or null when underivable.
|
||||
*/
|
||||
export function floridaCameraId(imageUrl) {
|
||||
let path;
|
||||
try {
|
||||
path = new URL(String(imageUrl ?? '')).pathname;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
const channel = /\/chan-(\d+)_[a-z]+\.jpg$/i.exec(path);
|
||||
if (channel) return `fl-${channel[1]}`;
|
||||
const slug = path
|
||||
.replace(/^\/+|\.[a-z0-9]+$/gi, '')
|
||||
.replace(/[^a-zA-Z0-9]+/g, '-')
|
||||
.toLowerCase();
|
||||
return slug ? `fl-${slug}` : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* One FL511 ArcGIS feature -> one catalog source, or null. DIRECTION is a
|
||||
* one-letter code the free-text parser won't read, so a table maps it.
|
||||
*
|
||||
* @param {object} feature - ArcGIS feature from the FL511 camera layer.
|
||||
* @returns {?object}
|
||||
*/
|
||||
export function floridaCameraToSource(feature) {
|
||||
const a = feature?.attributes;
|
||||
if (!a || typeof a !== 'object') return null;
|
||||
const lat = toFiniteNumber(a.LATITUDE);
|
||||
const lon = toFiniteNumber(a.LONGITUDE);
|
||||
if (!isLikelyFloridaCoordinate(lat, lon)) return null;
|
||||
|
||||
const imageUrl = String(a.IMAGE || '');
|
||||
let image;
|
||||
try {
|
||||
image = new URL(imageUrl);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
// Official image hosts only, HTTPS, no embedded credentials.
|
||||
if (
|
||||
image.protocol !== 'https:' ||
|
||||
!FLORIDA_IMAGE_HOSTS.includes(image.host) ||
|
||||
image.username ||
|
||||
image.password
|
||||
)
|
||||
return null;
|
||||
|
||||
const cameraId = floridaCameraId(imageUrl);
|
||||
if (!cameraId) return null;
|
||||
|
||||
const heading =
|
||||
FLORIDA_DIRECTION_HEADINGS[
|
||||
String(a.DIRECTION || '')
|
||||
.trim()
|
||||
.toUpperCase()
|
||||
];
|
||||
const hasHeading = Number.isFinite(heading);
|
||||
|
||||
return {
|
||||
id: cameraId,
|
||||
name: String(a.DESCRIPT || '').trim() || `FL511 ${cameraId.slice(3)}`,
|
||||
city: String(a.COUNTY ? `${a.COUNTY} County` : 'Florida'),
|
||||
cityId: 'florida',
|
||||
provider: 'FL511',
|
||||
lat,
|
||||
lon,
|
||||
headingDeg: hasHeading ? heading : fallbackHeadingFromId(cameraId),
|
||||
headingConfidence: hasHeading ? 'high' : 'low',
|
||||
pitchDeg: hasHeading ? -24 : -18,
|
||||
fovDeg: hasHeading ? 56 : 44,
|
||||
rangeM: hasHeading ? 210 : 145,
|
||||
mountHeightM: hasHeading ? 10 : 8,
|
||||
groundElevationM: 5, // Coastal prior; the client's ground snap corrects.
|
||||
feedType: 'image',
|
||||
url: imageUrl,
|
||||
snapshotUrl: imageUrl,
|
||||
sourceKind: 'fl511-open-data',
|
||||
license:
|
||||
'FL511 (FDOT), individual personal use only; other use needs FDOT written consent',
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch FL511 (Florida DOT) cameras, keyless: an ArcGIS FeatureServer paged
|
||||
* up to 2,000 rows at a time, ordered by the layer's object id so offsets are
|
||||
* stable. Paging follows `exceededTransferLimit` and advances by the rows each
|
||||
* page returned, so a lower server-side max record count still reads the whole
|
||||
* layer. Frames are stills on the Divas Cloud hosts. A redirect on any page
|
||||
* drops the pack; any other failed page keeps the rows already read.
|
||||
*
|
||||
* @returns {Promise<Array<object>>} Normalized camera source objects.
|
||||
*/
|
||||
export async function loadFloridaSourcesFromOpenData() {
|
||||
try {
|
||||
const cameras = [];
|
||||
const seen = new Set();
|
||||
const pageSize = 2000;
|
||||
let offset = 0;
|
||||
// Page ceiling, in case the service ignores resultOffset.
|
||||
for (let page = 0; ; page += 1) {
|
||||
if (page >= FLORIDA_MAX_CATALOG_PAGES) {
|
||||
console.warn(
|
||||
`[CCTV] Florida catalog stopped at the ${FLORIDA_MAX_CATALOG_PAGES}-page ceiling; the catalog may be incomplete`,
|
||||
);
|
||||
break;
|
||||
}
|
||||
// OBJECTID_1 is this layer's object id field (a plain OBJECTID is a 400).
|
||||
const url = `${FLORIDA_CCTV_URL}?where=1%3D1&outFields=*&orderByFields=OBJECTID_1&f=json&resultRecordCount=${pageSize}&resultOffset=${offset}`;
|
||||
let resp;
|
||||
try {
|
||||
resp = await fetch(url, {
|
||||
headers: { Accept: 'application/json' },
|
||||
redirect: 'manual',
|
||||
signal: AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS),
|
||||
});
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
'[CCTV] Florida catalog page failed:',
|
||||
error?.message || error,
|
||||
);
|
||||
break;
|
||||
}
|
||||
// Release the body on every path that will not read it.
|
||||
const discard = async () => {
|
||||
try {
|
||||
await resp.body?.cancel();
|
||||
} catch {
|
||||
/* no-op */
|
||||
}
|
||||
};
|
||||
if (resp.status >= 300 && resp.status < 400) {
|
||||
console.warn(
|
||||
'[CCTV] Florida catalog redirected; redirects are not followed',
|
||||
);
|
||||
await discard();
|
||||
return [];
|
||||
}
|
||||
if (!resp.ok) {
|
||||
console.warn('[CCTV] Florida camera download failed:', resp.status);
|
||||
await discard();
|
||||
break;
|
||||
}
|
||||
let body;
|
||||
try {
|
||||
body = await readResponseJsonCapped(resp, FLORIDA_MAX_CATALOG_BYTES);
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
'[CCTV] Florida catalog page unreadable:',
|
||||
error?.message || error,
|
||||
);
|
||||
break;
|
||||
}
|
||||
const feats = Array.isArray(body?.features) ? body.features : [];
|
||||
for (const feature of feats) {
|
||||
const camera = floridaCameraToSource(feature);
|
||||
if (!camera || seen.has(camera.id)) continue;
|
||||
seen.add(camera.id);
|
||||
cameras.push(camera);
|
||||
}
|
||||
// ArcGIS sets exceededTransferLimit while more rows remain. A page may
|
||||
// be shorter than pageSize when the layer's max record count is lower.
|
||||
if (body?.exceededTransferLimit !== true || feats.length === 0) break;
|
||||
offset += feats.length;
|
||||
}
|
||||
|
||||
const maxRaw = Number(
|
||||
process.env.CCTV_FLORIDA_MAX_SOURCES || DEFAULT_FLORIDA_MAX_SOURCES,
|
||||
);
|
||||
const maxCount = Number.isFinite(maxRaw)
|
||||
? Math.max(8, Math.min(4200, Math.floor(maxRaw)))
|
||||
: DEFAULT_FLORIDA_MAX_SOURCES;
|
||||
const prioritized = prioritizeSources(cameras, maxCount, FLORIDA_ANCHORS);
|
||||
console.log(
|
||||
`[CCTV] Loaded Florida camera sources: ${cameras.length} (using nearest ${prioritized.length})`,
|
||||
);
|
||||
return prioritized;
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
'[CCTV] Florida camera download error:',
|
||||
error?.message || error,
|
||||
);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
@@ -14,16 +14,34 @@ async function readRequestBodyCapped(req, maxBytes) {
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
function readRequestBody(req, maxBytes = 1024 * 1024) {
|
||||
/** Optional cancellation discards buffered data while the request drains. */
|
||||
function readRequestBody(req, maxBytes = 1024 * 1024, signal) {
|
||||
return new Promise((resolve, reject) => {
|
||||
let total = 0;
|
||||
let bodyTooLarge = false;
|
||||
let settled = false;
|
||||
const chunks = [];
|
||||
req.on('data', (chunk) => {
|
||||
const stopReading = () => {
|
||||
if (!signal) return;
|
||||
req.off('data', data);
|
||||
req.off('end', end);
|
||||
signal.removeEventListener('abort', abort);
|
||||
};
|
||||
const fail = (error) => {
|
||||
if (signal) {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
chunks.length = 0;
|
||||
signal.removeEventListener('abort', abort);
|
||||
}
|
||||
reject(error);
|
||||
};
|
||||
const abort = () => fail(signal.reason);
|
||||
const data = (chunk) => {
|
||||
// Keep draining after the cap without retaining more bytes. Destroying
|
||||
// the request also destroys its socket, so the handler cannot deliver
|
||||
// the fixed error response it constructs for BODY_TOO_LARGE.
|
||||
if (bodyTooLarge) return;
|
||||
if (bodyTooLarge || settled) return;
|
||||
total += chunk.length;
|
||||
if (total > maxBytes) {
|
||||
// Same marker readRequestBodyCapped sets, so a caller can tell an
|
||||
@@ -32,15 +50,42 @@ function readRequestBody(req, maxBytes = 1024 * 1024) {
|
||||
chunks.length = 0;
|
||||
const error = new Error(`Request body exceeds ${maxBytes} bytes`);
|
||||
error.code = 'BODY_TOO_LARGE';
|
||||
reject(error);
|
||||
fail(error);
|
||||
return;
|
||||
}
|
||||
chunks.push(chunk);
|
||||
});
|
||||
req.on('end', () => {
|
||||
};
|
||||
const end = () => {
|
||||
if (settled) {
|
||||
stopReading();
|
||||
return;
|
||||
}
|
||||
if (!bodyTooLarge) resolve(Buffer.concat(chunks).toString('utf8'));
|
||||
});
|
||||
req.on('error', reject);
|
||||
if (signal) {
|
||||
settled = true;
|
||||
chunks.length = 0;
|
||||
stopReading();
|
||||
}
|
||||
};
|
||||
const close = () => {
|
||||
fail(
|
||||
new DOMException('Request closed before its body ended', 'AbortError'),
|
||||
);
|
||||
stopReading();
|
||||
req.off('error', fail);
|
||||
req.off('close', close);
|
||||
};
|
||||
if (signal) {
|
||||
// Abort/overflow release retained bytes but keep draining. Retain an
|
||||
// error handler until close, including a late error after end/abort.
|
||||
req.on('close', close);
|
||||
signal.addEventListener('abort', abort, { once: true });
|
||||
if (signal.aborted) abort();
|
||||
}
|
||||
req.on('data', data);
|
||||
req.on('end', end);
|
||||
req.on('error', fail);
|
||||
if (signal && req.destroyed) close();
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { EventEmitter, getEventListeners } from 'node:events';
|
||||
import { readRequestBody } from './request.js';
|
||||
|
||||
function fixture() {
|
||||
const req = new EventEmitter();
|
||||
req.pause = () => assert.fail('The reader must keep draining');
|
||||
req.destroy = () =>
|
||||
assert.fail('The reader must preserve the response socket');
|
||||
const owner = new AbortController();
|
||||
return { req, owner };
|
||||
}
|
||||
|
||||
function finish(req, owner) {
|
||||
assert.doesNotThrow(() => req.emit('error', new Error('late stream error')));
|
||||
req.emit('end');
|
||||
assert.doesNotThrow(() => req.emit('error', new Error('error before close')));
|
||||
req.emit('close');
|
||||
for (const event of ['data', 'end', 'error', 'close'])
|
||||
assert.equal(req.listenerCount(event), 0, event);
|
||||
assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
|
||||
}
|
||||
|
||||
test('default and cancellable readers preserve byte caps and UTF-8 across chunks', async () => {
|
||||
for (const useSignal of [false, true]) {
|
||||
for (const text of ['', 'hello', 'é']) {
|
||||
const { req, owner } = fixture();
|
||||
const bytes = Buffer.from(text);
|
||||
const body = readRequestBody(
|
||||
req,
|
||||
bytes.length,
|
||||
useSignal ? owner.signal : undefined,
|
||||
);
|
||||
for (const byte of bytes) req.emit('data', Buffer.from([byte]));
|
||||
req.emit('end');
|
||||
assert.equal(await body, text);
|
||||
if (useSignal) finish(req, owner);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
test('default overflow rejects before end and drains without pausing or destroying', async () => {
|
||||
const { req } = fixture();
|
||||
const rejection = assert.rejects(readRequestBody(req, 2), {
|
||||
code: 'BODY_TOO_LARGE',
|
||||
});
|
||||
req.emit('data', Buffer.from('é'));
|
||||
req.emit('data', Buffer.from('!'));
|
||||
await rejection;
|
||||
req.emit('data', Buffer.alloc(4096));
|
||||
assert.doesNotThrow(() => req.emit('error', new Error('late')));
|
||||
req.emit('end');
|
||||
});
|
||||
|
||||
test('partial abort settles before end, discards late data and guards errors until close', async () => {
|
||||
const { req, owner } = fixture();
|
||||
const reason = new Error('request deadline');
|
||||
const rejection = assert.rejects(
|
||||
readRequestBody(req, 1024, owner.signal),
|
||||
(error) => error === reason,
|
||||
);
|
||||
req.emit('data', Buffer.from('partial'));
|
||||
owner.abort(reason);
|
||||
await rejection;
|
||||
assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
|
||||
req.emit('data', Buffer.alloc(4096));
|
||||
finish(req, owner);
|
||||
});
|
||||
|
||||
test('already-aborted and already-closed requests settle and release their listeners', async () => {
|
||||
for (const closed of [false, true]) {
|
||||
const { req, owner } = fixture();
|
||||
if (closed) req.destroyed = true;
|
||||
else owner.abort();
|
||||
await assert.rejects(readRequestBody(req, 1024, owner.signal), {
|
||||
name: 'AbortError',
|
||||
});
|
||||
if (!closed) finish(req, owner);
|
||||
for (const event of ['data', 'end', 'error', 'close'])
|
||||
assert.equal(req.listenerCount(event), 0);
|
||||
assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
|
||||
}
|
||||
});
|
||||
|
||||
test('overflow keeps its first failure through later abort and stream errors', async () => {
|
||||
const { req, owner } = fixture();
|
||||
const rejection = assert.rejects(readRequestBody(req, 2, owner.signal), {
|
||||
code: 'BODY_TOO_LARGE',
|
||||
});
|
||||
req.emit('data', Buffer.from('é!'));
|
||||
await rejection;
|
||||
owner.abort(new Error('later deadline'));
|
||||
req.emit('data', Buffer.alloc(4096));
|
||||
finish(req, owner);
|
||||
});
|
||||
|
||||
test('early close and request error reject once; successful end wins over later abort', async () => {
|
||||
for (const event of ['close', 'error', 'end']) {
|
||||
const { req, owner } = fixture();
|
||||
const reason = new Error('request failed');
|
||||
const body = readRequestBody(req, 32, owner.signal);
|
||||
const observed =
|
||||
event === 'end'
|
||||
? body.then((value) => assert.equal(value, 'ok'))
|
||||
: assert.rejects(
|
||||
body,
|
||||
event === 'close'
|
||||
? { name: 'AbortError' }
|
||||
: (error) => error === reason,
|
||||
);
|
||||
req.emit('data', Buffer.from('ok'));
|
||||
req.emit(event, reason);
|
||||
owner.abort(new Error('later abort'));
|
||||
await observed;
|
||||
if (event !== 'close') finish(req, owner);
|
||||
assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
|
||||
}
|
||||
});
|
||||
|
||||
test('a prior data observer can abort synchronously without retaining or resolving that chunk', async () => {
|
||||
const { req, owner } = fixture();
|
||||
const reason = new Error('observer stopped request');
|
||||
const external = () => owner.abort(reason);
|
||||
req.on('data', external);
|
||||
const rejection = assert.rejects(
|
||||
readRequestBody(req, 1024, owner.signal),
|
||||
(error) => error === reason,
|
||||
);
|
||||
req.emit('data', Buffer.from('must be discarded'));
|
||||
await rejection;
|
||||
req.emit('end');
|
||||
req.emit('close');
|
||||
assert.deepEqual(req.listeners('data'), [external]);
|
||||
assert.equal(getEventListeners(owner.signal, 'abort').length, 0);
|
||||
req.off('data', external);
|
||||
});
|
||||
@@ -41,3 +41,48 @@ export function sameSiteGated(handler) {
|
||||
return handler(req, res, next);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Proxy header values arrive client-controlled; read one only when it is a
|
||||
* single scheme token, never a comma-separated chain.
|
||||
* @param {import('node:http').IncomingMessage} req
|
||||
* @returns {'http:'|'https:'}
|
||||
*/
|
||||
function forwardedProtocol(req) {
|
||||
if (req.socket?.encrypted) return 'https:';
|
||||
const proto = String(req.headers?.['x-forwarded-proto'] || '')
|
||||
.trim()
|
||||
.toLowerCase();
|
||||
return proto === 'https' ? 'https:' : 'http:';
|
||||
}
|
||||
|
||||
/**
|
||||
* Cross-site gate for keyless routes the map needs in order to render
|
||||
* (`/api/terrain/heights`). It refuses the same cross-site browser shapes as
|
||||
* `admitSameSite` (a `Sec-Fetch-Site` other than same-origin/none, a foreign
|
||||
* or opaque `Origin`) but does not refuse reverse-proxy forwarding headers:
|
||||
* those protect credentials and spend, which a keyless route has neither of,
|
||||
* and refusing them would break rendering for an instance served behind the
|
||||
* reverse proxy the README documents. A single `X-Forwarded-Proto` supplies
|
||||
* the scheme an `Origin` is compared against. That header cannot be attached
|
||||
* to a cross-site browser request without a CORS preflight this server never
|
||||
* answers, and `Sec-Fetch-Site` is still enforced regardless.
|
||||
* @param {import('node:http').IncomingMessage} req
|
||||
* @param {import('node:http').ServerResponse} res
|
||||
* @returns {boolean} True when a 403 was sent.
|
||||
*/
|
||||
export function admitSameSiteKeyless(req, res) {
|
||||
const verdict = admitSameSiteRequest({
|
||||
hostHeader: req.headers?.host,
|
||||
protocol: forwardedProtocol(req),
|
||||
origin: req.headers?.origin,
|
||||
secFetchSite: req.headers?.['sec-fetch-site'],
|
||||
proxyHeaders: {},
|
||||
});
|
||||
if (verdict.ok) return false;
|
||||
res.statusCode = verdict.status;
|
||||
res.setHeader('Content-Type', 'application/json');
|
||||
res.setHeader('Cache-Control', 'no-store');
|
||||
res.end(JSON.stringify({ error: verdict.error }));
|
||||
return true;
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
import { createGeminiTokenHandler } from './gemini/realtime.js';
|
||||
|
||||
/** Mint constrained Live API credentials; the permanent key stays on Node. */
|
||||
function geminiLiveProxy({ annotationGuidance, realtime = {} } = {}) {
|
||||
function install(middlewares) {
|
||||
middlewares.use(
|
||||
'/api/gemini/token',
|
||||
createGeminiTokenHandler({ ...realtime, annotationGuidance }),
|
||||
);
|
||||
}
|
||||
return {
|
||||
name: 'gemini-live-proxy',
|
||||
configureServer(server) {
|
||||
install(server.middlewares);
|
||||
},
|
||||
configurePreviewServer(server) {
|
||||
install(server.middlewares);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export { geminiLiveProxy };
|
||||
@@ -0,0 +1,53 @@
|
||||
import { realtimeInstructions } from '../openai/instructions.js';
|
||||
import { GEV_REALTIME_TOOLS } from '../openai/tools.js';
|
||||
|
||||
export const GEMINI_LIVE_MODEL_DEFAULT = 'gemini-3.8-live';
|
||||
export const GEMINI_TOKEN_ENDPOINT =
|
||||
'https://generativelanguage.googleapis.com/v1beta/auth_tokens';
|
||||
export const GEMINI_LIVE_WEBSOCKET_URL =
|
||||
'wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContentConstrained';
|
||||
|
||||
export function normalizeGeminiModel(value) {
|
||||
const model = String(value ?? GEMINI_LIVE_MODEL_DEFAULT)
|
||||
.trim()
|
||||
.replace(/^models\//, '');
|
||||
return /^gemini-[a-z0-9][a-z0-9.-]{0,119}$/.test(model) ? model : null;
|
||||
}
|
||||
|
||||
/** REST BidiGenerateContentSetup fields, excluding its model resource name. */
|
||||
export function createGeminiLiveConfig(
|
||||
annotationGuidance,
|
||||
inputMode = 'open-mic',
|
||||
tools = GEV_REALTIME_TOOLS,
|
||||
) {
|
||||
if (!['open-mic', 'push-to-talk'].includes(inputMode))
|
||||
throw new Error('Invalid Gemini input mode');
|
||||
const config = {
|
||||
generationConfig: { responseModalities: ['AUDIO'] },
|
||||
systemInstruction: {
|
||||
parts: [{ text: realtimeInstructions(annotationGuidance) }],
|
||||
},
|
||||
tools: [
|
||||
{
|
||||
functionDeclarations: tools.map(
|
||||
({ name, description, parameters }) => ({
|
||||
name,
|
||||
description,
|
||||
// GEV's canonical schemas are JSON Schema, not Google's Schema.
|
||||
parametersJsonSchema: structuredClone(parameters),
|
||||
// App actions must finish before the model confirms their result.
|
||||
behavior: 'BLOCKING',
|
||||
}),
|
||||
),
|
||||
},
|
||||
],
|
||||
inputAudioTranscription: {},
|
||||
outputAudioTranscription: {},
|
||||
contextWindowCompression: { slidingWindow: {} },
|
||||
};
|
||||
if (inputMode === 'push-to-talk')
|
||||
config.realtimeInputConfig = {
|
||||
automaticActivityDetection: { disabled: true },
|
||||
};
|
||||
return config;
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
import { waitForSignal } from '../../../src/sources/requestDeadline.js';
|
||||
import { clientKey, makeCostRateLimiter } from '../common/rate-limit.js';
|
||||
import { readResponseJsonCapped } from '../../../src/sources/httpBody.js';
|
||||
import {
|
||||
GEMINI_TOKEN_ENDPOINT,
|
||||
GEMINI_LIVE_WEBSOCKET_URL,
|
||||
GEMINI_LIVE_MODEL_DEFAULT,
|
||||
createGeminiLiveConfig,
|
||||
normalizeGeminiModel,
|
||||
} from './config.js';
|
||||
import {
|
||||
isGeminiRequestOriginAllowed,
|
||||
readGeminiTokenRequest,
|
||||
} from './request.js';
|
||||
|
||||
const UPSTREAM_RESPONSE_BYTES = 32 * 1024;
|
||||
const MINT_FAILURE = 'Gemini voice could not connect. Try again.';
|
||||
|
||||
export function createGeminiTokenHandler({
|
||||
annotationGuidance,
|
||||
tools,
|
||||
fetchImpl = (...args) => fetch(...args),
|
||||
resolveApiKey = () => process.env.GEMINI_API_KEY,
|
||||
resolveModel = () =>
|
||||
process.env.GEMINI_LIVE_MODEL || GEMINI_LIVE_MODEL_DEFAULT,
|
||||
resolveHost = () => process.env.HOST,
|
||||
resolveAllowedHosts = () => process.env.GEV_ALLOWED_HOSTS,
|
||||
resolveRateLimit = () => process.env.GEV_RATELIMIT_GEMINI_PER_MIN ?? '30',
|
||||
now = Date.now,
|
||||
timeoutMs = 15_000,
|
||||
} = {}) {
|
||||
let limiter;
|
||||
return async (req, res) => {
|
||||
if (req.aborted || res.destroyed || res.writableEnded) return;
|
||||
res.setHeader('Cache-Control', 'no-store');
|
||||
res.setHeader('Content-Type', 'application/json; charset=utf-8');
|
||||
res.setHeader('X-Content-Type-Options', 'nosniff');
|
||||
const reply = (status, errorOrPayload) => {
|
||||
if (res.destroyed || res.writableEnded) return;
|
||||
res.statusCode = status;
|
||||
if (status >= 400) res.setHeader('Connection', 'close');
|
||||
res.end(
|
||||
JSON.stringify(
|
||||
typeof errorOrPayload === 'string'
|
||||
? { error: errorOrPayload }
|
||||
: errorOrPayload,
|
||||
),
|
||||
);
|
||||
};
|
||||
if (req.method !== 'POST') {
|
||||
res.setHeader('Allow', 'POST');
|
||||
return reply(405, 'Method not allowed');
|
||||
}
|
||||
let apiKey;
|
||||
let model;
|
||||
try {
|
||||
if (
|
||||
!isGeminiRequestOriginAllowed(req, resolveHost(), resolveAllowedHosts())
|
||||
)
|
||||
return reply(403, 'Gemini voice requires a same-origin app request');
|
||||
apiKey = String(resolveApiKey() || '').trim();
|
||||
if (!apiKey)
|
||||
return reply(503, 'Add a Gemini API key in Provider Settings');
|
||||
model = normalizeGeminiModel(resolveModel());
|
||||
if (!model)
|
||||
return reply(503, 'The Gemini Live model configuration is invalid');
|
||||
if (limiter === undefined)
|
||||
limiter = makeCostRateLimiter(resolveRateLimit(), 30);
|
||||
if (limiter && !limiter(clientKey(req))) {
|
||||
res.setHeader('Retry-After', '60');
|
||||
return reply(
|
||||
429,
|
||||
'Gemini voice is starting too often. Try again shortly.',
|
||||
);
|
||||
}
|
||||
} catch {
|
||||
// Connect does not observe the handler's returned promise. A synchronous
|
||||
// configuration resolver must fail with a bounded, fixed response.
|
||||
return reply(503, MINT_FAILURE);
|
||||
}
|
||||
|
||||
const lifetime = new AbortController();
|
||||
let disconnected = false;
|
||||
const disconnect = () => {
|
||||
disconnected = true;
|
||||
lifetime.abort(new DOMException('Request closed', 'AbortError'));
|
||||
};
|
||||
const responseClosed = () => {
|
||||
if (!res.writableEnded) disconnect();
|
||||
};
|
||||
req.once('aborted', disconnect);
|
||||
res.once('close', responseClosed);
|
||||
const deadline = setTimeout(() => {
|
||||
lifetime.abort(
|
||||
new DOMException('Token request timed out', 'TimeoutError'),
|
||||
);
|
||||
}, timeoutMs);
|
||||
deadline.unref?.();
|
||||
let acquiredResponse;
|
||||
try {
|
||||
const inputMode = await readGeminiTokenRequest(req, lifetime.signal);
|
||||
lifetime.signal.throwIfAborted();
|
||||
const config = createGeminiLiveConfig(
|
||||
annotationGuidance,
|
||||
inputMode,
|
||||
tools,
|
||||
);
|
||||
const issuedAt = now();
|
||||
// REST uses bidiGenerateContentSetup. liveConnectConstraints is the SDK
|
||||
// input shape, converted before transport. With no fieldMask the complete
|
||||
// setup is locked to this server-owned configuration.
|
||||
const request = {
|
||||
uses: 1,
|
||||
expireTime: new Date(issuedAt + 30 * 60_000).toISOString(),
|
||||
newSessionExpireTime: new Date(issuedAt + 60_000).toISOString(),
|
||||
bidiGenerateContentSetup: { model: `models/${model}`, ...config },
|
||||
};
|
||||
const acquisition = Promise.resolve(
|
||||
fetchImpl(GEMINI_TOKEN_ENDPOINT, {
|
||||
method: 'POST',
|
||||
redirect: 'error',
|
||||
signal: lifetime.signal,
|
||||
headers: {
|
||||
'x-goog-api-key': apiKey,
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify(request),
|
||||
}),
|
||||
);
|
||||
// A fetch implementation can ignore abort. Bound this caller's wait and
|
||||
// discard a response that arrives after timeout/disconnection.
|
||||
void acquisition.then(
|
||||
(late) => {
|
||||
acquiredResponse = late;
|
||||
if (lifetime.signal.aborted) void late.body?.cancel().catch(() => {});
|
||||
},
|
||||
() => {},
|
||||
);
|
||||
const response = await waitForSignal(acquisition, lifetime.signal);
|
||||
lifetime.signal.throwIfAborted();
|
||||
if (
|
||||
response.redirected ||
|
||||
(response.url && response.url !== GEMINI_TOKEN_ENDPOINT)
|
||||
) {
|
||||
void response.body?.cancel().catch(() => {});
|
||||
throw new Error('Unexpected token response destination');
|
||||
}
|
||||
if (!response.ok) {
|
||||
// Never forward upstream bodies, headers or exception messages.
|
||||
void response.body?.cancel().catch(() => {});
|
||||
if (response.status === 429) {
|
||||
res.setHeader('Retry-After', '60');
|
||||
return reply(429, 'Gemini quota is unavailable. Try again later.');
|
||||
}
|
||||
if (response.status === 401 || response.status === 403)
|
||||
return reply(
|
||||
503,
|
||||
'Gemini access was refused. Check the API key and model access.',
|
||||
);
|
||||
if (response.status === 400 || response.status === 404)
|
||||
return reply(
|
||||
502,
|
||||
'Gemini Live is unavailable for the configured model.',
|
||||
);
|
||||
return reply(502, MINT_FAILURE);
|
||||
}
|
||||
const data = await readResponseJsonCapped(
|
||||
response,
|
||||
UPSTREAM_RESPONSE_BYTES,
|
||||
lifetime.signal,
|
||||
);
|
||||
lifetime.signal.throwIfAborted();
|
||||
if (
|
||||
!data ||
|
||||
Array.isArray(data) ||
|
||||
typeof data !== 'object' ||
|
||||
typeof data.name !== 'string' ||
|
||||
data.name.length > 8192 ||
|
||||
!/^auth_tokens\/[A-Za-z0-9._~+/=-]+$/.test(data.name) ||
|
||||
data.name.includes(apiKey)
|
||||
)
|
||||
throw new Error('Invalid ephemeral token');
|
||||
reply(200, {
|
||||
token: data.name,
|
||||
model,
|
||||
config,
|
||||
websocketUrl: GEMINI_LIVE_WEBSOCKET_URL,
|
||||
});
|
||||
} catch (error) {
|
||||
if (disconnected || req.aborted) return;
|
||||
if (error?.code === 'BODY_TOO_LARGE')
|
||||
return reply(413, 'Token request is too large');
|
||||
if (error?.code === 'BAD_CONTENT_TYPE')
|
||||
return reply(415, 'Use an empty JSON token request');
|
||||
if (error?.code === 'BAD_BODY')
|
||||
return reply(400, 'Use an empty JSON token request');
|
||||
reply(lifetime.signal.aborted ? 504 : 502, MINT_FAILURE);
|
||||
} finally {
|
||||
// Abort can happen after acquisition resolves but before this handler
|
||||
// resumes and passes ownership to the body reader.
|
||||
if (lifetime.signal.aborted)
|
||||
void acquiredResponse?.body?.cancel().catch(() => {});
|
||||
clearTimeout(deadline);
|
||||
req.off('aborted', disconnect);
|
||||
res.off('close', responseClosed);
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
import { readRequestBody } from '../common/request.js';
|
||||
import { admitSameSiteRequest } from '../../../src/localRequestGate.mjs';
|
||||
import {
|
||||
isAllowedHost,
|
||||
resolveAllowedHosts,
|
||||
} from '../../../build/allowedHosts.js';
|
||||
|
||||
/** Use the app's Host and request policies, with a required browser Origin. */
|
||||
export function isGeminiRequestOriginAllowed(
|
||||
req,
|
||||
configuredHost = '',
|
||||
configuredAllowedHosts,
|
||||
) {
|
||||
const host = req.headers?.host;
|
||||
const origin = req.headers?.origin;
|
||||
if (
|
||||
typeof host !== 'string' ||
|
||||
typeof origin !== 'string' ||
|
||||
!host ||
|
||||
!origin ||
|
||||
/[\s/?#@\\]/.test(host)
|
||||
)
|
||||
return false;
|
||||
const protocol = req.socket?.encrypted ? 'https:' : 'http:';
|
||||
if (
|
||||
!admitSameSiteRequest({
|
||||
hostHeader: host,
|
||||
protocol,
|
||||
origin,
|
||||
secFetchSite: req.headers?.['sec-fetch-site'],
|
||||
proxyHeaders: req.headers || {},
|
||||
}).ok
|
||||
)
|
||||
return false;
|
||||
// The shared gate compares a present Origin exactly with this authority;
|
||||
// this route also requires the Origin and an allowed Host.
|
||||
const boundHost = String(configuredHost || '')
|
||||
.trim()
|
||||
.toLowerCase();
|
||||
return isAllowedHost(host, resolveAllowedHosts(configuredAllowedHosts), [
|
||||
boundHost,
|
||||
]);
|
||||
}
|
||||
|
||||
/** Admit only the existing input gesture; callers cannot supply model/setup. */
|
||||
export async function readGeminiTokenRequest(req, signal, maxBytes = 1024) {
|
||||
const declared = Number(req.headers?.['content-length']);
|
||||
const fail = (code) =>
|
||||
Object.assign(new Error('Invalid token request'), { code });
|
||||
const body = await readRequestBody(
|
||||
req,
|
||||
maxBytes,
|
||||
Number.isFinite(declared) && declared > maxBytes
|
||||
? AbortSignal.abort(fail('BODY_TOO_LARGE'))
|
||||
: signal,
|
||||
);
|
||||
if (!body.trim()) return 'open-mic';
|
||||
if (
|
||||
!/^application\/json(?:\s*;|$)/i.test(req.headers?.['content-type'] || '')
|
||||
)
|
||||
throw fail('BAD_CONTENT_TYPE');
|
||||
let parsed;
|
||||
try {
|
||||
parsed = JSON.parse(body);
|
||||
} catch {
|
||||
throw fail('BAD_BODY');
|
||||
}
|
||||
if (!parsed || Array.isArray(parsed) || typeof parsed !== 'object')
|
||||
throw fail('BAD_BODY');
|
||||
const keys = Object.keys(parsed);
|
||||
if (!keys.length) return 'open-mic';
|
||||
if (
|
||||
keys.length === 1 &&
|
||||
keys[0] === 'inputMode' &&
|
||||
parsed.inputMode === 'push-to-talk'
|
||||
)
|
||||
return 'push-to-talk';
|
||||
throw fail('BAD_BODY');
|
||||
}
|
||||
@@ -19,6 +19,7 @@ import { adsbLolProxy } from './aircraft/adsb-lol.js';
|
||||
import { aisLiveProxy } from './vessels/ais-live.js';
|
||||
import { trackBackfillProxies } from './aircraft/tracks.js';
|
||||
import { openAiRealtimeProxy } from './openai.js';
|
||||
import { geminiLiveProxy } from './gemini.js';
|
||||
import { googlePlacesContextProxy } from './places.js';
|
||||
import { mapillaryProxy } from './mapillary.js';
|
||||
import { keySetupEndpoint } from '../standalone/key-setup.js';
|
||||
@@ -30,9 +31,9 @@ import { windProxy } from './wind.js';
|
||||
|
||||
/**
|
||||
* Construct the local provider plugins in their established order.
|
||||
* `realtime` configures the voice session token endpoint.
|
||||
* `realtime` configures OpenAI; `gemini` explicitly opts into Gemini Live.
|
||||
*/
|
||||
function localProviderPlugins({ realtime } = {}) {
|
||||
function localProviderPlugins({ realtime, gemini } = {}) {
|
||||
return [
|
||||
openSkyProxy(),
|
||||
celestrakProxy(),
|
||||
@@ -55,6 +56,7 @@ function localProviderPlugins({ realtime } = {}) {
|
||||
aisLiveProxy(),
|
||||
trackBackfillProxies(),
|
||||
openAiRealtimeProxy({ realtime }),
|
||||
...(gemini ? [geminiLiveProxy({ realtime: gemini })] : []),
|
||||
googlePlacesContextProxy(),
|
||||
windProxy(),
|
||||
weatherProxy(),
|
||||
|
||||
@@ -13,6 +13,8 @@ export function launchLibraryRequestHeaders(token = process.env.LL2_API_TOKEN) {
|
||||
const normalized = String(token || '').trim();
|
||||
return {
|
||||
Accept: 'application/json',
|
||||
'User-Agent':
|
||||
'gods-eye-view-launch-library-proxy/1.0 (+https://github.com/bilawalsidhu/gods-eye-view)',
|
||||
...(normalized ? { Authorization: `Token ${normalized}` } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -8,6 +8,8 @@ import {
|
||||
terrainPointKey,
|
||||
validTerrainResult,
|
||||
} from '../../src/data/terrainHeightsProxy.js';
|
||||
import { makeRateLimiter, clientKey } from './common/rate-limit.js';
|
||||
import { admitSameSiteKeyless } from './common/same-site.js';
|
||||
|
||||
/**
|
||||
* Re:Earth terrain point-height proxy: batched lon/lat → ellipsoidal height
|
||||
@@ -20,11 +22,31 @@ import {
|
||||
* Only missing/stale points go upstream; the response is rebuilt in exact
|
||||
* request order. Larger requests are chunked sequentially, and one failing
|
||||
* chunk does not discard the chunks that resolved.
|
||||
*
|
||||
* Bounds, because every distinct 5dp point is its own key: coordinates must be
|
||||
* in WGS-84 range (rejected at parse), the point cache has an entry ceiling
|
||||
* with oldest-first eviction, an over-large cache file is not read back, and
|
||||
* the route carries an always-on per-client limiter. The route also refuses
|
||||
* cross-site browser requests, so another web page cannot use the visitor's
|
||||
* browser to fill the cache or spend the visitor's own rate-limit quota.
|
||||
* Reverse-proxied same-origin requests are admitted: terrain is keyless and
|
||||
* needed for rendering, and a 403 would send every chunk to the geoid fallback.
|
||||
*/
|
||||
export function terrainHeightsProxy() {
|
||||
export function terrainHeightsProxy(options = {}) {
|
||||
// Destructured in the body, not the signature: `proxyErrorResponses.test.mjs`
|
||||
// extracts these proxies by slicing to the first closing brace in column
|
||||
// zero, so a multi-line parameter list would cut the function in half.
|
||||
// Overridable so the ceilings can be exercised as behaviour rather than
|
||||
// asserted as constants; production constructs this with no arguments.
|
||||
const {
|
||||
cacheMaxEntries = 50_000,
|
||||
cacheMaxDiskBytes = 24 * 1024 * 1024,
|
||||
cachePath,
|
||||
} = options;
|
||||
const TTL_MS = 30 * 24 * 3600_000;
|
||||
const CACHE_DIR = path.join(process.cwd(), '.gev-cache');
|
||||
const CACHE_PATH = path.join(CACHE_DIR, 'terrain-heights.json');
|
||||
const CACHE_PATH =
|
||||
cachePath || path.join(process.cwd(), '.gev-cache', 'terrain-heights.json');
|
||||
const CACHE_DIR = path.dirname(CACHE_PATH);
|
||||
// Sized against measured upstream latency, not the documented page cap.
|
||||
// Re:Earth serves 87-186 ms/point depending on load (observed 2026-09-12,
|
||||
// a 2x swing within one hour). 256 points therefore costs 22-48 s and blows
|
||||
@@ -34,9 +56,50 @@ export function terrainHeightsProxy() {
|
||||
// that range.
|
||||
const UPSTREAM_CHUNK = 64;
|
||||
const MAX_POINTS = 2000;
|
||||
/**
|
||||
* Entry ceiling for the point cache, and with it the file written from it.
|
||||
*
|
||||
* Every distinct 5dp point is a separate key, so without a ceiling the map
|
||||
* — and the JSON dumped from it verbatim — grow for as long as a caller
|
||||
* supplies new coordinates. Measured against real use rather than guessed:
|
||||
* this checkout's `.gev-cache/terrain-heights.json` held 6,189 entries in
|
||||
* 1.0 MB after ordinary sessions, i.e. ~168 bytes/entry. 50,000 entries is
|
||||
* eight times that working set and ~8 MB on disk, comfortably inside the
|
||||
* 24 MB ceiling `rocketLaunchesProxy` already applies to its own cache file.
|
||||
*/
|
||||
const CACHE_MAX_ENTRIES = cacheMaxEntries;
|
||||
/** Refuse to load a cache file larger than this (mirrors rocketLaunchesProxy). */
|
||||
const CACHE_MAX_DISK_BYTES = cacheMaxDiskBytes;
|
||||
/**
|
||||
* Requests/min/client. Terrain is keyless and free, so this guards the point
|
||||
* cache and the upstream community service rather than a bill. Sized from
|
||||
* measurement: across a full `npm run test:track` run — the heaviest
|
||||
* exercise this app has — the route served 10 requests, peaking at 8/min.
|
||||
* 90/min is an order of magnitude above that, and the band `/api/overpass`
|
||||
* and `/api/military-installations` already use.
|
||||
*/
|
||||
const RATE_PER_MIN = 90;
|
||||
const allow = makeRateLimiter({
|
||||
windowMs: 60_000,
|
||||
max: RATE_PER_MIN,
|
||||
globalMax: RATE_PER_MIN * 3,
|
||||
});
|
||||
|
||||
/** @type {Map<string, {at:number, result:object}>} keyed by canonical 5dp lon/lat. */
|
||||
const mem = new Map();
|
||||
|
||||
/**
|
||||
* Drop the oldest entries until the map is back inside its ceiling.
|
||||
* Oldest-first by fetch time, the same shape `trackBackfillProxies` uses.
|
||||
* Terrain does not move, so evicting an old point costs one refetch, never
|
||||
* a wrong answer.
|
||||
*/
|
||||
function evictOverflow() {
|
||||
if (mem.size <= CACHE_MAX_ENTRIES) return;
|
||||
const byAge = [...mem.entries()].sort((a, b) => a[1].at - b[1].at);
|
||||
for (let i = 0; i < byAge.length && mem.size > CACHE_MAX_ENTRIES; i += 1)
|
||||
mem.delete(byAge[i][0]);
|
||||
}
|
||||
/** @type {Map<string, Promise<Array<object>>>} single-flight per missing-point subset. */
|
||||
const inflight = new Map();
|
||||
let diskLoaded = false;
|
||||
@@ -47,6 +110,21 @@ export function terrainHeightsProxy() {
|
||||
if (diskLoaded) return;
|
||||
diskLoaded = true;
|
||||
try {
|
||||
// A cache file past the ceiling is not read back into memory: loading it
|
||||
// would reinstate exactly the unbounded map the ceiling exists to stop.
|
||||
// The next flush rewrites a bounded one.
|
||||
//
|
||||
// "Cannot measure it" is not "too large": if stat is unavailable, fall
|
||||
// through and let readFile decide, so an absent file behaves exactly as
|
||||
// it did before this guard existed.
|
||||
let sizeBytes = null;
|
||||
try {
|
||||
sizeBytes = (await fsp.stat(CACHE_PATH)).size;
|
||||
} catch {
|
||||
sizeBytes = null;
|
||||
}
|
||||
if (sizeBytes !== null && sizeBytes > CACHE_MAX_DISK_BYTES)
|
||||
throw new Error('cache file too large');
|
||||
const parsed = JSON.parse(await fsp.readFile(CACHE_PATH, 'utf8'));
|
||||
const pointEntries =
|
||||
parsed?.version === 2 &&
|
||||
@@ -87,6 +165,7 @@ export function terrainHeightsProxy() {
|
||||
}
|
||||
diskDirty = mem.size > 0;
|
||||
}
|
||||
evictOverflow();
|
||||
} catch {
|
||||
/* no disk cache yet */
|
||||
}
|
||||
@@ -157,12 +236,25 @@ export function terrainHeightsProxy() {
|
||||
|
||||
const installMiddleware = (server) => {
|
||||
server.middlewares.use('/api/terrain/heights', async (req, res) => {
|
||||
// Refuse cross-site browser requests (an <img> or fetch from another
|
||||
// site) before the limiter, cache or upstream see them. Same-origin app
|
||||
// requests, non-browser loopback callers (MCP panel_request) and the app
|
||||
// served behind a reverse proxy pass.
|
||||
if (admitSameSiteKeyless(req, res)) return;
|
||||
const send = (status, bodyObj) => {
|
||||
if (res.headersSent) return;
|
||||
res.writeHead(status, { 'Content-Type': 'application/json' });
|
||||
res.end(JSON.stringify(bodyObj));
|
||||
};
|
||||
try {
|
||||
if (!allow(clientKey(req))) {
|
||||
res.writeHead(429, {
|
||||
'Content-Type': 'application/json',
|
||||
'Retry-After': '10',
|
||||
});
|
||||
res.end(JSON.stringify({ error: 'Rate limit exceeded' }));
|
||||
return;
|
||||
}
|
||||
await loadDiskOnce();
|
||||
const parsedUrl = new URL(req.url || '', 'http://internal');
|
||||
const rawPoints = parsedUrl.searchParams.get('points');
|
||||
@@ -187,7 +279,10 @@ export function terrainHeightsProxy() {
|
||||
fetchMissing: fetchMissingSingleFlight,
|
||||
ttlMs: TTL_MS,
|
||||
});
|
||||
if (outcome.cacheChanged) diskDirty = true;
|
||||
if (outcome.cacheChanged) {
|
||||
evictOverflow();
|
||||
diskDirty = true;
|
||||
}
|
||||
if (outcome.upstreamError) {
|
||||
console.warn(
|
||||
'[terrain-heights-proxy] refresh incomplete' +
|
||||
|
||||
@@ -16,9 +16,10 @@ export default defineConfig(({ command, mode }) => {
|
||||
for (const [key, value] of Object.entries(loaded)) {
|
||||
if (process.env[key] === undefined) process.env[key] = value;
|
||||
}
|
||||
const tools = standaloneVoiceTools();
|
||||
return createBrowserViteConfig({
|
||||
plugins: [
|
||||
...localProviderPlugins({ realtime: { tools: standaloneVoiceTools() } }),
|
||||
...localProviderPlugins({ realtime: { tools }, gemini: { tools } }),
|
||||
localMcpPlugin(),
|
||||
apiNotFoundPlugin(),
|
||||
unwatchedRuntimeDirsPlugin(),
|
||||
|
||||
Reference in new issue
Block a user