1use sqlx::PgPool;
2use sqlx::postgres::PgPoolOptions;
3use std::time::Duration;
4use tracing::instrument;
5
6#[instrument]
7pub async fn create_pool(database_url: &str) -> Result<PgPool, sqlx::Error> {
8 let max_retries: u32 = std::env::var("DB_CONNECT_RETRIES")
9 .ok()
10 .and_then(|v| v.parse().ok())
11 .unwrap_or(10);
12 let base_delay_ms: u64 = std::env::var("DB_CONNECT_BASE_DELAY_MS")
13 .ok()
14 .and_then(|v| v.parse().ok())
15 .unwrap_or(1000);
16
17 let mut attempt = 0u32;
18 loop {
19 attempt += 1;
20 tracing::info!(attempt, max_retries, "connecting to database");
21
22 match PgPoolOptions::new()
23 .max_connections(10)
24 .acquire_timeout(Duration::from_secs(5))
25 .connect(database_url)
26 .await
27 {
28 Ok(pool) => {
29 tracing::info!("database pool created");
30 return Ok(pool);
31 }
32 Err(e) if attempt < max_retries => {
33 let delay = Duration::from_millis(base_delay_ms * u64::from(attempt));
34 tracing::warn!(
35 attempt,
36 error = %e,
37 retry_in_ms = delay.as_millis(),
38 "database connection failed, retrying"
39 );
40 tokio::time::sleep(delay).await;
41 }
42 Err(e) => {
43 tracing::error!(attempt, max_retries, error = %e, "all database connection attempts exhausted");
44 return Err(e);
45 }
46 }
47 }
48}
49
50#[instrument(skip(pool))]
51pub async fn run_migrations(pool: &PgPool) -> Result<(), sqlx::migrate::MigrateError> {
52 tracing::info!("running database migrations");
53 sqlx::migrate!("./migrations").run(pool).await
54}