Aggregate frontend snapshots across views (#447)

2026-08-07 09:22:53 +02:00 · 2025-11-13 22:02:42 +01:00
parent 382e2609c9
commit 631455237f
6 changed files with 528 additions and 55 deletions
@@ -45,7 +45,7 @@ function createResponse(status, body) {
 test('refreshNodeInformation merges telemetry metrics when the base node lacks them', async () => {
  const calls = [];
  const responses = new Map([
-    ['/api/nodes/!test', createResponse(200, {
+    ['/api/nodes/!test?limit=7', createResponse(200, {
      node_id: '!test',
      short_name: 'TST',
      battery_level: null,
@@ -53,14 +53,14 @@ test('refreshNodeInformation merges telemetry metrics when the base node lacks t
      modem_preset: 'MediumFast',
      lora_freq: '868.1',
    })],
-    ['/api/telemetry/!test?limit=1', createResponse(200, [{
+    ['/api/telemetry/!test?limit=7', createResponse(200, [{
      node_id: '!test',
      battery_level: 73.5,
      rx_time: 1_200,
      telemetry_time: 1_180,
      voltage: 4.1,
    }])],
-    ['/api/positions/!test?limit=1', createResponse(200, [{
+    ['/api/positions/!test?limit=7', createResponse(200, [{
      node_id: '!test',
      latitude: 52.5,
      longitude: 13.4,
@@ -115,12 +115,12 @@ test('refreshNodeInformation merges telemetry metrics when the base node lacks t

 test('refreshNodeInformation preserves fallback metrics when telemetry is unavailable', async () => {
  const responses = new Map([
-    ['/api/nodes/42', createResponse(200, {
+    ['/api/nodes/42?limit=7', createResponse(200, {
      node_id: '!num',
      short_name: 'NUM',
    })],
-    ['/api/telemetry/42?limit=1', createResponse(404, { error: 'not found' })],
-    ['/api/positions/42?limit=1', createResponse(404, { error: 'not found' })],
+    ['/api/telemetry/42?limit=7', createResponse(404, { error: 'not found' })],
+    ['/api/positions/42?limit=7', createResponse(404, { error: 'not found' })],
    ['/api/neighbors/42?limit=1000', createResponse(404, { error: 'not found' })],
  ]);
  const fetchImpl = async (url, options) => {
@@ -147,15 +147,15 @@ test('refreshNodeInformation requires a node identifier', async () => {

 test('refreshNodeInformation handles missing node records by falling back to telemetry data', async () => {
  const responses = new Map([
-    ['/api/nodes/!missing', createResponse(404, { error: 'not found' })],
-    ['/api/telemetry/!missing?limit=1', createResponse(200, [{
+    ['/api/nodes/!missing?limit=7', createResponse(404, { error: 'not found' })],
+    ['/api/telemetry/!missing?limit=7', createResponse(200, [{
      node_id: '!missing',
      node_num: 77,
      battery_level: 66,
      rx_time: 2_000,
      telemetry_time: 1_950,
    }])],
-    ['/api/positions/!missing?limit=1', createResponse(200, [{
+    ['/api/positions/!missing?limit=7', createResponse(200, [{
      node_id: '!missing',
      latitude: 1.23,
      longitude: 3.21,
@@ -0,0 +1,121 @@
+/*
+ * Copyright © 2025-26 l5yth & contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+import {
+  SNAPSHOT_WINDOW,
+  aggregateSnapshots,
+  aggregateNodeSnapshots,
+  aggregateTelemetrySnapshots,
+  aggregatePositionSnapshots,
+  aggregateNeighborSnapshots,
+} from '../snapshot-aggregator.js';
+
+const SAMPLE_NODE_ID = '!node';
+
+function keyById(entry) {
+  return entry && typeof entry.id === 'string' ? entry.id : null;
+}
+
+test('aggregateSnapshots merges snapshots in chronological order', () => {
+  const snapshots = [
+    { id: 'alpha', metric: null, label: 'latest', ts: 30, ignored: Number.NaN },
+    { id: 'alpha', metric: 5, label: null, ts: 20 },
+    { id: 'alpha', metric: 1, legacy: 'keep', ts: 10 },
+  ];
+  const aggregated = aggregateSnapshots(snapshots, { keySelector: keyById, limit: 3 });
+  assert.equal(aggregated.length, 1);
+  const record = aggregated[0];
+  assert.equal(record.metric, 5);
+  assert.equal(record.label, 'latest');
+  assert.equal(record.legacy, 'keep');
+  assert.deepEqual(record.snapshots.map(s => s.ts), [10, 20, 30]);
+  assert.equal(record.latestSnapshot.ts, 30);
+  assert.equal(Object.prototype.propertyIsEnumerable.call(record, 'snapshots'), false);
+  assert.equal(Object.prototype.propertyIsEnumerable.call(record, 'latestSnapshot'), false);
+  assert.equal('ignored' in record, false);
+});
+
+test('aggregateSnapshots enforces key selectors and respects limits', () => {
+  assert.throws(() => aggregateSnapshots([{ id: 'noop' }], {}), /keySelector/);
+  const snapshots = [
+    { id: 'beta', value: 'newest', ts: 30 },
+    { id: 'beta', value: 'mid', ts: 20 },
+    { id: 'beta', value: 'oldest', ts: 10 },
+  ];
+  const aggregated = aggregateSnapshots(snapshots, { keySelector: keyById, limit: 2 });
+  assert.equal(aggregated[0].snapshots.length, 2);
+  assert.deepEqual(aggregated[0].snapshots.map(s => s.ts), [20, 30]);
+});
+
+test('aggregateNodeSnapshots reconciles identifiers and fills missing values', () => {
+  const entries = [
+    { nodeId: SAMPLE_NODE_ID, voltage: 4.2, battery_level: null, rx_time: 250 },
+    { node_id: SAMPLE_NODE_ID, node_num: 42, battery_level: 20, rx_time: 200 },
+    { node_num: 42, short_name: 'Legacy', battery_level: 15, rx_time: 100 },
+  ];
+  const aggregated = aggregateNodeSnapshots(entries, { limit: SNAPSHOT_WINDOW });
+  assert.equal(aggregated.length, 1);
+  const node = aggregated[0];
+  assert.equal(node.node_id, SAMPLE_NODE_ID);
+  assert.equal(node.node_num, 42);
+  assert.equal(node.short_name, 'Legacy');
+  assert.equal(node.battery_level, 20);
+  assert.equal(node.voltage, 4.2);
+  assert.equal(node.snapshots.length, 3);
+});
+
+test('aggregateTelemetrySnapshots and aggregatePositionSnapshots mirror node aggregation', () => {
+  const telemetryEntries = [
+    { node_id: SAMPLE_NODE_ID, node_num: 5, temperature: null, rx_time: 20 },
+    { node_num: 5, temperature: 21.5, humidity: 52, rx_time: 10 },
+  ];
+  const positionEntries = [
+    { node_id: SAMPLE_NODE_ID, node_num: 5, longitude: 13.4, rx_time: 25 },
+    { node_num: 5, latitude: 52.5, rx_time: 15 },
+  ];
+  const telemetryAggregated = aggregateTelemetrySnapshots(telemetryEntries, { limit: 3 });
+  const positionAggregated = aggregatePositionSnapshots(positionEntries, { limit: 3 });
+  assert.equal(telemetryAggregated.length, 1);
+  assert.equal(positionAggregated.length, 1);
+  assert.equal(telemetryAggregated[0].temperature, 21.5);
+  assert.equal(telemetryAggregated[0].humidity, 52);
+  assert.equal(positionAggregated[0].latitude, 52.5);
+  assert.equal(positionAggregated[0].longitude, 13.4);
+});
+
+test('aggregateNeighborSnapshots groups by node pairs', () => {
+  const neighborSnapshots = [
+    { node_id: '!src', node_num: 101, neighbor_id: '!dst', neighbor_num: 202, snr: null, rx_time: 180 },
+    { node_id: '!src', node_num: 101, neighbor_id: '!dst', neighbor_num: 202, snr: -5, rx_time: 150 },
+    { node_num: 101, neighbor_num: 202, snr: -11, rx_time: 100 },
+    null,
+  ];
+  const aggregated = aggregateNeighborSnapshots(neighborSnapshots, { limit: 5 });
+  assert.equal(aggregated.length, 1);
+  const connection = aggregated[0];
+  assert.equal(connection.node_id, '!src');
+  assert.equal(connection.neighbor_id, '!dst');
+  assert.equal(connection.snr, -5);
+  assert.equal(connection.snapshots.length, 3);
+});
+
+test('aggregateSnapshots returns an empty array when no entries are provided', () => {
+  assert.deepEqual(aggregateSnapshots(null, { keySelector: () => 'noop' }), []);
+  assert.deepEqual(aggregateNodeSnapshots([], {}), []);
+});
@@ -42,6 +42,23 @@ export const CHAT_LOG_ENTRY_TYPES = Object.freeze({
  MESSAGE_ENCRYPTED: 'message-encrypted'
 });

+/**
+ * Resolve the chronological snapshots associated with an aggregated entry.
+ *
+ * @param {*} entry Candidate snapshot or aggregate.
+ * @returns {Array<Object>} Chronologically ordered snapshots.
+ */
+function resolveSnapshotList(entry) {
+  if (!entry || typeof entry !== 'object') {
+    return [];
+  }
+  const snapshots = entry.snapshots;
+  if (Array.isArray(snapshots) && snapshots.length > 0) {
+    return snapshots;
+  }
+  return [entry];
+}
+
 /**
 * Build a data model describing the content for chat tabs.
 *
@@ -97,37 +114,46 @@ export function buildChatTabModel({
  }

  for (const telemetryEntry of telemetry || []) {
-    if (!telemetryEntry) continue;
-    const ts = resolveTimestampSeconds(
-      telemetryEntry.rx_time ?? telemetryEntry.rxTime ?? telemetryEntry.telemetry_time ?? telemetryEntry.telemetryTime,
-      telemetryEntry.rx_iso ?? telemetryEntry.rxIso ?? telemetryEntry.telemetry_time_iso ?? telemetryEntry.telemetryTimeIso
-    );
-    if (ts == null || ts < cutoff) continue;
-    const nodeId = normaliseNodeId(telemetryEntry);
-    const nodeNum = normaliseNodeNum(telemetryEntry);
-    logEntries.push({ ts, type: CHAT_LOG_ENTRY_TYPES.TELEMETRY, telemetry: telemetryEntry, nodeId, nodeNum });
+    const snapshots = resolveSnapshotList(telemetryEntry);
+    for (const snapshot of snapshots) {
+      if (!snapshot) continue;
+      const ts = resolveTimestampSeconds(
+        snapshot.rx_time ?? snapshot.rxTime ?? snapshot.telemetry_time ?? snapshot.telemetryTime,
+        snapshot.rx_iso ?? snapshot.rxIso ?? snapshot.telemetry_time_iso ?? snapshot.telemetryTimeIso
+      );
+      if (ts == null || ts < cutoff) continue;
+      const nodeId = normaliseNodeId(snapshot);
+      const nodeNum = normaliseNodeNum(snapshot);
+      logEntries.push({ ts, type: CHAT_LOG_ENTRY_TYPES.TELEMETRY, telemetry: snapshot, nodeId, nodeNum });
+    }
  }

  for (const positionEntry of positions || []) {
-    if (!positionEntry) continue;
-    const ts = resolveTimestampSeconds(
-      positionEntry.rx_time ?? positionEntry.rxTime ?? positionEntry.position_time ?? positionEntry.positionTime,
-      positionEntry.rx_iso ?? positionEntry.rxIso ?? positionEntry.position_time_iso ?? positionEntry.positionTimeIso
-    );
-    if (ts == null || ts < cutoff) continue;
-    const nodeId = normaliseNodeId(positionEntry);
-    const nodeNum = normaliseNodeNum(positionEntry);
-    logEntries.push({ ts, type: CHAT_LOG_ENTRY_TYPES.POSITION, position: positionEntry, nodeId, nodeNum });
+    const snapshots = resolveSnapshotList(positionEntry);
+    for (const snapshot of snapshots) {
+      if (!snapshot) continue;
+      const ts = resolveTimestampSeconds(
+        snapshot.rx_time ?? snapshot.rxTime ?? snapshot.position_time ?? snapshot.positionTime,
+        snapshot.rx_iso ?? snapshot.rxIso ?? snapshot.position_time_iso ?? snapshot.positionTimeIso
+      );
+      if (ts == null || ts < cutoff) continue;
+      const nodeId = normaliseNodeId(snapshot);
+      const nodeNum = normaliseNodeNum(snapshot);
+      logEntries.push({ ts, type: CHAT_LOG_ENTRY_TYPES.POSITION, position: snapshot, nodeId, nodeNum });
+    }
  }

  for (const neighborEntry of neighbors || []) {
-    if (!neighborEntry) continue;
-    const ts = resolveTimestampSeconds(neighborEntry.rx_time ?? neighborEntry.rxTime, neighborEntry.rx_iso ?? neighborEntry.rxIso);
-    if (ts == null || ts < cutoff) continue;
-    const nodeId = normaliseNodeId(neighborEntry);
-    const nodeNum = normaliseNodeNum(neighborEntry);
-    const neighborId = normaliseNeighborId(neighborEntry);
-    logEntries.push({ ts, type: CHAT_LOG_ENTRY_TYPES.NEIGHBOR, neighbor: neighborEntry, nodeId, nodeNum, neighborId });
+    const snapshots = resolveSnapshotList(neighborEntry);
+    for (const snapshot of snapshots) {
+      if (!snapshot) continue;
+      const ts = resolveTimestampSeconds(snapshot.rx_time ?? snapshot.rxTime, snapshot.rx_iso ?? snapshot.rxIso);
+      if (ts == null || ts < cutoff) continue;
+      const nodeId = normaliseNodeId(snapshot);
+      const nodeNum = normaliseNodeNum(snapshot);
+      const neighborId = normaliseNeighborId(snapshot);
+      logEntries.push({ ts, type: CHAT_LOG_ENTRY_TYPES.NEIGHBOR, neighbor: snapshot, nodeId, nodeNum, neighborId });
+    }
  }

  const encryptedLogEntries = [];
@@ -47,6 +47,13 @@ import { renderChatTabs } from './chat-tabs.js';
 import { formatPositionHighlights, formatTelemetryHighlights } from './chat-log-highlights.js';
 import { filterChatModel, normaliseChatFilterQuery } from './chat-search.js';
 import { buildMessageBody, buildMessageIndex, resolveReplyPrefix } from './message-replies.js';
+import {
+  SNAPSHOT_WINDOW,
+  aggregateNeighborSnapshots,
+  aggregateNodeSnapshots,
+  aggregatePositionSnapshots,
+  aggregateTelemetrySnapshots,
+} from './snapshot-aggregator.js';

 /**
 * Entry point for the interactive dashboard. Wires up event listeners,
@@ -151,6 +158,7 @@ let messagesById = new Map();
    logger: console,
  });
  const NODE_LIMIT = 1000;
+  const SNAPSHOT_LIMIT = SNAPSHOT_WINDOW;
  const CHAT_LIMIT = MESSAGE_LIMIT;
  const CHAT_RECENT_WINDOW_SECONDS = 7 * 24 * 60 * 60;
  const REFRESH_MS = config.refreshMs;
@@ -3047,6 +3055,23 @@ let messagesById = new Map();
    return `${Math.floor(diff/86400)}d ${Math.floor((diff%86400)/3600)}h`;
  }

+  /**
+   * Determine how many snapshots should be requested from the API to build a
+   * richer aggregate.
+   *
+   * @param {number} requestedLimit Desired number of unique entities.
+   * @param {number} [maxLimit=NODE_LIMIT] Maximum rows accepted by the API.
+   * @returns {number} Effective request limit honouring {@link SNAPSHOT_LIMIT}.
+   */
+  function resolveSnapshotLimit(requestedLimit, maxLimit = NODE_LIMIT) {
+    const base = Number.isFinite(requestedLimit) && requestedLimit > 0
+      ? Math.floor(requestedLimit)
+      : maxLimit;
+    const expanded = base * SNAPSHOT_LIMIT;
+    const candidate = expanded > base ? expanded : base;
+    return Math.min(candidate, maxLimit);
+  }
+
  /**
   * Fetch the latest nodes from the JSON API.
   *
@@ -3054,7 +3079,8 @@ let messagesById = new Map();
   * @returns {Promise<Array<Object>>} Parsed node payloads.
   */
  async function fetchNodes(limit = NODE_LIMIT) {
-    const r = await fetch(`/api/nodes?limit=${limit}`, { cache: 'no-store' });
+    const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT);
+    const r = await fetch(`/api/nodes?limit=${effectiveLimit}`, { cache: 'no-store' });
    if (!r.ok) throw new Error('HTTP ' + r.status);
    return r.json();
  }
@@ -3102,7 +3128,8 @@ let messagesById = new Map();
   * @returns {Promise<Array<Object>>} Parsed neighbour payloads.
   */
  async function fetchNeighbors(limit = NODE_LIMIT) {
-    const r = await fetch(`/api/neighbors?limit=${limit}`, { cache: 'no-store' });
+    const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT);
+    const r = await fetch(`/api/neighbors?limit=${effectiveLimit}`, { cache: 'no-store' });
    if (!r.ok) throw new Error('HTTP ' + r.status);
    return r.json();
  }
@@ -3114,7 +3141,8 @@ let messagesById = new Map();
   * @returns {Promise<Array<Object>>} Parsed telemetry payloads.
   */
  async function fetchTelemetry(limit = NODE_LIMIT) {
-    const r = await fetch(`/api/telemetry?limit=${limit}`, { cache: 'no-store' });
+    const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT);
+    const r = await fetch(`/api/telemetry?limit=${effectiveLimit}`, { cache: 'no-store' });
    if (!r.ok) throw new Error('HTTP ' + r.status);
    return r.json();
  }
@@ -3126,7 +3154,8 @@ let messagesById = new Map();
   * @returns {Promise<Array<Object>>} Parsed position payloads.
   */
  async function fetchPositions(limit = NODE_LIMIT) {
-    const r = await fetch(`/api/positions?limit=${limit}`, { cache: 'no-store' });
+    const effectiveLimit = resolveSnapshotLimit(limit, NODE_LIMIT);
+    const r = await fetch(`/api/positions?limit=${effectiveLimit}`, { cache: 'no-store' });
    if (!r.ok) throw new Error('HTTP ' + r.status);
    return r.json();
  }
@@ -3763,11 +3792,15 @@ let messagesById = new Map();
        telemetryPromise,
        encryptedMessagesPromise
      ]);
-      nodes.forEach(applyNodeNameFallback);
-      mergePositionsIntoNodes(nodes, positions);
-      computeDistances(nodes);
-      mergeTelemetryIntoNodes(nodes, telemetryEntries);
-      allNodes = nodes;
+      const aggregatedNodes = aggregateNodeSnapshots(nodes);
+      const aggregatedPositions = aggregatePositionSnapshots(positions);
+      const aggregatedNeighbors = aggregateNeighborSnapshots(neighborTuples);
+      const aggregatedTelemetry = aggregateTelemetrySnapshots(telemetryEntries);
+      aggregatedNodes.forEach(applyNodeNameFallback);
+      mergePositionsIntoNodes(aggregatedNodes, aggregatedPositions);
+      computeDistances(aggregatedNodes);
+      mergeTelemetryIntoNodes(aggregatedNodes, aggregatedTelemetry);
+      allNodes = aggregatedNodes;
      rebuildNodeIndex(allNodes);
      const [chatMessages, encryptedChatMessages] = await Promise.all([
        messageNodeHydrator.hydrate(messages, nodesById),
@@ -3775,9 +3808,9 @@ let messagesById = new Map();
      ]);
      allMessages = Array.isArray(chatMessages) ? chatMessages : [];
      allEncryptedMessages = Array.isArray(encryptedChatMessages) ? encryptedChatMessages : [];
-      allTelemetryEntries = Array.isArray(telemetryEntries) ? telemetryEntries : [];
-      allPositionEntries = Array.isArray(positions) ? positions : [];
-      allNeighbors = Array.isArray(neighborTuples) ? neighborTuples : [];
+      allTelemetryEntries = aggregatedTelemetry;
+      allPositionEntries = aggregatedPositions;
+      allNeighbors = aggregatedNeighbors;
      applyFilter();
      if (statusEl) {
        statusEl.textContent = 'updated ' + new Date().toLocaleTimeString();
@@ -15,10 +15,17 @@
 */

 import { extractModemMetadata } from './node-modem-metadata.js';
+import {
+  SNAPSHOT_WINDOW,
+  aggregateNeighborSnapshots,
+  aggregateNodeSnapshots,
+  aggregatePositionSnapshots,
+  aggregateTelemetrySnapshots,
+} from './snapshot-aggregator.js';

 const DEFAULT_FETCH_OPTIONS = Object.freeze({ cache: 'no-store' });
-const TELEMETRY_LIMIT = 1;
-const POSITION_LIMIT = 1;
+const TELEMETRY_LIMIT = SNAPSHOT_WINDOW;
+const POSITION_LIMIT = SNAPSHOT_WINDOW;
 const NEIGHBOR_LIMIT = 1000;

 /**
@@ -351,7 +358,7 @@ export async function refreshNodeInformation(reference, options = {}) {

  const [nodeRecord, telemetryRecords, positionRecords, neighborRecords] = await Promise.all([
    (async () => {
-      const response = await fetchImpl(`/api/nodes/${encodedId}`, DEFAULT_FETCH_OPTIONS);
+      const response = await fetchImpl(`/api/nodes/${encodedId}?limit=${SNAPSHOT_WINDOW}`, DEFAULT_FETCH_OPTIONS);
      if (response.status === 404) return null;
      if (!response.ok) {
        throw new Error(`Failed to load node information (HTTP ${response.status})`);
@@ -384,17 +391,36 @@ export async function refreshNodeInformation(reference, options = {}) {
    })(),
  ]);

-  const telemetryEntry = Array.isArray(telemetryRecords) ? telemetryRecords[0] ?? null : telemetryRecords ?? null;
-  const positionEntry = Array.isArray(positionRecords) ? positionRecords[0] ?? null : positionRecords ?? null;
-  const neighborEntries = Array.isArray(neighborRecords) ? neighborRecords.filter(isObject) : [];
+  const nodeCandidates = Array.isArray(nodeRecord)
+    ? nodeRecord.filter(isObject)
+    : (isObject(nodeRecord) ? [nodeRecord] : []);
+  const aggregatedNodeRecords = aggregateNodeSnapshots(nodeCandidates);
+  const nodeRecordEntry = aggregatedNodeRecords[0] ?? null;
+
+  const telemetryCandidates = Array.isArray(telemetryRecords)
+    ? telemetryRecords
+    : (isObject(telemetryRecords) ? [telemetryRecords] : []);
+  const aggregatedTelemetry = aggregateTelemetrySnapshots(telemetryCandidates);
+  const telemetryEntry = aggregatedTelemetry[0] ?? null;
+
+  const positionCandidates = Array.isArray(positionRecords)
+    ? positionRecords
+    : (isObject(positionRecords) ? [positionRecords] : []);
+  const aggregatedPositions = aggregatePositionSnapshots(positionCandidates);
+  const positionEntry = aggregatedPositions[0] ?? null;
+
+  const neighborCandidates = Array.isArray(neighborRecords)
+    ? neighborRecords
+    : (isObject(neighborRecords) ? [neighborRecords] : []);
+  const neighborEntries = aggregateNeighborSnapshots(neighborCandidates);

  const node = { neighbors: neighborEntries };

  if (normalized.fallback) {
    mergeNodeFields(node, normalized.fallback);
  }
-  if (nodeRecord) {
-    mergeNodeFields(node, nodeRecord);
+  if (nodeRecordEntry) {
+    mergeNodeFields(node, nodeRecordEntry);
  }
  if (normalized.nodeId && !node.nodeId) {
    node.nodeId = normalized.nodeId;
@@ -420,7 +446,7 @@ export async function refreshNodeInformation(reference, options = {}) {
  }

  node.rawSources = {
-    node: nodeRecord,
+    node: nodeRecordEntry,
    telemetry: telemetryEntry,
    position: positionEntry,
    neighbors: neighborEntries,
@@ -0,0 +1,267 @@
+/*
+ * Copyright © 2025-26 l5yth & contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Number of snapshots to merge for each entity when aggregating records.
+ *
+ * @type {number}
+ */
+export const SNAPSHOT_WINDOW = 7;
+
+/**
+ * Determine whether a candidate behaves like an object.
+ *
+ * @param {*} value Candidate value to inspect.
+ * @returns {boolean} ``true`` when the value is a non-null object.
+ */
+function isObject(value) {
+  return value != null && typeof value === 'object';
+}
+
+/**
+ * Convert a raw identifier into a trimmed canonical string.
+ *
+ * @param {*} value Raw identifier.
+ * @returns {string|null} Normalised identifier or ``null`` when blank.
+ */
+function normaliseId(value) {
+  if (typeof value !== 'string') return null;
+  const trimmed = value.trim();
+  return trimmed.length === 0 ? null : trimmed;
+}
+
+/**
+ * Convert a raw numeric identifier into a finite number.
+ *
+ * @param {*} value Raw numeric identifier.
+ * @returns {number|null} Finite number or ``null`` when coercion fails.
+ */
+function normaliseNum(value) {
+  if (typeof value === 'number') {
+    return Number.isFinite(value) ? value : null;
+  }
+  if (value == null || value === '') return null;
+  const parsed = Number(value);
+  return Number.isFinite(parsed) ? parsed : null;
+}
+
+/**
+ * Merge snapshot fields into the destination object, skipping ``null`` values.
+ *
+ * @param {Object} target Destination object mutated in-place.
+ * @param {Object} snapshot Snapshot payload merged into ``target``.
+ * @returns {void}
+ */
+function mergeSnapshotFields(target, snapshot) {
+  if (!isObject(target) || !isObject(snapshot)) return;
+  for (const key of Object.keys(snapshot)) {
+    const value = snapshot[key];
+    if (value == null) continue;
+    if (typeof value === 'number' && Number.isNaN(value)) continue;
+    target[key] = value;
+  }
+}
+
+/**
+ * Build a key resolver that keeps node identifiers and numeric references
+ * associated with a single aggregate key.
+ *
+ * @returns {(entry: Object) => string|null} Key resolver function.
+ */
+function createNodeKeyResolver() {
+  const byId = new Map();
+  const byNum = new Map();
+  return entry => {
+    if (!isObject(entry)) return null;
+    const nodeId = normaliseId(entry.node_id ?? entry.nodeId);
+    const nodeNum = normaliseNum(entry.node_num ?? entry.nodeNum ?? entry.num);
+    if (nodeId && byId.has(nodeId)) {
+      const key = byId.get(nodeId);
+      if (nodeNum != null && !byNum.has(nodeNum)) {
+        byNum.set(nodeNum, key);
+      }
+      return key;
+    }
+    if (nodeNum != null && byNum.has(nodeNum)) {
+      const key = byNum.get(nodeNum);
+      if (nodeId) byId.set(nodeId, key);
+      return key;
+    }
+    let key = null;
+    if (nodeId) {
+      key = `id:${nodeId}`;
+    } else if (nodeNum != null) {
+      key = `num:${nodeNum}`;
+    }
+    if (key) {
+      if (nodeId) byId.set(nodeId, key);
+      if (nodeNum != null) byNum.set(nodeNum, key);
+    }
+    return key;
+  };
+}
+
+/**
+ * Ensure a property is attached to the aggregate object without exposing it
+ * through enumeration.
+ *
+ * @param {Object} target Destination aggregate object.
+ * @param {string} key Property name to assign.
+ * @param {*} value Property value.
+ * @returns {void}
+ */
+function defineHiddenProperty(target, key, value) {
+  Object.defineProperty(target, key, {
+    value,
+    enumerable: false,
+    configurable: false,
+    writable: false,
+  });
+}
+
+/**
+ * Aggregate a collection of snapshots by key, merging up to
+ * {@link SNAPSHOT_WINDOW} entries for each logical entity.
+ *
+ * The supplied ``keySelector`` determines which entries belong to the same
+ * aggregate. Snapshots are merged in chronological order (oldest to newest),
+ * allowing recent values to override stale ones while retaining older data for
+ * fields that may be absent in the latest packet.
+ *
+ * @template T
+ * @param {Array<Object>} entries Raw snapshot entries.
+ * @param {{
+ *   keySelector: (entry: Object) => string|null,
+ *   limit?: number,
+ *   merge?: (target: Object, snapshot: Object) => void,
+ *   baseFactory?: (snapshot: Object) => T
+ * }} options Aggregation behaviour overrides.
+ * @returns {Array<T>} Aggregated snapshots.
+ */
+export function aggregateSnapshots(entries, {
+  keySelector,
+  limit = SNAPSHOT_WINDOW,
+  merge = mergeSnapshotFields,
+  baseFactory = () => ({}),
+} = {}) {
+  if (typeof keySelector !== 'function') {
+    throw new TypeError('aggregateSnapshots requires a keySelector function');
+  }
+  if (!Array.isArray(entries) || entries.length === 0) {
+    return [];
+  }
+  const groups = new Map();
+  const maxSnapshots = Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : SNAPSHOT_WINDOW;
+  for (const entry of entries) {
+    if (!isObject(entry)) continue;
+    const key = keySelector(entry);
+    if (!key) continue;
+    let group = groups.get(key);
+    if (!group) {
+      group = [];
+      groups.set(key, group);
+    }
+    if (group.length >= maxSnapshots) continue;
+    group.push(entry);
+  }
+  const aggregates = [];
+  for (const group of groups.values()) {
+    if (!Array.isArray(group) || group.length === 0) continue;
+    const baseSnapshot = group[group.length - 1];
+    const target = baseFactory(isObject(baseSnapshot) ? { ...baseSnapshot } : {});
+    const orderedSnapshots = [];
+    for (let idx = group.length - 1; idx >= 0; idx -= 1) {
+      const snapshot = group[idx];
+      if (!isObject(snapshot)) continue;
+      const clone = { ...snapshot };
+      orderedSnapshots.push(clone);
+      merge(target, clone);
+    }
+    defineHiddenProperty(target, 'snapshots', orderedSnapshots);
+    defineHiddenProperty(target, 'latestSnapshot', orderedSnapshots[orderedSnapshots.length - 1] ?? null);
+    aggregates.push(target);
+  }
+  return aggregates;
+}
+
+/**
+ * Aggregate node records into enriched snapshots keyed by identifier.
+ *
+ * @param {Array<Object>} entries Node records fetched from the API.
+ * @param {{ limit?: number }} [options] Aggregation options.
+ * @returns {Array<Object>} Aggregated node payloads.
+ */
+export function aggregateNodeSnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}) {
+  const resolveKey = createNodeKeyResolver();
+  return aggregateSnapshots(entries, { keySelector: resolveKey, limit });
+}
+
+/**
+ * Aggregate telemetry packets for each node.
+ *
+ * @param {Array<Object>} entries Telemetry payloads.
+ * @param {{ limit?: number }} [options] Aggregation options.
+ * @returns {Array<Object>} Aggregated telemetry data.
+ */
+export function aggregateTelemetrySnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}) {
+  const resolveKey = createNodeKeyResolver();
+  return aggregateSnapshots(entries, { keySelector: resolveKey, limit });
+}
+
+/**
+ * Aggregate position packets for each node.
+ *
+ * @param {Array<Object>} entries Position payloads.
+ * @param {{ limit?: number }} [options] Aggregation options.
+ * @returns {Array<Object>} Aggregated position data.
+ */
+export function aggregatePositionSnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}) {
+  const resolveKey = createNodeKeyResolver();
+  return aggregateSnapshots(entries, { keySelector: resolveKey, limit });
+}
+
+/**
+ * Aggregate neighbour packets for each node pair.
+ *
+ * @param {Array<Object>} entries Neighbour payloads.
+ * @param {{ limit?: number }} [options] Aggregation options.
+ * @returns {Array<Object>} Aggregated neighbour data.
+ */
+export function aggregateNeighborSnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}) {
+  const resolveSourceKey = createNodeKeyResolver();
+  const resolveNeighborKey = createNodeKeyResolver();
+  return aggregateSnapshots(entries, {
+    limit,
+    keySelector: entry => {
+      if (!isObject(entry)) return null;
+      const sourceKey = resolveSourceKey(entry);
+      const neighborId = entry.neighbor_id ?? entry.neighborId;
+      const neighborNum = entry.neighbor_num ?? entry.neighborNum;
+      const neighborKey = resolveNeighborKey({ node_id: neighborId, node_num: neighborNum });
+      if (!sourceKey || !neighborKey) return null;
+      return `${sourceKey}->${neighborKey}`;
+    },
+  });
+}
+
+export const __testUtils = {
+  isObject,
+  normaliseId,
+  normaliseNum,
+  mergeSnapshotFields,
+  createNodeKeyResolver,
+  defineHiddenProperty,
+};