use crate::quote::quote_ident;
use crate::{builder, safety};
use pgrx::prelude::*;

pub(crate) fn registered_schema_drift_reason(
    tables: &[builder::RegisteredTable],
    edges: &[builder::RegisteredEdge],
    filter_columns: &[builder::RegisteredFilterColumn],
) -> Option<String> {
    for table in tables {
        let oid = table.table_oid;
        if let Err(err) = validate_registered_table(
            oid,
            &table.id_columns.as_catalog_text(),
            Some(table.columns.as_slice()),
            table.tenant_column.as_deref(),
        ) {
            return Some(format!(
                "registered table '{}' no longer matches graph catalog: {}",
                table.table_name, err
            ));
        }
    }

    for edge in edges {
        let from_oid = edge.from_table_oid;
        let to_oid = edge.to_table_oid;
        match primary_key_columns(from_oid) {
            Ok(columns) if columns == edge.source_key_columns.columns() => {}
            Ok(columns) => {
                return Some(format!(
                    "registered edge '{}' source key no longer matches table OID {}: catalog [{}], current [{}]",
                    edge.label,
                    from_oid,
                    edge.source_key_columns.as_catalog_text(),
                    columns.join(",")
                ));
            }
            Err(err) => {
                return Some(format!(
                    "registered edge '{}' source key is invalid: {}",
                    edge.label, err
                ));
            }
        }
        if let Err(err) = validate_column_exists(from_oid, &edge.from_column) {
            return Some(format!(
                "registered edge '{}' source column '{}.{}' is invalid: {}",
                edge.label, edge.from_table, edge.from_column, err
            ));
        }
        let from_table_registered = tables
            .iter()
            .any(|table| table.table_oid == edge.from_table_oid);
        if let Err(err) = validate_edge_endpoint_columns(
            from_oid,
            &edge.from_table,
            &edge.from_column,
            to_oid,
            &edge.to_table,
            &edge.to_column,
            from_table_registered,
        ) {
            return Some(format!(
                "registered edge '{}' endpoint columns are invalid: {}",
                edge.label, err
            ));
        }
        if let Some(weight_column) = edge.weight_column.as_deref() {
            if let Err(err) = validate_numeric_column(from_oid, weight_column) {
                return Some(format!(
                    "registered edge '{}' weight column '{}.{}' is invalid: {}",
                    edge.label, edge.from_table, weight_column, err
                ));
            }
        }
        if let Some(label_column) = edge.label_column.as_deref() {
            if let Err(err) = validate_column_exists(from_oid, label_column) {
                return Some(format!(
                    "registered edge '{}' label column '{}.{}' is invalid: {}",
                    edge.label, edge.from_table, label_column, err
                ));
            }
        }
    }

    for filter in filter_columns {
        let table_oid = filter.table_oid;
        if let Err(err) =
            validate_filter_column_type(table_oid, &filter.column_name, &filter.column_type)
        {
            return Some(format!(
                "registered filter column '{}.{}' is invalid: {}",
                filter.table_name, filter.column_name, err
            ));
        }
    }

    None
}

#[cfg(test)]
mod tests {
    use crate::builder::split_catalog_columns;

    /// Covers catalog column-list parsing used by discovery functions and the
    /// `id_columns` compatibility layer.
    #[test]
    fn split_catalog_columns_ignores_empty_segments_and_whitespace() {
        assert_eq!(
            split_catalog_columns(" id, , tenant_id "),
            vec!["id".to_string(), "tenant_id".to_string()]
        );
        assert!(split_catalog_columns("").is_empty());
    }
}

pub(crate) fn regclass_text(table_oid: u32) -> safety::GraphResult<String> {
    Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select("SELECT $1::oid::regclass::text", None, &[table_oid.into()])
            .map_err(|e| safety::GraphError::Internal(format!("regclass lookup failed: {}", e)))?;
        let row = result.first();
        row.get::<String>(1)
            .map_err(|e| safety::GraphError::Internal(format!("regclass read failed: {}", e)))?
            .ok_or_else(|| {
                safety::GraphError::Internal(format!("NULL regclass for OID {}", table_oid))
            })
    })
}

/// Return the relation label exposed by pgGraph's public result types.
///
/// SQL generation must use [`regclass_text`] because it needs an unambiguous,
/// schema-qualified relation reference. Public APIs historically expose the
/// relation name itself, so keep that representation independent from the
/// security-definer function search path.
pub(crate) fn relation_name(table_oid: u32) -> safety::GraphResult<String> {
    Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(
                "SELECT relname::text FROM pg_catalog.pg_class WHERE oid = $1::oid",
                None,
                &[table_oid.into()],
            )
            .map_err(|e| {
                safety::GraphError::Internal(format!("relation name lookup failed: {e}"))
            })?;
        result
            .first()
            .get::<String>(1)
            .map_err(|e| safety::GraphError::Internal(format!("relation name read failed: {e}")))?
            .ok_or_else(|| {
                safety::GraphError::Internal(format!("NULL relation name for OID {table_oid}"))
            })
    })
}

#[derive(Debug, Clone)]
pub(crate) struct SqlTableName {
    sql: String,
}

impl SqlTableName {
    pub(crate) fn as_sql(&self) -> &str {
        &self.sql
    }
}

/// Render a registered relation directly from its stable PostgreSQL OID.
///
/// This avoids resolving a previously-read display name after a concurrent DDL
/// change has renamed, dropped, or recreated the relation.
pub(crate) fn sql_table_name_from_oid(table_oid: u32) -> safety::GraphResult<SqlTableName> {
    let sql = regclass_text(table_oid)?;
    Ok(SqlTableName { sql })
}

pub(crate) fn table_oid_from_name(table_name: &str) -> safety::GraphResult<u32> {
    Spi::connect(|client| {
        let result = client
            .select(
                "SELECT to_regclass($1)::oid::integer",
                None,
                &[table_name.into()],
            )
            .map_err(|e| {
                safety::GraphError::Internal(format!(
                    "table lookup failed for {}: {}",
                    table_name, e
                ))
            })?;
        let row = result.first();
        row.get::<i32>(1)
            .map_err(|e| safety::GraphError::Internal(format!("table OID read failed: {}", e)))?
            .map(|oid| oid as u32)
            .ok_or_else(|| {
                safety::GraphError::Internal(format!("relation not found: {}", table_name))
            })
    })
}

/// Resolve one foreign-key column to its canonical referenced relation.
///
/// PostgreSQL creates inherited constraint rows for partitions. Canonicalizing
/// both sides to their partition roots keeps those physical rows equivalent to
/// the single logical relationship mapping. Multiple distinct referenced roots
/// remain ambiguous and return `None` rather than selecting an arbitrary table.
pub(crate) fn foreign_key_target_table_oid(
    relation_oid: u32,
    column: &str,
) -> safety::GraphResult<Option<u32>> {
    Spi::connect(|client| {
        let rows = client
            .select(
                "SELECT pg_catalog.min(
                            COALESCE(
                                pg_catalog.pg_partition_root(foreign_key.confrelid),
                                foreign_key.confrelid
                            )::oid::integer
                        ),
                        pg_catalog.count(DISTINCT COALESCE(
                            pg_catalog.pg_partition_root(foreign_key.confrelid),
                            foreign_key.confrelid
                        ))::bigint
                   FROM pg_catalog.pg_constraint AS foreign_key
                   JOIN pg_catalog.unnest(foreign_key.conkey) WITH ORDINALITY
                        AS source_key(attnum, ordinality) ON true
                   JOIN pg_catalog.pg_attribute AS source_attribute
                     ON source_attribute.attrelid = foreign_key.conrelid
                    AND source_attribute.attnum = source_key.attnum
                  WHERE foreign_key.contype = 'f'
                    AND COALESCE(
                            pg_catalog.pg_partition_root(foreign_key.conrelid),
                            foreign_key.conrelid
                        ) = $1::oid
                    AND source_attribute.attname = $2",
                None,
                &[
                    pgrx::pg_sys::Oid::from_u32(relation_oid).into(),
                    column.into(),
                ],
            )
            .map_err(|err| {
                safety::GraphError::Internal(format!(
                    "foreign-key target lookup failed for relation OID {relation_oid}: {err}"
                ))
            })?;
        let row = rows.first();
        let target_count = row.get::<i64>(2).map_err(|err| {
            safety::GraphError::Internal(format!(
                "foreign-key target count read failed for relation OID {relation_oid}: {err}"
            ))
        })?;
        if target_count != Some(1) {
            return Ok(None);
        }
        row.get::<i32>(1)
            .map_err(|err| {
                safety::GraphError::Internal(format!(
                    "foreign-key target read failed for relation OID {relation_oid}: {err}"
                ))
            })
            .map(|oid| oid.map(|oid| oid as u32))
    })
}

pub(crate) fn estimated_table_rows(table_name: &str) -> safety::GraphResult<i64> {
    let table_oid = table_oid_from_name(table_name)?;
    let sql_table_name = sql_table_name_from_oid(table_oid)?;
    Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(
                "SELECT COALESCE(reltuples, 0)::bigint FROM pg_catalog.pg_class WHERE oid = $1::oid",
                None,
                &[table_oid.into()],
            )
            .map_err(|e| {
                safety::GraphError::Internal(format!(
                    "reltuples estimate failed for {}: {}",
                    table_name, e
                ))
            })?;
        let row = result.first();
        let estimate = row
            .get::<i64>(1)
            .map_err(|e| safety::GraphError::Internal(format!("reltuples read failed: {}", e)))?
            .unwrap_or(0)
            .max(0);
        if estimate > 0 {
            return Ok(estimate);
        }

        // Newly created, truncated, or never-analyzed tables can report zero
        // even when they contain rows. An exact preflight scan is slower but
        // prevents unknown statistics from authorizing an unbounded build.
        let exact_sql = format!("SELECT count(*)::bigint FROM {}", sql_table_name.as_sql());
        let exact = client.select(&exact_sql, None, &[]).map_err(|e| {
            safety::GraphError::Internal(format!(
                "exact row-count preflight failed for {}: {}",
                table_name, e
            ))
        })?;
        exact
            .first()
            .get::<i64>(1)
            .map_err(|e| safety::GraphError::Internal(format!("exact row count failed: {}", e)))?
            .ok_or_else(|| {
                safety::GraphError::Internal(format!(
                    "exact row-count preflight returned NULL for {}",
                    table_name
                ))
            })
    })
}

pub(crate) fn primary_key_expr(alias: &str, primary_key: &builder::PrimaryKeySpec) -> String {
    if primary_key.columns().len() > 1 {
        let parts = primary_key
            .columns()
            .iter()
            .map(|col| format!("{}.{}::text", alias, quote_ident(col)))
            .collect::<Vec<_>>()
            .join(", ");
        format!("jsonb_build_array({})::text", parts)
    } else if let Some(column) = primary_key.columns().first() {
        format!("{}.{}::text", alias, quote_ident(column))
    } else {
        "NULL::text".to_string()
    }
}

/// Return the declared primary-key columns for a relation in index order.
///
/// Relationship mappings use this tuple as their authoritative source-row
/// identity. A unique index is not substituted here: PostgreSQL primary keys
/// give the durable, non-null contract required by relationship hydration,
/// visibility, and mutation paths.
pub(crate) fn primary_key_columns(table_oid: u32) -> safety::GraphResult<Vec<String>> {
    Spi::connect(|client| {
        let result = client
            .select(
                "SELECT array_agg(attribute.attname::text ORDER BY key_column.ordinality)::text[]
                   FROM pg_catalog.pg_index AS index
                   JOIN pg_catalog.unnest(index.indkey) WITH ORDINALITY
                        AS key_column(attnum, ordinality) ON true
                   JOIN pg_catalog.pg_attribute AS attribute
                     ON attribute.attrelid = index.indrelid
                    AND attribute.attnum = key_column.attnum
                  WHERE index.indrelid = $1::oid
                    AND index.indisprimary
                  GROUP BY index.indexrelid",
                None,
                &[pgrx::pg_sys::Oid::from_u32(table_oid).into()],
            )
            .map_err(|err| {
                safety::GraphError::Internal(format!("primary-key lookup failed: {err}"))
            })?;
        result
            .first()
            .get::<Vec<String>>(1)
            .map_err(|err| safety::GraphError::Internal(format!("primary-key read failed: {err}")))?
            .filter(|columns| !columns.is_empty())
            .ok_or_else(|| safety::GraphError::InvalidFilter {
                reason: format!(
                    "relationship source table OID {table_oid} must have a primary key"
                ),
            })
    })
}

pub(crate) fn validate_column_exists(table_oid: u32, column: &str) -> safety::GraphResult<()> {
    let exists = Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(
                "SELECT EXISTS (
                SELECT 1
                FROM pg_catalog.pg_attribute
                WHERE attrelid = $1::oid
                  AND attname = $2
                  AND attnum > 0
                  AND NOT attisdropped
            )",
                None,
                &[table_oid.into(), column.into()],
            )
            .map_err(|e| {
                safety::GraphError::Internal(format!("column validation failed: {}", e))
            })?;
        let row = result.first();
        Ok::<_, safety::GraphError>(
            row.get::<bool>(1)
                .map_err(|e| {
                    safety::GraphError::Internal(format!("column validation read failed: {}", e))
                })?
                .unwrap_or(false),
        )
    })?;
    if exists {
        Ok(())
    } else {
        Err(safety::GraphError::Internal(format!(
            "column '{}' does not exist on table OID {}",
            column, table_oid
        )))
    }
}

#[allow(
    clippy::too_many_arguments,
    reason = "edge validation needs both endpoint table/column pairs for clear diagnostics"
)]
pub(crate) fn validate_edge_endpoint_columns(
    from_oid: u32,
    from_table: &str,
    from_column: &str,
    to_oid: u32,
    to_table: &str,
    to_column: &str,
    from_table_registered: bool,
) -> safety::GraphResult<()> {
    validate_column_exists(from_oid, from_column)?;
    if from_table_registered {
        validate_column_exists(to_oid, to_column).map_err(|err| {
            safety::GraphError::Internal(format!(
                "FK-style edge registration from source table '{}' requires to_column '{}' on target table '{}': {}",
                from_table, to_column, to_table, err
            ))
        })
    } else {
        validate_column_exists(from_oid, to_column).map_err(|err| {
            safety::GraphError::Internal(format!(
                "edge-table registration from source table '{}' requires to_column '{}' on the source edge table because both endpoint values are read from '{}': {}",
                from_table, to_column, from_table, err
            ))
        })
    }
}

pub(crate) fn validate_numeric_column(table_oid: u32, column: &str) -> safety::GraphResult<()> {
    let is_numeric = Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(
                "SELECT t.typcategory = 'N'
             FROM pg_catalog.pg_attribute a
             JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
             WHERE a.attrelid = $1::oid
               AND a.attname = $2
               AND a.attnum > 0
               AND NOT a.attisdropped",
                None,
                &[table_oid.into(), column.into()],
            )
            .map_err(|e| {
                safety::GraphError::Internal(format!("numeric validation failed: {}", e))
            })?;
        let row = result.first();
        Ok::<_, safety::GraphError>(
            row.get::<bool>(1)
                .map_err(|e| {
                    safety::GraphError::Internal(format!("numeric validation read failed: {}", e))
                })?
                .unwrap_or(false),
        )
    })?;

    if is_numeric {
        Ok(())
    } else {
        Err(safety::GraphError::Internal(format!(
            "column '{}' on table OID {} is not numeric",
            column, table_oid
        )))
    }
}

pub(crate) fn validate_filter_column_type(
    table_oid: u32,
    column: &str,
    column_type: &str,
) -> safety::GraphResult<()> {
    match column_type.trim().to_ascii_lowercase().as_str() {
        "numeric" => validate_numeric_column(table_oid, column),
        "text" | "boolean" | "date" | "timestamptz" | "uuid" => {
            validate_column_exists(table_oid, column)
        }
        other => Err(safety::GraphError::InvalidFilter {
            reason: format!(
                "unsupported filter column_type '{}'; expected text, numeric, boolean, date, timestamptz, or uuid",
                other
            ),
        }),
    }
}

pub(crate) fn validate_registered_table(
    table_oid: u32,
    id_column: &str,
    columns: Option<&[String]>,
    tenant_column: Option<&str>,
) -> safety::GraphResult<()> {
    let candidate_keys = Spi::connect(|client| {
        let query = format!(
            "SELECT
                i.indisprimary,
                array_agg(a.attname::text ORDER BY ord.n)::text[] AS columns,
                bool_and(a.attnotnull) AS all_not_null
             FROM pg_catalog.pg_index i
             JOIN unnest(i.indkey) WITH ORDINALITY AS ord(attnum, n) ON true
             JOIN pg_catalog.pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ord.attnum
             WHERE i.indrelid = {}::oid
               AND (i.indisprimary OR i.indisunique)
               AND i.indpred IS NULL
             GROUP BY i.indexrelid, i.indisprimary
             ORDER BY i.indisprimary DESC, i.indexrelid",
            table_oid
        );
        let result = client.select(&query, None, &[]).map_err(|e| {
            safety::GraphError::Internal(format!("identifier validation failed: {}", e))
        })?;
        let mut keys = Vec::new();
        for row in result {
            let is_primary = row
                .get::<bool>(1)
                .map_err(|err| safety::GraphError::Internal(err.to_string()))?
                .unwrap_or(false);
            let columns = row
                .get::<Vec<String>>(2)
                .map_err(|err| safety::GraphError::Internal(err.to_string()))?
                .unwrap_or_default();
            let all_not_null = row
                .get::<bool>(3)
                .map_err(|err| safety::GraphError::Internal(err.to_string()))?
                .unwrap_or(false);
            keys.push((is_primary, columns, all_not_null));
        }
        Ok::<_, safety::GraphError>(keys)
    })?;

    let provided_cols = id_column
        .split(',')
        .map(str::trim)
        .filter(|col| !col.is_empty())
        .collect::<Vec<_>>();

    if provided_cols.is_empty() {
        return Err(safety::GraphError::InvalidFilter {
            reason: "id_columns must contain at least one column".to_string(),
        });
    }

    let matches_identifier =
        candidate_keys
            .iter()
            .any(|(_is_primary, key_columns, all_not_null)| {
                *all_not_null
                    && key_columns.len() == provided_cols.len()
                    && key_columns
                        .iter()
                        .zip(provided_cols.iter())
                        .all(|(actual, provided)| actual == provided)
            });

    if !matches_identifier {
        let available = candidate_keys
            .iter()
            .filter(|(_is_primary, _columns, all_not_null)| *all_not_null)
            .map(|(is_primary, columns, _all_not_null)| {
                let kind = if *is_primary {
                    "primary key"
                } else {
                    "unique NOT NULL index"
                };
                format!("{} ({})", kind, columns.join(","))
            })
            .collect::<Vec<_>>()
            .join("; ");
        return Err(safety::GraphError::InvalidFilter {
            reason: if available.is_empty() {
                format!(
                    "id_columns '{}' must match a primary key or unique NOT NULL index on table OID {}",
                    id_column, table_oid
                )
            } else {
                format!(
                    "id_columns '{}' must match a primary key or unique NOT NULL index on table OID {}; available identifiers: {}",
                    id_column, table_oid, available
                )
            },
        });
    }

    for col in &provided_cols {
        validate_column_exists(table_oid, col)?;
    }
    if let Some(cols) = columns {
        for property in cols {
            validate_registered_property(table_oid, property)?;
        }
    }
    if let Some(column) = tenant_column {
        validate_column_exists(table_oid, column)?;
    }
    Ok(())
}

fn validate_registered_property(table_oid: u32, property: &str) -> safety::GraphResult<()> {
    let Some((base_column, _path)) = property.split_once('.') else {
        return validate_column_exists(table_oid, property);
    };
    if base_column.is_empty() {
        return Err(safety::GraphError::InvalidFilter {
            reason: format!("jsonb property path '{property}' is missing a base column"),
        });
    }
    validate_jsonb_column(table_oid, base_column).map_err(|err| match err {
        safety::GraphError::Internal(reason) => safety::GraphError::InvalidFilter {
            reason: format!(
                "jsonb property path '{property}' requires base column '{base_column}' to exist and have type jsonb: {reason}"
            ),
        },
        other => other,
    })
}

fn validate_jsonb_column(table_oid: u32, column: &str) -> safety::GraphResult<()> {
    let is_jsonb = Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(
                "SELECT a.atttypid = 'jsonb'::regtype
                 FROM pg_catalog.pg_attribute a
                 WHERE a.attrelid = $1::oid
                   AND a.attname = $2
                   AND a.attnum > 0
                   AND NOT a.attisdropped",
                None,
                &[table_oid.into(), column.into()],
            )
            .map_err(|e| {
                safety::GraphError::Internal(format!("jsonb column validation failed: {}", e))
            })?;
        let row = result.first();
        Ok::<_, safety::GraphError>(
            row.get::<bool>(1)
                .map_err(|e| {
                    safety::GraphError::Internal(format!(
                        "jsonb column validation read failed: {}",
                        e
                    ))
                })?
                .unwrap_or(false),
        )
    })?;
    if is_jsonb {
        Ok(())
    } else {
        Err(safety::GraphError::Internal(format!(
            "column '{}' on table OID {} is not jsonb",
            column, table_oid
        )))
    }
}
