From 3812a92ef46b444e06188a7e1a2a7d83dc86c73a Mon Sep 17 00:00:00 2001 From: Yosemite <78393994+yosemite01@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:28:08 +0100 Subject: [PATCH 1/2] fix: repair syntax errors blocking compilation on main Four unrelated pre-existing bugs kept the crate from compiling at all: misplaced inner doc comments in queue/mod.rs, #[allow(...)] attributes placed mid method-chain in realtime_broadcaster.rs and shutdown.rs, and a duplicated/dead code block in rpc/stellar.rs that left a brace unclosed. Filed Stellar-Insightss/backend#51 for the remaining dependency/API-mismatch errors this does not address. --- src/queue/mod.rs | 27 ++++++++++++++------------- src/rpc/stellar.rs | 7 ------- src/services/realtime_broadcaster.rs | 6 +++--- src/shutdown.rs | 11 +++++------ 4 files changed, 22 insertions(+), 29 deletions(-) diff --git a/src/queue/mod.rs b/src/queue/mod.rs index 42915c9..d127187 100644 --- a/src/queue/mod.rs +++ b/src/queue/mod.rs @@ -16,19 +16,20 @@ pub mod types; pub use replay::{QueueProcessor, QueueReplayError}; pub use types::{QueueMessage, QueueMessageStatus, QueueMessageType}; -//! Idempotent offline-action replay queue. -//! -//! Mobile and frontend clients persist mutations locally while offline (see -//! `mobile/src/services/database.ts`'s `sync_queue` table and the replay -//! logic in `mobile/src/hooks/useOfflineCaching.ts`) and resubmit them once -//! connectivity returns. A replay can be interrupted and retried, so every -//! action carries a client-generated `id` that this queue treats as an -//! idempotency key: applying the same id twice runs the underlying mutation -//! at most once, even under concurrent replay (e.g. two devices, or a retry -//! racing the original request). -//! -//! See `docs/offline-sync.md` for the full client/server reconciliation -//! contract this module implements one half of. + +// Idempotent offline-action replay queue. +// +// Mobile and frontend clients persist mutations locally while offline (see +// `mobile/src/services/database.ts`'s `sync_queue` table and the replay +// logic in `mobile/src/hooks/useOfflineCaching.ts`) and resubmit them once +// connectivity returns. A replay can be interrupted and retried, so every +// action carries a client-generated `id` that this queue treats as an +// idempotency key: applying the same id twice runs the underlying mutation +// at most once, even under concurrent replay (e.g. two devices, or a retry +// racing the original request). +// +// See `docs/offline-sync.md` for the full client/server reconciliation +// contract this module implements one half of. use chrono::{DateTime, Utc}; use dashmap::mapref::entry::Entry; diff --git a/src/rpc/stellar.rs b/src/rpc/stellar.rs index 67c382c..cb80b92 100644 --- a/src/rpc/stellar.rs +++ b/src/rpc/stellar.rs @@ -902,13 +902,6 @@ impl StellarRpcClient { return Err(RpcError::ParseError( "internal: pagination field missing or malformed".to_string(), )); - // We just inserted "pagination" as an object above; this nested - // modification is safe, but we use if-let to avoid any .expect. - if let Some(obj) = params - .get_mut("pagination") - .and_then(|v| v.as_object_mut()) - { - obj.insert("cursor".to_string(), json!(c)); } } else if let Some(start) = start_ledger { params.insert("startLedger".to_string(), json!(start)); diff --git a/src/services/realtime_broadcaster.rs b/src/services/realtime_broadcaster.rs index 6520d65..6b506b5 100644 --- a/src/services/realtime_broadcaster.rs +++ b/src/services/realtime_broadcaster.rs @@ -84,12 +84,12 @@ impl RealtimeBroadcaster { pub async fn start(&mut self) { info!("Starting RealtimeBroadcaster service"); + // SAFETY: `start()` must only be called once per broadcaster instance. + // Taking twice is a programming error that should abort immediately. + #[allow(clippy::expect_used)] let shutdown_rx = self .shutdown_rx .take() - // SAFETY: `start()` must only be called once per broadcaster instance. - // Taking twice is a programming error that should abort immediately. - #[allow(clippy::expect_used)] .expect("Shutdown receiver already taken"); let corridor_task = self.start_corridor_broadcast_task(); diff --git a/src/shutdown.rs b/src/shutdown.rs index 0293c2e..46f5bac 100644 --- a/src/shutdown.rs +++ b/src/shutdown.rs @@ -112,14 +112,13 @@ pub async fn wait_for_signal() { { use tokio::signal::unix::{signal, SignalKind}; + // SAFETY: Failing to install a signal handler means the OS rejected + // the request — panicking at startup is the correct response. + #[allow(clippy::expect_used)] let mut sigterm = - // SAFETY: Failing to install a signal handler means the OS rejected - // the request — panicking at startup is the correct response. - #[allow(clippy::expect_used)] signal(SignalKind::terminate()).expect("Failed to install SIGTERM handler"); - let mut sigint = - #[allow(clippy::expect_used)] - signal(SignalKind::interrupt()).expect("Failed to install SIGINT handler"); + #[allow(clippy::expect_used)] + let mut sigint = signal(SignalKind::interrupt()).expect("Failed to install SIGINT handler"); tokio::select! { _ = sigterm.recv() => { From f8acc65c34aeadfb83869aa75011326965647f59 Mon Sep 17 00:00:00 2001 From: Yosemite <78393994+yosemite01@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:28:08 +0100 Subject: [PATCH 2/2] feat: add wallet activity-calendar and largest-transfers endpoints Closes #28 and #29. - GET /api/v1/wallets/{address}/activity-calendar: per-UTC-day transaction count for an address, from the ingested payments table. - GET /api/v1/wallets/{address}/largest-transfers: the address's largest transfers (in or out) ranked by USD value via the price feed, since comparing raw asset amounts across assets of different unit value is meaningless. Ranking is computed over the address's most recent 1000 transfers to keep the query and price-feed conversion pass bounded. Also adds PriceFeedClient::with_provider() so handlers that depend on price conversion can be tested with an injected provider instead of making real network calls to CoinGecko. --- src/api/mod.rs | 1 + src/api/v1/mod.rs | 6 +- src/api/wallets.rs | 356 +++++++++++++++++++++++++++++++++++++ src/openapi.rs | 9 + src/services/price_feed.rs | 18 ++ 5 files changed, 389 insertions(+), 1 deletion(-) create mode 100644 src/api/wallets.rs diff --git a/src/api/mod.rs b/src/api/mod.rs index 484a147..23141dd 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -35,4 +35,5 @@ pub mod transactions; pub mod trustlines; pub mod v1; pub mod verification_rewards; +pub mod wallets; pub mod webhooks; diff --git a/src/api/v1/mod.rs b/src/api/v1/mod.rs index fdbc1a3..eacbc09 100644 --- a/src/api/v1/mod.rs +++ b/src/api/v1/mod.rs @@ -1,6 +1,6 @@ use crate::api::{ account_merges, anchors, cache_stats, corridors, cost_calculator, fee_bump, liquidity_pools, - metrics, oauth, price_feed as price_feed_api, rpc, webhooks, + metrics, oauth, price_feed as price_feed_api, rpc, wallets, webhooks, }; use crate::auth_middleware::auth_middleware; use crate::cache::CacheManager; @@ -92,6 +92,9 @@ pub fn routes( ) .with_state(cached_state); + // Captured before `app_state` is moved into `protected_routes` below. + let wallets_db = app_state.db.clone(); + // 2. Public anchor routes let public_anchor_routes = Router::new() .route("/health", get(crate::handlers::health_check)) @@ -150,6 +153,7 @@ pub fn routes( ) .nest("/liquidity-pools", liquidity_pools::routes(lp_analyzer)) .nest("/prices", price_feed_api::routes(price_feed.clone())) + .nest("/wallets", wallets::routes(wallets_db, price_feed.clone())) .nest("/cost-calculator", cost_calculator::routes(price_feed)) .nest("/cache/stats", cache_stats::routes(cache.clone())) .nest("/metrics", metrics::routes(cache.clone())) diff --git a/src/api/wallets.rs b/src/api/wallets.rs new file mode 100644 index 0000000..d7d2d94 --- /dev/null +++ b/src/api/wallets.rs @@ -0,0 +1,356 @@ +use axum::{ + extract::{Path, Query, State}, + routing::get, + Json, Router, +}; +use serde::{Deserialize, Serialize}; +use std::collections::HashSet; +use std::sync::Arc; +use utoipa::{IntoParams, ToSchema}; + +use crate::database::Database; +use crate::error::{ApiError, ApiResult}; +use crate::services::price_feed::PriceFeedClient; +use crate::validation::validate_stellar_address; + +const DEFAULT_ACTIVITY_WINDOW_DAYS: i64 = 365; +const MAX_ACTIVITY_WINDOW_DAYS: i64 = 365; + +const DEFAULT_TRANSFERS_LIMIT: i64 = 20; +const MAX_TRANSFERS_LIMIT: i64 = 100; + +/// Bound on how many of the address's most recent transfers are considered +/// when ranking by USD value. Keeps the query and the price-feed conversion +/// pass cheap; addresses with more history than this get ranked over their +/// most recent `TRANSFER_CANDIDATE_POOL` transfers rather than their entire +/// lifetime. +const TRANSFER_CANDIDATE_POOL: i64 = 1000; + +#[derive(Clone)] +pub struct WalletState { + pub db: Arc, + pub price_feed: Arc, +} + +// --------------------------------------------------------------------------- +// Activity calendar +// --------------------------------------------------------------------------- + +#[derive(Debug, Serialize, ToSchema)] +pub struct ActivityDay { + /// UTC calendar date (YYYY-MM-DD) + #[schema(example = "2026-07-20")] + pub date: String, + /// Number of transactions touching this address on that day + #[schema(example = 12)] + pub count: i64, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct ActivityCalendarResponse { + pub address: String, + pub activity: Vec, +} + +#[derive(Debug, Deserialize, IntoParams)] +#[into_params(parameter_in = Query)] +pub struct ActivityCalendarQuery { + /// Trailing window size in days (1-365, default 365) + #[param(example = 365)] + pub days: Option, +} + +#[derive(sqlx::FromRow)] +struct ActivityDayRow { + day: String, + count: i64, +} + +/// Get per-day transaction activity for a wallet +/// +/// Returns a count of transactions touching the address (as source or +/// destination) for each UTC calendar day in the requested window, suitable +/// for rendering an activity calendar/heatmap. +#[utoipa::path( + get, + path = "/api/v1/wallets/{address}/activity-calendar", + params( + ("address" = String, Path, description = "Stellar account address (G...)"), + ActivityCalendarQuery + ), + responses( + (status = 200, description = "Per-day activity counts", body = ActivityCalendarResponse), + (status = 400, description = "Invalid Stellar address"), + (status = 500, description = "Internal server error") + ), + tag = "Wallets" +)] +pub async fn get_activity_calendar( + State(state): State, + Path(address): Path, + Query(params): Query, +) -> ApiResult> { + validate_stellar_address(&address)?; + let days = params + .days + .unwrap_or(DEFAULT_ACTIVITY_WINDOW_DAYS) + .clamp(1, MAX_ACTIVITY_WINDOW_DAYS); + + // date(created_at) truncates the stored RFC3339 timestamp to a UTC + // calendar date, matching the boundary convention used elsewhere in the + // wallet dashboard endpoints. + let rows = sqlx::query_as::<_, ActivityDayRow>( + r" + SELECT date(created_at) AS day, COUNT(*) AS count + FROM payments + WHERE (source_account = ?1 OR destination_account = ?1) + AND date(created_at) >= date('now', ?2) + GROUP BY day + ORDER BY day ASC + ", + ) + .bind(&address) + .bind(format!("-{days} days")) + .fetch_all(state.db.pool()) + .await + .map_err(|e| { + ApiError::internal( + "wallet_activity_calendar_query_failed", + format!("Failed to load activity calendar: {e}"), + ) + })?; + + let activity = rows + .into_iter() + .map(|r| ActivityDay { + date: r.day, + count: r.count, + }) + .collect(); + + Ok(Json(ActivityCalendarResponse { address, activity })) +} + +// --------------------------------------------------------------------------- +// Largest transfers +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Copy, Serialize, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub enum TransferDirection { + In, + Out, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct LargestTransfer { + /// Payment/operation id + pub id: String, + /// Transaction hash + pub transaction_hash: String, + /// The other party in the transfer + #[schema(example = "GA...")] + pub counterparty: String, + /// Whether the transfer was incoming or outgoing relative to the queried address + pub direction: TransferDirection, + /// Raw amount in the asset's own units + #[schema(example = 25000.0)] + pub amount: f64, + /// Asset code (e.g. "XLM", "USDC") + pub asset_code: String, + /// Asset issuer, `None` for native XLM + pub asset_issuer: Option, + /// Amount converted to USD via the price feed, used to rank "largest" + #[schema(example = 3125.5)] + pub amount_usd: f64, + /// RFC3339 timestamp + pub timestamp: String, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct LargestTransfersResponse { + pub address: String, + pub transfers: Vec, +} + +#[derive(Debug, Deserialize, IntoParams)] +#[into_params(parameter_in = Query)] +pub struct LargestTransfersQuery { + /// Number of transfers to return (1-100, default 20) + #[param(example = 20)] + pub limit: Option, +} + +#[derive(sqlx::FromRow)] +struct PaymentCandidateRow { + id: String, + transaction_hash: String, + source_account: String, + destination_account: String, + asset_type: String, + asset_code: Option, + asset_issuer: Option, + amount: f64, + created_at: String, +} + +/// Builds the price-feed asset identifier (e.g. `"XLM:native"` or +/// `"USDC:GISSUER..."`) for a payment row. +fn asset_identifier(asset_type: &str, code: Option<&str>, issuer: Option<&str>) -> String { + if asset_type == "native" { + "XLM:native".to_string() + } else { + format!( + "{}:{}", + code.unwrap_or("UNKNOWN"), + issuer.unwrap_or("unknown") + ) + } +} + +/// Get the largest transfers for a wallet by USD value +/// +/// Returns the address's largest transfers (incoming or outgoing), ranked by +/// value converted to a common USD basis via the price feed -- comparing raw +/// asset amounts directly would be meaningless across assets of very +/// different unit value (e.g. a $10,000 XLM transfer vs. a $10 USDC +/// transfer). Assets with no known price mapping are excluded from the +/// ranking, since "largest" is undefined for them without a conversion rate. +#[utoipa::path( + get, + path = "/api/v1/wallets/{address}/largest-transfers", + params( + ("address" = String, Path, description = "Stellar account address (G...)"), + LargestTransfersQuery + ), + responses( + (status = 200, description = "Largest transfers by USD value", body = LargestTransfersResponse), + (status = 400, description = "Invalid Stellar address"), + (status = 500, description = "Internal server error") + ), + tag = "Wallets" +)] +pub async fn get_largest_transfers( + State(state): State, + Path(address): Path, + Query(params): Query, +) -> ApiResult> { + validate_stellar_address(&address)?; + let limit = params + .limit + .unwrap_or(DEFAULT_TRANSFERS_LIMIT) + .clamp(1, MAX_TRANSFERS_LIMIT); + + let candidates = sqlx::query_as::<_, PaymentCandidateRow>( + r" + SELECT id, transaction_hash, source_account, destination_account, + asset_type, asset_code, asset_issuer, amount, created_at + FROM payments + WHERE source_account = ?1 OR destination_account = ?1 + ORDER BY created_at DESC + LIMIT ?2 + ", + ) + .bind(&address) + .bind(TRANSFER_CANDIDATE_POOL) + .fetch_all(state.db.pool()) + .await + .map_err(|e| { + ApiError::internal( + "wallet_largest_transfers_query_failed", + format!("Failed to load transfer candidates: {e}"), + ) + })?; + + let asset_ids: Vec = { + let mut seen = HashSet::new(); + candidates + .iter() + .map(|c| { + asset_identifier( + &c.asset_type, + c.asset_code.as_deref(), + c.asset_issuer.as_deref(), + ) + }) + .filter(|id| seen.insert(id.clone())) + .collect() + }; + + let prices = state.price_feed.get_prices(&asset_ids).await; + + let mut transfers: Vec = candidates + .into_iter() + .filter_map(|c| { + let asset_id = asset_identifier( + &c.asset_type, + c.asset_code.as_deref(), + c.asset_issuer.as_deref(), + ); + let price = *prices.get(&asset_id)?; + let amount_usd = c.amount * price; + + let direction = if c.source_account == address { + TransferDirection::Out + } else { + TransferDirection::In + }; + let counterparty = if direction == TransferDirection::Out { + c.destination_account + } else { + c.source_account + }; + + Some(LargestTransfer { + id: c.id, + transaction_hash: c.transaction_hash, + counterparty, + direction, + amount: c.amount, + asset_code: c.asset_code.unwrap_or_else(|| "XLM".to_string()), + asset_issuer: c.asset_issuer, + amount_usd, + timestamp: c.created_at, + }) + }) + .collect(); + + transfers.sort_by(|a, b| b.amount_usd.total_cmp(&a.amount_usd)); + transfers.truncate(limit as usize); + + Ok(Json(LargestTransfersResponse { address, transfers })) +} + +pub fn routes(db: Arc, price_feed: Arc) -> Router { + let state = WalletState { db, price_feed }; + Router::new() + .route("/:address/activity-calendar", get(get_activity_calendar)) + .route("/:address/largest-transfers", get(get_largest_transfers)) + .with_state(state) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_asset_identifier_native() { + assert_eq!(asset_identifier("native", None, None), "XLM:native"); + } + + #[test] + fn test_asset_identifier_credit_asset() { + assert_eq!( + asset_identifier("credit_alphanum4", Some("USDC"), Some("GISSUER")), + "USDC:GISSUER" + ); + } + + #[test] + fn test_asset_identifier_missing_code_falls_back() { + assert_eq!( + asset_identifier("credit_alphanum4", None, None), + "UNKNOWN:unknown" + ); + } +} diff --git a/src/openapi.rs b/src/openapi.rs index 0e6009c..71771d6 100644 --- a/src/openapi.rs +++ b/src/openapi.rs @@ -162,6 +162,9 @@ use utoipa::OpenApi; crate::api::verification_rewards::get_leaderboard, crate::api::verification_rewards::get_user_verifications, crate::api::verification_rewards::get_public_user_stats, + // Wallets + crate::api::wallets::get_activity_calendar, + crate::api::wallets::get_largest_transfers, ), components( schemas( @@ -182,6 +185,11 @@ use utoipa::OpenApi; crate::api::cost_calculator::RouteEstimate, crate::api::cost_calculator::CostCalculationResponse, crate::api::cost_calculator::ErrorResponse, + crate::api::wallets::ActivityDay, + crate::api::wallets::ActivityCalendarResponse, + crate::api::wallets::TransferDirection, + crate::api::wallets::LargestTransfer, + crate::api::wallets::LargestTransfersResponse, ) ), tags( @@ -214,6 +222,7 @@ use utoipa::OpenApi; (name = "SEP-10", description = "SEP-10 authentication endpoints"), (name = "SEP-24", description = "SEP-24 hosted deposit/withdrawal endpoints"), (name = "Verification Rewards", description = "Snapshot verification reward endpoints"), + (name = "Wallets", description = "Per-wallet dashboard endpoints (balance, activity, transfers)"), ) )] pub struct ApiDoc; diff --git a/src/services/price_feed.rs b/src/services/price_feed.rs index e260a98..8cc9c8b 100644 --- a/src/services/price_feed.rs +++ b/src/services/price_feed.rs @@ -209,6 +209,24 @@ impl PriceFeedClient { } } + /// Create a price feed client with an explicit provider, bypassing the + /// `CoinGecko`-only construction in [`Self::new`]. Used to inject a + /// deterministic provider in tests so handlers that depend on price + /// conversion don't make real network calls. + #[must_use] + pub fn with_provider( + provider: Arc, + config: PriceFeedConfig, + asset_mapping: HashMap, + ) -> Self { + Self { + provider, + cache: Arc::new(RwLock::new(HashMap::new())), + asset_mapping: Arc::new(asset_mapping), + config, + } + } + /// Get price for a Stellar asset, returns USD value pub async fn get_price(&self, stellar_asset: &str) -> Result { // Check cache first