db.rs (4986B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2025, 2026 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 17 use std::{ 18 io::ErrorKind, 19 path::{Path, PathBuf}, 20 sync::Arc, 21 }; 22 23 use sqlx::{ 24 AssertSqlSafe, Connection, PgConnection, PgPool, Postgres, Row, Transaction, 25 postgres::{PgConnectOptions, PgPoolOptions, PgRow}, 26 }; 27 use taler_macros::EnumMeta; 28 use tracing::{debug, info}; 29 30 #[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type, EnumMeta)] 31 #[allow(non_camel_case_types)] 32 #[sqlx(type_name = "incoming_type")] 33 #[enum_meta(Str)] 34 pub enum IncomingType { 35 reserve, 36 kyc, 37 map, 38 } 39 40 /* ----- Pool ----- */ 41 42 pub async fn pool(cfg: PgConnectOptions, schema: &str) -> sqlx::Result<PgPool> { 43 pool_with_options(cfg, schema, PgPoolOptions::new()).await 44 } 45 46 pub async fn pool_with_options( 47 cfg: PgConnectOptions, 48 schema: &str, 49 options: PgPoolOptions, 50 ) -> sqlx::Result<PgPool> { 51 let init_sql = Arc::new(format!( 52 "SET search_path TO {schema};SET default_transaction_isolation = 'serializable';" 53 )); 54 let pool = options 55 .after_connect(move |conn, _meta| { 56 let init_sql = init_sql.clone(); 57 Box::pin(async move { 58 sqlx::raw_sql(AssertSqlSafe(init_sql)).execute(conn).await?; 59 60 Ok(()) 61 }) 62 }) 63 .connect_with(cfg) 64 .await?; 65 66 // TODO check postgresql version ? 67 68 Ok(pool) 69 } 70 71 /* ----- Migration----- */ 72 73 #[derive(Debug, thiserror::Error)] 74 pub enum MigrationErr { 75 #[error("could not read patch at '{}': {}", .0.to_string_lossy(), .1.kind())] 76 Io(PathBuf, std::io::Error), 77 #[error(transparent)] 78 Sql(#[from] sqlx::Error), 79 } 80 81 pub async fn dbinit( 82 conn: &mut PgConnection, 83 sql_dir: &Path, 84 prefix: &str, 85 reset: bool, 86 ) -> anyhow::Result<()> { 87 dbinit_setup(conn, sql_dir, prefix, reset, async |_| Ok(())).await 88 } 89 90 pub async fn dbinit_setup( 91 conn: &mut PgConnection, 92 sql_dir: &Path, 93 prefix: &str, 94 reset: bool, 95 setup: impl AsyncFn(&mut Transaction<Postgres>) -> anyhow::Result<()>, 96 ) -> anyhow::Result<()> { 97 let mut tx = conn.begin().await?; 98 99 let exec_sql_file = 100 async |conn: &mut PgConnection, file: &str, action: &str| -> Result<(), MigrationErr> { 101 let path = sql_dir.join(file); 102 match std::fs::read_to_string(&path) { 103 Ok(content) => { 104 info!(target: "dbinit", "applying {action}"); 105 sqlx::raw_sql(AssertSqlSafe(content)).execute(conn).await?; 106 Ok(()) 107 } 108 Err(e) => Err(MigrationErr::Io(path, e)), 109 } 110 }; 111 112 if reset { 113 info!(target: "dbinit", "reset, sqlqir '{}'", sql_dir.to_string_lossy()); 114 exec_sql_file(&mut *tx, &format!("{prefix}-drop.sql"), "drop").await?; 115 } 116 117 info!(target: "dbinit", "initialization, sqlqir '{}'", sql_dir.to_string_lossy()); 118 119 exec_sql_file(&mut *tx, "versioning.sql", "versioning").await?; 120 121 let applied: Vec<String> = sqlx::query("SELECT patch_name FROM _v.patches") 122 .try_map(|r: PgRow| r.try_get(0)) 123 .fetch_all(&mut *tx) 124 .await?; 125 for n in 1..9999 { 126 let patch = format!("{prefix}-{n:0>4}"); 127 if applied.contains(&patch) { 128 debug!(target: "dbinit", "patch {patch} already applied"); 129 continue; 130 } 131 132 if let Err(e) = 133 exec_sql_file(&mut *tx, &format!("{patch}.sql"), &format!("patch {patch}")).await 134 { 135 if let MigrationErr::Io(path, e) = &e 136 && e.kind() == ErrorKind::NotFound 137 { 138 debug!( 139 target: "dbinit", 140 "path '{}' doesn't exist anymore, stopping", 141 path.to_string_lossy() 142 ); 143 break; 144 } 145 return Err(e.into()); 146 } 147 } 148 149 if let Err(e) = exec_sql_file(&mut *tx, &format!("{prefix}-procedures.sql"), "procedures").await 150 { 151 if let MigrationErr::Io(_, e) = &e 152 && e.kind() == ErrorKind::NotFound 153 { 154 debug!( 155 target: "dbinit", 156 "no procedures.sql for the SQL collection: '{prefix}'" 157 ); 158 } else { 159 return Err(e.into()); 160 } 161 } 162 163 setup(&mut tx).await?; 164 165 tx.commit().await?; 166 167 Ok(()) 168 }