// Rust guideline compliant 2025-10-17
use std::collections::{HashMap, HashSet};

use libsql::params;

use super::connection::Database;
use crate::errors::{Result, TokenSaveError};
use crate::types::*;

// ---------------------------------------------------------------------------
// Helper: build SQL placeholder string `?, ?, ?, …` in one allocation.
// ---------------------------------------------------------------------------

/// Returns a SQL placeholder string of `n` anonymous `?` markers separated by
/// `, `. Used to construct `IN ($qmarks)` clauses without allocating one
/// `String` per id (`format!("?{i}")` previously did that).
fn build_qmark_placeholders(n: usize) -> String {
    debug_assert!(n > 0, "build_qmark_placeholders called with n == 0");
    // Each "?, " occupies 3 bytes; the last one drops the trailing ", ".
    let mut s = String::with_capacity(n * 3);
    for i in 0..n {
        if i > 0 {
            s.push_str(", ");
        }
        s.push('?');
    }
    s
}

// ---------------------------------------------------------------------------
// Helper: map a libsql row to domain types (by column index)
// ---------------------------------------------------------------------------

/// Maps a row from the `nodes` table to a `Node`.
///
/// Expected column order: id(0), kind(1), name(2), `qualified_name(3)`,
/// `file_path(4)`, `start_line(5)`, `end_line(6)`, `start_column(7)`, `end_column(8)`,
/// docstring(9), signature(10), visibility(11), `is_async(12)`,
/// branches(13), loops(14), returns(15), `max_nesting(16)`,
/// `unsafe_blocks(17)`, `unchecked_calls(18)`, assertions(19), `updated_at(20)`,
/// `attrs_start_line(21)`.
fn row_to_node(row: &libsql::Row) -> std::result::Result<Node, libsql::Error> {
    let kind_str = get_string_lossy(row, 1)?;
    let vis_str = get_string_lossy(row, 11)?;
    let is_async_int = row.get::<i64>(12)?;
    let start_line = row.get::<u32>(5)?;
    // Pre-v7 rows may have attrs_start_line == 0 (default); fall back to start_line.
    let attrs_raw = row.get::<u32>(21).unwrap_or(0);
    let attrs_start_line = if attrs_raw == 0 {
        start_line
    } else {
        attrs_raw
    };
    // `parent_id` is column 22 in v9+ SELECT lists. Older SELECTs in this
    // file don't request it; the .ok().flatten() chain swallows the missing-
    // column error and yields None.
    let parent_id = get_opt_string_lossy(row, 22).ok().flatten();

    Ok(Node {
        id: get_string_lossy(row, 0)?,
        kind: NodeKind::from_str(&kind_str).unwrap_or(NodeKind::Function),
        name: get_string_lossy(row, 2)?,
        qualified_name: get_string_lossy(row, 3)?,
        file_path: get_string_lossy(row, 4)?,
        start_line,
        attrs_start_line,
        end_line: row.get::<u32>(6)?,
        start_column: row.get::<u32>(7)?,
        end_column: row.get::<u32>(8)?,
        signature: get_opt_string_lossy(row, 10)?,
        docstring: get_opt_string_lossy(row, 9)?,
        visibility: Visibility::from_str(&vis_str).unwrap_or_default(),
        is_async: is_async_int != 0,
        branches: row.get::<u32>(13)?,
        loops: row.get::<u32>(14)?,
        returns: row.get::<u32>(15)?,
        max_nesting: row.get::<u32>(16)?,
        unsafe_blocks: row.get::<u32>(17)?,
        unchecked_calls: row.get::<u32>(18)?,
        assertions: row.get::<u32>(19)?,
        updated_at: row.get::<u64>(20)?,
        parent_id,
    })
}

/// Reads a text column as String, replacing invalid UTF-8 bytes with U+FFFD.
/// This prevents crashes when source files with non-UTF-8 encoding (e.g. Latin-1)
/// have their signatures or docstrings stored in the database.
///
/// libsql's `get::<String>()` panics on Blob values via `unreachable!()`, so we
/// must read as `Value` first and convert.
fn get_string_lossy(row: &libsql::Row, idx: i32) -> std::result::Result<String, libsql::Error> {
    let val = row.get::<libsql::Value>(idx)?;
    match val {
        libsql::Value::Text(s) => Ok(s),
        libsql::Value::Blob(bytes) => Ok(String::from_utf8_lossy(&bytes).into_owned()),
        libsql::Value::Null => Ok(String::new()),
        libsql::Value::Integer(i) => Ok(i.to_string()),
        libsql::Value::Real(f) => Ok(f.to_string()),
    }
}

/// Like `get_string_lossy` but for nullable columns.
fn get_opt_string_lossy(
    row: &libsql::Row,
    idx: i32,
) -> std::result::Result<Option<String>, libsql::Error> {
    let val = row.get::<libsql::Value>(idx)?;
    match val {
        libsql::Value::Null => Ok(None),
        libsql::Value::Text(s) => Ok(Some(s)),
        libsql::Value::Blob(bytes) => Ok(Some(String::from_utf8_lossy(&bytes).into_owned())),
        libsql::Value::Integer(i) => Ok(Some(i.to_string())),
        libsql::Value::Real(f) => Ok(Some(f.to_string())),
    }
}

/// Maps a row from the `edges` table to an `Edge`.
///
/// Expected column order: source(0), target(1), kind(2), line(3).
fn row_to_edge(row: &libsql::Row) -> std::result::Result<Edge, libsql::Error> {
    let kind_str = row.get::<String>(2)?;
    let line = row.get::<Option<u32>>(3)?;

    Ok(Edge {
        source: row.get::<String>(0)?,
        target: row.get::<String>(1)?,
        kind: EdgeKind::from_str(&kind_str).unwrap_or(EdgeKind::Uses),
        line,
    })
}

/// Maps a row from the `files` table to a `FileRecord`.
///
/// Expected column order: path(0), `content_hash(1)`, size(2), `modified_at(3)`,
/// `indexed_at(4)`, `node_count(5)`.
fn row_to_file(row: &libsql::Row) -> std::result::Result<FileRecord, libsql::Error> {
    Ok(FileRecord {
        path: row.get::<String>(0)?,
        content_hash: row.get::<String>(1)?,
        size: row.get::<u64>(2)?,
        modified_at: row.get::<i64>(3)?,
        indexed_at: row.get::<i64>(4)?,
        node_count: row.get::<u32>(5)?,
    })
}

/// Maps a row from the `unresolved_refs` table to an `UnresolvedRef`.
///
/// Expected column order: `from_node_id(0)`, `reference_name(1)`,
/// `reference_kind(2)`, line(3), col(4), `file_path(5)`.
fn row_to_unresolved_ref(row: &libsql::Row) -> std::result::Result<UnresolvedRef, libsql::Error> {
    let kind_str = row.get::<String>(2)?;

    Ok(UnresolvedRef {
        from_node_id: row.get::<String>(0)?,
        reference_name: row.get::<String>(1)?,
        reference_kind: EdgeKind::from_str(&kind_str).unwrap_or(EdgeKind::Uses),
        line: row.get::<u32>(3)?,
        column: row.get::<u32>(4)?,
        file_path: row.get::<String>(5)?,
    })
}

// ---------------------------------------------------------------------------
// Node operations
// ---------------------------------------------------------------------------

impl Database {
    /// Inserts or replaces a single node.
    pub async fn insert_node(&self, node: &Node) -> Result<()> {
        self.conn()
            .execute(
                "INSERT OR REPLACE INTO nodes
                (id, kind, name, qualified_name, file_path,
                 start_line, end_line, start_column, end_column,
                 docstring, signature, visibility, is_async,
                 branches, loops, returns, max_nesting,
                 unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23)",
                params![
                    node.id.as_str(),
                    node.kind.as_str(),
                    node.name.as_str(),
                    node.qualified_name.as_str(),
                    node.file_path.as_str(),
                    i64::from(node.start_line),
                    i64::from(node.end_line),
                    i64::from(node.start_column),
                    i64::from(node.end_column),
                    opt_str(node.docstring.as_deref()),
                    opt_str(node.signature.as_deref()),
                    node.visibility.as_str(),
                    i64::from(node.is_async),
                    i64::from(node.branches),
                    i64::from(node.loops),
                    i64::from(node.returns),
                    i64::from(node.max_nesting),
                    i64::from(node.unsafe_blocks),
                    i64::from(node.unchecked_calls),
                    i64::from(node.assertions),
                    node.updated_at as i64,
                    i64::from(node.attrs_start_line),
                    opt_str(node.parent_id.as_deref()),
                ],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to insert node: {e}"),
                operation: "insert_node".to_string(),
            })?;
        Ok(())
    }

    /// Inserts all nodes, edges, and file records in a single `execute_batch` call.
    /// This minimizes transaction overhead by combining everything into one SQL string.
    ///
    /// `Contains` edges are denormalized at insert time: their `(source, target)`
    /// pair is folded into the target node's `parent_id` column, and the edge
    /// itself is not persisted. Extractors keep emitting `Contains` edges as
    /// before; the conversion happens here, in one place.
    pub async fn insert_all(
        &self,
        nodes: &[Node],
        edges: &[Edge],
        files: &[FileRecord],
    ) -> Result<()> {
        // Pull every Contains edge out: build target_id -> parent_id map, then
        // filter the surviving edges list. When a node has multiple incoming
        // Contains rows (extractor anomaly), the first one wins — matching
        // the migration's `LIMIT 1` backfill behavior.
        let mut parent_map: std::collections::HashMap<&str, &str> =
            std::collections::HashMap::new();
        let mut surviving_edges: Vec<&Edge> = Vec::with_capacity(edges.len());
        for edge in edges {
            if edge.kind == crate::types::EdgeKind::Contains {
                parent_map
                    .entry(edge.target.as_str())
                    .or_insert(edge.source.as_str());
            } else {
                surviving_edges.push(edge);
            }
        }
        // Apply the hoisted parents to the node slice without cloning every
        // node: we materialize only when parent_map has something to say.
        let nodes_owned: Vec<Node>;
        let nodes_ref: &[Node] = if parent_map.is_empty() {
            nodes
        } else {
            nodes_owned = nodes
                .iter()
                .map(|n| {
                    if let Some(parent) = parent_map.get(n.id.as_str()) {
                        let mut copy = n.clone();
                        copy.parent_id = Some((*parent).to_string());
                        copy
                    } else {
                        n.clone()
                    }
                })
                .collect();
            &nodes_owned
        };

        let mut sql = String::with_capacity(
            nodes_ref.len() * 400 + surviving_edges.len() * 120 + files.len() * 120,
        );
        sql.push_str("BEGIN;\n");

        // Nodes
        for chunk in nodes_ref.chunks(200) {
            sql.push_str(
                "INSERT OR REPLACE INTO nodes \
                 (id,kind,name,qualified_name,file_path,\
                 start_line,end_line,start_column,end_column,\
                 docstring,signature,visibility,is_async,\
                 branches,loops,returns,max_nesting,\
                 unsafe_blocks,unchecked_calls,assertions,updated_at,attrs_start_line,parent_id) VALUES ",
            );
            for (i, node) in chunk.iter().enumerate() {
                if i > 0 {
                    sql.push(',');
                }
                sql.push('(');
                push_quoted(&mut sql, &node.id);
                sql.push(',');
                push_quoted(&mut sql, node.kind.as_str());
                sql.push(',');
                push_quoted(&mut sql, &node.name);
                sql.push(',');
                push_quoted(&mut sql, &node.qualified_name);
                sql.push(',');
                push_quoted(&mut sql, &node.file_path);
                sql.push(',');
                push_int(&mut sql, i64::from(node.start_line));
                sql.push(',');
                push_int(&mut sql, i64::from(node.end_line));
                sql.push(',');
                push_int(&mut sql, i64::from(node.start_column));
                sql.push(',');
                push_int(&mut sql, i64::from(node.end_column));
                sql.push(',');
                push_opt_quoted(&mut sql, node.docstring.as_deref());
                sql.push(',');
                push_opt_quoted(&mut sql, node.signature.as_deref());
                sql.push(',');
                push_quoted(&mut sql, node.visibility.as_str());
                sql.push(',');
                push_int(&mut sql, i64::from(node.is_async));
                sql.push(',');
                push_int(&mut sql, i64::from(node.branches));
                sql.push(',');
                push_int(&mut sql, i64::from(node.loops));
                sql.push(',');
                push_int(&mut sql, i64::from(node.returns));
                sql.push(',');
                push_int(&mut sql, i64::from(node.max_nesting));
                sql.push(',');
                push_int(&mut sql, i64::from(node.unsafe_blocks));
                sql.push(',');
                push_int(&mut sql, i64::from(node.unchecked_calls));
                sql.push(',');
                push_int(&mut sql, i64::from(node.assertions));
                sql.push(',');
                push_int(&mut sql, node.updated_at as i64);
                sql.push(',');
                push_int(&mut sql, i64::from(node.attrs_start_line));
                sql.push(',');
                push_opt_quoted(&mut sql, node.parent_id.as_deref());
                sql.push(')');
            }
            sql.push_str(";\n");
        }

        // Edges (Contains has already been hoisted out into parent_id)
        for chunk in surviving_edges.chunks(500) {
            sql.push_str("INSERT OR IGNORE INTO edges (source,target,kind,line) VALUES ");
            for (i, edge) in chunk.iter().enumerate() {
                if i > 0 {
                    sql.push(',');
                }
                sql.push('(');
                push_quoted(&mut sql, &edge.source);
                sql.push(',');
                push_quoted(&mut sql, &edge.target);
                sql.push(',');
                push_quoted(&mut sql, edge.kind.as_str());
                sql.push(',');
                match edge.line {
                    Some(l) => push_int(&mut sql, i64::from(l)),
                    None => sql.push_str("NULL"),
                }
                sql.push(')');
            }
            sql.push_str(";\n");
        }

        // Files
        for chunk in files.chunks(500) {
            sql.push_str(
                "INSERT OR REPLACE INTO files \
                 (path,content_hash,size,modified_at,indexed_at,node_count) VALUES ",
            );
            for (i, file) in chunk.iter().enumerate() {
                if i > 0 {
                    sql.push(',');
                }
                sql.push('(');
                push_quoted(&mut sql, &file.path);
                sql.push(',');
                push_quoted(&mut sql, &file.content_hash);
                sql.push(',');
                push_int(&mut sql, file.size as i64);
                sql.push(',');
                push_int(&mut sql, file.modified_at);
                sql.push(',');
                push_int(&mut sql, file.indexed_at);
                sql.push(',');
                push_int(&mut sql, i64::from(file.node_count));
                sql.push(')');
            }
            sql.push_str(";\n");
        }

        sql.push_str("COMMIT;\n");

        self.conn()
            .execute_batch(&sql)
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to bulk insert: {e}"),
                operation: "insert_all".to_string(),
            })?;
        Ok(())
    }

    /// Inserts nodes using a prepared statement: parse SQL once, then
    /// bind+execute+reset for each row — zero SQL parsing after the first call.
    pub async fn insert_nodes(&self, nodes: &[Node]) -> Result<()> {
        if nodes.is_empty() {
            return Ok(());
        }

        self.conn()
            .execute("BEGIN", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to begin: {e}"),
                operation: "insert_nodes".to_string(),
            })?;

        let stmt = self.conn()
            .prepare(
                "INSERT OR REPLACE INTO nodes \
                 (id,kind,name,qualified_name,file_path,\
                 start_line,end_line,start_column,end_column,\
                 docstring,signature,visibility,is_async,\
                 branches,loops,returns,max_nesting,\
                 unsafe_blocks,unchecked_calls,assertions,updated_at,attrs_start_line,parent_id) \
                 VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,?23)"
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to prepare: {e}"),
                operation: "insert_nodes".to_string(),
            })?;

        for node in nodes {
            stmt.execute(params![
                node.id.as_str(),
                node.kind.as_str(),
                node.name.as_str(),
                node.qualified_name.as_str(),
                node.file_path.as_str(),
                i64::from(node.start_line),
                i64::from(node.end_line),
                i64::from(node.start_column),
                i64::from(node.end_column),
                opt_str(node.docstring.as_deref()),
                opt_str(node.signature.as_deref()),
                node.visibility.as_str(),
                i64::from(node.is_async),
                i64::from(node.branches),
                i64::from(node.loops),
                i64::from(node.returns),
                i64::from(node.max_nesting),
                i64::from(node.unsafe_blocks),
                i64::from(node.unchecked_calls),
                i64::from(node.assertions),
                node.updated_at as i64,
                i64::from(node.attrs_start_line),
                opt_str(node.parent_id.as_deref()),
            ])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to insert node: {e}"),
                operation: "insert_nodes".to_string(),
            })?;
            stmt.reset();
        }

        self.conn()
            .execute("COMMIT", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to commit: {e}"),
                operation: "insert_nodes".to_string(),
            })?;
        Ok(())
    }

    /// Retrieves a node by its unique ID, returning `None` if not found.
    pub async fn get_node_by_id(&self, id: &str) -> Result<Option<Node>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT id, kind, name, qualified_name, file_path,
                        start_line, end_line, start_column, end_column,
                        docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes WHERE id = ?1",
                params![id],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query node by id: {e}"),
                operation: "get_node_by_id".to_string(),
            })?;

        match rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read node row: {e}"),
            operation: "get_node_by_id".to_string(),
        })? {
            Some(row) => {
                let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                    message: format!("failed to map node row: {e}"),
                    operation: "get_node_by_id".to_string(),
                })?;
                Ok(Some(node))
            }
            None => Ok(None),
        }
    }

    /// Returns nodes by their IDs in a single batch query.
    /// IDs not found are silently omitted. Results are returned in arbitrary order.
    pub async fn get_nodes_by_ids(&self, ids: &[String]) -> Result<Vec<Node>> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }
        // Build `?, ?, ?, …` in one allocation instead of `Vec<String>` of
        // `?1`/`?2`/`?N`. libsql binds anonymous `?` parameters in order, so
        // dropping the numbered form changes nothing for the driver. Large
        // BFS frontiers (`traverse_bfs` calls this once per level) hit this
        // path often enough that the per-id `format!` allocations showed up
        // on profiles.
        let placeholders = build_qmark_placeholders(ids.len());
        let sql = format!(
            "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
             FROM nodes WHERE id IN ({placeholders})",
        );
        let param_values: Vec<libsql::Value> = ids
            .iter()
            .map(|id| libsql::Value::Text(id.clone()))
            .collect();
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to batch query nodes: {e}"),
                operation: "get_nodes_by_ids".to_string(),
            })?;
        collect_rows(&mut rows, row_to_node, "get_nodes_by_ids").await
    }

    /// Returns all nodes for a given file, ordered by start line.
    pub async fn get_nodes_by_file(&self, file_path: &str) -> Result<Vec<Node>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes WHERE file_path = ?1 ORDER BY start_line",
                params![file_path],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query nodes by file: {e}"),
                operation: "get_nodes_by_file".to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, "get_nodes_by_file").await
    }

    /// Returns every node whose `parent_id` matches `parent_id`. Replaces
    /// the v8 pattern of querying outgoing `Contains` edges; after v9 the
    /// edges table no longer carries that information.
    pub async fn get_children_of(&self, parent_id: &str) -> Result<Vec<Node>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes WHERE parent_id = ?1 ORDER BY start_line",
                params![parent_id],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query children: {e}"),
                operation: "get_children_of".to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, "get_children_of").await
    }

    /// Returns all nodes of a given kind.
    pub async fn get_nodes_by_kind(&self, kind: NodeKind) -> Result<Vec<Node>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes WHERE kind = ?1",
                params![kind.as_str()],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query nodes by kind: {e}"),
                operation: "get_nodes_by_kind".to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, "get_nodes_by_kind").await
    }

    /// Returns every node in the database.
    pub async fn get_all_nodes(&self) -> Result<Vec<Node>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes",
                (),
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query all nodes: {e}"),
                operation: "get_all_nodes".to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, "get_all_nodes").await
    }

    /// Deletes all nodes (and cascading edges, unresolved refs, vectors) for a file.
    pub async fn delete_nodes_by_file(&self, file_path: &str) -> Result<()> {
        debug_assert!(
            !file_path.is_empty(),
            "delete_nodes_by_file called with empty file_path"
        );
        debug_assert!(
            !file_path.starts_with('/'),
            "delete_nodes_by_file expects relative path, got absolute"
        );
        // Gather node IDs for the file first.
        let node_ids: Vec<String> = {
            let mut rows = self
                .conn()
                .query(
                    "SELECT id FROM nodes WHERE file_path = ?1",
                    params![file_path],
                )
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to query node ids: {e}"),
                    operation: "delete_nodes_by_file".to_string(),
                })?;

            let mut ids = Vec::new();
            while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
                message: format!("failed to read node id: {e}"),
                operation: "delete_nodes_by_file".to_string(),
            })? {
                ids.push(row.get::<String>(0).map_err(|e| TokenSaveError::Database {
                    message: format!("failed to read node id value: {e}"),
                    operation: "delete_nodes_by_file".to_string(),
                })?);
            }
            ids
        };

        if node_ids.is_empty() {
            return Ok(());
        }

        let tx = self
            .conn()
            .transaction()
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to begin transaction: {e}"),
                operation: "delete_nodes_by_file".to_string(),
            })?;

        for id in &node_ids {
            tx.execute(
                "DELETE FROM edges WHERE source = ?1 OR target = ?1",
                params![id.as_str()],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to delete edges: {e}"),
                operation: "delete_nodes_by_file".to_string(),
            })?;

            tx.execute(
                "DELETE FROM unresolved_refs WHERE from_node_id = ?1",
                params![id.as_str()],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to delete unresolved refs: {e}"),
                operation: "delete_nodes_by_file".to_string(),
            })?;

            tx.execute(
                "DELETE FROM vectors WHERE node_id = ?1",
                params![id.as_str()],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to delete vectors: {e}"),
                operation: "delete_nodes_by_file".to_string(),
            })?;
        }

        tx.execute("DELETE FROM nodes WHERE file_path = ?1", params![file_path])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to delete nodes: {e}"),
                operation: "delete_nodes_by_file".to_string(),
            })?;

        tx.commit().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to commit transaction: {e}"),
            operation: "delete_nodes_by_file".to_string(),
        })
    }
}

// ---------------------------------------------------------------------------
// Edge operations
// ---------------------------------------------------------------------------

impl Database {
    /// Inserts a single edge, skipping silently if either endpoint is missing.
    pub async fn insert_edge(&self, edge: &Edge) -> Result<()> {
        // Contains is denormalized to nodes.parent_id since v9. Fold the
        // edge into an UPDATE rather than writing a row to the edges table.
        if edge.kind == EdgeKind::Contains {
            self.conn()
                .execute(
                    "UPDATE nodes SET parent_id = ?1 WHERE id = ?2",
                    params![edge.source.as_str(), edge.target.as_str()],
                )
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to set parent_id: {e}"),
                    operation: "insert_edge".to_string(),
                })?;
            return Ok(());
        }
        self.conn()
            .execute(
                "INSERT OR IGNORE INTO edges (source, target, kind, line) \
                 SELECT ?1, ?2, ?3, ?4 \
                 WHERE EXISTS (SELECT 1 FROM nodes WHERE id = ?1) \
                   AND EXISTS (SELECT 1 FROM nodes WHERE id = ?2)",
                params![
                    edge.source.as_str(),
                    edge.target.as_str(),
                    edge.kind.as_str(),
                    edge.line.map(i64::from)
                ],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to insert edge: {e}"),
                operation: "insert_edge".to_string(),
            })?;
        Ok(())
    }

    /// Inserts a batch of edges inside a single transaction.
    ///
    /// Edges whose source or target node does not yet exist are silently
    /// skipped (#58). They will be picked up on a future sync once the
    /// referenced file is indexed. `Contains` edges are denormalized into
    /// `nodes.parent_id` via UPDATE; they do not produce edge rows.
    pub async fn insert_edges(&self, edges: &[Edge]) -> Result<()> {
        if edges.is_empty() {
            return Ok(());
        }

        self.conn()
            .execute("BEGIN", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to begin: {e}"),
                operation: "insert_edges".to_string(),
            })?;

        // Conditional INSERT: only insert when both endpoints exist in
        // `nodes`. This avoids FK violations during incremental sync
        // when an edge references a node from a not-yet-indexed file.
        let stmt = self
            .conn()
            .prepare(
                "INSERT OR IGNORE INTO edges (source, target, kind, line) \
                 SELECT ?1, ?2, ?3, ?4 \
                 WHERE EXISTS (SELECT 1 FROM nodes WHERE id = ?1) \
                   AND EXISTS (SELECT 1 FROM nodes WHERE id = ?2)",
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to prepare: {e}"),
                operation: "insert_edges".to_string(),
            })?;

        let parent_stmt = self
            .conn()
            .prepare("UPDATE nodes SET parent_id = ?1 WHERE id = ?2")
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to prepare parent update: {e}"),
                operation: "insert_edges".to_string(),
            })?;

        for edge in edges {
            if edge.kind == EdgeKind::Contains {
                parent_stmt
                    .execute(params![edge.source.as_str(), edge.target.as_str()])
                    .await
                    .map_err(|e| TokenSaveError::Database {
                        message: format!("failed to set parent_id: {e}"),
                        operation: "insert_edges".to_string(),
                    })?;
                parent_stmt.reset();
                continue;
            }
            stmt.execute(params![
                edge.source.as_str(),
                edge.target.as_str(),
                edge.kind.as_str(),
                edge.line.map(i64::from),
            ])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to insert edge: {e}"),
                operation: "insert_edges".to_string(),
            })?;
            stmt.reset();
        }

        self.conn()
            .execute("COMMIT", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to commit: {e}"),
                operation: "insert_edges".to_string(),
            })?;
        Ok(())
    }

    /// Returns outgoing edges from a source node, optionally filtered by edge kinds.
    ///
    /// If `kinds` is empty, all outgoing edges are returned.
    pub async fn get_outgoing_edges(
        &self,
        source_id: &str,
        kinds: &[EdgeKind],
    ) -> Result<Vec<Edge>> {
        if kinds.is_empty() {
            let mut rows = self
                .conn()
                .query(
                    "SELECT source, target, kind, line FROM edges WHERE source = ?1",
                    params![source_id],
                )
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to query outgoing edges: {e}"),
                    operation: "get_outgoing_edges".to_string(),
                })?;

            collect_rows(&mut rows, row_to_edge, "get_outgoing_edges").await
        } else {
            let placeholders: Vec<String> = kinds
                .iter()
                .enumerate()
                .map(|(i, _)| format!("?{}", i + 2))
                .collect();
            let sql = format!(
                "SELECT source, target, kind, line FROM edges WHERE source = ?1 AND kind IN ({})",
                placeholders.join(", ")
            );

            let mut param_values: Vec<libsql::Value> = Vec::new();
            param_values.push(libsql::Value::Text(source_id.to_string()));
            for k in kinds {
                param_values.push(libsql::Value::Text(k.as_str().to_string()));
            }

            let mut rows = self
                .conn()
                .query(&sql, libsql::params_from_iter(param_values))
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to query outgoing edges: {e}"),
                    operation: "get_outgoing_edges".to_string(),
                })?;

            collect_rows(&mut rows, row_to_edge, "get_outgoing_edges").await
        }
    }

    /// Returns incoming edges to a target node, optionally filtered by edge kinds.
    ///
    /// If `kinds` is empty, all incoming edges are returned.
    pub async fn get_incoming_edges(
        &self,
        target_id: &str,
        kinds: &[EdgeKind],
    ) -> Result<Vec<Edge>> {
        if kinds.is_empty() {
            let mut rows = self
                .conn()
                .query(
                    "SELECT source, target, kind, line FROM edges WHERE target = ?1",
                    params![target_id],
                )
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to query incoming edges: {e}"),
                    operation: "get_incoming_edges".to_string(),
                })?;

            collect_rows(&mut rows, row_to_edge, "get_incoming_edges").await
        } else {
            let placeholders: Vec<String> = kinds
                .iter()
                .enumerate()
                .map(|(i, _)| format!("?{}", i + 2))
                .collect();
            let sql = format!(
                "SELECT source, target, kind, line FROM edges WHERE target = ?1 AND kind IN ({})",
                placeholders.join(", ")
            );

            let mut param_values: Vec<libsql::Value> = Vec::new();
            param_values.push(libsql::Value::Text(target_id.to_string()));
            for k in kinds {
                param_values.push(libsql::Value::Text(k.as_str().to_string()));
            }

            let mut rows = self
                .conn()
                .query(&sql, libsql::params_from_iter(param_values))
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to query incoming edges: {e}"),
                    operation: "get_incoming_edges".to_string(),
                })?;

            collect_rows(&mut rows, row_to_edge, "get_incoming_edges").await
        }
    }

    /// Returns all incoming edges for many target nodes in a single query.
    ///
    /// Used by the bulk `callers_for` MCP tool: clients pass a list of item
    /// IDs and get back, for each id, the set of nodes pointing at it via
    /// the requested edge kinds. One round-trip replaces N round-trips
    /// through `get_incoming_edges`.
    ///
    /// When `kinds` is empty, all edge kinds are returned.
    pub async fn get_incoming_edges_bulk(
        &self,
        target_ids: &[String],
        kinds: &[EdgeKind],
    ) -> Result<Vec<Edge>> {
        if target_ids.is_empty() {
            return Ok(Vec::new());
        }

        let target_placeholders: Vec<String> =
            (1..=target_ids.len()).map(|i| format!("?{i}")).collect();
        let mut param_values: Vec<libsql::Value> = target_ids
            .iter()
            .map(|id| libsql::Value::Text(id.clone()))
            .collect();

        let sql = if kinds.is_empty() {
            format!(
                "SELECT source, target, kind, line FROM edges WHERE target IN ({})",
                target_placeholders.join(", ")
            )
        } else {
            let kind_placeholders: Vec<String> = (1..=kinds.len())
                .map(|i| format!("?{}", target_ids.len() + i))
                .collect();
            for k in kinds {
                param_values.push(libsql::Value::Text(k.as_str().to_string()));
            }
            format!(
                "SELECT source, target, kind, line FROM edges \
                 WHERE target IN ({}) AND kind IN ({})",
                target_placeholders.join(", "),
                kind_placeholders.join(", ")
            )
        };

        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query bulk incoming edges: {e}"),
                operation: "get_incoming_edges_bulk".to_string(),
            })?;

        collect_rows(&mut rows, row_to_edge, "get_incoming_edges_bulk").await
    }

    /// Returns the subset of `candidate_ids` that are annotated with `#[test]`
    /// (i.e. targeted by an `Annotates` edge from an `annotation_usage` node
    /// named `"test"`).
    pub async fn get_test_annotated_node_ids(
        &self,
        candidate_ids: &[String],
    ) -> Result<HashSet<String>> {
        if candidate_ids.is_empty() {
            return Ok(HashSet::new());
        }
        let placeholders: Vec<String> =
            (1..=candidate_ids.len()).map(|i| format!("?{i}")).collect();
        let sql = format!(
            "SELECT DISTINCT e.target \
             FROM edges e \
             JOIN nodes n ON e.source = n.id \
             WHERE n.kind = 'annotation_usage' \
               AND n.name = 'test' \
               AND e.kind = 'annotates' \
               AND e.target IN ({})",
            placeholders.join(", ")
        );
        let param_values: Vec<libsql::Value> = candidate_ids
            .iter()
            .map(|id| libsql::Value::Text(id.clone()))
            .collect();
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query test-annotated nodes: {e}"),
                operation: "get_test_annotated_node_ids".to_string(),
            })?;
        let mut result = HashSet::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read test-annotated row: {e}"),
            operation: "get_test_annotated_node_ids".to_string(),
        })? {
            if let Ok(id) = row.get::<String>(0) {
                result.insert(id);
            }
        }
        Ok(result)
    }

    /// Returns all file paths that contain at least one node annotated with
    /// `#[test]` (useful for detecting inline test modules in source files).
    pub async fn get_files_with_test_annotations(&self) -> Result<HashSet<String>> {
        let sql = "SELECT DISTINCT t.file_path \
                   FROM edges e \
                   JOIN nodes n ON e.source = n.id \
                   JOIN nodes t ON e.target = t.id \
                   WHERE n.kind = 'annotation_usage' \
                     AND n.name = 'test' \
                     AND e.kind = 'annotates' \
                     AND t.kind IN ('function', 'method')";
        let mut rows = self
            .conn()
            .query(sql, ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query test-annotation files: {e}"),
                operation: "get_files_with_test_annotations".to_string(),
            })?;
        let mut result = HashSet::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read test-annotation file row: {e}"),
            operation: "get_files_with_test_annotations".to_string(),
        })? {
            if let Ok(path) = row.get::<String>(0) {
                result.insert(path);
            }
        }
        Ok(result)
    }

    /// Returns all node IDs whose docstring contains `skip-test-coverage`.
    pub async fn get_skip_test_coverage_node_ids(&self) -> Result<HashSet<String>> {
        let sql = "SELECT id FROM nodes WHERE docstring LIKE '%skip-test-coverage%'";
        let mut rows = self
            .conn()
            .query(sql, ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query skip-test-coverage nodes: {e}"),
                operation: "get_skip_test_coverage_node_ids".to_string(),
            })?;
        let mut result = HashSet::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read skip-test-coverage row: {e}"),
            operation: "get_skip_test_coverage_node_ids".to_string(),
        })? {
            if let Ok(id) = row.get::<String>(0) {
                result.insert(id);
            }
        }
        Ok(result)
    }

    /// Returns all nodes whose `qualified_name` matches the given string.
    ///
    /// Multiple rows can share a qualified name (overloads, generic
    /// specialisations, separate `impl Trait for T` blocks). Uses the
    /// `idx_nodes_qualified_name` index for cross-run lookups by name,
    /// independent of content-hash IDs that change on edits.
    pub async fn get_nodes_by_qualified_name(&self, qname: &str) -> Result<Vec<Node>> {
        // Exact match first — preserves the precise-lookup contract.
        let exact_sql = "SELECT id, kind, name, qualified_name, file_path,
                          start_line, end_line, start_column, end_column,
                          docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                   FROM nodes
                   WHERE qualified_name = ?1";
        let mut rows = self
            .conn()
            .query(exact_sql, params![qname])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query by qualified_name: {e}"),
                operation: "get_nodes_by_qualified_name".to_string(),
            })?;

        let exact: Vec<Node> =
            collect_rows(&mut rows, row_to_node, "get_nodes_by_qualified_name").await?;
        if !exact.is_empty() {
            return Ok(exact);
        }

        // Fallback strategy depends on whether the user passed a qualified
        // form or just a bare identifier:
        //
        // - `Type::method` (contains `::`) → suffix match. Recovers from
        //   extractor quirks (duplicated path segments, file-path prefixes
        //   the caller doesn't know about) and lets callers pass partial
        //   module paths. The leading `%` defeats `idx_nodes_qualified_name`,
        //   so this is a full table scan bounded by `LIMIT 50` — cheap at
        //   typical graph sizes.
        //
        // - `foo` (no `::`) → exact `name = ?` match. Uses `idx_nodes_name`,
        //   so it stays fast. Multiple nodes may share a name (overloads,
        //   `new()` constructors), `LIMIT 50` is a safety net.
        let (sql, pattern) = if qname.contains("::") {
            (
                "SELECT id, kind, name, qualified_name, file_path,
                        start_line, end_line, start_column, end_column,
                        docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes
                 WHERE qualified_name LIKE ?1
                 LIMIT 50",
                format!("%::{qname}"),
            )
        } else {
            (
                "SELECT id, kind, name, qualified_name, file_path,
                        start_line, end_line, start_column, end_column,
                        docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes
                 WHERE name = ?1
                 LIMIT 50",
                qname.to_string(),
            )
        };
        let mut fallback_rows = self
            .conn()
            .query(sql, params![pattern.as_str()])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query by qualified_name fallback: {e}"),
                operation: "get_nodes_by_qualified_name".to_string(),
            })?;
        collect_rows(
            &mut fallback_rows,
            row_to_node,
            "get_nodes_by_qualified_name",
        )
        .await
    }

    /// Returns nodes ranked by edge count for a given edge kind and direction,
    /// optionally filtered by node kind.
    ///
    /// When `incoming` is true, ranks target nodes by incoming edge count
    /// (e.g. "most implemented interface"). When false, ranks source nodes
    /// by outgoing edge count (e.g. "class that implements the most interfaces").
    ///
    /// The query is performed entirely in SQL for efficiency — no need to load
    /// all edges into memory. Results are ordered by count descending.
    pub async fn get_ranked_nodes_by_edge_kind(
        &self,
        edge_kind: &EdgeKind,
        node_kind: Option<&NodeKind>,
        incoming: bool,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(Node, u64)>> {
        debug_assert!(
            limit > 0,
            "get_ranked_nodes_by_edge_kind limit must be positive"
        );
        debug_assert!(
            !edge_kind.as_str().is_empty(),
            "edge_kind must not be empty"
        );
        let (join_col, group_col) = if incoming {
            ("e.target", "e.target")
        } else {
            ("e.source", "e.source")
        };

        let mut conditions = vec!["e.kind = ?1".to_string()];
        let mut param_values: Vec<libsql::Value> =
            vec![libsql::Value::Text(edge_kind.as_str().to_string())];
        let mut param_idx = 2;

        if let Some(nk) = node_kind {
            conditions.push(format!("n.kind = ?{param_idx}"));
            param_values.push(libsql::Value::Text(nk.as_str().to_string()));
            param_idx += 1;
        }
        if let Some(prefix) = path_prefix {
            conditions.push(format!("n.file_path LIKE ?{param_idx}"));
            param_values.push(libsql::Value::Text(format!("{prefix}%")));
            param_idx += 1;
        }

        let where_clause = conditions.join(" AND ");
        let sql = format!(
            "SELECT n.id, n.kind, n.name, n.qualified_name, n.file_path,
                    n.start_line, n.end_line, n.start_column, n.end_column,
                    n.docstring, n.signature, n.visibility, n.is_async, n.branches, n.loops, n.returns, n.max_nesting, n.unsafe_blocks, n.unchecked_calls, n.assertions, n.updated_at, n.attrs_start_line, n.parent_id,
                    COUNT(*) AS cnt
             FROM edges e
             JOIN nodes n ON {join_col} = n.id
             WHERE {where_clause}
             GROUP BY {group_col}
             ORDER BY cnt DESC
             LIMIT ?{param_idx}"
        );
        param_values.push(libsql::Value::Integer(limit as i64));

        let op = "get_ranked_nodes_by_edge_kind";
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query ranked nodes: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map row: {e}"),
                operation: op.to_string(),
            })?;
            let count = row.get::<u64>(23).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read count column: {e}"),
                operation: op.to_string(),
            })?;
            items.push((node, count));
        }

        Ok(items)
    }

    /// Returns nodes ranked by line span (`end_line` - `start_line` + 1), optionally
    /// filtered by node kind. Results are ordered by size descending.
    pub async fn get_largest_nodes(
        &self,
        node_kind: Option<&NodeKind>,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(Node, u32)>> {
        let mut conditions: Vec<String> = Vec::new();
        let mut param_values: Vec<libsql::Value> = Vec::new();
        let mut param_idx = 1;

        if let Some(nk) = node_kind {
            conditions.push(format!("kind = ?{param_idx}"));
            param_values.push(libsql::Value::Text(nk.as_str().to_string()));
            param_idx += 1;
        }
        if let Some(prefix) = path_prefix {
            conditions.push(format!("file_path LIKE ?{param_idx}"));
            param_values.push(libsql::Value::Text(format!("{prefix}%")));
            param_idx += 1;
        }

        let where_clause = if conditions.is_empty() {
            String::new()
        } else {
            format!("WHERE {}", conditions.join(" AND "))
        };

        let sql = format!(
            "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id,
                    (end_line - start_line + 1) AS lines
             FROM nodes
             {where_clause}
             ORDER BY lines DESC
             LIMIT ?{param_idx}"
        );
        param_values.push(libsql::Value::Integer(limit as i64));

        let op = "get_largest_nodes";
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query largest nodes: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map row: {e}"),
                operation: op.to_string(),
            })?;
            let lines = row.get::<u32>(23).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read lines column: {e}"),
                operation: op.to_string(),
            })?;
            items.push((node, lines));
        }

        Ok(items)
    }

    /// Returns files ranked by coupling (number of distinct other files connected
    /// via cross-file edges). `fan_in` mode counts how many files depend on each
    /// file; `fan_out` counts how many files each file depends on.
    ///
    /// Only `calls`, `uses`, `implements`, and `extends` edges are considered.
    pub async fn get_file_coupling(
        &self,
        fan_in: bool,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(String, u64)>> {
        let (group_alias, count_alias) = if fan_in {
            ("n_tgt", "n_src")
        } else {
            ("n_src", "n_tgt")
        };

        let path_filter = match path_prefix {
            Some(prefix) => format!("AND {group_alias}.file_path LIKE '{prefix}%'"),
            None => String::new(),
        };

        let sql = format!(
            "SELECT {group_alias}.file_path, COUNT(DISTINCT {count_alias}.file_path) AS coupling
             FROM edges e
             JOIN nodes n_src ON e.source = n_src.id
             JOIN nodes n_tgt ON e.target = n_tgt.id
             WHERE e.kind IN ('calls', 'uses', 'implements', 'extends')
               AND n_src.file_path != n_tgt.file_path
               {path_filter}
             GROUP BY {group_alias}.file_path
             ORDER BY coupling DESC
             LIMIT ?1"
        );

        let op = "get_file_coupling";
        let mut rows = self
            .conn()
            .query(&sql, params![limit as i64])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query file coupling: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let file_path = row.get::<String>(0).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read file_path: {e}"),
                operation: op.to_string(),
            })?;
            let count = row.get::<u64>(1).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read coupling count: {e}"),
                operation: op.to_string(),
            })?;
            items.push((file_path, count));
        }

        Ok(items)
    }

    /// Returns the maximum inheritance depth for classes/interfaces reachable
    /// via `extends` edges. Uses a recursive CTE to walk the hierarchy.
    ///
    /// Each result is a (`leaf_node`, depth) pair where depth is the number of
    /// `extends` hops from the leaf to the root of its hierarchy.
    pub async fn get_inheritance_depth(
        &self,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(Node, u64)>> {
        let path_filter = match path_prefix {
            Some(prefix) => format!("WHERE n.file_path LIKE '{prefix}%'"),
            None => String::new(),
        };

        // Track visited node IDs in `path` to avoid blowing up on cycles in the
        // `extends` graph. Without this guard, a cycle (or trait bound that
        // points back to itself through generics, common in Rust workspaces
        // like polkadot-sdk) makes the CTE explore the cycle up to the depth
        // bound, multiplied by every entry point — `get_inheritance_depth` then
        // takes >60s on polkadot vs 0.3s with cycle detection.
        //
        // Note the predicate order in the recursive step: `h.depth < 50` is a
        // cheap integer compare and is evaluated before the path `instr`
        // string-scan, so cycles still under the depth bound short-circuit
        // without paying for the substring search. Reducing the hierarchy to
        // `(leaf_id, max_depth)` in an inner subquery before joining `nodes`
        // means the `LIKE` path filter only runs against distinct leaves,
        // not against the (potentially huge) full hierarchy table.
        let sql = format!(
            "WITH RECURSIVE hierarchy(leaf_id, current_id, depth, path) AS (
                 SELECT e.source, e.target, 1,
                        ',' || e.source || ',' || e.target || ','
                 FROM edges e
                 WHERE e.kind = 'extends'
                 UNION ALL
                 SELECT h.leaf_id, e.target, h.depth + 1,
                        h.path || e.target || ','
                 FROM hierarchy h
                 JOIN edges e ON e.source = h.current_id AND e.kind = 'extends'
                 WHERE h.depth < 50
                   AND instr(h.path, ',' || e.target || ',') = 0
             ),
             leaf_depths AS (
                 SELECT leaf_id, MAX(depth) AS max_depth
                 FROM hierarchy
                 GROUP BY leaf_id
             )
             SELECT n.id, n.kind, n.name, n.qualified_name, n.file_path,
                    n.start_line, n.end_line, n.start_column, n.end_column,
                    n.docstring, n.signature, n.visibility, n.is_async, n.branches, n.loops, n.returns, n.max_nesting, n.unsafe_blocks, n.unchecked_calls, n.assertions, n.updated_at, n.attrs_start_line, n.parent_id,
                    ld.max_depth
             FROM leaf_depths ld
             JOIN nodes n ON ld.leaf_id = n.id
             {path_filter}
             ORDER BY ld.max_depth DESC
             LIMIT ?1"
        );

        let op = "get_inheritance_depth";
        let mut rows = self
            .conn()
            .query(&sql, params![limit as i64])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query inheritance depth: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map row: {e}"),
                operation: op.to_string(),
            })?;
            let depth = row.get::<u64>(23).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read depth column: {e}"),
                operation: op.to_string(),
            })?;
            items.push((node, depth));
        }

        Ok(items)
    }

    /// Returns node kind counts grouped by file or directory prefix.
    ///
    /// If `path_prefix` is provided, only files under that path are included.
    /// Results are grouped by (`file_path`, kind) and ordered by file then count.
    pub async fn get_node_distribution(
        &self,
        path_prefix: Option<&str>,
    ) -> Result<Vec<(String, String, u64)>> {
        let (sql, param_values): (&str, Vec<libsql::Value>) = match path_prefix {
            Some(prefix) => (
                "SELECT file_path, kind, COUNT(*) AS cnt
                 FROM nodes
                 WHERE file_path LIKE ?1
                 GROUP BY file_path, kind
                 ORDER BY file_path, cnt DESC",
                vec![libsql::Value::Text(format!("{prefix}%"))],
            ),
            None => (
                "SELECT file_path, kind, COUNT(*) AS cnt
                 FROM nodes
                 GROUP BY file_path, kind
                 ORDER BY file_path, cnt DESC",
                vec![],
            ),
        };

        let op = "get_node_distribution";
        let mut rows = self
            .conn()
            .query(sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query node distribution: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let file_path = row.get::<String>(0).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read file_path: {e}"),
                operation: op.to_string(),
            })?;
            let kind = row.get::<String>(1).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read kind: {e}"),
                operation: op.to_string(),
            })?;
            let count = row.get::<u64>(2).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read count: {e}"),
                operation: op.to_string(),
            })?;
            items.push((file_path, kind, count));
        }

        Ok(items)
    }

    /// Returns all `calls` edges for cycle detection in the call graph.
    ///
    /// Returns `(source_id, target_id)` pairs for every `calls` edge.
    pub async fn get_call_edges(&self, path_prefix: Option<&str>) -> Result<Vec<(String, String)>> {
        let op = "get_call_edges";
        let (sql, param_values): (String, Vec<libsql::Value>) = match path_prefix {
            Some(prefix) => (
                "SELECT e.source, e.target FROM edges e
                 JOIN nodes n ON e.source = n.id
                 WHERE e.kind = 'calls' AND n.file_path LIKE ?1"
                    .to_string(),
                vec![libsql::Value::Text(format!("{prefix}%"))],
            ),
            None => (
                "SELECT source, target FROM edges WHERE kind = 'calls'".to_string(),
                vec![],
            ),
        };
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query call edges: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let source = row.get::<String>(0).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read source: {e}"),
                operation: op.to_string(),
            })?;
            let target = row.get::<String>(1).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read target: {e}"),
                operation: op.to_string(),
            })?;
            items.push((source, target));
        }

        Ok(items)
    }

    /// Returns all `calls` edges with their source line for cycle detection.
    ///
    /// Returns `(source_id, target_id, line)` tuples for every `calls` edge.
    pub async fn get_call_edges_with_lines(
        &self,
        path_prefix: Option<&str>,
    ) -> Result<Vec<(String, String, Option<u32>)>> {
        let op = "get_call_edges_with_lines";
        let (sql, param_values): (String, Vec<libsql::Value>) = match path_prefix {
            Some(prefix) => (
                "SELECT e.source, e.target, e.line FROM edges e
                 JOIN nodes n ON e.source = n.id
                 WHERE e.kind = 'calls' AND n.file_path LIKE ?1"
                    .to_string(),
                vec![libsql::Value::Text(format!("{prefix}%"))],
            ),
            None => (
                "SELECT source, target, line FROM edges WHERE kind = 'calls'".to_string(),
                vec![],
            ),
        };
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query call edges with lines: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let source = row.get::<String>(0).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read source: {e}"),
                operation: op.to_string(),
            })?;
            let target = row.get::<String>(1).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read target: {e}"),
                operation: op.to_string(),
            })?;
            let line = row.get::<u32>(2).ok();
            items.push((source, target, line));
        }

        Ok(items)
    }

    /// Returns functions/methods ranked by a composite complexity score.
    ///
    /// Complexity = `line_count` + (`call_fan_out` * 3) + `call_fan_in`.
    /// Line count reflects size, fan-out reflects cognitive load, fan-in
    /// reflects coupling. Results are ordered by score descending.
    pub async fn get_complexity_ranked(
        &self,
        node_kind: Option<&NodeKind>,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(Node, u32, u64, u64, u64)>> {
        debug_assert!(limit > 0, "get_complexity_ranked limit must be positive");
        let mut conditions: Vec<String> = Vec::new();
        let mut param_values: Vec<libsql::Value> = Vec::new();
        let mut param_idx = 1;

        match node_kind {
            Some(nk) => {
                conditions.push(format!("n.kind = ?{param_idx}"));
                param_values.push(libsql::Value::Text(nk.as_str().to_string()));
                param_idx += 1;
            }
            None => {
                conditions.push("n.kind IN ('function', 'method')".to_string());
            }
        }
        if let Some(prefix) = path_prefix {
            conditions.push(format!("n.file_path LIKE ?{param_idx}"));
            param_values.push(libsql::Value::Text(format!("{prefix}%")));
            param_idx += 1;
        }

        let where_clause = conditions.join(" AND ");
        let sql = format!(
            "SELECT n.id, n.kind, n.name, n.qualified_name, n.file_path,
                    n.start_line, n.end_line, n.start_column, n.end_column,
                    n.docstring, n.signature, n.visibility, n.is_async, n.branches, n.loops, n.returns, n.max_nesting, n.unsafe_blocks, n.unchecked_calls, n.assertions, n.updated_at, n.attrs_start_line, n.parent_id,
                    (n.end_line - n.start_line + 1) AS lines,
                    COALESCE(out_calls.cnt, 0) AS fan_out,
                    COALESCE(in_calls.cnt, 0) AS fan_in,
                    ((n.end_line - n.start_line + 1) + COALESCE(out_calls.cnt, 0) * 3 + COALESCE(in_calls.cnt, 0)) AS score
             FROM nodes n
             LEFT JOIN (SELECT source, COUNT(*) AS cnt FROM edges WHERE kind = 'calls' GROUP BY source) out_calls ON out_calls.source = n.id
             LEFT JOIN (SELECT target, COUNT(*) AS cnt FROM edges WHERE kind = 'calls' GROUP BY target) in_calls ON in_calls.target = n.id
             WHERE {where_clause}
             ORDER BY score DESC
             LIMIT ?{param_idx}"
        );
        param_values.push(libsql::Value::Integer(limit as i64));

        let op = "get_complexity_ranked";
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query complexity ranking: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map row: {e}"),
                operation: op.to_string(),
            })?;
            let lines = row.get::<u32>(23).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read lines: {e}"),
                operation: op.to_string(),
            })?;
            let fan_out = row.get::<u64>(24).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read fan_out: {e}"),
                operation: op.to_string(),
            })?;
            let fan_in = row.get::<u64>(25).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read fan_in: {e}"),
                operation: op.to_string(),
            })?;
            let score = row.get::<u64>(26).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read score: {e}"),
                operation: op.to_string(),
            })?;
            items.push((node, lines, fan_out, fan_in, score));
        }

        Ok(items)
    }

    /// Returns public symbols that are missing docstrings.
    ///
    /// Filters to kinds that conventionally carry per-declaration docs
    /// (functions, methods, types, fields, variants, constants, modules, …).
    /// Excludes `namespace` and `package` because they are aggregators that
    /// almost never carry their own doc — reporting them would drown
    /// actionable items in noise. Checks for `NULL` or empty docstring.
    pub async fn get_undocumented_public_symbols(
        &self,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<Node>> {
        const DOC_COVERAGE_KINDS: &str = "'function', 'method', 'class', 'interface', 'trait', \
            'struct', 'enum', 'module', 'field', 'enum_variant', 'const', 'static', 'type_alias', \
            'property', 'csharp_property', 'record', 'data_class', 'sealed_class', 'object', \
            'case_class', 'kotlin_object', 'inner_class', 'abstract_method', 'constructor', \
            'struct_method', 'val', 'var', 'mixin', 'extension', 'union', 'typedef'";

        let (sql, param_values): (String, Vec<libsql::Value>) = match path_prefix {
            Some(prefix) => (
                format!(
                    "SELECT id, kind, name, qualified_name, file_path,
                            start_line, end_line, start_column, end_column,
                            docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                     FROM nodes
                     WHERE visibility = 'public'
                       AND (docstring IS NULL OR docstring = '')
                       AND kind IN ({DOC_COVERAGE_KINDS})
                       AND file_path LIKE ?1
                     ORDER BY file_path, start_line
                     LIMIT ?2"
                ),
                vec![
                    libsql::Value::Text(format!("{prefix}%")),
                    libsql::Value::Integer(limit as i64),
                ],
            ),
            None => (
                format!(
                    "SELECT id, kind, name, qualified_name, file_path,
                            start_line, end_line, start_column, end_column,
                            docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                     FROM nodes
                     WHERE visibility = 'public'
                       AND (docstring IS NULL OR docstring = '')
                       AND kind IN ({DOC_COVERAGE_KINDS})
                     ORDER BY file_path, start_line
                     LIMIT ?1"
                ),
                vec![libsql::Value::Integer(limit as i64)],
            ),
        };

        let op = "get_undocumented_public_symbols";
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query undocumented symbols: {e}"),
                operation: op.to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, op).await
    }

    /// Returns classes/structs ranked by number of contained members
    /// (methods, fields, constructors). Identifies "god classes" with
    /// excessive responsibility.
    pub async fn get_god_classes(
        &self,
        path_prefix: Option<&str>,
        limit: usize,
    ) -> Result<Vec<(Node, u64, u64, u64)>> {
        let path_filter = match path_prefix {
            Some(prefix) => format!("AND n.file_path LIKE '{prefix}%'"),
            None => String::new(),
        };

        // After v9, containment is `nodes.parent_id`, not Contains edges.
        // Join each candidate container directly to its children via parent_id.
        let sql = format!(
            "SELECT n.id, n.kind, n.name, n.qualified_name, n.file_path,
                    n.start_line, n.end_line, n.start_column, n.end_column,
                    n.docstring, n.signature, n.visibility, n.is_async, n.branches, n.loops, n.returns, n.max_nesting, n.unsafe_blocks, n.unchecked_calls, n.assertions, n.updated_at, n.attrs_start_line, n.parent_id,
                    SUM(CASE WHEN c.kind IN ('method', 'abstract_method', 'constructor') THEN 1 ELSE 0 END) AS methods,
                    SUM(CASE WHEN c.kind = 'field' THEN 1 ELSE 0 END) AS fields,
                    COUNT(*) AS total
             FROM nodes n
             JOIN nodes c ON c.parent_id = n.id
             WHERE n.kind IN ('class', 'struct', 'inner_class', 'object')
               {path_filter}
             GROUP BY n.id
             ORDER BY total DESC
             LIMIT ?1"
        );

        let op = "get_god_classes";
        let mut rows = self
            .conn()
            .query(&sql, params![limit as i64])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query god classes: {e}"),
                operation: op.to_string(),
            })?;

        let mut items = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read row: {e}"),
            operation: op.to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map row: {e}"),
                operation: op.to_string(),
            })?;
            let methods = row.get::<u64>(23).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read methods: {e}"),
                operation: op.to_string(),
            })?;
            let fields = row.get::<u64>(24).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read fields: {e}"),
                operation: op.to_string(),
            })?;
            let total = row.get::<u64>(25).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read total: {e}"),
                operation: op.to_string(),
            })?;
            items.push((node, methods, fields, total));
        }

        Ok(items)
    }

    /// Returns every edge in the database.
    pub async fn get_all_edges(&self) -> Result<Vec<Edge>> {
        let mut rows = self
            .conn()
            .query("SELECT source, target, kind, line FROM edges", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query all edges: {e}"),
                operation: "get_all_edges".to_string(),
            })?;

        collect_rows(&mut rows, row_to_edge, "get_all_edges").await
    }

    /// Deletes all edges originating from a given source node.
    pub async fn delete_edges_by_source(&self, source_id: &str) -> Result<()> {
        self.conn()
            .execute("DELETE FROM edges WHERE source = ?1", params![source_id])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to delete edges by source: {e}"),
                operation: "delete_edges_by_source".to_string(),
            })?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// File operations
// ---------------------------------------------------------------------------

impl Database {
    /// Inserts or replaces a file record.
    /// Batch upserts multiple file records using raw SQL for throughput.
    pub async fn upsert_files(&self, files: &[FileRecord]) -> Result<()> {
        if files.is_empty() {
            return Ok(());
        }

        self.conn()
            .execute("BEGIN", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to begin: {e}"),
                operation: "upsert_files".to_string(),
            })?;

        let stmt = self.conn()
            .prepare("INSERT OR REPLACE INTO files (path,content_hash,size,modified_at,indexed_at,node_count) VALUES (?1,?2,?3,?4,?5,?6)")
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to prepare: {e}"),
                operation: "upsert_files".to_string(),
            })?;

        for file in files {
            stmt.execute(params![
                file.path.as_str(),
                file.content_hash.as_str(),
                file.size as i64,
                file.modified_at,
                file.indexed_at,
                i64::from(file.node_count),
            ])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to upsert file: {e}"),
                operation: "upsert_files".to_string(),
            })?;
            stmt.reset();
        }

        self.conn()
            .execute("COMMIT", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to commit: {e}"),
                operation: "upsert_files".to_string(),
            })?;
        Ok(())
    }

    pub async fn upsert_file(&self, file: &FileRecord) -> Result<()> {
        self.conn()
            .execute(
                "INSERT OR REPLACE INTO files
                (path, content_hash, size, modified_at, indexed_at, node_count)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    file.path.as_str(),
                    file.content_hash.as_str(),
                    file.size as i64,
                    file.modified_at,
                    file.indexed_at,
                    i64::from(file.node_count),
                ],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to upsert file: {e}"),
                operation: "upsert_file".to_string(),
            })?;
        Ok(())
    }

    /// Retrieves a file record by path, returning `None` if not found.
    pub async fn get_file(&self, path: &str) -> Result<Option<FileRecord>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT path, content_hash, size, modified_at, indexed_at, node_count
                 FROM files WHERE path = ?1",
                params![path],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query file: {e}"),
                operation: "get_file".to_string(),
            })?;

        match rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read file row: {e}"),
            operation: "get_file".to_string(),
        })? {
            Some(row) => {
                let file = row_to_file(&row).map_err(|e| TokenSaveError::Database {
                    message: format!("failed to map file row: {e}"),
                    operation: "get_file".to_string(),
                })?;
                Ok(Some(file))
            }
            None => Ok(None),
        }
    }

    /// Returns all file records.
    pub async fn get_all_files(&self) -> Result<Vec<FileRecord>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT path, content_hash, size, modified_at, indexed_at, node_count FROM files",
                (),
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query all files: {e}"),
                operation: "get_all_files".to_string(),
            })?;

        collect_rows(&mut rows, row_to_file, "get_all_files").await
    }

    /// Deletes a file record and cascades to delete its nodes first.
    pub async fn delete_file(&self, path: &str) -> Result<()> {
        self.delete_nodes_by_file(path).await?;
        self.conn()
            .execute("DELETE FROM files WHERE path = ?1", params![path])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to delete file: {e}"),
                operation: "delete_file".to_string(),
            })?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Unresolved reference operations
// ---------------------------------------------------------------------------

impl Database {
    /// Inserts a single unresolved reference.
    pub async fn insert_unresolved_ref(&self, uref: &UnresolvedRef) -> Result<()> {
        self.conn()
            .execute(
                "INSERT INTO unresolved_refs
                (from_node_id, reference_name, reference_kind, line, col, file_path)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    uref.from_node_id.as_str(),
                    uref.reference_name.as_str(),
                    uref.reference_kind.as_str(),
                    i64::from(uref.line),
                    i64::from(uref.column),
                    uref.file_path.as_str(),
                ],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to insert unresolved ref: {e}"),
                operation: "insert_unresolved_ref".to_string(),
            })?;
        Ok(())
    }

    /// Inserts a batch of unresolved references using a prepared statement.
    pub async fn insert_unresolved_refs(&self, refs: &[UnresolvedRef]) -> Result<()> {
        if refs.is_empty() {
            return Ok(());
        }

        self.conn()
            .execute("BEGIN", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to begin: {e}"),
                operation: "insert_unresolved_refs".to_string(),
            })?;

        let stmt = self.conn()
            .prepare("INSERT INTO unresolved_refs (from_node_id,reference_name,reference_kind,line,col,file_path) VALUES (?1,?2,?3,?4,?5,?6)")
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to prepare: {e}"),
                operation: "insert_unresolved_refs".to_string(),
            })?;

        for uref in refs {
            stmt.execute(params![
                uref.from_node_id.as_str(),
                uref.reference_name.as_str(),
                uref.reference_kind.as_str(),
                i64::from(uref.line),
                i64::from(uref.column),
                uref.file_path.as_str(),
            ])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to insert unresolved ref: {e}"),
                operation: "insert_unresolved_refs".to_string(),
            })?;
            stmt.reset();
        }

        self.conn()
            .execute("COMMIT", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to commit: {e}"),
                operation: "insert_unresolved_refs".to_string(),
            })?;
        Ok(())
    }

    /// Returns all unresolved references.
    pub async fn get_unresolved_refs(&self) -> Result<Vec<UnresolvedRef>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT from_node_id, reference_name, reference_kind, line, col, file_path
                 FROM unresolved_refs",
                (),
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query unresolved refs: {e}"),
                operation: "get_unresolved_refs".to_string(),
            })?;

        collect_rows(&mut rows, row_to_unresolved_ref, "get_unresolved_refs").await
    }

    /// Removes all unresolved references.
    pub async fn clear_unresolved_refs(&self) -> Result<()> {
        self.conn()
            .execute("DELETE FROM unresolved_refs", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to clear unresolved refs: {e}"),
                operation: "clear_unresolved_refs".to_string(),
            })?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Search
// ---------------------------------------------------------------------------

impl Database {
    /// Searches nodes by name, qualified name, docstring, or signature.
    ///
    /// Attempts an FTS5 prefix match first. If the FTS index is corrupted,
    /// it is automatically rebuilt and the query retried. If FTS returns no
    /// results, falls back to a `LIKE` query.
    pub async fn search_nodes(&self, query: &str, limit: usize) -> Result<Vec<SearchResult>> {
        debug_assert!(!query.is_empty(), "search_nodes called with empty query");
        debug_assert!(limit > 0, "search_nodes limit must be positive");
        // Sanitize query for FTS5: wrap each word in double quotes to escape
        // special characters (*, ?, :, etc.) and join with spaces (implicit OR).
        let fts_query: String = query
            .split_whitespace()
            .filter(|w| !w.is_empty())
            .map(|w| {
                let sanitized: String = w.chars().filter(|c| *c != '"').collect();
                format!("\"{sanitized}\"*")
            })
            .collect::<Vec<_>>()
            .join(" OR ");

        if fts_query.is_empty() {
            return Ok(Vec::new());
        }

        // Try FTS search, with one self-healing retry on corruption.
        let fts_result = self.search_nodes_fts(&fts_query, limit).await;
        match fts_result {
            Ok(ref results) if !results.is_empty() => return fts_result,
            Ok(_) => {} // empty — fall through to LIKE
            Err(ref e) if Self::is_corruption_error(e) => {
                eprintln!("[tokensave] FTS index corruption detected — rebuilding…");
                if self.rebuild_fts().await.is_ok() {
                    match self.search_nodes_fts(&fts_query, limit).await {
                        Ok(results) if !results.is_empty() => return Ok(results),
                        Ok(_) => {} // fall through to LIKE
                        Err(e) => return Err(e),
                    }
                }
                // rebuild_fts failed — fall through to LIKE as last resort
            }
            Err(e) => return Err(e),
        }

        // Fallback: LIKE query
        let like_pattern = format!("%{query}%");
        let mut rows = self
            .conn()
            .query(
                "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async, branches, loops, returns, max_nesting, unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
                 FROM nodes
                 WHERE name LIKE ?1 OR qualified_name LIKE ?1 OR docstring LIKE ?1 OR signature LIKE ?1
                 LIMIT ?2",
                params![like_pattern.as_str(), limit as i64],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to execute LIKE query: {e}"),
                operation: "search_nodes".to_string(),
            })?;

        let mut results = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read search result: {e}"),
            operation: "search_nodes".to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map search result: {e}"),
                operation: "search_nodes".to_string(),
            })?;
            results.push(SearchResult { node, score: 1.0 });
        }
        Ok(results)
    }

    /// Executes the FTS5 query and returns ranked results.
    async fn search_nodes_fts(&self, fts_query: &str, limit: usize) -> Result<Vec<SearchResult>> {
        let mut rows = self
            .conn()
            .query(
                "SELECT n.id, n.kind, n.name, n.qualified_name, n.file_path,
                    n.start_line, n.end_line, n.start_column, n.end_column,
                    n.docstring, n.signature, n.visibility, n.is_async, n.branches, n.loops, n.returns, n.max_nesting, n.unsafe_blocks, n.unchecked_calls, n.assertions, n.updated_at, n.attrs_start_line, n.parent_id,
                    bm25(nodes_fts, 10.0, 5.0, 1.0, 2.0) AS rank
                 FROM nodes_fts
                 JOIN nodes n ON nodes_fts.rowid = n.rowid
                 WHERE nodes_fts MATCH ?1
                 ORDER BY bm25(nodes_fts, 10.0, 5.0, 1.0, 2.0)
                 LIMIT ?2",
                params![fts_query, limit as i64],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to execute FTS query: {e}"),
                operation: "search_nodes".to_string(),
            })?;

        let mut results = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read search result: {e}"),
            operation: "search_nodes".to_string(),
        })? {
            let node = row_to_node(&row).map_err(|e| TokenSaveError::Database {
                message: format!("failed to map search result: {e}"),
                operation: "search_nodes".to_string(),
            })?;
            let rank: f64 = row.get::<f64>(23).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read rank: {e}"),
                operation: "search_nodes".to_string(),
            })?;
            results.push(SearchResult { node, score: -rank });
        }
        Ok(results)
    }

    /// Returns a map of `node_id` → incoming "calls" edge count for the given IDs.
    /// IDs not found in any edge target are omitted from the result.
    pub async fn batch_incoming_call_counts(
        &self,
        node_ids: &[String],
    ) -> Result<std::collections::HashMap<String, u64>> {
        let mut counts = std::collections::HashMap::new();
        if node_ids.is_empty() {
            return Ok(counts);
        }
        let placeholders = build_qmark_placeholders(node_ids.len());
        let sql = format!(
            "SELECT target, COUNT(*) AS cnt FROM edges WHERE target IN ({placeholders}) AND kind = 'calls' GROUP BY target",
        );
        let param_values: Vec<libsql::Value> = node_ids
            .iter()
            .map(|id| libsql::Value::Text(id.clone()))
            .collect();
        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to batch count incoming calls: {e}"),
                operation: "batch_incoming_call_counts".to_string(),
            })?;
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read batch call count row: {e}"),
            operation: "batch_incoming_call_counts".to_string(),
        })? {
            let id: String = row.get(0).unwrap_or_default();
            let cnt: u64 = row.get::<u64>(1).unwrap_or(0);
            counts.insert(id, cnt);
        }
        Ok(counts)
    }

    /// Finds nodes whose `name` column exactly matches one of the given names
    /// (case-insensitive). Used to supplement FTS results so that perfect
    /// matches are never buried by BM25 noise.
    pub async fn search_nodes_by_exact_name(
        &self,
        names: &[String],
        limit: usize,
    ) -> Result<Vec<Node>> {
        if names.is_empty() || limit == 0 {
            return Ok(Vec::new());
        }
        let placeholders = build_qmark_placeholders(names.len());
        let sql = format!(
            "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async,
                    branches, loops, returns, max_nesting,
                    unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
             FROM nodes
             WHERE LOWER(name) IN ({placeholders})
             LIMIT ?",
        );
        let mut param_values: Vec<libsql::Value> = names
            .iter()
            .map(|n| libsql::Value::Text(n.to_lowercase()))
            .collect();
        param_values.push(libsql::Value::Integer(limit as i64));

        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to search by exact name: {e}"),
                operation: "search_nodes_by_exact_name".to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, "search_nodes_by_exact_name").await
    }

    /// Returns `true` if the error indicates `SQLite` database corruption.
    pub fn is_corruption_error(e: &TokenSaveError) -> bool {
        match e {
            TokenSaveError::Database { message, .. } => {
                message.contains("malformed")
                    || message.contains("corrupt")
                    || message.contains("disk image")
            }
            _ => false,
        }
    }
}

// ---------------------------------------------------------------------------
// Statistics
// ---------------------------------------------------------------------------

impl Database {
    /// Returns aggregate statistics about the code graph.
    pub async fn get_stats(&self) -> Result<GraphStats> {
        // Single query for all scalar counts: nodes, edges, files, last_updated, total_source_bytes
        let mut counts_rows = self
            .conn()
            .query(
                "SELECT \
                   (SELECT COUNT(*) FROM nodes), \
                   (SELECT COUNT(*) FROM edges), \
                   (SELECT COUNT(*) FROM files), \
                   (SELECT COALESCE(MAX(indexed_at), 0) FROM files), \
                   (SELECT COALESCE(SUM(size), 0) FROM files)",
                (),
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query counts: {e}"),
                operation: "get_stats".to_string(),
            })?;
        let counts_row = counts_rows
            .next()
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to read counts row: {e}"),
                operation: "get_stats".to_string(),
            })?;
        let (node_count, edge_count, file_count, last_updated, total_source_bytes) =
            match counts_row {
                Some(r) => {
                    let nc: i64 = r.get(0).unwrap_or(0);
                    let ec: i64 = r.get(1).unwrap_or(0);
                    let fc: i64 = r.get(2).unwrap_or(0);
                    let lu: i64 = r.get(3).unwrap_or(0);
                    let ts: i64 = r.get(4).unwrap_or(0);
                    (nc as u64, ec as u64, fc as u64, lu as u64, ts as u64)
                }
                None => (0, 0, 0, 0, 0),
            };

        // Nodes grouped by kind
        let nodes_by_kind = query_kind_counts(
            self.conn(),
            "SELECT kind, COUNT(*) FROM nodes GROUP BY kind",
        )
        .await?;

        // Edges grouped by kind
        let edges_by_kind = query_kind_counts(
            self.conn(),
            "SELECT kind, COUNT(*) FROM edges GROUP BY kind",
        )
        .await?;

        let db_size_bytes = self.size().await.unwrap_or(0);

        // Files grouped by language (derived from file extension)
        let files_by_language = query_kind_counts(
            self.conn(),
            "SELECT \
               CASE \
                 WHEN path LIKE '%.rs' THEN 'Rust' \
                 WHEN path LIKE '%.go' THEN 'Go' \
                 WHEN path LIKE '%.java' THEN 'Java' \
                 WHEN path LIKE '%.scala' OR path LIKE '%.sc' THEN 'Scala' \
                 ELSE 'Other' \
               END AS lang, \
               COUNT(*) \
             FROM files GROUP BY lang",
        )
        .await?;

        let last_sync_at = self
            .get_metadata("last_sync_at")
            .await?
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(0);
        let last_full_sync_at = self
            .get_metadata("last_full_sync_at")
            .await?
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(0);
        let last_sync_duration_ms = self
            .get_metadata("last_sync_duration_ms")
            .await?
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(0);

        Ok(GraphStats {
            node_count,
            edge_count,
            file_count,
            nodes_by_kind,
            edges_by_kind,
            db_size_bytes,
            last_updated,
            total_source_bytes,
            files_by_language,
            last_sync_at,
            last_full_sync_at,
            last_sync_duration_ms,
        })
    }

    /// Returns the most recent `indexed_at` timestamp across all files,
    /// or 0 if the files table is empty.
    pub async fn last_index_time(&self) -> Result<i64> {
        query_scalar_i64(
            self.conn(),
            "SELECT COALESCE(MAX(indexed_at), 0) FROM files",
            "last_index_time",
        )
        .await
    }
}

// ---------------------------------------------------------------------------
// Clear
// ---------------------------------------------------------------------------

impl Database {
    /// Removes all data from every table.
    pub async fn clear(&self) -> Result<()> {
        self.conn()
            .execute_batch(
                "DELETE FROM vectors;
                 DELETE FROM unresolved_refs;
                 DELETE FROM edges;
                 DELETE FROM nodes;
                 DELETE FROM files;",
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to clear database: {e}"),
                operation: "clear".to_string(),
            })?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Metadata
// ---------------------------------------------------------------------------

impl Database {
    /// Reads a metadata value by key, returning `None` if not set.
    pub async fn get_metadata(&self, key: &str) -> Result<Option<String>> {
        let mut rows = self
            .conn()
            .query("SELECT value FROM metadata WHERE key = ?1", params![key])
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query metadata: {e}"),
                operation: "get_metadata".to_string(),
            })?;

        match rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read metadata row: {e}"),
            operation: "get_metadata".to_string(),
        })? {
            Some(row) => {
                let value: String = row.get(0).map_err(|e| TokenSaveError::Database {
                    message: format!("failed to read metadata value: {e}"),
                    operation: "get_metadata".to_string(),
                })?;
                Ok(Some(value))
            }
            None => Ok(None),
        }
    }

    /// Sets a metadata value, creating or replacing the entry.
    pub async fn set_metadata(&self, key: &str, value: &str) -> Result<()> {
        self.conn()
            .execute(
                "INSERT OR REPLACE INTO metadata (key, value) VALUES (?1, ?2)",
                params![key, value],
            )
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to set metadata: {e}"),
                operation: "set_metadata".to_string(),
            })?;
        Ok(())
    }

    /// Returns all nodes under a directory prefix filtered by kinds.
    ///
    /// Uses `LIKE dir || '%'` for the path prefix and an `IN` clause for kinds.
    pub async fn get_nodes_by_dir(&self, dir: &str, kinds: &[NodeKind]) -> Result<Vec<Node>> {
        if kinds.is_empty() {
            return Ok(Vec::new());
        }

        let kind_placeholders: Vec<String> = kinds
            .iter()
            .enumerate()
            .map(|(i, _)| format!("?{}", i + 2))
            .collect();
        let sql = format!(
            "SELECT id, kind, name, qualified_name, file_path,
                    start_line, end_line, start_column, end_column,
                    docstring, signature, visibility, is_async,
                    branches, loops, returns, max_nesting,
                    unsafe_blocks, unchecked_calls, assertions, updated_at, attrs_start_line, parent_id
             FROM nodes
             WHERE file_path LIKE ?1 || '%' AND kind IN ({})
             ORDER BY file_path, start_line",
            kind_placeholders.join(", ")
        );

        let mut param_values: Vec<libsql::Value> = Vec::new();
        param_values.push(libsql::Value::Text(dir.to_string()));
        for k in kinds {
            param_values.push(libsql::Value::Text(k.as_str().to_string()));
        }

        let mut rows = self
            .conn()
            .query(&sql, libsql::params_from_iter(param_values))
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query nodes by dir: {e}"),
                operation: "get_nodes_by_dir".to_string(),
            })?;

        collect_rows(&mut rows, row_to_node, "get_nodes_by_dir").await
    }

    /// Returns edges where both source and target are in the given node ID set.
    ///
    /// Batches queries in groups of 500 IDs to avoid SQL parameter limits.
    pub async fn get_internal_edges(&self, node_ids: &[String]) -> Result<Vec<Edge>> {
        const BATCH_SIZE: usize = 500;
        if node_ids.is_empty() {
            return Ok(Vec::new());
        }

        // Build a set of IDs for filtering targets in memory, then query
        // edges from each batch of sources.
        let id_set: std::collections::HashSet<&str> =
            node_ids.iter().map(std::string::String::as_str).collect();
        let mut all_edges = Vec::new();
        let mut offset = 0;
        while offset < node_ids.len() {
            let end = (offset + BATCH_SIZE).min(node_ids.len());
            let batch = &node_ids[offset..end];

            let placeholders: Vec<String> = batch
                .iter()
                .enumerate()
                .map(|(i, _)| format!("?{}", i + 1))
                .collect();
            let sql = format!(
                "SELECT source, target, kind, line FROM edges WHERE source IN ({})",
                placeholders.join(", ")
            );

            let param_values: Vec<libsql::Value> = batch
                .iter()
                .map(|id| libsql::Value::Text(id.clone()))
                .collect();

            let mut rows = self
                .conn()
                .query(&sql, libsql::params_from_iter(param_values))
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to query internal edges: {e}"),
                    operation: "get_internal_edges".to_string(),
                })?;

            let batch_edges: Vec<Edge> =
                collect_rows(&mut rows, row_to_edge, "get_internal_edges").await?;

            // Keep only edges whose target is also in the node set.
            for edge in batch_edges {
                if id_set.contains(edge.target.as_str()) {
                    all_edges.push(edge);
                }
            }

            offset = end;
        }

        Ok(all_edges)
    }

    /// Resolves the set of `annotation_usage` node ids whose name marks a
    /// function as a test (`#[test]`, `#[tokio::test]`, `#[async_std::test]`,
    /// `#[wasm_bindgen_test]`, …). Runs the leading-wildcard `LIKE` scan
    /// exactly once over the `kind = 'annotation_usage'` partition.
    ///
    /// `find_dead_code` uses this in a two-step "resolve + use" pattern
    /// (push the ids into a TEMP table, then probe by id) so the LIKE never
    /// runs in a correlated subquery — the per-row degenerate plan from the
    /// reverted 4.14.8 CTE attempt timed out at >60s on scirs; the original
    /// JOIN+LIKE form times out at >25s on chromium. Both pathologies stem
    /// from re-running the wildcard scan per candidate row.
    pub async fn collect_test_marker_ids(&self) -> Result<Vec<String>> {
        let op = "collect_test_marker_ids";
        let sql = "SELECT id FROM nodes
                   WHERE kind = 'annotation_usage'
                     AND (
                         name = 'test'
                         OR name LIKE '%::test'
                         OR name = 'wasm_bindgen_test'
                         OR name LIKE '%::wasm_bindgen_test'
                     )";
        let mut rows = self
            .conn()
            .query(sql, ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to query test marker ids: {e}"),
                operation: op.to_string(),
            })?;
        let mut ids = Vec::new();
        while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
            message: format!("failed to read marker id row: {e}"),
            operation: op.to_string(),
        })? {
            let id: String = row.get(0).map_err(|e| TokenSaveError::Database {
                message: format!("failed to read marker id column: {e}"),
                operation: op.to_string(),
            })?;
            ids.push(id);
        }
        Ok(ids)
    }

    /// Drops, recreates, and bulk-inserts `ids` into `temp.test_markers`.
    ///
    /// The temp table has a `PRIMARY KEY` on `id` so `SQLite` builds a real
    /// rowid B-tree — `IN (SELECT id FROM temp.test_markers)` in downstream
    /// queries probes via that index, not a wildcard scan. Inserts are
    /// chunked under `SQLite`'s 999-parameter limit.
    ///
    /// Always drops first, so a previous call on the same connection
    /// (e.g. consecutive `find_dead_code` from the same MCP client) does
    /// not collide. The caller should also drop the table when done — see
    /// `find_dead_code` for the wrapping pattern.
    pub async fn populate_test_marker_temp_table(&self, ids: &[String]) -> Result<()> {
        // `SQLite`'s default parameter limit is 999. Chunk well under that.
        const CHUNK_SIZE: usize = 500;

        let op = "populate_test_marker_temp_table";
        let conn = self.conn();

        // Drop + recreate so we always start from an empty table.
        conn.execute("DROP TABLE IF EXISTS temp.test_markers", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to drop temp.test_markers: {e}"),
                operation: op.to_string(),
            })?;
        conn.execute("CREATE TEMP TABLE test_markers (id TEXT PRIMARY KEY)", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to create temp.test_markers: {e}"),
                operation: op.to_string(),
            })?;

        if ids.is_empty() {
            return Ok(());
        }

        for chunk in ids.chunks(CHUNK_SIZE) {
            let mut sql = String::from("INSERT INTO temp.test_markers (id) VALUES ");
            for i in 0..chunk.len() {
                if i > 0 {
                    sql.push_str(", ");
                }
                sql.push_str("(?)");
            }
            let params: Vec<libsql::Value> = chunk
                .iter()
                .map(|id| libsql::Value::Text(id.clone()))
                .collect();
            conn.execute(&sql, libsql::params_from_iter(params))
                .await
                .map_err(|e| TokenSaveError::Database {
                    message: format!("failed to bulk-insert test markers: {e}"),
                    operation: op.to_string(),
                })?;
        }
        Ok(())
    }

    /// Drops `temp.test_markers` if it exists. Used as cleanup by
    /// `find_dead_code` so the table does not leak to other queries on the
    /// same connection.
    ///
    /// Safe to call even if the table doesn't exist (uses `IF EXISTS`).
    pub async fn drop_test_marker_temp_table(&self) -> Result<()> {
        self.conn()
            .execute("DROP TABLE IF EXISTS temp.test_markers", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to drop temp.test_markers: {e}"),
                operation: "drop_test_marker_temp_table".to_string(),
            })?;
        Ok(())
    }

    /// Materialises the set of node ids that are targets of a test-marker
    /// `annotates` edge into `temp.test_annotated_targets`.
    ///
    /// This is the second step of the dead-code test-exclusion pipeline:
    /// 1. `populate_test_marker_temp_table` fills `temp.test_markers`.
    /// 2. THIS fn pre-resolves "which nodes are annotated by any test
    ///    marker" into a small lookup table with a PK on `target`.
    /// 3. `find_dead_code`'s outer SELECT then uses
    ///    `id NOT IN (SELECT target FROM temp.test_annotated_targets)` —
    ///    an indexed PK probe per candidate.
    ///
    /// Why two tables instead of `IN (SELECT id FROM temp.test_markers)`
    /// inside a correlated `NOT EXISTS`: on chromium (~13 K markers,
    /// ~134 K dead-code candidates, ~411 K annotates edges) `SQLite` picked
    /// `idx_edges_unique (source, target, kind)` for the correlated
    /// subquery, iterating every marker as the outer driver for every
    /// candidate. That's ~1.7 billion index probes and a >25 s timeout
    /// on the MCP probe. Pre-materialising the *target* set means the
    /// per-candidate probe becomes a single indexed lookup against a
    /// table with ~15 K rows. Real measurement on chromium 7.5 GB DB:
    /// 0.75 s end-to-end (vs. >60 s for the single-temp-table form).
    pub async fn populate_test_annotated_targets_temp_table(&self) -> Result<()> {
        let op = "populate_test_annotated_targets_temp_table";
        let conn = self.conn();

        // Drop + recreate so we always start from an empty table — same
        // hygiene as the test_markers temp table.
        conn.execute("DROP TABLE IF EXISTS temp.test_annotated_targets", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to drop temp.test_annotated_targets: {e}"),
                operation: op.to_string(),
            })?;
        conn.execute(
            "CREATE TEMP TABLE test_annotated_targets (target TEXT PRIMARY KEY)",
            (),
        )
        .await
        .map_err(|e| TokenSaveError::Database {
            message: format!("failed to create temp.test_annotated_targets: {e}"),
            operation: op.to_string(),
        })?;

        // `INSERT OR IGNORE` because a single function can have multiple
        // test markers (e.g. `#[test] #[cfg(target_os = "linux")]`) — one
        // row per target, not per (target, marker) pair.
        conn.execute(
            "INSERT OR IGNORE INTO temp.test_annotated_targets (target)
             SELECT e.target FROM edges e
             WHERE e.kind = 'annotates'
               AND e.source IN (SELECT id FROM temp.test_markers)",
            (),
        )
        .await
        .map_err(|e| TokenSaveError::Database {
            message: format!("failed to populate temp.test_annotated_targets: {e}"),
            operation: op.to_string(),
        })?;
        Ok(())
    }

    /// Drops `temp.test_annotated_targets` if it exists. Cleanup pair for
    /// `populate_test_annotated_targets_temp_table`.
    pub async fn drop_test_annotated_targets_temp_table(&self) -> Result<()> {
        self.conn()
            .execute("DROP TABLE IF EXISTS temp.test_annotated_targets", ())
            .await
            .map_err(|e| TokenSaveError::Database {
                message: format!("failed to drop temp.test_annotated_targets: {e}"),
                operation: "drop_test_annotated_targets_temp_table".to_string(),
            })?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Shared helpers
// ---------------------------------------------------------------------------

/// Converts `Option<String>` to a `libsql::Value` for use in params.
fn opt_str(opt: Option<&str>) -> libsql::Value {
    match opt {
        Some(s) => libsql::Value::Text(s.to_string()),
        None => libsql::Value::Null,
    }
}

/// Appends a SQL-safe single-quoted string to `buf`, escaping `'` as `''`.
fn push_quoted(buf: &mut String, s: &str) {
    buf.push('\'');
    for ch in s.chars() {
        if ch == '\'' {
            buf.push_str("''");
        } else {
            buf.push(ch);
        }
    }
    buf.push('\'');
}

/// Appends a SQL-safe quoted string or NULL for Option<String>.
fn push_opt_quoted(buf: &mut String, opt: Option<&str>) {
    match opt {
        Some(s) => push_quoted(buf, s),
        None => buf.push_str("NULL"),
    }
}

/// Appends an integer literal to the buffer.
fn push_int(buf: &mut String, val: i64) {
    use std::fmt::Write;
    let _ = write!(buf, "{val}");
}

/// Collects all rows from a `Rows` iterator into a `Vec<T>` using the given
/// row-mapping function.
async fn collect_rows<T>(
    rows: &mut libsql::Rows,
    map_fn: fn(&libsql::Row) -> std::result::Result<T, libsql::Error>,
    operation: &str,
) -> Result<Vec<T>> {
    let mut items = Vec::new();
    while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
        message: format!("failed to read row: {e}"),
        operation: operation.to_string(),
    })? {
        items.push(map_fn(&row).map_err(|e| TokenSaveError::Database {
            message: format!("failed to map row: {e}"),
            operation: operation.to_string(),
        })?);
    }
    Ok(items)
}

/// Executes a `SELECT label, COUNT(*) ... GROUP BY` query and returns
/// the results as a `HashMap<String, u64>`.
async fn query_kind_counts(conn: &libsql::Connection, sql: &str) -> Result<HashMap<String, u64>> {
    let mut map = HashMap::new();
    let mut rows = conn
        .query(sql, ())
        .await
        .map_err(|e| TokenSaveError::Database {
            message: format!("failed to query kind counts: {e}"),
            operation: "get_stats".to_string(),
        })?;
    while let Some(row) = rows.next().await.map_err(|e| TokenSaveError::Database {
        message: format!("failed to read kind count row: {e}"),
        operation: "get_stats".to_string(),
    })? {
        let kind: String = row.get(0).map_err(|e| TokenSaveError::Database {
            message: format!("failed to read kind: {e}"),
            operation: "get_stats".to_string(),
        })?;
        let count: i64 = row.get(1).map_err(|e| TokenSaveError::Database {
            message: format!("failed to read count: {e}"),
            operation: "get_stats".to_string(),
        })?;
        if count > 0 {
            map.insert(kind, count as u64);
        }
    }
    Ok(map)
}

/// Executes a scalar query returning a single `i64` value.
async fn query_scalar_i64(conn: &libsql::Connection, sql: &str, operation: &str) -> Result<i64> {
    let mut rows = conn
        .query(sql, ())
        .await
        .map_err(|e| TokenSaveError::Database {
            message: format!("failed to execute scalar query: {e}"),
            operation: operation.to_string(),
        })?;

    let row = rows
        .next()
        .await
        .map_err(|e| TokenSaveError::Database {
            message: format!("failed to read scalar row: {e}"),
            operation: operation.to_string(),
        })?
        .ok_or_else(|| TokenSaveError::Database {
            message: "no result from scalar query".to_string(),
            operation: operation.to_string(),
        })?;

    row.get::<i64>(0).map_err(|e| TokenSaveError::Database {
        message: format!("failed to read scalar value: {e}"),
        operation: operation.to_string(),
    })
}
