"""Shared message bus + loop-prevention core for the archy-messh Phase 0 bridge. Each protocol adapter runs on its own thread/event loop (Meshtastic's pubsub callbacks fire from its internal reader thread, MeshCore is asyncio-native, Reticulum/LXMF invoke callbacks from RNS's internal transport thread). All three push into this single thread-safe queue so one dispatcher can fan messages out across protocols without each adapter needing to know about the others. """ import hashlib import queue import threading import time from dataclasses import dataclass @dataclass class BridgeMessage: network: str # "meshtastic" | "meshcore" | "reticulum" sender: str text: str received_at: float class MessageBus: """Thread-safe inbox with content-hash dedup to prevent repeat loops. Dedup is content-hash based (not per-network packet-id based), since the whole point of the bridge is that the same text shows up natively on all three networks once relayed. A short TTL window is enough to stop the obvious loop (bridge re-hears its own relayed copy) without needing any cross-protocol message-id scheme, which doesn't exist. """ def __init__(self, dedup_ttl_seconds: float = 300.0): self._queue: "queue.Queue[BridgeMessage]" = queue.Queue() self._dedup_ttl = dedup_ttl_seconds self._seen: dict[str, float] = {} self._lock = threading.Lock() @staticmethod def _content_hash(text: str) -> str: return hashlib.sha256(text.strip().encode("utf-8")).hexdigest() def publish(self, message: BridgeMessage) -> None: self._queue.put(message) def is_duplicate(self, text: str) -> bool: h = self._content_hash(text) now = time.monotonic() with self._lock: self._seen = {k: v for k, v in self._seen.items() if v > now} if h in self._seen: return True self._seen[h] = now + self._dedup_ttl return False def mark_seen(self, text: str) -> None: """Record text as already-seen without reporting duplicate status. The dispatcher relays a *prefixed* copy of each message (e.g. "[meshtastic/x] hello"), not the original text. Dedup is content-hash based, so unless the exact on-air relayed string is recorded here, the bridge won't recognise its own relayed copy when it echoes back from another network — it would re-relay it, stacking another prefix each hop (an unbounded loop). Marking every outbound string seen closes that loop. """ h = self._content_hash(text) now = time.monotonic() with self._lock: self._seen[h] = now + self._dedup_ttl def get(self, timeout: float | None = None) -> BridgeMessage: return self._queue.get(timeout=timeout)