""" Device event callbacks for MeshCore GUI. Handles ``CHANNEL_MSG_RECV``, ``CONTACT_MSG_RECV`` and ``RX_LOG_DATA`` events from the MeshCore library. Extracted from ``SerialWorker`` so the worker only deals with connection lifecycle. BBS routing ~~~~~~~~~~~ Direct Messages (``CONTACT_MSG_RECV``) whose text starts with ``!`` are forwarded to :class:`~meshcore_gui.services.bbs_service.BbsCommandHandler` **before** any other DM processing. This path is completely independent of :class:`~meshcore_gui.services.bot.MeshBot`. """ from typing import TYPE_CHECKING, Callable, Dict, List, Optional from meshcore_gui.config import debug_print from meshcore_gui.core.models import Message, RxLogEntry from meshcore_gui.core.protocols import SharedDataWriter from meshcore_gui.ble.packet_decoder import PacketDecoder, PayloadType from meshcore_gui.services.bot import MeshBot from meshcore_gui.services.dedup import DualDeduplicator if TYPE_CHECKING: from meshcore_gui.services.bbs_service import BbsCommandHandler class EventHandler: """Processes device events and writes results to shared data. Args: shared: SharedDataWriter for storing messages and RX log. decoder: PacketDecoder for raw LoRa packet decryption. dedup: DualDeduplicator for message deduplication. bot: MeshBot for auto-reply logic. """ # Maximum entries in the path cache before oldest are evicted. _PATH_CACHE_MAX = 200 def __init__( self, shared: SharedDataWriter, decoder: PacketDecoder, dedup: DualDeduplicator, bot: MeshBot, bbs_handler: Optional["BbsCommandHandler"] = None, command_sink: Optional[Callable[[Dict], None]] = None, ) -> None: self._shared = shared self._decoder = decoder self._dedup = dedup self._bot = bot self._bbs_handler = bbs_handler self._command_sink = command_sink # Cache: message_hash → path_hashes (from RX_LOG decode). # Used by on_channel_msg fallback to recover hashes that the # CHANNEL_MSG_RECV event does not provide. self._path_cache: Dict[str, list] = {} # Secondary path cache keyed by "sender:text[:100]". # Fallback for on_channel_msg when the message_hash computed by # meshcoredecoder differs from the one in CHANNEL_MSG_RECV # (two independent libraries may disagree on the hash format). self._path_cache_by_content: Dict[str, list] = {} # ------------------------------------------------------------------ # Helpers — resolve names at receive time # ------------------------------------------------------------------ def _resolve_path_names(self, path_hashes: list) -> list: """Resolve path hashes to display names. Performs a contact lookup for each hash *now* so the names are captured at receive time and stored in the archive. Supports 1-byte (2 hex chars), 2-byte (4 hex chars) and 3-byte (6 hex chars) path hashes as introduced in firmware v1.14. Contact lookup uses ``startswith`` matching and is hash-size agnostic. Args: path_hashes: List of hex strings, 2–6 chars each (1–3 bytes). Returns: List of display names (same length as *path_hashes*). Unknown hashes become their uppercase hex value. """ names = [] for h in path_hashes: if not h or len(h) < 2: names.append('-') continue name = self._shared.get_contact_name_by_prefix(h) # startswith matching is hash-size agnostic (2/4/6-char hashes). if name and name != h[:8]: names.append(name) else: names.append(h.upper()) return names @staticmethod def _looks_like_hex_identifier(value: str) -> bool: """Return True when *value* looks like a pubkey/hash prefix.""" if not value: return False probe = str(value).strip() if len(probe) < 6: return False return all(ch in '0123456789abcdefABCDEF' for ch in probe) # ------------------------------------------------------------------ # RX_LOG_DATA — the single source of truth for path info # ------------------------------------------------------------------ def on_rx_log(self, event) -> None: """Handle RX log data events.""" payload = event.payload # Extract basic RX log info time_str = Message.now_timestamp() snr = payload.get('snr', 0) rssi = payload.get('rssi', 0) payload_type = '?' hops = payload.get('path_len', 0) # Try to decode payload to get message_hash message_hash = "" rx_path_hashes: list = [] rx_path_names: list = [] rx_sender: str = "" rx_receiver: str = self._shared.get_device_name() or "" payload_hex = payload.get('payload', '') decoded = None if payload_hex: decoded = self._decoder.decode(payload_hex) if decoded is not None: message_hash = decoded.message_hash payload_type = self._decoder.get_payload_type_text(decoded.payload_type) # Capture path info for all packet types if decoded.path_hashes: rx_path_hashes = decoded.path_hashes rx_path_names = self._resolve_path_names(decoded.path_hashes) # Use decoded path_length (from packet body) — more # reliable than the frame-header path_len which can be 0. if decoded.path_length: hops = decoded.path_length # Capture sender name when available (GroupText only) if decoded.sender: rx_sender = decoded.sender # Cache path_hashes for correlation with on_channel_msg. # Primary key: message_hash (fastest lookup). if decoded.path_hashes and message_hash: self._path_cache[message_hash] = decoded.path_hashes # Evict oldest entries if cache is too large if len(self._path_cache) > self._PATH_CACHE_MAX: oldest = next(iter(self._path_cache)) del self._path_cache[oldest] # Secondary key: "sender:text[:100]" (fallback when the # message_hash from meshcore and meshcoredecoder disagree). # Only populated for decrypted GroupText packets so we have # reliable sender + text to form the key. if decoded.path_hashes and decoded.is_decrypted and decoded.sender: ck = f"{decoded.sender}:{decoded.text[:100]}" self._path_cache_by_content[ck] = decoded.path_hashes if len(self._path_cache_by_content) > self._PATH_CACHE_MAX: oldest_ck = next(iter(self._path_cache_by_content)) del self._path_cache_by_content[oldest_ck] # Process decoded message if it's a group text if decoded.payload_type == PayloadType.GroupText and decoded.is_decrypted: if decoded.channel_idx is None: # The channel hash could not be resolved to a channel index # (PacketDecoder._hash_to_idx lookup returned None). # Marking dedup here would suppress on_channel_msg, which # carries a valid channel_idx from the device event — the only # path through which the bot can pass Guard 2 and respond. # Skip the entire block; on_channel_msg handles message + bot. # Path info is already in _path_cache for on_channel_msg to use. debug_print( f"RX_LOG → GroupText decrypted but channel_idx unresolved " f"(hash={decoded.message_hash}); deferring to on_channel_msg" ) else: self._dedup.mark_hash(decoded.message_hash) self._dedup.mark_content( decoded.sender, decoded.channel_idx, decoded.text, ) # Channel-agnostic sentinel: prevents on_channel_msg from # storing a duplicate even if its channel_idx or # message_hash differ from what PacketDecoder resolved. # Uses sentinel '*' so it cannot clash with real int indices. self._dedup.mark_content(decoded.sender, '*', decoded.text) sender_pubkey = '' if decoded.sender: match = self._shared.get_contact_by_name(decoded.sender) if match: sender_pubkey, _contact = match snr_msg = self._extract_snr(payload) self._shared.add_message(Message.incoming( decoded.sender, decoded.text, decoded.channel_idx, time=time_str, snr=snr_msg, path_len=decoded.path_length, sender_pubkey=sender_pubkey, path_hashes=decoded.path_hashes, path_names=rx_path_names, message_hash=decoded.message_hash, )) debug_print( f"RX_LOG → message: hash={decoded.message_hash}, " f"sender={decoded.sender!r}, ch={decoded.channel_idx}, " f"path={decoded.path_hashes}, " f"path_names={rx_path_names}" ) self._bot.check_and_reply( sender=decoded.sender, text=decoded.text, channel_idx=decoded.channel_idx, snr=snr_msg, path_len=decoded.path_length, path_hashes=decoded.path_hashes, sender_pubkey=sender_pubkey, ) # BBS channel hook: auto-whitelist sender and reply # for '!bbs' on a configured BBS channel. # Must run here because on_channel_msg is suppressed # by content-dedup when on_rx_log already stored the # message (the common path for resolved channel_idx). if ( self._bbs_handler is not None and self._command_sink is not None ): bbs_reply = self._bbs_handler.handle_channel_msg( channel_idx=decoded.channel_idx, sender=decoded.sender, sender_key=sender_pubkey, text=decoded.text, ) if bbs_reply is not None: debug_print( f"BBS channel reply (rx_log) on " f"ch{decoded.channel_idx} to " f"{decoded.sender!r}: {bbs_reply[:60]}" ) self._command_sink({ "action": "send_message", "channel": decoded.channel_idx, "text": bbs_reply, }) # Add RX log entry with message_hash and path info (if available) # ── Fase 1 Observer: raw packet metadata ── raw_packet_len = len(payload_hex) // 2 if payload_hex else 0 raw_payload_len = max(0, raw_packet_len - 1 - hops) if payload_hex else 0 raw_route_type = "D" if hops > 0 else ("F" if payload_hex else "") raw_packet_type_num = -1 if payload_hex and decoded is not None: try: raw_packet_type_num = decoded.payload_type.value except (AttributeError, ValueError): pass self._shared.add_rx_log(RxLogEntry( time=time_str, snr=snr, rssi=rssi, payload_type=payload_type, hops=hops, message_hash=message_hash, path_hashes=rx_path_hashes, path_names=rx_path_names, sender=rx_sender, receiver=rx_receiver, raw_payload=payload_hex, packet_len=raw_packet_len, payload_len=raw_payload_len, route_type=raw_route_type, packet_type_num=raw_packet_type_num, )) # ------------------------------------------------------------------ # CHANNEL_MSG_RECV — fallback when RX_LOG decode missed it # ------------------------------------------------------------------ def on_channel_msg(self, event) -> None: """Handle channel message events.""" payload = event.payload # DIAGNOSTIC — remove after root cause confirmed. # If two lines appear for one incoming packet (different ch_idx), # the meshcore library fires CHANNEL_MSG_RECV once per subscribed # channel, which is the confirmed cause of cross-channel duplicates. debug_print( f"CHANNEL_MSG_RECV: ch_idx={payload.get('channel_idx')!r}, " f"msg_hash={str(payload.get('message_hash', ''))[:12]!r}, " f"text={str(payload.get('text', ''))[:30]!r}" ) debug_print(f"Channel msg payload keys: {list(payload.keys())}") # Dedup via hash msg_hash = payload.get('message_hash', '') if msg_hash and self._dedup.is_hash_seen(msg_hash): debug_print(f"Channel msg suppressed (hash): {msg_hash}") return # Parse sender from "SenderName: message body" format raw_text = payload.get('text', '') sender, msg_text = '', raw_text if ': ' in raw_text: name_part, body_part = raw_text.split(': ', 1) sender = name_part.strip() msg_text = body_part elif raw_text: msg_text = raw_text # Channel-agnostic guard: on_rx_log already stored this message # (possibly under a different channel_idx due to index-scheme # differences between meshcoredecoder and the meshcore library). # The '*' sentinel is set by on_rx_log only when it successfully # stored the message, so this guard never fires for the deferred # case (channel_idx unresolved in on_rx_log). if sender and self._dedup.is_content_seen(sender, '*', msg_text): debug_print( f"Channel msg suppressed (rxlog stored, channel-agnostic): {sender!r}" ) return # Dedup via content ch_idx = payload.get('channel_idx') if self._dedup.is_content_seen(sender, ch_idx, msg_text): debug_print(f"Channel msg suppressed (content): {sender!r}") return debug_print( f"Channel msg (fallback): sender={sender!r}, " f"text={msg_text[:40]!r}" ) sender_pubkey = ( payload.get('pubkey_prefix') or payload.get('sender_pubkey') or payload.get('signature') or '' ) if not sender_pubkey and sender: match = self._shared.get_contact_by_name(sender) if match: sender_pubkey, _contact = match snr = self._extract_snr(payload) # Recover path_hashes from RX_LOG cache (CHANNEL_MSG_RECV # does not carry them, but the preceding RX_LOG decode does). # Primary lookup: by message_hash. path_hashes = self._path_cache.pop(msg_hash, []) if msg_hash else [] # Fallback lookup: by "sender:text[:100]" content key. # Handles the case where meshcoredecoder and the meshcore library # compute different message_hash values for the same packet. if not path_hashes and sender and msg_text: ck = f"{sender}:{msg_text[:100]}" path_hashes = self._path_cache_by_content.pop(ck, []) if path_hashes: debug_print( f"on_channel_msg: path_hashes recovered via content key: " f"{path_hashes}" ) path_names = self._resolve_path_names(path_hashes) self._shared.add_message(Message.incoming( sender, msg_text, ch_idx, snr=snr, path_len=payload.get('path_len', 0), sender_pubkey=sender_pubkey, path_hashes=path_hashes, path_names=path_names, message_hash=msg_hash, )) # BBS channel hook: auto-whitelist sender + bootstrap reply for !-commands. # Runs on every message on a configured BBS channel, independent of the bot. if self._bbs_handler is not None and self._command_sink is not None: bbs_reply = self._bbs_handler.handle_channel_msg( channel_idx=ch_idx, sender=sender, sender_key=sender_pubkey, text=msg_text, ) if bbs_reply is not None: debug_print(f"BBS channel reply on ch{ch_idx} to {sender!r}: {bbs_reply[:60]}") self._command_sink({ "action": "send_message", "channel": ch_idx, "text": bbs_reply, }) self._bot.check_and_reply( sender=sender, text=msg_text, channel_idx=ch_idx, snr=snr, path_len=payload.get('path_len', 0), sender_pubkey=sender_pubkey, ) # ------------------------------------------------------------------ # CONTACT_MSG_RECV — DMs # ------------------------------------------------------------------ def on_contact_msg(self, event) -> None: """Handle direct message and room message events. Room Server traffic also arrives as ``CONTACT_MSG_RECV``. In practice the payload is not stable enough to rely only on ``signature`` + ``pubkey_prefix``. Incoming room messages from *other* participants may omit ``signature`` and may carry the room key in receiver-style fields instead of ``pubkey_prefix``. To keep the rest of the GUI unchanged, room messages are stored with ``sender`` = actual author name and ``sender_pubkey`` = room public key. The Room Server panel already filters on ``sender_pubkey`` to decide to which room a message belongs. """ payload = event.payload or {} pubkey = payload.get('pubkey_prefix', '') txt_type = payload.get('txt_type', 0) signature = payload.get('signature', '') debug_print( "DM payload keys: " f"{list(payload.keys())}; txt_type={txt_type}; " f"pubkey_prefix={pubkey[:12]}; " f"receiver={(payload.get('receiver') or '')[:12]}; " f"room_pubkey={(payload.get('room_pubkey') or '')[:12]}; " f"signature={(signature or '')[:12]}" ) # Common fields for both Room and DM messages msg_hash = payload.get('message_hash', '') path_hashes = self._path_cache.pop(msg_hash, []) if msg_hash else [] path_names = self._resolve_path_names(path_hashes) raw_path_len = payload.get('path_len', 0) path_len = raw_path_len if raw_path_len < 255 else 0 if path_hashes: path_len = len(path_hashes) room_pubkey = ( payload.get('room_pubkey') or payload.get('receiver') or payload.get('receiver_pubkey') or payload.get('receiver_pubkey_prefix') or pubkey or '' ) is_room_message = txt_type == 2 if is_room_message: author = '' explicit_name = ( payload.get('author') or payload.get('sender_name') or payload.get('name') or '' ) if explicit_name and not self._looks_like_hex_identifier(explicit_name): author = explicit_name sender_field = str(payload.get('sender') or '').strip() if not author and sender_field and not self._looks_like_hex_identifier(sender_field): author = sender_field author_key = ( signature or payload.get('sender_pubkey') or payload.get('author_pubkey') or (sender_field if self._looks_like_hex_identifier(sender_field) else '') or '' ) if not author and author_key: author = self._shared.get_contact_name_by_prefix(author_key) if not author: author = ( explicit_name or sender_field or (author_key[:8] if author_key else '') or '?' ) self._shared.add_message(Message.incoming( author, payload.get('text', ''), None, snr=self._extract_snr(payload), path_len=path_len, sender_pubkey=room_pubkey, path_hashes=path_hashes, path_names=path_names, message_hash=msg_hash, )) debug_print( f"Room msg from {author} via room {room_pubkey[:12]} " f"(sig={signature[:12] if signature else '-'}): " f"{payload.get('text', '')[:30]}" ) return # --- Regular DM --- sender = '' if pubkey: sender = self._shared.get_contact_name_by_prefix(pubkey) if not sender: sender = ( payload.get('name') or payload.get('sender') or (pubkey[:8] if pubkey else '') ) dm_text = payload.get('text', '') # BBS routing: DMs starting with '!' go directly to BbsCommandHandler. # This path is independent of the bot (MeshBot is for channel messages only). if ( self._bbs_handler is not None and self._command_sink is not None and dm_text.strip().startswith("!") ): bbs_reply = self._bbs_handler.handle_dm( sender=sender, sender_key=pubkey, text=dm_text, ) if bbs_reply is not None: debug_print(f"BBS DM reply to {sender} ({pubkey[:8]}): {bbs_reply[:60]}") self._command_sink({ "action": "send_dm", "pubkey": pubkey, "text": bbs_reply, }) # Always store the incoming DM in the message archive too self._shared.add_message(Message.incoming( sender, dm_text, None, snr=self._extract_snr(payload), path_len=path_len, sender_pubkey=pubkey, path_hashes=path_hashes, path_names=path_names, message_hash=msg_hash, )) debug_print(f"BBS DM stored from {sender}: {dm_text[:30]}") return self._shared.add_message(Message.incoming( sender, dm_text, None, snr=self._extract_snr(payload), path_len=path_len, sender_pubkey=pubkey, path_hashes=path_hashes, path_names=path_names, message_hash=msg_hash, )) debug_print(f"DM received from {sender}: {dm_text[:30]}") # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @staticmethod def _extract_snr(payload: Dict) -> Optional[float]: """Extract SNR from a payload dict (handles 'SNR' and 'snr' keys).""" raw = payload.get('SNR') or payload.get('snr') if raw is not None: try: return float(raw) except (ValueError, TypeError): pass return None