archy-messh/host-bridge/bridge_core.py

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