fix(live-db): Fix three connectivity bugs in Live DB Mode

1. Replace quote_plus with quote(safe='') for URL-encoding credentials
   in the connection string. SQLAlchemy decodes with urllib.parse.unquote
   (not unquote_plus), so quote_plus-encoded spaces ('+') were passed
   literally to psycopg2 and caused authentication failures.

2. Use a single engine.connect() context in get_external_schema() and
   inspect(conn) instead of inspect(engine). In SQLAlchemy 2.0, inspect(engine)
   opens a new DB connection for every reflection method call (get_table_names,
   get_columns per table), leading to N+1 connection attempts. Using a
   single connection context eliminates this overhead and gives a well-defined
   connection lifecycle.

3. Remove pool_pre_ping=True from both get_external_schema() and
   execute_sql_on_external(). These engines are created, used once, and
   disposed — pool_pre_ping adds a redundant SELECT 1 round-trip before
   every use. On cold Neon instances this extra round-trip consumed part of
   the 10-second connect_timeout budget, causing spurious timeouts.

https://claude.ai/code/session_016qjAVBXoVuhQD9MNdHQ6Fd

Author: claude

backend/app/queries/router.py

@@ -1,5 +1,5 @@
 import json
-from urllib.parse import quote_plus
+from urllib.parse import quote
 from fastapi import APIRouter, Depends, HTTPException
 
 from app.auth.utils import CurrentUser, get_current_user, require_role
@@ -155,10 +155,12 @@ def run_live_query(
 ):
     sb = get_supabase()
 
-    # URL-encode user + password so special chars (@, :, #, +, etc.) don't break the URI
+    # URL-encode user, password, and db_name so special chars don't break the URI.
+    # Use quote() with safe='' (not quote_plus) because SQLAlchemy decodes with
+    # urllib.parse.unquote, which does NOT convert '+' back to space.
     conn_str = (
-        f"postgresql+psycopg2://{quote_plus(payload.db_user)}:{quote_plus(payload.db_password)}"
-        f"@{payload.db_host}:{payload.db_port}/{payload.db_name}"
+        f"postgresql+psycopg2://{quote(payload.db_user, safe='')}:{quote(payload.db_password, safe='')}"
+        f"@{payload.db_host}:{payload.db_port}/{quote(payload.db_name, safe='')}"
     )
 
     schema_text = get_external_schema(conn_str, ssl_required=payload.ssl_required)

backend/app/queries/sql_executor.py

@@ -113,7 +113,6 @@ def execute_sql_on_external(
     try:
         engine = create_engine(
             connection_string,
-            pool_pre_ping=True,
             connect_args={"connect_timeout": 10, "sslmode": sslmode},
         )
         with engine.connect() as conn:

backend/app/schema_service/service.py

@@ -42,15 +42,17 @@ def get_external_schema(connection_string: str, ssl_required: bool = True) -> st
     try:
         engine = create_engine(
             connection_string,
-            pool_pre_ping=True,
             connect_args={"connect_timeout": 10, "sslmode": sslmode},
         )
-        inspector = inspect(engine)
         parts = []
-        for table_name in inspector.get_table_names(schema="public"):
-            cols = inspector.get_columns(table_name, schema="public")
-            col_defs = [f"  {c['name']} {c['type']}" for c in cols]
-            parts.append(f"Table: {table_name}\nColumns:\n" + "\n".join(col_defs))
+        # Use a single connection for all reflection calls to avoid the overhead
+        # of opening a new connection per method when using inspect(engine).
+        with engine.connect() as conn:
+            inspector = inspect(conn)
+            for table_name in inspector.get_table_names(schema="public"):
+                cols = inspector.get_columns(table_name, schema="public")
+                col_defs = [f"  {c['name']} {c['type']}" for c in cols]
+                parts.append(f"Table: {table_name}\nColumns:\n" + "\n".join(col_defs))
         return "\n\n".join(parts)
     except Exception as exc:
         msg = str(exc)