From 96421c346df464555a9eee89883b8f823e532f52 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sat, 14 Feb 2026 14:29:05 +0100 Subject: [PATCH] ingestor: report self id per packet (#635) * ingestor: report self id per packet * ingestor: address review comments * ingestor: address review comments * ingestor: address review comments * ingestor: address review comments --- data/mesh_ingestor/handlers.py | 5 + data/messages.sql | 3 +- data/neighbors.sql | 1 + data/positions.sql | 3 +- data/telemetry.sql | 3 +- data/traces.sql | 3 +- tests/test_mesh.py | 10 + .../application/data_processing.rb | 80 ++++-- web/lib/potato_mesh/application/database.rb | 32 +++ web/lib/potato_mesh/application/queries.rb | 2 +- web/spec/app_spec.rb | 265 +++++++++++++++++- web/spec/database_spec.rb | 59 ++++ 12 files changed, 439 insertions(+), 27 deletions(-) diff --git a/data/mesh_ingestor/handlers.py b/data/mesh_ingestor/handlers.py index ca4572c..e9fcd75 100644 --- a/data/mesh_ingestor/handlers.py +++ b/data/mesh_ingestor/handlers.py @@ -424,6 +424,7 @@ def store_position_packet(packet: Mapping, decoded: Mapping) -> None: "hop_limit": hop_limit, "bitfield": bitfield, "payload_b64": payload_b64, + "ingestor": host_node_id(), } if raw_payload: position_payload["raw"] = raw_payload @@ -568,6 +569,7 @@ def store_traceroute_packet(packet: Mapping, decoded: Mapping) -> None: "rssi": rssi, "snr": snr, "elapsed_ms": elapsed_ms, + "ingestor": host_node_id(), } _queue_post_json( @@ -935,6 +937,7 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None: "rssi": rssi, "hop_limit": hop_limit, "payload_b64": payload_b64, + "ingestor": host_node_id(), } if battery_level is not None: @@ -1263,6 +1266,7 @@ def store_neighborinfo_packet(packet: Mapping, decoded: Mapping) -> None: "neighbors": neighbor_entries, "rx_time": rx_time, "rx_iso": _iso(rx_time), + "ingestor": host_node_id(), } if node_broadcast_interval is not None: @@ -1520,6 +1524,7 @@ def store_packet_dict(packet: Mapping) -> None: "hop_limit": int(hop) if hop is not None else None, "reply_id": reply_id, "emoji": emoji, + "ingestor": host_node_id(), } if not encrypted_flag and channel_name_value: diff --git a/data/messages.sql b/data/messages.sql index c401799..6803f62 100644 --- a/data/messages.sql +++ b/data/messages.sql @@ -29,7 +29,8 @@ CREATE TABLE IF NOT EXISTS messages ( modem_preset TEXT, channel_name TEXT, reply_id INTEGER, - emoji TEXT + emoji TEXT, + ingestor TEXT ); CREATE INDEX IF NOT EXISTS idx_messages_rx_time ON messages(rx_time); diff --git a/data/neighbors.sql b/data/neighbors.sql index 5b5c07d..debc75c 100644 --- a/data/neighbors.sql +++ b/data/neighbors.sql @@ -17,6 +17,7 @@ CREATE TABLE IF NOT EXISTS neighbors ( neighbor_id TEXT NOT NULL, snr REAL, rx_time INTEGER NOT NULL, + ingestor TEXT, 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/positions.sql b/data/positions.sql index 8a4dd2f..6542ad7 100644 --- a/data/positions.sql +++ b/data/positions.sql @@ -33,7 +33,8 @@ CREATE TABLE IF NOT EXISTS positions ( rssi INTEGER, hop_limit INTEGER, bitfield INTEGER, - payload_b64 TEXT + payload_b64 TEXT, + ingestor TEXT ); CREATE INDEX IF NOT EXISTS idx_positions_rx_time ON positions(rx_time); diff --git a/data/telemetry.sql b/data/telemetry.sql index f23ea7a..933e263 100644 --- a/data/telemetry.sql +++ b/data/telemetry.sql @@ -53,7 +53,8 @@ CREATE TABLE IF NOT EXISTS telemetry ( rainfall_1h REAL, rainfall_24h REAL, soil_moisture INTEGER, - soil_temperature REAL + soil_temperature REAL, + ingestor TEXT ); CREATE INDEX IF NOT EXISTS idx_telemetry_rx_time ON telemetry(rx_time); diff --git a/data/traces.sql b/data/traces.sql index f003aa7..c606f32 100644 --- a/data/traces.sql +++ b/data/traces.sql @@ -21,7 +21,8 @@ CREATE TABLE IF NOT EXISTS traces ( rx_iso TEXT NOT NULL, rssi INTEGER, snr REAL, - elapsed_ms INTEGER + elapsed_ms INTEGER, + ingestor TEXT ); CREATE TABLE IF NOT EXISTS trace_hops ( diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 5d33195..f376a25 100644 --- a/tests/test_mesh.py +++ b/tests/test_mesh.py @@ -788,6 +788,7 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch): mesh.config.LORA_FREQ = 868 mesh.config.MODEM_PRESET = "MediumFast" + mesh.register_host_node_id("!f00dbabe") packet = { "id": 123, @@ -823,6 +824,7 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch): assert payload["rssi"] == -70 assert payload["reply_id"] is None assert payload["emoji"] is None + assert payload["ingestor"] == "!f00dbabe" assert payload["lora_freq"] == 868 assert payload["modem_preset"] == "MediumFast" assert priority == mesh._MESSAGE_POST_PRIORITY @@ -879,6 +881,7 @@ def test_store_packet_dict_posts_position(mesh_module, monkeypatch): mesh.config.LORA_FREQ = 868 mesh.config.MODEM_PRESET = "MediumFast" + mesh.register_host_node_id("!f00dbabe") packet = { "id": 200498337, @@ -946,6 +949,7 @@ def test_store_packet_dict_posts_position(mesh_module, monkeypatch): ) assert payload["lora_freq"] == 868 assert payload["modem_preset"] == "MediumFast" + assert payload["ingestor"] == "!f00dbabe" assert payload["raw"]["time"] == 1_758_624_189 @@ -960,6 +964,7 @@ def test_store_packet_dict_posts_neighborinfo(mesh_module, monkeypatch): mesh.config.LORA_FREQ = 868 mesh.config.MODEM_PRESET = "MediumFast" + mesh.register_host_node_id("!f00dbabe") packet = { "id": 2049886869, @@ -1004,6 +1009,7 @@ def test_store_packet_dict_posts_neighborinfo(mesh_module, monkeypatch): assert neighbors[2]["neighbor_num"] == 0x0BAD_C0DE assert payload["lora_freq"] == 868 assert payload["modem_preset"] == "MediumFast" + assert payload["ingestor"] == "!f00dbabe" def test_store_packet_dict_handles_nodeinfo_packet(mesh_module, monkeypatch): @@ -2282,6 +2288,7 @@ def test_store_packet_dict_handles_telemetry_packet(mesh_module, monkeypatch): mesh.config.LORA_FREQ = 868 mesh.config.MODEM_PRESET = "MediumFast" + mesh.register_host_node_id("!f00dbabe") packet = { "id": 1_256_091_342, @@ -2334,6 +2341,7 @@ def test_store_packet_dict_handles_telemetry_packet(mesh_module, monkeypatch): assert payload["current"] == pytest.approx(0.0715) assert payload["lora_freq"] == 868 assert payload["modem_preset"] == "MediumFast" + assert payload["ingestor"] == "!f00dbabe" def test_store_packet_dict_handles_environment_telemetry(mesh_module, monkeypatch): @@ -2477,6 +2485,7 @@ def test_store_packet_dict_handles_traceroute_packet(mesh_module, monkeypatch): mesh.config.LORA_FREQ = 915 mesh.config.MODEM_PRESET = "LongFast" + mesh.register_host_node_id("!f00dbabe") packet = { "id": 2_934_054_466, @@ -2518,6 +2527,7 @@ def test_store_packet_dict_handles_traceroute_packet(mesh_module, monkeypatch): assert "elapsed_ms" in payload assert payload["lora_freq"] == 915 assert payload["modem_preset"] == "LongFast" + assert payload["ingestor"] == "!f00dbabe" def test_traceroute_hop_normalization_supports_mappings(mesh_module, monkeypatch): diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index b1f20a8..c8fc9e4 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -616,6 +616,7 @@ 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, @@ -639,13 +640,14 @@ module PotatoMesh hop_limit, bitfield, payload_b64, + ingestor, ] 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) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + precision_bits,sats_in_view,pdop,ground_speed,ground_track,snr,rssi,hop_limit,bitfield,payload_b64,ingestor) + 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), @@ -666,7 +668,8 @@ module PotatoMesh rssi=COALESCE(excluded.rssi,positions.rssi), hop_limit=COALESCE(excluded.hop_limit,positions.hop_limit), bitfield=COALESCE(excluded.bitfield,positions.bitfield), - payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64) + payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64), + ingestor=COALESCE(NULLIF(positions.ingestor,''), excluded.ingestor) SQL end @@ -721,6 +724,7 @@ module PotatoMesh 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 : [] @@ -759,19 +763,40 @@ module PotatoMesh ensure_unknown_node(db, neighbor_id || neighbor_num, neighbor_num, heard_time: entry_rx_time) touch_node_last_seen(db, neighbor_id || neighbor_num, neighbor_num, rx_time: entry_rx_time, source: :neighborinfo) - neighbor_entries << [neighbor_id, snr, entry_rx_time] + neighbor_entries << [neighbor_id, snr, entry_rx_time, ingestor] end with_busy_retry do db.transaction do - db.execute("DELETE FROM neighbors WHERE node_id = ?", [node_id]) - neighbor_entries.each do |neighbor_id, snr_value, heard_time| + if neighbor_entries.empty? + db.execute("DELETE FROM neighbors WHERE node_id = ?", [node_id]) + else + expected_neighbors = neighbor_entries.map(&:first).uniq + existing_neighbors = db.execute( + "SELECT neighbor_id FROM neighbors WHERE node_id = ?", + [node_id], + ).flatten + stale_neighbors = existing_neighbors - expected_neighbors + stale_neighbors.each_slice(500) do |slice| + placeholders = slice.map { "?" }.join(",") + db.execute( + "DELETE FROM neighbors WHERE node_id = ? AND neighbor_id IN (#{placeholders})", + [node_id] + slice, + ) + end + end + + neighbor_entries.each do |neighbor_id, snr_value, heard_time, reporter_id| db.execute( <<~SQL, - INSERT OR REPLACE INTO neighbors(node_id, neighbor_id, snr, rx_time) - VALUES (?, ?, ?, ?) + INSERT INTO neighbors(node_id, neighbor_id, snr, rx_time, ingestor) + 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) SQL - [node_id, neighbor_id, snr_value, heard_time], + [node_id, neighbor_id, snr_value, heard_time, reporter_id], ) end end @@ -981,6 +1006,7 @@ module PotatoMesh payload_b64 = string_or_nil(payload["payload_b64"] || payload["payload"]) 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"]) telemetry_section = normalize_json_object(payload["telemetry"]) device_metrics = normalize_json_object(payload["device_metrics"] || payload["deviceMetrics"]) @@ -1310,6 +1336,7 @@ module PotatoMesh rainfall_24h, soil_moisture, soil_temperature, + ingestor, ] placeholders = Array.new(row.length, "?").join(",") @@ -1317,7 +1344,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) + 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) VALUES (#{placeholders}) ON CONFLICT(id) DO UPDATE SET node_id=COALESCE(excluded.node_id,telemetry.node_id), @@ -1359,7 +1386,8 @@ module PotatoMesh rainfall_1h=COALESCE(excluded.rainfall_1h,telemetry.rainfall_1h), 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) + soil_temperature=COALESCE(excluded.soil_temperature,telemetry.soil_temperature), + ingestor=COALESCE(NULLIF(telemetry.ingestor,''), excluded.ingestor) SQL end @@ -1410,6 +1438,7 @@ module PotatoMesh metrics&.[]("latency_ms") || metrics&.[]("latencyMs"), ) + ingestor = string_or_nil(payload["ingestor"]) hops_value = payload.key?("hops") ? payload["hops"] : payload["path"] hops = normalize_trace_hops(hops_value) @@ -1421,9 +1450,9 @@ module PotatoMesh end with_busy_retry do - db.execute <<~SQL, [trace_identifier, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms] - INSERT INTO traces(id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms) - VALUES(?,?,?,?,?,?,?,?,?) + 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(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET request_id=COALESCE(excluded.request_id,traces.request_id), src=COALESCE(excluded.src,traces.src), @@ -1432,7 +1461,8 @@ module PotatoMesh rx_iso=excluded.rx_iso, rssi=COALESCE(excluded.rssi,traces.rssi), snr=COALESCE(excluded.snr,traces.snr), - elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms) + elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms), + ingestor=COALESCE(NULLIF(traces.ingestor,''), excluded.ingestor) SQL trace_id = trace_identifier || db.last_insert_row_id @@ -1593,6 +1623,7 @@ module PotatoMesh channel_name = string_or_nil(message["channel_name"] || message["channelName"]) reply_id = coerce_integer(message["reply_id"] || message["replyId"]) emoji = string_or_nil(message["emoji"]) + ingestor = string_or_nil(message["ingestor"]) row = [ msg_id, @@ -1612,11 +1643,12 @@ module PotatoMesh channel_name, reply_id, emoji, + ingestor, ] 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 FROM messages WHERE id = ?", + "SELECT from_id, to_id, text, encrypted, lora_freq, modem_preset, channel_name, reply_id, emoji, portnum, ingestor FROM messages WHERE id = ?", [msg_id], ) if existing @@ -1714,6 +1746,12 @@ module PotatoMesh updates["emoji"] = emoji if should_update end + if ingestor + existing_ingestor = existing.is_a?(Hash) ? existing["ingestor"] : existing[10] + existing_ingestor = string_or_nil(existing_ingestor) + updates["ingestor"] = ingestor if existing_ingestor.nil? + end + unless updates.empty? assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ") db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [msg_id]) @@ -1723,12 +1761,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) - 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) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) SQL rescue SQLite3::ConstraintException existing_row = db.get_first_row( - "SELECT text, encrypted FROM messages WHERE id = ?", + "SELECT text, encrypted, ingestor FROM messages WHERE id = ?", [msg_id], ) existing_text = existing_row.is_a?(Hash) ? existing_row["text"] : existing_row&.[](0) @@ -1736,6 +1774,8 @@ module PotatoMesh allow_encrypted_update = existing_text_str.nil? || existing_text_str.strip.empty? existing_encrypted = existing_row.is_a?(Hash) ? existing_row["encrypted"] : existing_row&.[](1) 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) decrypted_precedence = text && (clear_encrypted || (existing_encrypted_str && !existing_encrypted_str.strip.empty?)) fallback_updates = {} @@ -1763,6 +1803,7 @@ module PotatoMesh end 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? 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]) @@ -1904,6 +1945,7 @@ module PotatoMesh "lora_freq" => coerce_integer(message["lora_freq"] || message["loraFrequency"]), "modem_preset" => string_or_nil(message["modem_preset"] || message["modemPreset"]), "payload_b64" => payload_b64, + "ingestor" => string_or_nil(message["ingestor"]), } case decoded["type"] diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index a94a884..dea857d 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -149,6 +149,9 @@ module PotatoMesh db.execute("ALTER TABLE messages ADD COLUMN emoji TEXT") message_columns << "emoji" end + unless message_columns.include?("ingestor") + db.execute("ALTER TABLE messages ADD COLUMN ingestor TEXT") + end reply_index_exists = db.get_first_value( @@ -188,6 +191,31 @@ module PotatoMesh db.execute("ALTER TABLE telemetry ADD COLUMN #{name} #{type}") telemetry_columns << name end + unless telemetry_columns.include?("ingestor") + db.execute("ALTER TABLE telemetry ADD COLUMN ingestor TEXT") + end + + position_tables = + db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='positions'").flatten + if position_tables.empty? + positions_schema = File.expand_path("../../../../data/positions.sql", __dir__) + db.execute_batch(File.read(positions_schema)) + end + position_columns = db.execute("PRAGMA table_info(positions)").map { |row| row[1] } + unless position_columns.include?("ingestor") + db.execute("ALTER TABLE positions ADD COLUMN ingestor TEXT") + end + + neighbor_tables = + db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='neighbors'").flatten + if neighbor_tables.empty? + neighbors_schema = File.expand_path("../../../../data/neighbors.sql", __dir__) + db.execute_batch(File.read(neighbors_schema)) + end + neighbor_columns = db.execute("PRAGMA table_info(neighbors)").map { |row| row[1] } + unless neighbor_columns.include?("ingestor") + db.execute("ALTER TABLE neighbors ADD COLUMN ingestor TEXT") + end trace_tables = db.execute( @@ -197,6 +225,10 @@ module PotatoMesh traces_schema = File.expand_path("../../../../data/traces.sql", __dir__) db.execute_batch(File.read(traces_schema)) end + trace_columns = db.execute("PRAGMA table_info(traces)").map { |row| row[1] } + unless trace_columns.include?("ingestor") + db.execute("ALTER TABLE traces ADD COLUMN ingestor TEXT") + end ingestor_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestors'").flatten diff --git a/web/lib/potato_mesh/application/queries.rb b/web/lib/potato_mesh/application/queries.rb index 1d268b0..f400246 100644 --- a/web/lib/potato_mesh/application/queries.rb +++ b/web/lib/potato_mesh/application/queries.rb @@ -357,7 +357,7 @@ module PotatoMesh 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.reply_id, m.emoji, m.ingestor FROM messages m SQL sql += " WHERE #{where_clauses.join(" AND ")}\n" diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index baf9947..1c082af 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -2929,7 +2929,24 @@ RSpec.describe "Potato Mesh Sinatra app" do describe "POST /api/messages" do SELECT_MESSAGE_ENCRYPTED_SQL = "SELECT encrypted FROM messages WHERE id = ?".freeze + SELECT_NEIGHBOR_COUNT_BY_NODE_SQL = "SELECT COUNT(*) FROM neighbors WHERE node_id = ?".freeze NODE_INFO_LONG_NAME = "Node Info".freeze + FIRST_MESSAGE_INGESTOR_ID = "!1111aaaa".freeze + SHARED_TEST_INGESTOR_ID = "!aaaa1111".freeze + DEADBEEF_NODE_ID = "!deadbeef".freeze + NEIGHBOR_EMPTY_UPDATE_ROOT_ID = "!cafed00d".freeze + NEIGHBOR_ROOT_ID = "!1a2b3c01".freeze + NEIGHBOR_PRIMARY_ID = "!1a2b3c02".freeze + NEIGHBOR_SNR_CLEAR_ROOT_ID = "!1a2b3c10".freeze + NEIGHBOR_SNR_CLEAR_PEER_ID = "!1a2b3c11".freeze + NEIGHBOR_CHUNK_ROOT_ID = "!1a2b3c30".freeze + + def post_twice_for_ingestor(endpoint, first_payload, second_payload) + post endpoint, first_payload.to_json, auth_headers + expect(last_response).to be_ok + post endpoint, second_payload.to_json, auth_headers + expect(last_response).to be_ok + end it "persists messages from fixture data" do import_nodes_fixture @@ -3014,6 +3031,36 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(reaction_row["emoji"]).to eq("🔥") end + it "stores message ingestor and preserves the first reporter" do + first_payload = { + "id" => 77_001, + "rx_time" => reference_time.to_i - 10, + "from_id" => "!ingmsg01", + "channel" => 0, + "portnum" => "TEXT_MESSAGE_APP", + "text" => "first reporter", + "ingestor" => FIRST_MESSAGE_INGESTOR_ID, + } + second_payload = first_payload.merge( + "text" => "updated text", + "ingestor" => "!2222bbbb", + ) + + post_twice_for_ingestor("/api/messages", first_payload, second_payload) + + with_db(readonly: true) do |db| + db.results_as_hash = true + row = db.get_first_row("SELECT text, ingestor FROM messages WHERE id = ?", [first_payload["id"]]) + expect(row["text"]).to eq("updated text") + expect(row["ingestor"]).to eq(FIRST_MESSAGE_INGESTOR_ID) + end + + get "/api/messages?limit=10" + expect(last_response).to be_ok + row = JSON.parse(last_response.body).find { |entry| entry["id"] == first_payload["id"] } + expect(row["ingestor"]).to eq(FIRST_MESSAGE_INGESTOR_ID) + end + it "creates hidden nodes for unknown message senders" do payload = { "id" => 9_999, @@ -3078,7 +3125,7 @@ RSpec.describe "Potato Mesh Sinatra app" do "id" => 10_001, "rx_time" => reference_time.to_i, "from_id" => "!cafef00d", - "to_id" => "!deadbeef", + "to_id" => DEADBEEF_NODE_ID, "channel" => 0, "portnum" => "TEXT_MESSAGE_APP", "text" => "Spec participant placeholder", @@ -3095,12 +3142,12 @@ RSpec.describe "Potato Mesh Sinatra app" do <<~SQL, SELECT node_id, num, short_name, long_name, role, last_heard, first_heard FROM nodes - WHERE node_id IN ("!cafef00d", "!deadbeef") + WHERE node_id IN ("!cafef00d", "#{DEADBEEF_NODE_ID}") ORDER BY node_id SQL ) - expect(rows.map { |row| row["node_id"] }).to contain_exactly("!cafef00d", "!deadbeef") + expect(rows.map { |row| row["node_id"] }).to contain_exactly("!cafef00d", DEADBEEF_NODE_ID) rows.each do |row| expect(row["num"]).to be_an(Integer) expect(row["role"]).to eq("CLIENT_HIDDEN") @@ -3299,6 +3346,30 @@ RSpec.describe "Potato Mesh Sinatra app" do end end + it "stores position ingestor and preserves the first reporter" do + first_payload = { + "id" => 19_001, + "node_id" => "!ingpos01", + "rx_time" => reference_time.to_i - 80, + "latitude" => 52.1, + "longitude" => 13.2, + "ingestor" => SHARED_TEST_INGESTOR_ID, + } + second_payload = first_payload.merge( + "latitude" => 53.3, + "ingestor" => "!bbbb2222", + ) + + post_twice_for_ingestor("/api/positions", first_payload, second_payload) + + with_db(readonly: true) do |db| + db.results_as_hash = true + row = db.get_first_row("SELECT latitude, ingestor FROM positions WHERE id = ?", [first_payload["id"]]) + expect_same_value(row["latitude"], 53.3) + expect(row["ingestor"]).to eq(SHARED_TEST_INGESTOR_ID) + end + end + it "fills first_heard when updating an existing node without one" do node_id = "!specposfh" rx_time = reference_time.to_i - 90 @@ -3472,6 +3543,127 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(JSON.parse(last_response.body)).to be_empty end + it "removes stored neighbors when a later packet contains no neighbors" do + seed_payload = { + "node_id" => NEIGHBOR_EMPTY_UPDATE_ROOT_ID, + "rx_time" => reference_time.to_i - 50, + "neighbors" => [ + { "node_id" => DEADBEEF_NODE_ID, "snr" => -2.0 }, + ], + "ingestor" => SHARED_TEST_INGESTOR_ID, + } + empty_payload = { + "node_id" => NEIGHBOR_EMPTY_UPDATE_ROOT_ID, + "rx_time" => reference_time.to_i - 10, + "neighbors" => [], + "ingestor" => "!bbbb2222", + } + + post "/api/neighbors", seed_payload.to_json, auth_headers + expect(last_response).to be_ok + post "/api/neighbors", empty_payload.to_json, auth_headers + expect(last_response).to be_ok + + with_db(readonly: true) do |db| + remaining = db.get_first_value(SELECT_NEIGHBOR_COUNT_BY_NODE_SQL, [NEIGHBOR_EMPTY_UPDATE_ROOT_ID]) + expect(remaining).to eq(0) + end + end + + it "stores neighbor ingestor and preserves the first reporter per tuple" do + base = { + "node_id" => NEIGHBOR_ROOT_ID, + "rx_time" => reference_time.to_i - 45, + "neighbors" => [ + { "node_id" => NEIGHBOR_PRIMARY_ID, "snr" => -1.5 }, + { "node_id" => "!1a2b3c03", "snr" => -2.5 }, + ], + "ingestor" => "!aaaa9999", + } + update = { + "node_id" => NEIGHBOR_ROOT_ID, + "rx_time" => reference_time.to_i - 30, + "neighbors" => [ + { "node_id" => NEIGHBOR_PRIMARY_ID, "snr" => -0.5 }, + ], + "ingestor" => "!bbbb8888", + } + + post_twice_for_ingestor("/api/neighbors", base, update) + + with_db(readonly: true) do |db| + db.results_as_hash = true + rows = db.execute("SELECT neighbor_id, snr, ingestor FROM neighbors WHERE node_id = ? ORDER BY neighbor_id", [NEIGHBOR_ROOT_ID]) + expect(rows.size).to eq(1) + expect(rows.first["neighbor_id"]).to eq(NEIGHBOR_PRIMARY_ID) + expect_same_value(rows.first["snr"], -0.5) + expect(rows.first["ingestor"]).to eq("!aaaa9999") + end + end + + it "clears stored neighbor snr when an updated entry omits snr" do + initial = { + "node_id" => NEIGHBOR_SNR_CLEAR_ROOT_ID, + "rx_time" => reference_time.to_i - 40, + "neighbors" => [ + { "node_id" => NEIGHBOR_SNR_CLEAR_PEER_ID, "snr" => -3.25 }, + ], + } + update = { + "node_id" => NEIGHBOR_SNR_CLEAR_ROOT_ID, + "rx_time" => reference_time.to_i - 20, + "neighbors" => [ + { "node_id" => NEIGHBOR_SNR_CLEAR_PEER_ID }, + ], + } + + post "/api/neighbors", initial.to_json, auth_headers + expect(last_response).to be_ok + post "/api/neighbors", update.to_json, auth_headers + expect(last_response).to be_ok + + with_db(readonly: true) do |db| + db.results_as_hash = true + row = db.get_first_row( + "SELECT snr, rx_time FROM neighbors WHERE node_id = ? AND neighbor_id = ?", + [NEIGHBOR_SNR_CLEAR_ROOT_ID, NEIGHBOR_SNR_CLEAR_PEER_ID], + ) + expect(row["snr"]).to be_nil + expect(row["rx_time"]).to eq(update["rx_time"]) + end + end + + it "removes stale neighbors in chunked deletes" do + initial_neighbors = Array.new(1_100) do |i| + { "node_id" => format("!%08x", 0x2000_0000 + i), "snr" => -2.0 } + end + initial = { + "node_id" => NEIGHBOR_CHUNK_ROOT_ID, + "rx_time" => reference_time.to_i - 35, + "neighbors" => initial_neighbors, + } + update = { + "node_id" => NEIGHBOR_CHUNK_ROOT_ID, + "rx_time" => reference_time.to_i - 25, + "neighbors" => [ + { "node_id" => "!20000000", "snr" => -1.0 }, + ], + } + + post "/api/neighbors", initial.to_json, auth_headers + expect(last_response).to be_ok + post "/api/neighbors", update.to_json, auth_headers + expect(last_response).to be_ok + + with_db(readonly: true) do |db| + count = db.get_first_value( + SELECT_NEIGHBOR_COUNT_BY_NODE_SQL, + [NEIGHBOR_CHUNK_ROOT_ID], + ) + expect(count).to eq(1) + end + end + it "returns 400 when more than 1000 neighbor packets are provided" do payload = Array.new(1001) do |i| { "node_id" => format("!%08x", i), "rx_time" => reference_time.to_i - i } @@ -3487,6 +3679,30 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(count).to eq(0) end end + + it "handles large neighbor lists without SQLite bind overflows" do + neighbors = Array.new(1_100) do |i| + { "node_id" => format("!%08x", 0x1000_0000 + i), "snr" => -1.0 } + end + payload = { + "node_id" => "!1a2b3c20", + "rx_time" => reference_time.to_i - 15, + "neighbors" => neighbors, + } + + post "/api/neighbors", payload.to_json, auth_headers + + expect(last_response).to be_ok + expect(JSON.parse(last_response.body)).to eq("status" => "ok") + + with_db(readonly: true) do |db| + count = db.get_first_value( + SELECT_NEIGHBOR_COUNT_BY_NODE_SQL, + ["!1a2b3c20"], + ) + expect(count).to eq(1_100) + end + end end describe "POST /api/telemetry" do @@ -3644,6 +3860,27 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(JSON.parse(last_response.body)).to eq("error" => "invalid JSON") end + it "stores telemetry ingestor and preserves the first reporter" do + payload = { + "id" => 23_001, + "node_id" => "!ingtel01", + "rx_time" => reference_time.to_i - 70, + "telemetry" => { "deviceMetrics" => { "batteryLevel" => 90 } }, + "battery_level" => 90, + "ingestor" => "!1111bbbb", + } + updated = payload.merge("battery_level" => 80, "ingestor" => "!2222cccc") + + post_twice_for_ingestor("/api/telemetry", payload, updated) + + with_db(readonly: true) do |db| + db.results_as_hash = true + row = db.get_first_row("SELECT battery_level, ingestor FROM telemetry WHERE id = ?", [payload["id"]]) + expect_same_value(row["battery_level"], 80.0) + expect(row["ingestor"]).to eq("!1111bbbb") + end + end + it "returns 400 when more than 1000 telemetry packets are provided" do payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } } @@ -3763,6 +4000,28 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(JSON.parse(last_response.body)).to eq("error" => "invalid JSON") end + it "stores trace ingestor and preserves the first reporter" do + payload = { + "id" => 31_001, + "request_id" => 77, + "src" => 0x10000001, + "dest" => 0x10000002, + "rx_time" => reference_time.to_i - 50, + "hops" => [0x10000001, 0x10000002], + "ingestor" => "!aaaa0001", + } + update = payload.merge("snr" => 7.5, "ingestor" => "!bbbb0002") + + post_twice_for_ingestor("/api/traces", payload, update) + + with_db(readonly: true) do |db| + db.results_as_hash = true + row = db.get_first_row("SELECT snr, ingestor FROM traces WHERE id = ?", [payload["id"]]) + expect_same_value(row["snr"], 7.5) + expect(row["ingestor"]).to eq("!aaaa0001") + end + end + it "returns 400 when more than 1000 traces are provided" do payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } } diff --git a/web/spec/database_spec.rb b/web/spec/database_spec.rb index 316dd44..b5864f6 100644 --- a/web/spec/database_spec.rb +++ b/web/spec/database_spec.rb @@ -184,6 +184,65 @@ RSpec.describe PotatoMesh::App::Database do expect(hop_columns).to include("trace_id", "hop_index", "node_id") end + it "creates positions and neighbors tables when absent" do + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + db.execute("CREATE TABLE nodes(node_id TEXT)") + db.execute("CREATE TABLE messages(id INTEGER PRIMARY KEY)") + db.execute("CREATE TABLE telemetry(id INTEGER PRIMARY KEY, rx_time INTEGER, rx_iso TEXT)") + end + + expect(column_names_for("positions")).to be_empty + expect(column_names_for("neighbors")).to be_empty + + harness_class.ensure_schema_upgrades + + positions_columns = column_names_for("positions") + expect(positions_columns).to include("id", "node_id", "rx_time", "ingestor") + + neighbors_columns = column_names_for("neighbors") + expect(neighbors_columns).to include("node_id", "neighbor_id", "rx_time", "ingestor") + end + + it "adds ingestor columns to legacy positions neighbors and traces tables" do + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + db.execute("CREATE TABLE nodes(node_id TEXT)") + db.execute("CREATE TABLE messages(id INTEGER PRIMARY KEY)") + db.execute("CREATE TABLE telemetry(id INTEGER PRIMARY KEY, rx_time INTEGER, rx_iso TEXT)") + db.execute <<~SQL + CREATE TABLE positions ( + id INTEGER PRIMARY KEY, + rx_time INTEGER, + rx_iso TEXT, + node_id TEXT + ) + SQL + db.execute <<~SQL + CREATE TABLE neighbors ( + node_id TEXT, + neighbor_id TEXT, + rx_time INTEGER + ) + SQL + db.execute <<~SQL + CREATE TABLE traces ( + id INTEGER PRIMARY KEY, + request_id INTEGER, + src TEXT, + dest TEXT, + rx_time INTEGER, + rx_iso TEXT + ) + SQL + db.execute("CREATE TABLE trace_hops(trace_id INTEGER, hop_index INTEGER, node_id TEXT)") + end + + harness_class.ensure_schema_upgrades + + expect(column_names_for("positions")).to include("ingestor") + expect(column_names_for("neighbors")).to include("ingestor") + expect(column_names_for("traces")).to include("ingestor") + end + it "adds the contact_link column to existing instances tables" do SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| db.execute("CREATE TABLE nodes(node_id TEXT)")