From 4d0d6f8565a1b01540e7153db18b6a2026ce0364 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sun, 29 Mar 2026 11:48:32 +0200 Subject: [PATCH] web: implement a 'protocol' field across systems (#655) * web: implement a 'protocol' field across systems * web: address review feedback on multi-protocol support - Rebase on main (pick up coordinate-clearing bugfix from #654) - P1: prevent cross-protocol message merges on shared packet IDs - P2: exclude "ingestor" key when enforcing /api/nodes batch limit - Extract append_protocol_filter helper + PROTOCOL_CLAUSE constant to reduce cognitive complexity and deduplicate SQL fragment in queries.rb - Extract coerce_bool helper to reduce upsert_node cognitive complexity - Merge nested if in insert_message protocol update path (Sonar) - Add explicit UPDATE backfill in ensure_schema_upgrades so any pre-existing NULL/empty protocol rows are set to meshtastic on upgrade - Rename migration file to 20260328_ (correct year) - Expand protocol_spec.rb: filter tests for all 7 endpoints, cross-protocol non-merge test, batch limit test, Sonar constant fixes, ENV.fetch, P1 regression test * web: address review comments --- data/ingestors.sql | 3 +- data/mesh_ingestor/CONTRACTS.md | 13 +- data/messages.sql | 3 +- .../20260328_add_protocol_column.sql | 29 + data/neighbors.sql | 1 + data/nodes.sql | 3 +- data/positions.sql | 3 +- data/telemetry.sql | 3 +- data/traces.sql | 3 +- .../application/data_processing.rb | 170 ++++-- web/lib/potato_mesh/application/database.rb | 35 ++ web/lib/potato_mesh/application/queries.rb | 59 +- web/lib/potato_mesh/application/routes/api.rb | 23 +- .../potato_mesh/application/routes/ingest.rb | 23 +- web/spec/protocol_spec.rb | 518 ++++++++++++++++++ 15 files changed, 797 insertions(+), 92 deletions(-) create mode 100644 data/migrations/20260328_add_protocol_column.sql create mode 100644 web/spec/protocol_spec.rb diff --git a/data/ingestors.sql b/data/ingestors.sql index 810846e..7e18654 100644 --- a/data/ingestors.sql +++ b/data/ingestors.sql @@ -20,7 +20,8 @@ CREATE TABLE IF NOT EXISTS ingestors ( last_seen_time INTEGER NOT NULL, version TEXT, lora_freq INTEGER, - modem_preset TEXT + modem_preset TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' ); CREATE INDEX IF NOT EXISTS idx_ingestors_last_seen ON ingestors(last_seen_time); diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 8153025..c152bc4 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -24,9 +24,11 @@ Future providers should emit payloads that match these shapes (keys + types), wh #### `POST /api/nodes` -Payload is a mapping keyed by canonical node id: +Payload is a mapping keyed by canonical node id, with an optional top-level `”ingestor”` key: -- `{ "!abcdef01": { ... node fields ... } }` +- `{ “!abcdef01”: { ... node fields ... }, “ingestor”: “!ingestornodeid” }` + +When `”ingestor”` is present the protocol is inherited from the registered ingestor (see `POST /api/ingestors`); omitting it defaults to `”meshtastic”`. Node entry fields are “Meshtastic-ish” (camelCase) and may include: @@ -104,4 +106,11 @@ Heartbeat payload: - `start_time` (int), `last_seen_time` (int) - `version` (string) - Optional: `lora_freq`, `modem_preset` +- Optional: `protocol` (string; e.g. `"meshtastic"`, `"meshcore"`) — declares the mesh backend for this ingestor; defaults to `"meshtastic"` when absent + +**Protocol propagation**: all event records (`messages`, `positions`, `telemetry`, `traces`, `neighbors`) that reference this ingestor via their `ingestor` field will inherit its `protocol` value at write time. + +### GET endpoint filtering + +All collection GET endpoints (`/api/nodes`, `/api/messages`, `/api/positions`, `/api/telemetry`, `/api/traces`, `/api/neighbors`, `/api/ingestors`) accept an optional `?protocol=` query parameter. When present, only records whose `protocol` column matches the given value are returned. The `protocol` field is included in all GET responses. diff --git a/data/messages.sql b/data/messages.sql index 6803f62..2d45c47 100644 --- a/data/messages.sql +++ b/data/messages.sql @@ -30,7 +30,8 @@ CREATE TABLE IF NOT EXISTS messages ( channel_name TEXT, reply_id INTEGER, emoji TEXT, - ingestor TEXT + ingestor TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' ); CREATE INDEX IF NOT EXISTS idx_messages_rx_time ON messages(rx_time); diff --git a/data/migrations/20260328_add_protocol_column.sql b/data/migrations/20260328_add_protocol_column.sql new file mode 100644 index 0000000..f1e4211 --- /dev/null +++ b/data/migrations/20260328_add_protocol_column.sql @@ -0,0 +1,29 @@ +-- Copyright © 2025-26 l5yth & contributors +-- +-- Licensed under the Apache License, Version 2.0 (the "License"); +-- you may not use this file except in compliance with the License. +-- You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +-- Add a protocol column to every entity and event table so records from +-- different mesh backends (meshtastic, meshcore, reticulum, …) can co-exist +-- in the same database and be queried independently. +-- +-- Existing rows default to 'meshtastic' for backward compatibility. + +BEGIN; +ALTER TABLE ingestors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +ALTER TABLE nodes ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +ALTER TABLE messages ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +ALTER TABLE positions ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +ALTER TABLE telemetry ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +ALTER TABLE traces ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +ALTER TABLE neighbors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; +COMMIT; diff --git a/data/neighbors.sql b/data/neighbors.sql index debc75c..20e1675 100644 --- a/data/neighbors.sql +++ b/data/neighbors.sql @@ -18,6 +18,7 @@ CREATE TABLE IF NOT EXISTS neighbors ( snr REAL, rx_time INTEGER NOT NULL, ingestor TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic', PRIMARY KEY (node_id, neighbor_id), FOREIGN KEY (node_id) REFERENCES nodes(node_id) ON DELETE CASCADE, FOREIGN KEY (neighbor_id) REFERENCES nodes(node_id) ON DELETE CASCADE diff --git a/data/nodes.sql b/data/nodes.sql index 629363e..abf4458 100644 --- a/data/nodes.sql +++ b/data/nodes.sql @@ -41,7 +41,8 @@ CREATE TABLE IF NOT EXISTS nodes ( longitude REAL, altitude REAL, lora_freq INTEGER, - modem_preset TEXT + modem_preset TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' ); CREATE INDEX IF NOT EXISTS idx_nodes_last_heard ON nodes(last_heard); diff --git a/data/positions.sql b/data/positions.sql index 6542ad7..ce060c7 100644 --- a/data/positions.sql +++ b/data/positions.sql @@ -34,7 +34,8 @@ CREATE TABLE IF NOT EXISTS positions ( hop_limit INTEGER, bitfield INTEGER, payload_b64 TEXT, - ingestor TEXT + ingestor TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' ); CREATE INDEX IF NOT EXISTS idx_positions_rx_time ON positions(rx_time); diff --git a/data/telemetry.sql b/data/telemetry.sql index 933e263..09275a4 100644 --- a/data/telemetry.sql +++ b/data/telemetry.sql @@ -54,7 +54,8 @@ CREATE TABLE IF NOT EXISTS telemetry ( rainfall_24h REAL, soil_moisture INTEGER, soil_temperature REAL, - ingestor TEXT + ingestor TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' ); CREATE INDEX IF NOT EXISTS idx_telemetry_rx_time ON telemetry(rx_time); diff --git a/data/traces.sql b/data/traces.sql index c606f32..0040195 100644 --- a/data/traces.sql +++ b/data/traces.sql @@ -22,7 +22,8 @@ CREATE TABLE IF NOT EXISTS traces ( rssi INTEGER, snr REAL, elapsed_ms INTEGER, - ingestor TEXT + ingestor TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' ); CREATE TABLE IF NOT EXISTS trace_hops ( diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index 630971d..1187934 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -17,6 +17,19 @@ module PotatoMesh module App module DataProcessing + # Coerce a Ruby boolean into a SQLite integer (1/0) while passing through + # any other value unchanged. Used when writing boolean node fields. + # + # @param value [Boolean, Object] value to coerce. + # @return [Integer, Object] 1, 0, or the original value. + def coerce_bool(value) + case value + when true then 1 + when false then 0 + else value + end + end + def resolve_node_num(node_id, payload) raw = payload["num"] @@ -118,7 +131,7 @@ module PotatoMesh normalized == "ffffffff" end - def ensure_unknown_node(db, node_ref, fallback_num = nil, heard_time: nil) + def ensure_unknown_node(db, node_ref, fallback_num = nil, heard_time: nil, protocol: "meshtastic") parts = canonical_node_parts(node_ref, fallback_num) return unless parts @@ -131,7 +144,8 @@ module PotatoMesh ) return if existing - long_name = "Meshtastic #{short_id}" + protocol_label = protocol.split(/[-_]/).map(&:capitalize).join + long_name = "#{protocol_label} #{short_id}" heard_time = coerce_integer(heard_time) inserted = false @@ -254,11 +268,12 @@ module PotatoMesh return false unless version lora_freq = coerce_integer(payload["lora_freq"]) modem_preset = string_or_nil(payload["modem_preset"]) + protocol = string_or_nil(payload["protocol"]) || "meshtastic" with_busy_retry do - db.execute <<~SQL, [node_id, start_time, last_seen_time, version, lora_freq, modem_preset] - INSERT INTO ingestors(node_id, start_time, last_seen_time, version, lora_freq, modem_preset) - VALUES(?,?,?,?,?,?) + db.execute <<~SQL, [node_id, start_time, last_seen_time, version, lora_freq, modem_preset, protocol] + INSERT INTO ingestors(node_id, start_time, last_seen_time, version, lora_freq, modem_preset, protocol) + VALUES(?,?,?,?,?,?,?) ON CONFLICT(node_id) DO UPDATE SET start_time = CASE WHEN excluded.start_time > ingestors.start_time THEN excluded.start_time @@ -270,7 +285,8 @@ module PotatoMesh END, version = COALESCE(excluded.version, ingestors.version), lora_freq = COALESCE(excluded.lora_freq, ingestors.lora_freq), - modem_preset = COALESCE(excluded.modem_preset, ingestors.modem_preset) + modem_preset = COALESCE(excluded.modem_preset, ingestors.modem_preset), + protocol = excluded.protocol SQL end @@ -286,7 +302,7 @@ module PotatoMesh false end - def upsert_node(db, node_id, n) + def upsert_node(db, node_id, n, protocol: "meshtastic") user = n["user"] || {} met = n["deviceMetrics"] || {} pos = n["position"] || {} @@ -298,13 +314,6 @@ module PotatoMesh lh = now if lh && lh > now lh = pt if pt && (!lh || lh < pt) lh ||= now - bool = ->(v) { - case v - when true then 1 - when false then 0 - else v - end - } node_num = resolve_node_num(node_id, n) update_prometheus_metrics(node_id, user, role, met, pos) @@ -321,8 +330,8 @@ module PotatoMesh user["hwModel"] || n["hwModel"], role, user["publicKey"], - bool.call(user["isUnmessagable"]), - bool.call(n["isFavorite"]), + coerce_bool(user["isUnmessagable"]), + coerce_bool(n["isFavorite"]), n["hopsAway"], n["snr"], lh, @@ -344,13 +353,14 @@ module PotatoMesh pos["altitude"], lora_freq, modem_preset, + protocol, ] with_busy_retry do db.execute <<~SQL, row INSERT INTO nodes(node_id,num,short_name,long_name,macaddr,hw_model,role,public_key,is_unmessagable,is_favorite, hops_away,snr,last_heard,first_heard,battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds, - position_time,location_source,precision_bits,latitude,longitude,altitude,lora_freq,modem_preset) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + position_time,location_source,precision_bits,latitude,longitude,altitude,lora_freq,modem_preset,protocol) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(node_id) DO UPDATE SET num=excluded.num, short_name=excluded.short_name, long_name=excluded.long_name, macaddr=excluded.macaddr, hw_model=excluded.hw_model, role=excluded.role, public_key=excluded.public_key, is_unmessagable=excluded.is_unmessagable, @@ -364,7 +374,8 @@ module PotatoMesh latitude=COALESCE(excluded.latitude, nodes.latitude), longitude=COALESCE(excluded.longitude, nodes.longitude), altitude=COALESCE(excluded.altitude, nodes.altitude), - lora_freq=excluded.lora_freq, modem_preset=excluded.modem_preset + lora_freq=excluded.lora_freq, modem_preset=excluded.modem_preset, + protocol=COALESCE(NULLIF(nodes.protocol,'meshtastic'), excluded.protocol) WHERE COALESCE(excluded.last_heard,0) >= COALESCE(nodes.last_heard,0) SQL end @@ -498,7 +509,7 @@ module PotatoMesh end end - def insert_position(db, payload) + def insert_position(db, payload, protocol_cache: nil) pos_id = coerce_integer(payload["id"] || payload["packet_id"]) return unless pos_id @@ -529,8 +540,10 @@ module PotatoMesh lora_freq = coerce_integer(payload["lora_freq"] || payload["loraFrequency"]) modem_preset = string_or_nil(payload["modem_preset"] || payload["modemPreset"]) + ingestor = string_or_nil(payload["ingestor"]) + protocol = resolve_protocol(db, ingestor, cache: protocol_cache) - ensure_unknown_node(db, node_id || node_num, node_num, heard_time: rx_time) + ensure_unknown_node(db, node_id || node_num, node_num, heard_time: rx_time, protocol: protocol) touch_node_last_seen( db, node_id || node_num, @@ -621,7 +634,6 @@ module PotatoMesh payload_b64 = string_or_nil(payload["payload_b64"] || payload["payload"]) payload_b64 ||= string_or_nil(position_section.dig("payload", "__bytes_b64__")) - ingestor = string_or_nil(payload["ingestor"]) row = [ pos_id, @@ -646,13 +658,14 @@ module PotatoMesh bitfield, payload_b64, ingestor, + protocol, ] with_busy_retry do db.execute <<~SQL, row INSERT INTO positions(id,node_id,node_num,rx_time,rx_iso,position_time,to_id,latitude,longitude,altitude,location_source, - precision_bits,sats_in_view,pdop,ground_speed,ground_track,snr,rssi,hop_limit,bitfield,payload_b64,ingestor) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + precision_bits,sats_in_view,pdop,ground_speed,ground_track,snr,rssi,hop_limit,bitfield,payload_b64,ingestor,protocol) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET node_id=COALESCE(excluded.node_id,positions.node_id), node_num=COALESCE(excluded.node_num,positions.node_num), @@ -674,7 +687,8 @@ module PotatoMesh hop_limit=COALESCE(excluded.hop_limit,positions.hop_limit), bitfield=COALESCE(excluded.bitfield,positions.bitfield), payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64), - ingestor=COALESCE(NULLIF(positions.ingestor,''), excluded.ingestor) + ingestor=COALESCE(NULLIF(positions.ingestor,''), excluded.ingestor), + protocol=COALESCE(NULLIF(positions.protocol,'meshtastic'), excluded.protocol) SQL end @@ -693,7 +707,7 @@ module PotatoMesh ) end - def insert_neighbors(db, payload) + def insert_neighbors(db, payload, protocol_cache: nil) return unless payload.is_a?(Hash) now = Time.now.to_i @@ -725,11 +739,13 @@ module PotatoMesh node_id = "!#{node_id.delete_prefix("!").downcase}" if node_id.start_with?("!") - ensure_unknown_node(db, node_id || node_num, node_num, heard_time: rx_time) + ingestor = string_or_nil(payload["ingestor"]) + protocol = resolve_protocol(db, ingestor, cache: protocol_cache) + + ensure_unknown_node(db, node_id || node_num, node_num, heard_time: rx_time, protocol: protocol) touch_node_last_seen(db, node_id || node_num, node_num, rx_time: rx_time, source: :neighborinfo) neighbor_entries = [] - ingestor = string_or_nil(payload["ingestor"]) neighbors_payload = payload["neighbors"] neighbors_list = neighbors_payload.is_a?(Array) ? neighbors_payload : [] @@ -765,9 +781,9 @@ module PotatoMesh entry_rx_time = now if entry_rx_time && entry_rx_time > now snr = coerce_float(neighbor["snr"]) - ensure_unknown_node(db, neighbor_id || neighbor_num, neighbor_num, heard_time: entry_rx_time) + ensure_unknown_node(db, neighbor_id || neighbor_num, neighbor_num, heard_time: entry_rx_time, protocol: protocol) - neighbor_entries << [neighbor_id, snr, entry_rx_time, ingestor] + neighbor_entries << [neighbor_id, snr, entry_rx_time, ingestor, protocol] end with_busy_retry do @@ -790,17 +806,18 @@ module PotatoMesh end end - neighbor_entries.each do |neighbor_id, snr_value, heard_time, reporter_id| + neighbor_entries.each do |neighbor_id, snr_value, heard_time, reporter_id, proto| db.execute( <<~SQL, - INSERT INTO neighbors(node_id, neighbor_id, snr, rx_time, ingestor) - VALUES (?, ?, ?, ?, ?) + INSERT INTO neighbors(node_id, neighbor_id, snr, rx_time, ingestor, protocol) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(node_id, neighbor_id) DO UPDATE SET snr = excluded.snr, rx_time = excluded.rx_time, - ingestor = COALESCE(NULLIF(neighbors.ingestor,''), excluded.ingestor) + ingestor = COALESCE(NULLIF(neighbors.ingestor,''), excluded.ingestor), + protocol = COALESCE(NULLIF(neighbors.protocol,'meshtastic'), excluded.protocol) SQL - [node_id, neighbor_id, snr_value, heard_time, reporter_id], + [node_id, neighbor_id, snr_value, heard_time, reporter_id, proto], ) end end @@ -814,7 +831,8 @@ module PotatoMesh rx_time, metrics = {}, lora_freq: nil, - modem_preset: nil + modem_preset: nil, + protocol: "meshtastic" ) num = coerce_integer(node_num) id = string_or_nil(node_id) @@ -824,7 +842,7 @@ module PotatoMesh id ||= format("!%08x", num & 0xFFFFFFFF) if num return unless id - ensure_unknown_node(db, id, num, heard_time: rx_time) + ensure_unknown_node(db, id, num, heard_time: rx_time, protocol: protocol) touch_node_last_seen( db, id, @@ -926,6 +944,35 @@ module PotatoMesh private :resolve_numeric_metric + # Look up the protocol registered by a given ingestor node. + # + # @param db [SQLite3::Database] open database handle. + # @param ingestor_node_id [String, nil] the node_id of the reporting ingestor. + # @param cache [Hash, nil] optional per-request memoization hash; pass a shared + # Hash instance across a batch to avoid redundant DB lookups per record. + # @return [String] protocol string; defaults to "meshtastic" when absent or unknown. + def resolve_protocol(db, ingestor_node_id, cache: nil) + return "meshtastic" if ingestor_node_id.nil? || ingestor_node_id.to_s.strip.empty? + + if cache + return cache[ingestor_node_id] if cache.key?(ingestor_node_id) + + result = db.get_first_value( + "SELECT protocol FROM ingestors WHERE node_id = ? LIMIT 1", + [ingestor_node_id], + ) || "meshtastic" + cache[ingestor_node_id] = result + return result + end + + db.get_first_value( + "SELECT protocol FROM ingestors WHERE node_id = ? LIMIT 1", + [ingestor_node_id], + ) || "meshtastic" + end + + private :resolve_protocol + # Normalise a traceroute hop entry to a numeric node identifier. # # @param hop [Object] raw hop entry from the payload. @@ -964,7 +1011,7 @@ module PotatoMesh hop_entries.filter_map { |entry| coerce_trace_node_id(entry) } end - def insert_telemetry(db, payload) + def insert_telemetry(db, payload, protocol_cache: nil) return unless payload.is_a?(Hash) telemetry_id = coerce_integer(payload["id"] || payload["packet_id"]) @@ -1011,6 +1058,7 @@ module PotatoMesh lora_freq = coerce_integer(payload["lora_freq"] || payload["loraFrequency"]) modem_preset = string_or_nil(payload["modem_preset"] || payload["modemPreset"]) ingestor = string_or_nil(payload["ingestor"]) + protocol = resolve_protocol(db, ingestor, cache: protocol_cache) telemetry_section = normalize_json_object(payload["telemetry"]) device_metrics = normalize_json_object(payload["device_metrics"] || payload["deviceMetrics"]) @@ -1341,6 +1389,7 @@ module PotatoMesh soil_moisture, soil_temperature, ingestor, + protocol, ] placeholders = Array.new(row.length, "?").join(",") @@ -1348,7 +1397,7 @@ module PotatoMesh with_busy_retry do db.execute <<~SQL, row INSERT INTO telemetry(id,node_id,node_num,from_id,to_id,rx_time,rx_iso,telemetry_time,channel,portnum,hop_limit,snr,rssi,bitfield,payload_b64, - battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds,temperature,relative_humidity,barometric_pressure,gas_resistance,current,iaq,distance,lux,white_lux,ir_lux,uv_lux,wind_direction,wind_speed,weight,wind_gust,wind_lull,radiation,rainfall_1h,rainfall_24h,soil_moisture,soil_temperature,ingestor) + battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds,temperature,relative_humidity,barometric_pressure,gas_resistance,current,iaq,distance,lux,white_lux,ir_lux,uv_lux,wind_direction,wind_speed,weight,wind_gust,wind_lull,radiation,rainfall_1h,rainfall_24h,soil_moisture,soil_temperature,ingestor,protocol) VALUES (#{placeholders}) ON CONFLICT(id) DO UPDATE SET node_id=COALESCE(excluded.node_id,telemetry.node_id), @@ -1391,7 +1440,8 @@ module PotatoMesh rainfall_24h=COALESCE(excluded.rainfall_24h,telemetry.rainfall_24h), soil_moisture=COALESCE(excluded.soil_moisture,telemetry.soil_moisture), soil_temperature=COALESCE(excluded.soil_temperature,telemetry.soil_temperature), - ingestor=COALESCE(NULLIF(telemetry.ingestor,''), excluded.ingestor) + ingestor=COALESCE(NULLIF(telemetry.ingestor,''), excluded.ingestor), + protocol=COALESCE(NULLIF(telemetry.protocol,'meshtastic'), excluded.protocol) SQL end @@ -1409,6 +1459,7 @@ module PotatoMesh }, lora_freq: lora_freq, modem_preset: modem_preset, + protocol: protocol, ) end @@ -1417,7 +1468,7 @@ module PotatoMesh # @param db [SQLite3::Database] open database handle. # @param payload [Hash] traceroute payload as produced by the ingestor. # @return [void] - def insert_trace(db, payload) + def insert_trace(db, payload, protocol_cache: nil) return unless payload.is_a?(Hash) trace_identifier = coerce_integer(payload["id"] || payload["packet_id"] || payload["packetId"]) @@ -1443,20 +1494,21 @@ module PotatoMesh metrics&.[]("latencyMs"), ) ingestor = string_or_nil(payload["ingestor"]) + protocol = resolve_protocol(db, ingestor, cache: protocol_cache) hops_value = payload.key?("hops") ? payload["hops"] : payload["path"] hops = normalize_trace_hops(hops_value) all_nodes = [src, dest, *hops].compact.uniq all_nodes.each do |node| - ensure_unknown_node(db, node, node, heard_time: rx_time) + ensure_unknown_node(db, node, node, heard_time: rx_time, protocol: protocol) touch_node_last_seen(db, node, node, rx_time: rx_time, source: :trace) end with_busy_retry do - db.execute <<~SQL, [trace_identifier, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, ingestor] - INSERT INTO traces(id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, ingestor) - VALUES(?,?,?,?,?,?,?,?,?,?) + db.execute <<~SQL, [trace_identifier, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, ingestor, protocol] + INSERT INTO traces(id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, ingestor, protocol) + VALUES(?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET request_id=COALESCE(excluded.request_id,traces.request_id), src=COALESCE(excluded.src,traces.src), @@ -1466,7 +1518,8 @@ module PotatoMesh rssi=COALESCE(excluded.rssi,traces.rssi), snr=COALESCE(excluded.snr,traces.snr), elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms), - ingestor=COALESCE(NULLIF(traces.ingestor,''), excluded.ingestor) + ingestor=COALESCE(NULLIF(traces.ingestor,''), excluded.ingestor), + protocol=COALESCE(NULLIF(traces.protocol,'meshtastic'), excluded.protocol) SQL trace_id = trace_identifier || db.last_insert_row_id @@ -1534,7 +1587,7 @@ module PotatoMesh } end - def insert_message(db, message) + def insert_message(db, message, protocol_cache: nil) return unless message.is_a?(Hash) msg_id = coerce_integer(message["id"] || message["packet_id"]) @@ -1628,6 +1681,7 @@ module PotatoMesh reply_id = coerce_integer(message["reply_id"] || message["replyId"]) emoji = string_or_nil(message["emoji"]) ingestor = string_or_nil(message["ingestor"]) + protocol = resolve_protocol(db, ingestor, cache: protocol_cache) row = [ msg_id, @@ -1648,11 +1702,12 @@ module PotatoMesh reply_id, emoji, ingestor, + protocol, ] with_busy_retry do existing = db.get_first_row( - "SELECT from_id, to_id, text, encrypted, lora_freq, modem_preset, channel_name, reply_id, emoji, portnum, ingestor FROM messages WHERE id = ?", + "SELECT from_id, to_id, text, encrypted, lora_freq, modem_preset, channel_name, reply_id, emoji, portnum, ingestor, protocol FROM messages WHERE id = ?", [msg_id], ) if existing @@ -1756,6 +1811,10 @@ module PotatoMesh updates["ingestor"] = ingestor if existing_ingestor.nil? end + existing_protocol = existing.is_a?(Hash) ? existing["protocol"] : existing[11] + return if existing_protocol && existing_protocol != "meshtastic" && existing_protocol != protocol + updates["protocol"] = protocol if (existing_protocol.nil? || existing_protocol == "meshtastic") && protocol != "meshtastic" + unless updates.empty? assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ") db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [msg_id]) @@ -1765,12 +1824,12 @@ module PotatoMesh begin db.execute <<~SQL, row - INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,portnum,text,encrypted,snr,rssi,hop_limit,lora_freq,modem_preset,channel_name,reply_id,emoji,ingestor) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,portnum,text,encrypted,snr,rssi,hop_limit,lora_freq,modem_preset,channel_name,reply_id,emoji,ingestor,protocol) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) SQL rescue SQLite3::ConstraintException existing_row = db.get_first_row( - "SELECT text, encrypted, ingestor FROM messages WHERE id = ?", + "SELECT text, encrypted, ingestor, protocol FROM messages WHERE id = ?", [msg_id], ) existing_text = existing_row.is_a?(Hash) ? existing_row["text"] : existing_row&.[](0) @@ -1780,6 +1839,10 @@ module PotatoMesh existing_encrypted_str = existing_encrypted&.to_s existing_ingestor = existing_row.is_a?(Hash) ? existing_row["ingestor"] : existing_row&.[](2) existing_ingestor = string_or_nil(existing_ingestor) + existing_fallback_protocol = existing_row.is_a?(Hash) ? existing_row["protocol"] : existing_row&.[](3) + # Guard against cross-protocol contamination in the constraint fallback path, + # mirroring the same guard applied in the primary update path above. + return if existing_fallback_protocol && existing_fallback_protocol != "meshtastic" && existing_fallback_protocol != protocol decrypted_precedence = text && (clear_encrypted || (existing_encrypted_str && !existing_encrypted_str.strip.empty?)) fallback_updates = {} @@ -1808,6 +1871,7 @@ module PotatoMesh fallback_updates["reply_id"] = reply_id unless reply_id.nil? fallback_updates["emoji"] = emoji if emoji fallback_updates["ingestor"] = ingestor if ingestor && existing_ingestor.nil? + fallback_updates["protocol"] = protocol if (existing_fallback_protocol.nil? || existing_fallback_protocol == "meshtastic") && protocol != "meshtastic" unless fallback_updates.empty? assignments = fallback_updates.keys.map { |column| "#{column} = ?" }.join(", ") db.execute("UPDATE messages SET #{assignments} WHERE id = ?", fallback_updates.values + [msg_id]) @@ -1860,7 +1924,7 @@ module PotatoMesh should_touch_message = !stored_decrypted if should_touch_message - ensure_unknown_node(db, from_id || raw_from_id, message["from_num"], heard_time: rx_time) + ensure_unknown_node(db, from_id || raw_from_id, message["from_num"], heard_time: rx_time, protocol: protocol) touch_node_last_seen( db, from_id || raw_from_id || message["from_num"], @@ -1871,7 +1935,7 @@ module PotatoMesh modem_preset: modem_preset, ) - ensure_unknown_node(db, to_id || raw_to_id, message["to_num"], heard_time: rx_time) if to_id || raw_to_id + ensure_unknown_node(db, to_id || raw_to_id, message["to_num"], heard_time: rx_time, protocol: protocol) if to_id || raw_to_id if to_id || raw_to_id || message.key?("to_num") touch_node_last_seen( db, diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index dea857d..d307a6a 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -126,6 +126,11 @@ module PotatoMesh db.execute("ALTER TABLE nodes ADD COLUMN modem_preset TEXT") end + unless node_columns.include?("protocol") + db.execute("ALTER TABLE nodes ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE nodes SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end + message_columns = db.execute("PRAGMA table_info(messages)").map { |row| row[1] } unless message_columns.include?("lora_freq") @@ -153,6 +158,11 @@ module PotatoMesh db.execute("ALTER TABLE messages ADD COLUMN ingestor TEXT") end + unless message_columns.include?("protocol") + db.execute("ALTER TABLE messages ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE messages SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end + reply_index_exists = db.get_first_value( "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='idx_messages_reply_id'", @@ -195,6 +205,11 @@ module PotatoMesh db.execute("ALTER TABLE telemetry ADD COLUMN ingestor TEXT") end + unless telemetry_columns.include?("protocol") + db.execute("ALTER TABLE telemetry ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE telemetry SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end + position_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='positions'").flatten if position_tables.empty? @@ -206,6 +221,11 @@ module PotatoMesh db.execute("ALTER TABLE positions ADD COLUMN ingestor TEXT") end + unless position_columns.include?("protocol") + db.execute("ALTER TABLE positions ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE positions SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end + neighbor_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='neighbors'").flatten if neighbor_tables.empty? @@ -217,6 +237,11 @@ module PotatoMesh db.execute("ALTER TABLE neighbors ADD COLUMN ingestor TEXT") end + unless neighbor_columns.include?("protocol") + db.execute("ALTER TABLE neighbors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE neighbors SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end + trace_tables = db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('traces','trace_hops')", @@ -230,6 +255,11 @@ module PotatoMesh db.execute("ALTER TABLE traces ADD COLUMN ingestor TEXT") end + unless trace_columns.include?("protocol") + db.execute("ALTER TABLE traces ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE traces SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end + ingestor_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestors'").flatten if ingestor_tables.empty? @@ -246,6 +276,11 @@ module PotatoMesh unless ingestor_columns.include?("modem_preset") db.execute("ALTER TABLE ingestors ADD COLUMN modem_preset TEXT") end + + unless ingestor_columns.include?("protocol") + db.execute("ALTER TABLE ingestors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") + db.execute("UPDATE ingestors SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") + end end rescue SQLite3::SQLException, Errno::ENOENT => e warn_log( diff --git a/web/lib/potato_mesh/application/queries.rb b/web/lib/potato_mesh/application/queries.rb index 9f969a2..ca6449b 100644 --- a/web/lib/potato_mesh/application/queries.rb +++ b/web/lib/potato_mesh/application/queries.rb @@ -20,6 +20,7 @@ module PotatoMesh MAX_QUERY_LIMIT = 1000 DEFAULT_TELEMETRY_WINDOW_SECONDS = 86_400 DEFAULT_TELEMETRY_BUCKET_SECONDS = 300 + PROTOCOL_CLAUSE = "protocol = ?".freeze TELEMETRY_ZERO_INVALID_COLUMNS = %w[battery_level voltage].freeze TELEMETRY_AGGREGATE_COLUMNS = %w[ @@ -95,6 +96,22 @@ module PotatoMesh value end + # Append a protocol equality clause to an existing WHERE clause list when a + # protocol filter is specified. Mutates +where_clauses+ and +params+ in place. + # + # @param where_clauses [Array] accumulating WHERE conditions. + # @param params [Array] accumulating bind parameters. + # @param protocol [String, nil] optional protocol value to filter by. + # @param table_alias [String, nil] optional table alias prefix (e.g. "m" → "m.protocol = ?"). + # @return [void] + def append_protocol_filter(where_clauses, params, protocol, table_alias: nil) + return unless protocol + + clause = table_alias ? "#{table_alias}.#{PROTOCOL_CLAUSE}" : PROTOCOL_CLAUSE + where_clauses << clause + params << protocol + end + # Normalise a caller-provided limit to a sane, positive integer. # # @param limit [Object] value coerced to an integer. @@ -252,7 +269,7 @@ module PotatoMesh # @param node_ref [String, Integer, nil] optional node reference to narrow results. # @param since [Integer] unix timestamp threshold applied in addition to the rolling window for collections. # @return [Array] compacted node rows suitable for API responses. - def query_nodes(limit, node_ref: nil, since: 0) + def query_nodes(limit, node_ref: nil, since: 0, protocol: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) db.results_as_hash = true @@ -277,12 +294,14 @@ module PotatoMesh where_clauses << "(role IS NULL OR role <> 'CLIENT_HIDDEN')" end + append_protocol_filter(where_clauses, params, protocol) + sql = <<~SQL SELECT node_id, short_name, long_name, hw_model, role, snr, battery_level, voltage, last_heard, first_heard, uptime_seconds, channel_utilization, air_util_tx, position_time, location_source, precision_bits, - latitude, longitude, altitude, lora_freq, modem_preset + latitude, longitude, altitude, lora_freq, modem_preset, protocol FROM nodes SQL sql += " WHERE #{where_clauses.join(" AND ")}\n" if where_clauses.any? @@ -323,22 +342,26 @@ module PotatoMesh # @param limit [Integer] maximum number of ingestors to return. # @param since [Integer] unix timestamp threshold applied in addition to the rolling window for collections. # @return [Array] compacted ingestor rows suitable for API responses. - def query_ingestors(limit, since: 0) + def query_ingestors(limit, since: 0, protocol: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) db.results_as_hash = true now = Time.now.to_i cutoff = now - PotatoMesh::Config.week_seconds since_threshold = normalize_since_threshold(since, floor: cutoff) + where_clauses = ["last_seen_time >= ?"] + params = [since_threshold] + append_protocol_filter(where_clauses, params, protocol) sql = <<~SQL - SELECT node_id, start_time, last_seen_time, version, lora_freq, modem_preset + SELECT node_id, start_time, last_seen_time, version, lora_freq, modem_preset, protocol FROM ingestors - WHERE last_seen_time >= ? + WHERE #{where_clauses.join(" AND ")} ORDER BY last_seen_time DESC LIMIT ? SQL + params << limit - rows = db.execute(sql, [since_threshold, limit]) + rows = db.execute(sql, params) rows.each do |row| row.delete_if { |key, _| key.is_a?(Integer) } start_time = coerce_integer(row["start_time"]) @@ -366,7 +389,7 @@ module PotatoMesh # @param include_encrypted [Boolean] when true, include encrypted payloads in the response. # @param since [Integer] unix timestamp threshold; messages with rx_time older than this are excluded. # @return [Array] compacted message rows safe for API responses. - def query_messages(limit, node_ref: nil, include_encrypted: false, since: 0) + def query_messages(limit, node_ref: nil, include_encrypted: false, since: 0, protocol: nil) limit = coerce_query_limit(limit) since_threshold = normalize_since_threshold(since, floor: 0) db = open_database(readonly: true) @@ -390,11 +413,13 @@ module PotatoMesh params.concat(clause.last) end + append_protocol_filter(where_clauses, params, protocol, table_alias: "m") + sql = <<~SQL SELECT m.id, m.rx_time, m.rx_iso, m.from_id, m.to_id, m.channel, m.portnum, m.text, m.encrypted, m.rssi, m.hop_limit, m.lora_freq, m.modem_preset, m.channel_name, m.snr, - m.reply_id, m.emoji, m.ingestor + m.reply_id, m.emoji, m.ingestor, m.protocol FROM messages m SQL sql += " WHERE #{where_clauses.join(" AND ")}\n" @@ -455,7 +480,7 @@ module PotatoMesh # @param node_ref [String, Integer, nil] optional node reference to scope results. # @param since [Integer] unix timestamp threshold applied in addition to the rolling window. # @return [Array] compacted position rows suitable for API responses. - def query_positions(limit, node_ref: nil, since: 0) + def query_positions(limit, node_ref: nil, since: 0, protocol: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) db.results_as_hash = true @@ -475,6 +500,8 @@ module PotatoMesh params.concat(clause.last) end + append_protocol_filter(where_clauses, params, protocol) + sql = <<~SQL SELECT * FROM positions SQL @@ -514,7 +541,7 @@ module PotatoMesh # @param node_ref [String, Integer, nil] optional node reference to scope results. # @param since [Integer] unix timestamp threshold applied in addition to the rolling window for collections. # @return [Array] compacted neighbor rows suitable for API responses. - def query_neighbors(limit, node_ref: nil, since: 0) + def query_neighbors(limit, node_ref: nil, since: 0, protocol: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) db.results_as_hash = true @@ -534,6 +561,8 @@ module PotatoMesh params.concat(clause.last) end + append_protocol_filter(where_clauses, params, protocol) + sql = <<~SQL SELECT * FROM neighbors SQL @@ -562,7 +591,7 @@ module PotatoMesh # @param node_ref [String, Integer, nil] optional node reference to scope results. # @param since [Integer] unix timestamp threshold applied in addition to the rolling window for collections. # @return [Array] compacted telemetry rows suitable for API responses. - def query_telemetry(limit, node_ref: nil, since: 0) + def query_telemetry(limit, node_ref: nil, since: 0, protocol: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) db.results_as_hash = true @@ -582,6 +611,8 @@ module PotatoMesh params.concat(clause.last) end + append_protocol_filter(where_clauses, params, protocol) + sql = <<~SQL SELECT * FROM telemetry SQL @@ -771,7 +802,7 @@ module PotatoMesh # @param node_ref [String, Integer, nil] optional node reference to scope results. # @param since [Integer] unix timestamp threshold applied in addition to the rolling window. # @return [Array] compacted trace rows suitable for API responses. - def query_traces(limit, node_ref: nil, since: 0) + def query_traces(limit, node_ref: nil, since: 0, protocol: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) db.results_as_hash = true @@ -798,8 +829,10 @@ module PotatoMesh 3.times { params.concat(numeric_values) } end + append_protocol_filter(where_clauses, params, protocol) + sql = <<~SQL - SELECT id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms + SELECT id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms, protocol FROM traces SQL sql += " WHERE #{where_clauses.join(" AND ")}\n" if where_clauses.any? diff --git a/web/lib/potato_mesh/application/routes/api.rb b/web/lib/potato_mesh/application/routes/api.rb index f3fd5f9..a713951 100644 --- a/web/lib/potato_mesh/application/routes/api.rb +++ b/web/lib/potato_mesh/application/routes/api.rb @@ -64,7 +64,7 @@ module PotatoMesh app.get "/api/nodes" do content_type :json limit = [params["limit"]&.to_i || 200, 1000].min - query_nodes(limit, since: params["since"]).to_json + query_nodes(limit, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/stats" do @@ -88,7 +88,7 @@ module PotatoMesh app.get "/api/ingestors" do content_type :json limit = coerce_query_limit(params["limit"]) - query_ingestors(limit, since: params["since"]).to_json + query_ingestors(limit, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/messages" do @@ -97,7 +97,7 @@ module PotatoMesh include_encrypted = coerce_boolean(params["encrypted"]) || false since = coerce_integer(params["since"]) since = 0 if since.nil? || since.negative? - query_messages(limit, include_encrypted: include_encrypted, since: since).to_json + query_messages(limit, include_encrypted: include_encrypted, since: since, protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/messages/:id" do @@ -113,13 +113,14 @@ module PotatoMesh node_ref: node_ref, include_encrypted: include_encrypted, since: since, + protocol: string_or_nil(params["protocol"]), ).to_json end app.get "/api/positions" do content_type :json limit = [params["limit"]&.to_i || 200, 1000].min - query_positions(limit, since: params["since"]).to_json + query_positions(limit, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/positions/:id" do @@ -127,13 +128,13 @@ module PotatoMesh node_ref = string_or_nil(params["id"]) halt 400, { error: "missing node id" }.to_json unless node_ref limit = [params["limit"]&.to_i || 200, 1000].min - query_positions(limit, node_ref: node_ref, since: params["since"]).to_json + query_positions(limit, node_ref: node_ref, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/neighbors" do content_type :json limit = [params["limit"]&.to_i || 200, 1000].min - query_neighbors(limit, since: params["since"]).to_json + query_neighbors(limit, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/neighbors/:id" do @@ -141,13 +142,13 @@ module PotatoMesh node_ref = string_or_nil(params["id"]) halt 400, { error: "missing node id" }.to_json unless node_ref limit = [params["limit"]&.to_i || 200, 1000].min - query_neighbors(limit, node_ref: node_ref, since: params["since"]).to_json + query_neighbors(limit, node_ref: node_ref, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/telemetry" do content_type :json limit = [params["limit"]&.to_i || 200, 1000].min - query_telemetry(limit, since: params["since"]).to_json + query_telemetry(limit, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/telemetry/aggregated" do @@ -190,13 +191,13 @@ module PotatoMesh node_ref = string_or_nil(params["id"]) halt 400, { error: "missing node id" }.to_json unless node_ref limit = [params["limit"]&.to_i || 200, 1000].min - query_telemetry(limit, node_ref: node_ref, since: params["since"]).to_json + query_telemetry(limit, node_ref: node_ref, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/traces" do content_type :json limit = [params["limit"]&.to_i || 200, 1000].min - query_traces(limit, since: params["since"]).to_json + query_traces(limit, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/traces/:id" do @@ -204,7 +205,7 @@ module PotatoMesh node_ref = string_or_nil(params["id"]) halt 400, { error: "missing node id" }.to_json unless node_ref limit = [params["limit"]&.to_i || 200, 1000].min - query_traces(limit, node_ref: node_ref, since: params["since"]).to_json + query_traces(limit, node_ref: node_ref, since: params["since"], protocol: string_or_nil(params["protocol"])).to_json end app.get "/api/instances" do diff --git a/web/lib/potato_mesh/application/routes/ingest.rb b/web/lib/potato_mesh/application/routes/ingest.rb index 52fcf57..5851f35 100644 --- a/web/lib/potato_mesh/application/routes/ingest.rb +++ b/web/lib/potato_mesh/application/routes/ingest.rb @@ -35,10 +35,14 @@ module PotatoMesh unless data.is_a?(Hash) halt 400, { error: "invalid payload" }.to_json end - halt 400, { error: "too many nodes" }.to_json if data.size > 1000 + node_count = data.count { |k, _| k != "ingestor" } + halt 400, { error: "too many nodes" }.to_json if node_count > 1000 db = open_database + ingestor_node_id = string_or_nil(data["ingestor"]) + protocol = resolve_protocol(db, ingestor_node_id) data.each do |node_id, node| - upsert_node(db, node_id, node) + next if node_id == "ingestor" + upsert_node(db, node_id, node, protocol: protocol) end PotatoMesh::App::Prometheus::NODES_GAUGE.set(query_nodes(1000).length) { status: "ok" }.to_json @@ -57,8 +61,9 @@ module PotatoMesh messages = data.is_a?(Array) ? data : [data] halt 400, { error: "too many messages" }.to_json if messages.size > 1000 db = open_database + protocol_cache = {} messages.each do |msg| - insert_message(db, msg) + insert_message(db, msg, protocol_cache: protocol_cache) end { status: "ok" }.to_json ensure @@ -305,8 +310,9 @@ module PotatoMesh positions = data.is_a?(Array) ? data : [data] halt 400, { error: "too many positions" }.to_json if positions.size > 1000 db = open_database + protocol_cache = {} positions.each do |pos| - insert_position(db, pos) + insert_position(db, pos, protocol_cache: protocol_cache) end { status: "ok" }.to_json ensure @@ -324,8 +330,9 @@ module PotatoMesh neighbor_payloads = data.is_a?(Array) ? data : [data] halt 400, { error: "too many neighbor packets" }.to_json if neighbor_payloads.size > 1000 db = open_database + protocol_cache = {} neighbor_payloads.each do |packet| - insert_neighbors(db, packet) + insert_neighbors(db, packet, protocol_cache: protocol_cache) end { status: "ok" }.to_json ensure @@ -343,8 +350,9 @@ module PotatoMesh telemetry_packets = data.is_a?(Array) ? data : [data] halt 400, { error: "too many telemetry packets" }.to_json if telemetry_packets.size > 1000 db = open_database + protocol_cache = {} telemetry_packets.each do |packet| - insert_telemetry(db, packet) + insert_telemetry(db, packet, protocol_cache: protocol_cache) end { status: "ok" }.to_json ensure @@ -362,8 +370,9 @@ module PotatoMesh trace_packets = data.is_a?(Array) ? data : [data] halt 400, { error: "too many traces" }.to_json if trace_packets.size > 1000 db = open_database + protocol_cache = {} trace_packets.each do |packet| - insert_trace(db, packet) + insert_trace(db, packet, protocol_cache: protocol_cache) end { status: "ok" }.to_json ensure diff --git a/web/spec/protocol_spec.rb b/web/spec/protocol_spec.rb new file mode 100644 index 0000000..94e48bf --- /dev/null +++ b/web/spec/protocol_spec.rb @@ -0,0 +1,518 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# frozen_string_literal: true + +require "spec_helper" +require "json" +require "time" + +RSpec.describe "Multi-protocol support" do + let(:app) { Sinatra::Application } + let(:api_token) { "test-token" } + let(:auth_headers) do + { + "CONTENT_TYPE" => "application/json", + "HTTP_AUTHORIZATION" => "Bearer #{api_token}", + } + end + let(:now) { Time.now.to_i } + + MESHCORE_INGESTOR_ID = "!11223344".freeze + ALT_NODE_ID = "!aabbccdd".freeze + ALT_NODE_ID2 = "!ccddee00".freeze + MESH_NODE_ID = "!mesh0001".freeze + CORE_NODE_ID = "!core0001".freeze + MESH_INGESTOR_ID = "!mesh9999".freeze + SELECT_INGESTOR_PROTOCOL_SQL = "SELECT protocol FROM ingestors WHERE node_id = ?".freeze + + before do + @original_token = ENV.fetch("API_TOKEN", nil) + ENV["API_TOKEN"] = api_token + clear_tables + end + + after do + ENV["API_TOKEN"] = @original_token + clear_tables + end + + # Open a database connection for direct inspection. + # + # @param readonly [Boolean] whether to open in read-only mode. + # @yieldparam db [SQLite3::Database] open database handle. + # @return [void] + def with_db(readonly: false) + db = PotatoMesh::Application.open_database(readonly: readonly) + db.results_as_hash = true + yield db + ensure + db&.close + end + + # Remove all rows from tables exercised by these tests. + # + # @return [void] + def clear_tables + with_db do |db| + db.execute("DELETE FROM trace_hops") + db.execute("DELETE FROM traces") + db.execute("DELETE FROM neighbors") + db.execute("DELETE FROM messages") + db.execute("DELETE FROM positions") + db.execute("DELETE FROM telemetry") + db.execute("DELETE FROM nodes") + db.execute("DELETE FROM ingestors") + end + end + + # Register an ingestor via the API and return the response. + # + # @param node_id [String] canonical ingestor node identifier. + # @param protocol [String, nil] mesh protocol string; omit to test default. + # @return [Rack::MockResponse] the POST response. + def register_ingestor(node_id, protocol: nil) + payload = { + node_id: node_id, + start_time: now - 60, + last_seen_time: now, + version: "0.5.11", + } + payload[:protocol] = protocol if protocol + post "/api/ingestors", payload.to_json, auth_headers + last_response + end + + describe "POST /api/ingestors" do + it "stores protocol when provided" do + register_ingestor(MESHCORE_INGESTOR_ID, protocol: "meshcore") + + expect(last_response.status).to eq(200) + with_db(readonly: true) do |db| + row = db.get_first_row(SELECT_INGESTOR_PROTOCOL_SQL, [MESHCORE_INGESTOR_ID]) + expect(row["protocol"]).to eq("meshcore") + end + end + + it "defaults protocol to meshtastic when field is absent" do + register_ingestor("!aabbccdd") + + expect(last_response.status).to eq(200) + with_db(readonly: true) do |db| + row = db.get_first_row(SELECT_INGESTOR_PROTOCOL_SQL, ["!aabbccdd"]) + expect(row["protocol"]).to eq("meshtastic") + end + end + + it "updates protocol on re-registration" do + register_ingestor(MESHCORE_INGESTOR_ID, protocol: "meshtastic") + register_ingestor(MESHCORE_INGESTOR_ID, protocol: "meshcore") + + with_db(readonly: true) do |db| + row = db.get_first_row(SELECT_INGESTOR_PROTOCOL_SQL, [MESHCORE_INGESTOR_ID]) + expect(row["protocol"]).to eq("meshcore") + end + end + end + + describe "protocol propagation to event tables" do + before do + register_ingestor(MESHCORE_INGESTOR_ID, protocol: "meshcore") + end + + it "writes meshcore protocol to messages that reference a meshcore ingestor" do + msg = { + id: 42, + rx_time: now - 10, + rx_iso: Time.at(now - 10).utc.iso8601, + text: "hello from meshcore", + ingestor: MESHCORE_INGESTOR_ID, + } + post "/api/messages", [msg].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM messages WHERE id = ?", [42]) + expect(row["protocol"]).to eq("meshcore") + end + end + + it "writes meshcore protocol to positions that reference a meshcore ingestor" do + pos = { + id: 100, + rx_time: now - 5, + rx_iso: Time.at(now - 5).utc.iso8601, + node_id: ALT_NODE_ID, + latitude: 1.0, + longitude: 2.0, + ingestor: MESHCORE_INGESTOR_ID, + } + post "/api/positions", [pos].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM positions WHERE id = ?", [100]) + expect(row["protocol"]).to eq("meshcore") + end + end + + it "writes meshcore protocol to telemetry that references a meshcore ingestor" do + tel = { + id: 200, + rx_time: now - 5, + rx_iso: Time.at(now - 5).utc.iso8601, + node_id: ALT_NODE_ID, + battery_level: 80, + ingestor: MESHCORE_INGESTOR_ID, + } + post "/api/telemetry", [tel].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM telemetry WHERE id = ?", [200]) + expect(row["protocol"]).to eq("meshcore") + end + end + + it "writes meshcore protocol to traces that reference a meshcore ingestor" do + trace = { + id: 300, + src: 0x11223344, + dest: 0xaabbccdd, + rx_time: now - 5, + rx_iso: Time.at(now - 5).utc.iso8601, + hops: [], + ingestor: MESHCORE_INGESTOR_ID, + } + post "/api/traces", [trace].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM traces WHERE id = ?", [300]) + expect(row["protocol"]).to eq("meshcore") + end + end + + it "uses protocol-derived long_name for auto-created placeholder nodes" do + msg = { + id: 43, + rx_time: now - 10, + rx_iso: Time.at(now - 10).utc.iso8601, + from_id: "!11223300", + text: "unknown sender", + ingestor: MESHCORE_INGESTOR_ID, + } + post "/api/messages", [msg].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT long_name FROM nodes WHERE node_id = ?", ["!11223300"]) + expect(row["long_name"]).to eq("Meshcore 3300") + end + end + + it "does not merge a message update from a different protocol" do + msg = { + id: 500, + rx_time: now - 10, + rx_iso: Time.at(now - 10).utc.iso8601, + text: "meshcore original", + ingestor: MESHCORE_INGESTOR_ID, + } + post "/api/messages", [msg].to_json, auth_headers + expect(last_response.status).to eq(200) + + # Meshtastic ingestor posts same ID — should be ignored + meshtastic_msg = { + id: 500, + rx_time: now - 5, + rx_iso: Time.at(now - 5).utc.iso8601, + text: "meshtastic impostor", + } + post "/api/messages", [meshtastic_msg].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT text, protocol FROM messages WHERE id = ?", [500]) + expect(row["text"]).to eq("meshcore original") + expect(row["protocol"]).to eq("meshcore") + end + end + + it "does not overwrite a meshcore message via the constraint-fallback path" do + # Seed the message directly in the DB so the first INSERT triggers a + # constraint exception, exercising the rescue SQLite3::ConstraintException + # fallback path rather than the primary update branch. + with_db do |db| + db.execute( + "INSERT INTO messages(id, rx_time, rx_iso, text, protocol) VALUES(?,?,?,?,?)", + [501, now - 20, Time.at(now - 20).utc.iso8601, "meshcore seeded", "meshcore"], + ) + end + + # A Meshtastic payload arrives with the same packet ID and new text. + # The fallback path must not overwrite the existing meshcore record. + meshtastic_msg = { + id: 501, + rx_time: now - 5, + rx_iso: Time.at(now - 5).utc.iso8601, + text: "meshtastic fallback attempt", + } + post "/api/messages", [meshtastic_msg].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT text, protocol FROM messages WHERE id = ?", [501]) + expect(row["text"]).to eq("meshcore seeded") + expect(row["protocol"]).to eq("meshcore") + end + end + end + + describe "POST /api/nodes with ingestor key" do + it "inherits protocol from registered ingestor" do + register_ingestor(MESHCORE_INGESTOR_ID, protocol: "meshcore") + with_db do |db| + db.execute( + "INSERT INTO nodes(node_id, num, last_heard, first_heard) VALUES(?,?,?,?)", + [ALT_NODE_ID, 0xaabbccdd, now - 100, now - 200], + ) + end + + payload = { + ALT_NODE_ID => { "num" => 0xaabbccdd, "lastHeard" => now - 10 }, + "ingestor" => MESHCORE_INGESTOR_ID, + } + post "/api/nodes", payload.to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM nodes WHERE node_id = ?", [ALT_NODE_ID]) + expect(row["protocol"]).to eq("meshcore") + end + end + + it "defaults to meshtastic when ingestor key is absent" do + with_db do |db| + db.execute( + "INSERT INTO nodes(node_id, num, last_heard, first_heard) VALUES(?,?,?,?)", + [ALT_NODE_ID2, 0xccddee00, now - 100, now - 200], + ) + end + + payload = { ALT_NODE_ID2 => { "num" => 0xccddee00, "lastHeard" => now - 10 } } + post "/api/nodes", payload.to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM nodes WHERE node_id = ?", [ALT_NODE_ID2]) + expect(row["protocol"]).to eq("meshtastic") + end + end + + it "does not count the ingestor key against the node batch limit" do + # Build exactly 1000 node entries plus the ingestor key — should succeed + nodes = (1..1000).each_with_object({}) do |i, h| + h[format("!%08x", i)] = { "num" => i, "lastHeard" => now - 1 } + end + nodes["ingestor"] = MESHCORE_INGESTOR_ID + post "/api/nodes", nodes.to_json, auth_headers + + expect(last_response.status).to eq(200) + end + end + + describe "GET ?protocol= filter" do + before do + register_ingestor(MESHCORE_INGESTOR_ID, protocol: "meshcore") + with_db do |db| + db.execute( + "INSERT INTO nodes(node_id, num, last_heard, first_heard, protocol) VALUES(?,?,?,?,?)", + [MESH_NODE_ID, 1, now - 10, now - 20, "meshtastic"], + ) + db.execute( + "INSERT INTO nodes(node_id, num, last_heard, first_heard, protocol) VALUES(?,?,?,?,?)", + [CORE_NODE_ID, 2, now - 10, now - 20, "meshcore"], + ) + db.execute( + "INSERT INTO messages(id, rx_time, rx_iso, text, protocol) VALUES(?,?,?,?,?)", + [1001, now - 5, Time.at(now - 5).utc.iso8601, "meshtastic msg", "meshtastic"], + ) + db.execute( + "INSERT INTO messages(id, rx_time, rx_iso, text, protocol) VALUES(?,?,?,?,?)", + [1002, now - 5, Time.at(now - 5).utc.iso8601, "meshcore msg", "meshcore"], + ) + db.execute( + "INSERT INTO positions(id, rx_time, rx_iso, node_id, protocol) VALUES(?,?,?,?,?)", + [2001, now - 5, Time.at(now - 5).utc.iso8601, MESH_NODE_ID, "meshtastic"], + ) + db.execute( + "INSERT INTO positions(id, rx_time, rx_iso, node_id, protocol) VALUES(?,?,?,?,?)", + [2002, now - 5, Time.at(now - 5).utc.iso8601, CORE_NODE_ID, "meshcore"], + ) + db.execute( + "INSERT INTO neighbors(node_id, neighbor_id, rx_time, protocol) VALUES(?,?,?,?)", + [MESH_NODE_ID, CORE_NODE_ID, now - 5, "meshtastic"], + ) + db.execute( + "INSERT INTO neighbors(node_id, neighbor_id, rx_time, protocol) VALUES(?,?,?,?)", + [CORE_NODE_ID, MESH_NODE_ID, now - 5, "meshcore"], + ) + db.execute( + "INSERT INTO telemetry(id, rx_time, rx_iso, node_id, protocol) VALUES(?,?,?,?,?)", + [3001, now - 5, Time.at(now - 5).utc.iso8601, MESH_NODE_ID, "meshtastic"], + ) + db.execute( + "INSERT INTO telemetry(id, rx_time, rx_iso, node_id, protocol) VALUES(?,?,?,?,?)", + [3002, now - 5, Time.at(now - 5).utc.iso8601, CORE_NODE_ID, "meshcore"], + ) + db.execute( + "INSERT INTO traces(id, rx_time, rx_iso, protocol) VALUES(?,?,?,?)", + [4001, now - 5, Time.at(now - 5).utc.iso8601, "meshtastic"], + ) + db.execute( + "INSERT INTO traces(id, rx_time, rx_iso, protocol) VALUES(?,?,?,?)", + [4002, now - 5, Time.at(now - 5).utc.iso8601, "meshcore"], + ) + end + end + + it "filters /api/nodes by protocol" do + get "/api/nodes?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + ids = JSON.parse(last_response.body).map { |r| r["node_id"] } + expect(ids).to include(CORE_NODE_ID) + expect(ids).not_to include(MESH_NODE_ID) + end + + it "filters /api/messages by protocol" do + get "/api/messages?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + texts = JSON.parse(last_response.body).map { |r| r["text"] } + expect(texts).to include("meshcore msg") + expect(texts).not_to include("meshtastic msg") + end + + it "filters /api/positions by protocol" do + get "/api/positions?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + ids = JSON.parse(last_response.body).map { |r| r["id"] } + expect(ids).to include(2002) + expect(ids).not_to include(2001) + end + + it "filters /api/neighbors by protocol" do + get "/api/neighbors?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + rows = JSON.parse(last_response.body) + expect(rows.any? { |r| r["node_id"] == CORE_NODE_ID }).to be(true) + expect(rows.none? { |r| r["node_id"] == MESH_NODE_ID }).to be(true) + end + + it "filters /api/telemetry by protocol" do + get "/api/telemetry?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + ids = JSON.parse(last_response.body).map { |r| r["id"] } + expect(ids).to include(3002) + expect(ids).not_to include(3001) + end + + it "filters /api/traces by protocol" do + get "/api/traces?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + ids = JSON.parse(last_response.body).map { |r| r["id"] } + expect(ids).to include(4002) + expect(ids).not_to include(4001) + end + + it "filters /api/ingestors by protocol" do + with_db do |db| + db.execute( + "INSERT INTO ingestors(node_id, start_time, last_seen_time, version, protocol) VALUES(?,?,?,?,?)", + [MESH_INGESTOR_ID, now - 60, now, "0.5.11", "meshtastic"], + ) + end + + get "/api/ingestors?protocol=meshcore", {}, auth_headers + + expect(last_response.status).to eq(200) + ids = JSON.parse(last_response.body).map { |r| r["node_id"] } + expect(ids).to include(MESHCORE_INGESTOR_ID) + expect(ids).not_to include(MESH_INGESTOR_ID) + end + + it "returns all records when protocol param is absent" do + get "/api/nodes", {}, auth_headers + + expect(last_response.status).to eq(200) + ids = JSON.parse(last_response.body).map { |r| r["node_id"] } + expect(ids).to include(MESH_NODE_ID) + expect(ids).to include(CORE_NODE_ID) + end + + it "includes protocol field in GET /api/messages responses" do + get "/api/messages", {}, auth_headers + + expect(last_response.status).to eq(200) + rows = JSON.parse(last_response.body) + expect(rows.all? { |r| r.key?("protocol") }).to be(true) + end + + it "includes protocol field in GET /api/nodes responses" do + get "/api/nodes", {}, auth_headers + + expect(last_response.status).to eq(200) + rows = JSON.parse(last_response.body) + expect(rows.all? { |r| r.key?("protocol") }).to be(true) + end + end + + describe "backward compatibility" do + it "existing payloads without protocol field default to meshtastic" do + msg = { + id: 999, + rx_time: now - 10, + rx_iso: Time.at(now - 10).utc.iso8601, + text: "legacy message", + } + post "/api/messages", [msg].to_json, auth_headers + expect(last_response.status).to eq(200) + + with_db(readonly: true) do |db| + row = db.get_first_row("SELECT protocol FROM messages WHERE id = ?", [999]) + expect(row["protocol"]).to eq("meshtastic") + end + end + + it "existing ingestor registrations without protocol default to meshtastic in GET responses" do + with_db do |db| + db.execute( + "INSERT INTO ingestors(node_id, start_time, last_seen_time, version, protocol) VALUES(?,?,?,?,?)", + ["!legacy00", now - 120, now - 10, "0.5.0", "meshtastic"], + ) + end + + get "/api/ingestors", {}, auth_headers + expect(last_response.status).to eq(200) + entry = JSON.parse(last_response.body).find { |r| r["node_id"] == "!legacy00" } + expect(entry["protocol"]).to eq("meshtastic") + end + end +end