Merge main (82022306) into fix/flights-unpositioned-rows-complete

Resolve conflicts with main by keeping both sides:
- CHANGELOG.md: keep this PR's flights entry above main's newer entries.
- src/layers/flights/records.test.mjs: keep this PR's position-less
  retention test alongside main's civil emergency-status test.

src/sources/live/aircraft.js merged cleanly: openSkySnapshot keeps main's
emergencyByIcao normalizer and passes openSkyUnpositionedId to admitRecords.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/sources/gbfsStations.js

@@ -50,6 +50,26 @@ export function extractStationsArray(payload) {
   return [];
 }
 
+/** Read legacy text or a GBFS v3 Localized String, preferring usable English text. */
+function stationName(value) {
+  if (typeof value === 'string') return value.trim();
+  if (!Array.isArray(value)) return '';
+
+  let fallback = '';
+  for (const translation of value) {
+    if (typeof translation?.text !== 'string') continue;
+    const text = translation.text.trim();
+    if (!text) continue;
+    if (!fallback) fallback = text;
+    if (
+      typeof translation.language === 'string' &&
+      /^en(?:-|$)/i.test(translation.language.trim())
+    )
+      return text;
+  }
+  return fallback;
+}
+
 /**
  * Parse a GBFS station_information payload into a Map of station metadata.
  * Handles field-name variations across different GBFS providers
@@ -63,7 +83,7 @@ export function parseStationInformation(payload) {
   const stationMap = new Map();
 
   for (const raw of stations) {
-    // Accept both GBFS 2.x (station_id) and alternate (id) field names
+    // Accept both GBFS (station_id) and alternate (id) field names
     const stationId = String(raw?.station_id ?? raw?.id ?? '').trim();
     const lat = Number(raw?.lat ?? raw?.latitude);
     const lon = Number(raw?.lon ?? raw?.longitude);
@@ -71,7 +91,7 @@ export function parseStationInformation(payload) {
 
     stationMap.set(stationId, {
       stationId,
-      name: String(raw?.name || raw?.short_name || '').trim(),
+      name: stationName(raw?.name) || stationName(raw?.short_name),
       lat,
       lon,
       capacity: toNonNegativeInteger(raw?.capacity),

src/sources/gbfsStations.test.mjs

@@ -0,0 +1,199 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { parseStationInformation } from './gbfsStations.js';
+
+const station = (fields = {}) => ({
+  station_id: 's1',
+  lat: 30.2681,
+  lon: -97.7469,
+  capacity: 13,
+  ...fields,
+});
+const parseName = (fields, version) =>
+  parseStationInformation({
+    version,
+    data: { stations: [station(fields)] },
+  }).get('s1').name;
+
+test('GBFS v2 station names remain trimmed strings', () => {
+  assert.equal(
+    parseName({ name: ' Republic Square ', short_name: 'Short' }, '2.3'),
+    'Republic Square',
+  );
+});
+
+test('GBFS v3 station names prefer English over the first translation', () => {
+  assert.equal(
+    parseName(
+      {
+        name: [
+          { language: 'hu', text: 'Köztársaság tér' },
+          { language: 'en', text: ' Republic Square ' },
+        ],
+      },
+      '3.0',
+    ),
+    'Republic Square',
+  );
+});
+
+test('English preference accepts case differences and regional language tags', () => {
+  for (const language of ['EN', 'en-GB', 'En-US']) {
+    assert.equal(
+      parseName({
+        name: [
+          { language: 'hu', text: 'Hungarian' },
+          { language, text: 'English' },
+        ],
+      }),
+      'English',
+    );
+  }
+});
+
+test('without usable English text the first usable translation wins', () => {
+  assert.equal(
+    parseName({
+      name: [
+        null,
+        { language: 'en', text: '  ' },
+        { language: 'en', text: 42 },
+        { language: 'hu', text: ' Első ' },
+        { language: 'fr', text: 'Deuxième' },
+      ],
+    }),
+    'Első',
+  );
+});
+
+test('malformed language fields cannot abort translation selection', () => {
+  const name = [
+    null,
+    false,
+    'bad',
+    {},
+    { language: 42, text: 'Fallback' },
+    { language: { unexpected: true }, text: 'Other' },
+    { language: 'en', text: 'English' },
+  ];
+  assert.equal(parseName({ name }), 'English');
+  assert.equal(
+    parseName({ name: [{ text: ' Usable text without a locale ' }] }),
+    'Usable text without a locale',
+  );
+});
+
+test('a usable name takes precedence over an English short name', () => {
+  assert.equal(
+    parseName({
+      name: [{ language: 'hu', text: 'Full name' }],
+      short_name: [{ language: 'en', text: 'Short name' }],
+    }),
+    'Full name',
+  );
+});
+
+test('missing, empty and malformed names fall back to localized short_name', () => {
+  for (const name of [
+    undefined,
+    null,
+    '',
+    '  ',
+    [],
+    {},
+    0,
+    42,
+    false,
+    [null, {}, { language: 'en', text: [] }, { language: 'hu', text: ' ' }],
+  ]) {
+    assert.equal(
+      parseName({
+        name,
+        short_name: [
+          { language: 'hu', text: 'Rövid' },
+          { language: 'en', text: ' Short ' },
+        ],
+      }),
+      'Short',
+    );
+  }
+});
+
+test('localized names can fall back to legacy short_name strings', () => {
+  assert.equal(parseName({ name: [], short_name: ' Short ' }), 'Short');
+});
+
+test('no usable name leaves the station available with an empty name', () => {
+  for (const value of [
+    undefined,
+    null,
+    '',
+    [],
+    {},
+    42,
+    true,
+    [null, { text: false }, { language: 'en', text: ' ' }],
+  ]) {
+    assert.equal(parseName({ name: value, short_name: value }), '');
+  }
+});
+
+test('name shape is normalized even when a feed omits or misstates its version', () => {
+  const name = [{ language: 'en', text: 'Republic Square' }];
+  for (const version of [undefined, '2.3', '3.0', '3.1', 'unknown']) {
+    assert.equal(parseName({ name }, version), 'Republic Square');
+    assert.equal(parseName({ name: 'Legacy name' }, version), 'Legacy name');
+  }
+});
+
+test('bad names do not lose valid stations or their metadata', () => {
+  const information = parseStationInformation({
+    version: '3.0',
+    data: {
+      stations: [
+        station({ station_id: 'bad', name: [null, {}] }),
+        station({
+          station_id: 'short',
+          name: [],
+          short_name: [{ language: 'en', text: 'Short' }],
+        }),
+        station({
+          station_id: 'good',
+          name: [{ language: 'en', text: 'Republic Square' }],
+        }),
+      ],
+    },
+  });
+  assert.equal(information.size, 3);
+  assert.equal(information.get('bad').name, '');
+  assert.equal(information.get('short').name, 'Short');
+  assert.deepEqual(information.get('good'), {
+    stationId: 'good',
+    name: 'Republic Square',
+    lat: 30.2681,
+    lon: -97.7469,
+    capacity: 13,
+    isInstalled: true,
+    isRenting: true,
+    isReturning: true,
+  });
+});
+
+test('normalizing a localized name leaves the feed document unchanged', () => {
+  const payload = {
+    version: '3.0',
+    data: {
+      stations: [
+        station({
+          name: [
+            { language: 'hu', text: 'Magyar' },
+            { language: 'en', text: ' English ' },
+          ],
+        }),
+      ],
+    },
+  };
+  const original = structuredClone(payload);
+  parseStationInformation(payload);
+  assert.deepEqual(payload, original);
+});

src/sources/httpBody.js

@@ -55,7 +55,9 @@ export async function readResponseJsonCapped(response, maxBytes, signal) {
 
 /**
  * Read a fetch() Response body as bytes with the same hard cap as
- * readResponseTextCapped — for protobuf upstreams (GTFS-Realtime).
+ * readResponseTextCapped — for protobuf and tile upstreams (GTFS-Realtime,
+ * TomTom flow). Same early Content-Length rejection, running cap while
+ * streaming and cancellation on abort, so callers can treat the two alike.
  * Throws { code:'RESPONSE_TOO_LARGE' }.
  */
 export async function readResponseBytesCapped(response, maxBytes, signal) {

src/sources/httpBody.test.mjs

@@ -0,0 +1,133 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { readResponseBytesCapped } from './httpBody.js';
+
+/**
+ * A Response that streams `chunk` forever until the reader cancels it. The
+ * zero high-water mark keeps the stream from pulling eagerly at construction,
+ * so `pulls()` counts only what a reader actually asked for.
+ */
+function endlessResponse(chunk, onCancel, headers = {}) {
+  let pulls = 0;
+  const stream = new ReadableStream(
+    {
+      pull(controller) {
+        pulls += 1;
+        controller.enqueue(chunk);
+      },
+      cancel: onCancel,
+    },
+    { highWaterMark: 0 },
+  );
+  return { response: new Response(stream, { headers }), pulls: () => pulls };
+}
+
+function chunkedResponse(chunks, headers = {}) {
+  const stream = new ReadableStream(
+    {
+      start(controller) {
+        for (const chunk of chunks) controller.enqueue(chunk);
+        controller.close();
+      },
+    },
+    { highWaterMark: 0 },
+  );
+  return new Response(stream, { headers });
+}
+
+const bytes = (n, fill) => new Uint8Array(n).fill(fill);
+
+test('returns the exact bytes of a body that fits, untouched by text decoding', async () => {
+  // 0xFF and 0xFE are invalid UTF-8 lead bytes; a text path would mangle them.
+  const payload = Uint8Array.from([0xff, 0xfe, 0x00, 0x7f, 0x80, 0x01]);
+  const buf = await readResponseBytesCapped(chunkedResponse([payload]), 1024);
+  assert.ok(buf instanceof Uint8Array);
+  assert.deepEqual([...buf], [...payload]);
+});
+
+test('joins chunks in order and reports the total length', async () => {
+  const buf = await readResponseBytesCapped(
+    chunkedResponse([bytes(3, 1), bytes(2, 2), bytes(1, 3)]),
+    1024,
+  );
+  assert.deepEqual([...buf], [1, 1, 1, 2, 2, 3]);
+});
+
+test('refuses on a declared Content-Length without pulling the body', async () => {
+  let cancelled = false;
+  const { response, pulls } = endlessResponse(
+    bytes(512, 0),
+    () => {
+      cancelled = true;
+    },
+    { 'content-length': '4096' },
+  );
+  await assert.rejects(readResponseBytesCapped(response, 1024), {
+    code: 'RESPONSE_TOO_LARGE',
+  });
+  assert.equal(pulls(), 0);
+  assert.ok(cancelled);
+});
+
+test('refuses a chunked body the moment it passes the cap and stops pulling', async () => {
+  let cancelled = false;
+  const { response, pulls } = endlessResponse(bytes(512, 0), () => {
+    cancelled = true;
+  });
+  await assert.rejects(readResponseBytesCapped(response, 1024), {
+    code: 'RESPONSE_TOO_LARGE',
+  });
+  // Three 512-byte chunks is the first total over 1024; an uncapped read of
+  // this never-ending stream would not terminate at all.
+  assert.ok(pulls() <= 3, `pulled ${pulls()} chunks`);
+  assert.ok(cancelled);
+});
+
+test('accepts a body exactly at the cap', async () => {
+  const buf = await readResponseBytesCapped(
+    chunkedResponse([bytes(512, 7), bytes(512, 7)]),
+    1024,
+  );
+  assert.equal(buf.byteLength, 1024);
+});
+
+test('cancels a body still streaming when the signal aborts, and never returns a partial one', async () => {
+  const controller = new AbortController();
+  let cancelled = false;
+  let pulls = 0;
+  const stream = new ReadableStream(
+    {
+      pull(ctrl) {
+        pulls += 1;
+        ctrl.enqueue(bytes(64, 0));
+        if (pulls === 2) controller.abort();
+      },
+      cancel() {
+        cancelled = true;
+      },
+    },
+    { highWaterMark: 0 },
+  );
+  await assert.rejects(
+    readResponseBytesCapped(new Response(stream), 1 << 20, controller.signal),
+  );
+  assert.ok(cancelled);
+  assert.ok(pulls <= 3, `pulled ${pulls} chunks after abort`);
+});
+
+test('refuses immediately when the signal is already aborted', async () => {
+  const controller = new AbortController();
+  controller.abort();
+  const { response, pulls } = endlessResponse(bytes(8, 0), () => {});
+  await assert.rejects(
+    readResponseBytesCapped(response, 1024, controller.signal),
+  );
+  assert.equal(pulls(), 0);
+});
+
+test('falls back to arrayBuffer() for a body with no stream reader', async () => {
+  const response = new Response(bytes(16, 9));
+  const buf = await readResponseBytesCapped(response, 1024);
+  assert.equal(buf.byteLength, 16);
+  assert.equal(buf[0], 9);
+});

src/sources/live/aircraft.js

@@ -1,3 +1,4 @@
+import { normalizeSquawk } from '../../data/aircraftEmergency.js';
 import {
   admitRecords,
   cleanText,
@@ -32,6 +33,9 @@ export function normalizeOpenSkyAircraft(row) {
     typeCode: null,
     registration: null,
     operator: null,
+    squawk: normalizeSquawk(row[14]),
+    // State vectors carry no ADS-B emergency field; the squawk stands alone.
+    emergency: undefined,
   };
 }
 
@@ -70,6 +74,8 @@ export function normalizeReadsbAircraft(row, snapshotTimeMs) {
     typeCode: cleanText(row.t),
     registration: cleanText(row.r),
     operator: cleanText(row.ownOp),
+    squawk: normalizeSquawk(row.squawk),
+    emergency: cleanText(row.emergency).toLowerCase() || null,
   };
 }
 
@@ -94,7 +100,13 @@ export function openSkySnapshot(
 ) {
   const admitted = admitRecords(
     payload?.states,
-    normalizeOpenSkyAircraft,
+    (row) => {
+      const record = normalizeOpenSkyAircraft(row);
+      const statuses = payload?.emergencyByIcao;
+      if (record && statuses && Object.hasOwn(statuses, record.id))
+        record.emergency = cleanText(statuses[record.id]).toLowerCase() || null;
+      return record;
+    },
     'OpenSky',
     openSkyUnpositionedId,
   );

src/sources/live/contract.test.mjs

@@ -357,3 +357,23 @@ test('identity lookup validates its response and honors body-parse cancellation'
     name: 'AbortError',
   });
 });
+
+test('squawks are validated and omitted OpenSky status differs from an explicit ADS-B clear', () => {
+  const squawking = [...aircraft, '7700'];
+  const civil = openSkySnapshot({ states: [squawking] }, { now }).records[0];
+  assert.equal(civil.squawk, '7700');
+  assert.equal(civil.emergency, undefined);
+  assert.equal(openSkySnapshot({ states: [aircraft] }).records[0].squawk, null);
+  const mil = normalizeReadsbAircraft(
+    { hex: 'abc123', lat: 30, lon: -97, squawk: '7600', emergency: 'NORDO' },
+    now,
+  );
+  assert.equal(mil.squawk, '7600');
+  assert.equal(mil.emergency, 'nordo');
+  const quiet = normalizeReadsbAircraft(
+    { hex: 'abc123', lat: 30, lon: -97 },
+    now,
+  );
+  assert.equal(quiet.squawk, null);
+  assert.equal(quiet.emergency, null);
+});

src/sources/reference.js

@@ -1,12 +1,16 @@
 import { createUsgsEarthquakeSource } from '../layers/earthquakes/source.js';
+import { createEcccAirQualitySource } from '../layers/airQuality/source.js';
 import { createWfigsPerimeterSource } from '../layers/perimeters/source.js';
 import { createBundledCableSource } from '../layers/submarineCables/bundledSource.js';
+import { createAdsbGnssSource } from '../layers/gnss/source.js';
 
 /** Construct the existing reference feeds independently of application setup. */
 export function createReferenceSources() {
   return {
     earthquakes: createUsgsEarthquakeSource(),
+    'air-quality': createEcccAirQualitySource(),
     'fire-perimeters': createWfigsPerimeterSource(),
     cables: createBundledCableSource(),
+    'gnss-integrity': createAdsbGnssSource(),
   };
 }

src/sources/reference.test.mjs

@@ -14,14 +14,20 @@ test('reference factories retain compatibility without starting acquisition or s
   const second = createReferenceSources();
   assert.deepEqual(Object.keys(first), [
     'earthquakes',
+    'air-quality',
     'fire-perimeters',
     'cables',
+    'gnss-integrity',
   ]);
   assert.notEqual(first.earthquakes, second.earthquakes);
+  assert.notEqual(first['air-quality'], second['air-quality']);
   assert.notEqual(first['fire-perimeters'], second['fire-perimeters']);
   assert.notEqual(first.cables, second.cables);
+  assert.notEqual(first['gnss-integrity'], second['gnss-integrity']);
   assert.equal(typeof first.earthquakes.getSnapshot, 'function');
+  assert.equal(typeof first['air-quality'].getSnapshot, 'function');
   assert.equal(typeof first['fire-perimeters'].getSnapshot, 'function');
   assert.equal(typeof first.cables.fetch, 'function');
+  assert.equal(typeof first['gnss-integrity'].getSnapshot, 'function');
   assert.equal(requests, 0);
 });