postrust-graphql 3.0.0

//! Integration tests for the GraphQL surface: reads, writes, and the schema
//! shape for subscriptions.
//!
//! These execute real GraphQL documents against a real PostgreSQL database,
//! which is the only place resolver behaviour can be verified -- the SQL the
//! resolvers build is not observable from a unit test.
//!
//! Each test creates its own PostgreSQL schema containing a single `widgets`
//! table and exposes only that schema, so tests cannot disturb each other and
//! the generated field names are predictable. That also exercises GraphQL
//! against a non-`public` schema. Run with:
//!
//! ```text
//! DATABASE_URL="postgres://postgres:postgres@localhost:5432/postrust_test" \
//!   cargo test --package postrust-graphql --test graphql_integration -- --ignored
//! ```

use async_graphql::Request;
use postrust_auth::AuthResult;
use postrust_core::schema_cache::{SchemaCache, SchemaCacheRef};
use postrust_graphql::context::GraphQLContext;
use postrust_graphql::handler::GraphQLState;
use postrust_graphql::schema::SchemaConfig;
use sqlx::postgres::PgPoolOptions;
use sqlx::AssertSqlSafe;
use sqlx::{Executor, PgPool};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;

/// Connect as a role that can create and mutate the test tables.
const TEST_ROLE: &str = "postgres";

static TABLE_COUNTER: AtomicU32 = AtomicU32::new(0);

fn database_url() -> String {
    std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "postgres://postgres:postgres@localhost:5432/postrust_test".to_string())
}

/// Name for a throwaway schema dedicated to one test.
fn unique_schema_name(prefix: &str) -> String {
    let id = TABLE_COUNTER.fetch_add(1, Ordering::SeqCst);
    let stamp = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis();
    format!("gql_{}_{}_{}", prefix, stamp, id)
}

async fn connect() -> PgPool {
    PgPoolOptions::new()
        .max_connections(2)
        .connect(&database_url())
        .await
        .expect("failed to connect to test database")
}

/// Create a dedicated schema holding a `widgets` table with a mix of column
/// types, so type coercion is exercised too.
async fn create_widgets_schema(pool: &PgPool, schema: &str) {
    pool.execute(AssertSqlSafe(format!(
        "DROP SCHEMA IF EXISTS {} CASCADE",
        schema
    )))
    .await
    .expect("drop schema failed");
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("create schema failed");

    pool.execute(AssertSqlSafe(format!(
        r#"
            CREATE TABLE {}.widgets (
                id SERIAL PRIMARY KEY,
                name TEXT NOT NULL,
                category TEXT NOT NULL,
                price NUMERIC(10,2) NOT NULL,
                stock INTEGER NOT NULL,
                is_active BOOLEAN NOT NULL DEFAULT true
            )
            "#,
        schema
    )))
    .await
    .expect("create failed");

    pool.execute(AssertSqlSafe(format!(
        r#"
            INSERT INTO {}.widgets (name, category, price, stock, is_active) VALUES
                ('alpha', 'books', 10.50, 5, true),
                ('bravo', 'tools', 20.00, 0, true),
                ('charlie', 'books', 30.25, 12, false),
                ('delta', 'tools', 40.75, 7, true)
            "#,
        schema
    )))
    .await
    .expect("seed failed");
}

async fn create_widgets_schema_with_computed_field(pool: &PgPool, schema: &str) {
    create_widgets_schema(pool, schema).await;
    pool.execute(AssertSqlSafe(format!(
        r#"
            CREATE FUNCTION {schema}.widget_label(widget_row {schema}.widgets)
            RETURNS TEXT
            LANGUAGE SQL
            STABLE
            AS $$
                SELECT widget_row.name || ':' || widget_row.category
            $$
            "#
    )))
    .await
    .expect("create computed field failed");
}

async fn create_composite_key_schema(pool: &PgPool, schema: &str) {
    pool.execute(AssertSqlSafe(format!(
        "DROP SCHEMA IF EXISTS {} CASCADE",
        schema
    )))
    .await
    .expect("drop schema failed");
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("create schema failed");

    pool.execute(AssertSqlSafe(format!(
        r#"
            CREATE TABLE {}.inventory (
                warehouse_id INTEGER NOT NULL,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL,
                PRIMARY KEY (warehouse_id, sku)
            );
            INSERT INTO {}.inventory (warehouse_id, sku, quantity) VALUES
                (1, 'alpha', 5),
                (1, 'bravo', 10),
                (2, 'alpha', 15)
            "#,
        schema, schema
    )))
    .await
    .expect("create composite key fixture failed");
}

async fn create_equivalent_entity_keys_schema(pool: &PgPool, schema: &str) {
    pool.execute("CREATE EXTENSION IF NOT EXISTS citext")
        .await
        .expect("create citext extension failed");
    pool.execute(AssertSqlSafe(format!(
        "DROP SCHEMA IF EXISTS {schema} CASCADE"
    )))
    .await
    .expect("drop schema failed");
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
        .await
        .expect("create schema failed");

    pool.execute(AssertSqlSafe(format!(
        r#"
            CREATE TABLE {schema}.uuid_entities (
                id UUID PRIMARY KEY,
                label TEXT NOT NULL
            );
            CREATE TABLE {schema}.citext_entities (
                id CITEXT PRIMARY KEY,
                label TEXT NOT NULL
            );
            CREATE TABLE {schema}.timestamptz_entities (
                id TIMESTAMPTZ PRIMARY KEY,
                label TEXT NOT NULL
            );
            INSERT INTO {schema}.uuid_entities VALUES
                ('550e8400-e29b-41d4-a716-446655440000', 'uuid row');
            INSERT INTO {schema}.citext_entities VALUES ('alice', 'citext row');
            INSERT INTO {schema}.timestamptz_entities VALUES
                ('2024-01-01T00:00:00+00:00', 'timestamptz row')
            "#
    )))
    .await
    .expect("create equivalent entity key fixture failed");
}

async fn create_decimal_entity_keys_schema(pool: &PgPool, schema: &str) {
    pool.execute(AssertSqlSafe(format!(
        "DROP SCHEMA IF EXISTS {schema} CASCADE"
    )))
    .await
    .expect("drop schema failed");
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
        .await
        .expect("create schema failed");
    pool.execute(AssertSqlSafe(format!(
        r#"
            CREATE TABLE {schema}.bigint_entities (
                id BIGINT PRIMARY KEY,
                label TEXT NOT NULL
            );
            CREATE TABLE {schema}.numeric_entities (
                id NUMERIC PRIMARY KEY,
                label TEXT NOT NULL
            );
            INSERT INTO {schema}.bigint_entities VALUES
                (7, 'seven'),
                (9007199254740993, 'past JavaScript integer precision');
            INSERT INTO {schema}.numeric_entities VALUES
                (9007199254740993.123456789, 'exact decimal')
            "#
    )))
    .await
    .expect("create decimal entity key fixture failed");
}

async fn create_mixed_entity_schema(pool: &PgPool, schema: &str) {
    pool.execute(AssertSqlSafe(format!(
        "DROP SCHEMA IF EXISTS {} CASCADE",
        schema
    )))
    .await
    .expect("drop schema failed");
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("create schema failed");

    for statement in [
        format!("CREATE TABLE {schema}.authors (id SERIAL PRIMARY KEY, name TEXT NOT NULL)"),
        format!(
            "CREATE TABLE {schema}.author_items (id SERIAL PRIMARY KEY, title TEXT NOT NULL, \
             author_id INTEGER NOT NULL REFERENCES {schema}.authors(id))"
        ),
        format!("CREATE TABLE {schema}.shops (id SERIAL PRIMARY KEY, code TEXT NOT NULL)"),
        format!(
            "CREATE TABLE {schema}.shop_items (id SERIAL PRIMARY KEY, sku TEXT NOT NULL, \
             shop_id INTEGER NOT NULL REFERENCES {schema}.shops(id))"
        ),
        format!(
            "CREATE FUNCTION {schema}.code(author_row {schema}.authors, prefix TEXT) \
             RETURNS TEXT LANGUAGE SQL STABLE AS \
             'SELECT prefix || author_row.name'"
        ),
        format!("INSERT INTO {schema}.authors (name) VALUES ('ada')"),
        format!("INSERT INTO {schema}.author_items (title, author_id) VALUES ('paper', 1)"),
        format!("INSERT INTO {schema}.shops (code) VALUES ('shop-code')"),
        format!("INSERT INTO {schema}.shop_items (sku, shop_id) VALUES ('sku-1', 1)"),
    ] {
        pool.execute(AssertSqlSafe(statement))
            .await
            .expect("create mixed entity fixture failed");
    }
}

async fn drop_schema(pool: &PgPool, schema: &str) {
    let _ = pool
        .execute(AssertSqlSafe(format!(
            "DROP SCHEMA IF EXISTS {} CASCADE",
            schema
        )))
        .await;
}

/// Build GraphQL state over the current database schema.
async fn build_state(
    pool: &PgPool,
    schema: &str,
    max_rows: Option<i64>,
    subscriptions: bool,
) -> Arc<GraphQLState> {
    let schemas = vec![schema.to_string()];
    let cache = SchemaCache::load(pool, &schemas)
        .await
        .expect("failed to load schema cache");

    let mut config = SchemaConfig::default();
    config.exposed_schemas = schemas.clone();
    config.enable_mutations = true;
    config.enable_subscriptions = subscriptions;
    config.max_rows = max_rows;

    Arc::new(
        GraphQLState::new(pool.clone(), Arc::new(cache), config)
            .expect("failed to build GraphQL schema"),
    )
}

/// Build GraphQL state with federation enabled and a metadata document.
async fn build_federated_state_with_names(
    pool: &PgPool,
    schema: &str,
    names: &str,
) -> Arc<GraphQLState> {
    let schemas = vec![schema.to_string()];
    let cache = SchemaCache::load(pool, &schemas)
        .await
        .expect("failed to load schema cache");

    let mut config = SchemaConfig::default();
    config.exposed_schemas = schemas.clone();
    config.enable_mutations = true;
    config.enable_federation = true;
    config.names = postrust_graphql::names::NameOverrides::parse(names).expect("the names parse");

    Arc::new(
        GraphQLState::new(pool.clone(), Arc::new(cache), config)
            .expect("failed to build GraphQL schema"),
    )
}

fn shared_table_names(schema: &str, table: &str) -> String {
    format!(r#"{{"tables": {{"{schema}.{table}": {{"federation": {{"shared": true}}}}}}}}"#)
}

fn mixed_entity_names(schema: &str) -> String {
    format!(
        r#"{{"tables": {{
            "{schema}.authors": {{
                "federation": {{"shared": true}},
                "relationships": {{"author_items_author_id_fkey": "items"}}
            }},
            "{schema}.shops": {{
                "federation": {{"shared": true}},
                "relationships": {{"shop_items_shop_id_fkey": "items"}}
            }}
        }}}}"#
    )
}

/// Execute a GraphQL document and return the whole response.
async fn execute(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
) -> async_graphql::Response {
    execute_with_auth(
        state,
        pool,
        schema,
        query,
        AuthResult {
            role: TEST_ROLE.to_string(),
            claims: HashMap::new(),
        },
    )
    .await
}

/// Execute a GraphQL document with an explicit verified identity.
async fn execute_with_auth(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
    auth: AuthResult,
) -> async_graphql::Response {
    let cache = SchemaCache::load(pool, &[schema.to_string()])
        .await
        .expect("failed to load schema cache");

    // Every write in one operation shares a transaction, and the caller is what
    // settles it -- so a test that only executes and never settles would leave
    // its rows uncommitted, exactly as the server would.
    let write: postrust_graphql::context::SharedWrite = Default::default();
    let ctx = GraphQLContext::new(pool.clone(), SchemaCacheRef::from_static(cache), auth)
        .with_write(std::sync::Arc::clone(&write));

    let request = Request::new(query).data(ctx).data(pool.clone());
    let response = state.schema.execute(request).await;
    if let Some(tx) = write.lock().await.take() {
        if response.errors.is_empty() {
            tx.commit().await.expect("the write could not be committed");
        } else {
            tx.rollback().await.expect("the write could not be undone");
        }
    }
    response
}

/// Execute and require success, returning the `data` payload as JSON.
async fn execute_ok(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
) -> serde_json::Value {
    let response = execute(state, pool, schema, query).await;
    assert!(
        response.errors.is_empty(),
        "expected no GraphQL errors for {} -- got: {:?}",
        query,
        response.errors
    );
    serde_json::to_value(&response.data).expect("data was not serialisable")
}

/// Execute and require failure, returning the joined error messages.
async fn execute_err(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
) -> String {
    let response = execute(state, pool, schema, query).await;
    assert!(
        !response.errors.is_empty(),
        "expected a GraphQL error for {} -- got data: {:?}",
        query,
        response.data
    );
    response
        .errors
        .iter()
        .map(|e| e.message.clone())
        .collect::<Vec<_>>()
        .join("; ")
}

fn ids_of(rows: &serde_json::Value, field: &str) -> Vec<i64> {
    rows.get(field)
        .and_then(|v| v.as_array())
        .unwrap_or_else(|| panic!("expected a list at {} -- got {}", field, rows))
        .iter()
        .map(|row| {
            row.get("id")
                .and_then(|v| v.as_i64())
                .unwrap_or_else(|| panic!("row missing integer id: {}", row))
        })
        .collect()
}

// ===========================================================================
// Reads
// ===========================================================================

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn list_query_returns_rows_with_typed_columns() {
    let pool = connect().await;
    let schema = unique_schema_name("list");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: asc}]) { id name stock is_active } }",
    )
    .await;

    let rows = data
        .get("widgets")
        .and_then(|v| v.as_array())
        .unwrap()
        .clone();
    assert_eq!(rows.len(), 4);

    // Columns must come back with their real types, not stringified or null.
    assert_eq!(rows[0].get("id").and_then(|v| v.as_i64()), Some(1));
    assert_eq!(rows[0].get("name").and_then(|v| v.as_str()), Some("alpha"));
    assert_eq!(rows[0].get("stock").and_then(|v| v.as_i64()), Some(5));
    assert_eq!(
        rows[0].get("is_active").and_then(|v| v.as_bool()),
        Some(true)
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn by_pk_query_returns_the_requested_row() {
    // Regression: the by-PK resolver built no WHERE clause at all, fetched the
    // whole table and returned whichever row came back first.
    let pool = connect().await;
    let schema = unique_schema_name("bypk");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets_by_pk(id: 3) { id name } }",
    )
    .await;

    let row = data.get("widgets_by_pk").expect("missing by-pk field");
    assert_eq!(row.get("id").and_then(|v| v.as_i64()), Some(3));
    assert_eq!(row.get("name").and_then(|v| v.as_str()), Some("charlie"));

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn by_pk_query_returns_null_for_a_missing_key() {
    let pool = connect().await;
    let schema = unique_schema_name("bypknull");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets_by_pk(id: 9999) { id name } }",
    )
    .await;

    assert_eq!(
        data.get("widgets_by_pk"),
        Some(&serde_json::Value::Null),
        "a key that matches nothing must resolve to null"
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_resolve_shared_rows_by_key() {
    let pool = connect().await;
    let schema = unique_schema_name("fedentity");
    create_widgets_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "widgets");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "widgets", id: 3},
            {__typename: "widgets", id: 1}
          ]) {
            __typename
            ... on widgets {
              id
              name
            }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .expect("entities list");
    assert_eq!(rows.len(), 2);
    assert_eq!(
        rows[0].get("__typename").and_then(|value| value.as_str()),
        Some("widgets")
    );
    assert_eq!(rows[0].get("id").and_then(|value| value.as_i64()), Some(3));
    assert_eq!(
        rows[0].get("name").and_then(|value| value.as_str()),
        Some("charlie")
    );
    assert_eq!(
        rows[1].get("__typename").and_then(|value| value.as_str()),
        Some("widgets")
    );
    assert_eq!(rows[1].get("id").and_then(|value| value.as_i64()), Some(1));
    assert_eq!(
        rows[1].get("name").and_then(|value| value.as_str()),
        Some("alpha")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_use_postgres_key_equality() {
    let pool = connect().await;
    let schema = unique_schema_name("fedequality");
    create_equivalent_entity_keys_schema(&pool, &schema).await;

    let names = format!(
        r#"{{"tables": {{
            "{schema}.uuid_entities": {{"federation": {{"shared": true}}}},
            "{schema}.citext_entities": {{"federation": {{"shared": true}}}},
            "{schema}.timestamptz_entities": {{"federation": {{"shared": true}}}}
        }}}}"#
    );
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "uuid_entities", id: "550E8400-E29B-41D4-A716-446655440000"},
            {__typename: "uuid_entities", id: "550e8400-e29b-41d4-a716-446655440000"},
            {__typename: "citext_entities", id: "ALICE"},
            {__typename: "timestamptz_entities", id: "2024-01-01T00:00:00Z"}
          ]) {
            ... on uuid_entities { label }
            ... on citext_entities { label }
            ... on timestamptz_entities { label }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(serde_json::Value::as_array)
        .expect("entities list");
    assert_eq!(rows.len(), 4);
    assert_eq!(rows[0]["label"], "uuid row");
    assert_eq!(rows[1]["label"], "uuid row");
    assert_eq!(rows[2]["label"], "citext row");
    assert_eq!(rows[3]["label"], "timestamptz row");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_accept_string_bigint_and_numeric_keys() {
    let pool = connect().await;
    let schema = unique_schema_name("feddecimalkeys");
    create_decimal_entity_keys_schema(&pool, &schema).await;

    let names = format!(
        r#"{{"tables": {{
            "{schema}.bigint_entities": {{"federation": {{"shared": true}}}},
            "{schema}.numeric_entities": {{"federation": {{"shared": true}}}}
        }}}}"#
    );
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "bigint_entities", id: "7"},
            {__typename: "bigint_entities", id: 7},
            {__typename: "bigint_entities", id: "9007199254740993"},
            {__typename: "numeric_entities", id: "9007199254740993.123456789"}
          ]) {
            ... on bigint_entities { label }
            ... on numeric_entities { label }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(serde_json::Value::as_array)
        .expect("entities list");
    assert_eq!(rows.len(), 4);
    assert_eq!(rows[0]["label"], "seven");
    assert_eq!(rows[1]["label"], "seven");
    assert_eq!(rows[2]["label"], "past JavaScript integer precision");
    assert_eq!(rows[3]["label"], "exact decimal");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_integer_keys_reject_string_representations_before_querying() {
    let pool = connect().await;
    let schema = unique_schema_name("fedkeytype");
    create_widgets_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "widgets");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "widgets", id: "2"}]) {
            ... on widgets { id name }
          }
        }
        "#,
    )
    .await;

    assert!(
        errors.contains(
            "invalid key field \"id\": expected a value compatible with PostgreSQL type \"int4\", found a string"
        ),
        "unexpected error: {}",
        errors
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_integer_backed_ids_reuse_prepared_statements_across_both_input_forms() {
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(&database_url())
        .await
        .expect("failed to connect to test database");
    let schema = unique_schema_name("fedidnormalize");
    create_widgets_schema(&pool, &schema).await;

    let names = format!(
        r#"{{"tables": {{"{schema}.widgets": {{
            "column_types": {{"id": "ID"}},
            "federation": {{"shared": true}}
        }}}}}}"#
    );
    let state = build_federated_state_with_names(&pool, &schema, &names).await;

    for id in ["2", "\"2\"", "2"] {
        let query = format!(
            r#"{{
              _entities(representations: [{{__typename: "widgets", id: {id}}}]) {{
                ... on widgets {{ id name }}
              }}
            }}"#
        );
        let data = execute_ok(&state, &pool, &schema, &query).await;
        let entity = data
            .get("_entities")
            .and_then(|value| value.as_array())
            .and_then(|values| values.first())
            .expect("resolved entity");
        assert_eq!(
            entity.get("name").and_then(|value| value.as_str()),
            Some("bravo")
        );
    }

    // A different projection produces a fresh prepared statement. Start this
    // one with the string spelling so the opposite cache order is covered.
    for id in ["\"3\"", "3"] {
        let query = format!(
            r#"{{
              _entities(representations: [{{__typename: "widgets", id: {id}}}]) {{
                ... on widgets {{ id category }}
              }}
            }}"#
        );
        let data = execute_ok(&state, &pool, &schema, &query).await;
        let entity = data
            .get("_entities")
            .and_then(|value| value.as_array())
            .and_then(|values| values.first())
            .expect("resolved entity");
        assert_eq!(
            entity.get("category").and_then(|value| value.as_str()),
            Some("books")
        );
    }

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_return_null_for_missing_rows() {
    let pool = connect().await;
    let schema = unique_schema_name("fedmissing");
    create_widgets_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "widgets");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "widgets", id: 9999},
            {__typename: "widgets", id: 2}
          ]) {
            __typename
            ... on widgets {
              id
              name
            }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .expect("entity results");
    assert_eq!(rows.len(), 2);
    assert!(rows[0].is_null());
    assert_eq!(rows[1].get("id").and_then(|value| value.as_i64()), Some(2));
    assert_eq!(
        rows[1].get("name").and_then(|value| value.as_str()),
        Some("bravo")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_reject_unknown_typenames() {
    let pool = connect().await;
    let schema = unique_schema_name("fedunknown");
    create_widgets_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "widgets");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "not_widgets", id: 1}]) {
            __typename
          }
        }
        "#,
    )
    .await;

    assert!(
        errors.contains("unknown federation entity \"not_widgets\""),
        "unexpected error: {}",
        errors
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_reject_missing_key_fields() {
    let pool = connect().await;
    let schema = unique_schema_name("fednokey");
    create_widgets_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "widgets");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "widgets"}]) {
            __typename
            ... on widgets {
              id
            }
          }
        }
        "#,
    )
    .await;

    assert!(
        errors.contains("missing key field \"id\""),
        "unexpected error: {}",
        errors
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_use_renamed_key_fields() {
    let pool = connect().await;
    let schema = unique_schema_name("fedrename");
    create_widgets_schema(&pool, &schema).await;

    let names = format!(
        r#"{{"tables": {{"{schema}.widgets": {{
            "columns": {{"id": "widget_id"}},
            "federation": {{"shared": true}}
        }}}}}}"#
    );
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "widgets", widget_id: 4}]) {
            __typename
            ... on widgets {
              widget_id
              name
            }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .expect("entities list");
    assert_eq!(rows.len(), 1);
    assert_eq!(
        rows[0].get("widget_id").and_then(|value| value.as_i64()),
        Some(4)
    );
    assert_eq!(
        rows[0].get("name").and_then(|value| value.as_str()),
        Some("delta")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_resolve_composite_primary_keys() {
    let pool = connect().await;
    let schema = unique_schema_name("fedcomposite");
    create_composite_key_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "inventory");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "inventory", warehouse_id: 2, sku: "alpha"},
            {__typename: "inventory", warehouse_id: 1, sku: "bravo"}
          ]) {
            __typename
            ... on inventory {
              warehouse_id
              sku
              quantity
            }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .expect("entities list");
    assert_eq!(rows.len(), 2);
    assert_eq!(
        rows[0].get("warehouse_id").and_then(|value| value.as_i64()),
        Some(2)
    );
    assert_eq!(
        rows[0].get("sku").and_then(|value| value.as_str()),
        Some("alpha")
    );
    assert_eq!(
        rows[0].get("quantity").and_then(|value| value.as_i64()),
        Some(15)
    );
    assert_eq!(
        rows[1].get("warehouse_id").and_then(|value| value.as_i64()),
        Some(1)
    );
    assert_eq!(
        rows[1].get("sku").and_then(|value| value.as_str()),
        Some("bravo")
    );
    assert_eq!(
        rows[1].get("quantity").and_then(|value| value.as_i64()),
        Some(10)
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_do_not_bypass_select_permissions() {
    let pool = connect().await;
    let schema = unique_schema_name("fedperm");
    create_widgets_schema(&pool, &schema).await;

    let names = format!(
        r#"{{"tables": {{"{schema}.widgets": {{
            "federation": {{"shared": true}},
            "permissions": {{"listener": {{"select": {{
                "columns": "*",
                "filter": {{"category": {{"_eq": "books"}}}}
            }}}}}}
        }}}}}}"#
    );
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_as(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "widgets", id: 2},
            {__typename: "widgets", id: 3}
          ]) {
            __typename
            ... on widgets {
              id
              name
            }
          }
        }
        "#,
    )
    .await;

    let rows = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .expect("entity results");
    assert_eq!(rows.len(), 2);
    assert!(rows[0].is_null());
    assert_eq!(rows[1].get("id").and_then(|value| value.as_i64()), Some(3));
    assert_eq!(
        rows[1].get("name").and_then(|value| value.as_str()),
        Some("charlie")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_can_select_relationships() {
    let pool = connect().await;
    let schema = unique_schema_name("fedrel");
    create_related_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "authors");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "authors", id: 1}]) {
            __typename
            ... on authors {
              id
              name
              books(order_by: [{id: asc}]) {
                id
                title
              }
            }
          }
        }
        "#,
    )
    .await;

    let row = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .and_then(|rows| rows.first())
        .expect("first entity");
    assert_eq!(
        row.get("name").and_then(|value| value.as_str()),
        Some("ada")
    );
    let books = row
        .get("books")
        .and_then(|value| value.as_array())
        .expect("books list");
    assert_eq!(books.len(), 2);
    assert_eq!(
        books[0].get("title").and_then(|value| value.as_str()),
        Some("a-one")
    );
    assert_eq!(
        books[1].get("title").and_then(|value| value.as_str()),
        Some("a-two")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entities_can_select_computed_fields() {
    let pool = connect().await;
    let schema = unique_schema_name("fedcomputed");
    create_widgets_schema_with_computed_field(&pool, &schema).await;

    let names = shared_table_names(&schema, "widgets");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "widgets", id: 1}]) {
            __typename
            ... on widgets {
              id
              widget_label
            }
          }
        }
        "#,
    )
    .await;

    let row = data
        .get("_entities")
        .and_then(|value| value.as_array())
        .and_then(|rows| rows.first())
        .expect("first entity");
    assert_eq!(
        row.get("widget_label").and_then(|value| value.as_str()),
        Some("alpha:books")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entity_relationships_respect_fragment_type_conditions() {
    let pool = connect().await;
    let schema = unique_schema_name("fedrelfragments");
    create_mixed_entity_schema(&pool, &schema).await;

    let names = mixed_entity_names(&schema);
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "authors", id: 1},
            {__typename: "shops", id: 1}
          ]) {
            ... on authors {
              id
              items { title }
            }
            ... on shops {
              id
              items { sku }
            }
          }
        }
        "#,
    )
    .await;

    let rows = data["_entities"].as_array().expect("entities list");
    assert_eq!(rows[0]["items"][0]["title"], "paper");
    assert_eq!(rows[1]["items"][0]["sku"], "sku-1");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entity_computed_fields_respect_fragment_type_conditions() {
    let pool = connect().await;
    let schema = unique_schema_name("fedcomputedfragments");
    create_mixed_entity_schema(&pool, &schema).await;

    let names = mixed_entity_names(&schema);
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [
            {__typename: "authors", id: 1},
            {__typename: "shops", id: 1}
          ]) {
            ... on authors { id }
            ... on shops { id code }
          }
        }
        "#,
    )
    .await;

    let rows = data["_entities"].as_array().expect("entities list");
    assert_eq!(rows[0]["id"], 1);
    assert_eq!(rows[1]["code"], "shop-code");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn filter_argument_narrows_results() {
    // Regression: `filter` was a declared argument that the resolver ignored,
    // so a filtered query silently returned the entire table.
    let pool = connect().await;
    let schema = unique_schema_name("filter");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {category: {_eq: \"books\"}}, order_by: [{id: asc}]) { id } }",
    )
    .await;

    assert_eq!(ids_of(&data, "widgets"), vec![1, 3]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn filter_supports_comparison_operators() {
    let pool = connect().await;
    let schema = unique_schema_name("filtercmp");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {stock: {_gt: 5}}, order_by: [{id: asc}]) { id } }",
    )
    .await;

    assert_eq!(ids_of(&data, "widgets"), vec![3, 4]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn order_by_sorts_ascending_and_descending() {
    // Regression: `orderBy` was declared and ignored.
    let pool = connect().await;
    let schema = unique_schema_name("order");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;

    let ascending = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: asc}]) { id } }",
    )
    .await;
    let descending = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: desc}]) { id } }",
    )
    .await;

    assert_eq!(ids_of(&ascending, "widgets"), vec![1, 2, 3, 4]);
    assert_eq!(ids_of(&descending, "widgets"), vec![4, 3, 2, 1]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn order_by_sorts_on_a_non_key_column() {
    let pool = connect().await;
    let schema = unique_schema_name("ordercol");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{stock: asc}]) { id stock } }",
    )
    .await;

    // stock: bravo 0, alpha 5, delta 7, charlie 12
    assert_eq!(ids_of(&data, "widgets"), vec![2, 1, 4, 3]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn order_by_rejects_an_unknown_column() {
    // A column name reaches SQL, so it has to be one the table has. It is
    // refused by validation now rather than by the builder -- `<table>_order_by`
    // is a generated input, so a name that is not a column is not a field --
    // which is a better place to refuse it: the client is told before the
    // request runs, and nothing crafted can be written there at all.
    let pool = connect().await;
    let schema = unique_schema_name("orderbad");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{no_such_column: asc}]) { id } }",
    )
    .await;
    assert!(
        errors.contains("no_such_column"),
        "expected the unknown column to be named, got: {}",
        errors
    );

    // The table must still be intact.
    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 4);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn order_by_rejects_an_invalid_direction() {
    let pool = connect().await;
    let schema = unique_schema_name("orderdir");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: sideways}]) { id } }",
    )
    .await;
    assert!(
        errors.contains("sideways"),
        "expected the direction to be named, got: {}",
        errors
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn limit_and_offset_paginate() {
    let pool = connect().await;
    let schema = unique_schema_name("page");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: asc}], limit: 2, offset: 1) { id } }",
    )
    .await;

    assert_eq!(ids_of(&data, "widgets"), vec![2, 3]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn max_rows_caps_a_query_with_no_limit() {
    // Regression: a GraphQL query with no `limit` selected the whole table.
    let pool = connect().await;
    let schema = unique_schema_name("maxrows");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, Some(2), false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: asc}]) { id } }",
    )
    .await;

    assert_eq!(ids_of(&data, "widgets"), vec![1, 2]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn max_rows_bounds_a_larger_requested_limit() {
    let pool = connect().await;
    let schema = unique_schema_name("maxrowslim");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, Some(2), false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(order_by: [{id: asc}], limit: 100) { id } }",
    )
    .await;

    assert_eq!(ids_of(&data, "widgets").len(), 2);

    drop_schema(&pool, &schema).await;
}

// ===========================================================================
// Writes
// ===========================================================================

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn insert_mutation_creates_a_row() {
    let pool = connect().await;
    let schema = unique_schema_name("insert");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { insert_widgets(objects: [{name: \"echo\", category: \"tools\", price: 5.5, stock: 3}]) \
          { affected_rows returning { id name } } }",
    )
    .await;

    let response = data.get("insert_widgets").expect("insert returned nothing");
    assert_eq!(
        response.get("affected_rows").and_then(|v| v.as_i64()),
        Some(1)
    );
    let rows = response
        .get("returning")
        .and_then(|v| v.as_array())
        .expect("insert returned no rows");
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].get("name").and_then(|v| v.as_str()), Some("echo"));

    let total: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(total, 5);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn update_mutation_with_where_changes_only_matching_rows() {
    let pool = connect().await;
    let schema = unique_schema_name("update");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { update_widgets(where: {id: {_eq: 2}}, _set: {name: \"renamed\"}) \
          { affected_rows returning { id name } } }",
    )
    .await;

    let renamed: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets WHERE name = 'renamed'",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(renamed, 1, "exactly one row should have been updated");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn update_mutation_without_where_is_refused() {
    // Regression: an absent `where` produced `UPDATE <table> SET ...` with no
    // WHERE clause, rewriting every row.
    let pool = connect().await;
    let schema = unique_schema_name("updateall");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        "mutation { update_widgets(_set: {name: \"clobbered\"}) { returning { id } } }",
    )
    .await;
    // Refused by validation now rather than by the resolver: `where` is
    // non-null on a bulk write, so a request without one never runs.
    assert!(
        errors.contains("where"),
        "expected a refusal naming `where`, got: {}",
        errors
    );

    let clobbered: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets WHERE name = 'clobbered'",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(clobbered, 0, "no row should have been updated");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn delete_mutation_with_where_removes_only_matching_rows() {
    let pool = connect().await;
    let schema = unique_schema_name("delete");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets(where: {category: {_eq: \"books\"}}) \
          { affected_rows returning { id } } }",
    )
    .await;

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 2, "only the two 'books' rows should be gone");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn delete_mutation_without_where_is_refused() {
    // Regression: this emitted `DELETE FROM <table> RETURNING ...` and emptied
    // the table.
    let pool = connect().await;
    let schema = unique_schema_name("deleteall");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets { returning { id } } }",
    )
    .await;
    // Refused by validation now rather than by the resolver: `where` is
    // non-null on a bulk write, so a request without one never runs.
    assert!(
        errors.contains("where"),
        "expected a refusal naming `where`, got: {}",
        errors
    );

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 4, "the table must be untouched");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn mutation_value_is_not_interpreted_as_sql() {
    let pool = connect().await;
    let schema = unique_schema_name("inject");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { update_widgets(where: {name: {_eq: \"alpha\"}}, _set: {name: \"x'); DROP TABLE widgets;--\"}) \
          { returning { id } } }",
    )
    .await;

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("table should still exist");
    assert_eq!(remaining, 4);

    drop_schema(&pool, &schema).await;
}

// ===========================================================================
// Schema shape: subscriptions, queries, mutations
// ===========================================================================

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn subscription_type_is_present_when_enabled() {
    let pool = connect().await;
    let schema = unique_schema_name("sub");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, true).await;
    let sdl = state.schema.sdl();
    assert!(
        sdl.contains("type subscription_root"),
        "subscriptions enabled but no Subscription type in the schema"
    );
    assert!(
        !state.subscription_fields.is_empty(),
        "no subscription fields were generated"
    );

    // The subscription root mirrors the query root: a live query answers with
    // the same rows the query does, under the same arguments.
    let root = sdl
        .split("type subscription_root")
        .nth(1)
        .expect("the subscription root is in the SDL");
    let root = root.split("\n}").next().expect("the root type closes");
    for expected in [
        "widgets(",
        "widgets_by_pk(",
        "widgets_aggregate(",
        "where: widgets_bool_exp",
        "distinct_on: [widgets_select_column!]",
        "): [widgets!]!",
    ] {
        assert!(
            root.contains(expected),
            "expected `{}` on the subscription root, got:{}",
            expected,
            root
        );
    }

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn subscription_type_is_absent_when_disabled() {
    let pool = connect().await;
    let schema = unique_schema_name("nosub");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    assert!(
        !state.schema.sdl().contains("type Subscription"),
        "subscriptions are disabled but a Subscription type was generated"
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn schema_exposes_read_and_write_fields_for_each_table() {
    let pool = connect().await;
    let schema = unique_schema_name("shape");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let sdl = state.schema.sdl();

    for field in [
        "widgets",
        "widgets_by_pk",
        "insert_widgets",
        "update_widgets",
        "delete_widgets",
    ] {
        assert!(sdl.contains(field), "schema is missing field {}", field);
    }

    drop_schema(&pool, &schema).await;
}

// ===========================================================================
// Multi-schema naming
// ===========================================================================

/// Create a schema holding a `widgets` table with a single identifying row.
async fn create_marker_schema(pool: &PgPool, schema: &str, marker: &str) {
    let _ = pool
        .execute(AssertSqlSafe(format!(
            "DROP SCHEMA IF EXISTS {} CASCADE",
            schema
        )))
        .await;
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("create schema failed");
    pool.execute(AssertSqlSafe(format!(
        "CREATE TABLE {}.widgets (id SERIAL PRIMARY KEY, marker TEXT NOT NULL)",
        schema
    )))
    .await
    .expect("create table failed");
    pool.execute(AssertSqlSafe(format!(
        "INSERT INTO {}.widgets (marker) VALUES ('{}')",
        schema, marker
    )))
    .await
    .expect("seed failed");
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn same_table_name_in_two_schemas_gets_distinct_fields() {
    // Regression: type and field names were derived from the table name alone,
    // so a `widgets` table in two exposed schemas produced identical names and
    // one silently replaced the other.
    let pool = connect().await;
    let default_schema = unique_schema_name("multi_default");
    let other_schema = unique_schema_name("multi_other");

    create_marker_schema(&pool, &default_schema, "from-default").await;
    create_marker_schema(&pool, &other_schema, "from-other").await;

    let schemas = vec![default_schema.clone(), other_schema.clone()];
    let cache = SchemaCache::load(&pool, &schemas)
        .await
        .expect("failed to load schema cache");
    let mut config = SchemaConfig::default();
    config.exposed_schemas = schemas.clone();
    config.enable_mutations = true;
    config.max_rows = None;
    let state = Arc::new(
        GraphQLState::new(pool.clone(), Arc::new(cache), config)
            .expect("failed to build GraphQL schema"),
    );

    // Both tables must be reachable: the default schema keeps the bare name,
    // the other is prefixed with its schema.
    let other_field = format!("{}_widgets", other_schema);
    let sdl = state.schema.sdl();
    assert!(
        sdl.contains("widgets"),
        "default-schema table missing from the schema"
    );
    assert!(
        sdl.contains(&other_field),
        "second-schema table missing; expected a field named {} in:\n{}",
        other_field,
        sdl.lines()
            .filter(|l| l.contains("idgets"))
            .collect::<Vec<_>>()
            .join("\n")
    );

    // And each must read from its own table, not the same one twice.
    let ctx_schema = default_schema.clone();
    let default_rows = execute_ok(&state, &pool, &ctx_schema, "{ widgets { id marker } }").await;
    assert_eq!(
        default_rows["widgets"][0]["marker"].as_str(),
        Some("from-default")
    );

    let other_rows = execute_ok(
        &state,
        &pool,
        &ctx_schema,
        &format!("{{ {} {{ id marker }} }}", other_field),
    )
    .await;
    assert_eq!(
        other_rows[other_field.as_str()][0]["marker"].as_str(),
        Some("from-other"),
        "the prefixed field must read the other schema's table"
    );

    drop_schema(&pool, &default_schema).await;
    drop_schema(&pool, &other_schema).await;
}

// ===========================================================================
// Filter operators
//
// Regression: `build_where_clause` silently skipped any operator it did not
// recognise, so an advertised-but-unimplemented operator (`in`, `isNull`)
// produced no condition at all and the query returned every row.
// ===========================================================================

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn filter_in_operator_matches_a_set() {
    let pool = connect().await;
    let schema = unique_schema_name("filterin");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {id: {_in: [1, 3]}}, order_by: [{id: asc}]) { id } }",
    )
    .await;

    assert_eq!(ids_of(&data, "widgets"), vec![1, 3]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn filter_in_operator_with_an_empty_list_matches_nothing() {
    let pool = connect().await;
    let schema = unique_schema_name("filterinempty");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {id: {_in: []}}) { id } }",
    )
    .await;

    assert!(
        ids_of(&data, "widgets").is_empty(),
        "an empty `in` set must match nothing, not everything"
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn filter_is_null_operator_accepts_camel_case() {
    // The schema advertises `isNull`; the executor only understood `is_null`.
    let pool = connect().await;
    let schema = unique_schema_name("filterisnull");
    create_widgets_schema(&pool, &schema).await;
    // Two statements, issued separately: sqlx prepares each query, and a
    // prepared statement cannot carry multiple commands.
    pool.execute(AssertSqlSafe(format!(
        "ALTER TABLE {}.widgets ADD COLUMN note TEXT",
        schema
    )))
    .await
    .expect("alter failed");
    pool.execute(AssertSqlSafe(format!(
        "UPDATE {}.widgets SET note = 'x' WHERE id = 1",
        schema
    )))
    .await
    .expect("update failed");

    let state = build_state(&pool, &schema, None, false).await;

    let null_rows = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {note: {_is_null: true}}, order_by: [{id: asc}]) { id } }",
    )
    .await;
    assert_eq!(ids_of(&null_rows, "widgets"), vec![2, 3, 4]);

    let non_null_rows = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {note: {_is_null: false}}) { id } }",
    )
    .await;
    assert_eq!(ids_of(&non_null_rows, "widgets"), vec![1]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn filter_rejects_an_unsupported_operator() {
    // Silently ignoring it would return every row.
    let pool = connect().await;
    let schema = unique_schema_name("filterbadop");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        "{ widgets(where: {stock: {_between: 5}}) { id } }",
    )
    .await;

    assert!(
        errors.contains("_between"),
        "expected a rejection, got: {}",
        errors
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn mutation_with_unsupported_where_operator_is_refused() {
    // Before the operator check, an unrecognised operator produced an empty
    // WHERE clause -- which for a delete meant every row.
    let pool = connect().await;
    let schema = unique_schema_name("mutbadop");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let errors = execute_err(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets(where: {stock: {_between: 5}}) { returning { id } } }",
    )
    .await;
    assert!(
        errors.contains("_between"),
        "expected a rejection, got: {}",
        errors
    );

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 4, "the table must be untouched");

    drop_schema(&pool, &schema).await;
}

// ===========================================================================
// By-PK mutations
//
// Regression: `updateXByPk` / `deleteXByPk` declared a free-form `where` and
// simply returned the first affected row, so they were bulk mutations wearing
// a by-key name.
// ===========================================================================

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn update_by_pk_targets_exactly_one_row() {
    let pool = connect().await;
    let schema = unique_schema_name("updbypk");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { update_widgets_by_pk(pk_columns: {id: 2}, _set: {name: \"renamed\"}) { id name } }",
    )
    .await;

    let row = data.get("update_widgets_by_pk").expect("missing result");
    assert_eq!(row.get("id").and_then(|v| v.as_i64()), Some(2));
    assert_eq!(row.get("name").and_then(|v| v.as_str()), Some("renamed"));

    let renamed: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets WHERE name = 'renamed'",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(renamed, 1, "a by-PK update must touch exactly one row");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn delete_by_pk_removes_exactly_one_row() {
    let pool = connect().await;
    let schema = unique_schema_name("delbypk");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets_by_pk(id: 3) { id name } }",
    )
    .await;

    let row = data.get("delete_widgets_by_pk").expect("missing result");
    assert_eq!(row.get("id").and_then(|v| v.as_i64()), Some(3));

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 3, "a by-PK delete must remove exactly one row");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn by_pk_mutations_require_the_key_and_reject_where() {
    let pool = connect().await;
    let schema = unique_schema_name("bypkargs");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;

    // The key argument is required.
    let missing = execute_err(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets_by_pk { id } }",
    )
    .await;
    assert!(
        !missing.is_empty(),
        "a by-PK delete without its key must fail"
    );

    // `where` is no longer part of a by-PK mutation's signature.
    let rejected = execute_err(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets_by_pk(where: {id: {_eq: 1}}) { id } }",
    )
    .await;
    assert!(
        rejected.contains("where"),
        "expected `where` to be rejected on a by-PK mutation, got: {}",
        rejected
    );

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 4, "nothing should have been deleted");

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn by_pk_mutation_with_an_unknown_key_affects_nothing() {
    let pool = connect().await;
    let schema = unique_schema_name("bypkmiss");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { delete_widgets_by_pk(id: 9999) { id } }",
    )
    .await;

    assert_eq!(
        data.get("delete_widgets_by_pk"),
        Some(&serde_json::Value::Null),
        "a key that matches nothing must resolve to null"
    );

    let remaining: i64 = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT count(*) FROM {}.widgets",
        schema
    )))
    .fetch_one(&pool)
    .await
    .expect("count failed");
    assert_eq!(remaining, 4);

    drop_schema(&pool, &schema).await;
}

// ===========================================================================
// Relationship embedding
//
// Relationship metadata was generated but never wired into the schema, so
// nested fields did not exist at all.
// ===========================================================================

/// A schema with a parent/child pair joined by a foreign key.
async fn create_related_schema(pool: &PgPool, schema: &str) {
    let _ = pool
        .execute(AssertSqlSafe(format!(
            "DROP SCHEMA IF EXISTS {} CASCADE",
            schema
        )))
        .await;
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("create schema failed");

    for stmt in [
        format!(
            "CREATE TABLE {}.authors (id SERIAL PRIMARY KEY, name TEXT NOT NULL)",
            schema
        ),
        format!(
            "CREATE TABLE {}.books (id SERIAL PRIMARY KEY, title TEXT NOT NULL, \
             in_print BOOLEAN NOT NULL DEFAULT true, \
             author_id INTEGER NOT NULL REFERENCES {}.authors(id))",
            schema, schema
        ),
        format!(
            "CREATE TABLE {}.chapters (id SERIAL PRIMARY KEY, heading TEXT NOT NULL, \
             book_id INTEGER NOT NULL REFERENCES {}.books(id))",
            schema, schema
        ),
        format!(
            "INSERT INTO {}.authors (name) VALUES ('ada'), ('grace'), ('lonely')",
            schema
        ),
        format!(
            "INSERT INTO {}.books (title, in_print, author_id) VALUES \
             ('a-one', true, 1), ('a-two', false, 1), ('g-one', true, 2)",
            schema
        ),
        format!(
            "INSERT INTO {}.chapters (heading, book_id) VALUES \
             ('a-one-c1', 1), ('a-one-c2', 1), ('g-one-c1', 3)",
            schema
        ),
    ] {
        pool.execute(AssertSqlSafe(stmt))
            .await
            .expect("setup failed");
    }
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn to_many_relationship_is_embedded() {
    let pool = connect().await;
    let schema = unique_schema_name("relmany");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(order_by: [{id: asc}]) { id name books { id title } } }",
    )
    .await;

    let authors = data["authors"].as_array().expect("expected a list");
    assert_eq!(authors.len(), 3);

    let ada_books = authors[0]["books"]
        .as_array()
        .expect("books must be a list");
    assert_eq!(ada_books.len(), 2, "ada has two books");
    assert_eq!(ada_books[0]["title"].as_str(), Some("a-one"));

    assert_eq!(
        authors[2]["books"],
        serde_json::json!([]),
        "an author with no books must get an empty list, not null"
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn to_one_relationship_is_embedded() {
    let pool = connect().await;
    let schema = unique_schema_name("relone");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ books(order_by: [{id: asc}]) { id title author { id name } } }",
    )
    .await;

    let books = data["books"].as_array().expect("expected a list");
    assert_eq!(books.len(), 3);
    assert_eq!(
        books[0]["author"]["name"].as_str(),
        Some("ada"),
        "a to-one relationship must resolve to its single parent"
    );
    assert_eq!(books[2]["author"]["name"].as_str(), Some("grace"));

    drop_schema(&pool, &schema).await;
}

/// `__typename` inside a relationship is answered by the schema, as it is at
/// the root. Apollo Client asks for it in every selection set, and a
/// Federation router asks for it on every entity reference it follows, so a
/// relationship that read it as a column failed the whole query.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn typename_inside_relationships_is_answered_by_the_schema() {
    let pool = connect().await;
    let schema = unique_schema_name("reltypename");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ books(order_by: [{id: asc}], limit: 1) { __typename id \
           author { __typename id books(order_by: [{id: asc}]) { __typename title } } \
           chapters(order_by: [{id: asc}]) { __typename } } }",
    )
    .await;
    assert_eq!(
        data["books"][0],
        serde_json::json!({
            "__typename": "books",
            "id": 1,
            "author": {
                "__typename": "authors",
                "id": 1,
                "books": [
                    { "__typename": "books", "title": "a-one" },
                    { "__typename": "books", "title": "a-two" }
                ]
            },
            "chapters": [{ "__typename": "chapters" }, { "__typename": "chapters" }]
        })
    );

    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "mutation { insert_books_one(object: {title: \"new\", author_id: 2}) { \
           __typename title author { __typename name } } }",
    )
    .await;
    assert_eq!(
        data["insert_books_one"],
        serde_json::json!({
            "__typename": "books",
            "title": "new",
            "author": { "__typename": "authors", "name": "grace" }
        })
    );

    drop_schema(&pool, &schema).await;
}

/// `__typename` inside an aggregate is answered by the schema too, at every
/// level of it -- the aggregate object, each function's object, and the rows --
/// both at the root and through a relationship. A function's sub-fields were
/// all read as columns, so `max { __typename id }` became `max("__typename")`.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn typename_inside_aggregates_is_answered_by_the_schema() {
    let pool = connect().await;
    let schema = unique_schema_name("aggtypename");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ books_aggregate(order_by: [{id: asc}], limit: 1) { __typename \
           aggregate { __typename count max { __typename id } min { __typename } } \
           nodes { __typename id } } }",
    )
    .await;
    assert_eq!(
        data["books_aggregate"],
        serde_json::json!({
            "__typename": "books_aggregate",
            "aggregate": {
                "__typename": "books_aggregate_fields",
                "count": 1,
                "max": { "__typename": "books_max_fields", "id": 1 },
                "min": { "__typename": "books_min_fields" }
            },
            "nodes": [{ "__typename": "books", "id": 1 }]
        })
    );

    // Nothing but `__typename`, at each level: still objects, not nulls.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ books_aggregate { aggregate { __typename } } }",
    )
    .await;
    assert_eq!(
        data["books_aggregate"]["aggregate"],
        serde_json::json!({ "__typename": "books_aggregate_fields" })
    );

    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(where: {id: {_eq: 1}}) { __typename \
           books_aggregate(order_by: [{id: asc}]) { __typename \
             aggregate { __typename count max { __typename id } sum { __typename } } \
             nodes { __typename title } } } }",
    )
    .await;
    assert_eq!(
        data["authors"][0],
        serde_json::json!({
            "__typename": "authors",
            "books_aggregate": {
                "__typename": "books_aggregate",
                "aggregate": {
                    "__typename": "books_aggregate_fields",
                    "count": 2,
                    "max": { "__typename": "books_max_fields", "id": 2 },
                    "sum": { "__typename": "books_sum_fields" }
                },
                "nodes": [
                    { "__typename": "books", "title": "a-one" },
                    { "__typename": "books", "title": "a-two" }
                ]
            }
        })
    );

    drop_schema(&pool, &schema).await;
}

/// The query a Federation router sends when a field of another subgraph's
/// entity is reached through a relationship: the reference, with its
/// `__typename`, from inside an `_entities` answer.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn federation_entity_references_carry_their_typename() {
    let pool = connect().await;
    let schema = unique_schema_name("fedreftypename");
    create_related_schema(&pool, &schema).await;

    let names = shared_table_names(&schema, "authors");
    let state = build_federated_state_with_names(&pool, &schema, &names).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        r#"
        {
          _entities(representations: [{__typename: "authors", id: 2}]) {
            ... on authors { books { __typename id author { __typename id } } }
          }
        }
        "#,
    )
    .await;
    let books = &data["_entities"][0]["books"];
    assert_eq!(books.as_array().map(Vec::len), Some(1), "{data}");
    assert_eq!(
        books[0]["author"],
        serde_json::json!({"__typename": "authors", "id": 2})
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn nested_relationships_recurse_two_levels() {
    let pool = connect().await;
    let schema = unique_schema_name("relnest");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(order_by: [{id: asc}]) { id books { id chapters { id heading } } } }",
    )
    .await;

    let authors = data["authors"].as_array().unwrap();
    let ada_books = authors[0]["books"].as_array().unwrap();
    let first_book_chapters = ada_books[0]["chapters"].as_array().expect("chapters list");

    assert_eq!(first_book_chapters.len(), 2, "a-one has two chapters");
    assert_eq!(first_book_chapters[0]["heading"].as_str(), Some("a-one-c1"));
    assert_eq!(
        ada_books[1]["chapters"],
        serde_json::json!([]),
        "a-two has no chapters"
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn embedding_works_from_a_by_pk_query() {
    let pool = connect().await;
    let schema = unique_schema_name("relbypk");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors_by_pk(id: 1) { id name books { title } } }",
    )
    .await;

    let books = data["authors_by_pk"]["books"]
        .as_array()
        .expect("books list");
    assert_eq!(books.len(), 2);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn relationship_fields_appear_in_the_schema() {
    let pool = connect().await;
    let schema = unique_schema_name("relsdl");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    let sdl = state.schema.sdl();

    assert!(
        // The field carries the four arguments an embedded list takes, so it
        // is matched by what it answers with rather than by its whole
        // signature.
        sdl.contains("): [books!]!"),
        "expected a to-many relationship field on authors, got:\n{}",
        sdl.lines()
            .filter(|l| l.contains("book") || l.contains("author"))
            .collect::<Vec<_>>()
            .join("\n")
    );

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn a_predicate_over_an_aggregate_selects_by_how_many_children() {
    let pool = connect().await;
    let schema = unique_schema_name("aggpred");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;

    // ada has two books, grace one, lonely none.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(where: {books_aggregate: {count: {predicate: {_gte: 2}}}}, \
         order_by: [{id: asc}]) { id } }",
    )
    .await;
    assert_eq!(ids_of(&data, "authors"), vec![1]);

    // An author with no books at all: `count` over no rows is zero, which is
    // the whole reason this is a scalar subselect and not an EXISTS.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(where: {books_aggregate: {count: {predicate: {_eq: 0}}}}) { id } }",
    )
    .await;
    assert_eq!(ids_of(&data, "authors"), vec![3]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn a_predicate_over_an_aggregate_narrows_what_it_counts() {
    let pool = connect().await;
    let schema = unique_schema_name("aggpredf");
    create_related_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;

    // ada has two books and one of them is in print; grace's one is.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(where: {books_aggregate: {count: \
         {filter: {in_print: {_eq: true}}, predicate: {_gte: 2}}}}) { id } }",
    )
    .await;
    assert_eq!(ids_of(&data, "authors"), Vec::<i64>::new());

    // `bool_and` is true only where every book is in print, which is grace.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(where: {books_aggregate: \
         {bool_and: {arguments: in_print, predicate: {_eq: true}}}}, \
         order_by: [{id: asc}]) { id } }",
    )
    .await;
    assert_eq!(ids_of(&data, "authors"), vec![2]);

    // `bool_or` is true where any of them is: ada and grace, not lonely --
    // over no rows the fold is null, which is not true.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ authors(where: {books_aggregate: \
         {bool_or: {arguments: in_print, predicate: {_eq: true}}}}, \
         order_by: [{id: asc}]) { id } }",
    )
    .await;
    assert_eq!(ids_of(&data, "authors"), vec![1, 2]);

    drop_schema(&pool, &schema).await;
}

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn count_answers_each_alias_with_what_that_alias_asked_for() {
    let pool = connect().await;
    let schema = unique_schema_name("aggcount");
    create_widgets_schema(&pool, &schema).await;

    let state = build_state(&pool, &schema, None, false).await;
    // Four widgets, in two categories.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets_aggregate { aggregate { \
         count \
         named: count(columns: [category]) \
         categories: count(columns: [category], distinct: true) \
         } } }",
    )
    .await;
    let aggregate = &data["widgets_aggregate"]["aggregate"];
    assert_eq!(aggregate["count"], serde_json::json!(4));
    assert_eq!(aggregate["named"], serde_json::json!(4));
    assert_eq!(aggregate["categories"], serde_json::json!(2));

    drop_schema(&pool, &schema).await;
}

/// A ceiling bounds the page and not the count.
///
/// `PGRST_MAX_ROWS` and a permission's `limit` both exist to bound how many
/// rows travel; neither is an answer to "how many are there". Hasura's own
/// corpus proves this for the permission half -- a role limited to one row of
/// `article` still counts three -- and the configured ceiling is treated the
/// same way, which is what this pins.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn a_ceiling_bounds_the_page_and_not_the_count() {
    let pool = connect().await;
    let schema = unique_schema_name("aggceiling");
    create_widgets_schema(&pool, &schema).await;

    // Four widgets, and a server that will send at most two of them.
    let state = build_state(&pool, &schema, Some(2), false).await;
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets_aggregate(order_by: [{id: asc}])          { aggregate { count } nodes { id } } }",
    )
    .await;

    assert_eq!(
        data["widgets_aggregate"]["aggregate"]["count"],
        serde_json::json!(4),
        "the count is of what is there"
    );
    assert_eq!(
        data["widgets_aggregate"]["nodes"].as_array().map(Vec::len),
        Some(2),
        "the page is of what may be sent"
    );

    // A limit the request asked for is a different thing, and does reach the
    // count: `widgets_aggregate(limit: 3)` is a question about three rows.
    let data = execute_ok(
        &state,
        &pool,
        &schema,
        "{ widgets_aggregate(order_by: [{id: asc}], limit: 3) { aggregate { count } } }",
    )
    .await;
    assert_eq!(
        data["widgets_aggregate"]["aggregate"]["count"],
        serde_json::json!(3)
    );

    drop_schema(&pool, &schema).await;
}

/// A live query answers now, and answers again when what it answered stops
/// being true. This drives the refresh rather than the notifications, because
/// the test schema carries no trigger -- which is exactly the case the refresh
/// is there for.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn a_subscription_is_a_live_query() {
    use futures::StreamExt;

    let pool = connect().await;
    let schema = unique_schema_name("livequery");
    create_widgets_schema(&pool, &schema).await;

    let schemas = vec![schema.clone()];
    let cache = SchemaCache::load(&pool, &schemas)
        .await
        .expect("failed to load schema cache");
    let mut config = SchemaConfig::default();
    config.exposed_schemas = schemas.clone();
    config.enable_mutations = true;
    config.enable_subscriptions = true;
    config.subscription_refresh_seconds = 1;
    let state = Arc::new(
        GraphQLState::new(pool.clone(), Arc::new(cache), config)
            .expect("failed to build GraphQL schema"),
    );

    let cache = SchemaCache::load(&pool, &schemas)
        .await
        .expect("failed to load schema cache");
    let ctx = GraphQLContext::new(
        pool.clone(),
        SchemaCacheRef::from_static(cache),
        AuthResult {
            role: TEST_ROLE.to_string(),
            claims: HashMap::new(),
        },
    );
    let request = Request::new(
        "subscription { widgets(where: {category: {_eq: \"books\"}}, order_by: [{id: asc}]) \
         { id name } }",
    )
    .data(ctx)
    .data(pool.clone());

    let mut stream = state.schema.execute_stream(request);

    // The answer now, before anything has changed: a live query has no window
    // in which the client knows nothing.
    let first = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
        .await
        .expect("the first answer did not arrive")
        .expect("the stream ended");
    assert!(first.errors.is_empty(), "{:?}", first.errors);
    let names = |response: &async_graphql::Response| -> Vec<String> {
        serde_json::to_value(&response.data).expect("serialisable")["widgets"]
            .as_array()
            .expect("a list of widgets")
            .iter()
            .map(|row| row["name"].as_str().unwrap_or_default().to_string())
            .collect()
    };
    assert_eq!(names(&first), vec!["alpha", "charlie"]);

    // A row the subscription does not select is a wake and not a message: the
    // refresh reads the query again, sees the same answer, and sends nothing.
    pool.execute(AssertSqlSafe(format!(
        "INSERT INTO {}.widgets (name, category, price, stock) \
             VALUES ('echo', 'tools', 1, 1)",
        schema
    )))
    .await
    .expect("insert failed");
    let quiet = tokio::time::timeout(std::time::Duration::from_secs(3), stream.next()).await;
    assert!(
        quiet.is_err(),
        "a change outside the subscription was sent to it: {:?}",
        quiet.ok().flatten().map(|r| serde_json::to_value(&r.data))
    );

    // One it does select is answered again.
    pool.execute(AssertSqlSafe(format!(
        "INSERT INTO {}.widgets (name, category, price, stock) \
             VALUES ('foxtrot', 'books', 2, 2)",
        schema
    )))
    .await
    .expect("insert failed");

    let next = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
        .await
        .expect("the changed answer did not arrive")
        .expect("the stream ended");
    assert!(next.errors.is_empty(), "{:?}", next.errors);
    assert_eq!(names(&next), vec!["alpha", "charlie", "foxtrot"]);

    drop(stream);
    drop_schema(&pool, &schema).await;
}

/// A permission's filter is the server's own predicate, not the caller's.
///
/// Two things this pins, both of which used to make such a permission refuse
/// itself rather than apply:
///
///   * it may follow a relationship to a table the role has no permission on
///     at all -- consulting a table is not exposing it, which is the point of
///     writing the permission that way; and
///   * it may compare two columns, either on one table or across the
///     relationship it followed, with `["$", "name"]` meaning the table the
///     permission is written on.
///
/// Both are checked against real SQL, because both are about what the
/// generated statement refers to and nothing else can see that.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn a_permission_may_consult_a_table_the_role_cannot_read() {
    let pool = connect().await;
    let schema = unique_schema_name("permfilter");
    create_permission_schema(&pool, &schema).await;

    // `listener` reads tracks, and only those whose artist is the caller's and
    // whose artist outranks what the track asks for. It is granted nothing on
    // `artist`.
    let names = format!(
        r#"{{"tables": {{
             "{schema}.track": {{"permissions": {{"listener": {{"select": {{
                "columns": "*",
                "filter": {{"_and": [
                   {{"artist": {{"id": "X-Hasura-Artist-Id"}}}},
                   {{"artist": {{"rank": {{"_cgte": ["$", "min_rank"]}}}}}},
                   {{"price": {{"_cgte": "floor"}}}}
                ]}}
             }}}}}}}},
             "{schema}.playlist": {{"permissions": {{"listener": {{"select": {{
                "columns": "*", "filter": {{}}
             }}}}}}}}
           }}}}"#
    );
    let state = build_state_with_names(&pool, &schema, &names).await;

    let data = execute_as(
        &state,
        &pool,
        &schema,
        "{ track(order_by: [{id: asc}]) { id } }",
    )
    .await;
    assert_eq!(
        ids_of(&data, "track"),
        vec![1],
        "artist 1, rank at least the track's floor, and price at or above it"
    );

    // And the table the filter consulted is still not one this role can read.
    let refused = execute_as_err(&state, &pool, &schema, "{ artist { id } }").await;
    assert!(
        refused.contains("artist"),
        "the consulted table stays hidden -- got: {}",
        refused
    );

    // Reached through somebody else's filter, `$` still names the table the
    // permission is written on. Playlist 2 holds only track 5, whose artist
    // does not outrank *the track's* `min_rank` -- but does outrank the
    // playlist's, so a `$` bound to the query's root would let it through.
    let data = execute_as(
        &state,
        &pool,
        &schema,
        "{ playlist(where: {tracks: {id: {_gt: 0}}}, order_by: [{id: asc}]) { id } }",
    )
    .await;
    assert_eq!(
        ids_of(&data, "playlist"),
        vec![1],
        "the far side's permission is read against its own table"
    );

    drop_schema(&pool, &schema).await;
}

/// Verified JWT claims must reach the same transaction as the GraphQL query.
///
/// Role switching alone is insufficient for policies that distinguish users
/// sharing one database role. Individual claim settings are scoped to the
/// transaction so a pooled connection cannot retain the prior identity.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn verified_jwt_claims_are_available_to_graphql_rls_without_leaking() {
    let pool = connect().await;
    let schema = unique_schema_name("jwtclaims");
    let reader_role = format!("{}_reader", schema);

    drop_schema(&pool, &schema).await;
    pool.execute(AssertSqlSafe(format!(
        "DROP ROLE IF EXISTS {}",
        reader_role
    )))
    .await
    .expect("failed to drop stale reader role");
    pool.execute(AssertSqlSafe(format!(
        "CREATE ROLE {} NOLOGIN",
        reader_role
    )))
    .await
    .expect("failed to create reader role");
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("failed to create schema");
    pool.execute(AssertSqlSafe(format!(
        "CREATE TABLE {schema}.documents (
                 id integer PRIMARY KEY,
                 owner_id text NOT NULL,
                 title text NOT NULL
             );
             INSERT INTO {schema}.documents (id, owner_id, title) VALUES
                 (1, 'user-one', 'first'),
                 (2, 'user-two', 'second');
             ALTER TABLE {schema}.documents ENABLE ROW LEVEL SECURITY;
             CREATE POLICY documents_owner_read ON {schema}.documents
                 FOR SELECT TO {reader_role}
                 USING (
                     owner_id = NULLIF(
                         current_setting('request.jwt.claims.user_id', true), ''
                     )
                 );
             GRANT USAGE ON SCHEMA {schema} TO {reader_role};
             GRANT SELECT ON {schema}.documents TO {reader_role};"
    )))
    .await
    .expect("failed to create JWT RLS fixture");

    let state = build_state(&pool, &schema, None, false).await;
    let query = "{ documents(order_by: [{id: asc}]) { id owner_id } }";
    let auth_for = |user_id: Option<&str>| {
        let mut claims = HashMap::new();
        claims.insert("role".to_string(), serde_json::json!(reader_role.clone()));
        if let Some(user_id) = user_id {
            claims.insert("user_id".to_string(), serde_json::json!(user_id));
        }
        AuthResult {
            role: reader_role.clone(),
            claims,
        }
    };

    let first = execute_with_auth(&state, &pool, &schema, query, auth_for(Some("user-one"))).await;
    let second = execute_with_auth(&state, &pool, &schema, query, auth_for(Some("user-two"))).await;
    let claimless = execute_with_auth(&state, &pool, &schema, query, auth_for(None)).await;

    drop(state);
    drop_schema(&pool, &schema).await;
    pool.execute(AssertSqlSafe(format!("DROP ROLE {}", reader_role)))
        .await
        .expect("failed to drop reader role");

    let successful_data = |response: async_graphql::Response| {
        assert!(
            response.errors.is_empty(),
            "expected no GraphQL errors -- got: {:?}",
            response.errors
        );
        serde_json::to_value(response.data).expect("data was not serialisable")
    };
    assert_eq!(ids_of(&successful_data(first), "documents"), vec![1]);
    assert_eq!(ids_of(&successful_data(second), "documents"), vec![2]);
    assert!(
        ids_of(&successful_data(claimless), "documents").is_empty(),
        "a later claimless request inherited a pooled connection's JWT identity"
    );
}

/// Two tables, one key between them, and the columns a comparison needs.
async fn create_permission_schema(pool: &PgPool, schema: &str) {
    drop_schema(pool, schema).await;
    pool.execute(AssertSqlSafe(format!("CREATE SCHEMA {}", schema)))
        .await
        .expect("failed to create schema");
    pool.execute(AssertSqlSafe(format!(
        "CREATE TABLE {schema}.artist (
                 id integer PRIMARY KEY,
                 rank integer NOT NULL
             );
             CREATE TABLE {schema}.playlist (
                 id integer PRIMARY KEY,
                 min_rank integer NOT NULL
             );
             CREATE TABLE {schema}.track (
                 id integer PRIMARY KEY,
                 artist_id integer NOT NULL REFERENCES {schema}.artist(id),
                 playlist_id integer NOT NULL REFERENCES {schema}.playlist(id),
                 min_rank integer NOT NULL,
                 price numeric NOT NULL,
                 floor numeric NOT NULL
             );
             INSERT INTO {schema}.artist (id, rank) VALUES (1, 5), (2, 5);
             -- The second playlist asks for a rank every artist clears,
             -- while the only track on it asks for one none of them does.
             -- That is what tells the two roots apart: read against the
             -- playlist the track passes, read against the track it does not.
             INSERT INTO {schema}.playlist (id, min_rank) VALUES (1, 3), (2, 1);
             INSERT INTO {schema}.track
                 (id, artist_id, playlist_id, min_rank, price, floor) VALUES
                 (1, 1, 1, 3, 10, 10),
                 (2, 1, 1, 9, 10, 10),
                 (3, 1, 1, 3,  9, 10),
                 (4, 2, 1, 3, 10, 10),
                 (5, 1, 2, 9, 10, 10);"
    )))
    .await
    .expect("failed to create the permission fixture");
}

/// Build state over the schema with a permission document in play.
async fn build_state_with_names(pool: &PgPool, schema: &str, names: &str) -> Arc<GraphQLState> {
    let schemas = vec![schema.to_string()];
    let cache = SchemaCache::load(pool, &schemas)
        .await
        .expect("failed to load schema cache");
    let mut config = SchemaConfig::default();
    config.exposed_schemas = schemas.clone();
    config.enable_mutations = true;
    config.names = postrust_graphql::names::NameOverrides::parse(names).expect("the names parse");
    Arc::new(
        GraphQLState::new(pool.clone(), Arc::new(cache), config)
            .expect("failed to build GraphQL schema"),
    )
}

/// Execute as the `listener` role, carrying the session the filter reads.
async fn execute_as_response(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
) -> async_graphql::Response {
    let cache = SchemaCache::load(pool, &[schema.to_string()])
        .await
        .expect("failed to load schema cache");
    let mut session = HashMap::new();
    session.insert("artist_id".to_string(), "1".to_string());
    let ctx = GraphQLContext::new(
        pool.clone(),
        SchemaCacheRef::from_static(cache),
        AuthResult {
            role: TEST_ROLE.to_string(),
            claims: HashMap::new(),
        },
    )
    .with_session(session)
    .with_identity(Some("listener".to_string()), false);

    let schema_for_role = state
        .schema_for(Some("listener"), false)
        .expect("the role has a schema");
    schema_for_role
        .execute(Request::new(query).data(ctx).data(pool.clone()))
        .await
}

async fn execute_as(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
) -> serde_json::Value {
    let response = execute_as_response(state, pool, schema, query).await;
    assert!(
        response.errors.is_empty(),
        "expected no GraphQL errors for {} -- got: {:?}",
        query,
        response.errors
    );
    serde_json::to_value(&response.data).expect("data was not serialisable")
}

async fn execute_as_err(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    schema: &str,
    query: &str,
) -> String {
    let response = execute_as_response(state, pool, schema, query).await;
    assert!(
        !response.errors.is_empty(),
        "expected a GraphQL error for {} -- got data: {:?}",
        query,
        response.data
    );
    response
        .errors
        .iter()
        .map(|e| e.message.clone())
        .collect::<Vec<_>>()
        .join("; ")
}

/// Execute a GraphQL document on `pool` in a context of the caller's making,
/// settling its writes as the server does.
async fn execute_in(
    state: &Arc<GraphQLState>,
    pool: &PgPool,
    ctx: GraphQLContext,
    query: &str,
) -> async_graphql::Response {
    let write: postrust_graphql::context::SharedWrite = Default::default();
    let ctx = ctx.with_write(Arc::clone(&write));
    let response = state
        .schema
        .execute(Request::new(query).data(ctx).data(pool.clone()))
        .await;
    if let Some(tx) = write.lock().await.take() {
        if response.errors.is_empty() {
            tx.commit().await.expect("the write could not be committed");
        } else {
            tx.rollback().await.expect("the write could not be undone");
        }
    }
    response
}

/// A role's own `statement_timeout` holds for a GraphQL query, as it does for
/// a REST request: the transaction is set up by the same `Session`. It did
/// not before -- GraphQL set the role and the session variables and nothing
/// else -- so a role an operator had limited could run as long as it liked
/// here.
///
/// The same query, under the same configuration, as a role with no setting
/// of its own, is the control: what stops the query is the role's setting
/// and not the query.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn a_roles_statement_timeout_applies_to_a_graphql_query() {
    let pool = connect().await;
    let schema = unique_schema_name("roletimeout");
    let limited = format!("{schema}_limited");

    drop_schema(&pool, &schema).await;
    pool.execute(AssertSqlSafe(format!("DROP ROLE IF EXISTS {limited}")))
        .await
        .expect("failed to drop a stale role");
    pool.execute(AssertSqlSafe(format!(
        "CREATE ROLE {limited} NOLOGIN;
             CREATE SCHEMA {schema};
             CREATE TABLE {schema}.ticks (id integer PRIMARY KEY);
             INSERT INTO {schema}.ticks VALUES (1);
             -- Half a second of sleep, once, whoever reads it.
             CREATE VIEW {schema}.slow_ticks AS
                 SELECT id FROM {schema}.ticks WHERE (SELECT true FROM pg_sleep(0.5));
             GRANT USAGE ON SCHEMA {schema} TO {limited};
             GRANT SELECT ON ALL TABLES IN SCHEMA {schema} TO {limited};"
    )))
    .await
    .expect("failed to create the timeout fixture");

    let state = build_state(&pool, &schema, None, false).await;
    let mut config = postrust_core::AppConfig::default();
    let mut settings = postrust_core::config::RoleSettings::default();
    settings.statement_timeout = Some(50);
    config.role_settings.insert(limited.clone(), settings);
    let config = Arc::new(config);

    let run_as = |role: &str| {
        let role = role.to_string();
        let state = Arc::clone(&state);
        let pool = pool.clone();
        let config = Arc::clone(&config);
        let schema = schema.clone();
        async move {
            let cache = SchemaCache::load(&pool, &[schema])
                .await
                .expect("failed to load schema cache");
            let ctx = GraphQLContext::new(
                pool.clone(),
                SchemaCacheRef::from_static(cache),
                AuthResult {
                    role,
                    claims: HashMap::new(),
                },
            )
            .with_config(config);
            execute_in(&state, &pool, ctx, "{ slow_ticks { id } }").await
        }
    };
    let limited_response = run_as(&limited).await;
    let unlimited_response = run_as(TEST_ROLE).await;

    drop(state);
    drop_schema(&pool, &schema).await;
    pool.execute(AssertSqlSafe(format!("DROP ROLE {limited}")))
        .await
        .expect("failed to drop the role");

    let messages = limited_response
        .errors
        .iter()
        .map(|e| e.message.clone())
        .collect::<Vec<_>>()
        .join("; ");
    assert!(
        messages.contains("statement timeout"),
        "the role's statement_timeout did not stop the query: {messages:?}, data {:?}",
        limited_response.data
    );
    assert!(
        unlimited_response.errors.is_empty(),
        "a role with no timeout of its own was stopped: {:?}",
        unlimited_response.errors
    );
    assert_eq!(
        serde_json::to_value(unlimited_response.data).unwrap(),
        serde_json::json!({"slow_ticks": [{"id": 1}]})
    );
}

/// Through a `NOINHERIT` authenticator -- the setup PostgREST recommends,
/// where the login role holds nothing until it becomes another -- a query and
/// a mutation each run as the token's database role, never as the Hasura role
/// the caller speaks as, and both a policy and a function read the `hasura.*`
/// session variables and the verified claims.
///
/// The session now sets the role inside its one `set_config` statement,
/// before the claims and the `hasura.*` variables rather than after them as
/// `SET LOCAL ROLE` used to come. This is what says that order still works
/// for a role that may do nothing but become another: every setting after
/// the role is one any role may set on itself. And once the operation is
/// over, the authenticator's pooled connection is its own again.
#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn the_session_reaches_policies_and_functions_through_a_noinherit_authenticator() {
    use sqlx::postgres::PgConnectOptions;
    use std::str::FromStr;

    let admin = connect().await;
    let schema = unique_schema_name("gqlsession");
    let reader = format!("{schema}_reader");
    let authenticator = format!("{schema}_auth");
    let password = "postrust_test_gql_session";

    drop_schema(&admin, &schema).await;
    for role in [&authenticator, &reader] {
        admin
            .execute(AssertSqlSafe(format!("DROP ROLE IF EXISTS {role}")))
            .await
            .expect("failed to drop a stale role");
    }
    admin
        .execute(AssertSqlSafe(format!(
            "CREATE ROLE {reader} NOLOGIN;
                 CREATE ROLE {authenticator} LOGIN NOINHERIT PASSWORD '{password}';
                 GRANT {reader} TO {authenticator};
                 CREATE SCHEMA {schema};
                 CREATE TABLE {schema}.notes (
                     id integer PRIMARY KEY,
                     owner text NOT NULL,
                     body text NOT NULL
                 );
                 INSERT INTO {schema}.notes VALUES
                     (1, 'user-one', 'mine'),
                     (2, 'user-two', 'theirs');
                 ALTER TABLE {schema}.notes ENABLE ROW LEVEL SECURITY;
                 CREATE POLICY notes_owner ON {schema}.notes TO {reader}
                     USING (owner = current_setting('hasura.user_id', true))
                     WITH CHECK (owner = current_setting('hasura.user_id', true));
                 -- A computed field, so the function runs inside the
                 -- transaction the session set up.
                 CREATE FUNCTION {schema}.seen_as(note_row {schema}.notes)
                 RETURNS text LANGUAGE sql STABLE AS $$
                     SELECT concat_ws('|',
                         current_user,
                         current_setting('hasura.user_id', true),
                         current_setting('hasura.role', true),
                         current_setting('hasura.session', true)::json->>'x-hasura-user-id',
                         current_setting('request.jwt.claims', true)::json->>'sub',
                         current_setting('request.jwt.claims.sub', true))
                 $$;
                 GRANT USAGE ON SCHEMA {schema} TO {reader};
                 GRANT SELECT, INSERT ON {schema}.notes TO {reader};"
        )))
        .await
        .expect("failed to create the session fixture");

    // The schema is read by the superuser; the operations go through the
    // authenticator, on one connection so the last check below is on the
    // connection the operations used.
    let state = build_state(&admin, &schema, None, false).await;
    let options = PgConnectOptions::from_str(&database_url())
        .expect("DATABASE_URL parses")
        .username(&authenticator)
        .password(password);
    let through = PgPoolOptions::new()
        .max_connections(1)
        .connect_with(options)
        .await
        .expect("the authenticator could not log in");

    let context = || {
        let schema = schema.clone();
        let admin = admin.clone();
        let through = through.clone();
        let reader = reader.clone();
        async move {
            let cache = SchemaCache::load(&admin, &[schema])
                .await
                .expect("failed to load schema cache");
            let mut claims = HashMap::new();
            claims.insert("role".to_string(), serde_json::json!(reader));
            claims.insert("sub".to_string(), serde_json::json!("alice"));
            let mut session = HashMap::new();
            session.insert("user_id".to_string(), "user-one".to_string());
            GraphQLContext::new(
                through,
                SchemaCacheRef::from_static(cache),
                AuthResult {
                    role: reader,
                    claims,
                },
            )
            .with_session(session)
            // An administrator asking to be answered with the session
            // above: `admin` is a Hasura role, not a database one, and has to
            // reach the function as `hasura.role` while the transaction runs
            // as the token's role.
            .with_identity(Some("admin".to_string()), true)
        }
    };

    let read = execute_in(
        &state,
        &through,
        context().await,
        "{ notes(order_by: [{id: asc}]) { id seen_as } }",
    )
    .await;
    let written = execute_in(
        &state,
        &through,
        context().await,
        "mutation { insert_notes_one(object: {id: 3, owner: \"user-one\", body: \"new\"}) \
         { id seen_as } }",
    )
    .await;
    let refused = execute_in(
        &state,
        &through,
        context().await,
        "mutation { insert_notes_one(object: {id: 4, owner: \"user-two\", body: \"no\"}) \
         { id } }",
    )
    .await;
    let (user, left_behind): (String, Option<String>) = sqlx::query_as(
        "SELECT current_user::text, nullif(current_setting('hasura.user_id', true), '')",
    )
    .fetch_one(&through)
    .await
    .expect("the authenticator's connection answers");
    let stored: Vec<i32> = sqlx::query_scalar(AssertSqlSafe(format!(
        "SELECT id FROM {schema}.notes ORDER BY id"
    )))
    .fetch_all(&admin)
    .await
    .expect("the notes can be read");

    through.close().await;
    drop(state);
    drop_schema(&admin, &schema).await;
    for role in [&authenticator, &reader] {
        admin
            .execute(AssertSqlSafe(format!("DROP ROLE {role}")))
            .await
            .expect("failed to drop a role");
    }

    let seen = format!("{reader}|user-one|admin|user-one|alice|alice");
    assert!(read.errors.is_empty(), "{:?}", read.errors);
    assert_eq!(
        serde_json::to_value(read.data).unwrap(),
        serde_json::json!({"notes": [{"id": 1, "seen_as": seen}]}),
        "the policy narrows to the session's user, and the function sees the database \
         role, the hasura.* variables and the claims"
    );
    assert!(written.errors.is_empty(), "{:?}", written.errors);
    assert_eq!(
        serde_json::to_value(written.data).unwrap(),
        serde_json::json!({"insert_notes_one": {"id": 3, "seen_as": seen}}),
        "a mutation's shared transaction is set up the same way"
    );
    let refusal = refused
        .errors
        .iter()
        .map(|e| e.message.clone())
        .collect::<Vec<_>>()
        .join("; ");
    assert!(
        refusal.contains("row-level security"),
        "the policy's check reads hasura.user_id inside the write: {refusal:?}"
    );
    assert_eq!(stored, vec![1, 2, 3], "the refused row was not written");
    assert_eq!(
        (user.as_str(), left_behind),
        (authenticator.as_str(), None),
        "the role and the session variables ended with their transactions"
    );
}