mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-04 10:24:02 +00:00
Merge pull request #322: split aircraft and vessel server providers
This commit is contained in:
21 files changed
+2368
-1675
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
|
||||
+5
-1
@@ -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"
|
||||
}
|
||||
}
|
||||
@@ -31,5 +31,18 @@
|
||||
"build/vite.js",
|
||||
"server/standalone/vite.config.js",
|
||||
"vite.config.js",
|
||||
"src/tooling/viteBuild.test.mjs"
|
||||
"src/tooling/viteBuild.test.mjs",
|
||||
"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",
|
||||
"server/providers/live.js",
|
||||
"src/tooling/liveProviders.test.mjs",
|
||||
"src/data/adsbLolFallback.js",
|
||||
"src/data/aisStreamAdapter.js",
|
||||
"src/data/aisWatchdog.js"
|
||||
]
|
||||
@@ -27,5 +27,29 @@
|
||||
"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": []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
/**
|
||||
* 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' }));
|
||||
}
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
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' });
|
||||
}
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,699 @@
|
||||
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<string|null>|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<string|null>} 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<string,string>} 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' }));
|
||||
}
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
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=<hex6> — 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=<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<string, {at:number,status:number,body:string}>} */
|
||||
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);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
/**
|
||||
* 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 };
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 };
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
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';
|
||||
+10
-1592
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,406 @@
|
||||
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<typeof createAisStreamAdapter>|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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,288 @@
|
||||
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<string,object>} */
|
||||
const _aisStreamVessels = new Map();
|
||||
/** @type {Map<string,object>} */
|
||||
const _aisStreamStatic = new Map();
|
||||
/** @type {Map<string,{lats:Float32Array,lons:Float32Array,times:Uint32Array,head:number,len:number}>} mmsi -> track ring buffer */
|
||||
const _aisStreamTracks = new Map();
|
||||
/** @type {Map<string,{lat:number,lon:number,epochSec:number}>} 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();
|
||||
}
|
||||
@@ -9,7 +9,9 @@ function finiteNumber(value) {
|
||||
}
|
||||
|
||||
function emitterCategory(value) {
|
||||
const category = String(value || '').trim().toUpperCase();
|
||||
const category = String(value || '')
|
||||
.trim()
|
||||
.toUpperCase();
|
||||
const categories = {
|
||||
A1: 2,
|
||||
A2: 3,
|
||||
@@ -36,18 +38,24 @@ function emitterCategory(value) {
|
||||
* @returns {Array|null} OpenSky-compatible state vector, or null when invalid.
|
||||
*/
|
||||
export function normalizeAdsbLolAircraftState(aircraft, nowSeconds) {
|
||||
const hex = String(aircraft?.hex || '').trim().toLowerCase();
|
||||
const hex = String(aircraft?.hex || '')
|
||||
.trim()
|
||||
.toLowerCase();
|
||||
const latitude = finiteNumber(aircraft?.lat);
|
||||
const longitude = finiteNumber(aircraft?.lon);
|
||||
if (!hex || latitude === null || longitude === null) return null;
|
||||
|
||||
const seenPosition = Math.max(0, finiteNumber(aircraft?.seen_pos) ?? finiteNumber(aircraft?.seen) ?? 0);
|
||||
const seenPosition = Math.max(
|
||||
0,
|
||||
finiteNumber(aircraft?.seen_pos) ?? finiteNumber(aircraft?.seen) ?? 0,
|
||||
);
|
||||
const seen = Math.max(0, finiteNumber(aircraft?.seen) ?? seenPosition);
|
||||
const onGround = aircraft?.alt_baro === 'ground';
|
||||
const barometricFeet = onGround ? null : finiteNumber(aircraft?.alt_baro);
|
||||
const geometricFeet = finiteNumber(aircraft?.alt_geom);
|
||||
const groundSpeedKnots = finiteNumber(aircraft?.gs);
|
||||
const verticalRateFpm = finiteNumber(aircraft?.baro_rate) ?? finiteNumber(aircraft?.geom_rate);
|
||||
const verticalRateFpm =
|
||||
finiteNumber(aircraft?.baro_rate) ?? finiteNumber(aircraft?.geom_rate);
|
||||
const track = finiteNumber(aircraft?.track);
|
||||
|
||||
return [
|
||||
@@ -80,9 +88,12 @@ export function normalizeAdsbLolAircraftState(aircraft, nowSeconds) {
|
||||
*/
|
||||
export function normalizeAdsbLolPointResponse(payload) {
|
||||
const responseNow = finiteNumber(payload?.now);
|
||||
const nowSeconds = responseNow === null
|
||||
? Math.floor(Date.now() / 1000)
|
||||
: Math.floor(responseNow > 10_000_000_000 ? responseNow / 1000 : responseNow);
|
||||
const nowSeconds =
|
||||
responseNow === null
|
||||
? Math.floor(Date.now() / 1000)
|
||||
: Math.floor(
|
||||
responseNow > 10_000_000_000 ? responseNow / 1000 : responseNow,
|
||||
);
|
||||
const states = (Array.isArray(payload?.ac) ? payload.ac : [])
|
||||
.map((aircraft) => normalizeAdsbLolAircraftState(aircraft, nowSeconds))
|
||||
.filter(Boolean);
|
||||
|
||||
+103
-53
@@ -26,10 +26,12 @@ const DEFAULT_CLOCK = Object.freeze({
|
||||
});
|
||||
|
||||
/** Upstream text that identifies a credential rejection. */
|
||||
const AUTH_TEXT = /(unauthoriz|unauthoris|forbidden|invalid\s*api[\s_-]*key|invalid\s*key|bad\s*api[\s_-]*key|authentic|api\s*key\s*(is\s*)?(invalid|required|missing|not\s*valid))/i;
|
||||
const AUTH_TEXT =
|
||||
/(unauthoriz|unauthoris|forbidden|invalid\s*api[\s_-]*key|invalid\s*key|bad\s*api[\s_-]*key|authentic|api\s*key\s*(is\s*)?(invalid|required|missing|not\s*valid))/i;
|
||||
|
||||
/** Upstream text that identifies a rate limit / connection cap. */
|
||||
const RATE_TEXT = /(rate[\s_-]*limit|too\s*many\s*(requests|connections)|quota\s*exceeded|429)/i;
|
||||
const RATE_TEXT =
|
||||
/(rate[\s_-]*limit|too\s*many\s*(requests|connections)|quota\s*exceeded|429)/i;
|
||||
|
||||
/**
|
||||
* Parse an HTTP `Retry-After` header into milliseconds.
|
||||
@@ -61,7 +63,8 @@ export function parseRetryAfterMs(raw, nowMs = Date.now()) {
|
||||
*/
|
||||
export function classifyAisFailure(input = {}) {
|
||||
const text = String(input.message || '').trim();
|
||||
const fromHeader = () => parseRetryAfterMs(input.retryAfterHeader, input.nowMs);
|
||||
const fromHeader = () =>
|
||||
parseRetryAfterMs(input.retryAfterHeader, input.nowMs);
|
||||
|
||||
let status = Number(input.httpStatus);
|
||||
if (!Number.isFinite(status)) {
|
||||
@@ -72,7 +75,10 @@ export function classifyAisFailure(input = {}) {
|
||||
}
|
||||
|
||||
if (status === 401 || status === 403) {
|
||||
return { kind: 'auth', message: `AISStream rejected the API key (HTTP ${status})` };
|
||||
return {
|
||||
kind: 'auth',
|
||||
message: `AISStream rejected the API key (HTTP ${status})`,
|
||||
};
|
||||
}
|
||||
if (status === 429) {
|
||||
return {
|
||||
@@ -82,10 +88,14 @@ export function classifyAisFailure(input = {}) {
|
||||
};
|
||||
}
|
||||
if (Number.isFinite(status) && status >= 400) {
|
||||
return { kind: 'transport', message: `AISStream upgrade failed (HTTP ${status})` };
|
||||
return {
|
||||
kind: 'transport',
|
||||
message: `AISStream upgrade failed (HTTP ${status})`,
|
||||
};
|
||||
}
|
||||
if (AUTH_TEXT.test(text)) return { kind: 'auth', message: text };
|
||||
if (RATE_TEXT.test(text)) return { kind: 'rate-limit', message: text, retryAfterMs: fromHeader() };
|
||||
if (RATE_TEXT.test(text))
|
||||
return { kind: 'rate-limit', message: text, retryAfterMs: fromHeader() };
|
||||
return { kind: 'transport', message: text || 'AISStream websocket error' };
|
||||
}
|
||||
|
||||
@@ -151,7 +161,8 @@ export function aisFrameByteLength(data) {
|
||||
if (typeof data === 'string') return data.length;
|
||||
if (data instanceof ArrayBuffer) return data.byteLength;
|
||||
if (ArrayBuffer.isView(data)) return data.byteLength;
|
||||
if (Array.isArray(data)) return data.reduce((sum, part) => sum + aisFrameByteLength(part), 0);
|
||||
if (Array.isArray(data))
|
||||
return data.reduce((sum, part) => sum + aisFrameByteLength(part), 0);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -168,33 +179,35 @@ export function aisFrameByteLength(data) {
|
||||
*
|
||||
* If AISStream adds a type, add it here — do not infer it.
|
||||
*/
|
||||
export const AIS_RECOGNIZED_MESSAGE_TYPES = Object.freeze(new Set([
|
||||
'PositionReport',
|
||||
'UnknownMessage',
|
||||
'AddressedSafetyMessage',
|
||||
'AddressedBinaryMessage',
|
||||
'AidsToNavigationReport',
|
||||
'AssignedModeCommand',
|
||||
'BaseStationReport',
|
||||
'BinaryAcknowledge',
|
||||
'BinaryBroadcastMessage',
|
||||
'ChannelManagement',
|
||||
'CoordinatedUTCInquiry',
|
||||
'DataLinkManagementMessage',
|
||||
'DataLinkManagementMessageData',
|
||||
'ExtendedClassBPositionReport',
|
||||
'GroupAssignmentCommand',
|
||||
'GnssBroadcastBinaryMessage',
|
||||
'Interrogation',
|
||||
'LongRangeAisBroadcastMessage',
|
||||
'MultiSlotBinaryMessage',
|
||||
'SafetyBroadcastMessage',
|
||||
'ShipStaticData',
|
||||
'SingleSlotBinaryMessage',
|
||||
'StandardClassBPositionReport',
|
||||
'StandardSearchAndRescueAircraftReport',
|
||||
'StaticDataReport',
|
||||
]));
|
||||
export const AIS_RECOGNIZED_MESSAGE_TYPES = Object.freeze(
|
||||
new Set([
|
||||
'PositionReport',
|
||||
'UnknownMessage',
|
||||
'AddressedSafetyMessage',
|
||||
'AddressedBinaryMessage',
|
||||
'AidsToNavigationReport',
|
||||
'AssignedModeCommand',
|
||||
'BaseStationReport',
|
||||
'BinaryAcknowledge',
|
||||
'BinaryBroadcastMessage',
|
||||
'ChannelManagement',
|
||||
'CoordinatedUTCInquiry',
|
||||
'DataLinkManagementMessage',
|
||||
'DataLinkManagementMessageData',
|
||||
'ExtendedClassBPositionReport',
|
||||
'GroupAssignmentCommand',
|
||||
'GnssBroadcastBinaryMessage',
|
||||
'Interrogation',
|
||||
'LongRangeAisBroadcastMessage',
|
||||
'MultiSlotBinaryMessage',
|
||||
'SafetyBroadcastMessage',
|
||||
'ShipStaticData',
|
||||
'SingleSlotBinaryMessage',
|
||||
'StandardClassBPositionReport',
|
||||
'StandardSearchAndRescueAircraftReport',
|
||||
'StaticDataReport',
|
||||
]),
|
||||
);
|
||||
|
||||
/**
|
||||
* Resolve an envelope's MMSI, or null.
|
||||
@@ -221,9 +234,14 @@ export function aisEnvelopeMmsi(envelope, body = {}) {
|
||||
* @returns {boolean}
|
||||
*/
|
||||
export function isRecognizedAisEnvelope(envelope) {
|
||||
if (!envelope || typeof envelope !== 'object' || Array.isArray(envelope)) return false;
|
||||
if (!envelope || typeof envelope !== 'object' || Array.isArray(envelope))
|
||||
return false;
|
||||
const messageType = envelope.MessageType;
|
||||
if (typeof messageType !== 'string' || !AIS_RECOGNIZED_MESSAGE_TYPES.has(messageType)) return false;
|
||||
if (
|
||||
typeof messageType !== 'string' ||
|
||||
!AIS_RECOGNIZED_MESSAGE_TYPES.has(messageType)
|
||||
)
|
||||
return false;
|
||||
const body = envelope.Message?.[messageType];
|
||||
if (!body || typeof body !== 'object') return false;
|
||||
return aisEnvelopeMmsi(envelope, body) !== null;
|
||||
@@ -281,7 +299,10 @@ export function createAisStreamAdapter(options) {
|
||||
/** (Re)build the state machine, preserving the generation namespace. */
|
||||
function setWatchdogOptions(watchdogOptions = {}) {
|
||||
if (watchdog) {
|
||||
generationHighWater = Math.max(generationHighWater, watchdog.highWaterGeneration());
|
||||
generationHighWater = Math.max(
|
||||
generationHighWater,
|
||||
watchdog.highWaterGeneration(),
|
||||
);
|
||||
}
|
||||
watchdog = createAisWatchdog({
|
||||
...watchdogOptions,
|
||||
@@ -306,7 +327,9 @@ export function createAisStreamAdapter(options) {
|
||||
try {
|
||||
socket.terminate();
|
||||
} catch (error) {
|
||||
warn(`[AISStream] terminate(${reason || 'unknown'}) failed: ${error?.message || error}`);
|
||||
warn(
|
||||
`[AISStream] terminate(${reason || 'unknown'}) failed: ${error?.message || error}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -325,7 +348,8 @@ export function createAisStreamAdapter(options) {
|
||||
*/
|
||||
function runActions(owner, actions) {
|
||||
for (const action of actions || []) {
|
||||
if (action.type === 'terminate') terminateGeneration(action.generation, action.reason);
|
||||
if (action.type === 'terminate')
|
||||
terminateGeneration(action.generation, action.reason);
|
||||
else if (action.type === 'connect') openSocket(owner, action.generation);
|
||||
}
|
||||
}
|
||||
@@ -349,11 +373,18 @@ export function createAisStreamAdapter(options) {
|
||||
try {
|
||||
socket = createSocket(resolveUrl());
|
||||
} catch (error) {
|
||||
failGeneration(owner, generation, classifyAisFailure({ message: error?.message }));
|
||||
failGeneration(
|
||||
owner,
|
||||
generation,
|
||||
classifyAisFailure({ message: error?.message }),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (!socket) {
|
||||
failGeneration(owner, generation, { kind: 'transport', message: 'socket factory returned nothing' });
|
||||
failGeneration(owner, generation, {
|
||||
kind: 'transport',
|
||||
message: 'socket factory returned nothing',
|
||||
});
|
||||
return;
|
||||
}
|
||||
sockets.set(generation, socket);
|
||||
@@ -386,7 +417,11 @@ export function createAisStreamAdapter(options) {
|
||||
try {
|
||||
socket.send(JSON.stringify(buildSubscription()));
|
||||
} catch (error) {
|
||||
failGeneration(owner, generation, classifyAisFailure({ message: error?.message }));
|
||||
failGeneration(
|
||||
owner,
|
||||
generation,
|
||||
classifyAisFailure({ message: error?.message }),
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
@@ -408,7 +443,9 @@ export function createAisStreamAdapter(options) {
|
||||
return;
|
||||
}
|
||||
handleMessage(owner, generation, socket, data).catch((error) => {
|
||||
warn(`[AISStream] message handling failed: ${error?.message || error}`);
|
||||
warn(
|
||||
`[AISStream] message handling failed: ${error?.message || error}`,
|
||||
);
|
||||
});
|
||||
} catch (error) {
|
||||
// A frame that cannot even be decoded is a malformed frame: dropped,
|
||||
@@ -427,7 +464,9 @@ export function createAisStreamAdapter(options) {
|
||||
});
|
||||
try {
|
||||
response?.destroy?.();
|
||||
} catch { /* already gone */ }
|
||||
} catch {
|
||||
/* already gone */
|
||||
}
|
||||
releaseSocketEntry(generation, socket);
|
||||
abort(socket, 'unexpected-response');
|
||||
failGeneration(owner, generation, detail);
|
||||
@@ -440,10 +479,14 @@ export function createAisStreamAdapter(options) {
|
||||
abort(socket, 'orphan-error');
|
||||
return;
|
||||
}
|
||||
failGeneration(owner, generation, classifyAisFailure({
|
||||
message: error?.message || String(error || ''),
|
||||
nowMs: clock.wall(),
|
||||
}));
|
||||
failGeneration(
|
||||
owner,
|
||||
generation,
|
||||
classifyAisFailure({
|
||||
message: error?.message || String(error || ''),
|
||||
nowMs: clock.wall(),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
on('close', () => {
|
||||
@@ -482,10 +525,14 @@ export function createAisStreamAdapter(options) {
|
||||
if (parsed.kind === 'error') {
|
||||
// An error envelope is the OPPOSITE of liveness. Classifying it keeps a
|
||||
// rejected key out of the fast ladder.
|
||||
failGeneration(owner, generation, classifyAisFailure({
|
||||
message: parsed.message,
|
||||
nowMs: clock.wall(),
|
||||
}));
|
||||
failGeneration(
|
||||
owner,
|
||||
generation,
|
||||
classifyAisFailure({
|
||||
message: parsed.message,
|
||||
nowMs: clock.wall(),
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -504,7 +551,10 @@ export function createAisStreamAdapter(options) {
|
||||
/** Tear down sockets and machine state, preserving the generation namespace. */
|
||||
function dispose() {
|
||||
if (watchdog) {
|
||||
generationHighWater = Math.max(generationHighWater, watchdog.highWaterGeneration());
|
||||
generationHighWater = Math.max(
|
||||
generationHighWater,
|
||||
watchdog.highWaterGeneration(),
|
||||
);
|
||||
runActions(watchdog, watchdog.dispose());
|
||||
}
|
||||
for (const [generation, socket] of [...sockets]) {
|
||||
|
||||
+51
-20
@@ -63,7 +63,11 @@ export const AIS_WATCHDOG_STATUSES = Object.freeze([
|
||||
]);
|
||||
|
||||
/** Failure classes. Only 'transport' is allowed to walk the fast ladder. */
|
||||
export const AIS_FAILURE_KINDS = Object.freeze(['transport', 'auth', 'rate-limit']);
|
||||
export const AIS_FAILURE_KINDS = Object.freeze([
|
||||
'transport',
|
||||
'auth',
|
||||
'rate-limit',
|
||||
]);
|
||||
|
||||
/** Statuses that must not flip back to a hopeful 'connecting' on a retry. */
|
||||
const QUIET_TERMINAL = new Set(['down', 'auth-failed']);
|
||||
@@ -92,8 +96,8 @@ export function parseSilenceTimeoutEnv(raw, warn) {
|
||||
if (!text) return { kind: 'default' };
|
||||
if (!/^\d+(\.\d+)?$/.test(text)) {
|
||||
warn?.(
|
||||
`[AISStream] Ignoring AISSTREAM_SILENCE_TIMEOUT_MS="${raw}" (not a non-negative number); `
|
||||
+ `using the ${AIS_WATCHDOG_DEFAULTS.staleMs}ms default.`,
|
||||
`[AISStream] Ignoring AISSTREAM_SILENCE_TIMEOUT_MS="${raw}" (not a non-negative number); ` +
|
||||
`using the ${AIS_WATCHDOG_DEFAULTS.staleMs}ms default.`,
|
||||
);
|
||||
return { kind: 'default' };
|
||||
}
|
||||
@@ -132,11 +136,18 @@ export function createAisWatchdog(options = {}) {
|
||||
staleMs,
|
||||
positiveOr(options.recycleAfterMs, AIS_WATCHDOG_DEFAULTS.recycleAfterMs),
|
||||
);
|
||||
const backoffMs = Array.isArray(options.backoffMs) && options.backoffMs.length
|
||||
? options.backoffMs.map((ms) => positiveOr(ms, 5_000))
|
||||
: [...AIS_WATCHDOG_DEFAULTS.backoffMs];
|
||||
const downRetryMs = positiveOr(options.downRetryMs, AIS_WATCHDOG_DEFAULTS.downRetryMs);
|
||||
const authProbeMs = positiveOr(options.authProbeMs, AIS_WATCHDOG_DEFAULTS.authProbeMs);
|
||||
const backoffMs =
|
||||
Array.isArray(options.backoffMs) && options.backoffMs.length
|
||||
? options.backoffMs.map((ms) => positiveOr(ms, 5_000))
|
||||
: [...AIS_WATCHDOG_DEFAULTS.backoffMs];
|
||||
const downRetryMs = positiveOr(
|
||||
options.downRetryMs,
|
||||
AIS_WATCHDOG_DEFAULTS.downRetryMs,
|
||||
);
|
||||
const authProbeMs = positiveOr(
|
||||
options.authProbeMs,
|
||||
AIS_WATCHDOG_DEFAULTS.authProbeMs,
|
||||
);
|
||||
const clock = options.clock || DEFAULT_CLOCK;
|
||||
|
||||
let status = 'idle';
|
||||
@@ -184,7 +195,8 @@ export function createAisWatchdog(options = {}) {
|
||||
// unproven, and hammering on an unproven key is the behaviour this whole
|
||||
// state exists to prevent. Only valid data or a key rotation leaves
|
||||
// auth-failed.
|
||||
const effective = status === 'auth-failed' && kind !== 'auth' ? 'auth' : kind;
|
||||
const effective =
|
||||
status === 'auth-failed' && kind !== 'auth' ? 'auth' : kind;
|
||||
reconnectAttempt += 1;
|
||||
|
||||
if (effective === 'auth') {
|
||||
@@ -201,7 +213,8 @@ export function createAisWatchdog(options = {}) {
|
||||
reconnectAttempt = Math.max(reconnectAttempt, backoffMs.length);
|
||||
const wait = positiveOr(retryAfterMs, backoffMs[backoffMs.length - 1]);
|
||||
status = isExhausted() ? 'down' : 'reconnecting';
|
||||
nextAttemptMono = monoNow + (isExhausted() ? Math.max(wait, downRetryMs) : wait);
|
||||
nextAttemptMono =
|
||||
monoNow + (isExhausted() ? Math.max(wait, downRetryMs) : wait);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -251,7 +264,8 @@ export function createAisWatchdog(options = {}) {
|
||||
// A changed credential is the one event that can plausibly fix an auth
|
||||
// rejection, so it clears the terminal state immediately.
|
||||
const nextFingerprint = env.keyFingerprint ?? null;
|
||||
const credentialChanged = keyFingerprint !== null && nextFingerprint !== keyFingerprint;
|
||||
const credentialChanged =
|
||||
keyFingerprint !== null && nextFingerprint !== keyFingerprint;
|
||||
keyFingerprint = nextFingerprint;
|
||||
if (credentialChanged) {
|
||||
// Any socket in flight belongs to the OLD key — it subscribed with that
|
||||
@@ -353,9 +367,12 @@ export function createAisWatchdog(options = {}) {
|
||||
*/
|
||||
function onOpen(eventGeneration) {
|
||||
if (!ownsGeneration(eventGeneration)) {
|
||||
return [{ type: 'terminate', generation: eventGeneration, reason: 'orphan' }];
|
||||
return [
|
||||
{ type: 'terminate', generation: eventGeneration, reason: 'orphan' },
|
||||
];
|
||||
}
|
||||
if (!QUIET_TERMINAL.has(status) && status !== 'stale') status = 'connecting';
|
||||
if (!QUIET_TERMINAL.has(status) && status !== 'stale')
|
||||
status = 'connecting';
|
||||
return [];
|
||||
}
|
||||
|
||||
@@ -367,7 +384,9 @@ export function createAisWatchdog(options = {}) {
|
||||
*/
|
||||
function onMessage(eventGeneration) {
|
||||
if (!ownsGeneration(eventGeneration)) {
|
||||
return [{ type: 'terminate', generation: eventGeneration, reason: 'orphan' }];
|
||||
return [
|
||||
{ type: 'terminate', generation: eventGeneration, reason: 'orphan' },
|
||||
];
|
||||
}
|
||||
const monoNow = clock.mono();
|
||||
silenceSinceMono = monoNow;
|
||||
@@ -397,7 +416,9 @@ export function createAisWatchdog(options = {}) {
|
||||
*/
|
||||
function onFailure(eventGeneration, detail = {}) {
|
||||
if (!ownsGeneration(eventGeneration)) return [];
|
||||
const kind = AIS_FAILURE_KINDS.includes(detail.kind) ? detail.kind : 'transport';
|
||||
const kind = AIS_FAILURE_KINDS.includes(detail.kind)
|
||||
? detail.kind
|
||||
: 'transport';
|
||||
// A probe dying of some unrelated fault while the key is refused keeps the
|
||||
// credential message — surfacing "ECONNRESET" would send the operator
|
||||
// chasing the network instead of the key.
|
||||
@@ -433,7 +454,8 @@ export function createAisWatchdog(options = {}) {
|
||||
function snapshot() {
|
||||
const monoNow = clock.mono();
|
||||
const wallNow = clock.wall();
|
||||
const silentForMs = owned !== null ? Math.max(0, monoNow - silenceSinceMono) : null;
|
||||
const silentForMs =
|
||||
owned !== null ? Math.max(0, monoNow - silenceSinceMono) : null;
|
||||
const waitingMs = Math.max(0, nextAttemptMono - monoNow);
|
||||
return {
|
||||
status,
|
||||
@@ -442,9 +464,12 @@ export function createAisWatchdog(options = {}) {
|
||||
silentForMs,
|
||||
reconnectAttempt,
|
||||
// Projected onto the wall clock purely for display.
|
||||
nextAttemptAt: (status === 'reconnecting' || status === 'down' || status === 'auth-failed')
|
||||
? wallNow + waitingMs
|
||||
: null,
|
||||
nextAttemptAt:
|
||||
status === 'reconnecting' ||
|
||||
status === 'down' ||
|
||||
status === 'auth-failed'
|
||||
? wallNow + waitingMs
|
||||
: null,
|
||||
watchdog: silenceWatchArmed ? 'armed' : 'custom-subscription-off',
|
||||
staleAfterMs: staleMs,
|
||||
};
|
||||
@@ -452,7 +477,13 @@ export function createAisWatchdog(options = {}) {
|
||||
|
||||
/** Test/diagnostic view of internal ownership. */
|
||||
function debugState() {
|
||||
return { status, owned, generation, reconnectAttempt, lastMessageAt: lastMessageWall };
|
||||
return {
|
||||
status,
|
||||
owned,
|
||||
generation,
|
||||
reconnectAttempt,
|
||||
lastMessageAt: lastMessageWall,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
|
||||
@@ -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 <html>';
|
||||
|
||||
// Execute the production middleware with isolated upstreams and cache storage.
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
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}`
|
||||
);
|
||||
process.env.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, Math.floor(Date.now() / 1000) - 120],
|
||||
[30.01, Math.floor(Date.now() / 1000) - 60],
|
||||
]) {
|
||||
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);
|
||||
});
|
||||
Reference in new issue
Block a user