From 06fb90513ff5c4d1bdd08c04d4949ce7d7b51cf9 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sun, 14 Dec 2025 18:42:17 +0100 Subject: [PATCH] data: track ingestors heartbeat (#549) * data: track ingestors heartbeat * data: address review comments * cover missing unit test vectors * cover missing unit test vectors --- data/ingestors.sql | 26 +++ data/mesh_ingestor/__init__.py | 38 +++- data/mesh_ingestor/config.py | 5 + data/mesh_ingestor/daemon.py | 44 ++++- data/mesh_ingestor/ingestors.py | 139 +++++++++++++ data/mesh_ingestor/queue.py | 2 + tests/test_mesh.py | 131 +++++++++++++ .../application/data_processing.rb | 60 ++++++ web/lib/potato_mesh/application/database.rb | 24 ++- web/lib/potato_mesh/application/queries.rb | 35 ++++ web/lib/potato_mesh/application/routes/api.rb | 6 + .../potato_mesh/application/routes/ingest.rb | 19 ++ web/spec/app_spec.rb | 1 + web/spec/ingestors_spec.rb | 182 ++++++++++++++++++ 14 files changed, 706 insertions(+), 6 deletions(-) create mode 100644 data/ingestors.sql create mode 100644 data/mesh_ingestor/ingestors.py create mode 100644 web/spec/ingestors_spec.rb diff --git a/data/ingestors.sql b/data/ingestors.sql new file mode 100644 index 0000000..810846e --- /dev/null +++ b/data/ingestors.sql @@ -0,0 +1,26 @@ +-- 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. + +PRAGMA journal_mode=WAL; + +CREATE TABLE IF NOT EXISTS ingestors ( + node_id TEXT PRIMARY KEY, + start_time INTEGER NOT NULL, + last_seen_time INTEGER NOT NULL, + version TEXT, + lora_freq INTEGER, + modem_preset TEXT +); + +CREATE INDEX IF NOT EXISTS idx_ingestors_last_seen ON ingestors(last_seen_time); diff --git a/data/mesh_ingestor/__init__.py b/data/mesh_ingestor/__init__.py index 881adf2..694e756 100644 --- a/data/mesh_ingestor/__init__.py +++ b/data/mesh_ingestor/__init__.py @@ -21,7 +21,17 @@ import threading as threading # re-exported for compatibility import sys import types -from . import channels, config, daemon, handlers, interfaces, queue, serialization +from .. import VERSION as _PACKAGE_VERSION +from . import ( + channels, + config, + daemon, + handlers, + ingestors, + interfaces, + queue, + serialization, +) __all__: list[str] = [] @@ -40,7 +50,15 @@ def _export_constants() -> None: __all__.extend(["json", "urllib", "glob", "threading", "signal"]) -for _module in (channels, daemon, handlers, interfaces, queue, serialization): +for _module in ( + channels, + daemon, + handlers, + interfaces, + queue, + serialization, + ingestors, +): _reexport(_module) _export_constants() @@ -58,6 +76,7 @@ _CONFIG_ATTRS = { "_RECONNECT_INITIAL_DELAY_SECS", "_RECONNECT_MAX_DELAY_SECS", "_CLOSE_TIMEOUT_SECS", + "_INGESTOR_HEARTBEAT_SECS", "_debug_log", } @@ -71,9 +90,16 @@ _HANDLER_ATTRS = set(handlers.__all__) _DAEMON_ATTRS = set(daemon.__all__) _SERIALIZATION_ATTRS = set(serialization.__all__) _INTERFACE_EXPORTS = set(interfaces.__all__) +_INGESTOR_ATTRS = set(ingestors.__all__) + +# Re-export the package version for callers that previously referenced +# data.mesh_ingestor.VERSION directly. +VERSION = _PACKAGE_VERSION +__all__.append("VERSION") __all__.extend(sorted(_CONFIG_ATTRS)) __all__.extend(sorted(_INTERFACE_ATTRS)) +__all__.append("VERSION") class _MeshIngestorModule(types.ModuleType): @@ -88,6 +114,10 @@ class _MeshIngestorModule(types.ModuleType): return getattr(interfaces, name) if name in _INTERFACE_EXPORTS: return getattr(interfaces, name) + if name in _INGESTOR_ATTRS: + return getattr(ingestors, name) + if name == "VERSION": + return VERSION raise AttributeError(name) def __setattr__(self, name: str, value): # type: ignore[override] @@ -122,6 +152,10 @@ class _MeshIngestorModule(types.ModuleType): setattr(serialization, name, value) super().__setattr__(name, getattr(serialization, name, value)) handled = True + if name in _INGESTOR_ATTRS: + setattr(ingestors, name, value) + super().__setattr__(name, getattr(ingestors, name, value)) + handled = True if handled: return super().__setattr__(name, value) diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index bea54c7..91a03c1 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -46,6 +46,9 @@ DEFAULT_ENERGY_ONLINE_DURATION_SECS = 300.0 DEFAULT_ENERGY_SLEEP_SECS = float(6 * 60 * 60) """Sleep duration used when energy saving mode is active.""" +DEFAULT_INGESTOR_HEARTBEAT_SECS = float(60 * 60) +"""Interval between ingestor heartbeat announcements.""" + CONNECTION = os.environ.get("CONNECTION") or os.environ.get("MESH_SERIAL") """Optional connection target for the mesh interface. @@ -134,6 +137,7 @@ _CLOSE_TIMEOUT_SECS = DEFAULT_CLOSE_TIMEOUT_SECS _INACTIVITY_RECONNECT_SECS = DEFAULT_INACTIVITY_RECONNECT_SECS _ENERGY_ONLINE_DURATION_SECS = DEFAULT_ENERGY_ONLINE_DURATION_SECS _ENERGY_SLEEP_SECS = DEFAULT_ENERGY_SLEEP_SECS +_INGESTOR_HEARTBEAT_SECS = DEFAULT_INGESTOR_HEARTBEAT_SECS # Backwards compatibility shim for legacy imports. PORT = CONNECTION @@ -190,6 +194,7 @@ __all__ = [ "_INACTIVITY_RECONNECT_SECS", "_ENERGY_ONLINE_DURATION_SECS", "_ENERGY_SLEEP_SECS", + "_INGESTOR_HEARTBEAT_SECS", "_debug_log", ] diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index 90c008a..6a6af11 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -23,7 +23,7 @@ import time from pubsub import pub -from . import config, handlers, interfaces +from . import config, handlers, ingestors, interfaces _RECEIVE_TOPICS = ( "meshtastic.receive", @@ -169,6 +169,41 @@ def _is_ble_interface(iface_obj) -> bool: return "ble_interface" in module_name +def _process_ingestor_heartbeat(iface, *, ingestor_announcement_sent: bool) -> bool: + """Send ingestor liveness heartbeats when a host id is known. + + Parameters: + iface: Active mesh interface used to extract a host node id when absent. + ingestor_announcement_sent: Whether an initial heartbeat has already + been sent during the current session. + + Returns: + Updated ``ingestor_announcement_sent`` flag reflecting whether an + initial heartbeat was transmitted. + """ + + host_id = handlers.host_node_id() + if host_id is None and iface is not None: + extracted = interfaces._extract_host_node_id(iface) + if extracted: + handlers.register_host_node_id(extracted) + host_id = handlers.host_node_id() + + if host_id: + ingestors.set_ingestor_node_id(host_id) + heartbeat_sent = ingestors.queue_ingestor_heartbeat( + force=not ingestor_announcement_sent + ) + if heartbeat_sent and not ingestor_announcement_sent: + return True + return ingestor_announcement_sent + iface_cls = getattr(iface_obj, "__class__", None) + if iface_cls is None: + return False + module_name = getattr(iface_cls, "__module__", "") or "" + return "ble_interface" in module_name + + def _connected_state(candidate) -> bool | None: """Return the connection state advertised by ``candidate``. @@ -233,6 +268,7 @@ def main(existing_interface=None) -> None: inactivity_reconnect_secs = max( 0.0, getattr(config, "_INACTIVITY_RECONNECT_SECS", 0.0) ) + ingestor_announcement_sent = False energy_saving_enabled = config.ENERGY_SAVING energy_online_secs = max(0.0, config._ENERGY_ONLINE_DURATION_SECS) @@ -288,6 +324,7 @@ def main(existing_interface=None) -> None: handlers.register_host_node_id( interfaces._extract_host_node_id(iface) ) + ingestors.set_ingestor_node_id(handlers.host_node_id()) retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) initial_snapshot_sent = False if not announced_target and resolved_target: @@ -501,6 +538,10 @@ def main(existing_interface=None) -> None: iface_connected_at = None continue + ingestor_announcement_sent = _process_ingestor_heartbeat( + iface, ingestor_announcement_sent=ingestor_announcement_sent + ) + retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) stop.wait(config.SNAPSHOT_SECS) except KeyboardInterrupt: # pragma: no cover - interactive only @@ -520,6 +561,7 @@ __all__ = [ "_node_items_snapshot", "_subscribe_receive_topics", "_is_ble_interface", + "_process_ingestor_heartbeat", "_connected_state", "main", ] diff --git a/data/mesh_ingestor/ingestors.py b/data/mesh_ingestor/ingestors.py new file mode 100644 index 0000000..fa1f3e4 --- /dev/null +++ b/data/mesh_ingestor/ingestors.py @@ -0,0 +1,139 @@ +# 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. + +"""Helpers for tracking ingestor identity and liveness announcements.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Callable + +from .. import VERSION as INGESTOR_VERSION +from . import config, queue +from .serialization import _canonical_node_id + +HEARTBEAT_INTERVAL_SECS = 60 * 60 +"""Default interval between ingestor heartbeat announcements.""" + + +@dataclass +class _IngestorState: + """Mutable ingestor identity and heartbeat tracking data.""" + + start_time: int = field(default_factory=lambda: int(time.time())) + last_heartbeat: int | None = None + node_id: str | None = None + + +STATE = _IngestorState() +"""Shared ingestor identity state.""" +# Alias retained for clarity without exporting into the top-level mesh module to +# avoid colliding with the HTTP queue state. +INGESTOR_STATE = STATE + + +def ingestor_start_time() -> int: + """Return the unix timestamp representing when the ingestor booted.""" + + return STATE.start_time + + +def set_ingestor_node_id(node_id: str | None) -> str | None: + """Record the canonical host node identifier for the ingestor. + + Parameters: + node_id: Raw node identifier reported by the connected device. + + Returns: + Canonical node identifier in ``!xxxxxxxx`` form or ``None`` when the + provided value cannot be normalised. + """ + + canonical = _canonical_node_id(node_id) + if canonical is None: + return None + + if STATE.node_id != canonical: + STATE.node_id = canonical + STATE.last_heartbeat = None + + return canonical + + +def queue_ingestor_heartbeat( + *, + force: bool = False, + send: Callable[[str, dict], None] | None = None, + node_id: str | None = None, +) -> bool: + """Queue a heartbeat payload advertising ingestor liveness. + + Parameters: + force: When ``True``, bypasses the heartbeat interval guard so an + announcement is queued immediately. + send: Optional transport callable used for tests; defaults to the queue + dispatcher. + node_id: Optional node identifier to register before sending. When + omitted the previously recorded identifier is reused. + + Returns: + ``True`` when a heartbeat payload was queued, ``False`` otherwise. + """ + + canonical = _canonical_node_id(node_id) if node_id is not None else None + if canonical: + set_ingestor_node_id(canonical) + canonical = STATE.node_id + + if canonical is None: + return False + + now = int(time.time()) + interval = max( + 0, int(getattr(config, "_INGESTOR_HEARTBEAT_SECS", HEARTBEAT_INTERVAL_SECS)) + ) + last = STATE.last_heartbeat + if not force and last is not None and now - last < interval: + return False + + payload = { + "node_id": canonical, + "start_time": STATE.start_time, + "last_seen_time": now, + "version": INGESTOR_VERSION, + } + if getattr(config, "LORA_FREQ", None) is not None: + payload["lora_freq"] = config.LORA_FREQ + if getattr(config, "MODEM_PRESET", None) is not None: + payload["modem_preset"] = config.MODEM_PRESET + queue._queue_post_json( + "/api/ingestors", + payload, + priority=getattr( + queue, "_INGESTOR_POST_PRIORITY", queue._DEFAULT_POST_PRIORITY + ), + send=send, + ) + STATE.last_heartbeat = now + return True + + +__all__ = [ + "HEARTBEAT_INTERVAL_SECS", + "INGESTOR_STATE", + "ingestor_start_time", + "queue_ingestor_heartbeat", + "set_ingestor_node_id", +] diff --git a/data/mesh_ingestor/queue.py b/data/mesh_ingestor/queue.py index d601088..74c74df 100644 --- a/data/mesh_ingestor/queue.py +++ b/data/mesh_ingestor/queue.py @@ -74,6 +74,7 @@ def _payload_key_value_pairs(payload: Mapping[str, object]) -> str: _MESSAGE_POST_PRIORITY = 10 +_INGESTOR_POST_PRIORITY = 80 _NEIGHBOR_POST_PRIORITY = 20 _TRACE_POST_PRIORITY = 25 _POSITION_POST_PRIORITY = 30 @@ -259,6 +260,7 @@ __all__ = [ "QueueState", "_DEFAULT_POST_PRIORITY", "_MESSAGE_POST_PRIORITY", + "_INGESTOR_POST_PRIORITY", "_NEIGHBOR_POST_PRIORITY", "_NODE_POST_PRIORITY", "_POSITION_POST_PRIORITY", diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 785bd47..173b0e5 100644 --- a/tests/test_mesh.py +++ b/tests/test_mesh.py @@ -20,6 +20,7 @@ import re import sys import threading import types +import time """End-to-end tests covering the mesh ingestion package.""" @@ -214,6 +215,9 @@ def mesh_module(monkeypatch): if attr in module.__dict__: delattr(module, attr) module.channels._reset_channel_cache() + module.ingestors.STATE.start_time = int(time.time()) + module.ingestors.STATE.last_heartbeat = None + module.ingestors.STATE.node_id = None yield module @@ -2661,6 +2665,133 @@ def test_queue_post_json_skips_when_active(mesh_module, monkeypatch): mesh._clear_post_queue() +def test_process_ingestor_heartbeat_updates_flag(mesh_module, monkeypatch): + mesh = mesh_module + mesh.ingestors.STATE.last_heartbeat = None + mesh.ingestors.STATE.node_id = None + mesh.handlers.register_host_node_id(None) + recorded = {"force": None, "count": 0} + + def fake_queue_ingestor_heartbeat(*, force): + recorded["force"] = force + recorded["count"] += 1 + return True + + monkeypatch.setattr( + mesh.ingestors, "queue_ingestor_heartbeat", fake_queue_ingestor_heartbeat + ) + + class DummyIface: + def __init__(self): + self.myNodeNum = 0xCAFEBABE + + updated = mesh._process_ingestor_heartbeat( + DummyIface(), ingestor_announcement_sent=False + ) + + assert updated is True + assert recorded["force"] is True + assert recorded["count"] == 1 + assert mesh.handlers.host_node_id() == "!cafebabe" + + +def test_process_ingestor_heartbeat_skips_without_host(mesh_module, monkeypatch): + mesh = mesh_module + mesh.handlers.register_host_node_id(None) + mesh.ingestors.STATE.node_id = None + mesh.ingestors.STATE.last_heartbeat = None + + monkeypatch.setattr(mesh.ingestors, "queue_ingestor_heartbeat", lambda **_: False) + + updated = mesh._process_ingestor_heartbeat(None, ingestor_announcement_sent=False) + + assert updated is False + assert mesh.ingestors.STATE.node_id is None + assert mesh.ingestors.STATE.last_heartbeat is None + + +def test_ingestor_heartbeat_respects_interval_override(mesh_module, monkeypatch): + mesh = mesh_module + mesh.ingestors.STATE.start_time = 100 + mesh.ingestors.STATE.last_heartbeat = 1_000 + mesh.ingestors.STATE.node_id = "!abcd0001" + mesh._INGESTOR_HEARTBEAT_SECS = 10_000 + monkeypatch.setattr(mesh.ingestors.time, "time", lambda: 2_000) + sent = mesh.ingestors.queue_ingestor_heartbeat() + assert sent is False + assert mesh.ingestors.STATE.last_heartbeat == 1_000 + + +def test_setting_ingestor_attr_propagates(mesh_module): + mesh = mesh_module + mesh._INGESTOR_HEARTBEAT_SECS = 123 + assert mesh.config._INGESTOR_HEARTBEAT_SECS == 123 + + +def test_queue_ingestor_heartbeat_requires_node_id(mesh_module, monkeypatch): + mesh = mesh_module + captured = [] + + monkeypatch.setattr( + mesh.queue, + "_queue_post_json", + lambda path, payload, *, priority, send=None: captured.append( + (path, payload, priority) + ), + ) + + mesh.ingestors.STATE.node_id = None + mesh.ingestors.STATE.last_heartbeat = None + + queued = mesh.ingestors.queue_ingestor_heartbeat(force=True) + + assert queued is False + assert captured == [] + + +def test_queue_ingestor_heartbeat_enqueues_and_throttles(mesh_module, monkeypatch): + mesh = mesh_module + captured = [] + + monkeypatch.setattr( + mesh.queue, + "_queue_post_json", + lambda path, payload, *, priority, send=None: captured.append( + (path, payload, priority) + ), + ) + + mesh.ingestors.STATE.start_time = 1_700_000_000 + mesh.ingestors.STATE.last_heartbeat = None + mesh.ingestors.STATE.node_id = None + mesh.config.LORA_FREQ = 915 + mesh.config.MODEM_PRESET = "LongFast" + + mesh.ingestors.set_ingestor_node_id("!CAFEBABE") + first = mesh.ingestors.queue_ingestor_heartbeat(force=True) + second = mesh.ingestors.queue_ingestor_heartbeat() + + assert first is True + assert second is False + assert len(captured) == 1 + path, payload, priority = captured[0] + assert path == "/api/ingestors" + assert payload["node_id"] == "!cafebabe" + assert payload["start_time"] == 1_700_000_000 + assert payload["last_seen_time"] >= payload["start_time"] + assert payload["version"] == mesh.VERSION + assert payload["lora_freq"] == 915 + assert payload["modem_preset"] == "LongFast" + assert priority == mesh.queue._INGESTOR_POST_PRIORITY + + +def test_mesh_version_export_matches_package(mesh_module): + import data + + mesh = mesh_module + assert mesh.VERSION == data.VERSION + + def test_node_to_dict_handles_proto_fallback(mesh_module, monkeypatch): mesh = mesh_module diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index 9a5eba9..435703a 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -199,6 +199,66 @@ module PotatoMesh updated end + # Insert or update an ingestor heartbeat payload. + # + # @param db [SQLite3::Database] open database handle. + # @param payload [Hash] ingestor payload from the collector. + # @return [Boolean] true when persistence succeeded. + def upsert_ingestor(db, payload) + return false unless payload.is_a?(Hash) + + parts = canonical_node_parts(payload["node_id"] || payload["id"]) + return false unless parts + + node_id, = parts + now = Time.now.to_i + + start_time = coerce_integer(payload["start_time"] || payload["startTime"]) || now + last_seen_time = + coerce_integer(payload["last_seen_time"] || payload["lastSeenTime"]) || start_time + + start_time = 0 if start_time.negative? + last_seen_time = 0 if last_seen_time.negative? + start_time = now if start_time > now + last_seen_time = now if last_seen_time > now + last_seen_time = start_time if last_seen_time < start_time + + version = string_or_nil(payload["version"] || payload["ingestorVersion"]) + return false unless version + lora_freq = coerce_integer(payload["lora_freq"]) + modem_preset = string_or_nil(payload["modem_preset"]) + + 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(?,?,?,?,?,?) + ON CONFLICT(node_id) DO UPDATE SET + start_time = CASE + WHEN excluded.start_time > ingestors.start_time THEN excluded.start_time + ELSE ingestors.start_time + END, + last_seen_time = CASE + WHEN excluded.last_seen_time > ingestors.last_seen_time THEN excluded.last_seen_time + ELSE ingestors.last_seen_time + 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) + SQL + end + + true + rescue SQLite3::SQLException => e + warn_log( + "Failed to upsert ingestor record", + context: "data_processing.ingestors", + node_id: node_id, + error_class: e.class.name, + error_message: e.message, + ) + false + end + def upsert_node(db, node_id, n) user = n["user"] || {} met = n["deviceMetrics"] || {} diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index 07e4685..a94a884 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -81,10 +81,10 @@ module PotatoMesh return false unless File.exist?(PotatoMesh::Config.db_path) db = open_database(readonly: true) - required = %w[nodes messages positions telemetry neighbors instances traces trace_hops] + required = %w[nodes messages positions telemetry neighbors instances traces trace_hops ingestors] tables = db.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions','telemetry','neighbors','instances','traces','trace_hops')", + "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions','telemetry','neighbors','instances','traces','trace_hops','ingestors')", ).flatten (required - tables).empty? rescue SQLite3::Exception @@ -99,7 +99,7 @@ module PotatoMesh def init_db FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path)) db = open_database - %w[nodes messages positions telemetry neighbors instances traces].each do |schema| + %w[nodes messages positions telemetry neighbors instances traces ingestors].each do |schema| sql_file = File.expand_path("../../../../data/#{schema}.sql", __dir__) db.execute_batch(File.read(sql_file)) end @@ -197,6 +197,24 @@ module PotatoMesh traces_schema = File.expand_path("../../../../data/traces.sql", __dir__) db.execute_batch(File.read(traces_schema)) end + + ingestor_tables = + db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestors'").flatten + if ingestor_tables.empty? + ingestors_schema = File.expand_path("../../../../data/ingestors.sql", __dir__) + db.execute_batch(File.read(ingestors_schema)) + else + ingestor_columns = db.execute("PRAGMA table_info(ingestors)").map { |row| row[1] } + unless ingestor_columns.include?("version") + db.execute("ALTER TABLE ingestors ADD COLUMN version TEXT") + end + unless ingestor_columns.include?("lora_freq") + db.execute("ALTER TABLE ingestors ADD COLUMN lora_freq INTEGER") + end + unless ingestor_columns.include?("modem_preset") + db.execute("ALTER TABLE ingestors ADD COLUMN modem_preset TEXT") + end + end rescue SQLite3::SQLException, Errno::ENOENT => e warn_log( "Failed to apply schema upgrade", diff --git a/web/lib/potato_mesh/application/queries.rb b/web/lib/potato_mesh/application/queries.rb index ef1a3b9..5c0a123 100644 --- a/web/lib/potato_mesh/application/queries.rb +++ b/web/lib/potato_mesh/application/queries.rb @@ -262,6 +262,41 @@ module PotatoMesh db&.close end + def query_ingestors(limit) + 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 + sql = <<~SQL + SELECT node_id, start_time, last_seen_time, version, lora_freq, modem_preset + FROM ingestors + WHERE last_seen_time >= ? + ORDER BY last_seen_time DESC + LIMIT ? + SQL + + rows = db.execute(sql, [cutoff, limit]) + rows.each do |row| + row.delete_if { |key, _| key.is_a?(Integer) } + start_time = coerce_integer(row["start_time"]) + last_seen_time = coerce_integer(row["last_seen_time"]) + start_time = now if start_time && start_time > now + last_seen_time = now if last_seen_time && last_seen_time > now + if start_time && last_seen_time && last_seen_time < start_time + last_seen_time = start_time + end + row["start_time"] = start_time + row["last_seen_time"] = last_seen_time + row["start_time_iso"] = Time.at(start_time).utc.iso8601 if start_time + row["last_seen_iso"] = Time.at(last_seen_time).utc.iso8601 if last_seen_time + end + + rows.map { |row| compact_api_row(row) } + ensure + db&.close + end + # Fetch chat messages with optional filtering. # # @param limit [Integer] maximum number of rows to return. diff --git a/web/lib/potato_mesh/application/routes/api.rb b/web/lib/potato_mesh/application/routes/api.rb index a558bbc..c336e86 100644 --- a/web/lib/potato_mesh/application/routes/api.rb +++ b/web/lib/potato_mesh/application/routes/api.rb @@ -77,6 +77,12 @@ module PotatoMesh rows.first.to_json end + app.get "/api/ingestors" do + content_type :json + limit = coerce_query_limit(params["limit"]) + query_ingestors(limit).to_json + end + app.get "/api/messages" do content_type :json limit = [params["limit"]&.to_i || 200, 1000].min diff --git a/web/lib/potato_mesh/application/routes/ingest.rb b/web/lib/potato_mesh/application/routes/ingest.rb index c625450..e8f5636 100644 --- a/web/lib/potato_mesh/application/routes/ingest.rb +++ b/web/lib/potato_mesh/application/routes/ingest.rb @@ -65,6 +65,25 @@ module PotatoMesh db&.close end + app.post "/api/ingestors" do + require_token! + content_type :json + begin + payload = JSON.parse(read_json_body) + rescue JSON::ParserError + halt 400, { error: "invalid JSON" }.to_json + end + unless payload.is_a?(Hash) + halt 400, { error: "invalid payload" }.to_json + end + db = open_database + stored = upsert_ingestor(db, payload) + halt 400, { error: "invalid payload" }.to_json unless stored + { status: "ok" }.to_json + ensure + db&.close + end + app.post "/api/instances" do content_type :json begin diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 8a2d13b..0ca33d8 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -103,6 +103,7 @@ RSpec.describe "Potato Mesh Sinatra app" do db.execute("DELETE FROM nodes") db.execute("DELETE FROM positions") db.execute("DELETE FROM telemetry") + db.execute("DELETE FROM ingestors") end ensure_self_instance_record! end diff --git a/web/spec/ingestors_spec.rb b/web/spec/ingestors_spec.rb new file mode 100644 index 0000000..a6cd732 --- /dev/null +++ b/web/spec/ingestors_spec.rb @@ -0,0 +1,182 @@ +# 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 "Ingestor endpoints" do + let(:app) { Sinatra::Application } + let(:api_token) { "secret-token" } + let(:auth_headers) do + { + "CONTENT_TYPE" => "application/json", + "HTTP_AUTHORIZATION" => "Bearer #{api_token}", + } + end + + before do + @original_token = ENV["API_TOKEN"] + ENV["API_TOKEN"] = api_token + clear_ingestors_table + end + + after do + ENV["API_TOKEN"] = @original_token + clear_ingestors_table + end + + def clear_ingestors_table + with_db do |db| + db.execute("DELETE FROM ingestors") + db.execute("VACUUM") + end + end + + def with_db(readonly: false) + db = PotatoMesh::Application.open_database(readonly: readonly) + db.busy_timeout = PotatoMesh::Config.db_busy_timeout_ms + db.execute("PRAGMA foreign_keys = ON") + yield db + ensure + db&.close + end + + def ingestor_payload(overrides = {}) + now = Time.now.to_i + { + node_id: "!abc12345", + start_time: now - 120, + last_seen_time: now - 60, + version: "0.5.7", + lora_freq: 915, + modem_preset: "LongFast", + }.merge(overrides) + end + + describe "POST /api/ingestors" do + it "requires a bearer token" do + post "/api/ingestors", ingestor_payload.to_json, { "CONTENT_TYPE" => "application/json" } + + expect(last_response.status).to eq(403) + end + + it "upserts ingestor state without regressing start time" do + payload = ingestor_payload + post "/api/ingestors", payload.to_json, auth_headers + + expect(last_response.status).to eq(200) + + newer_last_seen = payload[:last_seen_time] + 3_600 + older_start = payload[:start_time] - 500 + post "/api/ingestors", + payload.merge(last_seen_time: newer_last_seen, start_time: older_start).to_json, + auth_headers + + expect(last_response.status).to eq(200) + with_db(readonly: true) do |db| + row = db.get_first_row( + "SELECT node_id, start_time, last_seen_time, version, lora_freq, modem_preset FROM ingestors WHERE node_id = ?", + [payload[:node_id]], + ) + expect(row[0]).to eq(payload[:node_id]) + expect(row[1]).to eq(payload[:start_time]) + expect(row[2]).to be >= payload[:last_seen_time] + expect(row[2]).to be <= Time.now.to_i + expect(row[3]).to eq(payload[:version]) + expect(row[4]).to eq(payload[:lora_freq]) + expect(row[5]).to eq(payload[:modem_preset]) + end + end + + it "rejects payloads missing required fields" do + post "/api/ingestors", { node_id: "!abcd0001" }.to_json, auth_headers + + expect(last_response.status).to eq(400) + end + + it "rejects invalid JSON" do + post "/api/ingestors", "{", auth_headers + + expect(last_response.status).to eq(400) + end + + it "rejects payloads missing version" do + post "/api/ingestors", ingestor_payload(version: nil).to_json, auth_headers + + expect(last_response.status).to eq(400) + end + + it "rejects non-object payloads" do + post "/api/ingestors", [].to_json, auth_headers + + expect(last_response.status).to eq(400) + end + end + + describe "GET /api/ingestors" do + it "returns recent ingestors and omits stale rows" do + now = Time.now.to_i + with_db do |db| + db.execute( + "INSERT INTO ingestors(node_id, start_time, last_seen_time, version) VALUES(?,?,?,?)", + ["!fresh000", now - 100, now - 10, "0.5.7"], + ) + db.execute( + "INSERT INTO ingestors(node_id, start_time, last_seen_time, version) VALUES(?,?,?,?)", + ["!stale000", now - (9 * 24 * 60 * 60), now - (9 * 24 * 60 * 60), "0.5.6"], + ) + db.execute( + "INSERT INTO ingestors(node_id, start_time, last_seen_time, version, lora_freq, modem_preset) VALUES(?,?,?,?,?,?)", + ["!rich000", now - 200, now - 100, "0.5.8", 915, "MediumFast"], + ) + end + + get "/api/ingestors" + + expect(last_response.status).to eq(200) + payload = JSON.parse(last_response.body) + expect(payload).to all(include("node_id", "start_time", "last_seen_time", "version")) + node_ids = payload.map { |entry| entry["node_id"] } + expect(node_ids).to include("!fresh000") + expect(node_ids).not_to include("!stale000") + rich = payload.find { |row| row["node_id"] == "!rich000" } + expect(rich["lora_freq"]).to eq(915) + expect(rich["modem_preset"]).to eq("MediumFast") + expect(rich["start_time_iso"]).to be_a(String) + expect(rich["last_seen_iso"]).to be_a(String) + end + end + + describe "schema migrations" do + it "creates the ingestors table with frequency and modem columns" do + tmp_db = File.join(SPEC_TMPDIR, "ingestor-migrate.db") + FileUtils.rm_f(tmp_db) + original = PotatoMesh::Config.db_path + allow(PotatoMesh::Config).to receive(:db_path).and_return(tmp_db) + + begin + PotatoMesh::Application.init_db + with_db(readonly: true) do |db| + columns = db.execute("PRAGMA table_info(ingestors)").map { |row| row[1] } + expect(columns).to include("lora_freq", "modem_preset", "version") + end + ensure + allow(PotatoMesh::Config).to receive(:db_path).and_return(original) + end + end + end +end