Merge remote-tracking branch 'upstream/main' into feat/opensky-compress-cached-body

# Conflicts:
#	server/providers/aircraft/opensky.js
#	src/tooling/liveProviders.test.mjs
This commit is contained in:
daikaginza committed 2026-10-23 13:30:47 +07:00
commit a160a9e953
571 files changed
+51737 -19428

No files matched your search

+64 -47
View File
@@ -1,4 +1,7 @@
import { readResponseTextCapped } from '../common/http.js';
import {
coalesceProxyRequest,
readResponseTextCapped,
} from '../common/http.js';
/**
* Vite plugin: adsb.lol military aircraft proxy with 12 s response cache.
@@ -19,6 +22,7 @@ import { readResponseTextCapped } from '../common/http.js';
export function adsbLolProxy() {
/** @type {string|null} Cached upstream JSON body. */
let _cache = null;
const inFlight = new Map();
/** @type {number} Epoch-ms when the cache was populated. */
let _cacheAt = 0;
/** @type {number} Epoch-ms until which upstream is not contacted. */
@@ -104,60 +108,73 @@ export function adsbLolProxy() {
);
return;
}
const signal = AbortSignal.timeout(UPSTREAM_TIMEOUT_MS);
const upstream = await fetch('https://api.adsb.lol/v2/mil', {
headers: { 'User-Agent': 'gods-eye-view-adsblol-proxy/1.0' },
signal,
});
if (upstream.ok) {
const refresh = coalesceProxyRequest(inFlight, 'global', async () => {
const result = (status, body, cacheStatus, extra = {}) => ({
status,
body,
cacheStatus,
extra,
});
const signal = AbortSignal.timeout(UPSTREAM_TIMEOUT_MS);
const upstream = await fetch('https://api.adsb.lol/v2/mil', {
headers: { 'User-Agent': 'gods-eye-view-adsblol-proxy/1.0' },
signal,
});
if (upstream.ok) {
const body = await readResponseTextCapped(
upstream,
ADSBLOL_MIL_MAX_RESPONSE_BYTES,
signal,
);
_cache = body;
_cacheAt = Date.now();
_cooldownUntil = 0;
_cooldownStatus = 0;
return result(200, body, 'MISS');
}
if (upstream.status === 429 || upstream.status >= 500) {
const failedAt = Date.now();
_cooldownUntil = failedAt + cooldownFor(upstream, failedAt);
_cooldownStatus = upstream.status;
console.warn(
`[adsb.lol Proxy] upstream ${upstream.status}; cooling down ${Math.round((_cooldownUntil - Date.now()) / 1000)} s${_cache ? ', serving stale' : ''}`,
);
if (_cache) {
// Do not wait for a failing response body before serving usable data.
upstream.body?.cancel().catch(() => {});
return result(200, _cache, 'STALE', {
'X-ADS-B-Upstream-Status': String(upstream.status),
'X-ADS-B-Retry-After-Seconds': String(
Math.ceil((_cooldownUntil - Date.now()) / 1000),
),
});
}
}
const body = await readResponseTextCapped(
upstream,
ADSBLOL_MIL_MAX_RESPONSE_BYTES,
signal,
);
_cache = body;
_cacheAt = Date.now();
_cooldownUntil = 0;
_cooldownStatus = 0;
serve(res, 200, body, 'MISS');
return;
}
if (upstream.status === 429 || upstream.status >= 500) {
const failedAt = Date.now();
_cooldownUntil = failedAt + cooldownFor(upstream, failedAt);
_cooldownStatus = upstream.status;
console.warn(
`[adsb.lol Proxy] upstream ${upstream.status}; cooling down ${Math.round((_cooldownUntil - Date.now()) / 1000)} s${_cache ? ', serving stale' : ''}`,
return result(
upstream.status,
body,
'MISS',
_cooldownUntil > Date.now()
? {
'Retry-After': String(
Math.ceil((_cooldownUntil - Date.now()) / 1000),
),
}
: {},
);
if (_cache) {
// Do not wait for a failing response body before serving usable data.
upstream.body?.cancel().catch(() => {});
serve(res, 200, _cache, 'STALE', {
'X-ADS-B-Upstream-Status': String(upstream.status),
'X-ADS-B-Retry-After-Seconds': String(
Math.ceil((_cooldownUntil - Date.now()) / 1000),
),
});
return;
}
}
const body = await readResponseTextCapped(
upstream,
ADSBLOL_MIL_MAX_RESPONSE_BYTES,
signal,
);
});
const response = await refresh.promise;
serve(
res,
upstream.status,
body,
'MISS',
_cooldownUntil > Date.now()
? {
'Retry-After': String(
Math.ceil((_cooldownUntil - Date.now()) / 1000),
),
}
: {},
response.status,
response.body,
response.cacheStatus,
response.extra,
);
} catch (e) {
console.error('[adsb.lol Proxy]', e.message);
+149 -124
View File
@@ -74,15 +74,24 @@ async function fetchOpenSkyStates(headers) {
try {
const signal = AbortSignal.timeout(OPENSKY_ATTEMPT_TIMEOUT_MS);
const upstream = await fetch(OPENSKY_STATES_URL, { headers, signal });
if (upstream.status === 429) {
if (upstream.status === 429 || upstream.status >= 500) {
// Start the cooldown from the headers and drop the rejected body
// unread, so a stalled or oversized 429 body cannot skip it.
_openskyCooldownUntil =
Date.now() + openSkyRateLimitCooldownMs(upstream);
Date.now() +
(upstream.status === 429
? openSkyRateLimitCooldownMs(upstream)
: 15000);
_openskyCooldownStatus = upstream.status;
void upstream.body?.cancel().catch(() => {});
return {
upstream,
body: JSON.stringify({ error: 'OpenSky rate limited.' }),
body: JSON.stringify({
error:
upstream.status === 429
? 'OpenSky rate limited.'
: 'OpenSky upstream unavailable.',
}),
};
}
const body = await readResponseTextCapped(
@@ -116,6 +125,8 @@ async function fetchOpenSkyStates(headers) {
let _openskyTtlMs = OPENSKY_CACHE_MS;
/** @type {number} Epoch-ms before which no upstream fetch is attempted. */
let _openskyCooldownUntil = 0;
let _openskyCooldownStatus = 429;
const _openskySnapshotInFlight = new Map();
/**
* Picks the cache TTL from the remaining daily credit budget.
* Client polls every 30 s, so tiers ≤30 s cost the same 480 credits/h; the
@@ -478,7 +489,9 @@ export function openSkyProxy() {
requestedMode: cachedMeta.requestedMode || requestedMode,
usedMode: cachedMeta.usedMode || 'unknown',
reason: isStale
? 'rate_limited_serving_stale'
? _openskyCooldownStatus === 429
? 'rate_limited_serving_stale'
: `upstream_http_${_openskyCooldownStatus}_serving_stale`
: cachedMeta.reason || 'cached',
staleSeconds: isStale
? (now - _openskyCacheTime) / 1000
@@ -506,80 +519,145 @@ export function openSkyProxy() {
)
return;
res.writeHead(
429,
_openskyCooldownStatus,
buildOpenSkyHeaders({
cacheStatus: 'COOLDOWN',
requestedMode,
usedMode: 'none',
reason: 'rate_limited',
reason:
_openskyCooldownStatus === 429
? 'rate_limited'
: `upstream_http_${_openskyCooldownStatus}`,
retryAfterSeconds: (_openskyCooldownUntil - now) / 1000,
}),
);
res.end(
JSON.stringify({
error: 'OpenSky rate limited; proxy cooling down.',
error: 'OpenSky upstream unavailable; proxy cooling down.',
}),
);
return;
}
const basicUser = process.env.OPENSKY_USERNAME || '';
const basicPass = process.env.OPENSKY_PASSWORD || '';
const hasBasicCreds = Boolean(basicUser && basicPass);
const headers = { Accept: 'application/json' };
let usedMode = 'anon';
let reason = 'forced_anonymous';
const refresh = coalesceProxyRequest(
_openskySnapshotInFlight,
'global',
async () => {
const basicUser = process.env.OPENSKY_USERNAME || '';
const basicPass = process.env.OPENSKY_PASSWORD || '';
const hasBasicCreds = Boolean(basicUser && basicPass);
const headers = { Accept: 'application/json' };
let usedMode = 'anon';
let reason = 'forced_anonymous';
if (requestedMode === 'basic') {
if (hasBasicCreds) {
headers.Authorization = `Basic ${Buffer.from(`${basicUser}:${basicPass}`).toString('base64')}`;
usedMode = 'basic';
reason = 'basic_credentials';
} else {
reason = 'missing_basic_creds';
}
} else if (requestedMode === 'oauth') {
const token = await getOpenSkyToken();
if (token) {
headers.Authorization = `Bearer ${token}`;
usedMode = 'oauth';
reason = 'oauth_token';
} else {
reason = 'oauth_invalid_or_missing';
}
} else if (requestedMode === 'auto') {
const token = await getOpenSkyToken();
if (token) {
headers.Authorization = `Bearer ${token}`;
usedMode = 'oauth';
reason = 'oauth_token';
} else if (hasBasicCreds) {
headers.Authorization = `Basic ${Buffer.from(`${basicUser}:${basicPass}`).toString('base64')}`;
usedMode = 'basic';
reason = 'oauth_unavailable_fallback_basic';
} else {
reason = 'missing_oauth_and_basic_creds';
}
}
if (requestedMode === 'basic') {
if (hasBasicCreds) {
headers.Authorization = `Basic ${Buffer.from(`${basicUser}:${basicPass}`).toString('base64')}`;
usedMode = 'basic';
reason = 'basic_credentials';
} else {
reason = 'missing_basic_creds';
}
} else if (requestedMode === 'oauth') {
const token = await getOpenSkyToken();
if (token) {
headers.Authorization = `Bearer ${token}`;
usedMode = 'oauth';
reason = 'oauth_token';
} else {
reason = 'oauth_invalid_or_missing';
}
} else if (requestedMode === 'auto') {
const token = await getOpenSkyToken();
if (token) {
headers.Authorization = `Bearer ${token}`;
usedMode = 'oauth';
reason = 'oauth_token';
} else if (hasBasicCreds) {
headers.Authorization = `Basic ${Buffer.from(`${basicUser}:${basicPass}`).toString('base64')}`;
usedMode = 'basic';
reason = 'oauth_unavailable_fallback_basic';
} else {
reason = 'missing_oauth_and_basic_creds';
}
}
let { upstream, body } = await fetchOpenSkyStates(headers);
// Auto-mode fallback: if OAuth was rejected, retry with Basic credentials
if (
(upstream.status === 401 || upstream.status === 403) &&
requestedMode === 'auto' &&
usedMode === 'oauth' &&
hasBasicCreds
) {
const retryHeaders = {
Accept: 'application/json',
Authorization: `Basic ${Buffer.from(`${basicUser}:${basicPass}`).toString('base64')}`,
};
({ upstream, body } = await fetchOpenSkyStates(retryHeaders));
usedMode = 'basic';
reason = 'oauth_rejected_fallback_basic';
}
let { upstream, body } = await fetchOpenSkyStates(headers);
// Auto-mode fallback: if OAuth was rejected, retry with Basic credentials
if (
(upstream.status === 401 || upstream.status === 403) &&
requestedMode === 'auto' &&
usedMode === 'oauth' &&
hasBasicCreds
) {
const retryHeaders = {
Accept: 'application/json',
Authorization: `Basic ${Buffer.from(`${basicUser}:${basicPass}`).toString('base64')}`,
};
({ upstream, body } = await fetchOpenSkyStates(retryHeaders));
usedMode = 'basic';
reason = 'oauth_rejected_fallback_basic';
}
const sourceEpochMs = upstream.ok ? openSkySourceEpochMs(body) : null;
const sourceEpochMs = upstream.ok
? openSkySourceEpochMs(body)
: null;
// Refine the reason string to reflect the actual outcome
if (upstream.ok && reason === 'forced_anonymous') {
reason = 'anonymous_ok';
} else if (
upstream.ok &&
usedMode === 'basic' &&
reason === 'basic_credentials'
) {
reason = 'basic_ok';
} else if (
upstream.ok &&
usedMode === 'oauth' &&
reason === 'oauth_token'
) {
reason = 'oauth_ok';
}
// Only cache successful responses — error responses (401/403/429/5xx)
// should not be served from cache on subsequent requests
if (upstream.ok) {
// The frame is built before either slot is written so the two
// are always published together: a request arriving mid-refresh
// reads the previous body with the previous body's frame.
const frame = await gzipBody(body);
_openskyCacheBody = body;
_openskyCacheGzip = frame;
_openskyCacheStatus = upstream.status;
_openskyCacheTime = now;
_openskyCacheSourceEpochMs = sourceEpochMs;
_openskyCacheMeta = {
requestedMode,
usedMode,
reason,
};
// Credit governor: adapt the cache TTL to the remaining daily
// budget so a continuously-open app stretches its polls instead of
// exhausting the quota mid-day. Success also clears any cooldown.
const remaining = Number(
upstream.headers.get('x-rate-limit-remaining'),
);
_openskyTtlMs = openskyAdaptiveTtlMs(remaining);
_openskyCooldownUntil = 0;
}
return {
upstream,
body,
usedMode,
reason,
hasBasicCreds,
sourceEpochMs,
};
},
);
let { upstream, body, usedMode, reason, hasBasicCreds, sourceEpochMs } =
await refresh.promise;
if (
upstream.ok &&
openSkySourceIsStale(sourceEpochMs, now) &&
@@ -590,25 +668,16 @@ export function openSkyProxy() {
'opensky_snapshot_stale_regional_fallback',
))
) {
// Keep the last global snapshot available as a fail-soft cache,
// but do not label or render it as a fresh live result.
// The frame is built before either slot is written so the two are
// always published together: a request arriving mid-refresh reads the
// previous body with the previous body's frame, never a mismatch.
const frame = await gzipBody(body);
_openskyCacheBody = body;
_openskyCacheGzip = frame;
_openskyCacheStatus = upstream.status;
_openskyCacheTime = now;
_openskyCacheSourceEpochMs = sourceEpochMs;
_openskyCacheMeta = { requestedMode, usedMode, reason };
return;
}
if (upstream.status === 429) {
reason = 'rate_limited';
if (upstream.status === 429 || upstream.status >= 500) {
reason =
upstream.status === 429
? 'rate_limited'
: `upstream_http_${upstream.status}`;
// Credit governor: fetchOpenSkyStates already started the cooldown
// from OpenSky's retry-after — no upstream attempts until then.
const cooldownMs = openSkyRateLimitCooldownMs(upstream);
const cooldownMs = Math.max(0, _openskyCooldownUntil - Date.now());
// Serve the last-good body instead of the 429 when we have one —
// the layer keeps rendering (STALE-cued) instead of dying.
if (_openskyCacheBody && _openskyCacheStatus === 200) {
@@ -622,7 +691,7 @@ export function openSkyProxy() {
cacheStatus: 'STALE',
requestedMode,
usedMode,
reason: 'rate_limited_serving_stale',
reason: `${reason}_serving_stale`,
staleSeconds: (now - _openskyCacheTime) / 1000,
retryAfterSeconds: cooldownMs / 1000,
}),
@@ -681,53 +750,9 @@ export function openSkyProxy() {
}
}
// Refine the reason string to reflect the actual outcome
if (upstream.ok && reason === 'forced_anonymous') {
reason = 'anonymous_ok';
} else if (
upstream.ok &&
usedMode === 'basic' &&
reason === 'basic_credentials'
) {
reason = 'basic_ok';
} else if (
upstream.ok &&
usedMode === 'oauth' &&
reason === 'oauth_token'
) {
reason = 'oauth_ok';
}
// Only cache successful responses — error responses (401/403/429/5xx)
// should not be served from cache on subsequent requests
if (upstream.ok) {
// Published as a pair, for the reason given at the other cache fill.
const frame = await gzipBody(body);
_openskyCacheBody = body;
_openskyCacheGzip = frame;
_openskyCacheStatus = upstream.status;
_openskyCacheTime = now;
_openskyCacheSourceEpochMs = sourceEpochMs;
_openskyCacheMeta = {
requestedMode,
usedMode,
reason,
};
// Credit governor: adapt the cache TTL to the remaining daily
// budget so a continuously-open app stretches its polls instead of
// exhausting the quota mid-day. Success also clears any cooldown.
const remaining = Number(
upstream.headers.get('x-rate-limit-remaining'),
);
_openskyTtlMs = openskyAdaptiveTtlMs(remaining);
_openskyCooldownUntil = 0;
}
// Only the branch above caches, so the gzip slot belongs to this body
// only when this body is the one it just stored. An error response
// (401/403/429/5xx) is not cached and must not borrow the previous
// snapshot's frame — identity comparison, not `upstream.ok`, because it
// stays correct if the caching condition above ever changes.
// Only the coalesced refresh caches, so the gzip slot belongs to this
// body only when this body is the one it stored. An error response is
// not cached and must not borrow the previous snapshot's frame.
const encoded = encodedBody(
req,
body,
+65 -35
View File
@@ -1,11 +1,9 @@
import { createCctvCatalog } from './cctv/catalog.js';
import {
normalizeFeedType,
isVideoFeedType,
toFiniteNumber,
} from './cctv/normalize.js';
import { normalizeFeedType, isVideoFeedType } from './cctv/normalize.js';
import {
buildSyntheticCctvSvg,
cctvLiveMediaType,
cctvRasterImageType,
proxyMediaResponse,
fetchCctvImageFromUpstream,
fetchTxdotSnapshot,
@@ -17,6 +15,8 @@ import {
CCTV_MAX_SOURCES_CEILING,
} from './cctv/constants.js';
import { sanitizeCctvRangeHeader } from './cctv/range.js';
import { admitSameSiteKeyless } from './common/same-site.js';
import { PROXY_SIGNALS } from '../../src/localRequestGate.mjs';
import { createHlsPuller } from './cctv/stream.js';
import { googleServerApiKey } from './places/google-key.js';
export { CCTV_FRAME_FETCH_TIMEOUT_MS, fetchCctvImageFromUpstream };
@@ -116,8 +116,15 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
headers: { 'User-Agent': 'gods-eye-view-cctv-proxy/1.0' },
signal: AbortSignal.timeout(CCTV_FRAME_FETCH_TIMEOUT_MS),
});
const svType = svResp.headers.get('content-type') || '';
if (!svResp.ok || !svType.startsWith('image/')) return null;
const svType = cctvRasterImageType(svResp.headers.get('content-type'));
if (!svResp.ok || !svType) {
try {
await svResp.body?.cancel();
} catch {
/* no-op */
}
return null;
}
return {
ok: true,
@@ -135,15 +142,19 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
});
server.middlewares.use('/api/cctv', async (req, res) => {
try {
const url = new URL(req.url || '/', 'http://localhost');
// Refuse cross-site requests before any catalog or upstream work. The
// keyless gate keeps the routes reachable behind a reverse proxy; the
// key-bearing Street View fallback below has its own check.
if (admitSameSiteKeyless(req, res)) return;
const sources = await getCctvSources();
const sourceById = new Map(
sources.map((source) => [source.id, source]),
);
const url = new URL(req.url || '/', 'http://localhost');
if (url.pathname === '/sources') {
const body = {
trimmedPacks: getCctvSources.trimmedPacks?.() ?? [],
trimmedPacks: getCctvSources.trimmedPacks?.(sources) ?? [],
sources: sources.map((source) => ({
id: source.id,
pack: source.pack,
@@ -217,8 +228,8 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
const mediaUrl = source?.url || '';
const feedType = normalizeFeedType(source?.feedType || 'image');
// Live feeds only. A still camera is served by /frame, whose fetch
// keeps redirects on the registered origin, accepts only image/* and
// caps the body; relaying it here would bypass all three.
// keeps redirects on the registered origin, accepts only raster
// images and caps the body; relaying it here would bypass all three.
if (!isVideoFeedType(feedType) && feedType !== 'mjpeg') {
res.writeHead(404, {
'Content-Type': 'application/json',
@@ -382,7 +393,6 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
}
return;
}
const contentType = upstream.headers.get('content-type') || '';
if (!upstream.ok) {
try {
await upstream.body?.cancel();
@@ -407,31 +417,42 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
return;
}
if (
isVideoFeedType(feedType) &&
!(
contentType.startsWith('video/') ||
contentType.includes('mpegurl')
)
) {
// The relayed type is the feed kind's own, checked before the
// camera counts as healthy and before any byte is relayed.
const contentType = cctvLiveMediaType(
feedType,
upstream.headers.get('content-type'),
);
if (!contentType) {
try {
await upstream.body?.cancel();
} catch {
/* already closed */
}
setHealth(cameraId, {
status: 'degraded',
sourceKind: 'upstream',
label: source?.provider || 'Configured source',
message: `Unexpected media type ${contentType || 'unknown'}`,
message: 'Unexpected media type',
});
} else {
setHealth(cameraId, {
status: 'ok',
sourceKind: isVideoFeedType(feedType) ? 'live' : 'snapshot',
label: source?.provider || 'Configured source',
message: isVideoFeedType(feedType)
? 'Live stream connected'
: 'Snapshot feed connected',
res.writeHead(502, {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
});
res.end(JSON.stringify({ error: 'Media proxy failed' }));
return;
}
setHealth(cameraId, {
status: 'ok',
sourceKind: isVideoFeedType(feedType) ? 'live' : 'snapshot',
label: source?.provider || 'Configured source',
message: isVideoFeedType(feedType)
? 'Live stream connected'
: 'Snapshot feed connected',
});
await proxyMediaResponse(res, upstream, {
contentType,
sourceHeader: isVideoFeedType(feedType)
? 'live-media'
: 'upstream-image',
@@ -524,13 +545,22 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
return;
}
const sv = await streetViewFallback({
lat,
lon,
heading,
fov,
pitch,
});
// The Street View fallback spends the Google key, so a forwarded
// request (whose client this server cannot see) gets the placeholder,
// as if no key were configured. Any forwarding header counts, even an
// empty one.
const forwarded = PROXY_SIGNALS.some(
(name) => req.headers?.[name] !== undefined,
);
const sv = forwarded
? null
: await streetViewFallback({
lat,
lon,
heading,
fov,
pitch,
});
if (sv?.ok) {
setHealth(cameraId, {
status: 'degraded',
+59 -12
View File
@@ -5,6 +5,7 @@ import { allocateSourceCap, resolveCatalogCap } from './cap.js';
import { loadGroundHeights, joinGroundHeights } from './groundHeights.js';
import { cameraRegion, normalizeSourceItem } from './normalize.js';
import { loadLithuaniaSources } from './lithuania.js';
import { loadTaiwanSources } from './taiwan.js';
import {
loadAustinSourcesFromOpenData,
loadCaltransSourcesFromOpenData,
@@ -19,12 +20,14 @@ import {
loadNswSourcesFromOpenData,
loadCalgarySourcesFromOpenData,
loadCataloniaSourcesFromOpenData,
loadHkSourcesFromOpenData,
loadDelDOTSourcesFromOpenData,
loadKingCountySourcesFromOpenData,
loadWsdotSourcesFromOpenData,
loadVegvesenSourcesFromOpenData,
loadNycSourcesFromOpenData,
loadTorontoSourcesFromOpenData,
loadFloridaSourcesFromOpenData,
} from './sources.js';
/** Env kill switch: unset or anything but "0" means enabled. */
@@ -105,6 +108,11 @@ const LIVE_PACKS = [
enabled: () => envEnabled('CCTV_CATALONIA_ENABLED'),
load: loadCataloniaSourcesFromOpenData,
},
{
name: 'hk',
enabled: () => envEnabled('CCTV_HK_ENABLED'),
load: loadHkSourcesFromOpenData,
},
{
name: 'deldot',
enabled: () => envEnabled('CCTV_DELDOT_ENABLED'),
@@ -144,6 +152,19 @@ 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,
},
{
name: 'taiwan',
enabled: () => envEnabled('CCTV_TAIWAN_ENABLED'),
load: loadTaiwanSources,
},
];
/**
* Load CCTV sources from a local JSON file (CCTV_SOURCES_FILE env or default).
@@ -195,16 +216,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 +243,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 +357,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 +371,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;
}
+102 -2
View File
@@ -18,7 +18,7 @@ export const CCTV_MAX_SOURCES_CEILING = 5000;
export const AUSTIN_DOWNTOWN = { lat: 30.2672, lon: -97.7431 };
/** Caltrans CCTV: one JSON feed per district, identical schema statewide. */
/** TxDOT ITS: one keyless JSON catalog per district (25 districts statewide). */
export const TXDOT_ORIGIN = 'https://its.txdot.gov';
const TXDOT_ORIGIN = 'https://its.txdot.gov';
export const TXDOT_CCTV_STATUS_URL = (district) =>
`${TXDOT_ORIGIN}/its/DistrictIts/GetCctvStatusListByDistrict?districtCode=${encodeURIComponent(district)}`;
/** Per-camera frame. Returns JSON `{snippet:<base64 jpeg>}`, not an image body;
@@ -113,13 +113,34 @@ 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 =
'https://s3-eu-west-1.amazonaws.com/jamcams.tfl.gov.uk/';
export const DEFAULT_TFL_MAX_SOURCES = 250;
export const LONDON_CENTER = { lat: 51.5074, lon: -0.1278 };
/** Ontario 511: keyless CARS/511 camera catalog; frame URLs are still images. */
/** Ontario 511: CARS/511 camera catalog, which needs a developer key (ONTARIO_511_API_KEY); frame URLs are still images. */
export const ONTARIO_511_CAMERAS_URL =
'https://511on.ca/api/v2/get/cameras?format=json&lang=en';
export const ONTARIO_511_IMAGE_ORIGIN = 'https://511on.ca/map/Cctv/';
@@ -188,6 +209,8 @@ export const TARKTEE_IMAGES_URL =
'https://tarktee.transpordiamet.ee/api/v1/datex/roadCameraImages';
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;
export const TARKTEE_ANCHORS = [
{ lat: 59.437, lon: 24.753 }, // Tallinn
{ lat: 58.378, lon: 26.729 }, // Tartu
@@ -286,6 +309,36 @@ export const CATALONIA_ANCHORS = [
{ lat: 41.5636, lon: 2.0111 }, // Terrassa
];
/** Hong Kong Transport Department traffic snapshot cameras: keyless English
* locations XML on DATA.GOV.HK; frames are stills on the TD CCTV host. */
export const DEFAULT_HK_CAMERAS_URL =
'https://static.data.gov.hk/td/traffic-snapshot-images/code/Traffic_Camera_Locations_En.xml';
/** The only origin HK camera frames may come from. Verified 2026-10-11: still
* URLs answer 200 on this host with no redirect hop. */
export const HK_IMAGE_ORIGIN = 'https://tdcctv.data.one.gov.hk/';
/** Above the full TD English catalog (~1,013 cameras) so a default install
* keeps nearly all of them. Catalog-wide round-robin still applies when the
* sum of packs exceeds DEFAULT_CCTV_MAX_SOURCES (Norway-style). */
export const DEFAULT_HK_MAX_SOURCES = 1100;
/** Prioritization anchors across Hong Kong Island, Kowloon and the New
* Territories so a lowered cap keeps territory-wide coverage rather than
* only the harbour corridor. */
export const HK_ANCHORS = [
{ lat: 22.2819, lon: 114.158 }, // Central
{ lat: 22.2976, lon: 114.1722 }, // Tsim Sha Tsui
{ lat: 22.3827, lon: 114.19 }, // Sha Tin
{ lat: 22.3701, lon: 114.114 }, // Tsuen Wan
{ lat: 22.4503, lon: 114.1688 }, // Tai Po
{ lat: 22.493, lon: 114.139 }, // Fanling
{ lat: 22.391, lon: 113.977 }, // Tuen Mun
{ lat: 22.4445, lon: 114.022 }, // Yuen Long
{ lat: 22.307, lon: 114.26 }, // Tseung Kwan O
];
/** Hard ceiling on the locations XML body. The live English list is ~400 KB
* for ~1,000 cameras; this only exists so an unbounded upstream cannot be
* buffered without limit. */
export const HK_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
/** Statens vegvesen (Norway) road cameras: the keyless OGC API Features view
* of the official DATEX 3.1 CCTV site table, one GeoJSON request for the whole
* country (~900 cameras). The raw DATEX node needs an account; this view does
@@ -317,6 +370,49 @@ export const NORWAY_ANCHORS = [
{ lat: 69.6492, lon: 18.9553 }, // Tromsø
];
/** Kartverket's keyless point-height API (CC BY 4.0): terrain height above
* sea level (NN2000), used as the Vegvesen cameras' ground prior because the
* camera feed's points are 2D. */
export const KARTVERKET_HEIGHT_URL =
'https://ws.geonorge.no/hoydedata/v1/punkt';
/** Points per request. The API takes up to 50, but above ~20 it answers the
* extra points from contour lines (`hoydekurver`, metres to ~20 m off) or not
* at all; at 20 nearly every camera resolves from the 1 m terrain model. */
export const KARTVERKET_MAX_POINTS = 20;
/** Requests in flight at once. The whole country is ~36 requests of
* 0.6-9.5 s each, so a cold lookup takes ~6-13 s. */
export const KARTVERKET_CONCURRENCY = 8;
/** Deadline for one complete lookup, shared by every request. A lookup cut
* short by it is resumed on the next refresh, without a backoff. */
export const KARTVERKET_LOOKUP_TIMEOUT_MS = 30 * 1000;
/** One request that takes longer than this counts as a failing service. */
export const KARTVERKET_REQUEST_TIMEOUT_MS = 10 * 1000;
/** The Vegvesen lane's budget, measured from the start of the loader (catalog
* fetch included), for waiting on heights. Under the 15 s per-provider fetch
* timeout so this pack is never the catalog's slowest lane. A lookup still
* running after it keeps filling the cache for the next refresh. */
export const KARTVERKET_CATALOG_WAIT_MS = 12 * 1000;
/** The lane only waits when more than this share of its cameras has no
* answer yet (a cold cache). A few new cameras or expired misses are looked
* up in the background instead of holding the refresh. */
export const KARTVERKET_WAIT_MISSING_SHARE = 0.1;
/** A point Kartverket has no height for (some ferry quays and bridges) is not
* asked about again for this long. */
export const KARTVERKET_NULL_TTL_MS = 6 * 60 * 60 * 1000;
/** After a failing request (408, 429, 5xx, a network error, a request
* timeout or an unreadable body), no lookup starts for this long, or for the
* Retry-After the service sent, bounded to [1 min, 1 h]. Longer than the
* 15-minute catalog refresh (CCTV_SOURCE_CACHE_MS), so a failing service
* costs one attempt per backoff, not one per refresh. */
export const KARTVERKET_FAILURE_BACKOFF_MS = 30 * 60 * 1000;
export const KARTVERKET_MAX_BACKOFF_MS = 60 * 60 * 1000;
export const KARTVERKET_MAX_RESPONSE_BYTES = 256 * 1024;
/** Plausible Norwegian terrain: below sea level only at the shore, and no road
* camera sits above ~1,500 m. Anything outside is treated as no height. */
export const KARTVERKET_HEIGHT_RANGE_M = Object.freeze({ min: -20, max: 2500 });
/** Same identity the other CCTV upstream requests send. */
export const KARTVERKET_USER_AGENT = 'gods-eye-view-cctv-proxy/1.0';
/** NYC DOT Traffic Management Center cameras: one keyless JSON list for the
* five boroughs (~980 cameras). Frames are stills synthesized from each
* camera UUID on the city's webcam host — never taken from the payload. */
@@ -439,6 +535,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. */
+1 -1
View File
@@ -3,7 +3,7 @@ import path from 'node:path';
import { poseHash } from '../../../src/data/cctvFootprint.js';
/** Sidecar written by scripts/precompute-cctv-heights.mjs. */
export const DEFAULT_GROUND_HEIGHTS_FILE =
const DEFAULT_GROUND_HEIGHTS_FILE =
'src/data/local_data/cctv_ground_heights/cctv_ground_heights.json';
/** @type {{path:string, mtimeMs:number, cameras:object}|null} */
+131 -10
View File
@@ -1,13 +1,21 @@
import { Readable } from 'node:stream';
import { hashSeed, escapeXml } from './normalize.js';
import { hashSeed, escapeXml, pinnedCameraUrl } from './normalize.js';
import {
CALGARY_IMAGE_ORIGIN,
CCTV_FRAME_FETCH_TIMEOUT_MS,
CCTV_FRAME_MAX_BODY_BYTES,
CCTV_MEDIA_FETCH_TIMEOUT_MS,
CCTV_MEDIA_IDLE_TIMEOUT_MS,
CCTV_MEDIA_MAX_BODY_BYTES,
HK_IMAGE_ORIGIN,
KING_COUNTY_IMAGE_ORIGIN,
NSW_IMAGE_ORIGIN,
NSW_IMAGE_USER_AGENT,
TALLINN_IMAGE_ORIGIN,
TARKTEE_IMAGE_ORIGIN,
TFL_IMAGE_ORIGIN,
WARENDORF_IMAGE_ORIGINS,
WSDOT_IMAGE_ORIGIN,
} from './constants.js';
/**
* Generate a synthetic SVG billboard image for a CCTV camera placeholder.
@@ -84,7 +92,7 @@ export function buildSyntheticCctvSvg({ cameraId, label, city, status }) {
* @param {ReadableStream|NodeJS.ReadableStream|null} body
* @returns {import('stream').Readable|null}
*/
export function toReadable(body) {
function toReadable(body) {
if (!body) return null;
if (typeof body.pipe === 'function') return body;
if (typeof body.getReader === 'function') {
@@ -93,6 +101,73 @@ export function toReadable(body) {
return null;
}
/**
* Split a Content-Type value into its lower-case essence and parameters.
*
* @param {string|null|undefined} value
* @returns {{essence:string,params:Map<string,string>}|null} Null when the
* value is missing or is not a `type/subtype` media type.
*/
function parseMediaType(value) {
const [head, ...rest] = String(value ?? '').split(';');
const essence = head.trim().toLowerCase();
if (!/^[a-z0-9!#$&^_.+-]+\/[a-z0-9!#$&^_.+-]+$/.test(essence)) return null;
const params = new Map();
for (const part of rest) {
const eq = part.indexOf('=');
if (eq <= 0) continue;
const name = part.slice(0, eq).trim().toLowerCase();
let paramValue = part.slice(eq + 1).trim();
if (/^".*"$/.test(paramValue)) paramValue = paramValue.slice(1, -1);
if (name && !params.has(name)) params.set(name, paramValue);
}
return { essence, params };
}
/** Still-image types a camera frame may be served as. */
const CCTV_RASTER_IMAGE_TYPES = new Set([
'image/jpeg',
'image/png',
'image/gif',
'image/webp',
'image/avif',
]);
/**
* The canonical type of a raster camera frame, or null for any other type.
*
* @param {string|null|undefined} contentType
* @returns {string|null}
*/
export function cctvRasterImageType(contentType) {
const essence = parseMediaType(contentType)?.essence;
return CCTV_RASTER_IMAGE_TYPES.has(essence) ? essence : null;
}
/**
* The type to relay a live feed as, or null when the upstream type does not
* match the feed kind. MJPEG keeps its part boundary, without which the body
* cannot be split into frames.
*
* @param {string} feedType - Normalized feed type (mp4, webm or mjpeg).
* @param {string|null|undefined} contentType
* @returns {string|null}
*/
export function cctvLiveMediaType(feedType, contentType) {
const parsed = parseMediaType(contentType);
if (!parsed) return null;
if (feedType === 'mp4' || feedType === 'webm')
return parsed.essence === `video/${feedType}` ? parsed.essence : null;
if (feedType !== 'mjpeg' || parsed.essence !== 'multipart/x-mixed-replace')
return null;
const boundary = parsed.params.get('boundary') || '';
if (!/^[0-9A-Za-z'()+_,./:=? -]{0,69}[0-9A-Za-z'()+_,./:=?-]$/.test(boundary))
return null;
return /^[0-9A-Za-z'+_.-]+$/.test(boundary)
? `multipart/x-mixed-replace; boundary=${boundary}`
: `multipart/x-mixed-replace; boundary="${boundary}"`;
}
/**
* Pipe an upstream fetch Response (image or video) to the client HTTP response.
*
@@ -104,6 +179,8 @@ export function toReadable(body) {
* @param {Response} upstream - fetch() Response object.
* @param {object} [opts]
* @param {string} [opts.sourceHeader='upstream'] - Value for X-CCTV-Source header.
* @param {string} [opts.contentType] - Validated type to send instead of the
* upstream's own.
* @param {number} [opts.idleTimeoutMs=CCTV_MEDIA_IDLE_TIMEOUT_MS] - Silence
* allowed between upstream chunks before the stream is released. Injectable
* only to keep the deadline unit-testable.
@@ -114,10 +191,10 @@ export async function proxyMediaResponse(
{
sourceHeader = 'upstream',
idleTimeoutMs = CCTV_MEDIA_IDLE_TIMEOUT_MS,
contentType = upstream.headers.get('content-type') ||
'application/octet-stream',
} = {},
) {
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');
@@ -340,7 +417,14 @@ export async function fetchCctvMediaUpstream(
if (downstream?.aborted) controller.abort();
else downstream?.addEventListener?.('abort', onDownstreamAbort);
try {
return await fetchImpl(url, { headers, signal: controller.signal });
// Redirects stay on the registered origin, as for frames.
const upstream = await fetchWithinHost(
url,
{ headers, signal: controller.signal },
fetchImpl,
);
if (!upstream) throw new Error('Media redirect refused');
return upstream;
} finally {
clearTimeout(timeoutId);
downstream?.removeEventListener?.('abort', onDownstreamAbort);
@@ -428,9 +512,40 @@ export async function fetchTxdotSnapshot(
/** Redirect hops the frame path will follow, and only within the same host. */
const MAX_SAME_HOST_REDIRECTS = 2;
/**
* Directories camera catalogs are pinned to, by host. A redirect on one of
* these hosts must stay inside the pinned directory too, so a same-origin
* redirect cannot leave it (another bucket on a shared storage host).
*/
const CCTV_PINNED_PREFIXES_BY_HOST = new Map();
for (const prefix of [
TFL_IMAGE_ORIGIN,
TALLINN_IMAGE_ORIGIN,
TARKTEE_IMAGE_ORIGIN,
...WARENDORF_IMAGE_ORIGINS,
NSW_IMAGE_ORIGIN,
CALGARY_IMAGE_ORIGIN,
HK_IMAGE_ORIGIN,
KING_COUNTY_IMAGE_ORIGIN,
WSDOT_IMAGE_ORIGIN,
]) {
const host = new URL(prefix).hostname;
CCTV_PINNED_PREFIXES_BY_HOST.set(host, [
...(CCTV_PINNED_PREFIXES_BY_HOST.get(host) || []),
prefix,
]);
}
/** Whether a redirect target stays inside its host's pinned directory. */
function withinPinnedPrefix(url) {
const prefixes = CCTV_PINNED_PREFIXES_BY_HOST.get(url.hostname);
return !prefixes || prefixes.some((p) => pinnedCameraUrl(url.href, p));
}
/**
* Fetch a registered frame URL following redirects ONLY within the original
* origin (scheme, host and port; at most MAX_SAME_HOST_REDIRECTS hops).
* origin (scheme, host and port; at most MAX_SAME_HOST_REDIRECTS hops) and,
* on a pinned camera host, within its pinned directory.
* 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.
@@ -471,7 +586,7 @@ export async function fetchWithinHost(url, init, fetchImpl = fetch) {
} catch {
return null;
}
if (next.origin !== origin) return null;
if (next.origin !== origin || !withinPinnedPrefix(next)) return null;
current = next;
}
return null;
@@ -543,9 +658,15 @@ export async function fetchCctvImageFromUpstream(
fetchImpl,
);
if (!upstream) return null;
const contentType = upstream.headers.get('content-type') || '';
if (!upstream.ok || !contentType.startsWith('image/')) {
controller.abort();
const contentType = cctvRasterImageType(
upstream.headers.get('content-type'),
);
if (!upstream.ok || !contentType) {
try {
await upstream.body?.cancel();
} catch {
/* no-op */
}
return null;
}
const body = await readCappedResponseBytes(upstream, maxBytes);
+60 -3
View File
@@ -64,7 +64,7 @@ export function toFiniteNumber(value, fallback = NaN) {
* @param {string} text
* @returns {string}
*/
export function normalizeKey(text) {
function normalizeKey(text) {
return String(text || '')
.trim()
.toLowerCase()
@@ -80,7 +80,7 @@ export function normalizeKey(text) {
* @param {string} value
* @returns {{lat:number, lon:number}}
*/
export function parsePointString(value) {
function parsePointString(value) {
const match = String(value || '').match(
/POINT\s*\(\s*(-?\d+(?:\.\d+)?)\s+(-?\d+(?:\.\d+)?)\s*\)/i,
);
@@ -100,7 +100,7 @@ export function parsePointString(value) {
* @param {string|object|null} value
* @returns {{lat:number, lon:number}}
*/
export function coerceLatLon(value) {
function coerceLatLon(value) {
if (!value) return { lat: NaN, lon: NaN };
if (typeof value === 'string') {
@@ -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 (
@@ -388,6 +399,17 @@ export function isLikelyNycCoordinate(lat, lon) {
);
}
/** Hong Kong SAR extent, with slack for the outlying islands. */
export function isLikelyHongKongCoordinate(lat, lon) {
return (
isPlausibleLatLon(lat, lon) &&
lat >= 22.15 &&
lat <= 22.6 &&
lon >= 113.8 &&
lon <= 114.45
);
}
/** Mainland Norway, with slack for the coast and the Finnmark border. */
export function isLikelyNorwayCoordinate(lat, lon) {
return (
@@ -637,3 +659,38 @@ export function isLikelyTorontoCoordinate(lat, lon) {
lon <= -79.1
);
}
/**
* A camera URL under a pinned directory, by parsed URL rather than string
* prefix: the same scheme and exact host as `prefix`, no credentials, the
* default port, and a normalized path inside the prefix's directory. Dot
* segments (including encoded ones) and backslashes are resolved before the
* path is compared, and an encoded slash or backslash is refused.
*
* @param {string} raw - Camera URL from a catalog or redirect.
* @param {string} prefix - Pinned origin and directory, ending in `/`.
* @returns {string|null} The normalized URL, or null when it is not pinned.
*/
export function pinnedCameraUrl(raw, prefix) {
let url;
let pin;
try {
url = new URL(String(raw ?? '').trim());
pin = new URL(prefix);
} catch {
return null;
}
if (
url.protocol !== pin.protocol ||
url.hostname !== pin.hostname ||
url.username ||
url.password ||
url.port ||
!url.pathname.startsWith(
pin.pathname.endsWith('/') ? pin.pathname : `${pin.pathname}/`,
) ||
/%2f|%5c/i.test(url.pathname)
)
return null;
return url.href;
}
+129
View File
@@ -0,0 +1,129 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import {
_resetOntarioKeyWarningForTest,
loadOntarioSourcesFromOpenData,
} from './sources.js';
const ROW = {
Id: '42',
Latitude: 43.6532,
Longitude: -79.3832,
Location: 'Hwy 401 near Yonge St',
Direction: 'Eastbound',
Views: [
{
Status: 'Enabled',
Url: 'https://511on.ca/map/Cctv/4201',
Description: 'Eastbound',
},
],
};
function captureConsole(t) {
const lines = [];
const capture = (...args) => lines.push(args.join(' '));
t.mock.method(console, 'warn', capture);
t.mock.method(console, 'log', capture);
return lines;
}
test.beforeEach(() => _resetOntarioKeyWarningForTest());
test('without ONTARIO_511_API_KEY the Ontario pack is skipped, logged once', async (t) => {
const lines = captureConsole(t);
let fetches = 0;
t.mock.method(globalThis, 'fetch', async () => {
fetches += 1;
return Response.json([ROW]);
});
assert.deepEqual(await loadOntarioSourcesFromOpenData({}, {}), []);
assert.deepEqual(
await loadOntarioSourcesFromOpenData({}, { ONTARIO_511_API_KEY: ' ' }),
[],
);
assert.equal(fetches, 0);
assert.equal(lines.length, 1);
assert.match(lines[0], /ONTARIO_511_API_KEY/);
});
test('with a key the camera list is requested with it and parsed', async (t) => {
const lines = captureConsole(t);
const requested = [];
t.mock.method(globalThis, 'fetch', async (url) => {
requested.push(String(url));
return Response.json([ROW]);
});
const cameras = await loadOntarioSourcesFromOpenData(
{ sourceRoot: '/ignored' },
{ ONTARIO_511_API_KEY: ' test-key ' },
);
assert.equal(requested.length, 1);
const url = new URL(requested[0]);
assert.equal(url.host, '511on.ca');
assert.equal(url.searchParams.get('key'), 'test-key');
assert.equal(url.searchParams.get('format'), 'json');
assert.equal(cameras.length, 1);
assert.equal(cameras[0].id, 'on-42');
assert.equal(cameras[0].url, 'https://511on.ca/map/Cctv/4201');
// The records go to the browser through /api/cctv; the key must stay on
// the server-side catalog request.
assert.doesNotMatch(JSON.stringify(cameras), /test-key/);
assert.doesNotMatch(lines.join('\n'), /test-key/);
});
test('the source cap is read from the same environment as the key', async (t) => {
captureConsole(t);
const rows = Array.from({ length: 12 }, (_, i) => ({
...ROW,
Id: String(100 + i),
Latitude: ROW.Latitude + i * 0.01,
Views: [{ ...ROW.Views[0], Url: `https://511on.ca/map/Cctv/${100 + i}` }],
}));
t.mock.method(globalThis, 'fetch', async () => Response.json(rows));
const cameras = await loadOntarioSourcesFromOpenData(
{},
{ ONTARIO_511_API_KEY: 'test-key', CCTV_ONTARIO_MAX_SOURCES: '8' },
);
assert.equal(cameras.length, 8);
});
for (const status of [400, 401, 403]) {
test(`a ${status} points at the key by name, never by value`, async (t) => {
const lines = captureConsole(t);
t.mock.method(globalThis, 'fetch', async () =>
Response.json({ Message: 'Invalid Key' }, { status }),
);
const cameras = await loadOntarioSourcesFromOpenData(
{},
{ ONTARIO_511_API_KEY: 'secret-key-value' },
);
assert.deepEqual(cameras, []);
assert.match(
lines.join('\n'),
new RegExp(`failed: ${status} \\(check ONTARIO_511_API_KEY\\)`),
);
assert.doesNotMatch(lines.join('\n'), /secret-key-value/);
});
}
for (const status of [429, 500, 503]) {
test(`a ${status} is reported without blaming the key`, async (t) => {
const lines = captureConsole(t);
t.mock.method(globalThis, 'fetch', async () =>
Response.json({}, { status }),
);
const cameras = await loadOntarioSourcesFromOpenData(
{},
{ ONTARIO_511_API_KEY: 'secret-key-value' },
);
assert.deepEqual(cameras, []);
assert.equal(lines.length, 1);
assert.equal(
lines[0],
`[CCTV] Ontario 511 camera download failed: ${status}`,
);
assert.doesNotMatch(lines[0], /secret-key-value/);
});
}
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -70,7 +70,7 @@ export async function fetchHlsBytes(
}
}
export function sameOriginHlsUrl(value, base) {
function sameOriginHlsUrl(value, base) {
const url = new URL(value, base);
if (
!['https:', 'http:'].includes(url.protocol) ||
+303
View File
@@ -0,0 +1,303 @@
import { CCTV_SOURCE_FETCH_TIMEOUT_MS } from './constants.js';
import {
cameraDisplayCode,
fallbackHeadingFromId,
prioritizeSources,
} from './normalize.js';
import { readResponseTextCapped } from '../common/http.js';
/** Freeway Bureau (MOTC) CCTV static table in the national MOTC 2.0 XML
* standard, published keyless and refreshed daily (~1,800 cameras). */
export const TAIWAN_FREEWAY_CCTV_URL =
'https://tisvcloud.freeway.gov.tw/history/motc20/CCTV.xml';
/** The whole network is ~1.2 MB. */
export const TAIWAN_MAX_CATALOG_BYTES = 6 * 1024 * 1024;
const DEFAULT_TAIWAN_MAX_SOURCES = 600;
/** After a failed download, wait this long before asking again. The catalog
* host does not answer from some networks outside Taiwan, and every attempt
* there holds the shared catalog refresh until the connect timeout. */
export const TAIWAN_RETRY_AFTER_FAILURE_MS = 60 * 60 * 1000;
/** The table is republished daily; a good copy stands in for a failed
* download for up to this long. */
const TAIWAN_LAST_GOOD_MAX_AGE_MS = 24 * 60 * 60 * 1000;
// The table publishes each camera's MJPEG stream. The same camera servers
// answer a single JPEG for the same camera ID on a sibling path, which is what
// the frame proxy can relay; only these exact pairs are ever registered.
const STREAM_TO_FRAME = new Map([
['cctvn.freeway.gov.tw', ['/abs2mjpg/bmjpg', '/abs2mjpg/jpg']],
['cctvn5.freeway.gov.tw', ['/abs2mjpg/bmjpg', '/abs2mjpg/jpg']],
['cctvc.freeway.gov.tw', ['/abs2mjpg/bmjpg', '/abs2mjpg/jpg']],
[
'cctvs.freeway.gov.tw',
['/live-view/mjpg/video.cgi', '/live-view/jpg/image.cgi'],
],
]);
const DIRECTIONS = { N: 'N', S: 'S', E: 'E', W: 'W' };
/** Prioritization anchors, north to south, so a cap keeps the whole island
* rather than only greater Taipei. */
const ANCHORS = [
{ lat: 25.033, lon: 121.5654 }, // Taipei
{ lat: 24.9936, lon: 121.301 }, // Taoyuan
{ lat: 24.8138, lon: 120.9675 }, // Hsinchu
{ lat: 24.7021, lon: 121.7378 }, // Yilan
{ lat: 24.1477, lon: 120.6736 }, // Taichung
{ lat: 22.9999, lon: 120.227 }, // Tainan
{ lat: 22.6273, lon: 120.3014 }, // Kaohsiung
];
/** Taiwan main island with slack for the coast; the freeway network has no
* offshore-island cameras. */
function isLikelyTaiwanCoordinate(lat, lon) {
return (
Number.isFinite(lat) &&
Number.isFinite(lon) &&
lat >= 21.8 &&
lat <= 25.4 &&
lon >= 119.9 &&
lon <= 122.1
);
}
const XML_ENTITIES = {
amp: '&',
lt: '<',
gt: '>',
quot: '"',
apos: "'",
};
function codePoint(value, radix) {
const code = parseInt(value, radix);
return Number.isInteger(code) && code > 0 && code <= 0x10ffff
? String.fromCodePoint(code)
: '';
}
function decodeXml(text) {
return String(text ?? '')
.replace(
/&(?:(amp|lt|gt|quot|apos)|#(\d{1,7})|#x([0-9a-fA-F]{1,6}));/g,
(_, name, dec, hex) =>
name ? XML_ENTITIES[name] : codePoint(dec ?? hex, dec ? 10 : 16),
)
.trim();
}
function tag(block, name) {
const match = new RegExp(`<${name}>([^<]*)</${name}>`).exec(block);
return match ? decodeXml(match[1]) : '';
}
/**
* Parse the MOTC 2.0 `CCTVList` document into flat rows. Regex-based like the
* Catalonia and Tarktee parsers: the schema is flat, and only the fields used
* below are read.
*
* @param {string} xml
* @returns {Array<object>}
*/
export function parseTaiwanCctvXml(xml) {
const rows = [];
const blockRe = /<CCTV>([\s\S]*?)<\/CCTV>/g;
let match;
while ((match = blockRe.exec(String(xml || ''))) !== null) {
const block = match[1];
rows.push({
cctvId: tag(block, 'CCTVID'),
streamUrl: tag(block, 'VideoStreamURL'),
lat: Number(tag(block, 'PositionLat')),
lon: Number(tag(block, 'PositionLon')),
roadName: tag(block, 'RoadName'),
direction: tag(block, 'RoadDirection'),
sectionStart: tag(block, 'Start'),
sectionEnd: tag(block, 'End'),
mile: tag(block, 'LocationMile'),
});
}
return rows;
}
/**
* Published MJPEG stream URL -> the same camera's still, plus the host and
* camera number that identify it, or null.
*
* @param {string} streamUrl
* @returns {?{frameUrl: string, host: string, camera: string}}
*/
export function taiwanFrame(streamUrl) {
let url;
try {
url = new URL(String(streamUrl ?? ''));
} catch {
return null;
}
if (url.protocol !== 'https:' || url.username || url.password || url.port)
return null;
const paths = STREAM_TO_FRAME.get(url.hostname);
if (!paths || url.pathname !== paths[0]) return null;
const camera = url.searchParams.get('camera') || '';
if (!/^[A-Za-z0-9-]{1,40}$/.test(camera)) return null;
return {
// Reconstruct so extra query parameters cannot alter the request.
frameUrl: `https://${url.hostname}${paths[1]}?camera=${camera}`,
host: url.hostname.split('.')[0],
camera,
};
}
/**
* One parsed Freeway Bureau row -> one catalog source, or null.
*
* @param {object} row - Output of parseTaiwanCctvXml.
* @returns {?object}
*/
export function taiwanCameraToSource(row) {
if (!row || typeof row !== 'object') return null;
const { lat, lon } = row;
if (!isLikelyTaiwanCoordinate(lat, lon)) return null;
const frame = taiwanFrame(row.streamUrl);
if (!frame) return null;
const imageUrl = frame.frameUrl;
// CCTVID carries Chinese place names and a few IDs repeat across distinct
// cameras, so the stable key is the camera server plus its camera number.
const rawId = String(row.cctvId ?? '')
.trim()
.slice(0, 80);
const cameraId = `tw-freeway-${frame.host}-${frame.camera.toLowerCase()}`;
const road = String(row.roadName ?? '').trim();
const direction = DIRECTIONS[String(row.direction ?? '').trim()] || '';
const mile = String(row.mile ?? '').trim();
const section = [row.sectionStart, row.sectionEnd]
.map((part) => String(part ?? '').trim())
.filter(Boolean)
.join('→');
// A bare carriageway letter is not a name; fall back to the published ID.
const name =
(road || mile || section
? [road, direction, mile, section].filter(Boolean).join(' ')
: '') ||
rawId ||
`Freeway camera ${frame.camera}`;
return {
id: cameraId,
name: name.slice(0, 160),
// One island-wide category in the camera picker, like Finland and Norway.
city: 'Taiwan / 臺灣',
cityId: 'taiwan',
provider: 'Freeway Bureau, MOTC',
lat,
lon,
// The table gives the carriageway direction, not the camera's bearing,
// so the pose uses the shared low-confidence prior.
headingDeg: fallbackHeadingFromId(cameraId),
headingConfidence: 'low',
pitchDeg: -18,
fovDeg: 44,
rangeM: 145,
mountHeightM: 8,
feedType: 'image',
url: imageUrl,
snapshotUrl: imageUrl,
sourceKind: 'tw-freeway-motc',
license:
'Freeway Bureau, MOTC open data, Open Government Data License, Taiwan (Version 1.0)',
code: cameraDisplayCode(
`${road} ${mile}`.trim().toUpperCase() || frame.camera,
),
};
}
let lastGood = null;
let lastGoodAt = 0;
let retryAfter = 0;
/** Forget the last good table and any failure backoff (tests only). */
export function resetTaiwanSourcesCache() {
lastGood = null;
lastGoodAt = 0;
retryAfter = 0;
}
function fallbackSources() {
return lastGood && Date.now() - lastGoodAt <= TAIWAN_LAST_GOOD_MAX_AGE_MS
? lastGood
: [];
}
function taiwanSourceCap() {
const rawCap = Number(
process.env.CCTV_TAIWAN_MAX_SOURCES || DEFAULT_TAIWAN_MAX_SOURCES,
);
return Number.isFinite(rawCap)
? Math.max(1, Math.min(2000, Math.floor(rawCap)))
: DEFAULT_TAIWAN_MAX_SOURCES;
}
/**
* Fetch Taiwan Freeway Bureau cameras from the keyless MOTC 2.0 XML table.
* One bounded request per shared catalog refresh. A failed download serves the
* last good table (up to a day old) and is not retried for an hour, so a host
* that cannot reach the Freeway Bureau does not wait on it every refresh.
*
* @returns {Promise<Array<object>>}
*/
export async function loadTaiwanSources() {
if (Date.now() < retryAfter) return fallbackSources();
try {
const signal = AbortSignal.timeout(CCTV_SOURCE_FETCH_TIMEOUT_MS);
const response = await fetch(
process.env.CCTV_TAIWAN_URL || TAIWAN_FREEWAY_CCTV_URL,
{
headers: { Accept: 'application/xml,text/xml' },
redirect: 'manual',
signal,
},
);
if (!response.ok) {
await response.body?.cancel().catch(() => {});
throw new Error(
response.status >= 300 && response.status < 400
? 'catalog redirected; redirects are not followed'
: `HTTP ${response.status}`,
);
}
const xml = await readResponseTextCapped(
response,
TAIWAN_MAX_CATALOG_BYTES,
signal,
);
const cameras = new Map();
for (const row of parseTaiwanCctvXml(xml)) {
const camera = taiwanCameraToSource(row);
if (camera && !cameras.has(camera.id)) cameras.set(camera.id, camera);
}
if (!cameras.size) throw new Error('no usable cameras in the table');
const sources = prioritizeSources(
[...cameras.values()],
taiwanSourceCap(),
ANCHORS,
);
console.log(
`[CCTV] Loaded Taiwan camera sources: ${cameras.size} (using ${sources.length})`,
);
lastGood = sources;
lastGoodAt = Date.now();
retryAfter = 0;
return sources;
} catch (error) {
retryAfter = Date.now() + TAIWAN_RETRY_AFTER_FAILURE_MS;
const fallback = fallbackSources();
console.warn(
`[CCTV] Taiwan camera download failed (${error?.message || error}); ` +
(fallback.length
? `serving the last good ${fallback.length} cameras`
: 'no Taiwan cameras') +
', next attempt in 1 h',
);
return fallback;
}
}
+53 -8
View File
@@ -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();
});
}
+137
View File
@@ -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);
});
+45
View File
@@ -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;
}
+15 -2
View File
@@ -236,13 +236,26 @@ export function gbfsProxy({ ratePerMinute = GBFS_RATE_PER_MIN } = {}) {
}
const upstream = await fetchGbfsUpstream(upstreamUrl.toString());
// Only JSON is relayed, re-serialized under a fixed type: neither the
// upstream's text nor its declared type reaches the browser.
let payload;
try {
payload = JSON.parse(upstream.body);
} catch {
res.writeHead(502, {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
});
res.end(JSON.stringify({ error: 'GBFS upstream returned non-JSON' }));
return;
}
res.writeHead(upstream.status, {
'Content-Type': upstream.contentType,
'Content-Type': 'application/json; charset=utf-8',
'Cache-Control': gbfsCacheControl(upstreamUrl.pathname),
'X-GBFS-Upstream': upstreamUrl.hostname,
'X-GBFS-Cache': 'MISS',
});
res.end(upstream.body);
res.end(JSON.stringify(payload));
} catch (error) {
if (error?.name === 'AbortError') {
res.writeHead(504, {
+22
View File
@@ -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 };
+53
View File
@@ -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;
}
+208
View File
@@ -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);
}
};
}
+79
View File
@@ -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');
}
+9 -9
View File
@@ -27,13 +27,13 @@ import { normalizeDump1090Aircraft } from '../../src/sources/adsbRecords.js';
*/
export const LOCAL_RECEIVERS_ROUTE = '/api/local-receivers/aircraft';
export const LOCAL_RECEIVER_FEEDS_ENV = 'LOCAL_RECEIVER_FEEDS';
export const LOCAL_RECEIVER_BANDS = Object.freeze(['1090', '978']);
export const LOCAL_RECEIVER_TIMEOUT_MS = 2_000;
export const LOCAL_RECEIVER_MAX_BODY_BYTES = 2 * 1024 * 1024;
export const LOCAL_RECEIVER_CACHE_MS = 1_000;
const LOCAL_RECEIVER_FEEDS_ENV = 'LOCAL_RECEIVER_FEEDS';
const LOCAL_RECEIVER_BANDS = Object.freeze(['1090', '978']);
const LOCAL_RECEIVER_TIMEOUT_MS = 2_000;
const LOCAL_RECEIVER_MAX_BODY_BYTES = 2 * 1024 * 1024;
const LOCAL_RECEIVER_CACHE_MS = 1_000;
/** A feed whose own `now` is older than this is reported stale. */
export const LOCAL_RECEIVER_STALE_MS = 10_000;
const LOCAL_RECEIVER_STALE_MS = 10_000;
const MAX_FEEDS = 8;
const LOG_PREFIX = '[local-receivers]';
@@ -139,7 +139,7 @@ class FeedError extends Error {
* @param {string} address Resolved IP address.
* @returns {boolean}
*/
export function isLocalReceiverAddress(address) {
function isLocalReceiverAddress(address) {
const value = String(address || '').toLowerCase();
if (value === '::1') return true;
const mapped = /^::ffff:(\d{1,3}(?:\.\d{1,3}){3})$/.exec(value);
@@ -152,7 +152,7 @@ export function isLocalReceiverAddress(address) {
* @param {typeof dnsLookup} lookupImpl `dns.promises.lookup`-compatible.
* @returns {Promise<Array<{address:string, family:number}>>}
*/
export async function resolveLocalReceiverAddresses(hostname, lookupImpl) {
async function resolveLocalReceiverAddresses(hostname, lookupImpl) {
let resolved;
try {
resolved = await lookupImpl(hostname, { all: true, verbatim: true });
@@ -274,7 +274,7 @@ function fetchLocalReceiverFeed(url, options) {
* @param {object} options
* @returns {Promise<object>} Parsed JSON document.
*/
export async function fetchLocalReceiverDocument(
async function fetchLocalReceiverDocument(
url,
{
fetchImpl = fetchLocalReceiverFeed,
+4 -2
View File
@@ -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(),
+101 -14
View File
@@ -1,19 +1,106 @@
import { installMapillaryRoutes } from './mapillary/routes.js';
import { sameSiteGated } from './common/same-site.js';
import { makeRateLimiter, clientKey } from './common/rate-limit.js';
import {
fetchTile,
mapillaryToken,
parseTilePath,
retryAfterWait,
TileUpstreamError,
} from './mapillary/tiles.js';
/**
* Vite plugin: cached Mapillary coverage tiles (token added server-side) and a
* status route that says whether a token is configured.
* Tile requests per client IP per minute, cache hits included: over 60 new
* 9-tile views a minute, more than a person flying the camera reaches.
*/
function mapillaryProxy() {
return {
name: 'mapillary-proxy',
configureServer(server) {
installMapillaryRoutes(server.middlewares);
},
configurePreviewServer(server) {
installMapillaryRoutes(server.middlewares);
},
};
export const TILE_ROUTE_MAX_PER_MIN = 600;
function sendJson(res, status, payload) {
res.statusCode = status;
res.setHeader('Content-Type', 'application/json; charset=utf-8');
res.setHeader('Cache-Control', 'no-store');
res.end(JSON.stringify(payload));
}
export { mapillaryProxy };
/** GET /api/mapillary/status — whether a token is configured, never its value. */
function handleStatus(req, res) {
if (req.method !== 'GET')
return sendJson(res, 405, { error: 'Method not allowed' });
sendJson(res, 200, { configured: Boolean(mapillaryToken()) });
}
/** GET /api/mapillary/tiles/coverage/{z}/{x}/{y} — trimmed protobuf tile. */
async function handleTile(req, res, allow) {
if (req.method !== 'GET')
return sendJson(res, 405, { error: 'Method not allowed' });
if (!allow(clientKey(req))) {
res.setHeader('Retry-After', '5');
return sendJson(res, 429, {
error: 'Too many tile requests',
retryAfter: 5,
});
}
const address = parseTilePath((req.url || '').split('?')[0]);
if (!address)
return sendJson(res, 400, {
error: 'Tile path must be /coverage/{z}/{x}/{y} with z 11–14',
});
if (!mapillaryToken())
return sendJson(res, 503, { error: 'no_key', keyRequired: true });
// A client that leaves (a tile the camera moved past) cancels its fetch.
const left = new AbortController();
res.on?.('close', () => {
if (!res.writableEnded) left.abort();
});
try {
const { bytes, source } = await fetchTile(address, { signal: left.signal });
res.statusCode = bytes.length ? 200 : 204;
res.setHeader('Content-Type', 'application/x-protobuf');
res.setHeader('Cache-Control', 'public, max-age=3600');
res.setHeader('X-Gev-Cache', source);
res.end(bytes.length ? bytes : undefined);
} catch (error) {
if (left.signal.aborted) return;
const status = error instanceof TileUpstreamError ? error.status : 0;
// A rejected token is a key problem the panel can name, not a fault.
if (status === 401 || status === 403)
return sendJson(res, 403, {
error: 'Mapillary rejected the access token',
keyRejected: true,
});
if (status === 429) {
const retryAfter = retryAfterWait(error.retryAfter);
res.setHeader('Retry-After', String(retryAfter));
return sendJson(res, 429, {
error: 'Mapillary is rate-limiting tile requests',
retryAfter,
});
}
// An upstream 400 must not read as a malformed request to this proxy.
sendJson(res, 502, { error: error?.message || 'Tile fetch failed' });
}
}
/**
* Vite plugin: Mapillary coverage tiles with the token added server-side, and
* a status route. Both refuse cross-site requests so another page cannot
* spend the token.
*/
export function mapillaryProxy() {
const install = (middlewares) => {
const allow = makeRateLimiter({
windowMs: 60_000,
max: TILE_ROUTE_MAX_PER_MIN,
globalMax: TILE_ROUTE_MAX_PER_MIN * 4,
});
middlewares.use('/api/mapillary/status', sameSiteGated(handleStatus));
middlewares.use(
'/api/mapillary/tiles',
sameSiteGated((req, res) => handleTile(req, res, allow)),
);
};
return {
name: 'mapillary-proxy',
configureServer: (server) => install(server.middlewares),
configurePreviewServer: (server) => install(server.middlewares),
};
}
-72
View File
@@ -1,72 +0,0 @@
import path from 'node:path';
/** Mapillary vector tile root; the path template is /{layer}/2/{z}/{x}/{y}. */
export const MAPILLARY_TILE_HOST = 'https://tiles.mapillary.com/maps/vtp';
/** Public tile layer names this proxy exposes, mapped to Mapillary's ids. */
export const TILE_LAYERS = Object.freeze({
// Region-sized z6–10 tiles are refused; the app never asks for them. The
// z14 `image` layer is ~98% of a tile and unused (positions come from the
// graph API), so it is dropped in transit.
coverage: Object.freeze({
upstream: 'mly1_public',
zoomRanges: Object.freeze([Object.freeze([0, 5]), Object.freeze([11, 14])]),
dropLayers: Object.freeze(['image']),
}),
});
/** Disk cache root, alongside the other providers' caches. */
export const MAPILLARY_CACHE_DIR = path.join(
process.cwd(),
'.gev-cache',
'mapillary',
);
export const TILE_DISK_DIR = path.join(MAPILLARY_CACHE_DIR, 'tiles');
/**
* Longer than the 15-minute upstream Cache-Control: tiles change only when new
* imagery is processed, and a day avoids re-downloading a city per camera move.
*/
export const TILE_DISK_TTL_MS = 24 * 60 * 60 * 1000;
/**
* A tile write starts a background sweep at most this often: expired files go
* first, then the oldest until the cache fits in TILE_DISK_MAX_BYTES.
*/
export const TILE_DISK_SWEEP_INTERVAL_MS = 15 * 60 * 1000;
export const TILE_DISK_MAX_BYTES = 1024 * 1024 * 1024;
/** Per-file cost on top of its size, so empty tiles still count. */
export const TILE_DISK_FILE_OVERHEAD_BYTES = 4096;
/** A z14 image tile over a dense city is ~11 MB; anything past this is wrong. */
export const TILE_MAX_BYTES = 48 * 1024 * 1024;
export const TILE_MEMORY_BUDGET_BYTES = 96 * 1024 * 1024;
/** Per-entry cost on top of its bytes, so empty tiles still count. */
export const TILE_MEMORY_ENTRY_OVERHEAD_BYTES = 1024;
export const TILE_FETCH_TIMEOUT_MS = 60_000;
/**
* Upstream fetches at once; further misses queue. A view needs at most 25
* tiles, and the cap stops a burst of misses opening thousands of requests.
*/
export const TILE_UPSTREAM_CONCURRENCY = 6;
/**
* Tile requests per client IP per minute, cache hits included. 600 is 24 new
* 25-tile views a minute, more than a person flying the camera reaches, but it
* stops one page draining the shared token into a Mapillary rate limit.
*/
export const TILE_ROUTE_MAX_PER_MIN = 600;
/**
* How long a 401/403 answers tile misses before asking Mapillary again. A
* changed token is tried at once.
*/
export const TILE_KEY_REJECTED_HOLD_MS = 5 * 60 * 1000;
/** After a 429 without a usable Retry-After, hold tile misses this long. */
export const TILE_RATE_LIMIT_HOLD_MS = 60_000;
/** The longest Retry-After honoured, so a bad header cannot stall coverage. */
export const TILE_RATE_LIMIT_MAX_HOLD_MS = 10 * 60 * 1000;
/** The client token lives in the browser by design; the server adds it to tile URLs too. */
export function mapillaryToken() {
return String(process.env.MAPILLARY_CLIENT_TOKEN || '').trim();
}
-95
View File
@@ -1,95 +0,0 @@
import { sameSiteGated } from '../common/same-site.js';
import { makeRateLimiter, clientKey } from '../common/rate-limit.js';
import { fetchTile, TileRequestError, TileUpstreamError } from './tiles.js';
import { mapillaryToken, TILE_ROUTE_MAX_PER_MIN } from './constants.js';
function sendJson(res, status, payload) {
res.statusCode = status;
res.setHeader('Content-Type', 'application/json; charset=utf-8');
res.setHeader('Cache-Control', 'no-store');
res.end(JSON.stringify(payload));
}
/** GET /api/mapillary/status — whether a token is configured, never its value. */
function handleStatus(req, res) {
if (req.method !== 'GET')
return sendJson(res, 405, { error: 'Method not allowed' });
sendJson(res, 200, { configured: Boolean(mapillaryToken()) });
}
/** GET /api/mapillary/tiles/coverage/{z}/{x}/{y} — cached protobuf tile. */
async function handleTile(req, res, allow) {
if (req.method !== 'GET')
return sendJson(res, 405, { error: 'Method not allowed' });
if (!allow(clientKey(req))) {
res.setHeader('Retry-After', '5');
return sendJson(res, 429, {
error: 'Too many tile requests',
retryAfter: 5,
});
}
const match = /^\/(coverage)\/(\d{1,2})\/(\d{1,6})\/(\d{1,6})$/.exec(
(req.url || '').split('?')[0],
);
if (!match)
return sendJson(res, 400, {
error: 'Tile path must be /coverage/{z}/{x}/{y}',
});
if (!mapillaryToken())
return sendJson(res, 503, { error: 'no_key', keyRequired: true });
const abandoned = new AbortController();
req.on?.('aborted', () => abandoned.abort());
res.on?.('close', () => abandoned.abort());
try {
const { bytes, source } = await fetchTile(
{ layer: match[1], z: match[2], x: match[3], y: match[4] },
{ signal: abandoned.signal },
);
if (res.writableEnded) return;
res.statusCode = bytes.length ? 200 : 204;
res.setHeader('Content-Type', 'application/x-protobuf');
res.setHeader('Cache-Control', 'public, max-age=3600');
res.setHeader('X-Gev-Cache', source);
res.end(bytes.length ? bytes : undefined);
} catch (error) {
if (res.writableEnded || abandoned.signal.aborted) return;
if (error instanceof TileRequestError)
return sendJson(res, error.status, { error: error.message });
if (error instanceof TileUpstreamError) {
// A rejected token is a key problem the panel can name, not a fault.
if (error.keyRejected)
return sendJson(res, 403, {
error: 'Mapillary rejected the access token',
keyRejected: true,
});
if (error.status === 429) {
const retryAfter = error.retryAfterSec || 60;
res.setHeader('Retry-After', String(retryAfter));
return sendJson(res, 429, {
error: 'Mapillary is rate-limiting tile requests',
retryAfter,
});
}
// An upstream 400 must not read as a malformed request to this proxy.
return sendJson(res, 502, { error: error.message });
}
sendJson(res, 502, { error: error?.message || 'Tile fetch failed' });
}
}
/**
* Attach the Mapillary routes. Both refuse cross-site requests so another page
* cannot spend the token or provoke a 429 that holds every tile miss.
*/
export function installMapillaryRoutes(middlewares) {
const allow = makeRateLimiter({
windowMs: 60_000,
max: TILE_ROUTE_MAX_PER_MIN,
globalMax: TILE_ROUTE_MAX_PER_MIN * 4,
});
middlewares.use('/api/mapillary/status', sameSiteGated(handleStatus));
middlewares.use(
'/api/mapillary/tiles',
sameSiteGated((req, res) => handleTile(req, res, allow)),
);
}
+216 -469
View File
@@ -1,518 +1,265 @@
import path from 'node:path';
import { promises as fsp } from 'node:fs';
import { readResponseBytesCapped } from '../common/http.js';
import { stripTileLayers } from './trim.js';
import {
MAPILLARY_TILE_HOST,
TILE_LAYERS,
TILE_DISK_DIR,
TILE_DISK_TTL_MS,
TILE_DISK_SWEEP_INTERVAL_MS,
TILE_DISK_MAX_BYTES,
TILE_DISK_FILE_OVERHEAD_BYTES,
TILE_MAX_BYTES,
TILE_MEMORY_BUDGET_BYTES,
TILE_MEMORY_ENTRY_OVERHEAD_BYTES,
TILE_FETCH_TIMEOUT_MS,
TILE_UPSTREAM_CONCURRENCY,
TILE_KEY_REJECTED_HOLD_MS,
TILE_RATE_LIMIT_HOLD_MS,
TILE_RATE_LIMIT_MAX_HOLD_MS,
mapillaryToken,
} from './constants.js';
/** Log without the stack; the token is never in scope here. */
function warn(what, error) {
console.warn(`[Mapillary Proxy] ${what}:`, error?.message || error);
/** Mapillary vector tiles, /{layer}/2/{z}/{x}/{y}: the only host the token goes to. */
const TILE_HOST = 'https://tiles.mapillary.com/maps/vtp';
/** Only the street zooms the app draws are proxied. */
const MIN_ZOOM = 11;
const MAX_ZOOM = 14;
/**
* The z14 `image` layer is ~98% of a tile (12 MB over a dense city) and unused
* (positions come from the Graph API), so it is dropped in transit.
*/
const DROP_LAYERS = ['image'];
/** A z14 tile with its image layer is ~12 MB; anything past this is wrong. */
export const TILE_MAX_BYTES = 48 * 1024 * 1024;
const FETCH_TIMEOUT_MS = 60_000;
/** Upstream fetches at once; more wait their turn. */
export const TILE_UPSTREAM_CONCURRENCY = 6;
/** Tiles change only when new imagery is processed. */
export const TILE_TTL_MS = 24 * 60 * 60 * 1000;
/** Trimmed tiles are ~80 KB: a dozen 9-tile views, well under 100 MB. */
export const TILE_MEMORY_MAX_TILES = 120;
/** A trimmed tile is ~80 KB; a body this big failed to trim and is not kept. */
export const TILE_MEMORY_MAX_TILE_BYTES = 4 * 1024 * 1024;
/** The longest upstream Retry-After honoured, in seconds. */
export const MAX_RETRY_AFTER_SEC = 600;
/** The client token lives in the browser by design; the server adds it to tile URLs too. */
export function mapillaryToken() {
return String(process.env.MAPILLARY_CLIENT_TOKEN || '').trim();
}
/** Thrown for a request this proxy refuses before contacting Mapillary. */
export class TileRequestError extends Error {
constructor(message, status = 400) {
super(message);
this.name = 'TileRequestError';
this.status = status;
}
}
/** Mapillary answered with an error status; 401/403 set `keyRejected`. */
/** Mapillary answered with an error status (or a size this proxy refuses). */
export class TileUpstreamError extends Error {
constructor(status, message, { retryAfterSec = null } = {}) {
constructor(status, message, retryAfter = null) {
super(message || `Mapillary tiles HTTP ${status}`);
this.name = 'TileUpstreamError';
this.status = status;
this.keyRejected = status === 401 || status === 403;
this.retryAfterSec = retryAfterSec;
/** The upstream Retry-After header, as sent. */
this.retryAfter = retryAfter;
}
}
/**
* A rejected token or rate limit that answers every tile miss until it expires
* (or the token changes). Cached tiles are still served meanwhile.
* @type {{status: number, until: number, token: string}|null}
*/
let _upstreamHold = null;
/** Milliseconds to wait from a Retry-After header (seconds or an HTTP date). */
function retryAfterMs(header, now = Date.now()) {
/** Seconds a Retry-After header asks for (an integer or an HTTP date), else null. */
function retryAfterSeconds(header, now = Date.now()) {
const text = String(header ?? '').trim();
let ms = NaN;
if (/^\d+$/.test(text)) ms = Number(text) * 1000;
else if (text) ms = Date.parse(text) - now;
if (!Number.isFinite(ms) || ms <= 0) return TILE_RATE_LIMIT_HOLD_MS;
return Math.min(ms, TILE_RATE_LIMIT_MAX_HOLD_MS);
if (!text) return null;
const seconds = /^\d+$/.test(text)
? Number(text)
: (Date.parse(text) - now) / 1000;
return Number.isFinite(seconds) && seconds > 0 ? Math.ceil(seconds) : null;
}
function heldRefusal(token) {
const hold = _upstreamHold;
if (!hold) return null;
const left = hold.until - Date.now();
if (left <= 0 || hold.token !== token) {
_upstreamHold = null;
/** The wait a 429 asks for, bounded; a missing or malformed header means a minute. */
export function retryAfterWait(header, now = Date.now()) {
return Math.min(retryAfterSeconds(header, now) ?? 60, MAX_RETRY_AFTER_SEC);
}
/**
* While Mapillary rate-limits, tile misses are refused here until the wait is
* over, so the rest of a view (and the fetches queued behind the first 429)
* do not go on asking. Scoped to the token that was refused.
* @type {{until: number, token: string}|null}
*/
let refusal = null;
/** The refusal still in force for `token`, as the error to answer with. */
function refusedNow(token) {
if (!refusal) return null;
const left = refusal.until - Date.now();
if (left <= 0 || refusal.token !== token) {
refusal = null;
return null;
}
return new TileUpstreamError(hold.status, undefined, {
retryAfterSec: Math.ceil(left / 1000),
});
return new TileUpstreamError(429, undefined, String(Math.ceil(left / 1000)));
}
/**
* Validate a tile address against the layer's zoom ranges.
* @returns {{layer:string, upstream:string, z:number, x:number, y:number, key:string}}
* Parse `/coverage/{z}/{x}/{y}`; null unless z is 11–14 and x/y are on the grid.
* @returns {{z: number, x: number, y: number, key: string}|null}
*/
export function normalizeTileAddress({ layer, z, x, y }) {
const spec = Object.hasOwn(TILE_LAYERS, layer) ? TILE_LAYERS[layer] : null;
if (!spec) throw new TileRequestError(`Unknown tile layer: ${layer}`);
const zi = Number(z);
const xi = Number(x);
const yi = Number(y);
if (
![zi, xi, yi].every((v) => Number.isInteger(v) && v >= 0) ||
!spec.zoomRanges.some(([min, max]) => zi >= min && zi <= max)
)
throw new TileRequestError(
`Tile zoom for ${layer} must be ${spec.zoomRanges
.map(([min, max]) => `${min}–${max}`)
.join(' or ')}`,
);
const n = 2 ** zi;
if (xi >= n || yi >= n)
throw new TileRequestError('Tile address out of range');
return {
layer,
upstream: spec.upstream,
dropLayers: spec.dropLayers || [],
z: zi,
x: xi,
y: yi,
key: `${layer}/${zi}/${xi}/${yi}`,
};
export function parseTilePath(pathname) {
const match = /^\/coverage\/(\d{1,2})\/(\d{1,6})\/(\d{1,6})$/.exec(pathname);
if (!match) return null;
const [z, x, y] = match.slice(1).map(Number);
if (z < MIN_ZOOM || z > MAX_ZOOM || x >= 2 ** z || y >= 2 ** z) return null;
return { z, x, y, key: `${z}/${x}/${y}` };
}
/** @type {Map<string, {bytes: Buffer, at: number}>} insertion-ordered LRU */
const _memory = new Map();
let _memoryBytes = 0;
const memoryCost = (bytes) => bytes.length + TILE_MEMORY_ENTRY_OVERHEAD_BYTES;
/**
* Upstream fetches shared by every request for the same tile; the last waiter
* to leave cancels it (see `joinFlight`).
* @type {Map<string, {controller: AbortController, promise: Promise<Buffer>, waiters: number}>}
*/
const _inFlight = new Map();
function memoryGet(key) {
const hit = _memory.get(key);
if (!hit) return null;
// Same life as the disk copy, so a long-running server never serves stale
// coverage.
if (Date.now() - hit.at > TILE_DISK_TTL_MS) {
_memory.delete(key);
_memoryBytes -= memoryCost(hit.bytes);
return null;
}
_memory.delete(key);
_memory.set(key, hit);
return hit.bytes;
}
/** `at` is when Mapillary served the bytes (a disk tile's mtime). */
function memoryPut(key, bytes, at = Date.now()) {
if (memoryCost(bytes) > TILE_MEMORY_BUDGET_BYTES / 2) return;
const existing = _memory.get(key);
if (existing) _memoryBytes -= memoryCost(existing.bytes);
_memory.set(key, { bytes, at });
_memoryBytes += memoryCost(bytes);
while (_memoryBytes > TILE_MEMORY_BUDGET_BYTES && _memory.size) {
const [oldest, entry] = _memory.entries().next().value;
_memory.delete(oldest);
_memoryBytes -= memoryCost(entry.bytes);
}
}
/** Disk cache root; tests point it at a temporary directory. */
let _diskDir = TILE_DISK_DIR;
function diskPath({ layer, z, x, y }, root = _diskDir) {
return path.join(root, layer, String(z), `${x}-${y}.pbf`);
}
async function readDisk(address) {
const file = diskPath(address);
function trim(bytes) {
try {
const stat = await fsp.stat(file);
if (Date.now() - stat.mtimeMs > TILE_DISK_TTL_MS) return null;
return { bytes: await fsp.readFile(file), at: stat.mtimeMs };
} catch {
return null;
}
}
/** Unsettled background writes and sweeps; requests never wait on them. */
const _background = new Set();
function track(task) {
const settled = task
.catch(() => {})
.finally(() => _background.delete(settled));
_background.add(settled);
return settled;
}
/** @type {Map<string, number>} cache file to its newest write's sequence */
const _diskWrites = new Map();
let _diskWriteSeq = 0;
function diskWritePending(address) {
return _diskWrites.has(diskPath(address));
}
/**
* Write a tile in the background. `at` keeps a rewrite's original fetch time
* so it does not extend the tile's life. An overtaken write steps aside.
*/
function writeDisk(address, bytes, at = null) {
// Sweep this write's root even if the cache dir has moved since.
const root = _diskDir;
const file = diskPath(address, root);
const seq = ++_diskWriteSeq;
const tmp = `${file}.${process.pid}-${seq}.tmp`;
const newest = () => _diskWrites.get(file) === seq;
_diskWrites.set(file, seq);
return track(
fsp
.mkdir(path.dirname(file), { recursive: true })
.then(() => fsp.writeFile(tmp, bytes))
.then(() => at != null && fsp.utimes(tmp, new Date(), new Date(at)))
.then(() => {
if (!newest()) return fsp.rm(tmp, { force: true });
return fsp.rename(tmp, file).then(() => scheduleSweep(root));
})
.catch(async (error) => {
await fsp.rm(tmp, { force: true }).catch(() => {});
warn('tile cache write failed', error);
})
.finally(() => {
if (newest()) _diskWrites.delete(file);
}),
);
}
let _lastSweepAt = 0;
let _sweeping = null;
let _sweepWarned = false;
function scheduleSweep(root) {
if (_sweeping || Date.now() - _lastSweepAt < TILE_DISK_SWEEP_INTERVAL_MS)
return;
_lastSweepAt = Date.now();
const sweep = sweepTileDisk({ root })
.catch((error) => {
if (_sweepWarned) return;
_sweepWarned = true;
warn('tile cache sweep failed', error);
})
.finally(() => {
if (_sweeping === tracked) _sweeping = null;
});
const tracked = track(sweep);
_sweeping = tracked;
}
/** Every file under `dir` with size and mtime; a missing dir is empty. */
async function listCacheFiles(dir) {
let entries;
try {
entries = await fsp.readdir(dir, { withFileTypes: true });
return stripTileLayers(bytes, DROP_LAYERS);
} catch (error) {
if (error?.code === 'ENOENT') return [];
throw error;
}
const files = [];
for (const entry of entries) {
const file = path.join(dir, entry.name);
if (entry.isDirectory()) files.push(...(await listCacheFiles(file)));
else if (entry.isFile()) {
const stat = await fsp.stat(file).catch(() => null);
if (stat) files.push({ file, size: stat.size, at: stat.mtimeMs });
}
}
return files;
}
/**
* Delete expired tiles, then the oldest until the rest fit in `maxBytes`.
* @returns {Promise<{removed: number, bytes: number}>} files deleted, charged bytes kept
*/
export async function sweepTileDisk({
root = _diskDir,
maxBytes = TILE_DISK_MAX_BYTES,
} = {}) {
const now = Date.now();
const files = (await listCacheFiles(root)).sort((a, b) => a.at - b.at);
const cost = (size) => size + TILE_DISK_FILE_OVERHEAD_BYTES;
let bytes = files.reduce((sum, { size }) => sum + cost(size), 0);
let removed = 0;
for (const { file, size, at } of files) {
if (now - at <= TILE_DISK_TTL_MS && bytes <= maxBytes) break;
await fsp.rm(file, { force: true });
bytes -= cost(size);
removed++;
}
return { removed, bytes };
}
/** The one origin the tile token may be sent to, at every redirect hop. */
const TILE_ORIGIN = new URL(MAPILLARY_TILE_HOST).origin;
export const TILE_MAX_REDIRECTS = 3;
const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);
/**
* Follow redirects by hand so every hop stays HTTPS on the tile origin and the
* token in the query never reaches another host. A bad redirect is a 502.
*/
async function fetchPinned(url, init) {
let next = url;
for (let hop = 0; ; hop++) {
const response = await fetch(next, { ...init, redirect: 'manual' });
if (!REDIRECT_STATUSES.has(response.status)) return response;
await response.body?.cancel().catch(() => {});
if (hop >= TILE_MAX_REDIRECTS)
throw new TileUpstreamError(502, 'Mapillary tile redirected too often');
let target = null;
try {
target = new URL(response.headers.get('location') ?? '', next);
} catch {
/* unparseable Location */
}
if (
!response.headers.get('location') ||
target?.protocol !== 'https:' ||
target.origin !== TILE_ORIGIN ||
target.username ||
target.password
)
throw new TileUpstreamError(
502,
'Mapillary tile redirect left the tile origin',
);
next = target.href;
}
}
async function fetchUpstream(address, signal) {
const token = mapillaryToken();
if (!token) throw new TileRequestError('Mapillary token not configured', 503);
const held = heldRefusal(token);
if (held) throw held;
const url = `${MAPILLARY_TILE_HOST}/${address.upstream}/2/${address.z}/${address.x}/${address.y}?access_token=${encodeURIComponent(token)}`;
const timeout = AbortSignal.timeout(TILE_FETCH_TIMEOUT_MS);
const oversize = new AbortController();
const response = await fetchPinned(url, {
signal: AbortSignal.any([signal, timeout, oversize.signal].filter(Boolean)),
headers: { Accept: 'application/x-protobuf' },
});
if (response.status === 404 || response.status === 204) {
await response.body?.cancel().catch(() => {});
return Buffer.alloc(0);
}
if (!response.ok) {
await response.body?.cancel().catch(() => {});
const error = new TileUpstreamError(response.status);
if (error.keyRejected)
_upstreamHold = {
status: response.status,
until: Date.now() + TILE_KEY_REJECTED_HOLD_MS,
token,
};
else if (response.status === 429) {
const wait = retryAfterMs(response.headers.get('retry-after'));
_upstreamHold = { status: 429, until: Date.now() + wait, token };
error.retryAfterSec = Math.ceil(wait / 1000);
}
throw error;
}
// Chunked or compressed bodies have no Content-Length: the cap applies as
// the body streams, and an oversized one also aborts the request.
try {
const bytes = await readResponseBytesCapped(
response,
TILE_MAX_BYTES,
signal,
);
return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength);
} catch (error) {
if (error?.code !== 'RESPONSE_TOO_LARGE') throw error;
oversize.abort();
throw new TileUpstreamError(502, 'Mapillary tile exceeds size cap');
}
}
/** @type {{active: number, queue: Array<() => void>}} upstream slots */
let _slots = { active: 0, queue: [] };
/**
* Resolve with a release function once a slot is free. An abort while
* queued leaves the queue.
* @returns {Promise<() => void>}
*/
function acquireUpstreamSlot(signal) {
const slots = _slots;
return new Promise((resolve, reject) => {
if (signal.aborted) return reject(signal.reason);
const onAbort = () => {
const index = slots.queue.indexOf(grant);
if (index !== -1) slots.queue.splice(index, 1);
reject(signal.reason);
};
function grant() {
signal.removeEventListener('abort', onAbort);
slots.active++;
let released = false;
resolve(() => {
if (released) return;
released = true;
slots.active--;
slots.queue.shift()?.();
});
}
if (slots.active < TILE_UPSTREAM_CONCURRENCY) return grant();
signal.addEventListener('abort', onAbort, { once: true });
slots.queue.push(grant);
});
}
function trim(address, bytes) {
if (!address.dropLayers.length || !bytes.length) return bytes;
try {
return stripTileLayers(bytes, address.dropLayers);
} catch (error) {
warn('tile trim failed, serving raw', error);
console.warn('[Mapillary Proxy] tile trim failed:', error?.message);
return bytes;
}
}
/**
* Fetch one tile from memory, disk, a shared flight, then Mapillary. Empty
* bytes mean no coverage.
* @param {{layer:string,z:number|string,x:number|string,y:number|string}} request
* @param {{signal?: AbortSignal}} [options]
* @returns {Promise<{bytes: Buffer, source: 'memory'|'disk'|'inflight'|'upstream', address: object}>}
*/
export async function fetchTile(request, { signal } = {}) {
const address = normalizeTileAddress(request);
const memory = memoryGet(address.key);
if (memory) return { bytes: memory, source: 'memory', address };
const disk = await readDisk(address);
if (disk) {
// Older cache files may be untrimmed: rewrite once, keeping the fetch time.
const bytes = trim(address, disk.bytes);
if (bytes !== disk.bytes && !diskWritePending(address))
writeDisk(address, bytes, disk.at);
memoryPut(address.key, bytes, disk.at);
return { bytes, source: 'disk', address };
/** One tile from Mapillary, trimmed; empty bytes mean no coverage. */
async function fetchUpstream({ z, x, y }, signal) {
const token = mapillaryToken();
const refused = refusedNow(token);
if (refused) throw refused;
const response = await fetch(
`${TILE_HOST}/mly1_public/2/${z}/${x}/${y}?access_token=${encodeURIComponent(token)}`,
{
signal: AbortSignal.any([signal, AbortSignal.timeout(FETCH_TIMEOUT_MS)]),
// The token is in the query: a redirect is never followed (a 3xx is a 502).
redirect: 'manual',
headers: { Accept: 'application/x-protobuf' },
},
);
if (!response.ok || response.status === 204) {
await response.body?.cancel().catch(() => {});
if (response.status === 204 || response.status === 404)
return Buffer.alloc(0);
const retryAfter = response.headers.get('retry-after');
if (response.status === 429)
refusal = {
token,
until: Date.now() + retryAfterWait(retryAfter) * 1000,
};
throw new TileUpstreamError(response.status, undefined, retryAfter);
}
try {
const bytes = await readResponseBytesCapped(response, TILE_MAX_BYTES);
return trim(Buffer.from(bytes.buffer, bytes.byteOffset, bytes.length));
} catch (error) {
if (error?.code !== 'RESPONSE_TOO_LARGE') throw error;
throw new TileUpstreamError(502, 'Mapillary tile exceeds size cap');
}
// A caller gone while the disk was read must not start a fetch nobody joins.
signal?.throwIfAborted();
let flight = _inFlight.get(address.key);
const joined = Boolean(flight) && !flight.controller.signal.aborted;
if (!joined) flight = startFlight(address);
const bytes = await joinFlight(flight, signal);
return { bytes, source: joined ? 'inflight' : 'upstream', address };
}
/** Start the shared upstream fetch; it holds a slot until the body is read. */
let upstreamActive = 0;
/** @type {Array<() => void>} fetches waiting for a slot, oldest first */
const upstreamWaiting = [];
/**
* Run `task` in an upstream slot, waiting for one unless `signal` aborts
* first. A freed slot passes straight to the oldest waiter, still counted, so
* a request arriving meanwhile cannot take it first.
*/
async function withUpstreamSlot(signal, task) {
if (upstreamActive < TILE_UPSTREAM_CONCURRENCY) upstreamActive++;
else
await new Promise((resolve, reject) => {
const wake = () => {
signal.removeEventListener('abort', leave);
resolve();
};
const leave = () => {
const at = upstreamWaiting.indexOf(wake);
if (at !== -1) upstreamWaiting.splice(at, 1);
reject(signal.reason);
};
upstreamWaiting.push(wake);
signal.addEventListener('abort', leave, { once: true });
});
try {
// Handed a slot after leaving: it goes on to the next waiter.
signal.throwIfAborted();
return await task();
} finally {
const next = upstreamWaiting.shift();
if (next) next();
else upstreamActive--;
}
}
/** `promise`, or a rejection as soon as `signal` aborts. */
function unlessAborted(promise, signal) {
if (!signal) return promise;
signal.throwIfAborted();
return new Promise((resolve, reject) => {
const leave = () => reject(signal.reason);
signal.addEventListener('abort', leave, { once: true });
promise
.then(resolve, reject)
.finally(() => signal.removeEventListener('abort', leave));
});
}
/** @type {Map<string, {bytes: Buffer, at: number}>} least recently used first */
const memory = new Map();
/**
* Upstream fetches shared per tile, with the requests still waiting on each.
* @type {Map<string, {promise: Promise<Buffer>, controller: AbortController, waiters: number}>}
*/
const inFlight = new Map();
function remember(key, bytes, at = Date.now()) {
if (bytes.length > TILE_MEMORY_MAX_TILE_BYTES) return;
memory.delete(key);
memory.set(key, { bytes, at });
for (const oldest of memory.keys()) {
if (memory.size <= TILE_MEMORY_MAX_TILES) break;
memory.delete(oldest);
}
}
function recall(key) {
const hit = memory.get(key);
if (!hit) return null;
if (Date.now() - hit.at > TILE_TTL_MS) {
memory.delete(key);
return null;
}
remember(key, hit.bytes, hit.at); // most recently used again
return hit.bytes;
}
/** Start the upstream fetch for a tile, shared until its last request leaves. */
function startFlight(address) {
const controller = new AbortController();
const flight = { controller, waiters: 0, promise: null };
flight.promise = acquireUpstreamSlot(controller.signal)
.then((release) =>
fetchUpstream(address, controller.signal).finally(release),
)
.then((raw) => trim(address, raw))
const flight = { controller, waiters: 0 };
flight.promise = withUpstreamSlot(controller.signal, () =>
fetchUpstream(address, controller.signal),
)
.then((bytes) => {
memoryPut(address.key, bytes);
writeDisk(address, bytes);
remember(address.key, bytes);
return bytes;
})
.finally(() => {
if (_inFlight.get(address.key) === flight) _inFlight.delete(address.key);
if (inFlight.get(address.key) === flight) inFlight.delete(address.key);
});
// Every waiter may have left before it settles; nobody awaits it then.
// Every request may have left; the rejection is then nobody's.
flight.promise.catch(() => {});
_inFlight.set(address.key, flight);
inFlight.set(address.key, flight);
return flight;
}
/**
* Wait on a shared flight. A caller's abort rejects only that caller; the
* fetch is cancelled when the last waiter leaves.
* One tile from memory, a shared in-flight fetch, or Mapillary. `signal`
* aborts when the request's client leaves; the fetch stops once all have.
* @param {{z: number, x: number, y: number, key: string}} address from parseTilePath
* @param {{signal?: AbortSignal}} [options]
* @returns {Promise<{bytes: Buffer, source: 'memory'|'inflight'|'upstream'}>}
*/
function joinFlight(flight, signal) {
signal?.throwIfAborted();
export async function fetchTile(address, { signal } = {}) {
const cached = recall(address.key);
if (cached) return { bytes: cached, source: 'memory' };
let flight = inFlight.get(address.key);
const shared = Boolean(flight);
flight ||= startFlight(address);
flight.waiters++;
return new Promise((resolve, reject) => {
let done = false;
const leave = () => {
if (done) return false;
done = true;
signal?.removeEventListener('abort', onAbort);
flight.waiters--;
return true;
};
const onAbort = () => {
if (!leave()) return;
if (flight.waiters === 0) flight.controller.abort();
reject(signal.reason);
};
signal?.addEventListener('abort', onAbort, { once: true });
flight.promise.then(
(bytes) => leave() && resolve(bytes),
(error) => leave() && reject(error),
);
});
try {
const bytes = await unlessAborted(flight.promise, signal);
return { bytes, source: shared ? 'inflight' : 'upstream' };
} finally {
if (--flight.waiters === 0) {
// A later request for the tile must not join a cancelled fetch.
if (inFlight.get(address.key) === flight) inFlight.delete(address.key);
flight.controller.abort();
}
}
}
/** Test seam: forget every cached tile held in memory. */
export function _resetTileMemoryForTest() {
_memory.clear();
_memoryBytes = 0;
_inFlight.clear();
_upstreamHold = null;
_slots = { active: 0, queue: [] };
}
/** Test seam: how many tiles memory holds and the bytes charged for them. */
export function _tileMemoryForTest() {
return { entries: _memory.size, bytes: _memoryBytes };
}
/** Test seam: wait for every background write and the sweeps they start. */
export async function _settleTileWritesForTest() {
while (_background.size) await Promise.allSettled([..._background]);
}
/** Test seam: keep the disk cache in `dir` (null restores the default). */
export function _setTileCacheDirForTest(dir) {
_diskDir = dir || TILE_DISK_DIR;
_lastSweepAt = 0;
/** Test seam: forget every cached and in-flight tile. */
export function _resetTileCacheForTest() {
memory.clear();
inFlight.clear();
refusal = null;
upstreamActive = 0;
upstreamWaiting.length = 0;
}
-13
View File
@@ -41,16 +41,3 @@ export function stripTileLayers(bytes, dropNames) {
}, null);
return dropped ? Buffer.from(writer.finish()) : bytes;
}
/** Names of the layers inside a vector tile, in order. */
export function listTileLayers(bytes) {
const names = [];
if (!bytes?.length) return names;
new PbfReader(bytes).readFields((tag, _result, pbf) => {
if (tag !== TILE_LAYER_FIELD) return;
const end = pbf.readVarint() + pbf.pos;
names.push(layerName(bytes.subarray(pbf.pos, end)));
pbf.pos = end;
}, null);
return names;
}
+1 -1
View File
@@ -145,7 +145,7 @@ const CODEX_LOGIN_ENV =
/^(?:PATH|PATHEXT|HOME|USER|USERNAME|LOGNAME|SHELL|LANG|LANGUAGE|LC_[A-Z_]+|TERM|TZ|TMPDIR|TEMP|TMP|XDG_[A-Z_]+|DISPLAY|WAYLAND_DISPLAY|DBUS_SESSION_BUS_ADDRESS|BROWSER|WSL_[A-Z_]+|WSLENV|(?:HTTPS?|ALL|NO)_PROXY|SSL_CERT_(?:FILE|DIR)|NODE_EXTRA_CA_CERTS|SYSTEMROOT|WINDIR|COMSPEC|USERPROFILE|APPDATA|LOCALAPPDATA|PROGRAMDATA|HOMEDRIVE|HOMEPATH|CODEX_[A-Z_]+)$/i;
/** The environment `codex login` runs with: the allowlist, then `CODEX_HOME`. */
export function codexLoginEnvironment(environment, codexHome) {
function codexLoginEnvironment(environment, codexHome) {
const env = {};
for (const [name, value] of Object.entries(environment || {}))
if (CODEX_LOGIN_ENV.test(name) && value !== undefined) env[name] = value;
+1 -1
View File
@@ -157,4 +157,4 @@ function createDebugLogHandler({
};
}
export { createDebugLogHandler, omitVoiceContent };
export { createDebugLogHandler };
+20
View File
@@ -1,5 +1,6 @@
import { makeRateLimiter, clientKey } from './common/rate-limit.js';
import { readRequestBodyCapped } from './common/request.js';
import { admitSameSite } from './common/same-site.js';
import {
OVERPASS_MAX_BODY_BYTES,
OVERPASS_MAX_CONCURRENT,
@@ -27,6 +28,14 @@ const _overpassInFlight = new Map();
let _overpassConcurrent = 0;
// Bound body-reading and validation work, including malformed requests and
// cache hits. Keep the tighter upstream budget below independent of this gate.
const _overpassRequestLimiter = makeRateLimiter({
windowMs: 60_000,
max: 300,
globalMax: 900,
});
const _overpassRateLimiter = makeRateLimiter({
windowMs: 60_000,
max: 90,
@@ -94,6 +103,17 @@ function overpassProxy({ routing = {} } = {}) {
return;
}
if (admitSameSite(req, res)) return;
if (!_overpassRequestLimiter(clientKey(req))) {
res.writeHead(429, {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
'Retry-After': '60',
});
res.end(JSON.stringify({ error: 'Rate limit exceeded' }));
return;
}
// Collect POST body with a hard byte cap (Overpass QL queries are small)
let body;
try {
+31 -5
View File
@@ -18,6 +18,29 @@ function isOverpassBoundaryQuery(cacheKey) {
return /is_in\s*\(|\bpivot\b/i.test(String(cacheKey || ''));
}
// Input has already had quoted literals and comments removed. Consume each
// character once: a regex that searches for a closing bracket from every `[` is
// quadratic on an unterminated filter. Nested or unmatched brackets are invalid.
function stripTagFilters(text) {
const parts = [];
let start = 0;
let inFilter = false;
for (let i = 0; i < text.length; i += 1) {
if (text[i] === '[') {
if (inFilter) return null;
parts.push(text.slice(start, i), ' ');
inFilter = true;
} else if (text[i] === ']') {
if (!inFilter) return null;
inFilter = false;
start = i + 1;
}
}
if (inFilter) return null;
parts.push(text.slice(start));
return parts.join('');
}
/**
* Validate + clamp an Overpass form body. Defends the generic proxy against
* planet-scale abuse: requires exactly one `data` query in which EVERY element
@@ -50,6 +73,10 @@ function sanitizeOverpassBody(rawBody) {
// Blank quoted literals + strip comments in one lexer pass so a fake bound or a
// `//` inside a string can't hide an unbounded selector (or satisfy a bound).
const stripped = stripOverpassNoise(data);
const withoutTags = stripTagFilters(stripped);
if (withoutTags === null) {
return { ok: false, error: 'Malformed Overpass tag filter' };
}
// Reject oversized radii in EVERY around form — point `around:r,lat,lon` AND the
// input-set form `around.set:r` — and parse the full numeric token so scientific
@@ -93,20 +120,19 @@ function sanitizeOverpassBody(rawBody) {
// statement. So `way[...]->.a` (global assigned to a set) is rejected, while the
// app's `is_in(...)->.a; area.a[...]` and `area(id)->.x; rel(pivot.x)` validate.
const boundedSets = new Set();
for (let stmt of stripped.split(';')) {
for (let stmt of withoutTags.split(';')) {
stmt = stmt.trim();
if (!stmt || stmt.startsWith('[') || /^out\b/.test(stmt)) continue;
// Strip output-set assignments (NOT input bounds), then strip bracket tag
// filters so a tag KEY/value (e.g. `way[is_in]`, `node[around]`) can never be
// Strip output-set assignments (NOT input bounds). Bracket tag filters were
// removed above so a tag KEY/value (e.g. `way[is_in]`, `node[around]`) cannot be
// misread as a spatial bound. Bounds live in (...) / function calls / set
// refs, never inside [...], so the probe loses nothing real.
const outSets = [];
const body = stmt.replace(/->\s*\.(\w+)/g, (_, name) => {
const probe = stmt.replace(/->\s*\.(\w+)/g, (_, name) => {
outSets.push(name);
return ' ';
});
const probe = body.replace(/\[[^\]]*\]/g, ' ');
// Reject element-in-area scans on the TAG-STRIPPED probe, so a tag filter
// between the selector and the area filter (way["highway"](area.a)) can't hide
@@ -0,0 +1,178 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { Readable } from 'node:stream';
import { spawnSync } from 'node:child_process';
import { overpassProxy } from '../overpass.js';
import { sanitizeOverpassBody } from './query.js';
const form = (query) => `data=${encodeURIComponent(query)}`;
test('bracket validation rejects malformed filters without trusting their contents as bounds', () => {
for (const query of [
'node[',
'node]',
'node[[name]](around:20,1,1);out;',
'node[name=around:20];out;',
'node[is_in];out;',
'node[name="(around:20,1,1)"];out;',
'node(around:20,1,1);way[name];out;',
'way[highway](area.a);out;',
'node[->.a](around:20,1,1);way.a;out;',
]) {
assert.equal(sanitizeOverpassBody(form(query)).ok, false, query);
}
});
test('bounded queries preserve quoted brackets, comments, multiple tags and timeout clamping', () => {
for (const query of [
'[out:json][timeout:999];node[name="[Town]; ]"](around:20,1,1);out;',
'/* [ */way[highway][name](30.1,-97.9,30.5,-97.5);out geom;',
'is_in(30.2,-97.7)->.a;area.a[boundary="administrative"];out tags;',
'area(3600039152)->.x;rel(pivot.x);out geom;',
]) {
const result = sanitizeOverpassBody(form(query));
assert.equal(result.ok, true, query);
assert.equal(
new URLSearchParams(result.body).get('data'),
query.replace('[timeout:999]', '[timeout:30]'),
);
}
});
test('a burst of maximum-size unterminated filters completes within a bounded process deadline', () => {
// A child deadline can stop synchronous regex backtracking; a test timeout in
// the same event loop cannot. The old parser takes several seconds for this.
const result = spawnSync(
process.execPath,
[
'--input-type=module',
'-e',
`import { sanitizeOverpassBody } from ${JSON.stringify(new URL('./query.js', import.meta.url).href)};
for (let i = 0; i < 20; i++) {
if (sanitizeOverpassBody('data=node' + '['.repeat(24000)).ok) process.exit(1);
}`,
],
{ timeout: 2000, encoding: 'utf8' },
);
assert.ifError(result.error);
assert.equal(result.status, 0, result.stderr);
});
function handler(hook = 'configureServer') {
let route;
overpassProxy()[hook]({
middlewares: {
use(path, fn) {
if (path === '/api/overpass') route = fn;
},
},
});
return route;
}
async function call(
route,
{
headers = {},
peer = '127.0.0.1',
body = 'data=node[',
forbidRead = false,
method = 'POST',
url = '/',
} = {},
) {
let reads = 0;
const req = new Readable({
read() {
reads++;
assert.equal(forbidRead, false, 'refused request must not read its body');
this.push(Buffer.from(body));
this.push(null);
},
});
Object.assign(req, {
method,
url,
headers: { host: 'localhost:4173', ...headers },
socket: { remoteAddress: peer },
});
const res = {
statusCode: 200,
headers: {},
setHeader(key, value) {
this.headers[key.toLowerCase()] = value;
},
writeHead(status, values) {
this.statusCode = status;
for (const [key, value] of Object.entries(values))
this.setHeader(key, value);
},
end(value) {
this.body = value;
},
};
await route(req, res);
return { ...res, reads };
}
for (const hook of ['configureServer', 'configurePreviewServer']) {
test(`${hook}: cross-site requests are rejected before consuming a body or fetching`, async (t) => {
t.mock.method(globalThis, 'fetch', () =>
assert.fail('unexpected upstream fetch'),
);
const route = handler(hook);
for (const headers of [
{ origin: 'https://foreign.example', 'content-type': 'text/plain' },
{ 'sec-fetch-site': 'cross-site' },
{ 'sec-fetch-site': 'same-site' },
{ origin: 'null' },
{ 'x-forwarded-for': '127.0.0.1' },
]) {
const response = await call(route, { headers, forbidRead: true });
assert.equal(response.statusCode, 403);
assert.equal(response.reads, 0);
assert.equal(response.headers['cache-control'], 'no-store');
}
});
}
test('same-origin, LAN and non-browser requests still reach validation; GET status stays available', async () => {
const route = handler();
for (const headers of [
{},
{ origin: 'http://localhost:4173', 'sec-fetch-site': 'same-origin' },
{ host: '192.168.1.10:4173', origin: 'http://192.168.1.10:4173' },
]) {
const response = await call(route, { headers });
assert.equal(response.statusCode, 400);
assert.equal(response.reads, 1);
assert.match(response.body, /Malformed Overpass tag filter/);
}
const response = await call(route, {
method: 'GET',
url: '/status',
forbidRead: true,
});
assert.equal(response.statusCode, 200);
assert.equal(typeof JSON.parse(response.body).configured, 'boolean');
});
test('invalid requests spend admission quota before parsing, including on an unconfigured server', async (t) => {
let now = Date.now() + 120000;
t.mock.method(Date, 'now', () => now);
t.mock.method(globalThis, 'fetch', () =>
assert.fail('unexpected upstream fetch'),
);
const route = handler();
const peer = 'admission-limit-test';
for (let i = 0; i < 300; i++) {
assert.equal((await call(route, { peer })).statusCode, 400);
}
const denied = await call(route, { peer, forbidRead: true });
assert.equal(denied.statusCode, 429);
assert.equal(denied.reads, 0);
assert.equal(denied.headers['retry-after'], '60');
assert.equal((await call(route, { peer: 'other-peer' })).statusCode, 400);
now += 60001;
assert.equal((await call(route, { peer })).statusCode, 400);
});
+1 -1
View File
@@ -26,7 +26,7 @@ import {
RADIO_UUID_RE,
RADIO_FALLBACK_MIRRORS,
} from './constants.js';
export async function mapRadioConcurrent(values, concurrency, mapper) {
async function mapRadioConcurrent(values, concurrency, mapper) {
const results = new Array(values.length);
let cursor = 0;
const workers = Array.from(
+2
View File
@@ -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}` } : {}),
};
}
+104 -9
View File
@@ -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
@@ -18,13 +20,33 @@ import {
* cache and serve-stale shape. Cache entries and stale fallback are keyed per
* 5dp point, so reordered and partially overlapping batches reuse prior work.
* 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.
* request order. Each request is bounded to one 64-point upstream chunk;
* larger client workloads are split by the browser terrain service.
*
* 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
@@ -33,10 +55,51 @@ export function terrainHeightsProxy() {
// 2026-08-21). 64 points costs 5.6-12 s, inside the timeout at both ends of
// that range.
const UPSTREAM_CHUNK = 64;
const MAX_POINTS = 2000;
const MAX_POINTS = UPSTREAM_CHUNK;
/**
* 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,13 +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 {
await loadDiskOnce();
if (!allow(clientKey(req))) {
res.writeHead(429, {
'Content-Type': 'application/json',
'Retry-After': '10',
});
res.end(JSON.stringify({ error: 'Rate limit exceeded' }));
return;
}
const parsedUrl = new URL(req.url || '', 'http://internal');
const rawPoints = parsedUrl.searchParams.get('points');
const points = parseTerrainPoints(rawPoints);
@@ -175,19 +266,23 @@ export function terrainHeightsProxy() {
return;
}
if (points.length > MAX_POINTS) {
send(500, {
send(413, {
error: `too many points (${points.length}); max ${MAX_POINTS} per request`,
});
return;
}
await loadDiskOnce();
const outcome = await resolveTerrainHeightRequest({
points,
cache: mem,
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' +
-1
View File
@@ -1,5 +1,4 @@
export {
createTransitHistory,
TRANSIT_HISTORY_LIMITS,
validTransitIdentifier,
} from '../../src/sources/transitHistoryStore.js';
+1 -1
View File
@@ -3,7 +3,7 @@ import { decodeWindGribMessage } from './decode.js';
import { resampleWeatherSnapshot, weatherScalarMetadata } from './grid.js';
/** Keyless ECMWF Open Data root for real-time IFS forecasts. */
export const IFS_BASE = 'https://data.ecmwf.int/forecasts';
const IFS_BASE = 'https://data.ecmwf.int/forecasts';
/**
* Select the latest IFS cycle that should be available.
+2 -1
View File
@@ -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(),