diff --git a/modules/service_plugins/channel_bridge_utils.py b/modules/service_plugins/channel_bridge_utils.py new file mode 100644 index 0000000..805251f --- /dev/null +++ b/modules/service_plugins/channel_bridge_utils.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +""" +Shared base for one-way bridges that post MeshCore channel messages elsewhere +(Discord, Telegram). The file name ends in ``_utils`` so the service loader +does not take ChannelBridgeBase for a service of its own. +""" + +import asyncio +import contextlib +import copy +import time +from collections import deque +from typing import Any + +from meshcore import EventType + +from ..profanity_filter import censor, contains_profanity +from .base_service import BaseServicePlugin + + +class ChannelBridgeBase(BaseServicePlugin): + """Lifecycle, message intake and the retrying send queue shared by channel bridges. + + Subclasses provide the destination: which targets a channel maps to, how a + message is formatted and queued, the rate limit, and the HTTP post. Log + wording that differs between bridges is kept per subclass through the + ``*_log`` templates. + """ + + # "Discord" or "Telegram": names the bridge in shared log lines + bridge_label = "" + # Payload key holding the message text (for the dropped-after-retries line) + payload_text_key = "text" + old_message_log = "Dropping old message from queue [{channel}]: age {age:.1f}s > {max_age}s" + retry_log = "Message failed, retry in {delay:.1f}s ({retry}/{max_retries}) [{channel}]" + processor_error_log = "Error in queue processor: {error}" + + # Set by subclasses in __init__ (annotations for type checkers) + filter_profanity: str + bridge_bot_responses: bool + http_session: Any + message_queues: dict[str, Any] + send_times: dict[str, Any] + max_retries: int + retry_delay_base: float + max_queue_age: float + _queue_processor_task: Any + + @property + def _bridge_mappings(self) -> dict[str, Any]: + """Configured channel name -> destination(s).""" + raise NotImplementedError + + def _open_http_session(self) -> None: + raise NotImplementedError + + def _init_queues(self) -> None: + raise NotImplementedError + + def _targets_for(self, channel_name: str) -> Any: + """The configured destination(s) for a channel, or None.""" + raise NotImplementedError + + def _prepare_message_text(self, message_text: str) -> str: + """Hook applied to the message text before the profanity filter.""" + return message_text + + async def _deliver(self, targets: Any, sender_name: str, message_text: str, channel_name: str) -> None: + raise NotImplementedError + + def _throttled(self, key: str, queue: Any, current_time: float) -> bool: + """True when the destination's rate limit says to wait.""" + raise NotImplementedError + + async def _send_queued(self, queued_msg: Any) -> bool: + raise NotImplementedError + + async def start(self) -> None: + if not self.enabled: + self.logger.info(f"{self.bridge_label} bridge service is disabled") + return + if not self._bridge_mappings: + self.logger.warning(f"{self.bridge_label} bridge enabled but no channels configured") + return + + self.logger.info(f"Starting {self.bridge_label} bridge service...") + self._open_http_session() + + # Subscribe to channel message events + # NOTE: We do NOT subscribe to CONTACT_MSG_RECV (DMs are never bridged) + if hasattr(self.bot, 'meshcore') and self.bot.meshcore: + self._subscribe(self.bot.meshcore, EventType.CHANNEL_MSG_RECV, self._on_mesh_channel_message) + self.logger.info("Subscribed to CHANNEL_MSG_RECV events") + else: + self.logger.error("Cannot subscribe to events - meshcore not available") + if self.http_session is not None: + await self.http_session.close() + self.http_session = None + return + + # Register for bot-sent channel messages so bot responses are bridged too + if self.bridge_bot_responses and getattr(self.bot, 'channel_sent_listeners', None) is not None: + self.bot.channel_sent_listeners.append(self._on_mesh_channel_message) + self.logger.info("Registered for bot channel-sent events (bridge_bot_responses=true)") + + self._init_queues() + + # Start background queue processor task + self._queue_processor_task = asyncio.create_task(self._process_message_queues()) + + self._running = True + self.logger.info( + f"{self.bridge_label} bridge service started (bridging {len(self._bridge_mappings)} channels)" + ) + + async def on_transport_reconnected(self) -> None: + """Re-subscribe to channel messages on the new meshcore instance.""" + if not self._running or not getattr(self.bot, 'meshcore', None): + return + self._unsubscribe_all() + self._subscribe(self.bot.meshcore, EventType.CHANNEL_MSG_RECV, self._on_mesh_channel_message) + self.logger.info(f"{self.bridge_label} bridge re-subscribed to CHANNEL_MSG_RECV after transport reconnect") + + async def stop(self) -> None: + self.logger.info(f"Stopping {self.bridge_label} bridge service...") + self._running = False + self._unsubscribe_all() + + # Unregister bot channel-sent listener + if getattr(self.bot, 'channel_sent_listeners', None) is not None: + with contextlib.suppress(ValueError): + self.bot.channel_sent_listeners.remove(self._on_mesh_channel_message) + + await self._cancel_tasks(self._queue_processor_task) + if self.http_session: + await self.http_session.close() + self.http_session = None + self.logger.info(f"{self.bridge_label} bridge service stopped") + + async def _on_mesh_channel_message(self, event: Any, metadata: Any = None) -> None: + """Bridge one mesh channel message. DMs are never bridged.""" + try: + # Copy payload immediately to avoid segfault if event is freed + payload = copy.deepcopy(event.payload) if hasattr(event, 'payload') else None + if payload is None: + self.logger.warning("Channel message event has no payload") + return + + channel_idx = payload.get('channel_idx', 0) + channel_name = self.bot.channel_manager.get_channel_name(channel_idx) + text = payload.get('text', '') + + # NEVER bridge DMs (double-check for safety) + if not channel_name or channel_name.lower() in ('dm', 'direct', 'private'): + self.logger.debug("Ignoring DM (DMs are never bridged)") + return + + targets = self._targets_for(channel_name) + if not targets: + self.logger.debug(f"Channel '{channel_name}' not configured for {self.bridge_label} bridge") + return + + # Sender is embedded in the text ("sender: message") + if ':' in text and not text.startswith('http'): + parts = text.split(':', 1) + sender_name = parts[0].strip() + message_text = parts[1].strip() if len(parts) > 1 else text + else: + sender_name = 'Unknown' + message_text = text + + message_text = self._prepare_message_text(message_text) + + # Profanity filter: drop (don't bridge), censor (replace with ****), or off + if self.filter_profanity == 'drop': + if contains_profanity(sender_name, self.logger) or contains_profanity(message_text, self.logger): + self.logger.debug( + f"{self.bridge_label} bridge: dropping message with profanity from [{channel_name}]" + ) + return + elif self.filter_profanity == 'censor': + sender_name = censor(sender_name, self.logger) + message_text = censor(message_text, self.logger) + + await self._deliver(targets, sender_name, message_text, channel_name) + except Exception as e: + self.logger.error(f"Error handling mesh channel message: {e}", exc_info=True) + + async def _process_message_queues(self) -> None: + """Send queued messages, respecting the rate limit, with exponential backoff on failure.""" + while self._running: + try: + current_time = time.time() + for key, queue in list(self.message_queues.items()): + if not queue: + continue + if self._throttled(key, queue, current_time): + continue + + # Next message whose retry delay has passed + queued_msg = None + for msg in queue: + if current_time >= msg.next_retry_at: + queued_msg = msg + break + if queued_msg is None: + continue + + age = current_time - queued_msg.first_queued + if age > self.max_queue_age: + queue.remove(queued_msg) + self.logger.warning(self.old_message_log.format( + channel=queued_msg.channel_name, age=age, max_age=self.max_queue_age + )) + continue + + success = await self._send_queued(queued_msg) + if success: + queue.remove(queued_msg) + if key not in self.send_times: + self.send_times[key] = deque() + self.send_times[key].append(current_time) + else: + queued_msg.retry_count += 1 + if queued_msg.retry_count > self.max_retries: + queue.remove(queued_msg) + self.logger.error( + f"Dropping message after {self.max_retries} retries " + f"[{queued_msg.channel_name}]: {queued_msg.payload[self.payload_text_key][:50]}..." + ) + else: + delay = self.retry_delay_base * (2 ** (queued_msg.retry_count - 1)) + queued_msg.next_retry_at = current_time + delay + self.logger.debug(self.retry_log.format( + delay=delay, + retry=queued_msg.retry_count, + max_retries=self.max_retries, + channel=queued_msg.channel_name, + )) + + # Small delay to prevent tight loop + await asyncio.sleep(0.1) + except asyncio.CancelledError: + break + except Exception as e: + self.logger.error(self.processor_error_log.format(error=e), exc_info=True) + await asyncio.sleep(1.0) diff --git a/modules/service_plugins/discord_bridge_service.py b/modules/service_plugins/discord_bridge_service.py index 51c20d9..4d772f5 100644 --- a/modules/service_plugins/discord_bridge_service.py +++ b/modules/service_plugins/discord_bridge_service.py @@ -5,7 +5,6 @@ Posts MeshCore channel messages to Discord via webhooks (one-way, read-only) """ import asyncio -import copy import time from collections import deque from collections.abc import Mapping @@ -13,9 +12,6 @@ from dataclasses import dataclass from datetime import datetime from typing import Any, Optional -# Import meshcore -from meshcore import EventType - # Try to import aiohttp for async HTTP (preferred) try: import aiohttp @@ -32,16 +28,14 @@ except ImportError: requests = None # type: ignore[assignment] REQUESTS_AVAILABLE = False -# Import base service -import contextlib - from ..bridge_outbound import ( DISCORD_WEBHOOK_ALLOWED_MENTIONS, neutralize_discord_mention_content, ) -from ..profanity_filter import censor, contains_profanity +from ..profanity_filter import censor, contains_profanity # noqa: F401 re-exported from ..security_utils import sanitize_name -from .base_service import BaseServicePlugin +from .base_service import BaseServicePlugin # noqa: F401 re-exported +from .channel_bridge_utils import ChannelBridgeBase @dataclass @@ -61,7 +55,7 @@ class QueuedMessage: self.next_retry_at = time.time() -class DiscordBridgeService(BaseServicePlugin): +class DiscordBridgeService(ChannelBridgeBase): """Discord bridge service. Posts MeshCore channel messages to Discord channels via webhooks. @@ -176,6 +170,76 @@ class DiscordBridgeService(BaseServicePlugin): self.logger.warning("No Discord channel mappings configured. Discord bridge will not post any messages.") self.logger.info("Add channel mappings in config: bridge. = ") + bridge_label = "Discord" + payload_text_key = "content" + old_message_log = "Dropping old message from queue [{channel}]: age {age:.1f}s exceeds max {max_age}s" + retry_log = "Message failed, will retry in {delay:.1f}s (attempt {retry}/{max_retries}) [{channel}]" + processor_error_log = "Error in message queue processor: {error}" + + @property + def _bridge_mappings(self) -> dict[str, list[str]]: + return self.channel_webhooks + + def _open_http_session(self) -> None: + # Create aiohttp session if available + if AIOHTTP_AVAILABLE: + self.http_session = aiohttp.ClientSession() + self.logger.debug("Using aiohttp for async HTTP requests") + else: + self.logger.debug("Using requests library for HTTP requests (fallback)") + + def _init_queues(self) -> None: + # Initialize message queues for each webhook + for webhook_urls in self.channel_webhooks.values(): + for webhook_url in webhook_urls: + self.message_queues[webhook_url] = deque() + self.send_times[webhook_url] = deque() + + def _targets_for(self, channel_name: str) -> Optional[list[str]]: + # Check if this channel is configured for bridging (case-insensitive) + for config_channel, urls in self.channel_webhooks.items(): + if config_channel.lower() == channel_name.lower(): + return urls + return None + + def _prepare_message_text(self, message_text: str) -> str: + # Clean up MeshCore @ mentions: @[username] → **@username** + message_text = self._format_mentions(message_text) + return neutralize_discord_mention_content(message_text) + + async def _deliver(self, targets: list[str], sender_name: str, message_text: str, channel_name: str) -> None: + # Fan out to all configured webhooks for this channel + for webhook_url in targets: + await self._queue_message(webhook_url, message_text, channel_name, sender_name) + + def _throttled(self, key: str, queue: Any, current_time: float) -> bool: + # Clean up old send times (outside rate limit window) + if key in self.send_times: + send_times = self.send_times[key] + while send_times and (current_time - send_times[0]) > self.rate_limit_window: + send_times.popleft() + + # Check if we can send (proactive rate limiting) + can_send = True + if key in self.send_times: + recent_sends = len(self.send_times[key]) + if recent_sends >= self.rate_limit_max: + can_send = False + # Calculate wait time until oldest message expires + oldest_send = self.send_times[key][0] + wait_time = (oldest_send + self.rate_limit_window) - current_time + if wait_time > 0: + self.logger.debug(f"Rate limit throttling [{queue[0].channel_name}]: waiting {wait_time:.1f}s") + return not can_send + + async def _send_queued(self, queued_msg: QueuedMessage) -> bool: + return await self._post_to_webhook( + queued_msg.webhook_url, + queued_msg.payload, + queued_msg.channel_name, + queued_msg + ) + def _load_channel_mappings(self) -> None: """Load channel webhook mappings from config. @@ -309,170 +373,6 @@ class DiscordBridgeService(BaseServicePlugin): return '/'.join(parts) return url[:50] + '...' - async def start(self) -> None: - """Start the Discord bridge service. - - Sets up message event handlers and initializes HTTP session. - """ - if not self.enabled: - self.logger.info("Discord bridge service is disabled") - return - - if not self.channel_webhooks: - self.logger.warning("Discord bridge enabled but no channels configured") - return - - self.logger.info("Starting Discord bridge service...") - - # Create aiohttp session if available - if AIOHTTP_AVAILABLE: - self.http_session = aiohttp.ClientSession() - self.logger.debug("Using aiohttp for async HTTP requests") - else: - self.logger.debug("Using requests library for HTTP requests (fallback)") - - # Subscribe to channel message events - # NOTE: We do NOT subscribe to CONTACT_MSG_RECV (DMs are never bridged) - if hasattr(self.bot, 'meshcore') and self.bot.meshcore: - self._subscribe(self.bot.meshcore, EventType.CHANNEL_MSG_RECV, self._on_mesh_channel_message) - self.logger.info("Subscribed to CHANNEL_MSG_RECV events") - else: - self.logger.error("Cannot subscribe to events - meshcore not available") - if self.http_session is not None: - await self.http_session.close() - self.http_session = None - return - - # Register for bot-sent channel messages so bot responses are bridged too - if self.bridge_bot_responses and getattr(self.bot, 'channel_sent_listeners', None) is not None: - self.bot.channel_sent_listeners.append(self._on_mesh_channel_message) - self.logger.info("Registered for bot channel-sent events (bridge_bot_responses=true)") - - # Initialize message queues for each webhook - for webhook_urls in self.channel_webhooks.values(): - for webhook_url in webhook_urls: - self.message_queues[webhook_url] = deque() - self.send_times[webhook_url] = deque() - - # Start background queue processor task - self._queue_processor_task = asyncio.create_task(self._process_message_queues()) - - self._running = True - self.logger.info(f"Discord bridge service started (bridging {len(self.channel_webhooks)} channels)") - - async def on_transport_reconnected(self) -> None: - """Re-subscribe to channel messages on the new meshcore instance.""" - if not self._running or not getattr(self.bot, 'meshcore', None): - return - self._unsubscribe_all() - self._subscribe(self.bot.meshcore, EventType.CHANNEL_MSG_RECV, self._on_mesh_channel_message) - self.logger.info("Discord bridge re-subscribed to CHANNEL_MSG_RECV after transport reconnect") - - async def stop(self) -> None: - """Stop the Discord bridge service. - - Cleans up HTTP session and event handlers. - """ - self.logger.info("Stopping Discord bridge service...") - self._running = False - self._unsubscribe_all() - - # Unregister bot channel-sent listener - if getattr(self.bot, 'channel_sent_listeners', None) is not None: - with contextlib.suppress(ValueError): - self.bot.channel_sent_listeners.remove(self._on_mesh_channel_message) - - # Cancel background tasks - await self._cancel_tasks(self._queue_processor_task) - - # Close aiohttp session - if self.http_session: - await self.http_session.close() - self.http_session = None - - self.logger.info("Discord bridge service stopped") - - async def _on_mesh_channel_message(self, event, metadata=None) -> None: - """Handle incoming mesh channel messages. - - Posts messages to corresponding Discord channels via webhooks. - DMs are explicitly ignored for privacy. - - Args: - event: The MeshCore event object containing the message payload. - metadata: Optional metadata dictionary associated with the event. - """ - try: - # Copy payload immediately to avoid segfault if event is freed - payload = copy.deepcopy(event.payload) if hasattr(event, 'payload') else None - if payload is None: - self.logger.warning("Channel message event has no payload") - return - - # Extract channel index and convert to channel name - channel_idx = payload.get('channel_idx', 0) - channel_name = self.bot.channel_manager.get_channel_name(channel_idx) - - # Extract sender and text - # Sender is embedded in the text (format: "sender: message") - text = payload.get('text', '') - sender = 'Unknown' - - # Try to extract sender from text - if ':' in text and not text.startswith('http'): - parts = text.split(':', 1) - sender = parts[0].strip() - # Don't modify text - keep it as is with sender included - - # NEVER bridge DMs (double-check for safety) - if not channel_name or channel_name.lower() in ['dm', 'direct', 'private']: - self.logger.debug("Ignoring DM (DMs are never bridged)") - return - - # Check if this channel is configured for bridging (case-insensitive) - webhook_urls = None - for config_channel, url in self.channel_webhooks.items(): - if config_channel.lower() == channel_name.lower(): - webhook_urls = url - break - - if not webhook_urls: - self.logger.debug(f"Channel '{channel_name}' not configured for Discord bridge") - return - - # Extract sender and message for better Discord formatting - # Format the message for better visual separation - if ':' in text and not text.startswith('http'): - # Split on first colon to separate sender from message - parts = text.split(':', 1) - sender_name = parts[0].strip() - message_text = parts[1].strip() if len(parts) > 1 else text - else: - # No clear sender format, use whole text - sender_name = sender # From earlier extraction - message_text = text - - # Clean up MeshCore @ mentions: @[username] → **@username** - message_text = self._format_mentions(message_text) - message_text = neutralize_discord_mention_content(message_text) - - # Profanity filter: drop (don't bridge), censor (replace with ****), or off - if self.filter_profanity == 'drop': - if contains_profanity(sender_name, self.logger) or contains_profanity(message_text, self.logger): - self.logger.debug(f"Discord bridge: dropping message with profanity from [{channel_name}]") - return - elif self.filter_profanity == 'censor': - sender_name = censor(sender_name, self.logger) - message_text = censor(message_text, self.logger) - - # Queue message for posting (with rate limiting and retry logic) - # Fan out to all configured webhooks for this channel - for webhook_url in webhook_urls: - await self._queue_message(webhook_url, message_text, channel_name, sender_name) - - except Exception as e: - self.logger.error(f"Error handling mesh channel message: {e}", exc_info=True) - async def _queue_message(self, webhook_url: str, message: str, channel_name: str, sender_name: Optional[str] = None) -> None: """Queue a message for posting to Discord webhook. @@ -516,108 +416,6 @@ class DiscordBridgeService(BaseServicePlugin): except Exception as e: self.logger.error(f"Failed to queue message for Discord webhook [{channel_name}]: {e}", exc_info=True) - async def _process_message_queues(self) -> None: - """Background task to process message queues with rate limiting and retries. - - Processes messages from queues, respecting rate limits and retrying failed messages. - """ - while self._running: - try: - current_time = time.time() - - # Process each webhook's queue - for webhook_url, queue in list(self.message_queues.items()): - if not queue: - continue - - # Clean up old send times (outside rate limit window) - if webhook_url in self.send_times: - send_times = self.send_times[webhook_url] - while send_times and (current_time - send_times[0]) > self.rate_limit_window: - send_times.popleft() - - # Check if we can send (proactive rate limiting) - can_send = True - if webhook_url in self.send_times: - recent_sends = len(self.send_times[webhook_url]) - if recent_sends >= self.rate_limit_max: - can_send = False - # Calculate wait time until oldest message expires - oldest_send = self.send_times[webhook_url][0] - wait_time = (oldest_send + self.rate_limit_window) - current_time - if wait_time > 0: - self.logger.debug(f"Rate limit throttling [{queue[0].channel_name}]: waiting {wait_time:.1f}s") - - if not can_send: - continue - - # Find next message ready to be sent (not waiting for retry delay) - queued_msg = None - for msg in queue: - if current_time >= msg.next_retry_at: - queued_msg = msg - break - - # If no message is ready, skip this webhook - if queued_msg is None: - continue - - # Check if message is too old - age = current_time - queued_msg.first_queued - if age > self.max_queue_age: - # Remove old message from queue - queue.remove(queued_msg) - self.logger.warning( - f"Dropping old message from queue [{queued_msg.channel_name}]: " - f"age {age:.1f}s exceeds max {self.max_queue_age}s" - ) - continue - - # Try to send the message - success = await self._post_to_webhook( - queued_msg.webhook_url, - queued_msg.payload, - queued_msg.channel_name, - queued_msg - ) - - if success: - # Success - remove from queue - queue.remove(queued_msg) - # Track send time for rate limiting - if webhook_url not in self.send_times: - self.send_times[webhook_url] = deque() - self.send_times[webhook_url].append(current_time) - else: - # Failed - increment retry count and schedule retry - queued_msg.retry_count += 1 - if queued_msg.retry_count > self.max_retries: - # Max retries exceeded - drop message - queue.remove(queued_msg) - self.logger.error( - f"Dropping message after {self.max_retries} retries " - f"[{queued_msg.channel_name}]: {queued_msg.payload['content'][:50]}..." - ) - else: - # Calculate exponential backoff delay - delay = self.retry_delay_base * (2 ** (queued_msg.retry_count - 1)) - queued_msg.next_retry_at = current_time + delay - self.logger.debug( - f"Message failed, will retry in {delay:.1f}s " - f"(attempt {queued_msg.retry_count}/{self.max_retries}) " - f"[{queued_msg.channel_name}]" - ) - # Message stays in queue, will be retried later - - # Small delay to prevent tight loop - await asyncio.sleep(0.1) - - except asyncio.CancelledError: - break - except Exception as e: - self.logger.error(f"Error in message queue processor: {e}", exc_info=True) - await asyncio.sleep(1.0) # Wait a bit before retrying on error - async def _post_to_webhook(self, webhook_url: str, payload: dict[str, Any], channel_name: str, queued_msg: Optional[QueuedMessage] = None) -> bool: """Post message to Discord webhook. diff --git a/modules/service_plugins/telegram_bridge_service.py b/modules/service_plugins/telegram_bridge_service.py index df4646c..ef40d84 100644 --- a/modules/service_plugins/telegram_bridge_service.py +++ b/modules/service_plugins/telegram_bridge_service.py @@ -5,7 +5,6 @@ Posts MeshCore channel messages to Telegram via the Bot API (one-way, read-only) """ import asyncio -import copy import html import os import re @@ -14,8 +13,6 @@ from collections import deque from dataclasses import dataclass from typing import Any, Optional -from meshcore import EventType - try: import aiohttp AIOHTTP_AVAILABLE = True @@ -30,10 +27,10 @@ except ImportError: requests = None # type: ignore[assignment] REQUESTS_AVAILABLE = False -import contextlib -from ..profanity_filter import censor, contains_profanity -from .base_service import BaseServicePlugin +from ..profanity_filter import censor, contains_profanity # noqa: F401 re-exported +from .base_service import BaseServicePlugin # noqa: F401 re-exported +from .channel_bridge_utils import ChannelBridgeBase # Telegram API TELEGRAM_API_BASE = "https://api.telegram.org/bot" @@ -58,7 +55,7 @@ class QueuedMessage: self.next_retry_at = time.time() -class TelegramBridgeService(BaseServicePlugin): +class TelegramBridgeService(ChannelBridgeBase): """Telegram bridge service. Posts MeshCore channel messages to Telegram channels/groups via the Bot API. @@ -161,6 +158,58 @@ class TelegramBridgeService(BaseServicePlugin): "Add bridge. = in [TelegramBridge]" ) + bridge_label = "Telegram" + payload_text_key = "text" + old_message_log = "Dropping old message from queue [{channel}]: age {age:.1f}s > {max_age}s" + retry_log = "Message failed, retry in {delay:.1f}s ({retry}/{max_retries}) [{channel}]" + processor_error_log = "Error in Telegram queue processor: {error}" + + @property + def _bridge_mappings(self) -> dict[str, str]: + return self.channel_chat_ids + + def _open_http_session(self) -> None: + if AIOHTTP_AVAILABLE: + self.http_session = aiohttp.ClientSession() + else: + self.logger.debug("Using requests for HTTP (fallback)") + + def _init_queues(self) -> None: + for chat_id in self.channel_chat_ids.values(): + self.message_queues[chat_id] = [] + self.send_times[chat_id] = deque() + + def _targets_for(self, channel_name: str) -> Optional[str]: + # Normalize: strip leading # and compare case-insensitively so bridge.HowlTest matches #howltest + channel_key = channel_name.lstrip('#').lower() + for config_channel, cid in self.channel_chat_ids.items(): + if config_channel.lstrip('#').lower() == channel_key: + return cid + return None + + async def _deliver(self, targets: str, sender_name: str, message_text: str, channel_name: str) -> None: + full_text = self._build_message_text(sender_name, message_text, channel_name) + full_text = self._truncate_text(full_text) + await self._queue_message(targets, full_text, channel_name) + + def _throttled(self, key: str, queue: Any, current_time: float) -> bool: + # Enforce min interval per chat + if key in self.send_times: + st = self.send_times[key] + while st and (current_time - st[0]) > self.rate_limit_min_interval: + st.popleft() + if st and (current_time - st[-1]) < self.rate_limit_min_interval: + return True + return False + + async def _send_queued(self, queued_msg: QueuedMessage) -> bool: + return await self._send_to_telegram( + queued_msg.chat_id, + queued_msg.payload, + queued_msg.channel_name, + queued_msg, + ) + def _load_channel_mappings(self) -> None: """Load bridge. = chat_id from config.""" if not self.bot.config.has_section('TelegramBridge'): @@ -220,124 +269,6 @@ class TelegramBridgeService(BaseServicePlugin): self.logger.debug(f"Truncating message from {len(text)} to {self.max_message_length} chars") return text[: self.max_message_length - 1].rstrip() + "…" - async def start(self) -> None: - if not self.enabled: - self.logger.info("Telegram bridge service is disabled") - return - if not self.channel_chat_ids: - self.logger.warning("Telegram bridge enabled but no channels configured") - return - - self.logger.info("Starting Telegram bridge service...") - if AIOHTTP_AVAILABLE: - self.http_session = aiohttp.ClientSession() - else: - self.logger.debug("Using requests for HTTP (fallback)") - - if hasattr(self.bot, 'meshcore') and self.bot.meshcore: - self._subscribe(self.bot.meshcore, EventType.CHANNEL_MSG_RECV, self._on_mesh_channel_message) - self.logger.info("Subscribed to CHANNEL_MSG_RECV events") - else: - self.logger.error("Cannot subscribe to events - meshcore not available") - if self.http_session is not None: - await self.http_session.close() - self.http_session = None - return - - # Register for bot-sent channel messages so bot responses are bridged too - if self.bridge_bot_responses and getattr(self.bot, 'channel_sent_listeners', None) is not None: - self.bot.channel_sent_listeners.append(self._on_mesh_channel_message) - self.logger.info("Registered for bot channel-sent events (bridge_bot_responses=true)") - - for chat_id in self.channel_chat_ids.values(): - self.message_queues[chat_id] = [] - self.send_times[chat_id] = deque() - - self._queue_processor_task = asyncio.create_task(self._process_message_queues()) - self._running = True - self.logger.info( - f"Telegram bridge service started (bridging {len(self.channel_chat_ids)} channels)" - ) - - async def on_transport_reconnected(self) -> None: - """Re-subscribe to channel messages on the new meshcore instance.""" - if not self._running or not getattr(self.bot, 'meshcore', None): - return - self._unsubscribe_all() - self._subscribe(self.bot.meshcore, EventType.CHANNEL_MSG_RECV, self._on_mesh_channel_message) - self.logger.info("Telegram bridge re-subscribed to CHANNEL_MSG_RECV after transport reconnect") - - async def stop(self) -> None: - self.logger.info("Stopping Telegram bridge service...") - self._running = False - self._unsubscribe_all() - - # Unregister bot channel-sent listener - if getattr(self.bot, 'channel_sent_listeners', None) is not None: - with contextlib.suppress(ValueError): - self.bot.channel_sent_listeners.remove(self._on_mesh_channel_message) - - await self._cancel_tasks(self._queue_processor_task) - if self.http_session: - await self.http_session.close() - self.http_session = None - self.logger.info("Telegram bridge service stopped") - - async def _on_mesh_channel_message(self, event, metadata=None) -> None: - try: - payload = copy.deepcopy(event.payload) if hasattr(event, 'payload') else None - if payload is None: - self.logger.warning("Channel message event has no payload") - return - - channel_idx = payload.get('channel_idx', 0) - channel_name = self.bot.channel_manager.get_channel_name(channel_idx) - text = payload.get('text', '') - sender = 'Unknown' - - if ':' in text and not text.startswith('http'): - parts = text.split(':', 1) - sender = parts[0].strip() - - if not channel_name or channel_name.lower() in ('dm', 'direct', 'private'): - self.logger.debug("Ignoring DM (DMs are never bridged)") - return - - chat_id = None - # Normalize: strip leading # and compare case-insensitively so bridge.HowlTest matches #howltest - channel_key = channel_name.lstrip('#').lower() - for config_channel, cid in self.channel_chat_ids.items(): - if config_channel.lstrip('#').lower() == channel_key: - chat_id = cid - break - if not chat_id: - self.logger.debug(f"Channel '{channel_name}' not configured for Telegram bridge") - return - - if ':' in text and not text.startswith('http'): - parts = text.split(':', 1) - sender_name = parts[0].strip() - message_text = parts[1].strip() if len(parts) > 1 else text - else: - sender_name = sender - message_text = text - - # Profanity filter: drop (don't bridge), censor (replace with ****), or off - if self.filter_profanity == 'drop': - if contains_profanity(sender_name, self.logger) or contains_profanity(message_text, self.logger): - self.logger.debug(f"Telegram bridge: dropping message with profanity from [{channel_name}]") - return - elif self.filter_profanity == 'censor': - sender_name = censor(sender_name, self.logger) - message_text = censor(message_text, self.logger) - - full_text = self._build_message_text(sender_name, message_text, channel_name) - full_text = self._truncate_text(full_text) - - await self._queue_message(chat_id, full_text, channel_name) - except Exception as e: - self.logger.error(f"Error handling mesh channel message: {e}", exc_info=True) - async def _queue_message(self, chat_id: str, text: str, channel_name: str) -> None: payload: dict[str, Any] = { "chat_id": chat_id, @@ -358,72 +289,6 @@ class TelegramBridgeService(BaseServicePlugin): self.message_queues[chat_id].append(queued) self.logger.debug(f"Queued message for Telegram [{channel_name}]: {text[:50]}...") - async def _process_message_queues(self) -> None: - while self._running: - try: - current_time = time.time() - for chat_id, queue in list(self.message_queues.items()): - if not queue: - continue - # Enforce min interval per chat - if chat_id in self.send_times: - st = self.send_times[chat_id] - while st and (current_time - st[0]) > self.rate_limit_min_interval: - st.popleft() - if st and (current_time - st[-1]) < self.rate_limit_min_interval: - continue - - queued_msg = None - for msg in queue: - if current_time >= msg.next_retry_at: - queued_msg = msg - break - if queued_msg is None: - continue - - age = current_time - queued_msg.first_queued - if age > self.max_queue_age: - queue.remove(queued_msg) - self.logger.warning( - f"Dropping old message from queue [{queued_msg.channel_name}]: " - f"age {age:.1f}s > {self.max_queue_age}s" - ) - continue - - success = await self._send_to_telegram( - queued_msg.chat_id, - queued_msg.payload, - queued_msg.channel_name, - queued_msg, - ) - if success: - queue.remove(queued_msg) - if chat_id not in self.send_times: - self.send_times[chat_id] = deque() - self.send_times[chat_id].append(current_time) - else: - queued_msg.retry_count += 1 - if queued_msg.retry_count > self.max_retries: - queue.remove(queued_msg) - self.logger.error( - f"Dropping message after {self.max_retries} retries " - f"[{queued_msg.channel_name}]: {queued_msg.payload['text'][:50]}..." - ) - else: - delay = self.retry_delay_base * (2 ** (queued_msg.retry_count - 1)) - queued_msg.next_retry_at = current_time + delay - self.logger.debug( - f"Message failed, retry in {delay:.1f}s " - f"({queued_msg.retry_count}/{self.max_retries}) [{queued_msg.channel_name}]" - ) - - await asyncio.sleep(0.1) - except asyncio.CancelledError: - break - except Exception as e: - self.logger.error(f"Error in Telegram queue processor: {e}", exc_info=True) - await asyncio.sleep(1.0) - async def _send_to_telegram( self, chat_id: str,