From 631455237fa4122b2a0c2ce9502d7b3c2031f643 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Thu, 13 Nov 2025 22:02:42 +0100 Subject: [PATCH] Aggregate frontend snapshots across views (#447) --- .../js/app/__tests__/node-details.test.js | 18 +- .../app/__tests__/snapshot-aggregator.test.js | 121 ++++++++ web/public/assets/js/app/chat-log-tabs.js | 76 +++-- web/public/assets/js/app/main.js | 57 +++- web/public/assets/js/app/node-details.js | 44 ++- .../assets/js/app/snapshot-aggregator.js | 267 ++++++++++++++++++ 6 files changed, 528 insertions(+), 55 deletions(-) create mode 100644 web/public/assets/js/app/__tests__/snapshot-aggregator.test.js create mode 100644 web/public/assets/js/app/snapshot-aggregator.js diff --git a/web/public/assets/js/app/__tests__/node-details.test.js b/web/public/assets/js/app/__tests__/node-details.test.js index d5c4b3c..828bcdc 100644 --- a/web/public/assets/js/app/__tests__/node-details.test.js +++ b/web/public/assets/js/app/__tests__/node-details.test.js @@ -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, diff --git a/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js b/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js new file mode 100644 index 0000000..4a603b2 --- /dev/null +++ b/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js @@ -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([], {}), []); +}); diff --git a/web/public/assets/js/app/chat-log-tabs.js b/web/public/assets/js/app/chat-log-tabs.js index 8d205c3..7a40069 100644 --- a/web/public/assets/js/app/chat-log-tabs.js +++ b/web/public/assets/js/app/chat-log-tabs.js @@ -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} 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 = []; diff --git a/web/public/assets/js/app/main.js b/web/public/assets/js/app/main.js index 9eb4636..bd79462 100644 --- a/web/public/assets/js/app/main.js +++ b/web/public/assets/js/app/main.js @@ -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>} 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>} 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>} 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>} 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(); diff --git a/web/public/assets/js/app/node-details.js b/web/public/assets/js/app/node-details.js index 747ca91..f2f0ee3 100644 --- a/web/public/assets/js/app/node-details.js +++ b/web/public/assets/js/app/node-details.js @@ -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, diff --git a/web/public/assets/js/app/snapshot-aggregator.js b/web/public/assets/js/app/snapshot-aggregator.js new file mode 100644 index 0000000..c7ec800 --- /dev/null +++ b/web/public/assets/js/app/snapshot-aggregator.js @@ -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} 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} 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} entries Node records fetched from the API. + * @param {{ limit?: number }} [options] Aggregation options. + * @returns {Array} 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} entries Telemetry payloads. + * @param {{ limit?: number }} [options] Aggregation options. + * @returns {Array} 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} entries Position payloads. + * @param {{ limit?: number }} [options] Aggregation options. + * @returns {Array} 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} entries Neighbour payloads. + * @param {{ limit?: number }} [options] Aggregation options. + * @returns {Array} 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, +};