style: format extracted live provider modules

This commit is contained in:
Sameh Khamis committed 2026-09-11 19:06:10 -07:00
1 parent 0e221441b4
commit f9a85528fe
14 files changed
+778 -362

No files matched your search

+18 -7
View File
@@ -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
View File
@@ -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
View File
@@ -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 {