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)
        }
    }
}

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)
    }
}

/// Unit tests that don't require a PostgreSQL connection.
#[cfg(test)]
mod unit_tests {
    use super::*;

    #[test]
    fn test_is_indexed_column_name_human_readable() {
        // Valid human-readable patterns
        assert!(PostgresIntrospector::is_indexed_column_name("items__product"));
        assert!(PostgresIntrospector::is_indexed_column_name("items__product__category"));
        assert!(PostgresIntrospector::is_indexed_column_name("items__product__category__code"));
        assert!(PostgresIntrospector::is_indexed_column_name("order_items__product_name"));

        // Invalid patterns
        assert!(!PostgresIntrospector::is_indexed_column_name("items"));
        assert!(!PostgresIntrospector::is_indexed_column_name("items_product")); // single underscore
        assert!(!PostgresIntrospector::is_indexed_column_name("__items")); // empty first segment
        assert!(!PostgresIntrospector::is_indexed_column_name("items__")); // empty last segment
    }

    #[test]
    fn test_is_indexed_column_name_entity_id() {
        // Valid entity ID patterns
        assert!(PostgresIntrospector::is_indexed_column_name("f200100__code"));
        assert!(PostgresIntrospector::is_indexed_column_name("f1__name"));
        assert!(PostgresIntrospector::is_indexed_column_name("f123456789__field"));

        // Invalid entity ID patterns (that also aren't valid human-readable)
        assert!(!PostgresIntrospector::is_indexed_column_name("f__code")); // no digits after 'f', and 'f' alone is reserved

        // Note: fx123__code IS valid as a human-readable pattern (fx123 is a valid identifier)
        assert!(PostgresIntrospector::is_indexed_column_name("fx123__code")); // valid as human-readable
    }
}

/// Integration tests that require a PostgreSQL connection.
#[cfg(all(test, feature = "test-postgres"))]
mod integration_tests {
    use deadpool_postgres::{Config, ManagerConfig, RecyclingMethod, Runtime};
    use tokio_postgres::NoTls;

    use super::*;
    use crate::postgres::PostgresAdapter;

    const TEST_DB_URL: &str =
        "postgresql://fraiseql_test:fraiseql_test_password@localhost:5433/test_fraiseql";

    // Helper to create test introspector
    async fn create_test_introspector() -> PostgresIntrospector {
        let _adapter =
            PostgresAdapter::new(TEST_DB_URL).await.expect("Failed to create test adapter");

        // Extract pool from adapter (we need a way to get the pool)
        // For now, create a new pool directly

        let mut cfg = Config::new();
        cfg.url = Some(TEST_DB_URL.to_string());
        cfg.manager = Some(ManagerConfig {
            recycling_method: RecyclingMethod::Fast,
        });
        cfg.pool = Some(deadpool_postgres::PoolConfig::new(10));

        let pool = cfg.create_pool(Some(Runtime::Tokio1), NoTls).expect("Failed to create pool");

        PostgresIntrospector::new(pool)
    }

    #[tokio::test]
    async fn test_get_columns_tf_sales() {
        let introspector = create_test_introspector().await;

        let columns = introspector.get_columns("tf_sales").await.expect("Failed to get columns");

        // Should have: id, revenue, quantity, cost, discount, data, customer_id, product_id,
        // occurred_at, created_at
        assert!(columns.len() >= 10, "Expected at least 10 columns, got {}", columns.len());

        // Check for key columns
        let column_names: Vec<String> = columns.iter().map(|(name, _, _)| name.clone()).collect();
        assert!(column_names.contains(&"revenue".to_string()));
        assert!(column_names.contains(&"quantity".to_string()));
        assert!(column_names.contains(&"data".to_string()));
        assert!(column_names.contains(&"customer_id".to_string()));
    }

    #[tokio::test]
    async fn test_get_indexed_columns_tf_sales() {
        let introspector = create_test_introspector().await;

        let indexed = introspector
            .get_indexed_columns("tf_sales")
            .await
            .expect("Failed to get indexed columns");

        // Should have indexes on: id (PK), customer_id, product_id, occurred_at, data (GIN)
        assert!(indexed.len() >= 4, "Expected at least 4 indexed columns, got {}", indexed.len());

        assert!(indexed.contains(&"customer_id".to_string()));
        assert!(indexed.contains(&"product_id".to_string()));
        assert!(indexed.contains(&"occurred_at".to_string()));
    }

    #[tokio::test]
    async fn test_database_type() {
        let introspector = create_test_introspector().await;
        assert_eq!(introspector.database_type(), DatabaseType::PostgreSQL);
    }
}