#[pg_test]
fn trigger_sync_adds_edge_overlay_after_explicit_apply_sync() {
    reset_and_create_fixtures();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_sync_edges_pgtest CASCADE")
        .expect("drop sync edge table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_sync_edges_pgtest (
                id TEXT PRIMARY KEY,
                friend_id TEXT REFERENCES public.graph_test_sync_edges_pgtest(id),
                name TEXT NOT NULL
            )",
    )
    .expect("create sync edge table failed");
    Spi::run(
        "INSERT INTO public.graph_test_sync_edges_pgtest (id, friend_id, name)
             VALUES ('a', NULL, 'A'), ('b', NULL, 'B')",
    )
    .expect("insert sync edge rows failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_sync_edges_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add sync edge table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_sync_edges_pgtest'::regclass,
                'friend_id',
                'graph_test_sync_edges_pgtest'::regclass,
                'id',
                'friend',
                bidirectional := false
            )",
    )
    .expect("add sync edge failed");
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");

    Spi::run(
        "INSERT INTO public.graph_test_sync_edges_pgtest (id, friend_id, name)
             VALUES ('c', 'a', 'C')",
    )
    .expect("insert c failed");
    Spi::run("SELECT * FROM graph.apply_sync()").expect("explicit apply_sync failed");

    let reaches_a = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse('graph_test_sync_edges_pgtest'::regclass, 'c', 1, hydrate := false)
             WHERE node_id = 'a'",
    )
    .expect("sync edge traverse failed")
    .unwrap_or(0);
    assert_eq!(reaches_a, 1);
}

#[pg_test]
fn sync_mode_trigger_installs_and_manual_removes_graph_triggers() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");

    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger mode failed");
    Spi::run("SELECT * FROM graph.build()").expect("trigger build failed");
    let trigger_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM pg_trigger
             WHERE tgrelid = 'graph_test_users_pgtest'::regclass
               AND tgname LIKE 'graph_sync_%'",
    )
    .expect("trigger count failed")
    .unwrap_or(0);

    Spi::run("SET graph.sync_mode = 'manual'").expect("set manual mode failed");
    Spi::run("SELECT * FROM graph.build()").expect("manual build failed");
    let manual_trigger_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM pg_trigger
             WHERE tgrelid = 'graph_test_users_pgtest'::regclass
               AND tgname LIKE 'graph_sync_%'",
    )
    .expect("manual trigger count failed")
    .unwrap_or(0);

    assert_eq!(trigger_count, 4);
    assert_eq!(manual_trigger_count, 0);
}

#[pg_test]
fn build_fails_closed_when_sync_checkpoint_read_fails() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("ALTER TABLE graph._sync_log RENAME COLUMN id TO broken_id")
        .expect("malform sync log failed");

    assert!(sql_raises("SELECT * FROM graph.build()"));
}

#[pg_test]
fn sync_mode_wal_fails_with_reserved_message() {
    reset_and_create_fixtures();
    Spi::run("SET graph.sync_mode = 'wal'").expect("set wal mode failed");

    let err = super::current_sync_mode().expect_err("wal mode should be reserved");

    assert!(matches!(
        err,
        crate::safety::GraphError::InvalidFilter { .. }
    ));
    assert_eq!(
            err.to_string(),
            "Invalid filter condition: graph.sync_mode = 'wal' is reserved for roadmap work; please use 'trigger' or 'manual'"
        );
}

#[pg_test]
fn build_accepts_rls_enabled_table_without_acknowledgement() {
    reset_and_create_fixtures();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_rls_pgtest CASCADE")
        .expect("drop rls table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_rls_pgtest (
                id TEXT PRIMARY KEY,
                owner TEXT NOT NULL,
                name TEXT NOT NULL
            )",
    )
    .expect("create rls table failed");
    Spi::run("ALTER TABLE public.graph_test_rls_pgtest ENABLE ROW LEVEL SECURITY")
        .expect("enable row level security failed");
    Spi::run(
        "INSERT INTO public.graph_test_rls_pgtest (id, owner, name)
             VALUES ('a1', 'alice', 'Alice Row')",
    )
    .expect("insert rls row failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_rls_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add rls table failed");

    Spi::run("SELECT * FROM graph.build()")
        .expect("build should accept an RLS-enabled source table by default");
    let node_count = Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
        .expect("status query failed")
        .unwrap_or(0);

    Spi::run("SET graph.allow_rls_tables = on").expect("set deprecated compatibility GUC failed");
    Spi::run("SELECT * FROM graph.build()")
        .expect("deprecated allow_rls_tables value must not change build acceptance");

    Spi::run("RESET graph.allow_rls_tables").expect("reset allow_rls_tables failed");

    assert_eq!(node_count, 1);
}

#[pg_test]
fn build_accepts_rls_enabled_edge_endpoint_table() {
    reset_and_create_fixtures();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_rls_edge_pgtest CASCADE")
        .expect("drop rls edge table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_rls_edge_pgtest (
                id TEXT PRIMARY KEY,
                friend_id TEXT REFERENCES public.graph_test_rls_edge_pgtest(id),
                name TEXT NOT NULL
            )",
    )
    .expect("create rls edge table failed");
    Spi::run(
        "INSERT INTO public.graph_test_rls_edge_pgtest (id, friend_id, name)
             VALUES ('b', NULL, 'Bob'), ('a', 'b', 'Alice')",
    )
    .expect("insert rls edge rows failed");
    Spi::run("ALTER TABLE public.graph_test_rls_edge_pgtest ENABLE ROW LEVEL SECURITY")
        .expect("enable row level security failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_rls_edge_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add rls edge table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_rls_edge_pgtest'::regclass,
                'friend_id',
                'graph_test_rls_edge_pgtest'::regclass,
                'id',
                'friend'
            )",
    )
    .expect("add rls edge failed");

    Spi::run("SELECT * FROM graph.build()")
        .expect("build should accept an edge mapping sourced from an RLS-enabled table");
    let edge_count = Spi::get_one::<i64>("SELECT edge_count FROM graph.status()")
        .expect("status query failed")
        .unwrap_or(0);
    assert_eq!(edge_count, 2);
}

fn setup_sync_log_retention_fixture() {
    reset_and_create_fixtures();
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persist_on_build failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");
    Spi::run("DELETE FROM graph._sync_watermarks").expect("clear stale watermarks failed");
    Spi::run("UPDATE graph._graphs SET sync_log_pruned_before_id = 0")
        .expect("reset prune watermark failed");
}

#[pg_test]
fn sync_health_reports_retention_diagnostics_columns() {
    setup_sync_log_retention_fixture();

    let (floor, prune_recommended, active_backends) = Spi::connect(|client| {
        let result = client
            .select(
                "SELECT sync_log_retention_floor, sync_log_prune_recommended,
                        active_sync_watermark_backends
                   FROM graph.sync_health()",
                None,
                &[],
            )
            .expect("sync_health query failed");
        let row = result.first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<i64>(1).expect("floor read failed"),
            row.get::<bool>(2)
                .expect("prune_recommended read failed")
                .unwrap_or(false),
            row.get::<i32>(3)
                .expect("active_backends read failed")
                .unwrap_or(-1),
        ))
    })
    .expect("sync_health diagnostics query failed");

    // Calling sync_health() itself registers this backend's heartbeat.
    assert!(active_backends >= 1);
    // A fresh graph has applied everything it has seen; the floor may be
    // Some(0) (nothing to prune yet) or None, but must never be negative.
    assert!(floor.is_none_or(|value| value >= 0));
    let _ = prune_recommended;
}

#[pg_test]
fn maintenance_prunes_sync_log_when_this_backend_is_the_only_heartbeat() {
    setup_sync_log_retention_fixture();

    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name) VALUES ('u10', 'Ten')",
    )
    .expect("insert sync row failed");
    Spi::run("SELECT * FROM graph.apply_sync()").expect("apply_sync failed");
    let before_prune = Spi::get_one::<i64>("SELECT count(*) FROM graph._sync_log")
        .expect("sync log count query failed")
        .unwrap_or(0);
    assert!(before_prune > 0, "fixture must have produced sync log rows");

    Spi::run("SELECT * FROM graph.maintenance(concurrently := false)")
        .expect("maintenance failed");

    let after_prune = Spi::get_one::<i64>("SELECT count(*) FROM graph._sync_log")
        .expect("sync log count query failed")
        .unwrap_or(-1);

    assert!(
        after_prune < before_prune,
        "maintenance should prune applied rows when this backend is the only heartbeat: before={before_prune} after={after_prune}"
    );
}

#[pg_test]
fn maintenance_does_not_prune_past_a_lagging_backend_heartbeat() {
    setup_sync_log_retention_fixture();

    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name) VALUES ('u11', 'Eleven')",
    )
    .expect("insert sync row failed");
    Spi::run("SELECT * FROM graph.apply_sync()").expect("apply_sync failed");

    // Simulate a second, lagging backend that has applied nothing yet, still
    // within its heartbeat TTL.
    let graph_id = Spi::get_one::<String>(
        "SELECT graph_id::text FROM graph._graphs WHERE graph_name = 'default'",
    )
    .expect("graph id query failed")
    .unwrap_or_default();
    Spi::run_with_args(
        "INSERT INTO graph._sync_watermarks (
             graph_id, backend_pid, database_oid, applied_sync_id, heartbeat_at, expires_at
         )
         VALUES ($1::uuid, -4242, (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()),
                 0, now(), now() + interval '5 minutes')",
        &[graph_id.clone().into()],
    )
    .expect("simulate lagging backend heartbeat failed");

    let before_prune = Spi::get_one::<i64>("SELECT count(*) FROM graph._sync_log")
        .expect("sync log count query failed")
        .unwrap_or(0);

    Spi::run("SELECT * FROM graph.maintenance(concurrently := false)")
        .expect("maintenance failed");

    let after_prune = Spi::get_one::<i64>("SELECT count(*) FROM graph._sync_log")
        .expect("sync log count query failed")
        .unwrap_or(-1);

    assert_eq!(
        after_prune, before_prune,
        "maintenance must not prune past a lagging backend's applied_sync_id=0 heartbeat"
    );

    Spi::run("DELETE FROM graph._sync_watermarks WHERE backend_pid = -4242")
        .expect("cleanup simulated heartbeat failed");
}

#[pg_test]
fn expired_backend_heartbeat_does_not_block_pruning() {
    setup_sync_log_retention_fixture();

    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name) VALUES ('u12', 'Twelve')",
    )
    .expect("insert sync row failed");
    Spi::run("SELECT * FROM graph.apply_sync()").expect("apply_sync failed");

    // An expired heartbeat from a backend that never cleaned up must be
    // excluded from the floor computation once its TTL has passed.
    let graph_id = Spi::get_one::<String>(
        "SELECT graph_id::text FROM graph._graphs WHERE graph_name = 'default'",
    )
    .expect("graph id query failed")
    .unwrap_or_default();
    Spi::run_with_args(
        "INSERT INTO graph._sync_watermarks (
             graph_id, backend_pid, database_oid, applied_sync_id, heartbeat_at, expires_at
         )
         VALUES ($1::uuid, -4243, (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()),
                 0, now() - interval '1 hour', now() - interval '55 minutes')",
        &[graph_id.into()],
    )
    .expect("simulate expired backend heartbeat failed");

    let before_prune = Spi::get_one::<i64>("SELECT count(*) FROM graph._sync_log")
        .expect("sync log count query failed")
        .unwrap_or(0);

    Spi::run("SELECT * FROM graph.maintenance(concurrently := false)")
        .expect("maintenance failed");

    let after_prune = Spi::get_one::<i64>("SELECT count(*) FROM graph._sync_log")
        .expect("sync log count query failed")
        .unwrap_or(-1);

    assert!(
        after_prune < before_prune,
        "an expired heartbeat must not block pruning: before={before_prune} after={after_prune}"
    );
}

#[pg_test]
fn stale_replay_position_fails_closed_after_a_prune() {
    setup_sync_log_retention_fixture();

    // Capture an intermediate, definitely-positive watermark this backend
    // fully applied, before a second row exists that also gets applied and
    // pruned. Captured explicitly rather than derived from
    // sync_log_pruned_before_id - 1, so this test does not depend on
    // incidental global _sync_log sequence state from other tests sharing
    // this backend's session.
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name) VALUES ('u13', 'Thirteen')",
    )
    .expect("insert first sync row failed");
    Spi::run("SELECT * FROM graph.apply_sync()").expect("first apply_sync failed");
    let intermediate_applied_sync_id =
        crate::ENGINE.with(|engine| engine.borrow().applied_sync_id);
    assert!(
        intermediate_applied_sync_id > 0,
        "first apply_sync() must advance applied_sync_id past 0"
    );

    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name) VALUES ('u14', 'Fourteen')",
    )
    .expect("insert second sync row failed");
    Spi::run("SELECT * FROM graph.apply_sync()").expect("second apply_sync failed");
    Spi::run("SELECT * FROM graph.maintenance(concurrently := false)")
        .expect("maintenance failed");

    let pruned_before_id = Spi::get_one::<i64>(
        "SELECT sync_log_pruned_before_id FROM graph._graphs WHERE graph_name = 'default'",
    )
    .expect("prune watermark query failed")
    .unwrap_or(0);
    assert!(
        pruned_before_id > intermediate_applied_sync_id,
        "maintenance must have pruned past the intermediate watermark: pruned_before_id={pruned_before_id} intermediate={intermediate_applied_sync_id}"
    );

    // Roll this backend's own applied_sync_id back to simulate resuming
    // after a long idle period whose heartbeat has since expired and been
    // pruned past.
    crate::ENGINE.with(|engine| {
        engine
            .borrow_mut()
            .record_applied_sync_id(intermediate_applied_sync_id);
    });

    let denied = sql_raises("SELECT * FROM graph.apply_sync()");
    assert!(
        denied,
        "apply_sync() must fail closed when applied_sync_id predates a completed prune"
    );
}

#[pg_test]
fn guc_contract_defaults_ranges_and_contexts_are_registered() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("RESET graph.sync_mode").expect("reset sync_mode failed");
    Spi::run("RESET graph.oom_action").expect("reset oom_action failed");
    Spi::run("RESET graph.tenant_setting").expect("reset tenant_setting failed");
    Spi::run("RESET graph.build_scan_mode").expect("reset build_scan_mode failed");
    Spi::run("RESET graph.default_projection_mode").expect("reset default_projection_mode failed");
    Spi::run("RESET graph.mutable_enabled").expect("reset mutable_enabled failed");
    Spi::run("RESET graph.enforce_tenant_scope").expect("reset enforce_tenant_scope failed");
    Spi::run("RESET graph.allow_rls_tables").expect("reset allow_rls_tables failed");
    Spi::run("RESET graph.rls_mode").expect("reset rls_mode failed");
    Spi::run("RESET graph.max_exact_path_count").expect("reset max_exact_path_count failed");
    Spi::run("RESET graph.build_batch_size").expect("reset build_batch_size failed");
    Spi::run("RESET graph.max_loaded_graphs_per_backend")
        .expect("reset max_loaded_graphs_per_backend failed");
    Spi::run("RESET graph.hot_eager_load").expect("reset hot_eager_load failed");
    Spi::run("RESET graph.idle_unload_secs").expect("reset idle_unload_secs failed");

    assert_eq!(crate::config::sync_mode(), "trigger");
    assert_eq!(
        crate::config::parsed_sync_mode(),
        Some(crate::config::SyncMode::Trigger)
    );
    assert_eq!(crate::config::oom_action(), crate::config::OomAction::Error);
    assert_eq!(
        crate::config::build_scan_mode(),
        crate::config::BuildScanMode::Select
    );
    assert_eq!(
        crate::config::default_projection_mode(),
        Some(crate::config::ProjectionMode::CsrReadonly)
    );
    assert!(!crate::config::MUTABLE_ENABLED.get());
    assert_eq!(crate::config::tenant_setting(), "");
    assert!(crate::config::ENFORCE_TENANT_SCOPE.get());
    assert_eq!(crate::config::parsed_rls_mode(), Some(crate::config::RlsMode::Enforce));
    assert_eq!(crate::config::MAX_EXACT_PATH_COUNT.get(), 100_000);
    assert_eq!(crate::config::BUILD_BATCH_SIZE.get(), 10_000);
    assert_eq!(crate::config::MAX_TX_DELTA_NODES.get(), 100_000);
    assert_eq!(crate::config::MAX_TX_DELTA_EDGES.get(), 100_000);
    assert_eq!(crate::config::MAX_OVERLAY_MEMORY_MB.get(), 256);
    assert_eq!(crate::config::COMPACTION_THRESHOLD.get(), 50_000);
    assert_eq!(crate::config::PROJECTION_RETENTION_GENERATIONS.get(), 2);
    assert_eq!(crate::config::MAX_LOADED_GRAPHS_PER_BACKEND.get(), 1);
    assert!(!crate::config::HOT_EAGER_LOAD.get());
    assert_eq!(crate::config::IDLE_UNLOAD_SECS.get(), 0);
    assert!(!crate::config::LOW_MEMORY_BUILD.get());

    let registered = Spi::get_one::<bool>(
        "WITH expected(name, context, min_val, max_val) AS (
                VALUES
                    ('graph.memory_limit_mb', 'superuser', '64', '32768'),
                    ('graph.default_max_depth', 'user', '1', '100'),
                    ('graph.max_nodes', 'user', '1', '10000000'),
                    ('graph.max_frontier', 'user', '1', '10000000'),
                    ('graph.max_exact_path_count', 'user', '1', '10000000'),
                    ('graph.build_batch_size', 'superuser', '1', '1000000'),
                    ('graph.edge_buffer_size', 'superuser', '1000', '10000000'),
                    ('graph.vacuum_interval_secs', 'superuser', '5', '86400'),
                    ('graph.max_tx_delta_nodes', 'user', '0', '10000000'),
                    ('graph.max_tx_delta_edges', 'user', '0', '10000000'),
                    ('graph.max_overlay_memory_mb', 'user', '1', '32768'),
                    ('graph.compaction_threshold', 'user', '1', '10000000'),
                    ('graph.projection_retention_generations', 'user', '1', '1000'),
                    ('graph.max_loaded_graphs_per_backend', 'superuser', '0', '1'),
                    ('graph.idle_unload_secs', 'superuser', '0', '86400'),
                    ('graph.low_memory_build', 'superuser', NULL, NULL),
                    ('graph.rls_mode', 'superuser', NULL, NULL)
             ),
             matched AS (
                SELECT e.name,
                       s.context = e.context
                       AND (e.min_val IS NULL OR s.min_val = e.min_val)
                       AND (e.max_val IS NULL OR s.max_val = e.max_val) AS ok
                FROM expected e
                JOIN pg_settings s ON s.name = e.name
             )
             SELECT count(*) = 17 AND bool_and(ok)
             FROM matched",
    )
    .expect("pg_settings inspection failed")
    .unwrap_or(false);
    assert!(registered);
    assert!(sql_raises("SET graph.memory_limit_mb = 63"));
    assert!(sql_raises("SET graph.max_exact_path_count = 0"));
    assert!(sql_raises("SET graph.build_batch_size = 0"));
    assert!(sql_raises("SET graph.edge_buffer_size = 999"));
    assert!(sql_raises("SET graph.max_tx_delta_nodes = -1"));
    assert!(sql_raises("SET graph.max_tx_delta_edges = -1"));
    assert!(sql_raises("SET graph.max_overlay_memory_mb = 0"));
    assert!(sql_raises("SET graph.compaction_threshold = 0"));
    assert!(sql_raises("SET graph.projection_retention_generations = 0"));
    assert!(sql_raises("SET graph.max_loaded_graphs_per_backend = 2"));
    assert!(sql_raises("SET graph.idle_unload_secs = 86401"));
    Spi::run("SET graph.rls_mode = 'unsupported'").expect("set unsupported rls_mode failed");
    assert_eq!(crate::config::parsed_rls_mode(), None);
    Spi::run("RESET graph.rls_mode").expect("reset rls_mode after invalid value failed");
}

#[pg_test]
fn projection_mode_build_and_status_contract() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");

    let readonly_mode = Spi::get_one::<String>(
        "SELECT projection_mode FROM graph.build(mode := 'csr_readonly')",
    )
    .expect("csr build failed")
    .expect("csr build row missing");
    let status_readonly_mode = Spi::get_one::<String>("SELECT projection_mode FROM graph.status()")
        .expect("status projection mode failed")
        .expect("status row missing");
    let delta_clean = Spi::get_one::<bool>(
        "SELECT NOT tx_delta_dirty
                AND overlay_tombstone_count = 0
                AND overlay_memory_bytes = 0
                AND NOT compaction_recommended
                AND tx_delta_added_nodes = 0
                AND tx_delta_deleted_nodes = 0
                AND tx_delta_added_edges = 0
                AND tx_delta_deleted_edges = 0
                AND tx_delta_memory_bytes = 0
         FROM graph.status()",
    )
    .expect("status delta fields failed")
    .unwrap_or(false);

    assert_eq!(readonly_mode, "csr_readonly");
    assert_eq!(status_readonly_mode, "csr_readonly");
    assert!(delta_clean);
    assert!(sql_raises(
        "SELECT projection_mode FROM graph.build(mode := 'mutable_overlay')"
    ));

    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    let mutable_job_id = super::create_build_job(crate::config::ProjectionMode::MutableOverlay)
        .expect("create mutable build job failed");
    Spi::run("SET graph.mutable_enabled = off").expect("disable mutable projection failed");
    super::run_build_job(&mutable_job_id).expect("run queued mutable build job failed");
    let completed_mutable_job = Spi::get_one::<bool>(&format!(
        "SELECT status = 'completed'
                AND projection_mode = 'mutable_overlay'
         FROM graph._build_jobs
         WHERE build_id = {}",
        super::sql_literal(&mutable_job_id)
    ))
    .expect("completed mutable job mode failed")
    .unwrap_or(false);

    Spi::run("SET graph.mutable_enabled = on").expect("re-enable mutable projection failed");
    let mutable_mode = Spi::get_one::<String>(
        "SELECT projection_mode FROM graph.build(mode := 'mutable_overlay')",
    )
    .expect("mutable build failed")
    .expect("mutable build row missing");
    let health_mode = Spi::get_one::<String>("SELECT projection_mode FROM graph.sync_health()")
        .expect("sync_health projection mode failed")
        .expect("sync_health row missing");

    assert!(completed_mutable_job);
    assert_eq!(mutable_mode, "mutable_overlay");
    assert_eq!(health_mode, "mutable_overlay");
    Spi::run("SET graph.mutable_enabled = off").expect("restore mutable projection failed");
}

#[pg_test]
fn every_oom_action_rejects_over_budget_build_before_allocation() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persist_on_build failed");
    Spi::run("SET graph.memory_limit_mb = 64").expect("set memory limit failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_test_oom_pgtest CASCADE")
        .expect("drop oom table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_oom_pgtest (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL
            )",
    )
    .expect("create oom table failed");
    Spi::run("INSERT INTO public.graph_test_oom_pgtest VALUES ('one', 'One')")
        .expect("insert oom row failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_oom_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add oom table failed");
    Spi::run(
        "UPDATE pg_class
             SET reltuples = 100000000
             WHERE oid = 'public.graph_test_oom_pgtest'::regclass",
    )
    .expect("inflate reltuples failed");

    Spi::run("SET graph.oom_action = 'error'").expect("set oom error failed");
    assert!(sql_raises("SELECT * FROM graph.build()"));

    Spi::run("SET graph.oom_action = 'readonly'").expect("set oom readonly failed");
    assert!(sql_raises("SELECT * FROM graph.build()"));
    let built = crate::ENGINE.with(|engine| engine.borrow().built);

    Spi::run("SET graph.oom_action = 'error'").expect("restore oom action failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");

    assert!(!built, "rejected builds must not install a read-only engine");
}

#[pg_test]
fn build_row_estimate_counts_exactly_when_statistics_are_unknown() {
    Spi::run("DROP TABLE IF EXISTS public.graph_test_unknown_stats_pgtest CASCADE")
        .expect("drop unknown-statistics table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_unknown_stats_pgtest (id bigint PRIMARY KEY);
         INSERT INTO public.graph_test_unknown_stats_pgtest VALUES (1), (2), (3);
         UPDATE pg_catalog.pg_class
            SET reltuples = 0
          WHERE oid = 'public.graph_test_unknown_stats_pgtest'::regclass;",
    )
    .expect("create unknown-statistics fixture failed");

    let rows = crate::catalog::estimated_table_rows("public.graph_test_unknown_stats_pgtest")
        .expect("unknown statistics should fall back to an exact count");

    assert_eq!(rows, 3);
    Spi::run("DROP TABLE public.graph_test_unknown_stats_pgtest")
        .expect("drop unknown-statistics fixture failed");
}

#[pg_test]
fn build_memory_headroom_accounts_for_existing_serving_graph() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persist_on_build failed");
    Spi::run("SET graph.memory_limit_mb = 64").expect("set memory limit failed");
    Spi::run("SET graph.oom_action = 'error'").expect("set oom error failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_headroom_pgtest CASCADE")
        .expect("drop headroom table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_headroom_pgtest (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL
            )",
    )
    .expect("create headroom table failed");
    Spi::run("INSERT INTO public.graph_test_headroom_pgtest VALUES ('one', 'One')")
        .expect("insert headroom row failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_headroom_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add headroom table failed");

    crate::ENGINE.with(|e| {
        let mut eng = crate::engine::Engine::new();
        for idx in 0..1_200_000u32 {
            eng.node_store.add_node(1, idx.to_string());
        }
        eng.built = true;
        *e.borrow_mut() = eng;
    });

    assert!(sql_raises("SELECT * FROM graph.build()"));

    Spi::run("SET graph.oom_action = 'error'").expect("restore oom action failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
}

#[pg_test]
fn low_memory_build_unloads_existing_backend_graph_before_rebuild() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persist_on_build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("set initial memory limit failed");
    Spi::run("SET graph.oom_action = 'error'").expect("set oom error failed");
    Spi::run("SET graph.low_memory_build = on").expect("enable low memory build failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_low_memory_pgtest CASCADE")
        .expect("drop low memory table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_low_memory_pgtest (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL
            )",
    )
    .expect("create low memory table failed");
    Spi::run("INSERT INTO public.graph_test_low_memory_pgtest VALUES ('one', 'One')")
        .expect("insert low memory row failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_low_memory_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add low memory table failed");
    Spi::run("SELECT * FROM graph.build()").expect("persisted baseline build failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persist_on_build failed");
    Spi::run("SET graph.memory_limit_mb = 65").expect("set low memory limit failed");

    crate::ENGINE.with(|e| {
        let mut eng = crate::engine::Engine::new();
        for idx in 0..1_200_000u32 {
            eng.node_store.add_node(1, idx.to_string());
        }
        eng.built = true;
        *e.borrow_mut() = eng;
    });

    let nodes = Spi::get_one::<i64>("SELECT nodes_loaded FROM graph.build()")
        .expect("low memory build failed")
        .unwrap_or(0);
    let read_only = Spi::get_one::<bool>("SELECT read_only FROM graph.status()")
        .expect("status read failed")
        .unwrap_or(true);

    Spi::run("SET graph.low_memory_build = off").expect("restore low memory build failed");
    Spi::run("SET graph.oom_action = 'error'").expect("restore oom action failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");

    assert_eq!(nodes, 1);
    assert!(!read_only);
}

#[pg_test]
fn replacement_faults_preserve_or_reconcile_the_published_generation() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persistence failed");
    Spi::run("SET graph.sync_mode = manual").expect("set manual sync failed");
    Spi::run("SET graph.low_memory_build = off").expect("disable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("set memory limit failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_replacement_fault_pgtest CASCADE")
        .expect("drop replacement table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_replacement_fault_pgtest (
            id TEXT PRIMARY KEY,
            name TEXT NOT NULL
        )",
    )
    .expect("create replacement table failed");
    Spi::run("INSERT INTO public.graph_test_replacement_fault_pgtest VALUES ('one', 'One')")
        .expect("insert baseline row failed");
    Spi::run(
        "SELECT graph.add_table(
            'graph_test_replacement_fault_pgtest'::regclass,
            id_column := 'id',
            columns := ARRAY['name']
        )",
    )
    .expect("register replacement table failed");
    Spi::run("SELECT * FROM graph.build()").expect("baseline build failed");

    let generation_a = Spi::get_one::<i64>(
        "SELECT manifest_generation FROM graph.projection_status()",
    )
    .expect("baseline generation query failed")
    .expect("baseline generation missing");
    Spi::run("INSERT INTO public.graph_test_replacement_fault_pgtest VALUES ('two', 'Two')")
        .expect("insert replacement row failed");

    Spi::run("SET graph.low_memory_build = on").expect("enable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 65").expect("set low memory limit failed");
    Spi::run("SELECT graph._test_arm_replacement_fault('source_scan')")
        .expect("arm low-memory source-scan fault failed");
    assert!(sql_raises("SELECT * FROM graph.build()"));
    #[cfg(feature = "development")]
    super::sql_facade::reset_query_start_probe_counts();
    assert_eq!(
        Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
            .expect("low-memory recovery status failed"),
        Some(1),
        "status must reload generation A after cancellation-style eviction"
    );
    #[cfg(feature = "development")]
    assert_eq!(
        super::sql_facade::query_start_probe_counts(),
        (1, 1),
        "replacement recovery must reuse the authoritative query-start catalog fingerprint"
    );
    assert_eq!(
        Spi::get_one::<i64>(
            "SELECT manifest_generation FROM graph.projection_status()",
        )
        .expect("generation A verification failed"),
        Some(generation_a)
    );

    Spi::run("SET graph.low_memory_build = off").expect("disable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
    for stage in [
        "candidate_write",
        "validation",
        "before_publication",
    ] {
        Spi::run(&format!(
            "SELECT graph._test_arm_replacement_fault('{}')",
            stage
        ))
        .expect("arm pre-publication fault failed");
        assert!(sql_raises("SELECT * FROM graph.build()"));
        assert_eq!(
            Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
                .expect("pre-publication recovery status failed"),
            Some(1),
            "pre-publication fault at {stage} must preserve generation A"
        );
        assert_eq!(
            Spi::get_one::<i64>(
                "SELECT manifest_generation FROM graph.projection_status()",
            )
            .expect("pre-publication generation query failed"),
            Some(generation_a)
        );
    }

    Spi::run("SELECT graph._test_arm_replacement_fault('after_publication')")
        .expect("arm post-publication fault failed");
    assert!(sql_raises("SELECT * FROM graph.build()"));
    assert_eq!(
        Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
            .expect("post-publication recovery status failed"),
        Some(2),
        "status must reconcile to generation B after publication"
    );
    let generation_b = Spi::get_one::<i64>(
        "SELECT manifest_generation FROM graph.projection_status()",
    )
    .expect("generation B query failed")
    .expect("generation B missing");
    assert!(generation_b > generation_a);
    Spi::run("SET graph.low_memory_build = off").expect("restore low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
}

#[pg_test]
fn low_memory_build_rejects_unrecoverable_resident_replacement() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persistence failed");
    Spi::run("SET graph.low_memory_build = off").expect("disable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("set memory limit failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_unrecoverable_build_pgtest CASCADE")
        .expect("drop unrecoverable table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_unrecoverable_build_pgtest (
            id TEXT PRIMARY KEY
        )",
    )
    .expect("create unrecoverable table failed");
    Spi::run("INSERT INTO public.graph_test_unrecoverable_build_pgtest VALUES ('one')")
        .expect("insert unrecoverable row failed");
    Spi::run(
        "SELECT graph.add_table(
            'graph_test_unrecoverable_build_pgtest'::regclass,
            id_column := 'id'
        )",
    )
    .expect("register unrecoverable table failed");
    Spi::run("SELECT * FROM graph.build()").expect("resident baseline build failed");

    Spi::run("SET graph.low_memory_build = on").expect("enable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 65").expect("set low memory limit failed");
    let message = sql_error_message("SELECT * FROM graph.build()")
        .expect("unrecoverable low-memory build should fail");
    assert!(message.contains("no recoverable persisted generation exists"));
    assert_eq!(
        Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
            .expect("resident status failed"),
        Some(1),
        "validation must reject before evicting the resident graph"
    );
    Spi::run("SET graph.low_memory_build = off").expect("restore low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
}

#[pg_test]
fn low_memory_build_rejects_corrupt_persisted_recovery_before_eviction() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persistence failed");
    Spi::run("SET graph.low_memory_build = off").expect("disable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("set memory limit failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_corrupt_recovery_pgtest CASCADE")
        .expect("drop corrupt recovery table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_corrupt_recovery_pgtest (
            id TEXT PRIMARY KEY
        );
        INSERT INTO public.graph_test_corrupt_recovery_pgtest VALUES ('one');
        SELECT graph.add_table(
            'graph_test_corrupt_recovery_pgtest'::regclass,
            id_column := 'id'
        )",
    )
    .expect("create corrupt recovery fixture failed");
    Spi::run("SELECT * FROM graph.build()").expect("persisted baseline build failed");

    let graph_path = crate::persistence::graph_file_path().expect("graph path resolves");
    let current_base = crate::persistence::current_base_artifact_path(&graph_path)
        .expect("current base resolves")
        .expect("current base exists");
    let original = std::fs::read(&current_base).expect("current base reads");
    std::fs::write(&current_base, b"truncated").expect("current base corruption writes");

    Spi::run("SET graph.low_memory_build = on").expect("enable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 65").expect("set low memory limit failed");
    let message = sql_error_message("SELECT * FROM graph.build()")
        .expect("corrupt recovery low-memory build should fail");
    let resident_count = Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
        .expect("resident status failed");
    std::fs::write(&current_base, original).expect("current base restores");

    assert!(message.contains("no recoverable persisted generation exists"));
    assert_eq!(
        resident_count,
        Some(1),
        "corrupt persisted recovery must be rejected before resident eviction"
    );
    Spi::run("SET graph.low_memory_build = off").expect("restore low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persistence failed");
}

#[pg_test]
fn rebuild_publishers_reconcile_prepublication_failures_and_retry() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persistence failed");
    Spi::run("SET graph.low_memory_build = off").expect("disable low memory build failed");
    Spi::run("SET graph.memory_limit_mb = 2048").expect("set memory limit failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_rebuild_publishers_pgtest CASCADE")
        .expect("drop rebuild publishers table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_rebuild_publishers_pgtest (
            id TEXT PRIMARY KEY
        );
        INSERT INTO public.graph_test_rebuild_publishers_pgtest VALUES ('baseline');
        SELECT graph.add_table(
            'graph_test_rebuild_publishers_pgtest'::regclass,
            id_column := 'id'
        );
        SELECT * FROM graph.build()",
    )
    .expect("create rebuild publisher fixture failed");
    let generation_a = Spi::get_one::<i64>(
        "SELECT manifest_generation FROM graph.projection_status()",
    )
    .expect("baseline generation query failed")
    .expect("baseline generation missing");

    for (id, statement) in [
        ("vacuum-row", "SELECT * FROM graph.vacuum()"),
        (
            "maintenance-row",
            "SELECT * FROM graph.maintenance(concurrently := false)",
        ),
    ] {
        Spi::run(&format!(
            "INSERT INTO public.graph_test_rebuild_publishers_pgtest VALUES ('{id}')"
        ))
        .expect("insert replacement source row failed");
        Spi::run("SELECT graph._test_arm_replacement_fault('before_publication')")
            .expect("arm rebuild publisher fault failed");
        assert!(sql_raises(statement), "{statement} should fail before publication");
        assert_eq!(
            Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
                .expect("rebuild publisher recovery status failed"),
            Some(1),
            "{statement} must keep generation A resident"
        );
        assert_eq!(
            Spi::get_one::<i64>(
                "SELECT manifest_generation FROM graph.projection_status()",
            )
            .expect("rebuild publisher generation query failed"),
            Some(generation_a),
            "{statement} must not publish its candidate"
        );
    }

    Spi::run(
        "INSERT INTO public.graph_test_rebuild_publishers_pgtest VALUES ('background-row')",
    )
    .expect("insert background replacement row failed");
    let job_id = super::create_maintenance_job().expect("create maintenance job failed");
    Spi::run("SELECT graph._test_arm_replacement_fault('before_publication')")
        .expect("arm background maintenance fault failed");
    let worker_error = Spi::get_one::<String>(&format!(
        "SELECT graph._test_run_maintenance_job('{}')",
        job_id.replace('\'', "''")
    ))
    .expect("run background maintenance test job failed");
    assert!(worker_error.is_some(), "background maintenance should record failure");
    assert_eq!(
        Spi::get_one::<i64>("SELECT node_count FROM graph.status()")
            .expect("background recovery status failed"),
        Some(1)
    );
    assert_eq!(
        Spi::get_one::<i64>(
            "SELECT manifest_generation FROM graph.projection_status()",
        )
        .expect("background generation query failed"),
        Some(generation_a)
    );

    let nodes = Spi::get_one::<i64>(
        "SELECT nodes_after FROM graph.maintenance(concurrently := false)",
    )
    .expect("successful maintenance retry failed");
    assert_eq!(nodes, Some(4));
    Spi::run("SET graph.persist_on_build = off").expect("restore persistence failed");
}

#[pg_test]
fn build_resource_status_reports_byte_adaptive_pressure_for_long_rows() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persistence failed");
    Spi::run("SET graph.memory_limit_mb = 64").expect("set memory limit failed");
    Spi::run("SET graph.build_batch_size = 10000").expect("set row batch limit failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_long_build_rows_pgtest CASCADE")
        .expect("drop long-row table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_long_build_rows_pgtest (
             id text PRIMARY KEY,
             payload text NOT NULL
         );
         INSERT INTO public.graph_test_long_build_rows_pgtest
         SELECT repeat('k', 1800) || lpad(value::text, 8, '0'),
                repeat('v', 64) || value::text
         FROM generate_series(1, 3000) AS value;",
    )
    .expect("create long-row fixture failed");
    Spi::run(
        "SELECT graph.add_table(
             'graph_test_long_build_rows_pgtest'::regclass,
             id_column := 'id',
             columns := ARRAY['payload']
         );",
    )
    .expect("register long-row fixture failed");

    let nodes = Spi::get_one::<i64>("SELECT nodes_loaded FROM graph.build()")
        .expect("long-row build failed")
        .unwrap_or(0);
    let status = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT budget_bytes, peak_bytes, peak_phase, pressure_events
                   FROM graph.build_resource_status()",
                None,
                &[],
            )?
            .first();
        Ok::<_, pgrx::spi::Error>(
            (
                row.get::<i64>(1)?.unwrap_or(0),
                row.get::<i64>(2)?.unwrap_or(0),
                row.get::<String>(3)?,
                row.get::<i64>(4)?.unwrap_or(0),
            ),
        )
    })
    .expect("read build resource status failed");

    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
    Spi::run("SET graph.build_batch_size = 10000").expect("restore row batch limit failed");
    Spi::run("SET graph.persist_on_build = off").expect("restore persistence failed");

    assert_eq!(nodes, 3000);
    assert_eq!(status.0, 60 * 1_048_576);
    assert!(status.1 > 0 && status.1 <= status.0);
    assert!(status.2.is_some());
    assert!(status.3 > 0, "long rows should trigger byte-pressure flushes");
}

#[pg_test]
fn traversal_work_limit_returns_stable_resource_diagnostic_and_status() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()") .expect("reset failed");
    Spi::run("SET LOCAL graph.auto_load = off").expect("disable auto-load failed");
    clear_graph_catalog_for_test();
    Spi::run(
        "DROP TABLE IF EXISTS public.graph_test_resource_paths CASCADE;
         CREATE TABLE public.graph_test_resource_paths (
             id text PRIMARY KEY,
             parent_id text REFERENCES public.graph_test_resource_paths(id)
         );
         INSERT INTO public.graph_test_resource_paths VALUES
             ('root', NULL), ('a', 'root'), ('b', 'root');
         SELECT graph.add_table(
             'graph_test_resource_paths'::regclass,
             id_column := 'id'
         );
         SELECT graph.add_edge(
             'graph_test_resource_paths'::regclass,
             'parent_id',
             'graph_test_resource_paths'::regclass,
             'id',
             'parent',
             bidirectional := false
         );
         SELECT * FROM graph.build();
         SET LOCAL graph.query_work_limit = 1;",
    )
    .expect("resource path fixture failed");

    let statement = "SELECT * FROM graph.traverse('graph_test_resource_paths'::regclass, 'root', 3, direction := 'in', hydrate := false)";
    assert_eq!(sqlstate_for_error(statement).as_deref(), Some("54000"));
    assert_eq!(
        sql_error_detail(statement).as_deref(),
        Some("pgGraph diagnostic: PG007")
    );
    let status = Spi::connect(|client| {
        let row = client
            .select(
                "SELECT operation, work_units FROM graph.resource_status()",
                None,
                &[],
            )?
            .first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<String>(1)?.unwrap_or_default(),
            row.get::<i64>(2)?.unwrap_or_default(),
        ))
    })
    .expect("resource status read failed");
    assert_eq!(status.0, "query");
    assert_eq!(status.1, 1);
}

#[pg_test]
fn stale_positive_statistics_are_stopped_by_runtime_build_budget() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persistence failed");
    Spi::run("SET graph.memory_limit_mb = 64").expect("set memory limit failed");
    clear_graph_catalog_for_test();
    Spi::run("DROP TABLE IF EXISTS public.graph_test_stale_build_stats_pgtest CASCADE")
        .expect("drop stale-statistics table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_stale_build_stats_pgtest (
             id bigint PRIMARY KEY
         );
         INSERT INTO public.graph_test_stale_build_stats_pgtest
         SELECT value FROM generate_series(1, 450000) AS value;
         UPDATE pg_catalog.pg_class
            SET reltuples = 1
          WHERE oid = 'public.graph_test_stale_build_stats_pgtest'::regclass;
         SELECT graph.add_table(
             'graph_test_stale_build_stats_pgtest'::regclass,
             id_column := 'id'
         );",
    )
    .expect("create stale-statistics fixture failed");

    assert!(sql_raises("SELECT * FROM graph.build()"));
    let built = crate::ENGINE.with(|engine| engine.borrow().built);

    Spi::run("SET graph.memory_limit_mb = 2048").expect("restore memory limit failed");
    assert!(!built, "runtime rejection must not install a partial engine");
}

#[pg_test]
fn catalog_drift_requires_rebuild() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");
    Spi::run("SELECT graph.add_filter_column('graph_test_users_pgtest'::regclass, 'age')")
        .expect("add filter after build failed");

    let result = super::ensure_current_graph();
    assert!(result.is_err());
}

#[cfg(feature = "development")]
#[pg_test]
fn query_start_state_reads_catalog_and_pending_sync_once() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
            'graph_test_users_pgtest'::regclass,
            id_column := 'id',
            columns := ARRAY['name']
         )",
    )
    .expect("add query-start table failed");
    Spi::run(
        "SELECT graph.add_edge(
            'graph_test_friendships_pgtest'::regclass,
            'user_id',
            'graph_test_users_pgtest'::regclass,
            'friend_id',
            'friend'
         )",
    )
    .expect("add query-start edge failed");
    Spi::run("SELECT * FROM graph.build() ").expect("build query-start fixture failed");

    super::sql_facade::reset_query_start_probe_counts();
    let row_count = Spi::get_one::<i64>(
        "SELECT count(*)
           FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                0,
                hydrate := false
           )",
    )
    .expect("public fixed-work traversal failed")
    .unwrap_or(0);
    let counts = super::sql_facade::query_start_probe_counts();

    assert_eq!(row_count, 1);
    assert_eq!(counts, (1, 1));

    super::sql_facade::reset_query_start_probe_counts();
    let hydrated_filtered_count = Spi::get_one::<i64>(
        "SELECT count(*)
           FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                0,
                filter := '{\"where\": {\"age\": {\"gte\": 0}}}'::jsonb,
                hydrate := true
           )
          WHERE node IS NOT NULL",
    )
    .expect("public hydrated filtered traversal failed")
    .unwrap_or(0);

    assert_eq!(hydrated_filtered_count, 1);
    assert_eq!(
        super::sql_facade::query_start_probe_counts(),
        (1, 1),
        "hydration and unindexed filter resolution must reuse query-start catalog rows"
    );
}

#[pg_test]
fn persisted_direct_build_matches_owned_build_for_public_queries() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO graph_test_users_pgtest (id, name, age)
         VALUES ('u4', 'Dana', 29), ('u3', 'Carol', 53);
         INSERT INTO graph_test_friendships_pgtest (id, user_id, friend_id)
         VALUES ('f3', 'u1', 'u4'), ('f2', 'u1', 'u3')",
    )
    .expect("add deterministic-order fixtures failed");
    Spi::run(
        "SELECT graph.add_table(
            'graph_test_users_pgtest'::regclass,
            id_column := 'id',
            columns := ARRAY['name']
         )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
            'graph_test_friendships_pgtest'::regclass,
            'user_id',
            'graph_test_users_pgtest'::regclass,
            'friend_id',
            'friend',
            bidirectional := false
         )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT graph.add_filter_column('graph_test_users_pgtest'::regclass, 'age')")
        .expect("add age filter failed");

    Spi::run("SET graph.persist_on_build = off").expect("disable persistence failed");
    Spi::run("SELECT * FROM graph.build() ").expect("owned build failed");
    let owned = direct_build_public_snapshot();

    Spi::run("SET graph.persist_on_build = on").expect("enable persistence failed");
    Spi::run("SELECT * FROM graph.build() ").expect("direct persisted build failed");
    let persisted = direct_build_public_snapshot();
    let capped_before_rewrite = direct_build_capped_snapshot();

    assert_eq!(persisted, owned);

    Spi::run(
        "CREATE INDEX graph_test_users_desc_idx
             ON graph_test_users_pgtest (id DESC);
         CREATE INDEX graph_test_friendships_desc_idx
             ON graph_test_friendships_pgtest (id DESC);
         CLUSTER graph_test_users_pgtest USING graph_test_users_desc_idx;
         CLUSTER graph_test_friendships_pgtest USING graph_test_friendships_desc_idx",
    )
    .expect("rewrite fixture heap order failed");
    Spi::run("SELECT * FROM graph.build() ").expect("direct rebuild after heap rewrite failed");
    assert_eq!(direct_build_capped_snapshot(), capped_before_rewrite);
}

fn direct_build_capped_snapshot() -> String {
    Spi::get_one::<String>(
        "SELECT string_agg(node_id || ':' || depth::text, ',' ORDER BY depth, node_id)
         FROM graph.traverse(
             'graph_test_users_pgtest'::regclass,
             'u1',
             1,
             edge_types := ARRAY['friend'],
             hydrate := false,
             max_rows := 2
         )",
    )
    .expect("capped traversal snapshot failed")
    .unwrap_or_default()
}

fn direct_build_public_snapshot() -> (String, i64, String) {
    let traversal = Spi::get_one::<String>(
        "SELECT string_agg(
             node_id || ':' || depth::text || ':' || edge_path::text,
             ',' ORDER BY depth, node_id
         )
         FROM graph.traverse(
             'graph_test_users_pgtest'::regclass,
             'u1',
             2,
             edge_types := ARRAY['friend'],
             hydrate := false
         )",
    )
    .expect("traversal snapshot failed")
    .unwrap_or_default();
    let filtered = Spi::get_one::<i64>(
        "SELECT count(*)
         FROM graph.traverse(
             'graph_test_users_pgtest'::regclass,
             'u1',
             2,
             filter := graph.all(ARRAY[graph.greater_than('age', 40)]),
             hydrate := false
         )",
    )
    .expect("filter snapshot failed")
    .unwrap_or_default();
    let gql = Spi::get_one::<String>(
        "SELECT string_agg(
             (row #>> '{u,_id,id}') || '>' || (row #>> '{v,_id,id}'),
             ',' ORDER BY row #>> '{u,_id,id}', row #>> '{v,_id,id}'
         )
         FROM graph.gql(
             'MATCH (u:graph_test_users_pgtest)-[:friend]->(v:graph_test_users_pgtest) RETURN u, v'
         )",
    )
    .expect("GQL snapshot failed")
    .unwrap_or_default();
    (traversal, filtered, gql)
}

#[pg_test]
fn schema_drift_detects_live_ddl_changes() {
    Spi::run("SELECT pg_advisory_xact_lock(1918928211, 1735552872)")
        .expect("test fixture lock failed");
    Spi::run("SELECT graph.reset()").expect("reset failed");
    Spi::run("SET graph.auto_load = off").expect("disable auto_load failed");
    Spi::run("SET graph.persist_on_build = off").expect("disable persist_on_build failed");
    Spi::run("SET graph.sync_mode = 'manual'").expect("reset sync mode failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_test_schema_drift_pgtest CASCADE")
        .expect("drop drift table failed");
    Spi::run(
        "CREATE TABLE public.graph_test_schema_drift_pgtest (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                score INT NOT NULL,
                parent_id TEXT,
                weight INT,
                edge_label TEXT
            )",
    )
    .expect("create drift table failed");
    Spi::run(
        "INSERT INTO public.graph_test_schema_drift_pgtest
                (id, name, score, parent_id, weight, edge_label)
             VALUES
                ('a', 'A', 10, NULL, NULL, NULL),
                ('b', 'B', 20, 'a', 7, 'parent')",
    )
    .expect("insert drift rows failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_schema_drift_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add drift table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_schema_drift_pgtest'::regclass,
                'parent_id',
                'graph_test_schema_drift_pgtest'::regclass,
                'id',
                'parent',
                bidirectional := false,
                weight_column := 'weight',
                label_column := 'edge_label'
            )",
    )
    .expect("add drift edge failed");
    Spi::run(
        "SELECT graph.add_filter_column(
                'graph_test_schema_drift_pgtest'::regclass,
                'score',
                column_type := 'numeric'
            )",
    )
    .expect("add drift filter failed");
    Spi::run("SELECT * FROM graph.build()").expect("build drift fixture failed");

    Spi::run(
        "ALTER TABLE public.graph_test_schema_drift_pgtest
             ALTER COLUMN score TYPE text USING score::text",
    )
    .expect("ddl drift mutation failed");
    let reason = Spi::get_one::<String>(
        "SELECT COALESCE(invalid_reason, '')
             FROM graph.status()
             WHERE schema_status = 'invalid'
               AND needs_rebuild",
    )
    .expect("status drift inspection failed")
    .unwrap_or_default();

    assert!(reason.contains("filter column"));
}

#[pg_test]
fn status_reports_dropped_registered_relation_as_invalid_schema() {
    reset_and_create_fixtures();
    Spi::run("SET graph.persist_on_build = off").expect("disable persistence failed");
    Spi::run("DROP TABLE IF EXISTS public.graph_test_dropped_relation_pgtest CASCADE")
        .expect("drop prior relation fixture failed");
    Spi::run(
        "CREATE TABLE public.graph_test_dropped_relation_pgtest (
            id text PRIMARY KEY,
            name text NOT NULL
        );
        INSERT INTO public.graph_test_dropped_relation_pgtest VALUES ('n1', 'Node')",
    )
    .expect("create relation fixture failed");
    Spi::run(
        "SELECT graph.add_table(
            'graph_test_dropped_relation_pgtest'::regclass,
            id_column := 'id',
            columns := ARRAY['name']
        )",
    )
    .expect("register relation fixture failed");
    Spi::run("SELECT * FROM graph.build() ").expect("build relation fixture failed");
    Spi::run("DROP TABLE public.graph_test_dropped_relation_pgtest CASCADE")
        .expect("drop registered relation failed");

    let status = Spi::get_two::<bool, String>(
        "SELECT needs_rebuild, schema_status
           FROM graph.status()",
    )
    .expect("inspect dropped-relation status failed");

    assert_eq!(status.0, Some(true));
    assert_eq!(status.1.as_deref(), Some("invalid"));
}

#[pg_test]
fn build_scan_mode_select_works_and_copy_fails_clearly() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SET graph.build_scan_mode = 'select'").expect("set select mode failed");
    Spi::run("SET graph.build_batch_size = 1").expect("set small build batch failed");
    let shape =
        Spi::get_one::<bool>("SELECT nodes_loaded = 2 AND edges_loaded = 2 FROM graph.build()")
            .expect("select build failed")
            .unwrap_or(false);
    assert!(shape);

    Spi::run("SET graph.build_scan_mode = 'copy'").expect("set copy mode failed");
    let (tables, edges, filter_columns) = super::read_catalog().expect("catalog read failed");
    let copy_result = crate::builder::build_graph(&tables, &edges, &filter_columns);
    assert!(copy_result.is_err());
    Spi::run("SET graph.build_scan_mode = 'select'").expect("restore select mode failed");
    Spi::run("SET graph.build_batch_size = 10000").expect("restore build batch failed");
}
