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/<region>/<id>/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)
This commit is contained in:
you committed 2026-03-20 07:29:01 +00:00
1 parent 4ff72935ca
commit 2713d501b4
3 files changed
+104 -6

No files matched your search

+2 -1
View File
@@ -16,7 +16,8 @@
"username": "your-username",
"password": "your-password",
"rejectUnauthorized": false,
"topics": ["meshcore/#"]
"topics": ["meshcore/+/+/packets", "meshcore/+/+/status"],
"iataFilter": ["SJC", "SFO", "OAK"]
}
],
"channelKeys": {
+69 -5
View File
@@ -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 };
+33
View File
@@ -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/<region>/<observer_id>/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/<region>/<observer>/packets, payload: { raw, SNR, RSSI, hash }
if (msg.raw && typeof msg.raw === 'string') {