From df7cfbe0ed71a677f0cc09e859f1a46cf10208a4 Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:53:29 +0000 Subject: [PATCH] Fix canonical repeater identity projection --- GNOME-REPEATER-2026-08-06.md | 86 ++ backend/src/api/hopreachCompatibility.ts | 17 +- backend/src/api/routes/activityTimeline.ts | 4 +- backend/src/api/routes/exports.ts | 4 +- backend/src/api/routes/misc.ts | 8 +- backend/src/api/routes/nodeStatus.ts | 68 +- backend/src/api/utils/networkFilters.ts | 22 +- backend/src/db/index.ts | 66 +- .../db/migrations/036_node_identity_merge.sql | 273 ++++++ backend/src/db/nodeIdentity.test.ts | 125 +++ backend/src/db/nodeIdentity.ts | 840 ++++++++++++++++++ backend/src/index.ts | 23 +- backend/src/owner/ownerRepository.ts | 97 +- backend/src/repositories/networkAnalysis.ts | 16 +- backend/src/repositories/nodes.ts | 32 +- backend/src/repositories/productFeatures.ts | 20 +- backend/src/stats/statsRepository.ts | 66 +- 17 files changed, 1596 insertions(+), 171 deletions(-) create mode 100644 GNOME-REPEATER-2026-08-06.md create mode 100644 backend/src/db/migrations/036_node_identity_merge.sql create mode 100644 backend/src/db/nodeIdentity.test.ts create mode 100644 backend/src/db/nodeIdentity.ts diff --git a/GNOME-REPEATER-2026-08-06.md b/GNOME-REPEATER-2026-08-06.md new file mode 100644 index 0000000..bebc899 --- /dev/null +++ b/GNOME-REPEATER-2026-08-06.md @@ -0,0 +1,86 @@ +# GNOME repeater identity investigation and fix + +Date: 2026-08-06 +Branch: `main` +Scope: backend data layer only; no frontend files or owner-dashboard UI were changed. + +## Finding + +The previous architecture finding is confirmed in both source and the live database. + +- `backend/src/mqtt/client.ts:546-590` uses the exact public key in the MQTT topic as the status observer identity. A status envelope is persisted with that observer key. +- `backend/src/mqtt/client.ts:681-752` decodes an advert sender public key and upserts/increments that decoded sender row. Packet persistence at `:829-852` keeps the topic observer as `rx_node_id` and the decoded sender as `src_node_id`. +- `backend/src/db/index.ts:323-334` increments one exact `nodes.node_id`; the table remains keyed by the exact 64-hex ID. The raw packet, status, link, and sighting tables likewise retain exact IDs. + +That produced the expected live split: + +| Device | Status/observer identity | Advert/position identity | Live evidence | +| --- | --- | --- | --- | +| GNOME-MSG-RPT | `230F10D0AE10E1F3FC9DBA1B32C48A242EEDAAAF89D440934AEB1D85FD719E5E` | `D7696E95DDE9A236C90322C3B7AEA756F91005E7B2A084D2F1EFDAD207F344A4` | Observer row has status but no position/adverts; advert row is positioned and had 2,736 adverts at investigation time. | +| GNOME-MSG-RPT Lanops.co | — | `D7696E95DDE9A236C90322C3B7AEA756F91605E7B2A084D2F1EFDAD207F344A4` | Same position as the active advert row and differs from it by one key nibble; one stale/duplicate advert row. | +| GNOME-STKTN-RPT | `6E9F3F33EE2E955BE8BBDDF7EC043C7B5B4A8CFD13E53456A382124B75E29F3E` | `65821B80039F42EEDA42C693D503873E3CCF2E83D1DAB85FB94A4433CBA104F8` | Observer status plus direct observer-to-advert packet evidence; V4 advert row is positioned. | +| GNOME-STKTN-RPT (stale) | `ADAE43BE6BDA85D606A842E755037B8319935F3B019B50D9924CE70AFB3E64E` | — | Same identity name, stale metadata-only row. | +| GNOME-MOBILE-RPT | `CEF1D1C2D5519A67680EF97C3A13BE3C1432AA201CFE4785A7F2E7E33A51B0F8` | same raw row | Stale and has no independent pairing evidence; intentionally remains separate. | + +Before this change, public projections queried those exact rows independently, so the status key and advert key could not produce one repeater view. + +## General identity model + +Migration `backend/src/db/migrations/036_node_identity_merge.sql` adds: + +- `node_identity_aliases`: reversible `source_node_id -> canonical_node_id` mappings, confidence, reason, and JSON evidence. Raw rows are never deleted or rewritten. +- `node_identity_match_evidence`: accepted and ambiguous candidate pairs, including score, confidence, reason, and evidence. Ambiguous pairs are retained for review but do not become aliases. +- `meshcore_canonical_node_id(text)`: a stable SQL resolver used by the data layer. +- Canonical views for nodes, sightings, statuses, packets, links, and link-radio reports. + +`node_identity_nodes` emits one row per canonical identity. It selects a positioned repeater representative first, then a repeater row, advert evidence, last-seen time, and key as tie-breakers; it aggregates member liveness, timestamps, and advert counts and exposes `identity_source_ids` and `identity_member_count`. The other views canonicalize IDs while preserving the original telemetry/packet observations. Canonical links are grouped as unordered endpoint pairs and direction counters are rotated when required. + +The backend refreshes automatic evidence at startup and every 30 minutes. Existing automatic aliases are replaced atomically inside a transaction; manual aliases, if added later, are preserved by `source_kind`. + +## Merge rule and confidence policy + +The rule is deliberately evidence-based and applies to all public repeaters, not to a GNOME-specific name list. + +1. Names are normalized with Unicode NFKC, upper-cased, punctuation-separated, `RPT` normalized to `REPEATER`, and version suffixes such as `V4` removed. Candidates must be in the same public network family, have compatible repeater roles, and share a normalized name bucket. Name similarity alone never merges anything. +2. A high-confidence pair is accepted when one of these independent signals exists in addition to the common name/network/role checks: + - keys differ in at most one hexadecimal nibble and valid positions are within 3 km; + - there are at least three packet observations and three decoded adverts directly pairing a metadata-only observer row with an active positioned advert row, with status evidence on the pair; or + - both keys have at least three self-adverts, their activity intervals do not overlap, the handover gap is at most 14 days, and status plus an active positioned repeater support the handover. +3. A medium-confidence pair is a metadata-only status row (no role, valid position, or adverts) plus exactly one unique active positioned repeater with the same normalized name/network and status evidence. Near-key duplicate active rows are collapsed when establishing that uniqueness. +4. A shared base name with a changed final ordinal/directional token is a veto for the weaker evidence paths. This protects genuinely separate colocated devices such as `Dunston-1/2`, `NorthMesh RPT-1/2`, and `Reach Yagi E/NWW`. Active-active packet hearing is not treated as identity evidence: one repeater hearing another is normal mesh behavior. The one-nibble-key-plus-position rule is the explicit strong-signal exception to the name-variant veto. +5. A group may not contain positioned members more than 3 km apart. Test-network rows are excluded from automatic grouping. + +The resulting canonical key is the best active positioned/advert-backed member, with deterministic last-seen and key tie-breakers. Stale and duplicate rows are therefore hidden from canonical public projections, not destroyed. Raw tables remain available for audit and rollback, while ambiguous candidates remain visible in `node_identity_match_evidence` rather than being silently merged. + +## Data-layer wiring + +The canonical projections are used by the public node/map/repeater APIs, status latest/history, packet history/adverts/recent packet/event projections, network filters, viable links and radio reports, topology/network analysis, stats, exports, HopReach compatibility, and owner live-data repository queries. The owner-dashboard UI itself was not changed. The latest status route resolves canonical members first and uses the existing per-node/time telemetry index, so the new merge does not turn the status endpoint into an unbounded full-view scan. + +## Live result after deployment + +The live canonical view now reports: + +| Canonical name | Canonical ID | Members | Source IDs | Position | Advert count | +| --- | --- | ---: | --- | --- | ---: | +| GNOME-MSG-RPT | `D7696E95DDE9A236C90322C3B7AEA756F91005E7B2A084D2F1EFDAD207F344A4` | 3 | `230F…195E`, active `D769…F344A4`, Lanops `D769…F916…F344A4` | 54.521892, -1.473446 | 2,738 | +| GNOME-STKTN-RPT-V4 | `65821B80039F42EEDA42C693D503873E3CCF2E83D1DAB85FB94A4433CBA104F8` | 3 | active `6582…A104F8`, `6E9F…9F3E`, stale `ADAE…E64E` | 54.562317, -1.366457 | 352 | +| GNOME-MOBILE-RPT | `CEF1D1C2D5519A67680EF97C3A13BE3C1432AA201CFE4785A7F2E7E33A51B0F8` | 1 | itself | no valid position (0,0) | 42 | + +The live API checks returned: + +- `/api/nodes?network=ukmesh`: HTTP 200; one positioned row for each active GNOME repeater, plus the intentionally separate stale mobile row. +- `/api/node-status/latest?network=ukmesh`: HTTP 200; GNOME status rows are returned under the canonical advert IDs. The MSG row had a current status timestamp and uptime; STKTN likewise returned under `65821…A104F8`. +- `/api/node-status/history?...nodeId=D7696E95...`: HTTP 200; 2,885 merged MSG telemetry points over 24 hours. +- `/hopreach/healthz`: HTTP 200, `{"status":"ok"}`. +- `/api/health`: HTTP 200, `{"status":"healthy","incidents":[]}` after the normal post-restart health snapshot refresh; the backend container health probe is healthy. + +The refresh currently records 20 accepted automatic pairs and 15,342 ambiguous same-name candidates. The false-merge regression checks for the colocated Dunston, NorthMesh, Reach Yagi, and NE33 families found zero aliases. This is the intended behavior: high-signal identities merge, uncertain same-name candidates remain separate and auditable. + +## Deployment and verification + +- Migration 036 was applied to the live database by the backend migration runner. +- Backend image was built with `BACKEND_IMAGE=meshcore-analytics-backend:gnome-fix` and deployed as digest `sha256:03e54021c9ea0ebcd6febcfec28c5b67fd222af28cb3c5c56a3589ab316cf7ca`. The host `.env` pin was updated to that digest. +- Only the backend was recreated with `docker compose -f docker-compose.yml -f docker-compose.live.yml up -d --no-deps --force-recreate backend`. Mosquitto, TimescaleDB, Redis, HopReach, workers, Discord bot, and beacon services were not restarted or changed. +- `npm run typecheck` passed. The six focused node-identity tests passed. The full backend suite passed 269 of 270 tests; the one failure is the existing unrelated cache-policy assertion for `src/mqtt/channelRegistry.ts#channelCache`. + +No screenshot was taken because the frontend was not changed; the frontend-visible data is verified through the same node/status APIs it consumes. diff --git a/backend/src/api/hopreachCompatibility.ts b/backend/src/api/hopreachCompatibility.ts index 14f3856..bc28218 100644 --- a/backend/src/api/hopreachCompatibility.ts +++ b/backend/src/api/hopreachCompatibility.ts @@ -167,7 +167,7 @@ const NODE_WHERE = ` AND ( n.network = 'ukmesh' OR EXISTS ( - SELECT 1 FROM node_network_sightings sighting + SELECT 1 FROM node_identity_sightings sighting WHERE sighting.node_id = n.node_id AND sighting.network = 'ukmesh' ) )`; @@ -175,11 +175,12 @@ const NODE_WHERE = ` async function fetchLinks(query: QueryFn, publicKeys: string[], days: number) { if (publicKeys.length === 0) return [] as HopReachLinkRow[]; const result = await query( - `WITH requested(source_id) AS ( - SELECT unnest($1::text[]) + `WITH requested(requested_id, source_id) AS ( + SELECT id, meshcore_canonical_node_id(id) + FROM unnest($1::text[]) AS ids(id) ) SELECT - requested.source_id, + requested.requested_id AS source_id, peer.node_id AS pubkey, peer.name, peer.lat, @@ -191,9 +192,9 @@ async function fetchLinks(query: QueryFn, publicKeys: string[], days: number) { END::int AS bottleneck, (nl.count_a_to_b > 0 AND nl.count_b_to_a > 0) AS bidir FROM requested - JOIN node_links nl + JOIN node_identity_links nl ON nl.node_a_id = requested.source_id OR nl.node_b_id = requested.source_id - JOIN nodes peer + JOIN node_identity_nodes peer ON peer.node_id = CASE WHEN nl.node_a_id = requested.source_id THEN nl.node_b_id ELSE nl.node_a_id @@ -244,7 +245,7 @@ export function createHopReachCompatibilityRoutes(query: QueryFn): Router { const [count, page] = await Promise.all([ loadCached(nodeCountCache, nodeCountInFlight, 'all-repeaters', async () => { const result = await query<{ total: string }>( - `SELECT COUNT(*)::text AS total FROM nodes n WHERE ${NODE_WHERE}`, + `SELECT COUNT(*)::text AS total FROM node_identity_nodes n WHERE ${NODE_WHERE}`, ); return Number(result.rows[0]?.total ?? 0); }), @@ -253,7 +254,7 @@ export function createHopReachCompatibilityRoutes(query: QueryFn): Router { `SELECT n.node_id AS public_key, n.name, n.lat, n.lon, n.last_seen AS last_heard, n.created_at AS first_seen, n.advert_count - FROM nodes n + FROM node_identity_nodes n WHERE ${NODE_WHERE} ORDER BY n.node_id LIMIT $1 OFFSET $2`, diff --git a/backend/src/api/routes/activityTimeline.ts b/backend/src/api/routes/activityTimeline.ts index 324b2b0..88b273e 100644 --- a/backend/src/api/routes/activityTimeline.ts +++ b/backend/src/api/routes/activityTimeline.ts @@ -41,8 +41,8 @@ export function registerActivityTimelineRoutes(router: Router, deps: Deps): void SELECT date_bin(${bucketParam}::interval, p.time, TIMESTAMPTZ '2000-01-01') AS bucket, p.packet_hash, - p.rx_node_id, - p.src_node_id + meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id, + meshcore_canonical_node_id(p.src_node_id) AS src_node_id FROM packets p WHERE p.time > NOW() - ${windowParam}::interval ${filters.packetsAlias('p')} diff --git a/backend/src/api/routes/exports.ts b/backend/src/api/routes/exports.ts index 8223e2b..cc8bf4e 100644 --- a/backend/src/api/routes/exports.ts +++ b/backend/src/api/routes/exports.ts @@ -63,7 +63,7 @@ export function registerExportRoutes(router: Router, deps: Deps): void { ) SELECT DISTINCT ON (h.ord) n.node_id, n.name, n.lat, n.lon, h.ord FROM hops h - JOIN nodes n ON upper(n.node_id) LIKE h.hash || '%' + JOIN node_identity_nodes n ON upper(n.node_id) LIKE h.hash || '%' WHERE n.lat IS NOT NULL AND n.lon IS NOT NULL AND (n.name IS NULL OR n.name NOT LIKE '%🚫%') ${filters.nodesAlias('n')} @@ -114,7 +114,7 @@ export function registerExportRoutes(router: Router, deps: Deps): void { role: number | null; iata: string | null; last_seen: string | null; hardware_model: string | null; }>( `SELECT node_id, name, lat, lon, role, iata, last_seen::text, hardware_model - FROM nodes + FROM node_identity_nodes WHERE lat IS NOT NULL AND lon IS NOT NULL AND (name IS NULL OR name NOT LIKE '%🚫%') ${filters.nodes} diff --git a/backend/src/api/routes/misc.ts b/backend/src/api/routes/misc.ts index ab7a835..17681a4 100644 --- a/backend/src/api/routes/misc.ts +++ b/backend/src/api/routes/misc.ts @@ -200,15 +200,15 @@ export function registerMiscRoutes(router: Router, deps: MiscRouteDeps): void { nss.tx_air_secs, nss.stats, COALESCE(pc.packet_count, 0) AS packets_24h - FROM node_status_samples nss - LEFT JOIN nodes n ON n.node_id = nss.node_id + FROM node_identity_status_samples nss + LEFT JOIN node_identity_nodes n ON n.node_id = nss.node_id LEFT JOIN ( - SELECT rx_node_id, COUNT(*) AS packet_count + SELECT meshcore_canonical_node_id(rx_node_id) AS rx_node_id, COUNT(*) AS packet_count FROM packets WHERE time > NOW() - INTERVAL '24 hours' AND rx_node_id IS NOT NULL ${packetNetworkClause} - GROUP BY rx_node_id + GROUP BY meshcore_canonical_node_id(rx_node_id) ) pc ON pc.rx_node_id = nss.node_id WHERE nss.time > NOW() - INTERVAL '15 minutes' AND (n.name IS NULL OR n.name NOT LIKE '%🚫%') diff --git a/backend/src/api/routes/nodeStatus.ts b/backend/src/api/routes/nodeStatus.ts index c75c177..095642c 100644 --- a/backend/src/api/routes/nodeStatus.ts +++ b/backend/src/api/routes/nodeStatus.ts @@ -16,16 +16,16 @@ router.get('/node-status/latest', async (req, res) => { const networkValues = expandResolverScope(network); const observer = normalizeObserverQuery(req.query['observer']); - const params: unknown[] = []; - const conditions: string[] = []; - params.push(networkValues); - conditions.push(`nss.network = ANY($${params.length}::text[])`); + const params: unknown[] = [networkValues]; + const identityConditions: string[] = [ + `(n.name IS NULL OR n.name NOT LIKE '%🚫%')`, + ]; if (observer) { params.push(observer); - conditions.push(`nss.node_id = $${params.length}`); + identityConditions.push(`n.node_id = meshcore_canonical_node_id($${params.length})`); } - const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; + const identityWhereClause = `WHERE ${identityConditions.join(' AND ')}`; const result = await query<{ time: string; node_id: string; @@ -42,10 +42,20 @@ router.get('/node-status/latest', async (req, res) => { hardware_model: string | null; firmware_version: string | null; }>( - `SELECT * FROM ( - SELECT DISTINCT ON (nss.node_id) + `WITH identity_sources AS MATERIALIZED ( + SELECT n.node_id AS canonical_id, + n.name, + n.iata, + n.hardware_model, + n.firmware_version, + source.source_node_id + FROM node_identity_nodes n + CROSS JOIN LATERAL unnest(n.identity_source_ids) AS source(source_node_id) + ${identityWhereClause} + ), latest AS ( + SELECT DISTINCT ON (s.canonical_id) nss.time::text, - nss.node_id, + s.canonical_id AS node_id, nss.network, nss.battery_mv, nss.uptime_secs, @@ -54,16 +64,23 @@ router.get('/node-status/latest', async (req, res) => { nss.channel_utilization, nss.air_util_tx, nss.stats, - n.name, - n.iata, - n.hardware_model, - n.firmware_version - FROM node_status_samples nss - LEFT JOIN nodes n ON n.node_id = nss.node_id - ${whereClause} - ${whereClause ? 'AND' : 'WHERE'} (n.name IS NULL OR n.name NOT LIKE '%🚫%') - ORDER BY nss.node_id, nss.time DESC - ) latest + s.name, + s.iata, + s.hardware_model, + s.firmware_version + FROM identity_sources s + JOIN LATERAL ( + SELECT time, network, battery_mv, uptime_secs, tx_air_secs, + rx_air_secs, channel_utilization, air_util_tx, stats + FROM node_status_samples + WHERE node_id = s.source_node_id + AND network = ANY($1::text[]) + ORDER BY time DESC + LIMIT 1 + ) nss ON TRUE + ORDER BY s.canonical_id, nss.time DESC + ) + SELECT * FROM latest ORDER BY time DESC`, params, ); @@ -101,13 +118,13 @@ router.get('/node-status/history', async (req, res) => { conditions.push(`nss.network = ANY($${params.length}::text[])`); if (observer) { params.push(observer); - conditions.push(`nss.node_id = $${params.length}`); + conditions.push(`nss.node_id = meshcore_canonical_node_id($${params.length})`); } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; const latestNode = await query<{ node_id: string }>( `SELECT nss.node_id - FROM node_status_samples nss - JOIN nodes n ON n.node_id = nss.node_id + FROM node_identity_status_samples nss + JOIN node_identity_nodes n ON n.node_id = nss.node_id ${whereClause} AND (n.name IS NULL OR n.name NOT LIKE '%🚫%') ORDER BY nss.time DESC @@ -169,9 +186,10 @@ router.get('/node-status/history', async (req, res) => { WHEN jsonb_typeof(stats->'tx_queue_depth_peak') = 'number' THEN (stats->>'tx_queue_depth_peak')::double precision ELSE NULL END AS tx_queue_depth_peak - FROM node_status_samples nss - JOIN nodes n ON n.node_id = nss.node_id - WHERE nss.node_id = $1 + FROM node_identity_nodes n + CROSS JOIN LATERAL unnest(n.identity_source_ids) AS source(source_node_id) + JOIN node_status_samples nss ON nss.node_id = source.source_node_id + WHERE n.node_id = meshcore_canonical_node_id($1) AND nss.network = ANY($3::text[]) AND (n.name IS NULL OR n.name NOT LIKE '%🚫%') AND nss.time > NOW() - ($2::text || ' hours')::interval diff --git a/backend/src/api/utils/networkFilters.ts b/backend/src/api/utils/networkFilters.ts index 0fbcba4..796e27a 100644 --- a/backend/src/api/utils/networkFilters.ts +++ b/backend/src/api/utils/networkFilters.ts @@ -96,7 +96,12 @@ export function networkFilters(network?: string, observer?: string): NetworkFilt packetConditions.push(excludesLegacyTestTopic('')); packetConditions.push(`COALESCE(rx_node_id, '') NOT IN (SELECT node_id FROM nodes WHERE network = 'test')`); } - if (observerParam) packetConditions.push(`rx_node_id = ${observerParam}`); + if (observerParam) { + packetConditions.push( + `(rx_node_id = ${observerParam} + OR meshcore_canonical_node_id(rx_node_id) = meshcore_canonical_node_id(${observerParam}))`, + ); + } packetConditions.push(...publicPacketPrivacyConditions('')); const nodeConditions = (alias?: string) => { @@ -119,7 +124,7 @@ export function networkFilters(network?: string, observer?: string): NetworkFilt ${prefix}network IS DISTINCT FROM 'test' AND EXISTS ( SELECT 1 - FROM node_network_sightings s + FROM node_identity_sightings s WHERE s.node_id = ${nodeRef} AND s.network ${netMatch} AND s.last_seen_at > NOW() - INTERVAL '30 days' @@ -137,11 +142,11 @@ export function networkFilters(network?: string, observer?: string): NetworkFilt // 7-day window keeps the packet scan inside recent chunks. conditions.push( `( - ${prefix}node_id = ${observerParam} + meshcore_canonical_node_id(${prefix}node_id) = meshcore_canonical_node_id(${observerParam}) OR ${nodeRef} IN ( - SELECT p.src_node_id + SELECT meshcore_canonical_node_id(p.src_node_id) FROM packets p - WHERE p.rx_node_id = ${observerParam} + WHERE meshcore_canonical_node_id(p.rx_node_id) = meshcore_canonical_node_id(${observerParam}) AND p.time > NOW() - INTERVAL '7 days' AND p.src_node_id IS NOT NULL ${pNetCond} @@ -172,7 +177,12 @@ export function networkFilters(network?: string, observer?: string): NetworkFilt conditions.push(excludesLegacyTestTopic(prefix)); conditions.push(`COALESCE(${prefix}rx_node_id, '') NOT IN (SELECT node_id FROM nodes WHERE network = 'test')`); } - if (observerParam) conditions.push(`${prefix}rx_node_id = ${observerParam}`); + if (observerParam) { + conditions.push( + `(${prefix}rx_node_id = ${observerParam} + OR meshcore_canonical_node_id(${prefix}rx_node_id) = meshcore_canonical_node_id(${observerParam}))`, + ); + } conditions.push(...publicPacketPrivacyConditions(prefix)); return conditions.length > 0 ? `AND ${conditions.join(' AND ')}` : ''; }, diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 1c7ea53..57ceaba 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -210,7 +210,9 @@ function buildPacketScopeClause( ); } if (placeholders.observerParam) { - conditions.push(`${prefix}rx_node_id = ${placeholders.observerParam}`); + conditions.push( + `meshcore_canonical_node_id(${prefix}rx_node_id) = meshcore_canonical_node_id(${placeholders.observerParam})`, + ); } return conditions.length > 0 ? ` AND ${conditions.join(' AND ')}` : ''; } @@ -245,7 +247,7 @@ function buildNodeScopeClause( ${prefix}network IS DISTINCT FROM 'test' AND EXISTS ( SELECT 1 - FROM node_network_sightings s + FROM node_identity_sightings s WHERE s.node_id = ${nodeRef} AND s.network ${netMatch} AND s.last_seen_at > NOW() - INTERVAL '30 days' @@ -268,11 +270,11 @@ function buildNodeScopeClause( // 7-day window matches the observer_meta lookback and keeps the packet // scan inside recent chunks (~700ms vs 5min unbounded). const observerNodeScope = [ - `${prefix}node_id = ${placeholders.observerParam}`, + `meshcore_canonical_node_id(${nodeRef}) = meshcore_canonical_node_id(${placeholders.observerParam})`, `OR ${nodeRef} IN ( - SELECT p.src_node_id + SELECT meshcore_canonical_node_id(p.src_node_id) FROM packets p - WHERE p.rx_node_id = ${placeholders.observerParam} + WHERE meshcore_canonical_node_id(p.rx_node_id) = meshcore_canonical_node_id(${placeholders.observerParam}) AND p.time > NOW() - INTERVAL '7 days' AND p.src_node_id IS NOT NULL`, netCond, @@ -708,7 +710,7 @@ export async function getNodes( ${nodeEffectiveOnlineSql('n')} AS is_online, n.advert_count ${optionalFields} - FROM nodes n + FROM node_identity_nodes n ${whereClause} ORDER BY ${nodeEffectiveLastSeenSql('n')} DESC`, scope.params @@ -719,9 +721,11 @@ export async function getNodes( export async function getNodeHistory(nodeId: string, hours = 24, network = 'ukmesh') { const scope = buildScopePlaceholders(3, network); const res = await pool.query( - `SELECT time, packet_hash, src_node_id, topic, packet_type, hop_count, rssi, snr, payload + `SELECT time, packet_hash, meshcore_canonical_node_id(src_node_id) AS src_node_id, + topic, packet_type, hop_count, rssi, snr, payload FROM packets p - WHERE p.rx_node_id = $1 AND p.time > NOW() - INTERVAL '1 hour' * $2 + WHERE meshcore_canonical_node_id(p.rx_node_id) = meshcore_canonical_node_id($1) + AND p.time > NOW() - INTERVAL '1 hour' * $2 ${buildPacketScopeClause(scope, 'p', network)} ${buildPublicPacketPrivacyClause('p')} ORDER BY time DESC LIMIT 500`, @@ -738,7 +742,8 @@ export async function getNodeAdverts(nodePublicKey: string, hours = 24, limit = `SELECT time, packet_hash FROM packets p WHERE p.packet_type = 4 - AND p.payload->>'publicKey' = $1 + AND meshcore_canonical_node_id(COALESCE(p.src_node_id, p.payload->>'publicKey')) + = meshcore_canonical_node_id($1) AND p.time > NOW() - INTERVAL '1 hour' * $2 ${buildPacketScopeClause(scope, 'p', network)} ${buildPublicPacketPrivacyClause('p')} @@ -760,7 +765,9 @@ export async function getRecentPackets( const res = await pool.query( `WITH recent_packets AS ( SELECT DISTINCT ON (p.packet_hash) - p.time, p.packet_hash, p.rx_node_id, p.src_node_id, p.topic, + p.time, p.packet_hash, + meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id, + meshcore_canonical_node_id(p.src_node_id) AS src_node_id, p.topic, p.topic_prefix, p.iata, p.packet_type, p.route_type, p.hop_count, p.rssi, p.snr, COALESCE(p.payload->>'_summary', pd.summary) AS summary, p.advert_count, p.path_hashes, p.path_hash_size_bytes, @@ -780,7 +787,9 @@ export async function getRecentPackets( packet_stats AS ( SELECT packet_hash, - ARRAY_AGG(DISTINCT rx_node_id ORDER BY rx_node_id) FILTER (WHERE rx_node_id IS NOT NULL) AS observer_node_ids, + ARRAY_AGG(DISTINCT meshcore_canonical_node_id(rx_node_id) + ORDER BY meshcore_canonical_node_id(rx_node_id)) + FILTER (WHERE rx_node_id IS NOT NULL) AS observer_node_ids, ARRAY_AGG(DISTINCT iata ORDER BY iata) FILTER (WHERE NULLIF(TRIM(iata), '') IS NOT NULL) AS observer_iatas, COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') <> 'tx')::int AS rx_count, COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') = 'tx')::int AS tx_count @@ -818,7 +827,9 @@ export async function getRecentMessages(limit = 50, network?: string, observer?: const res = await pool.query( `WITH recent_msgs AS ( SELECT DISTINCT ON (p.packet_hash) - p.time, p.packet_hash, p.rx_node_id, p.src_node_id, p.topic, + p.time, p.packet_hash, + meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id, + meshcore_canonical_node_id(p.src_node_id) AS src_node_id, p.topic, p.iata, p.packet_type, p.hop_count, p.rssi, p.snr, p.payload, COALESCE(p.payload->>'_summary', pd.summary) AS summary, @@ -837,7 +848,9 @@ export async function getRecentMessages(limit = 50, network?: string, observer?: msg_stats AS ( SELECT packet_hash, - ARRAY_AGG(DISTINCT rx_node_id ORDER BY rx_node_id) FILTER (WHERE rx_node_id IS NOT NULL) AS observer_node_ids, + ARRAY_AGG(DISTINCT meshcore_canonical_node_id(rx_node_id) + ORDER BY meshcore_canonical_node_id(rx_node_id)) + FILTER (WHERE rx_node_id IS NOT NULL) AS observer_node_ids, ARRAY_AGG(DISTINCT iata ORDER BY iata) FILTER (WHERE NULLIF(TRIM(iata), '') IS NOT NULL) AS observer_iatas, COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') <> 'tx')::int AS rx_count, COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') = 'tx')::int AS tx_count @@ -867,7 +880,9 @@ export async function getRecentPacketEvents(limit = 200, network?: string, obser const params: unknown[] = [limit, ...scope.params]; const res = await pool.query( `SELECT - p.time, p.packet_hash, p.rx_node_id, p.src_node_id, p.topic, p.iata, + p.time, p.packet_hash, + meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id, + meshcore_canonical_node_id(p.src_node_id) AS src_node_id, p.topic, p.iata, p.packet_type, p.hop_count, p.rssi, p.snr, p.payload, p.payload->>'_summary' AS summary, p.advert_count, p.path_hashes, p.path_hash_size_bytes @@ -901,7 +916,8 @@ export async function getPacketDetail(hash: string, network = 'ukmesh') { [hash, ...scope.params], ), pool.query( - `SELECT p.rx_node_id, p.iata, p.time, p.rssi, p.snr, p.hop_count + `SELECT meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id, + p.iata, p.time, p.rssi, p.snr, p.hop_count FROM packets p WHERE p.packet_hash = $1 ${buildPacketScopeClause(scope, 'p', network)} @@ -1110,9 +1126,9 @@ export async function getMultibytePathSegments(network?: string, observer?: stri b.lat AS b_lat, b.lon AS b_lon, nl.multibyte_observed_count AS count - FROM node_links nl - JOIN nodes a ON a.node_id = nl.node_a_id - JOIN nodes b ON b.node_id = nl.node_b_id + FROM node_identity_links nl + JOIN node_identity_nodes a ON a.node_id = nl.node_a_id + JOIN node_identity_nodes b ON b.node_id = nl.node_b_id WHERE nl.multibyte_observed_count > 0 AND nl.itm_viable = true AND a.lat IS NOT NULL @@ -1172,7 +1188,7 @@ export async function getViableLinks(network?: string, observer?: string): Promi const scopedNetworks = network === 'ukmesh' ? UKMESH_NETWORKS : [network]; const res = await pool.query( `WITH net_nodes AS ( - SELECT DISTINCT node_id FROM nodes + SELECT DISTINCT node_id FROM node_identity_nodes WHERE network = ANY($1::text[]) AND (name IS NULL OR name NOT LIKE '%🚫%') ) @@ -1187,12 +1203,12 @@ export async function getViableLinks(network?: string, observer?: string): Promi nl.itm_path_loss_db, nl.count_a_to_b, nl.count_b_to_a - FROM node_links nl + FROM node_identity_links nl LEFT JOIN ( SELECT node_a_id, node_b_id, SUM(sample_count)::int AS neighbor_report_count, MAX(best_snr_db) AS neighbor_best_snr_db - FROM node_link_radio_reports + FROM node_identity_link_radio_reports GROUP BY node_a_id, node_b_id ) nr ON nr.node_a_id = nl.node_a_id AND nr.node_b_id = nl.node_b_id WHERE (nl.itm_viable = true OR nl.force_viable = true) @@ -1218,16 +1234,16 @@ export async function getViableLinks(network?: string, observer?: string): Promi nl.itm_path_loss_db, nl.count_a_to_b, nl.count_b_to_a - FROM node_links nl + FROM node_identity_links nl LEFT JOIN ( SELECT node_a_id, node_b_id, SUM(sample_count)::int AS neighbor_report_count, MAX(best_snr_db) AS neighbor_best_snr_db - FROM node_link_radio_reports + FROM node_identity_link_radio_reports GROUP BY node_a_id, node_b_id ) nr ON nr.node_a_id = nl.node_a_id AND nr.node_b_id = nl.node_b_id - JOIN nodes a ON a.node_id = nl.node_a_id - JOIN nodes b ON b.node_id = nl.node_b_id + JOIN node_identity_nodes a ON a.node_id = nl.node_a_id + JOIN node_identity_nodes b ON b.node_id = nl.node_b_id WHERE (nl.itm_viable = true OR nl.force_viable = true) AND (a.name IS NULL OR a.name NOT LIKE '%🚫%') AND (b.name IS NULL OR b.name NOT LIKE '%🚫%') diff --git a/backend/src/db/migrations/036_node_identity_merge.sql b/backend/src/db/migrations/036_node_identity_merge.sql new file mode 100644 index 0000000..3c93bf5 --- /dev/null +++ b/backend/src/db/migrations/036_node_identity_merge.sql @@ -0,0 +1,273 @@ +-- Reversible identity projection for MQTT observer keys and decoded advert keys. +-- Raw node, packet, and telemetry rows remain intact; the data layer exposes +-- canonical views and the backend periodically refreshes accepted aliases. + +CREATE TABLE IF NOT EXISTS node_identity_aliases ( + source_node_id TEXT PRIMARY KEY + CHECK (source_node_id ~ '^[0-9A-F]{64}$'), + canonical_node_id TEXT NOT NULL + CHECK (canonical_node_id ~ '^[0-9A-F]{64}$'), + confidence TEXT NOT NULL + CHECK (confidence IN ('high', 'medium')), + reason TEXT NOT NULL, + evidence JSONB NOT NULL DEFAULT '{}'::jsonb, + source_kind TEXT NOT NULL DEFAULT 'automatic' + CHECK (source_kind IN ('automatic', 'manual')), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CHECK (source_node_id <> canonical_node_id) +); + +CREATE INDEX IF NOT EXISTS node_identity_aliases_canonical_idx + ON node_identity_aliases (canonical_node_id); + +CREATE TABLE IF NOT EXISTS node_identity_match_evidence ( + node_a_id TEXT NOT NULL + CHECK (node_a_id ~ '^[0-9A-F]{64}$'), + node_b_id TEXT NOT NULL + CHECK (node_b_id ~ '^[0-9A-F]{64}$'), + decision TEXT NOT NULL + CHECK (decision IN ('accepted', 'ambiguous')), + confidence TEXT NOT NULL + CHECK (confidence IN ('high', 'medium', 'low')), + score INTEGER NOT NULL DEFAULT 0 + CHECK (score >= 0), + reason TEXT NOT NULL, + evidence JSONB NOT NULL DEFAULT '{}'::jsonb, + source_kind TEXT NOT NULL DEFAULT 'automatic' + CHECK (source_kind IN ('automatic', 'manual')), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (node_a_id, node_b_id), + CHECK (node_a_id < node_b_id) +); + +ALTER TABLE node_identity_match_evidence + ADD COLUMN IF NOT EXISTS score INTEGER NOT NULL DEFAULT 0; +ALTER TABLE node_identity_match_evidence + ADD COLUMN IF NOT EXISTS source_kind TEXT NOT NULL DEFAULT 'automatic'; + +CREATE INDEX IF NOT EXISTS node_identity_match_evidence_decision_idx + ON node_identity_match_evidence (decision, updated_at DESC); + +CREATE OR REPLACE FUNCTION meshcore_canonical_node_id(input_node_id TEXT) +RETURNS TEXT +LANGUAGE SQL +STABLE +PARALLEL SAFE +SET search_path = public +AS $$ + SELECT CASE + WHEN input_node_id IS NULL OR btrim(input_node_id) = '' THEN input_node_id + ELSE COALESCE( + ( + SELECT alias.canonical_node_id + FROM node_identity_aliases alias + WHERE alias.source_node_id = upper(btrim(input_node_id)) + ), + upper(btrim(input_node_id)) + ) + END +$$; + +-- One row per canonical identity. The representative is the best positioned +-- repeater row, then the row with the strongest advert evidence. Aggregate +-- liveness, telemetry timestamps, and advert counts across all members. +CREATE VIEW node_identity_nodes AS +WITH mapped AS ( + SELECT n.*, + meshcore_canonical_node_id(n.node_id) AS canonical_node_id + FROM nodes n +), ranked AS ( + SELECT m.*, + row_number() OVER ( + PARTITION BY m.canonical_node_id + ORDER BY + CASE + WHEN m.role = 2 + AND m.lat BETWEEN -90 AND 90 + AND m.lon BETWEEN -180 AND 180 + AND NOT (ABS(m.lat) < 1e-9 AND ABS(m.lon) < 1e-9) + THEN 0 + WHEN m.role = 2 THEN 1 + WHEN COALESCE(m.advert_count, 0) > 0 THEN 2 + ELSE 3 + END, + COALESCE(m.advert_count, 0) DESC, + m.last_seen DESC NULLS LAST, + m.node_id + ) AS representative_rank + FROM mapped m +), aggregated AS ( + SELECT canonical_node_id, + bool_or(COALESCE(is_online, FALSE)) AS is_online, + max(last_seen) AS last_seen, + max(last_predicted_online_at) AS last_predicted_online_at, + max(last_path_evidence_at) AS last_path_evidence_at, + max(last_mqtt_observer_seen_at) AS last_mqtt_observer_seen_at, + max(last_rx_at) AS last_rx_at, + max(last_status_at) AS last_status_at, + min(created_at) AS created_at, + sum(COALESCE(advert_count, 0))::integer AS advert_count, + array_agg(node_id ORDER BY node_id) AS identity_source_ids, + count(*)::integer AS identity_member_count + FROM mapped + GROUP BY canonical_node_id +) +SELECT r.canonical_node_id AS node_id, + r.name, + r.lat, + r.lon, + r.last_seen, + a.is_online, + r.hardware_model, + r.firmware_version, + r.public_key, + a.created_at, + r.iata, + r.role, + a.advert_count, + r.elevation_m, + r.network, + a.last_predicted_online_at, + a.last_path_evidence_at, + a.last_mqtt_observer_seen_at, + a.last_rx_at, + a.last_status_at, + r.observer_iata, + a.identity_source_ids, + a.identity_member_count + FROM ranked r + JOIN aggregated a ON a.canonical_node_id = r.canonical_node_id + WHERE r.representative_rank = 1; + +CREATE VIEW node_identity_sightings AS +SELECT meshcore_canonical_node_id(node_id) AS node_id, + network, + min(first_seen_at) AS first_seen_at, + max(last_seen_at) AS last_seen_at + FROM node_network_sightings + GROUP BY meshcore_canonical_node_id(node_id), network; + +CREATE VIEW node_identity_status_samples AS +SELECT time, + meshcore_canonical_node_id(node_id) AS node_id, + network, + battery_mv, + uptime_secs, + tx_air_secs, + rx_air_secs, + channel_utilization, + air_util_tx, + stats + FROM node_status_samples; + +-- Packet observations retain all raw columns but expose canonical observer and +-- source keys. Public privacy queries may continue to use raw packets when +-- they need the original private-prefix evidence. +CREATE VIEW node_identity_packets AS +SELECT time, + packet_hash, + meshcore_canonical_node_id(rx_node_id) AS rx_node_id, + meshcore_canonical_node_id(src_node_id) AS src_node_id, + topic, + packet_type, + route_type, + hop_count, + rssi, + snr, + payload, + raw_hex, + advert_count, + path_hashes, + network, + path_hash_size_bytes, + transport_codes, + region_scope, + companion_sender, + topic_prefix, + iata, + is_private, + visibility_ok + FROM packets; + +-- Link rows are collapsed by the canonical unordered endpoint pair. Direction +-- counters are rotated when the raw key order differs from the canonical order. +CREATE VIEW node_identity_links AS +WITH mapped AS ( + SELECT meshcore_canonical_node_id(node_a_id) AS raw_a, + meshcore_canonical_node_id(node_b_id) AS raw_b, + observed_count, + last_observed, + itm_path_loss_db, + itm_viable, + itm_computed_at, + count_a_to_b, + count_b_to_a, + force_viable, + multibyte_observed_count, + terrain_profile_json + FROM node_links +), oriented AS ( + SELECT LEAST(raw_a, raw_b) AS node_a_id, + GREATEST(raw_a, raw_b) AS node_b_id, + observed_count, + last_observed, + itm_path_loss_db, + itm_viable, + itm_computed_at, + CASE WHEN raw_a <= raw_b THEN count_a_to_b ELSE count_b_to_a END AS count_a_to_b, + CASE WHEN raw_a <= raw_b THEN count_b_to_a ELSE count_a_to_b END AS count_b_to_a, + force_viable, + multibyte_observed_count, + terrain_profile_json + FROM mapped + WHERE raw_a <> raw_b +) +SELECT node_a_id, + node_b_id, + sum(observed_count)::integer AS observed_count, + max(last_observed) AS last_observed, + min(itm_path_loss_db) AS itm_path_loss_db, + bool_or(itm_viable) AS itm_viable, + max(itm_computed_at) AS itm_computed_at, + sum(count_a_to_b)::integer AS count_a_to_b, + sum(count_b_to_a)::integer AS count_b_to_a, + bool_or(force_viable) AS force_viable, + sum(multibyte_observed_count)::integer AS multibyte_observed_count, + (array_agg(terrain_profile_json ORDER BY last_observed DESC NULLS LAST) + FILTER (WHERE terrain_profile_json IS NOT NULL))[1] AS terrain_profile_json + FROM oriented + GROUP BY node_a_id, node_b_id; + +CREATE VIEW node_identity_link_radio_reports AS +WITH mapped AS ( + SELECT meshcore_canonical_node_id(node_a_id) AS raw_a, + meshcore_canonical_node_id(node_b_id) AS raw_b, + meshcore_canonical_node_id(reporter_node_id) AS reporter_node_id, + meshcore_canonical_node_id(peer_node_id) AS peer_node_id, + last_snr_db, + best_snr_db, + last_seen, + sample_count + FROM node_link_radio_reports +), oriented AS ( + SELECT LEAST(raw_a, raw_b) AS node_a_id, + GREATEST(raw_a, raw_b) AS node_b_id, + reporter_node_id, + peer_node_id, + last_snr_db, + best_snr_db, + last_seen, + sample_count + FROM mapped + WHERE raw_a <> raw_b +) +SELECT node_a_id, + node_b_id, + reporter_node_id, + peer_node_id, + (array_agg(last_snr_db ORDER BY last_seen DESC NULLS LAST))[1] AS last_snr_db, + max(best_snr_db) AS best_snr_db, + max(last_seen) AS last_seen, + sum(sample_count)::integer AS sample_count + FROM oriented + GROUP BY node_a_id, node_b_id, reporter_node_id, peer_node_id; diff --git a/backend/src/db/nodeIdentity.test.ts b/backend/src/db/nodeIdentity.test.ts new file mode 100644 index 0000000..43fdb16 --- /dev/null +++ b/backend/src/db/nodeIdentity.test.ts @@ -0,0 +1,125 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + assessIdentityPair, + buildIdentityGroups, + type IdentityEvidence, + type IdentityEvidenceMaps, + type IdentityNode, +} from './nodeIdentity.js'; + +const key = (character: string) => character.repeat(64); + +const evidence = (overrides: Partial = {}): IdentityEvidence => ({ + statusCount: 0, + statusFirstAt: null, + statusLastAt: null, + selfAdvertCount: 0, + selfFirstAt: null, + selfLastAt: null, + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + ...overrides, +}); + +const node = (overrides: Partial = {}): IdentityNode => ({ + node_id: key('A'), + name: 'GNOME-MSG-RPT', + lat: 54.52, + lon: -1.47, + role: 2, + advert_count: 100, + last_seen: '2026-08-06T00:00:00Z', + network: 'ukmesh', + hardware_model: null, + firmware_version: null, + ...overrides, +}); + +const emptyMaps = (): IdentityEvidenceMaps => ({ + statuses: new Map(), + selfAdverts: new Map(), + pairs: new Map(), +}); + +test('accepts a same-position one-nibble key rotation', () => { + const assessment = assessIdentityPair( + node(), + node({ node_id: `${'A'.repeat(63)}B` }), + evidence(), + evidence(), + ); + + assert.equal(assessment.accepted, true); + assert.equal(assessment.confidence, 'high'); + assert.match(assessment.reason, /one-nibble-key-rotation/); +}); + +test('accepts a status observer paired with a positioned advert identity', () => { + const assessment = assessIdentityPair( + node({ node_id: key('B'), role: null, lat: null, lon: null, advert_count: 0 }), + node({ node_id: key('C') }), + evidence({ statusCount: 4 }), + evidence(), + { pairAdvertCount: 6, pairPacketCount: 6, pairFirstAt: Date.parse('2026-07-01'), pairLastAt: Date.parse('2026-07-02') }, + ); + + assert.equal(assessment.accepted, true); + assert.equal(assessment.confidence, 'high'); + assert.match(assessment.reason, /observer-advert-pair/); +}); + +test('does not merge same-name positioned repeaters without independent evidence', () => { + const left = node({ node_id: key('D') }); + const right = node({ node_id: key('E'), lat: 55.1, lon: -1.47 }); + const result = buildIdentityGroups([left, right], emptyMaps()); + + assert.equal(result.aliases.length, 0); + assert.equal(result.assessments.length, 1); + assert.equal(result.assessments[0]?.accepted, false); + assert.equal(result.assessments[0]?.confidence, 'low'); +}); + +test('does not merge colocated ordinal repeaters from observer packet pairing', () => { + const assessment = assessIdentityPair( + node({ node_id: key('J'), name: 'Dunston-1' }), + node({ node_id: key('K'), name: 'Dunston-2' }), + evidence({ statusCount: 10 }), + evidence({ statusCount: 10 }), + { pairAdvertCount: 20, pairPacketCount: 20, pairFirstAt: Date.parse('2026-07-01'), pairLastAt: Date.parse('2026-08-01') }, + ); + + assert.equal(assessment.accepted, false); + assert.match(assessment.reason, /ordinal-or-direction-variant/); +}); + +test('folds a metadata-only status row only when the active repeater is unique', () => { + const stale = node({ node_id: key('F'), role: null, lat: null, lon: null, advert_count: 0 }); + const active = node({ node_id: key('G') }); + const assessment = assessIdentityPair( + stale, + active, + evidence({ statusCount: 12 }), + evidence(), + undefined, + { uniqueActiveRepeaterId: active.node_id }, + ); + + assert.equal(assessment.accepted, true); + assert.equal(assessment.confidence, 'medium'); +}); + +test('canonical selection prefers the active positioned repeater row', () => { + const stale = node({ node_id: key('H'), role: null, lat: null, lon: null, advert_count: 0 }); + const active = node({ node_id: key('I') }); + const maps = emptyMaps(); + maps.statuses.set(stale.node_id, { count: 1, firstAt: Date.parse('2026-01-01'), lastAt: Date.parse('2026-08-01') }); + + const result = buildIdentityGroups([stale, active], maps); + assert.equal(result.aliases.length, 1); + assert.equal(result.aliases[0]?.sourceNodeId, stale.node_id); + assert.equal(result.aliases[0]?.canonicalNodeId, active.node_id); + assert.equal(result.aliases[0]?.confidence, 'medium'); +}); diff --git a/backend/src/db/nodeIdentity.ts b/backend/src/db/nodeIdentity.ts new file mode 100644 index 0000000..940e12d --- /dev/null +++ b/backend/src/db/nodeIdentity.ts @@ -0,0 +1,840 @@ +import type pg from 'pg'; + +export type IdentityConfidence = 'high' | 'medium'; + +export type IdentityNode = { + node_id: string; + name: string | null; + lat: number | null; + lon: number | null; + role: number | null; + advert_count: number | null; + last_seen: Date | string | null; + network: string | null; + hardware_model: string | null; + firmware_version: string | null; +}; + +export type IdentityEvidence = { + statusCount: number; + statusFirstAt: number | null; + statusLastAt: number | null; + selfAdvertCount: number; + selfFirstAt: number | null; + selfLastAt: number | null; + pairAdvertCount: number; + pairPacketCount: number; + pairFirstAt: number | null; + pairLastAt: number | null; +}; + +export type IdentityPairAssessment = { + nodeAId: string; + nodeBId: string; + accepted: boolean; + confidence: IdentityConfidence | 'low'; + score: number; + reason: string; + reasons: string[]; + evidence: Record; +}; + +export type IdentityAlias = { + sourceNodeId: string; + canonicalNodeId: string; + confidence: IdentityConfidence; + reason: string; + evidence: Record; +}; + +export type IdentityRefreshResult = { + nodesConsidered: number; + candidatePairs: number; + acceptedPairs: number; + aliasesWritten: number; + ambiguousPairs: number; + canonicalGroups: number; +}; + +const MAX_NAME_DISTANCE_METERS = 3_000; +const MAX_HANDOVER_GAP_MS = 14 * 24 * 60 * 60 * 1_000; +const MAX_EVIDENCE_AGE_DAYS = 365; + +type PairKey = `${string}:${string}`; + +type PairEvidence = { + pairAdvertCount: number; + pairPacketCount: number; + pairFirstAt: number | null; + pairLastAt: number | null; +}; + +export type IdentityEvidenceMaps = { + statuses: Map; + selfAdverts: Map; + pairs: Map; +}; + +type IdentityEdge = IdentityPairAssessment & { + nodeA: IdentityNode; + nodeB: IdentityNode; +}; + +type IdentityGroup = { + members: IdentityNode[]; + edges: IdentityEdge[]; +}; + +function timeMs(value: unknown): number | null { + if (value == null) return null; + const timestamp = value instanceof Date ? value.getTime() : Date.parse(String(value)); + return Number.isFinite(timestamp) ? timestamp : null; +} + +function validPosition(node: Pick): boolean { + return node.lat != null + && node.lon != null + && Number.isFinite(node.lat) + && Number.isFinite(node.lon) + && Math.abs(node.lat) <= 90 + && Math.abs(node.lon) <= 180 + && !(Math.abs(node.lat) < 1e-9 && Math.abs(node.lon) < 1e-9); +} + +function distanceMeters(a: IdentityNode, b: IdentityNode): number | null { + if (!validPosition(a) || !validPosition(b)) return null; + const midLat = ((a.lat! + b.lat!) / 2) * Math.PI / 180; + const dLat = (b.lat! - a.lat!) * 111_320; + const dLon = (b.lon! - a.lon!) * 111_320 * Math.cos(midLat); + return Math.sqrt(dLat ** 2 + dLon ** 2); +} + +function normalizedNameTokens(value: string | null): string[] { + if (!value) return []; + return value + .normalize('NFKC') + .toUpperCase() + .replace(/V\d+(?:\.\d+)?/g, ' ') + .replace(/[^\p{L}\p{N}]+/gu, ' ') + .trim() + .split(/\s+/) + .filter(Boolean) + .map((token) => token === 'RPT' ? 'REPEATER' : token); +} + +/** + * A shared site name with a changed ordinal/directional suffix is commonly a + * pair of real colocated repeaters, not a key rotation. Keep that distinction + * as an explicit veto for evidence that is weaker than a near-identical key. + * Version suffixes (V2/V3/V4) are removed by normalizedNameTokens and are not + * treated as ordinals here. + */ +function nameVariantConflict(a: string | null, b: string | null): boolean { + const tokens = (value: string | null) => { + if (!value) return [] as string[]; + return value + .normalize('NFKC') + .toUpperCase() + .replace(/V\d+(?:\.\d+)?/g, ' ') + .replace(/[^\p{L}\p{N}]+/gu, ' ') + .trim() + .split(/\s+/) + .filter(Boolean) + .map((token) => token === 'RPT' ? 'REPEATER' : token); + }; + const left = tokens(a); + const right = tokens(b); + if (left.length < 2 || right.length < 2 || left.length !== right.length) return false; + return left.slice(0, -1).join('|') === right.slice(0, -1).join('|') + && left[left.length - 1] !== right[right.length - 1]; +} + +function identityNameBucket(tokens: string[]): string { + const significant = tokens.filter((token) => !['REPEATER', 'NODE'].includes(token)); + return (significant.length > 0 ? significant : tokens).slice(0, 2).join('|'); +} + +function nameSimilarity(a: string | null, b: string | null): number { + const aTokens = normalizedNameTokens(a); + const bTokens = normalizedNameTokens(b); + if (aTokens.length === 0 || bTokens.length === 0) return 0; + const aSet = new Set(aTokens); + const bSet = new Set(bTokens); + const intersection = [...aSet].filter((token) => bSet.has(token)).length; + const union = new Set([...aSet, ...bSet]).size; + return union > 0 ? intersection / union : 0; +} + +function namesMatch(a: string | null, b: string | null): boolean { + if (!a || !b || a.includes('🚫') || b.includes('🚫')) return false; + const aTokens = normalizedNameTokens(a); + const bTokens = normalizedNameTokens(b); + if (aTokens.length === 0 || bTokens.length === 0) return false; + if (identityNameBucket(aTokens) !== identityNameBucket(bTokens)) return false; + const similarity = nameSimilarity(a, b); + // A one-token name is deliberately permitted only when another independent + // identity signal (key, position, or packet evidence) is also present. + return similarity >= 0.6; +} + +function publicNetworkClass(network: string | null): string | null { + if (!network || network === 'test') return null; + if (network === 'ukmesh' || network === 'northeast' || network === 'teesside') return 'public'; + return network; +} + +function repeaterLike(node: IdentityNode): boolean { + return node.role == null || node.role === 2; +} + +function activePositionedRepeater(node: IdentityNode): boolean { + return node.role === 2 && validPosition(node) && Number(node.advert_count ?? 0) > 0; +} + +function roleCompatible(a: IdentityNode, b: IdentityNode): boolean { + if (!repeaterLike(a) || !repeaterLike(b)) return false; + return a.role == null || b.role == null || a.role === b.role; +} + +function nearKeyAndPosition(a: IdentityNode, b: IdentityNode): boolean { + const keyDistance = hexNibbleDistance(a.node_id, b.node_id); + const distance = distanceMeters(a, b); + return keyDistance <= 1 && distance != null && distance <= MAX_NAME_DISTANCE_METERS; +} + +function uniqueActiveRepeaterId(nodes: IdentityNode[]): string | null { + const representatives: IdentityNode[] = []; + for (const candidate of nodes.filter(activePositionedRepeater)) { + if (!representatives.some((representative) => nearKeyAndPosition(candidate, representative))) { + representatives.push(candidate); + } + } + return representatives.length === 1 ? representatives[0]!.node_id : null; +} + +function hexNibbleDistance(a: string, b: string): number { + const left = a.toUpperCase(); + const right = b.toUpperCase(); + if (!/^[0-9A-F]{64}$/.test(left) || !/^[0-9A-F]{64}$/.test(right)) return 64; + let distance = 0; + for (let index = 0; index < left.length; index += 1) { + if (left[index] !== right[index]) distance += 1; + } + return distance; +} + +function keyFor(a: string, b: string): PairKey { + return a < b ? `${a}:${b}` : `${b}:${a}`; +} + +function intervalGapMs( + aStart: number | null, + aEnd: number | null, + bStart: number | null, + bEnd: number | null, +): number | null { + if (aStart == null || aEnd == null || bStart == null || bEnd == null) return null; + if (aStart <= bEnd && bStart <= aEnd) return 0; + return aStart > bEnd ? aStart - bEnd : bStart - aEnd; +} + +function nodeActivityInterval(node: IdentityNode, evidence: IdentityEvidence): [number | null, number | null] { + const nodeLast = timeMs(node.last_seen); + const first = evidence.selfFirstAt ?? evidence.pairFirstAt ?? timeMs(node.last_seen); + const last = Math.max( + evidence.selfLastAt ?? 0, + evidence.pairLastAt ?? 0, + nodeLast ?? 0, + ) || null; + return [first, last]; +} + +function evidenceRecord( + a: IdentityNode, + b: IdentityNode, + evidenceA: IdentityEvidence, + evidenceB: IdentityEvidence, + pair: PairEvidence, + details: Record, +): Record { + return { + node_a: a.node_id, + node_b: b.node_id, + name_a: a.name, + name_b: b.name, + name_similarity: nameSimilarity(a.name, b.name), + key_nibble_distance: hexNibbleDistance(a.node_id, b.node_id), + distance_m: distanceMeters(a, b), + status_a: evidenceA, + status_b: evidenceB, + packet_pair: pair, + ...details, + }; +} + +/** + * Explain and score one possible identity edge. This is intentionally pure so + * the merge rule can be tested without a database. + */ +export function assessIdentityPair( + a: IdentityNode, + b: IdentityNode, + evidenceA: IdentityEvidence, + evidenceB: IdentityEvidence, + pair: PairEvidence = { + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + }, + options: { uniqueActiveRepeaterId?: string | null } = {}, +): IdentityPairAssessment { + const reasons: string[] = []; + const keyDistance = hexNibbleDistance(a.node_id, b.node_id); + const distance = distanceMeters(a, b); + const nameMatch = namesMatch(a.name, b.name); + const samePublicNetwork = publicNetworkClass(a.network) !== null + && publicNetworkClass(a.network) === publicNetworkClass(b.network); + const roleMatch = roleCompatible(a, b); + const statusA = evidenceA.statusCount > 0; + const statusB = evidenceB.statusCount > 0; + const directAdvertPair = pair.pairAdvertCount >= 3 && pair.pairPacketCount >= 3; + const selfA = evidenceA.selfAdvertCount >= 3; + const selfB = evidenceB.selfAdvertCount >= 3; + const [aStart, aEnd] = nodeActivityInterval(a, evidenceA); + const [bStart, bEnd] = nodeActivityInterval(b, evidenceB); + const activityGap = intervalGapMs(aStart, aEnd, bStart, bEnd); + const handover = selfA + && selfB + && activityGap != null + && activityGap <= MAX_HANDOVER_GAP_MS + && ((aEnd != null && bStart != null && aEnd < bStart) + || (bEnd != null && aStart != null && bEnd < aStart)); + const variantConflict = nameVariantConflict(a.name, b.name); + const metadataOnlyA = a.role == null && !validPosition(a) && Number(a.advert_count ?? 0) === 0; + const metadataOnlyB = b.role == null && !validPosition(b) && Number(b.advert_count ?? 0) === 0; + if (!samePublicNetwork) reasons.push('different-network-family'); + if (!roleMatch) reasons.push('incompatible-role'); + if (!nameMatch) reasons.push('name-not-similar'); + if (keyDistance <= 1) reasons.push('one-nibble-key-rotation'); + if (distance != null && distance <= MAX_NAME_DISTANCE_METERS) reasons.push('same-position'); + if (directAdvertPair) reasons.push('observer-advert-pair'); + if (handover) reasons.push('temporal-self-advert-handover'); + if (variantConflict) reasons.push('ordinal-or-direction-variant'); + if (statusA && activePositionedRepeater(b)) reasons.push('status-to-positioned-repeater'); + if (statusB && activePositionedRepeater(a)) reasons.push('status-to-positioned-repeater'); + + // A key rotation with matching advert self-reception, or direct observer → + // advert observations, is stronger than name similarity alone. A one-nibble + // key match additionally requires the same physical position. + const nearKeyAndPositionMatch = nearKeyAndPosition(a, b); + const acceptedHigh = samePublicNetwork + && roleMatch + && nameMatch + && ( + nearKeyAndPositionMatch + || (directAdvertPair + && !variantConflict + && (metadataOnlyA || metadataOnlyB) + && (statusA || statusB) + && (activePositionedRepeater(a) || activePositionedRepeater(b))) + || (handover + && !variantConflict + && (statusA || statusB) + && (activePositionedRepeater(a) || activePositionedRepeater(b))) + ); + + // A metadata-only, stale MQTT status row may be folded into a unique active + // repeater with the same identity name. The caller supplies that uniqueness; + // generic names with multiple positioned repeaters never reach this branch. + const uniqueActivePair = options.uniqueActiveRepeaterId != null + && (metadataOnlyA || metadataOnlyB) + && (activePositionedRepeater(a) || activePositionedRepeater(b)); + const acceptedMedium = samePublicNetwork + && roleMatch + && nameMatch + && !variantConflict + && uniqueActivePair + && (statusA || statusB) + && options.uniqueActiveRepeaterId === (activePositionedRepeater(a) ? a.node_id : b.node_id); + + const accepted = acceptedHigh || acceptedMedium; + const confidence: IdentityPairAssessment['confidence'] = acceptedHigh + ? 'high' + : acceptedMedium + ? 'medium' + : 'low'; + const score = [ + nameMatch, + samePublicNetwork, + roleMatch, + keyDistance <= 1, + distance != null && distance <= MAX_NAME_DISTANCE_METERS, + directAdvertPair, + handover, + statusA || statusB, + ].filter(Boolean).length; + + return { + nodeAId: a.node_id, + nodeBId: b.node_id, + accepted, + confidence, + score, + reason: accepted ? reasons.join('+') : (reasons.length > 0 ? reasons.join('+') : 'insufficient-evidence'), + reasons, + evidence: evidenceRecord(a, b, evidenceA, evidenceB, pair, { + activity_gap_ms: activityGap, + direct_observer_advert_pair: directAdvertPair, + temporal_handover: handover, + name_variant_conflict: variantConflict, + metadata_only_alias: acceptedMedium, + }), + }; +} + +function findRoot(parent: Map, nodeId: string): string { + let root = nodeId; + while (parent.get(root) !== root) root = parent.get(root)!; + let current = nodeId; + while (parent.get(current) !== current) { + const next = parent.get(current)!; + parent.set(current, root); + current = next; + } + return root; +} + +function canJoinGroups( + a: IdentityGroup, + b: IdentityGroup, +): boolean { + const positioned = [...a.members, ...b.members].filter(validPosition); + for (let left = 0; left < positioned.length; left += 1) { + for (let right = left + 1; right < positioned.length; right += 1) { + const distance = distanceMeters(positioned[left]!, positioned[right]!); + if (distance != null && distance > MAX_NAME_DISTANCE_METERS) return false; + } + } + return true; +} + +function chooseCanonicalNode(nodes: IdentityNode[]): IdentityNode { + return [...nodes].sort((a, b) => { + const activeDiff = Number(activePositionedRepeater(a)) - Number(activePositionedRepeater(b)); + if (activeDiff !== 0) return -activeDiff; + const advertDiff = Number(b.advert_count ?? 0) - Number(a.advert_count ?? 0); + if (advertDiff !== 0) return advertDiff; + const lastDiff = (timeMs(b.last_seen) ?? 0) - (timeMs(a.last_seen) ?? 0); + if (lastDiff !== 0) return lastDiff; + return a.node_id.localeCompare(b.node_id); + })[0]!; +} + +export type IdentityGroupingResult = { + aliases: IdentityAlias[]; + assessments: IdentityPairAssessment[]; + groups: IdentityGroup[]; +}; + +/** Build accepted components from the candidate evidence graph. */ +export function buildIdentityGroups( + nodes: IdentityNode[], + evidence: IdentityEvidenceMaps, +): IdentityGroupingResult { + const byBucket = new Map(); + for (const node of nodes) { + if (publicNetworkClass(node.network) == null || !repeaterLike(node)) continue; + const tokens = normalizedNameTokens(node.name); + if (tokens.length === 0) continue; + const bucket = `${publicNetworkClass(node.network)}:${identityNameBucket(tokens)}`; + const values = byBucket.get(bucket) ?? []; + values.push(node); + byBucket.set(bucket, values); + } + + const assessments: IdentityPairAssessment[] = []; + const edges: IdentityEdge[] = []; + for (const bucketNodes of byBucket.values()) { + for (let left = 0; left < bucketNodes.length; left += 1) { + for (let right = left + 1; right < bucketNodes.length; right += 1) { + const a = bucketNodes[left]!; + const b = bucketNodes[right]!; + const evidenceA = evidence.statuses.has(a.node_id) + ? { + statusCount: evidence.statuses.get(a.node_id)!.count, + statusFirstAt: evidence.statuses.get(a.node_id)!.firstAt, + statusLastAt: evidence.statuses.get(a.node_id)!.lastAt, + selfAdvertCount: evidence.selfAdverts.get(a.node_id)?.count ?? 0, + selfFirstAt: evidence.selfAdverts.get(a.node_id)?.firstAt ?? null, + selfLastAt: evidence.selfAdverts.get(a.node_id)?.lastAt ?? null, + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + } + : { + statusCount: 0, + statusFirstAt: null, + statusLastAt: null, + selfAdvertCount: evidence.selfAdverts.get(a.node_id)?.count ?? 0, + selfFirstAt: evidence.selfAdverts.get(a.node_id)?.firstAt ?? null, + selfLastAt: evidence.selfAdverts.get(a.node_id)?.lastAt ?? null, + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + }; + const evidenceB = evidence.statuses.has(b.node_id) + ? { + statusCount: evidence.statuses.get(b.node_id)!.count, + statusFirstAt: evidence.statuses.get(b.node_id)!.firstAt, + statusLastAt: evidence.statuses.get(b.node_id)!.lastAt, + selfAdvertCount: evidence.selfAdverts.get(b.node_id)?.count ?? 0, + selfFirstAt: evidence.selfAdverts.get(b.node_id)?.firstAt ?? null, + selfLastAt: evidence.selfAdverts.get(b.node_id)?.lastAt ?? null, + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + } + : { + statusCount: 0, + statusFirstAt: null, + statusLastAt: null, + selfAdvertCount: evidence.selfAdverts.get(b.node_id)?.count ?? 0, + selfFirstAt: evidence.selfAdverts.get(b.node_id)?.firstAt ?? null, + selfLastAt: evidence.selfAdverts.get(b.node_id)?.lastAt ?? null, + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + }; + const pair = evidence.pairs.get(keyFor(a.node_id, b.node_id)) ?? { + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + }; + const activeId = uniqueActiveRepeaterId(bucketNodes); + const assessment = assessIdentityPair(a, b, evidenceA, evidenceB, pair, { + uniqueActiveRepeaterId: activeId, + }); + assessments.push(assessment); + if (assessment.accepted) edges.push({ ...assessment, nodeA: a, nodeB: b }); + } + } + } + + const parent = new Map(nodes.map((node) => [node.node_id, node.node_id])); + const groups = new Map(); + for (const node of nodes) groups.set(node.node_id, { members: [node], edges: [] }); + + const sortedEdges = [...edges].sort((a, b) => { + const confidenceDiff = Number(b.confidence === 'high') - Number(a.confidence === 'high'); + return confidenceDiff || b.score - a.score || a.nodeAId.localeCompare(b.nodeAId); + }); + for (const edge of sortedEdges) { + const rootA = findRoot(parent, edge.nodeAId); + const rootB = findRoot(parent, edge.nodeBId); + if (rootA === rootB) continue; + const groupA = groups.get(rootA)!; + const groupB = groups.get(rootB)!; + if (!canJoinGroups(groupA, groupB)) continue; + parent.set(rootB, rootA); + groupA.members.push(...groupB.members); + groupA.edges.push(...groupB.edges, edge); + groups.delete(rootB); + } + + const aliases: IdentityAlias[] = []; + const acceptedPairKeys = new Set(); + for (const group of groups.values()) { + if (group.members.length < 2) continue; + const canonical = chooseCanonicalNode(group.members); + const groupConfidence: IdentityConfidence = group.edges.some((edge) => edge.confidence === 'high') + ? 'high' + : 'medium'; + const groupEvidence = { + canonical_node_id: canonical.node_id, + members: group.members.map((member) => member.node_id).sort(), + edges: group.edges.map((edge) => edge.evidence), + }; + for (const edge of group.edges) acceptedPairKeys.add(keyFor(edge.nodeAId, edge.nodeBId)); + for (const member of group.members) { + if (member.node_id === canonical.node_id) continue; + const memberEdges = group.edges.filter((edge) => ( + edge.nodeAId === member.node_id || edge.nodeBId === member.node_id + )); + const memberConfidence: IdentityConfidence = memberEdges.some((edge) => edge.confidence === 'high') + ? 'high' + : 'medium'; + aliases.push({ + sourceNodeId: member.node_id, + canonicalNodeId: canonical.node_id, + confidence: memberEdges.length > 0 ? memberConfidence : groupConfidence, + reason: (memberEdges.length > 0 ? memberEdges : group.edges) + .map((edge) => edge.reason) + .join(';'), + evidence: { + ...groupEvidence, + supporting_edges: (memberEdges.length > 0 ? memberEdges : group.edges) + .map((edge) => edge.evidence), + }, + }); + } + } + + // Mark rejected/uncertain same-name candidates as ambiguous for operator + // review, but never make an alias from name similarity alone. + const finalAssessments = assessments.map((assessment) => ({ + ...assessment, + accepted: acceptedPairKeys.has(keyFor(assessment.nodeAId, assessment.nodeBId)), + confidence: acceptedPairKeys.has(keyFor(assessment.nodeAId, assessment.nodeBId)) + ? assessment.confidence + : 'low' as const, + })); + return { + aliases, + assessments: finalAssessments, + groups: [...groups.values()].filter((group) => group.members.length > 1), + }; +} + +type DbClient = { + query(text: string, values?: unknown[]): Promise<{ rows: T[]; rowCount: number | null }>; + release(): void; +}; + +export async function refreshNodeIdentityAliases( + pool: { connect(): Promise }, +): Promise { + const client = await pool.connect(); + const refreshStartedAt = new Date(); + try { + await client.query('BEGIN'); + await client.query(`SELECT pg_advisory_xact_lock(hashtext('meshcore-node-identity-refresh'))`); + + const nodesResult = await client.query( + `SELECT node_id, name, lat, lon, role, advert_count, last_seen, + network, hardware_model, firmware_version + FROM nodes + WHERE network IS DISTINCT FROM 'test'`, + ); + const nodes = nodesResult.rows.map((row) => ({ + ...row, + node_id: row.node_id.toUpperCase(), + })); + const candidatePairs: Array<[string, string]> = []; + const candidateIds = new Set(); + const buckets = new Map(); + for (const node of nodes) { + if (!repeaterLike(node) || !node.name || node.name.includes('🚫')) continue; + const tokens = normalizedNameTokens(node.name); + if (tokens.length === 0) continue; + const bucket = `${publicNetworkClass(node.network)}:${identityNameBucket(tokens)}`; + const bucketNodes = buckets.get(bucket) ?? []; + bucketNodes.push(node); + buckets.set(bucket, bucketNodes); + } + for (const bucketNodes of buckets.values()) { + for (let left = 0; left < bucketNodes.length; left += 1) { + for (let right = left + 1; right < bucketNodes.length; right += 1) { + const a = bucketNodes[left]!; + const b = bucketNodes[right]!; + if (!namesMatch(a.name, b.name) || !roleCompatible(a, b)) continue; + candidatePairs.push([a.node_id, b.node_id]); + candidateIds.add(a.node_id); + candidateIds.add(b.node_id); + } + } + } + + const ids = [...candidateIds]; + const statuses = new Map(); + const selfAdverts = new Map(); + const pairs = new Map(); + if (ids.length > 0) { + const statusResult = await client.query<{ + node_id: string; + count: string; + first_at: Date | string | null; + last_at: Date | string | null; + }>( + `SELECT node_id, COUNT(*)::text AS count, + MIN(time) AS first_at, MAX(time) AS last_at + FROM node_status_samples + WHERE node_id = ANY($1::text[]) + GROUP BY node_id`, + [ids], + ); + for (const row of statusResult.rows) { + statuses.set(row.node_id, { + count: Number(row.count ?? 0), + firstAt: timeMs(row.first_at), + lastAt: timeMs(row.last_at), + }); + } + + const selfResult = await client.query<{ + node_id: string; + count: string; + first_at: Date | string | null; + last_at: Date | string | null; + }>( + `SELECT rx_node_id AS node_id, COUNT(*)::text AS count, + MIN(time) AS first_at, MAX(time) AS last_at + FROM packets + WHERE rx_node_id = src_node_id + AND packet_type = 4 + AND rx_node_id = ANY($1::text[]) + AND time > NOW() - ($2::text || ' days')::interval + GROUP BY rx_node_id`, + [ids, String(MAX_EVIDENCE_AGE_DAYS)], + ); + for (const row of selfResult.rows) { + selfAdverts.set(row.node_id, { + count: Number(row.count ?? 0), + firstAt: timeMs(row.first_at), + lastAt: timeMs(row.last_at), + }); + } + + const pairResult = await client.query<{ + rx_node_id: string; + src_node_id: string; + packet_count: string; + advert_count: string; + first_at: Date | string | null; + last_at: Date | string | null; + }>( + `SELECT rx_node_id, src_node_id, + COUNT(*)::text AS packet_count, + COUNT(*) FILTER (WHERE packet_type = 4)::text AS advert_count, + MIN(time) AS first_at, MAX(time) AS last_at + FROM packets + WHERE rx_node_id = ANY($1::text[]) + AND src_node_id = ANY($1::text[]) + AND rx_node_id <> src_node_id + AND time > NOW() - ($2::text || ' days')::interval + GROUP BY rx_node_id, src_node_id`, + [ids, String(MAX_EVIDENCE_AGE_DAYS)], + ); + for (const row of pairResult.rows) { + const key = keyFor(row.rx_node_id, row.src_node_id); + const current = pairs.get(key) ?? { + pairAdvertCount: 0, + pairPacketCount: 0, + pairFirstAt: null, + pairLastAt: null, + }; + current.pairAdvertCount += Number(row.advert_count ?? 0); + current.pairPacketCount += Number(row.packet_count ?? 0); + const firstAt = timeMs(row.first_at); + const lastAt = timeMs(row.last_at); + current.pairFirstAt = current.pairFirstAt == null || (firstAt != null && firstAt < current.pairFirstAt) + ? firstAt + : current.pairFirstAt; + current.pairLastAt = current.pairLastAt == null || (lastAt != null && lastAt > current.pairLastAt) + ? lastAt + : current.pairLastAt; + pairs.set(key, current); + } + } + + const grouped = buildIdentityGroups(nodes, { statuses, selfAdverts, pairs }); + + await client.query(`DELETE FROM node_identity_aliases WHERE source_kind = 'automatic'`); + if (grouped.aliases.length > 0) { + const params: unknown[] = []; + const values = grouped.aliases.map((alias, index) => { + const offset = index * 6; + params.push( + alias.sourceNodeId, + alias.canonicalNodeId, + alias.confidence, + alias.reason, + JSON.stringify(alias.evidence), + refreshStartedAt, + ); + return `($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}::jsonb, 'automatic', $${offset + 6})`; + }).join(', '); + await client.query( + `INSERT INTO node_identity_aliases + (source_node_id, canonical_node_id, confidence, reason, evidence, source_kind, updated_at) + VALUES ${values} + ON CONFLICT (source_node_id) DO UPDATE SET + canonical_node_id = EXCLUDED.canonical_node_id, + confidence = EXCLUDED.confidence, + reason = EXCLUDED.reason, + evidence = EXCLUDED.evidence, + updated_at = EXCLUDED.updated_at + WHERE node_identity_aliases.source_kind = 'automatic'`, + params, + ); + } + + await client.query( + `DELETE FROM node_identity_match_evidence + WHERE source_kind = 'automatic' AND updated_at < $1`, + [refreshStartedAt], + ); + if (grouped.assessments.length > 0) { + // Keep each statement well below PostgreSQL's bind-parameter limit. A + // common-name bucket can legitimately produce thousands of ambiguous + // pairs, all of which are retained as audit evidence. + const rowsPerWrite = 250; + for (let start = 0; start < grouped.assessments.length; start += rowsPerWrite) { + const chunk = grouped.assessments.slice(start, start + rowsPerWrite); + const params: unknown[] = []; + const values = chunk.map((assessment, index) => { + const [nodeAId, nodeBId] = assessment.nodeAId < assessment.nodeBId + ? [assessment.nodeAId, assessment.nodeBId] + : [assessment.nodeBId, assessment.nodeAId]; + const offset = index * 9; + params.push( + nodeAId, + nodeBId, + assessment.accepted ? 'accepted' : 'ambiguous', + assessment.confidence, + assessment.reason, + JSON.stringify(assessment.evidence), + 'automatic', + refreshStartedAt, + assessment.score, + ); + return `($${offset + 1}, $${offset + 2}, $${offset + 3}, $${offset + 4}, $${offset + 5}, $${offset + 6}::jsonb, $${offset + 7}, $${offset + 8}, $${offset + 9})`; + }).join(', '); + await client.query( + `INSERT INTO node_identity_match_evidence + (node_a_id, node_b_id, decision, confidence, reason, evidence, source_kind, updated_at, score) + VALUES ${values} + ON CONFLICT (node_a_id, node_b_id) DO UPDATE SET + decision = EXCLUDED.decision, + confidence = EXCLUDED.confidence, + reason = EXCLUDED.reason, + evidence = EXCLUDED.evidence, + updated_at = EXCLUDED.updated_at, + score = EXCLUDED.score + WHERE node_identity_match_evidence.source_kind = 'automatic'`, + params, + ); + } + } + + await client.query('COMMIT'); + return { + nodesConsidered: nodes.length, + candidatePairs: candidatePairs.length, + acceptedPairs: grouped.assessments.filter((assessment) => assessment.accepted).length, + aliasesWritten: grouped.aliases.length, + ambiguousPairs: grouped.assessments.filter((assessment) => !assessment.accepted).length, + canonicalGroups: grouped.groups.length, + }; + } catch (error) { + await client.query('ROLLBACK'); + throw error; + } finally { + client.release(); + } +} diff --git a/backend/src/index.ts b/backend/src/index.ts index daeb337..630ad98 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -3,7 +3,8 @@ import http from 'node:http'; import express from 'express'; import compression from 'compression'; import cors from 'cors'; -import { closeDb, initDb, query } from './db/index.js'; +import { closeDb, initDb, pool, query } from './db/index.js'; +import { refreshNodeIdentityAliases } from './db/nodeIdentity.js'; import { closeOwnerAuthDb, getOwnerAclReadiness, initOwnerAuthDb } from './db/ownerAuth.js'; import { getMqttRuntimeStatus, startMqttClient, stopMqttClient, onPacket, onNodeSeen, onNodeUpsert } from './mqtt/client.js'; import { @@ -58,9 +59,11 @@ const SHUTDOWN_DEADLINE_MS = Math.min( 30_000, Math.max(1_000, Number(process.env['SHUTDOWN_DEADLINE_MS'] ?? 30_000) || 30_000), ); +const NODE_IDENTITY_REFRESH_INTERVAL_MS = 30 * 60_000; const lifecycle = new LifecycleCoordinator(SHUTDOWN_DEADLINE_MS); let shutdownExitCode = 0; let forceCloseHttpConnections = () => {}; +let nodeIdentityRefreshTimer: NodeJS.Timeout | null = null; lifecycle.register({ name: 'queue-admission', @@ -117,6 +120,14 @@ lifecycle.register({ stage: 40, close: closeDb, }); +lifecycle.register({ + name: 'node-identity-refresh', + stage: 10, + close: () => { + if (nodeIdentityRefreshTimer) clearInterval(nodeIdentityRefreshTimer); + nodeIdentityRefreshTimer = null; + }, +}); async function shutdown(reason: string, exitCode: number): Promise { shutdownExitCode = Math.max(shutdownExitCode, exitCode); @@ -155,6 +166,16 @@ process.on('uncaughtException', (error) => { async function main() { // 1. Initialise DB schema + retention policy await initDb(); + const identityRefresh = await refreshNodeIdentityAliases(pool); + console.log('[node-identity] refreshed', identityRefresh); + nodeIdentityRefreshTimer = setInterval(() => { + void refreshNodeIdentityAliases(pool) + .then((result) => console.log('[node-identity] refreshed', result)) + .catch((error: unknown) => { + console.error('[node-identity] refresh failed:', error instanceof Error ? error.message : error); + }); + }, NODE_IDENTITY_REFRESH_INTERVAL_MS); + nodeIdentityRefreshTimer.unref(); await initOwnerAuthDb(); await startOwnerAuthorizationReconciler(); diff --git a/backend/src/owner/ownerRepository.ts b/backend/src/owner/ownerRepository.ts index 57e7434..3f9f3a6 100644 --- a/backend/src/owner/ownerRepository.ts +++ b/backend/src/owner/ownerRepository.ts @@ -35,11 +35,11 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { }>( `WITH clashing_prefixes AS ( SELECT LEFT(node_id, 2) AS prefix - FROM nodes + FROM node_identity_nodes WHERE role IN (1, 3) INTERSECT SELECT LEFT(node_id, 2) AS prefix - FROM nodes + FROM node_identity_nodes WHERE role = 2 ), owner_packets AS ( @@ -64,8 +64,11 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { THEN UPPER(p.path_hashes[array_length(p.path_hashes, 1)]) ELSE NULL END AS receiver_side_hash - FROM packets p - WHERE p.rx_node_id = ANY($1::text[]) + FROM node_identity_packets p + WHERE p.rx_node_id = ANY( + SELECT meshcore_canonical_node_id(id) + FROM unnest($1::text[]) AS selected(id) + ) AND ${timeFilter} AND p.packet_type NOT IN (8, 9) AND p.src_node_id != p.rx_node_id @@ -93,7 +96,7 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { COUNT(*) OVER (PARTITION BY uh.rx_node_id, uh.receiver_side_hash) AS match_count, ROW_NUMBER() OVER (PARTITION BY uh.rx_node_id, uh.receiver_side_hash ORDER BY n.node_id) AS rn FROM unique_receiver_targets uh - JOIN nodes n + JOIN node_identity_nodes n ON (n.role IS NULL OR n.role NOT IN (1, 3)) AND UPPER(n.node_id) LIKE uh.receiver_side_hash || '%' ), @@ -116,11 +119,11 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { n.node_id ) AS rn FROM unique_receiver_targets uh - JOIN nodes rx ON rx.node_id = uh.rx_node_id - JOIN nodes n + JOIN node_identity_nodes rx ON rx.node_id = uh.rx_node_id + JOIN node_identity_nodes n ON (n.role IS NULL OR n.role NOT IN (1, 3)) AND UPPER(LEFT(n.node_id, 2)) = UPPER(LEFT(uh.receiver_side_hash, 2)) - LEFT JOIN node_links nl + LEFT JOIN node_identity_links nl ON ( (nl.node_a_id = uh.rx_node_id AND nl.node_b_id = n.node_id) OR (nl.node_b_id = uh.rx_node_id AND nl.node_a_id = n.node_id) @@ -137,7 +140,10 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { CASE WHEN op.hop_count = 0 AND op.src_node_id IS NOT NULL - AND NOT (op.src_node_id = ANY($2::text[])) + AND NOT (op.src_node_id = ANY( + SELECT meshcore_canonical_node_id(id) + FROM unnest($2::text[]) AS owned(id) + )) AND src.node_id IS NOT NULL AND (src.role IS NULL OR src.role NOT IN (1, 3)) THEN op.src_node_id @@ -148,7 +154,10 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { CASE WHEN op.hop_count = 0 AND op.src_node_id IS NOT NULL - AND NOT (op.src_node_id = ANY($2::text[])) + AND NOT (op.src_node_id = ANY( + SELECT meshcore_canonical_node_id(id) + FROM unnest($2::text[]) AS owned(id) + )) AND src.node_id IS NOT NULL AND (src.role IS NULL OR src.role NOT IN (1, 3)) THEN COALESCE(src.name, op.src_node_id) @@ -159,7 +168,10 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { CASE WHEN op.hop_count = 0 AND op.src_node_id IS NOT NULL - AND NOT (op.src_node_id = ANY($2::text[])) + AND NOT (op.src_node_id = ANY( + SELECT meshcore_canonical_node_id(id) + FROM unnest($2::text[]) AS owned(id) + )) AND src.node_id IS NOT NULL AND (src.role IS NULL OR src.role NOT IN (1, 3)) THEN 'direct' @@ -170,7 +182,7 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { op.snr, op.rssi FROM owner_packets op - LEFT JOIN nodes src ON src.node_id = op.src_node_id + LEFT JOIN node_identity_nodes src ON src.node_id = op.src_node_id LEFT JOIN resolved_last_hop rl ON rl.rx_node_id = op.rx_node_id AND rl.receiver_side_hash = op.receiver_side_hash @@ -219,8 +231,8 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { role: number | null; }>( `SELECT node_id, name, network, iata, advert_count, last_seen, lat, lon, role - FROM nodes - WHERE node_id = $1 + FROM node_identity_nodes + WHERE node_id = meshcore_canonical_node_id($1) LIMIT 1`, [selectedNodeId], ), @@ -243,12 +255,12 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { n.lon, COUNT(*)::int AS packets_24h, MAX(p.time)::text AS last_seen - FROM packets p - LEFT JOIN nodes n ON n.node_id = p.src_node_id - WHERE p.rx_node_id = $1 + FROM node_identity_packets p + LEFT JOIN node_identity_nodes n ON n.node_id = p.src_node_id + WHERE p.rx_node_id = meshcore_canonical_node_id($1) AND p.hop_count = 0 AND p.src_node_id IS NOT NULL - AND p.src_node_id <> $1 + AND p.src_node_id <> meshcore_canonical_node_id($1) AND p.time > NOW() - INTERVAL '24 hours' GROUP BY p.src_node_id, n.name, n.network, n.iata, n.lat, n.lon ORDER BY packets_24h DESC @@ -290,8 +302,8 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { ) ORDER BY p.time DESC ) AS rn - FROM packets p - WHERE p.rx_node_id = $1 + FROM node_identity_packets p + WHERE p.rx_node_id = meshcore_canonical_node_id($1) ) SELECT r.time::text AS time, @@ -310,7 +322,7 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { r.payload->>'message' ) AS body FROM ranked r - LEFT JOIN nodes src ON src.node_id = r.src_node_id + LEFT JOIN node_identity_nodes src ON src.node_id = r.src_node_id WHERE r.rn = 1 ORDER BY r.time DESC LIMIT 9`, @@ -339,11 +351,11 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { COUNT(DISTINCT p.packet_hash)::int AS packets_7d, MAX(p.time)::text AS last_seen, MIN(p.hop_count) AS best_hops - FROM packets p - LEFT JOIN nodes n ON n.node_id = p.rx_node_id - WHERE p.src_node_id = $1 + FROM node_identity_packets p + LEFT JOIN node_identity_nodes n ON n.node_id = p.rx_node_id + WHERE p.src_node_id = meshcore_canonical_node_id($1) AND p.rx_node_id IS NOT NULL - AND p.rx_node_id <> $1 + AND p.rx_node_id <> meshcore_canonical_node_id($1) AND p.time > NOW() - INTERVAL '7 days' GROUP BY p.rx_node_id, n.name, n.network, n.iata, n.lat, n.lon ORDER BY packets_24h DESC, packets_7d DESC, last_seen DESC @@ -363,19 +375,26 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { last_observed: string | null; }>( `SELECT - CASE WHEN nl.node_a_id = $1 THEN nl.node_b_id ELSE nl.node_a_id END AS peer_node_id, + CASE WHEN nl.node_a_id = meshcore_canonical_node_id($1) + THEN nl.node_b_id ELSE nl.node_a_id END AS peer_node_id, peer.name AS peer_name, peer.network AS peer_network, - CASE WHEN nl.node_a_id = $1 THEN nl.count_a_to_b ELSE nl.count_b_to_a END AS owner_to_peer, - CASE WHEN nl.node_a_id = $1 THEN nl.count_b_to_a ELSE nl.count_a_to_b END AS peer_to_owner, + CASE WHEN nl.node_a_id = meshcore_canonical_node_id($1) + THEN nl.count_a_to_b ELSE nl.count_b_to_a END AS owner_to_peer, + CASE WHEN nl.node_a_id = meshcore_canonical_node_id($1) + THEN nl.count_b_to_a ELSE nl.count_a_to_b END AS peer_to_owner, nl.observed_count, nl.itm_path_loss_db, nl.itm_viable, nl.force_viable, nl.last_observed::text AS last_observed - FROM node_links nl - JOIN nodes peer ON peer.node_id = CASE WHEN nl.node_a_id = $1 THEN nl.node_b_id ELSE nl.node_a_id END - WHERE (nl.node_a_id = $1 OR nl.node_b_id = $1) + FROM node_identity_links nl + JOIN node_identity_nodes peer ON peer.node_id = CASE + WHEN nl.node_a_id = meshcore_canonical_node_id($1) THEN nl.node_b_id + ELSE nl.node_a_id + END + WHERE (nl.node_a_id = meshcore_canonical_node_id($1) + OR nl.node_b_id = meshcore_canonical_node_id($1)) AND ( nl.force_viable = true OR nl.itm_viable = true @@ -393,8 +412,8 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { `SELECT time_bucket('1 hour', time)::text AS bucket, COUNT(DISTINCT packet_hash)::int AS adverts - FROM packets - WHERE src_node_id = $1 + FROM node_identity_packets + WHERE src_node_id = meshcore_canonical_node_id($1) AND packet_type = 4 AND time > NOW() - INTERVAL '24 hours' GROUP BY bucket @@ -433,23 +452,23 @@ export function createOwnerRepository(deps: OwnerRepositoryDeps) { WHEN jsonb_typeof(stats->'tx_publish_calls') = 'number' THEN (stats->>'tx_publish_calls')::double precision ELSE NULL END AS tx_publish_calls - FROM node_status_samples - WHERE node_id = $1 + FROM node_identity_status_samples + WHERE node_id = meshcore_canonical_node_id($1) AND time > NOW() - INTERVAL '24 hours' ORDER BY time ASC`, [selectedNodeId], ), query<{ packets_24h: number }>( `SELECT COUNT(*)::int AS packets_24h - FROM packets - WHERE src_node_id = $1 + FROM node_identity_packets + WHERE src_node_id = meshcore_canonical_node_id($1) AND time > NOW() - INTERVAL '24 hours'`, [selectedNodeId], ), query<{ packets_24h: number }>( `SELECT COUNT(*)::int AS packets_24h - FROM packets - WHERE rx_node_id = $1 + FROM node_identity_packets + WHERE rx_node_id = meshcore_canonical_node_id($1) AND time > NOW() - INTERVAL '24 hours'`, [selectedNodeId], ), diff --git a/backend/src/repositories/networkAnalysis.ts b/backend/src/repositories/networkAnalysis.ts index 6cf87ff..12a078a 100644 --- a/backend/src/repositories/networkAnalysis.ts +++ b/backend/src/repositories/networkAnalysis.ts @@ -53,9 +53,9 @@ export async function topologyRows( nl.multibyte_observed_count, nl.last_observed::text, nl.itm_path_loss_db - FROM node_links nl - JOIN nodes a ON a.node_id = nl.node_a_id - JOIN nodes b ON b.node_id = nl.node_b_id + FROM node_identity_links nl + JOIN node_identity_nodes a ON a.node_id = nl.node_a_id + JOIN node_identity_nodes b ON b.node_id = nl.node_b_id WHERE (nl.itm_viable = true OR nl.force_viable = true) AND nl.last_observed > NOW() - INTERVAL '30 days' AND (a.role IS NULL OR a.role = 2) @@ -73,14 +73,14 @@ export async function topologyRows( export async function standaloneTopologyRows(query: QueryFn, filters: NetworkFilters) { return query( `SELECT n.node_id, n.name, n.lat, n.lon, n.iata - FROM nodes n + FROM node_identity_nodes n WHERE n.last_seen > NOW() - INTERVAL '30 days' AND (n.role IS NULL OR n.role = 2) AND (n.name IS NULL OR n.name NOT LIKE '%🚫%') ${filters.nodesAlias('n')} AND NOT EXISTS ( SELECT 1 - FROM node_links nl + FROM node_identity_links nl WHERE (nl.node_a_id = n.node_id OR nl.node_b_id = n.node_id) AND (nl.itm_viable = true OR nl.force_viable = true) AND nl.last_observed > NOW() - INTERVAL '30 days' @@ -121,9 +121,9 @@ export async function rfValidationRows( WHEN nl.itm_viable = true AND nl.observed_count <= 2 AND nl.last_observed < NOW() - INTERVAL '7 days' THEN 'weak_model_evidence' ELSE 'match' END AS classification - FROM node_links nl - JOIN nodes a ON a.node_id = nl.node_a_id - JOIN nodes b ON b.node_id = nl.node_b_id + FROM node_identity_links nl + JOIN node_identity_nodes a ON a.node_id = nl.node_a_id + JOIN node_identity_nodes b ON b.node_id = nl.node_b_id WHERE nl.last_observed > NOW() - INTERVAL '30 days' AND (a.name IS NULL OR a.name NOT LIKE '%🚫%') AND (b.name IS NULL OR b.name NOT LIKE '%🚫%') diff --git a/backend/src/repositories/nodes.ts b/backend/src/repositories/nodes.ts index f5b964d..c9e501d 100644 --- a/backend/src/repositories/nodes.ts +++ b/backend/src/repositories/nodes.ts @@ -98,8 +98,8 @@ export function createNodeRepository(query: QueryFn): NodeRepository { n.iata, n.hardware_model, n.firmware_version - FROM node_status_samples nss - LEFT JOIN nodes n ON n.node_id = nss.node_id + FROM node_identity_status_samples nss + LEFT JOIN node_identity_nodes n ON n.node_id = nss.node_id WHERE nss.network = 'test' ORDER BY nss.node_id, nss.time DESC ) latest @@ -118,7 +118,7 @@ export function createNodeRepository(query: QueryFn): NodeRepository { channel_utilization, air_util_tx, stats - FROM node_status_samples + FROM node_identity_status_samples WHERE network = 'test' ORDER BY time DESC`, [], @@ -164,7 +164,7 @@ export function createNodeRepository(query: QueryFn): NodeRepository { WHEN jsonb_typeof(stats->'tx_queue_depth_peak') = 'number' THEN (stats->>'tx_queue_depth_peak')::double precision ELSE NULL END AS tx_queue_depth_peak - FROM node_status_samples + FROM node_identity_status_samples WHERE node_id = $1 AND network = 'test' AND time > NOW() - INTERVAL '24 hours' @@ -198,7 +198,7 @@ export function createNodeRepository(query: QueryFn): NodeRepository { .join(', '); const result = await query>( `SELECT ${selectedFields} - FROM nodes n + FROM node_identity_nodes n WHERE ${publicMapFreshPredicate('n', `$${snapshotParameter}::timestamptz`)} ${filters.nodesAlias('n')} AND ($${cursorParameter}::text IS NULL OR n.node_id > $${cursorParameter}) @@ -210,7 +210,7 @@ export function createNodeRepository(query: QueryFn): NodeRepository { }, async listAllNodeIds() { - return (await query<{ node_id: string }>('SELECT node_id FROM nodes')).rows; + return (await query<{ node_id: string }>('SELECT node_id FROM node_identity_nodes')).rows; }, async listInferredPackets(scope, limit) { @@ -233,20 +233,24 @@ export function createNodeRepository(query: QueryFn): NodeRepository { return (await query( `WITH source_node AS MATERIALIZED ( SELECT node_id - FROM nodes - WHERE node_id = ${idParam} + FROM node_identity_nodes + WHERE node_id = meshcore_canonical_node_id(${idParam}) AND (name IS NULL OR name NOT LIKE '%🚫%') ${filters.nodes} ), relevant_links AS MATERIALIZED ( SELECT - CASE WHEN nl.node_a_id = ${idParam} THEN nl.node_b_id ELSE nl.node_a_id END AS peer_id, + CASE WHEN nl.node_a_id = meshcore_canonical_node_id(${idParam}) + THEN nl.node_b_id ELSE nl.node_a_id END AS peer_id, nl.observed_count, nl.itm_path_loss_db, - CASE WHEN nl.node_a_id = ${idParam} THEN nl.count_a_to_b ELSE nl.count_b_to_a END AS count_this_to_peer, - CASE WHEN nl.node_a_id = ${idParam} THEN nl.count_b_to_a ELSE nl.count_a_to_b END AS count_peer_to_this - FROM node_links nl - WHERE (nl.node_a_id = ${idParam} OR nl.node_b_id = ${idParam}) + CASE WHEN nl.node_a_id = meshcore_canonical_node_id(${idParam}) + THEN nl.count_a_to_b ELSE nl.count_b_to_a END AS count_this_to_peer, + CASE WHEN nl.node_a_id = meshcore_canonical_node_id(${idParam}) + THEN nl.count_b_to_a ELSE nl.count_a_to_b END AS count_peer_to_this + FROM node_identity_links nl + WHERE (nl.node_a_id = meshcore_canonical_node_id(${idParam}) + OR nl.node_b_id = meshcore_canonical_node_id(${idParam})) AND (nl.itm_viable = TRUE OR nl.force_viable = TRUE) AND EXISTS (SELECT 1 FROM source_node) ) @@ -254,7 +258,7 @@ export function createNodeRepository(query: QueryFn): NodeRepository { rl.peer_id, peer.name AS peer_name, rl.observed_count, rl.itm_path_loss_db, rl.count_this_to_peer, rl.count_peer_to_this FROM relevant_links rl - JOIN nodes peer ON peer.node_id = rl.peer_id + JOIN node_identity_nodes peer ON peer.node_id = rl.peer_id WHERE (peer.name IS NULL OR peer.name NOT LIKE '%🚫%') ${filters.nodesAlias('peer')} ORDER BY rl.observed_count DESC`, diff --git a/backend/src/repositories/productFeatures.ts b/backend/src/repositories/productFeatures.ts index 980d7e5..6f28762 100644 --- a/backend/src/repositories/productFeatures.ts +++ b/backend/src/repositories/productFeatures.ts @@ -49,10 +49,11 @@ export async function observerHealthRows(query: QueryFn, networks: string[]) { unique_src_48h: string; }>( `WITH observer_activity AS ( - SELECT p.rx_node_id, + SELECT meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id, COUNT(DISTINCT date_trunc('hour', p.time)) AS active_hours, COUNT(*) AS packets_48h, - COUNT(DISTINCT p.src_node_id) FILTER (WHERE p.src_node_id IS NOT NULL) AS unique_src_48h + COUNT(DISTINCT meshcore_canonical_node_id(p.src_node_id)) + FILTER (WHERE p.src_node_id IS NOT NULL) AS unique_src_48h FROM packets p WHERE p.time > NOW() - INTERVAL '48 hours' AND p.network = ANY($1::text[]) @@ -62,7 +63,7 @@ export async function observerHealthRows(query: QueryFn, networks: string[]) { SELECT n.node_id, n.name, n.lat, n.lon, oa.active_hours::text, oa.packets_48h::text, oa.unique_src_48h::text FROM observer_activity oa - JOIN nodes n ON n.node_id = oa.rx_node_id + JOIN node_identity_nodes n ON n.node_id = oa.rx_node_id WHERE n.lat IS NOT NULL AND n.lon IS NOT NULL AND (n.name IS NULL OR n.name NOT LIKE '%🚫%')`, [networks], @@ -76,7 +77,7 @@ export async function visibleLinkNodeIds( ) { return query<{ node_id: string }>( `SELECT n.node_id - FROM nodes n + FROM node_identity_nodes n WHERE n.node_id = ANY($1::text[]) AND (n.name IS NULL OR n.name NOT LIKE '%🚫%') AND ( @@ -85,7 +86,7 @@ export async function visibleLinkNodeIds( n.network IS DISTINCT FROM 'test' AND EXISTS ( SELECT 1 - FROM node_network_sightings sighting + FROM node_identity_sightings sighting WHERE sighting.node_id = n.node_id AND sighting.network = ANY($2::text[]) AND sighting.last_seen_at > NOW() - INTERVAL '30 days' @@ -112,10 +113,11 @@ export async function linkHistoryRows( `SELECT reports.last_seen::text AS time, reports.last_snr_db AS snr, NULL::double precision AS rssi, links.itm_path_loss_db AS path_loss, reports.sample_count - FROM node_link_radio_reports reports - LEFT JOIN node_links links + FROM node_identity_link_radio_reports reports + LEFT JOIN node_identity_links links ON links.node_a_id = reports.node_a_id AND links.node_b_id = reports.node_b_id - WHERE reports.node_a_id = $1 AND reports.node_b_id = $2 + WHERE reports.node_a_id = meshcore_canonical_node_id($1) + AND reports.node_b_id = meshcore_canonical_node_id($2) AND reports.last_seen > NOW() - ($3::text || ' hours')::interval ORDER BY reports.last_seen ASC`, [nodeA, nodeB, String(hours)], @@ -127,7 +129,7 @@ export async function repeaterFirmwareRows(query: QueryFn, networks: string[]) { `SELECT COALESCE(hardware_model, 'Unknown') AS hardware_model, COALESCE(NULLIF(firmware_version, ''), 'Unknown') AS firmware_version, COUNT(*)::text AS count - FROM nodes + FROM node_identity_nodes WHERE network = ANY($1::text[]) AND (role IS NULL OR role = 2) AND last_seen > NOW() - INTERVAL '30 days' diff --git a/backend/src/stats/statsRepository.ts b/backend/src/stats/statsRepository.ts index d56c8e3..f6348a3 100644 --- a/backend/src/stats/statsRepository.ts +++ b/backend/src/stats/statsRepository.ts @@ -165,8 +165,9 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { COALESCE(NULLIF(TRIM(UPPER(p.iata)), ''), 'UNK') AS iata, COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.time > NOW() - INTERVAL '24 hours') AS packets_24h, COUNT(DISTINCT p.packet_hash) AS packets_7d, - COUNT(DISTINCT p.rx_node_id) FILTER (WHERE p.time > NOW() - INTERVAL '1 minute') AS active_observers, - COUNT(DISTINCT p.rx_node_id) AS observers, + COUNT(DISTINCT meshcore_canonical_node_id(p.rx_node_id)) + FILTER (WHERE p.time > NOW() - INTERVAL '1 minute') AS active_observers, + COUNT(DISTINCT meshcore_canonical_node_id(p.rx_node_id)) AS observers, MAX(p.time)::text AS last_packet_at FROM packets p WHERE p.time > NOW() - INTERVAL '7 days' @@ -732,7 +733,8 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { ) AS bucket ), counts AS ( - SELECT time_bucket('1 hour', p.time) AS bucket, COUNT(DISTINCT p.src_node_id)::int AS count + SELECT time_bucket('1 hour', p.time) AS bucket, + COUNT(DISTINCT meshcore_canonical_node_id(p.src_node_id))::int AS count FROM packets p WHERE p.time > NOW() - INTERVAL '24 hours' AND p.src_node_id IS NOT NULL @@ -745,7 +747,8 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { ORDER BY b.bucket `, filters.params), query(` - SELECT time_bucket('1 day', time) AS day, COUNT(DISTINCT src_node_id) AS count + SELECT time_bucket('1 day', time) AS day, + COUNT(DISTINCT meshcore_canonical_node_id(src_node_id)) AS count FROM packets WHERE time > NOW() - INTERVAL '7 days' AND src_node_id IS NOT NULL ${filters.packets} GROUP BY day ORDER BY day @@ -788,7 +791,10 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { SELECT (SELECT COUNT(*) FROM packets WHERE time > NOW() - INTERVAL '24 hours' ${filters.packets}) AS total_24h, (SELECT COUNT(*) FROM packets WHERE time > NOW() - INTERVAL '7 days' ${filters.packets}) AS total_7d, - (SELECT COUNT(DISTINCT src_node_id) FROM packets WHERE time > NOW() - INTERVAL '24 hours' AND src_node_id IS NOT NULL ${filters.packets}) AS unique_radios_24h + (SELECT COUNT(DISTINCT meshcore_canonical_node_id(src_node_id)) + FROM packets + WHERE time > NOW() - INTERVAL '24 hours' + AND src_node_id IS NOT NULL ${filters.packets}) AS unique_radios_24h `, filters.params), fetchObserverRegionSummary(network, observer), query(` @@ -846,7 +852,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { p.path_hashes, rx.role AS rx_role FROM packets p - LEFT JOIN nodes rx ON rx.node_id = p.rx_node_id + LEFT JOIN node_identity_nodes rx ON rx.node_id = meshcore_canonical_node_id(p.rx_node_id) WHERE p.time > NOW() - INTERVAL '24 hours' AND p.path_hash_size_bytes > 1 AND COALESCE(array_length(p.path_hashes, 1), 0) > 0 @@ -896,7 +902,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { UPPER(LEFT(n.node_id, 4)) AS hash, COUNT(*)::int AS match_count, MIN(n.node_id) AS node_id - FROM nodes n + FROM node_identity_nodes n JOIN distinct_hashes dh ON LENGTH(dh.hash) = 4 AND dh.hash = UPPER(LEFT(n.node_id, 4)) @@ -909,7 +915,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { UPPER(LEFT(n.node_id, 6)) AS hash, COUNT(*)::int AS match_count, MIN(n.node_id) AS node_id - FROM nodes n + FROM node_identity_nodes n JOIN distinct_hashes dh ON LENGTH(dh.hash) = 6 AND dh.hash = UPPER(LEFT(n.node_id, 6)) @@ -985,7 +991,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { ) ORDER BY dh.ord) FROM latest_fully_decoded l JOIN decoded_hops dh ON dh.obs_id = l.obs_id - LEFT JOIN nodes n ON n.node_id = dh.node_id + LEFT JOIN node_identity_nodes n ON n.node_id = dh.node_id ) AS latest_fully_decoded_nodes, (SELECT time::text FROM longest_fully_decoded) AS longest_fully_decoded_at, (SELECT packet_hash FROM longest_fully_decoded) AS longest_fully_decoded_hash, @@ -1002,7 +1008,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { ) ORDER BY dh.ord) FROM longest_fully_decoded l JOIN decoded_hops dh ON dh.obs_id = l.obs_id - LEFT JOIN nodes n ON n.node_id = dh.node_id + LEFT JOIN node_identity_nodes n ON n.node_id = dh.node_id ) AS longest_fully_decoded_nodes`, filters.params, ), @@ -1013,7 +1019,8 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { single_observer_packets: string; }>( `WITH per_packet AS ( - SELECT packet_hash, COUNT(DISTINCT rx_node_id)::int AS observer_count + SELECT packet_hash, + COUNT(DISTINCT meshcore_canonical_node_id(rx_node_id))::int AS observer_count FROM ( SELECT p.packet_hash, p.rx_node_id FROM packets p @@ -1100,7 +1107,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { p.path_hashes, rx.role AS rx_role FROM packets p - LEFT JOIN nodes rx ON rx.node_id = p.rx_node_id + LEFT JOIN node_identity_nodes rx ON rx.node_id = meshcore_canonical_node_id(p.rx_node_id) WHERE p.time > NOW() - INTERVAL '7 days' AND p.path_hash_size_bytes > 1 AND COALESCE(array_length(p.path_hashes, 1), 0) > 0 @@ -1155,7 +1162,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { UPPER(LEFT(n.node_id, 4)) AS hash, COUNT(*)::int AS match_count, MIN(n.node_id) AS node_id - FROM nodes n + FROM node_identity_nodes n JOIN distinct_hashes dh ON LENGTH(dh.hash) = 4 AND dh.hash = UPPER(LEFT(n.node_id, 4)) @@ -1168,7 +1175,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { UPPER(LEFT(n.node_id, 6)) AS hash, COUNT(*)::int AS match_count, MIN(n.node_id) AS node_id - FROM nodes n + FROM node_identity_nodes n JOIN distinct_hashes dh ON LENGTH(dh.hash) = 6 AND dh.hash = UPPER(LEFT(n.node_id, 6)) @@ -1275,18 +1282,21 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { const [mqttCount, packetCount, staleCount, mapNodeCount, totalNodeCount, longestHopCount, nodesDayCount, internationalCount] = await Promise.all([ network != null - ? query(`SELECT COUNT(DISTINCT rx_node_id) AS count + ? query(`SELECT COUNT(DISTINCT meshcore_canonical_node_id(rx_node_id)) AS count FROM packets WHERE time > NOW() - INTERVAL '10 minutes' AND rx_node_id IS NOT NULL ${filters.packets}`, filters.params) : query(` WITH test_active AS ( - SELECT rx_node_id FROM packets WHERE rx_node_id IS NOT NULL AND rx_node_id <> '' + SELECT meshcore_canonical_node_id(rx_node_id) AS rx_node_id + FROM packets + WHERE rx_node_id IS NOT NULL AND rx_node_id <> '' AND time > NOW() - INTERVAL '7 days' - GROUP BY rx_node_id HAVING MAX(time) = MAX(time) FILTER (WHERE network = 'test') + GROUP BY meshcore_canonical_node_id(rx_node_id) + HAVING MAX(time) = MAX(time) FILTER (WHERE network = 'test') ) - SELECT COUNT(DISTINCT rx_node_id) AS count + SELECT COUNT(DISTINCT meshcore_canonical_node_id(rx_node_id)) AS count FROM packets WHERE time > NOW() - INTERVAL '10 minutes' AND rx_node_id IS NOT NULL @@ -1294,29 +1304,29 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { ${filters.packets} `, filters.params), query(`SELECT COUNT(*) AS count FROM packets WHERE time > NOW() - INTERVAL '24 hours' ${filters.packets}`, filters.params), - query(`SELECT COUNT(*) AS count FROM nodes + query(`SELECT COUNT(*) AS count FROM node_identity_nodes nodes WHERE ${publicMapBasePredicate('nodes')} AND ${nodeEffectiveLastSeenSql('nodes')} <= NOW() - INTERVAL '14 days' AND ${nodeEffectiveLastSeenSql('nodes')} > NOW() - INTERVAL '28 days' ${filters.nodes}`, filters.params), - query(`SELECT COUNT(*) AS count FROM nodes + query(`SELECT COUNT(*) AS count FROM node_identity_nodes nodes WHERE ${publicMapFreshPredicate('nodes')} ${filters.nodes}`, filters.params), - query(`SELECT COUNT(*) AS count FROM nodes + query(`SELECT COUNT(*) AS count FROM node_identity_nodes nodes WHERE (name IS NULL OR name NOT LIKE '%🚫%') AND (role IS NULL OR role != 4) ${filters.nodes}`, filters.params), longestHopResult(), - query(`SELECT COUNT(DISTINCT src_node_id) AS count + query(`SELECT COUNT(DISTINCT meshcore_canonical_node_id(src_node_id)) AS count FROM packets WHERE time > NOW() - INTERVAL '24 hours' AND src_node_id IS NOT NULL ${filters.packets}`, filters.params), query(`WITH intl AS ( SELECT lat, lon, last_seen, advert_count - FROM nodes + FROM node_identity_nodes nodes WHERE lat IS NOT NULL AND lon IS NOT NULL AND lat != 0 AND lon != 0 AND last_seen > NOW() - INTERVAL '7 days' @@ -1363,24 +1373,24 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { const filters = networkFilters(network); return query<{ node_id: string; name: string | null; rx_24h: string; tx_24h: string; last_tx: string | null; last_rx: string | null }>( `WITH rx AS ( - SELECT p.rx_node_id AS node_id, + SELECT meshcore_canonical_node_id(p.rx_node_id) AS node_id, COUNT(p.packet_hash)::text AS rx_24h, MAX(p.time)::text AS last_rx FROM packets p WHERE p.time > NOW() - INTERVAL '24 hours' AND p.rx_node_id IS NOT NULL ${filters.packetsAlias('p')} - GROUP BY p.rx_node_id + GROUP BY meshcore_canonical_node_id(p.rx_node_id) ), tx AS ( - SELECT p.src_node_id AS node_id, + SELECT meshcore_canonical_node_id(p.src_node_id) AS node_id, COUNT(p.packet_hash)::text AS tx_24h, MAX(p.time)::text AS last_tx FROM packets p WHERE p.time > NOW() - INTERVAL '24 hours' AND p.src_node_id IS NOT NULL ${filters.packetsAlias('p')} - GROUP BY p.src_node_id + GROUP BY meshcore_canonical_node_id(p.src_node_id) ) SELECT n.node_id, @@ -1390,7 +1400,7 @@ export function createStatsRepository(deps: StatsRepositoryDeps) { tx.last_tx, rx.last_rx FROM rx - JOIN nodes n ON n.node_id = rx.node_id + JOIN node_identity_nodes n ON n.node_id = rx.node_id LEFT JOIN tx ON tx.node_id = rx.node_id WHERE n.name IS NULL OR n.name NOT LIKE '%🚫%' ORDER BY rx.rx_24h::bigint DESC`,