// Ordered-scan behavior, scan-state lifecycle, and build-budget policy
// pg_tests split from `hnsw_am.rs` for the source-hygiene size target.

#[pg_test]
fn hnsw_access_method_serves_ordered_index_scan() {
    Spi::run(
        "CREATE TABLE hnsw_explain_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            embedding vector NOT NULL
         )",
    )
    .expect("explain HNSW fixture table should be created");

    Spi::run(
        "INSERT INTO hnsw_explain_items (embedding)
         VALUES ('[9,9,9]'::vector), ('[1,1,1]'::vector), ('[3,3,3]'::vector)",
    )
    .expect("explain HNSW fixture rows should be inserted");

    Spi::run(
        "CREATE INDEX hnsw_explain_items_embedding_idx
            ON hnsw_explain_items USING pgcontext_hnsw (embedding)",
    )
    .expect("explain HNSW index should build");

    Spi::run("SET LOCAL enable_seqscan = off").expect("seqscan should be disabled for plan check");

    let ids = Spi::get_one::<String>(
        "SELECT string_agg(id::text, ',' ORDER BY distance_rank)
           FROM (
                SELECT id,
                       row_number() OVER (
                           ORDER BY embedding OPERATOR(pgcontext.<->) '[0,0,0]'::vector
                       ) AS distance_rank
                  FROM hnsw_explain_items
                 ORDER BY embedding OPERATOR(pgcontext.<->) '[0,0,0]'::vector
                 LIMIT 2
           ) ordered",
    )
    .expect("HNSW ordered query should succeed")
    .expect("HNSW ordered query should return rows");

    assert_eq!(ids, "2,3");

    let plan = Spi::connect(|client| {
        let result = client
            .select(
                "EXPLAIN (COSTS TRUE, FORMAT TEXT)
                 SELECT id
                   FROM hnsw_explain_items
                  ORDER BY embedding OPERATOR(pgcontext.<->) '[0,0,0]'::vector
                  LIMIT 1",
                None,
                &[],
            )
            .expect("HNSW EXPLAIN query should succeed");

        let mut lines = Vec::new();
        for row in result {
            lines.push(row.get::<String>(1)?.unwrap_or_default());
        }
        Ok::<_, spi::Error>(lines.join("\n"))
    })
    .expect("HNSW EXPLAIN plan should decode");

    assert!(
        plan.contains("Index Scan using hnsw_explain_items_embedding_idx"),
        "expected HNSW index scan for order-by query:\n{plan}"
    );
    assert!(
        plan.contains("cost="),
        "EXPLAIN should include planner costs:\n{plan}"
    );
}

#[pg_test]
fn hnsw_unordered_scan_remains_planner_disabled() {
    Spi::run(
        "CREATE TABLE hnsw_unordered_plan_items (
             embedding vector NOT NULL
         );
         INSERT INTO hnsw_unordered_plan_items
         VALUES ('[1,1]'::vector), ('[2,2]'::vector);
         CREATE INDEX hnsw_unordered_plan_items_idx
             ON hnsw_unordered_plan_items USING pgcontext_hnsw (embedding);
         SET LOCAL enable_seqscan = off;
         SET LOCAL enable_bitmapscan = off",
    )
    .expect("unordered HNSW planner fixture should be created");

    let plan = Spi::connect(|client| {
        let result = client
            .select(
                "EXPLAIN (COSTS TRUE, FORMAT TEXT)
                 SELECT count(*)
                   FROM hnsw_unordered_plan_items
                  WHERE embedding IS NOT NULL",
                None,
                &[],
            )
            .expect("unordered HNSW EXPLAIN query should succeed");

        let mut lines = Vec::new();
        for row in result {
            lines.push(row.get::<String>(1)?.unwrap_or_default());
        }
        Ok::<_, spi::Error>(lines.join("\n"))
    })
    .expect("unordered HNSW EXPLAIN plan should decode");

    assert!(
        plan.contains("Seq Scan on hnsw_unordered_plan_items"),
        "unordered HNSW must remain planner-disabled:\n{plan}"
    );
    assert!(
        !plan.contains("Index Only Scan"),
        "an AM without amcanreturn must not be selected for index-only scans:\n{plan}"
    );
}

#[pg_test]
fn hnsw_scan_state_memory_context_cleanup_runs_exactly_once() {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    struct DropProbe(Arc<AtomicUsize>);

    impl Drop for DropProbe {
        fn drop(&mut self) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }
    }

    let drops = Arc::new(AtomicUsize::new(0));
    let mut context = pgrx::PgMemoryContexts::new("pgcontext hnsw scan cleanup test");
    let _probe = context.leak_and_drop_on_delete(DropProbe(Arc::clone(&drops)));

    // SAFETY: The dedicated test context owns only the registered probe and no
    // pointer allocated from it is used after reset.
    unsafe { context.reset() };
    assert_eq!(drops.load(Ordering::SeqCst), 1);

    // SAFETY: The first reset removed and invoked the callback. Resetting the
    // still-live empty context proves the callback cannot double-drop state.
    unsafe { context.reset() };
    assert_eq!(drops.load(Ordering::SeqCst), 1);
}

#[pg_test]
fn hnsw_build_over_budget_reports_hint_with_suggested_setting() {
    Spi::run(
        "CREATE TABLE hnsw_over_budget (id bigint PRIMARY KEY, \
         embedding vector(384) NOT NULL)",
    )
    .expect("over-budget test table should be created");
    Spi::run(
        "INSERT INTO hnsw_over_budget \
         SELECT n, \
                (SELECT '[' || string_agg(((n * 31 + d) % 97)::text, ',') || ']' \
                   FROM generate_series(1, 384) d)::vector \
           FROM generate_series(1, 2000) n",
    )
    .expect("over-budget test rows should insert");
    Spi::run("SET maintenance_work_mem = '1MB'").expect("budget should apply");

    let caught = PgTryBuilder::new(|| {
        Spi::run(
            "CREATE INDEX hnsw_over_budget_idx ON hnsw_over_budget \
             USING pgcontext_hnsw (embedding pgcontext.vector_hnsw_cosine_ops)",
        )
        .expect("over-budget build should not succeed");
        false
    })
    .catch_when(PgSqlErrorCode::ERRCODE_INVALID_PARAMETER_VALUE, |error| {
        let message = caught_error_message(&error).to_owned();
        assert!(
            message.contains("exceeds maintenance_work_mem budget"),
            "unexpected over-budget message: {message}"
        );
        let hint = caught_error_hint(&error)
            .expect("over-budget build error should carry a sizing hint");
        assert!(
            hint.contains("SET maintenance_work_mem"),
            "hint should carry an actionable setting suggestion: {hint}"
        );
        true
    })
    .execute();

    assert!(caught, "expected over-budget HNSW build to fail with 22023");
    Spi::run("RESET maintenance_work_mem").expect("budget should reset");
}

fn caught_error_hint(error: &pg_sys::panic::CaughtError) -> Option<String> {
    match error {
        pg_sys::panic::CaughtError::PostgresError(report)
        | pg_sys::panic::CaughtError::ErrorReport(report) => report.hint().map(str::to_owned),
        pg_sys::panic::CaughtError::RustPanic { ereport, .. } => {
            ereport.hint().map(str::to_owned)
        }
    }
}
