commit 080c8587ba31f63a7ca8b929b06ef2808ce723ad
parent 43a28f56d4dfd8237c6c29c19de14ea526ec5245
Author: Antoine A <>
Date: Wed, 14 Oct 2026 17:42:05 +0200
wise: wake up on webhooks notification
Diffstat:
10 files changed, 263 insertions(+), 82 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -2714,6 +2714,7 @@ version = "1.5.0"
dependencies = [
"anyhow",
"aws-lc-rs",
+ "axum",
"clap",
"compact_str",
"http-client",
diff --git a/adapters/taler-wise/Cargo.toml b/adapters/taler-wise/Cargo.toml
@@ -22,7 +22,8 @@ taler-macros.workspace = true
url.workspace = true
http-client.workspace = true
serde.workspace = true
-serde_json.workspace = true
+serde_json = { workspace = true, features = ["raw_value"] }
+axum.workspace = true
hyper.workspace = true
aws-lc-rs.workspace = true
thiserror.workspace = true
diff --git a/adapters/taler-wise/src/api.rs b/adapters/taler-wise/src/api.rs
@@ -16,9 +16,19 @@
use std::{collections::BTreeMap, sync::Arc};
-use compact_str::format_compact;
+use aws_lc_rs::signature::ParsedPublicKey;
+use axum::{
+ body::Bytes,
+ extract::State,
+ http::{HeaderMap, HeaderName, StatusCode},
+ routing::post,
+};
+use compact_str::{CompactString, format_compact};
+use http_client::headers::HeaderParser;
use jiff::Timestamp;
use prometheus_client::{metrics::gauge::Gauge, registry::Registry};
+use serde::Deserialize;
+use serde_json::value::RawValue;
use sqlx::PgPool;
use taler_api::{
api::{
@@ -56,14 +66,25 @@ use crate::{
payto::FullWisePayto,
};
+#[allow(clippy::too_many_arguments)]
pub async fn start(
- pool: sqlx::PgPool,
+ pool: PgPool,
name: &str,
balances: Vec<WiseBalance>,
+ key: ParsedPublicKey,
+ profile_id: u64,
wire_gateway: &Option<ApiCfg>,
revenue: &Option<ApiCfg>,
observability: &Option<ApiCfg>,
) -> Router {
+ let mut main_router = Router::new()
+ .route("/wise-webhooks", post(receive_webhook))
+ .with_state(Arc::new(WebhookState {
+ pool: pool.clone(),
+ key,
+ profile: profile_id,
+ balances: balances.iter().map(|b| (b.id, b.currency)).collect(),
+ }));
let apis: Vec<_> = balances
.into_iter()
.map(|b| {
@@ -91,7 +112,6 @@ pub async fn start(
in_channels,
taler_in_channels,
));
- let mut main_router = Router::new();
for api in apis {
let api = Arc::new(api);
let mut balance_router = Router::new();
@@ -340,6 +360,87 @@ impl Observability for WiseBalanceApi {
}
}
+struct WebhookState {
+ pool: PgPool,
+ key: ParsedPublicKey,
+ profile: u64,
+ balances: BTreeMap<u32, Currency>,
+}
+
+#[derive(Deserialize)]
+struct WebhookEvent<'a> {
+ event_type: CompactString,
+ schema_version: CompactString,
+ subscription_id: uuid::Uuid,
+ sent_at: Timestamp,
+ #[serde(borrow)]
+ data: &'a RawValue,
+}
+
+#[derive(Deserialize)]
+struct BalanceEvent {
+ resource: Resource,
+ balance_id: u32,
+ currency: Currency,
+ occurred_at: Timestamp,
+ transfer_reference: Option<CompactString>,
+ transaction_type: CompactString,
+}
+
+#[derive(Deserialize)]
+struct Resource {
+ profile_id: u64,
+}
+
+async fn receive_webhook(
+ State(state): State<Arc<WebhookState>>,
+ headers: HeaderMap,
+ body: Bytes,
+) -> Result<StatusCode, StatusCode> {
+ // Verify signature
+ let signature = headers
+ .b64_header(HeaderName::from_static("x-signature-sha256"))
+ .map_err(|_| StatusCode::UNAUTHORIZED)?;
+ state
+ .key
+ .verify_sig(&body, &signature)
+ .map_err(|_| StatusCode::UNAUTHORIZED)?;
+
+ // Parse webhook event
+ let event: WebhookEvent<'_> =
+ serde_json::from_slice(&body).map_err(|_| StatusCode::BAD_REQUEST)?;
+ tracing::debug!(target: "webhook",
+ event_type = %event.event_type,
+ schema_version = %event.schema_version,
+ subscription_id = %event.subscription_id,
+ sent_at = %event.sent_at,
+ "received Wise webhook"
+ );
+
+ // We only handle balances update right now
+ if event.event_type != "balances#update" || event.schema_version != "4.0.0" {
+ return Ok(StatusCode::NO_CONTENT);
+ }
+ let balance: BalanceEvent =
+ serde_json::from_str(event.data.get()).map_err(|_| StatusCode::BAD_REQUEST)?;
+ if balance.transaction_type != "credit"
+ || balance.resource.profile_id != state.profile
+ || state.balances.get(&balance.balance_id) != Some(&balance.currency)
+ {
+ return Ok(StatusCode::NO_CONTENT);
+ }
+ sqlx::query("SELECT pg_notify('wise_balance_credit', $1)")
+ .bind(balance.balance_id.to_string())
+ .execute(&state.pool)
+ .await
+ .map_err(|err| {
+ tracing::error!(target: "webhook", "cannot notify balance change: {err}");
+ StatusCode::SERVICE_UNAVAILABLE
+ })?;
+ tracing::debug!(target: "webhook", balance_id = balance.balance_id, transfer_reference = ?balance.transfer_reference, occurred_at = %balance.occurred_at, "notified balance change");
+ Ok(StatusCode::NO_CONTENT)
+}
+
#[cfg(test)]
mod test {
@@ -374,7 +475,7 @@ mod test {
use crate::{
api,
- config::WiseBalance,
+ config::{ApiEnv, WiseBalance},
db::{TxIn, register_tx_in},
payto::{FullWisePayto, WiseAccount},
};
@@ -403,6 +504,8 @@ mod test {
pool.clone(),
"Exchange",
balances,
+ ApiEnv::sandbox.public_key(),
+ 7,
&Some(ApiCfg {
auth: AuthCfg::None,
}),
diff --git a/adapters/taler-wise/src/config.rs b/adapters/taler-wise/src/config.rs
@@ -16,7 +16,7 @@
use std::{str::FromStr as _, sync::LazyLock, time::Duration};
-use aws_lc_rs::rsa::PublicKey;
+use aws_lc_rs::signature::{ParsedPublicKey, RSA_PKCS1_2048_8192_SHA256};
use compact_str::CompactString;
use taler_api::{
Serve,
@@ -61,7 +61,7 @@ impl ApiEnv {
}
}
- pub fn public_key(self) -> PublicKey {
+ pub fn public_key(self) -> ParsedPublicKey {
let pem = match self {
ApiEnv::production => {
"-----BEGIN PUBLIC KEY-----
@@ -87,7 +87,7 @@ VwIDAQAB
}
};
let parsed = pem::parse(pem).unwrap();
- aws_lc_rs::rsa::PublicKey::from_der(parsed.contents()).unwrap()
+ ParsedPublicKey::new(&RSA_PKCS1_2048_8192_SHA256, parsed.contents()).unwrap()
}
}
@@ -150,7 +150,7 @@ impl SetupCfg {
let s = cfg.section("wise-worker");
Ok(Self {
api: ms.parse("env", "API_ENV").require()?,
- profile_id: s.number("PROFILE_ID").opt()?,
+ profile_id: ms.number("PROFILE_ID").opt()?,
token: s.str("TOKEN").require()?,
name: ms.cstr("NAME").opt()?,
balances: balances(cfg)?,
@@ -160,6 +160,8 @@ impl SetupCfg {
/// taler-wise httpd config
pub struct ServeCfg {
+ pub api: ApiEnv,
+ pub profile_id: u64,
pub name: CompactString,
pub serve: Serve,
pub balances: Vec<WiseBalance>,
@@ -172,6 +174,8 @@ impl ServeCfg {
pub fn parse(cfg: &Config) -> Result<Self, CfgErr> {
let s = cfg.section("wise");
Ok(Self {
+ api: s.parse("env", "API_ENV").require()?,
+ profile_id: s.number("PROFILE_ID").require()?,
name: s.cstr("NAME").require()?,
serve: Serve::parse(&cfg.section("wise-httpd"))?,
wire_gateway: ApiCfg::parse(cfg.section("wise-httpd-wire-gateway-api"))?,
@@ -193,10 +197,11 @@ pub struct WorkerCfg {
impl WorkerCfg {
pub fn parse(cfg: &Config) -> Result<Self, CfgErr> {
+ let ms = cfg.section("wise");
let s = cfg.section("wise-worker");
let cfg = Self {
- api: cfg.section("wise").parse("env", "API_ENV").require()?,
- profile_id: s.number("PROFILE_ID").require()?,
+ api: ms.parse("env", "API_ENV").require()?,
+ profile_id: ms.number("PROFILE_ID").require()?,
token: s.str("TOKEN").require()?,
frequency: s.duration("FREQUENCY").require()?,
balances: balances(cfg)?,
diff --git a/adapters/taler-wise/src/main.rs b/adapters/taler-wise/src/main.rs
@@ -80,13 +80,7 @@ async fn run(cmd: Command, cfg: &Config) -> anyhow::Result<()> {
}
Command::Serve { check } => {
if check {
- let cfg = ServeCfg::parse(cfg)?;
- if cfg.revenue.is_none()
- && cfg.wire_gateway.is_none()
- && cfg.observability.is_none()
- {
- std::process::exit(1);
- }
+ ServeCfg::parse(cfg)?;
} else {
let pool = pool(cfg).await?;
let cfg = ServeCfg::parse(cfg)?;
@@ -94,6 +88,8 @@ async fn run(cmd: Command, cfg: &Config) -> anyhow::Result<()> {
pool,
&cfg.name,
cfg.balances,
+ cfg.api.public_key(),
+ cfg.profile_id,
&cfg.wire_gateway,
&cfg.revenue,
&cfg.observability,
diff --git a/adapters/taler-wise/src/worker.rs b/adapters/taler-wise/src/worker.rs
@@ -14,19 +14,19 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
-use std::{sync::LazyLock, time::Duration};
+use std::{
+ collections::BTreeSet,
+ sync::{LazyLock, Mutex},
+ time::Duration,
+};
use http_client::ApiErr;
use jiff::{SignedDuration, Span, Timestamp, tz::TimeZone};
use regex::Regex;
use sqlx::PgPool;
use taler_api::subject::{IncomingSubject, parse_incoming_unstructured};
-use taler_common::{
- ExpoBackoffDecorr,
- config::Config,
- types::{payto::BankID, utils::now_sql_stable_ts},
-};
-use tokio::time::Instant;
+use taler_common::{ExpoBackoffDecorr, config::Config, types::payto::BankID};
+use tokio::sync::Notify;
use tracing::{debug, error, info, trace, warn};
use crate::{
@@ -91,10 +91,28 @@ impl SyncState {
}
}
- fn delay(&self, now: Timestamp, frequency: Duration) -> Duration {
- remaining(self.early_success, frequency, now)
- .min(remaining(self.sync_success, HOUR, now))
- .min(remaining(self.checkpoint_success, DAY, now))
+ fn schedule(
+ &self,
+ now: Timestamp,
+ frequency: Duration,
+ ) -> (Timestamp, Tier, Timestamp, Option<Timestamp>, Duration) {
+ let early = remaining(self.early_success, frequency, now);
+ let sync = remaining(self.sync_success, HOUR, now);
+ let checkpoint = remaining(self.checkpoint_success, DAY, now);
+ let next = now + early.min(sync).min(checkpoint);
+ let (tier, cursor, last, overlap) = if checkpoint.is_zero() {
+ (
+ Tier::Checkpoint,
+ self.checkpoint,
+ self.checkpoint_success,
+ 2 * DAY,
+ )
+ } else if sync.is_zero() {
+ (Tier::Sync, self.sync, self.sync_success, 2 * HOUR)
+ } else {
+ (Tier::Early, self.early, self.early_success, Duration::ZERO)
+ };
+ (next, tier, cursor, last, overlap)
}
}
@@ -121,41 +139,116 @@ pub async fn run_worker(
Ok(())
}
+/// Coalesced balance wakeups with a single consumer.
+#[derive(Default)]
+struct BalanceWakeups {
+ pending: Mutex<BTreeSet<u32>>,
+ ready: Notify,
+}
+
+impl BalanceWakeups {
+ fn insert(&self, id: u32) {
+ if self.pending.lock().unwrap().insert(id) {
+ self.ready.notify_one();
+ }
+ }
+
+ fn remove(&self, id: u32) -> bool {
+ self.pending.lock().unwrap().remove(&id)
+ }
+
+ async fn recv(&self) {
+ while self.pending.lock().unwrap().is_empty() {
+ // notify_one retains a permit if an insert races with this wait.
+ self.ready.notified().await;
+ }
+ }
+}
+
async fn work_balances(
cfg: &WorkerCfg,
pool: &PgPool,
client: &Client<'_>,
transient: bool,
) -> WorkerResult {
- let mut next = Instant::now();
+ if transient {
+ for balance in &cfg.balances {
+ sync_balance(cfg, balance, pool, client, true)
+ .await
+ .map_err(|err| {
+ error!(target: "worker", "synchronization for {} failed: {err}", balance.id);
+ err
+ })?;
+ }
+ } else {
+ let wake = BalanceWakeups::default();
+ tokio::try_join!(
+ listen_balances(cfg, pool, &wake),
+ sync_balances(cfg, pool, client, &wake),
+ )?;
+ }
+ Ok(())
+}
+
+async fn listen_balances(cfg: &WorkerCfg, pool: &PgPool, wake: &BalanceWakeups) -> WorkerResult {
+ let configured: BTreeSet<_> = cfg.balances.iter().map(|b| b.id).collect();
+ taler_api::notification::notification_listener!(pool,
+ "wise_balance_credit" => (id: u32) {
+ if configured.contains(&id) {
+ wake.insert(id);
+ }
+ }
+ )
+}
+
+async fn sync_balances(
+ cfg: &WorkerCfg,
+ pool: &PgPool,
+ client: &Client<'_>,
+ wake: &BalanceWakeups,
+) -> WorkerResult {
+ let mut next_poll = Timestamp::now();
let mut jitter = ExpoBackoffDecorr::default();
loop {
- let mut delay = Duration::MAX;
- let mut first_error = None;
+ let mut deadline = next_poll;
+ let wait = Timestamp::now()
+ .duration_until(next_poll)
+ .max(SignedDuration::ZERO)
+ .unsigned_abs();
+ let poll = tokio::select! {
+ biased;
+ _ = tokio::time::sleep(wait) => {
+ deadline = Timestamp::now() + cfg.frequency;
+ true
+ },
+ _ = wake.recv() => false,
+ };
+ let mut failed = false;
for balance in &cfg.balances {
- tokio::time::sleep_until(next).await;
- match sync_balance(cfg, balance, pool, client, transient).await {
- Ok(wait) => delay = delay.min(wait),
+ let notified = wake.remove(balance.id);
+ if !poll && !notified {
+ continue;
+ }
+ match sync_balance(cfg, balance, pool, client, notified).await {
+ Ok(next) => deadline = deadline.min(next),
Err(err @ WorkerError::Db(sqlx::Error::PoolClosed)) => return Err(err),
Err(err) => {
+ failed = true;
let mut wait = jitter.backoff();
if let WorkerError::Api(api) = &err
&& let WiseErr::TooManyRequest(retry_after) = api.err.as_ref()
{
wait = wait.max(retry_after.unwrap_or_default());
}
- next = Instant::now() + wait;
- error!(target: "worker", "synchronization pass for {} failed: {err}", balance.id);
- first_error.get_or_insert(err);
+ wake.insert(balance.id);
+ error!(target: "worker", "synchronization for {} failed: {err}", balance.id);
+ tokio::time::sleep(wait).await;
}
}
}
- if transient {
- return first_error.map_or(Ok(()), Err);
- }
- if first_error.is_none() {
+ if !failed {
+ next_poll = deadline;
jitter.reset();
- next = Instant::now() + delay;
}
}
}
@@ -165,39 +258,16 @@ async fn sync_balance(
balance: &WiseBalance,
pool: &PgPool,
client: &Client<'_>,
- transient: bool,
-) -> WorkerResult<Duration> {
+ force: bool,
+) -> WorkerResult<Timestamp> {
let mut state = load_sync_state(pool, balance.id)
.await?
.ok_or(WorkerError::UninitializedBalance(balance.id))?;
- let now = now_sql_stable_ts();
- if !transient {
- let delay = state.delay(now, cfg.frequency);
- if !delay.is_zero() {
- return Ok(delay);
- }
+ let now = Timestamp::now();
+ let (next, tier, cursor, last, overlap) = state.schedule(now, cfg.frequency);
+ if !force && next > now {
+ return Ok(next);
}
- // Statements can miss transactions: early tracks the latest seen, hourly/daily
- // scans replay with 2h/2d lag. The broadest due scan advances newer tiers only
- // after inserts succeed. Idempotency makes replay safe.
- // We currently warn on recovered payments to see if this mechanism is usefull.
- let (tier, cursor, last, overlap) = if remaining(state.checkpoint_success, DAY, now).is_zero() {
- (
- Tier::Checkpoint,
- state.checkpoint,
- state.checkpoint_success,
- 2 * DAY,
- )
- } else if remaining(state.sync_success, HOUR, now).is_zero() {
- (Tier::Sync, state.sync, state.sync_success, 2 * HOUR)
- } else {
- (
- Tier::Early,
- state.early,
- state.early_success,
- Duration::ZERO,
- )
- };
let mut start = cursor.max(last.map_or(cursor, |last| last - overlap));
let previous_early = state.early_success.map(|last| state.early.max(last));
while start < now {
@@ -240,7 +310,7 @@ async fn sync_balance(
}
state.complete(tier, now);
save_sync_state(pool, balance.id, &state).await?;
- Ok(state.delay(Timestamp::now(), cfg.frequency))
+ Ok(state.schedule(now, cfg.frequency).0)
}
async fn register_transaction(
@@ -279,9 +349,10 @@ async fn register_transaction(
if failure.is_some() {
result = register_tx_in(pool, &t, &None, &now).await?;
}
- let AddIncomingResult::Success { new, .. } = result else {
+ let AddIncomingResult::Success { new, row_id, .. } = result else {
unreachable!("registration without a Taler subject cannot fail subject checks");
};
+ debug!(target: "worker", incoming_tx_id = row_id, wise_ref = ?t.wise_ref, "incoming transaction registration");
if !new {
trace!(target: "worker", "in {t} already seen");
return Ok(None);
diff --git a/adapters/taler-wise/wise.conf b/adapters/taler-wise/wise.conf
@@ -2,13 +2,13 @@
# Legal entity that is associated with the Wise account
NAME =
+# Adapter Wise profile ID
+PROFILE_ID =
+
# Api environment, this can either be production or sandbox
API_ENV = production
[wise-worker]
-# Adapter Wise profile ID
-PROFILE_ID =
-
# How often should worker run when no notification is received
FREQUENCY = 10m
@@ -105,4 +105,4 @@ TOKEN =
CONFIG = postgres:///taler-wise
# Where are the SQL files to setup our tables?
-SQL_DIR = ${DATADIR}/sql/
-\ No newline at end of file
+SQL_DIR = ${DATADIR}/sql/
diff --git a/common/http-client/src/headers.rs b/common/http-client/src/headers.rs
@@ -15,6 +15,7 @@
*/
use hyper::{HeaderMap, header::HeaderName};
+use taler_common::encoding::base64;
#[derive(Debug, thiserror::Error)]
pub enum HeaderError {
@@ -42,6 +43,9 @@ pub trait HeaderParser<E> {
fn bool_header(&self, name: impl Into<HeaderName>) -> Result<bool, E> {
self.parse(name, "boolean", |s| s.parse().map_err(|_| ()))
}
+ fn b64_header(&self, name: impl Into<HeaderName>) -> Result<Vec<u8>, E> {
+ self.parse(name, "b64", |s| base64::decode(s).map_err(|_| ()))
+ }
}
impl HeaderParser<HeaderError> for HeaderMap {
diff --git a/debian/etc/taler-wise/conf.d/wise-system.conf b/debian/etc/taler-wise/conf.d/wise-system.conf
@@ -1,5 +1,9 @@
# Configuration for system aspects of the wise adapter.
+[wise]
+# Adapter Wise profile ID
+PROFILE_ID =
+
# Read secret sections into configuration, but only
# if we have permission to do so.
@inline-secret@ wisedb-postgres ../secrets/wise-db.secret.conf
diff --git a/debian/etc/taler-wise/conf.d/wise-worker.conf b/debian/etc/taler-wise/conf.d/wise-worker.conf
@@ -1,9 +1,6 @@
# Configuration the wise adapter worker.
[wise-worker]
-# Adapter Wise profile ID
-PROFILE_ID =
-
# How often should worker run when no notification is received
# FREQUENCY =
