//! # Discover — Schema auto-discovery
//!
//! Discovers tables, primary keys, and foreign keys from the information schema
//! for SQL auto-discovery wrappers to register.
//!
//! Composite primary keys are supported:
//! - **Junction tables** (all PK columns are FKs) → registered as edges, not nodes.
//! - **Composite entities** (≥1 PK column is not a FK) → registered as nodes with
//!   a typed primary-key column set. The builder generates
//!   `jsonb_build_array(col1::text, col2::text)::text` as the PK string.
//! - **Typed relationship tables** with a surrogate key, exactly two foreign
//!   keys, and a conventional relationship-type text column are registered as
//!   one dynamic edge mapping rather than as nodes.
//!
//! See: `docs/user_guide/schema-registration.mdx`

use pgrx::prelude::*;
use std::collections::HashSet;

use crate::builder::{PrimaryKeySpec, PropertyColumns, RegisteredEdge, RegisteredTable};
use crate::catalog::{regclass_text, validate_column_exists};
use crate::quote::quote_ident;
use crate::safety::{GraphError, GraphResult};

/// Result of auto-discovery for SQL output.
#[derive(Debug, Clone)]
pub struct DiscoveryResult {
    pub item_type: String, // "table", "edge", or "junction"
    pub item_name: String,
    pub details: String,
}

/// Info about a discovered table before classification.
struct DiscoveredTable {
    table_oid: Option<u32>,
    schema_name: String,
    table_name: String,
    pk_columns: Vec<String>,
    id_is_primary: bool,
    text_columns: Vec<String>,
    relationship_label_column: Option<String>,
}

/// Auto-discover tables and foreign keys from a schema.
pub fn discover_schema(
    schema_name: &str,
) -> GraphResult<(
    Vec<RegisteredTable>,
    Vec<RegisteredEdge>,
    Vec<DiscoveryResult>,
)> {
    let mut tables = Vec::new();
    let mut edges = Vec::new();
    let mut discoveries = Vec::new();

    // Step 1: Find all tables with their primary key columns (including composite PKs)
    let discovered_tables = discover_tables_with_pks(schema_name)?;

    // Step 2: Find all FK relationships in the schema (needed for both edge discovery
    // AND junction table classification)
    let schema_fks = discover_foreign_keys(schema_name)?;

    let mut relationship_oids = HashSet::new();

    // Step 3: Classify each table
    for table in &discovered_tables {
        if table.pk_columns.is_empty() {
            // No PK at all — skip
            continue;
        }

        let table_oid = required_table_oid(table)?;
        let label_column = dynamic_relationship_label_column(table);
        let is_typed_relationship = table.pk_columns.len() == 1
            && label_column.is_some()
            && has_binary_endpoint_shape(table_oid, &schema_fks);

        if is_typed_relationship {
            relationship_oids.insert(table_oid);
            discoveries.push(relationship_discovery(table, label_column));
        } else if table.pk_columns.len() == 1 {
            // Single-column PK tables are registered as node tables.
            let id_column = table.pk_columns[0].clone();
            discoveries.push(DiscoveryResult {
                item_type: "table".to_string(),
                item_name: table.table_name.clone(),
                details: format!(
                    "pk={}, columns=[{}]",
                    id_column,
                    table.text_columns.join(", ")
                ),
            });

            tables.push(RegisteredTable {
                table_oid: required_table_oid(table)?,
                table_name: format!(
                    "{}.{}",
                    quote_ident(schema_name),
                    quote_ident(&table.table_name)
                ),
                id_columns: PrimaryKeySpec::from_columns(vec![id_column]),
                columns: PropertyColumns::from_columns(table.text_columns.clone()),
                tenant_column: None,
            });
        } else {
            // Composite PK — classify as junction or entity
            let is_junction = classify_as_junction(
                table.table_oid,
                &table.table_name,
                &table.pk_columns,
                &schema_fks,
            );

            if is_junction {
                relationship_oids.insert(table_oid);
                pgrx::notice!(
                    "graph: table '{}' has composite PK ({}) where all columns are foreign keys — treated as a junction table (edges only, not a node)",
                    table.table_name,
                    table.pk_columns.join(", ")
                );
                discoveries.push(DiscoveryResult {
                    item_type: "junction".to_string(),
                    item_name: table.table_name.clone(),
                    details: format!(
                        "composite pk=[{}], all columns are FKs — registered as edges",
                        table.pk_columns.join(", ")
                    ),
                });
            } else {
                // Composite entity: at least one PK col is NOT a FK — register as node
                let id_columns = PrimaryKeySpec::from_columns(table.pk_columns.clone());
                let id_column = id_columns.as_catalog_text();
                discoveries.push(DiscoveryResult {
                    item_type: "table".to_string(),
                    item_name: table.table_name.clone(),
                    details: format!(
                        "composite pk=[{}], columns=[{}]",
                        id_column,
                        table.text_columns.join(", ")
                    ),
                });

                tables.push(RegisteredTable {
                    table_oid: required_table_oid(table)?,
                    table_name: format!(
                        "{}.{}",
                        quote_ident(schema_name),
                        quote_ident(&table.table_name)
                    ),
                    id_columns,
                    columns: PropertyColumns::from_columns(table.text_columns.clone()),
                    tenant_column: None,
                });
            }
        }
    }

    // Step 4: Register junction/relationship tables as one edge per endpoint
    // pair, then register ordinary foreign keys independently.
    for relationship_oid in &relationship_oids {
        let table = discovered_tables
            .iter()
            .find(|table| table.table_oid == Some(*relationship_oid))
            .ok_or_else(|| {
                GraphError::Internal(format!(
                    "relationship discovery lost table OID {}",
                    relationship_oid
                ))
            })?;
        register_relationship_edges(
            table,
            &schema_fks,
            dynamic_relationship_label_column(table),
            &mut edges,
            &mut discoveries,
        )?;
    }

    for fk in schema_fks.iter().filter(|fk| {
        fk.from_oid
            .is_none_or(|from_oid| !relationship_oids.contains(&from_oid))
    }) {
        let label = edge_label(&fk.from_column);

        discoveries.push(DiscoveryResult {
            item_type: "edge".to_string(),
            item_name: format!(
                "{}.{} → {}.{}",
                fk.from_table, fk.from_column, fk.to_table, fk.to_column
            ),
            details: format!("label={}, bidirectional=true", label),
        });

        edges.push(registered_edge(
            required_fk_oid(fk.from_oid, "source")?,
            format!(
                "{}.{}",
                quote_ident(schema_name),
                quote_ident(&fk.from_table)
            ),
            &fk.from_column,
            required_fk_oid(fk.to_oid, "target")?,
            format!("{}.{}", quote_ident(schema_name), quote_ident(&fk.to_table)),
            &fk.to_column,
            &fk.from_column,
        ));
    }

    Ok((tables, edges, discoveries))
}

/// Auto-discover graph metadata for an explicit set of tables.
pub fn discover_table_set(
    table_oids: &[u32],
    tenant_column: Option<&str>,
) -> GraphResult<(
    Vec<RegisteredTable>,
    Vec<RegisteredEdge>,
    Vec<DiscoveryResult>,
)> {
    validate_target_table_list(table_oids)?;

    let mut tables = Vec::new();
    let mut edges = Vec::new();
    let mut discoveries = Vec::new();
    let selected_oids = table_oids.iter().copied().collect::<HashSet<_>>();
    let discovered_tables = discover_tables_by_oid(table_oids)?;
    let selected_fks = discover_foreign_keys_for_tables(table_oids)?;
    let mut relationship_oids = HashSet::new();

    for table in &discovered_tables {
        if let Some(column) = tenant_column {
            validate_column_exists(required_table_oid(table)?, column)?;
        }

        let table_oid = required_table_oid(table)?;
        let label_column = dynamic_relationship_label_column(table);
        let is_typed_relationship = table.pk_columns.len() == 1
            && label_column.is_some()
            && has_binary_endpoint_shape(table_oid, &selected_fks);

        if is_typed_relationship {
            relationship_oids.insert(table_oid);
            discoveries.push(relationship_discovery(table, label_column));
        } else if table.pk_columns.len() == 1 {
            let id_column = table.pk_columns[0].clone();
            discoveries.push(DiscoveryResult {
                item_type: "table".to_string(),
                item_name: table.table_name.clone(),
                details: discovery_details("pk", &id_column, &table.text_columns, tenant_column),
            });
            tables.push(registered_table(
                table,
                PrimaryKeySpec::from_columns(vec![id_column]),
                tenant_column,
            )?);
        } else if table.id_is_primary
            && classify_as_junction(
                table.table_oid,
                &table.table_name,
                &table.pk_columns,
                &selected_fks,
            )
        {
            relationship_oids.insert(table_oid);
            pgrx::notice!(
                "graph: table '{}' has composite PK ({}) where all columns are foreign keys — treated as a junction table (edges only, not a node)",
                table.table_name,
                table.pk_columns.join(", ")
            );
            discoveries.push(DiscoveryResult {
                item_type: "junction".to_string(),
                item_name: table.table_name.clone(),
                details: format!(
                    "composite pk=[{}], all columns are FKs — registered as edges",
                    table.pk_columns.join(", ")
                ),
            });
        } else {
            let id_columns = PrimaryKeySpec::from_columns(table.pk_columns.clone());
            let id_column = id_columns.as_catalog_text();
            discoveries.push(DiscoveryResult {
                item_type: "table".to_string(),
                item_name: table.table_name.clone(),
                details: discovery_details(
                    if table.id_is_primary {
                        "composite pk"
                    } else {
                        "unique not-null key"
                    },
                    &id_column,
                    &table.text_columns,
                    tenant_column,
                ),
            });
            tables.push(registered_table(table, id_columns, tenant_column)?);
        }
    }

    for relationship_oid in &relationship_oids {
        let table = discovered_tables
            .iter()
            .find(|table| table.table_oid == Some(*relationship_oid))
            .ok_or_else(|| {
                GraphError::Internal(format!(
                    "relationship discovery lost table OID {}",
                    relationship_oid
                ))
            })?;
        register_relationship_edges(
            table,
            &selected_fks,
            dynamic_relationship_label_column(table),
            &mut edges,
            &mut discoveries,
        )?;
    }

    for fk in selected_fks
        .iter()
        .filter(|fk| {
            fk.from_oid
                .is_some_and(|from_oid| selected_oids.contains(&from_oid))
                && fk
                    .to_oid
                    .is_some_and(|to_oid| selected_oids.contains(&to_oid))
        })
        .filter(|fk| {
            fk.from_oid
                .is_none_or(|from_oid| !relationship_oids.contains(&from_oid))
        })
    {
        let label = edge_label(&fk.from_column);

        discoveries.push(DiscoveryResult {
            item_type: "edge".to_string(),
            item_name: format!(
                "{}.{} → {}.{}",
                fk.from_table, fk.from_column, fk.to_table, fk.to_column
            ),
            details: format!("label={}, bidirectional=true", label),
        });
        edges.push(registered_edge(
            required_fk_oid(fk.from_oid, "source")?,
            regclass_text(required_fk_oid(fk.from_oid, "source")?)?,
            &fk.from_column,
            required_fk_oid(fk.to_oid, "target")?,
            regclass_text(required_fk_oid(fk.to_oid, "target")?)?,
            &fk.to_column,
            &fk.from_column,
        ));
    }

    Ok((tables, edges, discoveries))
}

/// A foreign key relationship discovered from the schema.
struct DiscoveredFk {
    from_oid: Option<u32>,
    from_table: String,
    from_column: String,
    to_oid: Option<u32>,
    to_table: String,
    to_column: String,
}

/// Discover all tables and their primary key columns.
fn discover_tables_with_pks(schema_name: &str) -> GraphResult<Vec<DiscoveredTable>> {
    // Query all tables with their PK columns (supports composite PKs)
    let table_pk_query = "SELECT relation.oid::integer AS table_oid,
                t.table_name::text AS table_name,
                kcu.column_name::text AS pk_column,
                kcu.ordinal_position
         FROM information_schema.tables t
         JOIN pg_catalog.pg_namespace namespace ON namespace.nspname = t.table_schema
         JOIN pg_catalog.pg_class relation
           ON relation.relnamespace = namespace.oid AND relation.relname = t.table_name
         JOIN information_schema.table_constraints tc
           ON tc.table_schema = t.table_schema AND tc.table_name = t.table_name
           AND tc.constraint_type = 'PRIMARY KEY'
         JOIN information_schema.key_column_usage kcu
           ON kcu.constraint_name = tc.constraint_name AND kcu.table_schema = tc.table_schema
         WHERE t.table_schema = $1
           AND t.table_type = 'BASE TABLE'
         ORDER BY t.table_name, kcu.ordinal_position";

    let mut table_map: Vec<(u32, String, Vec<String>)> = Vec::new();

    Spi::connect(|client| {
        let result = client
            .select(table_pk_query, None, &[schema_name.into()])
            .map_err(|e| GraphError::Internal(format!("Schema discovery failed: {}", e)))?;

        for row in result {
            let table_oid = row
                .get::<i32>(1)
                .map_err(|e| GraphError::Internal(format!("Cannot read table_oid: {}", e)))?
                .ok_or_else(|| {
                    GraphError::Internal("Schema discovery returned a NULL table OID".to_string())
                })?;
            let table_name: String = row
                .get::<String>(2)
                .map_err(|e| GraphError::Internal(format!("Cannot read table_name: {}", e)))?
                .unwrap_or_default();
            let pk_column: String = row
                .get::<String>(3)
                .map_err(|e| GraphError::Internal(format!("Cannot read pk_column: {}", e)))?
                .unwrap_or_default();

            // Group PK columns by table name
            if let Some(last) = table_map.last_mut() {
                if last.0 == table_oid as u32 {
                    last.2.push(pk_column);
                    continue;
                }
            }
            table_map.push((table_oid as u32, table_name, vec![pk_column]));
        }
        Ok::<(), GraphError>(())
    })?;

    // For each table, discover text/varchar columns
    let mut discovered = Vec::new();
    for (table_oid, table_name, pk_columns) in table_map {
        let text_columns = discover_text_columns(schema_name, &table_name, &pk_columns)?;
        let relationship_label_column =
            discover_dynamic_relationship_label_column(schema_name, &table_name)?;
        discovered.push(DiscoveredTable {
            table_oid: Some(table_oid),
            schema_name: schema_name.to_string(),
            table_name,
            pk_columns,
            id_is_primary: true,
            text_columns,
            relationship_label_column,
        });
    }

    Ok(discovered)
}

/// Discover text/varchar columns for a table (excluding PK columns).
fn discover_text_columns(
    schema_name: &str,
    table_name: &str,
    pk_columns: &[String],
) -> GraphResult<Vec<String>> {
    let col_query = "SELECT column_name::text FROM information_schema.columns
         WHERE table_schema = $1 AND table_name = $2
           AND data_type IN ('text', 'character varying')
           AND (character_maximum_length IS NULL OR character_maximum_length <= 128)
           AND column_name <> ALL($3::text[])
         ORDER BY ordinal_position";

    let mut columns = Vec::new();
    Spi::connect(|client| {
        let col_result = client
            .select(
                col_query,
                None,
                &[
                    schema_name.into(),
                    table_name.into(),
                    pk_columns.to_vec().into(),
                ],
            )
            .map_err(|e| GraphError::Internal(format!("Column discovery failed: {}", e)))?;

        for col_row in col_result {
            if let Ok(Some(col_name)) = col_row.get::<String>(1) {
                columns.push(col_name);
            }
        }
        Ok::<(), GraphError>(())
    })?;

    Ok(columns)
}

fn discover_dynamic_relationship_label_column(
    schema_name: &str,
    table_name: &str,
) -> GraphResult<Option<String>> {
    Spi::connect(|client| {
        let rows = client
            .select(
                "SELECT column_name::text
                   FROM information_schema.columns
                  WHERE table_schema = $1
                    AND table_name = $2
                    AND lower(column_name) IN (
                        'relationship_name', 'relationship_type', 'edge_type', 'rel_type'
                    )
                    AND data_type IN ('text', 'character varying')
                  ORDER BY CASE lower(column_name)
                      WHEN 'relationship_name' THEN 1
                      WHEN 'relationship_type' THEN 2
                      WHEN 'edge_type' THEN 3
                      WHEN 'rel_type' THEN 4
                      ELSE 5
                  END
                  LIMIT 1",
                None,
                &[schema_name.into(), table_name.into()],
            )
            .map_err(|error| {
                GraphError::Internal(format!(
                    "relationship label-column discovery failed: {error}"
                ))
            })?;
        let Some(row) = rows.into_iter().next() else {
            return Ok(None);
        };
        row.get::<String>(1).map_err(|error| {
            GraphError::Internal(format!("relationship label-column read failed: {error}"))
        })
    })
}

/// Discover all foreign key relationships in a schema.
fn discover_foreign_keys(schema_name: &str) -> GraphResult<Vec<DiscoveredFk>> {
    let fk_query = "SELECT
            c.conrelid::oid::integer AS from_oid,
            from_class.relname::text AS from_table,
            from_attr.attname::text AS from_column,
            c.confrelid::oid::integer AS to_oid,
            to_class.relname::text AS to_table,
            to_attr.attname::text AS to_column
         FROM pg_catalog.pg_constraint c
         JOIN pg_catalog.pg_class from_class ON from_class.oid = c.conrelid
         JOIN pg_catalog.pg_namespace from_namespace ON from_namespace.oid = from_class.relnamespace
         JOIN pg_catalog.pg_class to_class ON to_class.oid = c.confrelid
         JOIN pg_catalog.unnest(c.conkey) WITH ORDINALITY AS fk_from(attnum, n) ON true
         JOIN pg_catalog.unnest(c.confkey) WITH ORDINALITY AS fk_to(attnum, n) ON fk_to.n = fk_from.n
         JOIN pg_catalog.pg_attribute from_attr ON from_attr.attrelid = c.conrelid AND from_attr.attnum = fk_from.attnum
         JOIN pg_catalog.pg_attribute to_attr ON to_attr.attrelid = c.confrelid AND to_attr.attnum = fk_to.attnum
         JOIN pg_catalog.pg_index to_primary
           ON to_primary.indrelid = c.confrelid
          AND to_primary.indisprimary
          AND to_primary.indnkeyatts = 1
          AND to_attr.attnum = ANY(to_primary.indkey::smallint[])
         WHERE c.contype = 'f'
           AND pg_catalog.cardinality(c.conkey) = 1
           AND pg_catalog.cardinality(c.confkey) = 1
           AND from_namespace.nspname = $1
         ORDER BY c.conrelid, c.oid, fk_from.n";

    let mut fks = Vec::new();
    Spi::connect(|client| {
        let result = client
            .select(fk_query, None, &[schema_name.into()])
            .map_err(|e| GraphError::Internal(format!("FK discovery failed: {}", e)))?;

        for row in result {
            let from_oid = row
                .get::<i32>(1)
                .map_err(|e| GraphError::Internal(format!("Cannot read from_oid: {}", e)))?
                .ok_or_else(|| {
                    GraphError::Internal("FK discovery returned a NULL source OID".to_string())
                })?;
            let from_table: String = row
                .get::<String>(2)
                .map_err(|e| GraphError::Internal(format!("Cannot read from_table: {}", e)))?
                .unwrap_or_default();
            let from_column: String = row
                .get::<String>(3)
                .map_err(|e| GraphError::Internal(format!("Cannot read from_column: {}", e)))?
                .unwrap_or_default();
            let to_oid = row
                .get::<i32>(4)
                .map_err(|e| GraphError::Internal(format!("Cannot read to_oid: {}", e)))?
                .ok_or_else(|| {
                    GraphError::Internal("FK discovery returned a NULL target OID".to_string())
                })?;
            let to_table: String = row
                .get::<String>(5)
                .map_err(|e| GraphError::Internal(format!("Cannot read to_table: {}", e)))?
                .unwrap_or_default();
            let to_column: String = row
                .get::<String>(6)
                .map_err(|e| GraphError::Internal(format!("Cannot read to_column: {}", e)))?
                .unwrap_or_default();

            fks.push(DiscoveredFk {
                from_oid: Some(from_oid as u32),
                from_table,
                from_column,
                to_oid: Some(to_oid as u32),
                to_table,
                to_column,
            });
        }
        Ok::<(), GraphError>(())
    })?;

    Ok(fks)
}

fn validate_target_table_list(table_oids: &[u32]) -> GraphResult<()> {
    if table_oids.is_empty() {
        return Err(GraphError::InvalidFilter {
            reason: "auto_discover_tables() requires at least one table".to_string(),
        });
    }

    let mut seen = HashSet::with_capacity(table_oids.len());
    for oid in table_oids {
        if !seen.insert(*oid) {
            return Err(GraphError::InvalidFilter {
                reason: format!(
                    "auto_discover_tables() table list contains duplicate table {}",
                    regclass_text(*oid).unwrap_or_else(|_| format!("OID {}", oid))
                ),
            });
        }
        validate_supported_relation(*oid)?;
    }
    Ok(())
}

fn validate_supported_relation(table_oid: u32) -> GraphResult<()> {
    let relkind = Spi::connect(|client| {
        let table_oid = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(
                "SELECT c.relkind::text
                 FROM pg_catalog.pg_class c
                 WHERE c.oid = $1::oid",
                None,
                &[table_oid.into()],
            )
            .map_err(|err| GraphError::Internal(format!("relation validation failed: {}", err)))?;
        let row = result.first();
        row.get::<String>(1)
            .map_err(|err| GraphError::Internal(format!("relation kind read failed: {}", err)))?
            .ok_or_else(|| {
                GraphError::Internal(format!("relation OID {} does not exist", table_oid))
            })
    })?;

    match relkind.as_str() {
        "r" | "p" => Ok(()),
        "v" => Err(GraphError::InvalidFilter {
            reason: format!(
                "auto_discover_tables() does not support views: {}",
                regclass_text(table_oid).unwrap_or_else(|_| format!("OID {}", table_oid))
            ),
        }),
        "m" => Err(GraphError::InvalidFilter {
            reason: format!(
                "auto_discover_tables() does not support materialized views: {}",
                regclass_text(table_oid).unwrap_or_else(|_| format!("OID {}", table_oid))
            ),
        }),
        _ => Err(GraphError::InvalidFilter {
            reason: format!(
                "auto_discover_tables() only supports base tables: {}",
                regclass_text(table_oid).unwrap_or_else(|_| format!("OID {}", table_oid))
            ),
        }),
    }
}

fn discover_tables_by_oid(table_oids: &[u32]) -> GraphResult<Vec<DiscoveredTable>> {
    let mut tables = Vec::with_capacity(table_oids.len());
    for oid in table_oids {
        let (schema_name, table_name, id_is_primary, pk_columns) = discover_identifier(*oid)?;
        let text_columns = discover_text_columns(&schema_name, &table_name, &pk_columns)?;
        let relationship_label_column =
            discover_dynamic_relationship_label_column(&schema_name, &table_name)?;
        tables.push(DiscoveredTable {
            table_oid: Some(*oid),
            schema_name,
            table_name,
            pk_columns,
            id_is_primary,
            text_columns,
            relationship_label_column,
        });
    }
    Ok(tables)
}

fn discover_identifier(table_oid: u32) -> GraphResult<(String, String, bool, Vec<String>)> {
    Spi::connect(|client| {
        let query = "SELECT
                n.nspname::text,
                c.relname::text,
                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_class c
             JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
             JOIN pg_catalog.pg_index i ON i.indrelid = c.oid
             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 c.oid = $1::oid
               AND (i.indisprimary OR i.indisunique)
               AND i.indpred IS NULL
             GROUP BY n.nspname, c.relname, i.indexrelid, i.indisprimary
             HAVING bool_and(a.attnotnull)
             ORDER BY i.indisprimary DESC, i.indexrelid
             LIMIT 1";
        let table_oid_arg = pgrx::pg_sys::Oid::from_u32(table_oid);
        let result = client
            .select(query, None, &[table_oid_arg.into()])
            .map_err(|err| {
                GraphError::Internal(format!("targeted table discovery failed: {}", err))
            })?;
        let row = result.first();
        let Some(schema_name) = row
            .get::<String>(1)
            .map_err(|err| GraphError::Internal(format!("schema name read failed: {}", err)))?
        else {
            return Err(GraphError::InvalidFilter {
                reason: format!(
                    "table {} must have a primary key or unique NOT NULL key",
                    regclass_text(table_oid).unwrap_or_else(|_| format!("OID {}", table_oid))
                ),
            });
        };
        let table_name = row
            .get::<String>(2)
            .map_err(|err| GraphError::Internal(format!("table name read failed: {}", err)))?
            .unwrap_or_default();
        let id_is_primary = row
            .get::<bool>(3)
            .map_err(|err| GraphError::Internal(format!("identifier kind read failed: {}", err)))?
            .unwrap_or(false);
        let columns = row
            .get::<Vec<String>>(4)
            .map_err(|err| {
                GraphError::Internal(format!("identifier columns read failed: {}", err))
            })?
            .unwrap_or_default();
        Ok((schema_name, table_name, id_is_primary, columns))
    })
}

fn discover_foreign_keys_for_tables(table_oids: &[u32]) -> GraphResult<Vec<DiscoveredFk>> {
    let oid_list = table_oids
        .iter()
        .map(|oid| i64::from(*oid))
        .collect::<Vec<_>>();
    let query = "SELECT
            c.conrelid::oid::integer AS from_oid,
            from_class.relname::text AS from_table,
            from_attr.attname::text AS from_column,
            c.confrelid::oid::integer AS to_oid,
            to_class.relname::text AS to_table,
            to_attr.attname::text AS to_column
         FROM pg_catalog.pg_constraint c
         JOIN pg_catalog.pg_class from_class ON from_class.oid = c.conrelid
         JOIN pg_catalog.pg_class to_class ON to_class.oid = c.confrelid
         JOIN unnest(c.conkey) WITH ORDINALITY AS fk_from(attnum, n) ON true
         JOIN unnest(c.confkey) WITH ORDINALITY AS fk_to(attnum, n) ON fk_to.n = fk_from.n
         JOIN pg_catalog.pg_attribute from_attr ON from_attr.attrelid = c.conrelid AND from_attr.attnum = fk_from.attnum
         JOIN pg_catalog.pg_attribute to_attr ON to_attr.attrelid = c.confrelid AND to_attr.attnum = fk_to.attnum
         JOIN pg_catalog.pg_index to_primary
           ON to_primary.indrelid = c.confrelid
          AND to_primary.indisprimary
          AND to_primary.indnkeyatts = 1
          AND to_attr.attnum = ANY(to_primary.indkey::smallint[])
         WHERE c.contype = 'f'
           AND pg_catalog.cardinality(c.conkey) = 1
           AND pg_catalog.cardinality(c.confkey) = 1
           AND c.conrelid::bigint = ANY($1::int8[])
           AND c.confrelid::bigint = ANY($1::int8[])
         ORDER BY c.conrelid, c.oid, fk_from.n";

    let mut fks = Vec::new();
    Spi::connect(|client| {
        let result = client
            .select(query, None, &[oid_list.into()])
            .map_err(|err| {
                GraphError::Internal(format!("targeted FK discovery failed: {}", err))
            })?;
        for row in result {
            fks.push(DiscoveredFk {
                from_oid: Some(
                    row.get::<i32>(1)
                        .map_err(|err| {
                            GraphError::Internal(format!("from_oid read failed: {}", err))
                        })?
                        .unwrap_or_default() as u32,
                ),
                from_table: row
                    .get::<String>(2)
                    .map_err(|err| {
                        GraphError::Internal(format!("from_table read failed: {}", err))
                    })?
                    .unwrap_or_default(),
                from_column: row
                    .get::<String>(3)
                    .map_err(|err| {
                        GraphError::Internal(format!("from_column read failed: {}", err))
                    })?
                    .unwrap_or_default(),
                to_oid: Some(
                    row.get::<i32>(4)
                        .map_err(|err| {
                            GraphError::Internal(format!("to_oid read failed: {}", err))
                        })?
                        .unwrap_or_default() as u32,
                ),
                to_table: row
                    .get::<String>(5)
                    .map_err(|err| GraphError::Internal(format!("to_table read failed: {}", err)))?
                    .unwrap_or_default(),
                to_column: row
                    .get::<String>(6)
                    .map_err(|err| GraphError::Internal(format!("to_column read failed: {}", err)))?
                    .unwrap_or_default(),
            });
        }
        Ok::<(), GraphError>(())
    })?;
    Ok(fks)
}

fn registered_table(
    table: &DiscoveredTable,
    id_columns: PrimaryKeySpec,
    tenant_column: Option<&str>,
) -> GraphResult<RegisteredTable> {
    Ok(RegisteredTable {
        table_oid: required_table_oid(table)?,
        table_name: format!(
            "{}.{}",
            quote_ident(&table.schema_name),
            quote_ident(&table.table_name)
        ),
        id_columns,
        columns: PropertyColumns::from_columns(table.text_columns.clone()),
        tenant_column: tenant_column.map(ToString::to_string),
    })
}

fn discovery_details(
    key_label: &str,
    id_column: &str,
    text_columns: &[String],
    tenant_column: Option<&str>,
) -> String {
    let tenant = tenant_column
        .map(|column| format!(", tenant_column={}", column))
        .unwrap_or_default();
    format!(
        "{}={}, columns=[{}]{}",
        key_label,
        id_column,
        text_columns.join(", "),
        tenant
    )
}

fn edge_label(label_source_column: &str) -> String {
    label_source_column
        .trim_end_matches("_id")
        .trim_end_matches("_fk")
        .to_string()
}

fn registered_edge(
    from_table_oid: u32,
    from_table: String,
    from_column: &str,
    to_table_oid: u32,
    to_table: String,
    to_column: &str,
    label_source_column: &str,
) -> RegisteredEdge {
    RegisteredEdge {
        // Discovery output is registered before it is built; catalog insertion
        // assigns the durable mapping identity.
        mapping_id: 0,
        from_table_oid,
        from_table,
        from_column: from_column.to_string(),
        // Registration resolves and persists the authoritative PK tuple using
        // the relation OID. Discovery values are intentionally not consumed
        // directly by projection construction.
        source_key_columns: PrimaryKeySpec::from_columns(Vec::new()),
        to_table_oid,
        to_table,
        to_column: to_column.to_string(),
        label: edge_label(label_source_column),
        bidirectional: true,
        weight_column: None,
        label_column: None,
    }
}

fn outgoing_foreign_keys(table_oid: u32, foreign_keys: &[DiscoveredFk]) -> Vec<&DiscoveredFk> {
    let mut outgoing = foreign_keys
        .iter()
        .filter(|foreign_key| foreign_key.from_oid == Some(table_oid))
        .collect::<Vec<_>>();
    outgoing.sort_by_key(|foreign_key| endpoint_order(&foreign_key.from_column));
    outgoing
}

fn has_binary_endpoint_shape(table_oid: u32, foreign_keys: &[DiscoveredFk]) -> bool {
    let outgoing = outgoing_foreign_keys(table_oid, foreign_keys);
    outgoing.len() == 2 && outgoing[0].from_column != outgoing[1].from_column
}

fn endpoint_order(column: &str) -> u8 {
    match column.to_ascii_lowercase().as_str() {
        "source_id" | "from_id" | "start_id" | "subject_id" => 0,
        "target_id" | "to_id" | "end_id" | "object_id" => 2,
        _ => 1,
    }
}

fn dynamic_relationship_label_column(table: &DiscoveredTable) -> Option<&str> {
    table.relationship_label_column.as_deref()
}

fn relationship_discovery(table: &DiscoveredTable, label_column: Option<&str>) -> DiscoveryResult {
    let details = label_column.map_or_else(
        || {
            format!(
                "composite pk=[{}], all columns are FKs — registered as edges",
                table.pk_columns.join(", ")
            )
        },
        |column| {
            format!(
                "two foreign-key endpoints, dynamic label column={} — registered as an edge",
                column
            )
        },
    );
    DiscoveryResult {
        item_type: "junction".to_string(),
        item_name: table.table_name.clone(),
        details,
    }
}

fn register_relationship_edges(
    table: &DiscoveredTable,
    foreign_keys: &[DiscoveredFk],
    label_column: Option<&str>,
    edges: &mut Vec<RegisteredEdge>,
    discoveries: &mut Vec<DiscoveryResult>,
) -> GraphResult<()> {
    let table_oid = required_table_oid(table)?;
    let relationship_fks = outgoing_foreign_keys(table_oid, foreign_keys);
    let Some(source_fk) = relationship_fks.first() else {
        return Ok(());
    };

    for target_fk in relationship_fks.iter().skip(1) {
        let label = edge_label(&target_fk.from_column);
        let label_details = label_column
            .map(|column| format!(", label_column={column}"))
            .unwrap_or_default();
        discoveries.push(DiscoveryResult {
            item_type: "edge".to_string(),
            item_name: format!(
                "{}.{} → {}.{}",
                source_fk.from_table,
                source_fk.from_column,
                target_fk.to_table,
                target_fk.from_column
            ),
            details: format!("label={}, bidirectional=true{}", label, label_details),
        });
        let target_oid = required_fk_oid(target_fk.to_oid, "relationship target")?;
        let mut edge = registered_edge(
            table_oid,
            regclass_text(table_oid)?,
            &source_fk.from_column,
            target_oid,
            regclass_text(target_oid)?,
            &target_fk.from_column,
            &target_fk.from_column,
        );
        edge.label_column = label_column.map(ToString::to_string);
        edges.push(edge);
    }
    Ok(())
}

fn required_table_oid(table: &DiscoveredTable) -> GraphResult<u32> {
    table.table_oid.ok_or_else(|| {
        GraphError::Internal(format!(
            "OID-based discovery lost table OID for {}.{}",
            table.schema_name, table.table_name
        ))
    })
}

fn required_fk_oid(oid: Option<u32>, label: &str) -> GraphResult<u32> {
    oid.ok_or_else(|| {
        GraphError::Internal(format!(
            "OID-based discovery lost {} foreign-key OID",
            label
        ))
    })
}

/// Classify a composite-PK table as a junction table or a composite entity.
///
/// A junction table has ALL of its PK columns participating as FK source columns.
/// A composite entity has at least one PK column that is NOT a FK.
fn classify_as_junction(
    table_oid: Option<u32>,
    table_name: &str,
    pk_columns: &[String],
    schema_fks: &[DiscoveredFk],
) -> bool {
    // Collect all FK source columns for this specific table
    let fk_source_columns: Vec<&str> = schema_fks
        .iter()
        .filter(|fk| match (table_oid, fk.from_oid) {
            (Some(table_oid), Some(from_oid)) => table_oid == from_oid,
            _ => fk.from_table == table_name,
        })
        .map(|fk| fk.from_column.as_str())
        .collect();

    // Only a two-endpoint relationship has an unambiguous binary graph shape.
    fk_source_columns
        .iter()
        .copied()
        .collect::<HashSet<_>>()
        .len()
        == 2
        && pk_columns
            .iter()
            .all(|pk_col| fk_source_columns.contains(&pk_col.as_str()))
}

#[cfg(test)]
mod tests {
    //! Covers schema discovery classification, especially junction-table
    //! detection from primary-key and foreign-key relationships.

    use super::{
        classify_as_junction, dynamic_relationship_label_column, edge_label, outgoing_foreign_keys,
        registered_edge, DiscoveredFk, DiscoveredTable,
    };

    fn fk(from_table: &str, from_column: &str, to_table: &str, to_column: &str) -> DiscoveredFk {
        DiscoveredFk {
            from_oid: None,
            from_table: from_table.to_string(),
            from_column: from_column.to_string(),
            to_oid: None,
            to_table: to_table.to_string(),
            to_column: to_column.to_string(),
        }
    }

    #[test]
    fn classify_as_junction_true_when_all_pk_columns_are_fk_sources() {
        let pk_columns = vec!["user_id".to_string(), "group_id".to_string()];
        let schema_fks = vec![
            fk("user_groups", "user_id", "users", "id"),
            fk("user_groups", "group_id", "groups", "id"),
        ];

        assert!(classify_as_junction(
            None,
            "user_groups",
            &pk_columns,
            &schema_fks
        ));
    }

    #[test]
    fn classify_as_junction_false_when_any_pk_column_is_not_fk_source() {
        let pk_columns = vec!["order_id".to_string(), "line_number".to_string()];
        let schema_fks = vec![fk("order_lines", "order_id", "orders", "id")];

        assert!(!classify_as_junction(
            None,
            "order_lines",
            &pk_columns,
            &schema_fks
        ));
    }

    #[test]
    fn classify_as_junction_rejects_non_binary_endpoint_sets() {
        let pk_columns = vec![
            "first_id".to_string(),
            "second_id".to_string(),
            "third_id".to_string(),
        ];
        let schema_fks = vec![
            fk("memberships", "first_id", "nodes", "id"),
            fk("memberships", "second_id", "nodes", "id"),
            fk("memberships", "third_id", "nodes", "id"),
        ];

        assert!(!classify_as_junction(
            None,
            "memberships",
            &pk_columns,
            &schema_fks
        ));
    }

    #[test]
    fn classify_as_junction_ignores_foreign_keys_from_other_tables() {
        let pk_columns = vec!["a_id".to_string(), "b_id".to_string()];
        let schema_fks = vec![
            fk("other_table", "a_id", "a", "id"),
            fk("other_table", "b_id", "b", "id"),
        ];

        assert!(!classify_as_junction(
            None,
            "junction_table",
            &pk_columns,
            &schema_fks
        ));
    }

    #[test]
    fn classify_as_junction_prefers_oid_when_present() {
        let pk_columns = vec!["user_id".to_string(), "group_id".to_string()];
        let schema_fks = vec![
            DiscoveredFk {
                from_oid: Some(11),
                from_table: "user_groups".to_string(),
                from_column: "user_id".to_string(),
                to_oid: Some(1),
                to_table: "users".to_string(),
                to_column: "id".to_string(),
            },
            DiscoveredFk {
                from_oid: Some(11),
                from_table: "user_groups".to_string(),
                from_column: "group_id".to_string(),
                to_oid: Some(2),
                to_table: "groups".to_string(),
                to_column: "id".to_string(),
            },
            DiscoveredFk {
                from_oid: Some(22),
                from_table: "user_groups".to_string(),
                from_column: "other_id".to_string(),
                to_oid: Some(3),
                to_table: "other".to_string(),
                to_column: "id".to_string(),
            },
        ];

        assert!(classify_as_junction(
            Some(11),
            "user_groups",
            &pk_columns,
            &schema_fks
        ));
    }

    #[test]
    fn registered_edge_uses_shared_defaults_and_label_source_column() {
        let edge = registered_edge(
            11,
            "public.user_groups".to_string(),
            "user_id",
            12,
            "public.groups".to_string(),
            "group_id",
            "group_id",
        );

        assert_eq!(edge.from_table, "public.user_groups");
        assert_eq!(edge.from_column, "user_id");
        assert_eq!(edge.to_table, "public.groups");
        assert_eq!(edge.to_column, "group_id");
        assert_eq!(edge.label, "group");
        assert!(edge.bidirectional);
        assert_eq!(edge.weight_column, None);
        assert_eq!(edge.label_column, None);
    }

    #[test]
    fn typed_relationship_conventions_select_label_and_endpoint_direction() {
        let table = DiscoveredTable {
            table_oid: Some(11),
            schema_name: "public".to_string(),
            table_name: "relationships".to_string(),
            pk_columns: vec!["id".to_string()],
            id_is_primary: true,
            text_columns: vec!["note".to_string(), "relationship_name".to_string()],
            relationship_label_column: Some("relationship_name".to_string()),
        };
        let foreign_keys = vec![
            DiscoveredFk {
                from_oid: Some(11),
                from_table: "relationships".to_string(),
                from_column: "target_id".to_string(),
                to_oid: Some(12),
                to_table: "nodes".to_string(),
                to_column: "id".to_string(),
            },
            DiscoveredFk {
                from_oid: Some(11),
                from_table: "relationships".to_string(),
                from_column: "source_id".to_string(),
                to_oid: Some(12),
                to_table: "nodes".to_string(),
                to_column: "id".to_string(),
            },
        ];

        assert_eq!(
            dynamic_relationship_label_column(&table),
            Some("relationship_name")
        );
        let ordered = outgoing_foreign_keys(11, &foreign_keys);
        assert_eq!(ordered[0].from_column, "source_id");
        assert_eq!(ordered[1].from_column, "target_id");
    }

    #[test]
    fn edge_label_strips_supported_fk_suffixes() {
        assert_eq!(edge_label("user_id"), "user");
        assert_eq!(edge_label("account_fk"), "account");
        assert_eq!(edge_label("owner"), "owner");
    }
}
