Skip to main content

ctx/analytics/
duckdb_impl.rs

1//! Analytics module using DuckDB for complex graph queries.
2//!
3//! This module provides:
4//! - Recursive call graph traversal
5//! - Impact analysis (reverse call graph)
6//! - Module dependency analysis
7//! - Codebase statistics aggregations
8
9use std::path::Path;
10use std::sync::Arc;
11
12use duckdb::types::ValueRef;
13use duckdb::{params, Connection, InterruptHandle, Result};
14
15use super::{CallGraphNode, ComplexityResult, FileStats, ImpactNode};
16use crate::index::{CTX_DIR, DB_FILE};
17
18/// DuckDB analytics engine for code intelligence.
19pub struct Analytics {
20    conn: Connection,
21}
22
23impl Analytics {
24    /// Open DuckDB and attach the SQLite database.
25    pub fn open(root: &Path) -> Result<Self> {
26        let ctx_dir = root.join(CTX_DIR);
27        let sqlite_path = ctx_dir.join(DB_FILE);
28
29        // Create in-memory DuckDB and attach SQLite
30        let conn = Connection::open_in_memory()?;
31
32        // Attach SQLite database with properly escaped path
33        // DuckDB uses single quotes for strings, so we need to escape any single quotes in the path
34        let path_str = sqlite_path.display().to_string();
35        let escaped_path = path_str.replace('\'', "''");
36        conn.execute(
37            &format!("ATTACH '{}' AS code (TYPE sqlite, READ_ONLY)", escaped_path),
38            [],
39        )?;
40
41        // Create materialized views for better performance
42        let analytics = Self { conn };
43        analytics.create_materialized_views()?;
44
45        Ok(analytics)
46    }
47
48    /// Create materialized views for common analytical queries.
49    fn create_materialized_views(&self) -> Result<()> {
50        // Call graph view - joins edges with symbols for fast traversal
51        self.conn.execute_batch(
52            r#"
53            CREATE OR REPLACE VIEW call_graph AS
54            SELECT 
55                e.source_id,
56                e.target_name,
57                e.target_id,
58                e.kind as edge_kind,
59                e.line,
60                e.context,
61                s.file_path as source_file,
62                s.name as source_name,
63                s.kind as source_kind,
64                t.file_path as target_file,
65                t.name as target_name_resolved,
66                t.kind as target_kind
67            FROM code.edges e
68            JOIN code.symbols s ON e.source_id = s.id
69            LEFT JOIN code.symbols t ON e.target_id = t.id
70            WHERE e.kind = 'calls';
71
72            -- File statistics view
73            CREATE OR REPLACE VIEW file_stats AS
74            SELECT 
75                file_path,
76                COUNT(*) as symbol_count,
77                COUNT(*) FILTER (WHERE kind IN ('function', 'method')) as functions,
78                COUNT(*) FILTER (WHERE kind = 'struct') as structs,
79                COUNT(*) FILTER (WHERE kind = 'enum') as enums,
80                COUNT(*) FILTER (WHERE visibility = 'public') as public_symbols
81            FROM code.symbols
82            GROUP BY file_path;
83
84            -- Symbol summary view
85            CREATE OR REPLACE VIEW symbol_summary AS
86            SELECT 
87                kind,
88                COUNT(*) as count,
89                COUNT(*) FILTER (WHERE visibility = 'public') as public_count
90            FROM code.symbols
91            GROUP BY kind
92            ORDER BY count DESC;
93            "#,
94        )?;
95
96        Ok(())
97    }
98
99    /// Get the full call graph starting from a symbol (forward traversal).
100    ///
101    /// The `start_name` can be:
102    /// - A simple name like "new" (matches first symbol with that name)
103    /// - A qualified name like "LocalProvider::new"
104    /// - A full ID like "src/embeddings/local.rs::LocalProvider::new@20"
105    pub fn call_graph(&self, start_name: &str, max_depth: i32) -> Result<Vec<CallGraphNode>> {
106        let mut stmt = self.conn.prepare(
107            r#"
108            WITH RECURSIVE start_symbol AS (
109                -- Resolve the starting symbol by ID, qualified name, or name
110                SELECT id, name, file_path, kind
111                FROM code.symbols
112                WHERE id = ?
113                   OR qualified_name = ?
114                   OR (name = ? AND NOT EXISTS (
115                       SELECT 1 FROM code.symbols s2 
116                       WHERE s2.id = ? OR s2.qualified_name = ?
117                   ))
118                ORDER BY file_path, line_start
119                LIMIT 1
120            ),
121            graph AS (
122                -- Base case: start from resolved symbol
123                SELECT 
124                    s.name,
125                    s.file_path,
126                    s.kind,
127                    1 as depth,
128                    s.id as current_id
129                FROM code.symbols s
130                JOIN start_symbol ss ON s.id = ss.id
131                
132                UNION ALL
133                
134                -- Recursive case: follow outgoing edges (only resolved edges)
135                SELECT 
136                    t.name,
137                    t.file_path,
138                    t.kind,
139                    g.depth + 1 as depth,
140                    t.id as current_id
141                FROM graph g
142                JOIN code.edges e ON e.source_id = g.current_id
143                JOIN code.symbols t ON e.target_id = t.id
144                WHERE g.depth < ?
145                  AND e.kind = 'calls'
146                  AND e.target_id IS NOT NULL
147            )
148            SELECT DISTINCT name, file_path, kind, MIN(depth) as depth
149            FROM graph
150            WHERE depth > 1  -- Exclude the starting symbol itself
151            GROUP BY name, file_path, kind
152            ORDER BY depth, name
153            "#,
154        )?;
155
156        let rows = stmt.query_map(
157            params![start_name, start_name, start_name, start_name, start_name, max_depth],
158            |row| {
159                Ok(CallGraphNode {
160                    name: row.get(0)?,
161                    file_path: row.get(1)?,
162                    kind: row.get(2)?,
163                    depth: row.get(3)?,
164                })
165            },
166        )?;
167
168        rows.collect()
169    }
170
171    /// Impact analysis: find all symbols that would be affected by changing the target.
172    ///
173    /// The `target_name` can be:
174    /// - A simple name like "new" (matches first symbol with that name)
175    /// - A qualified name like "LocalProvider::new"
176    /// - A full ID like "src/embeddings/local.rs::LocalProvider::new@20"
177    pub fn impact_analysis(&self, target_name: &str, max_depth: i32) -> Result<Vec<ImpactNode>> {
178        let mut stmt = self.conn.prepare(
179            r#"
180            WITH RECURSIVE target_symbol AS (
181                -- Resolve the target symbol by ID, qualified name, or name
182                SELECT id, name, file_path
183                FROM code.symbols
184                WHERE id = ?
185                   OR qualified_name = ?
186                   OR (name = ? AND NOT EXISTS (
187                       SELECT 1 FROM code.symbols s2 
188                       WHERE s2.id = ? OR s2.qualified_name = ?
189                   ))
190                ORDER BY file_path, line_start
191                LIMIT 1
192            ),
193            impact AS (
194                -- Base case: find direct callers of the specific target symbol
195                SELECT 
196                    s.name,
197                    s.file_path,
198                    s.kind,
199                    1 as distance,
200                    s.id as current_id
201                FROM code.edges e
202                JOIN code.symbols s ON e.source_id = s.id
203                JOIN target_symbol ts ON 
204                    -- Only traverse resolved edges to avoid cross-file false positives
205                    e.target_id IS NOT NULL AND e.target_id = ts.id
206                WHERE e.kind = 'calls'
207                
208                UNION ALL
209                
210                -- Recursive case: find callers of callers (reverse traversal)
211                SELECT 
212                    s.name,
213                    s.file_path,
214                    s.kind,
215                    i.distance + 1 as distance,
216                    s.id as current_id
217                FROM impact i
218                JOIN code.edges e ON 
219                    -- Only traverse resolved edges to avoid cross-file false positives
220                    e.target_id IS NOT NULL AND e.target_id = i.current_id
221                JOIN code.symbols s ON e.source_id = s.id
222                WHERE i.distance < ?
223                  AND e.kind = 'calls'
224            )
225            SELECT DISTINCT name, file_path, kind, MIN(distance) as distance
226            FROM impact
227            GROUP BY name, file_path, kind
228            ORDER BY distance, name
229            "#,
230        )?;
231
232        let rows = stmt.query_map(
233            params![
234                target_name,
235                target_name,
236                target_name,
237                target_name,
238                target_name,
239                max_depth
240            ],
241            |row| {
242                Ok(ImpactNode {
243                    name: row.get(0)?,
244                    file_path: row.get(1)?,
245                    kind: row.get(2)?,
246                    distance: row.get(3)?,
247                })
248            },
249        )?;
250
251        rows.collect()
252    }
253
254    /// Get file statistics for all indexed files.
255    pub fn file_statistics(&self) -> Result<Vec<FileStats>> {
256        let mut stmt = self.conn.prepare(
257            r#"
258            SELECT file_path, symbol_count, functions, structs, enums, public_symbols
259            FROM file_stats
260            ORDER BY symbol_count DESC
261            "#,
262        )?;
263
264        let rows = stmt.query_map([], |row| {
265            Ok(FileStats {
266                file_path: row.get(0)?,
267                symbol_count: row.get(1)?,
268                functions: row.get(2)?,
269                structs: row.get(3)?,
270                enums: row.get(4)?,
271                public_symbols: row.get(5)?,
272            })
273        })?;
274
275        rows.collect()
276    }
277
278    /// Get symbol counts by kind.
279    #[allow(dead_code)]
280    pub fn symbol_summary(&self) -> Result<Vec<(String, i64, i64)>> {
281        let mut stmt = self.conn.prepare(
282            r#"
283            SELECT kind, count, public_count
284            FROM symbol_summary
285            "#,
286        )?;
287
288        let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?;
289
290        rows.collect()
291    }
292
293    /// Find if a path exists between two symbols (simplified version).
294    ///
295    /// Both `from_name` and `to_name` can be symbol IDs, qualified names, or simple names.
296    #[allow(dead_code)]
297    pub fn has_path(&self, from_name: &str, to_name: &str, max_depth: i32) -> Result<bool> {
298        let mut stmt = self.conn.prepare(
299            r#"
300            WITH RECURSIVE source_symbol AS (
301                SELECT id, name FROM code.symbols
302                WHERE id = ? OR qualified_name = ? OR name = ?
303                ORDER BY file_path
304                LIMIT 1
305            ),
306            target_symbol AS (
307                SELECT id, name FROM code.symbols
308                WHERE id = ? OR qualified_name = ? OR name = ?
309                ORDER BY file_path
310                LIMIT 1
311            ),
312            reachable AS (
313                -- Base case: start from source
314                SELECT 
315                    s.name,
316                    s.id as current_id,
317                    1 as depth
318                FROM code.symbols s
319                JOIN source_symbol src ON s.id = src.id
320                
321                UNION
322                
323                -- Follow edges using only resolved target_id to avoid cross-file false positives
324                SELECT 
325                    t.name,
326                    t.id,
327                    r.depth + 1
328                FROM reachable r
329                JOIN code.edges e ON e.source_id = r.current_id
330                JOIN code.symbols t ON e.target_id = t.id
331                WHERE r.depth < ?
332                  AND e.kind = 'calls'
333                  AND e.target_id IS NOT NULL
334            )
335            SELECT COUNT(*) > 0
336            FROM reachable r
337            JOIN target_symbol tgt ON r.current_id = tgt.id
338            "#,
339        )?;
340
341        let result: bool = stmt.query_row(
342            params![from_name, from_name, from_name, to_name, to_name, to_name, max_depth],
343            |row| row.get(0),
344        )?;
345        Ok(result)
346    }
347
348    /// Get the most connected symbols (highest in/out degree).
349    ///
350    /// This correctly counts incoming edges per symbol ID, not by name,
351    /// avoiding over-counting for common names like "new".
352    pub fn most_connected(&self, limit: i32) -> Result<Vec<(String, String, i64, i64)>> {
353        let mut stmt = self.conn.prepare(
354            r#"
355            WITH outgoing AS (
356                SELECT source_id, COUNT(*) as out_degree
357                FROM code.edges
358                WHERE kind = 'calls'
359                GROUP BY source_id
360            ),
361            incoming AS (
362                -- Count incoming edges by target_id when available
363                SELECT target_id as id, COUNT(*) as in_degree
364                FROM code.edges
365                WHERE kind = 'calls' AND target_id IS NOT NULL
366                GROUP BY target_id
367            )
368            SELECT 
369                s.name,
370                s.file_path,
371                COALESCE(o.out_degree, 0) as out_degree,
372                COALESCE(i.in_degree, 0) as in_degree
373            FROM code.symbols s
374            LEFT JOIN outgoing o ON s.id = o.source_id
375            LEFT JOIN incoming i ON s.id = i.id
376            WHERE s.kind IN ('function', 'method')
377            ORDER BY (COALESCE(o.out_degree, 0) + COALESCE(i.in_degree, 0)) DESC
378            LIMIT ?
379            "#,
380        )?;
381
382        let rows = stmt.query_map(params![limit], |row| {
383            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
384        })?;
385
386        rows.collect()
387    }
388
389    /// Check if there are any self-recursive functions.
390    ///
391    /// Uses target_id when available for accurate recursion detection.
392    #[allow(dead_code)]
393    pub fn find_recursive_functions(&self) -> Result<Vec<(String, String)>> {
394        let mut stmt = self.conn.prepare(
395            r#"
396            SELECT DISTINCT s.name, s.file_path
397            FROM code.edges e
398            JOIN code.symbols s ON e.source_id = s.id
399            WHERE e.kind = 'calls'
400              AND (
401                  (e.target_id IS NOT NULL AND e.target_id = s.id)
402                  OR (e.target_id IS NULL AND e.target_name = s.name)
403              )
404            ORDER BY s.name
405            "#,
406        )?;
407
408        let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?;
409        rows.collect()
410    }
411
412    /// Get dependency graph between files (module-level).
413    ///
414    /// Uses target_id when available for accurate file resolution.
415    pub fn file_dependencies(&self) -> Result<Vec<(String, String, i64)>> {
416        let mut stmt = self.conn.prepare(
417            r#"
418            SELECT 
419                s.file_path as source_file,
420                COALESCE(t.file_path, 'external') as target_file,
421                COUNT(*) as call_count
422            FROM code.edges e
423            JOIN code.symbols s ON e.source_id = s.id
424            LEFT JOIN code.symbols t ON 
425                (e.target_id IS NOT NULL AND e.target_id = t.id)
426                OR (e.target_id IS NULL AND e.target_name = t.name)
427            WHERE e.kind = 'calls'
428              AND s.file_path != COALESCE(t.file_path, '')
429            GROUP BY s.file_path, COALESCE(t.file_path, 'external')
430            ORDER BY call_count DESC
431            "#,
432        )?;
433
434        let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?;
435
436        rows.collect()
437    }
438
439    /// Analyze code complexity based on fan-out (outgoing calls) and fan-in (incoming calls).
440    ///
441    /// This correctly counts incoming edges per symbol ID, not by name,
442    /// avoiding over-counting for common names like "new".
443    pub fn complexity_analysis(&self, threshold: i64) -> Result<Vec<ComplexityResult>> {
444        let mut stmt = self.conn.prepare(
445            r#"
446            WITH outgoing AS (
447                SELECT source_id, COUNT(*) as out_degree
448                FROM code.edges
449                WHERE kind = 'calls'
450                GROUP BY source_id
451            ),
452            incoming AS (
453                -- Count incoming edges by target_id when available
454                SELECT target_id as id, COUNT(*) as in_degree
455                FROM code.edges
456                WHERE kind = 'calls' AND target_id IS NOT NULL
457                GROUP BY target_id
458            )
459            SELECT 
460                s.name,
461                s.file_path,
462                s.line_start,
463                COALESCE(o.out_degree, 0) as fan_out,
464                COALESCE(i.in_degree, 0) as fan_in,
465                -- Complexity score: weighted combination of fan-out (more important) and fan-in
466                (COALESCE(o.out_degree, 0) * 2 + COALESCE(i.in_degree, 0)) as complexity_score,
467                CASE 
468                    WHEN COALESCE(o.out_degree, 0) > 50 THEN 'critical'
469                    WHEN COALESCE(o.out_degree, 0) > 30 THEN 'high'
470                    WHEN COALESCE(o.out_degree, 0) > ? THEN 'medium'
471                    ELSE 'low'
472                END as severity
473            FROM code.symbols s
474            LEFT JOIN outgoing o ON s.id = o.source_id
475            LEFT JOIN incoming i ON s.id = i.id
476            WHERE s.kind IN ('function', 'method')
477            ORDER BY complexity_score DESC
478            "#,
479        )?;
480
481        let rows = stmt.query_map(params![threshold], |row| {
482            Ok(ComplexityResult {
483                name: row.get(0)?,
484                file_path: row.get(1)?,
485                line: row.get(2)?,
486                fan_out: row.get(3)?,
487                fan_in: row.get(4)?,
488                complexity_score: row.get(5)?,
489                severity: row.get(6)?,
490            })
491        })?;
492
493        rows.collect()
494    }
495
496    /// Get the full call graph (all edges with resolved symbols).
497    ///
498    /// Uses target_id when available for accurate symbol resolution.
499    pub fn full_call_graph(
500        &self,
501        _max_depth: i32,
502    ) -> Result<Vec<(String, String, String, String)>> {
503        let mut stmt = self.conn.prepare(
504            r#"
505            SELECT DISTINCT
506                s.file_path as source_file,
507                s.name as source_name,
508                COALESCE(t.file_path, 'external') as target_file,
509                COALESCE(t.name, e.target_name) as target_name
510            FROM code.edges e
511            JOIN code.symbols s ON e.source_id = s.id
512            LEFT JOIN code.symbols t ON 
513                (e.target_id IS NOT NULL AND e.target_id = t.id)
514                OR (e.target_id IS NULL AND e.target_name = t.name)
515            WHERE e.kind = 'calls'
516            ORDER BY s.file_path, s.name
517            "#,
518        )?;
519
520        let rows = stmt.query_map([], |row| {
521            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
522        })?;
523
524        rows.collect()
525    }
526
527    // ========================================================================
528    // Raw SQL surface (`ctx sql`) — hardened, read-only query sandbox.
529    // ========================================================================
530
531    /// Open DuckDB for the `ctx sql` command: attach the SQLite index read-only,
532    /// build the public `v1` view layer, optionally materialize snapshot
533    /// partitions as `snap.*` tables, then lock the engine down so untrusted
534    /// user SQL cannot touch the filesystem, load extensions, or re-enable any
535    /// of that. Safety is enforced entirely by engine configuration — never by
536    /// inspecting the SQL text.
537    ///
538    /// `snapshots`, when set, is a directory of `sha=<sha>/` Parquet
539    /// partitions written by `ctx snapshot`; each partition's four files are
540    /// loaded into in-memory tables `snap.files`, `snap.symbols`,
541    /// `snap.dup_pairs`, and `snap.meta`.
542    pub fn open_sql_sandbox(root: &Path, snapshots: Option<&Path>) -> crate::error::Result<Self> {
543        // Attach the index and build the public `v1` contract views BEFORE
544        // hardening — creating views and reading the attached DB must happen
545        // while access is still allowed.
546        let analytics = Self::open_with_public_schema(root)?;
547
548        // Load snapshot partitions BEFORE hardening, as MATERIALIZED tables
549        // (CREATE TABLE ... AS), not views. This ordering is load-bearing:
550        // `enable_external_access = false` below disables all filesystem reads
551        // at query time, so a lazy view over read_parquet() would fail on its
552        // first use — the Parquet data must be fully read into memory now.
553        if let Some(dir) = snapshots {
554            if !Self::has_snapshot_partitions(dir) {
555                return Err(crate::error::CtxError::Other(format!(
556                    "no snapshots found under {}; run 'ctx snapshot' first",
557                    dir.display()
558                )));
559            }
560            // Single-quote-escape the glob path, like the ATTACH path above.
561            let escaped_dir = dir.display().to_string().replace('\'', "''");
562            analytics.conn.execute_batch("CREATE SCHEMA snap;")?;
563            for table in ["files", "symbols", "dup_pairs", "meta"] {
564                // `hive_partitioning = false`: every row already carries
565                // `commit_sha`; don't add a duplicate `sha` column from the
566                // partition directory name.
567                analytics.conn.execute_batch(&format!(
568                    "CREATE TABLE snap.{table} AS \
569                     SELECT * FROM read_parquet('{escaped_dir}/sha=*/{table}.parquet', \
570                     union_by_name = true, hive_partitioning = false);"
571                ))?;
572            }
573        }
574
575        // Engine-level hardening (order matters: memory + external-access first,
576        // then lock configuration, which blocks any further `SET`).
577        //
578        // `enable_external_access = false` disables the filesystem (COPY,
579        // read_csv, file-based ATTACH), extension installation, and config
580        // changes; `lock_configuration = true` prevents user SQL from re-enabling
581        // any of it. The SQLite index is attached READ_ONLY, so it cannot be
582        // mutated. Note: DuckDB 1.4 has no engine toggle to reject a bare
583        // in-memory `ATTACH ':memory:'`; that is permitted but inert (no
584        // filesystem access, the index stays read-only, and the one-result-set
585        // rule blocks chaining a write onto it). We do not add SQL-text filtering
586        // for it — safety is enforced by engine configuration, not by parsing.
587        let mem_limit =
588            std::env::var("CTX_SQL_MEMORY_LIMIT").unwrap_or_else(|_| "512MB".to_string());
589        let mem_escaped = mem_limit.replace('\'', "''");
590        analytics.conn.execute_batch(&format!(
591            "SET memory_limit = '{}';\n\
592             SET enable_external_access = false;\n\
593             SET lock_configuration = true;",
594            mem_escaped
595        ))?;
596
597        Ok(analytics)
598    }
599
600    /// Whether `dir` contains at least one `sha=<sha>/` snapshot partition.
601    fn has_snapshot_partitions(dir: &Path) -> bool {
602        std::fs::read_dir(dir)
603            .map(|entries| {
604                entries.filter_map(|e| e.ok()).any(|e| {
605                    e.file_name().to_string_lossy().starts_with("sha=") && e.path().is_dir()
606                })
607            })
608            .unwrap_or(false)
609    }
610
611    /// Open DuckDB for snapshot export (`ctx snapshot`): attach the SQLite
612    /// index read-only and build the same public `v1` view layer as
613    /// [`Analytics::open_sql_sandbox`], but with **no** engine hardening
614    /// (`enable_external_access` stays on so `COPY ... TO ... (FORMAT
615    /// PARQUET)` can write partition files).
616    ///
617    /// This is a trusted internal path: it only ever executes SQL composed by
618    /// ctx itself and is never fed user SQL. Anything user-facing must go
619    /// through the hardened [`Analytics::open_sql_sandbox`] instead.
620    pub fn open_export(root: &Path) -> Result<Self> {
621        Self::open_with_public_schema(root)
622    }
623
624    /// Shared constructor for [`Analytics::open_sql_sandbox`] and
625    /// [`Analytics::open_export`]: in-memory DuckDB with the SQLite index
626    /// attached read-only as `code` and the public `v1` views created.
627    /// Performs no hardening — callers decide the trust level.
628    fn open_with_public_schema(root: &Path) -> Result<Self> {
629        let ctx_dir = root.join(CTX_DIR);
630        let sqlite_path = ctx_dir.join(DB_FILE);
631
632        let conn = Connection::open_in_memory()?;
633
634        // Attach the SQLite index read-only (single-quote-escape the path).
635        let path_str = sqlite_path.display().to_string();
636        let escaped_path = path_str.replace('\'', "''");
637        conn.execute(
638            &format!("ATTACH '{}' AS code (TYPE sqlite, READ_ONLY)", escaped_path),
639            [],
640        )?;
641
642        let analytics = Self { conn };
643
644        let index_root = root.display().to_string();
645        analytics.create_public_schema_v1(env!("CARGO_PKG_VERSION"), &index_root)?;
646
647        Ok(analytics)
648    }
649
650    /// Raw connection access for trusted internal callers (snapshot export
651    /// runs ctx-composed DDL/COPY and uses `duckdb::Appender` directly).
652    /// Never expose this to user SQL.
653    pub(crate) fn connection(&self) -> &Connection {
654        &self.conn
655    }
656
657    /// Create the versioned public schema `v1` — the stable query surface.
658    ///
659    /// These views (not the physical `code.*` tables) are the contract. Column
660    /// lists here are the documented `v1` schema; derived columns (fan-in/out,
661    /// complexity) are computed in-view.
662    fn create_public_schema_v1(&self, ctx_version: &str, index_root: &str) -> Result<()> {
663        // Literals injected into `v1.meta`; single-quote-escape defensively.
664        let ctx_version_lit = ctx_version.replace('\'', "''");
665        let index_root_lit = index_root.replace('\'', "''");
666
667        self.conn.execute_batch(&format!(
668            r#"
669            CREATE SCHEMA v1;
670
671            CREATE VIEW v1.symbols AS
672            WITH fan_out_counts AS (
673                SELECT source_id AS id, COUNT(*) AS fan_out
674                FROM code.edges
675                WHERE kind = 'calls'
676                GROUP BY source_id
677            ),
678            fan_in_counts AS (
679                SELECT target_id AS id, COUNT(*) AS fan_in
680                FROM code.edges
681                WHERE kind = 'calls' AND target_id IS NOT NULL
682                GROUP BY target_id
683            )
684            SELECT
685                s.id,
686                s.name,
687                s.qualified_name,
688                s.kind,
689                s.file_path AS file,
690                s.line_start,
691                s.line_end,
692                (s.visibility = 'public') AS is_public,
693                (COALESCE(fo.fan_out, 0) * 2 + COALESCE(fi.fan_in, 0)) AS complexity,
694                COALESCE(fi.fan_in, 0) AS fan_in,
695                COALESCE(fo.fan_out, 0) AS fan_out,
696                COALESCE(s.docstring, s.brief) AS doc
697            FROM code.symbols s
698            LEFT JOIN fan_out_counts fo ON fo.id = s.id
699            LEFT JOIN fan_in_counts fi ON fi.id = s.id;
700
701            CREATE VIEW v1.edges AS
702            SELECT
703                e.source_id,
704                s.name AS source_name,
705                s.file_path AS source_file,
706                e.target_id,
707                e.target_name,
708                t.file_path AS target_file,
709                e.kind,
710                e.line
711            FROM code.edges e
712            JOIN code.symbols s ON e.source_id = s.id
713            LEFT JOIN code.symbols t ON e.target_id = t.id;
714
715            CREATE VIEW v1.files AS
716            SELECT
717                f.path,
718                f.language,
719                COALESCE(agg.symbol_count, 0) AS symbol_count,
720                COALESCE(agg.total_complexity, 0) AS total_complexity,
721                f.last_indexed AS indexed_at
722            FROM code.files f
723            LEFT JOIN (
724                SELECT file, COUNT(*) AS symbol_count, SUM(complexity) AS total_complexity
725                FROM v1.symbols
726                GROUP BY file
727            ) agg ON agg.file = f.path;
728
729            CREATE VIEW v1.meta AS
730            SELECT
731                1 AS schema_version,
732                '{ctx_version}' AS ctx_version,
733                (SELECT MIN(last_indexed) FROM code.files) AS index_created_at,
734                '{index_root}' AS index_root;
735            "#,
736            ctx_version = ctx_version_lit,
737            index_root = index_root_lit,
738        ))?;
739
740        Ok(())
741    }
742
743    /// A handle that can interrupt an in-flight query from another thread
744    /// (used to enforce `--timeout`). `InterruptHandle` is `Send + Sync`.
745    pub fn interrupt_handle(&self) -> Arc<InterruptHandle> {
746        self.conn.interrupt_handle()
747    }
748
749    /// Execute a non-final statement in a multi-statement submission.
750    ///
751    /// Returns `Ok(true)` if the statement produces a result set — the caller
752    /// rejects that (only the final statement may return rows, per the one
753    /// result-set rule). Statements that produce no result set (e.g.
754    /// `CREATE TEMP TABLE …`, `SET …`) are executed for their side effects and
755    /// return `Ok(false)`.
756    ///
757    /// (DuckDB only knows a statement's column count *after* execution, so the
758    /// statement is always run — harmless here, since the index is read-only
759    /// and the in-memory DuckDB is throwaway.)
760    pub fn exec_non_final_statement(&self, sql: &str) -> Result<bool> {
761        let mut stmt = self.conn.prepare(sql)?;
762        let rows = stmt.query([])?;
763        let produced = rows
764            .as_ref()
765            .map(|s| {
766                let n = s.column_count();
767                if n == 0 {
768                    return false;
769                }
770                // DuckDB reports DDL/DML — CREATE/INSERT/UPDATE/DELETE, including
771                // `CREATE TABLE … AS SELECT` — as a single BIGINT column named
772                // "Count" (rows affected). That is not a user-facing result set,
773                // so such statements are allowed to precede the final one.
774                !(n == 1 && s.column_name(0).map(|c| c == "Count").unwrap_or(false))
775            })
776            .unwrap_or(false);
777        Ok(produced)
778    }
779
780    /// Run the final (result-producing) statement, streaming rows and stopping
781    /// once `max_rows` is reached (`0` = unlimited). One extra row is fetched to
782    /// detect truncation, so the full result is never materialized in Rust.
783    pub fn run_final_statement(&self, sql: &str, max_rows: usize) -> Result<SqlResult> {
784        let mut stmt = self.conn.prepare(sql)?;
785        let mut rows = stmt.query([])?;
786
787        // Column metadata is only available after the statement has executed.
788        let columns: Vec<SqlColumn> = {
789            let executed = rows.as_ref().expect("statement is available after query()");
790            let col_count = executed.column_count();
791            let names = executed.column_names();
792            (0..col_count)
793                .map(|i| SqlColumn {
794                    name: names.get(i).cloned().unwrap_or_default(),
795                    type_name: duckdb_type_name(&executed.column_type(i)),
796                })
797                .collect()
798        };
799        let col_count = columns.len();
800
801        let cap = if max_rows == 0 { usize::MAX } else { max_rows };
802        let mut rows_out: Vec<Vec<serde_json::Value>> = Vec::new();
803        let mut truncated = false;
804
805        while let Some(row) = rows.next()? {
806            if rows_out.len() >= cap {
807                truncated = true;
808                break;
809            }
810            let mut record = Vec::with_capacity(col_count);
811            for i in 0..col_count {
812                record.push(value_ref_to_json(row.get_ref(i)?));
813            }
814            rows_out.push(record);
815        }
816
817        Ok(SqlResult {
818            columns,
819            rows: rows_out,
820            truncated,
821        })
822    }
823}
824
825/// A column in a `ctx sql` result: its name and DuckDB type name.
826#[derive(Debug, Clone)]
827pub struct SqlColumn {
828    pub name: String,
829    pub type_name: String,
830}
831
832/// The result of a `ctx sql` query: columns plus row-capped data.
833#[derive(Debug, Clone)]
834pub struct SqlResult {
835    pub columns: Vec<SqlColumn>,
836    pub rows: Vec<Vec<serde_json::Value>>,
837    pub truncated: bool,
838}
839
840/// Map an Arrow `DataType` (what duckdb-rs reports for a column via
841/// `column_type`) to a DuckDB type name for the JSON envelope (e.g. `VARCHAR`,
842/// `BIGINT`).
843fn duckdb_type_name(dt: &duckdb::arrow::datatypes::DataType) -> String {
844    use duckdb::arrow::datatypes::DataType as D;
845    match dt {
846        D::Null => "NULL",
847        D::Boolean => "BOOLEAN",
848        D::Int8 => "TINYINT",
849        D::Int16 => "SMALLINT",
850        D::Int32 => "INTEGER",
851        D::Int64 => "BIGINT",
852        D::UInt8 => "UTINYINT",
853        D::UInt16 => "USMALLINT",
854        D::UInt32 => "UINTEGER",
855        D::UInt64 => "UBIGINT",
856        D::Float16 | D::Float32 => "FLOAT",
857        D::Float64 => "DOUBLE",
858        D::Utf8 | D::LargeUtf8 | D::Utf8View => "VARCHAR",
859        D::Binary | D::LargeBinary | D::BinaryView => "BLOB",
860        D::Date32 | D::Date64 => "DATE",
861        D::Timestamp(_, _) => "TIMESTAMP",
862        D::Time32(_) | D::Time64(_) => "TIME",
863        D::Decimal128(_, _) | D::Decimal256(_, _) => "DECIMAL",
864        other => return format!("{:?}", other),
865    }
866    .to_string()
867}
868
869/// Convert a DuckDB `ValueRef` into a `serde_json::Value` for output. Exotic
870/// types (temporal, decimal, nested) fall back to a debug string so rendering
871/// never panics.
872fn value_ref_to_json(v: ValueRef<'_>) -> serde_json::Value {
873    use serde_json::Value as J;
874    match v {
875        ValueRef::Null => J::Null,
876        ValueRef::Boolean(b) => J::Bool(b),
877        ValueRef::TinyInt(n) => J::from(n),
878        ValueRef::SmallInt(n) => J::from(n),
879        ValueRef::Int(n) => J::from(n),
880        ValueRef::BigInt(n) => J::from(n),
881        ValueRef::UTinyInt(n) => J::from(n),
882        ValueRef::USmallInt(n) => J::from(n),
883        ValueRef::UInt(n) => J::from(n),
884        ValueRef::UBigInt(n) => J::from(n),
885        ValueRef::HugeInt(n) => J::String(n.to_string()),
886        ValueRef::Float(f) => serde_json::Number::from_f64(f as f64)
887            .map(J::Number)
888            .unwrap_or(J::Null),
889        ValueRef::Double(f) => serde_json::Number::from_f64(f)
890            .map(J::Number)
891            .unwrap_or(J::Null),
892        ValueRef::Text(bytes) => J::String(String::from_utf8_lossy(bytes).into_owned()),
893        ValueRef::Blob(bytes) => J::String(format!("<blob: {} bytes>", bytes.len())),
894        ValueRef::Timestamp(unit, raw) => {
895            use duckdb::types::TimeUnit;
896            let nanos = match unit {
897                TimeUnit::Second => (raw as i128) * 1_000_000_000,
898                TimeUnit::Millisecond => (raw as i128) * 1_000_000,
899                TimeUnit::Microsecond => (raw as i128) * 1_000,
900                TimeUnit::Nanosecond => raw as i128,
901            };
902            let formatted = time::OffsetDateTime::from_unix_timestamp_nanos(nanos)
903                .ok()
904                .and_then(|dt| {
905                    dt.format(&time::format_description::well_known::Rfc3339)
906                        .ok()
907                });
908            J::String(formatted.unwrap_or_else(|| format!("Timestamp({unit:?}, {raw})")))
909        }
910        other => J::String(format!("{:?}", other)),
911    }
912}
913
914#[cfg(test)]
915mod tests {
916    use super::*;
917    use duckdb::Connection;
918
919    fn setup_test_db() -> Result<Connection> {
920        let conn = Connection::open_in_memory()?;
921
922        // Create schema
923        conn.execute_batch(
924            r#"
925            CREATE SCHEMA IF NOT EXISTS code;
926            
927            CREATE TABLE code.symbols (
928                id TEXT PRIMARY KEY,
929                name TEXT NOT NULL,
930                qualified_name TEXT,
931                kind TEXT NOT NULL,
932                file_path TEXT NOT NULL,
933                line_start INTEGER NOT NULL,
934                line_end INTEGER NOT NULL,
935                visibility TEXT
936            );
937            
938            CREATE TABLE code.edges (
939                source_id TEXT NOT NULL,
940                target_id TEXT,
941                target_name TEXT NOT NULL,
942                kind TEXT NOT NULL,
943                context TEXT,
944                FOREIGN KEY (source_id) REFERENCES code.symbols(id)
945            );
946            
947            -- Insert test data: main -> run -> helper
948            INSERT INTO code.symbols VALUES 
949                ('main@1', 'main', NULL, 'function', 'test.rs', 1, 10, 'public'),
950                ('run@11', 'run', NULL, 'function', 'test.rs', 11, 20, 'private'),
951                ('helper@21', 'helper', NULL, 'function', 'test.rs', 21, 30, 'private');
952            
953            INSERT INTO code.edges VALUES 
954                ('main@1', 'run@11', 'run', 'calls', NULL),
955                ('run@11', 'helper@21', 'helper', 'calls', NULL);
956            "#,
957        )?;
958
959        Ok(conn)
960    }
961
962    #[test]
963    fn test_call_graph_syntax() {
964        let conn = setup_test_db().expect("Failed to setup test db");
965        let analytics = Analytics { conn };
966
967        let result = analytics.call_graph("main", 5);
968        assert!(
969            result.is_ok(),
970            "call_graph query failed: {:?}",
971            result.err()
972        );
973
974        let nodes = result.unwrap();
975        assert_eq!(nodes.len(), 2, "Expected 2 nodes (run, helper)");
976        assert_eq!(nodes[0].name, "run");
977        assert_eq!(nodes[1].name, "helper");
978    }
979
980    #[test]
981    fn test_impact_analysis_syntax() {
982        let conn = setup_test_db().expect("Failed to setup test db");
983        let analytics = Analytics { conn };
984
985        let result = analytics.impact_analysis("helper", 5);
986        assert!(
987            result.is_ok(),
988            "impact_analysis query failed: {:?}",
989            result.err()
990        );
991
992        let nodes = result.unwrap();
993        assert_eq!(nodes.len(), 2, "Expected 2 nodes (run, main)");
994        // run calls helper directly (distance 1)
995        // main calls run which calls helper (distance 2)
996        assert!(nodes.iter().any(|n| n.name == "run" && n.distance == 1));
997        assert!(nodes.iter().any(|n| n.name == "main" && n.distance == 2));
998    }
999
1000    #[test]
1001    fn test_has_path_syntax() {
1002        let conn = setup_test_db().expect("Failed to setup test db");
1003        let analytics = Analytics { conn };
1004
1005        // main -> run -> helper (path exists)
1006        let result = analytics.has_path("main", "helper", 5);
1007        assert!(result.is_ok(), "has_path query failed: {:?}", result.err());
1008        assert!(result.unwrap(), "Expected path from main to helper");
1009
1010        // helper -> main (no path in reverse direction)
1011        let result = analytics.has_path("helper", "main", 5);
1012        assert!(result.is_ok(), "has_path query failed: {:?}", result.err());
1013        assert!(!result.unwrap(), "Expected no path from helper to main");
1014    }
1015
1016    #[test]
1017    fn test_sql_injection_in_path_escaping() {
1018        // Test that paths with special SQL characters are properly escaped
1019        // This is a unit test for the escaping logic - the actual open() function
1020        // requires a real SQLite file, so we test the escaping directly
1021
1022        // Test paths that would cause SQL injection if not escaped
1023        let test_paths = [
1024            "normal/path.db",
1025            "path with spaces/file.db",
1026            "path'with'quotes/file.db",
1027            "path''with''double/file.db",
1028            "path;DROP TABLE code;/file.db",
1029            "path' OR '1'='1/file.db",
1030        ];
1031
1032        for path in &test_paths {
1033            // Simulate the escaping done in Analytics::open()
1034            let escaped = path.replace('\'', "''");
1035            let sql = format!("ATTACH '{}' AS code", escaped);
1036
1037            // The escaped SQL should have balanced quotes
1038            let quote_count = sql.chars().filter(|c| *c == '\'').count();
1039            assert_eq!(
1040                quote_count % 2,
1041                0,
1042                "SQL for path '{}' has unbalanced quotes: {}",
1043                path,
1044                sql
1045            );
1046
1047            // Single quotes in the path should be doubled
1048            if path.contains('\'') {
1049                assert!(
1050                    escaped.contains("''"),
1051                    "Path with quote should have doubled quotes: {}",
1052                    escaped
1053                );
1054            }
1055        }
1056    }
1057}