#[pg_test]
fn index_status_reports_ready_index_metadata() {
    Spi::run(
        "CREATE TABLE m10_index_status_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            value integer NOT NULL
         )",
    )
    .expect("index status fixture table should be created");
    Spi::run(
        "CREATE INDEX m10_index_status_items_value_idx
            ON m10_index_status_items (value)",
    )
    .expect("index status fixture index should be created");

    let row = Spi::connect(|client| {
        let rows = client.select(
            "SELECT index_schema,
                    index_name,
                    table_schema,
                    table_name,
                    access_method,
                    is_valid,
                    is_ready,
                    is_live,
                    status::text
               FROM pgcontext.index_status('public.m10_index_status_items_value_idx')",
            None,
            &[],
        )?;
        let row = rows.first();
        Ok::<_, spi::Error>((
            row.get::<String>(1)?.expect("index_schema should not be null"),
            row.get::<String>(2)?.expect("index_name should not be null"),
            row.get::<String>(3)?.expect("table_schema should not be null"),
            row.get::<String>(4)?.expect("table_name should not be null"),
            row.get::<String>(5)?.expect("access_method should not be null"),
            row.get::<bool>(6)?.expect("is_valid should not be null"),
            row.get::<bool>(7)?.expect("is_ready should not be null"),
            row.get::<bool>(8)?.expect("is_live should not be null"),
            row.get::<String>(9)?.expect("status should not be null"),
        ))
    })
    .expect("index status row should be returned");

    assert_eq!(row.0, "public");
    assert_eq!(row.1, "m10_index_status_items_value_idx");
    assert_eq!(row.2, "public");
    assert_eq!(row.3, "m10_index_status_items");
    assert_eq!(row.4, "btree");
    assert!(row.5);
    assert!(row.6);
    assert!(row.7);
    assert_eq!(row.8, "Ready");
}

#[pg_test]
#[should_panic(expected = "index does not exist: public.m10_index_status_missing_idx")]
fn index_status_rejects_missing_indexes() {
    Spi::run("SELECT * FROM pgcontext.index_status('public.m10_index_status_missing_idx')")
        .expect("missing index status should fail");
}

#[pg_test]
fn index_diagnostics_reports_ready_hnsw_index() {
    Spi::run(
        "CREATE TABLE m10_diag_ready_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            embedding vector NOT NULL
         )",
    )
    .expect("diagnostic ready fixture table should be created");
    Spi::run(
        "INSERT INTO m10_diag_ready_items (embedding)
         VALUES ('[1,2]'::vector), ('[2,3]'::vector)",
    )
    .expect("diagnostic ready fixture rows should be inserted");
    Spi::run(
        "CREATE INDEX m10_diag_ready_items_embedding_idx
            ON m10_diag_ready_items USING pgcontext_hnsw (embedding)",
    )
    .expect("diagnostic ready HNSW index should be created");

    let row = index_diagnostic_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                status::text,
                context_error,
                sqlstate,
                repair_advice
           FROM pgcontext.index_diagnostics('public.m10_diag_ready_items_embedding_idx')",
    );

    assert_eq!(row.0, "public");
    assert_eq!(row.1, "m10_diag_ready_items_embedding_idx");
    assert_eq!(row.2, "public");
    assert_eq!(row.3, "m10_diag_ready_items");
    assert_eq!(row.4, "pgcontext_hnsw");
    assert_eq!(row.5, "Ready");
    assert_eq!(row.6, None);
    assert_eq!(row.7, None);
    assert_eq!(row.8, "No repair action is currently recommended.");
}

#[pg_test]
fn index_diagnostics_reports_unsupported_access_method() {
    Spi::run(
        "CREATE TABLE m10_diag_unsupported_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            value integer NOT NULL
         )",
    )
    .expect("diagnostic unsupported fixture table should be created");
    Spi::run(
        "CREATE INDEX m10_diag_unsupported_items_value_idx
            ON m10_diag_unsupported_items (value)",
    )
    .expect("diagnostic unsupported btree index should be created");

    let row = index_diagnostic_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                status::text,
                context_error,
                sqlstate,
                repair_advice
           FROM pgcontext.index_diagnostics('public.m10_diag_unsupported_items_value_idx')",
    );

    assert_eq!(row.4, "btree");
    assert_eq!(row.5, "UnsupportedAccessMethod");
    assert_eq!(row.6, None);
    assert_eq!(row.7, None);
    assert_eq!(
        row.8,
        "Create a pgcontext_hnsw index for vector serving diagnostics."
    );
}

#[pg_test]
fn index_diagnostics_reports_not_ready_catalog_state() {
    create_hnsw_diagnostic_fixture(
        "m10_diag_not_ready_items",
        "m10_diag_not_ready_items_embedding_idx",
    );
    Spi::run(
        "UPDATE pg_catalog.pg_index
            SET indisready = false
          WHERE indexrelid = 'public.m10_diag_not_ready_items_embedding_idx'::regclass",
    )
    .expect("diagnostic fixture should allow catalog readiness mutation");

    let row = index_diagnostic_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                status::text,
                context_error,
                sqlstate,
                repair_advice
           FROM pgcontext.index_diagnostics('public.m10_diag_not_ready_items_embedding_idx')",
    );

    assert_eq!(row.5, "IndexNotReady");
    assert_eq!(row.6, Some("IndexNotReady".to_owned()));
    assert_eq!(row.7, Some("55000".to_owned()));
    assert_eq!(
        row.8,
        "Wait for index build completion before using the pgcontext_hnsw serving path."
    );
}

#[pg_test]
fn index_diagnostics_reports_corrupt_catalog_state() {
    create_hnsw_diagnostic_fixture("m10_diag_corrupt_items", "m10_diag_corrupt_items_embedding_idx");
    Spi::run(
        "UPDATE pg_catalog.pg_index
            SET indisvalid = false
          WHERE indexrelid = 'public.m10_diag_corrupt_items_embedding_idx'::regclass",
    )
    .expect("diagnostic fixture should allow catalog validity mutation");

    let row = index_diagnostic_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                status::text,
                context_error,
                sqlstate,
                repair_advice
           FROM pgcontext.index_diagnostics('public.m10_diag_corrupt_items_embedding_idx')",
    );

    assert_eq!(row.5, "IndexCorrupt");
    assert_eq!(row.6, Some("IndexCorrupt".to_owned()));
    assert_eq!(row.7, Some("XX001".to_owned()));
    assert_eq!(
        row.8,
        "Run REINDEX, or drop and recreate the pgcontext_hnsw index from the source table."
    );
}

#[pg_test]
fn index_diagnostics_rejects_missing_indexes_with_stable_sqlstate() {
    let caught = PgTryBuilder::new(|| {
        Spi::run("SELECT * FROM pgcontext.index_diagnostics('public.m10_diag_missing_idx')")
            .expect("missing index diagnostics should fail");
        false
    })
    .catch_when(PgSqlErrorCode::ERRCODE_UNDEFINED_OBJECT, |_| true)
    .execute();

    assert!(caught, "missing index diagnostics must use SQLSTATE 42704");
}

#[pg_test]
fn estimate_index_memory_reports_hnsw_projection() {
    Spi::run(
        "CREATE TABLE m10_memory_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            embedding vector NOT NULL
         )",
    )
    .expect("memory fixture table should be created");
    Spi::run(
        "INSERT INTO m10_memory_items (embedding)
         VALUES ('[1,2,3]'::vector), ('[2,3,4]'::vector), ('[3,4,5]'::vector)",
    )
    .expect("memory fixture rows should be inserted");
    Spi::run(
        "CREATE INDEX m10_memory_items_embedding_idx
            ON m10_memory_items USING pgcontext_hnsw (embedding)",
    )
    .expect("memory fixture HNSW index should be created");

    let row = estimate_index_memory_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                estimated_rows,
                dimensions,
                vector_bytes,
                link_bytes,
                total_bytes,
                status::text
           FROM pgcontext.estimate_index_memory('public.m10_memory_items_embedding_idx')",
    );

    assert_eq!(row.0, "public");
    assert_eq!(row.1, "m10_memory_items_embedding_idx");
    assert_eq!(row.2, "public");
    assert_eq!(row.3, "m10_memory_items");
    assert_eq!(row.4, "pgcontext_hnsw");
    assert_eq!(row.5, 3);
    assert_eq!(row.6, 3);
    assert_eq!(row.7, 36);
    assert!(row.8 > 0);
    assert_eq!(row.9, row.7 + row.8);
    assert_eq!(row.10, "Projected");
}

#[pg_test]
fn estimate_index_memory_reports_unsupported_access_method() {
    Spi::run(
        "CREATE TABLE m10_memory_unsupported_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            value integer NOT NULL
         )",
    )
    .expect("unsupported memory fixture table should be created");
    Spi::run(
        "CREATE INDEX m10_memory_unsupported_items_value_idx
            ON m10_memory_unsupported_items (value)",
    )
    .expect("unsupported memory fixture btree index should be created");

    let row = estimate_index_memory_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                estimated_rows,
                dimensions,
                vector_bytes,
                link_bytes,
                total_bytes,
                status::text
           FROM pgcontext.estimate_index_memory(
             'public.m10_memory_unsupported_items_value_idx'
           )",
    );

    assert_eq!(row.4, "btree");
    assert_eq!(row.5, 0);
    assert_eq!(row.9, 0);
    assert_eq!(row.10, "UnsupportedAccessMethod");
}

#[pg_test]
#[should_panic(expected = "index does not exist: public.m10_memory_missing_idx")]
fn estimate_index_memory_rejects_missing_indexes() {
    Spi::run("SELECT * FROM pgcontext.estimate_index_memory('public.m10_memory_missing_idx')")
        .expect("missing index memory estimate should fail");
}

#[pg_test]
fn optimization_status_reports_exact_only_collection() {
    create_optimization_collection("m10_opt_exact");
    Spi::run("SELECT pgcontext.upsert_points('m10_opt_exact', ARRAY['10', '20'])")
        .expect("optimization points should be registered");
    Spi::run("SELECT pgcontext.register_filter_column('m10_opt_exact', 'tenant_id', 'tenant_id')")
        .expect("optimization filter should be registered");

    let row = optimization_status_row(
        "SELECT collection_name,
                table_schema,
                table_name,
                has_source_table,
                source_table_exists,
                registered_vectors,
                active_points,
                filter_fields,
                hnsw_indexes,
                status::text
           FROM pgcontext.optimization_status('m10_opt_exact')",
    );

    assert_eq!(row.0, "m10_opt_exact");
    assert_eq!(row.1, Some("public".to_owned()));
    assert_eq!(row.2, Some("m10_opt_exact".to_owned()));
    assert!(row.3);
    assert!(row.4);
    assert_eq!(row.5, 1);
    assert_eq!(row.6, 2);
    assert_eq!(row.7, 1);
    assert_eq!(row.8, 0);
    assert_eq!(row.9, "ExactOnly");
}

#[pg_test]
fn optimization_status_reports_indexed_collection() {
    create_optimization_collection("m10_opt_indexed");
    Spi::run(
        "CREATE INDEX m10_opt_indexed_embedding_idx
            ON m10_opt_indexed USING pgcontext_hnsw (embedding)",
    )
    .expect("optimization HNSW index should be created");

    let row = optimization_status_row(
        "SELECT collection_name,
                table_schema,
                table_name,
                has_source_table,
                source_table_exists,
                registered_vectors,
                active_points,
                filter_fields,
                hnsw_indexes,
                status::text
           FROM pgcontext.optimization_status('m10_opt_indexed')",
    );

    assert_eq!(row.8, 1);
    assert_eq!(row.9, "Indexed");
}

#[pg_test]
fn optimization_status_reports_missing_artifacts() {
    Spi::run("SELECT pgcontext.create_collection('m10_opt_missing_artifacts')")
        .expect("collection without source table should be created");

    let row = optimization_status_row(
        "SELECT collection_name,
                table_schema,
                table_name,
                has_source_table,
                source_table_exists,
                registered_vectors,
                active_points,
                filter_fields,
                hnsw_indexes,
                status::text
           FROM pgcontext.optimization_status('m10_opt_missing_artifacts')",
    );

    assert_eq!(row.0, "m10_opt_missing_artifacts");
    assert_eq!(row.1, None);
    assert!(!row.3);
    assert!(!row.4);
    assert_eq!(row.5, 0);
    assert_eq!(row.9, "MissingArtifacts");
}

#[pg_test]
fn optimization_status_reports_stale_catalog() {
    create_optimization_collection("m10_opt_stale");
    Spi::run("DROP TABLE public.m10_opt_stale").expect("source table should be dropped");

    let row = optimization_status_row(
        "SELECT collection_name,
                table_schema,
                table_name,
                has_source_table,
                source_table_exists,
                registered_vectors,
                active_points,
                filter_fields,
                hnsw_indexes,
                status::text
           FROM pgcontext.optimization_status('m10_opt_stale')",
    );

    assert_eq!(row.0, "m10_opt_stale");
    assert!(row.3);
    assert!(!row.4);
    assert_eq!(row.5, 1);
    assert_eq!(row.9, "StaleCatalog");
}

#[pg_test]
#[should_panic(expected = "collection does not exist: m10_opt_missing")]
fn optimization_status_rejects_missing_collection() {
    Spi::run("SELECT * FROM pgcontext.optimization_status('m10_opt_missing')")
        .expect("missing optimization collection should fail");
}

#[pg_test]
fn vacuum_advice_reports_hnsw_index_stats() {
    Spi::run(
        "CREATE TABLE m10_vacuum_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            embedding vector NOT NULL
         )",
    )
    .expect("vacuum fixture table should be created");
    Spi::run(
        "INSERT INTO m10_vacuum_items (embedding)
         VALUES ('[1,2]'::vector), ('[2,3]'::vector)",
    )
    .expect("vacuum fixture rows should be inserted");
    Spi::run(
        "CREATE INDEX m10_vacuum_items_embedding_idx
            ON m10_vacuum_items USING pgcontext_hnsw (embedding)",
    )
    .expect("vacuum fixture HNSW index should be created");

    let row = vacuum_advice_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                estimated_index_tuples,
                index_pages,
                dead_table_tuples,
                status::text
           FROM pgcontext.vacuum_advice('public.m10_vacuum_items_embedding_idx')",
    );

    assert_eq!(row.0, "public");
    assert_eq!(row.1, "m10_vacuum_items_embedding_idx");
    assert_eq!(row.2, "public");
    assert_eq!(row.3, "m10_vacuum_items");
    assert_eq!(row.4, "pgcontext_hnsw");
    assert_eq!(row.5, 2);
    assert!(row.6 >= 1);
    assert_eq!(row.7, 0);
    assert_eq!(row.8, "Healthy");
}

#[pg_test]
fn vacuum_advice_reports_unsupported_access_method() {
    Spi::run(
        "CREATE TABLE m10_vacuum_unsupported_items (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            value integer NOT NULL
         )",
    )
    .expect("unsupported vacuum fixture table should be created");
    Spi::run(
        "CREATE INDEX m10_vacuum_unsupported_items_value_idx
            ON m10_vacuum_unsupported_items (value)",
    )
    .expect("unsupported vacuum fixture btree index should be created");

    let row = vacuum_advice_row(
        "SELECT index_schema,
                index_name,
                table_schema,
                table_name,
                access_method,
                estimated_index_tuples,
                index_pages,
                dead_table_tuples,
                status::text
           FROM pgcontext.vacuum_advice('public.m10_vacuum_unsupported_items_value_idx')",
    );

    assert_eq!(row.4, "btree");
    assert_eq!(row.8, "UnsupportedAccessMethod");
}

#[pg_test]
#[should_panic(expected = "index does not exist: public.m10_vacuum_missing_idx")]
fn vacuum_advice_rejects_missing_indexes() {
    Spi::run("SELECT * FROM pgcontext.vacuum_advice('public.m10_vacuum_missing_idx')")
        .expect("missing vacuum index should fail");
}

#[pg_test]
fn telemetry_reports_collection_rollups() {
    create_optimization_collection("m10_tel_active");
    Spi::run("SELECT pgcontext.upsert_points('m10_tel_active', ARRAY['10', '20'])")
        .expect("telemetry points should be registered");
    Spi::run("SELECT pgcontext.delete_points('m10_tel_active', ARRAY['20'])")
        .expect("telemetry point should be deleted");
    Spi::run("SELECT pgcontext.register_filter_column('m10_tel_active', 'tenant_id', 'tenant_id')")
        .expect("telemetry filter should be registered");
    Spi::run(
        "CREATE INDEX m10_tel_active_embedding_idx
            ON m10_tel_active USING pgcontext_hnsw (embedding)",
    )
    .expect("telemetry HNSW index should be created");

    Spi::run("SELECT pgcontext.create_collection('m10_tel_missing')")
        .expect("telemetry missing-artifacts collection should be created");

    create_optimization_collection("m10_tel_stale");
    Spi::run("DROP TABLE public.m10_tel_stale").expect("telemetry stale table should be dropped");

    let rows = telemetry_rows(
        "SELECT collection_name,
                table_schema,
                table_name,
                has_source_table,
                source_table_exists,
                registered_vectors,
                active_points,
                deleted_points,
                filter_fields,
                hnsw_indexes,
                status::text
           FROM pgcontext.telemetry()
          WHERE collection_name LIKE 'm10_tel_%'
          ORDER BY collection_name",
    );

    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].0, "m10_tel_active");
    assert_eq!(rows[0].1, Some("public".to_owned()));
    assert_eq!(rows[0].2, Some("m10_tel_active".to_owned()));
    assert!(rows[0].3);
    assert!(rows[0].4);
    assert_eq!(rows[0].5, 1);
    assert_eq!(rows[0].6, 1);
    assert_eq!(rows[0].7, 1);
    assert_eq!(rows[0].8, 1);
    assert_eq!(rows[0].9, 1);
    assert_eq!(rows[0].10, "Active");

    assert_eq!(rows[1].0, "m10_tel_missing");
    assert_eq!(rows[1].1, None);
    assert!(!rows[1].3);
    assert_eq!(rows[1].5, 0);
    assert_eq!(rows[1].10, "MissingArtifacts");

    assert_eq!(rows[2].0, "m10_tel_stale");
    assert!(rows[2].3);
    assert!(!rows[2].4);
    assert_eq!(rows[2].5, 1);
    assert_eq!(rows[2].10, "StaleCatalog");
}

#[pg_test]
fn recall_check_reports_passing_structured_counts() {
    let row = recall_check_row(
        "SELECT exact_count,
                candidate_count,
                intersection_count,
                recall,
                status::text
           FROM pgcontext.recall_check(ARRAY[1,2,3,3]::bigint[], ARRAY[2,3,4]::bigint[], 0.66)",
    );

    assert_eq!(row.0, 3);
    assert_eq!(row.1, 3);
    assert_eq!(row.2, 2);
    assert!((row.3 - 0.666_666_666_666).abs() < 0.000_000_001);
    assert_eq!(row.4, "Passing");
}

#[pg_test]
fn recall_check_reports_failing_status() {
    let row = recall_check_row(
        "SELECT exact_count,
                candidate_count,
                intersection_count,
                recall,
                status::text
           FROM pgcontext.recall_check(ARRAY[1,2,3]::bigint[], ARRAY[3]::bigint[], 0.5)",
    );

    assert_eq!(row.0, 3);
    assert_eq!(row.1, 1);
    assert_eq!(row.2, 1);
    assert_eq!(row.4, "Failing");
}

#[pg_test]
fn recall_check_accepts_exact_search_oracle_output() {
    let exact_rows = exact_oracle_rows(
        "[0,0]",
        "l2",
        3,
        &[(1, "[3,0]"), (2, "[1,0]"), (3, "[2,0]"), (4, "[9,0]")],
    );
    let exact_ids = exact_rows
        .iter()
        .map(|(point_id, _score)| point_id.to_string())
        .collect::<Vec<_>>()
        .join(",");

    let row = recall_check_row(&format!(
        "SELECT exact_count,
                candidate_count,
                intersection_count,
                recall,
                status::text
           FROM pgcontext.recall_check(ARRAY[{exact_ids}]::bigint[], ARRAY[2,3,99]::bigint[], 0.66)"
    ));

    assert_eq!(row.0, 3);
    assert_eq!(row.1, 3);
    assert_eq!(row.2, 2);
    assert!((row.3 - 0.666_666_666_666).abs() < 0.000_000_001);
    assert_eq!(row.4, "Passing");
}

#[pg_test]
fn recall_check_reports_empty_exact_status() {
    let row = recall_check_row(
        "SELECT exact_count,
                candidate_count,
                intersection_count,
                recall,
                status::text
           FROM pgcontext.recall_check(ARRAY[]::bigint[], ARRAY[1]::bigint[], 1.0)",
    );

    assert_eq!(row.0, 0);
    assert_eq!(row.1, 1);
    assert_eq!(row.2, 0);
    assert_eq!(row.3, 1.0);
    assert_eq!(row.4, "EmptyExact");
}

#[pg_test]
#[should_panic(expected = "min_recall must be between 0 and 1 inclusive: 1.5")]
fn recall_check_rejects_invalid_min_recall() {
    Spi::run(
        "SELECT * FROM pgcontext.recall_check(ARRAY[1]::bigint[], ARRAY[1]::bigint[], 1.5)",
    )
    .expect("invalid min_recall should fail");
}

#[pg_test]
#[should_panic(expected = "exact_point_ids must not contain negative point IDs: -1")]
fn recall_check_rejects_negative_point_ids() {
    Spi::run(
        "SELECT * FROM pgcontext.recall_check(ARRAY[-1]::bigint[], ARRAY[1]::bigint[], 1.0)",
    )
    .expect("negative point IDs should fail");
}

#[pg_test]
fn recall_check_rejects_inputs_above_policy_budget_with_stable_sqlstate() {
    let exact_ids = (1..=10_001)
        .map(|point_id| point_id.to_string())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT *
           FROM pgcontext.recall_check(ARRAY[{exact_ids}]::bigint[], ARRAY[1]::bigint[], 1.0)"
    );

    let caught = PgTryBuilder::new(|| {
        Spi::run(&sql).expect("over-budget recall check should fail");
        false
    })
    .catch_when(PgSqlErrorCode::ERRCODE_PROGRAM_LIMIT_EXCEEDED, |_| true)
    .execute();

    assert!(caught, "recall budget failures must use SQLSTATE 54000");
}

fn recall_check_row(sql: &str) -> (i64, i64, i64, f64, String) {
    Spi::connect(|client| {
        let rows = client.select(sql, None, &[])?;
        let row = rows.first();
        Ok::<_, spi::Error>((
            row.get::<i64>(1)?.expect("exact_count should not be null"),
            row.get::<i64>(2)?
                .expect("candidate_count should not be null"),
            row.get::<i64>(3)?
                .expect("intersection_count should not be null"),
            row.get::<f64>(4)?.expect("recall should not be null"),
            row.get::<String>(5)?.expect("status should not be null"),
        ))
    })
    .expect("recall check row should be returned")
}

fn create_hnsw_diagnostic_fixture(table_name: &str, index_name: &str) {
    Spi::run(&format!(
        "CREATE TABLE public.{table_name} (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            embedding vector NOT NULL
         )"
    ))
    .expect("diagnostic HNSW fixture table should be created");
    Spi::run(&format!(
        "INSERT INTO public.{table_name} (embedding)
         VALUES ('[1,2]'::vector), ('[2,3]'::vector)"
    ))
    .expect("diagnostic HNSW fixture rows should be inserted");
    Spi::run(&format!(
        "CREATE INDEX {index_name}
            ON public.{table_name} USING pgcontext_hnsw (embedding)"
    ))
    .expect("diagnostic HNSW fixture index should be created");
}

#[allow(
    clippy::type_complexity,
    reason = "test assertions mirror the SQL result shape"
)]
fn index_diagnostic_row(
    sql: &str,
) -> (
    String,
    String,
    String,
    String,
    String,
    String,
    Option<String>,
    Option<String>,
    String,
) {
    Spi::connect(|client| {
        let rows = client.select(sql, None, &[])?;
        let row = rows.first();
        Ok::<_, spi::Error>((
            row.get::<String>(1)?.expect("index_schema should not be null"),
            row.get::<String>(2)?.expect("index_name should not be null"),
            row.get::<String>(3)?.expect("table_schema should not be null"),
            row.get::<String>(4)?.expect("table_name should not be null"),
            row.get::<String>(5)?.expect("access_method should not be null"),
            row.get::<String>(6)?.expect("status should not be null"),
            row.get::<String>(7)?,
            row.get::<String>(8)?,
            row.get::<String>(9)?
                .expect("repair_advice should not be null"),
        ))
    })
    .expect("index diagnostic row should be returned")
}

#[allow(
    clippy::type_complexity,
    reason = "test assertions mirror the SQL result shape"
)]
fn estimate_index_memory_row(
    sql: &str,
) -> (
    String,
    String,
    String,
    String,
    String,
    i64,
    i32,
    i64,
    i64,
    i64,
    String,
) {
    Spi::connect(|client| {
        let rows = client.select(sql, None, &[])?;
        let row = rows.first();
        Ok::<_, spi::Error>((
            row.get::<String>(1)?.expect("index_schema should not be null"),
            row.get::<String>(2)?.expect("index_name should not be null"),
            row.get::<String>(3)?.expect("table_schema should not be null"),
            row.get::<String>(4)?.expect("table_name should not be null"),
            row.get::<String>(5)?.expect("access_method should not be null"),
            row.get::<i64>(6)?.expect("estimated_rows should not be null"),
            row.get::<i32>(7)?.expect("dimensions should not be null"),
            row.get::<i64>(8)?.expect("vector_bytes should not be null"),
            row.get::<i64>(9)?.expect("link_bytes should not be null"),
            row.get::<i64>(10)?.expect("total_bytes should not be null"),
            row.get::<String>(11)?.expect("status should not be null"),
        ))
    })
    .expect("index memory estimate row should be returned")
}

fn create_optimization_collection(collection_name: &str) {
    Spi::run(&format!(
        "CREATE TABLE public.{collection_name} (
             id bigint PRIMARY KEY,
             embedding vector NOT NULL,
             tenant_id text NOT NULL
         )"
    ))
    .expect("optimization source table should be created");
    Spi::run(&format!(
        "INSERT INTO public.{collection_name} (id, embedding, tenant_id)
         VALUES (10, '[1,2]'::vector, 'acme'), (20, '[2,3]'::vector, 'acme')"
    ))
    .expect("optimization source rows should be inserted");
    Spi::run(&format!(
        "SELECT pgcontext.create_collection('{collection_name}', 'public.{collection_name}')"
    ))
    .expect("optimization collection should be created");
    Spi::run(&format!(
        "SELECT pgcontext.register_vector('{collection_name}', 'embedding', 'embedding', 2, 'l2')"
    ))
    .expect("optimization vector should be registered");
}

#[allow(
    clippy::type_complexity,
    reason = "test assertions mirror the SQL result shape"
)]
fn optimization_status_row(
    sql: &str,
) -> (
    String,
    Option<String>,
    Option<String>,
    bool,
    bool,
    i64,
    i64,
    i64,
    i64,
    String,
) {
    Spi::connect(|client| {
        let rows = client.select(sql, None, &[])?;
        let row = rows.first();
        Ok::<_, spi::Error>((
            row.get::<String>(1)?
                .expect("collection_name should not be null"),
            row.get::<String>(2)?,
            row.get::<String>(3)?,
            row.get::<bool>(4)?
                .expect("has_source_table should not be null"),
            row.get::<bool>(5)?
                .expect("source_table_exists should not be null"),
            row.get::<i64>(6)?
                .expect("registered_vectors should not be null"),
            row.get::<i64>(7)?.expect("active_points should not be null"),
            row.get::<i64>(8)?.expect("filter_fields should not be null"),
            row.get::<i64>(9)?.expect("hnsw_indexes should not be null"),
            row.get::<String>(10)?.expect("status should not be null"),
        ))
    })
    .expect("optimization status row should be returned")
}

fn vacuum_advice_row(sql: &str) -> (String, String, String, String, String, i64, i64, i64, String) {
    Spi::connect(|client| {
        let rows = client.select(sql, None, &[])?;
        let row = rows.first();
        Ok::<_, spi::Error>((
            row.get::<String>(1)?.expect("index_schema should not be null"),
            row.get::<String>(2)?.expect("index_name should not be null"),
            row.get::<String>(3)?.expect("table_schema should not be null"),
            row.get::<String>(4)?.expect("table_name should not be null"),
            row.get::<String>(5)?.expect("access_method should not be null"),
            row.get::<i64>(6)?
                .expect("estimated_index_tuples should not be null"),
            row.get::<i64>(7)?.expect("index_pages should not be null"),
            row.get::<i64>(8)?
                .expect("dead_table_tuples should not be null"),
            row.get::<String>(9)?.expect("status should not be null"),
        ))
    })
    .expect("vacuum advice row should be returned")
}

type TelemetryTestRow = (
    String,
    Option<String>,
    Option<String>,
    bool,
    bool,
    i64,
    i64,
    i64,
    i64,
    i64,
    String,
);

fn telemetry_rows(sql: &str) -> Vec<TelemetryTestRow> {
    Spi::connect(|client| {
        let rows = client.select(sql, None, &[])?;
        let mut output = Vec::new();
        for row in rows {
            output.push((
                row.get::<String>(1)?
                    .expect("collection_name should not be null"),
                row.get::<String>(2)?,
                row.get::<String>(3)?,
                row.get::<bool>(4)?
                    .expect("has_source_table should not be null"),
                row.get::<bool>(5)?
                    .expect("source_table_exists should not be null"),
                row.get::<i64>(6)?
                    .expect("registered_vectors should not be null"),
                row.get::<i64>(7)?.expect("active_points should not be null"),
                row.get::<i64>(8)?.expect("deleted_points should not be null"),
                row.get::<i64>(9)?.expect("filter_fields should not be null"),
                row.get::<i64>(10)?.expect("hnsw_indexes should not be null"),
                row.get::<String>(11)?.expect("status should not be null"),
            ));
        }
        Ok::<_, spi::Error>(output)
    })
    .expect("telemetry rows should be returned")
}
