api.rs (17868B)
1 use std::collections::{BTreeMap, BTreeSet}; 2 3 use anyhow::{Context, Result}; 4 use axum::{ 5 Json, 6 extract::{Query, State}, 7 }; 8 9 #[cfg(test)] 10 use crate::domain::Ledger; 11 use crate::{ 12 adapters::p2p::P2pMetrics, 13 app::{NodeStatus, PeerInfo}, 14 domain::{BlindedTransaction, OutPoint, Transaction, TxOutput}, 15 }; 16 17 use super::{ 18 BlocksQuery, ConfigResponse, MempoolCounts, MetricsQuery, MetricsResponse, 19 NetworkHealthLocalState, NetworkHealthResponse, Page, PageQuery, UiBlock, UiTransaction, 20 WalletTransactionContext, WalletTransactionFilters, WalletTransactionRow, 21 WalletTransactionsQuery, WalletUtxoRow, 22 }; 23 use super::{ 24 DATASET_LIMIT, DATASET_PAGE_LIMIT, EXPLORER_LIMIT, EXPLORER_PAGE_LIMIT, HttpState, 25 add_pending_outputs, cached_chain_view, cached_ui_blocks_for_tip, metrics_response, 26 network_health, ui_blinded_reveal, ui_blinded_transaction, ui_blocks_from_indexes, 27 ui_pending_revealed_transaction, ui_transaction, wallet_transaction_row, 28 wallet_transaction_rows, 29 }; 30 31 pub(super) async fn api_status(State(state): State<HttpState>) -> Json<NodeStatus> { 32 let mut status = state.node.lock().await.status(); 33 status.stratum = state.stratum.clone(); 34 Json(status) 35 } 36 37 pub(super) async fn api_blocks( 38 State(state): State<HttpState>, 39 Query(query): Query<BlocksQuery>, 40 ) -> Json<Vec<UiBlock>> { 41 let limit = query 42 .limit 43 .unwrap_or(EXPLORER_PAGE_LIMIT) 44 .min(EXPLORER_LIMIT); 45 let (tip_hash, blocks) = { 46 let node = state.node.lock().await; 47 let blocks = match query.before_height { 48 Some(before_height) => node.blocks_before(before_height, limit), 49 None => node.recent_blocks(limit), 50 }; 51 (node.chain_tip_hash(), blocks) 52 }; 53 if let Some(blocks) = cached_ui_blocks_for_tip(&state, Some(tip_hash.as_str()), blocks).await { 54 return Json(blocks); 55 } 56 57 let (snapshot, blocks) = { 58 let node = state.node.lock().await; 59 let snapshot = node.chain_snapshot(); 60 let blocks = match query.before_height { 61 Some(before_height) => node.blocks_before(before_height, limit), 62 None => node.recent_blocks(limit), 63 }; 64 (snapshot, blocks) 65 }; 66 let view = cached_chain_view(&state, &snapshot) 67 .await 68 .unwrap_or_default(); 69 Json(ui_blocks_from_indexes( 70 blocks, 71 &view.outputs, 72 &view.revealed_by_height, 73 &view.burn_leader_ranks_by_hash, 74 )) 75 } 76 77 pub(super) async fn api_config(State(state): State<HttpState>) -> Json<ConfigResponse> { 78 Json(ConfigResponse { 79 config: state.ui_config.lock().await.clone(), 80 p2p_inbound_runtime_active: state.gossip.accepts_inbound().await, 81 p2p_runtime_bind_addr: state.gossip.listen_addr().to_string(), 82 }) 83 } 84 85 pub(super) async fn api_mempool( 86 State(state): State<HttpState>, 87 Query(query): Query<PageQuery>, 88 ) -> Json<Page<UiTransaction>> { 89 let ui_data_ready = ensure_ui_data_current(&state).await.is_ok(); 90 let (pending, pending_blinded, pending_reveals, pending_revealed) = { 91 let node = state.node.lock().await; 92 let pending = node.pending_transactions(); 93 let pending_blinded = node.pending_blinded_transactions(); 94 let pending_reveals = node.pending_blinded_reveals(); 95 let pending_revealed = node 96 .pending_revealed_blinded_transactions() 97 .into_iter() 98 .map(|revealed| (revealed.commitment.clone(), revealed)) 99 .collect::<BTreeMap<_, _>>(); 100 (pending, pending_blinded, pending_reveals, pending_revealed) 101 }; 102 let mut required_outputs = BTreeSet::new(); 103 collect_transaction_input_outpoints(pending.iter(), &mut required_outputs); 104 collect_blinded_input_outpoints(pending_blinded.iter(), &mut required_outputs); 105 collect_transaction_input_outpoints( 106 pending_revealed 107 .values() 108 .map(|revealed| &revealed.transaction), 109 &mut required_outputs, 110 ); 111 let mut outputs = if ui_data_ready { 112 load_outputs_for_outpoints(&state, required_outputs) 113 .await 114 .unwrap_or_default() 115 } else { 116 BTreeMap::new() 117 }; 118 add_pending_outputs(&mut outputs, &pending); 119 let mut items = pending 120 .iter() 121 .map(|tx| ui_transaction(tx, &outputs)) 122 .collect::<Vec<_>>(); 123 items.extend( 124 pending_blinded 125 .iter() 126 .map(|transaction| ui_blinded_transaction(transaction, &outputs)), 127 ); 128 items.extend(pending_reveals.iter().map(|reveal| { 129 pending_revealed 130 .get(&reveal.commitment) 131 .map(|revealed| ui_pending_revealed_transaction(revealed, &outputs)) 132 .unwrap_or_else(|| ui_blinded_reveal(reveal)) 133 })); 134 items.reverse(); 135 Json(page_items(items, query)) 136 } 137 138 pub(super) async fn api_wallet_transactions( 139 State(state): State<HttpState>, 140 Query(query): Query<WalletTransactionsQuery>, 141 ) -> Json<Page<WalletTransactionRow>> { 142 let page_query = query.page(); 143 let offset = page_query.offset.unwrap_or(0); 144 let limit = page_query 145 .limit 146 .unwrap_or(DATASET_PAGE_LIMIT) 147 .clamp(1, DATASET_LIMIT); 148 let filters = WalletTransactionFilters::from_query(query); 149 if ensure_ui_data_current(&state).await.is_err() { 150 return Json(Page { 151 items: Vec::new(), 152 offset, 153 limit, 154 total: 0, 155 has_more: false, 156 next_offset: None, 157 }); 158 } 159 let (wallet, pending, owned_blinded) = { 160 let node = state.node.lock().await; 161 ( 162 node.wallet_address().to_string(), 163 node.pending_transactions(), 164 node.owned_blinded_payloads(), 165 ) 166 }; 167 let mut pending_required_outputs = BTreeSet::new(); 168 collect_transaction_input_outpoints(pending.iter(), &mut pending_required_outputs); 169 collect_transaction_input_outpoints(owned_blinded.iter(), &mut pending_required_outputs); 170 let mut pending_outputs = load_outputs_for_outpoints(&state, pending_required_outputs) 171 .await 172 .unwrap_or_default(); 173 add_pending_outputs(&mut pending_outputs, &pending); 174 let pending_rows = wallet_transaction_rows( 175 &wallet, 176 pending.clone(), 177 owned_blinded.clone(), 178 &[], 179 &BTreeMap::new(), 180 &pending_outputs, 181 filters, 182 ); 183 let pending_total = pending_rows.len(); 184 let mut items = pending_rows 185 .into_iter() 186 .skip(offset.min(pending_total)) 187 .take(limit) 188 .collect::<Vec<_>>(); 189 190 let confirmed_offset = offset.saturating_sub(pending_total); 191 let remaining_limit = limit.saturating_sub(items.len()); 192 let kinds = wallet_transaction_filter_kinds(filters); 193 let store = state.ui_data_store.clone(); 194 let wallet_for_query = wallet.clone(); 195 let (confirmed_rows, confirmed_total) = if remaining_limit == 0 { 196 (Vec::new(), 0) 197 } else { 198 tokio::task::spawn_blocking(move || { 199 store.load_wallet_transactions( 200 &wallet_for_query, 201 &kinds, 202 confirmed_offset, 203 remaining_limit, 204 ) 205 }) 206 .await 207 .ok() 208 .and_then(Result::ok) 209 .unwrap_or_default() 210 }; 211 let mut confirmed_required_outputs = BTreeSet::new(); 212 collect_transaction_input_outpoints( 213 confirmed_rows.iter().map(|row| &row.transaction), 214 &mut confirmed_required_outputs, 215 ); 216 let confirmed_outputs = load_outputs_for_outpoints(&state, confirmed_required_outputs) 217 .await 218 .unwrap_or_default(); 219 items.extend(confirmed_rows.into_iter().filter_map(|row| { 220 wallet_transaction_row( 221 &wallet, 222 &row.transaction, 223 &confirmed_outputs, 224 &WalletTransactionContext { 225 status: "confirmed", 226 block_height: Some(row.block_height), 227 timestamp_ms: Some(row.timestamp_ms), 228 block_finalizer: Some(row.block_finalizer), 229 blinded: row.blinded, 230 }, 231 ) 232 })); 233 let total = pending_total + confirmed_total; 234 let next_offset = offset + items.len(); 235 Json(Page { 236 items, 237 offset: offset.min(total), 238 limit, 239 total, 240 has_more: next_offset < total, 241 next_offset: (next_offset < total).then_some(next_offset), 242 }) 243 } 244 245 async fn load_outputs_for_outpoints( 246 state: &HttpState, 247 outpoints: BTreeSet<OutPoint>, 248 ) -> Result<BTreeMap<OutPoint, TxOutput>> { 249 if outpoints.is_empty() { 250 return Ok(BTreeMap::new()); 251 } 252 let store = state.ui_data_store.clone(); 253 tokio::task::spawn_blocking(move || store.load_outputs(&outpoints)) 254 .await 255 .unwrap_or_else(|_| Ok(BTreeMap::new())) 256 } 257 258 async fn ensure_ui_data_current(state: &HttpState) -> Result<()> { 259 let Some(tip_hash) = current_real_chain_tip(state).await else { 260 return Ok(()); 261 }; 262 if ui_data_matches_tip(state, tip_hash).await? { 263 return Ok(()); 264 } 265 266 let _refresh_guard = state.ui_data_refresh.lock().await; 267 let Some(tip_hash) = current_real_chain_tip(state).await else { 268 return Ok(()); 269 }; 270 if ui_data_matches_tip(state, tip_hash).await? { 271 return Ok(()); 272 } 273 274 let (snapshot, tip_hash) = { 275 let node = state.node.lock().await; 276 if !node.has_real_chain() { 277 return Ok(()); 278 } 279 (node.chain_snapshot(), node.chain_tip_hash()) 280 }; 281 let keep_metrics = state.ui_config.lock().await.keep_track_of_metrics; 282 let chain_store = state.chain_store.clone(); 283 let ui_data_store = state.ui_data_store.clone(); 284 tokio::task::spawn_blocking(move || { 285 chain_store 286 .save(&snapshot) 287 .context("failed to persist chain before UI data catch-up")?; 288 ui_data_store 289 .project_snapshot(&snapshot, keep_metrics) 290 .context("failed to project UI data catch-up")?; 291 Ok::<(), anyhow::Error>(()) 292 }) 293 .await 294 .context("UI data catch-up worker failed")??; 295 296 ui_data_matches_tip(state, tip_hash) 297 .await? 298 .then_some(()) 299 .context("UI data catch-up completed but projection tip does not match the chain tip") 300 } 301 302 async fn current_real_chain_tip(state: &HttpState) -> Option<String> { 303 let node = state.node.lock().await; 304 node.has_real_chain().then(|| node.chain_tip_hash()) 305 } 306 307 async fn ui_data_matches_tip(state: &HttpState, tip_hash: String) -> Result<bool> { 308 let store = state.ui_data_store.clone(); 309 tokio::task::spawn_blocking(move || store.is_projected_to(&tip_hash)) 310 .await 311 .context("UI data projection metadata worker failed")? 312 } 313 314 fn collect_transaction_input_outpoints<'a>( 315 transactions: impl IntoIterator<Item = &'a Transaction>, 316 outpoints: &mut BTreeSet<OutPoint>, 317 ) { 318 for transaction in transactions { 319 match transaction { 320 Transaction::Transfer { inputs, .. } | Transaction::Burn { inputs, .. } => { 321 outpoints.extend(inputs.iter().map(|input| input.outpoint.clone())); 322 } 323 Transaction::Mine { .. } => {} 324 } 325 } 326 } 327 328 fn collect_blinded_input_outpoints<'a>( 329 transactions: impl IntoIterator<Item = &'a BlindedTransaction>, 330 outpoints: &mut BTreeSet<OutPoint>, 331 ) { 332 for transaction in transactions { 333 outpoints.extend( 334 transaction 335 .inputs 336 .iter() 337 .map(|input| input.outpoint.clone()), 338 ); 339 } 340 } 341 342 pub(super) async fn api_wallet_utxos( 343 State(state): State<HttpState>, 344 Query(query): Query<PageQuery>, 345 ) -> Json<Page<WalletUtxoRow>> { 346 if ensure_ui_data_current(&state).await.is_err() { 347 return Json(page_items(Vec::new(), query)); 348 } 349 let (wallet, pending_spent) = { 350 let node = state.node.lock().await; 351 ( 352 node.wallet_address().to_string(), 353 node.wallet_pending_spent_outpoints(), 354 ) 355 }; 356 let store = state.ui_data_store.clone(); 357 let utxos = tokio::task::spawn_blocking(move || store.load_wallet_utxos(&wallet)) 358 .await 359 .ok() 360 .and_then(Result::ok) 361 .unwrap_or_default(); 362 Json(page_items( 363 wallet_utxo_rows_from_ui_data(utxos, &pending_spent), 364 query, 365 )) 366 } 367 368 pub(super) async fn api_wallet_selectable_utxos( 369 State(state): State<HttpState>, 370 ) -> Json<Vec<WalletUtxoRow>> { 371 if ensure_ui_data_current(&state).await.is_err() { 372 return Json(Vec::new()); 373 } 374 let (wallet, pending_spent) = { 375 let node = state.node.lock().await; 376 ( 377 node.wallet_address().to_string(), 378 node.wallet_pending_spent_outpoints(), 379 ) 380 }; 381 let store = state.ui_data_store.clone(); 382 let utxos = tokio::task::spawn_blocking(move || store.load_wallet_utxos(&wallet)) 383 .await 384 .ok() 385 .and_then(Result::ok) 386 .unwrap_or_default(); 387 Json( 388 wallet_utxo_rows_from_ui_data(utxos, &pending_spent) 389 .into_iter() 390 .filter(|utxo| utxo.spendable) 391 .collect(), 392 ) 393 } 394 395 pub(super) fn page_items<T>(items: Vec<T>, query: PageQuery) -> Page<T> { 396 let total = items.len(); 397 let offset = query.offset.unwrap_or(0).min(total); 398 let limit = query 399 .limit 400 .unwrap_or(DATASET_PAGE_LIMIT) 401 .clamp(1, DATASET_LIMIT); 402 let page_items = items 403 .into_iter() 404 .skip(offset) 405 .take(limit) 406 .collect::<Vec<_>>(); 407 let next_offset = offset + page_items.len(); 408 Page { 409 items: page_items, 410 offset, 411 limit, 412 total, 413 has_more: next_offset < total, 414 next_offset: (next_offset < total).then_some(next_offset), 415 } 416 } 417 418 fn wallet_transaction_filter_kinds(filters: WalletTransactionFilters) -> Vec<&'static str> { 419 let mut kinds = Vec::new(); 420 if filters.transfer { 421 kinds.push("transfer"); 422 } 423 if filters.mine { 424 kinds.push("mine"); 425 } 426 if filters.burn { 427 kinds.push("burn"); 428 } 429 kinds 430 } 431 432 #[cfg(test)] 433 pub(super) fn wallet_utxo_rows(ledger: &Ledger, wallet: &str) -> Vec<WalletUtxoRow> { 434 let spendable_outpoints = ledger 435 .available_utxos_for_address(wallet) 436 .unwrap_or_default() 437 .into_iter() 438 .map(|(outpoint, _)| outpoint) 439 .collect::<BTreeSet<_>>(); 440 let mut utxos = ledger 441 .utxos_for_address(wallet) 442 .into_iter() 443 .map(|(outpoint, output)| { 444 let spendable = spendable_outpoints.contains(&outpoint); 445 WalletUtxoRow { 446 outpoint, 447 address: output.address, 448 amount: output.amount, 449 spendable, 450 } 451 }) 452 .collect::<Vec<_>>(); 453 utxos.sort_by(|left, right| { 454 right 455 .amount 456 .cmp(&left.amount) 457 .then_with(|| left.outpoint.txid.cmp(&right.outpoint.txid)) 458 .then_with(|| left.outpoint.index.cmp(&right.outpoint.index)) 459 }); 460 utxos 461 } 462 463 fn wallet_utxo_rows_from_ui_data( 464 utxos: Vec<(crate::domain::OutPoint, crate::domain::TxOutput)>, 465 pending_spent: &BTreeSet<crate::domain::OutPoint>, 466 ) -> Vec<WalletUtxoRow> { 467 utxos 468 .into_iter() 469 .map(|(outpoint, output)| { 470 let spendable = !pending_spent.contains(&outpoint); 471 WalletUtxoRow { 472 outpoint, 473 address: output.address, 474 amount: output.amount, 475 spendable, 476 } 477 }) 478 .collect() 479 } 480 481 #[cfg(test)] 482 pub(super) fn selectable_wallet_utxo_rows(ledger: &Ledger, wallet: &str) -> Vec<WalletUtxoRow> { 483 wallet_utxo_rows(ledger, wallet) 484 .into_iter() 485 .filter(|utxo| utxo.spendable) 486 .collect() 487 } 488 489 pub(super) async fn api_peers( 490 State(state): State<HttpState>, 491 Query(query): Query<PageQuery>, 492 ) -> Json<Page<PeerInfo>> { 493 Json(page_items(state.peers.lock().await.list(), query)) 494 } 495 496 pub(super) async fn api_p2p_metrics(State(state): State<HttpState>) -> Json<P2pMetrics> { 497 Json(state.gossip.metrics()) 498 } 499 500 pub(super) async fn api_metrics( 501 State(state): State<HttpState>, 502 Query(query): Query<MetricsQuery>, 503 ) -> Json<MetricsResponse> { 504 let enabled = state.ui_config.lock().await.keep_track_of_metrics; 505 if !enabled { 506 return Json(MetricsResponse { 507 enabled, 508 latest: None, 509 charts: Vec::new(), 510 }); 511 } 512 if ensure_ui_data_current(&state).await.is_err() { 513 return Json(MetricsResponse { 514 enabled, 515 latest: None, 516 charts: Vec::new(), 517 }); 518 } 519 let store = state.ui_data_store.clone(); 520 let rows = tokio::task::spawn_blocking(move || match query.limit { 521 Some(limit) => store.load_recent_metrics(limit.clamp(1, DATASET_LIMIT)), 522 None => store.load_metrics(), 523 }) 524 .await 525 .ok() 526 .and_then(Result::ok) 527 .unwrap_or_default(); 528 Json(metrics_response(enabled, rows)) 529 } 530 531 pub(super) async fn api_network_health( 532 State(state): State<HttpState>, 533 ) -> Json<NetworkHealthResponse> { 534 let (local, mempool) = { 535 let node = state.node.lock().await; 536 let mempool = MempoolCounts { 537 plain_transactions: node.pending_transactions().len(), 538 blinded_transactions: node.pending_blinded_transactions().len(), 539 blinded_reveals: node.pending_blinded_reveals().len(), 540 }; 541 ( 542 NetworkHealthLocalState { 543 height: node.chain_height(), 544 pending_transactions: mempool.total(), 545 }, 546 mempool, 547 ) 548 }; 549 let peers = state.peers.lock().await.list(); 550 Json(network_health(local, &peers, mempool)) 551 }