#!/usr/bin/env python3

# Copyright (C) 2025 l5yth
#
# 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.

"""Mesh daemon helpers for synchronising Meshtastic data.

This module wraps the Meshtastic serial interface and exposes helper
functions that serialise nodes and text messages to JSON before forwarding
them to the accompanying web API.  It also provides the long-running daemon
entry point that performs these synchronisation tasks.
"""

import base64
import dataclasses
import glob
import heapq
import inspect
import ipaddress
import itertools
import json
import math
import os
import re
import signal
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Mapping
from functools import lru_cache
from typing import TYPE_CHECKING

from meshtastic.serial_interface import SerialInterface
from meshtastic.tcp_interface import TCPInterface
from pubsub import pub
from google.protobuf.json_format import MessageToDict
from google.protobuf.message import Message as ProtoMessage
from google.protobuf.message import DecodeError

if TYPE_CHECKING:  # pragma: no cover - import only used for type checking
    from meshtastic.ble_interface import BLEInterface as _BLEInterface

# Exposed for tests and backward compatibility; resolved lazily in
# :func:`_load_ble_interface` so importing this module does not require the BLE
# extras to be installed.
BLEInterface = None

# --- Config (env overrides) ---------------------------------------------------
PORT = os.environ.get("MESH_SERIAL")
SNAPSHOT_SECS = int(os.environ.get("MESH_SNAPSHOT_SECS", "60"))
CHANNEL_INDEX = int(os.environ.get("MESH_CHANNEL_INDEX", "0"))
DEBUG = os.environ.get("DEBUG") == "1"
INSTANCE = os.environ.get("POTATOMESH_INSTANCE", "").rstrip("/")
API_TOKEN = os.environ.get("API_TOKEN", "")


# --- Serial interface helpers --------------------------------------------------


_DEFAULT_TCP_PORT = 4403
_DEFAULT_TCP_TARGET = "http://127.0.0.1"

_DEFAULT_SERIAL_PATTERNS = (
    "/dev/ttyACM*",
    "/dev/ttyUSB*",
    "/dev/tty.usbmodem*",
    "/dev/tty.usbserial*",
    "/dev/cu.usbmodem*",
    "/dev/cu.usbserial*",
)

_BLE_ADDRESS_RE = re.compile(r"^(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}$")


def _debug_log(message: str):
    """Print ``message`` with a UTC timestamp when ``DEBUG`` is enabled."""

    if not DEBUG:
        return

    timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    print(f"[{timestamp}] [debug] {message}")


# Reconnect configuration: retry delays are adjustable via environment
# variables to ease testing while keeping sensible defaults in production.
_RECONNECT_INITIAL_DELAY_SECS = float(os.environ.get("MESH_RECONNECT_INITIAL", "5"))
_RECONNECT_MAX_DELAY_SECS = float(os.environ.get("MESH_RECONNECT_MAX", "60"))
_CLOSE_TIMEOUT_SECS = float(os.environ.get("MESH_CLOSE_TIMEOUT", "5"))


def _event_wait_allows_default_timeout() -> bool:
    """Return ``True`` when :func:`threading.Event.wait` accepts no timeout."""

    try:
        wait_signature = inspect.signature(threading.Event.wait)
    except (TypeError, ValueError):  # pragma: no cover - built-ins without inspect
        return True

    parameters = list(wait_signature.parameters.values())
    if len(parameters) <= 1:
        return True

    timeout_parameter = parameters[1]
    if timeout_parameter.kind in (
        inspect.Parameter.VAR_POSITIONAL,
        inspect.Parameter.VAR_KEYWORD,
    ):
        return True

    return timeout_parameter.default is not inspect._empty


class _DummySerialInterface:
    """In-memory replacement for ``meshtastic.serial_interface.SerialInterface``.

    The GitHub Actions release tests run the ingestor container without access
    to a serial device.  When ``MESH_SERIAL`` is set to ``"mock"`` (or similar)
    we provide this stub interface so the daemon can start and exercise its
    background loop without failing due to missing hardware.
    """

    def __init__(self):
        self.nodes = {}

    def close(self):
        """Mirror the real interface API."""
        pass


def _parse_ble_target(value: str) -> str | None:
    """Return an uppercase BLE MAC address when ``value`` matches the format."""

    if not value:
        return None
    value = value.strip()
    if not value:
        return None
    if _BLE_ADDRESS_RE.fullmatch(value):
        return value.upper()
    return None


def _parse_network_target(value: str) -> tuple[str, int] | None:
    """Return ``(host, port)`` when ``value`` is an IP address string.

    The ingestor accepts values such as ``192.168.1.10`` or
    ``tcp://192.168.1.10:4500`` for ``MESH_SERIAL`` to support Meshtastic
    devices shared via TCP.  Serial device paths (``/dev/ttyACM0``) are ignored
    by returning ``None``.
    """

    if not value:
        return None

    value = value.strip()
    if not value:
        return None

    def _validated_result(host: str | None, port: int | None):
        if not host:
            return None
        try:
            ipaddress.ip_address(host)
        except ValueError:
            return None
        return host, port or _DEFAULT_TCP_PORT

    parsed_values = []
    if "://" in value:
        parsed_values.append(urllib.parse.urlparse(value, scheme="tcp"))
    parsed_values.append(urllib.parse.urlparse(f"//{value}", scheme="tcp"))

    for parsed in parsed_values:
        try:
            port = parsed.port
        except ValueError:
            port = None
        result = _validated_result(parsed.hostname, port)
        if result:
            return result

    if value.count(":") == 1 and not value.startswith("["):
        host, _, port_text = value.partition(":")
        try:
            port = int(port_text) if port_text else None
        except ValueError:
            port = None
        result = _validated_result(host, port)
        if result:
            return result

    return _validated_result(value, None)


@lru_cache(maxsize=1)
def _load_ble_interface():
    """Return :class:`meshtastic.ble_interface.BLEInterface` when available."""

    global BLEInterface
    if BLEInterface is not None:
        return BLEInterface

    try:
        from meshtastic.ble_interface import BLEInterface as _resolved_interface
    except ImportError as exc:  # pragma: no cover - exercised in non-BLE envs
        raise RuntimeError(
            "BLE interface requested but the Meshtastic BLE dependencies are not installed. "
            "Install the 'meshtastic[ble]' extra to enable BLE support."
        ) from exc
    BLEInterface = _resolved_interface
    return _resolved_interface


def _create_serial_interface(port: str) -> tuple[object, str]:
    """Return an appropriate mesh interface for ``port``.

    Passing ``mock`` (case-insensitive) or an empty value skips hardware access
    and returns :class:`_DummySerialInterface`.  This makes it possible to run
    the container in CI environments that do not expose serial devices while
    keeping production behaviour unchanged.
    """

    port_value = (port or "").strip()
    if port_value.lower() in {"", "mock", "none", "null", "disabled"}:
        _debug_log(f"using dummy serial interface for port={port_value!r}")
        return _DummySerialInterface(), "mock"
    ble_target = _parse_ble_target(port_value)
    if ble_target:
        _debug_log(f"using BLE interface for address={ble_target}")
        return _load_ble_interface()(address=ble_target), ble_target
    network_target = _parse_network_target(port_value)
    if network_target:
        host, tcp_port = network_target
        _debug_log(f"using TCP interface for host={host!r} port={tcp_port!r}")
        return (
            TCPInterface(hostname=host, portNumber=tcp_port),
            f"tcp://{host}:{tcp_port}",
        )
    _debug_log(f"using serial interface for port={port_value!r}")
    return SerialInterface(devPath=port_value), port_value


class NoAvailableMeshInterface(RuntimeError):
    """Raised when no default mesh interface can be created."""


def _default_serial_targets() -> list[str]:
    """Return a list of candidate serial device paths for auto-discovery."""

    candidates: list[str] = []
    seen: set[str] = set()
    for pattern in _DEFAULT_SERIAL_PATTERNS:
        for path in sorted(glob.glob(pattern)):
            if path not in seen:
                candidates.append(path)
                seen.add(path)
    if "/dev/ttyACM0" not in seen:
        candidates.append("/dev/ttyACM0")
    return candidates


def _create_default_interface() -> tuple[object, str]:
    """Attempt to create the default mesh interface, raising on failure."""

    errors: list[tuple[str, Exception]] = []
    for candidate in _default_serial_targets():
        try:
            return _create_serial_interface(candidate)
        except Exception as exc:  # pragma: no cover - hardware dependent
            errors.append((candidate, exc))
            _debug_log(f"failed to open serial candidate {candidate!r}: {exc}")
    try:
        return _create_serial_interface(_DEFAULT_TCP_TARGET)
    except Exception as exc:  # pragma: no cover - network dependent
        errors.append((_DEFAULT_TCP_TARGET, exc))
        _debug_log(f"failed to open TCP fallback {_DEFAULT_TCP_TARGET!r}: {exc}")
    if errors:
        summary = "; ".join(f"{target}: {error}" for target, error in errors)
        raise NoAvailableMeshInterface(
            f"no mesh interface available ({summary})"
        ) from errors[-1][1]
    raise NoAvailableMeshInterface("no mesh interface available")


# --- POST queue ----------------------------------------------------------------
_POST_QUEUE_LOCK = threading.Lock()
_POST_QUEUE = []
_POST_QUEUE_COUNTER = itertools.count()
_POST_QUEUE_ACTIVE = False

_MESSAGE_POST_PRIORITY = 10
_NEIGHBOR_POST_PRIORITY = 20
_POSITION_POST_PRIORITY = 30
_TELEMETRY_POST_PRIORITY = 40
_NODE_POST_PRIORITY = 50
_DEFAULT_POST_PRIORITY = 90

_RECEIVE_TOPICS = (
    "meshtastic.receive",
    "meshtastic.receive.text",
    "meshtastic.receive.position",
    "meshtastic.receive.user",
    "meshtastic.receive.POSITION_APP",
    "meshtastic.receive.NODEINFO_APP",
    "meshtastic.receive.NEIGHBORINFO_APP",
    "meshtastic.receive.TEXT_MESSAGE_APP",
    "meshtastic.receive.TELEMETRY_APP",
)


def _get(obj, key, default=None):
    """Return a key or attribute value from ``obj``.

    Args:
        obj: Mapping or object containing the desired value.
        key: Key or attribute name to look up.
        default: Value returned when the key is missing.

    Returns:
        The resolved value if present, otherwise ``default``.
    """
    if isinstance(obj, dict):
        return obj.get(key, default)
    return getattr(obj, key, default)


# --- HTTP helpers -------------------------------------------------------------
def _post_json(path: str, payload: dict):
    """Send a JSON payload to the configured web API.

    Args:
        path: API path relative to the configured ``INSTANCE``.
        payload: Mapping serialised to JSON for the request body.
    """

    if not INSTANCE:
        return
    url = f"{INSTANCE}{path}"
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url, data=data, headers={"Content-Type": "application/json"}
    )
    if API_TOKEN:
        req.add_header("Authorization", f"Bearer {API_TOKEN}")
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()
    except Exception as e:
        _debug_log(f"[warn] POST {url} failed: {e}")


def _enqueue_post_json(path: str, payload: dict, priority: int):
    """Store a POST request in the priority queue."""

    with _POST_QUEUE_LOCK:
        heapq.heappush(
            _POST_QUEUE, (priority, next(_POST_QUEUE_COUNTER), path, payload)
        )


def _drain_post_queue():
    """Process queued POST requests in priority order."""

    global _POST_QUEUE_ACTIVE
    while True:
        with _POST_QUEUE_LOCK:
            if not _POST_QUEUE:
                _POST_QUEUE_ACTIVE = False
                return
            _priority, _idx, path, payload = heapq.heappop(_POST_QUEUE)
        _post_json(path, payload)


def _queue_post_json(
    path: str, payload: dict, *, priority: int = _DEFAULT_POST_PRIORITY
):
    """Queue a POST request and start processing if idle."""

    global _POST_QUEUE_ACTIVE
    _enqueue_post_json(path, payload, priority)
    with _POST_QUEUE_LOCK:
        if _POST_QUEUE_ACTIVE:
            return
        _POST_QUEUE_ACTIVE = True
    _drain_post_queue()


def _clear_post_queue():
    """Clear the pending POST queue (used by tests)."""

    global _POST_QUEUE_ACTIVE
    with _POST_QUEUE_LOCK:
        _POST_QUEUE.clear()
        _POST_QUEUE_ACTIVE = False


# --- Node upsert --------------------------------------------------------------
def _node_to_dict(n) -> dict:
    """Convert Meshtastic node or user structures into plain dictionaries.

    Args:
        n: ``dict``, dataclass or protobuf message describing a node or user.

    Returns:
        JSON serialisable representation of ``n``.
    """

    def _convert(value):
        """Recursively convert dataclasses and protobuf messages."""
        if isinstance(value, dict):
            return {k: _convert(v) for k, v in value.items()}
        if isinstance(value, (list, tuple, set)):
            return [_convert(v) for v in value]
        if dataclasses.is_dataclass(value):
            return {k: _convert(getattr(value, k)) for k in value.__dataclass_fields__}
        if isinstance(value, ProtoMessage):
            try:
                return MessageToDict(
                    value,
                    preserving_proto_field_name=True,
                    use_integers_for_enums=False,
                )
            except Exception:
                if hasattr(value, "to_dict"):
                    try:
                        return value.to_dict()
                    except Exception:
                        pass
                try:
                    return json.loads(json.dumps(value, default=str))
                except Exception:
                    return str(value)
        if isinstance(value, bytes):
            try:
                return value.decode()
            except Exception:
                return value.hex()
        if isinstance(value, (str, int, float, bool)) or value is None:
            return value
        try:
            return json.loads(json.dumps(value, default=str))
        except Exception:
            return str(value)

    return _convert(n)


def upsert_node(node_id, n):
    """Forward a node snapshot to the web API.

    Args:
        node_id: Unique identifier of the node in the mesh.
        n: Node object obtained from the Meshtastic serial interface.
    """

    ndict = _node_to_dict(n)
    _queue_post_json("/api/nodes", {node_id: ndict}, priority=_NODE_POST_PRIORITY)

    if DEBUG:
        user = _get(ndict, "user") or {}
        short = _get(user, "shortName")
        long = _get(user, "longName")
        _debug_log(f"upserted node {node_id} shortName={short!r} longName={long!r}")


# --- Message logging via PubSub -----------------------------------------------
def _iso(ts: int | float) -> str:
    """Return an ISO-8601 timestamp string for ``ts``.

    Args:
        ts: POSIX timestamp as ``int`` or ``float``.

    Returns:
        Timestamp formatted with a trailing ``Z`` to denote UTC.
    """

    import datetime

    return (
        datetime.datetime.fromtimestamp(int(ts), datetime.UTC)
        .isoformat()
        .replace("+00:00", "Z")
    )


def _first(d, *names, default=None):
    """Return the first non-empty key from ``names`` (supports nested lookups).

    Keys that resolve to ``None`` or an empty string are skipped so callers can
    provide multiple potential field names without accidentally capturing an
    explicit ``null`` value.

    Args:
        d: Mapping or object to query.
        *names: Candidate field names using dotted paths for nesting.
        default: Value returned when all candidates are missing.

    Returns:
        The first matching value or ``default`` if none resolve to content.
    """

    def _mapping_get(obj, key):
        if isinstance(obj, Mapping) and key in obj:
            return True, obj[key]
        if hasattr(obj, "__getitem__"):
            try:
                return True, obj[key]
            except Exception:
                pass
        if hasattr(obj, key):
            return True, getattr(obj, key)
        return False, None

    for name in names:
        cur = d
        ok = True
        for part in name.split("."):
            ok, cur = _mapping_get(cur, part)
            if not ok:
                break
        if ok:
            if cur is None:
                continue
            if isinstance(cur, str) and cur == "":
                continue
            return cur
    return default


def _coerce_int(value):
    """Return ``value`` converted to ``int`` when possible."""

    if value is None:
        return None
    if isinstance(value, bool):
        return int(value)
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value) if math.isfinite(value) else None
    if isinstance(value, (str, bytes, bytearray)):
        text = value.decode() if isinstance(value, (bytes, bytearray)) else value
        stripped = text.strip()
        if not stripped:
            return None
        try:
            if stripped.lower().startswith("0x"):
                return int(stripped, 16)
            return int(stripped, 10)
        except ValueError:
            try:
                return int(float(stripped))
            except ValueError:
                return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _coerce_float(value):
    """Return ``value`` converted to ``float`` when possible."""

    if value is None:
        return None
    if isinstance(value, bool):
        return float(value)
    if isinstance(value, (int, float)):
        result = float(value)
        return result if math.isfinite(result) else None
    if isinstance(value, (str, bytes, bytearray)):
        text = value.decode() if isinstance(value, (bytes, bytearray)) else value
        stripped = text.strip()
        if not stripped:
            return None
        try:
            result = float(stripped)
        except ValueError:
            return None
        return result if math.isfinite(result) else None
    try:
        result = float(value)
    except (TypeError, ValueError):
        return None
    return result if math.isfinite(result) else None


def _pkt_to_dict(packet) -> dict:
    """Normalise a received packet into a JSON-friendly dictionary.

    Args:
        packet: Protobuf ``MeshPacket`` or dictionary received from the daemon.

    Returns:
        Packet data ready for JSON serialisation.
    """
    if isinstance(packet, dict):
        return packet
    if isinstance(packet, ProtoMessage):
        try:
            return MessageToDict(
                packet, preserving_proto_field_name=True, use_integers_for_enums=False
            )
        except Exception:
            if hasattr(packet, "to_dict"):
                try:
                    return packet.to_dict()
                except Exception:
                    pass
    # Last resort: try to read attributes
    try:
        return json.loads(json.dumps(packet, default=lambda o: str(o)))
    except Exception:
        return {"_unparsed": str(packet)}


def _canonical_node_id(value) -> str | None:
    """Normalise node identifiers to the canonical ``!deadbeef`` form."""

    if value is None:
        return None
    if isinstance(value, (int, float)):
        try:
            num = int(value)
        except (TypeError, ValueError):
            return None
        if num < 0:
            return None
        return f"!{num & 0xFFFFFFFF:08x}"
    if not isinstance(value, str):
        return None

    trimmed = value.strip()
    if not trimmed:
        return None
    if trimmed.startswith("^"):
        return trimmed
    if trimmed.startswith("!"):
        body = trimmed[1:]
    elif trimmed.lower().startswith("0x"):
        body = trimmed[2:]
    elif trimmed.isdigit():
        try:
            return f"!{int(trimmed, 10) & 0xFFFFFFFF:08x}"
        except ValueError:
            return None
    else:
        body = trimmed

    if not body:
        return None
    try:
        return f"!{int(body, 16) & 0xFFFFFFFF:08x}"
    except ValueError:
        return None


def _node_num_from_id(node_id) -> int | None:
    """Return the numeric node reference derived from ``node_id``."""

    if node_id is None:
        return None
    if isinstance(node_id, (int, float)):
        try:
            num = int(node_id)
        except (TypeError, ValueError):
            return None
        return num if num >= 0 else None
    if not isinstance(node_id, str):
        return None

    trimmed = node_id.strip()
    if not trimmed:
        return None
    if trimmed.startswith("!"):
        trimmed = trimmed[1:]
    if trimmed.lower().startswith("0x"):
        trimmed = trimmed[2:]
    try:
        return int(trimmed, 16)
    except ValueError:
        try:
            return int(trimmed, 10)
        except ValueError:
            return None


def _merge_mappings(base, extra):
    """Recursively merge mapping ``extra`` into ``base`` without mutation."""

    base_dict: dict
    if isinstance(base, Mapping):
        base_dict = dict(base)
    elif base:
        converted_base = _node_to_dict(base)
        base_dict = dict(converted_base) if isinstance(converted_base, Mapping) else {}
    else:
        base_dict = {}

    if not isinstance(extra, Mapping):
        converted_extra = _node_to_dict(extra)
        if not isinstance(converted_extra, Mapping):
            return base_dict
        extra = converted_extra

    for key, value in extra.items():
        if isinstance(value, Mapping):
            existing = base_dict.get(key)
            base_dict[key] = _merge_mappings(existing, value)
        else:
            base_dict[key] = _node_to_dict(value)
    return base_dict


def _extract_payload_bytes(decoded_section: Mapping) -> bytes | None:
    """Extract raw payload bytes from a decoded packet section."""

    if not isinstance(decoded_section, Mapping):
        return None
    payload = decoded_section.get("payload")
    if isinstance(payload, Mapping):
        data = payload.get("__bytes_b64__") or payload.get("bytes")
        if isinstance(data, str):
            try:
                return base64.b64decode(data)
            except Exception:
                return None
    if isinstance(payload, (bytes, bytearray)):
        return bytes(payload)
    if isinstance(payload, str):
        try:
            return base64.b64decode(payload)
        except Exception:
            return None
    return None


def _decode_nodeinfo_payload(payload_bytes):
    """Return a ``NodeInfo`` protobuf message parsed from ``payload_bytes``."""

    if not payload_bytes:
        return None
    try:
        from meshtastic.protobuf import mesh_pb2
    except Exception:
        return None

    node_info = mesh_pb2.NodeInfo()
    try:
        node_info.ParseFromString(payload_bytes)
        return node_info
    except DecodeError:
        try:
            user_msg = mesh_pb2.User()
            user_msg.ParseFromString(payload_bytes)
        except DecodeError:
            return None
        node_info = mesh_pb2.NodeInfo()
        node_info.user.CopyFrom(user_msg)
        return node_info


def _nodeinfo_metrics_dict(node_info) -> dict | None:
    """Convert ``NodeInfo.device_metrics`` into a JSON-friendly mapping."""

    if not node_info:
        return None
    metrics_field_names = {f[0].name for f in node_info.ListFields()}
    if "device_metrics" not in metrics_field_names:
        return None
    metrics = {}
    for field_desc, value in node_info.device_metrics.ListFields():
        name = field_desc.name
        if name == "battery_level":
            metrics["batteryLevel"] = float(value)
        elif name == "voltage":
            metrics["voltage"] = float(value)
        elif name == "channel_utilization":
            metrics["channelUtilization"] = float(value)
        elif name == "air_util_tx":
            metrics["airUtilTx"] = float(value)
        elif name == "uptime_seconds":
            metrics["uptimeSeconds"] = int(value)
    return metrics if metrics else None


def _nodeinfo_position_dict(node_info) -> dict | None:
    """Convert ``NodeInfo.position`` into a dictionary with decoded coordinates."""

    if not node_info:
        return None
    field_names = {f[0].name for f in node_info.ListFields()}
    if "position" not in field_names:
        return None
    position = {}
    for field_desc, value in node_info.position.ListFields():
        name = field_desc.name
        if name == "latitude_i":
            position["latitude"] = float(value) / 1e7
        elif name == "longitude_i":
            position["longitude"] = float(value) / 1e7
        elif name == "altitude":
            position["altitude"] = float(value)
        elif name == "time":
            position["time"] = int(value)
        elif name == "location_source":
            try:
                from meshtastic.protobuf import mesh_pb2

                position["locationSource"] = mesh_pb2.Position.LocSource.Name(value)
            except Exception:
                position["locationSource"] = value
    return position if position else None


def _nodeinfo_user_dict(node_info, decoded_user) -> dict | None:
    """Merge user details from the decoded packet and NodeInfo payload."""

    user_dict = None
    if node_info:
        field_names = {f[0].name for f in node_info.ListFields()}
        if "user" in field_names:
            try:
                from google.protobuf.json_format import MessageToDict

                user_dict = MessageToDict(
                    node_info.user,
                    preserving_proto_field_name=False,
                    use_integers_for_enums=False,
                )
            except Exception:
                user_dict = None

    if isinstance(decoded_user, ProtoMessage):
        try:
            from google.protobuf.json_format import MessageToDict

            decoded_user = MessageToDict(
                decoded_user,
                preserving_proto_field_name=False,
                use_integers_for_enums=False,
            )
        except Exception:
            decoded_user = _node_to_dict(decoded_user)

    if isinstance(decoded_user, Mapping):
        user_dict = _merge_mappings(user_dict, decoded_user)

    if isinstance(user_dict, Mapping):
        canonical = _canonical_node_id(user_dict.get("id"))
        if canonical:
            user_dict = dict(user_dict)
            user_dict["id"] = canonical
    return user_dict


def store_position_packet(packet: dict, decoded: Mapping):
    """Handle ``POSITION_APP`` packets and forward them to ``/api/positions``."""

    node_ref = _first(packet, "fromId", "from_id", "from", default=None)
    if node_ref is None:
        node_ref = _first(decoded, "num", default=None)
    node_id = _canonical_node_id(node_ref)
    if node_id is None:
        return

    node_num = _coerce_int(_first(decoded, "num", default=None))
    if node_num is None:
        node_num = _node_num_from_id(node_id)

    pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None))
    if pkt_id is None:
        return

    rx_time = _coerce_int(_first(packet, "rxTime", "rx_time", default=time.time()))
    if rx_time is None:
        rx_time = int(time.time())

    to_id = _first(packet, "toId", "to_id", "to", default=None)
    to_id = to_id if to_id not in {"", None} else None

    position_section = decoded.get("position") if isinstance(decoded, Mapping) else None
    if not isinstance(position_section, Mapping):
        position_section = {}

    latitude = _coerce_float(
        _first(position_section, "latitude", "raw.latitude", default=None)
    )
    if latitude is None:
        lat_i = _coerce_int(
            _first(
                position_section,
                "latitudeI",
                "latitude_i",
                "raw.latitude_i",
                default=None,
            )
        )
        if lat_i is not None:
            latitude = lat_i / 1e7

    longitude = _coerce_float(
        _first(position_section, "longitude", "raw.longitude", default=None)
    )
    if longitude is None:
        lon_i = _coerce_int(
            _first(
                position_section,
                "longitudeI",
                "longitude_i",
                "raw.longitude_i",
                default=None,
            )
        )
        if lon_i is not None:
            longitude = lon_i / 1e7

    altitude = _coerce_float(
        _first(position_section, "altitude", "raw.altitude", default=None)
    )
    position_time = _coerce_int(
        _first(position_section, "time", "raw.time", default=None)
    )
    location_source = _first(
        position_section,
        "locationSource",
        "location_source",
        "raw.location_source",
        default=None,
    )
    location_source = (
        str(location_source).strip() if location_source not in {None, ""} else None
    )

    precision_bits = _coerce_int(
        _first(
            position_section,
            "precisionBits",
            "precision_bits",
            "raw.precision_bits",
            default=None,
        )
    )
    sats_in_view = _coerce_int(
        _first(
            position_section,
            "satsInView",
            "sats_in_view",
            "raw.sats_in_view",
            default=None,
        )
    )
    pdop = _coerce_float(
        _first(position_section, "PDOP", "pdop", "raw.PDOP", "raw.pdop", default=None)
    )
    ground_speed = _coerce_float(
        _first(
            position_section,
            "groundSpeed",
            "ground_speed",
            "raw.ground_speed",
            default=None,
        )
    )
    ground_track = _coerce_float(
        _first(
            position_section,
            "groundTrack",
            "ground_track",
            "raw.ground_track",
            default=None,
        )
    )

    snr = _coerce_float(_first(packet, "snr", "rx_snr", "rxSnr", default=None))
    rssi = _coerce_int(_first(packet, "rssi", "rx_rssi", "rxRssi", default=None))
    hop_limit = _coerce_int(_first(packet, "hopLimit", "hop_limit", default=None))
    bitfield = _coerce_int(_first(decoded, "bitfield", default=None))

    payload_bytes = _extract_payload_bytes(decoded)
    payload_b64 = (
        base64.b64encode(payload_bytes).decode("ascii") if payload_bytes else None
    )

    raw_section = decoded.get("raw") if isinstance(decoded, Mapping) else None
    raw_payload = _node_to_dict(raw_section) if raw_section else None
    if raw_payload is None and position_section:
        raw_position = (
            position_section.get("raw")
            if isinstance(position_section, Mapping)
            else None
        )
        if raw_position:
            raw_payload = _node_to_dict(raw_position)

    position_payload = {
        "id": pkt_id,
        "node_id": node_id or raw_from,
        "node_num": node_num,
        "num": node_num,
        "from_id": node_id,
        "to_id": to_id,
        "rx_time": rx_time,
        "rx_iso": _iso(rx_time),
        "latitude": latitude,
        "longitude": longitude,
        "altitude": altitude,
        "position_time": position_time,
        "location_source": location_source,
        "precision_bits": precision_bits,
        "sats_in_view": sats_in_view,
        "pdop": pdop,
        "ground_speed": ground_speed,
        "ground_track": ground_track,
        "snr": snr,
        "rssi": rssi,
        "hop_limit": hop_limit,
        "bitfield": bitfield,
        "payload_b64": payload_b64,
    }
    if raw_payload:
        position_payload["raw"] = raw_payload

    _queue_post_json(
        "/api/positions", position_payload, priority=_POSITION_POST_PRIORITY
    )

    if DEBUG:
        _debug_log(f"stored position for {node_id} lat={latitude!r} lon={longitude!r}")


def store_telemetry_packet(packet: dict, decoded: Mapping):
    """Handle ``TELEMETRY_APP`` packets and forward them to ``/api/telemetry``."""

    telemetry_section = (
        decoded.get("telemetry") if isinstance(decoded, Mapping) else None
    )
    if not isinstance(telemetry_section, Mapping):
        return

    pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None))
    if pkt_id is None:
        return

    raw_from = _first(packet, "fromId", "from_id", "from", default=None)
    node_id = _canonical_node_id(raw_from)
    node_num = _coerce_int(_first(decoded, "num", "node_num", default=None))
    if node_num is None:
        node_num = _node_num_from_id(node_id or raw_from)

    to_id = _first(packet, "toId", "to_id", "to", default=None)

    raw_rx_time = _first(packet, "rxTime", "rx_time", default=time.time())
    try:
        rx_time = int(raw_rx_time)
    except (TypeError, ValueError):
        rx_time = int(time.time())
    rx_iso = _iso(rx_time)

    telemetry_time = _coerce_int(_first(telemetry_section, "time", default=None))

    channel = _coerce_int(_first(decoded, "channel", default=None))
    if channel is None:
        channel = _coerce_int(_first(packet, "channel", default=None))
    if channel is None:
        channel = 0

    bitfield = _coerce_int(_first(decoded, "bitfield", default=None))
    snr = _coerce_float(_first(packet, "snr", "rx_snr", "rxSnr", default=None))
    rssi = _coerce_int(_first(packet, "rssi", "rx_rssi", "rxRssi", default=None))
    hop_limit = _coerce_int(_first(packet, "hopLimit", "hop_limit", default=None))

    portnum_raw = _first(decoded, "portnum", default=None)
    portnum = str(portnum_raw) if portnum_raw is not None else None

    payload_bytes = _extract_payload_bytes(decoded)
    payload_b64 = (
        base64.b64encode(payload_bytes).decode("ascii") if payload_bytes else None
    )

    device_metrics_section = telemetry_section.get(
        "deviceMetrics"
    ) or telemetry_section.get("device_metrics")
    device_metrics = (
        _node_to_dict(device_metrics_section)
        if isinstance(device_metrics_section, Mapping)
        else None
    )

    environment_section = telemetry_section.get(
        "environmentMetrics"
    ) or telemetry_section.get("environment_metrics")
    environment_metrics = (
        _node_to_dict(environment_section)
        if isinstance(environment_section, Mapping)
        else None
    )

    metrics_lookup = lambda mapping, *names: _first(mapping or {}, *names, default=None)

    battery_level = _coerce_float(
        metrics_lookup(device_metrics, "batteryLevel", "battery_level")
    )
    voltage = _coerce_float(metrics_lookup(device_metrics, "voltage"))
    channel_utilization = _coerce_float(
        metrics_lookup(device_metrics, "channelUtilization", "channel_utilization")
    )
    air_util_tx = _coerce_float(
        metrics_lookup(device_metrics, "airUtilTx", "air_util_tx")
    )
    uptime_seconds = _coerce_int(
        metrics_lookup(device_metrics, "uptimeSeconds", "uptime_seconds")
    )

    temperature = _coerce_float(
        metrics_lookup(
            environment_metrics,
            "temperature",
            "temperatureC",
            "temperature_c",
            "tempC",
        )
    )
    relative_humidity = _coerce_float(
        metrics_lookup(
            environment_metrics,
            "relativeHumidity",
            "relative_humidity",
            "humidity",
        )
    )
    barometric_pressure = _coerce_float(
        metrics_lookup(
            environment_metrics,
            "barometricPressure",
            "barometric_pressure",
            "pressure",
        )
    )

    telemetry_payload = {
        "id": pkt_id,
        "node_id": node_id,
        "node_num": node_num,
        "from_id": node_id or raw_from,
        "to_id": to_id,
        "rx_time": rx_time,
        "rx_iso": rx_iso,
        "telemetry_time": telemetry_time,
        "channel": channel,
        "portnum": portnum,
        "bitfield": bitfield,
        "snr": snr,
        "rssi": rssi,
        "hop_limit": hop_limit,
        "payload_b64": payload_b64,
    }

    if battery_level is not None:
        telemetry_payload["battery_level"] = battery_level
    if voltage is not None:
        telemetry_payload["voltage"] = voltage
    if channel_utilization is not None:
        telemetry_payload["channel_utilization"] = channel_utilization
    if air_util_tx is not None:
        telemetry_payload["air_util_tx"] = air_util_tx
    if uptime_seconds is not None:
        telemetry_payload["uptime_seconds"] = uptime_seconds
    if temperature is not None:
        telemetry_payload["temperature"] = temperature
    if relative_humidity is not None:
        telemetry_payload["relative_humidity"] = relative_humidity
    if barometric_pressure is not None:
        telemetry_payload["barometric_pressure"] = barometric_pressure

    _queue_post_json(
        "/api/telemetry", telemetry_payload, priority=_TELEMETRY_POST_PRIORITY
    )

    if DEBUG:
        _debug_log(
            f"stored telemetry for {node_id!r} battery={battery_level!r} voltage={voltage!r}"
        )


def store_nodeinfo_packet(packet: dict, decoded: Mapping):
    """Handle ``NODEINFO_APP`` packets and forward them to ``/api/nodes``."""

    payload_bytes = _extract_payload_bytes(decoded)
    node_info = _decode_nodeinfo_payload(payload_bytes)
    decoded_user = decoded.get("user")
    user_dict = _nodeinfo_user_dict(node_info, decoded_user)

    node_info_fields = set()
    if node_info:
        node_info_fields = {field_desc.name for field_desc, _ in node_info.ListFields()}

    node_id = None
    if isinstance(user_dict, Mapping):
        node_id = _canonical_node_id(user_dict.get("id"))

    if node_id is None:
        node_id = _canonical_node_id(
            _first(packet, "fromId", "from_id", "from", default=None)
        )

    if node_id is None:
        return

    node_payload = {}
    if user_dict:
        node_payload["user"] = user_dict

    node_num = None
    if node_info and "num" in node_info_fields:
        try:
            node_num = int(node_info.num)
        except (TypeError, ValueError):
            node_num = None
    if node_num is None:
        decoded_num = decoded.get("num")
        if decoded_num is not None:
            try:
                node_num = int(decoded_num)
            except (TypeError, ValueError):
                try:
                    node_num = int(str(decoded_num).strip(), 0)
                except Exception:
                    node_num = None
    if node_num is None:
        node_num = _node_num_from_id(node_id)
    if node_num is not None:
        node_payload["num"] = node_num

    rx_time = int(_first(packet, "rxTime", "rx_time", default=time.time()))
    last_heard = None
    if node_info and "last_heard" in node_info_fields:
        try:
            last_heard = int(node_info.last_heard)
        except (TypeError, ValueError):
            last_heard = None
    if last_heard is None:
        decoded_last_heard = decoded.get("lastHeard")
        if decoded_last_heard is not None:
            try:
                last_heard = int(decoded_last_heard)
            except (TypeError, ValueError):
                last_heard = None
    if last_heard is None or last_heard < rx_time:
        last_heard = rx_time
    node_payload["lastHeard"] = last_heard

    snr = None
    if node_info and "snr" in node_info_fields:
        try:
            snr = float(node_info.snr)
        except (TypeError, ValueError):
            snr = None
    if snr is None:
        snr = _first(packet, "snr", "rx_snr", "rxSnr", default=None)
        if snr is not None:
            try:
                snr = float(snr)
            except (TypeError, ValueError):
                snr = None
    if snr is not None:
        node_payload["snr"] = snr

    hops = None
    if node_info and "hops_away" in node_info_fields:
        try:
            hops = int(node_info.hops_away)
        except (TypeError, ValueError):
            hops = None
    if hops is None:
        hops = decoded.get("hopsAway")
        if hops is not None:
            try:
                hops = int(hops)
            except (TypeError, ValueError):
                hops = None
    if hops is not None:
        node_payload["hopsAway"] = hops

    if node_info and "channel" in node_info_fields:
        try:
            node_payload["channel"] = int(node_info.channel)
        except (TypeError, ValueError):
            pass

    if node_info and "via_mqtt" in node_info_fields:
        node_payload["viaMqtt"] = bool(node_info.via_mqtt)

    if node_info and "is_favorite" in node_info_fields:
        node_payload["isFavorite"] = bool(node_info.is_favorite)
    elif "isFavorite" in decoded:
        node_payload["isFavorite"] = bool(decoded.get("isFavorite"))

    if node_info and "is_ignored" in node_info_fields:
        node_payload["isIgnored"] = bool(node_info.is_ignored)
    if node_info and "is_key_manually_verified" in node_info_fields:
        node_payload["isKeyManuallyVerified"] = bool(node_info.is_key_manually_verified)

    metrics = _nodeinfo_metrics_dict(node_info)
    decoded_metrics = decoded.get("deviceMetrics")
    if isinstance(decoded_metrics, Mapping):
        metrics = _merge_mappings(metrics, _node_to_dict(decoded_metrics))
    if metrics:
        node_payload["deviceMetrics"] = metrics

    position = _nodeinfo_position_dict(node_info)
    decoded_position = decoded.get("position")
    if isinstance(decoded_position, Mapping):
        position = _merge_mappings(position, _node_to_dict(decoded_position))
    if position:
        node_payload["position"] = position

    hop_limit = _first(packet, "hopLimit", "hop_limit", default=None)
    if hop_limit is not None and "hopLimit" not in node_payload:
        try:
            node_payload["hopLimit"] = int(hop_limit)
        except (TypeError, ValueError):
            pass

    _queue_post_json(
        "/api/nodes", {node_id: node_payload}, priority=_NODE_POST_PRIORITY
    )

    if DEBUG:
        short = None
        if isinstance(user_dict, Mapping):
            short = user_dict.get("shortName")
            long = user_dict.get("longName")
        else:
            long = None
        _debug_log(
            f"stored nodeinfo for {node_id} shortName={short!r} longName={long!r}"
        )


def store_neighborinfo_packet(packet: dict, decoded: Mapping):
    """Handle ``NEIGHBORINFO_APP`` packets for mesh health broadcasts."""

    neighbor_section = (
        decoded.get("neighborinfo") if isinstance(decoded, Mapping) else None
    )
    if not isinstance(neighbor_section, Mapping):
        return

    node_ref = _first(
        neighbor_section,
        "nodeId",
        "node_id",
        default=_first(packet, "fromId", "from_id", "from", default=None),
    )
    node_id = _canonical_node_id(node_ref)
    if node_id is None:
        return

    node_num = _coerce_int(_first(neighbor_section, "nodeId", "node_id", default=None))
    if node_num is None:
        node_num = _node_num_from_id(node_id)

    rx_time = _coerce_int(_first(packet, "rxTime", "rx_time", default=time.time()))
    if rx_time is None:
        rx_time = int(time.time())

    neighbors_payload = neighbor_section.get("neighbors")
    neighbors_iterable = (
        neighbors_payload if isinstance(neighbors_payload, list) else []
    )

    neighbor_entries: list[dict] = []
    for entry in neighbors_iterable:
        if not isinstance(entry, Mapping):
            continue
        neighbor_ref = _first(entry, "nodeId", "node_id", default=None)
        neighbor_id = _canonical_node_id(neighbor_ref)
        if neighbor_id is None:
            continue
        neighbor_num = _coerce_int(_first(entry, "nodeId", "node_id", default=None))
        if neighbor_num is None:
            neighbor_num = _node_num_from_id(neighbor_id)
        snr = _coerce_float(_first(entry, "snr", default=None))
        entry_rx_time = _coerce_int(_first(entry, "rxTime", "rx_time", default=None))
        if entry_rx_time is None:
            entry_rx_time = rx_time
        neighbor_entries.append(
            {
                "neighbor_id": neighbor_id,
                "neighbor_num": neighbor_num,
                "snr": snr,
                "rx_time": entry_rx_time,
            }
        )

    payload: dict[str, object] = {
        "node_id": node_id,
        "node_num": node_num,
        "rx_time": rx_time,
    }
    if neighbor_entries:
        payload["neighbors"] = neighbor_entries

    broadcast_interval = _coerce_int(
        _first(neighbor_section, "nodeBroadcastIntervalSecs", default=None)
    )
    if broadcast_interval is not None:
        payload["node_broadcast_interval_secs"] = broadcast_interval

    last_sent_by = _canonical_node_id(
        _first(neighbor_section, "lastSentById", "last_sent_by_id", default=None)
    )
    if last_sent_by is not None:
        payload["last_sent_by_id"] = last_sent_by

    _queue_post_json("/api/neighbors", payload, priority=_NEIGHBOR_POST_PRIORITY)

    if DEBUG:
        _debug_log(
            f"stored neighborinfo for {node_id} neighbors={len(neighbor_entries)}"
        )


def store_packet_dict(p: dict):
    """Persist packets extracted from a decoded payload.

    Node information packets are forwarded to the ``/api/nodes`` endpoint
    while text messages from the ``TEXT_MESSAGE_APP`` port continue to be
    stored via ``/api/messages``. Field lookups tolerate camelCase and
    snake_case variants for compatibility across Meshtastic releases.

    Args:
        p: Packet dictionary produced by ``_pkt_to_dict``.
    """
    dec = p.get("decoded") or {}

    portnum_raw = _first(dec, "portnum", default=None)
    portnum = str(portnum_raw).upper() if portnum_raw is not None else None
    portnum_int = _coerce_int(portnum_raw)

    telemetry_section = dec.get("telemetry") if isinstance(dec, Mapping) else None
    if (
        portnum == "TELEMETRY_APP"
        or portnum_int == 65
        or isinstance(telemetry_section, Mapping)
    ):
        store_telemetry_packet(p, dec)
        return

    if portnum in {"5", "NODEINFO_APP"}:
        store_nodeinfo_packet(p, dec)
        return

    if portnum in {"4", "POSITION_APP"}:
        store_position_packet(p, dec)
        return

    neighborinfo_section = dec.get("neighborinfo") if isinstance(dec, Mapping) else None
    if portnum == "NEIGHBORINFO_APP" or isinstance(neighborinfo_section, Mapping):
        store_neighborinfo_packet(p, dec)
        return

    text = _first(dec, "payload.text", "text", default=None)
    encrypted = _first(dec, "payload.encrypted", "encrypted", default=None)
    if encrypted is None:
        encrypted = _first(p, "encrypted", default=None)
    if not text and not encrypted:
        return  # ignore packets that lack text and encrypted payloads

    # port filter: only keep packets from the TEXT_MESSAGE_APP port
    if portnum and portnum not in {"1", "TEXT_MESSAGE_APP"}:
        return  # ignore non-text-message ports

    # channel (prefer decoded.channel if present; else top-level)
    ch = _first(dec, "channel", default=None)
    if ch is None:
        ch = _first(p, "channel", default=0)
    try:
        ch = int(ch)
    except Exception:
        ch = 0

    # timestamps & ids
    pkt_id = _first(p, "id", "packet_id", "packetId", default=None)
    if pkt_id is None:
        return  # ignore packets without an id
    rx_time = int(_first(p, "rxTime", "rx_time", default=time.time()))
    from_id = _first(p, "fromId", "from_id", "from", default=None)
    to_id = _first(p, "toId", "to_id", "to", default=None)

    if (from_id is None or str(from_id) == "") and DEBUG:
        try:
            raw = json.dumps(p, default=str)
        except Exception:
            raw = str(p)
        _debug_log(f"packet missing from_id: {raw}")

    # link metrics
    snr = _first(p, "snr", "rx_snr", "rxSnr", default=None)
    rssi = _first(p, "rssi", "rx_rssi", "rxRssi", default=None)
    hop = _first(p, "hopLimit", "hop_limit", default=None)

    msg = {
        "id": int(pkt_id),
        "rx_time": rx_time,
        "rx_iso": _iso(rx_time),
        "from_id": from_id,
        "to_id": to_id,
        "channel": ch,
        "portnum": str(portnum) if portnum is not None else None,
        "text": text,
        "encrypted": encrypted,
        "snr": float(snr) if snr is not None else None,
        "rssi": int(rssi) if rssi is not None else None,
        "hop_limit": int(hop) if hop is not None else None,
    }
    _queue_post_json("/api/messages", msg, priority=_MESSAGE_POST_PRIORITY)

    if DEBUG:
        from_label = _canonical_node_id(from_id) or from_id
        to_label = _canonical_node_id(to_id) or to_id
        if text is None and encrypted:
            payload = "Encrypted"
        else:
            payload = text
        _debug_log(
            f"stored message from {from_label!r} to {to_label!r} ch={ch} text={payload!r}"
        )


# PubSub receive handler
def on_receive(packet, interface):
    """PubSub callback that stores inbound text messages.

    Args:
        packet: Packet received from the Meshtastic interface.
        interface: Serial interface instance (unused).
    """

    if isinstance(packet, dict):
        if packet.get("_potatomesh_seen"):
            return
        packet["_potatomesh_seen"] = True

    p = None
    try:
        p = _pkt_to_dict(packet)
        store_packet_dict(p)
    except Exception as e:
        info = list(p.keys()) if isinstance(p, dict) else type(packet)
        print(f"[warn] failed to store packet: {e} | info: {info}")


def _subscribe_receive_topics() -> list[str]:
    """Subscribe ``on_receive`` to relevant PubSub topics."""

    subscribed = []
    for topic in _RECEIVE_TOPICS:
        try:
            pub.subscribe(on_receive, topic)
            subscribed.append(topic)
        except Exception as exc:  # pragma: no cover - pub may raise in prod only
            _debug_log(f"failed to subscribe to {topic!r}: {exc}")
    return subscribed


# --- Main ---------------------------------------------------------------------
def _node_items_snapshot(nodes_obj, retries: int = 3):
    """Return a snapshot list of ``(node_id, node)`` pairs.

    The Meshtastic ``SerialInterface`` updates ``iface.nodes`` from another
    thread. When that happens during iteration Python raises ``RuntimeError``.
    To keep the daemon quiet we retry a few times and, if it keeps changing,
    bail out for this loop.

    Args:
        nodes_obj: Container mapping node IDs to node objects.
        retries: Number of attempts performed before giving up.

    Returns:
        Snapshot of node entries or ``None`` when retries were exhausted because
        the container kept mutating.
    """

    if not nodes_obj:
        return []

    items_callable = getattr(nodes_obj, "items", None)
    if callable(items_callable):
        for _ in range(max(1, retries)):
            try:
                return list(items_callable())
            except RuntimeError as err:
                if "dictionary changed size during iteration" not in str(err):
                    raise
                time.sleep(0)
        return None

    if hasattr(nodes_obj, "__iter__") and hasattr(nodes_obj, "__getitem__"):
        for _ in range(max(1, retries)):
            try:
                keys = list(nodes_obj)
                return [(k, nodes_obj[k]) for k in keys]
            except RuntimeError as err:
                if "dictionary changed size during iteration" not in str(err):
                    raise
                time.sleep(0)
        return None

    return []


def main():
    """Run the mesh synchronisation daemon."""

    # Subscribe to PubSub topics (reliable in current meshtastic)
    subscribed = _subscribe_receive_topics()
    if DEBUG and subscribed:
        _debug_log(f"subscribed to receive topics: {', '.join(subscribed)}")

    def _close_interface(iface_obj):
        if iface_obj is None:
            return

        def _do_close():
            try:
                iface_obj.close()
            except Exception as exc:
                if DEBUG:
                    _debug_log(f"error while closing mesh interface: {exc}")

        if _CLOSE_TIMEOUT_SECS <= 0 or not _event_wait_allows_default_timeout():
            _do_close()
            return

        close_thread = threading.Thread(
            target=_do_close, name="mesh-close", daemon=True
        )
        close_thread.start()
        close_thread.join(_CLOSE_TIMEOUT_SECS)
        if close_thread.is_alive():
            print(
                "[warn] mesh interface did not close within "
                f"{_CLOSE_TIMEOUT_SECS:g}s; continuing shutdown"
            )

    iface = None
    resolved_target = None
    retry_delay = max(0.0, _RECONNECT_INITIAL_DELAY_SECS)

    stop = threading.Event()
    initial_snapshot_sent = False

    def handle_sigterm(*_):
        """Stop the daemon when a termination signal is received."""

        stop.set()

    def handle_sigint(signum, frame):
        """Handle ``SIGINT`` by requesting shutdown and escalating on repeat."""

        if stop.is_set():
            signal.default_int_handler(signum, frame)
            return

        stop.set()

    signal.signal(signal.SIGINT, handle_sigint)
    signal.signal(signal.SIGTERM, handle_sigterm)

    target = INSTANCE or "(no POTATOMESH_INSTANCE)"
    configured_port = PORT
    active_candidate = configured_port
    announced_target = False
    print(
        f"Mesh daemon: nodes+messages → {target} | port={configured_port or 'auto'} | channel={CHANNEL_INDEX}"
    )
    try:
        while not stop.is_set():
            if iface is None:
                try:
                    if active_candidate:
                        iface, resolved_target = _create_serial_interface(
                            active_candidate
                        )
                    else:
                        iface, resolved_target = _create_default_interface()
                        active_candidate = resolved_target
                    retry_delay = max(0.0, _RECONNECT_INITIAL_DELAY_SECS)
                    initial_snapshot_sent = False
                    if not announced_target and resolved_target:
                        print(f"[info] using mesh interface: {resolved_target}")
                        announced_target = True
                except NoAvailableMeshInterface as exc:
                    print(f"[error] {exc}")
                    _close_interface(iface)
                    raise SystemExit(1) from exc
                except Exception as exc:
                    candidate_desc = active_candidate or "auto"
                    print(
                        f"[warn] failed to create mesh interface ({candidate_desc}): {exc}"
                    )
                    if configured_port is None:
                        active_candidate = None
                        announced_target = False
                    stop.wait(retry_delay)
                    if _RECONNECT_MAX_DELAY_SECS > 0:
                        retry_delay = min(
                            (
                                retry_delay * 2
                                if retry_delay
                                else _RECONNECT_INITIAL_DELAY_SECS
                            ),
                            _RECONNECT_MAX_DELAY_SECS,
                        )
                    continue

            if not initial_snapshot_sent:
                try:
                    nodes = getattr(iface, "nodes", {}) or {}
                    node_items = _node_items_snapshot(nodes)
                    if node_items is None:
                        _debug_log(
                            "skipping node snapshot; nodes changed during iteration"
                        )
                    else:
                        processed_snapshot_item = False
                        for node_id, n in node_items:
                            processed_snapshot_item = True
                            try:
                                upsert_node(node_id, n)
                            except Exception as e:
                                print(
                                    f"[warn] failed to update node snapshot for {node_id}: {e}"
                                )
                                if DEBUG:
                                    _debug_log(f"node object: {n!r}")
                        if processed_snapshot_item:
                            initial_snapshot_sent = True
                except Exception as e:
                    print(f"[warn] failed to update node snapshot: {e}")
                    _close_interface(iface)
                    iface = None
                    stop.wait(retry_delay)
                    if _RECONNECT_MAX_DELAY_SECS > 0:
                        retry_delay = min(
                            (
                                retry_delay * 2
                                if retry_delay
                                else _RECONNECT_INITIAL_DELAY_SECS
                            ),
                            _RECONNECT_MAX_DELAY_SECS,
                        )
                    continue

            retry_delay = max(0.0, _RECONNECT_INITIAL_DELAY_SECS)
            stop.wait(SNAPSHOT_SECS)
    except KeyboardInterrupt:
        _debug_log("received KeyboardInterrupt; shutting down")
        stop.set()
    finally:
        _close_interface(iface)


if __name__ == "__main__":
    main()
