meshcore-bot/modules/web_viewer/live_stream.py

"""Live updates for the web viewer: SocketIO broadcasts to subscribed clients,
the bot log tail, and the database poll that feeds the live command, packet and
message streams."""

from __future__ import annotations

import os
import re
from contextlib import closing
from pathlib import Path
from typing import Any

from modules.utils import resolve_path

# colorlog and other handlers write ANSI SGR sequences; strip for web /logs display
_ANSI_ESCAPE_RE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")


def _strip_ansi_codes(text: str) -> str:
    """Remove ANSI color and reset codes from log lines for SocketIO web clients."""
    return _ANSI_ESCAPE_RE.sub("", text)


class LiveStreamMixin:
    """Mixed into BotDataViewer."""

    _clients_lock: Any
    _config_base: Any
    config: Any
    connected_clients: Any
    db_path: Any
    logger: Any
    socketio: Any

    def _subscribed_clients(self, flag: str) -> list:
        """Sids of connected clients with ``flag`` set, read under the clients lock."""
        with self._clients_lock:
            return [
                client_id for client_id, client_info in self.connected_clients.items()
                if client_info.get(flag, False)
            ]

    def _handle_command_data(self, command_data):
        """Handle incoming command data from bot"""
        try:
            # Broadcast to subscribed clients
            subscribed_clients = self._subscribed_clients('subscribed_commands')

            if subscribed_clients:
                self.socketio.emit('command_data', command_data, room=None)
                self.logger.debug(f"Broadcasted command data to {len(subscribed_clients)} clients")
        except Exception as e:
            self.logger.error(f"Error handling command data: {e}")

    def _handle_packet_data(self, packet_data):
        """Handle incoming packet data from bot"""
        try:
            # Broadcast to subscribed clients
            subscribed_clients = self._subscribed_clients('subscribed_packets')

            if subscribed_clients:
                self.socketio.emit('packet_data', packet_data, room=None)
                self.logger.debug(f"Broadcasted packet data to {len(subscribed_clients)} clients")
        except Exception as e:
            self.logger.error(f"Error handling packet data: {e}")

    def _handle_mesh_edge_data(self, edge_data):
        """Handle incoming mesh edge data from bot"""
        try:
            # Broadcast to subscribed clients
            subscribed_clients = self._subscribed_clients('subscribed_mesh')

            if subscribed_clients:
                event_type = 'mesh_edge_added' if edge_data.get('is_new', False) else 'mesh_edge_updated'
                self.socketio.emit(event_type, edge_data, room=None)
        except Exception as e:
            self.logger.error(f"Error handling mesh edge data: {e}", exc_info=True)

    def _handle_mesh_node_data(self, node_data):
        """Handle incoming mesh node data from bot"""
        try:
            # Broadcast to subscribed clients
            subscribed_clients = self._subscribed_clients('subscribed_mesh')

            if subscribed_clients:
                self.socketio.emit('mesh_node_added', node_data, room=None)
        except Exception as e:
            self.logger.error(f"Error handling mesh node data: {e}", exc_info=True)

    def _handle_message_data(self, msg_data):
        """Broadcast a captured channel message to subscribed clients."""
        try:
            subscribed_clients = self._subscribed_clients('subscribed_messages')
            if subscribed_clients:
                self.socketio.emit('message_data', msg_data, room=None)
        except Exception as e:
            self.logger.error(f"Error handling message data: {e}")

    def _handle_log_line(self, line: str) -> None:
        """Broadcast a log line to clients subscribed to the log stream."""
        try:
            subscribed = self._subscribed_clients('subscribed_logs')
            if subscribed:
                self.socketio.emit(
                    'log_line', {'line': _strip_ansi_codes(line.rstrip())}, room=None
                )
        except Exception as e:
            self.logger.error(f"Error broadcasting log line: {e}")

    def _start_log_tailing(self) -> None:
        """Start a background thread that tails the bot log file and emits SocketIO events."""
        import os
        import threading

        log_file = ''
        try:
            log_file = self.config.get('Logging', 'log_file', fallback='').strip()
            if log_file:
                log_file = str(resolve_path(log_file, self._config_base))
        except Exception:
            pass

        if not log_file:
            self.logger.info("Log tailing disabled: no log_file configured")
            return

        def tail_log():
            import time as _time
            self.logger.info(f"Log tail thread started: {log_file}")
            pos = 0
            # Start at end of file so we only stream new lines
            try:
                pos = os.path.getsize(log_file)
            except OSError:
                pass
            while True:
                try:
                    if not os.path.exists(log_file):
                        _time.sleep(2)
                        continue
                    current_size = os.path.getsize(log_file)
                    if current_size < pos:
                        # File rotated — start from beginning
                        pos = 0
                    if current_size > pos:
                        with open(log_file, encoding='utf-8', errors='replace') as fh:
                            fh.seek(pos)
                            for line in fh:
                                self._handle_log_line(line)
                            pos = fh.tell()
                except Exception as e:
                    self.logger.debug(f"Log tail error: {e}")
                _time.sleep(1)

        tail_thread = threading.Thread(target=tail_log, daemon=True)
        tail_thread.start()
        self.logger.info("Log tailing started")

    def _start_database_polling(self):
        """Start background thread to poll database for new data"""
        import threading

        def poll_database():
            import time as _time
            last_timestamp = _time.time() - 300  # start 5 min back; subscribe handlers replay full history
            consecutive_errors = 0
            max_consecutive_errors = 10

            while True:
                try:
                    import json
                    import sqlite3
                    import time

                    # Subscription handlers replay recent history themselves.
                    # With no live command/packet/message subscribers there is
                    # nothing to broadcast, so avoid opening SQLite and decoding
                    # every packet-stream row merely to discard it.
                    if not self._has_live_stream_subscribers():
                        last_timestamp = time.time()
                        # An idle period is not a failure. Without this, an
                        # error burst before the last subscriber left would
                        # still be counted against the first poll after the
                        # next one arrives, mis-escalating its log level and
                        # backoff.
                        consecutive_errors = 0
                        time.sleep(2.0)
                        continue

                    # Check if database file exists and is accessible
                    db_file = Path(self.db_path)
                    if not db_file.exists():
                        consecutive_errors += 1
                        if consecutive_errors == 1 or consecutive_errors % 10 == 0:
                            self.logger.warning(f"Database file does not exist: {self.db_path}")
                        time.sleep(5)
                        continue

                    if not os.access(self.db_path, os.R_OK):
                        consecutive_errors += 1
                        if consecutive_errors == 1 or consecutive_errors % 10 == 0:
                            self.logger.warning(f"Database file is not readable: {self.db_path}")
                        time.sleep(5)
                        continue

                    # Connect to database with timeout to prevent hanging
                    try:
                        with closing(sqlite3.connect(self.db_path, timeout=60, check_same_thread=False)) as conn:
                            conn.row_factory = sqlite3.Row
                            cursor = conn.cursor()

                            # Get new data since last poll
                            cursor.execute('''
                                SELECT timestamp, data, type FROM packet_stream
                                WHERE timestamp > ?
                                ORDER BY timestamp ASC
                            ''', (last_timestamp,))

                            rows = cursor.fetchall()

                            # Process new data
                            for row in rows:
                                try:
                                    data_json = row[1]
                                    data_type = row[2]
                                    data = json.loads(data_json)
                                    if isinstance(data, dict):
                                        data['stream_ts'] = row[0]

                                    # Broadcast based on type
                                    if data_type == 'command':
                                        self._handle_command_data(data)
                                    elif data_type == 'packet':
                                        self._handle_packet_data(data)
                                    elif data_type == 'routing':
                                        self._handle_packet_data(data)  # Treat routing as packet data
                                    elif data_type == 'message':
                                        self._handle_message_data(data)

                                except Exception as e:
                                    self.logger.warning(f"Error processing database data: {e}")

                            # Update last timestamp
                            if rows:
                                last_timestamp = rows[-1][0]

                            # Reset error counter on success
                            consecutive_errors = 0
                    except sqlite3.OperationalError as conn_error:
                        error_msg = str(conn_error)
                        if "locked" in error_msg.lower() or "database is locked" in error_msg.lower():
                            consecutive_errors += 1
                            if consecutive_errors == 1 or consecutive_errors % 10 == 0:
                                self.logger.warning(f"Database is locked, waiting: {self.db_path}")
                            time.sleep(2)
                            continue
                        raise  # Re-raise non-locked OperationalErrors for outer handler to log/backoff

                    # Sleep before next poll (back off to reduce lock contention with bot writes)
                    time.sleep(2.0)  # Poll every 2s

                except sqlite3.OperationalError as e:
                    consecutive_errors += 1
                    error_msg = str(e)

                    # Provide more diagnostic information on first error or periodic errors
                    if consecutive_errors == 1 or consecutive_errors % 10 == 0:
                        db_file = Path(self.db_path)
                        exists = db_file.exists()
                        readable = os.access(self.db_path, os.R_OK) if exists else False
                        writable = os.access(self.db_path, os.W_OK) if exists else False
                        self.logger.error(
                            f"Database polling error (attempt {consecutive_errors}): {error_msg}\n"
                            f"  Path: {self.db_path}\n"
                            f"  Exists: {exists}\n"
                            f"  Readable: {readable}\n"
                            f"  Writable: {writable}"
                        )

                    # Log at appropriate level based on error frequency
                    if consecutive_errors >= max_consecutive_errors:
                        if consecutive_errors == max_consecutive_errors:
                            self.logger.error(f"Database polling persistent error (attempt {consecutive_errors}): {error_msg}")
                        # Exponential backoff for persistent errors
                        time.sleep(min(60, 2 ** min(consecutive_errors - max_consecutive_errors, 5)))
                    elif consecutive_errors > 3:
                        self.logger.warning(f"Database polling error (attempt {consecutive_errors}): {error_msg}")
                        time.sleep(5)  # Wait longer on repeated errors
                    else:
                        self.logger.debug(f"Database polling error (attempt {consecutive_errors}): {error_msg}")
                        time.sleep(1)  # Wait longer on error

                except Exception as e:
                    consecutive_errors += 1
                    if consecutive_errors >= max_consecutive_errors:
                        if consecutive_errors == max_consecutive_errors:
                            self.logger.error(f"Database polling unexpected error (attempt {consecutive_errors}): {e}", exc_info=True)
                        time.sleep(min(60, 2 ** min(consecutive_errors - max_consecutive_errors, 5)))
                    else:
                        self.logger.warning(f"Database polling unexpected error (attempt {consecutive_errors}): {e}")
                        time.sleep(2)


        # Start polling thread
        polling_thread = threading.Thread(target=poll_database, daemon=True)
        polling_thread.start()
        self.logger.info("Database polling started")

    def _has_live_stream_subscribers(self) -> bool:
        """Return whether any client consumes a DB-backed live stream."""
        subscription_keys = (
            'subscribed_commands',
            'subscribed_packets',
            'subscribed_messages',
        )
        with self._clients_lock:
            return any(
                any(client.get(key, False) for key in subscription_keys)
                for client in self.connected_clients.values()
            )