diff --git a/CHANGELOG.md b/CHANGELOG.md index f87443c90..3f42ce3ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,11 @@ of current runtime behavior, see [`docs/CURRENT-STATE.md`](docs/CURRENT-STATE.md ## [Unreleased] +- Split aircraft and vessel server providers into focused modules for source + fetching, AIS records/tracks and shared request helpers; preserve existing + routes, local setup, fallback behavior and rendering. + + ### Changed - Separate explicit browser build settings from standalone environment loading and local provider middleware. Preserve provider behavior and root named exports. diff --git a/docs/CODE-BOUNDARIES.md b/docs/CODE-BOUNDARIES.md index 6970b768d..8533e9710 100644 --- a/docs/CODE-BOUNDARIES.md +++ b/docs/CODE-BOUNDARIES.md @@ -84,3 +84,19 @@ credential-store paths. Provider Settings writes to the same root `.env` or Pinokio store as before. `vite.config.js` preserves the default configuration and existing named provider exports for tools/tests. The provider module remains large; later extractions should split complete provider families and their tests. + +## Aircraft and vessel providers + +`server/providers/live.js` exports the existing Node middleware factories and +request helpers. `aircraft/` owns OpenSky state/fallback, military positions, +enrichment and track endpoints in separate files. `vessels/ais-live.js` owns +websocket setup and route responses; `vessels/ais-store.js` owns record ingestion, +static metadata and recent-track storage. Neither area imports globe rendering. +`common/http.js` owns capped reads/coalescing; `common/query.js` owns query values. + +These plugins retain their existing process-scoped caches and server lifetime. +Importing the entry does not start acquisition. The AIS plugin disposes its +socket/watchdog on server close and re-reads configuration after restart. +The portable `gods-eye-view/sources/adsb-lol` export normalizes existing aircraft +records without importing Node middleware or a renderer. Browser layer/controller +separation is outside this server extraction. diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index bca9a1d79..99f83d62c 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -174,6 +174,23 @@ Launch payloads with missing records now say PAYLOAD DATA UNAVAILABLE. Missing n Updated: August 24, 2026 +## Aircraft and vessel server modules + +Standalone aircraft routes now live in `server/providers/aircraft/`: OpenSky +state vectors and regional fallback, adsb.lol military positions, ADSBDB +enrichment and track backfill. Vessel routes and websocket/watchdog setup live +in `server/providers/vessels/ais-live.js`; AIS records and recent tracks live +in `ais-store.js`. Common response caps, request coalescing and query parsing +have their own modules. `server/providers/local.js` composes these with the +remaining providers and retains existing named compatibility exports. + +`gods-eye-view/server/providers/live` is a Node-only entry for the existing +plugins and shared request helpers. Importing it starts no sockets or timers. +The existing aircraft normalizer is separately available through the portable +`gods-eye-view/sources/adsb-lol` export. Provider URLs, local credentials, cache +policy, fallback behavior, response shapes and rendering remain unchanged. + + ## Control names for assistive technology Scope, Bloom, Sharpen and location search have explicit accessible names. diff --git a/package.json b/package.json index 8f47f8646..c8b6c6e87 100644 --- a/package.json +++ b/package.json @@ -69,6 +69,10 @@ "./application/viewer": "./src/app/viewer.js", "./build/vite": { "node": "./build/vite.js" - } + }, + "./server/providers/live": { + "node": "./server/providers/live.js" + }, + "./sources/adsb-lol": "./src/data/adsbLolFallback.js" } } diff --git a/scripts/package-boundaries.json b/scripts/package-boundaries.json index cda46b637..7727021dc 100644 --- a/scripts/package-boundaries.json +++ b/scripts/package-boundaries.json @@ -10,22 +10,72 @@ "src/data/localGeojsonCore.js", "src/data/localGeojsonLod.js" ], - "external": ["cesium"] + "external": [ + "cesium" + ] }, "application": { - "exports": ["./application"], - "modules": ["src/app/application.js"], + "exports": [ + "./application" + ], + "modules": [ + "src/app/application.js" + ], "external": [] }, "viewer": { - "exports": ["./application/viewer"], - "modules": ["src/app/viewer.js"], - "external": ["cesium"] + "exports": [ + "./application/viewer" + ], + "modules": [ + "src/app/viewer.js" + ], + "external": [ + "cesium" + ] }, "vite-build": { "runtime": "node", - "exports": ["./build/vite"], - "modules": ["build/vite.js"], - "external": ["vite-plugin-cesium"] + "exports": [ + "./build/vite" + ], + "modules": [ + "build/vite.js" + ], + "external": [ + "vite-plugin-cesium" + ] + }, + "live-providers": { + "runtime": "node", + "exports": [ + "./server/providers/live" + ], + "modules": [ + "server/providers/live.js", + "server/providers/common/http.js", + "server/providers/common/query.js", + "server/providers/aircraft/opensky.js", + "server/providers/aircraft/adsb-lol.js", + "server/providers/aircraft/enrichment.js", + "server/providers/aircraft/tracks.js", + "server/providers/vessels/ais-store.js", + "server/providers/vessels/ais-live.js", + "src/data/adsbLolFallback.js", + "src/data/aisStreamAdapter.js", + "src/data/aisWatchdog.js" + ], + "external": [ + "ws" + ] + }, + "adsb-lol-source": { + "exports": [ + "./sources/adsb-lol" + ], + "modules": [ + "src/data/adsbLolFallback.js" + ], + "external": [] } } diff --git a/server/providers/aircraft/adsb-lol.js b/server/providers/aircraft/adsb-lol.js new file mode 100644 index 000000000..e8ce60b0c --- /dev/null +++ b/server/providers/aircraft/adsb-lol.js @@ -0,0 +1,50 @@ +/** + * Vite plugin: adsb.lol military aircraft proxy with 12 s response cache. + * + * Proxies GET /api/adsblol/mil to https://api.adsb.lol/v2/mil. On upstream + * failure, serves a stale cached response if one exists. + * + * @returns {import('vite').Plugin} + */ +export function adsbLolProxy() { + /** @type {string|null} Cached upstream JSON body. */ + let _cache = null; + /** @type {number} Epoch-ms when the cache was populated. */ + let _cacheAt = 0; + /** Response cache TTL (ms). */ + const CACHE_MS = 12000; + return { + name: 'adsblol-proxy', + configureServer(server) { + server.middlewares.use('/api/adsblol/mil', async (req, res) => { + try { + const now = Date.now(); + if (_cache && now - _cacheAt < CACHE_MS) { + res.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'X-ADS-B-Cache': 'HIT' }); + res.end(_cache); + return; + } + const upstream = await fetch('https://api.adsb.lol/v2/mil', { + headers: { 'User-Agent': 'gods-eye-view-adsblol-proxy/1.0' }, + }); + const body = await upstream.text(); + if (upstream.ok) { + _cache = body; + _cacheAt = now; + } + res.writeHead(upstream.status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'X-ADS-B-Cache': 'MISS' }); + res.end(body); + } catch (e) { + console.error('[adsb.lol Proxy]', e.message); + if (_cache) { + res.writeHead(200, { 'Content-Type': 'application/json', 'X-ADS-B-Cache': 'STALE' }); + res.end(_cache); + return; + } + res.writeHead(502, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: 'ADS-B proxy error' })); + } + }); + }, + }; +} diff --git a/server/providers/aircraft/enrichment.js b/server/providers/aircraft/enrichment.js new file mode 100644 index 000000000..c07c4204f --- /dev/null +++ b/server/providers/aircraft/enrichment.js @@ -0,0 +1,123 @@ +import path from 'node:path'; +import { promises as fsp } from 'node:fs'; +/** + * adsbdb.com enrichment proxy: callsign → route (airline + origin/destination + * airports) and hex → aircraft type/registration. Free community API — cached + * aggressively: ONE upstream request per new key ever (404s negative-cached), + * persisted to disk so restarts don't re-hammer it. Adapted from skylight + * (MIT) server/src/enrich/routes.ts. + */ +export function adsbdbProxy() { + const TTL_MS = 24 * 3600_000; + const CACHE_PATH = path.join(process.cwd(), '.gev-cache', 'adsbdb.json'); + let cache = { routes: {}, aircraft: {} }; + let dirty = false; + let loaded = false; + const inflight = new Map(); + + async function loadOnce() { + if (loaded) return; + loaded = true; + try { + const parsed = JSON.parse(await fsp.readFile(CACHE_PATH, 'utf8')); + cache = { routes: parsed.routes ?? {}, aircraft: parsed.aircraft ?? {} }; + } catch { /* first run */ } + setInterval(async () => { + if (!dirty) return; + dirty = false; + try { + await fsp.mkdir(path.dirname(CACHE_PATH), { recursive: true }); + await fsp.writeFile(CACHE_PATH, JSON.stringify(cache), 'utf8'); + } catch { dirty = true; } // retry next tick + }, 15_000).unref?.(); + } + + const fresh = (e) => e && Date.now() - e.at < TTL_MS; + + function parseRoute(json) { + const fr = json?.response?.flightroute; + if (!fr?.origin || !fr?.destination) return null; + const airport = (a) => ({ + code: a.iata_code || a.icao_code || '', + name: a.municipality || a.name || '', + lat: Number.isFinite(a.latitude) ? a.latitude : null, + lon: Number.isFinite(a.longitude) ? a.longitude : null, + }); + return { airline: fr.airline?.name || null, origin: airport(fr.origin), destination: airport(fr.destination) }; + } + + function parseAircraft(json) { + const a = json?.response?.aircraft; + if (!a) return null; + return { + typeCode: a.icao_type || null, // ICAO designator, e.g. "B738" — feeds classifyAircraft + typeName: a.manufacturer && a.type ? `${a.manufacturer} ${a.type}` : (a.type || null), + registration: a.registration || null, + }; + } + + function lookup(kind, key) { + const store = kind === 'route' ? cache.routes : cache.aircraft; + if (fresh(store[key])) return Promise.resolve(store[key].data); + const ik = `${kind}:${key}`; + if (!inflight.has(ik)) { + inflight.set(ik, (async () => { + try { + const url = kind === 'route' + ? `https://api.adsbdb.com/v0/callsign/${encodeURIComponent(key)}` + : `https://api.adsbdb.com/v0/aircraft/${encodeURIComponent(key)}`; + const res = await fetch(url, { signal: AbortSignal.timeout(8000) }); + if (res.ok) { + const data = kind === 'route' ? parseRoute(await res.json()) : parseAircraft(await res.json()); + store[key] = { at: Date.now(), data }; // data may be null — negative cache + dirty = true; + return data; + } + if (res.status === 404) { + store[key] = { at: Date.now(), data: null }; // known-missing — cache the miss + dirty = true; + } + // other statuses: leave uncached so we retry later + return fresh(store[key]) ? store[key].data : null; + } catch { + return fresh(store[key]) ? store[key].data : null; // network error → stale if any + } finally { + inflight.delete(ik); + } + })()); + } + return inflight.get(ik); + } + + return { + name: 'adsbdb-proxy', + configureServer(server) { + server.middlewares.use('/api/adsbdb', async (req, res) => { + await loadOnce(); + const send = (status, obj) => { + res.writeHead(status, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify(obj)); + }; + try { + const [, kind, rawKey] = String(req.url || '').split('?')[0].split('/'); + if (kind === 'route') { + const cs = String(rawKey || '').toUpperCase(); + if (!/^[A-Z0-9]{2,8}$/.test(cs)) return send(400, { error: 'invalid callsign' }); + const data = await lookup('route', cs); + return send(200, data ? { found: true, ...data } : { found: false }); + } + if (kind === 'type') { + const hex = String(rawKey || '').toLowerCase(); + if (!/^[0-9a-f]{6}$/.test(hex)) return send(400, { error: 'invalid hex' }); + const data = await lookup('aircraft', hex); + return send(200, data ? { found: true, ...data } : { found: false }); + } + return send(404, { error: 'unknown endpoint' }); + } catch (err) { + console.error('[adsbdb-proxy] request failed'); + return send(500, { error: 'adsbdb proxy error' }); + } + }); + }, + }; +} diff --git a/server/providers/aircraft/opensky.js b/server/providers/aircraft/opensky.js new file mode 100644 index 000000000..c3ec896f1 --- /dev/null +++ b/server/providers/aircraft/opensky.js @@ -0,0 +1,596 @@ +import { normalizeAdsbLolPointResponse } from '../../../src/data/adsbLolFallback.js'; +import { coalesceProxyRequest, readResponseJsonCapped } from '../common/http.js'; +import { requiredFiniteQueryNumber } from '../common/query.js'; +// --------------------------------------------------------------------------- +// OpenSky OAuth2 token + response cache state +// --------------------------------------------------------------------------- +/** @type {string|null} Current OAuth2 bearer token. */ +let _openskyToken = null; +/** @type {number} Epoch-ms when the current token expires. */ +let _openskyTokenExpiry = 0; +/** @type {Promise|null} In-flight token refresh promise (coalesces concurrent callers). */ +let _openskyTokenPromise = null; +/** @type {string|null} Cached upstream response body (JSON text). */ +let _openskyCacheBody = null; +/** @type {number} HTTP status of the cached response. */ +let _openskyCacheStatus = 0; +/** @type {number} Epoch-ms when the response was cached. */ +let _openskyCacheTime = 0; +/** @type {{requestedMode:string,usedMode:string,reason:string}|null} Auth metadata for the cached response. */ +let _openskyCacheMeta = null; +/** @type {number|null} Source snapshot epoch from the cached OpenSky body. */ +let _openskyCacheSourceEpochMs = null; +/** TTL for the OpenSky response cache (ms). */ +const OPENSKY_CACHE_MS = 9000; +// --- OpenSky credit governor (field-test fix 2026-07-06) ------------------- +// The global /states/all this proxy fetches costs 4 CREDITS per call against +// OpenSky's ~4000/day authenticated budget — a day with the app open burned +// the whole quota in ~8h and the layer then hard-died until the daily reset +// ("rate limited for 48h" owner report; auth itself was fine). Three levers: +// 1. Adaptive TTL: OpenSky returns X-Rate-Limit-Remaining on success; as the +// budget thins, the proxy stretches its cache TTL so a full day of +// continuous use never exhausts it. +// 2. 429 cooldown: honor X-Rate-Limit-Retry-After-Seconds — no upstream +// attempts until it passes (bounded 30 s … 30 min). +// 3. Serve-stale: while rate-limited/cooling, serve the last-good body (200 + +// X-OpenSky-Stale) so the layer keeps rendering instead of dying. +/** @type {number} Current adaptive TTL (ms) — starts at the base cache TTL. */ +let _openskyTtlMs = OPENSKY_CACHE_MS; +/** @type {number} Epoch-ms before which no upstream fetch is attempted. */ +let _openskyCooldownUntil = 0; +/** + * 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 + * later tiers stretch the day: >2400 → ~3 h of full freshness, then 30 s + * (~2.5 h), 90 s (~5 h), 300 s (~8 h) ≈ 18+ h of continuous use per day. + * @param {number} remaining - X-Rate-Limit-Remaining header value. + * @returns {number} TTL in ms. + */ +function openskyAdaptiveTtlMs(remaining) { + if (!Number.isFinite(remaining)) return OPENSKY_CACHE_MS; + if (remaining > 2400) return OPENSKY_CACHE_MS; + if (remaining > 1200) return 30_000; + if (remaining > 400) return 90_000; + return 300_000; +} +/** @type {boolean} Guards duplicate auth-failure warnings in logs. */ +let _openskyAuthWarned = false; +/** @type {boolean} Guards duplicate invalid-auth-mode warnings. */ +let _openskyAuthModeWarned = false; +/** Default auth mode when OPENSKY_AUTH_MODE env is unset. */ +const OPENSKY_AUTH_MODE_DEFAULT = 'oauth'; +/** Set of valid OPENSKY_AUTH_MODE values. */ +const OPENSKY_AUTH_MODE_SET = new Set(['basic', 'oauth', 'auto', 'anon']); +/** Regional civilian fallback cache, keyed by a coarse 0.25° view anchor. */ +const _adsbLolPointCache = new Map(); +/** Per-anchor single-flight map for concurrent regional fallback requests. */ +const _adsbLolPointInFlight = new Map(); +const ADSBLOL_POINT_CACHE_MS = 12000; +const ADSBLOL_POINT_CACHE_MAX = 80; +const ADSBLOL_POINT_RADIUS_NM = 250; +const ADSBLOL_POINT_MAX_RESPONSE_BYTES = 8 * 1024 * 1024; +// A 200 response can still contain an old OpenSky snapshot. Past this point +// the viewport-scoped adsb.lol source is more honest and keeps local motion +// current instead of coasting a stale worldwide frame indefinitely. +const OPENSKY_SOURCE_STALE_MS = 120_000; + +/** + * Obtain a valid OpenSky OAuth2 bearer token, refreshing if needed. + * + * Uses the client_credentials grant against the OpenSky Keycloak realm. + * Concurrent callers share a single in-flight refresh promise so only + * one token request is issued at a time. + * + * @returns {Promise} Bearer token string, or null if unavailable. + */ +export async function getOpenSkyToken() { + const now = Date.now(); + // Return cached token if still valid (with 60 s safety margin) + if (_openskyToken && now < _openskyTokenExpiry - 60000) return _openskyToken; + + // Coalesce concurrent refresh requests — if a refresh is already in-flight, + // return the same promise instead of issuing a duplicate token request + if (_openskyTokenPromise) return _openskyTokenPromise; + + const clientId = process.env.OPENSKY_CLIENT_ID; + const clientSecret = process.env.OPENSKY_CLIENT_SECRET; + if (!clientId || !clientSecret) return null; + + // Wrap the async token fetch in a shared promise stored in _openskyTokenPromise + _openskyTokenPromise = (async () => { + try { + const res = await fetch( + 'https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token', + { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: `grant_type=client_credentials&client_id=${encodeURIComponent(clientId)}&client_secret=${encodeURIComponent(clientSecret)}`, + } + ); + + let data = null; + try { + data = await res.json(); + } catch { + data = null; + } + + const accessToken = data?.access_token; + const expiresIn = Number(data?.expires_in); + if (!res.ok || !accessToken) { + if (!_openskyAuthWarned) { + const detail = data?.error_description || data?.error || `HTTP ${res.status}`; + console.warn('[OpenSky] OAuth client_credentials failed:', detail); + _openskyAuthWarned = true; + } + _openskyToken = null; + _openskyTokenExpiry = 0; + return null; + } + + _openskyToken = accessToken; + // Default to 1800 s (30 min) if expires_in is missing or non-finite + _openskyTokenExpiry = Date.now() + (Number.isFinite(expiresIn) ? expiresIn : 1800) * 1000; + console.log('[OpenSky] OAuth token refreshed, expires in', Number.isFinite(expiresIn) ? expiresIn : 1800, 's'); + _openskyAuthWarned = false; + return _openskyToken; + } catch (err) { + if (!_openskyAuthWarned) { + console.warn('[OpenSky] OAuth token request failed:', err?.message || String(err)); + _openskyAuthWarned = true; + } + _openskyToken = null; + _openskyTokenExpiry = 0; + return null; + } finally { + // Clear the shared promise so the next caller can start a fresh refresh + _openskyTokenPromise = null; + } + })(); + + return _openskyTokenPromise; +} + +/** + * Validate and normalize the OPENSKY_AUTH_MODE env value. + * + * @param {string} value - Raw env value (e.g. 'basic', 'oauth', 'auto', 'anon'). + * @returns {string} One of the valid mode strings, or the default ('oauth'). + */ +function normalizeOpenSkyAuthMode(value) { + const raw = String(value || '').trim().toLowerCase(); + if (!raw) return OPENSKY_AUTH_MODE_DEFAULT; + if (OPENSKY_AUTH_MODE_SET.has(raw)) return raw; + if (!_openskyAuthModeWarned) { + console.warn( + `[OpenSky] Invalid OPENSKY_AUTH_MODE="${raw}", defaulting to "${OPENSKY_AUTH_MODE_DEFAULT}"` + ); + _openskyAuthModeWarned = true; + } + return OPENSKY_AUTH_MODE_DEFAULT; +} + +/** + * Build standard response headers for OpenSky proxy responses. + * + * Includes diagnostic X-OpenSky-* headers so the client can inspect + * cache hit/miss status and which auth mode was actually used. + * + * @param {object} opts + * @param {string} opts.cacheStatus - 'HIT', 'MISS', or 'STALE'. + * @param {string} opts.requestedMode - The auth mode the config requested. + * @param {string} opts.usedMode - The auth mode actually used for the upstream call. + * @param {string} opts.reason - Human-readable reason string for diagnostics. + * @returns {Record} Header object. + */ +function buildOpenSkyHeaders({ cacheStatus, requestedMode, usedMode, reason, staleSeconds, retryAfterSeconds }) { + const headers = { + 'Content-Type': 'application/json', + 'Cache-Control': 'no-store', + 'X-OpenSky-Cache': cacheStatus, + 'X-OpenSky-Auth': usedMode, + 'X-OpenSky-Auth-Mode-Requested': requestedMode, + 'X-OpenSky-Auth-Mode-Used': usedMode, + 'X-OpenSky-Auth-Reason': reason, + }; + // Credit-governor extras (field-test fix 2026-07-06): the client can show a + // STALE cue / countdown without parsing the body. + if (Number.isFinite(staleSeconds)) headers['X-OpenSky-Stale-Seconds'] = String(Math.round(staleSeconds)); + if (Number.isFinite(retryAfterSeconds)) headers['X-OpenSky-Retry-After-Seconds'] = String(Math.round(retryAfterSeconds)); + return headers; +} + +export function adsbLolFallbackAnchor(req) { + const incoming = new URL(req?.url || '', 'http://localhost'); + const latitude = requiredFiniteQueryNumber(incoming.searchParams, 'lat'); + const longitude = requiredFiniteQueryNumber(incoming.searchParams, 'lon'); + if (!Number.isFinite(latitude) || latitude < -90 || latitude > 90) return null; + if (!Number.isFinite(longitude) || longitude < -180 || longitude > 180) return null; + return { latitude, longitude }; +} + +async function fetchAdsbLolPointFallback(req) { + const anchor = adsbLolFallbackAnchor(req); + if (!anchor) return null; + const roundedLat = Math.round(anchor.latitude * 4) / 4; + const roundedLon = Math.round(anchor.longitude * 4) / 4; + const cacheKey = `${roundedLat.toFixed(2)},${roundedLon.toFixed(2)}`; + const cached = _adsbLolPointCache.get(cacheKey); + const now = Date.now(); + if (cached && now - cached.cachedAt < ADSBLOL_POINT_CACHE_MS) { + return { ...cached, cacheStatus: 'HIT' }; + } + + const request = coalesceProxyRequest(_adsbLolPointInFlight, cacheKey, async () => { + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), 10000); + try { + const upstream = await fetch( + `https://api.adsb.lol/v2/lat/${roundedLat}/lon/${roundedLon}/dist/${ADSBLOL_POINT_RADIUS_NM}`, + { + headers: { + Accept: 'application/json', + 'User-Agent': 'gods-eye-view-adsblol-regional-fallback/1.0', + }, + signal: controller.signal, + }, + ); + if (!upstream.ok) throw new Error(`upstream HTTP ${upstream.status}`); + const payload = await readResponseJsonCapped(upstream, ADSBLOL_POINT_MAX_RESPONSE_BYTES); + const normalized = normalizeAdsbLolPointResponse(payload); + const record = { + body: JSON.stringify(normalized), + cachedAt: Date.now(), + count: normalized.states.length, + }; + _adsbLolPointCache.delete(cacheKey); + _adsbLolPointCache.set(cacheKey, record); + while (_adsbLolPointCache.size > ADSBLOL_POINT_CACHE_MAX) { + _adsbLolPointCache.delete(_adsbLolPointCache.keys().next().value); + } + return record; + } finally { + clearTimeout(timeoutId); + } + }); + try { + const record = await request.promise; + return { ...record, cacheStatus: request.shared ? 'INFLIGHT' : 'MISS' }; + } catch (error) { + if (!request.shared && error?.name !== 'AbortError') { + console.warn('[adsb.lol Flights Fallback]', error?.message || error); + } + return cached ? { ...cached, cacheStatus: 'STALE' } : null; + } +} + +async function serveAdsbLolPointFallback(req, res, requestedMode, reason) { + const fallback = await fetchAdsbLolPointFallback(req); + if (!fallback) return false; + res.writeHead(200, { + ...buildOpenSkyHeaders({ + cacheStatus: fallback.cacheStatus, + requestedMode, + usedMode: 'adsblol-regional', + reason, + }), + 'X-Flight-Source': 'adsb.lol', + 'X-Flight-Coverage': `${ADSBLOL_POINT_RADIUS_NM}nm regional fallback`, + 'X-Flight-Count': String(fallback.count), + }); + res.end(fallback.body); + return true; +} + +function openSkySourceEpochMs(body) { + try { + const seconds = Number(JSON.parse(body)?.time); + return Number.isFinite(seconds) && seconds > 0 ? seconds * 1000 : null; + } catch { + return null; + } +} + +function openSkySourceIsStale(sourceEpochMs, now = Date.now()) { + return Number.isFinite(sourceEpochMs) + && now - sourceEpochMs > OPENSKY_SOURCE_STALE_MS; +} + +/** + * Vite plugin: OpenSky Network proxy with multi-mode auth and response caching. + * + * Supports four auth modes controlled by OPENSKY_AUTH_MODE env: + * - 'oauth' (default) — client_credentials bearer token + * - 'basic' — HTTP Basic with OPENSKY_USERNAME / OPENSKY_PASSWORD + * - 'auto' — try OAuth first, fall back to Basic, then anon + * - 'anon' — no credentials + * + * Successful responses are cached for OPENSKY_CACHE_MS (~9 s). On + * upstream failure the proxy serves a stale cached response if available, or + * a bounded 250 nm adsb.lol point snapshot around the current view anchor. + * + * @returns {import('vite').Plugin} + */ +export function openSkyProxy() { + return { + name: 'opensky-proxy', + configureServer(server) { + server.middlewares.use('/api/opensky', async (req, res) => { + try { + const requestedMode = normalizeOpenSkyAuthMode(process.env.OPENSKY_AUTH_MODE); + const now = Date.now(); + const inCooldown = now < _openskyCooldownUntil; + // Fresh-enough cache (adaptive TTL) OR any cache during a 429 + // cooldown: serve it without touching upstream. Stale-during-cooldown + // is deliberate (credit governor): last-good planes beat a dead layer. + if (_openskyCacheBody && (now - _openskyCacheTime < _openskyTtlMs || inCooldown)) { + if ( + openSkySourceIsStale(_openskyCacheSourceEpochMs, now) + && await serveAdsbLolPointFallback( + req, + res, + requestedMode, + 'opensky_snapshot_stale_regional_fallback', + ) + ) { + return; + } + const cachedMeta = _openskyCacheMeta || { + requestedMode, + usedMode: 'unknown', + reason: 'cached', + }; + const isStale = now - _openskyCacheTime >= _openskyTtlMs; + res.writeHead( + _openskyCacheStatus || 200, + buildOpenSkyHeaders({ + cacheStatus: isStale ? 'STALE' : 'HIT', + requestedMode: cachedMeta.requestedMode || requestedMode, + usedMode: cachedMeta.usedMode || 'unknown', + reason: isStale ? 'rate_limited_serving_stale' : (cachedMeta.reason || 'cached'), + staleSeconds: isStale ? (now - _openskyCacheTime) / 1000 : undefined, + retryAfterSeconds: inCooldown ? (_openskyCooldownUntil - now) / 1000 : undefined, + }) + ); + res.end(_openskyCacheBody); + return; + } + // Cooling down with nothing cached (cold start into a rate limit): + // synthesize the 429 locally — hammering upstream mid-cooldown can't + // succeed and just burns goodwill. + if (inCooldown) { + if (await serveAdsbLolPointFallback(req, res, requestedMode, 'opensky_cooldown_regional_fallback')) return; + res.writeHead(429, buildOpenSkyHeaders({ + cacheStatus: 'COOLDOWN', + requestedMode, + usedMode: 'none', + reason: 'rate_limited', + retryAfterSeconds: (_openskyCooldownUntil - now) / 1000, + })); + res.end(JSON.stringify({ error: 'OpenSky rate limited; 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'; + + 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 = await fetch('https://opensky-network.org/api/states/all?extended=1', { 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 = await fetch('https://opensky-network.org/api/states/all?extended=1', { headers: retryHeaders }); + usedMode = 'basic'; + reason = 'oauth_rejected_fallback_basic'; + } + + let body = await upstream.text(); + const sourceEpochMs = upstream.ok ? openSkySourceEpochMs(body) : null; + if ( + upstream.ok + && openSkySourceIsStale(sourceEpochMs, now) + && await serveAdsbLolPointFallback( + req, + res, + requestedMode, + '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. + _openskyCacheBody = body; + _openskyCacheStatus = upstream.status; + _openskyCacheTime = now; + _openskyCacheSourceEpochMs = sourceEpochMs; + _openskyCacheMeta = { requestedMode, usedMode, reason }; + return; + } + if (upstream.status === 429) { + reason = 'rate_limited'; + // Credit governor: honor OpenSky's retry-after (bounded 30 s … 30 min; + // 2 min when the header is absent) — no upstream attempts until then. + const retryAfterSec = Number(upstream.headers.get('x-rate-limit-retry-after-seconds')); + const cooldownMs = Math.min( + Math.max(Number.isFinite(retryAfterSec) ? retryAfterSec * 1000 : 120_000, 30_000), + 30 * 60_000 + ); + _openskyCooldownUntil = now + cooldownMs; + // 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) { + res.writeHead(200, buildOpenSkyHeaders({ + cacheStatus: 'STALE', + requestedMode, + usedMode, + reason: 'rate_limited_serving_stale', + staleSeconds: (now - _openskyCacheTime) / 1000, + retryAfterSeconds: cooldownMs / 1000, + })); + res.end(_openskyCacheBody); + return; + } + } + + if (!upstream.ok && !_openskyCacheBody) { + const servedFallback = await serveAdsbLolPointFallback( + req, + res, + requestedMode, + `opensky_http_${upstream.status}_regional_fallback`, + ); + if (servedFallback) return; + } + + if (upstream.status === 401 || upstream.status === 403) { + if (requestedMode === 'basic' && !hasBasicCreds) { + body = JSON.stringify({ + error: 'OpenSky auth missing. Basic mode requires OPENSKY_USERNAME and OPENSKY_PASSWORD.', + }); + reason = 'missing_basic_creds'; + } else if (requestedMode === 'oauth' && usedMode !== 'oauth') { + body = JSON.stringify({ + error: 'OpenSky auth invalid. OAuth mode requires valid OPENSKY_CLIENT_ID and OPENSKY_CLIENT_SECRET.', + }); + reason = 'oauth_invalid_or_missing'; + } else if (usedMode === 'basic') { + body = JSON.stringify({ + error: 'OpenSky auth invalid. Username/password were rejected.', + }); + reason = 'basic_invalid_credentials'; + } else if (usedMode === 'oauth') { + body = JSON.stringify({ + error: 'OpenSky auth invalid. OAuth client credentials were rejected.', + }); + reason = 'oauth_invalid_credentials'; + } else if (requestedMode === 'auto' && !hasBasicCreds) { + body = JSON.stringify({ + error: 'OpenSky auth missing. Provide basic credentials or valid OAuth client credentials.', + }); + reason = 'missing_oauth_and_basic_creds'; + } else { + body = JSON.stringify({ + error: 'OpenSky auth required.', + }); + reason = 'auth_required'; + } + } + + // 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) { + _openskyCacheBody = body; + _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; + } + + res.writeHead( + upstream.status, + buildOpenSkyHeaders({ + cacheStatus: 'MISS', + requestedMode, + usedMode, + reason, + }) + ); + res.end(body); + } catch (e) { + console.error('[OpenSky Proxy]', e.message); + if (_openskyCacheBody) { + const cachedMeta = _openskyCacheMeta || { + requestedMode: normalizeOpenSkyAuthMode(process.env.OPENSKY_AUTH_MODE), + usedMode: 'unknown', + reason: 'cached_stale', + }; + res.writeHead( + _openskyCacheStatus || 200, + buildOpenSkyHeaders({ + cacheStatus: 'STALE', + requestedMode: cachedMeta.requestedMode || OPENSKY_AUTH_MODE_DEFAULT, + usedMode: cachedMeta.usedMode || 'unknown', + reason: cachedMeta.reason || 'cached_stale', + }) + ); + res.end(_openskyCacheBody); + return; + } + const requestedMode = normalizeOpenSkyAuthMode(process.env.OPENSKY_AUTH_MODE); + if (await serveAdsbLolPointFallback(req, res, requestedMode, 'opensky_proxy_error_regional_fallback')) return; + res.writeHead( + 502, + buildOpenSkyHeaders({ + cacheStatus: 'MISS', + requestedMode, + usedMode: 'error', + reason: 'proxy_error', + }) + ); + res.end(JSON.stringify({ error: 'OpenSky proxy error' })); + } + }); + }, + }; +} diff --git a/server/providers/aircraft/tracks.js b/server/providers/aircraft/tracks.js new file mode 100644 index 000000000..2c11b278f --- /dev/null +++ b/server/providers/aircraft/tracks.js @@ -0,0 +1,114 @@ +import { getOpenSkyToken } from './opensky.js'; +import { readCappedResponseText } from '../common/http.js'; +/** + * Vite plugin: aircraft track-history backfill proxies (PRD WS-F F1/F2). + * + * /api/opensky-track?icao24= — OpenSky GET /tracks/all (experimental; + * own credit bucket, 4 credits per call on the free tier). OAuth via the + * shared coalesced token. 60s per-icao cache; 404/429 forwarded so the + * client can fall back to its accumulated trail silently. + * /api/adsblol/trace?hex= — adsb.lol tar1090 readsb trace + * (undocumented but live; no browser CORS, hence this proxy). Up to ~24h + * of real history per aircraft. Treat as best-effort; data is ODbL — + * credit "adsb.lol (ODbL)" in the UI. + */ +export function trackBackfillProxies() { + const TRACK_CACHE_MS = 60000; + const TRACK_CACHE_MAX = 200; + const RESPONSE_CAP_BYTES = 5 * 1024 * 1024; + /** @type {Map} */ + const cache = new Map(); + + function cachePut(key, entry) { + cache.set(key, entry); + if (cache.size > TRACK_CACHE_MAX) { + const oldest = [...cache.entries()].sort((a, b) => a[1].at - b[1].at)[0]; + if (oldest) cache.delete(oldest[0]); + } + } + + async function proxyJson(res, key, upstreamUrl, headers = {}) { + const cached = cache.get(key); + if (cached && Date.now() - cached.at < TRACK_CACHE_MS) { + res.statusCode = cached.status; + res.setHeader('Content-Type', 'application/json; charset=utf-8'); + res.setHeader('Cache-Control', 'no-store'); + res.end(cached.body); + return; + } + const upstream = await fetch(upstreamUrl, { headers, signal: AbortSignal.timeout(12000) }); + const { tooLarge, text } = await readCappedResponseText(upstream, RESPONSE_CAP_BYTES); + let body; + if (tooLarge) { + body = JSON.stringify({ error: 'Upstream track response too large' }); + } else if (!upstream.ok) { + // Sanitize upstream error surface; status code is signal enough + body = JSON.stringify({ error: `Track source HTTP ${upstream.status}` }); + } else { + body = text; + } + cachePut(key, { at: Date.now(), status: upstream.status, body }); + res.statusCode = upstream.status; + res.setHeader('Content-Type', 'application/json; charset=utf-8'); + res.setHeader('Cache-Control', 'no-store'); + res.end(body); + } + + function install(middlewares) { + middlewares.use('/api/opensky-track', async (req, res) => { + try { + const incoming = new URL(req.url || '', 'http://localhost'); + const icao24 = String(incoming.searchParams.get('icao24') || '').trim().toLowerCase(); + if (!/^[0-9a-f]{6}$/.test(icao24)) { + res.statusCode = 400; + res.setHeader('Content-Type', 'application/json'); + res.end(JSON.stringify({ error: 'icao24 must be a 6-char hex string' })); + return; + } + const token = await getOpenSkyToken(); + await proxyJson( + res, + `osky:${icao24}`, + `https://opensky-network.org/api/tracks/all?icao24=${icao24}&time=0`, + token ? { Authorization: `Bearer ${token}` } : {} + ); + } catch (error) { + res.statusCode = 502; + res.setHeader('Content-Type', 'application/json'); + res.end(JSON.stringify({ error: 'OpenSky track fetch failed' })); + } + }); + + middlewares.use('/api/adsblol/trace', async (req, res) => { + try { + const incoming = new URL(req.url || '', 'http://localhost'); + const hex = String(incoming.searchParams.get('hex') || '').trim().toLowerCase(); + if (!/^[0-9a-f~]{6,7}$/.test(hex)) { + res.statusCode = 400; + res.setHeader('Content-Type', 'application/json'); + res.end(JSON.stringify({ error: 'hex must be a 6-7 char hex string' })); + return; + } + await proxyJson( + res, + `lol:${hex}`, + `https://adsb.lol/data/traces/${hex.slice(-2)}/trace_full_${hex}.json` + ); + } catch (error) { + res.statusCode = 502; + res.setHeader('Content-Type', 'application/json'); + res.end(JSON.stringify({ error: 'adsb.lol trace fetch failed' })); + } + }); + } + + return { + name: 'track-backfill-proxies', + configureServer(server) { + install(server.middlewares); + }, + configurePreviewServer(server) { + install(server.middlewares); + }, + }; +} diff --git a/server/providers/common/http.js b/server/providers/common/http.js new file mode 100644 index 000000000..361ce928c --- /dev/null +++ b/server/providers/common/http.js @@ -0,0 +1,108 @@ +/** + * Read a fetch() Response body as text with a hard byte cap. Rejects early on an + * oversized Content-Length, then streams with a running cap so a chunked or + * length-omitted response cannot blow past the limit. Throws { code:'RESPONSE_TOO_LARGE' }. + */ +export async function readResponseTextCapped(response, maxBytes) { + const declared = Number(response.headers.get('content-length')); + if (Number.isFinite(declared) && declared > maxBytes) { + const err = new Error('Upstream response too large'); + err.code = 'RESPONSE_TOO_LARGE'; + throw err; + } + const reader = response.body?.getReader?.(); + if (!reader) { + const text = await response.text(); + if (Buffer.byteLength(text) > maxBytes) { + const err = new Error('Upstream response too large'); + err.code = 'RESPONSE_TOO_LARGE'; + throw err; + } + return text; + } + const decoder = new TextDecoder(); + let out = ''; + let total = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > maxBytes) { + try { await reader.cancel(); } catch { /* no-op */ } + const err = new Error('Upstream response too large'); + err.code = 'RESPONSE_TOO_LARGE'; + throw err; + } + out += decoder.decode(value, { stream: true }); + } + out += decoder.decode(); + return out; +} + +/** Parse a fetch() JSON response only after enforcing a hard byte cap. */ +export async function readResponseJsonCapped(response, maxBytes) { + return JSON.parse(await readResponseTextCapped(response, maxBytes)); +} + +/** + * Return the existing promise for a cache key, or create one and remove it + * only when that exact promise settles. + */ +export function coalesceProxyRequest(inFlight, key, create) { + const existing = inFlight.get(key); + if (existing) return { promise: existing, shared: true }; + let promise; + promise = Promise.resolve() + .then(create) + .finally(() => { + if (inFlight.get(key) === promise) inFlight.delete(key); + }); + inFlight.set(key, promise); + return { promise, shared: false }; +} + +/** + * Pipe an upstream fetch Response (image or video) to the client HTTP response. + * + * Forwards Content-Type, Content-Length, Content-Range, Accept-Ranges, and + * Cache-Control headers from the upstream. Falls back to buffered arrayBuffer + * if the body is not streamable. + * + * @param {import('http').ServerResponse} res + * @param {Response} upstream - fetch() Response object. + * @param {object} [opts] + * @param {string} [opts.sourceHeader='upstream'] - Value for X-CCTV-Source header. + */ +/** + * Read a fetch Response body as text while enforcing a hard byte cap during + * the read — so a malicious or buggy upstream that streams an unbounded body + * (no/oversized Content-Length, chunked) can't OOM the proxy. Returns + * { tooLarge, text }. Cancels the stream as soon as the cap is crossed. + * @param {Response} upstream - fetch() response. + * @param {number} maxBytes - hard ceiling on decoded bytes. + * @returns {Promise<{tooLarge: boolean, text: string}>} + */ +export async function readCappedResponseText(upstream, maxBytes) { + const declared = Number(upstream.headers.get('content-length')); + if (Number.isFinite(declared) && declared > maxBytes) { + try { await upstream.body?.cancel(); } catch { /* no-op */ } + return { tooLarge: true, text: '' }; + } + if (!upstream.body || typeof upstream.body[Symbol.asyncIterator] !== 'function') { + const text = await upstream.text(); + return text.length > maxBytes ? { tooLarge: true, text: '' } : { tooLarge: false, text }; + } + const decoder = new TextDecoder(); + let text = ''; + let total = 0; + for await (const chunk of upstream.body) { + total += chunk.length; + if (total > maxBytes) { + try { await upstream.body.cancel(); } catch { /* no-op */ } + return { tooLarge: true, text: '' }; + } + text += decoder.decode(chunk, { stream: true }); + } + text += decoder.decode(); + return { tooLarge: false, text }; +} diff --git a/server/providers/common/query.js b/server/providers/common/query.js new file mode 100644 index 000000000..9e23d2354 --- /dev/null +++ b/server/providers/common/query.js @@ -0,0 +1,12 @@ +export function requiredFiniteQueryNumber(params, key) { + const value = params.get(key); + if (value === null || value.trim() === '') return null; + const number = Number(value); + return Number.isFinite(number) ? number : null; +} + +export function clampInt(value, min, max, fallback) { + const number = Number.parseInt(value, 10); + if (!Number.isFinite(number)) return fallback; + return Math.max(min, Math.min(max, number)); +} diff --git a/server/providers/live.js b/server/providers/live.js new file mode 100644 index 000000000..c67109041 --- /dev/null +++ b/server/providers/live.js @@ -0,0 +1,7 @@ +export { readResponseTextCapped, readResponseJsonCapped, coalesceProxyRequest, readCappedResponseText } from './common/http.js'; +export { requiredFiniteQueryNumber, clampInt } from './common/query.js'; +export { adsbLolFallbackAnchor, openSkyProxy } from './aircraft/opensky.js'; +export { adsbLolProxy } from './aircraft/adsb-lol.js'; +export { adsbdbProxy } from './aircraft/enrichment.js'; +export { trackBackfillProxies } from './aircraft/tracks.js'; +export { aisLiveProxy } from './vessels/ais-live.js'; diff --git a/server/providers/local.js b/server/providers/local.js index 4e1916f00..7268d8282 100644 --- a/server/providers/local.js +++ b/server/providers/local.js @@ -25,6 +25,16 @@ */ import { resolveGoogleServerKey } from '../../scripts/google-server-key.mjs'; + +import { openSkyProxy, adsbLolFallbackAnchor } from './aircraft/opensky.js'; +import { adsbLolProxy } from './aircraft/adsb-lol.js'; +import { adsbdbProxy } from './aircraft/enrichment.js'; +import { trackBackfillProxies } from './aircraft/tracks.js'; +import { aisLiveProxy } from './vessels/ais-live.js'; +import { readResponseTextCapped, readResponseJsonCapped, coalesceProxyRequest, readCappedResponseText } from './common/http.js'; +import { requiredFiniteQueryNumber, clampInt } from './common/query.js'; +export { adsbLolFallbackAnchor, readResponseTextCapped, readResponseJsonCapped, coalesceProxyRequest, requiredFiniteQueryNumber }; + import fs from 'node:fs'; import os from 'node:os'; import { promises as fsp } from 'node:fs'; @@ -43,16 +53,12 @@ import { } from '../../src/data/tomtomTiles.js'; import { filterTrailing24h, parseFirmsCsv } from '../../src/data/firmsCsv.js'; import { fileURLToPath } from 'node:url'; -import { createRequire } from 'node:module'; import { normalizeRadioCountryInput } from '../../src/data/radioCountry.js'; import { normalizeRegionalArticles, normalizeRegionalPlace, normalizeRegionalWeather, } from '../../src/data/regionalBrief.js'; -import { normalizeAdsbLolPointResponse } from '../../src/data/adsbLolFallback.js'; -import { createAisStreamAdapter, isRecognizedAisEnvelope } from '../../src/data/aisStreamAdapter.js'; -import { parseSilenceTimeoutEnv } from '../../src/data/aisWatchdog.js'; import { keylessHudSummaryResponse } from '../../src/hudSummaryResponse.js'; import { parseEnv as parseDotenvText } from 'node:util'; import { readEnvironmentSource as readPinokioEnvironmentSource } from '../../scripts/pinokio-environment.mjs'; @@ -109,79 +115,6 @@ const DEV_FRESH_EXTERNAL_KEYS_AT_BOOT = new Set( .map((name) => name.trim()) .filter((name) => knownKeySetupEnvVars().has(name)), ); - -// --------------------------------------------------------------------------- -// OpenSky OAuth2 token + response cache state -// --------------------------------------------------------------------------- -/** @type {string|null} Current OAuth2 bearer token. */ -let _openskyToken = null; -/** @type {number} Epoch-ms when the current token expires. */ -let _openskyTokenExpiry = 0; -/** @type {Promise|null} In-flight token refresh promise (coalesces concurrent callers). */ -let _openskyTokenPromise = null; -/** @type {string|null} Cached upstream response body (JSON text). */ -let _openskyCacheBody = null; -/** @type {number} HTTP status of the cached response. */ -let _openskyCacheStatus = 0; -/** @type {number} Epoch-ms when the response was cached. */ -let _openskyCacheTime = 0; -/** @type {{requestedMode:string,usedMode:string,reason:string}|null} Auth metadata for the cached response. */ -let _openskyCacheMeta = null; -/** @type {number|null} Source snapshot epoch from the cached OpenSky body. */ -let _openskyCacheSourceEpochMs = null; -/** TTL for the OpenSky response cache (ms). */ -const OPENSKY_CACHE_MS = 9000; -// --- OpenSky credit governor (field-test fix 2026-07-06) ------------------- -// The global /states/all this proxy fetches costs 4 CREDITS per call against -// OpenSky's ~4000/day authenticated budget — a day with the app open burned -// the whole quota in ~8h and the layer then hard-died until the daily reset -// ("rate limited for 48h" owner report; auth itself was fine). Three levers: -// 1. Adaptive TTL: OpenSky returns X-Rate-Limit-Remaining on success; as the -// budget thins, the proxy stretches its cache TTL so a full day of -// continuous use never exhausts it. -// 2. 429 cooldown: honor X-Rate-Limit-Retry-After-Seconds — no upstream -// attempts until it passes (bounded 30 s … 30 min). -// 3. Serve-stale: while rate-limited/cooling, serve the last-good body (200 + -// X-OpenSky-Stale) so the layer keeps rendering instead of dying. -/** @type {number} Current adaptive TTL (ms) — starts at the base cache TTL. */ -let _openskyTtlMs = OPENSKY_CACHE_MS; -/** @type {number} Epoch-ms before which no upstream fetch is attempted. */ -let _openskyCooldownUntil = 0; -/** - * 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 - * later tiers stretch the day: >2400 → ~3 h of full freshness, then 30 s - * (~2.5 h), 90 s (~5 h), 300 s (~8 h) ≈ 18+ h of continuous use per day. - * @param {number} remaining - X-Rate-Limit-Remaining header value. - * @returns {number} TTL in ms. - */ -function openskyAdaptiveTtlMs(remaining) { - if (!Number.isFinite(remaining)) return OPENSKY_CACHE_MS; - if (remaining > 2400) return OPENSKY_CACHE_MS; - if (remaining > 1200) return 30_000; - if (remaining > 400) return 90_000; - return 300_000; -} -/** @type {boolean} Guards duplicate auth-failure warnings in logs. */ -let _openskyAuthWarned = false; -/** @type {boolean} Guards duplicate invalid-auth-mode warnings. */ -let _openskyAuthModeWarned = false; -/** Default auth mode when OPENSKY_AUTH_MODE env is unset. */ -const OPENSKY_AUTH_MODE_DEFAULT = 'oauth'; -/** Set of valid OPENSKY_AUTH_MODE values. */ -const OPENSKY_AUTH_MODE_SET = new Set(['basic', 'oauth', 'auto', 'anon']); -/** Regional civilian fallback cache, keyed by a coarse 0.25° view anchor. */ -const _adsbLolPointCache = new Map(); -/** Per-anchor single-flight map for concurrent regional fallback requests. */ -const _adsbLolPointInFlight = new Map(); -const ADSBLOL_POINT_CACHE_MS = 12000; -const ADSBLOL_POINT_CACHE_MAX = 80; -const ADSBLOL_POINT_RADIUS_NM = 250; -const ADSBLOL_POINT_MAX_RESPONSE_BYTES = 8 * 1024 * 1024; -// A 200 response can still contain an old OpenSky snapshot. Past this point -// the viewport-scoped adsb.lol source is more honest and keeps local motion -// current instead of coasting a stale worldwide frame indefinitely. -const OPENSKY_SOURCE_STALE_MS = 120_000; // --------------------------------------------------------------------------- // Overpass API proxy constants and cache state // --------------------------------------------------------------------------- @@ -726,69 +659,6 @@ async function readRequestBodyCapped(req, maxBytes) { return Buffer.concat(chunks); } -/** - * Read a fetch() Response body as text with a hard byte cap. Rejects early on an - * oversized Content-Length, then streams with a running cap so a chunked or - * length-omitted response cannot blow past the limit. Throws { code:'RESPONSE_TOO_LARGE' }. - */ -export async function readResponseTextCapped(response, maxBytes) { - const declared = Number(response.headers.get('content-length')); - if (Number.isFinite(declared) && declared > maxBytes) { - const err = new Error('Upstream response too large'); - err.code = 'RESPONSE_TOO_LARGE'; - throw err; - } - const reader = response.body?.getReader?.(); - if (!reader) { - const text = await response.text(); - if (Buffer.byteLength(text) > maxBytes) { - const err = new Error('Upstream response too large'); - err.code = 'RESPONSE_TOO_LARGE'; - throw err; - } - return text; - } - const decoder = new TextDecoder(); - let out = ''; - let total = 0; - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - total += value.byteLength; - if (total > maxBytes) { - try { await reader.cancel(); } catch { /* no-op */ } - const err = new Error('Upstream response too large'); - err.code = 'RESPONSE_TOO_LARGE'; - throw err; - } - out += decoder.decode(value, { stream: true }); - } - out += decoder.decode(); - return out; -} - -/** Parse a fetch() JSON response only after enforcing a hard byte cap. */ -export async function readResponseJsonCapped(response, maxBytes) { - return JSON.parse(await readResponseTextCapped(response, maxBytes)); -} - -/** - * Return the existing promise for a cache key, or create one and remove it - * only when that exact promise settles. - */ -export function coalesceProxyRequest(inFlight, key, create) { - const existing = inFlight.get(key); - if (existing) return { promise: existing, shared: true }; - let promise; - promise = Promise.resolve() - .then(create) - .finally(() => { - if (inFlight.get(key) === promise) inFlight.delete(key); - }); - inFlight.set(key, promise); - return { promise, shared: false }; -} - // --------------------------------------------------------------------------- // Radio Browser directory proxy // --------------------------------------------------------------------------- @@ -1329,46 +1199,6 @@ const GBFS_ALLOWED_HOSTS = new Set([ 'hon.publicbikesystem.net', 'chat.publicbikesystem.net', ]); - -// --------------------------------------------------------------------------- -// AISStream live vessel cache state -// --------------------------------------------------------------------------- -const AISSTREAM_URL = 'wss://stream.aisstream.io/v0/stream'; -const AISSTREAM_DEFAULT_BBOXES = [[[-90, -180], [90, 180]]]; -const AISSTREAM_DEFAULT_MESSAGE_TYPES = [ - 'PositionReport', - 'StandardClassBPositionReport', - 'ExtendedClassBPositionReport', - 'ShipStaticData', - 'StaticDataReport', -]; -const AISSTREAM_CACHE_MAX = 50000; -const AISSTREAM_STALE_MS = 30 * 60 * 1000; -// Per-MMSI recent-path ring buffers (PRD WS-F F3). Float32 lat/lon (~1m -// precision, fine for 25m thinning) + Uint32 epoch seconds ≈ 12B/sample; -// 64 samples × 50k MMSIs worst case ≈ 38MB. Tracks exist only while the dev -// server runs — this is "recent path", not voyage history. -const AIS_TRACK_SAMPLES = 64; -const AIS_TRACK_MIN_GAP_SEC = 30; -const AIS_TRACK_MIN_MOVE_M = 25; -// Watchdog budgets (policy lives in src/data/aisWatchdog.js). Silence is -// REPORTED quickly and ACTED ON slowly: a dead feed must read as dead within -// ~2 min, but recycling the socket is throttled so recovery can never become a -// reconnect cycle against AISStream's one-connection-per-key limit. -const AISSTREAM_SILENCE_REPORT_MS = 120_000; -/** Recycle threshold as a multiple of the report threshold. */ -const AISSTREAM_RECYCLE_RATIO = 2.5; -const AISSTREAM_BACKOFF_MS = Object.freeze([5_000, 15_000, 60_000, 300_000]); -/** Slow retry cadence once the ladder is spent and the feed reads DOWN. */ -const AISSTREAM_DOWN_RETRY_MS = 900_000; -/** - * Probe cadence while AISStream is rejecting the key. Retrying cannot fix a - * bad credential, so this exists only to recover from an upstream-side - * mistake — it must never approach the ladder's pace. - */ -const AISSTREAM_AUTH_PROBE_MS = 3_600_000; -/** How often the watchdog re-evaluates without request traffic. */ -const AISSTREAM_TICK_MS = 15_000; // Sourced from the shared voice-model registry so the client's cost estimate // can never be computed against a different model than the session runs on. const OPENAI_REALTIME_MODEL_DEFAULT = VOICE_MODELS.standard.id; @@ -1382,155 +1212,6 @@ const REALTIME_DEBUG_LOG_DIR = path.join(__dirname, '.gev-logs'); const REALTIME_DEBUG_LOG_FILE = path.join(REALTIME_DEBUG_LOG_DIR, 'realtime-conversations.jsonl'); const REALTIME_DEBUG_LOG_MAX_BYTES = 8 * 1024 * 1024; -/** - * @type {ReturnType|null} - * Module-lifetime: it owns the socket-generation namespace, which must never - * restart across a dev-server reload (see aisStreamAdapter.js ownership rules). - */ -let _aisAdapter = null; -/** @type {{silenceWatch:boolean,reportMs:number,recycleMs:number,url:string}|null} */ -let _aisWatchdogPolicy = null; -/** @type {number|null} */ -let _aisStreamTickTimer = null; -/** Set by dispose so the next ensure() re-derives budgets from a reloaded .env. */ -let _aisNeedsRearm = false; -/** @type {Function|null|undefined} `ws` constructor; null = unavailable, undefined = not yet probed. */ -let _aisWebSocketImpl; -/** @type {Map} */ -const _aisStreamVessels = new Map(); -/** @type {Map} */ -const _aisStreamStatic = new Map(); -/** @type {Map} mmsi -> track ring buffer */ -const _aisStreamTracks = new Map(); -/** @type {Map} mmsi -> first fix awaiting second (lazy buffer allocation) */ -const _aisStreamTrackPending = new Map(); - -/** - * Obtain a valid OpenSky OAuth2 bearer token, refreshing if needed. - * - * Uses the client_credentials grant against the OpenSky Keycloak realm. - * Concurrent callers share a single in-flight refresh promise so only - * one token request is issued at a time. - * - * @returns {Promise} Bearer token string, or null if unavailable. - */ -async function getOpenSkyToken() { - const now = Date.now(); - // Return cached token if still valid (with 60 s safety margin) - if (_openskyToken && now < _openskyTokenExpiry - 60000) return _openskyToken; - - // Coalesce concurrent refresh requests — if a refresh is already in-flight, - // return the same promise instead of issuing a duplicate token request - if (_openskyTokenPromise) return _openskyTokenPromise; - - const clientId = process.env.OPENSKY_CLIENT_ID; - const clientSecret = process.env.OPENSKY_CLIENT_SECRET; - if (!clientId || !clientSecret) return null; - - // Wrap the async token fetch in a shared promise stored in _openskyTokenPromise - _openskyTokenPromise = (async () => { - try { - const res = await fetch( - 'https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token', - { - method: 'POST', - headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, - body: `grant_type=client_credentials&client_id=${encodeURIComponent(clientId)}&client_secret=${encodeURIComponent(clientSecret)}`, - } - ); - - let data = null; - try { - data = await res.json(); - } catch { - data = null; - } - - const accessToken = data?.access_token; - const expiresIn = Number(data?.expires_in); - if (!res.ok || !accessToken) { - if (!_openskyAuthWarned) { - const detail = data?.error_description || data?.error || `HTTP ${res.status}`; - console.warn('[OpenSky] OAuth client_credentials failed:', detail); - _openskyAuthWarned = true; - } - _openskyToken = null; - _openskyTokenExpiry = 0; - return null; - } - - _openskyToken = accessToken; - // Default to 1800 s (30 min) if expires_in is missing or non-finite - _openskyTokenExpiry = Date.now() + (Number.isFinite(expiresIn) ? expiresIn : 1800) * 1000; - console.log('[OpenSky] OAuth token refreshed, expires in', Number.isFinite(expiresIn) ? expiresIn : 1800, 's'); - _openskyAuthWarned = false; - return _openskyToken; - } catch (err) { - if (!_openskyAuthWarned) { - console.warn('[OpenSky] OAuth token request failed:', err?.message || String(err)); - _openskyAuthWarned = true; - } - _openskyToken = null; - _openskyTokenExpiry = 0; - return null; - } finally { - // Clear the shared promise so the next caller can start a fresh refresh - _openskyTokenPromise = null; - } - })(); - - return _openskyTokenPromise; -} - -/** - * Validate and normalize the OPENSKY_AUTH_MODE env value. - * - * @param {string} value - Raw env value (e.g. 'basic', 'oauth', 'auto', 'anon'). - * @returns {string} One of the valid mode strings, or the default ('oauth'). - */ -function normalizeOpenSkyAuthMode(value) { - const raw = String(value || '').trim().toLowerCase(); - if (!raw) return OPENSKY_AUTH_MODE_DEFAULT; - if (OPENSKY_AUTH_MODE_SET.has(raw)) return raw; - if (!_openskyAuthModeWarned) { - console.warn( - `[OpenSky] Invalid OPENSKY_AUTH_MODE="${raw}", defaulting to "${OPENSKY_AUTH_MODE_DEFAULT}"` - ); - _openskyAuthModeWarned = true; - } - return OPENSKY_AUTH_MODE_DEFAULT; -} - -/** - * Build standard response headers for OpenSky proxy responses. - * - * Includes diagnostic X-OpenSky-* headers so the client can inspect - * cache hit/miss status and which auth mode was actually used. - * - * @param {object} opts - * @param {string} opts.cacheStatus - 'HIT', 'MISS', or 'STALE'. - * @param {string} opts.requestedMode - The auth mode the config requested. - * @param {string} opts.usedMode - The auth mode actually used for the upstream call. - * @param {string} opts.reason - Human-readable reason string for diagnostics. - * @returns {Record} Header object. - */ -function buildOpenSkyHeaders({ cacheStatus, requestedMode, usedMode, reason, staleSeconds, retryAfterSeconds }) { - const headers = { - 'Content-Type': 'application/json', - 'Cache-Control': 'no-store', - 'X-OpenSky-Cache': cacheStatus, - 'X-OpenSky-Auth': usedMode, - 'X-OpenSky-Auth-Mode-Requested': requestedMode, - 'X-OpenSky-Auth-Mode-Used': usedMode, - 'X-OpenSky-Auth-Reason': reason, - }; - // Credit-governor extras (field-test fix 2026-07-06): the client can show a - // STALE cue / countdown without parsing the body. - if (Number.isFinite(staleSeconds)) headers['X-OpenSky-Stale-Seconds'] = String(Math.round(staleSeconds)); - if (Number.isFinite(retryAfterSeconds)) headers['X-OpenSky-Retry-After-Seconds'] = String(Math.round(retryAfterSeconds)); - return headers; -} - /** * Vite plugin: CelesTrak TLE proxy. * @@ -2381,128 +2062,6 @@ function terrainHeightsProxy() { }; } -/** - * adsbdb.com enrichment proxy: callsign → route (airline + origin/destination - * airports) and hex → aircraft type/registration. Free community API — cached - * aggressively: ONE upstream request per new key ever (404s negative-cached), - * persisted to disk so restarts don't re-hammer it. Adapted from skylight - * (MIT) server/src/enrich/routes.ts. - */ -function adsbdbProxy() { - const TTL_MS = 24 * 3600_000; - const CACHE_PATH = path.join(process.cwd(), '.gev-cache', 'adsbdb.json'); - let cache = { routes: {}, aircraft: {} }; - let dirty = false; - let loaded = false; - const inflight = new Map(); - - async function loadOnce() { - if (loaded) return; - loaded = true; - try { - const parsed = JSON.parse(await fsp.readFile(CACHE_PATH, 'utf8')); - cache = { routes: parsed.routes ?? {}, aircraft: parsed.aircraft ?? {} }; - } catch { /* first run */ } - setInterval(async () => { - if (!dirty) return; - dirty = false; - try { - await fsp.mkdir(path.dirname(CACHE_PATH), { recursive: true }); - await fsp.writeFile(CACHE_PATH, JSON.stringify(cache), 'utf8'); - } catch { dirty = true; } // retry next tick - }, 15_000).unref?.(); - } - - const fresh = (e) => e && Date.now() - e.at < TTL_MS; - - function parseRoute(json) { - const fr = json?.response?.flightroute; - if (!fr?.origin || !fr?.destination) return null; - const airport = (a) => ({ - code: a.iata_code || a.icao_code || '', - name: a.municipality || a.name || '', - lat: Number.isFinite(a.latitude) ? a.latitude : null, - lon: Number.isFinite(a.longitude) ? a.longitude : null, - }); - return { airline: fr.airline?.name || null, origin: airport(fr.origin), destination: airport(fr.destination) }; - } - - function parseAircraft(json) { - const a = json?.response?.aircraft; - if (!a) return null; - return { - typeCode: a.icao_type || null, // ICAO designator, e.g. "B738" — feeds classifyAircraft - typeName: a.manufacturer && a.type ? `${a.manufacturer} ${a.type}` : (a.type || null), - registration: a.registration || null, - }; - } - - function lookup(kind, key) { - const store = kind === 'route' ? cache.routes : cache.aircraft; - if (fresh(store[key])) return Promise.resolve(store[key].data); - const ik = `${kind}:${key}`; - if (!inflight.has(ik)) { - inflight.set(ik, (async () => { - try { - const url = kind === 'route' - ? `https://api.adsbdb.com/v0/callsign/${encodeURIComponent(key)}` - : `https://api.adsbdb.com/v0/aircraft/${encodeURIComponent(key)}`; - const res = await fetch(url, { signal: AbortSignal.timeout(8000) }); - if (res.ok) { - const data = kind === 'route' ? parseRoute(await res.json()) : parseAircraft(await res.json()); - store[key] = { at: Date.now(), data }; // data may be null — negative cache - dirty = true; - return data; - } - if (res.status === 404) { - store[key] = { at: Date.now(), data: null }; // known-missing — cache the miss - dirty = true; - } - // other statuses: leave uncached so we retry later - return fresh(store[key]) ? store[key].data : null; - } catch { - return fresh(store[key]) ? store[key].data : null; // network error → stale if any - } finally { - inflight.delete(ik); - } - })()); - } - return inflight.get(ik); - } - - return { - name: 'adsbdb-proxy', - configureServer(server) { - server.middlewares.use('/api/adsbdb', async (req, res) => { - await loadOnce(); - const send = (status, obj) => { - res.writeHead(status, { 'Content-Type': 'application/json' }); - res.end(JSON.stringify(obj)); - }; - try { - const [, kind, rawKey] = String(req.url || '').split('?')[0].split('/'); - if (kind === 'route') { - const cs = String(rawKey || '').toUpperCase(); - if (!/^[A-Z0-9]{2,8}$/.test(cs)) return send(400, { error: 'invalid callsign' }); - const data = await lookup('route', cs); - return send(200, data ? { found: true, ...data } : { found: false }); - } - if (kind === 'type') { - const hex = String(rawKey || '').toLowerCase(); - if (!/^[0-9a-f]{6}$/.test(hex)) return send(400, { error: 'invalid hex' }); - const data = await lookup('aircraft', hex); - return send(200, data ? { found: true, ...data } : { found: false }); - } - return send(404, { error: 'unknown endpoint' }); - } catch (err) { - console.error('[adsbdb-proxy] request failed'); - return send(500, { error: 'adsbdb proxy error' }); - } - }); - }, - }; -} - /** * Detect whether an Overpass API response body indicates rate-limiting. * @@ -2901,401 +2460,6 @@ function overpassProxy() { }; } -export function adsbLolFallbackAnchor(req) { - const incoming = new URL(req?.url || '', 'http://localhost'); - const latitude = requiredFiniteQueryNumber(incoming.searchParams, 'lat'); - const longitude = requiredFiniteQueryNumber(incoming.searchParams, 'lon'); - if (!Number.isFinite(latitude) || latitude < -90 || latitude > 90) return null; - if (!Number.isFinite(longitude) || longitude < -180 || longitude > 180) return null; - return { latitude, longitude }; -} - -async function fetchAdsbLolPointFallback(req) { - const anchor = adsbLolFallbackAnchor(req); - if (!anchor) return null; - const roundedLat = Math.round(anchor.latitude * 4) / 4; - const roundedLon = Math.round(anchor.longitude * 4) / 4; - const cacheKey = `${roundedLat.toFixed(2)},${roundedLon.toFixed(2)}`; - const cached = _adsbLolPointCache.get(cacheKey); - const now = Date.now(); - if (cached && now - cached.cachedAt < ADSBLOL_POINT_CACHE_MS) { - return { ...cached, cacheStatus: 'HIT' }; - } - - const request = coalesceProxyRequest(_adsbLolPointInFlight, cacheKey, async () => { - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), 10000); - try { - const upstream = await fetch( - `https://api.adsb.lol/v2/lat/${roundedLat}/lon/${roundedLon}/dist/${ADSBLOL_POINT_RADIUS_NM}`, - { - headers: { - Accept: 'application/json', - 'User-Agent': 'gods-eye-view-adsblol-regional-fallback/1.0', - }, - signal: controller.signal, - }, - ); - if (!upstream.ok) throw new Error(`upstream HTTP ${upstream.status}`); - const payload = await readResponseJsonCapped(upstream, ADSBLOL_POINT_MAX_RESPONSE_BYTES); - const normalized = normalizeAdsbLolPointResponse(payload); - const record = { - body: JSON.stringify(normalized), - cachedAt: Date.now(), - count: normalized.states.length, - }; - _adsbLolPointCache.delete(cacheKey); - _adsbLolPointCache.set(cacheKey, record); - while (_adsbLolPointCache.size > ADSBLOL_POINT_CACHE_MAX) { - _adsbLolPointCache.delete(_adsbLolPointCache.keys().next().value); - } - return record; - } finally { - clearTimeout(timeoutId); - } - }); - try { - const record = await request.promise; - return { ...record, cacheStatus: request.shared ? 'INFLIGHT' : 'MISS' }; - } catch (error) { - if (!request.shared && error?.name !== 'AbortError') { - console.warn('[adsb.lol Flights Fallback]', error?.message || error); - } - return cached ? { ...cached, cacheStatus: 'STALE' } : null; - } -} - -async function serveAdsbLolPointFallback(req, res, requestedMode, reason) { - const fallback = await fetchAdsbLolPointFallback(req); - if (!fallback) return false; - res.writeHead(200, { - ...buildOpenSkyHeaders({ - cacheStatus: fallback.cacheStatus, - requestedMode, - usedMode: 'adsblol-regional', - reason, - }), - 'X-Flight-Source': 'adsb.lol', - 'X-Flight-Coverage': `${ADSBLOL_POINT_RADIUS_NM}nm regional fallback`, - 'X-Flight-Count': String(fallback.count), - }); - res.end(fallback.body); - return true; -} - -function openSkySourceEpochMs(body) { - try { - const seconds = Number(JSON.parse(body)?.time); - return Number.isFinite(seconds) && seconds > 0 ? seconds * 1000 : null; - } catch { - return null; - } -} - -function openSkySourceIsStale(sourceEpochMs, now = Date.now()) { - return Number.isFinite(sourceEpochMs) - && now - sourceEpochMs > OPENSKY_SOURCE_STALE_MS; -} - -/** - * Vite plugin: OpenSky Network proxy with multi-mode auth and response caching. - * - * Supports four auth modes controlled by OPENSKY_AUTH_MODE env: - * - 'oauth' (default) — client_credentials bearer token - * - 'basic' — HTTP Basic with OPENSKY_USERNAME / OPENSKY_PASSWORD - * - 'auto' — try OAuth first, fall back to Basic, then anon - * - 'anon' — no credentials - * - * Successful responses are cached for OPENSKY_CACHE_MS (~9 s). On - * upstream failure the proxy serves a stale cached response if available, or - * a bounded 250 nm adsb.lol point snapshot around the current view anchor. - * - * @returns {import('vite').Plugin} - */ -function openSkyProxy() { - return { - name: 'opensky-proxy', - configureServer(server) { - server.middlewares.use('/api/opensky', async (req, res) => { - try { - const requestedMode = normalizeOpenSkyAuthMode(process.env.OPENSKY_AUTH_MODE); - const now = Date.now(); - const inCooldown = now < _openskyCooldownUntil; - // Fresh-enough cache (adaptive TTL) OR any cache during a 429 - // cooldown: serve it without touching upstream. Stale-during-cooldown - // is deliberate (credit governor): last-good planes beat a dead layer. - if (_openskyCacheBody && (now - _openskyCacheTime < _openskyTtlMs || inCooldown)) { - if ( - openSkySourceIsStale(_openskyCacheSourceEpochMs, now) - && await serveAdsbLolPointFallback( - req, - res, - requestedMode, - 'opensky_snapshot_stale_regional_fallback', - ) - ) { - return; - } - const cachedMeta = _openskyCacheMeta || { - requestedMode, - usedMode: 'unknown', - reason: 'cached', - }; - const isStale = now - _openskyCacheTime >= _openskyTtlMs; - res.writeHead( - _openskyCacheStatus || 200, - buildOpenSkyHeaders({ - cacheStatus: isStale ? 'STALE' : 'HIT', - requestedMode: cachedMeta.requestedMode || requestedMode, - usedMode: cachedMeta.usedMode || 'unknown', - reason: isStale ? 'rate_limited_serving_stale' : (cachedMeta.reason || 'cached'), - staleSeconds: isStale ? (now - _openskyCacheTime) / 1000 : undefined, - retryAfterSeconds: inCooldown ? (_openskyCooldownUntil - now) / 1000 : undefined, - }) - ); - res.end(_openskyCacheBody); - return; - } - // Cooling down with nothing cached (cold start into a rate limit): - // synthesize the 429 locally — hammering upstream mid-cooldown can't - // succeed and just burns goodwill. - if (inCooldown) { - if (await serveAdsbLolPointFallback(req, res, requestedMode, 'opensky_cooldown_regional_fallback')) return; - res.writeHead(429, buildOpenSkyHeaders({ - cacheStatus: 'COOLDOWN', - requestedMode, - usedMode: 'none', - reason: 'rate_limited', - retryAfterSeconds: (_openskyCooldownUntil - now) / 1000, - })); - res.end(JSON.stringify({ error: 'OpenSky rate limited; 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'; - - 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 = await fetch('https://opensky-network.org/api/states/all?extended=1', { 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 = await fetch('https://opensky-network.org/api/states/all?extended=1', { headers: retryHeaders }); - usedMode = 'basic'; - reason = 'oauth_rejected_fallback_basic'; - } - - let body = await upstream.text(); - const sourceEpochMs = upstream.ok ? openSkySourceEpochMs(body) : null; - if ( - upstream.ok - && openSkySourceIsStale(sourceEpochMs, now) - && await serveAdsbLolPointFallback( - req, - res, - requestedMode, - '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. - _openskyCacheBody = body; - _openskyCacheStatus = upstream.status; - _openskyCacheTime = now; - _openskyCacheSourceEpochMs = sourceEpochMs; - _openskyCacheMeta = { requestedMode, usedMode, reason }; - return; - } - if (upstream.status === 429) { - reason = 'rate_limited'; - // Credit governor: honor OpenSky's retry-after (bounded 30 s … 30 min; - // 2 min when the header is absent) — no upstream attempts until then. - const retryAfterSec = Number(upstream.headers.get('x-rate-limit-retry-after-seconds')); - const cooldownMs = Math.min( - Math.max(Number.isFinite(retryAfterSec) ? retryAfterSec * 1000 : 120_000, 30_000), - 30 * 60_000 - ); - _openskyCooldownUntil = now + cooldownMs; - // 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) { - res.writeHead(200, buildOpenSkyHeaders({ - cacheStatus: 'STALE', - requestedMode, - usedMode, - reason: 'rate_limited_serving_stale', - staleSeconds: (now - _openskyCacheTime) / 1000, - retryAfterSeconds: cooldownMs / 1000, - })); - res.end(_openskyCacheBody); - return; - } - } - - if (!upstream.ok && !_openskyCacheBody) { - const servedFallback = await serveAdsbLolPointFallback( - req, - res, - requestedMode, - `opensky_http_${upstream.status}_regional_fallback`, - ); - if (servedFallback) return; - } - - if (upstream.status === 401 || upstream.status === 403) { - if (requestedMode === 'basic' && !hasBasicCreds) { - body = JSON.stringify({ - error: 'OpenSky auth missing. Basic mode requires OPENSKY_USERNAME and OPENSKY_PASSWORD.', - }); - reason = 'missing_basic_creds'; - } else if (requestedMode === 'oauth' && usedMode !== 'oauth') { - body = JSON.stringify({ - error: 'OpenSky auth invalid. OAuth mode requires valid OPENSKY_CLIENT_ID and OPENSKY_CLIENT_SECRET.', - }); - reason = 'oauth_invalid_or_missing'; - } else if (usedMode === 'basic') { - body = JSON.stringify({ - error: 'OpenSky auth invalid. Username/password were rejected.', - }); - reason = 'basic_invalid_credentials'; - } else if (usedMode === 'oauth') { - body = JSON.stringify({ - error: 'OpenSky auth invalid. OAuth client credentials were rejected.', - }); - reason = 'oauth_invalid_credentials'; - } else if (requestedMode === 'auto' && !hasBasicCreds) { - body = JSON.stringify({ - error: 'OpenSky auth missing. Provide basic credentials or valid OAuth client credentials.', - }); - reason = 'missing_oauth_and_basic_creds'; - } else { - body = JSON.stringify({ - error: 'OpenSky auth required.', - }); - reason = 'auth_required'; - } - } - - // 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) { - _openskyCacheBody = body; - _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; - } - - res.writeHead( - upstream.status, - buildOpenSkyHeaders({ - cacheStatus: 'MISS', - requestedMode, - usedMode, - reason, - }) - ); - res.end(body); - } catch (e) { - console.error('[OpenSky Proxy]', e.message); - if (_openskyCacheBody) { - const cachedMeta = _openskyCacheMeta || { - requestedMode: normalizeOpenSkyAuthMode(process.env.OPENSKY_AUTH_MODE), - usedMode: 'unknown', - reason: 'cached_stale', - }; - res.writeHead( - _openskyCacheStatus || 200, - buildOpenSkyHeaders({ - cacheStatus: 'STALE', - requestedMode: cachedMeta.requestedMode || OPENSKY_AUTH_MODE_DEFAULT, - usedMode: cachedMeta.usedMode || 'unknown', - reason: cachedMeta.reason || 'cached_stale', - }) - ); - res.end(_openskyCacheBody); - return; - } - const requestedMode = normalizeOpenSkyAuthMode(process.env.OPENSKY_AUTH_MODE); - if (await serveAdsbLolPointFallback(req, res, requestedMode, 'opensky_proxy_error_regional_fallback')) return; - res.writeHead( - 502, - buildOpenSkyHeaders({ - cacheStatus: 'MISS', - requestedMode, - usedMode: 'error', - reason: 'proxy_error', - }) - ); - res.end(JSON.stringify({ error: 'OpenSky proxy error' })); - } - }); - }, - }; -} - /** * Check whether a hostname is in the GBFS allowlist. * @@ -4376,52 +3540,6 @@ function toReadable(body) { return null; } -/** - * Pipe an upstream fetch Response (image or video) to the client HTTP response. - * - * Forwards Content-Type, Content-Length, Content-Range, Accept-Ranges, and - * Cache-Control headers from the upstream. Falls back to buffered arrayBuffer - * if the body is not streamable. - * - * @param {import('http').ServerResponse} res - * @param {Response} upstream - fetch() Response object. - * @param {object} [opts] - * @param {string} [opts.sourceHeader='upstream'] - Value for X-CCTV-Source header. - */ -/** - * Read a fetch Response body as text while enforcing a hard byte cap during - * the read — so a malicious or buggy upstream that streams an unbounded body - * (no/oversized Content-Length, chunked) can't OOM the proxy. Returns - * { tooLarge, text }. Cancels the stream as soon as the cap is crossed. - * @param {Response} upstream - fetch() response. - * @param {number} maxBytes - hard ceiling on decoded bytes. - * @returns {Promise<{tooLarge: boolean, text: string}>} - */ -async function readCappedResponseText(upstream, maxBytes) { - const declared = Number(upstream.headers.get('content-length')); - if (Number.isFinite(declared) && declared > maxBytes) { - try { await upstream.body?.cancel(); } catch { /* no-op */ } - return { tooLarge: true, text: '' }; - } - if (!upstream.body || typeof upstream.body[Symbol.asyncIterator] !== 'function') { - const text = await upstream.text(); - return text.length > maxBytes ? { tooLarge: true, text: '' } : { tooLarge: false, text }; - } - const decoder = new TextDecoder(); - let text = ''; - let total = 0; - for await (const chunk of upstream.body) { - total += chunk.length; - if (total > maxBytes) { - try { await upstream.body.cancel(); } catch { /* no-op */ } - return { tooLarge: true, text: '' }; - } - text += decoder.decode(chunk, { stream: true }); - } - text += decoder.decode(); - return { tooLarge: false, text }; -} - async function proxyMediaResponse(res, upstream, { sourceHeader = 'upstream' } = {}) { const contentType = upstream.headers.get('content-type') || 'application/octet-stream'; const cacheControl = upstream.headers.get('cache-control') || 'no-store'; @@ -4808,260 +3926,6 @@ function cctvProxy() { }; } -/** - * Vite plugin: adsb.lol military aircraft proxy with 12 s response cache. - * - * Proxies GET /api/adsblol/mil to https://api.adsb.lol/v2/mil. On upstream - * failure, serves a stale cached response if one exists. - * - * @returns {import('vite').Plugin} - */ -function adsbLolProxy() { - /** @type {string|null} Cached upstream JSON body. */ - let _cache = null; - /** @type {number} Epoch-ms when the cache was populated. */ - let _cacheAt = 0; - /** Response cache TTL (ms). */ - const CACHE_MS = 12000; - return { - name: 'adsblol-proxy', - configureServer(server) { - server.middlewares.use('/api/adsblol/mil', async (req, res) => { - try { - const now = Date.now(); - if (_cache && now - _cacheAt < CACHE_MS) { - res.writeHead(200, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'X-ADS-B-Cache': 'HIT' }); - res.end(_cache); - return; - } - const upstream = await fetch('https://api.adsb.lol/v2/mil', { - headers: { 'User-Agent': 'gods-eye-view-adsblol-proxy/1.0' }, - }); - const body = await upstream.text(); - if (upstream.ok) { - _cache = body; - _cacheAt = now; - } - res.writeHead(upstream.status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'X-ADS-B-Cache': 'MISS' }); - res.end(body); - } catch (e) { - console.error('[adsb.lol Proxy]', e.message); - if (_cache) { - res.writeHead(200, { 'Content-Type': 'application/json', 'X-ADS-B-Cache': 'STALE' }); - res.end(_cache); - return; - } - res.writeHead(502, { 'Content-Type': 'application/json' }); - res.end(JSON.stringify({ error: 'ADS-B proxy error' })); - } - }); - }, - }; -} - -/** - * Vite plugin: AISStream live vessel cache. - * - * AISStream does not support browser CORS and requires a private API key, so - * the Vite server keeps one backend websocket open and exposes a same-origin - * JSON snapshot to the Cesium layer. - */ -function aisLiveProxy() { - function install(middlewares) { - middlewares.use('/api/ais-live', async (req, res) => { - try { - ensureAisStreamConnection(); - const incoming = new URL(req.url || '', 'http://localhost'); - - // Track sub-route MUST be handled before the rows snapshot — this - // mount prefix-matches every subpath, so without this branch - // /api/ais-live/track would be silently answered with vessel rows. - if (incoming.pathname === '/track' || incoming.pathname.startsWith('/track/')) { - const mmsi = String(incoming.searchParams.get('mmsi') || '').trim(); - res.statusCode = /^\d{5,10}$/.test(mmsi) ? 200 : 400; - res.setHeader('Content-Type', 'application/json; charset=utf-8'); - res.setHeader('Cache-Control', 'no-store'); - if (res.statusCode !== 200) { - res.end(JSON.stringify({ error: 'mmsi query param required', samples: [] })); - return; - } - res.end(JSON.stringify({ - mmsi, - samples: readAisTrack(mmsi), - source: 'AISStream (accumulated since server start)', - retainedSec: Math.floor(AISSTREAM_STALE_MS / 1000), - })); - return; - } - - const maxRows = clampInt(incoming.searchParams.get('maxRows'), 1, AISSTREAM_CACHE_MAX, AISSTREAM_CACHE_MAX); - const rows = aisStreamRows(maxRows); - - const feed = aisStreamStatusSnapshot(); - - res.statusCode = process.env.AISSTREAM_API_KEY ? 200 : 503; - res.setHeader('Content-Type', 'application/json; charset=utf-8'); - res.setHeader('Cache-Control', 'no-store'); - res.end(JSON.stringify({ - rows, - source: 'AISStream', - status: feed.status, - error: feed.error, - refreshing: feed.status !== 'live', - newestPositionAt: newestAisPositionAt(rows), - lastMessageAt: feed.lastMessageAt, - // Honest-failure metadata: how long the feed has been quiet, which - // recovery attempt we are on, and when the next one lands. - silentForMs: feed.silentForMs, - reconnectAttempt: feed.reconnectAttempt, - nextAttemptAt: feed.nextAttemptAt, - staleAfterMs: feed.staleAfterMs, - watchdog: feed.watchdog, - })); - } catch (error) { - res.statusCode = 502; - res.setHeader('Content-Type', 'application/json; charset=utf-8'); - res.setHeader('Cache-Control', 'no-store'); - res.end(JSON.stringify({ error: error?.message || 'AIS live stream error', rows: [] })); - } - }); - } - - return { - name: 'ais-live-proxy', - configureServer(server) { - install(server.middlewares); - startAisStreamWatchdogTick(); - // Vite restarts the server in-process on a config change while this - // module's state survives; without teardown each reload stacks another - // interval and another socket. - server.httpServer?.on('close', disposeAisStream); - }, - configurePreviewServer(server) { - install(server.middlewares); - startAisStreamWatchdogTick(); - server.httpServer?.on('close', disposeAisStream); - }, - // Middleware-mode backstop: there is no httpServer to hang 'close' on. - closeBundle() { - disposeAisStream(); - }, - }; -} - -/** - * Vite plugin: aircraft track-history backfill proxies (PRD WS-F F1/F2). - * - * /api/opensky-track?icao24= — OpenSky GET /tracks/all (experimental; - * own credit bucket, 4 credits per call on the free tier). OAuth via the - * shared coalesced token. 60s per-icao cache; 404/429 forwarded so the - * client can fall back to its accumulated trail silently. - * /api/adsblol/trace?hex= — adsb.lol tar1090 readsb trace - * (undocumented but live; no browser CORS, hence this proxy). Up to ~24h - * of real history per aircraft. Treat as best-effort; data is ODbL — - * credit "adsb.lol (ODbL)" in the UI. - */ -function trackBackfillProxies() { - const TRACK_CACHE_MS = 60000; - const TRACK_CACHE_MAX = 200; - const RESPONSE_CAP_BYTES = 5 * 1024 * 1024; - /** @type {Map} */ - const cache = new Map(); - - function cachePut(key, entry) { - cache.set(key, entry); - if (cache.size > TRACK_CACHE_MAX) { - const oldest = [...cache.entries()].sort((a, b) => a[1].at - b[1].at)[0]; - if (oldest) cache.delete(oldest[0]); - } - } - - async function proxyJson(res, key, upstreamUrl, headers = {}) { - const cached = cache.get(key); - if (cached && Date.now() - cached.at < TRACK_CACHE_MS) { - res.statusCode = cached.status; - res.setHeader('Content-Type', 'application/json; charset=utf-8'); - res.setHeader('Cache-Control', 'no-store'); - res.end(cached.body); - return; - } - const upstream = await fetch(upstreamUrl, { headers, signal: AbortSignal.timeout(12000) }); - const { tooLarge, text } = await readCappedResponseText(upstream, RESPONSE_CAP_BYTES); - let body; - if (tooLarge) { - body = JSON.stringify({ error: 'Upstream track response too large' }); - } else if (!upstream.ok) { - // Sanitize upstream error surface; status code is signal enough - body = JSON.stringify({ error: `Track source HTTP ${upstream.status}` }); - } else { - body = text; - } - cachePut(key, { at: Date.now(), status: upstream.status, body }); - res.statusCode = upstream.status; - res.setHeader('Content-Type', 'application/json; charset=utf-8'); - res.setHeader('Cache-Control', 'no-store'); - res.end(body); - } - - function install(middlewares) { - middlewares.use('/api/opensky-track', async (req, res) => { - try { - const incoming = new URL(req.url || '', 'http://localhost'); - const icao24 = String(incoming.searchParams.get('icao24') || '').trim().toLowerCase(); - if (!/^[0-9a-f]{6}$/.test(icao24)) { - res.statusCode = 400; - res.setHeader('Content-Type', 'application/json'); - res.end(JSON.stringify({ error: 'icao24 must be a 6-char hex string' })); - return; - } - const token = await getOpenSkyToken(); - await proxyJson( - res, - `osky:${icao24}`, - `https://opensky-network.org/api/tracks/all?icao24=${icao24}&time=0`, - token ? { Authorization: `Bearer ${token}` } : {} - ); - } catch (error) { - res.statusCode = 502; - res.setHeader('Content-Type', 'application/json'); - res.end(JSON.stringify({ error: 'OpenSky track fetch failed' })); - } - }); - - middlewares.use('/api/adsblol/trace', async (req, res) => { - try { - const incoming = new URL(req.url || '', 'http://localhost'); - const hex = String(incoming.searchParams.get('hex') || '').trim().toLowerCase(); - if (!/^[0-9a-f~]{6,7}$/.test(hex)) { - res.statusCode = 400; - res.setHeader('Content-Type', 'application/json'); - res.end(JSON.stringify({ error: 'hex must be a 6-7 char hex string' })); - return; - } - await proxyJson( - res, - `lol:${hex}`, - `https://adsb.lol/data/traces/${hex.slice(-2)}/trace_full_${hex}.json` - ); - } catch (error) { - res.statusCode = 502; - res.setHeader('Content-Type', 'application/json'); - res.end(JSON.stringify({ error: 'adsb.lol trace fetch failed' })); - } - }); - } - - return { - name: 'track-backfill-proxies', - configureServer(server) { - install(server.middlewares); - }, - configurePreviewServer(server) { - install(server.middlewares); - }, - }; -} - /** * Vite plugin: OpenAI Realtime ephemeral client secret. * @@ -6305,405 +5169,6 @@ const GEV_REALTIME_TOOLS = [ }, ]; -/** - * Load the `ws` constructor once. - * - * Node's built-in WebSocket cannot be used here: it has no terminate(), and - * its close() waits forever for a close frame a black-holed peer never sends - * (verified in src/data/aisWatchdogTransport.test.mjs). A socket parked in - * CLOSING keeps holding AISStream's single per-key connection, which is how - * the reverted watchdog wedged. - * - * Loaded lazily rather than imported at the top of this file so a missing - * optional dependency degrades the vessel feed honestly instead of breaking - * the whole dev server and build. - * - * @returns {Function|null} - */ -function aisWebSocketImpl() { - if (_aisWebSocketImpl !== undefined) return _aisWebSocketImpl; - try { - _aisWebSocketImpl = createRequire(import.meta.url)('ws'); - } catch (error) { - _aisWebSocketImpl = null; - console.warn('[AISStream] `ws` is unavailable; the live vessel feed is off.', error?.message || ''); - } - return _aisWebSocketImpl; -} - -/** - * Resolve the watchdog policy from the environment, once. - * - * Read lazily because module evaluation happens before Vite's loadEnv() copies - * .env into process.env — the reverted watchdog read these at import time and - * silently ignored every .env value, including its own kill switch. - * - * A custom subscription (one harbor, one message type) can be legitimately - * silent for minutes, so the silence watch only self-arms for the default - * worldwide subscription. An operator with a narrow filter opts back in by - * setting AISSTREAM_SILENCE_TIMEOUT_MS to a value sized for that filter; 0 is - * an explicit kill switch. - */ -function aisWatchdogPolicy() { - if (_aisWatchdogPolicy) return _aisWatchdogPolicy; - const customSubscription = Boolean( - process.env.AISSTREAM_BOUNDING_BOXES || process.env.AISSTREAM_MESSAGE_TYPES, - ); - const override = parseSilenceTimeoutEnv( - process.env.AISSTREAM_SILENCE_TIMEOUT_MS, - (message) => console.warn(message), - ); - const reportMs = override.kind === 'timeout' ? override.value : AISSTREAM_SILENCE_REPORT_MS; - _aisWatchdogPolicy = { - silenceWatch: override.kind === 'off' ? false : (override.kind === 'timeout' || !customSubscription), - reportMs, - recycleMs: Math.round(reportMs * AISSTREAM_RECYCLE_RATIO), - // Overridable so the watchdog can be exercised end-to-end against a local - // stand-in upstream without opening a connection to AISStream (which - // allows only one per key). - url: process.env.AISSTREAM_URL || AISSTREAM_URL, - }; - return _aisWatchdogPolicy; -} - - -/** - * The transport adapter, built on first use and kept for the module lifetime. - * - * Never rebuilt: it owns the socket-generation namespace, and a restarted - * namespace would let a pre-disposal handler act on its successor's socket. - */ -function aisAdapter() { - if (_aisAdapter) return _aisAdapter; - _aisAdapter = createAisStreamAdapter({ - createSocket: (url) => { - const WebSocketCtor = aisWebSocketImpl(); - if (!WebSocketCtor) throw new Error('ws transport unavailable'); - return new WebSocketCtor(url); - }, - resolveUrl: () => aisWatchdogPolicy().url, - buildSubscription: aisStreamSubscription, - ingestEnvelope: ingestAisStreamEnvelope, - warn: (message) => console.warn(message), - }); - _aisAdapter.setWatchdogOptions(aisWatchdogBudgets()); - return _aisAdapter; -} - -/** Watchdog budgets derived from the resolved environment policy. */ -function aisWatchdogBudgets() { - const policy = aisWatchdogPolicy(); - return { - staleMs: policy.reportMs, - recycleAfterMs: policy.recycleMs, - backoffMs: [...AISSTREAM_BACKOFF_MS], - downRetryMs: AISSTREAM_DOWN_RETRY_MS, - authProbeMs: AISSTREAM_AUTH_PROBE_MS, - }; -} - -/** - * Fingerprint the credential so a key change can clear the terminal - * auth-failed state. Only a truncated digest is kept — never the key. - */ -function aisKeyFingerprint() { - const key = process.env.AISSTREAM_API_KEY; - if (!key) return null; - return createHash('sha256').update(String(key)).digest('hex').slice(0, 12); -} - -/** - * Drive the watchdog once. Called on every /api/ais-live request and on the - * background interval, so recovery does not depend on browser traffic. - */ -function ensureAisStreamConnection() { - const adapter = aisAdapter(); - if (_aisNeedsRearm) { - // Post-dispose re-arm, now that the restarted server's .env is loaded. The - // adapter keeps its generation namespace across this. - _aisNeedsRearm = false; - adapter.setWatchdogOptions(aisWatchdogBudgets()); - } - const policy = aisWatchdogPolicy(); - adapter.ensure({ - hasKey: Boolean(process.env.AISSTREAM_API_KEY), - hasTransport: Boolean(aisWebSocketImpl()), - silenceWatch: policy.silenceWatch, - keyFingerprint: aisKeyFingerprint(), - }); -} -/** Status metadata for /api/ais-live, safe to call before the first connect. */ -function aisStreamStatusSnapshot() { - const snapshot = _aisAdapter ? _aisAdapter.snapshot() : null; - if (snapshot) return snapshot; - return { - status: process.env.AISSTREAM_API_KEY ? 'idle' : 'missing-key', - error: process.env.AISSTREAM_API_KEY ? null : 'AISSTREAM_API_KEY is not set', - lastMessageAt: null, - silentForMs: null, - reconnectAttempt: 0, - nextAttemptAt: null, - watchdog: 'armed', - staleAfterMs: AISSTREAM_SILENCE_REPORT_MS, - }; -} - -/** - * Start the background watchdog tick. Unref'd so it never holds the dev server - * open, and idempotent so a Vite in-process restart cannot stack intervals. - */ -function startAisStreamWatchdogTick() { - if (_aisStreamTickTimer) return; - _aisStreamTickTimer = setInterval(() => { - try { - ensureAisStreamConnection(); - } catch (error) { - console.warn('[AISStream] watchdog tick failed', error?.message || ''); - } - }, AISSTREAM_TICK_MS); - _aisStreamTickTimer.unref?.(); -} - -/** - * Tear down every timer and socket this module owns. - * - * Vite restarts the dev server in-process on a config change while module - * state survives, so without this each reload stacked another interval and - * another reconnect chain. The cached policy is dropped too, so a restart - * re-reads .env. - * - * The adapter instance itself is deliberately KEPT: it owns the socket - * generation namespace, which must stay monotonic across restarts so a - * pre-disposal handler can never collide with a post-disposal socket. - */ -function disposeAisStream() { - if (_aisStreamTickTimer) { - clearInterval(_aisStreamTickTimer); - _aisStreamTickTimer = null; - } - if (_aisAdapter) _aisAdapter.dispose(); - // Drop the cached policy and re-arm LAZILY. Re-deriving budgets here would - // read process.env before the restarted server's loadEnv() has repopulated - // it, caching the outgoing configuration; the next ensure() runs after that. - _aisWatchdogPolicy = null; - _aisNeedsRearm = true; -} - -function aisStreamSubscription() { - return { - APIKey: process.env.AISSTREAM_API_KEY, - BoundingBoxes: parseJsonEnv('AISSTREAM_BOUNDING_BOXES', AISSTREAM_DEFAULT_BBOXES), - FilterMessageTypes: parseCsvOrJsonEnv('AISSTREAM_MESSAGE_TYPES', AISSTREAM_DEFAULT_MESSAGE_TYPES), - }; -} - -/** - * Store one parsed AIS envelope. - * - * The return value is the feed's ONLY liveness proof, so it is true strictly - * when the envelope carried a real AIS record. Malformed frames and error - * envelopes never reach here — the adapter classifies those — and a JSON - * object without an MMSI proves nothing about the feed. - * - * @param {Object} envelope Parsed, non-error AIS envelope. - * @returns {boolean} True when an AIS record was recognised. - */ -function ingestAisStreamEnvelope(envelope) { - // Single shared recognition rule (also used by the adapter's tests), so the - // liveness predicate that ships is the one under test. An envelope carrying - // only an MMSI is not proof the feed works. - if (!isRecognizedAisEnvelope(envelope)) return false; - - const messageType = envelope?.MessageType; - const message = envelope?.Message?.[messageType] || {}; - const metadata = envelope?.MetaData || envelope?.Metadata || {}; - const mmsi = stringValue(metadata.MMSI ?? message.UserID ?? message.UserId ?? message.Mmsi); - if (!mmsi) return false; - - if (messageType === 'ShipStaticData' || messageType === 'StaticDataReport') { - const staticData = { - name: vesselNameFromAis(metadata, message, _aisStreamStatic.get(mmsi)), - type: vesselTypeFromAis(message, _aisStreamStatic.get(mmsi)), - destination: stringValue(message.Destination), - imo: stringValue(message.ImoNumber ?? message.IMO), - }; - _aisStreamStatic.set(mmsi, staticData); - mergeAisStaticIntoLiveVessel(mmsi, staticData); - } - - const lat = numberValue(metadata.latitude ?? metadata.Latitude ?? message.Latitude); - const lon = numberValue(metadata.longitude ?? metadata.Longitude ?? message.Longitude); - // A positionless but well-formed record (static data) is still the feed - // delivering AIS traffic, so it counts as liveness. - if (!Number.isFinite(lat) || !Number.isFinite(lon)) return true; - - const staticData = _aisStreamStatic.get(mmsi) || {}; - _aisStreamVessels.set(mmsi, { - lat, - lon, - name: vesselNameFromAis(metadata, message, staticData) || `MMSI ${mmsi}`, - mmsi, - imo: stringValue(message.ImoNumber ?? message.IMO ?? staticData.imo), - type: vesselTypeFromAis(message, staticData), - destination: stringValue(message.Destination ?? staticData.destination), - speed: numberValue(message.Sog ?? message.SOG), - course: numberValue(message.Cog ?? message.COG), - heading: normalizedHeading(message.TrueHeading ?? message.Heading), - last_position_UTC: normalizeAisTimestamp(metadata.time_utc ?? metadata.TimeUtc), - // Use the AIS message's own report time, not server ingest wall-clock — - // trail spacing and dead reckoning depend on true fix epochs. - last_position_epoch: aisEpochSeconds(metadata.time_utc ?? metadata.TimeUtc), - _updatedAt: Date.now(), - }); - - appendAisTrackSample(mmsi, lat, lon, aisEpochSeconds(metadata.time_utc ?? metadata.TimeUtc)); - - pruneAisStreamCache(); - return true; -} - -/** - * Parses an AISStream UTC timestamp into epoch seconds (fallback: now). - */ -function aisEpochSeconds(value) { - const ms = Date.parse(normalizeAisTimestamp(value)); - return Number.isFinite(ms) ? Math.floor(ms / 1000) : Math.floor(Date.now() / 1000); -} - -/** - * Appends a thinned position sample to a vessel's track ring buffer. - * Buffers allocate lazily on the second fix (most MMSIs are seen once); - * samples are kept only when >=AIS_TRACK_MIN_GAP_SEC and - * >=AIS_TRACK_MIN_MOVE_M from the previous stored sample, so anchored - * vessels collapse to a single point. - */ -function appendAisTrackSample(mmsi, lat, lon, epochSec) { - let track = _aisStreamTracks.get(mmsi); - if (!track) { - const pending = _aisStreamTrackPending.get(mmsi); - if (!pending) { - _aisStreamTrackPending.set(mmsi, { lat, lon, epochSec }); - return; - } - if (epochSec - pending.epochSec < AIS_TRACK_MIN_GAP_SEC) return; - if (approxMetersBetween(pending.lat, pending.lon, lat, lon) < AIS_TRACK_MIN_MOVE_M) return; - track = { - lats: new Float32Array(AIS_TRACK_SAMPLES), - lons: new Float32Array(AIS_TRACK_SAMPLES), - times: new Uint32Array(AIS_TRACK_SAMPLES), - head: 0, - len: 0, - }; - _aisStreamTracks.set(mmsi, track); - _aisStreamTrackPending.delete(mmsi); - writeAisTrackSample(track, pending.lat, pending.lon, pending.epochSec); - writeAisTrackSample(track, lat, lon, epochSec); - return; - } - - const lastIdx = (track.head - 1 + AIS_TRACK_SAMPLES) % AIS_TRACK_SAMPLES; - const lastEpoch = track.times[lastIdx]; - if (epochSec - lastEpoch < AIS_TRACK_MIN_GAP_SEC) return; - if (approxMetersBetween(track.lats[lastIdx], track.lons[lastIdx], lat, lon) < AIS_TRACK_MIN_MOVE_M) return; - writeAisTrackSample(track, lat, lon, epochSec); -} - -function writeAisTrackSample(track, lat, lon, epochSec) { - track.lats[track.head] = lat; - track.lons[track.head] = lon; - track.times[track.head] = epochSec; - track.head = (track.head + 1) % AIS_TRACK_SAMPLES; - track.len = Math.min(track.len + 1, AIS_TRACK_SAMPLES); -} - -/** - * Reads a vessel's accumulated track in chronological order. - * @returns {Array<{lat:number,lon:number,t:number}>} - */ -function readAisTrack(mmsi) { - const track = _aisStreamTracks.get(mmsi); - if (!track || !track.len) return []; - const samples = []; - const start = (track.head - track.len + AIS_TRACK_SAMPLES) % AIS_TRACK_SAMPLES; - for (let i = 0; i < track.len; i++) { - const idx = (start + i) % AIS_TRACK_SAMPLES; - samples.push({ lat: track.lats[idx], lon: track.lons[idx], t: track.times[idx] }); - } - return samples; -} - -/** Equirectangular distance approximation — plenty for 25m thinning. */ -function approxMetersBetween(lat1, lon1, lat2, lon2) { - const dLat = (lat2 - lat1) * 111320; - const dLon = (lon2 - lon1) * 111320 * Math.cos(((lat1 + lat2) / 2) * (Math.PI / 180)); - return Math.hypot(dLat, dLon); -} - -function mergeAisStaticIntoLiveVessel(mmsi, staticData) { - const existing = _aisStreamVessels.get(mmsi); - if (!existing) return; - if (staticData.name && (!existing.name || existing.name === `MMSI ${mmsi}`)) existing.name = staticData.name; - if (staticData.type && !existing.type) existing.type = staticData.type; - if (staticData.destination && !existing.destination) existing.destination = staticData.destination; - if (staticData.imo && !existing.imo) existing.imo = staticData.imo; -} - -function vesselNameFromAis(metadata, message, staticData = {}) { - return stringValue( - metadata.ShipName - ?? message.Name - ?? message.ShipName - ?? message.ReportA?.Name - ?? staticData.name - ); -} - -function vesselTypeFromAis(message, staticData = {}) { - return stringValue( - message.Type - ?? message.ShipType - ?? message.ReportB?.ShipType - ?? staticData.type - ); -} - -function aisStreamRows(maxRows) { - const cutoff = Date.now() - AISSTREAM_STALE_MS; - const rows = []; - for (const row of _aisStreamVessels.values()) { - if (row._updatedAt >= cutoff) rows.push(row); - } - rows.sort((a, b) => b._updatedAt - a._updatedAt); - return rows.slice(0, maxRows).map(({ _updatedAt, ...row }) => row); -} - -function pruneAisStreamCache() { - const cutoff = Date.now() - AISSTREAM_STALE_MS; - for (const [mmsi, row] of _aisStreamVessels) { - if (row._updatedAt < cutoff) { - _aisStreamVessels.delete(mmsi); - _aisStreamTracks.delete(mmsi); - _aisStreamTrackPending.delete(mmsi); - } - } - // Pending single-fix entries for vessels never seen again must not leak - const pendingCutoffSec = Math.floor(cutoff / 1000); - for (const [mmsi, pending] of _aisStreamTrackPending) { - if (pending.epochSec < pendingCutoffSec) _aisStreamTrackPending.delete(mmsi); - } - if (_aisStreamVessels.size <= AISSTREAM_CACHE_MAX) return; - const ordered = [..._aisStreamVessels.entries()].sort((a, b) => a[1]._updatedAt - b[1]._updatedAt); - for (const [mmsi] of ordered.slice(0, _aisStreamVessels.size - AISSTREAM_CACHE_MAX)) { - _aisStreamVessels.delete(mmsi); - _aisStreamTracks.delete(mmsi); - _aisStreamTrackPending.delete(mmsi); - } -} - -function newestAisPositionAt(rows) { - return rows[0]?.last_position_UTC || null; -} - // --------------------------------------------------------------------------- // Military-installation context proxy // --------------------------------------------------------------------------- @@ -7074,13 +5539,6 @@ const _weatherEffectsRateLimiter = makeRateLimiter({ windowMs: 60_000, max: 45, let _nominatimQueue = Promise.resolve(); let _nominatimLastRequestAt = 0; -export function requiredFiniteQueryNumber(params, key) { - const value = params.get(key); - if (value === null || value.trim() === '') return null; - const number = Number(value); - return Number.isFinite(number) ? number : null; -} - export function validRegionalPoint(params) { const latitude = requiredFiniteQueryNumber(params, 'latitude'); const longitude = requiredFiniteQueryNumber(params, 'longitude'); @@ -7431,58 +5889,6 @@ function weatherEffectsProxy() { }; } -function parseJsonEnv(key, fallback) { - const value = process.env[key]; - if (!value) return fallback; - try { - return JSON.parse(value); - } catch { - console.warn(`[AISStream] Invalid ${key}; using default.`); - return fallback; - } -} - -function parseCsvOrJsonEnv(key, fallback) { - const value = process.env[key]; - if (!value) return fallback; - try { - const parsed = JSON.parse(value); - return Array.isArray(parsed) ? parsed : fallback; - } catch { - return value.split(',').map((entry) => entry.trim()).filter(Boolean); - } -} - -function clampInt(value, min, max, fallback) { - const number = Number.parseInt(value, 10); - if (!Number.isFinite(number)) return fallback; - return Math.max(min, Math.min(max, number)); -} - -function stringValue(value) { - if (value === undefined || value === null) return ''; - return String(value).trim(); -} - -function numberValue(value) { - if (value === null || value === undefined || value === '') return null; - const number = Number(value); - return Number.isFinite(number) ? number : null; -} - -function normalizedHeading(value) { - const heading = numberValue(value); - return heading !== null && heading >= 0 && heading <= 360 ? heading : null; -} - -function normalizeAisTimestamp(value) { - const text = stringValue(value); - if (!text) return new Date().toISOString(); - const normalized = text.replace(' +0000 UTC', 'Z').replace(' UTC', 'Z'); - const date = new Date(normalized); - return Number.isNaN(date.getTime()) ? new Date().toISOString() : date.toISOString(); -} - /** * In-app key setup ("POWER UP" panel) — dev-server only. * diff --git a/server/providers/vessels/ais-live.js b/server/providers/vessels/ais-live.js new file mode 100644 index 000000000..d40c04539 --- /dev/null +++ b/server/providers/vessels/ais-live.js @@ -0,0 +1,355 @@ +import { createHash } from 'node:crypto'; +import { createRequire } from 'node:module'; +import { createAisStreamAdapter } from '../../../src/data/aisStreamAdapter.js'; +import { parseSilenceTimeoutEnv } from '../../../src/data/aisWatchdog.js'; +import { clampInt } from '../common/query.js'; +import { AISSTREAM_CACHE_MAX, AISSTREAM_STALE_MS, ingestAisStreamEnvelope, readAisTrack, aisStreamRows, newestAisPositionAt } from './ais-store.js'; +// --------------------------------------------------------------------------- +// AISStream live vessel cache state +// --------------------------------------------------------------------------- +const AISSTREAM_URL = 'wss://stream.aisstream.io/v0/stream'; +const AISSTREAM_DEFAULT_BBOXES = [[[-90, -180], [90, 180]]]; +const AISSTREAM_DEFAULT_MESSAGE_TYPES = [ + 'PositionReport', + 'StandardClassBPositionReport', + 'ExtendedClassBPositionReport', + 'ShipStaticData', + 'StaticDataReport', +]; +// Watchdog budgets (policy lives in src/data/aisWatchdog.js). Silence is +// REPORTED quickly and ACTED ON slowly: a dead feed must read as dead within +// ~2 min, but recycling the socket is throttled so recovery can never become a +// reconnect cycle against AISStream's one-connection-per-key limit. +const AISSTREAM_SILENCE_REPORT_MS = 120_000; +/** Recycle threshold as a multiple of the report threshold. */ +const AISSTREAM_RECYCLE_RATIO = 2.5; +const AISSTREAM_BACKOFF_MS = Object.freeze([5_000, 15_000, 60_000, 300_000]); +/** Slow retry cadence once the ladder is spent and the feed reads DOWN. */ +const AISSTREAM_DOWN_RETRY_MS = 900_000; +/** + * Probe cadence while AISStream is rejecting the key. Retrying cannot fix a + * bad credential, so this exists only to recover from an upstream-side + * mistake — it must never approach the ladder's pace. + */ +const AISSTREAM_AUTH_PROBE_MS = 3_600_000; +/** How often the watchdog re-evaluates without request traffic. */ +const AISSTREAM_TICK_MS = 15_000; + +/** + * @type {ReturnType|null} + * Module-lifetime: it owns the socket-generation namespace, which must never + * restart across a dev-server reload (see aisStreamAdapter.js ownership rules). + */ +let _aisAdapter = null; +/** @type {{silenceWatch:boolean,reportMs:number,recycleMs:number,url:string}|null} */ +let _aisWatchdogPolicy = null; +/** @type {number|null} */ +let _aisStreamTickTimer = null; +/** Set by dispose so the next ensure() re-derives budgets from a reloaded .env. */ +let _aisNeedsRearm = false; +/** @type {Function|null|undefined} `ws` constructor; null = unavailable, undefined = not yet probed. */ +let _aisWebSocketImpl; + +/** + * Vite plugin: AISStream live vessel cache. + * + * AISStream does not support browser CORS and requires a private API key, so + * the Vite server keeps one backend websocket open and exposes a same-origin + * JSON snapshot to the Cesium layer. + */ +export function aisLiveProxy() { + function install(middlewares) { + middlewares.use('/api/ais-live', async (req, res) => { + try { + ensureAisStreamConnection(); + const incoming = new URL(req.url || '', 'http://localhost'); + + // Track sub-route MUST be handled before the rows snapshot — this + // mount prefix-matches every subpath, so without this branch + // /api/ais-live/track would be silently answered with vessel rows. + if (incoming.pathname === '/track' || incoming.pathname.startsWith('/track/')) { + const mmsi = String(incoming.searchParams.get('mmsi') || '').trim(); + res.statusCode = /^\d{5,10}$/.test(mmsi) ? 200 : 400; + res.setHeader('Content-Type', 'application/json; charset=utf-8'); + res.setHeader('Cache-Control', 'no-store'); + if (res.statusCode !== 200) { + res.end(JSON.stringify({ error: 'mmsi query param required', samples: [] })); + return; + } + res.end(JSON.stringify({ + mmsi, + samples: readAisTrack(mmsi), + source: 'AISStream (accumulated since server start)', + retainedSec: Math.floor(AISSTREAM_STALE_MS / 1000), + })); + return; + } + + const maxRows = clampInt(incoming.searchParams.get('maxRows'), 1, AISSTREAM_CACHE_MAX, AISSTREAM_CACHE_MAX); + const rows = aisStreamRows(maxRows); + + const feed = aisStreamStatusSnapshot(); + + res.statusCode = process.env.AISSTREAM_API_KEY ? 200 : 503; + res.setHeader('Content-Type', 'application/json; charset=utf-8'); + res.setHeader('Cache-Control', 'no-store'); + res.end(JSON.stringify({ + rows, + source: 'AISStream', + status: feed.status, + error: feed.error, + refreshing: feed.status !== 'live', + newestPositionAt: newestAisPositionAt(rows), + lastMessageAt: feed.lastMessageAt, + // Honest-failure metadata: how long the feed has been quiet, which + // recovery attempt we are on, and when the next one lands. + silentForMs: feed.silentForMs, + reconnectAttempt: feed.reconnectAttempt, + nextAttemptAt: feed.nextAttemptAt, + staleAfterMs: feed.staleAfterMs, + watchdog: feed.watchdog, + })); + } catch (error) { + res.statusCode = 502; + res.setHeader('Content-Type', 'application/json; charset=utf-8'); + res.setHeader('Cache-Control', 'no-store'); + res.end(JSON.stringify({ error: error?.message || 'AIS live stream error', rows: [] })); + } + }); + } + + return { + name: 'ais-live-proxy', + configureServer(server) { + install(server.middlewares); + startAisStreamWatchdogTick(); + // Vite restarts the server in-process on a config change while this + // module's state survives; without teardown each reload stacks another + // interval and another socket. + server.httpServer?.on('close', disposeAisStream); + }, + configurePreviewServer(server) { + install(server.middlewares); + startAisStreamWatchdogTick(); + server.httpServer?.on('close', disposeAisStream); + }, + // Middleware-mode backstop: there is no httpServer to hang 'close' on. + closeBundle() { + disposeAisStream(); + }, + }; +} + +/** + * Load the `ws` constructor once. + * + * Node's built-in WebSocket cannot be used here: it has no terminate(), and + * its close() waits forever for a close frame a black-holed peer never sends + * (verified in src/data/aisWatchdogTransport.test.mjs). A socket parked in + * CLOSING keeps holding AISStream's single per-key connection, which is how + * the reverted watchdog wedged. + * + * Loaded lazily rather than imported at the top of this file so a missing + * optional dependency degrades the vessel feed honestly instead of breaking + * the whole dev server and build. + * + * @returns {Function|null} + */ +function aisWebSocketImpl() { + if (_aisWebSocketImpl !== undefined) return _aisWebSocketImpl; + try { + _aisWebSocketImpl = createRequire(import.meta.url)('ws'); + } catch (error) { + _aisWebSocketImpl = null; + console.warn('[AISStream] `ws` is unavailable; the live vessel feed is off.', error?.message || ''); + } + return _aisWebSocketImpl; +} + +/** + * Resolve the watchdog policy from the environment, once. + * + * Read lazily because module evaluation happens before Vite's loadEnv() copies + * .env into process.env — the reverted watchdog read these at import time and + * silently ignored every .env value, including its own kill switch. + * + * A custom subscription (one harbor, one message type) can be legitimately + * silent for minutes, so the silence watch only self-arms for the default + * worldwide subscription. An operator with a narrow filter opts back in by + * setting AISSTREAM_SILENCE_TIMEOUT_MS to a value sized for that filter; 0 is + * an explicit kill switch. + */ +function aisWatchdogPolicy() { + if (_aisWatchdogPolicy) return _aisWatchdogPolicy; + const customSubscription = Boolean( + process.env.AISSTREAM_BOUNDING_BOXES || process.env.AISSTREAM_MESSAGE_TYPES, + ); + const override = parseSilenceTimeoutEnv( + process.env.AISSTREAM_SILENCE_TIMEOUT_MS, + (message) => console.warn(message), + ); + const reportMs = override.kind === 'timeout' ? override.value : AISSTREAM_SILENCE_REPORT_MS; + _aisWatchdogPolicy = { + silenceWatch: override.kind === 'off' ? false : (override.kind === 'timeout' || !customSubscription), + reportMs, + recycleMs: Math.round(reportMs * AISSTREAM_RECYCLE_RATIO), + // Overridable so the watchdog can be exercised end-to-end against a local + // stand-in upstream without opening a connection to AISStream (which + // allows only one per key). + url: process.env.AISSTREAM_URL || AISSTREAM_URL, + }; + return _aisWatchdogPolicy; +} + + +/** + * The transport adapter, built on first use and kept for the module lifetime. + * + * Never rebuilt: it owns the socket-generation namespace, and a restarted + * namespace would let a pre-disposal handler act on its successor's socket. + */ +function aisAdapter() { + if (_aisAdapter) return _aisAdapter; + _aisAdapter = createAisStreamAdapter({ + createSocket: (url) => { + const WebSocketCtor = aisWebSocketImpl(); + if (!WebSocketCtor) throw new Error('ws transport unavailable'); + return new WebSocketCtor(url); + }, + resolveUrl: () => aisWatchdogPolicy().url, + buildSubscription: aisStreamSubscription, + ingestEnvelope: ingestAisStreamEnvelope, + warn: (message) => console.warn(message), + }); + _aisAdapter.setWatchdogOptions(aisWatchdogBudgets()); + return _aisAdapter; +} + +/** Watchdog budgets derived from the resolved environment policy. */ +function aisWatchdogBudgets() { + const policy = aisWatchdogPolicy(); + return { + staleMs: policy.reportMs, + recycleAfterMs: policy.recycleMs, + backoffMs: [...AISSTREAM_BACKOFF_MS], + downRetryMs: AISSTREAM_DOWN_RETRY_MS, + authProbeMs: AISSTREAM_AUTH_PROBE_MS, + }; +} + +/** + * Fingerprint the credential so a key change can clear the terminal + * auth-failed state. Only a truncated digest is kept — never the key. + */ +function aisKeyFingerprint() { + const key = process.env.AISSTREAM_API_KEY; + if (!key) return null; + return createHash('sha256').update(String(key)).digest('hex').slice(0, 12); +} + +/** + * Drive the watchdog once. Called on every /api/ais-live request and on the + * background interval, so recovery does not depend on browser traffic. + */ +function ensureAisStreamConnection() { + const adapter = aisAdapter(); + if (_aisNeedsRearm) { + // Post-dispose re-arm, now that the restarted server's .env is loaded. The + // adapter keeps its generation namespace across this. + _aisNeedsRearm = false; + adapter.setWatchdogOptions(aisWatchdogBudgets()); + } + const policy = aisWatchdogPolicy(); + adapter.ensure({ + hasKey: Boolean(process.env.AISSTREAM_API_KEY), + hasTransport: Boolean(aisWebSocketImpl()), + silenceWatch: policy.silenceWatch, + keyFingerprint: aisKeyFingerprint(), + }); +} +/** Status metadata for /api/ais-live, safe to call before the first connect. */ +function aisStreamStatusSnapshot() { + const snapshot = _aisAdapter ? _aisAdapter.snapshot() : null; + if (snapshot) return snapshot; + return { + status: process.env.AISSTREAM_API_KEY ? 'idle' : 'missing-key', + error: process.env.AISSTREAM_API_KEY ? null : 'AISSTREAM_API_KEY is not set', + lastMessageAt: null, + silentForMs: null, + reconnectAttempt: 0, + nextAttemptAt: null, + watchdog: 'armed', + staleAfterMs: AISSTREAM_SILENCE_REPORT_MS, + }; +} + +/** + * Start the background watchdog tick. Unref'd so it never holds the dev server + * open, and idempotent so a Vite in-process restart cannot stack intervals. + */ +function startAisStreamWatchdogTick() { + if (_aisStreamTickTimer) return; + _aisStreamTickTimer = setInterval(() => { + try { + ensureAisStreamConnection(); + } catch (error) { + console.warn('[AISStream] watchdog tick failed', error?.message || ''); + } + }, AISSTREAM_TICK_MS); + _aisStreamTickTimer.unref?.(); +} + +/** + * Tear down every timer and socket this module owns. + * + * Vite restarts the dev server in-process on a config change while module + * state survives, so without this each reload stacked another interval and + * another reconnect chain. The cached policy is dropped too, so a restart + * re-reads .env. + * + * The adapter instance itself is deliberately KEPT: it owns the socket + * generation namespace, which must stay monotonic across restarts so a + * pre-disposal handler can never collide with a post-disposal socket. + */ +function disposeAisStream() { + if (_aisStreamTickTimer) { + clearInterval(_aisStreamTickTimer); + _aisStreamTickTimer = null; + } + if (_aisAdapter) _aisAdapter.dispose(); + // Drop the cached policy and re-arm LAZILY. Re-deriving budgets here would + // read process.env before the restarted server's loadEnv() has repopulated + // it, caching the outgoing configuration; the next ensure() runs after that. + _aisWatchdogPolicy = null; + _aisNeedsRearm = true; +} + +function aisStreamSubscription() { + return { + APIKey: process.env.AISSTREAM_API_KEY, + BoundingBoxes: parseJsonEnv('AISSTREAM_BOUNDING_BOXES', AISSTREAM_DEFAULT_BBOXES), + FilterMessageTypes: parseCsvOrJsonEnv('AISSTREAM_MESSAGE_TYPES', AISSTREAM_DEFAULT_MESSAGE_TYPES), + }; +} + +function parseJsonEnv(key, fallback) { + const value = process.env[key]; + if (!value) return fallback; + try { + return JSON.parse(value); + } catch { + console.warn(`[AISStream] Invalid ${key}; using default.`); + return fallback; + } +} + +function parseCsvOrJsonEnv(key, fallback) { + const value = process.env[key]; + if (!value) return fallback; + try { + const parsed = JSON.parse(value); + return Array.isArray(parsed) ? parsed : fallback; + } catch { + return value.split(',').map((entry) => entry.trim()).filter(Boolean); + } +} diff --git a/server/providers/vessels/ais-store.js b/server/providers/vessels/ais-store.js new file mode 100644 index 000000000..0ba5235f0 --- /dev/null +++ b/server/providers/vessels/ais-store.js @@ -0,0 +1,249 @@ +import { isRecognizedAisEnvelope } from '../../../src/data/aisStreamAdapter.js'; +export const AISSTREAM_CACHE_MAX = 50000; +export const AISSTREAM_STALE_MS = 30 * 60 * 1000; +// Per-MMSI recent-path ring buffers (PRD WS-F F3). Float32 lat/lon (~1m +// precision, fine for 25m thinning) + Uint32 epoch seconds ≈ 12B/sample; +// 64 samples × 50k MMSIs worst case ≈ 38MB. Tracks exist only while the dev +// server runs — this is "recent path", not voyage history. +const AIS_TRACK_SAMPLES = 64; +const AIS_TRACK_MIN_GAP_SEC = 30; +const AIS_TRACK_MIN_MOVE_M = 25; +/** @type {Map} */ +const _aisStreamVessels = new Map(); +/** @type {Map} */ +const _aisStreamStatic = new Map(); +/** @type {Map} mmsi -> track ring buffer */ +const _aisStreamTracks = new Map(); +/** @type {Map} mmsi -> first fix awaiting second (lazy buffer allocation) */ +const _aisStreamTrackPending = new Map(); + +/** + * Store one parsed AIS envelope. + * + * The return value is the feed's ONLY liveness proof, so it is true strictly + * when the envelope carried a real AIS record. Malformed frames and error + * envelopes never reach here — the adapter classifies those — and a JSON + * object without an MMSI proves nothing about the feed. + * + * @param {Object} envelope Parsed, non-error AIS envelope. + * @returns {boolean} True when an AIS record was recognised. + */ +export function ingestAisStreamEnvelope(envelope) { + // Single shared recognition rule (also used by the adapter's tests), so the + // liveness predicate that ships is the one under test. An envelope carrying + // only an MMSI is not proof the feed works. + if (!isRecognizedAisEnvelope(envelope)) return false; + + const messageType = envelope?.MessageType; + const message = envelope?.Message?.[messageType] || {}; + const metadata = envelope?.MetaData || envelope?.Metadata || {}; + const mmsi = stringValue(metadata.MMSI ?? message.UserID ?? message.UserId ?? message.Mmsi); + if (!mmsi) return false; + + if (messageType === 'ShipStaticData' || messageType === 'StaticDataReport') { + const staticData = { + name: vesselNameFromAis(metadata, message, _aisStreamStatic.get(mmsi)), + type: vesselTypeFromAis(message, _aisStreamStatic.get(mmsi)), + destination: stringValue(message.Destination), + imo: stringValue(message.ImoNumber ?? message.IMO), + }; + _aisStreamStatic.set(mmsi, staticData); + mergeAisStaticIntoLiveVessel(mmsi, staticData); + } + + const lat = numberValue(metadata.latitude ?? metadata.Latitude ?? message.Latitude); + const lon = numberValue(metadata.longitude ?? metadata.Longitude ?? message.Longitude); + // A positionless but well-formed record (static data) is still the feed + // delivering AIS traffic, so it counts as liveness. + if (!Number.isFinite(lat) || !Number.isFinite(lon)) return true; + + const staticData = _aisStreamStatic.get(mmsi) || {}; + _aisStreamVessels.set(mmsi, { + lat, + lon, + name: vesselNameFromAis(metadata, message, staticData) || `MMSI ${mmsi}`, + mmsi, + imo: stringValue(message.ImoNumber ?? message.IMO ?? staticData.imo), + type: vesselTypeFromAis(message, staticData), + destination: stringValue(message.Destination ?? staticData.destination), + speed: numberValue(message.Sog ?? message.SOG), + course: numberValue(message.Cog ?? message.COG), + heading: normalizedHeading(message.TrueHeading ?? message.Heading), + last_position_UTC: normalizeAisTimestamp(metadata.time_utc ?? metadata.TimeUtc), + // Use the AIS message's own report time, not server ingest wall-clock — + // trail spacing and dead reckoning depend on true fix epochs. + last_position_epoch: aisEpochSeconds(metadata.time_utc ?? metadata.TimeUtc), + _updatedAt: Date.now(), + }); + + appendAisTrackSample(mmsi, lat, lon, aisEpochSeconds(metadata.time_utc ?? metadata.TimeUtc)); + + pruneAisStreamCache(); + return true; +} + +/** + * Parses an AISStream UTC timestamp into epoch seconds (fallback: now). + */ +function aisEpochSeconds(value) { + const ms = Date.parse(normalizeAisTimestamp(value)); + return Number.isFinite(ms) ? Math.floor(ms / 1000) : Math.floor(Date.now() / 1000); +} + +/** + * Appends a thinned position sample to a vessel's track ring buffer. + * Buffers allocate lazily on the second fix (most MMSIs are seen once); + * samples are kept only when >=AIS_TRACK_MIN_GAP_SEC and + * >=AIS_TRACK_MIN_MOVE_M from the previous stored sample, so anchored + * vessels collapse to a single point. + */ +function appendAisTrackSample(mmsi, lat, lon, epochSec) { + let track = _aisStreamTracks.get(mmsi); + if (!track) { + const pending = _aisStreamTrackPending.get(mmsi); + if (!pending) { + _aisStreamTrackPending.set(mmsi, { lat, lon, epochSec }); + return; + } + if (epochSec - pending.epochSec < AIS_TRACK_MIN_GAP_SEC) return; + if (approxMetersBetween(pending.lat, pending.lon, lat, lon) < AIS_TRACK_MIN_MOVE_M) return; + track = { + lats: new Float32Array(AIS_TRACK_SAMPLES), + lons: new Float32Array(AIS_TRACK_SAMPLES), + times: new Uint32Array(AIS_TRACK_SAMPLES), + head: 0, + len: 0, + }; + _aisStreamTracks.set(mmsi, track); + _aisStreamTrackPending.delete(mmsi); + writeAisTrackSample(track, pending.lat, pending.lon, pending.epochSec); + writeAisTrackSample(track, lat, lon, epochSec); + return; + } + + const lastIdx = (track.head - 1 + AIS_TRACK_SAMPLES) % AIS_TRACK_SAMPLES; + const lastEpoch = track.times[lastIdx]; + if (epochSec - lastEpoch < AIS_TRACK_MIN_GAP_SEC) return; + if (approxMetersBetween(track.lats[lastIdx], track.lons[lastIdx], lat, lon) < AIS_TRACK_MIN_MOVE_M) return; + writeAisTrackSample(track, lat, lon, epochSec); +} + +function writeAisTrackSample(track, lat, lon, epochSec) { + track.lats[track.head] = lat; + track.lons[track.head] = lon; + track.times[track.head] = epochSec; + track.head = (track.head + 1) % AIS_TRACK_SAMPLES; + track.len = Math.min(track.len + 1, AIS_TRACK_SAMPLES); +} + +/** + * Reads a vessel's accumulated track in chronological order. + * @returns {Array<{lat:number,lon:number,t:number}>} + */ +export function readAisTrack(mmsi) { + const track = _aisStreamTracks.get(mmsi); + if (!track || !track.len) return []; + const samples = []; + const start = (track.head - track.len + AIS_TRACK_SAMPLES) % AIS_TRACK_SAMPLES; + for (let i = 0; i < track.len; i++) { + const idx = (start + i) % AIS_TRACK_SAMPLES; + samples.push({ lat: track.lats[idx], lon: track.lons[idx], t: track.times[idx] }); + } + return samples; +} + +/** Equirectangular distance approximation — plenty for 25m thinning. */ +function approxMetersBetween(lat1, lon1, lat2, lon2) { + const dLat = (lat2 - lat1) * 111320; + const dLon = (lon2 - lon1) * 111320 * Math.cos(((lat1 + lat2) / 2) * (Math.PI / 180)); + return Math.hypot(dLat, dLon); +} + +function mergeAisStaticIntoLiveVessel(mmsi, staticData) { + const existing = _aisStreamVessels.get(mmsi); + if (!existing) return; + if (staticData.name && (!existing.name || existing.name === `MMSI ${mmsi}`)) existing.name = staticData.name; + if (staticData.type && !existing.type) existing.type = staticData.type; + if (staticData.destination && !existing.destination) existing.destination = staticData.destination; + if (staticData.imo && !existing.imo) existing.imo = staticData.imo; +} + +function vesselNameFromAis(metadata, message, staticData = {}) { + return stringValue( + metadata.ShipName + ?? message.Name + ?? message.ShipName + ?? message.ReportA?.Name + ?? staticData.name + ); +} + +function vesselTypeFromAis(message, staticData = {}) { + return stringValue( + message.Type + ?? message.ShipType + ?? message.ReportB?.ShipType + ?? staticData.type + ); +} + +export function aisStreamRows(maxRows) { + const cutoff = Date.now() - AISSTREAM_STALE_MS; + const rows = []; + for (const row of _aisStreamVessels.values()) { + if (row._updatedAt >= cutoff) rows.push(row); + } + rows.sort((a, b) => b._updatedAt - a._updatedAt); + return rows.slice(0, maxRows).map(({ _updatedAt, ...row }) => row); +} + +function pruneAisStreamCache() { + const cutoff = Date.now() - AISSTREAM_STALE_MS; + for (const [mmsi, row] of _aisStreamVessels) { + if (row._updatedAt < cutoff) { + _aisStreamVessels.delete(mmsi); + _aisStreamTracks.delete(mmsi); + _aisStreamTrackPending.delete(mmsi); + } + } + // Pending single-fix entries for vessels never seen again must not leak + const pendingCutoffSec = Math.floor(cutoff / 1000); + for (const [mmsi, pending] of _aisStreamTrackPending) { + if (pending.epochSec < pendingCutoffSec) _aisStreamTrackPending.delete(mmsi); + } + if (_aisStreamVessels.size <= AISSTREAM_CACHE_MAX) return; + const ordered = [..._aisStreamVessels.entries()].sort((a, b) => a[1]._updatedAt - b[1]._updatedAt); + for (const [mmsi] of ordered.slice(0, _aisStreamVessels.size - AISSTREAM_CACHE_MAX)) { + _aisStreamVessels.delete(mmsi); + _aisStreamTracks.delete(mmsi); + _aisStreamTrackPending.delete(mmsi); + } +} + +export function newestAisPositionAt(rows) { + return rows[0]?.last_position_UTC || null; +} + +function stringValue(value) { + if (value === undefined || value === null) return ''; + return String(value).trim(); +} + +function numberValue(value) { + if (value === null || value === undefined || value === '') return null; + const number = Number(value); + return Number.isFinite(number) ? number : null; +} + +function normalizedHeading(value) { + const heading = numberValue(value); + return heading !== null && heading >= 0 && heading <= 360 ? heading : null; +} + +function normalizeAisTimestamp(value) { + const text = stringValue(value); + if (!text) return new Date().toISOString(); + const normalized = text.replace(' +0000 UTC', 'Z').replace(' UTC', 'Z'); + const date = new Date(normalized); + return Number.isNaN(date.getTime()) ? new Date().toISOString() : date.toISOString(); +} diff --git a/src/proxyErrorResponses.test.mjs b/src/proxyErrorResponses.test.mjs index 17abf4a33..6372f8366 100644 --- a/src/proxyErrorResponses.test.mjs +++ b/src/proxyErrorResponses.test.mjs @@ -3,7 +3,9 @@ import assert from 'node:assert/strict'; import { readFileSync } from 'node:fs'; import path from 'node:path'; -const source = readFileSync(new URL('../server/providers/local.js', import.meta.url), 'utf8'); +const source = ['local.js', 'common/http.js', 'aircraft/enrichment.js'] + .map(file => readFileSync(new URL(`../server/providers/${file}`, import.meta.url), 'utf8')) + .join('\n'); const detail = 'fixture-secret-token /internal/example '; // Execute the production middleware with isolated upstreams and cache storage. diff --git a/src/tooling/liveProviders.test.mjs b/src/tooling/liveProviders.test.mjs new file mode 100644 index 000000000..d1e53e43c --- /dev/null +++ b/src/tooling/liveProviders.test.mjs @@ -0,0 +1,163 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { once } from 'node:events'; +import { WebSocketServer } from 'ws'; +import { setTimeout as delay } from 'node:timers/promises'; +import * as providers from '../../server/providers/live.js'; +import * as portable from '../../src/data/adsbLolFallback.js'; + +function install(plugin, preview = false) { + const routes = new Map(); + plugin[preview ? 'configurePreviewServer' : 'configureServer']({ + middlewares: { use(route, handler) { routes.set(route, handler); } }, + }); + return async (route, url = '/', method = 'GET') => { + assert.ok(routes.has(route), `registered route: ${route}`); + const response = { + statusCode: 200, headers: {}, + setHeader(key, value) { this.headers[key.toLowerCase()] = value; }, + writeHead(status, headers = {}) { + this.statusCode = status; + for (const [key, value] of Object.entries(headers)) this.setHeader(key, value); + }, + end(body) { this.body = body; }, + }; + await routes.get(route)({ url, method }, response); + return response; + }; +} + +function environment(t, values) { + for (const [key, value] of Object.entries(values)) { + const original = process.env[key]; + if (value === undefined) delete process.env[key]; else process.env[key] = value; + t.after(() => { if (original === undefined) delete process.env[key]; else process.env[key] = original; }); + } +} + +test('live entry resolves in Node and aircraft normalization stays independently portable', async () => { + const entry = await import('gods-eye-view/server/providers/live'); + assert.equal(entry.openSkyProxy, providers.openSkyProxy); + assert.equal(entry.aisLiveProxy, providers.aisLiveProxy); + const normalizer = await import('gods-eye-view/sources/adsb-lol'); + assert.equal(normalizer.normalizeAdsbLolAircraftState, portable.normalizeAdsbLolAircraftState); +}); + +test('OpenSky state and track routes share tokens, retain cache and use regional fallback', async t => { + environment(t, { + OPENSKY_CLIENT_ID: 'fixture-client', OPENSKY_CLIENT_SECRET: 'fixture-secret', + OPENSKY_AUTH_MODE: 'oauth', OPENSKY_USERNAME: undefined, OPENSKY_PASSWORD: undefined, + }); + let now = Date.now(); + t.mock.method(Date, 'now', () => now); + t.mock.method(console, 'log', () => {}); + const calls = []; + t.mock.method(globalThis, 'fetch', async (url, options) => { + calls.push({ url, options }); + if (url.includes('/token')) return Response.json({ access_token: 'fixture-token', expires_in: 1800 }); + if (url.includes('/states/')) { + assert.equal(options.headers.Authorization, 'Bearer fixture-token'); + return Response.json({ time: Math.floor(now / 1000), states: [] }); + } + if (url.includes('/tracks/')) { + assert.equal(options.headers.Authorization, 'Bearer fixture-token'); + return Response.json({ path: [] }); + } + if (url.includes('/lat/')) return Response.json({ now: now / 1000, ac: [{ hex: 'abc123', lat: 30, lon: -97, alt_baro: 10000 }] }); + throw Error(`Unexpected URL: ${url}`); + }); + const states = install(providers.openSkyProxy()); + assert.equal((await states('/api/opensky', '?lat=30&lon=-97')).statusCode, 200); + assert.equal((await states('/api/opensky', '?lat=30&lon=-97')).headers['x-opensky-cache'], 'HIT'); + assert.equal(calls.length, 2); + const tracks = install(providers.trackBackfillProxies(), true); + assert.equal((await tracks('/api/opensky-track', '?icao24=ABC123')).statusCode, 200); + await tracks('/api/opensky-track', '?icao24=abc123'); + assert.equal(calls.filter(call => call.url.includes('/token')).length, 1); + assert.equal(calls.filter(call => call.url.includes('/tracks/')).length, 1); + assert.equal((await tracks('/api/adsblol/trace', '?hex=invalid')).statusCode, 400); + now += 130_000; + t.mock.method(globalThis, 'fetch', async url => { + if (url.includes('/states/')) return new Response('', { status: 503 }); + if (url.includes('/lat/')) return Response.json({ now: now / 1000, ac: [{ hex: 'abc123', lat: 30, lon: -97, alt_baro: 10000 }] }); + throw Error(`Unexpected URL: ${url}`); + }); + // A fresh request without a usable cached worldwide frame should use the regional feed. + const fresh = await import(`../../server/providers/aircraft/opensky.js?fallback=${now}`); + environment(t, { OPENSKY_AUTH_MODE: 'anon' }); + const fallback = await install(fresh.openSkyProxy())('/api/opensky', '?lat=30&lon=-97'); + assert.equal(fallback.statusCode, 200); + assert.equal(fallback.headers['x-flight-source'], 'adsb.lol'); + assert.equal(JSON.parse(fallback.body).states[0][0], 'abc123'); +}); + +test('military aircraft route preserves fresh cache and stale response after upstream failure', async t => { + let now = Date.now(); + let calls = 0; + t.mock.method(Date, 'now', () => now); + t.mock.method(console, 'error', () => {}); + t.mock.method(globalThis, 'fetch', async () => { + if (++calls > 1) throw Error('offline'); + return Response.json({ ac: [{ hex: 'abc123' }] }); + }); + const request = install(providers.adsbLolProxy()); + const first = await request('/api/adsblol/mil'); + assert.equal((await request('/api/adsblol/mil')).body, first.body); + assert.equal(calls, 1); + now += 13_000; + assert.equal((await request('/api/adsblol/mil')).body, first.body); + assert.equal(calls, 2); +}); + +test('AIS preview route ingests through the socket, returns tracks and disposes before restart', async t => { + const upstream = new WebSocketServer({ host: '127.0.0.1', port: 0 }); + await once(upstream, 'listening'); + const sockets = []; + t.after(async () => { + for (const socket of sockets) socket.terminate(); + await new Promise(resolve => upstream.close(resolve)); + }); + environment(t, { + AISSTREAM_API_KEY: 'fixture-key', + AISSTREAM_URL: `ws://127.0.0.1:${upstream.address().port}`, + AISSTREAM_BOUNDING_BOXES: undefined, AISSTREAM_MESSAGE_TYPES: undefined, + AISSTREAM_SILENCE_TIMEOUT_MS: '0', + }); + upstream.on('connection', socket => { + sockets.push(socket); + socket.on('message', raw => { + assert.equal(JSON.parse(raw).APIKey, 'fixture-key'); + for (const [lat, epoch] of [[30, 1_800_000_000], [30.01, 1_800_000_060]]) { + socket.send(JSON.stringify({ + MessageType: 'PositionReport', + MetaData: { MMSI: 123456789, latitude: lat, longitude: -97, time_utc: new Date(epoch * 1000).toISOString() }, + Message: { PositionReport: { UserID: 123456789, Sog: 10, Cog: 90, TrueHeading: 511 } }, + })); + } + }); + }); + const plugin = providers.aisLiveProxy(); + t.after(() => plugin.closeBundle()); + const request = install(plugin, true); + let res; + for (let i = 0; i < 100; i++) { + res = await request('/api/ais-live'); + if (JSON.parse(res.body).rows.length) break; + await delay(10); + } + const data = JSON.parse(res.body); + assert.equal(data.status, 'live'); + assert.equal(data.rows[0].mmsi, '123456789'); + assert.equal(data.rows[0].heading, null); + const history = await request('/api/ais-live', '/track?mmsi=123456789'); + assert.equal(JSON.parse(history.body).samples.length, 2); + assert.equal((await request('/api/ais-live', '/track?mmsi=bad')).statusCode, 400); + assert.equal(sockets.length, 1); + plugin.closeBundle(); + for (let i = 0; i < 100 && sockets[0].readyState !== 3; i++) await delay(10); + assert.equal(sockets[0].readyState, 3); + const restarted = install(plugin); + await restarted('/api/ais-live'); + for (let i = 0; i < 100 && sockets.length < 2; i++) await delay(10); + assert.equal(sockets.length, 2); +});