"""Schema migration framework for the code-review-graph SQLite database. Manages incremental schema changes via versioned migration functions. Each migration is idempotent (uses IF NOT EXISTS / column existence checks). """ from __future__ import annotations import logging import sqlite3 from typing import Callable logger = logging.getLogger(__name__) def get_schema_version(conn: sqlite3.Connection) -> int: """Read the current schema version from the metadata table. Returns: int: The schema version (0 if metadata table doesn't exist, 1 if not set). """ try: row = conn.execute( "SELECT value FROM metadata WHERE key = 'schema_version'" ).fetchone() if row is None: return 1 return int(row[0] if isinstance(row, (tuple, list)) else row["value"]) except sqlite3.OperationalError: # metadata table doesn't exist return 0 def _set_schema_version(conn: sqlite3.Connection, version: int) -> None: """Set the schema version in the metadata table.""" conn.execute( "INSERT OR REPLACE INTO metadata (key, value) VALUES ('schema_version', ?)", (str(version),), ) _KNOWN_TABLES = frozenset({ "nodes", "edges", "metadata", "communities", "flows", "flow_memberships", "nodes_fts", "community_summaries", "flow_snapshots", "risk_index", }) def _has_column(conn: sqlite3.Connection, table: str, column: str) -> bool: """Check if a column exists in a table.""" if table not in _KNOWN_TABLES: raise ValueError(f"Unknown table: {table}") cursor = conn.execute(f"PRAGMA table_info({table})") # noqa: S608 columns = [row[1] if isinstance(row, tuple) else row["name"] for row in cursor] return column in columns def _table_exists(conn: sqlite3.Connection, table: str) -> bool: """Check if a table exists.""" if table not in _KNOWN_TABLES: raise ValueError(f"Unknown table: {table}") row = conn.execute( "SELECT count(*) FROM sqlite_master WHERE type IN ('table', 'view') " "AND name = ?", (table,), ).fetchone() return row[0] > 0 # --------------------------------------------------------------------------- # Migration functions # --------------------------------------------------------------------------- def _migrate_v2(conn: sqlite3.Connection) -> None: """v2: Add signature column to nodes table.""" if not _has_column(conn, "nodes", "signature"): conn.execute("ALTER TABLE nodes ADD COLUMN signature TEXT") logger.info("Migration v2: added 'signature' column to nodes") def _migrate_v3(conn: sqlite3.Connection) -> None: """v3: Create flows and flow_memberships tables.""" conn.execute(""" CREATE TABLE IF NOT EXISTS flows ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, entry_point_id INTEGER NOT NULL, depth INTEGER NOT NULL, node_count INTEGER NOT NULL, file_count INTEGER NOT NULL, criticality REAL NOT NULL DEFAULT 0.0, path_json TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS flow_memberships ( flow_id INTEGER NOT NULL, node_id INTEGER NOT NULL, position INTEGER NOT NULL, PRIMARY KEY (flow_id, node_id) ) """) conn.execute( "CREATE INDEX IF NOT EXISTS idx_flows_criticality ON flows(criticality DESC)" ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_flows_entry ON flows(entry_point_id)" ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_flow_memberships_node ON flow_memberships(node_id)" ) logger.info("Migration v3: created flows and flow_memberships tables") def _migrate_v4(conn: sqlite3.Connection) -> None: """v4: Create communities table, add community_id to nodes.""" conn.execute(""" CREATE TABLE IF NOT EXISTS communities ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, level INTEGER NOT NULL DEFAULT 0, parent_id INTEGER, cohesion REAL NOT NULL DEFAULT 0.0, size INTEGER NOT NULL DEFAULT 0, dominant_language TEXT, description TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) ) """) if not _has_column(conn, "nodes", "community_id"): conn.execute("ALTER TABLE nodes ADD COLUMN community_id INTEGER") logger.info("Migration v4: added 'community_id' column to nodes") conn.execute( "CREATE INDEX IF NOT EXISTS idx_nodes_community ON nodes(community_id)" ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_communities_parent ON communities(parent_id)" ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_communities_cohesion ON communities(cohesion DESC)" ) logger.info("Migration v4: created communities table") def _migrate_v5(conn: sqlite3.Connection) -> None: """v5: Create FTS5 virtual table for nodes.""" if not _table_exists(conn, "nodes_fts"): conn.execute(""" CREATE VIRTUAL TABLE nodes_fts USING fts5( name, qualified_name, file_path, signature, content='nodes', content_rowid='rowid', tokenize='porter unicode61' ) """) logger.info("Migration v5: created nodes_fts FTS5 virtual table") def _migrate_v6(conn: sqlite3.Connection) -> None: """v6: Add pre-computed summary tables for token-efficient queries.""" conn.execute(""" CREATE TABLE IF NOT EXISTS community_summaries ( community_id INTEGER PRIMARY KEY, name TEXT NOT NULL, purpose TEXT DEFAULT '', key_symbols TEXT DEFAULT '[]', risk TEXT DEFAULT 'unknown', size INTEGER DEFAULT 0, dominant_language TEXT DEFAULT '', FOREIGN KEY (community_id) REFERENCES communities(id) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS flow_snapshots ( flow_id INTEGER PRIMARY KEY, name TEXT NOT NULL, entry_point TEXT NOT NULL, critical_path TEXT DEFAULT '[]', criticality REAL DEFAULT 0.0, node_count INTEGER DEFAULT 0, file_count INTEGER DEFAULT 0, FOREIGN KEY (flow_id) REFERENCES flows(id) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS risk_index ( node_id INTEGER PRIMARY KEY, qualified_name TEXT NOT NULL, risk_score REAL DEFAULT 0.0, caller_count INTEGER DEFAULT 0, test_coverage TEXT DEFAULT 'unknown', security_relevant INTEGER DEFAULT 0, last_computed TEXT DEFAULT '', FOREIGN KEY (node_id) REFERENCES nodes(id) ) """) conn.execute( "CREATE INDEX IF NOT EXISTS idx_risk_index_score " "ON risk_index(risk_score DESC)" ) logger.info("Migration v6: created summary tables " "(community_summaries, flow_snapshots, risk_index)") def _migrate_v7(conn: sqlite3.Connection) -> None: """v7: Add compound edge indexes for summary and risk queries.""" conn.execute( "CREATE INDEX IF NOT EXISTS idx_edges_target_kind " "ON edges(target_qualified, kind)" ) conn.execute( "CREATE INDEX IF NOT EXISTS idx_edges_source_kind " "ON edges(source_qualified, kind)" ) logger.info("Migration v7: added compound edge indexes") def _migrate_v8(conn: sqlite3.Connection) -> None: """v8: Add composite index on edges for upsert_edge performance.""" conn.execute(""" CREATE INDEX IF NOT EXISTS idx_edges_composite ON edges(kind, source_qualified, target_qualified, file_path, line) """) logger.info("Migration v8: created composite edge index") def _migrate_v9(conn: sqlite3.Connection) -> None: """v9: Add confidence scoring to edges.""" if not _has_column(conn, "edges", "confidence"): conn.execute( "ALTER TABLE edges ADD COLUMN confidence REAL DEFAULT 1.0" ) if not _has_column(conn, "edges", "confidence_tier"): conn.execute( "ALTER TABLE edges ADD COLUMN confidence_tier TEXT DEFAULT 'EXTRACTED'" ) logger.info("Migration v9: added edge confidence columns") # --------------------------------------------------------------------------- # Migration registry # --------------------------------------------------------------------------- MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = { 2: _migrate_v2, 3: _migrate_v3, 4: _migrate_v4, 5: _migrate_v5, 6: _migrate_v6, 7: _migrate_v7, 8: _migrate_v8, 9: _migrate_v9, } LATEST_VERSION = max(MIGRATIONS.keys()) def run_migrations(conn: sqlite3.Connection) -> None: """Run all pending migrations in order. Each migration runs in its own transaction. The schema_version metadata entry is updated after each successful migration. """ current = get_schema_version(conn) if current >= LATEST_VERSION: return logger.info("Schema version %d -> %d: running migrations", current, LATEST_VERSION) for version in sorted(MIGRATIONS.keys()): if version <= current: continue logger.info("Running migration v%d", version) try: MIGRATIONS[version](conn) _set_schema_version(conn, version) conn.commit() except sqlite3.Error: conn.rollback() logger.error("Migration v%d failed, rolling back", version, exc_info=True) raise logger.info("Migrations complete, now at schema version %d", LATEST_VERSION)