mcp-server-sqlite 1.0.0

//! The `list_indexes` tool: returns index metadata for the database. When given
//! a table name, only indexes on that table are returned; otherwise all indexes
//! are listed. Each entry includes the index name, owning table, uniqueness
//! flag, column list, and partial-index predicate (if any).

use rmcp::model::{Content, IntoContents};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

use super::ToolError;
use crate::{mcp::McpServerSqlite, traits::SqliteServerTool};

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Default,
    Serialize,
    Deserialize,
    JsonSchema,
)]
/// List indexes in the database. When a table name is provided, only indexes
/// belonging to that table are returned. Otherwise all indexes across every
/// table are listed. Each index entry includes its name, the table it belongs
/// to, whether it enforces uniqueness, the ordered list of indexed columns, and
/// the WHERE predicate for partial indexes.
pub struct ListIndexesTool;

impl SqliteServerTool for ListIndexesTool {
    const NAME: &str = "list_indexes";

    type Context = McpServerSqlite;
    type Error = ToolError<ListIndexesError>;

    type Input = ListIndexesInput;
    type Output = ListIndexesOutput;

    fn handle(
        ctx: &Self::Context,
        input: Self::Input,
    ) -> Result<Self::Output, Self::Error> {
        let conn = ctx
            .connection()
            .map_err(|source| ToolError::Connection { source })?;

        let raw_indexes = match &input.table_name {
            Some(table) => {
                let mut stmt = conn
                    .prepare(
                        "SELECT name, tbl_name, sql FROM sqlite_master \
                         WHERE type = 'index' AND tbl_name = ? \
                         ORDER BY tbl_name, name",
                    )
                    .map_err(|source| {
                        ToolError::Tool(ListIndexesError::Query { source })
                    })?;

                stmt.query_map([table], |row| {
                    Ok(RawIndex {
                        name: row.get(0)?,
                        table_name: row.get(1)?,
                        sql: row.get(2)?,
                    })
                })
                .map_err(|source| {
                    ToolError::Tool(ListIndexesError::Query { source })
                })?
                .collect::<Result<Vec<_>, _>>()
                .map_err(|source| {
                    ToolError::Tool(ListIndexesError::Query { source })
                })?
            }
            None => {
                let mut stmt = conn
                    .prepare(
                        "SELECT name, tbl_name, sql FROM sqlite_master \
                         WHERE type = 'index' \
                         ORDER BY tbl_name, name",
                    )
                    .map_err(|source| {
                        ToolError::Tool(ListIndexesError::Query { source })
                    })?;

                stmt.query_map([], |row| {
                    Ok(RawIndex {
                        name: row.get(0)?,
                        table_name: row.get(1)?,
                        sql: row.get(2)?,
                    })
                })
                .map_err(|source| {
                    ToolError::Tool(ListIndexesError::Query { source })
                })?
                .collect::<Result<Vec<_>, _>>()
                .map_err(|source| {
                    ToolError::Tool(ListIndexesError::Query { source })
                })?
            }
        };

        let indexes = raw_indexes
            .into_iter()
            .map(|raw| {
                let columns = query_index_columns(&conn, &raw.name)?;
                let unique = is_unique_index(&raw);
                let partial_predicate =
                    raw.sql.as_deref().and_then(extract_where_clause);

                Ok(IndexInfo {
                    name: raw.name,
                    table_name: raw.table_name,
                    unique,
                    columns,
                    partial_predicate,
                })
            })
            .collect::<Result<Vec<_>, ToolError<ListIndexesError>>>()?;

        Ok(ListIndexesOutput { indexes })
    }
}

/// Transient holder for the raw columns returned by the `sqlite_master` query
/// before enrichment with column names and uniqueness.
struct RawIndex {
    /// The index name from `sqlite_master.name`.
    name: String,
    /// The table this index belongs to.
    table_name: String,
    /// The `CREATE INDEX` SQL, or `None` for auto-generated indexes (e.g. from
    /// PRIMARY KEY or UNIQUE constraints).
    sql: Option<String>,
}

/// Queries `PRAGMA index_info` for the given index and returns the ordered list
/// of column names.
fn query_index_columns(
    conn: &rusqlite::Connection,
    index_name: &str,
) -> Result<Vec<String>, ToolError<ListIndexesError>> {
    let pragma = format!("PRAGMA index_info('{index_name}')");
    let mut stmt = conn.prepare(&pragma).map_err(|source| {
        ToolError::Tool(ListIndexesError::Query { source })
    })?;

    stmt.query_map([], |row| row.get::<_, String>(2))
        .map_err(|source| ToolError::Tool(ListIndexesError::Query { source }))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|source| ToolError::Tool(ListIndexesError::Query { source }))
}

/// Determines whether an index enforces uniqueness. Indexes created with
/// `CREATE UNIQUE INDEX` will contain "UNIQUE" in their DDL. Auto-indexes
/// generated by PRIMARY KEY or UNIQUE constraints have no DDL but use the
/// `sqlite_autoindex_` name prefix, so they are treated as unique.
fn is_unique_index(raw: &RawIndex) -> bool {
    match &raw.sql {
        Some(sql) => sql.to_uppercase().contains("UNIQUE"),
        None => raw.name.starts_with("sqlite_autoindex_"),
    }
}

/// Extracts the WHERE-clause predicate from a `CREATE INDEX` SQL statement,
/// returning the text after the last top-level `WHERE` keyword. Returns `None`
/// if the statement has no WHERE clause.
fn extract_where_clause(sql: &str) -> Option<String> {
    let upper = sql.to_uppercase();
    let where_pos = upper.rfind(" WHERE ")?;
    let predicate = sql[where_pos + " WHERE ".len()..].trim();
    if predicate.is_empty() {
        return None;
    }
    Some(predicate.to_owned())
}

/// The input parameters for the `list_indexes` tool.
#[derive(
    Clone,
    Debug,
    Default,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
pub struct ListIndexesInput {
    /// If provided, only indexes on this table are returned. Omit to list
    /// indexes across all tables.
    #[schemars(
        description = "Filter indexes to this table. Omit for all tables."
    )]
    pub table_name: Option<String>,
}

/// The result of listing indexes in the database.
#[derive(
    Clone,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
pub struct ListIndexesOutput {
    /// The indexes found, each with its metadata and column list.
    pub indexes: Vec<IndexInfo>,
}

/// Metadata for a single index in the database.
#[derive(
    Clone,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Serialize,
    Deserialize,
    schemars::JsonSchema,
)]
pub struct IndexInfo {
    /// The index name as recorded in `sqlite_master`.
    pub name: String,
    /// The table this index belongs to.
    pub table_name: String,
    /// Whether this index enforces a uniqueness constraint.
    pub unique: bool,
    /// The ordered list of column names covered by this index.
    pub columns: Vec<String>,
    /// The WHERE-clause predicate for partial indexes, or `None` for full
    /// indexes.
    pub partial_predicate: Option<String>,
}

/// Errors specific to the `list_indexes` tool.
#[derive(Debug, thiserror::Error)]
pub enum ListIndexesError {
    /// Failed to query index metadata from `sqlite_master` or `PRAGMA
    /// index_info`.
    #[error("failed to list indexes: {source}")]
    Query {
        /// The underlying rusqlite error.
        source: rusqlite::Error,
    },
}

/// Converts the list-indexes-specific error into MCP content by rendering the
/// display string as text.
impl IntoContents for ListIndexesError {
    fn into_contents(self) -> Vec<Content> {
        vec![Content::text(self.to_string())]
    }
}