feat: remove sqlx

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()
+}