Skip to content

Commit

Permalink
Merge pull request #1989 from scpwiki/sqlx
Browse files Browse the repository at this point in the history
[WJ-1248] Add sqlx to DEEPWELL
  • Loading branch information
emmiegit authored Jul 15, 2024
2 parents 7ec1092 + e3c09e9 commit 91bb502
Show file tree
Hide file tree
Showing 4 changed files with 21 additions and 11 deletions.
15 changes: 10 additions & 5 deletions deepwell/src/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ use jsonrpsee::types::error::ErrorObjectOwned;
use rsmq_async::PooledRsmq;
use s3::bucket::Bucket;
use sea_orm::{DatabaseConnection, TransactionTrait};
use sqlx::{Pool, Postgres};
use std::fmt::{self, Debug};
use std::sync::Arc;
use std::time::Duration;
Expand All @@ -51,7 +52,8 @@ pub type ServerState = Arc<ServerStateInner>;

pub struct ServerStateInner {
pub config: Config,
pub database: DatabaseConnection,
pub database_seaorm: DatabaseConnection,
pub database_sqlx: Pool<Postgres>,
pub redis: redis::Client,
pub rsmq: PooledRsmq,
pub localizations: Localizations,
Expand All @@ -63,7 +65,8 @@ impl Debug for ServerStateInner {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("ServerStateInner")
.field("config", &self.config)
.field("database", &self.database)
.field("database_seaorm", &self.database_seaorm)
.field("database_sqlx", &self.database_sqlx)
.field("redis", &self.redis)
.field("rsmq", &debug_pointer(&self.rsmq))
.field("localizations", &self.localizations)
Expand All @@ -79,7 +82,8 @@ pub async fn build_server_state(
) -> anyhow::Result<ServerState> {
// Connect to databases
info!("Connecting to PostgreSQL database");
let database = database::connect(&secrets.database_url).await?;
let (database_sqlx, database_seaorm) =
database::connect(&secrets.database_url).await?;

info!("Connecting to Redis");
let (redis, rsmq) = redis_db::connect(&secrets.redis_url).await?;
Expand Down Expand Up @@ -112,7 +116,8 @@ pub async fn build_server_state(
// Build server state
let state = Arc::new(ServerStateInner {
config,
database,
database_seaorm,
database_sqlx,
redis,
rsmq,
localizations,
Expand Down Expand Up @@ -157,7 +162,7 @@ async fn build_module(app_state: ServerState) -> anyhow::Result<RpcModule<Server
// At this level, we take the database-or-RPC error and make it just an RPC error.
let db_state = Arc::clone(&state);
db_state
.database
.database_seaorm
.transaction(move |txn| {
Box::pin(async move {
// Run the endpoint's implementation, and convert from
Expand Down
13 changes: 9 additions & 4 deletions deepwell/src/database/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,17 +27,22 @@ use sea_orm::{ConnectOptions, Database, DatabaseConnection};
use sqlx::{Pool, Postgres};
use std::time::Duration;

pub async fn connect<S: Into<String>>(database_uri: S) -> Result<DatabaseConnection> {
let mut options = ConnectOptions::new(database_uri.into());
pub async fn connect<S: Into<String>>(
database_uri: S,
) -> Result<(Pool<Postgres>, DatabaseConnection)> {
let database_uri = database_uri.into();
let sqlx_db = Pool::<Postgres>::connect(&database_uri).await?;

let mut options = ConnectOptions::new(database_uri);
options
.min_connections(4)
.max_connections(100)
.connect_timeout(Duration::from_secs(5))
.idle_timeout(Duration::from_secs(10))
.sqlx_logging(true);

let db = Database::connect(options).await?;
Ok(db)
let sea_orm_db = Database::connect(options).await?;
Ok((sqlx_db, sea_orm_db))
}

pub async fn migrate(database_uri: &str) -> Result<()> {
Expand Down
2 changes: 1 addition & 1 deletion deepwell/src/database/seeder/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ pub async fn seed(state: &ServerState) -> Result<()> {
info!("Running seeder...");

// Set up context
let txn = state.database.begin().await?;
let txn = state.database_seaorm.begin().await?;
let ctx = ServiceContext::new(state, &txn);

// Ensure seeding has not already been done
Expand Down
2 changes: 1 addition & 1 deletion deepwell/src/services/job/worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ impl JobWorker {

debug!("Received job from queue: {job:?}");
trace!("Setting up ServiceContext for job processing");
let txn = self.state.database.begin().await?;
let txn = self.state.database_seaorm.begin().await?;
let ctx = &ServiceContext::new(&self.state, &txn);

trace!("Beginning job processing");
Expand Down

0 comments on commit 91bb502

Please sign in to comment.