Cargo.toml

[package]
name = "pgrepro"
version = "0.1.0"
authors = ["Charles Samborski <demurgos@demurgos.net>"]
edition = "2018"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
futures-util = "0.3.13"
sqlx = { version = "0.5.1", default-features = false, features = ["offline", "postgres", "runtime-tokio-rustls"] }
tokio = { version = "1.4.0", features = ["full"]}

main.rs

use sqlx::{PgPool, Transaction, Postgres, Executor};
use sqlx::postgres::{PgPoolOptions, PgConnectOptions};
use std::error::Error;
use futures_util::stream::StreamExt;

const RESET: &str = r"DROP SCHEMA public CASCADE;
CREATE SCHEMA IF NOT EXISTS public;
CREATE EXTENSION IF NOT EXISTS pgcrypto;
CREATE EXTENSION IF NOT EXISTS btree_gist;
";

#[tokio::main]
async fn main() {
  let pool: PgPool = PgPoolOptions::new()
    .max_connections(5)
    .connect_with(
      PgConnectOptions::new()
        .host("localhost")
        .port(5432)
        .database("etwin.dev")
        .username("etwin.dev.admin")
        .password("dev"),
    )
    .await
    .unwrap();
  let mut tx = pool.begin().await.unwrap();
  migrate(&mut tx).await.unwrap();
  tx.commit().await.unwrap();
  pool.close().await;
}

async fn migrate(tx: &mut Transaction<'_, Postgres>) -> Result<(), Box<dyn Error>> {
  {
    let mut stream = tx.execute_many(RESET);
    while let Some(r) = stream.next().await {
      r.unwrap();
    }
    drop(stream);
  }
  tx_set_schema_meta(&mut *tx, 1).await?;
  tx_set_schema_meta(&mut *tx, 2).await?;
  Ok(())
}

async fn tx_set_schema_meta(
  tx: &mut Transaction<'_, Postgres>,
  version: u32,
) -> Result<(), Box<dyn Error>> {
  let create_meta_fn = format!("CREATE FUNCTION get_schema_meta() RETURNS schema_meta LANGUAGE sql IMMUTABLE STRICT PARALLEL SAFE AS $$ SELECT ROW({version}); $$;", version = version);

  let queries = [
    "DROP FUNCTION IF EXISTS get_schema_meta;",
    "DROP TYPE IF EXISTS schema_meta;",
    "DROP TYPE IF EXISTS raw_schema_meta;",
    "CREATE TYPE raw_schema_meta AS (version int4);",
    // "CREATE DOMAIN schema_meta AS raw_schema_meta;",
    // Uncommenting the line below causes an error looking like a memory corruption in Postgres
    "CREATE DOMAIN schema_meta AS raw_schema_meta CHECK ((value).version IS NOT NULL AND (value).version >= 1);",
    create_meta_fn.as_str(),
  ];

  for query in std::array::IntoIter::new(queries) {
    sqlx::query(&query).execute(&mut *tx).await?;
  }

  Ok(())
}