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 prevent compounding with retries
24
25    let pool = SqlitePoolOptions::new()
26        .max_connections(50)
27        .min_connections(10)
28        .acquire_timeout(Duration::from_secs(30))
29        .idle_timeout(Duration::from_secs(300))
30        .max_lifetime(Duration::from_secs(1800))
31        .connect_with(options)
32        .await?;
33
34    sqlx::migrate!("./migrations/sqlite")
35        .set_ignore_missing(true)
36        .run(&pool)
37        .await?;
38
39    pool.execute("PRAGMA cache_size = -64000").await?; // 64MB cache
40    pool.execute("PRAGMA temp_store = MEMORY").await?;
41    pool.execute("PRAGMA mmap_size = 268435456").await?; // 256MB mmap
42    pool.execute("PRAGMA auto_vacuum = INCREMENTAL").await?; // Enables incremental auto-vacuum
43    pool.execute("PRAGMA wal_autocheckpoint = 1000").await?; // Checkpoint every 1000 pages
44
45    Ok(pool)
46}
47
48pub async fn init_postgres_database(connection_string: &str) -> anyhow::Result<PgPool> {
49    let pool = PgPoolOptions::new()
50        .max_connections(50)
51        .min_connections(10)
52        .acquire_timeout(Duration::from_secs(30))
53        .idle_timeout(Duration::from_secs(300))
54        .max_lifetime(Duration::from_secs(1800))
55        .after_connect(|conn, _meta| {
56            Box::pin(async move {
57                sqlx::query("SET search_path TO public")
58                    .execute(conn)
59                    .await?;
60                Ok(())
61            })
62        })
63        .connect(connection_string)
64        .await?;
65
66    sqlx::migrate!("./migrations/postgres")
67        .set_ignore_missing(true)
68        .run(&pool)
69        .await?;
70
71    Ok(pool)
72}
pipedash_core/infrastructure/database/schema.rs

pipedash_core/infrastructure/database/
schema.rs
