fraiseql-core 2.16.0

#![allow(clippy::unwrap_used)] // Reason: test code, panics are acceptable

use super::{
    query::QueryDefinition,
    schema::{CompiledSchema, SubscribableEntity},
};
#[cfg(feature = "federation")]
use crate::schema::config_types::FederationEntity;
use crate::schema::{
    CURRENT_FRAISEQL_VERSION, MutationDefinition, ProducerVersion,
    config_types::{FederationConfig, NamingConvention},
    graphql_type_defs::TypeDefinition,
    observer_types::{ObserverDefinition, RetryConfig},
    security_config::{RoleDefinition, SecurityConfig},
};

#[test]
fn test_compiled_schema_with_observers() {
    let json = r#"{
        "types": [],
        "enums": [],
        "input_types": [],
        "interfaces": [],
        "unions": [],
        "queries": [],
        "mutations": [],
        "subscriptions": [],
        "observers": [
            {
                "name": "onHighValueOrder",
                "entity": "Order",
                "event": "INSERT",
                "condition": "total > 1000",
                "actions": [
                    {
                        "type": "webhook",
                        "url": "https://api.example.com/webhook"
                    }
                ],
                "retry": {
                    "max_attempts": 3,
                    "backoff_strategy": "exponential",
                    "initial_delay_ms": 1000,
                    "max_delay_ms": 60000
                }
            }
        ]
    }"#;

    let schema = CompiledSchema::from_json(json, false).unwrap();

    assert!(schema.has_observers());
    assert_eq!(schema.observer_count(), 1);

    let observer = schema.find_observer("onHighValueOrder").unwrap();
    assert_eq!(observer.entity, "Order");
    assert_eq!(observer.event, "INSERT");
    assert_eq!(observer.condition, Some("total > 1000".to_string()));
    assert_eq!(observer.actions.len(), 1);
    assert_eq!(observer.retry.max_attempts, 3);
    assert!(observer.retry.is_exponential());
}

#[test]
fn test_compiled_schema_backward_compatible() {
    // Schema without observers field should still load
    let json = r#"{
        "types": [],
        "enums": [],
        "input_types": [],
        "interfaces": [],
        "unions": [],
        "queries": [],
        "mutations": [],
        "subscriptions": []
    }"#;

    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert!(!schema.has_observers());
    assert_eq!(schema.observer_count(), 0);
}

#[test]
fn test_find_observers_for_entity() {
    let schema = CompiledSchema {
        observers: vec![
            ObserverDefinition::new("onOrderInsert", "Order", "INSERT"),
            ObserverDefinition::new("onOrderUpdate", "Order", "UPDATE"),
            ObserverDefinition::new("onUserInsert", "User", "INSERT"),
        ],
        ..Default::default()
    };

    let order_observers = schema.find_observers_for_entity("Order");
    assert_eq!(order_observers.len(), 2);

    let user_observers = schema.find_observers_for_entity("User");
    assert_eq!(user_observers.len(), 1);
}

#[test]
fn test_find_observers_for_event() {
    let schema = CompiledSchema {
        observers: vec![
            ObserverDefinition::new("onOrderInsert", "Order", "INSERT"),
            ObserverDefinition::new("onOrderUpdate", "Order", "UPDATE"),
            ObserverDefinition::new("onUserInsert", "User", "INSERT"),
        ],
        ..Default::default()
    };

    let insert_observers = schema.find_observers_for_event("INSERT");
    assert_eq!(insert_observers.len(), 2);

    let update_observers = schema.find_observers_for_event("UPDATE");
    assert_eq!(update_observers.len(), 1);
}

#[test]
fn test_observer_definition_builder() {
    let observer = ObserverDefinition::new("test", "Order", "INSERT")
        .with_condition("total > 1000")
        .with_action(serde_json::json!({"type": "webhook", "url": "https://example.com"}))
        .with_retry(RetryConfig::exponential(5, 1000, 60000));

    assert_eq!(observer.name, "test");
    assert_eq!(observer.entity, "Order");
    assert_eq!(observer.event, "INSERT");
    assert!(observer.has_condition());
    assert_eq!(observer.action_count(), 1);
    assert_eq!(observer.retry.max_attempts, 5);
}

#[test]
fn test_retry_config_types() {
    let exponential = RetryConfig::exponential(3, 1000, 60000);
    assert!(exponential.is_exponential());
    assert!(!exponential.is_linear());
    assert!(!exponential.is_fixed());

    let linear = RetryConfig::linear(3, 1000, 60000);
    assert!(!linear.is_exponential());
    assert!(linear.is_linear());
    assert!(!linear.is_fixed());

    let fixed = RetryConfig::fixed(3, 5000);
    assert!(!fixed.is_exponential());
    assert!(!fixed.is_linear());
    assert!(fixed.is_fixed());
    assert_eq!(fixed.initial_delay_ms, 5000);
    assert_eq!(fixed.max_delay_ms, 5000);
}

// =========================================================================
// content_hash tests
// =========================================================================

#[test]
fn test_content_hash_stable() {
    let schema = CompiledSchema::default();
    assert_eq!(
        schema.content_hash(),
        schema.content_hash(),
        "Same schema must produce same hash"
    );
}

#[test]
fn test_content_hash_length() {
    let hash = CompiledSchema::default().content_hash();
    assert_eq!(hash.len(), 32, "Hash must be 32 hex chars (16 bytes)");
    assert!(hash.chars().all(|c| c.is_ascii_hexdigit()), "Hash must be valid hex");
}

#[test]
fn test_content_hash_changes_on_field_rename() {
    let mut schema_a = CompiledSchema::default();
    schema_a
        .queries
        .push(QueryDefinition::new("users", "User").with_sql_source("v_user"));

    let mut schema_b = CompiledSchema::default();
    schema_b
        .queries
        .push(QueryDefinition::new("users", "User").with_sql_source("v_account")); // different view

    assert_ne!(
        schema_a.content_hash(),
        schema_b.content_hash(),
        "Schemas with different view names must produce different hashes"
    );
}

// =========================================================================
// #366 @subscribable tests
// =========================================================================

#[test]
fn subscribable_defaults_empty_when_absent() {
    // A compiled schema that predates #366 has no `subscribable` key — it must
    // deserialize to an empty list (back-compat), mirroring the `authorize` field.
    let json = r#"{ "types": [], "queries": [], "mutations": [], "subscriptions": [] }"#;
    let schema: CompiledSchema = serde_json::from_str(json).unwrap();
    assert!(schema.subscribable.is_empty(), "absent subscribable defaults to empty");
}

#[test]
fn subscribable_empty_is_not_serialized_so_hash_is_unchanged() {
    // skip_serializing_if = "Vec::is_empty" keeps a non-subscribable schema's JSON
    // (and therefore its content_hash) byte-identical to a pre-#366 schema — so
    // existing golden/insta fixtures need no re-blessing.
    let schema = CompiledSchema::default();
    let json = serde_json::to_string(&schema).unwrap();
    assert!(!json.contains("subscribable"), "empty subscribable is omitted: {json}");
}

#[test]
fn subscribable_round_trips_when_present() {
    let schema = CompiledSchema {
        subscribable: vec![SubscribableEntity {
            entity_type: "Post".to_string(),
            tables:      vec!["tb_post".to_string(), "public.tb_post_archive".to_string()],
            pre_image:   false,
        }],
        ..CompiledSchema::default()
    };
    let json = serde_json::to_string(&schema).unwrap();
    assert!(json.contains("subscribable"), "present subscribable is serialized");
    let back: CompiledSchema = serde_json::from_str(&json).unwrap();
    assert_eq!(back.subscribable, schema.subscribable, "subscribable round-trips");
}

// ── cascade flag (opt-in typed cascade surface, gate decision 3) ──────

#[test]
fn mutation_cascade_defaults_off() {
    let m = MutationDefinition::new("createPost", "Post");
    assert!(!m.cascade, "a new mutation has the cascade surface off by default");
}

#[test]
fn mutation_cascade_absent_is_not_serialized() {
    // skip_serializing_if keeps a non-cascade mutation byte-identical to one
    // authored before the flag existed, so the codegen schema hash is unchanged.
    let m = MutationDefinition::new("createPost", "Post");
    let json = serde_json::to_string(&m).unwrap();
    assert!(!json.contains("cascade"), "off cascade is omitted: {json}");
}

#[test]
fn mutation_cascade_round_trips_when_on() {
    let json = r#"{ "name": "createPost", "return_type": "Post", "cascade": true }"#;
    let m: MutationDefinition = serde_json::from_str(json).unwrap();
    assert!(m.cascade, "cascade=true deserializes");
    let back = serde_json::to_string(&m).unwrap();
    assert!(back.contains("\"cascade\":true"), "on cascade serializes: {back}");
}

#[test]
fn subscribable_participates_in_content_hash() {
    let plain = CompiledSchema::default();
    let with_sub = CompiledSchema {
        subscribable: vec![SubscribableEntity {
            entity_type: "Post".to_string(),
            tables:      vec!["tb_post".to_string()],
            pre_image:   false,
        }],
        ..CompiledSchema::default()
    };
    assert_ne!(
        plain.content_hash(),
        with_sub.content_hash(),
        "adding a subscribable entity changes the content hash"
    );
}

// =========================================================================
// has_rls_configured tests
// =========================================================================

#[test]
fn test_has_rls_configured_no_security() {
    let schema = CompiledSchema::default();
    assert!(
        !schema.has_rls_configured(),
        "Schema with no security section must return false"
    );
}

#[test]
fn test_has_rls_configured_with_empty_policies() {
    let sec = SecurityConfig {
        policies: vec![],
        ..SecurityConfig::default()
    };
    let schema = CompiledSchema {
        security: Some(sec),
        ..CompiledSchema::default()
    };
    assert!(!schema.has_rls_configured(), "Empty policies array must return false");
}

#[test]
fn test_has_rls_configured_with_rls_declaration() {
    let sec = SecurityConfig {
        rls: crate::schema::RlsConfig { enabled: true },
        ..SecurityConfig::default()
    };
    let schema = CompiledSchema {
        security: Some(sec),
        ..CompiledSchema::default()
    };
    assert!(schema.has_rls_configured(), "`[security.rls] enabled = true` must return true");
}

/// `security.policies` is *authorization* policy, not RLS, and #612 made a
/// non-empty `[security.policies]` a hard compile error — so counting it made
/// `has_rls_configured()` unconditionally false for every producible schema (#758).
#[test]
fn test_has_rls_configured_ignores_authorization_policies() {
    let sec = SecurityConfig {
        policies: vec![crate::schema::AuthorizationPolicy {
            name:              "tenant_isolation".to_string(),
            policy_type:       "RBAC".to_string(),
            rule:              Some("tenant_id = $1".to_string()),
            roles:             vec!["tenant".to_string()],
            strategy:          None,
            attributes:        vec![],
            description:       None,
            cache_ttl_seconds: None,
        }],
        ..SecurityConfig::default()
    };
    let schema = CompiledSchema {
        security: Some(sec),
        ..CompiledSchema::default()
    };
    assert!(!schema.has_rls_configured());
}

#[test]
fn test_has_rls_configured_no_policies_key() {
    let sec = SecurityConfig {
        rate_limiting: Some(crate::schema::RateLimitingSecurityConfig {
            enabled: true,
            ..crate::schema::RateLimitingSecurityConfig::default()
        }),
        ..SecurityConfig::default()
    };
    let schema = CompiledSchema {
        security: Some(sec),
        ..CompiledSchema::default()
    };
    assert!(!schema.has_rls_configured(), "Security without policies key must return false");
}

// ---------------------------------------------------------------------------
// schema.rs tests
// ---------------------------------------------------------------------------
// -------------------------------------------------------------------------
// Helpers
// -------------------------------------------------------------------------

fn make_type_def(name: &str) -> TypeDefinition {
    TypeDefinition {
        name:                name.into(),
        sql_source:          format!("v_{}", name.to_lowercase()).as_str().into(),
        jsonb_column:        "data".to_string(),
        fields:              vec![],
        description:         None,
        sql_projection_hint: None,
        implements:          vec![],
        requires_role:       None,
        is_error:            false,
        relay:               false,
        internal:            false,
        embedded:            false,
        relationships:       vec![],
        inject_params:       indexmap::IndexMap::new(),
        subscription_policy: None,
    }
}

fn make_query(name: &str, return_type: &str) -> QueryDefinition {
    QueryDefinition::new(name, return_type)
}

fn make_mutation(name: &str, return_type: &str) -> MutationDefinition {
    MutationDefinition::new(name, return_type)
}

// -------------------------------------------------------------------------
// Constructor behaviour
// -------------------------------------------------------------------------

#[test]
fn new_returns_empty_schema() {
    let schema = CompiledSchema::new();
    assert!(schema.types.is_empty());
    assert!(schema.queries.is_empty());
    assert!(schema.mutations.is_empty());
    assert!(schema.subscriptions.is_empty());
    assert!(schema.enums.is_empty());
    assert!(schema.interfaces.is_empty());
    assert!(schema.unions.is_empty());
}

#[test]
fn from_json_empty_array_fields() {
    let json = r#"{"types":[],"queries":[],"mutations":[],"subscriptions":[]}"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert_eq!(schema.types.len(), 0);
    assert_eq!(schema.queries.len(), 0);
    assert_eq!(schema.mutations.len(), 0);
    assert_eq!(schema.subscriptions.len(), 0);
}

#[test]
fn from_json_minimal_empty_object() {
    // All fields have #[serde(default)] — an empty JSON object is valid
    let schema = CompiledSchema::from_json("{}", false).unwrap();
    assert!(schema.types.is_empty());
    assert!(schema.queries.is_empty());
}

#[test]
fn from_json_invalid_returns_error() {
    let result = CompiledSchema::from_json("not json at all", false);
    assert!(result.is_err());
}

#[test]
fn from_json_builds_query_index() {
    let json = r#"{
        "types": [{"name":"User","sql_source":"v_user","fields":[]}],
        "queries": [{"name":"users","return_type":"User"}],
        "mutations": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert!(schema.query_index.contains_key("users"));
    assert_eq!(schema.query_index["users"], 0);
}

// ---------------------------------------------------------------------------
// #1142: type-level inject_params
// ---------------------------------------------------------------------------

/// A query and its return type may both scope the same column — but not from two
/// different sources. The `_entities` path merges the two declarations, and a merge
/// that picked a winner would silently enforce one of a contradictory pair while the
/// author believes the other holds.
#[test]
fn a_query_and_its_type_disagreeing_on_an_inject_column_is_refused_at_load() {
    let json = r#"{
        "types": [{
            "name":"User","sql_source":"v_user","fields":[],
            "inject_params": {"tenant_id": {"source":"jwt","claim":"org_id"}}
        }],
        "queries": [{
            "name":"users","return_type":"User","sql_source":"v_user",
            "inject_params": {"tenant_id": {"source":"jwt","claim":"tenant_id"}}
        }],
        "mutations": [],
        "subscriptions": []
    }"#;
    let err = CompiledSchema::from_json(json, false).unwrap_err();
    let msg = err.to_string();
    assert!(msg.contains("tenant_id"), "the message must name the column, got: {msg}");
    assert!(msg.contains("User"), "the message must name the type, got: {msg}");
    assert!(msg.contains("users"), "the message must name the query, got: {msg}");
}

/// Ruling Y 7: a scope is granted only by a role definition, and those live in the
/// `security` section. A schema that declares `requires_scope` without one declares a gate
/// no principal can pass and, until now, one the runtime silently switched off. Refused at
/// load, naming the field and the scope.
#[test]
fn a_scoped_field_without_a_security_section_is_refused_at_load() {
    let json = r#"{
        "types": [{
            "name":"User","sql_source":"v_user",
            "fields":[
                {"name":"id","field_type":"ID"},
                {"name":"salary","field_type":"Int","requires_scope":"read:salary"}
            ]
        }],
        "queries": [], "mutations": [], "subscriptions": []
    }"#;
    let err = CompiledSchema::from_json(json, false).unwrap_err();
    let msg = err.to_string();
    assert!(msg.contains("User.salary"), "the message must name the field, got: {msg}");
    assert!(msg.contains("read:salary"), "the message must name the scope, got: {msg}");
}

/// Control: the same field with a `security` section loads.
#[test]
fn a_scoped_field_with_a_security_section_loads() {
    let json = r#"{
        "types": [{
            "name":"User","sql_source":"v_user",
            "fields":[
                {"name":"id","field_type":"ID"},
                {"name":"salary","field_type":"Int","requires_scope":"read:salary"}
            ]
        }],
        "queries": [], "mutations": [], "subscriptions": [],
        "security": {}
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert!(schema.security.is_some());
}

/// Control: the same column from the *same* source is the redundant-but-consistent
/// case, and must load — otherwise the check would refuse every schema that states its
/// scoping in both places.
#[test]
fn a_query_and_its_type_agreeing_on_an_inject_column_loads() {
    let json = r#"{
        "types": [{
            "name":"User","sql_source":"v_user","fields":[],
            "inject_params": {"tenant_id": {"source":"jwt","claim":"tenant_id"}}
        }],
        "queries": [{
            "name":"users","return_type":"User","sql_source":"v_user",
            "inject_params": {"tenant_id": {"source":"jwt","claim":"tenant_id"}}
        }],
        "mutations": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert_eq!(schema.types[0].inject_params.len(), 1);
}

/// Control: declaring a column the query does not is the *point* of the field — the
/// two sets merge. Only a same-column disagreement is a contradiction.
#[test]
fn a_type_scoping_a_column_its_query_does_not_loads() {
    let json = r#"{
        "types": [{
            "name":"User","sql_source":"v_user","fields":[],
            "inject_params": {"owner_id": {"source":"jwt","claim":"sub"}}
        }],
        "queries": [{
            "name":"users","return_type":"User","sql_source":"v_user",
            "inject_params": {"tenant_id": {"source":"jwt","claim":"tenant_id"}}
        }],
        "mutations": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert_eq!(schema.types[0].inject_params.len(), 1);
    assert_eq!(schema.queries[0].inject_params.len(), 1);
}

/// The queryless shape #1142 exists for: the declaration survives a load/serialize
/// round trip with no backing query to carry it.
#[test]
fn a_queryless_type_carries_its_inject_params_through_a_round_trip() {
    let json = r#"{
        "types": [{
            "name":"User","sql_source":"v_user","fields":[],
            "inject_params": {"tenant_id": {"source":"jwt","claim":"tenant_id"}}
        }],
        "queries": [],
        "mutations": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    let reloaded = CompiledSchema::from_json(&schema.to_json().unwrap(), false).unwrap();
    assert_eq!(
        reloaded.types[0].inject_params.get("tenant_id"),
        Some(&crate::schema::InjectedParamSource::Jwt("tenant_id".to_string())),
        "the type's scoping must survive the compiled artefact it is written to"
    );
}

#[test]
fn from_json_builds_mutation_index() {
    let json = r#"{
        "types": [{"name":"User","sql_source":"v_user","fields":[]}],
        "mutations": [{"name":"createUser","return_type":"User"}],
        "queries": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert!(schema.mutation_index.contains_key("createUser"));
}

// -------------------------------------------------------------------------
// Serialization round-trip
// -------------------------------------------------------------------------

#[test]
fn to_json_and_back_is_identity() {
    let schema = CompiledSchema::new();
    let json = schema.to_json().unwrap();
    let schema2 = CompiledSchema::from_json(&json, false).unwrap();
    assert_eq!(schema, schema2);
}

#[test]
fn to_json_pretty_is_valid_json() {
    let schema = CompiledSchema::new();
    let pretty = schema.to_json_pretty().unwrap();
    // Should re-parse without error
    let _: serde_json::Value = serde_json::from_str(&pretty).unwrap();
}

// -------------------------------------------------------------------------
// Producer version (#1304): a compiled schema is a build artifact of the
// release that wrote it, so a runtime accepts its own build's artifact and
// nothing else. The two producers of a `CompiledSchema` — the compiler, which
// stamps what it writes, and in-process construction, which *is* this build —
// must stay distinguishable from the third case, an artifact carrying no stamp.
// -------------------------------------------------------------------------

#[test]
fn a_schema_built_in_process_is_this_build() {
    let schema = CompiledSchema::new();
    assert_eq!(schema.fraiseql_version.get(), Some(CURRENT_FRAISEQL_VERSION));
    assert!(schema.validate_producer_version().is_ok());
}

#[test]
fn an_artifact_from_another_build_is_refused() {
    let mut schema = CompiledSchema::new();
    schema.fraiseql_version = serde_json::from_value(serde_json::json!("2.14.0")).unwrap();

    let msg = schema.validate_producer_version().expect_err("another build must be refused");
    assert!(msg.contains("2.14.0"), "the message must name the producing build: {msg}");
    assert!(
        msg.contains(CURRENT_FRAISEQL_VERSION),
        "the message must name this runtime's build: {msg}"
    );
    assert!(msg.contains("Recompile"), "the message must tell an operator what to do: {msg}");
}

/// The hole the format-version check left open, closed: an artifact carrying no
/// stamp is the *older* case, and accepting it admitted the stalest artifacts
/// while refusing merely-stale ones.
#[test]
fn an_unstamped_artifact_is_refused() {
    let mut schema = CompiledSchema::new();
    schema.fraiseql_version = ProducerVersion::unstamped();

    let msg = schema
        .validate_producer_version()
        .expect_err("an unstamped artifact is refused");
    assert!(
        msg.contains("fraiseql_version"),
        "the message must name the missing stamp: {msg}"
    );
    assert!(msg.contains("Recompile"), "and the remedy: {msg}");
}

/// The discrimination the `Option<u32>` format version could not express. A
/// missing JSON key must not deserialize to whatever this build constructs in
/// process, or an artifact from any release would pass by looking freshly built.
#[test]
fn a_missing_key_does_not_deserialize_to_this_build() {
    let deserialized: CompiledSchema = serde_json::from_str(
        r#"{"types": [], "queries": [], "mutations": [], "subscriptions": []}"#,
    )
    .unwrap();

    assert!(deserialized.fraiseql_version.is_unstamped());
    assert_ne!(
        deserialized.fraiseql_version,
        ProducerVersion::default(),
        "an artifact that names no build must not equal one this build constructed"
    );
    assert!(deserialized.validate_producer_version().is_err());
}

/// A stamped schema writes the stamp, so an artifact this build compiles is
/// accepted by this build after a round trip — the compiler's whole contract.
#[test]
fn a_stamped_schema_round_trips_through_json() {
    let json = serde_json::to_string(&CompiledSchema::new()).unwrap();
    assert!(json.contains(CURRENT_FRAISEQL_VERSION), "the stamp must reach the artifact");

    let reloaded = CompiledSchema::from_json(&json, false).unwrap();
    assert!(reloaded.validate_producer_version().is_ok());
}

/// And an unstamped one writes no key, rather than a `null` a reader would have
/// to interpret.
#[test]
fn an_unstamped_schema_serializes_no_key() {
    let mut schema = CompiledSchema::new();
    schema.fraiseql_version = ProducerVersion::unstamped();

    let value = serde_json::to_value(&schema).unwrap();
    assert!(
        value.get("fraiseql_version").is_none(),
        "an absent stamp is an absent key: {value:?}"
    );
}

// -------------------------------------------------------------------------
// Index building
// -------------------------------------------------------------------------

#[test]
fn build_indexes_populates_all_three_maps() {
    let mut schema = CompiledSchema::new();
    schema.queries.push(make_query("getUser", "User"));
    schema.mutations.push(make_mutation("createUser", "User"));
    schema.build_indexes();
    assert!(schema.query_index.contains_key("getUser"));
    assert!(schema.mutation_index.contains_key("createUser"));
}

#[test]
fn build_indexes_multiple_queries() {
    let mut schema = CompiledSchema::new();
    schema.queries.push(make_query("alpha", "A"));
    schema.queries.push(make_query("beta", "B"));
    schema.queries.push(make_query("gamma", "C"));
    schema.build_indexes();
    assert_eq!(schema.query_index["alpha"], 0);
    assert_eq!(schema.query_index["beta"], 1);
    assert_eq!(schema.query_index["gamma"], 2);
}

// -------------------------------------------------------------------------
// Finder methods
// -------------------------------------------------------------------------

#[test]
fn find_type_returns_none_for_missing() {
    let schema = CompiledSchema::new();
    assert!(schema.find_type("Ghost").is_none());
}

#[test]
fn find_type_returns_existing() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("User"));
    assert!(schema.find_type("User").is_some());
    assert_eq!(schema.find_type("User").unwrap().name, "User");
}

#[test]
fn find_query_uses_index_when_populated() {
    let json = r#"{
        "types": [{"name":"User","sql_source":"v_user","fields":[]}],
        "queries": [{"name":"users","return_type":"User"}],
        "mutations": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    let q = schema.find_query("users");
    assert!(q.is_some());
    assert_eq!(q.unwrap().name, "users");
}

#[test]
fn find_query_falls_back_to_linear_scan_without_index() {
    // Build schema directly without calling build_indexes
    let mut schema = CompiledSchema::new();
    schema.queries.push(make_query("direct", "String"));
    // query_index is empty but queries is not — should fall back to linear scan
    let q = schema.find_query("direct");
    assert!(q.is_some());
}

#[test]
fn find_query_returns_none_for_missing() {
    let schema = CompiledSchema::from_json("{}", false).unwrap();
    assert!(schema.find_query("nope").is_none());
}

#[test]
fn find_mutation_returns_correct_entry() {
    let json = r#"{
        "types": [{"name":"User","sql_source":"v_user","fields":[]}],
        "mutations": [{"name":"createUser","return_type":"User"}],
        "queries": [],
        "subscriptions": []
    }"#;
    let schema = CompiledSchema::from_json(json, false).unwrap();
    assert!(schema.find_mutation("createUser").is_some());
    assert!(schema.find_mutation("nope").is_none());
}

#[test]
fn find_interface_returns_none_when_absent() {
    let schema = CompiledSchema::new();
    assert!(schema.find_interface("Node").is_none());
}

#[test]
fn find_implementors_filters_by_interface() {
    let mut schema = CompiledSchema::new();
    let mut user = make_type_def("User");
    user.implements = vec!["Node".to_string()];
    schema.types.push(user);
    schema.types.push(make_type_def("Product")); // does not implement Node

    let implementors = schema.find_implementors("Node");
    assert_eq!(implementors.len(), 1);
    assert_eq!(implementors[0].name, "User");
}

// -------------------------------------------------------------------------
// operation_count
// -------------------------------------------------------------------------

#[test]
fn operation_count_sums_all_three() {
    let mut schema = CompiledSchema::new();
    schema.queries.push(make_query("q1", "String"));
    schema.queries.push(make_query("q2", "String"));
    schema.mutations.push(make_mutation("m1", "String"));
    assert_eq!(schema.operation_count(), 3);
}

#[test]
fn operation_count_zero_for_empty_schema() {
    assert_eq!(CompiledSchema::new().operation_count(), 0);
}

// -------------------------------------------------------------------------
// Fact tables
// -------------------------------------------------------------------------

#[test]
fn fact_table_add_and_get() {
    use crate::compiler::fact_table::{DimensionColumn, FactTableMetadata};

    let mut schema = CompiledSchema::new();
    assert!(!schema.has_fact_tables());

    let meta = FactTableMetadata {
        table_name:               "tf_sales".to_string(),
        type_name:                None,
        measures:                 vec![],
        dimensions:               DimensionColumn {
            name:  "data".to_string(),
            paths: vec![],
        },
        denormalized_filters:     vec![],
        calendar_dimensions:      vec![],
        partial_period:           None,
        native_measures:          std::collections::HashMap::new(),
        native_dimension_mapping: std::collections::HashMap::new(),
    };
    schema.add_fact_table("tf_sales".to_string(), meta);

    assert!(schema.has_fact_tables());
    assert!(schema.get_fact_table("tf_sales").is_some());
    assert!(schema.get_fact_table("tf_missing").is_none());
}

#[test]
fn list_fact_tables_returns_all_names() {
    use crate::compiler::fact_table::{DimensionColumn, FactTableMetadata};

    let make_meta = |name: &str| FactTableMetadata {
        table_name:               name.to_string(),
        type_name:                None,
        measures:                 vec![],
        dimensions:               DimensionColumn {
            name:  "data".to_string(),
            paths: vec![],
        },
        denormalized_filters:     vec![],
        calendar_dimensions:      vec![],
        partial_period:           None,
        native_measures:          std::collections::HashMap::new(),
        native_dimension_mapping: std::collections::HashMap::new(),
    };

    let mut schema = CompiledSchema::new();
    schema.add_fact_table("tf_a".to_string(), make_meta("tf_a"));
    schema.add_fact_table("tf_b".to_string(), make_meta("tf_b"));

    let names = schema.list_fact_tables();
    assert_eq!(names.len(), 2);
    assert!(names.contains(&"tf_a"));
    assert!(names.contains(&"tf_b"));
}

// -------------------------------------------------------------------------
// Observers
// -------------------------------------------------------------------------

#[test]
fn has_observers_false_for_empty_schema() {
    assert!(!CompiledSchema::new().has_observers());
}

#[test]
fn find_observer_returns_by_name() {
    let mut schema = CompiledSchema::new();
    schema.observers.push(ObserverDefinition::new("onInsert", "Order", "INSERT"));
    assert!(schema.find_observer("onInsert").is_some());
    assert!(schema.find_observer("missing").is_none());
}

#[test]
fn find_observers_for_entity_filters_correctly() {
    let mut schema = CompiledSchema::new();
    schema.observers.push(ObserverDefinition::new("obs1", "Order", "INSERT"));
    schema.observers.push(ObserverDefinition::new("obs2", "Order", "UPDATE"));
    schema.observers.push(ObserverDefinition::new("obs3", "User", "INSERT"));

    let order_obs = schema.find_observers_for_entity("Order");
    assert_eq!(order_obs.len(), 2);
    let user_obs = schema.find_observers_for_entity("User");
    assert_eq!(user_obs.len(), 1);
}

#[test]
fn find_observers_for_event_filters_correctly() {
    let mut schema = CompiledSchema::new();
    schema.observers.push(ObserverDefinition::new("obs1", "Order", "INSERT"));
    schema.observers.push(ObserverDefinition::new("obs2", "User", "INSERT"));
    schema.observers.push(ObserverDefinition::new("obs3", "Order", "DELETE"));

    let inserts = schema.find_observers_for_event("INSERT");
    assert_eq!(inserts.len(), 2);
}

#[test]
fn observer_count_matches_vec_length() {
    let mut schema = CompiledSchema::new();
    assert_eq!(schema.observer_count(), 0);
    schema.observers.push(ObserverDefinition::new("o1", "A", "INSERT"));
    assert_eq!(schema.observer_count(), 1);
}

// -------------------------------------------------------------------------
// Security helpers
// -------------------------------------------------------------------------

#[test]
fn is_multi_tenant_false_by_default() {
    assert!(!CompiledSchema::new().is_multi_tenant());
}

#[test]
fn is_multi_tenant_true_when_configured() {
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.multi_tenant = true;
    schema.security = Some(sec);
    assert!(schema.is_multi_tenant());
}

// ── tenancy_mode / tenancy_config ──────────────────────────────────

#[test]
fn tenancy_mode_none_by_default() {
    use crate::schema::TenancyMode;
    assert_eq!(CompiledSchema::new().tenancy_mode(), TenancyMode::None);
}

#[test]
fn tenancy_mode_row_when_configured() {
    use crate::schema::{TenancyConfig, TenancyMode};
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.tenancy = TenancyConfig {
        mode:         TenancyMode::Row,
        tenant_claim: "tenant_id".to_string(),
    };
    schema.security = Some(sec);
    assert_eq!(schema.tenancy_mode(), TenancyMode::Row);
}

#[test]
fn tenancy_mode_schema_when_configured() {
    use crate::schema::{TenancyConfig, TenancyMode};
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.tenancy = TenancyConfig {
        mode:         TenancyMode::Schema,
        tenant_claim: "org_id".to_string(),
    };
    schema.security = Some(sec);
    assert_eq!(schema.tenancy_mode(), TenancyMode::Schema);
}

#[test]
fn tenancy_config_none_without_security() {
    assert!(CompiledSchema::new().tenancy_config().is_none());
}

#[test]
fn tenancy_config_returns_default_when_security_present() {
    use crate::schema::TenancyMode;
    let mut schema = CompiledSchema::new();
    schema.security = Some(SecurityConfig::new());
    let tc = schema.tenancy_config().unwrap();
    assert_eq!(tc.mode, TenancyMode::None);
    assert_eq!(tc.tenant_claim, "tenant_id");
}

#[test]
fn tenancy_round_trip_through_json() {
    use crate::schema::{TenancyConfig, TenancyMode};
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.tenancy = TenancyConfig {
        mode:         TenancyMode::Row,
        tenant_claim: "org_id".to_string(),
    };
    schema.security = Some(sec);

    let json = schema.to_json().unwrap();
    let restored = CompiledSchema::from_json(&json, false).unwrap();
    assert_eq!(restored.tenancy_mode(), TenancyMode::Row);
    assert_eq!(restored.tenancy_config().unwrap().tenant_claim, "org_id");
}

#[test]
fn find_role_returns_none_without_security_config() {
    assert!(CompiledSchema::new().find_role("admin").is_none());
}

#[test]
fn find_role_returns_defined_role() {
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.add_role(RoleDefinition::new("editor", vec!["read:*".to_string()]));
    schema.security = Some(sec);
    assert!(schema.find_role("editor").is_some());
}

#[test]
fn role_has_scope_false_without_security() {
    assert!(!CompiledSchema::new().role_has_scope("admin", "read:*"));
}

#[test]
fn role_has_scope_true_when_granted() {
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.add_role(RoleDefinition::new("admin", vec!["read:*".to_string()]));
    schema.security = Some(sec);
    assert!(schema.role_has_scope("admin", "read:anything"));
    assert!(!schema.role_has_scope("admin", "write:anything"));
}

/// #784: `read:*` is documented as "any scope starting with `read:`" — it must
/// not also grant `readwrite:` scopes that merely share the string prefix.
#[test]
fn role_scope_verb_wildcard_stops_at_the_delimiter() {
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.add_role(RoleDefinition::new("reader", vec!["read:*".to_string()]));
    schema.security = Some(sec);
    assert!(schema.role_has_scope("reader", "read:User.email"));
    assert!(!schema.role_has_scope("reader", "readwrite:User.email"));
    // A wildcard grant matches only namespaced scopes, never a bare verb.
    assert!(!schema.role_has_scope("reader", "read"));
}

/// #784: `read:User.*` grants fields of `User` only; a bare-prefix grant like
/// `read*` is not a wildcard form and must not match anything by string prefix.
#[test]
fn role_scope_type_wildcard_requires_boundary() {
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.add_role(RoleDefinition::new("user_reader", vec!["read:User.*".to_string()]));
    sec.add_role(RoleDefinition::new("sloppy", vec!["read*".to_string()]));
    schema.security = Some(sec);
    assert!(schema.role_has_scope("user_reader", "read:User.email"));
    assert!(!schema.role_has_scope("user_reader", "read:UserAudit.email"));
    assert!(!schema.role_has_scope("sloppy", "readwrite:User.email"));
}

#[test]
fn get_role_scopes_empty_for_missing_role() {
    let schema = CompiledSchema::new();
    assert!(schema.get_role_scopes("ghost").is_empty());
}

// -------------------------------------------------------------------------
// Federation metadata
// -------------------------------------------------------------------------

#[test]
fn federation_metadata_none_when_no_federation() {
    assert!(CompiledSchema::new().federation_metadata().is_none());
}

#[test]
fn federation_metadata_none_when_disabled() {
    let mut schema = CompiledSchema::new();
    schema.federation = Some(FederationConfig {
        enabled: false,
        ..Default::default()
    });
    assert!(schema.federation_metadata().is_none());
}

#[test]
#[cfg(feature = "federation")]
fn federation_metadata_some_when_enabled() {
    let mut schema = CompiledSchema::new();
    schema.federation = Some(FederationConfig {
        enabled: true,
        version: Some("v2".to_string()),
        entities: vec![FederationEntity {
            name: "User".to_string(),
            key_fields: vec!["id".to_string()],
            ..Default::default()
        }],
        ..Default::default()
    });
    let meta = schema.federation_metadata();
    assert!(meta.is_some());
    let meta = meta.unwrap();
    assert!(meta.enabled);
    assert_eq!(meta.types.len(), 1);
    assert_eq!(meta.types[0].name, "User");
}

// -------------------------------------------------------------------------
// entity_sources (federation _entities backing-relation map)
// -------------------------------------------------------------------------

#[test]
#[cfg(feature = "federation")]
fn entity_sources_query_wins_then_type_level_fallback() {
    let mut schema = CompiledSchema::new();

    // Owned entity "User": its relation rides on the root query. The compiled
    // type carries an EMPTY type-level sql_source (what the converter emits for
    // an owned type).
    schema
        .queries
        .push(QueryDefinition::new("user", "User").with_sql_source("v_user"));
    schema.types.push(TypeDefinition::new("User", ""));

    // Owner-split `extend type` entity "Organization": no local query, so its
    // relation rides on the type-level sql_source (#507).
    schema.types.push(TypeDefinition::new("Organization", "v_organization"));

    // "Account" has BOTH a backing query and a type-level sql_source → the
    // query-sourced relation wins.
    schema
        .queries
        .push(QueryDefinition::new("account", "Account").with_sql_source("v_account_query"));
    schema.types.push(TypeDefinition::new("Account", "v_account_type"));

    let sources = schema.entity_sources();

    let user = sources.get("User").expect("owned entity sourced from its query");
    assert_eq!(user.relation, "v_user");
    assert_eq!(
        user.jsonb_column.as_deref(),
        Some("data"),
        "the query's jsonb_column defaults to data"
    );

    let org = sources
        .get("Organization")
        .expect("extends entity sourced from type-level sql_source");
    assert_eq!(org.relation, "v_organization");
    assert_eq!(
        org.jsonb_column.as_deref(),
        Some("data"),
        "an extends entity defaults to the standard jsonb `data` column"
    );

    let account = sources.get("Account").expect("Account is present");
    assert_eq!(
        account.relation, "v_account_query",
        "a query-sourced relation wins over the type-level fallback"
    );
}

// -------------------------------------------------------------------------
// content_hash
// -------------------------------------------------------------------------

#[test]
fn content_hash_is_32_hex_chars() {
    let hash = CompiledSchema::new().content_hash();
    assert_eq!(hash.len(), 32);
    assert!(hash.chars().all(|c| c.is_ascii_hexdigit()));
}

#[test]
fn content_hash_is_stable() {
    let schema = CompiledSchema::new();
    assert_eq!(schema.content_hash(), schema.content_hash());
}

#[test]
fn content_hash_differs_for_different_schemas() {
    let s1 = CompiledSchema::new();
    let mut s2 = CompiledSchema::new();
    s2.queries.push(QueryDefinition::new("users", "User"));
    assert_ne!(s1.content_hash(), s2.content_hash());
}

/// The producing build is part of what a schema *is*, so two artifacts that
/// differ only in which release wrote them do not share a hash — a cache keyed
/// on it cannot carry a previous build's entries into this one.
#[test]
fn content_hash_covers_the_producing_build() {
    let stamped = CompiledSchema::new();
    let mut unstamped = CompiledSchema::new();
    unstamped.fraiseql_version = ProducerVersion::unstamped();
    assert_ne!(stamped.content_hash(), unstamped.content_hash());
}

// -------------------------------------------------------------------------
// has_rls_configured
// -------------------------------------------------------------------------

#[test]
fn has_rls_configured_false_without_security() {
    assert!(!CompiledSchema::new().has_rls_configured());
}

#[test]
fn has_rls_configured_false_when_policies_empty() {
    let mut schema = CompiledSchema::new();
    let sec = SecurityConfig {
        policies: vec![],
        ..SecurityConfig::default()
    };
    schema.security = Some(sec);
    assert!(!schema.has_rls_configured());
}

#[test]
fn has_rls_configured_true_when_declared() {
    let mut schema = CompiledSchema::new();
    let mut sec = SecurityConfig::new();
    sec.rls.enabled = true;
    schema.security = Some(sec);
    assert!(schema.has_rls_configured());
}

/// A non-`none` tenancy mode is itself a multi-tenant declaration; requiring a
/// second boolean to say so is what left both dependent gates permanently off
/// (#758).
#[test]
fn tenancy_mode_implies_multi_tenant() {
    for mode in [
        crate::schema::TenancyMode::Row,
        crate::schema::TenancyMode::Schema,
    ] {
        let mut schema = CompiledSchema::new();
        let mut sec = SecurityConfig::new();
        sec.tenancy.mode = mode;
        schema.security = Some(sec);
        assert!(schema.is_multi_tenant(), "{mode} must imply multi-tenant");
    }

    let mut single = CompiledSchema::new();
    single.security = Some(SecurityConfig::new());
    assert!(!single.is_multi_tenant(), "mode = none without the flag is single-tenant");
}

// -------------------------------------------------------------------------
// raw_schema
// -------------------------------------------------------------------------

#[test]
fn raw_schema_returns_sdl_when_set() {
    let mut schema = CompiledSchema::new();
    schema.schema_sdl = Some("type Query { ping: String }".to_string());
    assert_eq!(schema.raw_schema(), "type Query { ping: String }");
}

#[test]
fn raw_schema_generates_from_types_when_sdl_absent() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("User"));
    let sdl = schema.raw_schema();
    assert!(sdl.contains("User"));
}

#[test]
fn raw_schema_includes_root_query_and_mutation() {
    // Root operations live in `queries`/`mutations`, NOT as `Query`/`Mutation`
    // object types in `types`. The generated SDL must still render them — otherwise
    // it advertises no root fields, and the federation `_service` SDL built from it
    // fails gateway composition with NO_QUERIES.
    use crate::schema::{ArgumentDefinition, FieldType};

    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("User"));

    let mut users = QueryDefinition::new("users", "User");
    users.returns_list = true;
    let mut user = QueryDefinition::new("user", "User");
    user.nullable = true;
    user.arguments = vec![ArgumentDefinition::new("id", FieldType::Id)];
    schema.queries.push(users);
    schema.queries.push(user);

    let mut create = MutationDefinition::new("createUser", "User");
    create.arguments = vec![ArgumentDefinition::new("name", FieldType::String)];
    schema.mutations.push(create);

    let sdl = schema.raw_schema();

    assert!(sdl.contains("type Query {"), "SDL must declare a root Query type:\n{sdl}");
    assert!(sdl.contains("users: [User!]!"), "list query rendered as a list type:\n{sdl}");
    assert!(sdl.contains("user(id: ID!): User"), "nullable single query with arg:\n{sdl}");
    assert!(sdl.contains("type Mutation {"), "SDL must declare a root Mutation type:\n{sdl}");
    assert!(sdl.contains("createUser(name: String!): User"), "mutation field:\n{sdl}");
}

#[test]
fn raw_schema_renders_auto_params_as_arguments() {
    // `auto_params` are runtime flags, never stored in `arguments`; the rendered SDL
    // must still surface them as real GraphQL arguments so a federated gateway (and a
    // generated client) can paginate/filter. Regression test for #499.
    use crate::schema::AutoParams;

    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("Event"));

    let mut events = QueryDefinition::new("events", "Event").returning_list();
    events.auto_params = AutoParams {
        has_where:    true,
        has_order_by: true,
        has_limit:    true,
        has_offset:   false,
    };
    schema.queries.push(events);

    let sdl = schema.raw_schema();

    assert!(
        sdl.contains("events(where: JSON, orderBy: JSON, limit: Int): [Event!]!"),
        "auto_params must render as arguments:\n{sdl}"
    );
    // `Event` carries no fields, so nothing is derivable and `where`/`orderBy`
    // fall back to the `JSON` scalar — which must then be declared so the SDL is
    // type-complete for gateway composition. See
    // `raw_schema_renders_the_derived_filter_types_as_input_blocks` for the
    // other half, where the schema *can* adjudicate.
    assert!(sdl.contains("scalar JSON"), "JSON scalar must be declared:\n{sdl}");
    // `has_offset` is false → no `offset` argument leaks in.
    assert!(!sdl.contains("offset"), "disabled auto_param must not appear:\n{sdl}");
    // A scalar that only the synthesized arg references must still be flagged as a
    // FieldType, not a list/non-null marker (sanity on the scalar walk).
    assert!(!sdl.contains("scalar Int"), "Int is a builtin, never declared:\n{sdl}");
}

#[test]
fn graphql_arguments_synthesizes_and_dedups_auto_params() {
    use crate::schema::{ArgumentDefinition, AutoParams, FieldType};

    // This schema declares no types, so nothing is derivable and the auto-wired
    // shapes fall back to `JSON` — the argument *names* are what this test is
    // about.
    let schema = CompiledSchema::default();

    // No auto_params → arguments unchanged.
    let plain = QueryDefinition::new("user", "User");
    assert!(plain.graphql_arguments(&schema).is_empty());

    // Explicit arg of the same name wins — no duplicate synthesized.
    let mut q = QueryDefinition::new("users", "User").returning_list();
    q.arguments = vec![ArgumentDefinition::optional("limit", FieldType::Int)];
    q.auto_params = AutoParams::all();
    let args = q.graphql_arguments(&schema);
    let names: Vec<&str> = args.iter().map(|a| a.name.as_str()).collect();
    assert_eq!(names, vec!["limit", "where", "orderBy", "offset"], "no duplicate `limit`");

    // Relay queries keep their explicit args verbatim (pagination handled elsewhere).
    let mut relay = QueryDefinition::new("posts", "Post").returning_list();
    relay.relay = true;
    relay.auto_params = AutoParams::all();
    assert!(relay.graphql_arguments(&schema).is_empty(), "relay queries are untouched");
}

#[cfg(feature = "federation")]
#[test]
fn service_sdl_advertises_root_query_fields() {
    // End-to-end: the `_service { sdl }` a gateway composes must expose the root
    // query fields, the entity `@key`, and the federation plumbing together.
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("User"));
    let mut users = QueryDefinition::new("users", "User");
    users.returns_list = true;
    schema.queries.push(users);
    schema.federation = Some(FederationConfig {
        enabled: true,
        version: Some("v2".to_string()),
        entities: vec![FederationEntity {
            name: "User".to_string(),
            key_fields: vec!["id".to_string()],
            ..Default::default()
        }],
        ..Default::default()
    });

    let meta = schema.federation_metadata().expect("federation enabled");
    let sdl = crate::federation::generate_service_sdl(&schema.raw_schema(), &meta);

    assert!(
        sdl.contains("users: [User!]!"),
        "root query must be in the _service SDL:\n{sdl}"
    );
    assert!(sdl.contains("@key(fields: \"id\")"), "entity key must be present:\n{sdl}");
    assert!(sdl.contains("_service"), "federation plumbing must be present:\n{sdl}");
}

#[test]
fn raw_schema_renders_full_type_closure() {
    // The generated SDL must be type-complete: input objects, enums, unions, and
    // non-built-in scalar declarations the root operations reference. Otherwise a
    // gateway composing the subgraph hits `Unknown type CreateQuoteInput` (and the
    // same for every custom scalar / enum / result union).
    use crate::schema::{
        ArgumentDefinition, EnumDefinition, EnumValueDefinition, FieldType, InputFieldDefinition,
        InputObjectDefinition, UnionDefinition,
    };

    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("QuoteSuccess"));
    schema.types.push(make_type_def("MutationError"));

    let mut status = EnumDefinition::new("QuoteStatus");
    status.values = vec![
        EnumValueDefinition::new("DRAFT"),
        EnumValueDefinition::new("SENT"),
    ];
    schema.enums.push(status);

    let mut input = InputObjectDefinition::new("CreateQuoteInput");
    input.fields = vec![
        InputFieldDefinition::new("amount", "Decimal"), // non-built-in scalar
        InputFieldDefinition::new("status", "QuoteStatus"), // enum
    ];
    schema.input_types.push(input);

    schema.unions.push(UnionDefinition {
        name:         "CreateQuoteResult".to_string(),
        member_types: vec!["QuoteSuccess".to_string(), "MutationError".to_string()],
        description:  None,
    });

    let mut create = MutationDefinition::new("createQuote", "CreateQuoteResult");
    create.arguments = vec![ArgumentDefinition::new(
        "input",
        FieldType::Input("CreateQuoteInput".to_string()),
    )];
    schema.mutations.push(create);

    let sdl = schema.raw_schema();

    assert!(sdl.contains("enum QuoteStatus {"), "enum must be declared:\n{sdl}");
    assert!(sdl.contains("DRAFT"), "enum values must be rendered:\n{sdl}");
    assert!(
        sdl.contains("input CreateQuoteInput {"),
        "input object must be declared:\n{sdl}"
    );
    assert!(
        sdl.contains("union CreateQuoteResult = QuoteSuccess | MutationError"),
        "result union must be declared:\n{sdl}"
    );
    assert!(sdl.contains("scalar Decimal"), "non-built-in scalar must be declared:\n{sdl}");
    assert!(
        sdl.contains("createQuote(input: CreateQuoteInput!): CreateQuoteResult"),
        "mutation signature references the now-declared input + result:\n{sdl}"
    );
}

#[test]
fn raw_schema_declares_scalars_under_the_name_fields_reference() {
    // The `scalar` declaration and the field that references it must agree on one
    // name. Fields render scalar type names verbatim (`datetime`, `FloatRange`); the
    // declaration walk must declare those same names, or composition reports
    // `Unknown type datetime`.
    use crate::schema::{InputFieldDefinition, InputObjectDefinition};

    let mut schema = CompiledSchema::new();
    let mut input = InputObjectDefinition::new("EventFilter");
    input.fields = vec![
        InputFieldDefinition::new("at", "datetime"), // lowercase scalar name
        InputFieldDefinition::new("span", "FloatRange"), // rich scalar not previously declared
    ];
    schema.input_types.push(input);

    let sdl = schema.raw_schema();

    assert!(sdl.contains("at: datetime"), "field renders the verbatim scalar name:\n{sdl}");
    assert!(
        sdl.contains("scalar datetime"),
        "the referenced scalar must be declared under the SAME name:\n{sdl}"
    );
    assert!(sdl.contains("span: FloatRange"), "field renders FloatRange:\n{sdl}");
    assert!(sdl.contains("scalar FloatRange"), "FloatRange must be declared:\n{sdl}");
}

// -------------------------------------------------------------------------
// is_builtin_type (private fn — tested via validate())
// -------------------------------------------------------------------------

// ── Operation name normalization (issue #199) ────────────────────────

#[test]
fn find_query_exact_match() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("User"));
    schema.queries.push(make_query("users", "User"));
    schema.build_indexes();
    assert!(schema.find_query("users").is_some());
}

#[test]
fn find_query_camel_to_snake_fallback() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("DnsServer"));
    schema.queries.push(make_query("dns_servers", "DnsServer"));
    schema.build_indexes();
    // Exact match works
    assert!(schema.find_query("dns_servers").is_some());
    // camelCase fallback also works
    assert!(schema.find_query("dnsServers").is_some());
}

#[test]
fn find_query_camel_to_snake_fallback_without_index() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("DnsServer"));
    schema.queries.push(make_query("dns_servers", "DnsServer"));
    // No build_indexes() — exercises the linear scan fallback path
    assert!(schema.find_query("dnsServers").is_some());
}

#[test]
fn find_mutation_camel_to_snake_fallback() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("Location"));
    schema.mutations.push(make_mutation("create_location", "Location"));
    schema.build_indexes();
    assert!(schema.find_mutation("createLocation").is_some());
}

#[test]
fn find_query_returns_none_for_unknown() {
    let mut schema = CompiledSchema::new();
    schema.types.push(make_type_def("User"));
    schema.queries.push(make_query("users", "User"));
    schema.build_indexes();
    assert!(schema.find_query("nonexistent").is_none());
}

// ── NamingConvention (issue #216) ───────────────────────────────────

#[test]
fn display_name_preserve_returns_unchanged() {
    let schema = CompiledSchema::new(); // default = Preserve
    assert_eq!(schema.display_name("create_dns_server"), "create_dns_server");
    assert_eq!(schema.display_name("dns_servers"), "dns_servers");
}

#[test]
fn display_name_camel_case_converts() {
    let mut schema = CompiledSchema::new();
    schema.naming_convention = NamingConvention::CamelCase;
    assert_eq!(schema.display_name("create_dns_server"), "createDnsServer");
    assert_eq!(schema.display_name("dns_servers"), "dnsServers");
    assert_eq!(schema.display_name("delete_outreach_sequence"), "deleteOutreachSequence");
}

#[test]
fn camel_case_index_lookup() {
    let mut schema = CompiledSchema::new();
    schema.naming_convention = NamingConvention::CamelCase;
    schema.types.push(make_type_def("DnsServer"));
    schema.queries.push(make_query("dns_servers", "DnsServer"));
    schema.mutations.push(make_mutation("create_dns_server", "DnsServer"));
    schema.build_indexes();

    // camelCase lookup via index
    assert!(schema.find_query("dnsServers").is_some());
    assert!(schema.find_mutation("createDnsServer").is_some());

    // Original snake_case still works
    assert!(schema.find_query("dns_servers").is_some());
    assert!(schema.find_mutation("create_dns_server").is_some());
}

#[test]
fn preserve_convention_no_camel_index_entry() {
    let mut schema = CompiledSchema::new(); // Preserve
    schema.types.push(make_type_def("DnsServer"));
    schema.queries.push(make_query("dns_servers", "DnsServer"));
    schema.build_indexes();

    // Only 1 index entry (the original name); camelCase only works via fallback
    assert_eq!(schema.query_index.len(), 1);
    assert!(schema.query_index.contains_key("dns_servers"));
}

#[test]
fn naming_convention_serde_in_compiled_schema() {
    let mut schema = CompiledSchema::new();
    schema.naming_convention = NamingConvention::CamelCase;
    let json = schema.to_json().unwrap();
    let restored = CompiledSchema::from_json(&json, false).unwrap();
    assert_eq!(restored.naming_convention, NamingConvention::CamelCase);
}

#[test]
fn schema_integrity_verification() {
    let schema = CompiledSchema::new();
    let body = schema.to_json().unwrap();

    // Embed the hash exactly as `fraiseql-cli compile` does — through the shared
    // routine, not a hand-copy of it. The previous version of this test hashed
    // the *uncanonicalized* value and agreed with the verifier only because
    // `serde_json::Map` happened to be a sorted `BTreeMap` in the build it ran
    // in, so it failed under `--workspace --all-features` and passed under
    // `-p fraiseql-core` (#899).
    let value: serde_json::Value = serde_json::from_str(&body).unwrap();
    let hash_hex = super::content_hash_of(&value);

    // Build wrapped JSON with _content_hash as first field
    let obj = value.as_object().unwrap();
    let mut new_obj = serde_json::Map::new();
    new_obj.insert("_content_hash".to_string(), serde_json::Value::String(hash_hex));
    for (k, v) in obj {
        new_obj.insert(k.clone(), v.clone());
    }
    let wrapped_json = serde_json::to_string_pretty(&serde_json::Value::Object(new_obj)).unwrap();

    // from_json with strict=true should accept
    let restored = CompiledSchema::from_json(&wrapped_json, true).unwrap();
    assert_eq!(restored.types.len(), schema.types.len());

    // Test mismatch: change hash in Value
    let mut mismatch_value: serde_json::Value = serde_json::from_str(&wrapped_json).unwrap();
    let mismatch_obj = mismatch_value.as_object_mut().unwrap();
    mismatch_obj.insert(
        "_content_hash".to_string(),
        serde_json::Value::String("0000000000000000".to_string()),
    );
    let mismatch_json = serde_json::to_string_pretty(&mismatch_value).unwrap();
    let result = CompiledSchema::from_json(&mismatch_json, true);
    assert!(result.is_err(), "Expected validation error for hash mismatch");

    // Test missing hash with strict
    let result = CompiledSchema::from_json(&body, true);
    assert!(result.is_err(), "Expected error for missing hash in strict mode");

    // Test missing hash with non-strict
    let restored = CompiledSchema::from_json(&body, false).unwrap();
    assert_eq!(restored.types.len(), schema.types.len());
}

/// The hash must be a property of the schema, not of the build that computed it.
///
/// `serde_json::Map` is insertion-ordered here, so a digest taken over a
/// serialized `Value` is only stable if the value is canonicalized first. This
/// drives the same content through two different key orderings and requires one
/// answer — the property #899 says the integrity check must have.
#[test]
fn content_hash_is_independent_of_key_order() {
    let a: serde_json::Value =
        serde_json::from_str(r#"{"b": 1, "a": {"y": [2, {"q": 3, "p": 4}], "x": 5}}"#).unwrap();
    let b: serde_json::Value =
        serde_json::from_str(r#"{"a": {"x": 5, "y": [2, {"p": 4, "q": 3}]}, "b": 1}"#).unwrap();
    assert_eq!(super::content_hash_of(&a), super::content_hash_of(&b));

    // …and still distinguishes different content.
    let c: serde_json::Value = serde_json::from_str(r#"{"a": 1, "b": 2}"#).unwrap();
    assert_ne!(super::content_hash_of(&a), super::content_hash_of(&c));
}

/// Array order is content, not formatting: reordering `types` must change the hash.
#[test]
fn content_hash_is_sensitive_to_array_order() {
    let a: serde_json::Value = serde_json::from_str(r#"{"types": [{"n": 1}, {"n": 2}]}"#).unwrap();
    let b: serde_json::Value = serde_json::from_str(r#"{"types": [{"n": 2}, {"n": 1}]}"#).unwrap();
    assert_ne!(super::content_hash_of(&a), super::content_hash_of(&b));
}

// -------------------------------------------------------------------------
// `TypeDefinition.internal` serde (#665, Phase 01)
// -------------------------------------------------------------------------

/// The `internal` flag must survive a compiled-schema serialize→deserialize when set —
/// the server's schema loader and cli commands re-load compiled files, and Phase 02's
/// runtime guard reads `internal` off a re-loaded schema (its first runtime consumer).
/// A `#[serde(skip)]` flag would evaporate at that boundary; this pins that it does not.
#[test]
fn internal_flag_round_trips_when_true() {
    let mut ty = make_type_def("EntityChangeLog");
    ty.internal = true;

    let json = serde_json::to_string(&ty).unwrap();
    assert!(json.contains("\"internal\":true"), "internal=true must serialize: {json}");

    let back: TypeDefinition = serde_json::from_str(&json).unwrap();
    assert!(back.internal, "internal must survive round-trip");
    assert_eq!(ty, back);
}

/// Authored types never set `internal`; `skip_serializing_if` omits the key entirely so
/// existing compiled schemas are byte-identical, and the serde default restores `false`
/// on load. This is the honest proof the field is inert for every non-framework type.
#[test]
fn internal_flag_omitted_when_false() {
    let ty = make_type_def("User");
    assert!(!ty.internal, "authored types default to internal = false");

    let json = serde_json::to_string(&ty).unwrap();
    assert!(!json.contains("internal"), "internal=false must be omitted from JSON: {json}");

    let back: TypeDefinition = serde_json::from_str(&json).unwrap();
    assert!(!back.internal, "omitted key deserializes back to false");
    assert_eq!(ty, back);
}

// =============================================================================
// The derived filter/sort surface (§ 5.8.2, #1154)
// =============================================================================

/// A schema whose `where`/`orderBy` surface is derivable: one entity with
/// fields, one query that filters and sorts it.
fn schema_with_a_filterable_query() -> CompiledSchema {
    use crate::schema::{AutoParams, FieldDefinition, FieldType};

    let mut schema = CompiledSchema::default();
    let mut order = TypeDefinition::new("Order", "v_order");
    order.fields.push(FieldDefinition::new("reference", FieldType::String));
    order.fields.push(FieldDefinition::new("total", FieldType::Int));
    schema.types.push(order);

    let mut orders = QueryDefinition::new("orders", "Order");
    orders.returns_list = true;
    orders.auto_params = AutoParams::all();
    schema.queries.push(orders);
    schema
}

#[test]
fn build_indexes_materialises_the_derived_filter_surface() {
    let mut schema = schema_with_a_filterable_query();
    assert!(schema.find_input_type("OrderWhereInput").is_none(), "not yet derived");

    schema.build_indexes();

    assert!(
        schema.find_input_type("OrderWhereInput").is_some(),
        "`where` needs a type to name"
    );
    assert!(schema.find_input_type("OrderOrderByInput").is_some(), "`orderBy` needs one too");
    assert!(schema.find_input_type("StringFilter").is_some(), "and so does each leaf");
    assert!(schema.find_enum("SortDirection").is_some(), "the sort surface names an enum");
}

/// `build_indexes` is the documented "rebuild derived state" hook and is called
/// again after any direct mutation, so deriving twice must not double the
/// surface.
#[test]
fn build_indexes_is_idempotent() {
    let mut schema = schema_with_a_filterable_query();
    schema.build_indexes();
    let after_first = schema.input_types.len();
    let enums_after_first = schema.enums.len();

    schema.build_indexes();

    assert_eq!(schema.input_types.len(), after_first, "a second pass duplicated input types");
    assert_eq!(schema.enums.len(), enums_after_first, "a second pass duplicated enums");
}

#[test]
fn graphql_arguments_types_where_and_order_by_against_the_derived_surface() {
    use crate::schema::FieldType;

    let mut schema = schema_with_a_filterable_query();
    schema.build_indexes();
    let args = schema.queries[0].graphql_arguments(&schema);

    let arg = |name: &str| args.iter().find(|a| a.name == name).expect("argument present");
    assert_eq!(arg("where").arg_type, FieldType::Input("OrderWhereInput".to_string()));
    assert_eq!(
        arg("orderBy").arg_type,
        FieldType::List(Box::new(FieldType::Input("OrderOrderByInput".to_string()))),
        "the engine parses `[{{field, direction}}]`, so the published type is a list"
    );
    // Untouched: these were never dynamic shapes.
    assert_eq!(arg("limit").arg_type, FieldType::Int);
    assert_eq!(arg("offset").arg_type, FieldType::Int);
}

/// The rule that makes the surface impossible to dangle: an argument is typed
/// **iff** the schema carries the type it would name. A return type the schema
/// cannot adjudicate derives nothing, so the argument stays `JSON` rather than
/// naming a type nothing defines.
#[test]
fn an_auto_wired_argument_stays_json_when_no_type_backs_the_name() {
    use crate::schema::{AutoParams, FieldType};

    let mut schema = CompiledSchema::default();
    let mut ghosts = QueryDefinition::new("ghosts", "Ghost");
    ghosts.returns_list = true;
    ghosts.auto_params = AutoParams::all();
    schema.queries.push(ghosts);
    schema.build_indexes();

    let args = schema.queries[0].graphql_arguments(&schema);
    let arg = |name: &str| args.iter().find(|a| a.name == name).expect("argument present");
    assert_eq!(arg("where").arg_type, FieldType::Json, "no `GhostWhereInput` exists to name");
    assert_eq!(arg("orderBy").arg_type, FieldType::Json);
}

/// The invariant stated once and checked over the whole schema: every named
/// input type any query publishes is a type the schema defines.
#[test]
fn every_published_argument_type_resolves_against_the_schema() {
    let mut schema = schema_with_a_filterable_query();
    schema.build_indexes();

    let mut checked = 0_usize;
    for query in &schema.queries {
        for arg in query.graphql_arguments(&schema) {
            let mut ty = &arg.arg_type;
            while let crate::schema::FieldType::List(inner) = ty {
                ty = inner;
            }
            if let crate::schema::FieldType::Input(name) = ty {
                assert!(
                    schema.find_input_type(name).is_some(),
                    "query `{}` publishes `{name}`, which the schema does not define",
                    query.name
                );
                checked += 1;
            }
        }
    }
    assert!(checked > 0, "the walk found no input-typed arguments, so it proved nothing");
}

/// Every input field of every derived type must name something the schema
/// defines — a scalar, an enum, or another input type. A dangling reference
/// here is what turns a published filter surface into a fiction.
#[test]
fn every_derived_input_field_names_a_type_the_schema_carries() {
    let mut schema = schema_with_a_filterable_query();
    schema.build_indexes();

    let published = crate::schema::published_scalar_names();
    for input_type in &schema.input_types {
        for field in &input_type.fields {
            let leaf = field.field_type.trim_start_matches('[').trim_end_matches(['!', ']']);
            assert!(
                published.iter().any(|s| s == leaf)
                    || schema.find_enum(leaf).is_some()
                    || schema.find_input_type(leaf).is_some(),
                "`{}.{}` names `{leaf}`, which the schema does not define",
                input_type.name,
                field.name
            );
        }
    }
}

/// The other half of `raw_schema_renders_auto_params_as_arguments`: when the
/// schema *can* adjudicate the return type, the federation SDL declares the
/// derived types as `input` blocks and references them from the arguments.
///
/// `referenced_scalars` excludes anything in `input_types`, so getting this
/// wrong shows up as `scalar OrderWhereInput` — a gateway would compose a
/// subgraph whose filter argument is an opaque scalar.
#[test]
fn raw_schema_renders_the_derived_filter_types_as_input_blocks() {
    let mut schema = schema_with_a_filterable_query();
    schema.build_indexes();

    let sdl = schema.raw_schema();

    assert!(
        sdl.contains("orders(where: OrderWhereInput, orderBy: [OrderOrderByInput], "),
        "the arguments must reference the derived types:\n{sdl}"
    );
    assert!(
        sdl.contains("input OrderWhereInput {"),
        "the filter type must be declared:\n{sdl}"
    );
    assert!(sdl.contains("input OrderOrderByInput {"), "the sort item too:\n{sdl}");
    assert!(sdl.contains("input StringFilter {"), "and each leaf filter:\n{sdl}");
    assert!(sdl.contains("enum SortDirection {"), "and the direction enum:\n{sdl}");
    assert!(
        !sdl.contains("scalar OrderWhereInput"),
        "a derived input type must never be declared as a bare scalar:\n{sdl}"
    );
}

/// § 5.8.2 end-to-end over the real load path: the name a client writes is the
/// name the schema now carries, so a document that was refused before this
/// release validates, and a typo is still refused with a hint.
#[test]
fn the_derived_filter_types_are_acceptable_variable_types() {
    use crate::{graphql::parse_query, runtime::validate_variable_types};

    let mut schema = schema_with_a_filterable_query();
    schema.build_indexes();

    let var_defs = |source: &str| parse_query(source).expect("query parses").variables;

    validate_variable_types(
        &schema,
        Some("Q"),
        &var_defs("query Q($w: OrderWhereInput) { orders(where: $w) { reference } }"),
    )
    .expect("the schema publishes `OrderWhereInput`, so a client may write it");

    validate_variable_types(
        &schema,
        Some("Q"),
        &var_defs("query Q($o: [OrderOrderByInput!]) { orders(orderBy: $o) { reference } }"),
    )
    .expect("and `OrderOrderByInput`, list-wrapped");

    validate_variable_types(
        &schema,
        Some("Q"),
        &var_defs("query Q($d: SortDirection) { orders(orderBy: [{field: \"x\", direction: $d}]) { reference } }"),
    )
    .expect("and the direction enum");

    let err = validate_variable_types(
        &schema,
        Some("Q"),
        &var_defs("query Q($w: OrdrWhereInput) { orders(where: $w) { reference } }"),
    )
    .expect_err("a name the schema does not carry is still refused");
    let message = err.to_string();
    assert!(message.contains("OrdrWhereInput"), "message was: {message}");
    assert!(message.contains("Did you mean 'OrderWhereInput'?"), "message was: {message}");
}

// =============================================================================
// #1262 — a subscription filter that names no declared argument
// =============================================================================

/// The golden fixtures live outside this crate; resolve them the way the integration
/// suite does.
fn golden_fixtures_dir() -> std::path::PathBuf {
    std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
        .parent()
        .unwrap()
        .parent()
        .unwrap()
        .join("tests/fixtures/golden")
}

/// Every golden fixture must be a document `fraiseql compile` could actually emit.
///
/// `02-enum-input-subscription.json` filtered on `status` while declaring only `userId`,
/// and nothing noticed for as long as the fixture existed — the same dangling reference
/// #1262 is about, sitting in the repo's own canonical artifacts. A fixture that no
/// compiler run can produce tests a shape no deployment has.
///
/// Lives in the lib rather than beside the other golden tests because
/// `tests/golden_fixtures_test.rs` runs only in the `Dagger — integration` leg
/// (`--test '*'`), while `--lib` runs in `Dagger — test`, which is a required check.
///
/// Discovered from the directory rather than from a list, so a fixture added later is
/// covered without anyone remembering to add it here.
#[test]
fn no_golden_fixture_filters_on_an_argument_it_does_not_declare() {
    let mut checked = 0_usize;
    for entry in std::fs::read_dir(golden_fixtures_dir()).expect("fixtures dir readable") {
        let path = entry.expect("readable dir entry").path();
        if path.extension().and_then(|e| e.to_str()) != Some("json") {
            continue;
        }
        let name = path.file_name().unwrap().to_str().unwrap().to_string();
        let json = std::fs::read_to_string(&path).unwrap();
        // Deserialized structurally rather than through `from_json`, which now *refuses*
        // a dangling filter: routed through it, an offending fixture would fail to load,
        // be skipped as "not a CompiledSchema", and the per-fixture assertion below could
        // never fire. Only the count at the end would notice, and it could not say which
        // fixture or why.
        //
        // Not every fixture here is a whole CompiledSchema; the ones that are not cannot
        // carry a subscription filter either.
        let Ok(schema) = serde_json::from_str::<CompiledSchema>(&json) else {
            continue;
        };
        checked += 1;

        for sub in &schema.subscriptions {
            assert!(
                sub.filter_violation().is_none(),
                "{name}: {}",
                sub.filter_violation().unwrap_or_default()
            );
        }
    }
    assert!(
        checked >= 9,
        "expected the golden fixtures to be found and parsed, got {checked}"
    );
}

/// A compiled schema whose subscription filter names an undeclared argument is refused
/// at **load**, not carried into the process (#1262).
///
/// `fraiseql compile` refuses to emit this document, so reaching it means the artifact
/// was edited by hand — the one path a compile-time check cannot cover. The refusal
/// belongs on the load path every entry point shares (`from_json`), beside the role and
/// inject-param refusals that already live there: a schema whose filter cannot be
/// applied is one whose subscriptions would deliver every event on their topic.
#[test]
fn a_compiled_schema_with_a_dangling_subscription_filter_is_refused_at_load() {
    let json = r#"{
      "types": [{"name": "Order", "sql_source": "v_order",
                 "fields": [{"name": "id", "field_type": "ID", "nullable": false}]}],
      "queries": [], "mutations": [],
      "subscriptions": [{
        "name": "orderUpdated", "return_type": "Order", "topic": "order_events",
        "arguments": [{"name": "orderId", "arg_type": "ID", "nullable": true}],
        "filter": {"argument_paths": {"no_such_argument": "/id"}}
      }]
    }"#;

    let err = CompiledSchema::from_json(json, false)
        .expect_err("a filter that cannot be applied must not load");
    let message = err.to_string();
    assert!(
        message.contains("no_such_argument"),
        "the refusal names the reference: {message}"
    );
    assert!(message.contains("orderId"), "the refusal names what is declared: {message}");
}

/// The positive control for the load-time refusal: the same document with the reference
/// spelled the way the subscription declares it loads, and keeps its filter.
#[test]
fn a_compiled_schema_with_a_resolvable_subscription_filter_loads() {
    let json = r#"{
      "types": [{"name": "Order", "sql_source": "v_order",
                 "fields": [{"name": "id", "field_type": "ID", "nullable": false}]}],
      "queries": [], "mutations": [],
      "subscriptions": [{
        "name": "orderUpdated", "return_type": "Order", "topic": "order_events",
        "arguments": [{"name": "orderId", "arg_type": "ID", "nullable": true}],
        "filter": {"argument_paths": {"orderId": "/id"}}
      }]
    }"#;

    let schema = CompiledSchema::from_json(json, false).expect("a resolvable filter must load");
    let filter = schema.subscriptions[0].filter.as_ref().expect("filter survives the load");
    assert_eq!(filter.argument_paths.get("orderId"), Some(&"/id".to_string()));
}

// ---------------------------------------------------------------------------
// Relationships: the load-time half of #1266.
//
// `fraiseql compile` refuses to emit any of these documents — the same
// `relationship_violations` call runs in the converter's `validate` — so reaching
// one means the artifact was hand-edited, which is the case a compile-time check
// cannot cover. Every branch below is asserted by the message it produces, not by
// "load failed", so a check that started refusing everything would still be caught.
// ---------------------------------------------------------------------------

/// The document every case below mutates: `Author.posts` is a followable `OneToMany`.
fn relationship_fixture() -> serde_json::Value {
    serde_json::json!({
      "types": [
        {"name": "Author", "sql_source": "v_author",
         "fields": [{"name": "id", "field_type": "ID", "nullable": false},
                    {"name": "name", "field_type": "String", "nullable": false}],
         "relationships": [{"name": "posts", "target_type": "Post",
                            "cardinality": "OneToMany",
                            "foreign_key": "fk_author", "referenced_key": "id"}]},
        {"name": "Post", "sql_source": "v_post",
         "fields": [{"name": "id", "field_type": "ID", "nullable": false},
                    {"name": "fk_author", "field_type": "Int", "nullable": false}]}
      ],
      "queries": [
        {"name": "authors", "return_type": "Author", "returns_list": true, "sql_source": "v_author"},
        {"name": "posts", "return_type": "Post", "returns_list": true, "sql_source": "v_post"}
      ],
      "mutations": []
    })
}

/// Load `fixture` after `mutate` has been applied, expecting a refusal, and return the
/// message so each case can assert on the part that is its own.
fn refusal_for(mutate: impl FnOnce(&mut serde_json::Value)) -> String {
    let mut doc = relationship_fixture();
    mutate(&mut doc);
    let err = CompiledSchema::from_json(&doc.to_string(), false)
        .expect_err("an unfollowable relationship must not load");
    err.to_string()
}

/// The positive control. Without it every case below would pass against a check that
/// refused every schema carrying any relationship at all.
#[test]
fn a_followable_relationship_loads_and_survives() {
    let schema = CompiledSchema::from_json(&relationship_fixture().to_string(), false)
        .expect("a followable relationship must load");
    let rels = &schema.find_type("Author").expect("Author loads").relationships;
    assert_eq!(rels.len(), 1, "the relationship survives the load");
    assert_eq!(rels[0].target_type, "Post");
}

/// Asserted on this branch's own wording, not merely on "the load failed". An undeclared
/// target also has no list query, so the downstream check fires on the same document — a
/// test asserting only that `'Nope'` appears somewhere passes with this branch deleted,
/// which is what the mutation loop caught it doing.
#[test]
fn a_relationship_targeting_an_undeclared_type_is_refused_at_load() {
    let message = refusal_for(|d| d["types"][0]["relationships"][0]["target_type"] = "Nope".into());
    assert!(
        message.contains("targets type 'Nope', which the schema does not declare"),
        "the refusal diagnoses the missing type rather than a downstream symptom: {message}"
    );
    assert!(message.contains("Author.posts"), "and names the relationship: {message}");
}

/// The parent side of #1230, one layer earlier: `OneToMany` reads `referenced_key` off
/// the declaring type's already-projected row, so a column no field publishes is a key
/// the embed can never find.
#[test]
fn a_relationship_whose_parent_join_column_names_no_field_is_refused_at_load() {
    let message =
        refusal_for(|d| d["types"][0]["relationships"][0]["referenced_key"] = "no_such".into());
    assert!(message.contains("'no_such'"), "the refusal names the column: {message}");
    assert!(message.contains("of 'Author'"), "and the side it is missing from: {message}");
}

#[test]
fn a_relationship_whose_target_join_column_names_no_field_is_refused_at_load() {
    let message =
        refusal_for(|d| d["types"][0]["relationships"][0]["foreign_key"] = "no_such".into());
    assert!(message.contains("'no_such'"), "the refusal names the column: {message}");
    assert!(message.contains("'Post'"), "and the side it is missing from: {message}");
}

/// Which side each key is read from swaps with the cardinality, so flipping it alone —
/// touching no column name — must move the verdict. A check that resolved the columns
/// its own way rather than through `Relationship::parent_join_column` would pass this
/// document, and the executor would then read `fk_author` off an `Author` row.
#[test]
fn flipping_the_cardinality_alone_makes_the_same_columns_unfollowable() {
    let message =
        refusal_for(|d| d["types"][0]["relationships"][0]["cardinality"] = "ManyToOne".into());
    assert!(
        message.contains("joins on 'fk_author' of 'Author'"),
        "ManyToOne reads the foreign key off the declaring type: {message}"
    );
}

/// `foreign_key` and `referenced_key` carry `#[serde(default)]` on the compiled struct,
/// so an omitted one is an empty string rather than a parse error.
#[test]
fn a_relationship_with_an_empty_join_column_is_refused_at_load() {
    let message = refusal_for(|d| {
        d["types"][0]["relationships"][0]
            .as_object_mut()
            .expect("relationship is an object")
            .remove("foreign_key");
    });
    assert!(message.contains("`foreign_key`"), "the refusal names the key: {message}");
}

#[test]
fn a_type_declaring_one_relationship_name_twice_is_refused_at_load() {
    let message = refusal_for(|d| {
        let dup = d["types"][0]["relationships"][0].clone();
        d["types"][0]["relationships"].as_array_mut().expect("array").push(dup);
    });
    assert!(message.contains("more than once"), "the refusal names the shape: {message}");
}

/// `embed_into_single` sources its rows from the target's list query and sets the empty
/// default when there is none — before touching the database, so no amount of data makes
/// it work.
#[test]
fn a_relationship_whose_target_has_no_list_query_is_refused_at_load() {
    let message = refusal_for(|d| d["queries"][1]["returns_list"] = false.into());
    assert!(message.contains("no list query"), "the refusal names the cause: {message}");
}

/// The column→field lookup snake-cases both sides, so a schema whose fields are published
/// in `camelCase` — the default naming convention — resolves the SQL column the executor's
/// `declared_key` would resolve, and loads.
#[test]
fn a_join_column_resolves_against_a_camel_case_field_name() {
    let mut doc = relationship_fixture();
    doc["types"][0]["relationships"][0]["cardinality"] = "ManyToOne".into();
    doc["types"][0]["relationships"][0]["target_type"] = "Post".into();
    doc["types"][0]["fields"].as_array_mut().expect("array").push(serde_json::json!(
        {"name": "fkAuthor", "field_type": "Int", "nullable": false}));
    CompiledSchema::from_json(&doc.to_string(), false)
        .expect("`fk_author` is published as `fkAuthor`, which is the spelling the executor reads");
}

// ── A to-one embed joins on a key that identifies one row ────────────────────
//
// `ManyToOne`/`OneToOne` answer with one object, and the REST executor produces it by
// taking the first row of a `LIMIT max_page_size` read. These five fix where that is
// decided: in the compiled schema, so the runtime never has a second row to mishandle.
//
// Every case mutates one document, so what separates a refusal from a load is the single
// thing each test names -- and the two that must *load* are what stops the check from
// being "refuse every to-one".

/// `Post.author` targets `Author` through `referenced_key`, which is the key the check
/// asks about. `key` is the column on `Author` it joins to.
fn to_one_fixture(key: &str) -> serde_json::Value {
    let mut doc = relationship_fixture();
    doc["types"][1]["relationships"] = serde_json::json!([{
        "name": "author", "target_type": "Author", "cardinality": "ManyToOne",
        "foreign_key": "fk_author", "referenced_key": key
    }]);
    doc
}

/// The defect, stated at the point it becomes decidable: `Author.name` is an ordinary
/// column, so "the" author of a post is whichever row the plan happened to return first.
#[test]
fn a_to_one_joining_on_a_non_unique_key_is_refused_at_load() {
    let err = CompiledSchema::from_json(&to_one_fixture("name").to_string(), false)
        .expect_err("a to-one on a non-unique key must not load");
    let message = err.to_string();
    assert!(
        message.contains("not declared unique"),
        "the refusal names the missing fact rather than a downstream symptom: {message}"
    );
    assert!(message.contains("Post.author"), "and names the relationship: {message}");
    assert!(
        message.contains("unique = true"),
        "and names the remedy, which is the whole reason this is a refusal and not a \
         silent first-row: {message}"
    );
}

/// The first of the two positive controls. Without it the check above passes just as
/// happily when it refuses every to-one relationship ever declared.
#[test]
fn declaring_the_target_key_unique_makes_the_same_to_one_followable() {
    let mut doc = to_one_fixture("name");
    doc["types"][0]["fields"][1]["unique"] = true.into();
    CompiledSchema::from_json(&doc.to_string(), false)
        .expect("the author's `unique` declaration is the fact the check was missing");
}

/// The second. The entity identity carries no `unique` flag and never will -- ADR-0017
/// makes `id: ID!` the one global identity, which Relay `Node`, federation `@key` and
/// cache normalization already read as unique. Requiring a declaration here would refuse
/// every conventional schema in existence, since `id` is *the* `referenced_key` for a
/// to-one.
#[test]
fn the_entity_identity_needs_no_uniqueness_declaration() {
    CompiledSchema::from_json(&to_one_fixture("id").to_string(), false)
        .expect("`id: ID` is unique under ADR-0017 without being declared so");
}

/// And the line under that exemption: it reads the identity *contract*, not the spelling.
/// `examples/basic` authors `id: Int`, which ADR-0017 explicitly declines to canonicalize
/// to `ID` because exposing a serial pk as a global id is the thing the Trinity avoids --
/// so an `Int` `id` is a column named like an identity that is not one, and gets no
/// exemption. This is the test that keeps [`FieldDefinition::is_unique_key`] a reading of
/// the contract rather than an inference from a name.
#[test]
fn a_field_merely_named_id_is_not_taken_for_the_entity_identity() {
    let mut doc = to_one_fixture("id");
    doc["types"][0]["fields"][0]["field_type"] = "Int".into();
    let err = CompiledSchema::from_json(&doc.to_string(), false)
        .expect_err("an Int `id` is not the ADR-0017 identity");
    assert!(
        err.to_string().contains("not declared unique"),
        "the name alone buys no exemption: {err}"
    );
}

/// The discriminator for the scope of the whole check. A collection is exactly what a
/// non-unique key yields, so `OneToMany` must be untouched by it -- and this is the same
/// column, on the same document, that the first test refuses.
#[test]
fn a_to_many_on_the_same_non_unique_key_still_loads() {
    let mut doc = to_one_fixture("name");
    doc["types"][1]["relationships"][0]["cardinality"] = "OneToMany".into();
    // `OneToMany` swaps which side each key is read from: the declaring type `Post` is
    // now read on `referenced_key` and the target `Author` filtered on `foreign_key`,
    // so the columns have to swap with it for the document to be followable at all.
    doc["types"][1]["relationships"][0]["referenced_key"] = "fk_author".into();
    doc["types"][1]["relationships"][0]["foreign_key"] = "name".into();
    CompiledSchema::from_json(&doc.to_string(), false)
        .expect("a to-many on a non-unique key is a collection, which is the point of one");
}

// ── #1303 the compiled pagination order, on the wire ─────────────────────────

/// Both variants survive a round-trip, and the JSON spelling is the one a reader
/// of a compiled schema sees.
///
/// The spelling is asserted rather than only the round-trip because a round-trip
/// passes for any pair of matching serializer and deserializer, including one
/// that emits a shape no author could read.
#[test]
fn pagination_order_round_trips_with_a_legible_spelling() {
    use super::query::PaginationOrder;

    let json_identity = serde_json::to_value(PaginationOrder::JsonIdentity).unwrap();
    assert_eq!(json_identity, serde_json::json!("json_identity"));

    let column = serde_json::to_value(PaginationOrder::Column("pk_user".into())).unwrap();
    assert_eq!(column, serde_json::json!({ "column": "pk_user" }));

    for order in [
        PaginationOrder::JsonIdentity,
        PaginationOrder::Column("pk_user".into()),
    ] {
        let mut q = QueryDefinition::new("users", "User").returning_list();
        q.pagination_order = Some(order.clone());
        let back: QueryDefinition =
            serde_json::from_str(&serde_json::to_string(&q).unwrap()).unwrap();
        assert_eq!(back.pagination_order, Some(order));
    }
}

/// A query with no pagination order emits no key, and one with no key parses back
/// to no order — the two halves of the sibling contract `relay_cursor_column`
/// already has.
#[test]
fn a_query_with_no_pagination_order_emits_no_key() {
    let q = QueryDefinition::new("user", "User");
    let value = serde_json::to_value(&q).unwrap();
    assert!(
        value.get("pagination_order").is_none(),
        "an absent order must not be spelled out: {value}"
    );

    let back: QueryDefinition = serde_json::from_value(value).unwrap();
    assert_eq!(back.pagination_order, None);
}

// ---------------------------------------------------------------------------
// The load-time half of #596's subscription policy, and the duplicate-name
// checks that came with it (#1265).
//
// These lived in `CompiledSchema::validate()`, which had no caller outside tests:
// every entry point that loads a compiled schema calls `from_json` and stops.
// So #596's "load-time error, not a silent deliver-all at subscribe time" never
// ran anywhere, and `SubscriptionPolicy::validate` had exactly one caller — the
// unreachable one.
//
// They now run in `finish_load`, the chokepoint both `from_json` branches share.
// Every case asserts the message it produces, not merely "load failed", so a check
// that started refusing everything would still be caught.
// ---------------------------------------------------------------------------

/// The document each case below mutates: one type carrying a well-formed policy.
fn policy_fixture() -> serde_json::Value {
    serde_json::json!({
      "types": [
        {"name": "Order", "sql_source": "v_order",
         "fields": [{"name": "id", "field_type": "Int", "nullable": false},
                    {"name": "owner_id", "field_type": "String", "nullable": false}],
         "subscription_policy": {"owner_path": "$.owner_id",
                                 "identity_field": "user_id",
                                 "bypass_roles": ["admin"]}}
      ],
      "queries": [
        {"name": "orders", "return_type": "Order", "returns_list": true, "sql_source": "v_order"}
      ],
      "mutations": []
    })
}

fn policy_refusal_for(mutate: impl FnOnce(&mut serde_json::Value)) -> String {
    let mut doc = policy_fixture();
    mutate(&mut doc);
    CompiledSchema::from_json(&doc.to_string(), false)
        .expect_err("a malformed subscription policy must not load")
        .to_string()
}

/// The positive control. Without it every case below would pass against a check that
/// refused every schema carrying a `subscription_policy` at all.
#[test]
fn a_well_formed_subscription_policy_loads_and_survives() {
    let schema = CompiledSchema::from_json(&policy_fixture().to_string(), false)
        .expect("a well-formed policy must load");
    let policy = schema
        .find_type("Order")
        .expect("Order loads")
        .subscription_policy
        .as_ref()
        .expect("the policy survives the load");
    assert_eq!(policy.owner_field(), "owner_id");
    assert_eq!(policy.identity_field, "user_id");
}

/// The case #596's comment names. A nested path yields an RLS condition whose pointer
/// lookup misses, and the delivery loop returns `false` on a miss — so the subscription
/// goes silently *quiet* rather than leaking. Diagnosed at 3am instead of at boot,
/// which is what the refusal exists to prevent.
#[test]
fn a_nested_subscription_policy_owner_path_is_refused_at_load() {
    let message =
        policy_refusal_for(|d| d["types"][0]["subscription_policy"]["owner_path"] = "$.a.b".into());
    assert!(
        message.contains("must be a single-level"),
        "the refusal diagnoses the nested path: {message}"
    );
    assert!(message.contains("Order"), "and names the declaring type: {message}");
}

#[test]
fn an_empty_subscription_policy_identity_field_is_refused_at_load() {
    let message = policy_refusal_for(|d| {
        d["types"][0]["subscription_policy"]["identity_field"] = "".into();
    });
    assert!(
        message.contains("identity_field must not be empty"),
        "the refusal diagnoses the empty identity field: {message}"
    );
}

#[test]
fn a_subscription_policy_owner_path_naming_no_field_is_refused_at_load() {
    let message =
        policy_refusal_for(|d| d["types"][0]["subscription_policy"]["owner_path"] = "$.".into());
    assert!(
        message.contains("must name a field"),
        "the refusal diagnoses the empty owner field: {message}"
    );
}

/// A duplicate name is not a style problem: `build_indexes` keys by name, so one
/// definition silently shadows the other and the schema behaves as something nobody
/// wrote. `fraiseql compile` refuses to emit one; this is the hand-edited artifact.
#[test]
fn a_duplicate_type_name_is_refused_at_load() {
    let message = policy_refusal_for(|d| {
        let dup = d["types"][0].clone();
        d["types"].as_array_mut().expect("types is an array").push(dup);
    });
    assert!(message.contains("Duplicate type name: Order"), "{message}");
}

#[test]
fn a_duplicate_query_name_is_refused_at_load() {
    let message = policy_refusal_for(|d| {
        let dup = d["queries"][0].clone();
        d["queries"].as_array_mut().expect("queries is an array").push(dup);
    });
    assert!(message.contains("Duplicate query name: orders"), "{message}");
}

#[test]
fn a_duplicate_mutation_name_is_refused_at_load() {
    let message = policy_refusal_for(|d| {
        let m = serde_json::json!({"name": "createOrder", "return_type": "Order",
                                   "operation": {"Insert": {"table": "tb_order"}}});
        let arr = d["mutations"].as_array_mut().expect("mutations is an array");
        arr.push(m.clone());
        arr.push(m);
    });
    assert!(message.contains("Duplicate mutation name: createOrder"), "{message}");
}

/// **The check that must NOT move to the load path.** `CompiledSchema::validate()` also
/// carried a "references undefined type" check, and that check was wrong: it resolved
/// return types against `self.types` plus ten builtin scalar names, while the schema
/// holds `enums`, `interfaces` and `unions` in *separate* collections. Measured before
/// deleting it, it reported `Query 'orderStatus' references undefined type
/// 'OrderStatus'` for the document below — so moving it here would have refused, at
/// boot, every schema with an enum-returning query.
///
/// Nobody noticed in the years it existed, because nothing ran it. That is the argument
/// for having one reachable validation surface rather than two, and this test is what
/// stops the next person restoring the second one.
#[test]
fn a_query_returning_an_enum_or_union_still_loads() {
    let doc = serde_json::json!({
      "types": [
        {"name": "Order", "sql_source": "v_order",
         "fields": [{"name": "id", "field_type": "Int", "nullable": false}]}
      ],
      "enums": [{"name": "OrderStatus", "values": [{"name": "OPEN"}, {"name": "CLOSED"}]}],
      "queries": [
        {"name": "orderStatus", "return_type": "OrderStatus", "sql_source": "v_order"}
      ],
      "mutations": []
    });
    CompiledSchema::from_json(&doc.to_string(), false).expect(
        "a query returning an enum must load: the load path resolves return types \
         against types/enums/interfaces/unions or not at all, never against `types` alone",
    );
}

/// Ruling AB 1: a fact table linked to a type is read as that type, so every name it
/// declares must be a field of it — a declared column the type lacks would be a column no
/// gate can reach. Refused at load, naming the fact table, the type and the names.
#[test]
fn a_fact_table_linked_to_a_type_lacking_one_of_its_columns_is_refused_at_load() {
    let json = r#"{
        "types": [{
            "name":"Sale","sql_source":"v_sale",
            "fields":[{"name":"revenue","field_type":"Float"}]
        }],
        "queries": [], "mutations": [], "subscriptions": [],
        "fact_tables": {"tf_sales": {
            "table_name":"tf_sales","type_name":"Sale",
            "measures":[
                {"name":"revenue","sql_type":"Decimal","nullable":false},
                {"name":"margin","sql_type":"Decimal","nullable":true}
            ],
            "dimensions":{"name":"data","paths":[]},
            "denormalized_filters":[]
        }}
    }"#;
    let err = CompiledSchema::from_json(json, false).unwrap_err();
    let msg = err.to_string();
    assert!(msg.contains("tf_sales"), "the message must name the fact table, got: {msg}");
    assert!(msg.contains("Sale"), "the message must name the type, got: {msg}");
    assert!(msg.contains("margin"), "the message must name the missing field, got: {msg}");
}

/// Ruling AB 1: the linked type must exist.
#[test]
fn a_fact_table_linked_to_an_unknown_type_is_refused_at_load() {
    let json = r#"{
        "types": [], "queries": [], "mutations": [], "subscriptions": [],
        "fact_tables": {"tf_sales": {
            "table_name":"tf_sales","type_name":"Sael",
            "measures":[{"name":"revenue","sql_type":"Decimal","nullable":false}],
            "dimensions":{"name":"data","paths":[]},
            "denormalized_filters":[]
        }}
    }"#;
    let msg = CompiledSchema::from_json(json, false).unwrap_err().to_string();
    assert!(msg.contains("Sael"), "the message must name the type, got: {msg}");
}

/// Control: an unlinked fact table loads as before, and a complete link loads.
#[test]
fn an_unlinked_or_completely_linked_fact_table_loads() {
    let json = r#"{
        "types": [{
            "name":"Sale","sql_source":"v_sale",
            "fields":[{"name":"revenue","field_type":"Float"},{"name":"category","field_type":"String"}]
        }],
        "queries": [], "mutations": [], "subscriptions": [],
        "fact_tables": {
            "tf_sales": {
                "table_name":"tf_sales","type_name":"Sale",
                "measures":[{"name":"revenue","sql_type":"Decimal","nullable":false}],
                "dimensions":{"name":"data","paths":[
                    {"name":"category","json_path":"data->>'category'","data_type":"text"}
                ]},
                "denormalized_filters":[]
            },
            "tf_events": {
                "table_name":"tf_events",
                "measures":[{"name":"n","sql_type":"Int","nullable":false}],
                "dimensions":{"name":"data","paths":[]},
                "denormalized_filters":[]
            }
        }
    }"#;
    CompiledSchema::from_json(json, false).unwrap();
}

// ── #1498: a fact-table path column's hierarchy is checked at load ─────────────

mod fact_table_hierarchy {
    use std::collections::HashMap;

    use crate::{
        compiler::fact_table::{DimensionColumn, FactTableMetadata, FilterColumn, SqlType},
        schema::{CompiledSchema, HierarchyDefinition},
    };

    fn schema_with(sql_type: SqlType, hierarchy: &str, declared: bool) -> String {
        let mut schema = CompiledSchema::new();
        schema.add_fact_table(
            "tf_org".to_string(),
            FactTableMetadata {
                table_name:               "tf_org".to_string(),
                type_name:                None,
                measures:                 vec![],
                dimensions:               DimensionColumn {
                    name:  "data".to_string(),
                    paths: vec![],
                },
                denormalized_filters:     vec![FilterColumn {
                    name: "org_path".to_string(),
                    sql_type,
                    indexed: true,
                    hierarchy: Some(hierarchy.to_string()),
                }],
                calendar_dimensions:      vec![],
                partial_period:           None,
                native_measures:          HashMap::new(),
                native_dimension_mapping: HashMap::new(),
            },
        );
        if declared {
            schema.hierarchies_config = Some(HashMap::from([(
                "org".to_string(),
                HierarchyDefinition {
                    table:       "tb_org_unit".to_string(),
                    path_column: "path".to_string(),
                },
            )]));
        }
        serde_json::to_string(&schema).unwrap()
    }

    #[test]
    fn a_declared_hierarchy_on_an_ltree_column_loads() {
        CompiledSchema::from_json(&schema_with(SqlType::Ltree, "org", true), false).unwrap();
    }

    #[test]
    fn an_undeclared_hierarchy_is_refused() {
        let err = CompiledSchema::from_json(&schema_with(SqlType::Ltree, "org", false), false)
            .unwrap_err()
            .to_string();
        assert!(err.contains("tf_org.org_path") && err.contains("[hierarchies.org]"), "{err}");
    }

    #[test]
    fn a_hierarchy_on_a_column_that_is_not_a_path_is_refused() {
        let err = CompiledSchema::from_json(&schema_with(SqlType::Text, "org", true), false)
            .unwrap_err()
            .to_string();
        assert!(err.contains("tf_org.org_path") && err.contains("ltree"), "{err}");
    }
}