fraiseql-cli 2.19.0

#![allow(clippy::unwrap_used, clippy::panic)] // Reason: test code, panics are acceptable
#![allow(clippy::wildcard_imports)] // Reason: test modules use wildcard imports for conciseness

use fraiseql_core::schema::{NamingConvention, PaginationOrder};
use indexmap::IndexMap;

use super::*;
use crate::schema::intermediate::{
    IntermediateArgument, IntermediateAutoParams, IntermediateField, IntermediateQuery,
    IntermediateSchema, IntermediateType,
};

// ── #366 @subscribable threading ─────────────────────────────────────────────

#[test]
fn intermediate_type_subscribable_tables_defaults_none() {
    // A type authored before #366 has no `subscribable_tables` key.
    let json = r#"{ "name": "Post", "fields": [] }"#;
    let t: IntermediateType = serde_json::from_str(json).unwrap();
    assert!(t.subscribable_tables.is_none(), "absent subscribable_tables defaults to None");
}

#[test]
fn intermediate_type_reads_subscribable_tables() {
    let json = r#"{ "name": "Post", "fields": [], "subscribable_tables": ["tb_post"] }"#;
    let t: IntermediateType = serde_json::from_str(json).unwrap();
    assert_eq!(t.subscribable_tables, Some(vec!["tb_post".to_string()]));
}

#[test]
fn convert_aggregates_subscribable_tables_into_compiled() {
    let intermediate = IntermediateSchema {
        types: vec![
            IntermediateType {
                name: "Post".to_string(),
                subscribable_tables: Some(vec![
                    "tb_post".to_string(),
                    "public.tb_post_archive".to_string(),
                ]),
                ..Default::default()
            },
            // Not subscribable → excluded.
            IntermediateType {
                name: "Comment".to_string(),
                ..Default::default()
            },
            // Empty list → treated as not subscribable.
            IntermediateType {
                name: "Tag".to_string(),
                subscribable_tables: Some(vec![]),
                subscribable_pre_image: false,
                ..Default::default()
            },
        ],
        ..Default::default()
    };

    let compiled = SchemaConverter::convert(intermediate).expect("convert");
    assert_eq!(compiled.subscribable.len(), 1, "only Post is subscribable");
    assert_eq!(compiled.subscribable[0].entity_type, "Post");
    assert_eq!(
        compiled.subscribable[0].tables,
        vec!["tb_post".to_string(), "public.tb_post_archive".to_string()]
    );
    assert!(
        !compiled.subscribable[0].pre_image,
        "pre_image defaults off when subscribable_pre_image is unset"
    );
}

#[test]
fn convert_threads_subscribable_pre_image_into_compiled() {
    // changelog_pre_image out-of-band parity: subscribable_pre_image=true on a
    // @subscribable type threads into SubscribableEntity.pre_image, so the
    // generated capture triggers record OLD into object_data_before.
    let intermediate = IntermediateSchema {
        types: vec![IntermediateType {
            name: "Price".to_string(),
            subscribable_tables: Some(vec!["tb_price".to_string()]),
            subscribable_pre_image: true,
            ..Default::default()
        }],
        ..Default::default()
    };

    let compiled = SchemaConverter::convert(intermediate).expect("convert");
    assert_eq!(compiled.subscribable.len(), 1);
    assert!(compiled.subscribable[0].pre_image, "subscribable_pre_image threads through");
}

// ── #507 type-level sql_source threading (federation extends entities) ────────

#[test]
fn intermediate_type_reads_type_level_sql_source() {
    // The authoring SDK emits a type-level `sql_source` for an owner-split
    // `extend type … @key` federation entity that has no local backing query.
    let json = r#"{ "name": "Organization", "fields": [], "sql_source": "v_organization" }"#;
    let t: IntermediateType = serde_json::from_str(json).unwrap();
    assert_eq!(t.sql_source.as_deref(), Some("v_organization"));
}

#[test]
fn intermediate_type_sql_source_defaults_none() {
    // The common case: an owned type binds its relation on the query that returns
    // it, so the type-level key is absent.
    let json = r#"{ "name": "User", "fields": [] }"#;
    let t: IntermediateType = serde_json::from_str(json).unwrap();
    assert!(t.sql_source.is_none(), "absent sql_source defaults to None");
}

#[test]
fn convert_threads_type_level_sql_source_into_compiled() {
    // An extends entity's type-level sql_source flows to TypeDefinition.sql_source
    // so the federation `_entities` resolver can source its backing relation when
    // no root query returns the type (#507). An owned type without one keeps the
    // historical empty type-level sql_source.
    let intermediate = IntermediateSchema {
        types: vec![
            IntermediateType {
                name: "Organization".to_string(),
                sql_source: Some("v_organization".to_string()),
                ..Default::default()
            },
            IntermediateType {
                name: "User".to_string(),
                ..Default::default()
            },
        ],
        ..Default::default()
    };

    let compiled = SchemaConverter::convert(intermediate).expect("convert");
    let org = compiled
        .types
        .iter()
        .find(|t| t.name == "Organization")
        .expect("Organization type");
    assert_eq!(
        org.sql_source.as_str(),
        "v_organization",
        "type-level sql_source threads through to the compiled type"
    );
    assert_eq!(
        org.jsonb_column, "data",
        "an extends entity's fields default to the standard jsonb `data` column, symmetric \
         with the query path"
    );
    let user = compiled.types.iter().find(|t| t.name == "User").expect("User type");
    assert!(
        user.sql_source.as_str().is_empty(),
        "an owned type keeps the historical empty type-level sql_source"
    );
    assert!(
        user.jsonb_column.is_empty(),
        "an owned type keeps the historical empty type-level jsonb_column (optimizer heuristic \
         stays off, compiled output byte-identical)"
    );
}

// ── #1142 type-level inject_params threading (queryless entity scoping) ──────

#[test]
fn intermediate_type_reads_type_level_inject_params_in_both_shapes() {
    // The same two shapes the operation-level key accepts (#806): nested, which every
    // SDK emits, and flat, which is pleasanter to hand-author.
    let nested = r#"{ "name": "Organization", "fields": [],
        "inject_params": {"tenant_id": {"source": "jwt", "claim": "org_id"}} }"#;
    let t: IntermediateType = serde_json::from_str(nested).unwrap();
    assert_eq!(t.inject_params.get("tenant_id").map(String::as_str), Some("jwt:org_id"));

    let flat = r#"{ "name": "Organization", "fields": [],
        "inject_params": {"tenant_id": "jwt:org_id"} }"#;
    let t: IntermediateType = serde_json::from_str(flat).unwrap();
    assert_eq!(t.inject_params.get("tenant_id").map(String::as_str), Some("jwt:org_id"));
}

#[test]
fn intermediate_type_inject_params_defaults_empty() {
    let json = r#"{ "name": "User", "fields": [] }"#;
    let t: IntermediateType = serde_json::from_str(json).unwrap();
    assert!(t.inject_params.is_empty(), "absent inject_params defaults to empty");
}

#[test]
fn convert_threads_type_level_inject_params_into_compiled() {
    // The #1142 shape: an entity whose relation rides on the type-level sql_source and
    // which no query returns, so the type is the only place its scoping can live.
    let intermediate = IntermediateSchema {
        types: vec![IntermediateType {
            name: "Organization".to_string(),
            sql_source: Some("v_organization".to_string()),
            inject_params: std::iter::once(("tenant_id".to_string(), "jwt:org_id".to_string()))
                .collect(),
            ..Default::default()
        }],
        ..Default::default()
    };

    let compiled = SchemaConverter::convert(intermediate).expect("convert");
    assert_eq!(
        compiled.types[0].inject_params.get("tenant_id"),
        Some(&fraiseql_core::schema::InjectedParamSource::Jwt("org_id".to_string())),
        "type-level inject_params threads through to the compiled type"
    );
}

#[test]
fn a_type_level_inject_source_with_an_unknown_prefix_is_refused() {
    // Same refusal as the operation path: an unparseable source must not compile to a
    // type with no scoping at all, which is exactly the silent-empty-map failure #806
    // was filed for.
    let intermediate = IntermediateSchema {
        types: vec![IntermediateType {
            name: "Organization".to_string(),
            sql_source: Some("v_organization".to_string()),
            inject_params: std::iter::once(("tenant_id".to_string(), "header:x-org".to_string()))
                .collect(),
            ..Default::default()
        }],
        ..Default::default()
    };

    let err =
        SchemaConverter::convert(intermediate).expect_err("an unknown prefix must not compile");
    // `{:#}` renders the whole anyhow chain; `to_string()` shows only the outermost
    // context ("Failed to convert types"), which would pass for any conversion failure.
    let chain = format!("{err:#}");
    assert!(
        chain.contains("Unknown inject source prefix"),
        "the refusal must name the unsupported prefix: {chain}"
    );
    assert!(
        chain.contains("Organization") && chain.contains("tenant_id"),
        "the refusal must locate the bad source at its type and column: {chain}"
    );
}

#[test]
fn test_convert_minimal_schema() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    assert_eq!(compiled.types.len(), 0);
    assert_eq!(compiled.queries.len(), 0);
    assert_eq!(compiled.mutations.len(), 0);
}

#[test]
fn convert_field_maps_authorize_true() {
    let intermediate = IntermediateField {
        function: None,

        deprecated:      None,
        vector_config:   None,
        vector_distance: None,
        name:            "email".to_string(),
        field_type:      "String".to_string(),
        nullable:        true,
        description:     None,
        directives:      None,
        requires_scope:  None,
        on_deny:         None,
        authorize:       Some(true),
        hierarchy:       None,
        unique:          None,
        localized:       false,
    };
    let compiled = SchemaConverter::convert_field(
        intermediate,
        &crate::schema::converter::DeclaredTypeNames::default(),
    )
    .unwrap();
    assert!(compiled.authorize, "authorize: Some(true) must compile to authorize == true");
}

#[test]
fn convert_field_authorize_absent_defaults_false() {
    let intermediate = IntermediateField {
        function: None,

        deprecated:      None,
        vector_config:   None,
        vector_distance: None,
        name:            "id".to_string(),
        field_type:      "Int".to_string(),
        nullable:        false,
        description:     None,
        directives:      None,
        requires_scope:  None,
        on_deny:         None,
        authorize:       None,
        hierarchy:       None,
        unique:          None,
        localized:       false,
    };
    let compiled = SchemaConverter::convert_field(
        intermediate,
        &crate::schema::converter::DeclaredTypeNames::default(),
    )
    .unwrap();
    assert!(!compiled.authorize, "absent authorize must compile to authorize == false");
}

/// The authoring hop the load-time to-one check depends on. `unique` is declared on a
/// field and consumed by `relationship_violations`, two crates away, so a converter that
/// dropped it would compile a schema whose to-one relationships are all refused — with
/// nothing pointing at the converter.
#[test]
fn convert_field_maps_unique_true() {
    let intermediate = IntermediateField {
        function: None,

        deprecated:      None,
        vector_config:   None,
        vector_distance: None,
        name:            "identifier".to_string(),
        field_type:      "String".to_string(),
        nullable:        false,
        description:     None,
        directives:      None,
        requires_scope:  None,
        on_deny:         None,
        authorize:       None,
        hierarchy:       None,
        unique:          Some(true),
        localized:       false,
    };
    let compiled = SchemaConverter::convert_field(
        intermediate,
        &crate::schema::converter::DeclaredTypeNames::default(),
    )
    .unwrap();
    assert!(compiled.unique, "unique: Some(true) must compile to unique == true");
    assert!(
        compiled.is_unique_key(),
        "and must satisfy the predicate the to-one check actually reads"
    );
}

/// Absent is `false`, not "unknown": a field says nothing about uniqueness unless it says
/// `unique`, and the to-one check refuses on that silence rather than guessing.
#[test]
fn convert_field_unique_absent_defaults_false() {
    let intermediate = IntermediateField {
        function: None,

        deprecated:      None,
        vector_config:   None,
        vector_distance: None,
        name:            "name".to_string(),
        field_type:      "String".to_string(),
        nullable:        false,
        description:     None,
        directives:      None,
        requires_scope:  None,
        on_deny:         None,
        authorize:       None,
        hierarchy:       None,
        unique:          None,
        localized:       false,
    };
    let compiled = SchemaConverter::convert_field(
        intermediate,
        &crate::schema::converter::DeclaredTypeNames::default(),
    )
    .unwrap();
    assert!(!compiled.unique, "absent unique must compile to unique == false");
    assert!(!compiled.is_unique_key(), "and an ordinary column is not an identity");
}

// ── #434: list field types must compile to FieldType::List ──────────────
//
// `parse_field_type` matched built-in scalar names and routed everything else —
// including the SDL list string "[Foo!]" — to FieldType::Object, so a list field
// (or list query argument) became Object("[Foo!]"), a single object whose type
// name does not exist, and projected at runtime as one null object, not a list.

#[test]
fn parse_field_type_unwraps_list_of_objects() {
    assert_eq!(
        SchemaConverter::parse_field_type(
            "[Item!]",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::list(FieldType::Object("Item".to_string())),
    );
}

#[test]
fn parse_field_type_unwraps_nullable_element_list() {
    assert_eq!(
        SchemaConverter::parse_field_type(
            "[Item]",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::list_of_nullable(FieldType::Object("Item".to_string())),
    );
}

#[test]
fn parse_field_type_unwraps_list_of_scalars() {
    assert_eq!(
        SchemaConverter::parse_field_type(
            "[String!]",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::list(FieldType::String),
    );
}

#[test]
fn parse_field_type_strips_trailing_nonnull() {
    assert_eq!(
        SchemaConverter::parse_field_type(
            "Item!",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::Object("Item".to_string()),
    );
    assert_eq!(
        SchemaConverter::parse_field_type(
            "String!",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::String
    );
}

#[test]
fn parse_field_type_nested_list() {
    assert_eq!(
        SchemaConverter::parse_field_type(
            "[[Item!]!]",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::list(FieldType::list(FieldType::Object("Item".to_string()))),
    );
}

#[test]
fn parse_field_type_plain_scalar_and_object_unchanged() {
    assert_eq!(
        SchemaConverter::parse_field_type(
            "String",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::String
    );
    assert_eq!(
        SchemaConverter::parse_field_type(
            "User",
            &crate::schema::converter::DeclaredTypeNames::default()
        )
        .unwrap(),
        FieldType::Object("User".to_string()),
    );
}

#[test]
fn convert_field_list_type_compiles_to_list() {
    let intermediate = IntermediateField {
        function: None,

        deprecated:      None,
        vector_config:   None,
        vector_distance: None,
        name:            "items".to_string(),
        field_type:      "[Item!]".to_string(),
        nullable:        false,
        description:     None,
        directives:      None,
        requires_scope:  None,
        on_deny:         None,
        authorize:       None,
        hierarchy:       None,
        unique:          None,
        localized:       false,
    };
    let compiled = SchemaConverter::convert_field(
        intermediate,
        &crate::schema::converter::DeclaredTypeNames::default(),
    )
    .unwrap();
    assert_eq!(
        compiled.field_type,
        FieldType::list(FieldType::Object("Item".to_string())),
        "a list field must compile to FieldType::List, not Object(\"[Item!]\")"
    );
}

#[test]
fn test_convert_type_with_fields() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "User".to_string(),
            sql_source:             None,
            fields:                 vec![
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "id".to_string(),
                    field_type:      "Int".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "name".to_string(),
                    field_type:      "String".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
            ],
            description:            Some("User type".to_string()),
            implements:             vec![],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    assert_eq!(compiled.types.len(), 1);
    assert_eq!(compiled.types[0].name, "User");
    assert_eq!(compiled.types[0].fields.len(), 2);
    assert_eq!(compiled.types[0].fields[0].field_type, FieldType::Int);
    assert_eq!(compiled.types[0].fields[1].field_type, FieldType::String);
}

#[test]
fn test_validate_unknown_type_reference() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![IntermediateQuery {
            function: None,

            requires_actor:          Vec::new(),
            count:                   false,
            name:                    "users".to_string(),
            return_type:             "UnknownType".to_string(),
            returns_list:            true,
            nullable:                false,
            arguments:               vec![],
            description:             None,
            sql_source:              Some("v_user".to_string()),
            sql_source_kind:         fraiseql_core::schema::SourceKind::Relation,
            auto_params:             None,
            deprecated:              None,
            jsonb_column:            None,
            relay:                   false,
            inject:                  IndexMap::default(),
            exclude_inject_defaults: Vec::new(),
            read_routing:            fraiseql_core::db::types::ReadRouting::default(),
            cache_ttl_seconds:       None,
            additional_views:        vec![],
            requires_role:           None,
            relay_cursor_type:       None,
            rest:                    None,
            rest_stream:             false,
            pagination_order:        None,
        }],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let result = SchemaConverter::convert(intermediate);
    assert!(result.is_err(), "expected Err, got: {result:?}");
    assert!(result.expect_err("test").to_string().contains("unknown type 'UnknownType'"));
}

#[test]
fn test_convert_query_with_arguments() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "User".to_string(),
            sql_source:             None,
            fields:                 vec![],
            description:            None,
            implements:             vec![],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![IntermediateQuery {
            function: None,

            requires_actor:          Vec::new(),
            count:                   false,
            name:                    "users".to_string(),
            return_type:             "User".to_string(),
            returns_list:            true,
            nullable:                false,
            arguments:               vec![IntermediateArgument {
                name:        "limit".to_string(),
                arg_type:    "Int".to_string(),
                nullable:    false,
                default:     Some(serde_json::json!(10)),
                description: None,
                deprecated:  None,
                localized:   false,
            }],
            description:             Some("Get users".to_string()),
            sql_source:              Some("v_user".to_string()),
            sql_source_kind:         fraiseql_core::schema::SourceKind::Relation,
            auto_params:             Some(IntermediateAutoParams {
                limit:        Some(true),
                offset:       Some(true),
                where_clause: Some(false),
                order_by:     Some(false),
            }),
            deprecated:              None,
            jsonb_column:            None,
            relay:                   false,
            inject:                  IndexMap::default(),
            exclude_inject_defaults: Vec::new(),
            read_routing:            fraiseql_core::db::types::ReadRouting::default(),
            cache_ttl_seconds:       None,
            additional_views:        vec![],
            requires_role:           None,
            relay_cursor_type:       None,
            rest:                    None,
            rest_stream:             false,
            pagination_order:        None,
        }],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    assert_eq!(compiled.queries.len(), 1);
    assert_eq!(compiled.queries[0].arguments.len(), 1);
    assert_eq!(compiled.queries[0].arguments[0].arg_type, FieldType::Int);
    assert!(compiled.queries[0].auto_params.has_limit);
}

#[test]
fn test_list_query_without_auto_params_defaults_to_all() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "Item".to_string(),
            sql_source:             None,
            fields:                 vec![],
            description:            None,
            implements:             vec![],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![IntermediateQuery {
            function: None,

            requires_actor:          Vec::new(),
            count:                   false,
            name:                    "items".to_string(),
            return_type:             "Item".to_string(),
            returns_list:            true,
            nullable:                false,
            arguments:               vec![],
            description:             None,
            sql_source:              Some("v_item".to_string()),
            sql_source_kind:         fraiseql_core::schema::SourceKind::Relation,
            auto_params:             None,
            deprecated:              None,
            jsonb_column:            None,
            relay:                   false,
            inject:                  IndexMap::default(),
            exclude_inject_defaults: Vec::new(),
            read_routing:            fraiseql_core::db::types::ReadRouting::default(),
            cache_ttl_seconds:       None,
            additional_views:        vec![],
            requires_role:           None,
            relay_cursor_type:       None,
            rest:                    None,
            rest_stream:             false,
            pagination_order:        None,
        }],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    let params = &compiled.queries[0].auto_params;
    assert!(params.has_limit);
    assert!(params.has_offset);
    assert!(params.has_where);
    assert!(params.has_order_by);
}

#[test]
fn test_single_item_query_without_auto_params_defaults_to_none() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "Item".to_string(),
            sql_source:             None,
            fields:                 vec![],
            description:            None,
            implements:             vec![],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![IntermediateQuery {
            function: None,

            requires_actor:          Vec::new(),
            count:                   false,
            name:                    "item".to_string(),
            return_type:             "Item".to_string(),
            returns_list:            false,
            nullable:                true,
            arguments:               vec![],
            description:             None,
            sql_source:              Some("v_item".to_string()),
            sql_source_kind:         fraiseql_core::schema::SourceKind::Relation,
            auto_params:             None,
            deprecated:              None,
            jsonb_column:            None,
            relay:                   false,
            inject:                  IndexMap::default(),
            exclude_inject_defaults: Vec::new(),
            read_routing:            fraiseql_core::db::types::ReadRouting::default(),
            cache_ttl_seconds:       None,
            additional_views:        vec![],
            requires_role:           None,
            relay_cursor_type:       None,
            rest:                    None,
            rest_stream:             false,
            pagination_order:        None,
        }],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    let params = &compiled.queries[0].auto_params;
    assert!(!params.has_limit);
    assert!(!params.has_offset);
    assert!(!params.has_where);
    assert!(!params.has_order_by);
}

#[test]
fn test_convert_field_with_deprecated_directive() {
    use crate::schema::intermediate::IntermediateAppliedDirective;

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "User".to_string(),
            sql_source:             None,
            fields:                 vec![
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "oldId".to_string(),
                    field_type:      "Int".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      Some(vec![IntermediateAppliedDirective {
                        name:      "deprecated".to_string(),
                        arguments: Some(serde_json::json!({"reason": "Use 'id' instead"})),
                    }]),
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "id".to_string(),
                    field_type:      "Int".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
            ],
            description:            None,
            implements:             vec![],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    assert_eq!(compiled.types.len(), 1);
    assert_eq!(compiled.types[0].fields.len(), 2);

    // Check deprecated field
    let old_id_field = &compiled.types[0].fields[0];
    assert_eq!(old_id_field.name, "oldId");
    assert!(old_id_field.is_deprecated());
    assert_eq!(old_id_field.deprecation_reason(), Some("Use 'id' instead"));

    // Check non-deprecated field
    let id_field = &compiled.types[0].fields[1];
    assert_eq!(id_field.name, "id");
    assert!(!id_field.is_deprecated());
    assert_eq!(id_field.deprecation_reason(), None);
}

#[test]
fn test_convert_enum() {
    use crate::schema::intermediate::{
        IntermediateDeprecation, IntermediateEnum, IntermediateEnumValue,
    };

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![],
        enums:             vec![IntermediateEnum {
            name:        "OrderStatus".to_string(),
            values:      vec![
                IntermediateEnumValue {
                    name:        "PENDING".to_string(),
                    description: None,
                    deprecated:  None,
                    value:       None,
                },
                IntermediateEnumValue {
                    name:        "PROCESSING".to_string(),
                    description: Some("Currently being processed".to_string()),
                    deprecated:  None,
                    value:       None,
                },
                IntermediateEnumValue {
                    name:        "CANCELLED".to_string(),
                    description: None,
                    deprecated:  Some(IntermediateDeprecation {
                        reason: Some("Use VOIDED instead".to_string()),
                    }),
                    value:       None,
                },
            ],
            description: Some("Order status enum".to_string()),
        }],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    assert_eq!(compiled.enums.len(), 1);

    let status_enum = &compiled.enums[0];
    assert_eq!(status_enum.name, "OrderStatus");
    assert_eq!(status_enum.description, Some("Order status enum".to_string()));
    assert_eq!(status_enum.values.len(), 3);

    // Check PENDING value
    assert_eq!(status_enum.values[0].name, "PENDING");
    assert!(!status_enum.values[0].is_deprecated());

    // Check PROCESSING value with description
    assert_eq!(status_enum.values[1].name, "PROCESSING");
    assert_eq!(status_enum.values[1].description, Some("Currently being processed".to_string()));

    // Check CANCELLED deprecated value
    assert_eq!(status_enum.values[2].name, "CANCELLED");
    assert!(status_enum.values[2].is_deprecated());
}

#[test]
fn test_convert_input_object() {
    use crate::schema::intermediate::{
        IntermediateDeprecation, IntermediateInputField, IntermediateInputObject,
    };

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![],
        enums:             vec![],
        input_types:       vec![IntermediateInputObject {
            name:        "UserFilter".to_string(),
            fields:      vec![
                IntermediateInputField {
                    name:        "name".to_string(),
                    field_type:  "String".to_string(),
                    nullable:    false,
                    description: None,
                    default:     None,
                    deprecated:  None,
                    localized:   false,
                },
                IntermediateInputField {
                    name:        "active".to_string(),
                    field_type:  "Boolean".to_string(),
                    nullable:    true,
                    description: Some("Filter by active status".to_string()),
                    default:     Some(serde_json::json!(true)),
                    deprecated:  None,
                    localized:   false,
                },
                IntermediateInputField {
                    name:        "oldField".to_string(),
                    field_type:  "String".to_string(),
                    nullable:    true,
                    description: None,
                    default:     None,
                    deprecated:  Some(IntermediateDeprecation {
                        reason: Some("Use newField instead".to_string()),
                    }),
                    localized:   false,
                },
            ],
            description: Some("User filter input".to_string()),
        }],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    // Exactly the 1 user-defined input type: the compiler must not fabricate
    // input types the runtime cannot serve (#869).
    assert_eq!(compiled.input_types.len(), 1);

    // Find the UserFilter type
    let filter = compiled.input_types.iter().find(|t| t.name == "UserFilter").expect("test");
    assert_eq!(filter.name, "UserFilter");
    assert_eq!(filter.description, Some("User filter input".to_string()));
    assert_eq!(filter.fields.len(), 3);

    // Check name field — nullability must survive the conversion (#414).
    let name_field = filter.find_field("name").expect("test");
    assert_eq!(name_field.field_type, "String");
    assert!(!name_field.is_deprecated());
    assert!(
        !name_field.nullable,
        "non-null input field must compile to nullable=false (#414)"
    );
    assert!(name_field.is_required(), "non-null field with no default is required (#414)");

    // Check active field with default value
    let active_field = filter.find_field("active").expect("test");
    assert_eq!(active_field.field_type, "Boolean");
    assert_eq!(active_field.default_value, Some("true".to_string()));
    assert_eq!(active_field.description, Some("Filter by active status".to_string()));
    assert!(
        active_field.nullable,
        "nullable input field must compile to nullable=true (#414)"
    );
    assert!(!active_field.is_required(), "nullable field is not required (#414)");

    // Check deprecated field
    let old_field = filter.find_field("oldField").expect("test");
    assert!(old_field.is_deprecated());
}

/// #869 contract test — the compiler must not advertise WHERE operators the
/// runtime cannot serve.
///
/// The `<RichType>WhereInput` surface (48 rich scalar types x 35 operator
/// names) was emitted unconditionally while `WhereOperator::from_str` — the
/// only path by which a `where` operator becomes SQL — could parse almost none
/// of them, and two (`depthEq`, `overlaps`) silently bound to unrelated
/// ltree/inet operators. The emission is gone; this test keeps it gone, and
/// keeps every operator-shaped field any input type does advertise parseable
/// by the runtime.
#[test]
fn compiled_schema_advertises_no_unservable_where_operators() {
    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");

    // An empty authored schema compiles to zero input types — nothing fabricated.
    assert!(
        compiled.input_types.is_empty(),
        "compiler fabricated input types for an empty schema: {:?}",
        compiled.input_types.iter().map(|t| t.name.as_str()).collect::<Vec<_>>()
    );

    // And any `*WhereInput` type that IS compiled may only advertise operator
    // names the runtime WHERE parser accepts.
    for input_type in &compiled.input_types {
        if !input_type.name.ends_with("WhereInput") {
            continue;
        }
        for field in &input_type.fields {
            assert!(
                fraiseql_core::db::where_clause::WhereOperator::from_str(&field.name).is_ok(),
                "input type '{}' advertises operator '{}', which \
                 WhereOperator::from_str cannot parse — the request would fail at \
                 runtime with 'Unknown WHERE operator'",
                input_type.name,
                field.name
            );
        }
    }
}

#[test]
fn test_convert_interface() {
    use crate::schema::intermediate::{IntermediateField, IntermediateInterface};

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![IntermediateInterface {
            name:        "Node".to_string(),
            fields:      vec![IntermediateField {
                function: None,

                deprecated:      None,
                vector_config:   None,
                vector_distance: None,
                name:            "id".to_string(),
                field_type:      "ID".to_string(),
                nullable:        false,
                description:     None,
                directives:      None,
                requires_scope:  None,
                on_deny:         None,
                authorize:       None,
                hierarchy:       None,
                unique:          None,
                localized:       false,
            }],
            description: Some("An object with a globally unique ID".to_string()),
        }],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");
    assert_eq!(compiled.interfaces.len(), 1);

    let interface = &compiled.interfaces[0];
    assert_eq!(interface.name, "Node");
    assert_eq!(interface.description, Some("An object with a globally unique ID".to_string()));
    assert_eq!(interface.fields.len(), 1);
    assert_eq!(interface.fields[0].name, "id");
    assert_eq!(interface.fields[0].field_type, FieldType::Id);
}

#[test]
fn test_convert_type_implements_interface() {
    use crate::schema::intermediate::{IntermediateField, IntermediateInterface, IntermediateType};

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "User".to_string(),
            sql_source:             None,
            fields:                 vec![
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "id".to_string(),
                    field_type:      "ID".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "name".to_string(),
                    field_type:      "String".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
            ],
            description:            None,
            implements:             vec!["Node".to_string()],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![IntermediateInterface {
            name:        "Node".to_string(),
            fields:      vec![IntermediateField {
                function: None,

                deprecated:      None,
                vector_config:   None,
                vector_distance: None,
                name:            "id".to_string(),
                field_type:      "ID".to_string(),
                nullable:        false,
                description:     None,
                directives:      None,
                requires_scope:  None,
                on_deny:         None,
                authorize:       None,
                hierarchy:       None,
                unique:          None,
                localized:       false,
            }],
            description: None,
        }],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");

    // Check type implements interface
    assert_eq!(compiled.types.len(), 1);
    assert_eq!(compiled.types[0].implements, vec!["Node"]);

    // Check interface exists
    assert_eq!(compiled.interfaces.len(), 1);
    assert_eq!(compiled.interfaces[0].name, "Node");
}

#[test]
fn test_validate_unknown_interface() {
    use crate::schema::intermediate::{IntermediateField, IntermediateType};

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "User".to_string(),
            sql_source:             None,
            fields:                 vec![IntermediateField {
                function: None,

                deprecated:      None,
                vector_config:   None,
                vector_distance: None,
                name:            "id".to_string(),
                field_type:      "ID".to_string(),
                nullable:        false,
                description:     None,
                directives:      None,
                requires_scope:  None,
                on_deny:         None,
                authorize:       None,
                hierarchy:       None,
                unique:          None,
                localized:       false,
            }],
            description:            None,
            implements:             vec!["UnknownInterface".to_string()],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![], // No interface defined!
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let result = SchemaConverter::convert(intermediate);
    assert!(result.is_err(), "expected Err, got: {result:?}");
    assert!(result.expect_err("test").to_string().contains("unknown interface"));
}

#[test]
fn test_validate_missing_interface_field() {
    use crate::schema::intermediate::{IntermediateField, IntermediateInterface, IntermediateType};

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "User".to_string(),
            sql_source:             None,
            fields:                 vec![
                // Missing the required 'id' field from Node interface!
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "name".to_string(),
                    field_type:      "String".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
            ],
            description:            None,
            implements:             vec!["Node".to_string()],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![IntermediateInterface {
            name:        "Node".to_string(),
            fields:      vec![IntermediateField {
                function: None,

                deprecated:      None,
                vector_config:   None,
                vector_distance: None,
                name:            "id".to_string(),
                field_type:      "ID".to_string(),
                nullable:        false,
                description:     None,
                directives:      None,
                requires_scope:  None,
                on_deny:         None,
                authorize:       None,
                hierarchy:       None,
                unique:          None,
                localized:       false,
            }],
            description: None,
        }],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let result = SchemaConverter::convert(intermediate);
    assert!(result.is_err(), "expected Err, got: {result:?}");
    assert!(result.expect_err("test").to_string().contains("missing field 'id'"));
}

#[test]
fn test_convert_union() {
    use crate::schema::intermediate::{IntermediateField, IntermediateType, IntermediateUnion};

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![
            IntermediateType {
                subscription_policy:    None,
                name:                   "User".to_string(),
                sql_source:             None,
                fields:                 vec![IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "id".to_string(),
                    field_type:      "ID".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                }],
                description:            None,
                implements:             vec![],
                requires_role:          None,
                is_error:               false,
                is_input:               false,
                relay:                  false,
                embedded:               false,
                subscribable_tables:    None,
                subscribable_pre_image: false,
                inject_params:          indexmap::IndexMap::new(),
                relationships:          Vec::new(),
            },
            IntermediateType {
                subscription_policy:    None,
                name:                   "Post".to_string(),
                sql_source:             None,
                fields:                 vec![IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "id".to_string(),
                    field_type:      "ID".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                }],
                description:            None,
                implements:             vec![],
                requires_role:          None,
                is_error:               false,
                is_input:               false,
                relay:                  false,
                embedded:               false,
                subscribable_tables:    None,
                subscribable_pre_image: false,
                inject_params:          indexmap::IndexMap::new(),
                relationships:          Vec::new(),
            },
        ],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![IntermediateUnion {
            name:         "SearchResult".to_string(),
            member_types: vec!["User".to_string(), "Post".to_string()],
            description:  Some("Result from a search query".to_string()),
        }],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");

    // Check union exists
    assert_eq!(compiled.unions.len(), 1);
    let union_def = &compiled.unions[0];
    assert_eq!(union_def.name, "SearchResult");
    assert_eq!(union_def.member_types, vec!["User", "Post"]);
    assert_eq!(union_def.description, Some("Result from a search query".to_string()));
}

#[test]
fn test_convert_field_requires_scope() {
    use crate::schema::intermediate::{IntermediateField, IntermediateType};

    let intermediate = IntermediateSchema {
        grpc_config:       None,
        security:          None,
        auth:              None,
        version:           "2.0.0".to_string(),
        types:             vec![IntermediateType {
            subscription_policy:    None,
            name:                   "Employee".to_string(),
            sql_source:             None,
            fields:                 vec![
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "id".to_string(),
                    field_type:      "ID".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "name".to_string(),
                    field_type:      "String".to_string(),
                    nullable:        false,
                    description:     None,
                    directives:      None,
                    requires_scope:  None,
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "salary".to_string(),
                    field_type:      "Float".to_string(),
                    nullable:        false,
                    description:     Some("Employee salary - protected field".to_string()),
                    directives:      None,
                    requires_scope:  Some("read:Employee.salary".to_string()),
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
                IntermediateField {
                    function: None,

                    deprecated:      None,
                    vector_config:   None,
                    vector_distance: None,
                    name:            "ssn".to_string(),
                    field_type:      "String".to_string(),
                    nullable:        true,
                    description:     Some("Social Security Number - highly protected".to_string()),
                    directives:      None,
                    requires_scope:  Some("admin".to_string()),
                    on_deny:         None,
                    authorize:       None,
                    hierarchy:       None,
                    unique:          None,
                    localized:       false,
                },
            ],
            description:            None,
            implements:             vec![],
            requires_role:          None,
            is_error:               false,
            is_input:               false,
            relay:                  false,
            embedded:               false,
            subscribable_tables:    None,
            subscribable_pre_image: false,
            inject_params:          indexmap::IndexMap::new(),
            relationships:          Vec::new(),
        }],
        enums:             vec![],
        input_types:       vec![],
        interfaces:        vec![],
        unions:            vec![],
        queries:           vec![],
        mutations:         vec![],
        subscriptions:     vec![],
        fragments:         None,
        directives:        None,
        fact_tables:       None,
        aggregate_queries: None,
        observers:         None,

        sources:              None,
        functions:            None,
        functions_config:     None,
        custom_scalars:       None,
        observers_config:     None,
        subscriptions_config: None,
        validation_config:    None,
        federation_config:    None,
        debug_config:         None,
        mcp_config:           None,
        rest_config:          None,
        query_defaults:       None,
        inject_defaults:      None,
        naming_convention:    NamingConvention::default(),
        session_variables:    None,
        locale:               None,
        hierarchies_config:   None,
        changelog_config:     None,
    };

    let compiled = SchemaConverter::convert(intermediate).expect("test");

    assert_eq!(compiled.types.len(), 1);
    let employee_type = &compiled.types[0];
    assert_eq!(employee_type.name, "Employee");
    assert_eq!(employee_type.fields.len(), 4);

    // id field - no scope required
    assert_eq!(employee_type.fields[0].name, "id");
    assert!(employee_type.fields[0].requires_scope.is_none());

    // name field - no scope required
    assert_eq!(employee_type.fields[1].name, "name");
    assert!(employee_type.fields[1].requires_scope.is_none());

    // salary field - requires specific scope
    assert_eq!(employee_type.fields[2].name, "salary");
    assert_eq!(employee_type.fields[2].requires_scope, Some("read:Employee.salary".to_string()));

    // ssn field - requires admin scope
    assert_eq!(employee_type.fields[3].name, "ssn");
    assert_eq!(employee_type.fields[3].requires_scope, Some("admin".to_string()));
}

// ── promote_input_type_args (#456 Option 2: IR honesty) ──────────────────────

#[test]
fn promote_input_type_args_rewrites_object_to_input() {
    use fraiseql_core::schema::{
        ArgumentDefinition, CompiledSchema, FieldType, InputObjectDefinition, MutationDefinition,
    };

    let arg = |arg_type| ArgumentDefinition {
        name: "input".to_string(),
        arg_type,
        nullable: false,
        default_value: None,
        description: None,
        deprecation: None,
        localized: false,
    };

    let mut schema = CompiledSchema {
        input_types: vec![InputObjectDefinition {
            name:        "CreateOrderInput".to_string(),
            fields:      vec![],
            description: None,
            metadata:    None,
        }],
        ..Default::default()
    };
    // (1) arg naming an input type → Input
    let mut m1 = MutationDefinition::new("createOrder", "Order");
    m1.arguments = vec![arg(FieldType::Object("CreateOrderInput".to_string()))];
    // (2) arg naming an OUTPUT object type → stays Object
    let mut m2 = MutationDefinition::new("touchOrder", "Order");
    m2.arguments = vec![arg(FieldType::Object("Order".to_string()))];
    // (3) list of an input type → inner promoted to Input
    let mut m3 = MutationDefinition::new("bulkCreate", "Order");
    m3.arguments = vec![arg(FieldType::list(FieldType::Object(
        "CreateOrderInput".to_string(),
    )))];
    schema.mutations = vec![m1, m2, m3];

    SchemaConverter::promote_input_type_args(&mut schema);

    assert_eq!(
        schema.mutations[0].arguments[0].arg_type,
        FieldType::Input("CreateOrderInput".to_string()),
        "an Object naming a registered input type must become Input"
    );
    assert_eq!(
        schema.mutations[1].arguments[0].arg_type,
        FieldType::Object("Order".to_string()),
        "an Object naming an output type must stay Object"
    );
    assert_eq!(
        schema.mutations[2].arguments[0].arg_type,
        FieldType::list(FieldType::Input("CreateOrderInput".to_string())),
        "a list of an input type must promote its element to Input"
    );
}

// ── tenancy converter tests ─────────────────────────────────────────────────

mod tenancy_tests {
    use indexmap::IndexMap;

    use super::super::tenancy::{AnnotatedTypeIndex, validate_tenant_annotations};
    use crate::schema::intermediate::{
        IntermediateField, IntermediateMutation, IntermediateQuery, IntermediateSchema,
        IntermediateType, fragments::IntermediateAppliedDirective,
    };

    fn make_type(name: &str, fields: Vec<IntermediateField>) -> IntermediateType {
        IntermediateType {
            subscription_policy: None,
            name: name.to_string(),
            sql_source: None,
            fields,
            description: None,
            implements: vec![],
            requires_role: None,
            is_error: false,
            is_input: false,
            relay: false,
            embedded: false,
            subscribable_tables: None,
            subscribable_pre_image: false,
            inject_params: indexmap::IndexMap::new(),
            relationships: Vec::new(),
        }
    }

    fn make_field(name: &str, field_type: &str) -> IntermediateField {
        IntermediateField {
            function: None,

            deprecated:      None,
            vector_config:   None,
            vector_distance: None,
            name:            name.to_string(),
            field_type:      field_type.to_string(),
            nullable:        false,
            description:     None,
            directives:      None,
            requires_scope:  None,
            on_deny:         None,
            authorize:       None,
            hierarchy:       None,
            unique:          None,
            localized:       false,
        }
    }

    fn make_tenant_id_field(name: &str) -> IntermediateField {
        IntermediateField {
            function: None,

            deprecated:      None,
            vector_config:   None,
            vector_distance: None,
            name:            name.to_string(),
            field_type:      "String".to_string(),
            nullable:        false,
            description:     None,
            directives:      Some(vec![IntermediateAppliedDirective {
                name:      "tenant_id".to_string(),
                arguments: None,
            }]),
            requires_scope:  None,
            on_deny:         None,
            authorize:       None,
            hierarchy:       None,
            unique:          None,
            localized:       false,
        }
    }

    fn make_query(name: &str, return_type: &str) -> IntermediateQuery {
        IntermediateQuery {
            name: name.to_string(),
            return_type: return_type.to_string(),
            ..Default::default()
        }
    }

    fn make_mutation(name: &str, return_type: &str) -> IntermediateMutation {
        IntermediateMutation {
            name: name.to_string(),
            return_type: return_type.to_string(),
            ..Default::default()
        }
    }

    fn make_schema(
        types: Vec<IntermediateType>,
        queries: Vec<IntermediateQuery>,
        mutations: Vec<IntermediateMutation>,
    ) -> IntermediateSchema {
        IntermediateSchema {
            types,
            queries,
            mutations,
            ..Default::default()
        }
    }

    // ── AnnotatedTypeIndex ──────────────────────────────────────────────

    #[test]
    fn index_empty_when_no_annotations() {
        let types = vec![make_type("User", vec![make_field("id", "Int")])];
        let index = AnnotatedTypeIndex::build(&types);
        assert!(!index.has_annotations());
    }

    #[test]
    fn index_detects_tenant_id_field() {
        let types = vec![make_type(
            "User",
            vec![make_field("id", "Int"), make_tenant_id_field("tenant_id")],
        )];
        let index = AnnotatedTypeIndex::build(&types);
        assert!(index.has_annotations());
        let fields = index.fields_for_type("User").unwrap();
        assert!(fields.contains("tenant_id"));
    }

    #[test]
    fn index_multiple_types_independently() {
        let types = vec![
            make_type("User", vec![make_tenant_id_field("tenant_id")]),
            make_type("Post", vec![make_field("id", "Int")]),
            make_type("Order", vec![make_tenant_id_field("org_id")]),
        ];
        let index = AnnotatedTypeIndex::build(&types);
        assert!(index.fields_for_type("User").is_some());
        assert!(index.fields_for_type("Post").is_none());
        assert!(index.fields_for_type("Order").is_some());
        assert!(index.fields_for_type("Order").unwrap().contains("org_id"));
    }

    // ── Auto-injection ──────────────────────────────────────────────────

    /// #1386: row isolation that matches no type is refused, naming the scope of the
    /// no-op and the setting for isolation enforced by database RLS.
    #[test]
    fn row_tenancy_with_no_annotated_type_is_refused() {
        let mut schema = make_schema(
            vec![make_type("User", vec![make_field("id", "Int")])],
            vec![make_query("getUser", "User")],
            vec![],
        );
        let message =
            validate_tenant_annotations(&mut schema, "tenant_id").unwrap_err().to_string();
        assert!(message.contains("matched 0 of 1 types"), "{message}");
        assert!(message.contains("multi_tenant = true"), "{message}");
        assert!(schema.queries[0].inject.is_empty(), "nothing may be injected on refusal");
    }

    #[test]
    fn auto_injects_query_when_inject_empty() {
        let mut schema = make_schema(
            vec![make_type(
                "User",
                vec![make_field("id", "Int"), make_tenant_id_field("tenant_id")],
            )],
            vec![make_query("getUser", "User")],
            vec![],
        );
        validate_tenant_annotations(&mut schema, "tenant_id").unwrap();
        assert_eq!(schema.queries[0].inject.get("tenant_id"), Some(&"jwt:tenant_id".to_string()));
    }

    #[test]
    fn auto_injects_mutation_when_inject_empty() {
        let mut schema = make_schema(
            vec![make_type(
                "User",
                vec![make_field("id", "Int"), make_tenant_id_field("tenant_id")],
            )],
            vec![],
            vec![make_mutation("createUser", "User")],
        );
        validate_tenant_annotations(&mut schema, "tenant_id").unwrap();
        assert_eq!(schema.mutations[0].inject.get("tenant_id"), Some(&"jwt:tenant_id".to_string()));
    }

    // ── input_style threading (flatten default / explicit jsonb) ─────────

    /// An absent / unset `input_style` converts to the `Flatten` default.
    #[test]
    fn convert_mutation_defaults_input_style_to_flatten() {
        use fraiseql_core::schema::InputStyle;

        use crate::schema::SchemaConverter;
        let md = SchemaConverter::convert_mutation(
            make_mutation("createUser", "User"),
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap();
        assert_eq!(md.input_style, InputStyle::Flatten);
    }

    /// `input_style = jsonb` threads through orthogonally to the real DML verb:
    /// the mutation keeps its `Insert` operation (so the Change Spine logs the
    /// true verb) while opting into single-JSONB input passing.
    #[test]
    fn convert_mutation_threads_jsonb_input_style_with_real_verb() {
        use fraiseql_core::schema::{InputStyle, MutationOperation};

        use crate::schema::SchemaConverter;
        let im = IntermediateMutation {
            operation: Some("INSERT".to_string()),
            sql_source: Some("create_user".to_string()),
            input_style: InputStyle::Jsonb,
            ..make_mutation("createUser", "User")
        };
        let md = SchemaConverter::convert_mutation(
            im,
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap();
        assert_eq!(md.input_style, InputStyle::Jsonb);
        assert!(matches!(md.operation, MutationOperation::Insert { .. }));
    }

    /// The authoring JSON contract: lowercase `"flatten"`/`"jsonb"`, with an
    /// absent key defaulting to `flatten` (byte-identical to pre-`input_style`).
    #[test]
    fn intermediate_mutation_input_style_json_contract() {
        use fraiseql_core::schema::InputStyle;
        let absent: IntermediateMutation =
            serde_json::from_str(r#"{ "name": "m", "return_type": "R" }"#).unwrap();
        assert_eq!(absent.input_style, InputStyle::Flatten);
        let jsonb: IntermediateMutation =
            serde_json::from_str(r#"{ "name": "m", "return_type": "R", "input_style": "jsonb" }"#)
                .unwrap();
        assert_eq!(jsonb.input_style, InputStyle::Jsonb);
    }

    // ── changelog_pre_image threading (default off / explicit on) ─────────

    /// An absent / unset `changelog_pre_image` converts to `false` (after-image
    /// only — byte-identical to the behavior before the flag existed).
    #[test]
    fn convert_mutation_defaults_changelog_pre_image_to_false() {
        use crate::schema::SchemaConverter;
        let md = SchemaConverter::convert_mutation(
            make_mutation("createUser", "User"),
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap();
        assert!(!md.changelog_pre_image);
    }

    /// `changelog_pre_image = true` threads through to the compiled mutation so the
    /// outbox CTE records the entity's pre-image into `object_data_before`.
    #[test]
    fn convert_mutation_threads_changelog_pre_image() {
        use crate::schema::SchemaConverter;
        let im = IntermediateMutation {
            changelog_pre_image: true,
            ..make_mutation("updatePrice", "Price")
        };
        let md = SchemaConverter::convert_mutation(
            im,
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap();
        assert!(md.changelog_pre_image);
    }

    /// The authoring JSON contract: an absent key defaults to `false`; `true`
    /// threads through.
    #[test]
    fn intermediate_mutation_changelog_pre_image_json_contract() {
        let absent: IntermediateMutation =
            serde_json::from_str(r#"{ "name": "m", "return_type": "R" }"#).unwrap();
        assert!(!absent.changelog_pre_image);
        let on: IntermediateMutation = serde_json::from_str(
            r#"{ "name": "m", "return_type": "R", "changelog_pre_image": true }"#,
        )
        .unwrap();
        assert!(on.changelog_pre_image);
    }

    // ── cascade threading (default off / explicit on, gate decision 3) ────

    /// An absent / unset `cascade` converts to `false` (no typed cascade
    /// surface — byte-identical to the behavior before the flag existed).
    #[test]
    fn convert_mutation_defaults_cascade_to_false() {
        use crate::schema::SchemaConverter;
        let md = SchemaConverter::convert_mutation(
            make_mutation("createPost", "Post"),
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap();
        assert!(!md.cascade);
    }

    /// `cascade = true` threads through to the compiled mutation so the runtime
    /// exposes and enforces the typed cascade field. Before this, the compiler
    /// silently dropped the SDK flag (eval finding 4).
    #[test]
    fn convert_mutation_threads_cascade() {
        use crate::schema::SchemaConverter;
        let im = IntermediateMutation {
            cascade: true,
            ..make_mutation("createPost", "Post")
        };
        let md = SchemaConverter::convert_mutation(
            im,
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap();
        assert!(md.cascade);
    }

    /// The authoring JSON contract: an absent key defaults to `false`; `true`
    /// threads through. This is the key the Python/TS SDKs write for
    /// `@fraiseql.type(crud=True, cascade=True)`.
    #[test]
    fn intermediate_mutation_cascade_json_contract() {
        let absent: IntermediateMutation =
            serde_json::from_str(r#"{ "name": "m", "return_type": "R" }"#).unwrap();
        assert!(!absent.cascade);
        let on: IntermediateMutation =
            serde_json::from_str(r#"{ "name": "m", "return_type": "R", "cascade": true }"#)
                .unwrap();
        assert!(on.cascade);
    }

    // ── cascade type synthesis (typed payload-wrapper surface) ──

    /// A view-backed entity type with the given name (queryable — so it
    /// auto-implements `CascadeNode`).
    fn make_entity_type(name: &str) -> IntermediateType {
        IntermediateType {
            sql_source: Some(format!("v_{}", name.to_lowercase())),
            ..make_type(name, vec![make_field("id", "ID")])
        }
    }

    /// A `cascade = true` mutation synthesizes the spec-aligned typed surface: the
    /// `CascadeNode` interface (auto-implemented on every queryable entity), the
    /// `CascadeOperation` enum, the `UpdatedEntity`/`DeletedEntity`/`CascadeUpdates`
    /// envelope, and a `<Name>Payload { entity, cascade, updatedFields }` wrapper.
    #[test]
    fn cascade_synthesis_builds_typed_surface() {
        use fraiseql_core::schema::FieldType;

        use crate::schema::SchemaConverter;
        let create_post = IntermediateMutation {
            cascade: true,
            sql_source: Some("fn_create_post".to_string()),
            ..make_mutation("createPost", "Post")
        };
        let schema = make_schema(vec![make_entity_type("Post")], vec![], vec![create_post]);
        let compiled = SchemaConverter::convert(schema).expect("convert");

        // CascadeNode interface exists and the queryable entity implements it.
        assert!(compiled.interfaces.iter().any(|i| i.name == "CascadeNode"));
        let post = compiled.types.iter().find(|t| t.name.as_str() == "Post").unwrap();
        assert!(
            post.implements.iter().any(|i| i == "CascadeNode"),
            "Post implements CascadeNode"
        );

        // CascadeOperation enum with the spec's three values.
        let op = compiled.enums.iter().find(|e| e.name == "CascadeOperation").expect("enum");
        for v in ["CREATED", "UPDATED", "DELETED"] {
            assert!(op.has_value(v), "CascadeOperation has {v}");
        }

        // UpdatedEntity carries operation + a non-null typed entity; DeletedEntity
        // carries no entity body (a deleted row has nothing to project).
        let updated = compiled.types.iter().find(|t| t.name.as_str() == "UpdatedEntity").unwrap();
        assert!(updated.find_field("operation").is_some(), "UpdatedEntity.operation");
        let entity_field = updated.find_field("entity").expect("UpdatedEntity.entity");
        assert!(!entity_field.nullable, "UpdatedEntity.entity is non-null");
        let deleted = compiled.types.iter().find(|t| t.name.as_str() == "DeletedEntity").unwrap();
        assert!(deleted.find_field("entity").is_none(), "DeletedEntity has no entity body");
        assert!(deleted.find_field("deletedAt").is_some(), "DeletedEntity.deletedAt");

        // CascadeUpdates references the split entry types + a metadata envelope.
        let updates = compiled.types.iter().find(|t| t.name.as_str() == "CascadeUpdates").unwrap();
        let updated_field = updates.find_field("updated").expect("CascadeUpdates.updated");
        assert!(
            matches!(&updated_field.field_type, FieldType::List { item: inner, .. }
                if matches!(inner.as_ref(), FieldType::Object(n) if n == "UpdatedEntity")),
            "updated: [UpdatedEntity!]!"
        );
        assert!(updates.find_field("metadata").is_some(), "CascadeUpdates.metadata");
        assert!(updates.find_field("invalidations").is_some(), "CascadeUpdates.invalidations");
        let meta = compiled.types.iter().find(|t| t.name.as_str() == "CascadeMetadata").unwrap();
        for f in ["timestamp", "depth", "affectedCount", "truncated"] {
            assert!(meta.find_field(f).is_some(), "CascadeMetadata.{f}");
        }
        // The invalidation surface (type + its two enums) is synthesized.
        assert!(compiled.types.iter().any(|t| t.name.as_str() == "QueryInvalidation"));
        assert!(compiled.enums.iter().any(|e| e.name == "InvalidationStrategy"));
        assert!(compiled.enums.iter().any(|e| e.name == "InvalidationScope"));

        // The mutation returns CreatePostPayload { entity, cascade, updatedFields }.
        let m = compiled.mutations.iter().find(|m| m.name == "createPost").unwrap();
        assert_eq!(m.return_type, "CreatePostPayload");
        let payload =
            compiled.types.iter().find(|t| t.name.as_str() == "CreatePostPayload").unwrap();
        assert!(payload.find_field("entity").is_some(), "payload has entity");
        assert!(payload.find_field("cascade").is_some(), "payload has cascade");
        assert!(payload.find_field("updatedFields").is_some(), "payload rehomes updatedFields");
    }

    /// No `cascade = true` mutation ⇒ the pass is inert: no cascade types/enum, no
    /// `implements` churn, mutation return type unchanged (byte-identical to a
    /// schema compiled before the pass existed).
    #[test]
    fn cascade_synthesis_inert_without_cascade_mutations() {
        use crate::schema::SchemaConverter;
        let create_post = IntermediateMutation {
            sql_source: Some("fn_create_post".to_string()),
            ..make_mutation("createPost", "Post")
        };
        let schema = make_schema(vec![make_entity_type("Post")], vec![], vec![create_post]);
        let compiled = SchemaConverter::convert(schema).expect("convert");

        assert!(!compiled.interfaces.iter().any(|i| i.name == "CascadeNode"));
        assert!(!compiled.types.iter().any(|t| t.name.as_str() == "CascadeUpdates"));
        assert!(!compiled.enums.iter().any(|e| e.name == "CascadeOperation"));
        let m = compiled.mutations.iter().find(|m| m.name == "createPost").unwrap();
        assert_eq!(m.return_type, "Post");
        let post = compiled.types.iter().find(|t| t.name.as_str() == "Post").unwrap();
        assert!(!post.implements.iter().any(|i| i == "CascadeNode"));
    }

    // ── Entity-identity contract (ADR-0017) + Node conformance backstop ────

    /// A view-backed entity with a chosen `id` field type — for identity tests.
    fn entity_with_id(name: &str, id_ty: &str) -> IntermediateType {
        IntermediateType {
            sql_source: Some(format!("v_{}", name.to_lowercase())),
            ..make_type(name, vec![make_field("id", id_ty)])
        }
    }

    /// A cascade mutation returning `entity` — the opt-in that triggers synthesis.
    fn cascade_mut(name: &str, entity: &str) -> IntermediateMutation {
        IntermediateMutation {
            cascade: true,
            sql_source: Some(format!("fn_{}", name.to_lowercase())),
            ..make_mutation(name, entity)
        }
    }

    /// The identity contract: the Trinity external id `id: UUID` is canonicalized to
    /// `id: ID` (wire-transparent), independent of any interface. This is what makes
    /// every Trinity entity uniformly identity-bearing.
    #[test]
    fn entity_uuid_id_is_canonicalized_to_id() {
        use fraiseql_core::schema::FieldType;

        use crate::schema::SchemaConverter;
        let schema = make_schema(
            vec![entity_with_id("Order", "UUID")],
            vec![make_query("orders", "Order")],
            vec![],
        );
        let compiled = SchemaConverter::convert(schema).expect("convert");
        let order = compiled.types.iter().find(|t| t.name.as_str() == "Order").unwrap();
        let id = order.find_field("id").expect("id field");
        assert_eq!(id.field_type, FieldType::Id, "id: UUID canonicalizes to id: ID");
    }

    /// The reporter's case, resolved: a cascade entity with the Trinity `id: UUID`
    /// now canonicalizes to `id: ID`, satisfies `CascadeNode`, and compiles — no
    /// hand-editing 64 entities.
    #[test]
    fn cascade_uuid_id_entity_canonicalizes_and_compiles() {
        use crate::schema::SchemaConverter;
        let schema = make_schema(
            vec![entity_with_id("Order", "UUID")],
            vec![],
            vec![cascade_mut("createOrder", "Order")],
        );
        let compiled = SchemaConverter::convert(schema).expect("Trinity UUID cascade compiles");
        let order = compiled.types.iter().find(|t| t.name.as_str() == "Order").unwrap();
        assert!(order.implements.iter().any(|i| i == "CascadeNode"));
    }

    /// A cascade entity with no `id` field at all (e.g. keyed on another column)
    /// cannot be identity-bearing — a legible, actionable hard error (the backstop).
    #[test]
    fn cascade_missing_id_entity_fails_with_actionable_error() {
        use crate::schema::SchemaConverter;
        let order = IntermediateType {
            sql_source: Some("v_order".to_string()),
            ..make_type("Order", vec![make_field("orderNumber", "String")])
        };
        let schema = make_schema(vec![order], vec![], vec![cascade_mut("createOrder", "Order")]);
        let err = SchemaConverter::convert(schema).unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("Order") && msg.contains("no `id` field"), "{msg}");
        assert!(msg.contains("Fix:"), "offers a remedy: {msg}");
    }

    /// #653 (2.14 Phase 02). An embedded value object (`Money`) that the Python SDK gave a
    /// synthesized `sql_source` (`v_money`, a view that does not exist) is flagged as a
    /// cascade entity purely because `is_queryable_entity` keys off `!sql_source.is_empty()`.
    /// The error now names (3) *why* the type was classified an entity — its `sql_source`
    /// signal — and (4) the reference path a cascade mutation reaches it by, so the author
    /// sees the real fix ("Money should not declare a source") instead of the two wrong ones.
    /// (This was the Phase 00 baseline pin, flipped from "omits" to "reports".)
    #[test]
    fn cascade_embedded_value_object_error_reports_derivation_and_path() {
        use crate::schema::SchemaConverter;
        // Order is genuinely view-backed and identity-bearing — it must NOT be flagged.
        let order = IntermediateType {
            sql_source: Some("v_order".to_string()),
            ..make_type("Order", vec![make_field("id", "UUID"), make_field("total", "Money")])
        };
        // Money is an embedded value object: no independent identity, never queried on its
        // own. The SDK synthesized `sql_source = "v_money"` for it and no such view exists.
        let money = IntermediateType {
            sql_source: Some("v_money".to_string()),
            ..make_type(
                "Money",
                vec![
                    make_field("amount", "Int"),
                    make_field("currency", "String"),
                ],
            )
        };
        let schema =
            make_schema(vec![order, money], vec![], vec![cascade_mut("createOrder", "Order")]);
        let msg = format!("{:#}", SchemaConverter::convert(schema).unwrap_err());

        assert!(msg.contains("Money"), "names the flagged type: {msg}");
        assert!(msg.contains("no `id` field"), "states the failed assertion: {msg}");
        // #653(3): the message names the classification signal (its declared sql_source).
        assert!(
            msg.contains("sql_source") && msg.contains("v_money"),
            "#653(3): names the classification signal: {msg}"
        );
        // #653(4): the message names the reference path a cascade mutation reaches it by.
        assert!(
            msg.contains("createOrder → Order.total → Money"),
            "#653(4): names the reference path: {msg}"
        );
    }

    /// #653(4): a flagged queryable entity that no cascade mutation reaches gets the
    /// `sql_source` signal but NO "reached via" line — it is a top-level entity, not one
    /// dragged in by a cascade, so there is no path to name.
    #[test]
    fn cascade_flagged_entity_unreachable_from_a_mutation_has_no_reference_path() {
        use crate::schema::SchemaConverter;
        let order = IntermediateType {
            sql_source: Some("v_order".to_string()),
            ..make_type("Order", vec![make_field("id", "UUID")])
        };
        // Widget is view-backed but has no `id` and is not reachable from `createOrder`.
        let widget = IntermediateType {
            sql_source: Some("v_widget".to_string()),
            ..make_type("Widget", vec![make_field("label", "String")])
        };
        let schema =
            make_schema(vec![order, widget], vec![], vec![cascade_mut("createOrder", "Order")]);
        let msg = format!("{:#}", SchemaConverter::convert(schema).unwrap_err());

        assert!(msg.contains("Widget") && msg.contains("no `id` field"), "{msg}");
        assert!(msg.contains("v_widget"), "#653(3): signal present even off the cascade: {msg}");
        assert!(!msg.contains("reached via"), "#653(4): no path for an unreachable type: {msg}");
    }

    /// An `id: Int` (a serial pk exposed directly — not the Trinity external id) is
    /// not canonicalized and cannot back `id: ID!`; a cascade over it fails fast with
    /// the actual id type and the remedy (adopt a UUID/ID identity).
    #[test]
    fn cascade_int_id_entity_fails_with_actionable_error() {
        use crate::schema::SchemaConverter;
        let schema = make_schema(
            vec![entity_with_id("Order", "Int")],
            vec![],
            vec![cascade_mut("createOrder", "Order")],
        );
        let msg = format!("{:#}", SchemaConverter::convert(schema).unwrap_err());
        assert!(
            msg.contains("Order") && msg.contains("Int"),
            "names type + actual id type: {msg}"
        );
        assert!(msg.contains("id: ID!") && msg.contains("Fix:"), "{msg}");
        // #653(3): a wrong-typed `id` is a genuine entity, not a misclassified value
        // object — it must NOT get the "declare no source" hint (that guard is `has_id`).
        assert!(
            !msg.contains("embedded value object"),
            "wrong-typed id is a real entity, not a value object: {msg}"
        );
    }

    /// One error lists *every* offender, so a developer fixes them in one pass
    /// rather than whack-a-mole.
    #[test]
    fn cascade_conformance_error_aggregates_all_offenders() {
        use crate::schema::SchemaConverter;
        let schema = make_schema(
            vec![
                entity_with_id("Order", "Int"),
                entity_with_id("Invoice", "Int"),
            ],
            vec![],
            vec![cascade_mut("createOrder", "Order")],
        );
        let msg = format!("{:#}", SchemaConverter::convert(schema).unwrap_err());
        assert!(msg.contains("Order") && msg.contains("Invoice"), "lists both offenders: {msg}");
    }

    /// Happy-path guard: an explicit `id: ID!` cascade entity still synthesizes and
    /// implements `CascadeNode` (the contract must not over-correct).
    #[test]
    fn cascade_conformant_id_entity_still_implements_cascade_node() {
        use crate::schema::SchemaConverter;
        let schema = make_schema(
            vec![entity_with_id("Order", "ID")],
            vec![],
            vec![cascade_mut("createOrder", "Order")],
        );
        let compiled =
            SchemaConverter::convert(schema).expect("conformant cascade schema compiles");
        let order = compiled.types.iter().find(|t| t.name.as_str() == "Order").unwrap();
        assert!(order.implements.iter().any(|i| i == "CascadeNode"));
    }

    /// Relay `Node` shares the identity contract: a `relay = true` type with the
    /// Trinity `id: UUID` canonicalizes and compiles.
    #[test]
    fn relay_uuid_id_type_canonicalizes_and_compiles() {
        use crate::schema::SchemaConverter;
        let mut order = entity_with_id("Order", "UUID");
        order.relay = true;
        let schema = make_schema(vec![order], vec![make_query("orders", "Order")], vec![]);
        let compiled = SchemaConverter::convert(schema).expect("Trinity UUID relay type compiles");
        let order = compiled.types.iter().find(|t| t.name.as_str() == "Order").unwrap();
        assert!(order.implements.iter().any(|i| i == "Node"));
    }

    /// A `relay = true` type with a non-identity `id: Int` still fails legibly.
    #[test]
    fn relay_int_id_type_fails_with_actionable_error() {
        use crate::schema::SchemaConverter;
        let mut order = entity_with_id("Order", "Int");
        order.relay = true;
        let schema = make_schema(vec![order], vec![make_query("orders", "Order")], vec![]);
        let msg = format!("{:#}", SchemaConverter::convert(schema).unwrap_err());
        assert!(msg.contains("Order") && msg.contains("relay"), "{msg}");
        assert!(msg.contains("Fix:"), "offers a remedy: {msg}");
    }

    /// Phase-2 soundness invariant: no synthesis pass may emit IR that `validate()`
    /// rejects with a swallowed bail. Across id types × interface-forcing features,
    /// `convert()` must either succeed or fail with a *legible* synthesis error
    /// (a `Fix:` remedy) — never leak the raw validator "…is missing field 'id'".
    #[test]
    fn synthesis_never_leaks_a_raw_validator_bail() {
        use crate::schema::SchemaConverter;

        let mut cases: Vec<(String, bool, IntermediateSchema)> = Vec::new();
        // `ID` and the Trinity `UUID` (canonicalized to `ID`) conform; a non-identity
        // `Int` id does not and must fail *legibly*.
        for (id_ty, expect_ok) in [("ID", true), ("UUID", true), ("Int", false)] {
            cases.push((
                format!("cascade+{id_ty}"),
                expect_ok,
                make_schema(
                    vec![entity_with_id("Order", id_ty)],
                    vec![],
                    vec![cascade_mut("createOrder", "Order")],
                ),
            ));
            let mut order = entity_with_id("Order", id_ty);
            order.relay = true;
            cases.push((
                format!("relay+{id_ty}"),
                expect_ok,
                make_schema(vec![order], vec![make_query("orders", "Order")], vec![]),
            ));
        }

        for (label, expect_ok, schema) in cases {
            match SchemaConverter::convert(schema) {
                Ok(_) => assert!(expect_ok, "{label} unexpectedly compiled"),
                Err(e) => {
                    let msg = format!("{e:#}");
                    assert!(!expect_ok, "{label} unexpectedly failed: {msg}");
                    assert!(msg.contains("Fix:"), "{label} error is not legible: {msg}");
                    assert!(
                        !msg.contains("but is missing field"),
                        "{label} leaked the raw validator bail: {msg}"
                    );
                },
            }
        }
    }

    // ── validate(): interfaces are valid return types (latent-gap fix) ─────

    /// An interface named `Node` with a single `id: ID!` field, for return-type tests.
    fn node_interface() -> crate::schema::intermediate::IntermediateInterface {
        crate::schema::intermediate::IntermediateInterface {
            name:        "Node".to_string(),
            fields:      vec![make_field("id", "ID")],
            description: None,
        }
    }

    /// A query may return an interface (narrowed via inline fragments). Previously
    /// interfaces weren't in the return-type registry, so this failed the reference
    /// check (with a `warn!`); now it converts.
    #[test]
    fn query_returning_an_interface_is_valid() {
        use crate::schema::SchemaConverter;
        let mut schema = make_schema(vec![], vec![make_query("node", "Node")], vec![]);
        schema.interfaces = vec![node_interface()];
        SchemaConverter::convert(schema).expect("a query may return an interface type");
    }

    /// Same for a mutation return type — previously a *silent* bail (no `warn!`).
    #[test]
    fn mutation_returning_an_interface_is_valid() {
        use crate::schema::SchemaConverter;
        let mut schema = make_schema(vec![], vec![], vec![make_mutation("promote", "Node")]);
        schema.interfaces = vec![node_interface()];
        SchemaConverter::convert(schema).expect("a mutation may return an interface type");
    }

    #[test]
    fn auto_inject_uses_custom_claim() {
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![make_query("getUser", "User")],
            vec![],
        );
        validate_tenant_annotations(&mut schema, "org_id").unwrap();
        assert_eq!(schema.queries[0].inject.get("tenant_id"), Some(&"jwt:org_id".to_string()));
    }

    // ── Existing inject accepted ────────────────────────────────────────

    #[test]
    fn existing_inject_with_tenant_field_accepted() {
        let mut inject = IndexMap::new();
        inject.insert("tenant_id".to_string(), "jwt:tenant_id".to_string());
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![IntermediateQuery {
                name: "getUser".to_string(),
                return_type: "User".to_string(),
                inject,
                ..Default::default()
            }],
            vec![],
        );
        validate_tenant_annotations(&mut schema, "tenant_id").unwrap();
    }

    // ── Error: explicit inject missing tenant ───────────────────────────

    #[test]
    fn error_when_inject_overridden_without_tenant() {
        let mut inject = IndexMap::new();
        inject.insert("user_id".to_string(), "jwt:sub".to_string());
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![IntermediateQuery {
                name: "getUser".to_string(),
                return_type: "User".to_string(),
                inject,
                ..Default::default()
            }],
            vec![],
        );
        let err = validate_tenant_annotations(&mut schema, "tenant_id").unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("getUser"), "error should mention query name: {msg}");
        assert!(msg.contains("@tenant_id"), "error should mention directive: {msg}");
        assert!(msg.contains("tenant_id"), "error should mention field: {msg}");
    }

    #[test]
    fn error_mutation_inject_overridden_without_tenant() {
        let mut inject = IndexMap::new();
        inject.insert("user_id".to_string(), "jwt:sub".to_string());
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![],
            vec![IntermediateMutation {
                name: "createUser".to_string(),
                return_type: "User".to_string(),
                inject,
                ..Default::default()
            }],
        );
        let err = validate_tenant_annotations(&mut schema, "tenant_id").unwrap_err();
        assert!(err.to_string().contains("createUser"));
    }

    // ── No-op for non-annotated types ───────────────────────────────────

    #[test]
    fn query_on_non_annotated_type_unchanged() {
        let mut schema = make_schema(
            vec![
                make_type("Post", vec![make_field("id", "Int")]),
                make_type("User", vec![make_tenant_id_field("tenant_id")]),
            ],
            vec![make_query("getPosts", "Post")],
            vec![],
        );
        validate_tenant_annotations(&mut schema, "tenant_id").unwrap();
        assert!(schema.queries[0].inject.is_empty());
    }

    // ── A mutation passes the tenant as a function parameter (#1386) ────

    /// A mutation that already injects the tenant claim, under its function's own
    /// parameter name, is scoped: no argument is added for the annotated field.
    #[test]
    fn a_mutation_injecting_the_tenant_claim_under_another_name_is_left_alone() {
        let mut inject = IndexMap::new();
        inject.insert("p_tenant_id".to_string(), "jwt:tenant_id".to_string());
        inject.insert("p_user_id".to_string(), "jwt:sub".to_string());
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![],
            vec![IntermediateMutation {
                name: "createUser".to_string(),
                return_type: "User".to_string(),
                inject: inject.clone(),
                ..Default::default()
            }],
        );
        validate_tenant_annotations(&mut schema, "tenant_id").unwrap();
        assert_eq!(schema.mutations[0].inject, inject, "the function's arity is untouched");
    }

    /// The refusal for a mutation passing no tenant names the claim it must read.
    #[test]
    fn a_mutation_passing_no_tenant_is_refused_naming_the_claim() {
        let mut inject = IndexMap::new();
        inject.insert("p_user_id".to_string(), "jwt:sub".to_string());
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![],
            vec![IntermediateMutation {
                name: "createUser".to_string(),
                return_type: "User".to_string(),
                inject,
                ..Default::default()
            }],
        );
        let message = validate_tenant_annotations(&mut schema, "org_id").unwrap_err().to_string();
        assert!(message.contains("createUser") && message.contains("jwt:org_id"), "{message}");
    }

    // ── #1142: a @tenant_id type no operation returns ───────────────────
    //
    // Auto-injection reaches queries and mutations. A federation entity resolved only
    // through `_entities` has neither, so the annotation had nothing to be lowered onto
    // and the compile succeeded with the type unscoped on that path — the same hole as
    // a hand-written `inject_params`, reached through the primary tenancy surface.

    #[test]
    fn a_tenant_annotated_type_no_operation_returns_scopes_itself() {
        let mut schema = make_schema(
            vec![IntermediateType {
                sql_source: Some("v_organization".to_string()),
                ..make_type("Organization", vec![make_tenant_id_field("tenant_id")])
            }],
            vec![],
            vec![],
        );
        validate_tenant_annotations(&mut schema, "org_id").unwrap();
        assert_eq!(
            schema.types[0].inject_params.get("tenant_id").map(String::as_str),
            Some("jwt:org_id"),
            "a @tenant_id type with no operation to carry the scoping must carry it itself"
        );
    }

    /// Control: the type-level declaration is for the shape that has nowhere else. A type
    /// a query returns keeps its scoping on the query, so the compiled artefact does not
    /// grow a second copy of every tenancy annotation.
    #[test]
    fn a_tenant_annotated_type_a_query_returns_is_left_to_its_query() {
        let mut schema = make_schema(
            vec![make_type("User", vec![make_tenant_id_field("tenant_id")])],
            vec![make_query("getUser", "User")],
            vec![],
        );
        validate_tenant_annotations(&mut schema, "org_id").unwrap();
        assert_eq!(
            schema.queries[0].inject.get("tenant_id").map(String::as_str),
            Some("jwt:org_id"),
            "the query still carries it"
        );
        assert!(
            schema.types[0].inject_params.is_empty(),
            "the type does not duplicate a declaration its query already carries"
        );
    }

    /// The operation path errors when an explicit inject omits the annotated field rather
    /// than silently scoping nothing. The type path owes the author the same refusal.
    #[test]
    fn error_when_type_inject_overridden_without_tenant() {
        let mut schema = make_schema(
            vec![IntermediateType {
                sql_source: Some("v_organization".to_string()),
                inject_params: std::iter::once(("owner_id".to_string(), "jwt:sub".to_string()))
                    .collect(),
                ..make_type("Organization", vec![make_tenant_id_field("tenant_id")])
            }],
            vec![],
            vec![],
        );
        let err = validate_tenant_annotations(&mut schema, "org_id").unwrap_err();
        let msg = format!("{err:#}");
        assert!(
            msg.contains("Organization") && msg.contains("tenant_id"),
            "the refusal must name the type and the field it lacks: {msg}"
        );
    }
}

// ── converter types tests ───────────────────────────────────────────────────

mod types_tests {
    use super::super::SchemaConverter;

    #[test]
    fn test_is_safe_sql_identifier_simple() {
        assert!(SchemaConverter::is_safe_sql_identifier("v_user"));
    }

    #[test]
    fn test_is_safe_sql_identifier_schema_qualified() {
        assert!(SchemaConverter::is_safe_sql_identifier("public.v_user"));
    }

    #[test]
    fn test_is_safe_sql_identifier_three_part() {
        assert!(SchemaConverter::is_safe_sql_identifier("catalog.schema.table"));
    }

    #[test]
    fn test_is_safe_sql_identifier_empty_rejected() {
        assert!(!SchemaConverter::is_safe_sql_identifier(""));
    }

    #[test]
    fn test_is_safe_sql_identifier_leading_dot_rejected() {
        assert!(!SchemaConverter::is_safe_sql_identifier(".foo"));
    }

    #[test]
    fn test_is_safe_sql_identifier_trailing_dot_rejected() {
        assert!(!SchemaConverter::is_safe_sql_identifier("foo."));
    }

    #[test]
    fn test_is_safe_sql_identifier_double_dot_rejected() {
        assert!(!SchemaConverter::is_safe_sql_identifier("foo..bar"));
    }

    #[test]
    fn test_is_safe_sql_identifier_four_parts_rejected() {
        assert!(!SchemaConverter::is_safe_sql_identifier("a.b.c.d"));
    }

    #[test]
    fn test_is_safe_sql_identifier_special_chars_rejected() {
        assert!(!SchemaConverter::is_safe_sql_identifier("v_user; DROP TABLE"));
    }
}

mod changelog_validation_tests {
    use fraiseql_core::schema::ChangelogConfig;
    use serde_json::json;

    use crate::schema::{converter::SchemaConverter, intermediate::IntermediateSchema};

    fn intermediate_with(
        changelog: Option<ChangelogConfig>,
        observers_config: Option<serde_json::Value>,
    ) -> IntermediateSchema {
        IntermediateSchema {
            version: "2.0.0".to_string(),
            changelog_config: changelog,
            observers_config,
            ..IntermediateSchema::default()
        }
    }

    #[test]
    fn changelog_expose_without_observers_is_rejected() {
        let intermediate = intermediate_with(
            Some(ChangelogConfig {
                expose: true,
                ..Default::default()
            }),
            None,
        );
        let err = SchemaConverter::convert(intermediate).unwrap_err();
        assert!(
            err.to_string().contains("[observers]"),
            "error should explain the observers prerequisite, got: {err}"
        );
    }

    #[test]
    fn changelog_expose_with_disabled_observers_is_rejected() {
        let intermediate = intermediate_with(
            Some(ChangelogConfig {
                expose: true,
                ..Default::default()
            }),
            Some(json!({ "enabled": false, "backend": "postgres" })),
        );
        assert!(SchemaConverter::convert(intermediate).is_err());
    }

    #[test]
    fn changelog_expose_with_enabled_observers_is_allowed() {
        let intermediate = intermediate_with(
            Some(ChangelogConfig {
                expose: true,
                ..Default::default()
            }),
            Some(json!({ "enabled": true, "backend": "postgres" })),
        );
        let compiled = SchemaConverter::convert(intermediate).expect("should convert");
        assert!(compiled.changelog.is_some());
        assert!(compiled.changelog.as_ref().unwrap().expose);

        // The injection ran (after rich filters, before validate) and validate()
        // accepted the schema-qualified view sql_source + the generated operations.
        assert!(compiled.types.iter().any(|t| t.name == "EntityChangeLog"));
        assert!(compiled.types.iter().any(|t| t.name == "TransportCheckpoint"));
        assert!(compiled.queries.iter().any(|q| q.name == "entity_change_logs"));
        assert!(compiled.queries.iter().any(|q| q.name == "transport_checkpoint"));
        assert!(compiled.mutations.iter().any(|m| m.name == "upsert_transport_checkpoint"));
    }

    #[test]
    fn changelog_disabled_does_not_require_observers() {
        // expose = false must never trip the observers gate.
        let intermediate = intermediate_with(
            Some(ChangelogConfig {
                expose: false,
                ..Default::default()
            }),
            None,
        );
        assert!(SchemaConverter::convert(intermediate).is_ok());
    }

    #[test]
    fn changelog_expose_on_federation_subgraph_warns_but_compiles() {
        // #497: exposing the change-log on a federation subgraph is the single-owner
        // pattern — allowed. The cross-subgraph collision can't be detected here (each
        // subgraph compiles alone), so the guardrail is a compile warning, never an
        // error; the owning subgraph legitimately sets both.
        let intermediate = IntermediateSchema {
            version: "2.0.0".to_string(),
            changelog_config: Some(ChangelogConfig {
                expose: true,
                ..Default::default()
            }),
            observers_config: Some(json!({ "enabled": true, "backend": "postgres" })),
            federation_config: Some(json!({ "enabled": true })),
            ..IntermediateSchema::default()
        };
        let compiled =
            SchemaConverter::convert(intermediate).expect("federated expose warns, never errors");
        assert!(compiled.changelog.as_ref().unwrap().expose);
        assert!(compiled.federation.as_ref().unwrap().enabled);
        assert!(compiled.types.iter().any(|t| t.name == "EntityChangeLog"));
    }
}

// ── #665 changelog projections vs. cascade entity classification ─────────────

mod changelog_cascade_conformance_tests {
    //! Pins for the interaction between the framework-synthesized changelog
    //! projections and cascade entity classification (#665), plus the two exclusion
    //! legs `is_queryable_entity`'s doc-comment claims but which nothing pinned.

    use fraiseql_core::schema::ChangelogConfig;
    use serde_json::json;

    use crate::schema::{
        converter::SchemaConverter,
        intermediate::{
            IntermediateField, IntermediateMutation, IntermediateSchema, IntermediateType,
        },
    };

    fn field(name: &str, field_type: &str) -> IntermediateField {
        IntermediateField {
            function: None,

            deprecated:      None,
            vector_config:   None,
            vector_distance: None,
            name:            name.to_string(),
            field_type:      field_type.to_string(),
            nullable:        false,
            description:     None,
            directives:      None,
            requires_scope:  None,
            on_deny:         None,
            authorize:       None,
            hierarchy:       None,
            unique:          None,
            localized:       false,
        }
    }

    /// A view-backed entity with a conformant `id: ID`.
    fn entity(name: &str) -> IntermediateType {
        IntermediateType {
            name: name.to_string(),
            sql_source: Some(format!("v_{}", name.to_lowercase())),
            fields: vec![field("id", "ID"), field("title", "String")],
            ..IntermediateType::default()
        }
    }

    fn cascade_mutation(name: &str, entity_type: &str) -> IntermediateMutation {
        IntermediateMutation {
            name: name.to_string(),
            return_type: entity_type.to_string(),
            cascade: true,
            sql_source: Some(format!("fn_{}", name.to_lowercase())),
            ..IntermediateMutation::default()
        }
    }

    fn schema_with(
        types: Vec<IntermediateType>,
        mutations: Vec<IntermediateMutation>,
    ) -> IntermediateSchema {
        IntermediateSchema {
            version: "2.0.0".to_string(),
            types,
            mutations,
            ..IntermediateSchema::default()
        }
    }

    /// The same, with the change-log surface exposed (and its observers prerequisite).
    fn changelog_schema_with(
        types: Vec<IntermediateType>,
        mutations: Vec<IntermediateMutation>,
    ) -> IntermediateSchema {
        IntermediateSchema {
            changelog_config: Some(ChangelogConfig {
                expose: true,
                ..Default::default()
            }),
            observers_config: Some(json!({ "enabled": true, "backend": "postgres" })),
            ..schema_with(types, mutations)
        }
    }

    /// The cascade envelope and payload types are never cascade nodes: each is
    /// synthesized with an empty `sql_source`, which is the sole mechanism excluding
    /// them from `is_queryable_entity`.
    ///
    /// Pinned because the exclusion is load-bearing but was previously only implied by
    /// `synth_type`'s construction — no test asserted it.
    #[test]
    fn cascade_envelope_types_are_never_cascade_nodes() {
        let compiled = SchemaConverter::convert(schema_with(
            vec![entity("Post")],
            vec![cascade_mutation("createPost", "Post")],
        ))
        .expect("cascade-only schema compiles");

        // The user entity is a cascade node …
        let post = compiled.types.iter().find(|t| t.name.as_str() == "Post").unwrap();
        assert!(
            post.implements.iter().any(|i| i == "CascadeNode"),
            "Post implements CascadeNode"
        );

        // … and none of the synthesized machinery is.
        for name in [
            "UpdatedEntity",
            "DeletedEntity",
            "CascadeUpdates",
            "CascadeMetadata",
            "QueryInvalidation",
            "CreatePostPayload",
        ] {
            let ty = compiled.types.iter().find(|t| t.name.as_str() == name);
            assert!(ty.is_some(), "{name} is synthesized");
            let ty = ty.unwrap();
            assert!(ty.sql_source.as_str().is_empty(), "{name} declares no sql_source");
            assert!(
                !ty.implements.iter().any(|i| i == "CascadeNode"),
                "{name} does not implement CascadeNode"
            );
        }
    }

    /// Relay connection/edge wrappers are excluded from the cascade entity set for the
    /// same reason. `is_queryable_entity`'s doc-comment asserts this ("as are relay
    /// connection/edge wrappers"); until now nothing combined `relay = true` with
    /// `cascade = true` to prove it.
    #[test]
    fn relay_wrapper_types_are_never_cascade_nodes() {
        let post = IntermediateType {
            relay: true,
            ..entity("Post")
        };
        let compiled = SchemaConverter::convert(schema_with(
            vec![post],
            vec![cascade_mutation("createPost", "Post")],
        ))
        .expect("relay + cascade compiles");

        for name in ["PostConnection", "PostEdge", "PageInfo"] {
            let ty = compiled.types.iter().find(|t| t.name.as_str() == name);
            assert!(ty.is_some(), "{name} is synthesized by the relay pass");
            assert!(
                !ty.unwrap().implements.iter().any(|i| i == "CascadeNode"),
                "{name} is a pagination wrapper, not a cascade entity"
            );
        }
    }

    /// **#665 — the fix.** Exposing the change-log and opting a mutation into
    /// `cascade = true` now COMPILES. `inject_changelog` marks its two projections
    /// `internal` (Phase 01); `is_queryable_entity` excludes internal types (Phase 02),
    /// so the framework's own `TransportCheckpoint` — keyed by `transport_name`, no `id`
    /// by design — is no longer classified a cascade entity and no longer fails the
    /// `id: ID!` contract on a type the user never wrote.
    ///
    /// This is the flip of the Phase 00 expected-failure pin
    /// (`changelog_plus_cascade_fails_on_transport_checkpoint_today`). The whole error —
    /// including the misleading #659 "embedded value object" hint that misfired on a
    /// framework projection — is gone, because `validate()` now accepts the schema.
    #[test]
    fn changelog_plus_cascade_compiles_via_internal_exemption() {
        let compiled = SchemaConverter::convert(changelog_schema_with(
            vec![entity("Post")],
            vec![cascade_mutation("createPost", "Post")],
        ))
        .expect("#665: changelog + cascade now compiles (validate() passes)");

        // The two framework projections are `internal` — NOT cascade nodes. Excluding
        // them is the whole fix: `TransportCheckpoint` has no `id` and could never back
        // the CascadeNode contract, and `EntityChangeLog` is bookkeeping, not a
        // cascade-deliverable entity (gate 2).
        for name in ["TransportCheckpoint", "EntityChangeLog"] {
            let ty = compiled.types.iter().find(|t| t.name.as_str() == name);
            assert!(ty.is_some(), "{name} is injected");
            assert!(
                !ty.unwrap().implements.iter().any(|i| i == "CascadeNode"),
                "{name} is a framework projection, not a cascade entity"
            );
        }

        // The genuine user entity still IS a cascade node — enforcement is unchanged
        // for real entities.
        let post = compiled.types.iter().find(|t| t.name.as_str() == "Post").unwrap();
        assert!(
            post.implements.iter().any(|i| i == "CascadeNode"),
            "the user's Post entity still implements CascadeNode"
        );

        // The injected change-log surface survives untouched.
        assert!(compiled.queries.iter().any(|q| q.name == "entity_change_logs"));
        assert!(compiled.queries.iter().any(|q| q.name == "transport_checkpoint"));
        assert!(compiled.mutations.iter().any(|m| m.name == "upsert_transport_checkpoint"));
    }

    /// The change-log surface alone (no cascade mutation) compiles — the baseline the
    /// fix must not disturb.
    #[test]
    fn changelog_without_cascade_compiles() {
        let compiled = SchemaConverter::convert(changelog_schema_with(
            vec![entity("Post")],
            vec![IntermediateMutation {
                cascade: false,
                ..cascade_mutation("createPost", "Post")
            }],
        ))
        .expect("changelog-only schema compiles");

        assert!(compiled.types.iter().any(|t| t.name == "EntityChangeLog"));
        assert!(compiled.types.iter().any(|t| t.name == "TransportCheckpoint"));
        assert!(!compiled.interfaces.iter().any(|i| i.name == "CascadeNode"));
    }

    // ── #687 declared cascade-node identity — Phase 00 baseline pins ──────────
    //
    // These lock the three behaviors #687 will change, so Phase 01's diff shows
    // only intended changes against a green baseline. All GREEN today
    // (characterization); each is proven able to fail before landing.
    //
    // Cycle 0 inventory — what the existing suite already covers (do NOT duplicate):
    //   * id-less value object under cascade HARD-FAILS with the #659 signal + reference-path
    //     clauses: `tenancy_tests::cascade_embedded_value_object_error_reports_derivation_and_path`
    //     (asserts the `sql_source` / `v_money` / `createOrder → Order.total → Money` *fragments*).
    //     The Cycle-1 pin below tightens this to the exact signal clause; the Cycle-3 pin snapshots
    //     the whole diagnostic string.
    //   * a real view-backed `id: ID` entity implements `CascadeNode`:
    //     `tenancy_tests::cascade_conformant_id_entity_still_implements_cascade_node` and
    //     `changelog_cascade_conformance_tests::cascade_envelope_types_are_never_cascade_nodes`.
    //   * a wrong-typed / absent `id` cascade entity fails legibly:
    //     `tenancy_tests::cascade_int_id_entity_fails_with_actionable_error`,
    //     `tenancy_tests::cascade_missing_id_entity_fails_with_actionable_error`.
    //
    // #687-specific gaps added below: (1) the *exact* #659 signal clause on the
    // id-less value object; (2) a real `{ id, total }` entity reached by a cascade
    // stays classified (guards Phase 01's future `&& !ty.embedded` exemption from
    // over-reaching); (3) a full-string snapshot of the diagnostic Phase 01 Cycle 4
    // will deliberately re-point at `embedded=True`.

    /// A view-backed type with a synthesized `sql_source` (`v_<name>`) and the given
    /// fields — models both a real entity and the SDK-synthesized-source value object
    /// #687 distinguishes (the SDK gives every `@fraiseql.type` a source today, which
    /// is exactly why an id-less value object is misclassified as a cascade entity).
    fn view_type(name: &str, fields: Vec<IntermediateField>) -> IntermediateType {
        IntermediateType {
            name: name.to_string(),
            sql_source: Some(format!("v_{}", name.to_lowercase())),
            fields,
            ..IntermediateType::default()
        }
    }

    /// The realistic #653/#687 shape: an id-less value object (`Money`) the SDK gave a
    /// synthesized `sql_source = "v_money"`, embedded under a genuine entity (`Order`)
    /// that a `cascade = true` mutation returns.
    fn order_embedding_money() -> IntermediateSchema {
        schema_with(
            vec![
                view_type("Order", vec![field("id", "ID"), field("total", "Money")]),
                view_type("Money", vec![field("amount", "Int"), field("currency", "String")]),
            ],
            vec![cascade_mutation("createOrder", "Order")],
        )
    }

    /// **Cycle 1 — status quo we improve.** An id-less value object carrying an
    /// SDK-synthesized `sql_source` HARD-FAILS the cascade `id: ID!` contract *today*,
    /// with the #659 signal clause naming its declared source as the classification
    /// reason. Phase 01 will make this compile *when the type declares `embedded=True`*
    /// and will update this pin; the failure must persist for a non-embedded id-less
    /// type (Phase 01 Cycle 3).
    #[test]
    fn cascade_idless_value_object_hard_fails_naming_its_source() {
        let msg = format!("{:#}", SchemaConverter::convert(order_embedding_money()).unwrap_err());
        // The exact #659 signal clause — not just the `sql_source` / `v_money`
        // fragments the tenancy_tests pin checks.
        assert!(
            msg.contains(
                "classified as a cascade entity because it declares sql_source = \"v_money\""
            ),
            "names the classification signal verbatim: {msg}"
        );
        assert!(msg.contains("Type 'Money': no `id` field"), "states the failed contract: {msg}");
    }

    /// **Cycle 2 — real entities stay classified.** A genuine view-backed entity
    /// (`Order { id, total }`) reached by a cascade compiles, `implements CascadeNode`,
    /// and keeps its canonical `id: ID`. Guards Phase 01's forthcoming
    /// `&& !ty.embedded` exemption leg from accidentally exempting real entities
    /// (a real entity has `embedded=false`, so it must remain a cascade node).
    #[test]
    fn real_entity_reached_by_cascade_stays_a_cascade_node() {
        use fraiseql_core::schema::FieldType;
        let compiled = SchemaConverter::convert(schema_with(
            vec![view_type(
                "Order",
                vec![field("id", "ID"), field("total", "Int")],
            )],
            vec![cascade_mutation("createOrder", "Order")],
        ))
        .expect("a real id-bearing entity under a cascade compiles");

        let order = compiled.types.iter().find(|t| t.name.as_str() == "Order").unwrap();
        assert!(
            order.implements.iter().any(|i| i == "CascadeNode"),
            "the real entity implements CascadeNode"
        );
        assert_eq!(
            order.find_field("id").expect("id field").field_type,
            FieldType::Id,
            "its id is enforced as ID"
        );
    }

    /// A declared embedded value object (#687) that *also* carries a non-empty
    /// `sql_source` — the hand-authored contradiction the SDK cannot emit (it suppresses
    /// the source, and Phase 02 rejects the combination outright). The non-empty source
    /// is load-bearing: an embedded type with an *empty* source is already exempt via
    /// the `!sql_source.is_empty()` leg, so it would compile even before Phase 01 and
    /// prove nothing about the `&& !ty.embedded` guard.
    fn embedded_view_type(name: &str, fields: Vec<IntermediateField>) -> IntermediateType {
        IntermediateType {
            embedded: true,
            ..view_type(name, fields)
        }
    }

    /// **Phase 01 Cycle 2 — the declared exemption.** An id-less `Money` that declares
    /// `embedded=True` compiles even though it carries a source: the author stated it has
    /// no independent identity, so it is not a cascade entity, is not `id`-enforced, and
    /// never auto-implements `CascadeNode`.
    #[test]
    fn declared_embedded_value_object_is_exempt_from_cascade_entity_classification() {
        let compiled = SchemaConverter::convert(schema_with(
            vec![
                view_type("Order", vec![field("id", "ID"), field("total", "Money")]),
                embedded_view_type(
                    "Money",
                    vec![field("amount", "Int"), field("currency", "String")],
                ),
            ],
            vec![cascade_mutation("createOrder", "Order")],
        ))
        .expect("a declared embedded value object compiles under a cascade");

        let money = compiled.types.iter().find(|t| t.name.as_str() == "Money").unwrap();
        assert!(money.embedded, "the declaration survives into the compiled schema");
        assert!(
            !money.implements.iter().any(|i| i == "CascadeNode"),
            "a value object is never a cascade node: {:?}",
            money.implements
        );
        assert!(money.find_field("id").is_none(), "no `id` was synthesized onto it");

        let order = compiled.types.iter().find(|t| t.name.as_str() == "Order").unwrap();
        assert!(
            order.implements.iter().any(|i| i == "CascadeNode"),
            "its parent entity is still a cascade node"
        );
    }

    /// **Phase 01 Cycle 3 — the (b) regression guard.** The exemption is scoped to the
    /// *declaration* alone: a **non-`embedded`** id-less type carrying the same
    /// synthesized source still hard-fails. Enforcement is whole-schema (gated only on a
    /// cascade mutation existing, `cascade_types.rs`), so this holds whether or not the
    /// type sits under the payload — embedding it just adds the `reached via` clause.
    /// Locks the forgot-`id` hole (#687 failure mode 2) shut.
    #[test]
    fn non_embedded_idless_type_still_hard_fails_after_the_exemption_lands() {
        let err = SchemaConverter::convert(order_embedding_money())
            .expect_err("a non-embedded id-less type is still a cascade entity");
        let msg = format!("{err:#}");
        assert!(msg.contains("Type 'Money': no `id` field"), "still fails the contract: {msg}");
    }

    /// **Cycle 3 — diagnostic wording snapshot.** The full cascade `id` error for the
    /// id-less value object, verbatim: the signal clause + the reference-path clause +
    /// the `Fix:` line. Phase 01 Cycle 4 re-pointed the signal clause at `embedded=True`
    /// — it now names *both* exits (declare it embedded, or give it an `id`) instead of
    /// the pre-#687 "should declare no source", which described a state the author had no
    /// way to reach.
    #[test]
    fn cascade_idless_value_object_diagnostic_wording_snapshot() {
        let msg = format!("{:#}", SchemaConverter::convert(order_embedding_money()).unwrap_err());
        let expected = "cascade requires `id: ID!` on every cascade entity (the graphql-cascade \
CascadeNode interface requires it):
  - Type 'Money': no `id` field
      classified as a cascade entity because it declares sql_source = \"v_money\"; if it is an \
embedded value object, mark it embedded=True (it will declare no source and be exempt); if it is \
an entity, add id: ID!
      reached via createOrder → Order.total → Money
Fix: expose `id: ID!` on each type, or remove `cascade` from the mutations that return them.";
        assert_eq!(msg, expected, "diagnostic wording drifted:\n{msg}");
    }
}

// ── #573 scheduled ingress sources ───────────────────────────────────────────

#[test]
fn converts_sources_into_the_compiled_schema() {
    use fraiseql_core::schema::SourceDefinition;
    let intermediate = IntermediateSchema {
        sources: Some(vec![
            SourceDefinition::new("orders", "*/5 * * * *", "pollOrders").with_cursor("orders-hwm"),
        ]),
        ..Default::default()
    };
    let compiled = SchemaConverter::convert(intermediate).unwrap();
    assert_eq!(compiled.sources.len(), 1);
    assert_eq!(compiled.sources[0].name, "orders");
    assert_eq!(compiled.sources[0].cursor_name(), "orders-hwm");
}

#[test]
fn absent_sources_convert_to_an_empty_list() {
    let compiled = SchemaConverter::convert(IntermediateSchema::default()).unwrap();
    assert!(compiled.sources.is_empty());
}

// ===========================================================================
// The scalar-name table and the converter must not drift (#724 item 2)
// ===========================================================================

/// Every built-in scalar name round-trips: name → `FieldType` → the same name.
///
/// The two directions used to be three hand-written lists — the validator's known-type
/// registry, `parse_field_type` and `extract_type_name` — and they drifted twice. First
/// over spelling: the validator listed `"JSON"`, which `parse_field_type` does not
/// recognize, so it became `FieldType::Object("JSON")`, a reference to a nonexistent type,
/// while `"Json"`, the spelling every SDK emits, was not in the validator's list at all.
/// Then over the #959 vector types, which parsed correctly and were reported as
/// undeclared. All three now read `fraiseql_core::schema::BUILTIN_SCALARS`; this pins that
/// they still do, in both directions.
#[test]
fn builtin_scalar_names_round_trip_through_the_converter() {
    for (name, expected) in fraiseql_core::schema::BUILTIN_SCALARS {
        let parsed = SchemaConverter::parse_field_type(
            name,
            &crate::schema::converter::DeclaredTypeNames::default(),
        )
        .unwrap_or_else(|e| panic!("scalar {name:?} must parse: {e}"));
        assert_eq!(
            &parsed, expected,
            "{name:?} is listed as a built-in scalar denoting {expected:?} but the converter \
             parses it as {parsed:?}"
        );
        assert_eq!(
            SchemaConverter::extract_type_name(&parsed),
            *name,
            "{name:?} does not survive the round trip back to a type name, so a compiled \
             schema would name the type something an author cannot write"
        );
    }
}

/// A name that is *not* in the scalar table must parse as an object reference.
///
/// The reverse guard: the table must not be so broad that an ordinary type name is
/// swallowed as a scalar, which would make a reference to a real type unresolvable.
#[test]
fn an_undeclared_name_stays_an_object_reference() {
    assert!(
        matches!(
            SchemaConverter::parse_field_type("User", &crate::schema::converter::DeclaredTypeNames::default()).unwrap(),
            FieldType::Object(ref n) if n == "User"
        ),
        "an ordinary type name must parse as an object reference"
    );
}

/// The mutation verb is case-insensitive, because every authoring surface spells it
/// lowercase and the compiler only ever accepted uppercase.
///
/// `docs/authoring.md`, `docs/architecture/intermediate-schema.md`, the Python SDK's
/// `@fraiseql.mutation(operation="insert")`, the PHP `MutationBuilder`'s own class
/// docblock and the Java `OperationBuilder` all use the lowercase form. Every one of
/// them produced `Error: Unknown mutation operation: insert` — a schema the project's
/// own documentation told the author to write, rejected outright. The verb set stays
/// closed: an unrecognized word is still a hard error.
#[test]
fn mutation_operation_verbs_are_case_insensitive() {
    use fraiseql_core::schema::MutationOperation;

    for spelling in ["insert", "Insert", "INSERT", "create", "CREATE"] {
        assert!(
            matches!(
                SchemaConverter::parse_mutation_operation(Some(spelling), Some("fn_x")).unwrap(),
                MutationOperation::Insert { ref table } if table == "fn_x"
            ),
            "{spelling:?} must parse as an INSERT"
        );
    }
    for spelling in ["update", "UPDATE"] {
        assert!(
            matches!(
                SchemaConverter::parse_mutation_operation(Some(spelling), Some("fn_x")).unwrap(),
                MutationOperation::Update { .. }
            ),
            "{spelling:?} must parse as an UPDATE"
        );
    }
    for spelling in ["delete", "DELETE"] {
        assert!(
            matches!(
                SchemaConverter::parse_mutation_operation(Some(spelling), Some("fn_x")).unwrap(),
                MutationOperation::Delete { .. }
            ),
            "{spelling:?} must parse as a DELETE"
        );
    }
    for spelling in ["custom", "CUSTOM"] {
        assert!(
            matches!(
                SchemaConverter::parse_mutation_operation(Some(spelling), None).unwrap(),
                MutationOperation::Custom
            ),
            "{spelling:?} must parse as a CUSTOM operation"
        );
    }
}

/// Case-insensitivity must not become "accept anything".
#[test]
fn unknown_mutation_operation_verb_is_still_rejected() {
    let error = SchemaConverter::parse_mutation_operation(Some("upsert"), Some("fn_x"))
        .expect_err("`upsert` is not a supported verb and must be refused, not defaulted");
    assert!(
        error.to_string().contains("upsert"),
        "the diagnostic must name the offending verb, got: {error}"
    );
}

/// A custom scalar declaring validation rules is refused, because no compiled artifact
/// carries them.
///
/// `CompiledSchema.custom_scalars` is `#[serde(skip)]` — the converter registers the
/// scalar into an in-memory `CustomTypeRegistry` that is then dropped when the schema is
/// written to `schema.compiled.json`. Nothing in `fraiseql-server` reads it either; the
/// only mention is `reload_gate.rs`, which explicitly ignores the field.
///
/// So an author who declared `pattern` or `length` on a scalar got `✓ Schema compiled
/// successfully` and a server that validates nothing. Carrying the rules through the
/// converter without a runtime consumer would relocate the drop rather than fix it —
/// exactly the disposition `#779` got for observers.
///
/// The scalar *declaration* still works: the name is known to the compiler, so a field
/// typed with it resolves. Only the unenforceable rules are refused.
#[test]
fn custom_scalar_validation_rules_are_refused_not_dropped() {
    use fraiseql_core::validation::ValidationRule;

    use crate::schema::IntermediateScalar;

    let scalar = IntermediateScalar {
        name:             "EmailAddress".into(),
        description:      None,
        specified_by_url: None,
        validation_rules: vec![ValidationRule::Length {
            min: Some(3),
            max: Some(254),
        }],
        base_type:        Some("String".into()),
    };

    let error = SchemaConverter::convert_custom_scalar(scalar)
        .expect_err("a scalar declaring rules no artifact carries must be refused");
    let message = error.to_string();

    assert!(
        message.contains("EmailAddress"),
        "the diagnostic must name the scalar: {message}"
    );
    assert!(
        message.contains("validation_rules"),
        "the diagnostic must name the offending key: {message}"
    );
}

/// A custom scalar with no rules still compiles — the declaration is what makes the name
/// known to the compiler, and that half works.
#[test]
fn custom_scalar_without_rules_still_compiles() {
    use crate::schema::IntermediateScalar;

    let scalar = IntermediateScalar {
        name:             "EmailAddress".into(),
        description:      Some("An email address.".into()),
        specified_by_url: None,
        validation_rules: Vec::new(),
        base_type:        Some("String".into()),
    };

    let converted = SchemaConverter::convert_custom_scalar(scalar)
        .expect("a scalar without rules must still convert");
    assert_eq!(converted.name, "EmailAddress");
    assert_eq!(converted.base_type.as_deref(), Some("String"));
}

// ── #938 count sibling ───────────────────────────────────────────────────────

/// A list query carrying `count = true`, with everything else at its default.
fn count_query(name: &str, tweak: impl FnOnce(&mut IntermediateQuery)) -> IntermediateQuery {
    let mut q = IntermediateQuery {
        function: None,

        requires_actor:          Vec::new(),
        count:                   true,
        name:                    name.to_string(),
        return_type:             "User".to_string(),
        returns_list:            true,
        nullable:                false,
        arguments:               vec![],
        description:             None,
        sql_source:              Some("v_user".to_string()),
        sql_source_kind:         fraiseql_core::schema::SourceKind::Relation,
        auto_params:             None,
        deprecated:              None,
        jsonb_column:            None,
        relay:                   false,
        inject:                  IndexMap::default(),
        exclude_inject_defaults: Vec::new(),
        read_routing:            fraiseql_core::db::types::ReadRouting::default(),
        cache_ttl_seconds:       None,
        additional_views:        vec![],
        requires_role:           None,
        relay_cursor_type:       None,
        rest:                    None,
        rest_stream:             false,
        pagination_order:        None,
    };
    tweak(&mut q);
    q
}

fn convert_one(
    q: IntermediateQuery,
) -> Result<fraiseql_core::schema::QueryDefinition, anyhow::Error> {
    SchemaConverter::convert_query(
        q,
        &crate::schema::intermediate::IntermediateQueryDefaults::default(),
        &DeclaredTypeNames::default(),
    )
}

#[test]
fn count_on_a_single_item_query_is_refused() {
    let err = convert_one(count_query("user", |q| q.returns_list = false))
        .expect_err("a single-item query has nothing to count");
    assert!(
        err.to_string().contains("requires returns_list=true"),
        "the refusal must name the constraint: {err}"
    );
}

#[test]
fn count_without_a_sql_source_is_refused() {
    let err = convert_one(count_query("users", |q| q.sql_source = None))
        .expect_err("SELECT COUNT(*) needs a view to count");
    assert!(
        err.to_string().contains("requires sql_source"),
        "the refusal must name the constraint: {err}"
    );
}

/// Relay connections already carry `totalCount` over the same rows. Accepting
/// both would compile two different totals for one query and leave the author to
/// discover which one their client got.
#[test]
fn count_with_relay_is_refused_as_redundant() {
    let err = convert_one(count_query("users", |q| q.relay = true))
        .expect_err("relay already exposes totalCount");
    assert!(
        err.to_string().contains("redundant with relay=true"),
        "the refusal must explain the overlap: {err}"
    );
}

/// The derived sibling inherits every restriction of the list it counts. This is
/// the anti-drift property the whole design rests on: a count that kept the rows
/// but dropped the tenant filter returns no row and still discloses how many
/// exist (#1030's shape).
#[test]
fn count_sibling_inherits_the_lists_restrictions() {
    let mut list = fraiseql_core::schema::QueryDefinition::new("users", "User")
        .returning_list()
        .with_sql_source("v_user");
    list.auto_params = fraiseql_core::schema::AutoParams::all();
    list.requires_role = Some("auditor".to_string());
    list.inject_params.insert(
        "tenant_id".to_string(),
        fraiseql_core::schema::security_config::InjectedParamSource::Jwt("org_id".to_string()),
    );
    list.additional_views = vec!["v_org".to_string()];
    list.native_columns
        .insert("tenant_id".to_string(), fraiseql_core::schema::NativeColumn::nullable("uuid"));

    let count = list.count_sibling();

    assert_eq!(count.name, "usersCount");
    assert!(count.returns_count, "the sibling is a count");
    assert!(!count.returns_list, "a count is a scalar, not a list");
    assert!(!count.nullable, "Int!, so 'no rows' and 'no answer' stay distinct");

    // Same rows, same restrictions.
    assert_eq!(count.return_type, "User", "the entity type keys the filter machinery");
    assert_eq!(count.sql_source.as_deref(), Some("v_user"));
    assert_eq!(count.requires_role.as_deref(), Some("auditor"));
    assert_eq!(count.inject_params, list.inject_params, "tenant scoping must be inherited");
    assert_eq!(count.native_columns, list.native_columns, "same casts, so the same indexes");
    assert_eq!(count.additional_views, list.additional_views, "same cache invalidation");

    // The filter survives; the pagination does not.
    assert!(count.auto_params.has_where, "the count composes with the same filter");
    assert!(!count.auto_params.has_limit, "a total that moved with the page is not a total");
    assert!(!count.auto_params.has_offset);
    assert!(!count.auto_params.has_order_by, "ordering cannot change a count");
    assert!(!count.relay);

    // A bare schema: this test is about which argument *names* survive on a
    // count sibling, so the auto-wired shapes stay `JSON` and nothing is derived.
    let args = count.graphql_arguments(&fraiseql_core::schema::CompiledSchema::default());
    let arg_names: Vec<&str> = args.iter().map(|a| a.name.as_str()).collect();
    assert_eq!(arg_names, vec!["where"], "the exposed surface is `where` alone");
}

// ── #1266 relationships: the authoring surface, and the refusals it enables ───

/// The whole document as JSON, so each case can mutate one key and re-convert.
///
/// Authored as `schema.json` rather than by building `IntermediateType` literally: the
/// point of #1266 is that the *wire* document may now carry `relationships`, and
/// `IntermediateType` denies unknown fields — so a document that deserializes at all is
/// itself part of what is asserted.
fn relationship_ir() -> serde_json::Value {
    serde_json::json!({
      "types": [
        {"name": "User", "sql_source": "v_user",
         "fields": [{"name": "id", "type": "ID", "nullable": false}],
         "relationships": [{"name": "orders", "target_type": "Order",
                            "cardinality": "OneToMany",
                            "foreign_key": "fk_user", "referenced_key": "id"}]},
        {"name": "Order", "sql_source": "v_order",
         "fields": [{"name": "id", "type": "ID", "nullable": false},
                    {"name": "fkUser", "type": "ID", "nullable": false}]}
      ],
      "queries": [
        {"name": "users", "return_type": "User", "returns_list": true, "sql_source": "v_user"},
        {"name": "orders", "return_type": "Order", "returns_list": true, "sql_source": "v_order"}
      ]
    })
}

fn convert_ir(doc: &serde_json::Value) -> anyhow::Result<fraiseql_core::schema::CompiledSchema> {
    let intermediate: IntermediateSchema =
        serde_json::from_value(doc.clone()).expect("the fixture must deserialize");
    SchemaConverter::convert(intermediate)
}

/// The positive control, and the whole point of the issue: a relationship authored in
/// `schema.json` reaches `TypeDefinition.relationships` instead of being replaced by the
/// empty vector every converter used to write.
#[test]
fn an_authored_relationship_reaches_the_compiled_schema() {
    let compiled = convert_ir(&relationship_ir()).expect("a followable relationship compiles");
    let user = compiled.find_type("User").expect("User compiles");
    assert_eq!(user.relationships.len(), 1, "the relationship survives the conversion");
    let rel = &user.relationships[0];
    assert_eq!(rel.name, "orders");
    assert_eq!(rel.target_type, "Order");
    assert_eq!(rel.cardinality, fraiseql_core::schema::Cardinality::OneToMany);
    assert_eq!(rel.foreign_key, "fk_user");
    assert_eq!(rel.referenced_key, "id");
}

/// The column→field resolution is the compiler's, not the author's: `fk_user` is a SQL
/// column and `Order` publishes it as `fkUser`. A compiler matching literally would refuse
/// this document, which is why the fixture spells the two differently.
#[test]
fn a_join_column_resolves_against_the_camel_case_field_it_is_published_as() {
    let mut doc = relationship_ir();
    doc["types"][1]["fields"][1]["name"] = "fk_user".into();
    convert_ir(&doc).expect("the snake_case spelling of the same column also resolves");
    convert_ir(&relationship_ir()).expect("and so does the camelCase one");
}

/// The refusal is the compile-time half of #1266; `finish_load` runs the same check on
/// the hand-edited artifact this one cannot reach.
#[test]
fn a_relationship_targeting_an_undeclared_type_fails_the_compile() {
    let mut doc = relationship_ir();
    doc["types"][0]["relationships"][0]["target_type"] = "Nope".into();
    let err = convert_ir(&doc).expect_err("an undeclared target must not compile");
    assert!(
        err.to_string()
            .contains("targets type 'Nope', which the schema does not declare"),
        "{err}"
    );
}

#[test]
fn a_relationship_whose_join_column_names_no_field_fails_the_compile() {
    let mut doc = relationship_ir();
    doc["types"][0]["relationships"][0]["foreign_key"] = "no_such".into();
    let err = convert_ir(&doc).expect_err("an unresolvable join column must not compile");
    assert!(err.to_string().contains("'no_such'"), "{err}");
    assert!(err.to_string().contains("'Order'"), "the side it is missing from: {err}");
}

/// An embed sources its rows from the target's list query, so a target without one is a
/// relationship that answers `[]` before touching the database.
#[test]
fn a_relationship_whose_target_has_no_list_query_fails_the_compile() {
    let mut doc = relationship_ir();
    doc["queries"][1]["returns_list"] = false.into();
    let err = convert_ir(&doc).expect_err("a target with no list query must not compile");
    assert!(err.to_string().contains("no list query"), "{err}");
}

/// Before #1266 this was the *only* outcome: `IntermediateType` denied the key outright,
/// so an SDK emitting it had its whole document refused. Pinned as the state the issue
/// left behind, not as a regression risk.
#[test]
fn a_type_declaring_no_relationships_compiles_to_an_empty_vector() {
    let mut doc = relationship_ir();
    doc["types"][0].as_object_mut().expect("object").remove("relationships");
    let compiled = convert_ir(&doc).expect("a type with no relationships compiles");
    assert!(compiled.find_type("User").expect("User").relationships.is_empty());
}

// ── #1283: what the compiler says about a list query's auto-params ───────────

/// The three rules, as values, on the `AutoParams` each one is about.
///
/// `warn_auto_params` emitted these through `tracing::warn!` and nothing in this crate
/// observed them, so all three were unasserted — including the two that predate #1283.
#[test]
fn auto_param_warnings_state_each_rule_on_the_configuration_that_earns_it() {
    use fraiseql_core::schema::AutoParams;

    // The ordering a paginating query resolves to under the default posture — so
    // these cases read the same configuration the compiler produces for them.
    let ordered = PaginationOrder::Identity;
    let warn = |params| SchemaConverter::auto_param_warnings("things", &params, Some(&ordered));

    // Everything on: nothing to say.
    assert!(warn(AutoParams::all()).is_empty());

    // #1283: a list route that accepts no filter, which is legal and invisible to the
    // client until it sends one.
    let no_filter = AutoParams {
        has_where: false,
        ..AutoParams::all()
    };
    let messages = warn(no_filter);
    assert_eq!(messages.len(), 1, "{messages:?}");
    assert!(messages[0].contains("where_clause = false"), "{}", messages[0]);
    assert!(
        messages[0].contains("refused with 400"),
        "the warning states the consequence a client sees, not just the setting: {}",
        messages[0]
    );

    // The two that came before, still each on their own configuration.
    let unbounded = AutoParams {
        has_limit: false,
        ..AutoParams::all()
    };
    assert!(warn(unbounded)[0].contains("unbounded"));

    // #1303: `order_by = false` is no longer what earns the page-determinism
    // warning. A query whose pages carry a total order is deterministic whatever
    // its client-facing sort surface says, so this configuration is now silent.
    let no_client_sort = AutoParams {
        has_order_by: false,
        ..AutoParams::all()
    };
    let silent = warn(no_client_sort.clone());
    assert!(
        silent.is_empty(),
        "a paginating query with a page order has nothing to warn about: {silent:?}"
    );

    // What earns it now is the thing itself: it paginates, and nothing orders its
    // pages — reachable through a declared `none` or the `allow` posture. Both
    // sort surfaces earn it, because a client that *can* sort and does not is in
    // exactly the same position as one that cannot.
    for params in [AutoParams::all(), no_client_sort] {
        let messages = SchemaConverter::auto_param_warnings("things", &params, None);
        assert_eq!(messages.len(), 1, "{messages:?}");
        assert!(messages[0].contains("overlap and skip rows"), "{}", messages[0]);
        assert!(
            messages[0].contains("pagination_order"),
            "the remedy must be named: {}",
            messages[0]
        );
    }
}

// ── #1290 the fixture builder and the compiler agree on auto-params ──────────

/// `TestQueryBuilder` and `SchemaConverter` must resolve the same `AutoParams` for
/// the same authored query.
///
/// **Why this is a test and not a comment.** A fixture's job is to stand in for what
/// the compiler emits. `TestQueryBuilder::returns_list(true)` left `auto_params` at
/// `Default` — every flag `false` — while the compiler gives a list query every flag
/// `true`, and nothing compared the two. The divergence was only discoverable by
/// reading both defaults, and 34 call sites across 22 files were built on it.
///
/// The direction mattered: all-false is the *permissive* shape for an assertion about
/// filtering. A `?field=value` sent to a query without `has_where` was accepted by the
/// extractor, validated against the return type, built into `arguments["where"]` and
/// dropped by `resolve_root_read` — answering `200` with the whole relation. Three
/// tests passed that way, one of them (`an_export_without_a_dotted_parameter_still_streams`)
/// existing precisely to show that a plain filter survives a gate it was not surviving.
///
/// So the two producers are compared here rather than each asserted against a literal:
/// two literals that agree today are two literals, and #1268 is this repository's
/// record of what happens when a rule is written down three times.
mod auto_params_parity_tests {
    use fraiseql_core::schema::AutoParams;
    use fraiseql_test_utils::schema_builder::TestQueryBuilder;

    use super::*;

    /// The compiler's answer for one authored shape, with everything else minimal.
    ///
    /// `query_defaults: None` on purpose: the divergence is about what a project that
    /// configures *nothing* gets, which is the case every fixture stands in for.
    fn compiled_query(returns_list: bool, relay: bool) -> fraiseql_core::schema::QueryDefinition {
        let intermediate = IntermediateSchema {
            grpc_config:       None,
            security:          None,
            auth:              None,
            version:           "2.0.0".to_string(),
            types:             vec![IntermediateType {
                subscription_policy:    None,
                name:                   "Item".to_string(),
                sql_source:             Some("v_item".to_string()),
                fields:                 vec![],
                description:            None,
                implements:             vec![],
                requires_role:          None,
                is_error:               false,
                is_input:               false,
                relay:                  false,
                embedded:               false,
                subscribable_tables:    None,
                subscribable_pre_image: false,
                inject_params:          IndexMap::new(),
                relationships:          Vec::new(),
            }],
            enums:             vec![],
            input_types:       vec![],
            interfaces:        vec![],
            unions:            vec![],
            queries:           vec![IntermediateQuery {
                function: None,

                requires_actor: Vec::new(),
                count: false,
                name: "items".to_string(),
                return_type: "Item".to_string(),
                returns_list: returns_list || relay,
                nullable: false,
                arguments: vec![],
                description: None,
                sql_source: Some("v_item".to_string()),
                sql_source_kind: fraiseql_core::schema::SourceKind::Relation,
                auto_params: None,
                deprecated: None,
                jsonb_column: None,
                relay,
                inject: IndexMap::default(),
                exclude_inject_defaults: Vec::new(),
                read_routing: fraiseql_core::db::types::ReadRouting::default(),
                cache_ttl_seconds: None,
                additional_views: vec![],
                requires_role: None,
                relay_cursor_type: None,
                rest: None,
                rest_stream: false,
                pagination_order: None,
            }],
            mutations:         vec![],
            subscriptions:     vec![],
            fragments:         None,
            directives:        None,
            fact_tables:       None,
            aggregate_queries: None,
            observers:         None,

            sources:              None,
            functions:            None,
            functions_config:     None,
            custom_scalars:       None,
            observers_config:     None,
            subscriptions_config: None,
            validation_config:    None,
            federation_config:    None,
            debug_config:         None,
            mcp_config:           None,
            rest_config:          None,
            query_defaults:       None,
            inject_defaults:      None,
            naming_convention:    NamingConvention::default(),
            session_variables:    None,
            locale:               None,
            hierarchies_config:   None,
            changelog_config:     None,
        };
        SchemaConverter::convert(intermediate).expect("test schema converts").queries[0].clone()
    }

    /// The compiler's auto-params for one authored shape.
    fn compiled_auto_params(returns_list: bool, relay: bool) -> AutoParams {
        compiled_query(returns_list, relay).auto_params
    }

    /// The fixture builder's answer for the same shape.
    fn fixture_auto_params(returns_list: bool, relay: bool) -> AutoParams {
        TestQueryBuilder::new("items", "Item")
            .returns_list(returns_list || relay)
            .relay(relay)
            .with_sql_source("v_item")
            .build()
            .auto_params
    }

    #[test]
    fn a_list_query_gets_the_same_auto_params_from_both_producers() {
        assert_eq!(
            fixture_auto_params(true, false),
            compiled_auto_params(true, false),
            "TestQueryBuilder must describe the route a list query actually compiles to"
        );
    }

    #[test]
    fn a_single_item_query_gets_the_same_auto_params_from_both_producers() {
        assert_eq!(fixture_auto_params(false, false), compiled_auto_params(false, false));
    }

    #[test]
    fn a_relay_query_gets_the_same_auto_params_from_both_producers() {
        assert_eq!(
            fixture_auto_params(false, true),
            compiled_auto_params(false, true),
            "a connection is paged by first/after; limit and offset are a second, \
             unsynchronised cursor"
        );
    }

    /// #1303: and on the order its pages are cut in.
    ///
    /// The same divergence as the auto-params one, one field over: a fixture list
    /// query carrying no `pagination_order` describes a route the compiler cannot
    /// produce, and — worse than being merely unrealistic — it is the *pre-fix*
    /// route. Every runtime test built on such a fixture would go on asserting the
    /// unordered behaviour after the defect was fixed, and the suite would not move.
    #[test]
    fn both_producers_agree_on_the_pagination_order() {
        let compiled = |list, relay| compiled_query(list, relay).pagination_order;
        let fixture = |list: bool, relay: bool| {
            TestQueryBuilder::new("items", "Item")
                .returns_list(list || relay)
                .relay(relay)
                .with_sql_source("v_item")
                .build()
                .pagination_order
        };

        // The shape that has one, and the two that must not.
        assert_eq!(fixture(true, false), Some(PaginationOrder::Identity));
        assert_eq!(fixture(true, false), compiled(true, false));
        assert_eq!(fixture(false, false), None);
        assert_eq!(fixture(false, false), compiled(false, false));
        assert_eq!(fixture(false, true), None);
        assert_eq!(fixture(false, true), compiled(false, true));
    }

    /// The three shapes are genuinely different, so the equalities above cannot all be
    /// satisfied by one constant.
    ///
    /// Without this, a builder that returned `AutoParams::all()` unconditionally would
    /// fail two of the three — but a builder that returned `AutoParams::default()`
    /// unconditionally, which is exactly the defect, would fail only one, and a reader
    /// could not tell from a green run that the cases discriminate.
    #[test]
    fn the_three_shapes_do_not_collapse_to_one_answer() {
        let list = compiled_auto_params(true, false);
        let single = compiled_auto_params(false, false);
        let relay = compiled_auto_params(false, true);
        assert_ne!(list, single);
        assert_ne!(list, relay);
        assert_ne!(single, relay);
        assert_eq!(list, AutoParams::all());
        assert_eq!(single, AutoParams::none());
        assert_eq!(relay, AutoParams::relay());
    }
}

// ── #1303 the order an offset page is cut in ─────────────────────────────────

/// A list query, everything else at its default: `IntermediateQueryDefaults` turns
/// on `limit` and `offset`, so this paginates.
fn paginating_query(tweak: impl FnOnce(&mut IntermediateQuery)) -> IntermediateQuery {
    let mut q = IntermediateQuery {
        name: "users".to_string(),
        return_type: "User".to_string(),
        returns_list: true,
        sql_source: Some("v_user".to_string()),
        ..IntermediateQuery::default()
    };
    tweak(&mut q);
    q
}

/// The three postures, on the query shapes that tell them apart.
///
/// `refuse` and `identity` derive the same ordering and differ only on whether the
/// opt-out is permitted, so a test that only checked the derived value would find
/// them identical — which is why each case below reads the opt-out too.
mod posture {
    use fraiseql_core::schema::AutoParams;

    use super::*;
    use crate::config::toml_schema::PaginationPosture;

    fn resolve(
        authored: Option<&str>,
        posture: PaginationPosture,
    ) -> Result<Option<PaginationOrder>, anyhow::Error> {
        SchemaConverter::resolve_pagination_order(
            "users",
            authored,
            &AutoParams::all(),
            true,
            false,
            posture,
        )
    }

    #[test]
    fn identity_derives_the_default_and_permits_the_opt_out() {
        assert_eq!(
            resolve(None, PaginationPosture::Identity).unwrap(),
            Some(PaginationOrder::Identity)
        );
        assert_eq!(resolve(Some("none"), PaginationPosture::Identity).unwrap(), None);
    }

    #[test]
    fn refuse_derives_the_same_ordering_but_forbids_the_opt_out() {
        assert_eq!(
            resolve(None, PaginationPosture::Refuse).unwrap(),
            Some(PaginationOrder::Identity),
            "refuse is stricter than identity, not different from it"
        );
        let err = resolve(Some("none"), PaginationPosture::Refuse)
            .expect_err("the whole point of the posture");
        assert!(err.to_string().contains("no exceptions"), "{err}");

        // An authored *column* is still the way to satisfy a strict deployment.
        assert_eq!(
            resolve(Some("pk_user"), PaginationPosture::Refuse).unwrap(),
            Some(PaginationOrder::Column("pk_user".to_string()))
        );
    }

    #[test]
    fn allow_derives_nothing_and_still_honours_an_authored_column() {
        assert_eq!(
            resolve(None, PaginationPosture::Allow).unwrap(),
            None,
            "the pre-#1303 behaviour is the point of this posture"
        );
        assert_eq!(resolve(Some("none"), PaginationPosture::Allow).unwrap(), None);
        assert_eq!(
            resolve(Some("pk_user"), PaginationPosture::Allow).unwrap(),
            Some(PaginationOrder::Column("pk_user".to_string())),
            "an author who declared a column meant it, whatever the deployment prefers"
        );
    }

    #[test]
    fn the_three_postures_do_not_collapse_to_one_answer() {
        // Without this, a resolver that ignored the posture entirely would pass
        // two of the three cases above.
        let derived = |p| resolve(None, p).unwrap();
        assert_ne!(derived(PaginationPosture::Identity), derived(PaginationPosture::Allow));
        let opt_out = |p| resolve(Some("none"), p).is_ok();
        assert!(opt_out(PaginationPosture::Identity));
        assert!(!opt_out(PaginationPosture::Refuse));
    }

    /// The TOML spelling, because the posture is authored as a string and a
    /// rename would be a silently accepted default.
    #[test]
    fn the_toml_spelling_of_each_posture_round_trips() {
        for (text, expected) in [
            ("identity", PaginationPosture::Identity),
            ("refuse", PaginationPosture::Refuse),
            ("allow", PaginationPosture::Allow),
        ] {
            let parsed: PaginationPosture =
                serde_json::from_value(serde_json::json!(text)).expect(text);
            assert_eq!(parsed, expected);
        }
        assert!(
            serde_json::from_value::<PaginationPosture>(serde_json::json!("Identity")).is_err(),
            "the spelling is snake_case and a near-miss must be refused, not defaulted"
        );
    }
}

#[test]
fn a_paginating_query_with_no_authored_order_gets_the_identity() {
    // The answer a compile with no `--database` can always give: `id` is an
    // enforced invariant (ADR-0017), so the extraction is valid for every type.
    let q = convert_one(paginating_query(|_| {})).expect("a plain list query compiles");
    assert_eq!(q.pagination_order, Some(PaginationOrder::Identity));
}

#[test]
fn an_authored_pagination_column_wins_over_the_derived_identity() {
    let q = convert_one(paginating_query(|q| {
        q.pagination_order = Some("pk_user".to_string());
    }))
    .expect("an authored column compiles");
    assert_eq!(q.pagination_order, Some(PaginationOrder::Column("pk_user".to_string())));
}

#[test]
fn pagination_order_none_is_the_declared_opt_out() {
    // The escape hatch for a view carrying its own ORDER BY — which is the
    // remedy `auto_param_warnings` itself recommends, and which ordering
    // unconditionally would destroy.
    let q = convert_one(paginating_query(|q| {
        q.pagination_order = Some("none".to_string());
    }))
    .expect("`none` compiles");
    assert_eq!(q.pagination_order, None);
}

#[test]
fn a_query_that_cannot_paginate_gets_no_pagination_order() {
    // Three ways not to paginate, and none of them may carry an ordering for
    // pages that do not exist.
    let single = convert_one(paginating_query(|q| q.returns_list = false))
        .expect("a single-item query compiles");
    assert_eq!(single.pagination_order, None, "a single item is not a page");

    let unpaged = convert_one(paginating_query(|q| {
        q.auto_params = Some(IntermediateAutoParams {
            limit:        Some(false),
            offset:       Some(false),
            where_clause: None,
            order_by:     None,
        });
    }))
    .expect("a list query with limit and offset off compiles");
    assert_eq!(unpaged.pagination_order, None, "no limit and no offset is no page");

    let relay = convert_one(paginating_query(|q| q.relay = true)).expect("a relay query compiles");
    assert_eq!(
        relay.pagination_order, None,
        "relay pages by cursor, ordered by relay_cursor_column"
    );
    assert_eq!(relay.relay_cursor_column.as_deref(), Some("pk_user"));
}

#[test]
fn only_one_of_limit_and_offset_still_paginates() {
    // `?limit=` alone slices a prefix and `?offset=` alone slices a suffix; both
    // are a *slice of a sequence*, so both need one.
    for (limit, offset) in [(true, false), (false, true)] {
        let q = convert_one(paginating_query(|q| {
            q.auto_params = Some(IntermediateAutoParams {
                limit:        Some(limit),
                offset:       Some(offset),
                where_clause: None,
                order_by:     None,
            });
        }))
        .expect("compiles");
        assert_eq!(
            q.pagination_order,
            Some(PaginationOrder::Identity),
            "limit={limit} offset={offset}"
        );
    }
}

#[test]
fn an_unsafe_pagination_order_identifier_is_refused() {
    // The value is interpolated into ORDER BY, not bound — the same boundary
    // `additional_views` is validated at.
    let err = convert_one(paginating_query(|q| {
        q.pagination_order = Some("id; DROP TABLE tb_user".to_string());
    }))
    .expect_err("an injection payload is not an identifier");
    assert!(
        err.to_string().contains("not a valid SQL identifier"),
        "the refusal must name what is wrong: {err}"
    );
}

#[test]
fn a_pagination_order_on_a_query_that_cannot_paginate_is_refused() {
    // Refused rather than ignored: an ordering that silently applies to nothing
    // looks exactly like one that works.
    let err = convert_one(paginating_query(|q| {
        q.returns_list = false;
        q.pagination_order = Some("pk_user".to_string());
    }))
    .expect_err("a single-item query has no pages to order");
    assert!(
        err.to_string().contains("does not paginate"),
        "the refusal must name the constraint: {err}"
    );
    assert!(
        err.to_string().contains("returns a single item"),
        "and say which of the three reasons applies: {err}"
    );

    let relay_err = convert_one(paginating_query(|q| {
        q.relay = true;
        q.pagination_order = Some("pk_user".to_string());
    }))
    .expect_err("relay orders by its cursor column");
    assert!(
        relay_err.to_string().contains("relay_cursor_column"),
        "the refusal must point at the field that already decides this: {relay_err}"
    );
}

#[test]
fn the_count_sibling_carries_no_pagination_order() {
    // A `SELECT COUNT(*)` has no rows to order, and inheriting the parent's
    // ordering would add a sort to a scalar read.
    let parent = convert_one(paginating_query(|q| q.count = true)).expect("compiles");
    assert_eq!(parent.pagination_order, Some(PaginationOrder::Identity));
    assert_eq!(parent.count_sibling().pagination_order, None);
}

// ── #1418: `jwt_optional` — a mutation argument may be absent, a row filter may not ──

mod optional_claim {
    use fraiseql_core::schema::InjectedParamSource;

    use crate::schema::{
        SchemaConverter, converter::tenancy::validate_tenant_annotations,
        intermediate::IntermediateSchema,
    };

    fn schema(json: &str) -> IntermediateSchema {
        serde_json::from_str(json).unwrap()
    }

    const INVOICE: &str =
        r#"{"name": "Invoice", "fields": [{"name": "id", "type": "ID", "nullable": false}]}"#;

    #[test]
    fn a_mutation_may_inject_an_optional_claim_in_either_wire_shape() {
        for inject in [
            r#""jwt_optional:act""#,
            r#"{"source": "jwt_optional", "claim": "act"}"#,
        ] {
            let compiled = SchemaConverter::convert(schema(&format!(
                r#"{{"types": [{INVOICE}], "mutations": [{{"name": "createInvoice",
                    "return_type": "Invoice", "sql_source": "app.create_invoice",
                    "inject_params": {{"p_actor": {inject}}}}}]}}"#
            )))
            .expect("an optional claim is a valid mutation argument");
            assert_eq!(
                compiled.mutations[0].inject_params.get("p_actor"),
                Some(&InjectedParamSource::JwtOptional("act".to_string()))
            );
        }
    }

    #[test]
    fn a_query_may_not_filter_rows_on_an_optional_claim() {
        let err = SchemaConverter::convert(schema(&format!(
            r#"{{"types": [{INVOICE}], "queries": [{{"name": "invoices",
                "return_type": "Invoice", "returns_list": true, "sql_source": "v_invoice",
                "inject_params": {{"act": "jwt_optional:act"}}}}]}}"#
        )))
        .unwrap_err();
        let err = format!("{err:#}");
        assert!(err.contains("invoices") && err.contains("jwt_optional"), "{err}");
    }

    #[test]
    fn a_type_may_not_scope_its_rows_on_an_optional_claim() {
        let err = SchemaConverter::convert(schema(
            r#"{"types": [{"name": "Invoice", "fields": [{"name": "id", "type": "ID", "nullable": false}],
                "sql_source": "v_invoice", "inject_params": {"act": "jwt_optional:act"}}]}"#,
        ))
        .unwrap_err();
        let err = format!("{err:#}");
        assert!(err.contains("Invoice") && err.contains("jwt_optional"), "{err}");
    }

    #[test]
    fn the_tenant_claim_is_never_optional() {
        let mut intermediate = schema(&format!(
            r#"{{"types": [{INVOICE}], "mutations": [{{"name": "createInvoice",
                "return_type": "Invoice", "sql_source": "app.create_invoice",
                "inject_params": {{"p_tenant": "jwt_optional:org_id"}}}}]}}"#
        ));
        let err = validate_tenant_annotations(&mut intermediate, "org_id")
            .unwrap_err()
            .to_string();
        assert!(err.contains("createInvoice") && err.contains("org_id"), "{err}");
    }
}

// ── #1498: a fact table's path column declares its hierarchy ──────────────────

#[test]
fn a_fact_table_filter_carries_its_ltree_type_and_hierarchy_into_the_compiled_schema() {
    use fraiseql_core::compiler::fact_table::SqlType;

    let intermediate: IntermediateSchema = serde_json::from_str(
        r#"{"fact_tables": [{"table_name": "tf_org", "measures": [],
            "dimensions": {"name": "data", "paths": []},
            "denormalized_filters": [
                {"name": "org_path", "sql_type": "LTREE", "indexed": true, "hierarchy": "org"}
            ]}]}"#,
    )
    .unwrap();
    let compiled = SchemaConverter::convert(intermediate).unwrap();
    let filter = &compiled.fact_tables["tf_org"].denormalized_filters[0];
    assert_eq!(filter.sql_type, SqlType::Ltree);
    assert_eq!(filter.hierarchy.as_deref(), Some("org"));
}

// #1586: each list level records whether its items may be `null`, at any depth; the
// outermost `!` is the field's own nullability.
#[test]
fn parse_field_type_records_each_levels_item_nullability() {
    let declared = crate::schema::converter::DeclaredTypeNames::default();
    let parse = |t: &str| SchemaConverter::parse_field_type(t, &declared).unwrap();
    let (strict, loose) = (FieldType::list, FieldType::list_of_nullable);
    for (written, expected) in [
        ("[Float]", loose(FieldType::Float)),
        ("[Float]!", loose(FieldType::Float)),
        ("[Float!]", strict(FieldType::Float)),
        ("[Float!]!", strict(FieldType::Float)),
        ("Float", FieldType::Float),
        ("[[Float]]", loose(loose(FieldType::Float))),
        ("[[Float]!]", strict(loose(FieldType::Float))),
        ("[[Float!]]", loose(strict(FieldType::Float))),
        ("[[[Float!]]!]!", strict(loose(strict(FieldType::Float)))),
    ] {
        assert_eq!(parse(written), expected, "{written}");
    }
}

// ── #1657: subscription_policy is authorable ─────────────────────────────────

/// An authoring type `Order` with an `owner_id` field and the given `subscription_policy`.
fn order_with_policy(policy: &serde_json::Value) -> Result<IntermediateSchema, serde_json::Error> {
    serde_json::from_value(serde_json::json!({
        "types": [{
            "name": "Order",
            "sql_source": "v_order",
            "fields": [
                {"name": "id", "type": "ID", "nullable": false},
                {"name": "owner_id", "type": "String", "nullable": false}
            ],
            "subscription_policy": policy
        }]
    }))
}

#[test]
fn a_declared_subscription_policy_reaches_the_compiled_type() {
    let policy = serde_json::json!({"owner_path": "$.owner_id", "identity_field": "user_id"});
    let parsed = order_with_policy(&policy);
    assert!(parsed.is_ok(), "an authoring type must accept subscription_policy: {parsed:?}");
    let compiled = SchemaConverter::convert(parsed.unwrap()).unwrap();
    let order = compiled.types.iter().find(|t| t.name == "Order").unwrap();
    assert_eq!(
        order.subscription_policy,
        Some(serde_json::from_value(policy).unwrap()),
        "the declared policy must reach the compiled type"
    );
}

/// An `is_input` type's fields are rebuilt as input fields. Every attribute an output
/// field can carry either has an input-field slot and is carried, or is refused by name:
/// none is dropped.
mod is_input_field_attributes {
    use serde_json::{Value, json};

    use super::*;

    /// A schema whose input type `CreateUserInput` has one field, `email`, with `extra`
    /// merged into its declaration.
    fn schema_with_input_field(extra: &Value) -> IntermediateSchema {
        let mut field = json!({"name": "email", "type": "String", "nullable": false});
        for (key, value) in extra.as_object().unwrap() {
            field[key] = value.clone();
        }
        serde_json::from_value(json!({
            "types": [
                {"name": "User", "sql_source": "v_user",
                 "fields": [{"name": "id", "type": "ID", "nullable": false}]},
                {"name": "CreateUserInput", "is_input": true, "fields": [field]}
            ],
            "queries": [{"name": "users", "return_type": "User", "returns_list": true,
                         "sql_source": "v_user"}],
            "mutations": [{"name": "createUser", "return_type": "User",
                           "sql_source": "fn_create_user", "operation": "INSERT",
                           "arguments": [{"name": "input", "type": "CreateUserInput",
                                          "nullable": false}]}]
        }))
        .unwrap()
    }

    fn refusal(extra: &Value) -> String {
        format!(
            "{:#}",
            SchemaConverter::convert(schema_with_input_field(extra))
                .expect_err("the attribute must be refused, not dropped")
        )
    }

    fn compiled_email(extra: &Value) -> fraiseql_core::schema::InputFieldDefinition {
        let compiled = SchemaConverter::convert(schema_with_input_field(extra))
            .expect("the attribute must compile");
        let input = compiled.input_types.iter().find(|i| i.name == "CreateUserInput").unwrap();
        input.fields.iter().find(|f| f.name == "email").unwrap().clone()
    }

    #[test]
    fn each_output_only_attribute_is_refused_by_name() {
        let cases = [
            ("requires_scope", json!("read:User.email")),
            ("on_deny", json!("mask")),
            ("authorize", json!(true)),
            ("function", json!("fn_email")),
            ("hierarchy", json!("parent_id")),
            ("unique", json!(true)),
            ("vector_config", json!({"dimensions": 3})),
            ("vector_distance", json!("embedding")),
        ];
        for (key, value) in cases {
            let message = refusal(&json!({key: value}));
            assert!(message.contains("CreateUserInput"), "{key}: names the type: {message}");
            assert!(message.contains("'email'"), "{key}: names the field: {message}");
            assert!(message.contains(&format!("`{key}`")), "{key}: names the key: {message}");
        }
    }

    #[test]
    fn a_directive_with_no_input_field_slot_is_refused_by_name() {
        for name in ["pattern", "inaccessible", "tag"] {
            let message = refusal(&json!({"directives": [{"name": name}]}));
            assert!(message.contains("'email'"), "@{name}: names the field: {message}");
            assert!(message.contains(&format!("@{name}")), "names the directive: {message}");
        }
    }

    #[test]
    fn the_deprecated_key_is_carried() {
        let field = compiled_email(&json!({"deprecated": {"reason": "use contact"}}));
        assert_eq!(field.deprecation.and_then(|d| d.reason).as_deref(), Some("use contact"));
    }

    #[test]
    fn a_deprecated_directive_is_carried() {
        let field = compiled_email(
            &json!({"directives": [{"name": "deprecated", "arguments": {"reason": "old"}}]}),
        );
        assert_eq!(field.deprecation.and_then(|d| d.reason).as_deref(), Some("old"));
    }

    /// As on an output field, the first-class key wins over the directive.
    #[test]
    fn the_deprecated_key_wins_over_the_directive() {
        let field = compiled_email(&json!({
            "deprecated": {"reason": "key"},
            "directives": [{"name": "deprecated", "arguments": {"reason": "directive"}}]
        }));
        assert_eq!(field.deprecation.and_then(|d| d.reason).as_deref(), Some("key"));
    }

    /// `false` declares nothing, so there is nothing to refuse.
    #[test]
    fn a_flag_set_to_false_compiles() {
        compiled_email(&json!({"authorize": false, "unique": false}));
    }

    #[test]
    fn an_input_field_with_no_extra_attribute_compiles_undeprecated() {
        assert!(compiled_email(&json!({})).deprecation.is_none());
    }

    #[test]
    fn each_output_only_type_attribute_is_refused_by_name() {
        let cases = [
            ("sql_source", json!("v_x")),
            ("implements", json!(["Node"])),
            ("requires_role", json!("admin")),
            ("is_error", json!(true)),
            ("relay", json!(true)),
            ("subscribable_tables", json!(["tb_x"])),
        ];
        for (key, value) in cases {
            let mut schema = serde_json::to_value(schema_with_input_field(&json!({}))).unwrap();
            schema["types"][1][key] = value;
            let message = format!(
                "{:#}",
                SchemaConverter::convert(serde_json::from_value(schema).unwrap())
                    .expect_err("an output-only type attribute must be refused")
            );
            assert!(message.contains(&format!("`{key}`")), "{key}: {message}");
        }
    }
}