postrust-graphql 2.4.1

//! GraphQL schema generation from PostgreSQL schema cache.
//!
//! Builds a dynamic GraphQL schema from the database schema cache,
//! creating query and mutation types for each table.

pub mod aggregate;
pub mod object;
pub mod relationship;

use crate::input::mutation::{is_deletable, is_insertable, is_updatable};
use crate::schema::object::TableObjectType;
use crate::schema::relationship::RelationshipField;
use postrust_core::schema_cache::{SchemaCache, Table};
use postrust_core::QualifiedIdentifier;
use std::collections::{HashMap, HashSet};

/// Configuration for schema generation.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct SchemaConfig {
    /// Schemas to expose in GraphQL (e.g., ["public"])
    pub exposed_schemas: Vec<String>,
    /// Whether to generate mutation types
    pub enable_mutations: bool,
    /// Whether to generate subscription types
    pub enable_subscriptions: bool,

    /// How often a live query re-reads itself when nothing has notified it.
    ///
    /// A subscription is woken by the trigger on the table it reads, which
    /// costs nothing while nothing is written. A trigger cannot see
    /// everything: a view has none, an embedded row may live in a table that
    /// carries none, and a predicate written against the clock changes with
    /// no write at all. This is how often those are noticed anyway. Seconds;
    /// zero turns the refresh off and leaves only the notifications.
    pub subscription_refresh_seconds: u64,
    /// The members of each enum table, keyed by `schema.table`, as
    /// `(value, comment)`.
    ///
    /// These are rows rather than schema, so they cannot come from the schema
    /// cache: they are read at startup, and again whenever the schema is
    /// rebuilt, for the tables the configuration marks. A table marked as an enum whose values are absent is exposed as
    /// an ordinary table -- an empty GraphQL enum is not a legal type, and
    /// refusing to start over a table with no rows in it would be worse.
    pub enum_values: HashMap<String, Vec<(String, Option<String>)>>,
    /// Names given rather than derived.
    ///
    /// Empty by default, which is every name derived from the schema exactly
    /// as before.
    pub names: crate::names::NameOverrides,
    /// Ceiling on rows a single query may return (`PGRST_MAX_ROWS`).
    ///
    /// Applied when a query supplies no `limit` of its own, and as an upper
    /// bound when it supplies a larger one. `None` leaves queries unbounded.
    pub max_rows: Option<i64>,

    /// Whether this schema is emitted as an Apollo Federation subgraph.
    pub enable_federation: bool,
    /// Namespace applied to generated table types and root fields.
    pub type_prefix: Option<String>,

    /// Whose schema this is.
    ///
    /// `None` is the unrestricted one: what an administrator sees, and what
    /// every caller sees on a server with no permission document. `Some(role)`
    /// is built from a cache already reduced to what that role may see, so the
    /// only thing left for the builders to ask about is the aggregate root.
    pub role: Option<String>,
}

impl Default for SchemaConfig {
    fn default() -> Self {
        Self {
            exposed_schemas: vec!["public".to_string()],
            enum_values: HashMap::new(),
            names: crate::names::NameOverrides::default(),
            enable_mutations: true,
            enable_subscriptions: false,
            subscription_refresh_seconds: 30,
            max_rows: None,
            enable_federation: false,
            type_prefix: None,
            role: None,
        }
    }
}

impl SchemaConfig {
    /// Create a new schema config.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set the exposed schemas.
    pub fn with_schemas(mut self, schemas: Vec<String>) -> Self {
        self.exposed_schemas = schemas;
        self
    }

    /// Enable or disable mutations.
    pub fn with_mutations(mut self, enable: bool) -> Self {
        self.enable_mutations = enable;
        self
    }

    /// Enable or disable subscriptions.
    pub fn with_subscriptions(mut self, enable: bool) -> Self {
        self.enable_subscriptions = enable;
        self
    }

    /// Whether a table keeps its unprefixed type identity across subgraphs.
    pub fn is_shared_entity(&self, table: &QualifiedIdentifier) -> bool {
        self.enable_federation && self.names.federation_shared(&table.schema, &table.name)
    }

    /// Name a table type, applying the configured namespace unless it is shared.
    pub fn table_type_name(&self, table: &QualifiedIdentifier, base_name: &str) -> String {
        match self.type_prefix.as_deref() {
            Some(prefix) if !prefix.is_empty() && !self.is_shared_entity(table) => {
                format!("{prefix}_{base_name}")
            }
            _ => base_name.to_string(),
        }
    }

    /// Namespace a root field so independently generated subgraphs do not collide.
    pub fn root_field_name(&self, base_name: &str) -> String {
        match self.type_prefix.as_deref() {
            Some(prefix) if !prefix.is_empty() => format!("{prefix}_{base_name}"),
            _ => base_name.to_string(),
        }
    }

    /// How often a live query re-reads itself when nothing has notified it.
    pub fn subscription_refresh(&self) -> std::time::Duration {
        match self.subscription_refresh_seconds {
            // A zero interval is not a tick every instant; it is no tick at
            // all, which leaves the notifications alone to wake it.
            0 => std::time::Duration::from_secs(60 * 60 * 24 * 365),
            seconds => std::time::Duration::from_secs(seconds),
        }
    }

    /// Check if a schema is exposed.
    pub fn is_schema_exposed(&self, schema: &str) -> bool {
        self.exposed_schemas.iter().any(|s| s == schema)
    }

    /// The schema whose tables get unqualified GraphQL names.
    ///
    /// This is the first exposed schema, mirroring how the REST surface treats
    /// the first entry of `PGRST_DB_SCHEMAS` as the default.
    pub fn default_schema(&self) -> &str {
        self.exposed_schemas
            .first()
            .map(|s| s.as_str())
            .unwrap_or("public")
    }
}

/// Primary key columns of a table, as `(column name, PostgreSQL type)`.
///
/// `nominal_type` (the underlying `udt_name`) is used because it is always a
/// castable type name.
/// Whether the whole primary key is present on this table.
///
/// A `_by_pk` field addresses a row by its key, so a key the caller cannot see
/// is a field it cannot use: `books_by_pk(id: 1, book_name: "...")` for a role
/// granted `book_name` and not `id` asks for a row by half a key. Hasura does
/// not publish the field at all, and neither does this -- which is also what
/// stops `pk_columns_of` falling back to `text` for a column it cannot find
/// and typing the argument `String`.
fn has_whole_key(table: &Table) -> bool {
    !table.pk_cols.is_empty()
        && table
            .pk_cols
            .iter()
            .all(|column| table.get_column(column).is_some())
}

fn pk_columns_of(table: &Table) -> Vec<(String, String)> {
    table
        .pk_cols
        .iter()
        .map(|col_name| {
            let pg_type = table
                .get_column(col_name)
                .map(|c| c.nominal_type.clone())
                .unwrap_or_else(|| "text".to_string());
            (col_name.clone(), pg_type)
        })
        .collect()
}

/// Every key a relationship may be named under, most specific first.
///
/// More than one, because there is more than one way to identify the same
/// relationship and a document may be written by hand or converted from
/// Hasura's metadata, which uses a different one. A constraint names exactly
/// one relationship even where two of them point at the same table; a computed
/// relationship has no constraint and is named by its function; and Hasura
/// names both directions by a column -- the foreign key column on this table
/// for a relationship to one row, and `table.column` on the far side for a
/// relationship to many. All of them are accepted, so a converted document
/// needs no database to turn a column into the constraint that carries it.
pub(crate) fn relationship_keys(rel: &postrust_core::schema_cache::Relationship) -> Vec<String> {
    use postrust_core::schema_cache::{Cardinality, Relationship};

    if let Relationship::Computed { function, .. } = rel {
        return vec![function.name.clone()];
    }

    let mut keys = Vec::new();
    // A many-to-many has no single constraint of its own: it is inferred from
    // the two constraints on its junction table. The junction is the stable,
    // unambiguous name clients already use as a PostgREST relationship hint,
    // and either constituent constraint is useful to hand-written metadata.
    if let Relationship::ForeignKey {
        cardinality: Cardinality::M2M(junction),
        ..
    } = rel
    {
        keys.push(junction.table.name.clone());
        keys.push(junction.constraint1.clone());
        keys.push(junction.constraint2.clone());
    }
    match rel.constraint_name() {
        "" => {}
        constraint => keys.push(constraint.to_string()),
    }
    // The column that carries the key, on whichever side carries it. A
    // one-to-one may have it on either, and two relationships between the same
    // pair of tables can share the local column while differing in the far one
    // -- so the more specific spelling is offered first and the bare column
    // last.
    if let Some(column) = rel.single_foreign_column() {
        keys.push(format!("{}.{}", rel.foreign_table().name, column));
    }
    if !rel.is_one_to_many() {
        if let Some(column) = rel.single_local_column() {
            keys.push(column.to_string());
        }
    }
    // A key over more than one column, spelled as the columns in order. A
    // composite foreign key has no single column to be named by, and Hasura
    // names one by its columns rather than by its constraint -- so a
    // relationship over `(author_id1, author_id2)` is unreachable under any
    // of the spellings above. Sorted, because the order two sides list the
    // same columns in is not something either of them promises.
    let joined = |mut columns: Vec<String>| {
        columns.sort();
        columns.join(",")
    };
    let columns = rel.join_columns();
    if columns.len() > 1 {
        keys.push(format!(
            "{}.{}",
            rel.foreign_table().name,
            joined(columns.iter().map(|(_, foreign)| foreign.clone()).collect())
        ));
        if !rel.is_one_to_many() {
            keys.push(joined(
                columns.iter().map(|(local, _)| local.clone()).collect(),
            ));
        }
    }
    keys
}

/// A type name as the catalogue writes it, reduced to the table it names.
///
/// `pg_catalog.format_type` quotes anything that needs quoting and qualifies
/// anything outside the search path, so a function returning rows of a table
/// called `user` -- a reserved word -- reports `"user"`, and one returning rows
/// of `other.thing` reports `other.thing`. Neither is the table's name, and
/// comparing them against one finds nothing.
fn type_name_of(rendered: &str) -> (Option<String>, String) {
    let (schema, name) = match rendered.rsplit_once('.') {
        Some((schema, name)) => (Some(schema.trim_matches('"').to_string()), name),
        None => (None, rendered),
    };
    (schema, name.trim_matches('"').to_string())
}

/// Base name used to derive a table's GraphQL type and field names.
///
/// Tables in the default schema keep their bare name, so a single-schema
/// deployment is unaffected. Tables in any other exposed schema are prefixed
/// with the schema, because GraphQL has one flat namespace: without this, a
/// `users` table in both `public` and `api` would generate the same type and
/// field names and one would silently replace the other.
fn base_name_for(table: &Table, config: &SchemaConfig) -> String {
    // A name that was given is the answer; the rest of this is what to call a
    // table nobody named.
    if let Some(given) = config.names.base_name(&table.schema, &table.name) {
        return given.to_string();
    }
    if table.schema == config.default_schema() {
        table.name.clone()
    } else {
        format!("{}_{}", table.schema, table.name)
    }
}

/// Represents a generated GraphQL schema.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct GeneratedSchema {
    /// Object types for each table
    pub object_types: HashMap<String, TableObjectType>,
    /// Query fields
    pub query_fields: Vec<QueryField>,
    /// Mutation fields (if enabled)
    pub mutation_fields: Vec<MutationField>,
    /// Relationship fields for each type
    pub relationship_fields: HashMap<String, Vec<RelationshipField>>,
    /// GraphQL enums generated from enum tables: type name to
    /// `(member, description)`.
    pub enum_types: HashMap<String, Vec<(String, Option<String>)>>,
    /// Database functions exposed as root fields.
    pub function_fields: Vec<FunctionField>,
    /// Apollo Federation entities keyed by GraphQL type name.
    pub federation_entities: HashMap<String, FederationEntity>,
    /// Whether Apollo Federation system fields should be exposed.
    pub enable_federation: bool,
}

/// A table exposed as an Apollo Federation entity.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct FederationEntity {
    /// GraphQL type name of the entity.
    pub type_name: String,
    /// Table this entity resolves to.
    pub table: QualifiedIdentifier,
    /// Primary key field names as exposed in GraphQL.
    pub key_fields: Vec<String>,
    /// Whether the type is configured as shared across subgraphs.
    pub shared: bool,
}

impl FederationEntity {
    /// The `fields` argument used by Apollo Federation's `@key` directive.
    pub fn key_fields_sdl(&self) -> String {
        self.key_fields.join(" ")
    }
}

/// A database function returning rows of a table, exposed as a root field.
///
/// A function that returns `SETOF <table>` answers the same question a table
/// does, from a query somebody wrote in SQL rather than from the table
/// directly. It filters, orders and pages like the table it returns, and its
/// own arguments arrive under `args` so they cannot collide with those.
///
/// Where it appears follows from what PostgreSQL says it does: a function
/// declared VOLATILE may write, so it is a mutation; one declared STABLE or
/// IMMUTABLE may not, so it is a query. Nothing else is a safe place to draw
/// that line -- the alternative is trusting a name.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct FunctionField {
    /// The field's name, which is the function's.
    pub name: String,
    /// Schema the function lives in.
    pub schema_name: String,
    /// The function's own name, which the field is named after.
    pub function_name: String,
    /// The GraphQL type of the rows it returns.
    pub returns: String,
    /// Base name for helper types owned by the returned rows.
    pub local_returns: String,
    /// The table those rows belong to, as `(schema, table)`.
    pub returns_table: (String, String),
    /// Its arguments, as `(name, PostgreSQL type, required)`. The session
    /// argument is not among them: the client does not supply it.
    pub arguments: Vec<(String, String, bool)>,
    /// The name of its session argument, if it has one.
    ///
    /// Hasura's convention: a `hasura_session json` parameter is filled from
    /// the caller's session variables rather than taken from the request, so a
    /// function can read who is asking. Exposing it as an argument would let
    /// the caller name its own identity, which is the opposite of the point.
    pub session_argument: Option<String>,
    /// Whether it may write, and so belongs on the mutation root.
    pub volatile: bool,
    /// Description from the function's comment.
    pub description: Option<String>,
}

impl FunctionField {
    /// The argument list, as PostgreSQL would write it -- `a integer, b text`.
    ///
    /// Two overloads differ only here, so this is what distinguishes them in a
    /// warning and what orders them against each other.
    pub fn argument_signature(&self) -> String {
        self.arguments
            .iter()
            .map(|(name, sql_type, _)| format!("{name} {sql_type}"))
            .collect::<Vec<_>>()
            .join(", ")
    }
}

/// What a computed relationship's function takes from the caller.
///
/// Everything but the parent row, which is what makes it a relationship, and
/// the session, which is the server's to supply. Read from the routine rather
/// than from the relationship because the relationship loader knows only which
/// argument is the row -- the rest are the same parameters any function has,
/// and they are already loaded.
/// One relationship field per foreign key the catalogue carries.
fn reflected_fields(
    table: &Table,
    schema_cache: &SchemaCache,
    config: &SchemaConfig,
    base_names: &HashMap<(String, String), String>,
    type_names: &HashMap<(String, String), String>,
    local_type_names: &HashMap<(String, String), String>,
) -> Vec<RelationshipField> {
    schema_cache
        .get_relationships(&table.qualified_identifier(), &table.schema)
        .map(|relationships| {
            relationships
                .iter()
                .filter_map(|rel| {
                    // A relationship whose target is not exposed would
                    // reference a GraphQL type that was never registered.
                    let foreign = rel.foreign_table();
                    let target_key = (foreign.schema.clone(), foreign.name.clone());
                    let target_base = base_names.get(&target_key)?;
                    let target_type = type_names.get(&target_key)?;
                    let target_local_type = local_type_names.get(&target_key)?;
                    let mut field = RelationshipField::from_relationship_typed(
                        rel,
                        target_base,
                        target_type,
                        target_local_type,
                        relationship_keys(rel).iter().find_map(|key| {
                            config.names.relationship(&table.schema, &table.name, key)
                        }),
                    );
                    field.arguments = caller_arguments(rel, schema_cache);
                    Some(field)
                })
                .collect()
        })
        .unwrap_or_default()
}

/// One relationship field per relationship metadata declares, and no others.
///
/// A declaration reached by a key names a relationship reflection already
/// found, so the catalogue's own cardinality and columns are used -- a key
/// says which side has one row and which has many, and guessing that from a
/// name would be guessing. One reached by a column mapping is not in the
/// catalogue at all and is built from the mapping, with the direction taken
/// from the declaration: Hasura writes object and array relationships
/// separately, which is where `to_one` comes from.
///
/// A declaration this server cannot resolve is left out rather than guessed
/// at. It is a field that would not work; the ones beside it still do.
fn declared_fields(
    declared: &[crate::names::DeclaredRelationship],
    table: &Table,
    schema_cache: &SchemaCache,
    config: &SchemaConfig,
    base_names: &HashMap<(String, String), String>,
    type_names: &HashMap<(String, String), String>,
    local_type_names: &HashMap<(String, String), String>,
) -> Vec<RelationshipField> {
    use postrust_core::schema_cache::Relationship;

    let reflected = schema_cache
        .get_relationships(&table.qualified_identifier(), &table.schema)
        .map(|found| found.to_vec())
        .unwrap_or_default();

    let mut fields = Vec::with_capacity(declared.len());
    for entry in declared {
        // Reached by a key reflection already found.
        if let Some(key) = entry.using.as_deref() {
            let Some(rel) = reflected
                .iter()
                .find(|rel| relationship_keys(rel).iter().any(|found| found == key))
            else {
                tracing::warn!(
                    "GraphQL: {}.{} declares the relationship \"{}\" through \"{}\", \
                     which no foreign key carries; it is left out",
                    table.schema,
                    table.name,
                    entry.name,
                    key
                );
                continue;
            };
            let foreign = rel.foreign_table();
            let target_key = (foreign.schema.clone(), foreign.name.clone());
            let (Some(target_base), Some(target_type), Some(target_local_type)) = (
                base_names.get(&target_key),
                type_names.get(&target_key),
                local_type_names.get(&target_key),
            ) else {
                continue;
            };
            let mut field = RelationshipField::from_relationship_typed(
                rel,
                target_base,
                target_type,
                target_local_type,
                Some(&entry.name),
            );
            field.arguments = caller_arguments(rel, schema_cache);
            fields.push(field);
            continue;
        }

        // Reached by a column mapping, which is a join no key describes.
        let Some((schema, name)) = entry.target(config.default_schema()) else {
            tracing::warn!(
                "GraphQL: {}.{} declares the relationship \"{}\" with neither a key \
                 nor a table to join to; it is left out",
                table.schema,
                table.name,
                entry.name
            );
            continue;
        };
        let target_key = (schema, name);
        let (Some(target_base), Some(target_type), Some(target_local_type)) = (
            base_names.get(&target_key),
            type_names.get(&target_key),
            local_type_names.get(&target_key),
        ) else {
            continue;
        };
        match mapped_relationship_field(
            entry,
            table,
            schema_cache,
            config.default_schema(),
            target_base,
            target_type,
            target_local_type,
        ) {
            Some(field) => fields.push(field),
            None => tracing::warn!(
                "GraphQL: {}.{} declares the relationship \"{}\" with no columns to \
                 join on; it is left out",
                table.schema,
                table.name,
                entry.name
            ),
        }
    }

    // A computed relationship is declared by `add_computed_field`, not by
    // `create_object_relationship`, so it is not in this list and is not what
    // the list is exhaustive about. Leaving it to reflection is what keeps
    // `author.get_articles` on a table whose foreign keys are all declared.
    let already: std::collections::HashSet<String> =
        fields.iter().map(|field| field.name.clone()).collect();
    for rel in reflected
        .iter()
        .filter(|rel| matches!(rel, Relationship::Computed { .. }))
    {
        let foreign = rel.foreign_table();
        let target_key = (foreign.schema.clone(), foreign.name.clone());
        let (Some(target_base), Some(target_type), Some(target_local_type)) = (
            base_names.get(&target_key),
            type_names.get(&target_key),
            local_type_names.get(&target_key),
        ) else {
            continue;
        };
        let mut field = RelationshipField::from_relationship_typed(
            rel,
            target_base,
            target_type,
            target_local_type,
            relationship_keys(rel)
                .iter()
                .find_map(|key| config.names.relationship(&table.schema, &table.name, key)),
        );
        if already.contains(&field.name) {
            continue;
        }
        field.arguments = caller_arguments(rel, schema_cache);
        fields.push(field);
    }
    fields
}

/// One relationship field for a declaration that maps column to column.
///
/// The join is the declaration's, the direction is the declaration's, and the
/// constraint name is this server's -- nothing reads it back, and a mapped
/// join has no constraint to borrow. `None` where the declaration names no
/// columns to join on, which is not a join.
pub(crate) fn mapped_relationship_field(
    entry: &crate::names::DeclaredRelationship,
    table: &Table,
    schema_cache: &SchemaCache,
    default_schema: &str,
    target_base: &str,
    target_type: &str,
    target_local_type: &str,
) -> Option<RelationshipField> {
    use postrust_core::schema_cache::{Cardinality, Relationship};

    let (schema, name) = entry.target(default_schema)?;
    if entry.columns.is_empty() {
        return None;
    }
    let foreign_table = postrust_core::api_request::QualifiedIdentifier::new(&schema, &name);
    let here = table.qualified_identifier();
    let columns: Vec<(String, String)> = entry
        .columns
        .iter()
        .map(|(mine, theirs)| (mine.clone(), theirs.clone()))
        .collect();
    let constraint = format!("{}.{}", table.name, entry.name);
    // Which side holds the key, which is what a nested insert needs and a
    // column mapping does not say. `M2O` is "this row's column carries their
    // key", so the related row is written first; `O2O { is_parent: true }` is
    // the other way round, and is how `after_parent` is spelled here. A
    // relationship to many is written after its parent either way.
    let cardinality = match (entry.to_one, entry.after_parent) {
        (true, false) => Cardinality::M2O {
            constraint: constraint.clone(),
            columns,
        },
        (true, true) => Cardinality::O2O {
            constraint: constraint.clone(),
            columns,
            is_parent: true,
        },
        (false, _) => Cardinality::O2M {
            constraint: constraint.clone(),
            columns,
        },
    };
    let rel = Relationship::ForeignKey {
        is_self: here == foreign_table,
        table: here,
        foreign_table_is_view: schema_cache
            .get_table(&foreign_table)
            .is_some_and(|t| t.is_view),
        foreign_table,
        cardinality,
        table_is_view: table.is_view,
        constraint_name: constraint,
    };
    Some(RelationshipField::from_relationship_typed(
        &rel,
        target_base,
        target_type,
        target_local_type,
        Some(&entry.name),
    ))
}

fn caller_arguments(
    relationship: &postrust_core::schema_cache::Relationship,
    schema_cache: &SchemaCache,
) -> Vec<(String, String, bool)> {
    use postrust_core::schema_cache::Relationship;
    let Relationship::Computed {
        function,
        row_argument,
        ..
    } = relationship
    else {
        return Vec::new();
    };
    let Some(row_argument) = row_argument else {
        return Vec::new();
    };
    schema_cache
        .routines
        .get(function)
        .into_iter()
        .flatten()
        .find(|routine| routine.name == function.name)
        .map(|routine| {
            routine
                .params
                .iter()
                .filter(|param| &param.name != row_argument)
                .filter(|param| !is_session_argument(param))
                .filter(|param| !param.name.is_empty())
                .map(|param| (param.name.clone(), param.param_type.clone(), param.required))
                .collect()
        })
        .unwrap_or_default()
}

/// What a computed *column*'s function takes from the caller.
///
/// The same rule as [`caller_arguments`], one field kind along: everything but
/// the row, which is what makes it a field of this table, and the session,
/// which is the server's to supply.
pub(crate) fn computed_caller_arguments(
    computed: &postrust_core::schema_cache::ComputedColumn,
    schema_cache: &SchemaCache,
) -> Vec<(String, String, bool)> {
    let Some(row_argument) = &computed.row_argument else {
        return Vec::new();
    };
    schema_cache
        .routines
        .get(&computed.function)
        .into_iter()
        .flatten()
        .find(|routine| routine.name == computed.function.name)
        .map(|routine| {
            routine
                .params
                .iter()
                .filter(|param| &param.name != row_argument)
                .filter(|param| !is_session_argument(param))
                .filter(|param| !param.name.is_empty())
                .map(|param| (param.name.clone(), param.param_type.clone(), param.required))
                .collect()
        })
        .unwrap_or_default()
}

/// Whether a parameter is the one Hasura fills from the session.
///
/// Recognised by name and type, which is the only thing the database records:
/// Hasura writes `session_argument: hasura_session` into metadata, and every
/// function that has one calls it that.
fn is_session_argument(param: &postrust_core::schema_cache::RoutineParam) -> bool {
    param.name == "hasura_session" && matches!(param.param_type.as_str(), "json" | "jsonb")
}

impl GeneratedSchema {
    /// Get an object type by name.
    pub fn get_object_type(&self, name: &str) -> Option<&TableObjectType> {
        self.object_types.get(name)
    }

    /// Get query fields for a table.
    pub fn get_query_field(&self, table_name: &str) -> Option<&QueryField> {
        self.query_fields
            .iter()
            .find(|f| f.table_name == table_name)
    }

    /// Get mutation fields for a table.
    pub fn get_mutation_fields(&self, table_name: &str) -> Vec<&MutationField> {
        self.mutation_fields
            .iter()
            .filter(|f| f.table_name == table_name)
            .collect()
    }

    /// Get relationship fields for a type.
    pub fn get_relationship_fields(&self, type_name: &str) -> Option<&Vec<RelationshipField>> {
        self.relationship_fields.get(type_name)
    }

    /// Get federation entity metadata for a type.
    pub fn get_federation_entity(&self, type_name: &str) -> Option<&FederationEntity> {
        self.federation_entities.get(type_name)
    }

    /// Get all table names.
    pub fn table_names(&self) -> Vec<&str> {
        self.object_types
            .values()
            .map(|t| t.table.name.as_str())
            .collect()
    }

    /// Get all type names.
    pub fn type_names(&self) -> Vec<&str> {
        self.object_types.keys().map(|s| s.as_str()).collect()
    }
}

/// A query field for a table (e.g., users, userByPk).
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct QueryField {
    /// Field name (e.g., "users")
    pub name: String,
    /// Table name
    pub table_name: String,
    /// Schema the table lives in
    pub schema_name: String,
    /// GraphQL object type name (e.g., "Users")
    pub type_name: String,
    /// Base name for helper types owned by this table.
    pub local_type_name: String,
    /// GraphQL return type
    pub return_type: String,
    /// Whether this returns a list
    pub is_list: bool,
    /// Whether this is a "by PK" query
    pub is_by_pk: bool,
    /// Primary key columns, as `(column name, PostgreSQL type)`.
    ///
    /// Populated for by-PK queries so the resolver can filter on the table's
    /// actual key rather than assuming a column called `id`. Empty for list
    /// queries.
    pub pk_columns: Vec<(String, String)>,
    /// Field description
    pub description: Option<String>,
    /// The name given to this table's aggregate root, if one was.
    ///
    /// The aggregate field is built beside the list field rather than being a
    /// `QueryField` of its own, so the name it was given rides along with the
    /// list field that spawns it.
    pub aggregate_name: Option<String>,
    /// The description given to that aggregate root, if one was. `Some("")`
    /// means metadata said it has none.
    pub aggregate_description: Option<String>,
    /// Whether the aggregate root is built at all.
    ///
    /// The one thing a role's view of the schema cache cannot say. Dropping a
    /// table or a column is a fact about the database as this role sees it;
    /// whether that role may *count* the rows it can already read is a fact
    /// about the permission, and Hasura keeps the two apart because counting
    /// rows you cannot see is a way of seeing them. True unless a select
    /// permission says otherwise, which is what leaves an unconfigured server
    /// exactly as it was.
    pub aggregates: bool,
    /// Whether the rows themselves can be asked for.
    ///
    /// False for a table a role may count and not read: Hasura's way of
    /// granting "how many" without granting "which" is a select permission
    /// naming no columns, and such a table has an aggregate root and no row
    /// type to hang a list root on. The field is still carried so that the
    /// aggregate root beside it is built.
    pub rows: bool,
}

impl QueryField {
    /// Create a list query field (e.g., users), named after the table.
    pub fn list(table: &Table) -> Self {
        Self::list_named(table, &table.name)
    }

    /// Create a list query field using an explicit base name.
    pub fn list_named(table: &Table, base_name: &str) -> Self {
        Self::list_typed(table, base_name, base_name, base_name)
    }

    /// Create a list query field with separate root, row type and helper names.
    pub fn list_typed(
        table: &Table,
        root_base_name: &str,
        type_name: &str,
        local_type_name: &str,
    ) -> Self {
        let name = root_base_name.to_string();

        Self {
            name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            return_type: format!("[{}!]!", type_name),
            is_list: true,
            is_by_pk: false,
            pk_columns: Vec::new(),
            description: Some(format!("fetch data from the table: \"{}\"", table.name)),
            aggregate_name: None,
            aggregate_description: None,
            aggregates: true,
            rows: true,
        }
    }

    /// Create a by-PK query field (e.g., userByPk), named after the table.
    pub fn by_pk(table: &Table) -> Option<Self> {
        Self::by_pk_named(table, &table.name)
    }

    /// Create a by-PK query field using an explicit base name.
    pub fn by_pk_named(table: &Table, base_name: &str) -> Option<Self> {
        Self::by_pk_typed(table, base_name, base_name, base_name)
    }

    /// Create a by-PK query field with separate root, row type and helper names.
    pub fn by_pk_typed(
        table: &Table,
        root_base_name: &str,
        type_name: &str,
        local_type_name: &str,
    ) -> Option<Self> {
        if !has_whole_key(table) {
            return None;
        }

        let field_name = format!("{}_by_pk", root_base_name);

        // Carry the key columns and their types so the resolver can filter on
        // the real primary key.
        let pk_columns = pk_columns_of(table);

        Some(Self {
            name: field_name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            return_type: type_name.to_string(),
            is_list: false,
            is_by_pk: true,
            pk_columns,
            description: Some(format!(
                "fetch data from the table: \"{}\" using primary key columns",
                table.name
            )),
            aggregate_name: None,
            aggregate_description: None,
            aggregates: true,
            rows: true,
        })
    }
}

/// A mutation field for a table.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct MutationField {
    /// Field name (e.g., "insertUsers")
    pub name: String,
    /// Table name
    pub table_name: String,
    /// Schema the table lives in
    pub schema_name: String,
    /// Mutation type
    pub mutation_type: MutationType,
    /// Primary key columns, as `(column name, PostgreSQL type)`.
    ///
    /// Populated for by-PK mutations so the resolver can target the row by its
    /// key. Empty for bulk mutations.
    pub pk_columns: Vec<(String, String)>,
    /// GraphQL return type
    pub return_type: String,
    /// GraphQL row object type name.
    pub type_name: String,
    /// Base name for helper types owned by this table.
    pub local_type_name: String,
    /// Field description
    pub description: Option<String>,
}

/// Types of mutations.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MutationType {
    /// Insert multiple records
    Insert,
    /// Insert a single record
    InsertOne,
    /// Update records matching a filter
    Update,
    /// Update a single record by PK
    UpdateByPk,
    /// Several updates, each with its own filter, in one transaction
    UpdateMany,
    /// Delete records matching a filter
    Delete,
    /// Delete a single record by PK
    DeleteByPk,
}

impl MutationField {
    /// Create insert mutation fields for a table.
    pub fn insert_fields(table: &Table) -> Vec<Self> {
        Self::insert_fields_named(table, &table.name)
    }

    /// As [`Self::insert_fields`], with an explicit base name for the generated
    /// field and type names.
    pub fn insert_fields_named(table: &Table, base_name: &str) -> Vec<Self> {
        Self::insert_fields_typed(table, base_name, base_name)
    }

    /// Create insert mutation fields using separate helper and row type names.
    pub fn insert_fields_typed(table: &Table, local_type_name: &str, type_name: &str) -> Vec<Self> {
        if !is_insertable(table) {
            return vec![];
        }

        let mut fields = vec![];

        // insert_users (batch insert)
        let name = format!("insert_{}", local_type_name);
        fields.push(Self {
            name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            mutation_type: MutationType::Insert,
            pk_columns: Vec::new(),
            return_type: format!("{}_mutation_response", local_type_name),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            description: Some(format!("insert data into the table: \"{}\"", table.name)),
        });

        // insert_user_one (single insert)
        let name = format!("insert_{}_one", local_type_name);
        fields.push(Self {
            name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            mutation_type: MutationType::InsertOne,
            pk_columns: Vec::new(),
            return_type: type_name.to_string(),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            description: Some(format!(
                "insert a single row into the table: \"{}\"",
                table.name
            )),
        });

        fields
    }

    /// Create update mutation fields for a table.
    pub fn update_fields(table: &Table) -> Vec<Self> {
        Self::update_fields_named(table, &table.name)
    }

    /// As [`Self::update_fields`], with an explicit base name for the generated
    /// field and type names.
    pub fn update_fields_named(table: &Table, base_name: &str) -> Vec<Self> {
        Self::update_fields_typed(table, base_name, base_name)
    }

    /// Create update mutation fields using separate helper and row type names.
    pub fn update_fields_typed(table: &Table, local_type_name: &str, type_name: &str) -> Vec<Self> {
        if !is_updatable(table) {
            return vec![];
        }

        let mut fields = vec![];

        // update_users (batch update)
        let name = format!("update_{}", local_type_name);
        fields.push(Self {
            name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            mutation_type: MutationType::Update,
            pk_columns: Vec::new(),
            return_type: format!("{}_mutation_response", local_type_name),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            description: Some(format!("update data of the table: \"{}\"", table.name)),
        });

        // update_users_many (several filters, each with its own values)
        let name = format!("update_{}_many", local_type_name);
        fields.push(Self {
            name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            mutation_type: MutationType::UpdateMany,
            pk_columns: Vec::new(),
            return_type: format!("[{}_mutation_response]", local_type_name),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            description: Some(format!(
                "update multiples rows of table: \"{}\"",
                table.name
            )),
        });

        // update_user_by_pk (single update by PK)
        if has_whole_key(table) {
            let name = format!("update_{}_by_pk", local_type_name);
            fields.push(Self {
                name,
                table_name: table.name.clone(),
                schema_name: table.schema.clone(),
                mutation_type: MutationType::UpdateByPk,
                pk_columns: pk_columns_of(table),
                return_type: type_name.to_string(),
                type_name: type_name.to_string(),
                local_type_name: local_type_name.to_string(),
                description: Some(format!(
                    "update single row of the table: \"{}\"",
                    table.name
                )),
            });
        }

        fields
    }

    /// Create delete mutation fields for a table.
    pub fn delete_fields(table: &Table) -> Vec<Self> {
        Self::delete_fields_named(table, &table.name)
    }

    /// As [`Self::delete_fields`], with an explicit base name for the generated
    /// field and type names.
    pub fn delete_fields_named(table: &Table, base_name: &str) -> Vec<Self> {
        Self::delete_fields_typed(table, base_name, base_name)
    }

    /// Create delete mutation fields using separate helper and row type names.
    pub fn delete_fields_typed(table: &Table, local_type_name: &str, type_name: &str) -> Vec<Self> {
        if !is_deletable(table) {
            return vec![];
        }

        let mut fields = vec![];

        // delete_users (batch delete)
        let name = format!("delete_{}", local_type_name);
        fields.push(Self {
            name,
            table_name: table.name.clone(),
            schema_name: table.schema.clone(),
            mutation_type: MutationType::Delete,
            pk_columns: Vec::new(),
            return_type: format!("{}_mutation_response", local_type_name),
            type_name: type_name.to_string(),
            local_type_name: local_type_name.to_string(),
            description: Some(format!("delete data from the table: \"{}\"", table.name)),
        });

        // delete_user_by_pk (single delete by PK)
        if has_whole_key(table) {
            let name = format!("delete_{}_by_pk", local_type_name);
            fields.push(Self {
                name,
                table_name: table.name.clone(),
                schema_name: table.schema.clone(),
                mutation_type: MutationType::DeleteByPk,
                pk_columns: pk_columns_of(table),
                return_type: type_name.to_string(),
                type_name: type_name.to_string(),
                local_type_name: local_type_name.to_string(),
                description: Some(format!(
                    "delete single row from the table: \"{}\"",
                    table.name
                )),
            });
        }

        fields
    }
}

/// Build a GraphQL schema from a schema cache.
pub fn build_schema(schema_cache: &SchemaCache, config: &SchemaConfig) -> GeneratedSchema {
    let mut object_types = HashMap::new();
    let mut query_fields = Vec::new();
    let mut mutation_fields = Vec::new();
    let mut relationship_fields = HashMap::new();
    let mut federation_entities = HashMap::new();

    // Tables are visited in a stable order: the cache is a hash map, and any
    // name disambiguation below must not shift between restarts.
    let mut tables: Vec<&Table> = schema_cache
        .tables
        .values()
        .filter(|table| config.is_schema_exposed(&table.schema))
        .collect();
    tables.sort_by(|a, b| (&a.schema, &a.name).cmp(&(&b.schema, &b.name)));

    // Base names already carry the schema for non-default schemas, so a clash
    // here needs contrived naming (a `public.api_users` table alongside
    // `api.users`). Resolve it with a numeric suffix rather than letting one
    // table overwrite the other.
    let mut used_base_names: HashMap<String, u32> = HashMap::new();
    // (schema, table) -> resolved base name, needed when naming relationship
    // targets.
    let mut base_names: HashMap<(String, String), String> = HashMap::new();

    for table in &tables {
        let preferred = base_name_for(table, config);
        let base_name = match used_base_names.get_mut(&preferred) {
            None => {
                used_base_names.insert(preferred.clone(), 1);
                preferred
            }
            Some(count) => {
                *count += 1;
                let disambiguated = format!("{}_{}", preferred, count);
                tracing::warn!(
                    "GraphQL name collision: {}.{} would generate the same names as \
                     an earlier table; exposing it as \"{}\" instead",
                    table.schema,
                    table.name,
                    disambiguated
                );
                disambiguated
            }
        };

        base_names.insert(
            (table.schema.clone(), table.name.clone()),
            base_name.clone(),
        );
    }

    // Object and derived input types share one GraphQL namespace. Resolve the
    // prefixed names independently from field bases because shared entities
    // keep their type identity while their root fields remain namespaced.
    let mut used_type_names: HashMap<String, u32> = HashMap::new();
    let mut type_names: HashMap<(String, String), String> = HashMap::new();
    let mut local_type_names: HashMap<(String, String), String> = HashMap::new();
    for table in &tables {
        let key = (table.schema.clone(), table.name.clone());
        let base_name = base_names
            .get(&key)
            .expect("every visited table has a resolved base name");
        let preferred = config.table_type_name(&table.qualified_identifier(), base_name);
        let type_name = match used_type_names.get_mut(&preferred) {
            None => {
                used_type_names.insert(preferred.clone(), 1);
                preferred
            }
            Some(count) => {
                *count += 1;
                let disambiguated = format!("{}_{}", preferred, count);
                tracing::warn!(
                    "GraphQL type name collision: {}.{} would generate the same type as \
                     an earlier table; exposing it as \"{}\" instead",
                    table.schema,
                    table.name,
                    disambiguated
                );
                disambiguated
            }
        };
        let local_type_name = config.root_field_name(base_name);
        local_type_names.insert(key.clone(), local_type_name);
        type_names.insert(key, type_name);
    }

    // The tables this role may read. Needed after the loop, by
    // anything that would answer with a row of one.
    let mut readable_tables: HashSet<QualifiedIdentifier> = HashSet::new();

    for table in &tables {
        let base_name = base_names
            .get(&(table.schema.clone(), table.name.clone()))
            .expect("every visited table has a resolved base name")
            .clone();
        let type_name = type_names
            .get(&(table.schema.clone(), table.name.clone()))
            .expect("every visited table has a resolved type name")
            .clone();
        let local_type_name = local_type_names
            .get(&(table.schema.clone(), table.name.clone()))
            .expect("every visited table has a resolved local type name")
            .clone();

        // Create object type
        let mut obj_type = TableObjectType::from_table_named(table, &type_name, &config.names);
        obj_type.local_name = local_type_name.clone();
        // Whether this role can read the table at all. A table it may only
        // write has no GraphQL type -- a type with no fields is not a legal
        // one -- so nothing that would return that type is generated: no query
        // root, no `_by_pk`, no `insert_one`. The bulk write stays, and
        // answers with the count alone.
        //
        // Asked of the permission rather than of the fields, because the
        // fields are not split into the readable and the writable half until
        // every rewrite below has been applied to both.
        let readable = config.role.as_deref().is_none_or(|role| {
            config
                .names
                .permissions(&table.schema, &table.name, role)
                .is_none_or(|granted| crate::role::reads_anything(granted, table))
        });
        if readable {
            readable_tables.insert(table.qualified_identifier());
        }
        // What each computed field takes from the caller. Read from the
        // routine, for the reason `caller_arguments` gives.
        for field in &mut obj_type.fields {
            let Some(function) = config
                .names
                .computed_source(&table.schema, &table.name, &field.name)
                .or(Some(field.name.as_str()))
            else {
                continue;
            };
            let Some(computed) = table.get_computed_column(function) else {
                continue;
            };
            field.arguments = computed_caller_arguments(computed, schema_cache);
        }
        // Add query fields. Each root may be named separately -- Hasura names
        // them one at a time, and a set that does not agree on a base name has
        // nowhere else to be written down.
        // A root may be given a name, a description, or both. An empty
        // description is one that was given and is empty: metadata said the
        // field has none.
        let rename = |field: &mut QueryField, kind: &str| {
            if let Some(given) = config.names.root(&table.schema, &table.name, kind) {
                field.name = given.to_string();
            }
            if let Some(comment) = config.names.root_comment(&table.schema, &table.name, kind) {
                field.description = Some(comment.to_string()).filter(|c| !c.is_empty());
            }
        };
        let mut list = QueryField::list_typed(table, &base_name, &type_name, &local_type_name);
        list.name = base_name.clone();
        rename(&mut list, "select");
        list.name = config.root_field_name(&list.name);
        // The aggregate root is generated from the type name rather than being
        // a `QueryField` of its own, so its name travels with the list field
        // that spawns it.
        let aggregate_name = config
            .names
            .root(&table.schema, &table.name, "select_aggregate")
            .map(str::to_string)
            .unwrap_or_else(|| format!("{}_aggregate", base_name));
        list.aggregate_name = Some(config.root_field_name(&aggregate_name));
        list.aggregate_description = config
            .names
            .root_comment(&table.schema, &table.name, "select_aggregate")
            .map(str::to_string);
        // Whether this role may count what it can read. Only asked where a
        // role is building its own schema; the unrestricted one aggregates
        // whatever the database will.
        if let Some(role) = &config.role {
            list.aggregates =
                crate::role::allows_aggregations(&config.names, &table.schema, &table.name, role);
        }
        // A table with no readable field still gets its list entry when the
        // role may count it: the aggregate root is built beside that entry
        // rather than being one of its own, and `rows` is what says the list
        // root itself is not there. By key there is nothing to answer with at
        // all.
        list.rows = readable;
        if readable || list.aggregates {
            query_fields.push(list);
        }
        if readable {
            if let Some(mut by_pk) =
                QueryField::by_pk_typed(table, &base_name, &type_name, &local_type_name)
            {
                by_pk.name = format!("{}_by_pk", base_name);
                rename(&mut by_pk, "select_by_pk");
                by_pk.name = config.root_field_name(&by_pk.name);
                query_fields.push(by_pk);
            }
        }

        // Add mutation fields if enabled
        if config.enable_mutations {
            let mut fields: Vec<MutationField> = Vec::new();
            fields.extend(MutationField::insert_fields_typed(
                table,
                &local_type_name,
                &type_name,
            ));
            fields.extend(MutationField::update_fields_typed(
                table,
                &local_type_name,
                &type_name,
            ));
            fields.extend(MutationField::delete_fields_typed(
                table,
                &local_type_name,
                &type_name,
            ));
            // The three that answer with a row rather than with a count.
            if !readable {
                fields.retain(|field| {
                    !matches!(
                        field.mutation_type,
                        MutationType::InsertOne
                            | MutationType::UpdateByPk
                            | MutationType::DeleteByPk
                    )
                });
            }
            for field in &mut fields {
                let kind = match field.mutation_type {
                    MutationType::Insert => "insert",
                    MutationType::InsertOne => "insert_one",
                    MutationType::Update => "update",
                    MutationType::UpdateByPk => "update_by_pk",
                    MutationType::UpdateMany => "update_many",
                    MutationType::Delete => "delete",
                    MutationType::DeleteByPk => "delete_by_pk",
                };
                field.name = match field.mutation_type {
                    MutationType::Insert => format!("insert_{base_name}"),
                    MutationType::InsertOne => format!("insert_{base_name}_one"),
                    MutationType::Update => format!("update_{base_name}"),
                    MutationType::UpdateByPk => format!("update_{base_name}_by_pk"),
                    MutationType::UpdateMany => format!("update_{base_name}_many"),
                    MutationType::Delete => format!("delete_{base_name}"),
                    MutationType::DeleteByPk => format!("delete_{base_name}_by_pk"),
                };
                if let Some(given) = config.names.root(&table.schema, &table.name, kind) {
                    field.name = given.to_string();
                }
                field.name = config.root_field_name(&field.name);
                if let Some(comment) = config.names.root_comment(&table.schema, &table.name, kind) {
                    field.description = Some(comment.to_string()).filter(|c| !c.is_empty());
                }
            }
            mutation_fields.extend(fields);
        }

        // Add relationship fields. Where metadata declares them, they are
        // what the table has; otherwise reflection offers one per foreign
        // key, which is what a document saying nothing about a table has
        // always meant. See `TableNames::declared_relationships`.
        let rels = match config
            .names
            .declared_relationships(&table.schema, &table.name)
        {
            Some(declared) => declared_fields(
                declared,
                table,
                schema_cache,
                config,
                &base_names,
                &type_names,
                &local_type_names,
            ),
            None => reflected_fields(
                table,
                schema_cache,
                config,
                &base_names,
                &type_names,
                &local_type_names,
            ),
        };

        if !rels.is_empty() {
            relationship_fields.insert(type_name.clone(), rels);
        }

        if config.enable_federation && readable && has_whole_key(table) {
            let key_fields: Vec<String> = obj_type
                .fields
                .iter()
                .filter(|field| field.is_pk)
                .map(|field| field.name.clone())
                .collect();
            if key_fields.len() == table.pk_cols.len() {
                federation_entities.insert(
                    type_name.clone(),
                    FederationEntity {
                        type_name: type_name.clone(),
                        table: table.qualified_identifier(),
                        key_fields,
                        shared: config.is_shared_entity(&table.qualified_identifier()),
                    },
                );
            }
        }

        object_types.insert(type_name, obj_type);
    }

    // A table marked as a set of allowed values becomes a GraphQL enum, and
    // every column with a foreign key to it is typed as that enum rather than
    // as text. The values are rows, so they were read at startup rather than
    // reflected.
    let mut enum_types: HashMap<String, Vec<(String, Option<String>)>> = HashMap::new();
    let mut enum_type_of: HashMap<(String, String), String> = HashMap::new();

    for (schema_name, table_name) in config.names.enum_tables() {
        // A role that may not read the table's rows still sees the enum. The
        // values are a *type*, and a column typed by it is readable -- Hasura
        // publishes `colors_enum` to a role granted nothing on `colors`, and
        // the query `where: {favorite_color: {_eq: red}}` is one it answers.
        // So the name is derived rather than looked up, since a table the
        // role cannot read has no entry to look up.
        let table_key = (schema_name.clone(), table_name.clone());
        let base = match type_names.get(&table_key) {
            Some(found) => found.clone(),
            None => {
                let unprefixed = config
                    .names
                    .base_name(&schema_name, &table_name)
                    .map(str::to_string)
                    .unwrap_or_else(|| match schema_name == config.default_schema() {
                        true => table_name.clone(),
                        false => format!("{}_{}", schema_name, table_name),
                    });
                config.table_type_name(
                    &QualifiedIdentifier::new(&schema_name, &table_name),
                    &unprefixed,
                )
            }
        };
        let key = format!("{}.{}", schema_name, table_name);
        let members: Vec<(String, Option<String>)> = config
            .enum_values
            .get(&key)
            .map(|values| {
                values
                    .iter()
                    .filter(|(value, _)| is_graphql_name(value))
                    .cloned()
                    .collect()
            })
            .unwrap_or_default();

        if members.is_empty() {
            // No rows, or none whose value is a legal GraphQL name. An empty
            // enum is not a legal type, so the table stays an ordinary one.
            tracing::warn!(
                "{}.{} is marked as an enumeration but has no values that can name an \
                 enum member; exposing it as an ordinary table",
                schema_name,
                table_name
            );
            continue;
        }

        let type_name = format!("{}_enum", base);
        enum_type_of.insert((schema_name.clone(), table_name.clone()), type_name.clone());
        enum_types.insert(type_name, members);
    }

    // Retype the columns that point at one.
    //
    // Read from the catalogue's foreign keys rather than from the
    // relationships this schema exposes, which are two different questions: a
    // relationship *to* an enum table is never exposed at all -- see just
    // below -- and a column's type has nothing to do with whether the row at
    // the other end can be traversed to. Keying one on the other meant a table
    // whose relationships metadata had declared away lost its enum columns'
    // type with them.
    if !enum_type_of.is_empty() {
        for table in &tables {
            let Some(type_name) = type_names.get(&(table.schema.clone(), table.name.clone()))
            else {
                continue;
            };
            let found = schema_cache
                .get_relationships(&table.qualified_identifier(), &table.schema)
                .map(|found| found.to_vec())
                .unwrap_or_default();
            for relationship in &found {
                if !relationship.is_to_one() {
                    continue;
                }
                let foreign = relationship.foreign_table();
                let Some(enum_type) =
                    enum_type_of.get(&(foreign.schema.clone(), foreign.name.clone()))
                else {
                    continue;
                };
                let Some(column) = relationship.single_local_column() else {
                    continue;
                };
                if let Some(object) = object_types.get_mut(type_name) {
                    // Under the name the column is exposed as, which is not
                    // always its own.
                    let exposed = config
                        .names
                        .column(&object.table.schema, &object.table.name, column)
                        .unwrap_or(column)
                        .to_string();
                    for field in &mut object.fields {
                        if field.name == exposed {
                            field.graphql_type =
                                crate::types::GraphQLType::Custom(enum_type.clone());
                        }
                    }
                }
            }
        }
    }

    // A table marked as a set of allowed values is a set of *values*: the
    // column typed as the enum is the whole of what a client needs, and there
    // is nothing at the other end of a relationship to it worth traversing to.
    // Hasura exposes none, and a stray `user { color { ... } }` beside
    // `favorite_color` is a second spelling of the same fact.
    if !enum_type_of.is_empty() {
        for relationships in relationship_fields.values_mut() {
            relationships.retain(|relationship| {
                let foreign = relationship.relationship.foreign_table();
                !enum_type_of.contains_key(&(foreign.schema.clone(), foreign.name.clone()))
            });
        }
    }

    // Split each type's fields into the half that may be read and the half
    // that may be written. Last, so that everything above -- the computed
    // fields' arguments, the retyping of an enum table's columns -- has been
    // applied to the whole list before either half is taken out of it.
    //
    // A role may be granted a column to set that it is not granted to see, so
    // the cache carries the union and this is where the two part company. See
    // [`crate::role`].
    for object in object_types.values_mut() {
        // Taken unconditionally, even where no role narrows anything: the
        // write inputs are built from this list, and leaving it as it was when
        // the type was first made would leave it behind every rewrite since.
        object.writable_fields = object.fields.clone();
        if let Some(role) = &config.role {
            let (schema_name, table_name) =
                (object.table.schema.clone(), object.table.name.clone());
            let Some(granted) = config.names.permissions(&schema_name, &table_name, role) else {
                continue;
            };
            let table = object.table.clone();
            object.fields.retain(|field| {
                let column = config
                    .names
                    .column_source(&schema_name, &table_name, &field.name)
                    .unwrap_or(&field.name);
                match table.get_column(column) {
                    // A computed field rather than a column, and those were
                    // reduced to the granted ones with the cache.
                    None => true,
                    Some(_) => granted
                        .select
                        .as_ref()
                        .is_some_and(|select| select.columns.allows(column)),
                }
            });
        }
    }

    // Functions that answer with rows of a table it already exposes. One
    // returning anything else has no type here to return: a scalar-returning
    // function is not a root field, it is an RPC, which the REST surface
    // already offers.
    let mut function_fields = Vec::new();
    for routines in schema_cache.routines.values() {
        for routine in routines {
            if !config.is_schema_exposed(&routine.schema) || routine.is_procedure {
                continue;
            }
            let postrust_core::schema_cache::RetType::SetOf(returned) = &routine.return_type else {
                continue;
            };
            // A function taking a table's row is that table's computed field,
            // not a root field: `fetch_articles(author_row author)` is asked
            // of an author, and a row type is not something a client can send
            // as an argument. It is already exposed where it belongs.
            if routine.params.iter().any(|param| {
                let (_, named) = type_name_of(&param.param_type);
                base_names.keys().any(|(_, table)| table == &named)
            }) {
                continue;
            }
            // `SETOF <table>`, where the table is one this schema exposes. The
            // catalogue names the returned type without a schema, so the
            // function's own schema is tried first -- a function and the table
            // it returns usually live together, and two schemas may both have
            // a table of that name.
            let (returned_schema, returned_table) = type_name_of(returned);
            let found = base_names
                .iter()
                .find(|((schema, table), _)| {
                    table == &returned_table
                        && schema == returned_schema.as_ref().unwrap_or(&routine.schema)
                })
                .or_else(|| {
                    base_names
                        .iter()
                        .find(|((_, table), _)| table == &returned_table)
                });
            let Some(((target_schema, target_table), _)) = found else {
                continue;
            };
            let target = QualifiedIdentifier::new(target_schema, target_table);
            // A function returning rows of a table this role cannot read has
            // nowhere to put them: that table has no GraphQL type, and naming
            // one that was never registered is a schema that will not build.
            if !readable_tables.contains(&target) {
                continue;
            }
            let Some(return_type) = type_names.get(&(target_schema.clone(), target_table.clone()))
            else {
                continue;
            };
            let Some(local_return_type) =
                local_type_names.get(&(target_schema.clone(), target_table.clone()))
            else {
                continue;
            };
            let returns_table = (target_schema.clone(), target_table.clone());

            // Placed by what PostgreSQL says it does, unless metadata said
            // otherwise: `track_function` with `configuration: {exposed_as:
            // query}` puts a VOLATILE function on the query root, which is a
            // decision a person made and no catalogue remembers.
            let volatile = match config.names.exposed_as(&routine.schema, &routine.name) {
                Some("query") => false,
                Some("mutation") => true,
                _ => matches!(
                    routine.volatility,
                    postrust_core::schema_cache::FuncVolatility::Volatile
                ),
            };
            // A mutation a role was never granted. Hasura infers a query
            // function's permission from the select permission on the table it
            // returns -- which is the check just above -- and infers nothing
            // for a mutation: reading a table is not permission to change it,
            // so the role has to be named. A role granted no mutation at all
            // then has no mutation root, and a request for one is answered
            // `no mutations exist` rather than by a field that is missing.
            if volatile {
                if let Some(role) = config.role.as_deref() {
                    if !config
                        .names
                        .function_grants(&routine.schema, &routine.name, role)
                    {
                        continue;
                    }
                }
            }

            function_fields.push(FunctionField {
                name: config.root_field_name(&routine.name),
                schema_name: routine.schema.clone(),
                function_name: routine.name.clone(),
                returns: return_type.clone(),
                local_returns: local_return_type.clone(),
                returns_table,
                arguments: routine
                    .params
                    .iter()
                    .filter(|param| !param.name.is_empty())
                    .filter(|param| !is_session_argument(param))
                    .map(|param| (param.name.clone(), param.param_type.clone(), param.required))
                    .collect(),
                session_argument: routine
                    .params
                    .iter()
                    .find(|param| is_session_argument(param))
                    .map(|param| param.name.clone()),
                volatile,
                description: routine.description.clone(),
            });
        }
    }
    // Sorted by name, and then by signature so that which overload survives
    // the deduplication below does not depend on hash order.
    function_fields.sort_by(|a, b| {
        a.name.cmp(&b.name).then_with(|| {
            a.arguments
                .len()
                .cmp(&b.arguments.len())
                .then_with(|| a.argument_signature().cmp(&b.argument_signature()))
        })
    });

    // PostgreSQL overloads functions; GraphQL does not overload fields. Two
    // `SETOF` overloads of one name therefore want the same root field, and
    // `dynamic::Object::field` *panics* on the second -- so this is not a
    // cosmetic choice, it is the difference between starting and not. A schema
    // with `f(a int)` and `f(a int, b text)` both returning `SETOF t` crashed
    // the server at boot.
    //
    // The one taking fewest arguments is kept, being the one most callers can
    // call, and every other is named in a warning: a function that is not
    // exposed should say so rather than be discovered missing.
    let mut seen: HashSet<String> = HashSet::new();
    function_fields.retain(|function| {
        if seen.insert(function.name.clone()) {
            return true;
        }
        tracing::warn!(
            "not exposing {}.{}({}) -- another overload of {} is already a root \
             field, and GraphQL has no overloading",
            function.schema_name,
            function.function_name,
            function.argument_signature(),
            function.name,
        );
        false
    });

    GeneratedSchema {
        object_types,
        query_fields,
        mutation_fields,
        relationship_fields,
        enum_types,
        function_fields,
        federation_entities,
        enable_federation: config.enable_federation,
    }
}

/// Whether a value can name a GraphQL enum member.
///
/// `[_A-Za-z][_0-9A-Za-z]*`, which is what the specification allows. A value
/// that cannot is left out rather than mangled into one: a client asking for
/// `light blue` should be told there is no such member, not handed
/// `light_blue` and left to wonder which row it means.
fn is_graphql_name(value: &str) -> bool {
    let mut chars = value.chars();
    match chars.next() {
        Some(first) if first == '_' || first.is_ascii_alphabetic() => {}
        _ => return false,
    }
    chars.all(|c| c == '_' || c.is_ascii_alphanumeric())
}

#[cfg(test)]
mod tests {
    use super::*;
    use indexmap::IndexMap;
    use postrust_core::schema_cache::Column;
    use pretty_assertions::assert_eq;

    fn create_test_table(name: &str, insertable: bool, updatable: bool, deletable: bool) -> Table {
        let mut columns = IndexMap::new();
        columns.insert(
            "id".into(),
            Column {
                name: "id".into(),
                description: None,
                nullable: false,
                data_type: "integer".into(),
                nominal_type: "int4".into(),
                max_len: None,
                default: Some("nextval('id_seq')".into()),
                enum_values: vec![],
                is_pk: true,
                position: 1,
                always_generated: false,
                domain_type: None,
            },
        );
        columns.insert(
            "name".into(),
            Column {
                name: "name".into(),
                description: None,
                nullable: false,
                data_type: "text".into(),
                nominal_type: "text".into(),
                max_len: None,
                default: None,
                enum_values: vec![],
                is_pk: false,
                position: 2,
                always_generated: false,
                domain_type: None,
            },
        );

        Table {
            schema: "public".into(),
            name: name.into(),
            description: None,
            is_view: false,
            insertable,
            updatable,
            deletable,
            pk_cols: vec!["id".into()],
            unique_constraints: Vec::new(),
            columns,
            computed_columns: Default::default(),
            is_partitioned: false,
        }
    }

    fn create_test_schema_cache() -> SchemaCache {
        use std::collections::{HashMap, HashSet};

        let mut tables = HashMap::new();

        let users = create_test_table("users", true, true, true);
        let posts = create_test_table("posts", true, true, true);
        let comments = create_test_table("comments", true, false, false);

        tables.insert(users.qualified_identifier(), users);
        tables.insert(posts.qualified_identifier(), posts);
        tables.insert(comments.qualified_identifier(), comments);

        SchemaCache {
            tables,
            relationships: HashMap::new(),
            routines: HashMap::new(),
            timezones: HashSet::new(),
            media_handlers: HashMap::new(),
            pg_version: 150000,
            representations: Default::default(),
        }
    }

    #[test]
    fn a_many_to_many_relationship_can_be_named_by_its_junction() {
        use postrust_core::schema_cache::{Cardinality, Junction, Relationship};

        let relationship = Relationship::ForeignKey {
            table: QualifiedIdentifier::new("public", "products"),
            foreign_table: QualifiedIdentifier::new("public", "product_research"),
            is_self: false,
            cardinality: Cardinality::M2M(Junction {
                table: QualifiedIdentifier::new("public", "product_research_links"),
                constraint1: "product_research_links_product_id_fkey".into(),
                constraint2: "product_research_links_research_id_fkey".into(),
                source_columns: vec![("id".into(), "product_id".into())],
                target_columns: vec![("research_id".into(), "id".into())],
            }),
            table_is_view: false,
            foreign_table_is_view: false,
            constraint_name: String::new(),
        };

        assert_eq!(
            relationship_keys(&relationship),
            vec![
                "product_research_links",
                "product_research_links_product_id_fkey",
                "product_research_links_research_id_fkey"
            ]
        );

        let mut cache = create_test_schema_cache();
        let products = create_test_table("products", true, true, true);
        let product_research = create_test_table("product_research", true, true, true);
        let products_key = products.qualified_identifier();
        cache.tables.insert(products_key.clone(), products);
        cache
            .tables
            .insert(product_research.qualified_identifier(), product_research);
        cache
            .relationships
            .insert((products_key, "public".into()), vec![relationship]);

        let names = crate::names::NameOverrides::parse(
            r#"{"tables": {"public.products": {
                "relationships": {"product_research_links": "research"}
            }}}"#,
        )
        .unwrap();
        let schema = build_schema(
            &cache,
            &SchemaConfig {
                names,
                ..SchemaConfig::default()
            },
        );
        let field = schema
            .get_relationship_fields("products")
            .and_then(|fields| fields.iter().find(|field| field.name == "research"))
            .expect("junction metadata renames the inferred relationship");

        assert_eq!(field.target_type, "product_research");
        assert!(field.is_list);
    }

    /// A `SETOF users` function, with the given argument names and types.
    fn setof_users_routine(
        name: &str,
        args: &[(&str, &str)],
    ) -> postrust_core::schema_cache::Routine {
        use postrust_core::schema_cache::{FuncVolatility, RetType, Routine, RoutineParam};
        Routine {
            schema: "public".into(),
            name: name.into(),
            description: None,
            params: args
                .iter()
                .map(|(pname, ptype)| RoutineParam {
                    name: (*pname).into(),
                    param_type: (*ptype).into(),
                    type_max_length: (*ptype).into(),
                    required: true,
                    variadic: false,
                })
                .collect(),
            return_type: RetType::SetOf("users".into()),
            returns_composite: true,
            volatility: FuncVolatility::Stable,
            has_variadic: false,
            isolation_level: None,
            settings: Vec::new(),
            is_procedure: false,
            media_type: None,
            media_base_type: None,
            output_columns: Vec::new(),
        }
    }

    /// Two `SETOF` overloads of one name want one root field, and
    /// `dynamic::Object::field` panics on the second -- so this is the
    /// difference between the server starting and not. Reproduced from
    /// PostgREST's `overloaded_default` fixture, which crashed it at boot.
    #[test]
    fn overloads_of_one_function_become_a_single_root_field() {
        let mut cache = create_test_schema_cache();
        let one = setof_users_routine("overloaded", &[("a", "integer")]);
        let two = setof_users_routine("overloaded", &[("a", "integer"), ("b", "text")]);
        cache
            .routines
            .insert(one.qualified_identifier(), vec![two, one]);

        let generated = build_schema(&cache, &SchemaConfig::default());
        let named: Vec<&str> = generated
            .function_fields
            .iter()
            .filter(|f| f.name == "overloaded")
            .map(|f| f.name.as_str())
            .collect();
        assert_eq!(named.len(), 1, "exactly one root field per function name");
    }

    /// And which overload survives cannot depend on hash order, or the schema
    /// differs between two runs of the same binary against one database.
    #[test]
    fn the_surviving_overload_is_the_one_taking_fewest_arguments() {
        let long = setof_users_routine("overloaded", &[("a", "integer"), ("b", "text")]);
        let short = setof_users_routine("overloaded", &[("a", "integer")]);

        for order in [vec![long.clone(), short.clone()], vec![short, long]] {
            let mut cache = create_test_schema_cache();
            cache
                .routines
                .insert(order[0].qualified_identifier(), order);
            let generated = build_schema(&cache, &SchemaConfig::default());
            let kept = generated
                .function_fields
                .iter()
                .find(|f| f.name == "overloaded")
                .expect("the function is exposed");
            assert_eq!(kept.argument_signature(), "a integer");
        }
    }

    // ============================================================================
    // SchemaConfig Tests
    // ============================================================================

    #[test]
    fn test_schema_config_default() {
        let config = SchemaConfig::default();
        assert!(config.is_schema_exposed("public"));
        assert!(!config.is_schema_exposed("private"));
        assert!(config.enable_mutations);
        assert!(!config.enable_subscriptions);
    }

    #[test]
    fn test_schema_config_with_schemas() {
        let config =
            SchemaConfig::new().with_schemas(vec!["api".to_string(), "public".to_string()]);
        assert!(config.is_schema_exposed("api"));
        assert!(config.is_schema_exposed("public"));
        assert!(!config.is_schema_exposed("private"));
    }

    #[test]
    fn test_schema_config_mutations_disabled() {
        let config = SchemaConfig::new().with_mutations(false);
        assert!(!config.enable_mutations);
    }

    // ============================================================================
    // QueryField Tests
    // ============================================================================

    #[test]
    fn test_query_field_list() {
        let table = create_test_table("users", true, true, true);
        let field = QueryField::list(&table);

        assert_eq!(field.name, "users");
        assert_eq!(field.return_type, "[users!]!");
        assert!(field.is_list);
        assert!(!field.is_by_pk);
    }

    #[test]
    fn test_query_field_by_pk() {
        let table = create_test_table("users", true, true, true);
        let field = QueryField::by_pk(&table).unwrap();

        assert_eq!(field.name, "users_by_pk");
        assert_eq!(field.return_type, "users");
        assert!(!field.is_list);
        assert!(field.is_by_pk);
    }

    #[test]
    fn test_query_field_by_pk_no_pk() {
        let mut table = create_test_table("users", true, true, true);
        table.pk_cols = vec![];
        let field = QueryField::by_pk(&table);

        assert!(field.is_none());
    }

    // ============================================================================
    // MutationField Tests
    // ============================================================================

    #[test]
    fn test_mutation_field_insert() {
        let table = create_test_table("users", true, true, true);
        let fields = MutationField::insert_fields(&table);

        assert_eq!(fields.len(), 2);
        assert_eq!(fields[0].name, "insert_users");
        assert_eq!(fields[0].mutation_type, MutationType::Insert);
        assert_eq!(fields[1].name, "insert_users_one");
        assert_eq!(fields[1].mutation_type, MutationType::InsertOne);
    }

    #[test]
    fn test_mutation_field_insert_not_insertable() {
        let table = create_test_table("users", false, true, true);
        let fields = MutationField::insert_fields(&table);

        assert!(fields.is_empty());
    }

    #[test]
    fn test_mutation_field_update() {
        let table = create_test_table("users", true, true, true);
        let fields = MutationField::update_fields(&table);

        assert_eq!(fields.len(), 3);
        assert_eq!(fields[0].name, "update_users");
        assert_eq!(fields[0].mutation_type, MutationType::Update);
        assert_eq!(fields[1].name, "update_users_many");
        assert_eq!(fields[1].mutation_type, MutationType::UpdateMany);
        assert_eq!(fields[2].name, "update_users_by_pk");
        assert_eq!(fields[2].mutation_type, MutationType::UpdateByPk);
    }

    #[test]
    fn test_mutation_field_update_not_updatable() {
        let table = create_test_table("users", true, false, true);
        let fields = MutationField::update_fields(&table);

        assert!(fields.is_empty());
    }

    #[test]
    fn test_mutation_field_delete() {
        let table = create_test_table("users", true, true, true);
        let fields = MutationField::delete_fields(&table);

        assert_eq!(fields.len(), 2);
        assert_eq!(fields[0].name, "delete_users");
        assert_eq!(fields[0].mutation_type, MutationType::Delete);
        assert_eq!(fields[1].name, "delete_users_by_pk");
        assert_eq!(fields[1].mutation_type, MutationType::DeleteByPk);
    }

    #[test]
    fn test_mutation_field_delete_not_deletable() {
        let table = create_test_table("users", true, true, false);
        let fields = MutationField::delete_fields(&table);

        assert!(fields.is_empty());
    }

    // ============================================================================
    // Singularize Tests
    // ============================================================================

    // ============================================================================
    // Build Schema Tests
    // ============================================================================

    #[test]
    fn test_build_schema_object_types() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        assert_eq!(schema.object_types.len(), 3);
        assert!(schema.get_object_type("users").is_some());
        assert!(schema.get_object_type("posts").is_some());
        assert!(schema.get_object_type("comments").is_some());
    }

    #[test]
    fn test_build_schema_query_fields() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        // 3 tables * 2 (list + byPk) = 6 query fields
        assert_eq!(schema.query_fields.len(), 6);

        // Check users query fields
        let users_field = schema.get_query_field("users").unwrap();
        assert_eq!(users_field.name, "users");
        assert!(users_field.is_list);
    }

    #[test]
    fn test_build_schema_mutation_fields() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        // users: 2 insert + 3 update + 2 delete = 7
        // posts: 2 insert + 3 update + 2 delete = 7
        // comments: 2 insert + 0 update + 0 delete = 2
        // Total: 16
        assert_eq!(schema.mutation_fields.len(), 16);

        let users_mutations = schema.get_mutation_fields("users");
        assert_eq!(users_mutations.len(), 7);
    }

    #[test]
    fn test_build_schema_mutations_disabled() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::new().with_mutations(false);
        let schema = build_schema(&cache, &config);

        assert!(schema.mutation_fields.is_empty());
    }

    #[test]
    fn test_build_schema_table_names() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        let names = schema.table_names();
        assert_eq!(names.len(), 3);
        assert!(names.contains(&"users"));
        assert!(names.contains(&"posts"));
        assert!(names.contains(&"comments"));
    }

    #[test]
    fn test_build_schema_type_names() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        let names = schema.type_names();
        assert_eq!(names.len(), 3);
        assert!(names.contains(&"users"));
        assert!(names.contains(&"posts"));
        assert!(names.contains(&"comments"));
    }

    #[test]
    fn type_prefix_namespaces_generated_types_and_roots() {
        let mut cache = create_test_schema_cache();
        let routine = setof_users_routine("search_users", &[("name", "text")]);
        cache
            .routines
            .insert(routine.qualified_identifier(), vec![routine]);
        let config = SchemaConfig {
            type_prefix: Some("test".into()),
            ..SchemaConfig::default()
        };

        let schema = build_schema(&cache, &config);

        assert!(schema.get_object_type("test_users").is_some());
        let users = schema
            .query_fields
            .iter()
            .find(|field| field.table_name == "users" && field.is_list)
            .expect("users list field");
        assert_eq!(users.name, "test_users");
        assert_eq!(users.type_name, "test_users");
        assert_eq!(
            users.aggregate_name.as_deref(),
            Some("test_users_aggregate")
        );
        assert!(schema
            .mutation_fields
            .iter()
            .any(|field| field.name == "test_insert_users"));
        let function = schema
            .function_fields
            .iter()
            .find(|field| field.function_name == "search_users")
            .expect("set-returning function field");
        assert_eq!(function.name, "test_search_users");
        assert_eq!(function.returns, "test_users");
    }

    #[test]
    fn shared_entities_keep_their_type_name_but_not_their_root_names() {
        use postrust_core::schema_cache::{Cardinality, Relationship};

        let mut cache = create_test_schema_cache();
        let posts = QualifiedIdentifier::new("public", "posts");
        cache.relationships.insert(
            (posts.clone(), "public".into()),
            vec![Relationship::ForeignKey {
                table: posts,
                foreign_table: QualifiedIdentifier::new("public", "users"),
                is_self: false,
                cardinality: Cardinality::M2O {
                    constraint: "posts_user_id_fkey".into(),
                    columns: vec![("id".into(), "id".into())],
                },
                table_is_view: false,
                foreign_table_is_view: false,
                constraint_name: "posts_user_id_fkey".into(),
            }],
        );
        let config = SchemaConfig {
            enable_federation: true,
            type_prefix: Some("test".into()),
            names: crate::names::NameOverrides::parse(
                r#"{"tables": {"public.users": {"federation": {"shared": true}}}}"#,
            )
            .expect("metadata"),
            ..SchemaConfig::default()
        };

        let schema = build_schema(&cache, &config);

        let users_object = schema.get_object_type("users").expect("users object");
        assert_eq!(users_object.local_name, "test_users");
        let posts_object = schema.get_object_type("test_posts").expect("posts object");
        assert_eq!(posts_object.local_name, "test_posts");
        let users = schema
            .query_fields
            .iter()
            .find(|field| field.table_name == "users" && field.is_list)
            .expect("users list field");
        assert_eq!(users.name, "test_users");
        assert_eq!(users.type_name, "users");
        assert_eq!(users.local_type_name, "test_users");
        let posts_relationships = schema
            .get_relationship_fields("test_posts")
            .expect("posts relationships");
        assert_eq!(posts_relationships[0].name, "user");
        assert_eq!(posts_relationships[0].target_type, "users");
        assert_eq!(posts_relationships[0].target_local_name, "test_users");
    }

    #[test]
    fn federation_entities_are_generated_only_when_enabled() {
        let cache = create_test_schema_cache();

        let disabled = build_schema(&cache, &SchemaConfig::default());
        assert!(!disabled.enable_federation);
        assert!(disabled.federation_entities.is_empty());

        let enabled = build_schema(
            &cache,
            &SchemaConfig {
                enable_federation: true,
                type_prefix: Some("test".into()),
                names: crate::names::NameOverrides::parse(
                    r#"{"tables": {"public.users": {"federation": {"shared": true}}}}"#,
                )
                .expect("metadata"),
                ..SchemaConfig::default()
            },
        );

        assert!(enabled.enable_federation);
        let users = enabled
            .get_federation_entity("users")
            .expect("shared users entity");
        assert_eq!(users.type_name, "users");
        assert_eq!(users.table, QualifiedIdentifier::new("public", "users"));
        assert_eq!(users.key_fields_sdl(), "id");
        assert!(users.shared);

        let posts = enabled
            .get_federation_entity("test_posts")
            .expect("prefixed posts entity");
        assert_eq!(posts.type_name, "test_posts");
        assert_eq!(posts.key_fields_sdl(), "id");
        assert!(!posts.shared);
    }

    #[test]
    fn test_build_schema_exposed_schemas() {
        let mut cache = create_test_schema_cache();

        // Add a table in a different schema
        let private_table = Table {
            schema: "private".into(),
            name: "secrets".into(),
            description: None,
            is_view: false,
            insertable: true,
            updatable: true,
            deletable: true,
            pk_cols: vec!["id".into()],
            unique_constraints: Vec::new(),
            columns: indexmap::IndexMap::new(),
            computed_columns: Default::default(),
            is_partitioned: false,
        };
        cache
            .tables
            .insert(private_table.qualified_identifier(), private_table);

        let config = SchemaConfig::default(); // Only exposes "public"
        let schema = build_schema(&cache, &config);

        // Should only have 3 tables from public schema
        assert_eq!(schema.object_types.len(), 3);
        assert!(schema.get_object_type("Secrets").is_none());
    }

    // ============================================================================
    // GeneratedSchema Tests
    // ============================================================================

    #[test]
    fn test_generated_schema_get_object_type() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        let users = schema.get_object_type("users").unwrap();
        assert_eq!(users.table.name, "users");
    }

    #[test]
    fn test_generated_schema_get_query_field() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        let field = schema.get_query_field("posts").unwrap();
        assert_eq!(field.table_name, "posts");
    }

    #[test]
    fn test_generated_schema_get_mutation_fields() {
        let cache = create_test_schema_cache();
        let config = SchemaConfig::default();
        let schema = build_schema(&cache, &config);

        let fields = schema.get_mutation_fields("comments");
        // comments is only insertable
        assert_eq!(fields.len(), 2); // insert_comments + insert_comments_one
    }
}