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/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/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 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() => {