fix(database): unify PostgreSQL selection with legacy indexing connections

DEVELOPER_DOCS.md

@@ -4,6 +4,30 @@ This document provides instructions for developing, testing, configuring, and ru
 
 ---
 
+## 🗄️ Relational Database Selection, Schema Migration & Rollback
+
+ContextCortex supports unified relational database selection across **SQLite** (default local embedded storage) and **PostgreSQL** (enterprise local deployment profiles).
+
+### Database Engine Selection
+- **SQLite (Default)**: Used when `DATABASE_URL` is unset or starts with `sqlite:///`. Relational metadata and index caches are stored at `/app/data/index_cache.db` (configurable via `CACHE_DB_PATH`).
+- **PostgreSQL**: Selected by setting `DATABASE_URL` to a valid PostgreSQL connection string (e.g., `postgresql+psycopg://user:password@postgres:5432/contextcortex`).
+  - When PostgreSQL is selected, all relational access across repositories, settings, credentials, indexing, file summaries, AST symbols, API routes, relationships, search, ADRs, and file deletion routes through the unified `PostgresConnectionAdapter`.
+  - **No Fallback SQLite Files**: When PostgreSQL is selected, the application operates exclusively against PostgreSQL and will never create or query an unrelated fallback SQLite file.
+
+### Schema Migration
+1. **Automated Schema Initialization**: Schema tables and initial seed data (custom prompts, system metadata, vector store configuration) are initialized on startup via `init_db()`.
+2. **Canonical SQLAlchemy Core Schema**: Table definitions are specified in `app/services/database/schema.py`. Running `init_db()` calls `metadata.create_all(bind=engine)` to idempotently create missing tables across SQLite and PostgreSQL.
+3. **Column Migration Checks**: `ensure_git_repositories_columns()` and `ensure_file_summaries_columns()` verify table columns on startup (e.g. `keep_shallow`, `summary_text`) using `information_schema` on PostgreSQL or `PRAGMA table_info` on SQLite.
+4. **pgvector Extension**: When initializing PostgreSQL, `init_db()` executes `CREATE EXTENSION IF NOT EXISTS vector;` to enable vector operations for pgvector storage.
+
+### Database Rollback Strategy
+If you need to roll back or switch storage engines between PostgreSQL and SQLite:
+1. **Switch Connection URL**: Set `DATABASE_URL` to the desired storage target (e.g., unset `DATABASE_URL` or set `DATABASE_URL="sqlite:////app/data/index_cache.db"` to revert to SQLite).
+2. **Initialize Target Schema**: Start the application (`python main.py`) or call `init_db()`. The database engine factory initializes target schema tables idempotently without modifying or destroying data in the previously active database.
+3. **Re-index Content**: Trigger re-indexing (`run_full_indexing()` or via the UI/API) to populate repository files, symbols, and summaries into the target relational database.
+
+---
+
 ## 🛠️ Prerequisites
 
 - **Python 3.11+**

REQUIREMENTS.md

@@ -2,7 +2,7 @@
 
 > **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`.
 
-**Test Verification Baseline:** **1052 Automated Tests** (699 Pytest Backend + 303 Vitest Frontend + 50 Playwright E2E).
+**Test Verification Baseline:** **1057 Automated Tests** (704 Pytest Backend + 303 Vitest Frontend + 50 Playwright E2E).
 
 ---
 
@@ -1179,6 +1179,16 @@ and leaves the prior indexed state intact without data loss._
 - `test_poller_worker_cycle`
 - `test_poller_worker_disabled_interval`
 
+#### `tests/test_postgres_unified_connection.py` (5 tests)
+- `test_dict_row_compatibility` - _Verify DictRow supports dict lookup, index lookup, attribute access, and dict conversion._
+- `test_convert_sql_placeholders` - _Verify placeholder conversion converts ? to :p1, :p2 and appends RETURNING id for inserts._
+- `test_postgres_selection_never_creates_fallback_sqlite` - _Asserts that when PostgreSQL is selected, get_db_connection() returns the PostgreSQL adapter
+and NEVER creates or queries the fallback SQLite index_cache.db file._
+- `test_postgres_full_relational_crud_and_indexing` - _Tests full relational lifecycle (repositories, settings, credentials, indexing,
+file summaries, AST symbols/routes/relationships, ADRs, search and deletion)
+over the PostgresConnectionAdapter using an initialized test engine._
+- `test_ensure_columns_postgres` - _Verify ensure_git_repositories_columns and ensure_file_summaries_columns run without error when on PostgreSQL._
+
 #### `tests/test_processor_caching.py` (6 tests)
 - `test_compute_text_hash`
 - `test_process_file_content_populates_embedding_cache`

app/services/database/connection.py

@@ -18,18 +18,36 @@
 logger = logging.getLogger("contextcortex.db")
 
 CACHE_DB_PATH = get_default_db_path()
-try:
-    os.makedirs(os.path.dirname(CACHE_DB_PATH), exist_ok=True)
-except Exception:
-    pass
+if not is_postgres():
+    try:
+        os.makedirs(os.path.dirname(CACHE_DB_PATH), exist_ok=True)
+    except Exception:
+        pass
 
+from app.services.database.postgres_compat import (
+    DictRow,
+    TABLES_WITH_ID,
+    _convert_sql_placeholders,
+    _SingleRowResult,
+    PostgresResultAdapter,
+    PostgresCursorAdapter,
+    PostgresConnectionAdapter,
+)
 
-def get_db_connection() -> sqlite3.Connection:
+def get_db_connection() -> Any:
     """
-    Returns a SQLite database connection for backward compatibility with legacy modules.
+    Returns a database connection (PostgreSQL adapter or SQLite connection)
+    based on database selection.
     """
+    if is_postgres():
+        return PostgresConnectionAdapter()
+
     from app.services.database.engine import get_current_db_path
     db_path = get_current_db_path()
+    try:
+        os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    except Exception:
+        pass
 
     conn = sqlite3.connect(db_path, timeout=15.0)
     conn.row_factory = sqlite3.Row
@@ -38,14 +56,12 @@ def get_db_connection() -> sqlite3.Connection:
     conn.execute("PRAGMA foreign_keys=ON;")
     return conn
 
-
 def init_db(vault_path: str = "/docs", engine=None):
     """
     Initializes database schema and seed data via SQLAlchemy Core engine.
     """
     return engine_init_db(vault_path=vault_path, engine=engine)
 
-
 def get_metadata(key: str, default: Optional[str] = None) -> Optional[str]:
     try:
         with get_connection() as conn:
@@ -56,7 +72,6 @@ def get_metadata(key: str, default: Optional[str] = None) -> Optional[str]:
     except Exception:
         return default
 
-
 def set_metadata(key: str, value: str):
     try:
         with get_connection() as conn:
@@ -70,7 +85,6 @@ def set_metadata(key: str, value: str):
     except Exception as e:
         logger.error(f"Failed to set metadata key {key}: {e}")
 
-
 def get_default_vector_storage_path() -> str:
     env_path = os.getenv("VECTOR_STORE_STORAGE_PATH")
     if env_path:
@@ -79,7 +93,6 @@ def get_default_vector_storage_path() -> str:
         return "/app/data/vector_storage"
     return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "vector_storage")
 
-
 def _resolve_default_vector_store_config(conn: Optional[Any] = None) -> Dict[str, str]:
     provider = os.getenv("VECTOR_STORE") or os.getenv("VECTOR_STORE_PROVIDER") or (
         "pgvector" if is_postgres() else "qdrant"
@@ -127,7 +140,6 @@ def _resolve_default_vector_store_config(conn: Optional[Any] = None) -> Dict[str
         "collection": collection.strip(),
     }
 
-
 def get_vector_store_db_config() -> Dict[str, str]:
     provider = get_metadata("vector_store_provider")
     mode = get_metadata("vector_store_mode")
@@ -145,7 +157,6 @@ def get_vector_store_db_config() -> Dict[str, str]:
         "collection": (collection or default_cfg["collection"]).strip(),
     }
 
-
 def set_vector_store_db_config(
     provider: Optional[str] = None,
     mode: Optional[str] = None,
@@ -164,7 +175,6 @@ def set_vector_store_db_config(
     if collection is not None:
         set_metadata("vector_store_collection", collection.strip())
 
-
 def detect_system_resources() -> Dict[str, Any]:
     """Detects available CPU cores and RAM in the execution environment, respecting container quotas."""
     import math
@@ -200,7 +210,6 @@ def detect_system_resources() -> Dict[str, Any]:
         pass
     return {"cpus": cpus, "memory_gb": mem_gb}
 
-
 def _resolve_default_embedding_config(conn: Optional[Any] = None) -> Dict[str, Any]:
     sys_res = detect_system_resources()
     default_threads = min(2, max(1, sys_res["cpus"]))
@@ -234,7 +243,6 @@ def _resolve_default_embedding_config(conn: Optional[Any] = None) -> Dict[str, A
         "system_memory_gb": sys_res["memory_gb"],
     }
 
-
 def get_embedding_db_config() -> Dict[str, Any]:
     provider = get_metadata("embedding_provider")
     dense_model = get_metadata("embedding_dense_model")
@@ -273,7 +281,6 @@ def get_embedding_db_config() -> Dict[str, Any]:
         "system_memory_gb": default_cfg["system_memory_gb"],
     }
 
-
 def set_embedding_db_config(
     provider: Optional[str] = None,
     dense_model: Optional[str] = None,
@@ -308,23 +315,20 @@ def set_embedding_db_config(
         set_metadata("chat_model", chat_model.strip())
         set_metadata("embedding_chat_model", chat_model.strip())
 
-
 def get_vision_ocr_model() -> str:
     """Returns stored vision OCR model or fallback to env/default."""
     stored = get_metadata("openai_vision_model") or get_metadata("vision_ocr_model") or get_metadata("embedding_vision_ocr_model")
     if stored and stored.strip():
         return stored.strip()
     return (os.getenv("VISION_OCR_MODEL") or "").strip()
 
-
 def get_chat_model() -> str:
     """Returns stored chat model or fallback to env/default."""
     stored = get_metadata("openai_chat_model") or get_metadata("chat_model") or get_metadata("embedding_chat_model")
     if stored and stored.strip():
         return stored.strip()
     return (os.getenv("CHAT_MODEL") or "").strip()
 
-
 def get_ai_gateway_config() -> Dict[str, Any]:
     """
     Returns OpenAI-compatible AI Gateway configuration from database.
@@ -366,7 +370,6 @@ def get_ai_gateway_config() -> Dict[str, Any]:
         "embedding_model": embedding_model,
     }
 
-
 def set_ai_gateway_config(
     url: Optional[str] = None,
     api_key: Optional[str] = None,
@@ -397,40 +400,57 @@ def set_ai_gateway_config(
 
     return get_ai_gateway_config()
 
-
 def ensure_git_repositories_columns(conn=None):
     """Ensures keep_shallow column exists on git_repositories table."""
     try:
-        raw_conn = get_db_connection()
-        try:
-            cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(git_repositories)").fetchall()]
-            if cols and "keep_shallow" not in cols:
-                raw_conn.execute("ALTER TABLE git_repositories ADD COLUMN keep_shallow INTEGER DEFAULT 0")
-                raw_conn.commit()
-                logger.info("Migrated git_repositories: added keep_shallow column")
-        finally:
-            raw_conn.close()
+        if is_postgres():
+            from sqlalchemy import text
+            with get_connection() as sa_conn:
+                res = sa_conn.execute(
+                    text("SELECT column_name FROM information_schema.columns WHERE table_name='git_repositories' AND column_name='keep_shallow'")
+                ).fetchone()
+                if not res:
+                    sa_conn.execute(text("ALTER TABLE git_repositories ADD COLUMN keep_shallow INTEGER DEFAULT 0"))
+                    sa_conn.commit()
+                    logger.info("Migrated git_repositories: added keep_shallow column")
+        else:
+            raw_conn = get_db_connection()
+            try:
+                cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(git_repositories)").fetchall()]
+                if cols and "keep_shallow" not in cols:
+                    raw_conn.execute("ALTER TABLE git_repositories ADD COLUMN keep_shallow INTEGER DEFAULT 0")
+                    raw_conn.commit()
+                    logger.info("Migrated git_repositories: added keep_shallow column")
+            finally:
+                raw_conn.close()
     except Exception as e:
         logger.debug(f"Migration error for git_repositories: {e}")
 
-
 def ensure_file_summaries_columns(conn=None):
     """Ensures summary_text column exists on file_summaries table and git_repositories has keep_shallow."""
     try:
         ensure_git_repositories_columns(conn)
-        raw_conn = get_db_connection()
-        try:
-            cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(file_summaries)").fetchall()]
-            if cols and "summary_text" not in cols:
-                raw_conn.execute("ALTER TABLE file_summaries ADD COLUMN summary_text TEXT")
-                raw_conn.commit()
-        finally:
-            raw_conn.close()
+        if is_postgres():
+            from sqlalchemy import text
+            with get_connection() as sa_conn:
+                res = sa_conn.execute(
+                    text("SELECT column_name FROM information_schema.columns WHERE table_name='file_summaries' AND column_name='summary_text'")
+                ).fetchone()
+                if not res:
+                    sa_conn.execute(text("ALTER TABLE file_summaries ADD COLUMN summary_text TEXT"))
+                    sa_conn.commit()
+        else:
+            raw_conn = get_db_connection()
+            try:
+                cols = [r["name"] for r in raw_conn.execute("PRAGMA table_info(file_summaries)").fetchall()]
+                if cols and "summary_text" not in cols:
+                    raw_conn.execute("ALTER TABLE file_summaries ADD COLUMN summary_text TEXT")
+                    raw_conn.commit()
+            finally:
+                raw_conn.close()
     except Exception as e:
         logger.debug(f"Migration error for file_summaries: {e}")
 
-
-
 def get_file_settings() -> Dict[str, Any]:
     """Retrieves file reading and large file summarization settings."""
     enabled_val = get_metadata("file_summary_enabled")
@@ -447,7 +467,6 @@ def get_file_settings() -> Dict[str, Any]:
         "summary_chat_model": (model_val or get_chat_model()).strip(),
     }
 
-
 def set_file_settings(payload: Dict[str, Any]) -> Dict[str, Any]:
     """Updates file reading and summarization settings."""
     if "summary_enabled" in payload:

app/services/database/credentials.py

@@ -33,8 +33,12 @@ def save_git_host_credential(host: str, provider: str, auth_token: str, auth_use
         with get_db_connection() as conn:
             cur = conn.cursor()
             cur.execute(
-                """INSERT OR REPLACE INTO git_host_credentials (host, provider, auth_user, auth_token)
-                   VALUES (?, ?, ?, ?)""",
+                """INSERT INTO git_host_credentials (host, provider, auth_user, auth_token)
+                   VALUES (?, ?, ?, ?)
+                   ON CONFLICT(host) DO UPDATE SET
+                       provider = EXCLUDED.provider,
+                       auth_user = EXCLUDED.auth_user,
+                       auth_token = EXCLUDED.auth_token""",
                 (clean_host, provider.lower().strip(), auth_user.strip() if auth_user else None, auth_token.strip())
             )
             conn.commit()

app/services/database/embedding_cache.py

@@ -47,8 +47,12 @@ def set_cached_embeddings_batch(items: List[Dict[str, Any]]) -> None:
         ))
     with get_db_connection() as conn:
         conn.executemany(
-            """INSERT OR REPLACE INTO embedding_cache (chunk_hash, dense_vector, sparse_indices, sparse_values, model_name)
-               VALUES (?, ?, ?, ?, ?)""",
+            """INSERT INTO embedding_cache (chunk_hash, dense_vector, sparse_indices, sparse_values, model_name)
+               VALUES (?, ?, ?, ?, ?)
+               ON CONFLICT(chunk_hash, model_name) DO UPDATE SET
+                   dense_vector = EXCLUDED.dense_vector,
+                   sparse_indices = EXCLUDED.sparse_indices,
+                   sparse_values = EXCLUDED.sparse_values""",
             rows_to_insert
         )
         conn.commit()