Implement database migration enhancements and connection configuration

- Added `_apply_sqlite_pragmas` method in `DBManager` to configure SQLite connection settings such as foreign keys, busy timeout, and journal mode.
- Updated `connection` methods in `DBManager` and `BotDataViewer` to utilize the new pragma settings.
- Introduced validation functions in `db_migrations.py` to ensure proper identifier formats and table existence checks.
- Created new migration functions for managing `packet_stream` and repeater-related tables, ensuring they are created and indexed correctly.
- Removed redundant table initialization code from `RepeaterManager` and `BotDataViewer`, relying on migrations for table setup.
- Enhanced tests to verify the creation of repeater tables and indexes during migrations.

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:

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]}"

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:

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:

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."""

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

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