fraiseql-db 2.16.0

//! Unit tests for the PostgreSQL adapter (no live database required).

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

use fraiseql_error::FraiseQLError;

use super::{
    HnswIterativeScan, IvfflatIterativeScan, PoolPrewarmConfig, PostgresAdapter, PostgresTlsConfig,
    SearchPath, VectorScanConfig, build_projection_select_sql, build_where_select_sql,
    build_where_select_sql_ordered, compose_startup_options,
};

// ── build_where_select_sql ─────────────────────────────────────────────────

#[test]
fn test_build_where_select_sql_no_clause() {
    let (sql, params) = build_where_select_sql("v_user", None, None, None).unwrap();
    assert_eq!(sql, r#"SELECT data FROM "v_user""#);
    assert!(params.is_empty());
}

#[test]
fn test_build_where_select_sql_with_limit_offset() {
    let (sql, params) = build_where_select_sql("v_order", None, Some(10), Some(20)).unwrap();
    // LIMIT takes $1, OFFSET takes $2.
    assert!(sql.contains("LIMIT $1"), "expected LIMIT $1 in: {sql}");
    assert!(sql.contains("OFFSET $2"), "expected OFFSET $2 in: {sql}");
    assert_eq!(params.len(), 2, "expected 2 params (limit + offset)");
}

#[test]
fn pool_prewarm_config_carries_all_fields() {
    let cfg = PoolPrewarmConfig {
        min_size:            5,
        max_size:            20,
        timeout_secs:        Some(30),
        search_path:         None,
        tls:                 PostgresTlsConfig::default(),
        read_replicas:       None,
        max_streaming_reads: None,
        vector_scan:         VectorScanConfig::default(),
    };
    assert_eq!(cfg.min_size, 5);
    assert_eq!(cfg.max_size, 20);
    assert_eq!(cfg.timeout_secs, Some(30));
}

#[test]
fn pool_prewarm_config_no_timeout_is_none() {
    let cfg = PoolPrewarmConfig {
        min_size:            0,
        max_size:            10,
        timeout_secs:        None,
        search_path:         None,
        tls:                 PostgresTlsConfig::default(),
        read_replicas:       None,
        max_streaming_reads: None,
        vector_scan:         VectorScanConfig::default(),
    };
    assert!(cfg.timeout_secs.is_none());
}

#[test]
fn pool_prewarm_config_min_zero_is_valid() {
    let cfg = PoolPrewarmConfig {
        min_size:            0,
        max_size:            5,
        timeout_secs:        None,
        search_path:         None,
        tls:                 PostgresTlsConfig::default(),
        read_replicas:       None,
        max_streaming_reads: None,
        vector_scan:         VectorScanConfig::default(),
    };
    assert_eq!(cfg.min_size, 0);
    assert_eq!(cfg.max_size, 5);
}

#[test]
fn pool_prewarm_config_min_equals_max_is_valid() {
    let cfg = PoolPrewarmConfig {
        min_size:            10,
        max_size:            10,
        timeout_secs:        Some(60),
        search_path:         None,
        tls:                 PostgresTlsConfig::default(),
        read_replicas:       None,
        max_streaming_reads: None,
        vector_scan:         VectorScanConfig::default(),
    };
    assert_eq!(cfg.min_size, cfg.max_size);
}

// ── SearchPath / startup options (#809) ────────────────────────────────────

#[test]
fn search_path_joins_schemas_in_order() {
    let path = SearchPath::new(["tenant_acme", "public"]).unwrap();
    assert_eq!(path.as_str(), "tenant_acme,public");
}

#[test]
fn search_path_rejects_non_identifiers() {
    // Rejected, never escaped: the value is emitted into a startup option where
    // whitespace is a separator and quoting has no meaning.
    for bad in [
        "public; DROP SCHEMA x",
        "a b",
        "\"quoted\"",
        "1leading",
        "",
        "with-dash",
    ] {
        assert!(SearchPath::new([bad]).is_err(), "`{bad}` must be refused");
    }
}

#[test]
fn search_path_rejects_an_empty_list() {
    assert!(SearchPath::new(Vec::<&str>::new()).is_err());
}

#[test]
fn search_path_rejects_an_over_long_identifier() {
    assert!(SearchPath::new(["a".repeat(64)]).is_err());
    assert!(SearchPath::new(["a".repeat(63)]).is_ok());
}

/// The vector settings that contribute nothing, so a test about the search path
/// is about the search path.
const NO_VECTOR_SCAN: VectorScanConfig = VectorScanConfig {
    hnsw:      HnswIterativeScan::Off,
    ivfflat:   IvfflatIterativeScan::Off,
    ef_search: None,
};

#[test]
fn startup_options_are_composed_when_the_url_carries_none() {
    let path = SearchPath::new(["tenant_acme", "public"]).unwrap();
    assert_eq!(
        compose_startup_options("postgres://u:p@h:5432/db", Some(&path), NO_VECTOR_SCAN),
        Some("-c search_path=tenant_acme,public".to_string())
    );
}

/// deadpool's struct-level `options` field *overrides* whatever the URL carried, so
/// composing by assignment alone would silently drop an operator's own `-c` settings.
#[test]
fn startup_options_preserve_operator_supplied_options() {
    let path = SearchPath::new(["tenant_acme", "public"]).unwrap();
    let composed = compose_startup_options(
        "postgres://u:p@h:5432/db?options=-c%20statement_timeout%3D5000",
        Some(&path),
        NO_VECTOR_SCAN,
    );
    assert_eq!(
        composed,
        Some("-c statement_timeout=5000 -c search_path=tenant_acme,public".to_string())
    );
}

/// An unparseable connection string must not panic here — the pool builder that
/// follows produces the real, actionable error.
#[test]
fn startup_options_tolerate_an_unparseable_url() {
    let path = SearchPath::new(["tenant_acme"]).unwrap();
    assert_eq!(
        compose_startup_options("not-a-postgres-url", Some(&path), NO_VECTOR_SCAN),
        Some("-c search_path=tenant_acme".to_string())
    );
}

// ── #1116: pgvector scan settings ride the same startup packet ──────────────

/// The default settings reach the startup packet on a pool with no search path.
///
/// Before #1116 `cfg.options` was assigned **only** inside `if let Some(path) =
/// search_path`, so a single-tenant server — which passes `search_path: None` —
/// had no route to the startup packet at all. This is the case that had none.
#[test]
fn vector_scan_settings_ride_the_startup_packet_without_a_search_path() {
    assert_eq!(
        compose_startup_options("postgres://u:p@h:5432/db", None, VectorScanConfig::default()),
        Some(
            "-c hnsw.iterative_scan=strict_order -c ivfflat.iterative_scan=relaxed_order"
                .to_string()
        )
    );
}

/// The default is `strict_order` for HNSW, and `relaxed_order` for `IVFFlat`
/// **because pgvector defines no `strict_order` for `IVFFlat`** — measured against
/// pgvector 0.8.6:
///
/// ```text
/// WARNING:  invalid value for parameter "ivfflat.iterative_scan": "strict_order"
/// HINT:  Available values: off, relaxed_order.
/// ```
///
/// after which the setting falls back to `off`. The types are separate so that
/// value is not expressible; this pins the defaults that follow from it.
#[test]
fn the_two_indexes_default_to_the_strongest_guarantee_each_defines() {
    assert_eq!(HnswIterativeScan::default(), HnswIterativeScan::StrictOrder);
    assert_eq!(IvfflatIterativeScan::default(), IvfflatIterativeScan::RelaxedOrder);
}

/// `Off` writes **nothing**, rather than `-c hnsw.iterative_scan=off`.
///
/// The difference is not cosmetic: an explicit `off` in the startup packet would
/// override a database-level `ALTER DATABASE app SET hnsw.iterative_scan =
/// relaxed_order`, which is the workaround operators were told to use before
/// #1116. "Turn FraiseQL's setting off" must mean "leave whatever is there
/// alone", not "force it off".
#[test]
fn turning_the_settings_off_writes_nothing_rather_than_forcing_off() {
    assert_eq!(NO_VECTOR_SCAN.startup_fragments(), Vec::<String>::new());
    assert_eq!(compose_startup_options("postgres://u:p@h/db", None, NO_VECTOR_SCAN), None);
}

/// Each setting is independent, and each renders the value pgvector spells.
#[test]
fn each_index_contributes_its_own_fragment() {
    let hnsw_only = VectorScanConfig {
        hnsw:      HnswIterativeScan::RelaxedOrder,
        ivfflat:   IvfflatIterativeScan::Off,
        ef_search: None,
    };
    assert_eq!(hnsw_only.startup_fragments(), vec!["-c hnsw.iterative_scan=relaxed_order"]);

    let ivfflat_only = VectorScanConfig {
        hnsw:      HnswIterativeScan::Off,
        ivfflat:   IvfflatIterativeScan::RelaxedOrder,
        ef_search: None,
    };
    assert_eq!(
        ivfflat_only.startup_fragments(),
        vec!["-c ivfflat.iterative_scan=relaxed_order"]
    );

    let ef_only = VectorScanConfig {
        hnsw:      HnswIterativeScan::Off,
        ivfflat:   IvfflatIterativeScan::Off,
        ef_search: Some(200),
    };
    assert_eq!(ef_only.startup_fragments(), vec!["-c hnsw.ef_search=200"]);
}

/// `ef_search` is unset by default, and setting it does not disturb the others.
///
/// Unset rather than raised because it is a latency/recall trade paid by every
/// search on the deployment — unlike `iterative_scan`, which the benchmark
/// measured as free on an unfiltered one. It exists because `iterative_scan`
/// alone does **not** guarantee `k` rows: measured over three independently
/// generated 20 000-row / 64-dimension datasets with a 1% filter, it returned
/// 10, 5 and 4 rows, while `ef_search = 1000` returned 10 every time and
/// `max_scan_tuples` raised 100× changed nothing.
#[test]
fn ef_search_is_unset_by_default_and_composes_with_the_scan_modes() {
    assert_eq!(VectorScanConfig::default().ef_search, None);
    assert!(
        !VectorScanConfig::default()
            .startup_fragments()
            .iter()
            .any(|f| f.contains("ef_search")),
        "the default must not choose an ef_search for the operator"
    );

    let all = VectorScanConfig {
        ef_search: Some(400),
        ..VectorScanConfig::default()
    };
    assert_eq!(
        all.startup_fragments(),
        vec![
            "-c hnsw.iterative_scan=strict_order",
            "-c ivfflat.iterative_scan=relaxed_order",
            "-c hnsw.ef_search=400",
        ]
    );
}

/// All three fragment sources compose, in a stable order, without dropping any.
///
/// The operator's own `-c` first (it came from the URL), then the search path,
/// then the vector settings — one packet carrying three independent decisions.
#[test]
fn the_search_path_the_operators_options_and_the_vector_settings_all_survive() {
    let path = SearchPath::new(["tenant_acme"]).unwrap();
    let composed = compose_startup_options(
        "postgres://u:p@h:5432/db?options=-c%20statement_timeout%3D5000",
        Some(&path),
        VectorScanConfig::default(),
    );
    let expected = "-c statement_timeout=5000 \
                    -c search_path=tenant_acme \
                    -c hnsw.iterative_scan=strict_order \
                    -c ivfflat.iterative_scan=relaxed_order";
    assert_eq!(composed.as_deref(), Some(expected));
}

// ── EP-5: Connection pool failure paths ───────────────────────────────────

#[tokio::test]
async fn test_new_with_malformed_url_returns_connection_pool_error() {
    // A completely unparseable URL causes deadpool-postgres to fail immediately
    // at pool creation or the initial `pool.get()`, both mapped to ConnectionPool.
    let result = PostgresAdapter::new("not-a-postgres-url").await;
    assert!(result.is_err(), "expected error for malformed URL");
    let err = result.expect_err("error confirmed above");
    assert!(
        matches!(err, FraiseQLError::ConnectionPool { .. }),
        "expected ConnectionPool error for malformed URL, got: {err:?}"
    );
}

#[tokio::test]
async fn test_with_pool_size_malformed_url_returns_connection_pool_error() {
    let result = PostgresAdapter::with_pool_size("://bad-url", 1).await;
    assert!(result.is_err(), "expected error for bad URL");
    let err = result.expect_err("error confirmed above");
    assert!(
        matches!(err, FraiseQLError::ConnectionPool { .. }),
        "expected ConnectionPool error for bad URL with custom pool size, got: {err:?}"
    );
}

// ── build_changelog_cte_sql (Change Spine in-txn outbox, phase-02) ──────────

#[test]
fn changelog_cte_runs_the_function_once_and_inserts_the_outbox_row() {
    let sql = super::database::build_changelog_cte_sql(r#""app"."fn_create_user""#, 2, false);

    // The function is materialised so a volatile mutation runs EXACTLY once even
    // though both the INSERT CTE and the primary SELECT read it.
    assert!(
        sql.contains(r#"WITH r AS MATERIALIZED (SELECT * FROM "app"."fn_create_user"($1, $2))"#),
        "function call must be a MATERIALIZED CTE: {sql}"
    );
    // One outbox row, into the framework-owned contract table.
    assert!(
        sql.contains("INSERT INTO core.tb_entity_change_log"),
        "must INSERT into the contract table: {sql}"
    );
    // The primary query returns the function's row unchanged to the caller.
    assert!(
        sql.trim_end().ends_with("SELECT * FROM r"),
        "must return the function row: {sql}"
    );
}

#[test]
fn changelog_cte_threads_object_type_fallback_and_modification_type_after_args() {
    // 2 function args => object_type fallback is $3, modification_type is $4.
    let sql = super::database::build_changelog_cte_sql(r#""fn_x""#, 2, false);
    assert!(
        sql.contains("COALESCE(r.entity_type, $3)"),
        "object_type falls back to $n+1 when entity_type is NULL: {sql}"
    );
    // The verb param ($4) is selected straight into modification_type.
    assert!(sql.contains("$4,"), "modification_type verb is $n+2: {sql}");

    // 0 args => fallback is $1, verb is $2.
    let zero = super::database::build_changelog_cte_sql(r#""fn_y""#, 0, false);
    assert!(
        zero.contains(r#"SELECT * FROM "fn_y"()"#),
        "no-arg call has empty parens: {zero}"
    );
    assert!(
        zero.contains("COALESCE(r.entity_type, $1)"),
        "fallback is $1 with no args: {zero}"
    );
    assert!(zero.contains("$2,"), "verb is $2 with no args: {zero}");
}

#[test]
fn changelog_cte_threads_actor_envelope_after_trace_context() {
    // 2 function args => trace_context is $8, actor_type $9, acting_for $10.
    let sql = super::database::build_changelog_cte_sql(r#""fn_x""#, 2, false);
    // Both actor columns are in the INSERT list, after trace_context and before
    // commit_time.
    assert!(
        sql.contains("actor_type, acting_for, commit_time)"),
        "actor_type/acting_for INSERT columns thread before commit_time: {sql}"
    );
    // acting_for is the last appended envelope param ($n+8) and cast to uuid.
    assert!(sql.contains("$9,"), "actor_type is $n+7: {sql}");
    assert!(sql.contains("$10::uuid,"), "acting_for is $n+8, bound as uuid: {sql}");
}

#[test]
fn changelog_cte_only_logs_effective_changes_and_stamps_the_duration_marker() {
    let sql = super::database::build_changelog_cte_sql(r#""fn_z""#, 1, false);
    // Only an effective change (succeeded AND state_changed) is logged — no-ops
    // and business-logic failures must NOT fan out to the spine.
    assert!(
        sql.contains("WHERE r.succeeded AND r.state_changed"),
        "must gate on an effective change: {sql}"
    );
    // duration_ms uses the canonical wall-clock computation, not the
    // EXTRACT(MILLISECONDS) trap, on the DB clock.
    assert!(sql.contains("EXTRACT(EPOCH"), "canonical duration computation: {sql}");
    assert!(
        !sql.to_uppercase().contains("MILLISECONDS"),
        "must avoid the MILLISECONDS trap: {sql}"
    );
    assert!(
        sql.contains("current_setting('fraiseql.started_at')"),
        "reads started_at from the txn-local GUC: {sql}"
    );
    // The data-quality marker lets #392 refuse to mix pre/post-fix rows.
    assert!(
        sql.contains("'duration_calc_version', 2::int"),
        "stamps the duration_calc_version marker: {sql}"
    );
}

#[test]
fn changelog_cte_maps_mutation_response_columns_to_contract_columns() {
    let sql = super::database::build_changelog_cte_sql(r#""fn_w""#, 0, false);
    // The changed-entity payload columns come straight off the function's
    // app.mutation_response row (object_id<-entity_id, object_data<-entity, …).
    for needle in [
        "(object_type, modification_type, object_id, object_data, updated_fields, cascade, \
         started_at, duration_ms, extra_metadata, tenant_id, trace_id, schema_version, \
         trace_context, actor_type, acting_for, commit_time)",
        "r.entity_id, r.entity, r.updated_fields, r.cascade",
    ] {
        assert!(sql.contains(needle), "expected `{needle}` in: {sql}");
    }
}

#[test]
fn changelog_cte_stamps_the_envelope_tenant_commit_time_and_lets_seq_default() {
    // 2 function args => tenant_id is the $5 envelope param ($n+3), appended
    // AFTER object_type ($3) and modification_type ($4) so the SQL text is stable
    // for prepare_cached regardless of the tenant value.
    let sql = super::database::build_changelog_cte_sql(r#""fn_e""#, 2, false);
    assert!(
        sql.contains("$5::uuid"),
        "tenant_id is the $n+3 envelope param, cast to uuid: {sql}"
    );
    // trace_id is the $n+4 envelope param (plain text, no cast — #375).
    assert!(sql.contains("$6,"), "trace_id is the $n+4 envelope param: {sql}");
    // schema_version is the $n+5 envelope param (plain text, no cast — #377).
    assert!(sql.contains("$7,"), "schema_version is the $n+5 envelope param: {sql}");
    // trace_context is the $n+6 envelope param, cast to jsonb (#375).
    assert!(
        sql.contains("$8::jsonb"),
        "trace_context is the $n+6 envelope param, cast jsonb: {sql}"
    );
    // actor_type is the $n+7 envelope param (plain text — #390).
    assert!(sql.contains("$9,"), "actor_type is the $n+7 envelope param: {sql}");
    // acting_for is the $n+8 envelope param, cast to uuid (#390).
    assert!(
        sql.contains("$10::uuid"),
        "acting_for is the $n+8 envelope param, cast uuid: {sql}"
    );
    // commit_time is the DB clock at INSERT (durable ordering basis).
    assert!(
        sql.contains("clock_timestamp()"),
        "commit_time stamped with clock_timestamp(): {sql}"
    );
    // seq is NOT in the INSERT column list — the table's SEQUENCE default fires,
    // so any INSERTer (incl. cooperative external producers) gets a monotonic
    // value. The column list ends at `commit_time)`.
    assert!(
        sql.contains("acting_for, commit_time)"),
        "INSERT column list ends at commit_time, omitting seq for its DEFAULT: {sql}"
    );
}

#[test]
fn changelog_cte_omits_object_data_before_when_pre_image_is_off() {
    // Default path (opt-out): no pre-image column, no entity_before read — the SQL
    // is byte-for-byte today's behavior. `object_data` stays the after-image.
    let sql = super::database::build_changelog_cte_sql(r#""fn_o""#, 2, false);
    assert!(
        !sql.contains("object_data_before"),
        "pre_image=false must NOT add the object_data_before column: {sql}"
    );
    assert!(
        !sql.contains("r.entity_before"),
        "pre_image=false must NOT read r.entity_before: {sql}"
    );
    // Still threads the after-image and the unchanged INSERT column list.
    assert!(
        sql.contains("object_id, object_data, updated_fields"),
        "after-image object_data column is unchanged: {sql}"
    );
    assert!(
        sql.contains("r.entity_id, r.entity, r.updated_fields"),
        "after-image r.entity is unchanged: {sql}"
    );
}

#[test]
fn changelog_cte_records_the_pre_image_when_opted_in() {
    // Opt-in (changelog_pre_image=true): the before-state is recorded into
    // object_data_before from the function's own entity_before column, ALONGSIDE
    // the after-image (object_data <- entity). object_data stays the after-image
    // for every consumer; the pre-image is a separate column, never an envelope.
    let sql = super::database::build_changelog_cte_sql(r#""fn_p""#, 2, true);
    assert!(
        sql.contains("object_id, object_data, object_data_before, updated_fields"),
        "pre_image=true adds object_data_before AFTER object_data in the INSERT list: {sql}"
    );
    assert!(
        sql.contains("r.entity_id, r.entity, r.entity_before, r.updated_fields"),
        "pre_image=true selects r.entity_before AFTER r.entity (after-image unchanged): {sql}"
    );
    // The positional envelope params are unaffected — the pre-image is a column on
    // `r`, not a bound parameter, so $n+1.. keep their meaning and the statement
    // still differs in TEXT from the off form (so prepare_cached keys them apart).
    assert!(
        sql.contains("COALESCE(r.entity_type, $3)"),
        "object_type fallback still $n+1: {sql}"
    );
    assert!(sql.contains("$10::uuid"), "acting_for still $n+8: {sql}");
}

// ── read-your-writes pin window (#407) ─────────────────────────────────────

#[test]
fn pin_is_active_inside_the_window() {
    assert!(super::pin_active(1_000, 900, 200), "100ms after a write with a 200ms pin");
}

#[test]
fn pin_expires_at_the_window_boundary() {
    assert!(!super::pin_active(1_000, 900, 100), "exactly pin_ms after a write is unpinned");
}

#[test]
fn pin_is_inactive_when_never_written() {
    // last_write = 0 (never armed) with a realistic clock: not pinned.
    assert!(!super::pin_active(1_700_000_000_000, 0, 5_000));
}

#[test]
fn pin_with_zero_window_never_activates() {
    assert!(!super::pin_active(1_000, 1_000, 0), "pin_after_write = 0 disables pinning");
}

#[test]
fn pin_saturates_on_clock_regression() {
    // A clock that steps backwards must land on the safe (pinned) side rather
    // than underflowing.
    assert!(super::pin_active(900, 1_000, 100));
}

// ── the offset-paginated builders order TOTALLY (#1287) ────────────────────
//
// These pin the WIRING, not the renderer: `order_by.rs` can render a total order
// and it is worth nothing unless the two builders that page by LIMIT/OFFSET ask
// for one. Both did not, and the result was a slice of no sequence — measured
// against PostgreSQL 16, walking three pages of 100 over 2 000 rows with four
// distinct statuses returned 300 rows of which 156 were distinct.

use crate::types::{
    DatabaseType,
    sql_hints::{OrderByClause, OrderDirection, SqlProjectionHint},
};

#[test]
fn build_where_select_sql_ordered_appends_the_identity() {
    let clauses = [OrderByClause::new(
        "status".to_string(),
        OrderDirection::Asc,
    )];
    let (sql, _) =
        build_where_select_sql_ordered("v_user", None, Some(10), Some(20), Some(&clauses)).unwrap();
    assert!(
        sql.contains("ORDER BY data->>'status' ASC, data->>'id' ASC"),
        "a paginated read must order totally: {sql}"
    );
    // And the tie-breaker is a column expression, so it must not disturb the
    // placeholder run that puts LIMIT/OFFSET last (#1284).
    assert!(sql.contains("LIMIT $1"), "{sql}");
    assert!(sql.contains("OFFSET $2"), "{sql}");
}

#[test]
fn build_projection_select_sql_appends_the_identity() {
    let hint = SqlProjectionHint {
        database:                    DatabaseType::PostgreSQL,
        projection_template:         "jsonb_build_object('id', data->>'id')".to_string(),
        estimated_reduction_percent: 50,
    };
    let clauses = [OrderByClause::new(
        "status".to_string(),
        OrderDirection::Asc,
    )];
    let (sql, _) =
        build_projection_select_sql(&hint, "v_user", None, Some(10), None, Some(&clauses)).unwrap();
    assert!(
        sql.contains("ORDER BY data->>'status' ASC, data->>'id' ASC"),
        "the projection path pages the same way and needs the same order: {sql}"
    );
}

#[test]
fn an_unordered_paginated_read_is_still_unordered() {
    // The boundary this change deliberately does not cross: a read that asked for
    // no ordering is not a sequence to begin with, and manufacturing one would
    // turn every unordered list read into a sort. `warn_auto_params` reports that
    // case at compile time.
    let (sql, _) = build_where_select_sql_ordered("v_user", None, Some(10), None, None).unwrap();
    assert!(!sql.contains("ORDER BY"), "{sql}");
}