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

@@ -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);

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)

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",
 ]
 

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",
 ]

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",
+]

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",

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
 

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"] || {}

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",

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.

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

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

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

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