worker.rs (12615B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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 collections::BTreeSet, 19 sync::{LazyLock, Mutex}, 20 time::Duration, 21 }; 22 23 use http_client::ApiErr; 24 use jiff::{SignedDuration, Span, Timestamp, tz::TimeZone}; 25 use regex::Regex; 26 use sqlx::PgPool; 27 use taler_api::subject::{IncomingSubject, parse_incoming_unstructured}; 28 use taler_common::{ExpoBackoffDecorr, config::Config, types::payto::BankID}; 29 use tokio::sync::Notify; 30 use tracing::{debug, error, info, trace, warn}; 31 32 use crate::{ 33 config::{WiseBalance, WorkerCfg}, 34 constants::{DAY, HOUR}, 35 db::{ 36 AddIncomingResult, SyncState, TxIn, load_sync_state, record_sync_execution, register_tx_in, 37 save_sync_state, 38 }, 39 payto::WiseAccount, 40 wise_api::{ 41 client::{Client, WiseErr}, 42 types::{Direction, Tx}, 43 }, 44 }; 45 46 #[derive(Debug, thiserror::Error)] 47 pub enum WorkerError { 48 #[error(transparent)] 49 Db(#[from] sqlx::Error), 50 #[error(transparent)] 51 Api(#[from] ApiErr<WiseErr>), 52 #[error("invalid Wise statement: {0}")] 53 InvalidStatement(String), 54 #[error("balance {0} has no synchronization state; run taler-wise setup")] 55 UninitializedBalance(u32), 56 } 57 58 pub type WorkerResult<T = ()> = Result<T, WorkerError>; 59 60 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] 61 enum Tier { 62 Early, 63 Sync, 64 Checkpoint, 65 } 66 67 fn remaining(last: Option<Timestamp>, period: Duration, now: Timestamp) -> Duration { 68 last.map_or(Duration::ZERO, |last| { 69 let deadline = last + period; 70 now.duration_until(deadline) 71 .max(SignedDuration::ZERO) 72 .unsigned_abs() 73 }) 74 } 75 76 impl SyncState { 77 fn advance(&mut self, tier: Tier, latest: Timestamp) { 78 self.early = self.early.max(latest); 79 if tier >= Tier::Sync { 80 self.sync = self.sync.max(latest - 2 * HOUR); 81 } 82 if tier >= Tier::Checkpoint { 83 self.checkpoint = self.checkpoint.max(latest - 2 * DAY); 84 } 85 } 86 87 fn complete(&mut self, tier: Tier, now: Timestamp) { 88 self.early_success = Some(now); 89 if tier >= Tier::Sync { 90 self.sync_success = Some(now); 91 } 92 if tier >= Tier::Checkpoint { 93 self.checkpoint_success = Some(now); 94 } 95 } 96 97 fn schedule( 98 &self, 99 now: Timestamp, 100 frequency: Duration, 101 ) -> (Timestamp, Tier, Timestamp, Option<Timestamp>, Duration) { 102 let early = remaining(self.early_success, frequency, now); 103 let sync = remaining(self.sync_success, HOUR, now); 104 let checkpoint = remaining(self.checkpoint_success, DAY, now); 105 let next = now + early.min(sync).min(checkpoint); 106 let (tier, cursor, last, overlap) = if checkpoint.is_zero() { 107 ( 108 Tier::Checkpoint, 109 self.checkpoint, 110 self.checkpoint_success, 111 2 * DAY, 112 ) 113 } else if sync.is_zero() { 114 (Tier::Sync, self.sync, self.sync_success, 2 * HOUR) 115 } else { 116 (Tier::Early, self.early, self.early_success, Duration::ZERO) 117 }; 118 (next, tier, cursor, last, overlap) 119 } 120 } 121 122 fn parse_account(account_str: &str) -> Option<WiseAccount> { 123 static IBAN_BIC_PATTERN: LazyLock<Regex> = 124 LazyLock::new(|| Regex::new(r"^\(([A-Z0-9]{8,11})\) ([A-Z0-9]+)$").unwrap()); 125 126 let caps = IBAN_BIC_PATTERN.captures(account_str)?; 127 Some(WiseAccount::IBAN(BankID { 128 iban: caps[2].parse().ok()?, 129 bic: Some(caps[1].parse().ok()?), 130 })) 131 } 132 133 pub async fn run_worker( 134 cfg: &Config, 135 pool: &PgPool, 136 client: &http_client::Client, 137 transient: bool, 138 ) -> anyhow::Result<()> { 139 let cfg = WorkerCfg::parse(cfg)?; 140 let client = Client::new(client, cfg.api, &cfg.token); 141 work_balances(&cfg, pool, &client, transient).await?; 142 Ok(()) 143 } 144 145 /// Coalesced balance wakeups with a single consumer. 146 #[derive(Default)] 147 struct BalanceWakeups { 148 pending: Mutex<BTreeSet<u32>>, 149 ready: Notify, 150 } 151 152 impl BalanceWakeups { 153 fn insert(&self, id: u32) { 154 if self.pending.lock().unwrap().insert(id) { 155 self.ready.notify_one(); 156 } 157 } 158 159 fn remove(&self, id: u32) -> bool { 160 self.pending.lock().unwrap().remove(&id) 161 } 162 163 async fn recv(&self) { 164 while self.pending.lock().unwrap().is_empty() { 165 // notify_one retains a permit if an insert races with this wait. 166 self.ready.notified().await; 167 } 168 } 169 } 170 171 async fn work_balances( 172 cfg: &WorkerCfg, 173 pool: &PgPool, 174 client: &Client<'_>, 175 transient: bool, 176 ) -> WorkerResult { 177 if transient { 178 for balance in &cfg.balances { 179 sync_balance(cfg, balance, pool, client, true) 180 .await 181 .map_err(|err| { 182 error!(target: "worker", "synchronization for {} failed: {err}", balance.id); 183 err 184 })?; 185 } 186 } else { 187 let wake = BalanceWakeups::default(); 188 tokio::try_join!( 189 listen_balances(cfg, pool, &wake), 190 sync_balances(cfg, pool, client, &wake), 191 )?; 192 } 193 Ok(()) 194 } 195 196 async fn listen_balances(cfg: &WorkerCfg, pool: &PgPool, wake: &BalanceWakeups) -> WorkerResult { 197 let configured: BTreeSet<_> = cfg.balances.iter().map(|b| b.id).collect(); 198 taler_api::notification::notification_listener!(pool, 199 "wise_balance_credit" => (id: u32) { 200 if configured.contains(&id) { 201 wake.insert(id); 202 } 203 } 204 ) 205 } 206 207 async fn sync_balances( 208 cfg: &WorkerCfg, 209 pool: &PgPool, 210 client: &Client<'_>, 211 wake: &BalanceWakeups, 212 ) -> WorkerResult { 213 let mut next_poll = Timestamp::now(); 214 let mut jitter = ExpoBackoffDecorr::default(); 215 loop { 216 let mut deadline = next_poll; 217 let wait = Timestamp::now() 218 .duration_until(next_poll) 219 .max(SignedDuration::ZERO) 220 .unsigned_abs(); 221 let poll = tokio::select! { 222 biased; 223 _ = tokio::time::sleep(wait) => { 224 deadline = Timestamp::now() + cfg.frequency; 225 true 226 }, 227 _ = wake.recv() => false, 228 }; 229 let mut failed = false; 230 for balance in &cfg.balances { 231 let notified = wake.remove(balance.id); 232 if !poll && !notified { 233 continue; 234 } 235 match sync_balance(cfg, balance, pool, client, notified).await { 236 Ok(next) => deadline = deadline.min(next), 237 Err(err @ WorkerError::Db(sqlx::Error::PoolClosed)) => return Err(err), 238 Err(err) => { 239 failed = true; 240 let mut wait = jitter.backoff(); 241 if let WorkerError::Api(api) = &err 242 && let WiseErr::TooManyRequest(retry_after) = api.err.as_ref() 243 { 244 wait = wait.max(retry_after.unwrap_or_default()); 245 } 246 wake.insert(balance.id); 247 error!(target: "worker", "synchronization for {} failed: {err}", balance.id); 248 tokio::time::sleep(wait).await; 249 } 250 } 251 } 252 if !failed { 253 next_poll = deadline; 254 jitter.reset(); 255 } 256 } 257 } 258 259 async fn sync_balance( 260 cfg: &WorkerCfg, 261 balance: &WiseBalance, 262 pool: &PgPool, 263 client: &Client<'_>, 264 force: bool, 265 ) -> WorkerResult<Timestamp> { 266 let mut state = load_sync_state(pool, balance.id) 267 .await? 268 .ok_or(WorkerError::UninitializedBalance(balance.id))?; 269 let now = Timestamp::now(); 270 let (next, tier, cursor, last, overlap) = state.schedule(now, cfg.frequency); 271 if !force && next > now { 272 return Ok(next); 273 } 274 record_sync_execution(pool, balance.id).await?; 275 let mut start = cursor.max(last.map_or(cursor, |last| last - overlap)); 276 let previous_early = state.early_success.map(|last| state.early.max(last)); 277 while start < now { 278 // Fetch at most one month at a time 279 let end = (start.to_zoned(TimeZone::UTC) + Span::new().months(1)) 280 .timestamp() 281 .min(now); 282 debug!(target: "worker", "fetching statement for {} {tier:?} from {start} to {end}", balance.id); 283 let stmt = client 284 .balance_statement(cfg.profile_id, balance.id, &balance.currency, start, end) 285 .await?; 286 let latest = stmt.transactions.iter().map(|tx| tx.date).max(); 287 for tx in stmt.transactions { 288 if tx.reference_number.trim().is_empty() { 289 return Err(WorkerError::InvalidStatement( 290 "transaction has an empty Wise reference".into(), 291 )); 292 } else if tx.amount.currency != balance.currency { 293 return Err(WorkerError::InvalidStatement(format!( 294 "transaction {} has currency {}, expected {}", 295 tx.reference_number, tx.amount.currency, balance.currency, 296 ))); 297 } 298 let Some(tx) = register_transaction(pool, balance, tx, now).await? else { 299 continue; 300 }; 301 if tier >= Tier::Sync 302 && let Some(early) = previous_early 303 && tx.value_at <= early 304 { 305 warn!(target: "worker", "in (recovered) {tx}"); 306 } 307 } 308 if let Some(latest) = latest { 309 state.advance(tier, latest); 310 save_sync_state(pool, balance.id, &state).await?; 311 trace!(target: "worker", "saved statement cursors for {}", balance.id); 312 } 313 start = end; 314 } 315 state.complete(tier, now); 316 save_sync_state(pool, balance.id, &state).await?; 317 Ok(state.schedule(now, cfg.frequency).0) 318 } 319 320 async fn register_transaction( 321 pool: &PgPool, 322 balance: &WiseBalance, 323 tx: Tx, 324 now: Timestamp, 325 ) -> WorkerResult<Option<TxIn>> { 326 if tx.direction == Direction::Debit { 327 // Its timestamp advances the cursors, but outgoing registration is not supported. 328 return Ok(None); 329 } 330 let subject = match parse_incoming_unstructured(&tx.details.payment_reference) { 331 Ok(IncomingSubject::Key(key)) => Some(key), 332 Ok(IncomingSubject::AdminBalanceAdjust) | Err(_) => None, 333 }; 334 let t = TxIn { 335 balance_id: balance.id, 336 wise_ref: Some(tx.reference_number), 337 amount: tx 338 .amount 339 .try_into() 340 .map_err(|err| WorkerError::InvalidStatement(format!("invalid amount: {err}")))?, 341 subject: tx.details.payment_reference, 342 name: tx.details.sender_name, 343 debtor: parse_account(&tx.details.sender_account), 344 value_at: tx.date, 345 }; 346 let mut result = register_tx_in(pool, &t, &subject, &now).await?; 347 let failure = match result { 348 AddIncomingResult::Success { .. } => None, 349 AddIncomingResult::ReservePubReuse => Some("reserve pub reuse"), 350 AddIncomingResult::UnknownMapping => Some("unknown mapping"), 351 AddIncomingResult::MappingReuse => Some("mapping reuse"), 352 }; 353 if failure.is_some() { 354 result = register_tx_in(pool, &t, &None, &now).await?; 355 } 356 let AddIncomingResult::Success { new, row_id, .. } = result else { 357 unreachable!("registration without a Taler subject cannot fail subject checks"); 358 }; 359 debug!(target: "worker", incoming_tx_id = row_id, wise_ref = ?t.wise_ref, "incoming transaction registration"); 360 if !new { 361 trace!(target: "worker", "in {t} already seen"); 362 return Ok(None); 363 } 364 if let Some(reason) = failure { 365 info!(target: "worker", "in {t} malformed: {reason}"); 366 } else { 367 info!(target: "worker", "in {t}"); 368 } 369 if t.debtor.is_none() { 370 warn!(target: "worker", "in {t} cannot parse debtor account '{}'", tx.details.sender_account); 371 } 372 Ok(Some(t)) 373 }