|
| 1 | +""" |
| 2 | +Neo4j read model for the repository dependency graph. |
| 3 | +
|
| 4 | +WHY THIS EXISTS |
| 5 | +The graph questions that matter for understanding a codebase are traversals: |
| 6 | +"what breaks if I change this file", "how does auth reach the database", "are there |
| 7 | +import cycles". Those are transitive, and the Python implementation cannot answer them |
| 8 | +-- hops is hard-capped at 2 in the API because each hop rescans the entire edge list, so |
| 9 | +depth is quadratic in Python and a single variable-length pattern in Cypher. |
| 10 | +
|
| 11 | +WHAT IT DOES NOT DO |
| 12 | +This is a read model, not a source of truth. code_dependencies (SQL) remains |
| 13 | +authoritative; this is projected from it at index time and can be rebuilt at any point |
| 14 | +by re-indexing. Every read falls back to the SQL path when Neo4j is unavailable, so |
| 15 | +enabling this cannot take the graph endpoint down. |
| 16 | +
|
| 17 | +Degree and centrality come from COUNT {} subqueries rather than Graph Data Science |
| 18 | +deliberately: the in-database GDS plugin is AuraDB Professional and above, and Aura |
| 19 | +Graph Analytics sessions are an offline batch shape (2GB, one concurrent session, |
| 20 | +30-minute TTL) that does not fit a synchronous request. |
| 21 | +""" |
| 22 | + |
| 23 | +from __future__ import annotations |
| 24 | + |
| 25 | +import logging |
| 26 | +from typing import Any, Dict, List, Optional, Sequence |
| 27 | + |
| 28 | +logger = logging.getLogger(__name__) |
| 29 | + |
| 30 | +# Applied once at startup. MERGE on (repo_id, path) is the hot write pattern, so the |
| 31 | +# uniqueness constraints are also the indexes that make it fast -- without them every |
| 32 | +# MERGE degrades to a label scan. |
| 33 | +SCHEMA_STATEMENTS: Sequence[str] = ( |
| 34 | + "CREATE CONSTRAINT file_unique IF NOT EXISTS " |
| 35 | + "FOR (f:File) REQUIRE (f.repo_id, f.path) IS UNIQUE", |
| 36 | + "CREATE CONSTRAINT module_unique IF NOT EXISTS " |
| 37 | + "FOR (m:Module) REQUIRE (m.repo_id, m.key) IS UNIQUE", |
| 38 | + "CREATE INDEX file_repo IF NOT EXISTS FOR (f:File) ON (f.repo_id)", |
| 39 | + "CREATE INDEX file_module IF NOT EXISTS FOR (f:File) ON (f.repo_id, f.module_key)", |
| 40 | +) |
| 41 | + |
| 42 | +# Batched so a large repository does not build one enormous transaction. |
| 43 | +_INGEST_BATCH = 500 |
| 44 | + |
| 45 | +_MERGE_FILES = """ |
| 46 | +UNWIND $rows AS row |
| 47 | +MERGE (f:File {repo_id: $repo_id, path: row.path}) |
| 48 | +SET f.filename = row.filename, |
| 49 | + f.extension = row.extension, |
| 50 | + f.language = row.language, |
| 51 | + f.loc = row.loc, |
| 52 | + f.module_key = row.module_key |
| 53 | +""" |
| 54 | + |
| 55 | +_MERGE_EDGES = """ |
| 56 | +UNWIND $rows AS row |
| 57 | +MATCH (s:File {repo_id: $repo_id, path: row.source}) |
| 58 | +MATCH (t:File {repo_id: $repo_id, path: row.target}) |
| 59 | +MERGE (s)-[r:IMPORTS {relation: row.relation}]->(t) |
| 60 | +SET r.weight = row.weight, r.confidence = row.confidence |
| 61 | +""" |
| 62 | + |
| 63 | +# COUNT {} rather than GDS degreeCentrality -- see the module docstring. |
| 64 | +_READ_NODES_WITH_DEGREE = """ |
| 65 | +MATCH (f:File {repo_id: $repo_id}) |
| 66 | +RETURN f.path AS path, |
| 67 | + f.language AS language, |
| 68 | + f.loc AS loc, |
| 69 | + f.module_key AS module_key, |
| 70 | + COUNT { (f)-[:IMPORTS]->(:File) } AS out_degree, |
| 71 | + COUNT { (f)<-[:IMPORTS]-(:File) } AS in_degree |
| 72 | +""" |
| 73 | + |
| 74 | +_READ_EDGES = """ |
| 75 | +MATCH (s:File {repo_id: $repo_id})-[r:IMPORTS]->(t:File {repo_id: $repo_id}) |
| 76 | +RETURN s.path AS source, t.path AS target, r.relation AS relation, |
| 77 | + r.weight AS weight, r.confidence AS confidence |
| 78 | +""" |
| 79 | + |
| 80 | +# The capability the Python path structurally cannot provide: arbitrary-depth |
| 81 | +# traversal. $hops is interpolated rather than parameterised because Cypher does not |
| 82 | +# allow a parameter inside a variable-length pattern bound; it is coerced to a bounded |
| 83 | +# int by the caller before it reaches here. |
| 84 | +_REACHABILITY = """ |
| 85 | +MATCH path = (s:File {repo_id: $repo_id, path: $path})-[:IMPORTS*1..%(hops)d]->(t:File) |
| 86 | +RETURN DISTINCT t.path AS path, length(path) AS distance |
| 87 | +ORDER BY distance, path |
| 88 | +""" |
| 89 | + |
| 90 | +_BLAST_RADIUS = """ |
| 91 | +MATCH path = (s:File {repo_id: $repo_id, path: $path})<-[:IMPORTS*1..%(hops)d]-(t:File) |
| 92 | +RETURN DISTINCT t.path AS path, length(path) AS distance |
| 93 | +ORDER BY distance, path |
| 94 | +""" |
| 95 | + |
| 96 | +_SHORTEST_PATH = """ |
| 97 | +MATCH (a:File {repo_id: $repo_id, path: $from_path}), |
| 98 | + (b:File {repo_id: $repo_id, path: $to_path}), |
| 99 | + p = shortestPath((a)-[:IMPORTS*..%(max_hops)d]->(b)) |
| 100 | +RETURN [n IN nodes(p) | n.path] AS path_nodes, length(p) AS distance |
| 101 | +""" |
| 102 | + |
| 103 | +# Import cycles: a real code-health signal the SQL path cannot express at all. |
| 104 | +_CYCLES = """ |
| 105 | +MATCH (f:File {repo_id: $repo_id}) |
| 106 | +MATCH p = (f)-[:IMPORTS*2..%(max_len)d]->(f) |
| 107 | +RETURN [n IN nodes(p) | n.path] AS cycle, length(p) AS size |
| 108 | +ORDER BY size, cycle |
| 109 | +LIMIT $limit |
| 110 | +""" |
| 111 | + |
| 112 | +_DELETE_REPO = """ |
| 113 | +MATCH (n {repo_id: $repo_id}) |
| 114 | +CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 1000 ROWS |
| 115 | +""" |
| 116 | + |
| 117 | + |
| 118 | +class Neo4jGraphStore: |
| 119 | + """ |
| 120 | + Thin async wrapper around the Neo4j driver. |
| 121 | +
|
| 122 | + Holds no state beyond the driver, so it is safe to construct per request; the driver |
| 123 | + itself is a long-lived singleton (see dependencies.get_graph_driver) because it owns |
| 124 | + the connection pool. |
| 125 | + """ |
| 126 | + |
| 127 | + def __init__(self, driver, database: Optional[str] = None): |
| 128 | + self._driver = driver |
| 129 | + self._database = database |
| 130 | + |
| 131 | + async def _run(self, query: str, **params) -> List[Dict[str, Any]]: |
| 132 | + records, _, _ = await self._driver.execute_query( |
| 133 | + query, database_=self._database, **params |
| 134 | + ) |
| 135 | + return [dict(r) for r in records] |
| 136 | + |
| 137 | + # --- lifecycle --------------------------------------------------------------- |
| 138 | + |
| 139 | + async def verify(self) -> bool: |
| 140 | + """True when the server is reachable and authenticated.""" |
| 141 | + try: |
| 142 | + await self._driver.verify_connectivity() |
| 143 | + return True |
| 144 | + except Exception as exc: |
| 145 | + logger.warning("Neo4j connectivity check failed: %s", exc) |
| 146 | + return False |
| 147 | + |
| 148 | + async def ensure_schema(self) -> None: |
| 149 | + """Idempotent; every statement is IF NOT EXISTS.""" |
| 150 | + for statement in SCHEMA_STATEMENTS: |
| 151 | + await self._run(statement) |
| 152 | + |
| 153 | + # --- ingest ------------------------------------------------------------------ |
| 154 | + |
| 155 | + async def sync_repository( |
| 156 | + self, |
| 157 | + repo_id: str, |
| 158 | + files: Sequence[Dict[str, Any]], |
| 159 | + edges: Sequence[Dict[str, Any]], |
| 160 | + ) -> Dict[str, int]: |
| 161 | + """ |
| 162 | + Project a repository's files and edges into the graph. |
| 163 | +
|
| 164 | + Replaces rather than merges: the previous subgraph is deleted first, because a |
| 165 | + MERGE-only sync would leave edges for files that no longer exist and silently |
| 166 | + accumulate a graph that no commit ever had. |
| 167 | + """ |
| 168 | + await self.delete_repository(repo_id) |
| 169 | + |
| 170 | + for start in range(0, len(files), _INGEST_BATCH): |
| 171 | + await self._run( |
| 172 | + _MERGE_FILES, repo_id=repo_id, rows=list(files[start:start + _INGEST_BATCH]) |
| 173 | + ) |
| 174 | + for start in range(0, len(edges), _INGEST_BATCH): |
| 175 | + await self._run( |
| 176 | + _MERGE_EDGES, repo_id=repo_id, rows=list(edges[start:start + _INGEST_BATCH]) |
| 177 | + ) |
| 178 | + |
| 179 | + logger.info( |
| 180 | + "Synced %d files and %d edges into Neo4j for repo %s", |
| 181 | + len(files), len(edges), repo_id, |
| 182 | + ) |
| 183 | + return {"files": len(files), "edges": len(edges)} |
| 184 | + |
| 185 | + async def delete_repository(self, repo_id: str) -> None: |
| 186 | + """Remove a repository's subgraph. Called on re-sync and on repo deletion.""" |
| 187 | + await self._run(_DELETE_REPO, repo_id=repo_id) |
| 188 | + |
| 189 | + # --- reads ------------------------------------------------------------------- |
| 190 | + |
| 191 | + async def nodes_with_degree(self, repo_id: str) -> List[Dict[str, Any]]: |
| 192 | + return await self._run(_READ_NODES_WITH_DEGREE, repo_id=repo_id) |
| 193 | + |
| 194 | + async def edges(self, repo_id: str) -> List[Dict[str, Any]]: |
| 195 | + return await self._run(_READ_EDGES, repo_id=repo_id) |
| 196 | + |
| 197 | + # --- traversals the SQL path cannot answer ----------------------------------- |
| 198 | + |
| 199 | + @staticmethod |
| 200 | + def _bounded(value: int, lo: int, hi: int) -> int: |
| 201 | + return max(lo, min(hi, int(value))) |
| 202 | + |
| 203 | + async def reachable_from(self, repo_id: str, path: str, hops: int = 3) -> List[Dict[str, Any]]: |
| 204 | + """What this file transitively imports.""" |
| 205 | + q = _REACHABILITY % {"hops": self._bounded(hops, 1, 10)} |
| 206 | + return await self._run(q, repo_id=repo_id, path=path) |
| 207 | + |
| 208 | + async def blast_radius(self, repo_id: str, path: str, hops: int = 3) -> List[Dict[str, Any]]: |
| 209 | + """What transitively imports this file -- i.e. what a change here can affect.""" |
| 210 | + q = _BLAST_RADIUS % {"hops": self._bounded(hops, 1, 10)} |
| 211 | + return await self._run(q, repo_id=repo_id, path=path) |
| 212 | + |
| 213 | + async def shortest_path( |
| 214 | + self, repo_id: str, from_path: str, to_path: str, max_hops: int = 10 |
| 215 | + ) -> Optional[Dict[str, Any]]: |
| 216 | + q = _SHORTEST_PATH % {"max_hops": self._bounded(max_hops, 1, 15)} |
| 217 | + rows = await self._run(q, repo_id=repo_id, from_path=from_path, to_path=to_path) |
| 218 | + return rows[0] if rows else None |
| 219 | + |
| 220 | + async def import_cycles( |
| 221 | + self, repo_id: str, max_length: int = 6, limit: int = 20 |
| 222 | + ) -> List[Dict[str, Any]]: |
| 223 | + q = _CYCLES % {"max_len": self._bounded(max_length, 2, 10)} |
| 224 | + return await self._run(q, repo_id=repo_id, limit=max(1, int(limit))) |
0 commit comments