fraiseql_db/postgres/

introspector.rs

/// ! PostgreSQL database introspection for fact tables.
use deadpool_postgres::Pool;
use fraiseql_error::{FraiseQLError, Result};
use tokio_postgres::Row;

use crate::{
    DatabaseType,
    introspector::{DatabaseIntrospector, RelationInfo, RelationKind},
};

/// PostgreSQL introspector for fact table metadata.
pub struct PostgresIntrospector {
    pool: Pool,
}

impl PostgresIntrospector {
    /// Create new PostgreSQL introspector from connection pool.
    #[must_use]
    pub const fn new(pool: Pool) -> Self {
        Self { pool }
    }
}

impl DatabaseIntrospector for PostgresIntrospector {
    async fn list_fact_tables(&self) -> Result<Vec<String>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Query information_schema for tables matching tf_* pattern across all schemas
        // in the current search path.
        let query = r"
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = ANY(current_schemas(false))
              AND table_type = 'BASE TABLE'
              AND table_name LIKE 'tf_%'
            ORDER BY table_name
        ";

        let rows: Vec<Row> =
            client.query(query, &[]).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to list fact tables: {e}"),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        let tables = rows
            .into_iter()
            .map(|row| {
                let name: String = row.get(0);
                name
            })
            .collect();

        Ok(tables)
    }

    async fn get_columns(&self, table_name: &str) -> Result<Vec<(String, String, bool)>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Support schema-qualified names like "benchmark.tv_post".
        // When no schema prefix is present, search across all schemas in the current
        // search path so that non-public schemas work without explicit qualification.
        let rows: Vec<Row> = if let Some(dot) = table_name.find('.') {
            let (schema, name) = (&table_name[..dot], &table_name[dot + 1..]);
            client
                .query(
                    r"SELECT column_name, data_type, is_nullable = 'YES' as is_nullable
                      FROM information_schema.columns
                      WHERE table_name = $1 AND table_schema = $2
                      ORDER BY ordinal_position",
                    &[&name, &schema],
                )
                .await
        } else {
            client
                .query(
                    r"SELECT column_name, data_type, is_nullable = 'YES' as is_nullable
                      FROM information_schema.columns
                      WHERE table_name = $1
                        AND table_schema = ANY(current_schemas(false))
                      ORDER BY ordinal_position",
                    &[&table_name],
                )
                .await
        }
        .map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to query column information: {e}"),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        let columns = rows
            .into_iter()
            .map(|row| {
                let name: String = row.get(0);
                let data_type: String = row.get(1);
                let is_nullable: bool = row.get(2);
                (name, data_type, is_nullable)
            })
            .collect();

        Ok(columns)
    }

    async fn get_indexed_columns(&self, table_name: &str) -> Result<Vec<String>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Support schema-qualified names like "benchmark.tv_post".
        // When no schema prefix is present, search across all schemas in the current
        // search path so that non-public schemas work without explicit qualification.
        let rows: Vec<Row> = if let Some(dot) = table_name.find('.') {
            let (schema, name) = (&table_name[..dot], &table_name[dot + 1..]);
            client
                .query(
                    r"SELECT DISTINCT a.attname AS column_name
                      FROM pg_index i
                      JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
                      JOIN pg_class t ON t.oid = i.indrelid
                      JOIN pg_namespace n ON n.oid = t.relnamespace
                      WHERE t.relname = $1 AND n.nspname = $2 AND a.attnum > 0
                      ORDER BY column_name",
                    &[&name, &schema],
                )
                .await
        } else {
            client
                .query(
                    r"SELECT DISTINCT a.attname AS column_name
                      FROM pg_index i
                      JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
                      JOIN pg_class t ON t.oid = i.indrelid
                      JOIN pg_namespace n ON n.oid = t.relnamespace
                      WHERE t.relname = $1
                        AND n.nspname = ANY(current_schemas(false))
                        AND a.attnum > 0
                      ORDER BY column_name",
                    &[&table_name],
                )
                .await
        }
        .map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to query index information: {e}"),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        let indexed_columns = rows
            .into_iter()
            .map(|row| {
                let name: String = row.get(0);
                name
            })
            .collect();

        Ok(indexed_columns)
    }

    fn database_type(&self) -> DatabaseType {
        DatabaseType::PostgreSQL
    }

    async fn list_relations(&self) -> Result<Vec<RelationInfo>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // List all tables and views visible in the current search path, excluding
        // system schemas. Uses current_schemas(false) to match the same scope as
        // get_columns / get_indexed_columns so that relation existence checks are
        // consistent with column lookups.
        let rows: Vec<Row> = client
            .query(
                r"SELECT table_schema, table_name,
                         CASE table_type WHEN 'VIEW' THEN 'view' ELSE 'table' END AS kind
                  FROM information_schema.tables
                  WHERE table_schema = ANY(current_schemas(false))
                  ORDER BY table_schema, table_name",
                &[],
            )
            .await
            .map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to list relations: {e}"),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        let relations = rows
            .into_iter()
            .map(|row| {
                let schema: String = row.get(0);
                let name: String = row.get(1);
                let kind_str: String = row.get(2);
                let kind = if kind_str == "view" {
                    RelationKind::View
                } else {
                    RelationKind::Table
                };
                RelationInfo { schema, name, kind }
            })
            .collect();

        Ok(relations)
    }

    async fn get_sample_jsonb(
        &self,
        table_name: &str,
        column_name: &str,
    ) -> Result<Option<serde_json::Value>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Query for a sample row with non-null JSON data
        // Use format! for identifiers (safe because we validate table_name pattern)
        let query = format!(
            r#"
            SELECT "{column}"::text
            FROM "{table}"
            WHERE "{column}" IS NOT NULL
            LIMIT 1
            "#,
            table = table_name,
            column = column_name
        );

        let rows: Vec<Row> =
            client.query(&query, &[]).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to query sample JSONB: {e}"),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        if rows.is_empty() {
            return Ok(None);
        }

        let json_text: Option<String> = rows[0].get(0);
        if let Some(text) = json_text {
            let value: serde_json::Value =
                serde_json::from_str(&text).map_err(|e| FraiseQLError::Parse {
                    message:  format!("Failed to parse JSONB sample: {e}"),
                    location: format!("{table_name}.{column_name}"),
                })?;
            Ok(Some(value))
        } else {
            Ok(None)
        }
    }

    async fn function_exists(&self, schema: Option<&str>, name: &str) -> Result<Option<bool>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Mirror `pg_catalog::resolve_functions` name/schema matching (verbatim,
        // schema-qualified or `current_schemas(false)`-scoped). Restrict to kinds a
        // mutation can actually call — regular function `f` or procedure `p` —
        // excluding aggregate `a` / window `w`, which `resolve_functions` does not
        // filter (a minor over-match there).
        let row = if let Some(schema) = schema {
            client
                .query_one(
                    "SELECT EXISTS(SELECT 1 FROM pg_proc p \
                       JOIN pg_namespace n ON n.oid = p.pronamespace \
                       WHERE n.nspname = $1 AND p.proname = $2 AND p.prokind IN ('f','p'))",
                    &[&schema, &name],
                )
                .await
        } else {
            client
                .query_one(
                    "SELECT EXISTS(SELECT 1 FROM pg_proc p \
                       JOIN pg_namespace n ON n.oid = p.pronamespace \
                       WHERE p.proname = $1 AND n.nspname = ANY(current_schemas(false)) \
                       AND p.prokind IN ('f','p'))",
                    &[&name],
                )
                .await
        }
        .map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to probe function existence: {e}"),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        Ok(Some(row.get(0)))
    }

    async fn qualified_relation_exists(&self, schema: &str, name: &str) -> Result<Option<bool>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Resolve exactly as the runtime does (`postgres/adapter/mod.rs` quotes each
        // component case-sensitively), so a mixed-case relation in an off-search_path
        // schema the runtime *can* serve is not falsely reported missing. `to_regclass`
        // takes the quoted identifier as a string and ignores `search_path` for a
        // schema-qualified name, returning NULL when the relation is genuinely absent.
        let quoted = format!(
            "{}.{}",
            crate::identifier::quote_postgres_identifier(schema),
            crate::identifier::quote_postgres_identifier(name),
        );
        let row = client
            .query_one("SELECT to_regclass($1) IS NOT NULL", &[&quoted])
            .await
            .map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to probe relation existence: {e}"),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        Ok(Some(row.get(0)))
    }
}

impl PostgresIntrospector {
    /// Get indexed columns for a view/table that match the nested path naming convention.
    ///
    /// This method introspects the database to find columns that follow the FraiseQL
    /// indexed column naming conventions:
    /// - Human-readable: `items__product__category__code` (double underscore separated)
    /// - Entity ID format: `f{entity_id}__{field_name}` (e.g., `f200100__code`)
    ///
    /// These columns are created by DBAs to optimize filtering on nested GraphQL paths
    /// by avoiding JSONB extraction at runtime.
    ///
    /// # Arguments
    ///
    /// * `view_name` - Name of the view or table to introspect
    ///
    /// # Returns
    ///
    /// Set of column names that match the indexed column naming conventions.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::ConnectionPool` if a connection cannot be acquired,
    /// or `FraiseQLError::Database` if the query fails.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use fraiseql_db::postgres::PostgresIntrospector;
    /// # use fraiseql_error::Result;
    /// use deadpool_postgres::Pool;
    ///
    /// # async fn example(pool: Pool) -> Result<()> {
    /// let introspector = PostgresIntrospector::new(pool);
    /// let indexed_cols = introspector.get_indexed_nested_columns("v_order_items").await?;
    /// // Returns: {"items__product__category__code", "f200100__code", ...}
    /// # Ok(())
    /// # }
    /// ```
    pub async fn get_indexed_nested_columns(
        &self,
        view_name: &str,
    ) -> Result<std::collections::HashSet<String>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        // Query information_schema for columns matching __ pattern.
        // This works for both views and tables across all schemas in the search path.
        let query = r"
            SELECT column_name
            FROM information_schema.columns
            WHERE table_name = $1
              AND table_schema = ANY(current_schemas(false))
              AND column_name LIKE '%__%'
            ORDER BY column_name
        ";

        let rows: Vec<Row> =
            client.query(query, &[&view_name]).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to query view columns: {e}"),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        let indexed_columns: std::collections::HashSet<String> = rows
            .into_iter()
            .map(|row| {
                let name: String = row.get(0);
                name
            })
            .filter(|name| {
                // Filter to only columns that match our naming conventions:
                // 1. Human-readable: path__to__field (at least one __ separator)
                // 2. Entity ID: f{digits}__field_name
                Self::is_indexed_column_name(name)
            })
            .collect();

        Ok(indexed_columns)
    }

    /// Check if a column name matches the indexed column naming convention.
    ///
    /// Valid patterns:
    /// - `items__product__category__code` (human-readable nested path)
    /// - `f200100__code` (entity ID format)
    fn is_indexed_column_name(name: &str) -> bool {
        // Must contain at least one double underscore
        if !name.contains("__") {
            return false;
        }

        // Check for entity ID format: f{digits}__field
        if let Some(rest) = name.strip_prefix('f') {
            if let Some(underscore_pos) = rest.find("__") {
                let digits = &rest[..underscore_pos];
                if digits.chars().all(|c| c.is_ascii_digit()) && !digits.is_empty() {
                    // Verify the field part is valid
                    let field_part = &rest[underscore_pos + 2..];
                    if !field_part.is_empty()
                        && field_part.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
                        && !field_part.starts_with(|c: char| c.is_ascii_digit())
                    {
                        return true;
                    }
                }
            }
        }

        // Human-readable format: split by __ and check each segment is valid identifier
        // Must have at least 2 segments, and first segment must NOT be just 'f'
        let segments: Vec<&str> = name.split("__").collect();
        if segments.len() < 2 {
            return false;
        }

        // Reject if first segment is just 'f' (reserved for entity ID format)
        if segments[0] == "f" {
            return false;
        }

        // Each segment should be a valid identifier (alphanumeric + underscore, not starting with
        // digit)
        segments.iter().all(|s| {
            !s.is_empty()
                && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
                && !s.starts_with(|c: char| c.is_ascii_digit())
        })
    }

    /// Get all column names for a view/table.
    ///
    /// # Arguments
    ///
    /// * `view_name` - Name of the view or table to introspect
    ///
    /// # Returns
    ///
    /// List of all column names in the view/table.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::ConnectionPool` if a connection cannot be acquired,
    /// or `FraiseQLError::Database` if the query fails.
    pub async fn get_view_columns(&self, view_name: &str) -> Result<Vec<String>> {
        let client = self.pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
            message: format!("Failed to acquire connection: {e}"),
        })?;

        let query = r"
            SELECT column_name
            FROM information_schema.columns
            WHERE table_name = $1
              AND table_schema = ANY(current_schemas(false))
            ORDER BY ordinal_position
        ";

        let rows: Vec<Row> =
            client.query(query, &[&view_name]).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to query view columns: {e}"),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        let columns = rows
            .into_iter()
            .map(|row| {
                let name: String = row.get(0);
                name
            })
            .collect();

        Ok(columns)
    }
}

#[cfg(test)]
mod tests;