worker.rs (6836B)
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::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 failure = 107 match register_tx_in(pool, &t, &subject.ok(), &now).await? { 108 AddIncomingResult::Success { new, .. } => { 109 if new { 110 info!(target: "worker", "in {t}"); 111 if t.debtor.is_none() { 112 warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account) 113 } 114 } else { 115 trace!(target: "worker", "in {t} already seen"); 116 } 117 continue; 118 } 119 AddIncomingResult::ReservePubReuse => "reserve pub reuse", 120 AddIncomingResult::UnknownMapping => "unknown mapping", 121 AddIncomingResult::MappingReuse => "mapping reuse", 122 }; 123 124 match register_tx_in(pool, &t, &None, &now).await? { 125 AddIncomingResult::Success { new, .. } => { 126 if new { 127 info!(target: "worker", "in {t}: {failure}"); 128 if t.debtor.is_none() { 129 warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account) 130 } 131 } else { 132 trace!(target: "worker", "in {t} already seen: {failure}"); 133 } 134 continue; 135 } 136 AddIncomingResult::ReservePubReuse 137 | AddIncomingResult::UnknownMapping 138 | AddIncomingResult::MappingReuse => unreachable!(), 139 }; 140 } 141 } 142 } 143 } 144 145 // then wait 146 if transient { 147 break Ok(()); 148 } 149 jitter.reset(); 150 tokio::time::sleep(cfg.frequency).await; 151 info!(target: "worker", "running at frequency"); 152 } 153 } 154 .await; 155 if transient { 156 res?; 157 return Ok(()); 158 } 159 let err = res.unwrap_err(); 160 error!(target: "worker", "{err}"); 161 tokio::time::sleep(jitter.backoff()).await; 162 } 163 }