feat: remove sqlx (#192)

Remove SQLx in favor of tokio-postgres.
SQLx has had many issues. From the poor design of the query engine to
issues with transcations and deserializing. The idea of compile time
queries is pretty good but the SQLx implementation is just very sub par.
Most features are not supported.

---------

Co-authored-by: Lennart Kloock <lennart.kloock@protonmail.com>

Cargo.toml

@@ -62,6 +62,8 @@ ffmpeg = { path = "ffmpeg" }
 # These patches are pending PRs to the upstream crates
 # TODO: Remove these once the PRs are merged
 [patch.crates-io]
+# https://github.com/remkop22/postgres-from-row/pull/9
+postgres-from-row = { git = "https://github.com/ScuffleTV/postgres-from-row.git", branch = "troy/from_fn" }
 # https://github.com/async-graphql/async-graphql/pull/1424
 async-graphql = { git = "https://github.com/ScuffleTV/async-graphql.git", branch = "troy/union-generics" }
 # https://github.com/aembke/fred.rs/pull/199

binary-helper/Cargo.toml

@@ -10,7 +10,6 @@ tokio = { version = "1.35", features = ["full"] }
 serde = { version = "1.0.1", features = ["derive"] }
 async-nats = "0.33"
 ulid = "1.1"
-sqlx = { version = "0.7", features = ["postgres"] }
 async-trait = "0.1"
 tonic = { version = "0.10", features = ["tls"] }
 anyhow = "1.0"
@@ -20,6 +19,7 @@ futures-util = "0.3"
 rustls = "0.22"
 rustls-pemfile = "2.0"
 fred = { version = "8.0.0", features = ["enable-rustls", "sentinel-client", "dns"] }
+tokio-postgres-rustls = "0.11"
 
 config = { workspace = true }
 common = { workspace = true, features = ["default"] }

binary-helper/src/global.rs

@@ -1,16 +1,16 @@
 use std::io;
-use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context as _;
 use async_nats::ServerAddr;
 use common::config::{DatabaseConfig, NatsConfig, RedisConfig};
+use common::database::deadpool_postgres::{ManagerConfig, PoolConfig, RecyclingMethod, Runtime};
+use common::database::tokio_postgres::NoTls;
+use common::database::Pool;
 use fred::interfaces::ClientLike;
 use fred::types::ServerConfig;
 use rustls::RootCertStore;
-use sqlx::postgres::PgConnectOptions;
-use sqlx::ConnectOptions;
 
 #[macro_export]
 macro_rules! impl_global_traits {
@@ -36,7 +36,7 @@ macro_rules! impl_global_traits {
 
 		impl common::global::GlobalDb for $struct {
 			#[inline(always)]
-			fn db(&self) -> &Arc<sqlx::PgPool> {
+			fn db(&self) -> &Arc<common::database::Pool> {
 				&self.db
 			}
 		}
@@ -89,16 +89,69 @@ pub async fn setup_nats(
 	Ok((nats, jetstream))
 }
 
-pub async fn setup_database(config: &DatabaseConfig) -> anyhow::Result<Arc<sqlx::PgPool>> {
-	Ok(Arc::new(
-		sqlx::PgPool::connect_with(
-			PgConnectOptions::from_str(&config.uri)
-				.context("failed to parse database uri")?
-				.disable_statement_logging()
-				.to_owned(),
+pub async fn setup_database(config: &DatabaseConfig) -> anyhow::Result<Arc<common::database::Pool>> {
+	let mut pg_config = config
+		.uri
+		.parse::<common::database::tokio_postgres::Config>()
+		.context("invalid database uri")?;
+
+	pg_config.ssl_mode(if config.tls.is_some() {
+		common::database::tokio_postgres::config::SslMode::Require
+	} else {
+		common::database::tokio_postgres::config::SslMode::Disable
+	});
+
+	let manager = if let Some(tls) = &config.tls {
+		let cert = tokio::fs::read(&tls.cert).await.context("failed to read redis client cert")?;
+		let key = tokio::fs::read(&tls.key)
+			.await
+			.context("failed to read redis client private key")?;
+
+		let key = rustls_pemfile::pkcs8_private_keys(&mut io::BufReader::new(io::Cursor::new(key)))
+			.next()
+			.ok_or_else(|| anyhow::anyhow!("failed to find private key in redis client private key file"))??
+			.into();
+
+		let certs = rustls_pemfile::certs(&mut io::BufReader::new(io::Cursor::new(cert))).collect::<Result<Vec<_>, _>>()?;
+
+		let mut cert_store = RootCertStore::empty();
+		if let Some(ca_cert) = &tls.ca_cert {
+			let ca_cert = tokio::fs::read(ca_cert).await.context("failed to read redis ca cert")?;
+			let ca_certs =
+				rustls_pemfile::certs(&mut io::BufReader::new(io::Cursor::new(ca_cert))).collect::<Result<Vec<_>, _>>()?;
+			for cert in ca_certs {
+				cert_store.add(cert).context("failed to add redis ca cert")?;
+			}
+		}
+
+		let tls = rustls::ClientConfig::builder()
+			.with_root_certificates(cert_store)
+			.with_client_auth_cert(certs, key)
+			.context("failed to create redis tls config")?;
+
+		common::database::deadpool_postgres::Manager::from_config(
+			pg_config,
+			tokio_postgres_rustls::MakeRustlsConnect::new(tls),
+			ManagerConfig {
+				recycling_method: RecyclingMethod::Fast,
+			},
+		)
+	} else {
+		common::database::deadpool_postgres::Manager::from_config(
+			pg_config,
+			NoTls,
+			ManagerConfig {
+				recycling_method: RecyclingMethod::Fast,
+			},
 		)
-		.await
-		.context("failed to connect to database")?,
+	};
+
+	Ok(Arc::new(
+		Pool::builder(manager)
+			.config(PoolConfig::default())
+			.runtime(Runtime::Tokio1)
+			.build()
+			.context("failed to create database pool")?,
 	))
 }
 

common/Cargo.toml

@@ -12,11 +12,11 @@ context = ["dep:tokio", "dep:tokio-util"]
 prelude = ["dep:tokio"]
 signal = ["tokio/signal", "tokio/process"]
 macros = []
-database = ["dep:sqlx", "dep:sqlx-postgres", "dep:prost", "dep:uuid", "dep:ulid"]
+database = ["dep:tokio-postgres", "dep:postgres-types", "dep:deadpool-postgres", "dep:postgres-from-row", "dep:prost", "ulid/postgres"]
 dataloader = ["dep:fnv", "dep:futures-util", "dep:futures-channel"]
 config = ["dep:config", "dep:serde", "logging"]
 ratelimiter = ["dep:fred"]
-global = ["context", "dep:fred", "dep:sqlx", "dep:async-nats"]
+global = ["context", "dep:fred", "database", "dep:async-nats"]
 http = ["dep:hyper", "dep:serde_json", "dep:bytes", "dep:http-body-util", "dep:pin-project", "dep:path-tree"]
 task = ["dep:tokio", "dep:thiserror"]
 s3 = ["dep:aws-sdk-s3", "dep:aws-credential-types", "dep:aws-config", "dep:aws-smithy-types", "dep:http-body"]
@@ -56,8 +56,10 @@ futures-channel = { version = "0.3", optional = true }
 
 const_format = { version = "0.2" }
 
-sqlx = { version = "0.7", features = ["postgres", "json", "chrono", "uuid"], optional = true }
-sqlx-postgres = { version = "0.7", optional = true }
+tokio-postgres = { version = "0.7", optional = true }
+postgres-types = { version = "0.2", optional = true, features = ["with-serde_json-1", "with-chrono-0_4", "derive"] }
+deadpool-postgres = { version = "0.12", optional = true }
+postgres-from-row = { version = "0.5", optional = true }
 prost = { version = "0.12", optional = true }
 uuid = { version = "1.6", features = ["v4"], optional = true }
 ulid = { version = "1.1", features = ["uuid"], optional = true}

common/src/config.rs

@@ -134,12 +134,16 @@ impl Default for RedisConfig {
 pub struct DatabaseConfig {
 	/// The database URL to use
 	pub uri: String,
+
+	/// The TLS configuration
+	pub tls: Option<TlsConfig>,
 }
 
 impl Default for DatabaseConfig {
 	fn default() -> Self {
 		Self {
 			uri: "postgres://localhost:5432".to_string(),
+			tls: None,
 		}
 	}
 }

common/src/database/mod.rs

@@ -1,7 +1,19 @@
-mod non_null;
 mod protobuf;
-mod ulid;
+mod query_builder;
 
-pub use non_null::*;
+pub use deadpool_postgres::Pool;
+pub use postgres_from_row::FromRow;
+pub use postgres_types::Json;
 pub use protobuf::*;
-pub use ulid::*;
+pub use query_builder::*;
+pub use {deadpool_postgres, postgres_from_row, postgres_types, tokio_postgres};
+
+#[inline]
+pub fn json<T>(row: Json<T>) -> T {
+	row.0
+}
+
+#[inline]
+pub fn non_null_vec<T>(vec: Vec<Option<T>>) -> Vec<T> {
+	vec.into_iter().flatten().collect()
+}