From e9a615e120a4a0dbe4e764c21e3788c5d6abff90 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 31 Jan 2026 09:25:17 -0800 Subject: [PATCH] feat: Add database query execution methods for connection reuse - Introduced `execute_query_on_connection` and `execute_update_on_connection` methods in DBManager to facilitate executing queries and updates on existing database connections. - Enhanced the `_recalculate_distance_if_needed`, `_get_location_by_public_key`, and `_get_location_by_prefix` methods in MeshGraph to support optional database connection and location caching, improving performance during batch operations. --- modules/commands/base_command.py | 6 +- modules/db_manager.py | 20 ++- modules/mesh_graph.py | 229 +++++++++++++++++++++++++------ 3 files changed, 208 insertions(+), 47 deletions(-) diff --git a/modules/commands/base_command.py b/modules/commands/base_command.py index 4d11320..cde4a55 100644 --- a/modules/commands/base_command.py +++ b/modules/commands/base_command.py @@ -337,19 +337,21 @@ class BaseCommand(ABC): # Check if channel matches allowed list return message_channel_normalized in allowed_normalized - def can_execute(self, message: MeshMessage) -> bool: + def can_execute(self, message: MeshMessage, skip_channel_check: bool = False) -> bool: """Check if this command can be executed with the given message. Checks channel permissions, DM requirements, cooldowns, and admin access. Args: message: The message to check execution for. + skip_channel_check: If True, skip channel check (used when a parent + command has already enforced its own channel override, e.g. wx delegating to gwx). Returns: bool: True if the command can be executed, False otherwise. """ # Check channel access (standardized channel override) - if not self.is_channel_allowed(message): + if not skip_channel_check and not self.is_channel_allowed(message): return False # Check if command requires DM and message is not DM diff --git a/modules/db_manager.py b/modules/db_manager.py index 7162541..f3f6801 100644 --- a/modules/db_manager.py +++ b/modules/db_manager.py @@ -534,7 +534,25 @@ class DBManager: except Exception as e: self.logger.error(f"Error executing update: {e}") return 0 - + + def execute_query_on_connection(self, conn: sqlite3.Connection, query: str, params: Tuple = ()) -> List[Dict]: + """Execute a query on an existing connection. Caller owns the connection.""" + cursor = conn.cursor() + cursor.execute(query, params) + rows = cursor.fetchall() + if conn.row_factory is sqlite3.Row: + return [dict(row) for row in rows] + desc = cursor.description + if not desc: + return [] + return [dict(zip([c[0] for c in desc], row)) for row in rows] + + def execute_update_on_connection(self, conn: sqlite3.Connection, query: str, params: Tuple = ()) -> int: + """Execute an update/insert/delete on an existing connection. Caller must commit.""" + cursor = conn.cursor() + cursor.execute(query, params) + return cursor.rowcount + # Bot metadata methods def set_metadata(self, key: str, value: str) -> None: """Set a metadata value for the bot. diff --git a/modules/mesh_graph.py b/modules/mesh_graph.py index a328d18..fddb3eb 100644 --- a/modules/mesh_graph.py +++ b/modules/mesh_graph.py @@ -224,45 +224,58 @@ class MeshGraph: except Exception as e: self.logger.debug(f"Error notifying web viewer of edge update: {e}") - def _recalculate_distance_if_needed(self, edge: Dict) -> Optional[float]: + def _recalculate_distance_if_needed( + self, + edge: Dict, + conn: Optional[sqlite3.Connection] = None, + location_cache: Optional[Dict[str, Tuple[float, float]]] = None, + ) -> Optional[float]: """Recalculate geographic distance using full public keys if available. This ensures we get the correct location when there are prefix collisions. Args: edge: Edge dictionary with prefix and optional public keys. + conn: Optional existing DB connection for batch operations. + location_cache: Optional cache for location lookups within a flush (keyed by pk: or prefix:). Returns: Optional[float]: Recalculated distance in km, or None if can't calculate. """ from .utils import calculate_distance - # Try to get locations using full public keys first (more accurate) - from_location = None - to_location = None - - # Get location for 'from' node + # Get location for 'from' node (conn optional for single-connection batch flush) if edge.get('from_public_key'): - # Use full public key for precise lookup - from_location = self._get_location_by_public_key(edge['from_public_key']) - + from_location = self._get_location_by_public_key( + edge['from_public_key'], conn=conn, location_cache=location_cache + ) + else: + from_location = None if not from_location: - # Fallback to prefix lookup - use to_location as reference for LoRa distance preference to_location_temp = None if edge.get('to_public_key'): - to_location_temp = self._get_location_by_public_key(edge['to_public_key']) + to_location_temp = self._get_location_by_public_key( + edge['to_public_key'], conn=conn, location_cache=location_cache + ) if not to_location_temp: - to_location_temp = self._get_location_by_prefix(edge['to_prefix']) - from_location = self._get_location_by_prefix(edge['from_prefix'], to_location_temp) + to_location_temp = self._get_location_by_prefix( + edge['to_prefix'], conn=conn, location_cache=location_cache + ) + from_location = self._get_location_by_prefix( + edge['from_prefix'], to_location_temp, conn=conn, location_cache=location_cache + ) # Get location for 'to' node if edge.get('to_public_key'): - # Use full public key for precise lookup - to_location = self._get_location_by_public_key(edge['to_public_key']) - + to_location = self._get_location_by_public_key( + edge['to_public_key'], conn=conn, location_cache=location_cache + ) + else: + to_location = None if not to_location: - # Fallback to prefix lookup - use from_location as reference for LoRa distance preference - to_location = self._get_location_by_prefix(edge['to_prefix'], from_location) + to_location = self._get_location_by_prefix( + edge['to_prefix'], from_location, conn=conn, location_cache=location_cache + ) # Calculate distance if we have both locations if from_location and to_location: @@ -273,11 +286,19 @@ class MeshGraph: return None - def _get_location_by_public_key(self, public_key: str) -> Optional[Tuple[float, float]]: + def _get_location_by_public_key( + self, + public_key: str, + conn: Optional[sqlite3.Connection] = None, + location_cache: Optional[Dict[str, Tuple[float, float]]] = None, + ) -> Optional[Tuple[float, float]]: """Get location for a full public key (more accurate than prefix lookup). Prefers starred repeaters if there are somehow multiple entries (shouldn't happen with full key). """ + cache_key = f"pk:{public_key}" if location_cache is not None else None + if cache_key is not None and cache_key in location_cache: + return location_cache[cache_key] try: query = ''' SELECT latitude, longitude @@ -289,18 +310,30 @@ class MeshGraph: ORDER BY is_starred DESC, COALESCE(last_advert_timestamp, last_heard) DESC LIMIT 1 ''' - results = self.db_manager.execute_query(query, (public_key,)) + if conn is not None: + results = self.db_manager.execute_query_on_connection(conn, query, (public_key,)) + else: + results = self.db_manager.execute_query(query, (public_key,)) if results: row = results[0] lat = row.get('latitude') lon = row.get('longitude') if lat is not None and lon is not None: - return (float(lat), float(lon)) + result = (float(lat), float(lon)) + if cache_key is not None: + location_cache[cache_key] = result + return result except Exception as e: self.logger.debug(f"Error getting location by public key {public_key[:16]}...: {e}") return None - def _get_location_by_prefix(self, prefix: str, reference_location: Optional[Tuple[float, float]] = None) -> Optional[Tuple[float, float]]: + def _get_location_by_prefix( + self, + prefix: str, + reference_location: Optional[Tuple[float, float]] = None, + conn: Optional[sqlite3.Connection] = None, + location_cache: Optional[Dict[str, Tuple[float, float]]] = None, + ) -> Optional[Tuple[float, float]]: """Get location for a prefix (fallback when full public key not available). For LoRa networks, prefers shorter distances when there are prefix collisions, @@ -309,7 +342,16 @@ class MeshGraph: Args: prefix: 2-character hex prefix. reference_location: Optional (lat, lon) to calculate distance from for LoRa preference. + conn: Optional existing DB connection for batch operations. + location_cache: Optional cache for location lookups within a flush. """ + if location_cache is not None: + if reference_location is not None: + cache_key = f"prefix:{prefix}:{reference_location[0]}:{reference_location[1]}" + else: + cache_key = f"prefix:{prefix}" + if cache_key in location_cache: + return location_cache[cache_key] try: prefix_pattern = f"{prefix}%" @@ -323,8 +365,10 @@ class MeshGraph: AND latitude != 0 AND longitude != 0 AND role IN ('repeater', 'roomserver') ''' - - results = self.db_manager.execute_query(query, (prefix_pattern,)) + if conn is not None: + results = self.db_manager.execute_query_on_connection(conn, query, (prefix_pattern,)) + else: + results = self.db_manager.execute_query(query, (prefix_pattern,)) if not results: return None @@ -358,7 +402,11 @@ class MeshGraph: lat = best_row.get('latitude') lon = best_row.get('longitude') if lat is not None and lon is not None: - return (float(lat), float(lon)) + result = (float(lat), float(lon)) + if location_cache is not None and reference_location is not None: + cache_key = f"prefix:{prefix}:{reference_location[0]}:{reference_location[1]}" + location_cache[cache_key] = result + return result # No reference location or single result - use standard ordering # Prefer starred, then most recent @@ -371,17 +419,29 @@ class MeshGraph: lat = row.get('latitude') lon = row.get('longitude') if lat is not None and lon is not None: - return (float(lat), float(lon)) + result = (float(lat), float(lon)) + if location_cache is not None: + cache_key = f"prefix:{prefix}" if reference_location is None else f"prefix:{prefix}:{reference_location[0]}:{reference_location[1]}" + location_cache[cache_key] = result + return result except Exception as e: self.logger.debug(f"Error getting location by prefix {prefix}: {e}") return None - def _write_edge_to_db(self, edge_key: Tuple[str, str], is_new: bool): + def _write_edge_to_db( + self, + edge_key: Tuple[str, str], + is_new: bool, + conn: Optional[sqlite3.Connection] = None, + location_cache: Optional[Dict[str, Tuple[float, float]]] = None, + ): """Write a single edge to the database. Args: edge_key: (from_prefix, to_prefix) tuple. is_new: True if this is a new edge, False if updating existing. + conn: Optional existing DB connection for batch operations (caller commits). + location_cache: Optional cache for location lookups within a flush. """ if edge_key not in self.edges: return @@ -391,7 +451,9 @@ class MeshGraph: # Recalculate distance using full public keys if available (more accurate) # This fixes issues where prefix collisions cause wrong locations to be used if edge.get('from_public_key') or edge.get('to_public_key'): - recalculated_distance = self._recalculate_distance_if_needed(edge) + recalculated_distance = self._recalculate_distance_if_needed( + edge, conn=conn, location_cache=location_cache + ) if recalculated_distance is not None: edge['geographic_distance'] = recalculated_distance self.logger.debug(f"Mesh graph: Recalculated distance for {edge_key} using public keys: {recalculated_distance:.1f} km") @@ -422,7 +484,9 @@ class MeshGraph: # Only update distance if we have at least one public key and current distance seems wrong current_distance = edge.get('geographic_distance') if (edge.get('from_public_key') or edge.get('to_public_key')) and current_distance: - recalculated = self._recalculate_distance_if_needed(edge) + recalculated = self._recalculate_distance_if_needed( + edge, conn=conn, location_cache=location_cache + ) if recalculated is not None: # Update if recalculated distance is significantly different (more than 20% difference) if abs(recalculated - current_distance) / max(current_distance, 1.0) > 0.2: @@ -433,14 +497,7 @@ class MeshGraph: # Always update public keys if provided (allows filling in missing keys on existing edges) from_key = edge.get('from_public_key') to_key = edge.get('to_public_key') - query = ''' - UPDATE mesh_connections - SET observation_count = ?, last_seen = ?, - avg_hop_position = ?, geographic_distance = ?, - from_public_key = CASE WHEN ? IS NOT NULL THEN ? ELSE from_public_key END, - to_public_key = CASE WHEN ? IS NOT NULL THEN ? ELSE to_public_key END - WHERE from_prefix = ? AND to_prefix = ? - ''' + query = self._MESH_EDGE_UPDATE_QUERY params = ( edge['observation_count'], edge['last_seen'].isoformat() if isinstance(edge['last_seen'], datetime) else edge['last_seen'], @@ -454,8 +511,10 @@ class MeshGraph: edge['to_prefix'] ) - # Use execute_update for INSERT/UPDATE operations (handles commit automatically) - rows_affected = self.db_manager.execute_update(query, params) + if conn is not None: + rows_affected = self.db_manager.execute_update_on_connection(conn, query, params) + else: + rows_affected = self.db_manager.execute_update(query, params) if rows_affected > 0: self.logger.debug(f"Mesh graph: Successfully wrote edge {edge_key} to database ({'INSERT' if is_new else 'UPDATE'}, {rows_affected} rows)") else: @@ -466,6 +525,62 @@ class MeshGraph: import traceback self.logger.debug(traceback.format_exc()) + # UPDATE statement used for both single-edge writes and batch executemany + _MESH_EDGE_UPDATE_QUERY = ''' + UPDATE mesh_connections + SET observation_count = ?, last_seen = ?, + avg_hop_position = ?, geographic_distance = ?, + from_public_key = CASE WHEN ? IS NOT NULL THEN ? ELSE from_public_key END, + to_public_key = CASE WHEN ? IS NOT NULL THEN ? ELSE to_public_key END + WHERE from_prefix = ? AND to_prefix = ? + ''' + + def _build_update_params_for_edge( + self, + edge_key: Tuple[str, str], + conn: Optional[sqlite3.Connection], + location_cache: Optional[Dict[str, Tuple[float, float]]], + ) -> Optional[Tuple]: + """Build UPDATE params for an edge (for batch executemany). Returns None to skip.""" + if edge_key not in self.edges: + return None + try: + edge = self.edges[edge_key] + # Recalculate distance if we have public keys (same logic as _write_edge_to_db) + if edge.get('from_public_key') or edge.get('to_public_key'): + recalculated_distance = self._recalculate_distance_if_needed( + edge, conn=conn, location_cache=location_cache + ) + if recalculated_distance is not None: + edge['geographic_distance'] = recalculated_distance + current_distance = edge.get('geographic_distance') + if (edge.get('from_public_key') or edge.get('to_public_key')) and current_distance: + recalculated = self._recalculate_distance_if_needed( + edge, conn=conn, location_cache=location_cache + ) + if recalculated is not None and abs(recalculated - current_distance) / max(current_distance, 1.0) > 0.2: + edge['geographic_distance'] = recalculated + from_key = edge.get('from_public_key') + to_key = edge.get('to_public_key') + last_seen = edge['last_seen'] + if isinstance(last_seen, datetime): + last_seen = last_seen.isoformat() + return ( + edge['observation_count'], + last_seen, + edge.get('avg_hop_position'), + edge.get('geographic_distance'), + from_key, + from_key, + to_key, + to_key, + edge['from_prefix'], + edge['to_prefix'], + ) + except Exception as e: + self.logger.debug(f"Error building update params for {edge_key}: {e}") + return None + def _start_batch_writer(self): """Start background task for batched writes.""" def batch_writer_loop(): @@ -481,7 +596,10 @@ class MeshGraph: self._batch_thread = batch_thread def _flush_pending_updates_sync(self): - """Flush all pending edge updates to database (synchronous version).""" + """Flush all pending edge updates to database (synchronous version). + Uses a single connection for the entire batch to avoid 'unable to open database file' + when many edges are written in quick succession. + """ with self.pending_lock: if not self.pending_updates: return @@ -489,10 +607,33 @@ class MeshGraph: updates = list(self.pending_updates) self.pending_updates.clear() - # Write all pending updates - for edge_key in updates: - if edge_key in self.edges: - self._write_edge_to_db(edge_key, False) + conn = None + location_cache: Dict[str, Tuple[float, float]] = {} + try: + conn = self.db_manager.get_connection() + params_list: List[Tuple] = [] + for edge_key in updates: + params_tuple = self._build_update_params_for_edge(edge_key, conn, location_cache) + if params_tuple is not None: + params_list.append(params_tuple) + if params_list: + cursor = conn.cursor() + cursor.executemany(self._MESH_EDGE_UPDATE_QUERY, params_list) + if conn: + conn.commit() + except Exception as e: + self.logger.warning(f"Error flushing graph updates: {e}") + if conn: + try: + conn.rollback() + except Exception: + pass + finally: + if conn: + try: + conn.close() + except Exception: + pass if updates: self.logger.debug(f"Flushed {len(updates)} pending graph edge updates")