""" Repeater-specific CompanionFrameServer with SQLite persistence. Thin subclass of :class:`openhop_core.companion.frame_server.CompanionFrameServer` that adds SQLite-backed message, contact, and channel persistence via a ``sqlite_handler`` dependency. """ from __future__ import annotations import asyncio import logging from typing import Optional from openhop_core.companion.constants import RESP_CODE_NO_MORE_MESSAGES from openhop_core.companion.frame_server import CompanionFrameServer as _BaseFrameServer from openhop_core.companion.models import QueuedMessage logger = logging.getLogger("CompanionFrameServer") class CompanionFrameServer(_BaseFrameServer): """Adds SQLite persistence for messages, contacts, and channels. Constructor signature is intentionally kept compatible with the previous monolithic implementation so ``main.py`` call-sites need zero changes. """ def __init__( self, bridge, companion_hash: str, port: int = 5000, bind_address: str = "0.0.0.0", # nosec B104 - intentional default for LAN reachability client_idle_timeout_sec: Optional[int] = 8 * 60 * 60, # 8 hours sqlite_handler=None, local_hash: Optional[int] = None, stats_getter=None, control_handler=None, ): super().__init__( bridge=bridge, companion_hash=companion_hash, port=port, bind_address=bind_address, client_idle_timeout_sec=client_idle_timeout_sec, device_model="openHop-Repeater-Companion", device_version=None, # use FIRMWARE_VER_CODE from openhop-core build_date="13 Feb 2026", local_hash=local_hash, stats_getter=stats_getter, control_handler=control_handler, ) self.sqlite_handler = sqlite_handler # ----------------------------------------------------------------- # Persistence hook overrides # ----------------------------------------------------------------- async def _persist_companion_message(self, msg_dict: dict) -> None: """Persist message to SQLite and pop from bridge queue. The bridge's ``offline_queue_size`` (``message_queue._max_size``) doubles as the SQLite retention limit: 0 disables offline storage entirely, so the message is dropped instead of persisted. """ if not self.sqlite_handler: return retention = getattr(self.bridge.message_queue, "_max_size", None) if retention == 0: self.bridge.message_queue.pop_last() return await asyncio.to_thread( self.sqlite_handler.companion_push_message, self.companion_hash, msg_dict, retention, ) self.bridge.message_queue.pop_last() def _sync_next_from_persistence(self) -> Optional[QueuedMessage]: """Retrieve next message from SQLite when bridge queue is empty.""" if not self.sqlite_handler: return None msg_dict = self.sqlite_handler.companion_pop_message(self.companion_hash) if not msg_dict: return None return QueuedMessage( sender_key=msg_dict.get("sender_key", b""), txt_type=msg_dict.get("txt_type", 0), timestamp=msg_dict.get("timestamp", 0), text=msg_dict.get("text", ""), is_channel=bool(msg_dict.get("is_channel", False)), channel_idx=msg_dict.get("channel_idx", 0), path_len=msg_dict.get("path_len", 0), ) # ----------------------------------------------------------------- # Non-blocking command overrides (keep event loop responsive) # ----------------------------------------------------------------- async def _cmd_sync_next_message(self, data: bytes) -> None: """Sync next message; run persistence read in thread so SQLite does not block.""" msg = self.bridge.sync_next_message() if msg is None: msg = await asyncio.to_thread(self._sync_next_from_persistence) if msg is None: self._write_frame(bytes([RESP_CODE_NO_MORE_MESSAGES])) return self._write_frame(self._build_message_frame(msg)) @staticmethod def _contact_to_dict(c) -> dict: """Convert a Contact object to a persistence dict.""" pk = c.public_key if isinstance(c.public_key, bytes) else bytes.fromhex(c.public_key) return { "pubkey": pk, "name": c.name, "adv_type": c.adv_type, "flags": c.flags, "out_path_len": c.out_path_len, "out_path": ( c.out_path if isinstance(c.out_path, bytes) else (bytes.fromhex(c.out_path) if c.out_path else b"") ), "last_advert_timestamp": c.last_advert_timestamp, "lastmod": c.lastmod, "gps_lat": c.gps_lat, "gps_lon": c.gps_lon, "sync_since": c.sync_since, } async def _persist_contact(self, contact) -> None: """Upsert a single contact to SQLite (non-blocking).""" if not self.sqlite_handler: return contact_dict = self._contact_to_dict(contact) await asyncio.to_thread( self.sqlite_handler.companion_upsert_contact, self.companion_hash, contact_dict, ) async def _save_contacts(self) -> None: """Persist all contacts to SQLite (non-blocking).""" if not self.sqlite_handler: return contacts = self.bridge.get_contacts() dicts = [self._contact_to_dict(c) for c in contacts] await asyncio.to_thread( self.sqlite_handler.companion_save_contacts, self.companion_hash, dicts, ) async def _save_channels(self) -> None: """Persist channels to SQLite (non-blocking).""" if not self.sqlite_handler: return channels = [] max_ch = getattr(getattr(self.bridge, "channels", None), "max_channels", 40) for idx in range(max_ch): ch = self.bridge.get_channel(idx) if ch is not None: channels.append( { "channel_idx": idx, "name": ch.name, "secret": ch.secret, } ) await asyncio.to_thread( self.sqlite_handler.companion_save_channels, self.companion_hash, channels, ) async def stop(self) -> None: """Persist contacts and channels before stopping (so they survive daemon restart).""" if self.sqlite_handler: try: await self._save_contacts() await self._save_channels() except Exception as e: logger.warning("Failed to persist contacts/channels on stop: %s", e) await super().stop()