oliphaunt-wasix 0.1.0

#![cfg(feature = "extensions")]

use anyhow::Result;
use oliphaunt_wasix::{Oliphaunt, OliphauntError, OliphauntServer, extensions};
use serde_json::json;
use sqlx::{Connection, Row};
use std::path::{Path, PathBuf};

struct TestTrace {
    name: &'static str,
}

impl TestTrace {
    fn new(name: &'static str) -> Self {
        eprintln!("extensions_smoke::{name} start");
        Self { name }
    }
}

impl Drop for TestTrace {
    fn drop(&mut self) {
        eprintln!("extensions_smoke::{} end", self.name);
    }
}

fn trace_expected(label: &str) {
    eprintln!("extensions_smoke::expected_sql_error exercising {label}");
}

fn first_f64(result: &oliphaunt_wasix::Results, column: &str) -> f64 {
    result.rows[0][column].as_f64().expect("floating result")
}

fn assert_oliphaunt_code(err: &anyhow::Error, expected_code: &str, message_contains: &str) {
    let pg_err = err
        .downcast_ref::<OliphauntError>()
        .expect("error should preserve Postgres fields");
    assert_eq!(pg_err.database_error().code.as_deref(), Some(expected_code));
    assert!(
        pg_err.database_error().message.contains(message_contains),
        "expected error message to contain {message_contains:?}, got {:?}",
        pg_err.database_error().message
    );
}

const PGCRYPTO_FUNCTIONAL_SMOKE_SQL: &str = "\
DO $$ \
DECLARE \
  hashed text; \
  encrypted bytea; \
  armored text; \
  header_count int; \
  crypto_key bytea := decode('000102030405060708090a0b0c0d0e0f', 'hex'); \
  crypto_iv bytea := decode('101112131415161718191a1b1c1d1e1f', 'hex'); \
BEGIN \
  IF encode(digest('abc', 'sha256'), 'hex') <> 'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad' THEN \
    RAISE EXCEPTION 'sha256 digest failed'; \
  END IF; \
  IF encode(hmac('test', 'key', 'sha1'), 'hex') <> '671f54ce0c540f78ffe1e26dcf9c2a047aea4fda' THEN \
    RAISE EXCEPTION 'hmac failed'; \
  END IF; \
  IF length(gen_random_bytes(16)) <> 16 THEN \
    RAISE EXCEPTION 'random bytes length failed'; \
  END IF; \
  IF gen_random_uuid()::text !~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' THEN \
    RAISE EXCEPTION 'random uuid format failed'; \
  END IF; \
  SELECT crypt('secret', gen_salt('bf', 4)) INTO hashed; \
  IF crypt('secret', hashed) <> hashed THEN \
    RAISE EXCEPTION 'password hash verify failed'; \
  END IF; \
  SELECT armor(digest('test', 'sha1'), ARRAY['Version'], ARRAY['oliphaunt']) INTO armored; \
  IF position('Version: oliphaunt' in armored) = 0 THEN \
    RAISE EXCEPTION 'armor header failed'; \
  END IF; \
  SELECT count(*) INTO header_count FROM pgp_armor_headers(armored); \
  IF header_count <> 1 THEN \
    RAISE EXCEPTION 'armor header count failed: %', header_count; \
  END IF; \
  SELECT pgp_sym_encrypt('oliphaunt secret', 'passphrase') INTO encrypted; \
  IF pgp_sym_decrypt(encrypted, 'passphrase') <> 'oliphaunt secret' THEN \
    RAISE EXCEPTION 'PGP symmetric decrypt failed'; \
  END IF; \
  IF pgp_key_id(encrypted) <> 'SYMKEY' THEN \
    RAISE EXCEPTION 'PGP symmetric key id failed'; \
  END IF; \
  SELECT encrypt(convert_to('oliphaunt raw cipher', 'UTF8'), crypto_key, 'aes') INTO encrypted; \
  IF convert_from(decrypt(encrypted, crypto_key, 'aes'), 'UTF8') <> 'oliphaunt raw cipher' THEN \
    RAISE EXCEPTION 'raw decrypt failed'; \
  END IF; \
  SELECT encrypt_iv(convert_to('oliphaunt iv cipher', 'UTF8'), crypto_key, crypto_iv, 'aes-cbc') INTO encrypted; \
  IF convert_from(decrypt_iv(encrypted, crypto_key, crypto_iv, 'aes-cbc'), 'UTF8') <> 'oliphaunt iv cipher' THEN \
    RAISE EXCEPTION 'raw iv decrypt failed'; \
  END IF; \
END $$";

fn assert_sqlx_code(err: &sqlx::Error, expected_code: &str) {
    assert_eq!(
        err.as_database_error().and_then(|db| db.code()).as_deref(),
        Some(expected_code)
    );
}

fn assert_only_requested_extension_assets_are_materialized(
    root: &Path,
    requested: &str,
    unrequested: &str,
) {
    let runtime = root.join("tmp/oliphaunt");
    assert!(
        runtime
            .join(format!("lib/postgresql/{requested}.so"))
            .is_file(),
        "requested extension side module should be materialized in the upper runtime layer"
    );
    assert!(
        runtime
            .join(format!("share/postgresql/extension/{requested}.control"))
            .is_file(),
        "requested extension control file should be materialized in the upper runtime layer"
    );
    assert!(
        !runtime
            .join(format!("lib/postgresql/{unrequested}.so"))
            .exists(),
        "unrequested extension side module should not be materialized"
    );
    let lib_files = relative_files(&runtime.join("lib/postgresql"));
    assert_eq!(
        lib_files,
        vec![PathBuf::from(format!("{requested}.so"))],
        "upper runtime library layer should contain only the requested extension side module"
    );
    let share_files = relative_files(&runtime.join("share/postgresql"));
    assert!(
        share_files.iter().all(|path| {
            path.parent() == Some(Path::new("extension"))
                && path
                    .file_name()
                    .and_then(|name| name.to_str())
                    .is_some_and(|name| {
                        name == format!("{requested}.control")
                            || name == format!("{requested}.sql")
                            || name.starts_with(&format!("{requested}--"))
                    })
        }),
        "upper runtime share layer should contain only requested extension metadata, got {share_files:?}"
    );
    assert!(
        !runtime.join("bin").exists(),
        "core binaries should stay in the lower cached runtime"
    );
    assert!(
        !runtime.join("lib/postgresql/plpgsql.so").exists(),
        "core runtime side modules should stay in the lower cached runtime"
    );
    assert!(
        !runtime.join("share/postgresql/postgres.bki").exists(),
        "core catalog files should stay in the lower cached runtime"
    );
}

fn relative_files(root: &Path) -> Vec<PathBuf> {
    fn walk(base: &Path, current: &Path, files: &mut Vec<PathBuf>) {
        if !current.exists() {
            return;
        }
        for entry in std::fs::read_dir(current).expect("read runtime test directory") {
            let entry = entry.expect("read runtime test directory entry");
            let path = entry.path();
            if path.is_dir() {
                walk(base, &path, files);
            } else if path.is_file() {
                files.push(
                    path.strip_prefix(base)
                        .expect("relative file")
                        .to_path_buf(),
                );
            }
        }
    }

    let mut files = Vec::new();
    walk(root, root, &mut files);
    files.sort();
    files
}

#[test]
fn vector_extension_direct_smoke() -> Result<()> {
    let _trace = TestTrace::new("vector_extension_direct_smoke");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extension(extensions::VECTOR)
        .open()?;

    db.exec("CREATE TEMP TABLE oxide_vec (embedding vector(3))", None)?;
    db.exec("INSERT INTO oxide_vec VALUES ('[1,2,3]')", None)?;
    let result = db.query(
        "SELECT embedding <-> '[1,2,4]'::vector AS distance FROM oxide_vec",
        &[],
        None,
    )?;
    assert_eq!(first_f64(&result, "distance"), 1.0);

    let version = db.query(
        "SELECT extversion, n.nspname AS schema_name \
         FROM pg_extension e \
         JOIN pg_namespace n ON n.oid = e.extnamespace \
         WHERE e.extname = 'vector'",
        &[],
        None,
    )?;
    let extversion = version.rows[0]["extversion"]
        .as_str()
        .expect("vector extversion");
    assert!(!extversion.is_empty());
    assert_eq!(version.rows[0]["schema_name"], json!("pg_catalog"));

    trace_expected("vector_direct division-by-zero");
    let err = db
        .query(
            "SELECT 10 / $1::int4 AS impossible_after_vector",
            &[serde_json::json!(0)],
            None,
        )
        .expect_err("division by zero after vector load should fail");
    assert_oliphaunt_code(&err, "22012", "division by zero");
    let recovered = db.query("SELECT 13::int AS recovered_after_vector_error", &[], None)?;
    assert_eq!(recovered.rows[0]["recovered_after_vector_error"], json!(13));

    trace_expected("vector_direct invalid-vector-literal");
    let invalid_vector = db
        .query(
            "SELECT $1::vector AS embedding",
            &[json!("[hello,1]")],
            None,
        )
        .expect_err("invalid vector literal should fail inside the vector extension");
    assert_oliphaunt_code(
        &invalid_vector,
        "22P02",
        "invalid input syntax for type vector",
    );
    let recovered = db.query(
        "SELECT 15::int AS recovered_after_invalid_vector",
        &[],
        None,
    )?;
    assert_eq!(
        recovered.rows[0]["recovered_after_invalid_vector"],
        json!(15)
    );

    trace_expected("vector_direct dimension-mismatch");
    let dimension_mismatch = db
        .query(
            "SELECT $1::vector <-> $2::vector AS distance",
            &[json!("[1,2]"), json!("[3]")],
            None,
        )
        .expect_err("vector distance should reject mismatched dimensions");
    assert_oliphaunt_code(&dimension_mismatch, "22000", "different vector dimensions");
    let recovered = db.query(
        "SELECT 16::int AS recovered_after_dimension_mismatch",
        &[],
        None,
    )?;
    assert_eq!(
        recovered.rows[0]["recovered_after_dimension_mismatch"],
        json!(16)
    );

    db.close()?;
    Ok(())
}

#[test]
fn pure_mountfs_materializes_only_requested_extension_assets() -> Result<()> {
    let _trace = TestTrace::new("pure_mountfs_materializes_only_requested_extension_assets");
    let root = tempfile::TempDir::new()?;
    {
        let mut db = Oliphaunt::builder()
            .path(root.path())
            .extension(extensions::VECTOR)
            .open()?;
        let result = db.query(
            "SELECT '[1,2,3]'::vector <-> '[1,2,4]'::vector AS distance",
            &[],
            None,
        )?;
        assert_eq!(first_f64(&result, "distance"), 1.0);
        db.close()?;
    }

    assert_only_requested_extension_assets_are_materialized(root.path(), "vector", "pg_trgm");
    Ok(())
}

#[test]
fn vector_extension_ports_pgvector_core_type_cases() -> Result<()> {
    let _trace = TestTrace::new("vector_extension_ports_pgvector_core_type_cases");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extension(extensions::VECTOR)
        .open()?;

    let valid = db.query(
        "SELECT \
            '[1,2,3]'::vector::text AS vector_text, \
            vector_dims('[1,2,3]'::vector)::int AS dims, \
            l2_distance('[0,0]'::vector, '[3,4]'::vector)::float8 AS distance",
        &[],
        None,
    )?;
    assert_eq!(valid.rows[0]["vector_text"], json!("[1,2,3]"));
    assert_eq!(valid.rows[0]["dims"], json!(3));
    assert_eq!(first_f64(&valid, "distance"), 5.0);

    for (sql, code, message) in [
        (
            "SELECT '[hello,1]'::vector",
            "22P02",
            "invalid input syntax for type vector",
        ),
        ("SELECT '[NaN,1]'::vector", "22000", "NaN not allowed"),
        (
            "SELECT '[1,2,3]'::vector(2)",
            "22000",
            "expected 2 dimensions, not 3",
        ),
        (
            "SELECT '[1,2]'::vector <-> '[3]'::vector",
            "22000",
            "different vector dimensions",
        ),
    ] {
        trace_expected(&format!("vector_core_type_cases {sql}"));
        let err = match db.query(sql, &[], None) {
            Ok(_) => panic!("{sql} should fail"),
            Err(err) => err,
        };
        assert_oliphaunt_code(&err, code, message);
        let recovered = db.query("SELECT 17::int AS recovered", &[], None)?;
        assert_eq!(recovered.rows[0]["recovered"], json!(17));
    }

    db.close()?;
    Ok(())
}

#[test]
fn vector_extension_direct_transaction_commit_rollback_and_error_recovery() -> Result<()> {
    let _trace =
        TestTrace::new("vector_extension_direct_transaction_commit_rollback_and_error_recovery");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extension(extensions::VECTOR)
        .open()?;
    db.exec(
        "CREATE TABLE vector_tx_items(id int PRIMARY KEY, embedding vector(3))",
        None,
    )?;

    db.transaction(|tx| {
        tx.query(
            "INSERT INTO vector_tx_items(id, embedding) VALUES ($1, $2::vector) \
             RETURNING embedding <-> '[1,2,4]'::vector AS distance",
            &[json!(1), json!("[1,2,3]")],
            None,
        )?;
        Ok::<_, anyhow::Error>(())
    })?;

    let rollback: anyhow::Result<()> = db.transaction(|tx| {
        tx.query(
            "INSERT INTO vector_tx_items(id, embedding) VALUES ($1, $2::vector)",
            &[json!(2), json!("[9,9,9]")],
            None,
        )?;
        Err(anyhow::anyhow!("force vector rollback"))
    });
    assert!(rollback.is_err());

    trace_expected("vector_direct_transaction invalid-vector-literal");
    let failed: anyhow::Result<()> = db.transaction(|tx| {
        tx.query(
            "INSERT INTO vector_tx_items(id, embedding) VALUES ($1, $2::vector)",
            &[json!(3), json!("[3,3,3]")],
            None,
        )?;
        tx.query(
            "SELECT $1::vector AS embedding",
            &[json!("[hello,1]")],
            None,
        )?;
        Ok(())
    });
    let failed = failed.expect_err("invalid vector should fail inside transaction");
    assert_oliphaunt_code(&failed, "22P02", "invalid input syntax for type vector");

    let result = db.query(
        "SELECT count(*)::int AS count, \
                min(embedding <-> '[1,2,4]'::vector)::float8 AS distance \
         FROM vector_tx_items",
        &[],
        None,
    )?;
    assert_eq!(result.rows[0]["count"], json!(1));
    assert_eq!(first_f64(&result, "distance"), 1.0);

    let recovered = db.query("SELECT 44::int AS recovered_after_vector_tx", &[], None)?;
    assert_eq!(recovered.rows[0]["recovered_after_vector_tx"], json!(44));

    db.close()?;
    Ok(())
}

#[test]
fn vector_extension_install_is_demand_driven_idempotent_and_persistent() -> Result<()> {
    let _trace =
        TestTrace::new("vector_extension_install_is_demand_driven_idempotent_and_persistent");
    let root = tempfile::TempDir::new()?;
    {
        let mut db = Oliphaunt::builder().path(root.path()).open()?;
        assert!(
            !db.paths()
                .pgroot
                .join("oliphaunt")
                .join("lib/postgresql/vector.so")
                .exists(),
            "vector side module should not be installed before it is requested"
        );

        db.enable_extension(extensions::VECTOR)?;
        db.enable_extension(extensions::VECTOR)?;
        assert!(
            db.paths()
                .pgroot
                .join("oliphaunt")
                .join("lib/postgresql/vector.so")
                .exists(),
            "vector side module should be installed after enable_extension"
        );

        let installed = db.query(
            "SELECT count(*)::int AS count FROM pg_extension WHERE extname = 'vector'",
            &[],
            None,
        )?;
        assert_eq!(installed.rows[0]["count"], json!(1));
        db.close()?;
    }

    {
        let mut reopened = Oliphaunt::builder().path(root.path()).open()?;
        let result = reopened.query("SELECT '[1,2,3]'::vector::text AS value", &[], None)?;
        assert_eq!(result.rows[0]["value"], json!("[1,2,3]"));
        reopened.close()?;
    }

    Ok(())
}

#[test]
fn pg_trgm_extension_direct_smoke() -> Result<()> {
    let _trace = TestTrace::new("pg_trgm_extension_direct_smoke");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extension(extensions::PG_TRGM)
        .open()?;

    let result = db.query(
        "SELECT similarity('postgres', 'postgrex') AS score",
        &[],
        None,
    )?;
    assert!(first_f64(&result, "score") > 0.5);

    let installed = db.query(
        "SELECT count(*)::int AS count, max(n.nspname) AS schema_name \
         FROM pg_extension e \
         JOIN pg_namespace n ON n.oid = e.extnamespace \
         WHERE e.extname = 'pg_trgm'",
        &[],
        None,
    )?;
    assert_eq!(installed.rows[0]["count"], json!(1));
    assert_eq!(installed.rows[0]["schema_name"], json!("pg_catalog"));

    db.close()?;
    Ok(())
}

#[test]
fn hstore_extension_direct_smoke() -> Result<()> {
    let _trace = TestTrace::new("hstore_extension_direct_smoke");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extension(extensions::HSTORE)
        .open()?;

    db.exec(
        "CREATE TEMP TABLE oxide_hstore (id serial PRIMARY KEY, data hstore)",
        None,
    )?;
    db.exec(
        "INSERT INTO oxide_hstore (data) VALUES ('\"name\"=>\"test1\"'), ('\"name\"=>\"test2\"')",
        None,
    )?;
    let result = db.query(
        "SELECT data::jsonb AS data FROM oxide_hstore WHERE data -> 'name' = 'test1'",
        &[],
        None,
    )?;
    assert_eq!(result.rows[0]["data"], json!({"name": "test1"}));

    let installed = db.query(
        "SELECT count(*)::int AS count, max(n.nspname) AS schema_name \
         FROM pg_extension e \
         JOIN pg_namespace n ON n.oid = e.extnamespace \
         WHERE e.extname = 'hstore'",
        &[],
        None,
    )?;
    assert_eq!(installed.rows[0]["count"], json!(1));
    assert_eq!(installed.rows[0]["schema_name"], json!("pg_catalog"));

    db.close()?;
    Ok(())
}

#[test]
fn hstore_extension_reopens_cleanly() -> Result<()> {
    let _trace = TestTrace::new("hstore_extension_reopens_cleanly");
    let root = tempfile::TempDir::new()?;
    {
        let mut db = Oliphaunt::builder()
            .path(root.path())
            .extension(extensions::HSTORE)
            .open()?;
        db.exec("CREATE TABLE oxide_hstore_restart (data hstore)", None)?;
        db.exec(
            "INSERT INTO oxide_hstore_restart VALUES ('\"name\"=>\"persisted\"')",
            None,
        )?;
        db.close()?;
    }

    {
        let mut reopened = Oliphaunt::builder().path(root.path()).open()?;
        let result = reopened.query(
            "SELECT data -> 'name' AS name FROM oxide_hstore_restart",
            &[],
            None,
        )?;
        assert_eq!(result.rows[0]["name"], json!("persisted"));
        reopened.close()?;
    }

    Ok(())
}

#[test]
fn pgcrypto_extension_direct_smoke() -> Result<()> {
    let _trace = TestTrace::new("pgcrypto_extension_direct_smoke");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extension(extensions::PGCRYPTO)
        .open()?;

    db.exec(PGCRYPTO_FUNCTIONAL_SMOKE_SQL, None)?;

    let installed = db.query(
        "SELECT count(*)::int AS count, max(n.nspname) AS schema_name \
         FROM pg_extension e \
         JOIN pg_namespace n ON n.oid = e.extnamespace \
         WHERE e.extname = 'pgcrypto'",
        &[],
        None,
    )?;
    assert_eq!(installed.rows[0]["count"], json!(1));
    assert_eq!(installed.rows[0]["schema_name"], json!("public"));

    trace_expected("pgcrypto_direct invalid digest algorithm");
    let invalid_algorithm = db
        .query("SELECT digest('abc', 'not-a-digest')", &[], None)
        .expect_err("invalid pgcrypto digest algorithm should fail");
    assert_oliphaunt_code(
        &invalid_algorithm,
        "22023",
        "Cannot use \"not-a-digest\": No such hash algorithm",
    );
    let recovered = db.query(
        "SELECT 23::int AS recovered_after_pgcrypto_error",
        &[],
        None,
    )?;
    assert_eq!(
        recovered.rows[0]["recovered_after_pgcrypto_error"],
        json!(23)
    );

    db.close()?;
    Ok(())
}

#[test]
fn pgcrypto_extension_reopens_cleanly() -> Result<()> {
    let _trace = TestTrace::new("pgcrypto_extension_reopens_cleanly");
    let root = tempfile::TempDir::new()?;
    {
        let mut db = Oliphaunt::builder()
            .path(root.path())
            .extension(extensions::PGCRYPTO)
            .open()?;
        db.exec(PGCRYPTO_FUNCTIONAL_SMOKE_SQL, None)?;
        db.exec(
            "CREATE TABLE oxide_pgcrypto_restart AS \
             SELECT encode(digest('persisted', 'sha256'), 'hex') AS digest_hex",
            None,
        )?;
        db.close()?;
    }

    {
        let mut reopened = Oliphaunt::builder().path(root.path()).open()?;
        let result = reopened.query(
            "SELECT digest_hex = encode(digest('persisted', 'sha256'), 'hex') AS ok \
             FROM oxide_pgcrypto_restart",
            &[],
            None,
        )?;
        assert_eq!(result.rows[0]["ok"], json!(true));
        reopened.close()?;
    }

    Ok(())
}

#[test]
fn pgcrypto_extension_materializes_only_requested_assets() -> Result<()> {
    let _trace = TestTrace::new("pgcrypto_extension_materializes_only_requested_assets");
    let root = tempfile::TempDir::new()?;
    {
        let mut db = Oliphaunt::builder()
            .path(root.path())
            .extension(extensions::PGCRYPTO)
            .open()?;
        db.exec(PGCRYPTO_FUNCTIONAL_SMOKE_SQL, None)?;
        db.close()?;
    }

    assert_only_requested_extension_assets_are_materialized(root.path(), "pgcrypto", "vector");
    Ok(())
}

#[test]
fn multiple_extension_set_direct_smoke() -> Result<()> {
    let _trace = TestTrace::new("multiple_extension_set_direct_smoke");
    let mut db = Oliphaunt::builder()
        .temporary()
        .extensions([extensions::VECTOR, extensions::PG_TRGM])
        .open()?;

    let result = db.query(
        "SELECT \
            '[1,2,3]'::vector <-> '[1,2,4]'::vector AS distance, \
            similarity('postgres', 'postgrex') AS score",
        &[],
        None,
    )?;
    assert_eq!(first_f64(&result, "distance"), 1.0);
    assert!(first_f64(&result, "score") > 0.5);

    let installed = db.query(
        "SELECT count(*)::int AS count \
         FROM pg_extension \
         WHERE extname IN ('vector', 'pg_trgm')",
        &[],
        None,
    )?;
    assert_eq!(installed.rows[0]["count"], json!(2));

    db.close()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn pg_trgm_extension_server_sqlx_smoke() -> Result<()> {
    let _trace = TestTrace::new("pg_trgm_extension_server_sqlx_smoke");
    let server = OliphauntServer::builder()
        .temporary()
        .extension(extensions::PG_TRGM)
        .start()?;
    let mut conn = sqlx::PgConnection::connect(&server.connection_uri()).await?;

    let row = sqlx::query("SELECT similarity('postgres', 'postgrex')::float8 AS score")
        .fetch_one(&mut conn)
        .await?;
    assert!(row.try_get::<f64, _>("score")? > 0.5);

    let row =
        sqlx::query("SELECT count(*)::int4 AS count FROM pg_extension WHERE extname = 'pg_trgm'")
            .fetch_one(&mut conn)
            .await?;
    assert_eq!(row.try_get::<i32, _>("count")?, 1);

    conn.close().await?;
    server.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn hstore_extension_server_sqlx_smoke() -> Result<()> {
    let _trace = TestTrace::new("hstore_extension_server_sqlx_smoke");
    let server = OliphauntServer::builder()
        .temporary()
        .extension(extensions::HSTORE)
        .start()?;
    let mut conn = sqlx::PgConnection::connect(&server.connection_uri()).await?;

    sqlx::query("CREATE TEMP TABLE oxide_hstore_sqlx (data hstore)")
        .execute(&mut conn)
        .await?;
    sqlx::query("INSERT INTO oxide_hstore_sqlx VALUES ('\"name\"=>\"test1\"')")
        .execute(&mut conn)
        .await?;
    let row = sqlx::query(
        "SELECT data -> 'name' AS name, \
         (SELECT count(*)::int4 FROM pg_extension WHERE extname = 'hstore') AS count \
         FROM oxide_hstore_sqlx",
    )
    .fetch_one(&mut conn)
    .await?;
    assert_eq!(row.try_get::<String, _>("name")?, "test1");
    assert_eq!(row.try_get::<i32, _>("count")?, 1);

    conn.close().await?;
    server.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn pgcrypto_extension_server_sqlx_smoke() -> Result<()> {
    let _trace = TestTrace::new("pgcrypto_extension_server_sqlx_smoke");
    let server = OliphauntServer::builder()
        .temporary()
        .extension(extensions::PGCRYPTO)
        .start()?;
    let mut conn = sqlx::PgConnection::connect(&server.connection_uri()).await?;

    sqlx::query(PGCRYPTO_FUNCTIONAL_SMOKE_SQL)
        .execute(&mut conn)
        .await?;
    let row =
        sqlx::query("SELECT count(*)::int4 AS count FROM pg_extension WHERE extname = 'pgcrypto'")
            .fetch_one(&mut conn)
            .await?;
    assert_eq!(row.try_get::<i32, _>("count")?, 1);

    conn.close().await?;
    server.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn vector_extension_server_sqlx_smoke() -> Result<()> {
    let _trace = TestTrace::new("vector_extension_server_sqlx_smoke");
    let server = OliphauntServer::builder()
        .temporary()
        .extension(extensions::VECTOR)
        .start()?;
    let mut conn = sqlx::PgConnection::connect(&server.connection_uri()).await?;

    sqlx::query("CREATE TABLE oxide_vec_server (embedding vector(3))")
        .execute(&mut conn)
        .await?;
    sqlx::query("INSERT INTO oxide_vec_server VALUES ('[1,2,3]')")
        .execute(&mut conn)
        .await?;
    let row =
        sqlx::query("SELECT embedding <-> '[1,2,4]'::vector AS distance FROM oxide_vec_server")
            .fetch_one(&mut conn)
            .await?;

    assert_eq!(row.try_get::<f64, _>("distance")?, 1.0);

    trace_expected("vector_server_sqlx division-by-zero");
    let err = sqlx::query("SELECT 10 / $1::int4 AS impossible_after_vector")
        .bind(0_i32)
        .fetch_one(&mut conn)
        .await
        .expect_err("division by zero after vector load should fail");
    assert_sqlx_code(&err, "22012");
    let row = sqlx::query("SELECT 14::int4 AS recovered_after_vector_error")
        .fetch_one(&mut conn)
        .await?;
    assert_eq!(row.try_get::<i32, _>("recovered_after_vector_error")?, 14);

    trace_expected("vector_server_sqlx invalid-vector-literal");
    let err = sqlx::query("SELECT $1::text::vector AS embedding")
        .bind("[hello,1]")
        .fetch_one(&mut conn)
        .await
        .expect_err("invalid vector input through SQLx should fail in the vector extension");
    assert_sqlx_code(&err, "22P02");
    let row = sqlx::query("SELECT 18::int4 AS recovered_after_invalid_vector")
        .fetch_one(&mut conn)
        .await?;
    assert_eq!(row.try_get::<i32, _>("recovered_after_invalid_vector")?, 18);

    trace_expected("vector_server_sqlx dimension-mismatch");
    let err = sqlx::query("SELECT $1::text::vector <-> $2::text::vector AS distance")
        .bind("[1,2]")
        .bind("[3]")
        .fetch_one(&mut conn)
        .await
        .expect_err("vector distance should reject mismatched dimensions through SQLx");
    assert_sqlx_code(&err, "22000");
    let row = sqlx::query("SELECT 19::int4 AS recovered_after_dimension_mismatch")
        .fetch_one(&mut conn)
        .await?;
    assert_eq!(
        row.try_get::<i32, _>("recovered_after_dimension_mismatch")?,
        19
    );

    conn.close().await?;
    server.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn vector_extension_server_sqlx_transaction_commit_rollback_and_error_recovery() -> Result<()>
{
    let _trace = TestTrace::new(
        "vector_extension_server_sqlx_transaction_commit_rollback_and_error_recovery",
    );
    let server = OliphauntServer::builder()
        .temporary()
        .extension(extensions::VECTOR)
        .start()?;
    let mut conn = sqlx::PgConnection::connect(&server.connection_uri()).await?;

    sqlx::query("CREATE TABLE vector_server_tx_items(id int PRIMARY KEY, embedding vector(3))")
        .execute(&mut conn)
        .await?;

    {
        let mut tx = conn.begin().await?;
        sqlx::query(
            "INSERT INTO vector_server_tx_items(id, embedding) VALUES ($1, $2::text::vector)",
        )
        .bind(1_i32)
        .bind("[1,2,3]")
        .execute(&mut *tx)
        .await?;
        let row = sqlx::query(
            "SELECT embedding <-> '[1,2,4]'::vector AS distance \
             FROM vector_server_tx_items WHERE id = 1",
        )
        .fetch_one(&mut *tx)
        .await?;
        assert_eq!(row.try_get::<f64, _>("distance")?, 1.0);
        tx.commit().await?;
    }

    {
        let mut tx = conn.begin().await?;
        sqlx::query(
            "INSERT INTO vector_server_tx_items(id, embedding) VALUES ($1, $2::text::vector)",
        )
        .bind(2_i32)
        .bind("[9,9,9]")
        .execute(&mut *tx)
        .await?;
        tx.rollback().await?;
    }

    {
        let mut tx = conn.begin().await?;
        sqlx::query(
            "INSERT INTO vector_server_tx_items(id, embedding) VALUES ($1, $2::text::vector)",
        )
        .bind(3_i32)
        .bind("[3,3,3]")
        .execute(&mut *tx)
        .await?;
        trace_expected("vector_server_sqlx_transaction invalid-vector-literal");
        let err = sqlx::query("SELECT $1::text::vector AS embedding")
            .bind("[hello,1]")
            .fetch_one(&mut *tx)
            .await
            .expect_err("invalid vector should fail inside SQLx transaction");
        assert_sqlx_code(&err, "22P02");
        trace_expected("vector_server_sqlx_transaction still-aborted");
        let aborted = sqlx::query("SELECT 1::int4 AS still_aborted")
            .fetch_one(&mut *tx)
            .await
            .expect_err("transaction should stay aborted after vector failure");
        assert_sqlx_code(&aborted, "25P02");
        tx.rollback().await?;
    }

    let row = sqlx::query(
        "SELECT count(*)::int4 AS count, \
                min(embedding <-> '[1,2,4]'::vector)::float8 AS distance \
         FROM vector_server_tx_items",
    )
    .fetch_one(&mut conn)
    .await?;
    assert_eq!(row.try_get::<i32, _>("count")?, 1);
    assert_eq!(row.try_get::<f64, _>("distance")?, 1.0);

    let row = sqlx::query("SELECT 45::int4 AS recovered_after_vector_tx")
        .fetch_one(&mut conn)
        .await?;
    assert_eq!(row.try_get::<i32, _>("recovered_after_vector_tx")?, 45);

    conn.close().await?;
    server.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn multiple_extension_set_server_sqlx_smoke() -> Result<()> {
    let _trace = TestTrace::new("multiple_extension_set_server_sqlx_smoke");
    let server = OliphauntServer::builder()
        .temporary()
        .extensions([extensions::VECTOR, extensions::PG_TRGM])
        .start()?;
    let mut conn = sqlx::PgConnection::connect(&server.connection_uri()).await?;

    let row = sqlx::query(
        "SELECT \
            '[1,2,3]'::vector <-> '[1,2,4]'::vector AS distance, \
            similarity('postgres', 'postgrex')::float8 AS score",
    )
    .fetch_one(&mut conn)
    .await?;
    assert_eq!(row.try_get::<f64, _>("distance")?, 1.0);
    assert!(row.try_get::<f64, _>("score")? > 0.5);

    let row = sqlx::query(
        "SELECT count(*)::int4 AS count \
         FROM pg_extension \
         WHERE extname IN ('vector', 'pg_trgm')",
    )
    .fetch_one(&mut conn)
    .await?;
    assert_eq!(row.try_get::<i32, _>("count")?, 2);

    conn.close().await?;
    server.shutdown()?;
    Ok(())
}