fix(meshtastic): persist dashboard-ready positions

opentakserver/controllers/meshtastic_controller.py

@@ -17,6 +17,8 @@
 
 from opentakserver.controllers.rabbitmq_client import RabbitMQClient
 from opentakserver.extensions import db, logger, socketio
+from opentakserver.functions import datetime_from_iso8601_string
+from opentakserver.models.CoT import CoT
 from opentakserver.models.EUD import EUD
 from opentakserver.models.Meshtastic import MeshtasticChannel
 from opentakserver.models.Point import Point
@@ -310,7 +312,13 @@ def cot(self, pb, from_id, to_id, portnum, how="m-g", cot_type="a-f-G-U-C", uid=
 
         return event
 
-    def insert_or_update_eud(self, uid: str, from_id: str, update_if_exists: bool = True):
+    def insert_or_update_eud(
+        self,
+        uid: str,
+        from_id: str,
+        update_if_exists: bool = True,
+        last_event_time: datetime.datetime | None = None,
+    ):
         eud = EUD()
         eud.uid = uid
         eud.callsign = self.meshtastic_devices[from_id]["long_name"]
@@ -321,6 +329,9 @@ def insert_or_update_eud(self, uid: str, from_id: str, update_if_exists: bool =
         eud.team_role = self.meshtastic_devices[from_id]["role"]
         eud.meshtastic_id = int(self.meshtastic_devices[from_id]["meshtastic_id"], 16)
         eud.meshtastic_macaddr = self.meshtastic_devices[from_id]["macaddr"]
+        if last_event_time:
+            eud.last_event_time = last_event_time
+            eud.last_status = "Connected"
 
         with self.context:
             socketio.emit("eud", eud.to_json(), namespace="/socket.io")
@@ -342,17 +353,46 @@ def insert_or_update_eud(self, uid: str, from_id: str, update_if_exists: bool =
                         sqlalchemy.update(EUD).where(EUD.uid == uid).values(**eud.serialize())
                     )
                     db.session.commit()
+                elif last_event_time:
+                    db.session.execute(
+                        sqlalchemy.update(EUD)
+                        .where(EUD.uid == uid)
+                        .values(last_event_time=last_event_time, last_status="Connected")
+                    )
+                    db.session.commit()
+
+    def resolve_position_uid(self, from_id: str) -> str:
+        device = self.meshtastic_devices[from_id]
+        mapped_uid = device["uid"]
+        if mapped_uid == from_id:
+            return from_id
+
+        requested_uid = mapped_uid or from_id
+        with self.context:
+            eud = db.session.execute(db.session.query(EUD).filter_by(uid=requested_uid)).scalar()
+            if not eud:
+                try:
+                    meshtastic_id = int(from_id, 16)
+                except ValueError:
+                    return requested_uid
+                eud = db.session.execute(
+                    db.session.query(EUD).filter_by(meshtastic_id=meshtastic_id)
+                ).scalar()
+
+        device["uid"] = eud.uid if eud else requested_uid
+        return device["uid"]
 
     def position(self, pb, from_id, to_id, portnum):
         try:
+            device_uid = self.resolve_position_uid(from_id)
             if (
                 portnum == "MAP_REPORT_APP"
                 and pb.firmware_version != self.meshtastic_devices[from_id]["firmware_version"]
             ):
                 try:
                     with self.context:
                         eud = self.db.session.execute(
-                            self.db.session.query(EUD).filter_by(uid=from_id)
+                            self.db.session.query(EUD).filter_by(uid=device_uid)
                         ).first()[0]
                         eud.version = pb.firmware_version
                         eud.device = pb.hw_model
@@ -406,27 +446,48 @@ def position(self, pb, from_id, to_id, portnum):
                     pb.ground_speed if pb.ground_speed else "0.0"
                 )
 
-            self.insert_or_update_eud(from_id, from_id, False)
+            event = self.cot(pb, from_id, to_id, portnum, uid=device_uid)
+            device_uid = event.attrib["uid"]
+            event_time = datetime_from_iso8601_string(event.attrib["time"])
+            stale_time = datetime_from_iso8601_string(event.attrib["stale"])
+
+            self.insert_or_update_eud(device_uid, from_id, False, last_event_time=event_time)
+
+            cot = CoT()
+            cot.how = event.attrib["how"]
+            cot.type = event.attrib["type"]
+            cot.uid = event.attrib["uid"]
+            cot.sender_uid = device_uid
+            cot.timestamp = event_time
+            cot.start = event_time
+            cot.stale = stale_time
+            cot.xml = tostring(event).decode("utf-8")
 
             point = Point()
-            point.uid = str(uuid.uuid4())
-            point.device_uid = from_id
+            point.uid = event.attrib["uid"]
+            point.device_uid = device_uid
             point.latitude = self.meshtastic_devices[from_id]["last_lat"]
             point.longitude = self.meshtastic_devices[from_id]["last_lon"]
             point.hae = self.meshtastic_devices[from_id]["last_alt"]
             point.course = self.meshtastic_devices[from_id]["course"]
             point.speed = self.meshtastic_devices[from_id]["speed"]
-            point.timestamp = datetime.datetime.now(tz=datetime.timezone.utc)
+            point.timestamp = event_time
             point.point = f"POINT({self.meshtastic_devices[from_id]["last_lon"]} {self.meshtastic_devices[from_id]["last_lat"]})"
 
             with self.context:
+                db.session.add(cot)
+                db.session.flush()
+                point.cot_id = cot.id
+                point.cot = cot
                 db.session.add(point)
                 db.session.commit()
 
                 socketio.emit("point", point.to_json(), namespace="/socket.io")
 
-            return self.cot(pb, from_id, to_id, portnum)
+            return event
         except BaseException as e:
+            with self.context:
+                db.session.rollback()
             self.logger.error("Failed to create CoT: {}".format(str(e)))
             self.logger.error(traceback.format_exc())
             return

tests/test_meshtastic_position_integrity.py

@@ -0,0 +1,251 @@
+import datetime
+import types
+import unittest
+from unittest.mock import MagicMock, patch
+
+from opentakserver.controllers import meshtastic_controller as module
+
+
+class _Context:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        return False
+
+
+class _Query:
+    def __init__(self):
+        self.filters = {}
+
+    def filter_by(self, **filters):
+        self.filters = filters
+        return self
+
+
+class _Result:
+    def __init__(self, value):
+        self.value = value
+
+    def scalar(self):
+        return self.value
+
+
+class _Session:
+    def __init__(self, commit_error=None, euds_by_uid=None, euds_by_meshtastic_id=None):
+        self.added = []
+        self.commit_error = commit_error
+        self.rolled_back = False
+        self.euds_by_uid = euds_by_uid or {}
+        self.euds_by_meshtastic_id = euds_by_meshtastic_id or {}
+
+    def query(self, model):
+        return _Query()
+
+    def execute(self, query):
+        if "uid" in query.filters:
+            return _Result(self.euds_by_uid.get(query.filters["uid"]))
+        return _Result(self.euds_by_meshtastic_id.get(query.filters.get("meshtastic_id")))
+
+    def add(self, value):
+        self.added.append(value)
+
+    def flush(self):
+        for value in self.added:
+            if isinstance(value, _CoT) and value.id is None:
+                value.id = 17
+
+    def commit(self):
+        if self.commit_error:
+            raise self.commit_error
+
+    def rollback(self):
+        self.rolled_back = True
+
+
+class _SocketIO:
+    def __init__(self):
+        self.emitted = []
+
+    def emit(self, event, payload, namespace=None):
+        self.emitted.append((event, payload, namespace))
+
+
+class _CoT:
+    def __init__(self):
+        self.id = None
+
+
+class _Point:
+    def __init__(self):
+        self.cot_id = None
+        self.cot = None
+
+    def to_json(self):
+        return {
+            "uid": self.uid,
+            "how": self.cot.how if self.cot else None,
+            "type": self.cot.type if self.cot else None,
+        }
+
+
+class MeshtasticPositionIntegrityTest(unittest.TestCase):
+    def make_controller(self, mapped_uid=None):
+        controller = object.__new__(module.MeshtasticController)
+        controller.context = _Context()
+        controller.logger = MagicMock()
+        controller.meshtastic_devices = {
+            "a1b2c3d4": {
+                "hw_model": "HELTEC_V3",
+                "long_name": "Mesh Alpha",
+                "short_name": "Alpha",
+                "macaddr": "AAECAwQF",
+                "firmware_version": "2.7.6",
+                "last_lat": "0.0",
+                "last_lon": "0.0",
+                "battery": 87,
+                "meshtastic_id": "a1b2c3d4",
+                "voltage": 0,
+                "uptime": 0,
+                "last_alt": "9999999.0",
+                "course": "0.0",
+                "speed": "0.0",
+                "team": "Cyan",
+                "role": "Team Member",
+                "uid": mapped_uid,
+            }
+        }
+        return controller
+
+    def protobuf(self):
+        return types.SimpleNamespace(
+            latitude_i=389123456,
+            longitude_i=-91234567,
+            altitude=123,
+            ground_track=45,
+            ground_speed=6,
+        )
+
+    def test_position_uses_one_canonical_uid_with_and_without_mapping(self):
+        raw_uid = "a1b2c3d4"
+        mapped_uid = "ATAK-MAPPED-UID"
+        raw_eud = types.SimpleNamespace(uid=raw_uid)
+        mapped_eud = types.SimpleNamespace(uid=mapped_uid)
+        scenarios = (
+            ("raw", None, {}, {}, raw_uid),
+            (
+                "restart-mapped",
+                None,
+                {},
+                {int(raw_uid, 16): mapped_eud},
+                mapped_uid,
+            ),
+            ("mapped-existing", mapped_uid, {mapped_uid: mapped_eud}, {}, mapped_uid),
+            (
+                "raw-existing",
+                mapped_uid,
+                {},
+                {int(raw_uid, 16): raw_eud},
+                raw_uid,
+            ),
+            ("mapped-new", mapped_uid, {}, {}, mapped_uid),
+        )
+        for name, memory_uid, by_uid, by_mesh_id, canonical_uid in scenarios:
+            with self.subTest(name=name):
+                controller = self.make_controller(memory_uid)
+                eud_updates = []
+
+                def record_eud_update(uid, from_id, update_if_exists=True, last_event_time=None):
+                    eud_updates.append((uid, from_id, update_if_exists, last_event_time))
+
+                controller.insert_or_update_eud = record_eud_update
+                session = _Session(
+                    euds_by_uid=by_uid,
+                    euds_by_meshtastic_id=by_mesh_id,
+                )
+                socketio = _SocketIO()
+
+                with (
+                    patch.object(module, "db", types.SimpleNamespace(session=session)),
+                    patch.object(module, "socketio", socketio),
+                    patch.object(module, "CoT", _CoT, create=True),
+                    patch.object(module, "Point", _Point),
+                ):
+                    event = controller.position(self.protobuf(), raw_uid, "all", "POSITION_APP")
+
+                self.assertEqual(event.attrib["uid"], canonical_uid)
+                self.assertEqual(controller.meshtastic_devices[raw_uid]["uid"], canonical_uid)
+                self.assertEqual(eud_updates[0][:3], (canonical_uid, raw_uid, False))
+                self.assertIsInstance(eud_updates[0][3], datetime.datetime)
+                self.assertEqual(eud_updates[0][3].tzinfo, datetime.timezone.utc)
+
+                cot = next(value for value in session.added if isinstance(value, _CoT))
+                point = next(value for value in session.added if isinstance(value, _Point))
+                self.assertEqual(cot.sender_uid, canonical_uid)
+                self.assertEqual(point.device_uid, canonical_uid)
+                self.assertEqual(point.uid, canonical_uid)
+                self.assertEqual(point.cot_id, cot.id)
+                self.assertIs(point.cot, cot)
+                self.assertEqual(point.to_json()["type"], "a-f-G-U-C")
+                self.assertEqual(point.to_json()["how"], "m-g")
+                self.assertEqual(socketio.emitted[0][1]["type"], "a-f-G-U-C")
+
+    def test_position_rolls_back_failed_cot_point_transaction(self):
+        controller = self.make_controller("ATAK-MAPPED-UID")
+        controller.insert_or_update_eud = lambda *args, **kwargs: None
+        session = _Session(commit_error=RuntimeError("database write failed"))
+        socketio = _SocketIO()
+
+        with (
+            patch.object(module, "db", types.SimpleNamespace(session=session)),
+            patch.object(module, "socketio", socketio),
+            patch.object(module, "CoT", _CoT, create=True),
+            patch.object(module, "Point", _Point),
+        ):
+            event = controller.position(self.protobuf(), "a1b2c3d4", "all", "POSITION_APP")
+
+        self.assertIsNone(event)
+        self.assertTrue(session.rolled_back)
+        self.assertEqual(socketio.emitted, [])
+
+    def test_existing_eud_conflict_updates_only_activity_after_rollback(self):
+        controller = self.make_controller("ATAK-MAPPED-UID")
+        event_time = datetime.datetime(2026, 8, 26, 14, 0, tzinfo=datetime.timezone.utc)
+        eud = MagicMock()
+        eud.to_json.return_value = {}
+        eud_type = MagicMock(return_value=eud)
+        eud_type.uid = MagicMock()
+        session = MagicMock()
+        session.execute.return_value.scalar.return_value = None
+        session.commit.side_effect = [
+            module.sqlalchemy.exc.IntegrityError(
+                "duplicate EUD", params=None, orig=RuntimeError("duplicate")
+            ),
+            None,
+        ]
+        update = MagicMock()
+
+        with (
+            patch.object(module, "db", types.SimpleNamespace(session=session)),
+            patch.object(module, "socketio", MagicMock()),
+            patch.object(module, "EUD", eud_type),
+            patch.object(module, "Team", MagicMock()),
+            patch.object(module.sqlalchemy, "update", return_value=update),
+        ):
+            controller.insert_or_update_eud(
+                "ATAK-MAPPED-UID",
+                "a1b2c3d4",
+                update_if_exists=False,
+                last_event_time=event_time,
+            )
+
+        session.rollback.assert_called_once_with()
+        self.assertEqual(session.commit.call_count, 2)
+        update.where.return_value.values.assert_called_once_with(
+            last_event_time=event_time,
+            last_status="Connected",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()