db.rs (11978B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2024-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::{str::FromStr, time::Duration}; 18 19 use jiff::{ 20 Timestamp, 21 civil::{Date, Time}, 22 tz::TimeZone, 23 }; 24 use sqlx::{ 25 Decode, Error, PgPool, Postgres, QueryBuilder, Row, Type, 26 error::BoxDynError, 27 postgres::PgRow, 28 query::{Query, QueryScalar}, 29 }; 30 use taler_common::{ 31 api::params::{History, Page, Pooling}, 32 types::{ 33 amount::{Amount, Currency, Decimal}, 34 iban::IBAN, 35 payto::PaytoURI, 36 utils::date_to_utc_ts, 37 }, 38 }; 39 use tokio::sync::watch::{self}; 40 use url::Url; 41 42 pub type PgQueryBuilder<'b> = QueryBuilder<'b, Postgres>; 43 44 /* ------ Serialization ----- */ 45 46 pub trait PgError { 47 const PG_SERIALIZATION_FAILURE: &str = "40001"; 48 const PG_DEADLOCK_DETECTED: &str = "40P01"; 49 const PG_UNIQUE_VIOLATION: &str = "23505"; 50 const PG_FOREIGN_KEY_VIOLATION: &str = "23503"; 51 52 fn is_retryable_err(&self) -> bool; 53 fn is_unique_err(&self) -> bool; 54 fn is_fk_err(&self) -> bool; 55 } 56 57 impl PgError for sqlx::error::Error { 58 fn is_retryable_err(&self) -> bool { 59 if let sqlx::Error::Database(e) = self { 60 return matches!( 61 e.downcast_ref::<sqlx::postgres::PgDatabaseError>().code(), 62 Self::PG_SERIALIZATION_FAILURE | Self::PG_DEADLOCK_DETECTED 63 ); 64 } 65 false 66 } 67 68 fn is_unique_err(&self) -> bool { 69 if let sqlx::Error::Database(e) = self { 70 return e.downcast_ref::<sqlx::postgres::PgDatabaseError>().code() 71 == Self::PG_UNIQUE_VIOLATION; 72 } 73 false 74 } 75 76 fn is_fk_err(&self) -> bool { 77 if let sqlx::Error::Database(e) = self { 78 return e.downcast_ref::<sqlx::postgres::PgDatabaseError>().code() 79 == Self::PG_FOREIGN_KEY_VIOLATION; 80 } 81 false 82 } 83 } 84 85 #[macro_export] 86 macro_rules! serialized { 87 ($logic:expr) => {{ 88 use $crate::db::PgError; 89 let mut attempts = 0; 90 const MAX_RETRIES: u32 = 5; 91 92 loop { 93 let res: sqlx::Result<_, sqlx::Error> = $logic.await; 94 if let Err(e) = &res 95 && e.is_retryable_err() 96 && attempts < MAX_RETRIES 97 { 98 attempts += 1; 99 tokio::task::yield_now().await; 100 continue; 101 } 102 break res; 103 } 104 }}; 105 } 106 107 /* ----- Routines ------ */ 108 109 pub async fn page<'a, 'b, R: Send + Unpin>( 110 db: &PgPool, 111 params: &Page, 112 id_col: &str, 113 prepare: impl Fn() -> QueryBuilder<'a, Postgres> + Copy, 114 map: impl Fn(PgRow) -> Result<R, Error> + Send + Copy, 115 ) -> Result<Vec<R>, Error> { 116 serialized!(async { 117 let mut builder = prepare(); 118 if let Some(offset) = params.offset { 119 builder 120 .push(format_args!( 121 " {id_col} {}", 122 if params.backward() { '<' } else { '>' } 123 )) 124 .push_bind(offset); 125 } else { 126 builder.push("TRUE"); 127 } 128 builder.push(format_args!( 129 " ORDER BY {id_col} {} LIMIT ", 130 if params.backward() { "DESC" } else { "ASC" } 131 )); 132 builder 133 .push_bind(params.limit()) 134 .build() 135 .try_map(map) 136 .fetch_all(db) 137 .await 138 }) 139 } 140 141 pub async fn pooling<R, N, F: Future<Output = sqlx::Result<R>>>( 142 params: &Pooling, 143 listen: impl FnOnce() -> watch::Receiver<N>, 144 filter: impl FnMut(&N) -> bool, 145 mut load: impl FnMut() -> F, 146 mut check: impl FnMut(&R) -> bool, 147 ) -> Result<R, Error> { 148 let timeout = params.timeout_ms.unwrap_or_default(); 149 if timeout > 0 { 150 let mut listener = listen(); 151 let init = load().await?; 152 // Long polling if we found no transactions 153 if !check(&init) { 154 let pooling = tokio::time::timeout(Duration::from_millis(timeout), async { 155 listener.wait_for(filter).await.ok(); 156 }) 157 .await; 158 match pooling { 159 Ok(_) => load().await, 160 Err(_) => Ok(init), 161 } 162 } else { 163 Ok(init) 164 } 165 } else { 166 load().await 167 } 168 } 169 170 pub async fn history<T: Send + Unpin>( 171 db: &PgPool, 172 id_col: &str, 173 params: &History, 174 listen: impl FnOnce() -> watch::Receiver<i64>, 175 prepare: impl Fn() -> QueryBuilder<'static, Postgres> + Copy, 176 map: impl Fn(PgRow) -> Result<T, Error> + Send + Copy, 177 ) -> Result<Vec<T>, Error> { 178 let load = async || page(db, ¶ms.page, id_col, prepare, map).await; 179 // When going backward there is always at least one transaction or none 180 let poll = if params.page.limit < 0 { 181 &Pooling::default() 182 } else { 183 ¶ms.pooling 184 }; 185 pooling( 186 poll, 187 listen, 188 |id| *id > params.page.offset.unwrap_or_default(), 189 load, 190 |init| !init.is_empty(), 191 ) 192 .await 193 } 194 195 /* ----- Bind ----- */ 196 197 pub trait BindHelper { 198 fn bind_timestamp(self, timestamp: &Timestamp) -> Self; 199 fn bind_date(self, date: &Date) -> Self; 200 } 201 202 impl<'q> BindHelper for Query<'q, Postgres, <Postgres as sqlx::Database>::Arguments<'q>> { 203 fn bind_timestamp(self, timestamp: &Timestamp) -> Self { 204 self.bind(timestamp.as_microsecond()) 205 } 206 207 fn bind_date(self, date: &Date) -> Self { 208 self.bind_timestamp(&date_to_utc_ts(date)) 209 } 210 } 211 212 impl<'q, T> BindHelper 213 for QueryScalar<'q, Postgres, T, <Postgres as sqlx::Database>::Arguments<'q>> 214 { 215 fn bind_timestamp(self, timestamp: &Timestamp) -> Self { 216 self.bind(timestamp.as_microsecond()) 217 } 218 219 fn bind_date(self, date: &Date) -> Self { 220 self.bind_timestamp(&date_to_utc_ts(date)) 221 } 222 } 223 224 /* ----- Get ----- */ 225 226 pub trait TypeHelper { 227 fn try_get_map< 228 'r, 229 I: sqlx::ColumnIndex<Self>, 230 T: Decode<'r, Postgres> + Type<Postgres>, 231 E: Into<BoxDynError>, 232 R, 233 M: FnOnce(T) -> Result<R, E>, 234 >( 235 &'r self, 236 index: I, 237 map: M, 238 ) -> sqlx::Result<R>; 239 fn try_get_opt_map< 240 'r, 241 I: sqlx::ColumnIndex<Self>, 242 T: Decode<'r, Postgres> + Type<Postgres>, 243 E: Into<BoxDynError>, 244 R, 245 M: FnOnce(T) -> Result<R, E>, 246 >( 247 &'r self, 248 index: I, 249 map: M, 250 ) -> sqlx::Result<Option<R>> { 251 self.try_get_map(index, |it: Option<T>| it.map(map).transpose()) 252 } 253 fn try_get_parse<I: sqlx::ColumnIndex<Self>, E: Into<BoxDynError>, T: FromStr<Err = E>>( 254 &self, 255 index: I, 256 ) -> sqlx::Result<T> { 257 self.try_get_map(index, |s: &str| s.parse()) 258 } 259 fn try_get_opt_parse<I: sqlx::ColumnIndex<Self>, E: Into<BoxDynError>, T: FromStr<Err = E>>( 260 &self, 261 index: I, 262 ) -> sqlx::Result<Option<T>> { 263 self.try_get_map(index, |s: Option<&str>| s.map(|s| s.parse()).transpose()) 264 } 265 fn try_get_timestamp<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<Timestamp> { 266 self.try_get_map(index, |micros| { 267 jiff::Timestamp::from_microsecond(micros) 268 .map_err(|e| format!("expected timestamp micros got overflowing {micros}: {e}")) 269 }) 270 } 271 fn try_get_opt_timestamp<I: sqlx::ColumnIndex<Self>>( 272 &self, 273 index: I, 274 ) -> sqlx::Result<Option<Timestamp>> { 275 self.try_get_map(index, |micros: Option<i64>| { 276 if let Some(micros) = micros { 277 Some(jiff::Timestamp::from_microsecond(micros).map_err(|e| { 278 format!("expected timestamp micros got overflowing {micros}: {e}") 279 })) 280 .transpose() 281 } else { 282 Ok(None) 283 } 284 }) 285 } 286 fn try_get_date<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<Date> { 287 let timestamp = self.try_get_timestamp(index)?; 288 let zoned = timestamp.to_zoned(TimeZone::UTC); 289 assert_eq!(zoned.time(), Time::midnight()); 290 Ok(zoned.date()) 291 } 292 fn try_get_u16<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<u16> { 293 self.try_get_map(index, |signed: i16| signed.try_into()) 294 } 295 fn try_get_opt_u16<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<Option<u16>> { 296 self.try_get_opt_map(index, |signed: i16| signed.try_into()) 297 } 298 fn try_get_u32<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<u32> { 299 self.try_get_map(index, |signed: i32| signed.try_into()) 300 } 301 fn try_get_opt_u32<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<Option<u32>> { 302 self.try_get_opt_map(index, |signed: i32| signed.try_into()) 303 } 304 fn try_get_u64<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<u64> { 305 self.try_get_map(index, |signed: i64| signed.try_into()) 306 } 307 fn try_get_opt_u64<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<Option<u64>> { 308 self.try_get_opt_map(index, |signed: i64| signed.try_into()) 309 } 310 fn try_get_url<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<Url> { 311 self.try_get_parse(index) 312 } 313 fn try_get_payto<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<PaytoURI> { 314 self.try_get_parse(index) 315 } 316 fn try_get_opt_payto<I: sqlx::ColumnIndex<Self>>( 317 &self, 318 index: I, 319 ) -> sqlx::Result<Option<PaytoURI>> { 320 self.try_get_opt_parse(index) 321 } 322 fn try_get_iban<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<IBAN> { 323 self.try_get_parse(index) 324 } 325 fn try_get_amount<I: sqlx::ColumnIndex<Self>>( 326 &self, 327 index: I, 328 currency: &Currency, 329 ) -> sqlx::Result<Amount>; 330 fn try_get_opt_amount<I: sqlx::ColumnIndex<Self>>( 331 &self, 332 index: I, 333 currency: &Currency, 334 ) -> sqlx::Result<Option<Amount>>; 335 336 /** Flag consider NULL and false to be the same */ 337 fn try_get_flag<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<bool>; 338 } 339 340 impl TypeHelper for PgRow { 341 fn try_get_map< 342 'r, 343 I: sqlx::ColumnIndex<Self>, 344 T: Decode<'r, Postgres> + Type<Postgres>, 345 E: Into<BoxDynError>, 346 R, 347 M: FnOnce(T) -> Result<R, E>, 348 >( 349 &'r self, 350 index: I, 351 map: M, 352 ) -> sqlx::Result<R> { 353 let primitive: T = self.try_get(&index)?; 354 map(primitive).map_err(|source| sqlx::Error::ColumnDecode { 355 index: format!("{index:?}"), 356 source: source.into(), 357 }) 358 } 359 360 fn try_get_amount<I: sqlx::ColumnIndex<Self>>( 361 &self, 362 index: I, 363 currency: &Currency, 364 ) -> sqlx::Result<Amount> { 365 let decimal: Decimal = self.try_get(index)?; 366 Ok(Amount::new_decimal(currency, decimal)) 367 } 368 369 fn try_get_opt_amount<I: sqlx::ColumnIndex<Self>>( 370 &self, 371 index: I, 372 currency: &Currency, 373 ) -> sqlx::Result<Option<Amount>> { 374 let decimal: Option<Decimal> = self.try_get(index)?; 375 Ok(decimal.map(|decimal| Amount::new_decimal(currency, decimal))) 376 } 377 378 fn try_get_flag<I: sqlx::ColumnIndex<Self>>(&self, index: I) -> sqlx::Result<bool> { 379 let opt_bool: Option<bool> = self.try_get(index)?; 380 Ok(opt_bool.unwrap_or(false)) 381 } 382 }