From 2713d501b4ada582be2ef8971e3eb6d1c06b570d Mon Sep 17 00:00:00 2001 From: you Date: Fri, 20 Mar 2026 07:29:01 +0000 Subject: [PATCH] feat: MQTT topic arrays, IATA filtering, observer status parsing - mqttSources[].topics is now an array of topic patterns - mqttSources[].iataFilter optionally restricts to specific regions - meshcore///status topic parsed for observer metadata: name, model, firmware, client_version, radio, battery, uptime, noise_floor - New observer columns with auto-migration for existing DBs - Status updates don't inflate packet_count (separate updateObserverStatus) --- config.example.json | 3 +- db.js | 74 ++++++++++++++++++++++++++++++++++++++++++--- server.js | 33 ++++++++++++++++++++ 3 files changed, 104 insertions(+), 6 deletions(-) diff --git a/config.example.json b/config.example.json index ce38a013..8c7549ac 100644 --- a/config.example.json +++ b/config.example.json @@ -16,7 +16,8 @@ "username": "your-username", "password": "your-password", "rejectUnauthorized": false, - "topics": ["meshcore/#"] + "topics": ["meshcore/+/+/packets", "meshcore/+/+/status"], + "iataFilter": ["SJC", "SFO", "OAK"] } ], "channelKeys": { diff --git a/db.js b/db.js index bc128089..8705285e 100644 --- a/db.js +++ b/db.js @@ -49,7 +49,14 @@ db.exec(` iata TEXT, last_seen TEXT, first_seen TEXT, - packet_count INTEGER DEFAULT 0 + packet_count INTEGER DEFAULT 0, + model TEXT, + firmware TEXT, + client_version TEXT, + radio TEXT, + battery_mv INTEGER, + uptime_secs INTEGER, + noise_floor INTEGER ); CREATE TABLE IF NOT EXISTS paths ( @@ -66,6 +73,16 @@ db.exec(` CREATE INDEX IF NOT EXISTS idx_observers_last_seen ON observers(last_seen); `); +// --- Migrations for existing DBs --- +const observerCols = db.pragma('table_info(observers)').map(c => c.name); +for (const col of ['model', 'firmware', 'client_version', 'radio', 'battery_mv', 'uptime_secs', 'noise_floor']) { + if (!observerCols.includes(col)) { + const type = ['battery_mv', 'uptime_secs', 'noise_floor'].includes(col) ? 'INTEGER' : 'TEXT'; + db.exec(`ALTER TABLE observers ADD COLUMN ${col} ${type}`); + console.log(`[migration] Added observers.${col}`); + } +} + // --- Prepared statements --- const stmts = { insertPacket: db.prepare(` @@ -85,13 +102,35 @@ const stmts = { advert_count = advert_count + 1 `), upsertObserver: db.prepare(` - INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) - VALUES (@id, @name, @iata, @last_seen, @first_seen, 1) + INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count, model, firmware, client_version, radio, battery_mv, uptime_secs, noise_floor) + VALUES (@id, @name, @iata, @last_seen, @first_seen, 1, @model, @firmware, @client_version, @radio, @battery_mv, @uptime_secs, @noise_floor) ON CONFLICT(id) DO UPDATE SET name = COALESCE(@name, name), iata = COALESCE(@iata, iata), last_seen = @last_seen, - packet_count = packet_count + 1 + packet_count = packet_count + 1, + model = COALESCE(@model, model), + firmware = COALESCE(@firmware, firmware), + client_version = COALESCE(@client_version, client_version), + radio = COALESCE(@radio, radio), + battery_mv = COALESCE(@battery_mv, battery_mv), + uptime_secs = COALESCE(@uptime_secs, uptime_secs), + noise_floor = COALESCE(@noise_floor, noise_floor) + `), + updateObserverStatus: db.prepare(` + INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count, model, firmware, client_version, radio, battery_mv, uptime_secs, noise_floor) + VALUES (@id, @name, @iata, @last_seen, @first_seen, 0, @model, @firmware, @client_version, @radio, @battery_mv, @uptime_secs, @noise_floor) + ON CONFLICT(id) DO UPDATE SET + name = COALESCE(@name, name), + iata = COALESCE(@iata, iata), + last_seen = @last_seen, + model = COALESCE(@model, model), + firmware = COALESCE(@firmware, firmware), + client_version = COALESCE(@client_version, client_version), + radio = COALESCE(@radio, radio), + battery_mv = COALESCE(@battery_mv, battery_mv), + uptime_secs = COALESCE(@uptime_secs, uptime_secs), + noise_floor = COALESCE(@noise_floor, noise_floor) `), getPacket: db.prepare(`SELECT * FROM packets WHERE id = ?`), getPathsForPacket: db.prepare(`SELECT * FROM paths WHERE packet_id = ? ORDER BY hop_index`), @@ -159,6 +198,31 @@ function upsertObserver(data) { iata: data.iata || null, last_seen: data.last_seen || now, first_seen: data.first_seen || now, + model: data.model || null, + firmware: data.firmware || null, + client_version: data.client_version || null, + radio: data.radio || null, + battery_mv: data.battery_mv || null, + uptime_secs: data.uptime_secs || null, + noise_floor: data.noise_floor || null, + }); +} + +function updateObserverStatus(data) { + const now = new Date().toISOString(); + stmts.updateObserverStatus.run({ + id: data.id, + name: data.name || null, + iata: data.iata || null, + last_seen: data.last_seen || now, + first_seen: data.first_seen || now, + model: data.model || null, + firmware: data.firmware || null, + client_version: data.client_version || null, + radio: data.radio || null, + battery_mv: data.battery_mv || null, + uptime_secs: data.uptime_secs || null, + noise_floor: data.noise_floor || null, }); } @@ -493,4 +557,4 @@ function getNodeAnalytics(pubkey, days) { }; } -module.exports = { db, insertPacket, insertPath, upsertNode, upsertObserver, getPackets, getPacket, getNodes, getNode, getObservers, getStats, seed, searchNodes, getNodeHealth, getNodeAnalytics }; +module.exports = { db, insertPacket, insertPath, upsertNode, upsertObserver, updateObserverStatus, getPackets, getPacket, getNodes, getNode, getObservers, getStats, seed, searchNodes, getNodeHealth, getNodeAnalytics }; diff --git a/server.js b/server.js index 5a65763e..4388457b 100644 --- a/server.js +++ b/server.js @@ -403,6 +403,39 @@ for (const source of mqttSources) { const parts = topic.split('/'); const now = new Date().toISOString(); + // IATA filter: if source has iataFilter, only accept matching regions + const region = parts[1] || null; + if (source.iataFilter && Array.isArray(source.iataFilter) && region) { + if (!source.iataFilter.includes(region)) return; + } + + // --- Status topic: meshcore///status --- + if (parts[3] === 'status' && parts[2]) { + const observerId = parts[2]; + const name = msg.origin || null; + const iata = region; + // Parse radio string: "freq,bw,sf,cr" + let radioInfo = null; + if (msg.radio) { + const rp = msg.radio.split(','); + radioInfo = { freq: parseFloat(rp[0]), bw: parseFloat(rp[1]), sf: parseInt(rp[2]), cr: parseInt(rp[3]) }; + } + db.updateObserverStatus({ + id: observerId, + name: name, + iata: iata, + model: msg.model || null, + firmware: msg.firmware_version || null, + client_version: msg.client_version || null, + radio: msg.radio || null, + battery_mv: msg.stats?.battery_mv || null, + uptime_secs: msg.stats?.uptime_secs || null, + noise_floor: msg.stats?.noise_floor || null, + }); + console.log(`MQTT [${tag}] status: ${name || observerId} (${iata}) - ${msg.status}`); + return; + } + // --- Format 1: Raw packet logging (meshcoretomqtt / Cisien format) --- // Topic: meshcore///packets, payload: { raw, SNR, RSSI, hash } if (msg.raw && typeof msg.raw === 'string') {