senax_pgsql_parser/

lib.rs

//! PostgreSQL Database Schema Parser
//!
//! This crate provides functionality to read PostgreSQL database schema information,
//! provide it as structured data, and generate DDL statements.
//!
//! # Main Features
//!
//! - Read table structures from PostgreSQL databases
//! - PostGIS extension support
//! - DDL generation (CREATE TABLE, DROP TABLE)
//! - Management of foreign key constraints, indexes, and CHECK constraints
//! - Accurate handling of Serial types
//! - Roundtrip testing functionality
//!
//! # Usage Example
//!
//! ```rust,no_run
//! use senax_pgsql_parser::{connect_to_database, get_database_schema};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     let pool = connect_to_database("postgresql://user:pass@localhost/db").await?;
//!     let schema = get_database_schema(&pool).await?;
//!     
//!     println!("Number of tables: {}", schema.tables.len());
//!     for table in &schema.tables {
//!         println!("Table: {}", table.table_name);
//!         println!("DDL: {}", table.to_ddl());
//!     }
//!     
//!     Ok(())
//! }
//! ```

use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Row};
use std::collections::HashMap;
use std::fmt;

/// Parse PostgreSQL array string and convert to Vec<String>
fn parse_postgres_array(array_str: &str) -> Vec<String> {
    if array_str.is_empty() || array_str == "{}" {
        return vec![];
    }

    // Parse string in "{item1,item2,item3}" format
    let trimmed = array_str.trim_start_matches('{').trim_end_matches('}');
    if trimmed.is_empty() {
        return vec![];
    }

    trimmed
        .split(',')
        .map(|s| s.trim().trim_matches('"').to_string())
        .collect()
}

/// Parsed default value types
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum DefaultValue {
    /// String literal value (unescaped)
    String(String),
    /// Integer value
    Integer(i64),
    /// Floating point value
    Float(f64),
    /// Boolean value
    Boolean(bool),
    /// NULL value
    Null,
    /// CURRENT_TIMESTAMP
    CurrentTimestamp,
    /// CURRENT_DATE
    CurrentDate,
    /// CURRENT_TIME
    CurrentTime,
    /// Function call or expression (e.g., nextval(...))
    Expression(String),
    /// Binary/bytea value
    Binary(String),
}

impl fmt::Display for DefaultValue {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            DefaultValue::String(s) => write!(f, "String: \"{}\"", s),
            DefaultValue::Integer(i) => write!(f, "Integer: {}", i),
            DefaultValue::Float(fl) => write!(f, "Float: {}", fl),
            DefaultValue::Boolean(b) => write!(f, "Boolean: {}", b),
            DefaultValue::Null => write!(f, "Null"),
            DefaultValue::CurrentTimestamp => write!(f, "CURRENT_TIMESTAMP"),
            DefaultValue::CurrentDate => write!(f, "CURRENT_DATE"),
            DefaultValue::CurrentTime => write!(f, "CURRENT_TIME"),
            DefaultValue::Expression(e) => write!(f, "Expression: {}", e),
            DefaultValue::Binary(b) => write!(f, "Binary: {}", b),
        }
    }
}

impl DefaultValue {
    /// Convert to DDL-compatible string representation
    pub fn to_ddl_string(&self) -> String {
        match self {
            DefaultValue::String(s) => {
                // Escape single quotes for SQL
                let escaped = s.replace('\'', "''");
                format!("'{}'", escaped)
            }
            DefaultValue::Integer(i) => i.to_string(),
            DefaultValue::Float(f) => f.to_string(),
            DefaultValue::Boolean(b) => b.to_string(),
            DefaultValue::Null => "NULL".to_string(),
            DefaultValue::CurrentTimestamp => "CURRENT_TIMESTAMP".to_string(),
            DefaultValue::CurrentDate => "CURRENT_DATE".to_string(),
            DefaultValue::CurrentTime => "CURRENT_TIME".to_string(),
            DefaultValue::Expression(e) => e.clone(),
            DefaultValue::Binary(b) => format!("'{}'::bytea", b),
        }
    }
}

/// Parse PostgreSQL default value expression
///
/// This function parses PostgreSQL default value strings and returns a structured representation.
///
/// # Examples
///
/// ```
/// use senax_pgsql_parser::{parse_default_value, DefaultValue};
///
/// // String with escaped quotes
/// let result = parse_default_value("'it''s a test'::character varying");
/// assert_eq!(result, DefaultValue::String("it's a test".to_string()));
///
/// // Simple string
/// let result = parse_default_value("'simple'::text");
/// assert_eq!(result, DefaultValue::String("simple".to_string()));
///
/// // Integer
/// let result = parse_default_value("42");
/// assert_eq!(result, DefaultValue::Integer(42));
///
/// // Float
/// let result = parse_default_value("99.99");
/// assert_eq!(result, DefaultValue::Float(99.99));
///
/// // Boolean
/// let result = parse_default_value("true");
/// assert_eq!(result, DefaultValue::Boolean(true));
///
/// // NULL
/// let result = parse_default_value("NULL");
/// assert_eq!(result, DefaultValue::Null);
///
/// // Current timestamp
/// let result = parse_default_value("CURRENT_TIMESTAMP");
/// assert_eq!(result, DefaultValue::CurrentTimestamp);
///
/// // Current date
/// let result = parse_default_value("CURRENT_DATE");
/// assert_eq!(result, DefaultValue::CurrentDate);
///
/// // Sequence
/// let result = parse_default_value("nextval('test_seq'::regclass)");
/// assert_eq!(result, DefaultValue::Expression("nextval('test_seq'::regclass)".to_string()));
/// ```
pub fn parse_default_value(default_str: &str) -> DefaultValue {
    let trimmed = default_str.trim();

    // Check for NULL
    if trimmed.eq_ignore_ascii_case("NULL") {
        return DefaultValue::Null;
    }

    // Check for boolean values
    if trimmed.eq_ignore_ascii_case("true") {
        return DefaultValue::Boolean(true);
    }
    if trimmed.eq_ignore_ascii_case("false") {
        return DefaultValue::Boolean(false);
    }

    // Check for CURRENT_* functions
    if trimmed.eq_ignore_ascii_case("CURRENT_TIMESTAMP") {
        return DefaultValue::CurrentTimestamp;
    }
    if trimmed.eq_ignore_ascii_case("CURRENT_DATE") {
        return DefaultValue::CurrentDate;
    }
    if trimmed.eq_ignore_ascii_case("CURRENT_TIME") {
        return DefaultValue::CurrentTime;
    }

    // Check for string literals (starts with single quote)
    if trimmed.starts_with('\'') {
        return parse_string_literal(trimmed);
    }

    // Check for numeric values
    if let Ok(int_val) = trimmed.parse::<i64>() {
        return DefaultValue::Integer(int_val);
    }
    if let Ok(float_val) = trimmed.parse::<f64>() {
        return DefaultValue::Float(float_val);
    }

    // Everything else is treated as an expression (functions, etc.)
    DefaultValue::Expression(trimmed.to_string())
}

/// Parse PostgreSQL string literal with type cast
///
/// Handles formats like:
/// - 'simple'::text
/// - 'it''s a test'::character varying
/// - '\x00'::bytea
fn parse_string_literal(input: &str) -> DefaultValue {
    let input = input.trim();

    // Find the closing quote
    let mut chars = input.chars().peekable();
    let mut result = String::new();
    let mut type_cast = String::new();
    let mut after_string = false;

    // Skip the opening quote
    if chars.next() != Some('\'') {
        return DefaultValue::Expression(input.to_string());
    }
    let mut in_string = true;

    while let Some(ch) = chars.next() {
        if in_string {
            if ch == '\'' {
                // Check if it's an escaped quote
                if chars.peek() == Some(&'\'') {
                    // Escaped quote: '' -> '
                    result.push('\'');
                    chars.next(); // consume the second quote
                } else {
                    // End of string
                    in_string = false;
                    after_string = true;
                }
            } else {
                result.push(ch);
            }
        } else if after_string {
            // Collect the rest (type cast)
            type_cast.push(ch);
        }
    }

    // Check if it's a bytea type
    if type_cast.contains("bytea") {
        return DefaultValue::Binary(result);
    }

    // Return as string
    DefaultValue::String(result)
}

/// PostGIS geometry types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum GeometryType {
    Point,
    LineString,
    Polygon,
    MultiPoint,
    MultiLineString,
    MultiPolygon,
    GeometryCollection,
}

impl fmt::Display for GeometryType {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let s = match self {
            GeometryType::Point => "POINT",
            GeometryType::LineString => "LINESTRING",
            GeometryType::Polygon => "POLYGON",
            GeometryType::MultiPoint => "MULTIPOINT",
            GeometryType::MultiLineString => "MULTILINESTRING",
            GeometryType::MultiPolygon => "MULTIPOLYGON",
            GeometryType::GeometryCollection => "GEOMETRYCOLLECTION",
        };
        write!(f, "{}", s)
    }
}

/// PostGIS geometry detailed information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct GeometryInfo {
    pub geometry_type: GeometryType,
    pub srid: Option<i32>,
    pub dimension: Option<i32>,
}

impl fmt::Display for GeometryInfo {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match (&self.srid, &self.dimension) {
            (Some(srid), Some(_dim)) => write!(f, "({}, {})", self.geometry_type, srid),
            (Some(srid), None) => write!(f, "({}, {})", self.geometry_type, srid),
            (None, Some(_dim)) => write!(f, "({})", self.geometry_type),
            (None, None) => write!(f, "({})", self.geometry_type),
        }
    }
}

/// PostgreSQL data types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PostgreSQLDataType {
    // Integer types
    SmallInt,
    Integer,
    BigInt,

    // Serial types
    SmallSerial,
    Serial,
    BigSerial,

    // Floating point types
    Real,
    DoublePrecision,

    // Numeric types
    Numeric(Option<i32>, Option<i32>), // precision, scale
    Decimal(Option<i32>, Option<i32>), // precision, scale

    // String types
    Char(Option<i32>),
    Varchar(Option<i32>),
    Text,

    // Binary type
    Bytea,

    // Date/time types
    Date,
    Time,
    TimeWithTimeZone,
    Timestamp,
    TimestampWithTimeZone,
    Interval,

    // Boolean type
    Boolean,

    // UUID type
    Uuid,

    // JSON types
    Json,
    Jsonb,

    // Array type
    Array(Box<PostgreSQLDataType>),

    // PostGIS types
    Geometry(Option<GeometryInfo>),
    Geography(Option<GeometryInfo>),

    // Custom types (ENUM, etc.)
    Custom(String),

    // Others
    Unknown(String),
}

impl fmt::Display for PostgreSQLDataType {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PostgreSQLDataType::SmallInt => write!(f, "smallint"),
            PostgreSQLDataType::Integer => write!(f, "integer"),
            PostgreSQLDataType::BigInt => write!(f, "bigint"),
            PostgreSQLDataType::SmallSerial => write!(f, "smallserial"),
            PostgreSQLDataType::Serial => write!(f, "serial"),
            PostgreSQLDataType::BigSerial => write!(f, "bigserial"),
            PostgreSQLDataType::Real => write!(f, "real"),
            PostgreSQLDataType::DoublePrecision => write!(f, "double precision"),
            PostgreSQLDataType::Numeric(p, s) => match (p, s) {
                (Some(precision), Some(scale)) => write!(f, "numeric({},{})", precision, scale),
                (Some(precision), None) => write!(f, "numeric({})", precision),
                _ => write!(f, "numeric"),
            },
            PostgreSQLDataType::Decimal(p, s) => match (p, s) {
                (Some(precision), Some(scale)) => write!(f, "numeric({},{})", precision, scale),
                (Some(precision), None) => write!(f, "numeric({})", precision),
                _ => write!(f, "numeric"),
            },
            PostgreSQLDataType::Char(len) => match len {
                Some(l) => write!(f, "char({})", l),
                None => write!(f, "char"),
            },
            PostgreSQLDataType::Varchar(len) => match len {
                Some(l) => write!(f, "varchar({})", l),
                None => write!(f, "varchar"),
            },
            PostgreSQLDataType::Text => write!(f, "text"),
            PostgreSQLDataType::Bytea => write!(f, "bytea"),
            PostgreSQLDataType::Date => write!(f, "date"),
            PostgreSQLDataType::Time => write!(f, "time"),
            PostgreSQLDataType::TimeWithTimeZone => write!(f, "time with time zone"),
            PostgreSQLDataType::Timestamp => write!(f, "timestamp"),
            PostgreSQLDataType::TimestampWithTimeZone => write!(f, "timestamp with time zone"),
            PostgreSQLDataType::Interval => write!(f, "interval"),
            PostgreSQLDataType::Boolean => write!(f, "boolean"),
            PostgreSQLDataType::Uuid => write!(f, "uuid"),
            PostgreSQLDataType::Json => write!(f, "json"),
            PostgreSQLDataType::Jsonb => write!(f, "jsonb"),
            PostgreSQLDataType::Array(inner) => write!(f, "{}[]", inner),
            PostgreSQLDataType::Geometry(info) => match info {
                Some(geo_info) => write!(f, "geometry{}", geo_info),
                None => write!(f, "geometry"),
            },
            PostgreSQLDataType::Geography(info) => match info {
                Some(geo_info) => write!(f, "geography{}", geo_info),
                None => write!(f, "geography"),
            },
            PostgreSQLDataType::Custom(type_name) => write!(f, "{}", type_name),
            PostgreSQLDataType::Unknown(type_name) => write!(f, "{}", type_name),
        }
    }
}

impl PostgreSQLDataType {
    /// Get string representation for DDL
    pub fn to_ddl_string(&self) -> String {
        match self {
            PostgreSQLDataType::Array(inner) => match inner.as_ref() {
                PostgreSQLDataType::Geometry(Some(info)) => {
                    format!("geometry{}[]", info)
                }
                PostgreSQLDataType::Geography(Some(info)) => {
                    format!("geography{}[]", info)
                }
                _ => format!("{}[]", inner.to_ddl_string()),
            },
            _ => self.to_string(),
        }
    }
}

/// Column information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ColumnInfo {
    pub column_name: String,
    pub data_type: PostgreSQLDataType,
    pub is_nullable: bool,
    pub column_default: Option<DefaultValue>,
    pub character_maximum_length: Option<i32>,
    pub numeric_precision: Option<i32>,
    pub numeric_scale: Option<i32>,
    pub is_primary_key: bool,
    pub comment: Option<String>,
    pub collate: Option<String>,
}

impl ColumnInfo {
    /// Generate DDL definition for column
    pub fn to_ddl(&self) -> String {
        let mut ddl = format!(
            "{} {}",
            escape_db_identifier(&self.column_name),
            self.data_type.to_ddl_string()
        );

        // Add COLLATE clause for string types
        if let Some(collate) = &self.collate {
            match &self.data_type {
                PostgreSQLDataType::Char(_)
                | PostgreSQLDataType::Varchar(_)
                | PostgreSQLDataType::Text => {
                    ddl.push_str(&format!(" COLLATE \"{}\"", collate));
                }
                _ => {}
            }
        }

        if !self.is_nullable {
            ddl.push_str(" NOT NULL");
        }

        // Omit DEFAULT clause for Serial types (auto-generated)
        if let Some(default) = &self.column_default {
            match &self.data_type {
                PostgreSQLDataType::SmallSerial
                | PostgreSQLDataType::Serial
                | PostgreSQLDataType::BigSerial => {
                    // For Serial types, DEFAULT clause is automatically added, so omit it
                }
                _ => {
                    ddl.push_str(&format!(" DEFAULT {}", default.to_ddl_string()));
                }
            }
        }

        ddl
    }
}

/// Foreign key constraint actions
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum ForeignKeyAction {
    NoAction,
    Restrict,
    Cascade,
    SetNull,
    SetDefault,
}

impl ForeignKeyAction {
    /// Convert string to ForeignKeyAction
    pub fn from_str(s: &str) -> Option<Self> {
        match s.to_uppercase().as_str() {
            "NO ACTION" => Some(ForeignKeyAction::NoAction),
            "RESTRICT" => Some(ForeignKeyAction::Restrict),
            "CASCADE" => Some(ForeignKeyAction::Cascade),
            "SET NULL" => Some(ForeignKeyAction::SetNull),
            "SET DEFAULT" => Some(ForeignKeyAction::SetDefault),
            _ => None,
        }
    }

    /// Convert ForeignKeyAction to string
    pub fn to_string(&self) -> String {
        match self {
            ForeignKeyAction::NoAction => "NO ACTION".to_string(),
            ForeignKeyAction::Restrict => "RESTRICT".to_string(),
            ForeignKeyAction::Cascade => "CASCADE".to_string(),
            ForeignKeyAction::SetNull => "SET NULL".to_string(),
            ForeignKeyAction::SetDefault => "SET DEFAULT".to_string(),
        }
    }
}

impl fmt::Display for ForeignKeyAction {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.to_string())
    }
}

/// Index types
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum IndexType {
    BTree,
    Hash,
    Gin,
    Gist,
    SpGist,
    Brin,
    Bloom,
}

impl IndexType {
    /// Convert string to IndexType
    pub fn from_str(s: &str) -> Option<Self> {
        match s.to_lowercase().as_str() {
            "btree" => Some(IndexType::BTree),
            "hash" => Some(IndexType::Hash),
            "gin" => Some(IndexType::Gin),
            "gist" => Some(IndexType::Gist),
            "spgist" => Some(IndexType::SpGist),
            "brin" => Some(IndexType::Brin),
            "bloom" => Some(IndexType::Bloom),
            _ => None,
        }
    }

    /// Convert IndexType to string
    pub fn to_string(&self) -> String {
        match self {
            IndexType::BTree => "btree".to_string(),
            IndexType::Hash => "hash".to_string(),
            IndexType::Gin => "gin".to_string(),
            IndexType::Gist => "gist".to_string(),
            IndexType::SpGist => "spgist".to_string(),
            IndexType::Brin => "brin".to_string(),
            IndexType::Bloom => "bloom".to_string(),
        }
    }
}

impl fmt::Display for IndexType {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.to_string())
    }
}

/// Foreign key constraint information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ForeignKeyConstraint {
    pub constraint_name: String,
    pub columns: Vec<String>,
    pub referenced_table: String,
    pub referenced_columns: Vec<String>,
    pub on_delete: Option<ForeignKeyAction>,
    pub on_update: Option<ForeignKeyAction>,
    pub is_deferrable: bool,
    pub initially_deferred: bool,
}

impl ForeignKeyConstraint {
    /// Generate DDL for foreign key constraint
    pub fn to_ddl(&self) -> String {
        let delete_action = match &self.on_delete {
            Some(action) => format!(" ON DELETE {}", action),
            None => String::new(),
        };
        let update_action = match &self.on_update {
            Some(action) => format!(" ON UPDATE {}", action),
            None => String::new(),
        };
        let deferrable = if self.is_deferrable {
            if self.initially_deferred {
                " DEFERRABLE INITIALLY DEFERRED"
            } else {
                " DEFERRABLE INITIALLY IMMEDIATE"
            }
        } else {
            ""
        };

        format!(
            "CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {} ({}){}{}{}",
            escape_db_identifier(&self.constraint_name),
            self.columns
                .iter()
                .map(|v| escape_db_identifier(v))
                .collect::<Vec<_>>()
                .join(", "),
            escape_db_identifier(&self.referenced_table),
            self.referenced_columns
                .iter()
                .map(|v| escape_db_identifier(v))
                .collect::<Vec<_>>()
                .join(", "),
            delete_action,
            update_action,
            deferrable
        )
    }
}

/// Index information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct IndexInfo {
    pub index_name: String,
    pub columns: Vec<String>,
    pub is_unique: bool,
    pub is_primary: bool,
    pub index_type: IndexType,
    pub condition: Option<String>, // partial index condition
}

impl IndexInfo {
    /// Generate DDL for index creation
    pub fn to_ddl(&self, table_schema: &str, table_name: &str) -> String {
        let unique_clause = if self.is_unique { "UNIQUE " } else { "" };
        let using_clause = if self.index_type != IndexType::BTree {
            format!(" USING {}", self.index_type.to_string().to_uppercase())
        } else {
            String::new()
        };
        let where_clause = match &self.condition {
            Some(condition) => format!(" WHERE {}", condition),
            None => String::new(),
        };

        format!(
            "CREATE {}INDEX {} ON {}.{}{} ({}){}",
            unique_clause,
            escape_db_identifier(&self.index_name),
            escape_db_identifier(&table_schema),
            escape_db_identifier(&table_name),
            using_clause,
            self.columns
                .iter()
                .map(|v| escape_db_identifier(v))
                .collect::<Vec<_>>()
                .join(", "),
            where_clause
        )
    }
}

/// CHECK constraint information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CheckConstraint {
    pub constraint_name: String,
    pub check_clause: String,
}

impl CheckConstraint {
    /// Generate DDL for CHECK constraint
    pub fn to_ddl(&self) -> String {
        format!(
            "CONSTRAINT {} CHECK ({})",
            escape_db_identifier(&self.constraint_name),
            self.check_clause
        )
    }
}

/// Table information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct TableInfo {
    pub table_schema: String,
    pub table_name: String,
    pub columns: Vec<ColumnInfo>,
    pub foreign_keys: Vec<ForeignKeyConstraint>,
    pub indexes: Vec<IndexInfo>,
    pub check_constraints: Vec<CheckConstraint>,
    pub comment: Option<String>,
}

impl TableInfo {
    /// Check if table uses PostGIS extension
    pub fn uses_postgis(&self) -> bool {
        self.columns.iter().any(|col| match &col.data_type {
            PostgreSQLDataType::Geometry(_) | PostgreSQLDataType::Geography(_) => true,
            PostgreSQLDataType::Array(inner) => matches!(
                inner.as_ref(),
                PostgreSQLDataType::Geometry(_) | PostgreSQLDataType::Geography(_)
            ),
            _ => false,
        })
    }

    /// Get details of columns using PostGIS types
    pub fn get_postgis_types(&self) -> HashMap<String, i32> {
        let mut postgis_types = HashMap::new();

        for col in &self.columns {
            match &col.data_type {
                PostgreSQLDataType::Geometry(_) => {
                    *postgis_types.entry("geometry".to_string()).or_insert(0) += 1;
                }
                PostgreSQLDataType::Geography(_) => {
                    *postgis_types.entry("geography".to_string()).or_insert(0) += 1;
                }
                PostgreSQLDataType::Array(inner) => match inner.as_ref() {
                    PostgreSQLDataType::Geometry(_) => {
                        *postgis_types.entry("geometry[]".to_string()).or_insert(0) += 1;
                    }
                    PostgreSQLDataType::Geography(_) => {
                        *postgis_types.entry("geography[]".to_string()).or_insert(0) += 1;
                    }
                    _ => {}
                },
                _ => {}
            }
        }

        postgis_types
    }

    /// Generate CREATE TABLE DDL
    pub fn to_ddl(&self) -> String {
        let mut ddl = format!(
            "CREATE TABLE {}.{} (\n",
            escape_db_identifier(&self.table_schema),
            escape_db_identifier(&self.table_name)
        );

        // Column definitions
        let column_definitions: Vec<String> = self
            .columns
            .iter()
            .map(|col| format!("  {}", col.to_ddl()))
            .collect();

        let mut constraints = Vec::new();

        // PRIMARY KEY constraints
        let primary_key_columns: Vec<&str> = self
            .columns
            .iter()
            .filter(|col| col.is_primary_key)
            .map(|col| col.column_name.as_str())
            .collect();

        if !primary_key_columns.is_empty() {
            constraints.push(format!(
                "  CONSTRAINT pk_{} PRIMARY KEY ({})",
                self.table_name,
                primary_key_columns
                    .iter()
                    .map(|v| escape_db_identifier(v))
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }

        // FOREIGN KEY constraints
        for fk in &self.foreign_keys {
            let delete_action = match &fk.on_delete {
                Some(action) => format!(" ON DELETE {}", action),
                None => String::new(),
            };
            let update_action = match &fk.on_update {
                Some(action) => format!(" ON UPDATE {}", action),
                None => String::new(),
            };
            let deferrable = if fk.is_deferrable {
                if fk.initially_deferred {
                    " DEFERRABLE INITIALLY DEFERRED"
                } else {
                    " DEFERRABLE INITIALLY IMMEDIATE"
                }
            } else {
                ""
            };

            constraints.push(format!(
                "  CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {} ({}){}{}{}",
                escape_db_identifier(&fk.constraint_name),
                fk.columns
                    .iter()
                    .map(|v| escape_db_identifier(v))
                    .collect::<Vec<_>>()
                    .join(", "),
                escape_db_identifier(&fk.referenced_table),
                fk.referenced_columns
                    .iter()
                    .map(|v| escape_db_identifier(v))
                    .collect::<Vec<_>>()
                    .join(", "),
                delete_action,
                update_action,
                deferrable
            ));
        }

        // CHECK constraints
        for check in &self.check_constraints {
            constraints.push(format!(
                "  CONSTRAINT {} CHECK ({})",
                escape_db_identifier(&check.constraint_name),
                check.check_clause
            ));
        }

        // Combine all definitions
        let mut all_definitions = column_definitions;
        all_definitions.extend(constraints);

        ddl.push_str(&all_definitions.join(",\n"));
        ddl.push_str("\n);");

        ddl
    }

    /// Generate DDL for index creation
    pub fn to_index_ddl(&self) -> Vec<String> {
        self.indexes
            .iter()
            .filter(|idx| !idx.is_primary) // Exclude PRIMARY KEY indexes
            .map(|idx| idx.to_ddl(&self.table_schema, &self.table_name))
            .collect()
    }

    /// Generate DDL for table and column comments
    pub fn to_comment_ddl(&self) -> Vec<String> {
        let mut comments = Vec::new();

        // Add table comment if exists
        if let Some(table_comment) = &self.comment {
            comments.push(format!(
                "COMMENT ON TABLE {}.{} IS '{}'",
                escape_db_identifier(&self.table_schema),
                escape_db_identifier(&self.table_name),
                table_comment.replace("'", "''") // Escape single quotes
            ));
        }

        // Add column comments
        for col in &self.columns {
            if let Some(comment) = &col.comment {
                comments.push(format!(
                    "COMMENT ON COLUMN {}.{}.{} IS '{}'",
                    escape_db_identifier(&self.table_schema),
                    escape_db_identifier(&self.table_name),
                    escape_db_identifier(&col.column_name),
                    comment.replace("'", "''") // Escape single quotes
                ));
            }
        }

        comments
    }

    /// Get list of referenced tables
    pub fn get_referenced_tables(&self) -> Vec<String> {
        self.foreign_keys
            .iter()
            .map(|fk| fk.referenced_table.clone())
            .collect()
    }

    /// Check if table has foreign keys
    pub fn has_foreign_keys(&self) -> bool {
        !self.foreign_keys.is_empty()
    }
}

/// Complete database schema information
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct DatabaseSchema {
    pub tables: Vec<TableInfo>,
}

impl DatabaseSchema {
    /// Check if PostGIS extension is required
    pub fn requires_postgis(&self) -> bool {
        self.tables.iter().any(|table| table.uses_postgis())
    }

    /// Get list of PostGIS tables
    pub fn get_postgis_tables(&self) -> Vec<&TableInfo> {
        self.tables
            .iter()
            .filter(|table| table.uses_postgis())
            .collect()
    }

    /// Get usage statistics for all PostGIS types
    pub fn get_all_postgis_types(&self) -> HashMap<String, i32> {
        let mut all_types = HashMap::new();

        for table in &self.tables {
            let table_types = table.get_postgis_types();
            for (type_name, count) in table_types {
                *all_types.entry(type_name).or_insert(0) += count;
            }
        }

        all_types
    }

    /// Generate CREATE DDL (considering dependencies)
    pub fn to_create_ddl(&self) -> String {
        let mut ddl_parts = Vec::new();

        // If PostGIS extension is required
        if self.requires_postgis() {
            ddl_parts.push("CREATE EXTENSION IF NOT EXISTS postgis;".to_string());
            ddl_parts.push("".to_string()); // Empty line
        }

        // Create tables without foreign keys first
        let mut created_tables = std::collections::HashSet::new();
        let mut remaining_tables: Vec<&TableInfo> = self.tables.iter().collect();

        // Create from tables without foreign keys
        while !remaining_tables.is_empty() {
            let mut progress = false;

            remaining_tables.retain(|table| {
                let can_create = !table.has_foreign_keys()
                    || table.get_referenced_tables().iter().all(|ref_table| {
                        created_tables.contains(ref_table) || ref_table == &table.table_name
                    });

                if can_create {
                    ddl_parts.push(table.to_ddl());
                    ddl_parts.push("".to_string()); // Empty line

                    // Add table and column comments
                    let comment_ddls = table.to_comment_ddl();
                    for comment_ddl in comment_ddls {
                        ddl_parts.push(comment_ddl);
                    }
                    if !table.to_comment_ddl().is_empty() {
                        ddl_parts.push("".to_string()); // Empty line after comments
                    }

                    // Add indexes
                    let index_ddls = table.to_index_ddl();
                    for index_ddl in index_ddls {
                        ddl_parts.push(index_ddl);
                    }
                    if !table.to_index_ddl().is_empty() {
                        ddl_parts.push("".to_string()); // Empty line after indexes
                    }

                    created_tables.insert(table.table_name.clone());
                    progress = true;
                    false // Remove this table from remaining_tables
                } else {
                    true // Keep this table in remaining_tables
                }
            });

            // If no progress (circular references, etc.), force create remaining tables
            if !progress && !remaining_tables.is_empty() {
                let table = remaining_tables.remove(0);
                ddl_parts.push(table.to_ddl());
                ddl_parts.push("".to_string());

                // Add table and column comments
                let comment_ddls = table.to_comment_ddl();
                for comment_ddl in comment_ddls {
                    ddl_parts.push(comment_ddl);
                }
                if !table.to_comment_ddl().is_empty() {
                    ddl_parts.push("".to_string()); // Empty line after comments
                }

                // Add indexes
                let index_ddls = table.to_index_ddl();
                for index_ddl in index_ddls {
                    ddl_parts.push(index_ddl);
                }
                if !table.to_index_ddl().is_empty() {
                    ddl_parts.push("".to_string()); // Empty line after indexes
                }

                created_tables.insert(table.table_name.clone());
            }
        }

        ddl_parts.join("\n")
    }

    /// Generate DROP DDL (reverse order of CREATE DDL)
    pub fn to_drop_ddl(&self) -> String {
        let mut ddl_parts = Vec::new();

        // Drop tables with foreign keys first
        let mut dropped_tables = std::collections::HashSet::new();
        let mut remaining_tables: Vec<&TableInfo> = self.tables.iter().collect();

        while !remaining_tables.is_empty() {
            let mut progress = false;

            remaining_tables.retain(|table| {
                let can_drop = self.tables.iter().all(|other_table| {
                    other_table.table_name == table.table_name
                        || dropped_tables.contains(&other_table.table_name)
                        || !other_table
                            .get_referenced_tables()
                            .contains(&table.table_name)
                });

                if can_drop {
                    ddl_parts.push(format!(
                        "DROP TABLE IF EXISTS {}.{} CASCADE;",
                        escape_db_identifier(&table.table_schema),
                        escape_db_identifier(&table.table_name)
                    ));
                    dropped_tables.insert(table.table_name.clone());
                    progress = true;
                    false
                } else {
                    true
                }
            });

            if !progress && !remaining_tables.is_empty() {
                let table = remaining_tables.remove(0);
                ddl_parts.push(format!(
                    "DROP TABLE IF EXISTS {}.{} CASCADE;",
                    escape_db_identifier(&table.table_schema),
                    escape_db_identifier(&table.table_name)
                ));
                dropped_tables.insert(table.table_name.clone());
            }
        }

        ddl_parts.join("\n")
    }
}

/// Connect to database
pub async fn connect_to_database(database_url: &str) -> Result<PgPool, sqlx::Error> {
    PgPool::connect(database_url).await
}

/// Get current schema name
pub async fn get_current_schema(pool: &PgPool) -> Result<String, sqlx::Error> {
    let row = sqlx::query("SELECT current_schema()")
        .fetch_one(pool)
        .await?;
    Ok(row.get(0))
}

/// Get complete database schema
pub async fn get_database_schema(pool: &PgPool) -> Result<DatabaseSchema, sqlx::Error> {
    let current_schema = get_current_schema(pool).await?;
    let table_names = get_table_names(pool, &current_schema).await?;

    let mut tables = Vec::new();
    for table_name in table_names {
        let columns = get_table_columns(pool, &current_schema, &table_name).await?;
        let foreign_keys = get_foreign_key_constraints(pool, &current_schema, &table_name).await?;
        let indexes = get_table_indexes(pool, &current_schema, &table_name).await?;
        let check_constraints = get_check_constraints(pool, &current_schema, &table_name).await?;
        let table_comment = get_table_comment(pool, &current_schema, &table_name).await?;

        tables.push(TableInfo {
            table_schema: current_schema.clone(),
            table_name,
            columns,
            foreign_keys,
            indexes,
            check_constraints,
            comment: table_comment,
        });
    }

    Ok(DatabaseSchema { tables })
}

// Private implementation functions below

pub async fn get_table_names(pool: &PgPool, schema: &str) -> Result<Vec<String>, sqlx::Error> {
    let rows = sqlx::query(
        "SELECT table_name FROM information_schema.tables WHERE table_schema = $1 AND table_type = 'BASE TABLE' ORDER BY table_name"
    )
    .bind(schema)
    .fetch_all(pool)
    .await?;

    Ok(rows.into_iter().map(|row| row.get(0)).collect())
}

pub async fn get_table_columns(
    pool: &PgPool,
    schema: &str,
    table: &str,
) -> Result<Vec<ColumnInfo>, sqlx::Error> {
    let query = r#"
        SELECT 
            c.column_name,
            c.data_type,
            c.is_nullable,
            c.column_default,
            c.character_maximum_length,
            c.numeric_precision,
            c.numeric_scale,
            c.udt_name,
            COALESCE(pk.is_primary, false) as is_primary_key,
            pgd.description as comment,
            c.collation_name
        FROM information_schema.columns c
        LEFT JOIN (
            SELECT 
                kcu.column_name,
                true as is_primary
            FROM information_schema.table_constraints tc
            JOIN information_schema.key_column_usage kcu 
                ON tc.constraint_name = kcu.constraint_name
                AND tc.table_schema = kcu.table_schema
            WHERE tc.constraint_type = 'PRIMARY KEY'
                AND tc.table_schema = $1
                AND tc.table_name = $2
        ) pk ON c.column_name = pk.column_name
        LEFT JOIN pg_class pgc ON pgc.relname = c.table_name AND pgc.relnamespace = (
            SELECT oid FROM pg_namespace WHERE nspname = c.table_schema
        )
        LEFT JOIN pg_attribute pga ON pga.attrelid = pgc.oid AND pga.attname = c.column_name
        LEFT JOIN pg_description pgd ON pgd.objoid = pgc.oid AND pgd.objsubid = pga.attnum
        WHERE c.table_schema = $1 AND c.table_name = $2
        ORDER BY c.ordinal_position
    "#;

    let rows = sqlx::query(query)
        .bind(schema)
        .bind(table)
        .fetch_all(pool)
        .await?;

    let mut columns = Vec::new();
    for row in rows {
        let column_name: String = row.get("column_name");
        let data_type_str: String = row.get("data_type");
        let udt_name: String = row.get("udt_name");
        let is_nullable_str: String = row.get("is_nullable");
        let is_nullable = is_nullable_str == "YES";
        let column_default_str: Option<String> = row.get("column_default");
        let character_maximum_length: Option<i32> = row.get("character_maximum_length");
        let numeric_precision: Option<i32> = row.get("numeric_precision");
        let numeric_scale: Option<i32> = row.get("numeric_scale");
        let is_primary_key: bool = row.get("is_primary_key");
        let comment: Option<String> = row.get("comment");
        let collation_name: Option<String> = row.get("collation_name");

        // Parse default value
        let column_default = column_default_str.as_ref().map(|s| parse_default_value(s));

        let data_type = parse_data_type(
            &data_type_str,
            &udt_name,
            character_maximum_length,
            numeric_precision,
            numeric_scale,
            &column_default_str,
            &column_name,
            table,
            pool,
            schema,
        )
        .await;

        columns.push(ColumnInfo {
            column_name,
            data_type,
            is_nullable,
            column_default,
            character_maximum_length,
            numeric_precision,
            numeric_scale,
            is_primary_key,
            comment,
            collate: collation_name,
        });
    }

    Ok(columns)
}

async fn parse_data_type(
    data_type_str: &str,
    udt_name: &str,
    character_maximum_length: Option<i32>,
    numeric_precision: Option<i32>,
    numeric_scale: Option<i32>,
    column_default: &Option<String>,
    column_name: &str,
    table_name: &str,
    pool: &PgPool,
    schema: &str,
) -> PostgreSQLDataType {
    // Check if this is a Serial type by examining the DEFAULT value
    let is_serial = if let Some(default_value) = column_default {
        default_value.contains("nextval")
            && (default_value.contains(&format!("{}_seq", column_name))
                || default_value.contains(&format!("{}_{}_seq", table_name, column_name)))
    } else {
        false
    };

    match data_type_str {
        "smallint" => {
            if is_serial {
                PostgreSQLDataType::SmallSerial
            } else {
                PostgreSQLDataType::SmallInt
            }
        }
        "integer" => {
            if is_serial {
                PostgreSQLDataType::Serial
            } else {
                PostgreSQLDataType::Integer
            }
        }
        "bigint" => {
            if is_serial {
                PostgreSQLDataType::BigSerial
            } else {
                PostgreSQLDataType::BigInt
            }
        }
        "real" => PostgreSQLDataType::Real,
        "double precision" => PostgreSQLDataType::DoublePrecision,
        "numeric" => PostgreSQLDataType::Numeric(numeric_precision, numeric_scale),
        "character varying" => PostgreSQLDataType::Varchar(character_maximum_length),
        "character" => PostgreSQLDataType::Char(character_maximum_length),
        "text" => PostgreSQLDataType::Text,
        "bytea" => PostgreSQLDataType::Bytea,
        "date" => PostgreSQLDataType::Date,
        "time without time zone" => PostgreSQLDataType::Time,
        "time with time zone" => PostgreSQLDataType::TimeWithTimeZone,
        "timestamp without time zone" => PostgreSQLDataType::Timestamp,
        "timestamp with time zone" => PostgreSQLDataType::TimestampWithTimeZone,
        "interval" => PostgreSQLDataType::Interval,
        "boolean" => PostgreSQLDataType::Boolean,
        "uuid" => PostgreSQLDataType::Uuid,
        "json" => PostgreSQLDataType::Json,
        "jsonb" => PostgreSQLDataType::Jsonb,
        "ARRAY" => {
            let base_type = parse_base_array_type(
                udt_name,
                column_default,
                column_name,
                table_name,
                pool,
                schema,
            )
            .await;
            PostgreSQLDataType::Array(Box::new(base_type))
        }
        "USER-DEFINED" => match udt_name {
            "geometry" => {
                let geo_info =
                    get_geometry_info(pool, schema, table_name, column_name, "geometry_columns")
                        .await;
                PostgreSQLDataType::Geometry(geo_info)
            }
            "geography" => {
                let geo_info =
                    get_geometry_info(pool, schema, table_name, column_name, "geography_columns")
                        .await;
                PostgreSQLDataType::Geography(geo_info)
            }
            _ => {
                if udt_name.starts_with("_") {
                    let base_udt = &udt_name[1..];
                    let base_type = match base_udt {
                        "geometry" => PostgreSQLDataType::Geometry(None),
                        "geography" => PostgreSQLDataType::Geography(None),
                        _ => PostgreSQLDataType::Custom(base_udt.to_string()),
                    };
                    PostgreSQLDataType::Array(Box::new(base_type))
                } else {
                    PostgreSQLDataType::Custom(udt_name.to_string())
                }
            }
        },
        _ => PostgreSQLDataType::Unknown(data_type_str.to_string()),
    }
}

async fn parse_base_array_type(
    udt_name: &str,
    _column_default: &Option<String>,
    _column_name: &str,
    _table_name: &str,
    _pool: &PgPool,
    _schema: &str,
) -> PostgreSQLDataType {
    if udt_name.starts_with("_") {
        let base_type = &udt_name[1..];
        match base_type {
            "int2" => PostgreSQLDataType::SmallInt,
            "int4" => PostgreSQLDataType::Integer,
            "int8" => PostgreSQLDataType::BigInt,
            "float4" => PostgreSQLDataType::Real,
            "float8" => PostgreSQLDataType::DoublePrecision,
            "numeric" => PostgreSQLDataType::Numeric(None, None),
            "varchar" => PostgreSQLDataType::Varchar(None),
            "char" => PostgreSQLDataType::Char(None),
            "text" => PostgreSQLDataType::Text,
            "bytea" => PostgreSQLDataType::Bytea,
            "bool" => PostgreSQLDataType::Boolean,
            "uuid" => PostgreSQLDataType::Uuid,
            "json" => PostgreSQLDataType::Json,
            "jsonb" => PostgreSQLDataType::Jsonb,
            "geometry" => PostgreSQLDataType::Geometry(None),
            "geography" => PostgreSQLDataType::Geography(None),
            _ => PostgreSQLDataType::Custom(base_type.to_string()),
        }
    } else {
        PostgreSQLDataType::Unknown(udt_name.to_string())
    }
}

async fn get_geometry_info(
    pool: &PgPool,
    schema: &str,
    table_name: &str,
    column_name: &str,
    metadata_table: &str,
) -> Option<GeometryInfo> {
    // geography_columns uses f_geography_column, geometry_columns uses f_geometry_column
    let column_field = if metadata_table == "geography_columns" {
        "f_geography_column"
    } else {
        "f_geometry_column"
    };

    let query = format!(
        "SELECT type, srid, coord_dimension FROM {} WHERE f_table_schema = $1 AND f_table_name = $2 AND {} = $3",
        metadata_table, column_field
    );

    if let Ok(row) = sqlx::query(&query)
        .bind(schema)
        .bind(table_name)
        .bind(column_name)
        .fetch_optional(pool)
        .await
    {
        if let Some(row) = row {
            let geometry_type_str: String = row.get("type");
            let srid: Option<i32> = row.get("srid");
            let dimension: Option<i32> = row.get("coord_dimension");

            let geometry_type = match geometry_type_str.to_uppercase().as_str() {
                "POINT" => GeometryType::Point,
                "LINESTRING" => GeometryType::LineString,
                "POLYGON" => GeometryType::Polygon,
                "MULTIPOINT" => GeometryType::MultiPoint,
                "MULTILINESTRING" => GeometryType::MultiLineString,
                "MULTIPOLYGON" => GeometryType::MultiPolygon,
                "GEOMETRYCOLLECTION" => GeometryType::GeometryCollection,
                _ => return None,
            };

            return Some(GeometryInfo {
                geometry_type,
                srid,
                dimension,
            });
        }
    }

    None
}

pub async fn get_foreign_key_constraints(
    pool: &PgPool,
    schema: &str,
    table: &str,
) -> Result<Vec<ForeignKeyConstraint>, sqlx::Error> {
    let query = r#"
        SELECT 
            con.conname as constraint_name,
            array_agg(att.attname ORDER BY u.attposition)::text as columns,
            ref_class.relname as referenced_table,
            array_agg(ref_att.attname ORDER BY u.attposition)::text as referenced_columns,
            CASE con.confdeltype
                WHEN 'a' THEN NULL
                WHEN 'r' THEN 'RESTRICT'
                WHEN 'c' THEN 'CASCADE'
                WHEN 'n' THEN 'SET NULL'
                WHEN 'd' THEN 'SET DEFAULT'
            END as on_delete,
            CASE con.confupdtype
                WHEN 'a' THEN NULL
                WHEN 'r' THEN 'RESTRICT'
                WHEN 'c' THEN 'CASCADE'
                WHEN 'n' THEN 'SET NULL'
                WHEN 'd' THEN 'SET DEFAULT'
            END as on_update,
            con.condeferrable as is_deferrable,
            con.condeferred as initially_deferred
        FROM pg_constraint con
        JOIN pg_class class ON con.conrelid = class.oid
        JOIN pg_namespace nsp ON class.relnamespace = nsp.oid
        JOIN pg_class ref_class ON con.confrelid = ref_class.oid
        CROSS JOIN LATERAL unnest(con.conkey) WITH ORDINALITY AS u(attnum, attposition)
        JOIN pg_attribute att ON att.attrelid = class.oid AND att.attnum = u.attnum
        CROSS JOIN LATERAL unnest(con.confkey) WITH ORDINALITY AS u_ref(ref_attnum, ref_attposition)
        JOIN pg_attribute ref_att ON ref_att.attrelid = ref_class.oid 
            AND ref_att.attnum = u_ref.ref_attnum 
            AND u.attposition = u_ref.ref_attposition
        WHERE nsp.nspname = $1
            AND class.relname = $2
            AND con.contype = 'f'
        GROUP BY con.conname, ref_class.relname, con.confdeltype, con.confupdtype, 
                 con.condeferrable, con.condeferred
        ORDER BY con.conname
    "#;

    let rows = sqlx::query(query)
        .bind(schema)
        .bind(table)
        .fetch_all(pool)
        .await?;

    let mut constraints = Vec::new();
    for row in rows {
        let constraint_name: String = row.get("constraint_name");
        let columns_str: String = row.get("columns");
        let columns = parse_postgres_array(&columns_str);

        let referenced_table: String = row.get("referenced_table");

        let referenced_columns_str: String = row.get("referenced_columns");
        let referenced_columns = parse_postgres_array(&referenced_columns_str);
        let on_delete: Option<String> = row.get("on_delete");
        let on_update: Option<String> = row.get("on_update");
        let is_deferrable: bool = row.get("is_deferrable");
        let initially_deferred: bool = row.get("initially_deferred");

        let on_delete_action = on_delete.as_deref().and_then(|action| {
            if action == "NO ACTION" {
                None
            } else {
                ForeignKeyAction::from_str(action)
            }
        });

        let on_update_action = on_update.as_deref().and_then(|action| {
            if action == "NO ACTION" {
                None
            } else {
                ForeignKeyAction::from_str(action)
            }
        });

        constraints.push(ForeignKeyConstraint {
            constraint_name,
            columns,
            referenced_table,
            referenced_columns,
            on_delete: on_delete_action,
            on_update: on_update_action,
            is_deferrable,
            initially_deferred,
        });
    }

    Ok(constraints)
}

pub async fn get_table_indexes(
    pool: &PgPool,
    schema: &str,
    table: &str,
) -> Result<Vec<IndexInfo>, sqlx::Error> {
    let query = r#"
        SELECT 
            i.relname as index_name,
            array_agg(a.attname ORDER BY array_position(ix.indkey, a.attnum))::text as columns,
            ix.indisunique as is_unique,
            ix.indisprimary as is_primary,
            am.amname as index_type,
            pg_get_expr(ix.indpred, ix.indrelid) as condition
        FROM pg_index ix
        JOIN pg_class i ON i.oid = ix.indexrelid
        JOIN pg_class t ON t.oid = ix.indrelid
        JOIN pg_namespace n ON n.oid = t.relnamespace
        JOIN pg_am am ON am.oid = i.relam
        JOIN pg_attribute a ON a.attrelid = ix.indrelid AND a.attnum = ANY(ix.indkey)
        WHERE n.nspname = $1
            AND t.relname = $2
        GROUP BY i.relname, ix.indisunique, ix.indisprimary, am.amname, ix.indpred, ix.indrelid
        ORDER BY i.relname
    "#;

    let rows = sqlx::query(query)
        .bind(schema)
        .bind(table)
        .fetch_all(pool)
        .await?;

    let mut indexes = Vec::new();
    for row in rows {
        let index_name: String = row.get("index_name");
        let columns_str: String = row.get("columns");
        let columns = parse_postgres_array(&columns_str);
        let is_unique: bool = row.get("is_unique");
        let is_primary: bool = row.get("is_primary");
        let index_type_str: String = row.get("index_type");
        let index_type = IndexType::from_str(&index_type_str).unwrap_or(IndexType::BTree);
        let condition: Option<String> = row.get("condition");

        indexes.push(IndexInfo {
            index_name,
            columns,
            is_unique,
            is_primary,
            index_type,
            condition,
        });
    }

    Ok(indexes)
}

pub async fn get_check_constraints(
    pool: &PgPool,
    schema: &str,
    table: &str,
) -> Result<Vec<CheckConstraint>, sqlx::Error> {
    let query = r#"
        SELECT 
            cc.constraint_name,
            cc.check_clause
        FROM information_schema.check_constraints cc
        JOIN information_schema.table_constraints tc
            ON cc.constraint_name = tc.constraint_name
            AND cc.constraint_schema = tc.constraint_schema
        WHERE tc.table_schema = $1
            AND tc.table_name = $2
            AND tc.constraint_type = 'CHECK'
        ORDER BY cc.constraint_name
    "#;

    let rows = sqlx::query(query)
        .bind(schema)
        .bind(table)
        .fetch_all(pool)
        .await?;

    let mut constraints = Vec::new();
    for row in rows {
        let constraint_name: String = row.get("constraint_name");
        let check_clause: String = row.get("check_clause");

        // Skip system-generated NOT NULL constraints
        if is_not_null_constraint(&constraint_name, &check_clause) {
            continue;
        }

        constraints.push(CheckConstraint {
            constraint_name,
            check_clause,
        });
    }

    Ok(constraints)
}

pub async fn get_table_comment(
    pool: &PgPool,
    schema: &str,
    table: &str,
) -> Result<Option<String>, sqlx::Error> {
    let query = r#"
        SELECT 
            pgd.description
        FROM pg_class pgc
        JOIN pg_namespace pgn ON pgn.oid = pgc.relnamespace
        LEFT JOIN pg_description pgd ON pgd.objoid = pgc.oid AND pgd.objsubid = 0
        WHERE pgn.nspname = $1 AND pgc.relname = $2
    "#;

    let row = sqlx::query(query)
        .bind(schema)
        .bind(table)
        .fetch_optional(pool)
        .await?;

    Ok(row.and_then(|r| r.get("description")))
}

/// Check if a constraint is a system-generated NOT NULL constraint
fn is_not_null_constraint(constraint_name: &str, check_clause: &str) -> bool {
    // System-generated NOT NULL constraints typically have names like:
    // - {schema_oid}_{table_oid}_{column_number}_not_null
    // And check clauses like: (column_name IS NOT NULL)

    if constraint_name.ends_with("_not_null") {
        // Check if the constraint name follows the pattern: numbers_numbers_numbers_not_null
        let parts: Vec<&str> = constraint_name.split('_').collect();

        // Pattern is: numbers_numbers_numbers_not_null (5 parts total)
        if parts.len() == 5 && parts[3] == "not" && parts[4] == "null" {
            // Check if first 3 parts are all numbers
            if parts[0].chars().all(|c| c.is_ascii_digit())
                && parts[1].chars().all(|c| c.is_ascii_digit())
                && parts[2].chars().all(|c| c.is_ascii_digit())
            {
                // Also check if the check clause looks like a NOT NULL constraint
                return check_clause.contains("IS NOT NULL");
            }
        }
    }

    false
}

/// Check if an index is system-generated
pub fn is_system_generated_index(idx: &IndexInfo) -> bool {
    // System-generated indexes typically have names like:
    // - table_column_key (unique constraints)
    // - table_pkey (primary keys)
    // - constraint names starting with numbers (system-generated)

    let name = &idx.index_name;

    // Skip primary key indexes
    if idx.is_primary {
        return true;
    }

    // Skip indexes with names that look system-generated
    if name.ends_with("_pkey")
        || name.contains("_not_null")
        || name.chars().next().map_or(false, |c| c.is_ascii_digit())
    {
        return true;
    }

    // Allow UNIQUE constraint indexes (ending with _key) as they are user-defined constraints
    false
}

/// Convert MySQL placeholder (?) to PostgreSQL placeholders ($1, $2, $3, ...)
///
/// This function efficiently converts MySQL-style question mark placeholders to PostgreSQL-style
/// numbered placeholders while respecting SQL string literals and comments.
///
/// # Escape Sequences
///
/// - `\?` is converted to `?` (not a placeholder)
/// - `??` is converted to `?` (not a placeholder)
/// - Single `?` is converted to `$1`, `$2`, etc.
///
/// # Examples
///
/// ```rust
/// use senax_pgsql_parser::convert_mysql_placeholders_to_postgresql;
///
/// // Basic placeholder conversion
/// let mysql_sql = "SELECT * FROM users WHERE id = ? AND name = ?";
/// let postgresql_sql = convert_mysql_placeholders_to_postgresql(mysql_sql);
/// assert_eq!(postgresql_sql, "SELECT * FROM users WHERE id = $1 AND name = $2");
///
/// // String literals are preserved
/// let mysql_sql = "SELECT * FROM users WHERE name = 'user?' AND id = ?";
/// let postgresql_sql = convert_mysql_placeholders_to_postgresql(mysql_sql);
/// assert_eq!(postgresql_sql, "SELECT * FROM users WHERE name = 'user?' AND id = $1");
///
/// // Escaped \? is converted to ? (useful for PostgreSQL JSON operators)
/// let mysql_sql = r"SELECT * FROM users WHERE json_data \? 'key' AND id = ?";
/// let postgresql_sql = convert_mysql_placeholders_to_postgresql(mysql_sql);
/// assert_eq!(postgresql_sql, "SELECT * FROM users WHERE json_data ? 'key' AND id = $1");
///
/// // Double ?? is converted to ? (useful for PostgreSQL JSON operators)
/// let mysql_sql = "SELECT * FROM users WHERE json_data ?? 'key' AND id = ?";
/// let postgresql_sql = convert_mysql_placeholders_to_postgresql(mysql_sql);
/// assert_eq!(postgresql_sql, "SELECT * FROM users WHERE json_data ? 'key' AND id = $1");
///
/// // Combined usage
/// let mysql_sql = r"SELECT \? as escaped, ?? as double, ? as param";
/// let postgresql_sql = convert_mysql_placeholders_to_postgresql(mysql_sql);
/// assert_eq!(postgresql_sql, "SELECT ? as escaped, ? as double, $1 as param");
/// ```
pub fn convert_mysql_placeholders_to_postgresql(sql: &str) -> String {
    let mut result = String::with_capacity(sql.len() + 32); // Pre-allocate with extra space for placeholders
    let mut chars = sql.chars().peekable();
    let mut placeholder_count = 0;
    let mut in_single_quote = false;
    let mut in_double_quote = false;
    let mut in_line_comment = false;
    let mut in_block_comment = false;

    while let Some(ch) = chars.next() {
        match ch {
            // Handle string literals
            '\'' if !in_double_quote && !in_line_comment && !in_block_comment => {
                in_single_quote = !in_single_quote;
                result.push(ch);
            }
            '"' if !in_single_quote && !in_line_comment && !in_block_comment => {
                in_double_quote = !in_double_quote;
                result.push(ch);
            }

            // Handle SQL comments
            '-' if !in_single_quote && !in_double_quote && !in_block_comment => {
                if chars.peek() == Some(&'-') {
                    chars.next(); // consume second '-'
                    in_line_comment = true;
                    result.push_str("--");
                } else {
                    result.push(ch);
                }
            }
            '/' if !in_single_quote && !in_double_quote && !in_line_comment => {
                if chars.peek() == Some(&'*') {
                    chars.next(); // consume '*'
                    in_block_comment = true;
                    result.push_str("/*");
                } else {
                    result.push(ch);
                }
            }
            '*' if in_block_comment => {
                if chars.peek() == Some(&'/') {
                    chars.next(); // consume '/'
                    in_block_comment = false;
                    result.push_str("*/");
                } else {
                    result.push(ch);
                }
            }
            '\n' | '\r' if in_line_comment => {
                in_line_comment = false;
                result.push(ch);
            }

            // Handle escape sequences in string literals
            '\\' if (in_single_quote || in_double_quote)
                && !in_line_comment
                && !in_block_comment =>
            {
                result.push(ch);
                if let Some(next_ch) = chars.next() {
                    result.push(next_ch); // Push the escaped character as-is
                }
            }

            // Handle \? outside string literals - convert to ?
            '\\' if !in_single_quote
                && !in_double_quote
                && !in_line_comment
                && !in_block_comment =>
            {
                if chars.peek() == Some(&'?') {
                    chars.next(); // consume '?'
                    result.push('?'); // Just push '?' without converting to placeholder
                } else {
                    result.push(ch);
                }
            }

            // Handle placeholder conversion
            '?' if !in_single_quote
                && !in_double_quote
                && !in_line_comment
                && !in_block_comment =>
            {
                // Handle ?? - convert to single ?
                if chars.peek() == Some(&'?') {
                    chars.next(); // consume second '?'
                    result.push('?'); // Just push single '?' without converting to placeholder
                } else {
                    // Convert ? to $N
                    placeholder_count += 1;
                    result.push('$');
                    result.push_str(&placeholder_count.to_string());
                }
            }

            // Default case: just push the character
            _ => {
                result.push(ch);
            }
        }
    }

    result
}

pub fn escape_db_identifier(s: &str) -> String {
    format!(r#""{}""#, s.replace('"', r#""""#))
}