pipedash_core/infrastructure/database/
schema.rs

1use std::path::PathBuf;
2use std::time::Duration;
3
4use sqlx::postgres::{
5    PgPool,
6    PgPoolOptions,
7};
8use sqlx::sqlite::{
9    SqliteConnectOptions,
10    SqlitePoolOptions,
11};
12use sqlx::{
13    Executor,
14    SqlitePool,
15};
16
17pub async fn init_database(path: PathBuf) -> anyhow::Result<SqlitePool> {
18    let options = SqliteConnectOptions::new()
19        .filename(&path)
20        .create_if_missing(true)
21        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
22        .synchronous(sqlx::sqlite::SqliteSynchronous::Normal)
23        .busy_timeout(Duration::from_secs(5)); // Reduced from 30s to 5s to
24                                               // prevent compounding with
25                                               // retries
26
27    let migrating = SqlitePoolOptions::new()
28        .max_connections(1)
29        .connect_with(options.clone())
30        .await?;
31    sqlx::migrate!("./migrations/sqlite")
32        .set_ignore_missing(true)
33        .run(&migrating)
34        .await?;
35    migrating.close().await;
36
37    let pool = SqlitePoolOptions::new()
38        .max_connections(50)
39        // In-process connections can't drop; pinging only adds a worker round trip.
40        .test_before_acquire(false)
41        .min_connections(10)
42        .acquire_timeout(Duration::from_secs(30))
43        .idle_timeout(Duration::from_secs(300))
44        .max_lifetime(Duration::from_secs(1800))
45        .connect_with(options)
46        .await?;
47
48    pool.execute("PRAGMA cache_size = -64000").await?; // 64MB cache
49    pool.execute("PRAGMA temp_store = MEMORY").await?;
50    pool.execute("PRAGMA mmap_size = 268435456").await?; // 256MB mmap
51    pool.execute("PRAGMA auto_vacuum = INCREMENTAL").await?; // Enables incremental auto-vacuum
52    pool.execute("PRAGMA wal_autocheckpoint = 1000").await?; // Checkpoint every
53                                                             // 1000 pages
54
55    Ok(pool)
56}
57
58pub async fn init_postgres_database(connection_string: &str) -> anyhow::Result<PgPool> {
59    let pool = PgPoolOptions::new()
60        .max_connections(50)
61        .min_connections(10)
62        .acquire_timeout(Duration::from_secs(30))
63        .idle_timeout(Duration::from_secs(300))
64        .max_lifetime(Duration::from_secs(1800))
65        .after_connect(|conn, _meta| {
66            Box::pin(async move {
67                sqlx::query("SET search_path TO public")
68                    .execute(conn)
69                    .await?;
70                Ok(())
71            })
72        })
73        .connect(connection_string)
74        .await?;
75
76    sqlx::migrate!("./migrations/postgres")
77        .set_ignore_missing(true)
78        .run(&pool)
79        .await?;
80
81    Ok(pool)
82}
pipedash_core/infrastructure/database/schema.rs

pipedash_core/infrastructure/database/
schema.rs
