worker.rs (7118B)
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::sync::LazyLock; 18 19 use http_client::ApiErr; 20 use jiff::Timestamp; 21 use regex::Regex; 22 use sqlx::PgPool; 23 use taler_api::subject::{IncomingSubject, parse_incoming_unstructured}; 24 use taler_common::{ExpoBackoffDecorr, config::Config, types::payto::BankID}; 25 use tracing::{error, info, trace, warn}; 26 27 use crate::{ 28 config::WorkerCfg, 29 db::{AddIncomingResult, TxIn, register_tx_in}, 30 payto::WiseAccount, 31 wise_api::{ 32 client::{Client, WiseErr}, 33 types::Direction, 34 }, 35 }; 36 37 #[derive(Debug, thiserror::Error)] 38 pub enum WorkerError { 39 #[error(transparent)] 40 Db(#[from] sqlx::Error), 41 #[error(transparent)] 42 Api(#[from] ApiErr<WiseErr>), 43 } 44 45 pub type WorkerResult = Result<(), WorkerError>; 46 47 fn parse_account(account_str: &str) -> Option<WiseAccount> { 48 static IBAN_BIC_PATTERN: LazyLock<Regex> = 49 LazyLock::new(|| Regex::new(r"^\(([A-Z0-9]{8,11})\) ([A-Z0-9]+)$").unwrap()); 50 51 if let Some(caps) = IBAN_BIC_PATTERN.captures(account_str) { 52 let bic = caps[1].parse().ok()?; 53 let iban = caps[2].parse().ok()?; 54 55 return Some(WiseAccount::IBAN(BankID { 56 iban, 57 bic: Some(bic), 58 })); 59 } 60 61 None 62 } 63 64 pub async fn run_worker( 65 cfg: &Config, 66 pool: &PgPool, 67 client: &http_client::Client, 68 transient: bool, 69 ) -> anyhow::Result<()> { 70 let cfg = WorkerCfg::parse(cfg)?; 71 let client = Client::new(client, &cfg.token); 72 let mut jitter = ExpoBackoffDecorr::default(); 73 loop { 74 if !transient { 75 info!(target: "worker", "running at initialisation"); 76 } 77 let res: WorkerResult = async { 78 loop { 79 // Sync 80 for balance in &cfg.balances { 81 let stmt = client 82 .balance_statement(cfg.profile_id, balance.id, &balance.currency, "2026-07-16T00:00:00.000Z".parse().unwrap(), Timestamp::now() ) 83 .await 84 .unwrap(); 85 let now = Timestamp::now(); 86 for tx in stmt.transactions { 87 match tx.direction { 88 Direction::Debit => { 89 // TODO support outgoing transaction 90 } 91 Direction::Credit => { 92 let subject = 93 parse_incoming_unstructured(&tx.details.payment_reference); 94 // Parse sender account 95 let payto = parse_account(&tx.details.sender_account); 96 // 97 let t = TxIn { 98 balance_id: balance.id, 99 wise_ref: Some(tx.reference_number), 100 amount: tx.amount.into(), 101 subject: tx.details.payment_reference, 102 name: tx.details.sender_name, 103 debtor: payto, 104 value_at: tx.date, 105 }; 106 let subject = &match subject { 107 Ok(IncomingSubject::Key(key)) => Some(key), 108 Ok(IncomingSubject::AdminBalanceAdjust) | Err(_) => None, 109 }; 110 let failure = 111 match register_tx_in(pool, &t, subject, &now).await? { 112 AddIncomingResult::Success { new, .. } => { 113 if new { 114 info!(target: "worker", "in {t}"); 115 if t.debtor.is_none() { 116 warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account) 117 } 118 } else { 119 trace!(target: "worker", "in {t} already seen"); 120 } 121 continue; 122 } 123 AddIncomingResult::ReservePubReuse => "reserve pub reuse", 124 AddIncomingResult::UnknownMapping => "unknown mapping", 125 AddIncomingResult::MappingReuse => "mapping reuse", 126 }; 127 128 match register_tx_in(pool, &t, &None, &now).await? { 129 AddIncomingResult::Success { new, .. } => { 130 if new { 131 info!(target: "worker", "in {t}: {failure}"); 132 if t.debtor.is_none() { 133 warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account) 134 } 135 } else { 136 trace!(target: "worker", "in {t} already seen: {failure}"); 137 } 138 continue; 139 } 140 AddIncomingResult::ReservePubReuse 141 | AddIncomingResult::UnknownMapping 142 | AddIncomingResult::MappingReuse => unreachable!(), 143 }; 144 } 145 } 146 } 147 } 148 149 // then wait 150 if transient { 151 break Ok(()); 152 } 153 jitter.reset(); 154 tokio::time::sleep(cfg.frequency).await; 155 info!(target: "worker", "running at frequency"); 156 } 157 } 158 .await; 159 if transient { 160 res?; 161 return Ok(()); 162 } 163 let err = res.unwrap_err(); 164 error!(target: "worker", "{err}"); 165 tokio::time::sleep(jitter.backoff()).await; 166 } 167 }