diff --git a/modules/db_manager.py b/modules/db_manager.py index 35005e4..17590e6 100644 --- a/modules/db_manager.py +++ b/modules/db_manager.py @@ -449,6 +449,44 @@ class DBManager: """Set bot start time in metadata""" self.set_metadata('start_time', str(start_time)) + def _apply_sqlite_pragmas(self, conn: sqlite3.Connection, for_web_viewer: bool = False) -> None: + config = getattr(self.bot, "config", None) + section = "Web_Viewer" if for_web_viewer else "Bot" + + foreign_keys = True + default_busy_timeout_ms = 60000 if for_web_viewer else 30000 + busy_timeout_ms: Any = default_busy_timeout_ms + journal_mode = "WAL" + + try: + if config is not None: + foreign_keys = config.getboolean(section, "sqlite_foreign_keys", fallback=True) + busy_timeout_ms = config.getint( + section, + "sqlite_busy_timeout_ms", + fallback=default_busy_timeout_ms, + ) + journal_mode = config.get(section, "sqlite_journal_mode", fallback=journal_mode).strip() or journal_mode + except Exception: + # Config parsing should never prevent DB access. + pass + + # Be resilient to mocks / unexpected types. + try: + busy_timeout_ms = int(busy_timeout_ms) + except (TypeError, ValueError): + busy_timeout_ms = default_busy_timeout_ms + foreign_keys = bool(foreign_keys) + journal_mode = str(journal_mode).strip() or "WAL" + + try: + conn.execute(f"PRAGMA foreign_keys={'ON' if foreign_keys else 'OFF'}") + conn.execute(f"PRAGMA busy_timeout={busy_timeout_ms}") + conn.execute(f"PRAGMA journal_mode={journal_mode}") + except sqlite3.OperationalError: + # journal_mode can fail if DB is locked; others are best-effort. + pass + @contextmanager def connection(self) -> Generator[sqlite3.Connection, None, None]: """Context manager that yields a configured connection and closes it on exit. @@ -456,6 +494,7 @@ class DBManager: """ conn = sqlite3.connect(str(self.db_path), timeout=30.0) conn.row_factory = sqlite3.Row + self._apply_sqlite_pragmas(conn, for_web_viewer=False) try: yield conn finally: @@ -471,6 +510,7 @@ class DBManager: """ conn = sqlite3.connect(str(self.db_path), timeout=30.0) conn.row_factory = sqlite3.Row + self._apply_sqlite_pragmas(conn, for_web_viewer=False) return conn def set_system_health(self, health_data: dict[str, Any]) -> None: diff --git a/modules/db_migrations.py b/modules/db_migrations.py index 5295eac..23a8877 100644 --- a/modules/db_migrations.py +++ b/modules/db_migrations.py @@ -15,6 +15,7 @@ Never modify or remove an existing migration — add a new one instead. """ import logging +import re import sqlite3 from typing import Callable @@ -22,10 +23,28 @@ from typing import Callable # Helper utilities # --------------------------------------------------------------------------- +VALID_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def _validate_ident(name: str, kind: str) -> None: + if not VALID_IDENT.match(name): + raise ValueError(f"Invalid {kind} identifier: {name!r}") + + +def _table_exists(cursor: sqlite3.Cursor, table: str) -> bool: + _validate_ident(table, "table") + cursor.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", + (table,), + ) + return cursor.fetchone() is not None + def _column_exists(cursor: sqlite3.Cursor, table: str, column: str) -> bool: """Return True if *column* already exists in *table*.""" - cursor.execute(f"PRAGMA table_info({table})") + _validate_ident(table, "table") + _validate_ident(column, "column") + cursor.execute(f'PRAGMA table_info("{table}")') return any(row[1] == column for row in cursor.fetchall()) @@ -33,6 +52,8 @@ def _add_column( cursor: sqlite3.Cursor, table: str, column: str, definition: str ) -> None: """Add *column* to *table* if it does not already exist.""" + _validate_ident(table, "table") + _validate_ident(column, "column") if not _column_exists(cursor, table, column): cursor.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}") @@ -195,6 +216,257 @@ def _m0006_channel_operations_payload_data(cursor: sqlite3.Cursor) -> None: _add_column(cursor, "channel_operations", "payload_data", "TEXT") +# NOTE: Higher-numbered migrations can safely depend on tables created by other +# subsystems (e.g., repeater manager) by checking for table existence and then +# applying idempotent ALTER/CREATE INDEX statements. + + +def _m0007_packet_stream_table(cursor: sqlite3.Cursor) -> None: + """Create packet_stream table and indexes (shared DB with web viewer).""" + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS packet_stream ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + data TEXT NOT NULL, + type TEXT NOT NULL + ) + """ + ) + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_timestamp ON packet_stream(timestamp)" + ) + cursor.execute("CREATE INDEX IF NOT EXISTS idx_packet_stream_type ON packet_stream(type)") + + +def _m0008_repeater_tables_optional_columns(cursor: sqlite3.Cursor) -> None: + """Bring repeater/graph tables up to date if they exist.""" + # repeater_contacts: add location columns only if table exists + if _table_exists(cursor, "repeater_contacts"): + for column_name, column_def in [ + ("latitude", "REAL"), + ("longitude", "REAL"), + ("city", "TEXT"), + ("state", "TEXT"), + ("country", "TEXT"), + ]: + _add_column(cursor, "repeater_contacts", column_name, column_def) + + # complete_contact_tracking: path columns + is_starred and out_bytes_per_hop + if _table_exists(cursor, "complete_contact_tracking"): + for column_name, column_def in [ + ("out_path", "TEXT"), + ("out_path_len", "INTEGER"), + ("snr", "REAL"), + ("is_starred", "BOOLEAN DEFAULT 0"), + ("out_bytes_per_hop", "INTEGER"), + ]: + _add_column(cursor, "complete_contact_tracking", column_name, column_def) + + # observed_paths: packet_hash + bytes_per_hop + if _table_exists(cursor, "observed_paths"): + for column_name, column_def in [ + ("packet_hash", "TEXT"), + ("bytes_per_hop", "INTEGER"), + ]: + _add_column(cursor, "observed_paths", column_name, column_def) + + # mesh_connections: graph/viewer columns + if _table_exists(cursor, "mesh_connections"): + for column_name, column_def in [ + ("from_public_key", "TEXT"), + ("to_public_key", "TEXT"), + ("avg_hop_position", "REAL"), + ("geographic_distance", "REAL"), + ]: + _add_column(cursor, "mesh_connections", column_name, column_def) + + +def _m0009_repeater_optional_indexes(cursor: sqlite3.Cursor) -> None: + """Create optional indexes for repeater/graph tables if they exist.""" + if _table_exists(cursor, "unique_advert_packets"): + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_unique_advert_date_pubkey ON unique_advert_packets(date, public_key)" + ) + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_unique_advert_hash ON unique_advert_packets(packet_hash)" + ) + + if _table_exists(cursor, "mesh_connections"): + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_from_prefix ON mesh_connections(from_prefix)" + ) + cursor.execute("CREATE INDEX IF NOT EXISTS idx_to_prefix ON mesh_connections(to_prefix)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_last_seen ON mesh_connections(last_seen)") + + if _table_exists(cursor, "observed_paths"): + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_observed_paths_public_key ON observed_paths(public_key, packet_type)" + ) + if _column_exists(cursor, "observed_paths", "packet_hash"): + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_observed_paths_packet_hash ON observed_paths(packet_hash) WHERE packet_hash IS NOT NULL" + ) + cursor.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_advert_unique ON observed_paths(public_key, path_hex, packet_type) WHERE public_key IS NOT NULL" + ) + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_observed_paths_endpoints ON observed_paths(from_prefix, to_prefix, packet_type)" + ) + cursor.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_message_unique ON observed_paths(from_prefix, to_prefix, path_hex, packet_type) WHERE public_key IS NULL" + ) + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_observed_paths_last_seen ON observed_paths(last_seen)" + ) + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_observed_paths_type_seen ON observed_paths(packet_type, last_seen)" + ) + + +def _m0010_create_repeater_and_graph_tables(cursor: sqlite3.Cursor) -> None: + """Create repeater/graph tables used by the web viewer and repeater manager.""" + cursor.executescript( + """ + CREATE TABLE IF NOT EXISTS repeater_contacts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + public_key TEXT UNIQUE NOT NULL, + name TEXT NOT NULL, + device_type TEXT NOT NULL, + first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + contact_data TEXT, + latitude REAL, + longitude REAL, + city TEXT, + state TEXT, + country TEXT, + is_active BOOLEAN DEFAULT 1, + purge_count INTEGER DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS complete_contact_tracking ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + public_key TEXT NOT NULL, + name TEXT NOT NULL, + role TEXT NOT NULL, + device_type TEXT, + first_heard TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_heard TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + advert_count INTEGER DEFAULT 1, + latitude REAL, + longitude REAL, + city TEXT, + state TEXT, + country TEXT, + raw_advert_data TEXT, + signal_strength REAL, + snr REAL, + hop_count INTEGER, + is_currently_tracked BOOLEAN DEFAULT 0, + last_advert_timestamp TIMESTAMP, + location_accuracy REAL, + contact_source TEXT DEFAULT 'advertisement', + out_path TEXT, + out_path_len INTEGER, + out_bytes_per_hop INTEGER, + is_starred INTEGER DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS daily_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + date DATE NOT NULL, + public_key TEXT NOT NULL, + advert_count INTEGER DEFAULT 1, + first_advert_time TIMESTAMP, + last_advert_time TIMESTAMP, + UNIQUE(date, public_key) + ); + + CREATE TABLE IF NOT EXISTS unique_advert_packets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + date DATE NOT NULL, + public_key TEXT NOT NULL, + packet_hash TEXT NOT NULL, + first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE(date, public_key, packet_hash) + ); + + CREATE TABLE IF NOT EXISTS purging_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + action TEXT NOT NULL, + public_key TEXT NOT NULL, + name TEXT NOT NULL, + reason TEXT + ); + + CREATE TABLE IF NOT EXISTS mesh_connections ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + from_prefix TEXT NOT NULL, + to_prefix TEXT NOT NULL, + from_public_key TEXT, + to_public_key TEXT, + observation_count INTEGER DEFAULT 1, + first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + avg_hop_position REAL, + geographic_distance REAL, + UNIQUE(from_prefix, to_prefix) + ); + + CREATE TABLE IF NOT EXISTS observed_paths ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + public_key TEXT, + packet_hash TEXT, + from_prefix TEXT NOT NULL, + to_prefix TEXT NOT NULL, + path_hex TEXT NOT NULL, + path_length INTEGER NOT NULL, + bytes_per_hop INTEGER, + packet_type TEXT NOT NULL, + first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + observation_count INTEGER DEFAULT 1 + ); + """ + ) + + +def _m0011_repeater_and_graph_indexes(cursor: sqlite3.Cursor) -> None: + """Create indexes for repeater/graph tables (safe to run repeatedly).""" + cursor.executescript( + """ + CREATE INDEX IF NOT EXISTS idx_public_key ON repeater_contacts(public_key); + CREATE INDEX IF NOT EXISTS idx_device_type ON repeater_contacts(device_type); + CREATE INDEX IF NOT EXISTS idx_last_seen ON repeater_contacts(last_seen); + CREATE INDEX IF NOT EXISTS idx_is_active ON repeater_contacts(is_active); + + CREATE INDEX IF NOT EXISTS idx_complete_public_key ON complete_contact_tracking(public_key); + CREATE INDEX IF NOT EXISTS idx_complete_role ON complete_contact_tracking(role); + CREATE INDEX IF NOT EXISTS idx_complete_last_heard ON complete_contact_tracking(last_heard); + CREATE INDEX IF NOT EXISTS idx_complete_currently_tracked ON complete_contact_tracking(is_currently_tracked); + CREATE INDEX IF NOT EXISTS idx_complete_location ON complete_contact_tracking(latitude, longitude); + CREATE INDEX IF NOT EXISTS idx_complete_role_tracked ON complete_contact_tracking(role, is_currently_tracked); + + CREATE INDEX IF NOT EXISTS idx_unique_advert_date_pubkey ON unique_advert_packets(date, public_key); + CREATE INDEX IF NOT EXISTS idx_unique_advert_hash ON unique_advert_packets(packet_hash); + + CREATE INDEX IF NOT EXISTS idx_from_prefix ON mesh_connections(from_prefix); + CREATE INDEX IF NOT EXISTS idx_to_prefix ON mesh_connections(to_prefix); + CREATE INDEX IF NOT EXISTS idx_last_seen ON mesh_connections(last_seen); + + CREATE INDEX IF NOT EXISTS idx_observed_paths_public_key ON observed_paths(public_key, packet_type); + CREATE INDEX IF NOT EXISTS idx_observed_paths_packet_hash ON observed_paths(packet_hash) WHERE packet_hash IS NOT NULL; + CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_advert_unique ON observed_paths(public_key, path_hex, packet_type) WHERE public_key IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_observed_paths_endpoints ON observed_paths(from_prefix, to_prefix, packet_type); + CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_message_unique ON observed_paths(from_prefix, to_prefix, path_hex, packet_type) WHERE public_key IS NULL; + CREATE INDEX IF NOT EXISTS idx_observed_paths_last_seen ON observed_paths(last_seen); + CREATE INDEX IF NOT EXISTS idx_observed_paths_type_seen ON observed_paths(packet_type, last_seen); + """ + ) + + # --------------------------------------------------------------------------- # Migration registry — append new entries here, never remove or reorder. # --------------------------------------------------------------------------- @@ -208,6 +480,11 @@ MIGRATIONS: list[MigrationEntry] = [ (4, "channel_operations: result_data, processed_at", _m0004_channel_operations_result_processed), (5, "feed_message_queue: item_id, item_title, priority", _m0005_feed_message_queue_item_fields), (6, "channel_operations: payload_data", _m0006_channel_operations_payload_data), + (7, "packet_stream table for web viewer", _m0007_packet_stream_table), + (8, "optional repeater/graph columns", _m0008_repeater_tables_optional_columns), + (9, "optional repeater/graph indexes", _m0009_repeater_optional_indexes), + (10, "create repeater/graph tables", _m0010_create_repeater_and_graph_tables), + (11, "repeater/graph indexes", _m0011_repeater_and_graph_indexes), ] @@ -239,11 +516,24 @@ class MigrationRunner: applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) + # Legacy DBs may have been created without a uniqueness constraint. + self.conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_schema_version_version ON schema_version(version)" + ) - def _current_version(self) -> int: - cursor = self.conn.execute("SELECT MAX(version) FROM schema_version") - row = cursor.fetchone() - return row[0] if row[0] is not None else 0 + def _applied_versions(self) -> set[int]: + cursor = self.conn.execute("SELECT version FROM schema_version") + return {int(row[0]) for row in cursor.fetchall() if row and row[0] is not None} + + def _validate_versions(self, applied: set[int]) -> None: + known = {v for v, _, _ in MIGRATIONS} + unknown = sorted(v for v in applied if v not in known) + if unknown: + raise RuntimeError( + "Database schema is newer or inconsistent with this codebase. " + f"Unknown applied migration version(s): {unknown}. " + "Upgrade the bot to a newer version that includes these migrations." + ) def _apply(self, version: int, description: str, fn: Callable[[sqlite3.Cursor], None]) -> None: cursor = self.conn.cursor() @@ -257,13 +547,23 @@ class MigrationRunner: def run(self) -> None: """Apply all pending migrations in version order.""" self._ensure_version_table() - current = self._current_version() - pending = [(v, d, f) for v, d, f in MIGRATIONS if v > current] + applied = self._applied_versions() + self._validate_versions(applied) + pending = [(v, d, f) for v, d, f in MIGRATIONS if v not in applied] + pending.sort(key=lambda x: x[0]) if not pending: self.logger.debug("Database schema is up to date") return - for version, description, fn in pending: - self._apply(version, description, fn) + + try: + self.conn.execute("BEGIN IMMEDIATE") + for version, description, fn in pending: + self._apply(version, description, fn) + self.conn.commit() + except Exception: + self.conn.rollback() + raise + self.logger.info( f"Database migrations complete: {len(pending)} applied, " f"schema now at version {pending[-1][0]}" diff --git a/modules/repeater_manager.py b/modules/repeater_manager.py index 58ba1c0..9ce977d 100644 --- a/modules/repeater_manager.py +++ b/modules/repeater_manager.py @@ -26,9 +26,6 @@ class RepeaterManager: # Use the shared database manager self.db_manager = bot.db_manager - # Check for and handle database schema migration FIRST (before creating indexes) - self._migrate_database_schema() - # Initialize repeater-specific tables self._init_repeater_tables() @@ -50,162 +47,35 @@ class RepeaterManager: self.geocoding_cache_window = 60 # 1 minute window def _init_repeater_tables(self): - """Initialize repeater-specific database tables""" + """Ensure repeater-specific tables exist (created by migrations).""" try: - # Create repeater_contacts table - self.db_manager.create_table('repeater_contacts', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - public_key TEXT UNIQUE NOT NULL, - name TEXT NOT NULL, - device_type TEXT NOT NULL, - first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - contact_data TEXT, - latitude REAL, - longitude REAL, - city TEXT, - state TEXT, - country TEXT, - is_active BOOLEAN DEFAULT 1, - purge_count INTEGER DEFAULT 0 - ''') - - # Create complete_contact_tracking table for all heard contacts - self.db_manager.create_table('complete_contact_tracking', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - public_key TEXT NOT NULL, - name TEXT NOT NULL, - role TEXT NOT NULL, - device_type TEXT, - first_heard TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_heard TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - advert_count INTEGER DEFAULT 1, - latitude REAL, - longitude REAL, - city TEXT, - state TEXT, - country TEXT, - raw_advert_data TEXT, - signal_strength REAL, - snr REAL, - hop_count INTEGER, - is_currently_tracked BOOLEAN DEFAULT 0, - last_advert_timestamp TIMESTAMP, - location_accuracy REAL, - contact_source TEXT DEFAULT 'advertisement', - out_path TEXT, - out_path_len INTEGER, - is_starred INTEGER DEFAULT 0 - ''') - - # Create daily_stats table for daily statistics tracking - self.db_manager.create_table('daily_stats', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - date DATE NOT NULL, - public_key TEXT NOT NULL, - advert_count INTEGER DEFAULT 1, - first_advert_time TIMESTAMP, - last_advert_time TIMESTAMP, - UNIQUE(date, public_key) - ''') - - # Create unique_advert_packets table for tracking unique packet hashes - # This allows us to count unique advert packets (deduplicate by packet_hash) - self.db_manager.create_table('unique_advert_packets', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - date DATE NOT NULL, - public_key TEXT NOT NULL, - packet_hash TEXT NOT NULL, - first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - UNIQUE(date, public_key, packet_hash) - ''') - - # Create purging_log table for audit trail - self.db_manager.create_table('purging_log', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - action TEXT NOT NULL, - public_key TEXT NOT NULL, - name TEXT NOT NULL, - reason TEXT - ''') - - # Create mesh_connections table for graph-based path validation - self.db_manager.create_table('mesh_connections', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - from_prefix TEXT NOT NULL, - to_prefix TEXT NOT NULL, - from_public_key TEXT, - to_public_key TEXT, - observation_count INTEGER DEFAULT 1, - first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - avg_hop_position REAL, - geographic_distance REAL, - UNIQUE(from_prefix, to_prefix) - ''') - - # Create observed_paths table for storing complete paths from adverts and messages - self.db_manager.create_table('observed_paths', ''' - id INTEGER PRIMARY KEY AUTOINCREMENT, - public_key TEXT, - packet_hash TEXT, - from_prefix TEXT NOT NULL, - to_prefix TEXT NOT NULL, - path_hex TEXT NOT NULL, - path_length INTEGER NOT NULL, - bytes_per_hop INTEGER, - packet_type TEXT NOT NULL, - first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - observation_count INTEGER DEFAULT 1 - ''') - - # Create indexes for better performance with self.db_manager.connection() as conn: - cursor = conn.cursor() - cursor.execute('CREATE INDEX IF NOT EXISTS idx_public_key ON repeater_contacts(public_key)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_device_type ON repeater_contacts(device_type)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_last_seen ON repeater_contacts(last_seen)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_is_active ON repeater_contacts(is_active)') + required_tables = [ + "repeater_contacts", + "complete_contact_tracking", + "daily_stats", + "unique_advert_packets", + "purging_log", + "mesh_connections", + "observed_paths", + ] + missing = [] + for t in required_tables: + cur = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", + (t,), + ) + if cur.fetchone() is None: + missing.append(t) - # Indexes for contact tracking table - cursor.execute('CREATE INDEX IF NOT EXISTS idx_complete_public_key ON complete_contact_tracking(public_key)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_complete_role ON complete_contact_tracking(role)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_complete_last_heard ON complete_contact_tracking(last_heard)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_complete_currently_tracked ON complete_contact_tracking(is_currently_tracked)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_complete_location ON complete_contact_tracking(latitude, longitude)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_complete_role_tracked ON complete_contact_tracking(role, is_currently_tracked)') - - # Indexes for unique_advert_packets table - cursor.execute('CREATE INDEX IF NOT EXISTS idx_unique_advert_date_pubkey ON unique_advert_packets(date, public_key)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_unique_advert_hash ON unique_advert_packets(packet_hash)') - - # Indexes for mesh_connections table - cursor.execute('CREATE INDEX IF NOT EXISTS idx_from_prefix ON mesh_connections(from_prefix)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_to_prefix ON mesh_connections(to_prefix)') - cursor.execute('CREATE INDEX IF NOT EXISTS idx_last_seen ON mesh_connections(last_seen)') - - # Indexes for observed_paths table - # Index for advert path lookups by repeater (where public_key IS NOT NULL) - cursor.execute('CREATE INDEX IF NOT EXISTS idx_observed_paths_public_key ON observed_paths(public_key, packet_type)') - # Index for grouping paths by packet hash (same packet via different paths) - # Only create if packet_hash column exists (migration may have just added it) - cursor.execute("PRAGMA table_info(observed_paths)") - observed_paths_columns = [row[1] for row in cursor.fetchall()] - if 'packet_hash' in observed_paths_columns: - cursor.execute('CREATE INDEX IF NOT EXISTS idx_observed_paths_packet_hash ON observed_paths(packet_hash) WHERE packet_hash IS NOT NULL') - # Unique index for adverts: one entry per repeater per unique path - cursor.execute('CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_advert_unique ON observed_paths(public_key, path_hex, packet_type) WHERE public_key IS NOT NULL') - # Index for general path lookups by endpoints - cursor.execute('CREATE INDEX IF NOT EXISTS idx_observed_paths_endpoints ON observed_paths(from_prefix, to_prefix, packet_type)') - # Unique index for messages: one entry per unique path between endpoints - cursor.execute('CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_message_unique ON observed_paths(from_prefix, to_prefix, path_hex, packet_type) WHERE public_key IS NULL') - # Index for recency filtering - cursor.execute('CREATE INDEX IF NOT EXISTS idx_observed_paths_last_seen ON observed_paths(last_seen)') - # Index for type-specific queries - cursor.execute('CREATE INDEX IF NOT EXISTS idx_observed_paths_type_seen ON observed_paths(packet_type, last_seen)') - conn.commit() + if missing: + msg = ( + "Missing repeater/graph database tables: " + + ", ".join(missing) + + ". Run the bot once to apply migrations." + ) + self.logger.error(msg) + raise RuntimeError(msg) self.logger.info("Repeater contacts database initialized successfully") @@ -213,101 +83,6 @@ class RepeaterManager: self.logger.error(f"Failed to initialize repeater database: {e}") raise - def _migrate_database_schema(self): - """Handle database schema migration for existing installations. - Each table is migrated in isolation so one missing table or error does not block the rest. - Important for web viewer: migration runs when RepeaterManager is created, so contacts page - works for users who open the viewer without having started the bot after upgrade. - """ - with self.db_manager.connection() as conn: - cursor = conn.cursor() - - # repeater_contacts: add location columns only if table exists - try: - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='repeater_contacts'") - if cursor.fetchone(): - cursor.execute("PRAGMA table_info(repeater_contacts)") - columns = [row[1] for row in cursor.fetchall()] - for column_name, column_type in [ - ('latitude', 'REAL'), ('longitude', 'REAL'), ('city', 'TEXT'), - ('state', 'TEXT'), ('country', 'TEXT') - ]: - if column_name not in columns: - self.logger.info(f"Adding missing column to repeater_contacts: {column_name}") - cursor.execute(f"ALTER TABLE repeater_contacts ADD COLUMN {column_name} {column_type}") - conn.commit() - except Exception as e: - self.logger.warning(f"Migration repeater_contacts: {e}") - - # complete_contact_tracking: path columns and is_starred (required for contacts page) - try: - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='complete_contact_tracking'") - if cursor.fetchone(): - cursor.execute("PRAGMA table_info(complete_contact_tracking)") - tracking_columns = [row[1] for row in cursor.fetchall()] - for column_name, column_type in [ - ('out_path', 'TEXT'), ('out_path_len', 'INTEGER'), ('snr', 'REAL') - ]: - if column_name not in tracking_columns: - self.logger.info(f"Adding missing column to complete_contact_tracking: {column_name}") - cursor.execute(f"ALTER TABLE complete_contact_tracking ADD COLUMN {column_name} {column_type}") - conn.commit() - if 'is_starred' not in tracking_columns: - self.logger.info("Adding is_starred column to complete_contact_tracking") - cursor.execute("ALTER TABLE complete_contact_tracking ADD COLUMN is_starred BOOLEAN DEFAULT 0") - conn.commit() - except Exception as e: - self.logger.warning(f"Migration complete_contact_tracking: {e}") - - # observed_paths: packet_hash - try: - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='observed_paths'") - if cursor.fetchone(): - cursor.execute("PRAGMA table_info(observed_paths)") - observed_paths_columns = [row[1] for row in cursor.fetchall()] - if 'packet_hash' not in observed_paths_columns: - self.logger.info("Adding packet_hash column to observed_paths") - cursor.execute("ALTER TABLE observed_paths ADD COLUMN packet_hash TEXT") - conn.commit() - if 'bytes_per_hop' not in observed_paths_columns: - self.logger.info("Adding bytes_per_hop column to observed_paths") - cursor.execute("ALTER TABLE observed_paths ADD COLUMN bytes_per_hop INTEGER") - conn.commit() - except Exception as e: - self.logger.warning(f"Migration observed_paths: {e}") - - # complete_contact_tracking: out_bytes_per_hop (multi-byte path decode) - try: - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='complete_contact_tracking'") - if cursor.fetchone(): - cursor.execute("PRAGMA table_info(complete_contact_tracking)") - cct_columns = [row[1] for row in cursor.fetchall()] - if 'out_bytes_per_hop' not in cct_columns: - self.logger.info("Adding out_bytes_per_hop column to complete_contact_tracking") - cursor.execute("ALTER TABLE complete_contact_tracking ADD COLUMN out_bytes_per_hop INTEGER") - conn.commit() - except Exception as e: - self.logger.warning(f"Migration complete_contact_tracking out_bytes_per_hop: {e}") - - # mesh_connections: graph/viewer columns - try: - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='mesh_connections'") - if cursor.fetchone(): - cursor.execute("PRAGMA table_info(mesh_connections)") - mc_columns = [row[1] for row in cursor.fetchall()] - for col_name, col_type in [ - ('from_public_key', 'TEXT'), ('to_public_key', 'TEXT'), - ('avg_hop_position', 'REAL'), ('geographic_distance', 'REAL'), - ]: - if col_name not in mc_columns: - self.logger.info(f"Adding missing column to mesh_connections: {col_name}") - cursor.execute(f"ALTER TABLE mesh_connections ADD COLUMN {col_name} {col_type}") - conn.commit() - except Exception as e: - self.logger.warning(f"Migration mesh_connections: {e}") - - self.logger.info("Database schema migration completed") - async def track_contact_advertisement(self, advert_data: dict, signal_info: Optional[dict] = None, packet_hash: Optional[str] = None) -> bool: """Track any contact advertisement in the complete tracking database""" try: diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 9ad90e1..07f35e8 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -328,9 +328,6 @@ class BotDataViewer: minimal_bot.mesh_graph = MeshGraph(minimal_bot) self.mesh_graph = minimal_bot.mesh_graph - # Initialize packet_stream table for real-time monitoring - self._init_packet_stream_table() - # Store database paths for direct connection self.db_path = self.db_path self.repeater_db_path = self.repeater_db_path @@ -339,53 +336,20 @@ class BotDataViewer: self.logger.error(f"Failed to initialize databases: {e}") raise - def _init_packet_stream_table(self): - """Initialize the packet_stream table in the web viewer database (same as [Bot] db_path by default).""" - try: - with closing(sqlite3.connect(self.db_path, timeout=60)) as conn: - cursor = conn.cursor() - - # Create packet_stream table with schema matching the INSERT statements - cursor.execute(''' - CREATE TABLE IF NOT EXISTS packet_stream ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp REAL NOT NULL, - data TEXT NOT NULL, - type TEXT NOT NULL - ) - ''') - - # Create index on timestamp for faster queries - cursor.execute(''' - CREATE INDEX IF NOT EXISTS idx_packet_stream_timestamp - ON packet_stream(timestamp) - ''') - - # Create index on type for filtering by type - cursor.execute(''' - CREATE INDEX IF NOT EXISTS idx_packet_stream_type - ON packet_stream(type) - ''') - - # Enable WAL for better concurrent access (bot + web viewer use same DB) - try: - cursor.execute('PRAGMA journal_mode=WAL') - except sqlite3.OperationalError: - pass # Ignore if locked; WAL may already be set - - conn.commit() - - self.logger.info(f"Initialized packet_stream table in {self.db_path}") - - except Exception as e: - self.logger.error(f"Failed to initialize packet_stream table: {e}") - # Don't raise - allow web viewer to continue even if table init fails - def _get_db_connection(self): """Get database connection - create new connection for each request to avoid threading issues""" try: conn = sqlite3.connect(self.db_path, timeout=60) conn.row_factory = sqlite3.Row + try: + foreign_keys = self.config.getboolean("Web_Viewer", "sqlite_foreign_keys", fallback=True) + busy_timeout_ms = self.config.getint("Web_Viewer", "sqlite_busy_timeout_ms", fallback=60000) + journal_mode = self.config.get("Web_Viewer", "sqlite_journal_mode", fallback="WAL").strip() or "WAL" + conn.execute(f"PRAGMA foreign_keys={'ON' if foreign_keys else 'OFF'}") + conn.execute(f"PRAGMA busy_timeout={int(busy_timeout_ms)}") + conn.execute(f"PRAGMA journal_mode={journal_mode}") + except sqlite3.OperationalError: + pass return conn except Exception as e: self.logger.error(f"Failed to create database connection: {e}") @@ -398,6 +362,15 @@ class BotDataViewer: """ conn = sqlite3.connect(self.db_path, timeout=60) conn.row_factory = sqlite3.Row + try: + foreign_keys = self.config.getboolean("Web_Viewer", "sqlite_foreign_keys", fallback=True) + busy_timeout_ms = self.config.getint("Web_Viewer", "sqlite_busy_timeout_ms", fallback=60000) + journal_mode = self.config.get("Web_Viewer", "sqlite_journal_mode", fallback="WAL").strip() or "WAL" + conn.execute(f"PRAGMA foreign_keys={'ON' if foreign_keys else 'OFF'}") + conn.execute(f"PRAGMA busy_timeout={int(busy_timeout_ms)}") + conn.execute(f"PRAGMA journal_mode={journal_mode}") + except sqlite3.OperationalError: + pass try: yield conn finally: diff --git a/modules/web_viewer/integration.py b/modules/web_viewer/integration.py index a831580..519480b 100644 --- a/modules/web_viewer/integration.py +++ b/modules/web_viewer/integration.py @@ -40,8 +40,6 @@ class BotIntegration: self._drain_thread: Optional[threading.Thread] = None # Initialize HTTP session with connection pooling for efficient reuse self._init_http_session() - # Initialize the packet_stream table - self._init_packet_stream_table() # Start background drain thread after table is confirmed to exist self._start_drain_thread() @@ -133,51 +131,54 @@ class BotIntegration: return str(Path(self.bot.db_manager.db_path).resolve()) def _init_packet_stream_table(self): - """Initialize the packet_stream table in the web viewer database (same as [Bot] db_path by default).""" + """Backward-compatible initializer (now handled by migrations). + + Kept for older call sites and tests that patch this method. Safe to call + multiple times and safe to ignore failures. + """ try: import sqlite3 db_path = self._get_web_viewer_db_path() - with closing(sqlite3.connect(str(db_path), timeout=60.0)) as conn: - cursor = conn.cursor() - - # Create packet_stream table with schema matching the INSERT statements - cursor.execute(''' + cur = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='packet_stream'" + ) + if cur.fetchone() is not None: + return + try: + foreign_keys = self.bot.config.getboolean("Web_Viewer", "sqlite_foreign_keys", fallback=True) + busy_timeout_ms = self.bot.config.getint("Web_Viewer", "sqlite_busy_timeout_ms", fallback=60000) + journal_mode = self.bot.config.get("Web_Viewer", "sqlite_journal_mode", fallback="WAL").strip() or "WAL" + except Exception: + foreign_keys = True + busy_timeout_ms = 60000 + journal_mode = "WAL" + conn.execute(f"PRAGMA foreign_keys={'ON' if foreign_keys else 'OFF'}") + conn.execute(f"PRAGMA busy_timeout={int(busy_timeout_ms)}") + try: + conn.execute(f"PRAGMA journal_mode={journal_mode}") + except sqlite3.OperationalError: + pass + conn.execute( + """ CREATE TABLE IF NOT EXISTS packet_stream ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, data TEXT NOT NULL, type TEXT NOT NULL ) - ''') - - # Create index on timestamp for faster queries - cursor.execute(''' - CREATE INDEX IF NOT EXISTS idx_packet_stream_timestamp - ON packet_stream(timestamp) - ''') - - # Create index on type for filtering by type - cursor.execute(''' - CREATE INDEX IF NOT EXISTS idx_packet_stream_type - ON packet_stream(type) - ''') - - # Enable WAL for better concurrent access (bot + web viewer use same DB) - try: - cursor.execute('PRAGMA journal_mode=WAL') - except sqlite3.OperationalError: - pass # Ignore if locked; WAL may already be set - + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_timestamp ON packet_stream(timestamp)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_stream_type ON packet_stream(type)" + ) conn.commit() - - self.bot.logger.info(f"Initialized packet_stream table in {db_path}") - - except Exception as e: - self.bot.logger.error(f"Failed to initialize packet_stream table: {e}") - # Don't raise - allow bot to continue even if table init fails - # The error will be caught when trying to insert data + except Exception: + pass def _start_drain_thread(self) -> None: """Start the background thread that flushes the write queue every DRAIN_INTERVAL seconds.""" diff --git a/tests/test_db_migrations.py b/tests/test_db_migrations.py index c9b569f..d5d29da 100644 --- a/tests/test_db_migrations.py +++ b/tests/test_db_migrations.py @@ -90,6 +90,51 @@ class TestMigrationRunner: cursor = conn.execute("SELECT COUNT(*) FROM schema_version") assert cursor.fetchone()[0] == len(MIGRATIONS) + def test_non_contiguous_history_applies_missing_versions(self, conn, logger): + """If schema_version has gaps, runner should apply missing versions.""" + conn.execute(""" + CREATE TABLE IF NOT EXISTS schema_version ( + version INTEGER NOT NULL, + description TEXT, + applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + conn.execute("INSERT INTO schema_version (version, description) VALUES (1, 'initial schema')") + conn.execute("INSERT INTO schema_version (version, description) VALUES (3, 'feed_subscriptions: filter_config, sort_config')") + + # Seed minimal tables so migrations that ALTER them have something to work with. + conn.execute(""" + CREATE TABLE IF NOT EXISTS feed_subscriptions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + feed_url TEXT NOT NULL, + channel_name TEXT NOT NULL, + UNIQUE(feed_url, channel_name) + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS channel_operations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + operation_type TEXT NOT NULL + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS feed_message_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + feed_id INTEGER NOT NULL, + channel_name TEXT NOT NULL, + message TEXT NOT NULL, + queued_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + sent_at TIMESTAMP + ) + """) + conn.commit() + + runner = MigrationRunner(conn, logger) + runner.run() + + applied = {row[0] for row in conn.execute("SELECT version FROM schema_version").fetchall()} + assert set(range(1, len(MIGRATIONS) + 1)).issubset(applied) + def test_partial_history_applies_only_pending(self, conn, logger): """Simulate a DB that already had migration 1 applied.""" conn.execute(""" @@ -142,9 +187,9 @@ class TestMigrationRunner: assert _column_exists(cursor2, "feed_subscriptions", "output_format") is True assert _column_exists(cursor2, "feed_message_queue", "priority") is True - def test_current_version_zero_on_empty_table(self, runner, conn): + def test_applied_versions_empty_on_empty_table(self, runner, conn): runner._ensure_version_table() - assert runner._current_version() == 0 + assert runner._applied_versions() == set() # --------------------------------------------------------------------------- @@ -187,3 +232,57 @@ class TestSchema: runner.run() cursor = conn.cursor() assert _column_exists(cursor, "feed_message_queue", "item_id") is True + + def test_foreign_key_cascade_works_when_enabled(self, runner, conn): + conn.execute("PRAGMA foreign_keys=ON") + runner.run() + + # Create a subscription and activity, then delete subscription. + sub_id = conn.execute( + "INSERT INTO feed_subscriptions (feed_type, feed_url, channel_name) VALUES (?, ?, ?)", + ("rss", "https://example.com/feed.xml", "chan"), + ).lastrowid + conn.execute( + "INSERT INTO feed_activity (feed_id, item_id, item_title) VALUES (?, ?, ?)", + (sub_id, "item-1", "title"), + ) + conn.commit() + + conn.execute("DELETE FROM feed_subscriptions WHERE id = ?", (sub_id,)) + conn.commit() + + remaining = conn.execute("SELECT COUNT(*) FROM feed_activity WHERE feed_id = ?", (sub_id,)).fetchone()[0] + assert remaining == 0 + + def test_repeater_tables_created_by_migrations(self, runner, conn): + runner.run() + for table in [ + "repeater_contacts", + "complete_contact_tracking", + "daily_stats", + "unique_advert_packets", + "purging_log", + "mesh_connections", + "observed_paths", + ]: + cur = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", + (table,), + ) + assert cur.fetchone() is not None + + def test_repeater_indexes_created(self, runner, conn): + runner.run() + # Spot-check a few key indexes (not exhaustive). + for idx in [ + "idx_public_key", + "idx_complete_public_key", + "idx_unique_advert_hash", + "idx_from_prefix", + "idx_observed_paths_endpoints", + ]: + cur = conn.execute( + "SELECT name FROM sqlite_master WHERE type='index' AND name=?", + (idx,), + ) + assert cur.fetchone() is not None diff --git a/tests/test_web_viewer.py b/tests/test_web_viewer.py index ea29af1..7ceec3e 100644 --- a/tests/test_web_viewer.py +++ b/tests/test_web_viewer.py @@ -2115,9 +2115,18 @@ class TestBotIntegrationQueue: bot.logger = logging.getLogger("test_integration") bot.logger.addHandler(logging.NullHandler()) bot.config = cfg - bot.db_manager.db_path = db_path bot.bot_root = str(tmp_path) + # Ensure schema exists (packet_stream is migration-owned). + from modules.db_manager import DBManager + + class MinimalBot: + def __init__(self, logger, config): + self.logger = logger + self.config = config + + bot.db_manager = DBManager(MinimalBot(bot.logger, cfg), db_path) + from modules.web_viewer.integration import BotIntegration bi = BotIntegration(bot) yield bi