ui_data_store.rs (58542B)
1 use std::{ 2 collections::{BTreeMap, BTreeSet}, 3 fs, 4 path::{Path, PathBuf}, 5 time::{SystemTime, UNIX_EPOCH}, 6 }; 7 8 use anyhow::{Context, Result}; 9 use rusqlite::{Connection, OptionalExtension, params, params_from_iter, types::Value}; 10 11 use serde::Serialize; 12 13 use crate::{ 14 adapters::ui_index::{UiChainIndex, build_ui_chain_index}, 15 domain::{ 16 Amount, BLINDED_COMMITTER_FEE_BPS, BLINDED_FEE_BPS_DENOMINATOR, 17 BLINDED_REVEAL_BUNDLE_SIGNER_FEE_BPS, BlindedTransaction, Block, BurnLeaderRank, 18 ChainSnapshot, Ledger, MINE_REWARD, OutPoint, REVEAL_COMMITTEE_SIZE, 19 RevealedBlindedTransaction, Transaction, TxInput, TxOutput, blinded_reveal_finalizer_fee, 20 hex_hash, reveal_committee_slot_count_for_height, revealed_blinded_transactions, 21 }, 22 }; 23 24 const SCHEMA: &str = r#" 25 CREATE TABLE IF NOT EXISTS block_metrics ( 26 height INTEGER PRIMARY KEY, 27 block_hash TEXT NOT NULL, 28 timestamp_ms INTEGER NOT NULL, 29 block_time_ms INTEGER, 30 mine_difficulty_bits INTEGER NOT NULL, 31 circulating_supply INTEGER NOT NULL, 32 known_wallet_addresses INTEGER NOT NULL DEFAULT 0, 33 transaction_count INTEGER NOT NULL, 34 transfer_count INTEGER NOT NULL, 35 burn_count INTEGER NOT NULL, 36 mine_count INTEGER NOT NULL, 37 burned_amount INTEGER NOT NULL, 38 total_burned_amount INTEGER NOT NULL, 39 fees_amount INTEGER NOT NULL, 40 reward_amount INTEGER NOT NULL, 41 vdf_rounds INTEGER NOT NULL, 42 finalizer_rank INTEGER NOT NULL 43 ); 44 45 CREATE TABLE IF NOT EXISTS ui_cache_meta ( 46 id INTEGER PRIMARY KEY CHECK (id = 1), 47 schema_version INTEGER NOT NULL, 48 tip_hash TEXT NOT NULL, 49 updated_at_ms INTEGER NOT NULL 50 ); 51 52 CREATE TABLE IF NOT EXISTS ui_output_index ( 53 txid TEXT NOT NULL, 54 output_index INTEGER NOT NULL, 55 address TEXT NOT NULL, 56 amount INTEGER NOT NULL, 57 PRIMARY KEY (txid, output_index) 58 ); 59 60 CREATE TABLE IF NOT EXISTS ui_utxos ( 61 txid TEXT NOT NULL, 62 output_index INTEGER NOT NULL, 63 address TEXT NOT NULL, 64 amount INTEGER NOT NULL, 65 PRIMARY KEY (txid, output_index) 66 ); 67 68 CREATE INDEX IF NOT EXISTS idx_ui_utxos_address 69 ON ui_utxos(address); 70 71 CREATE TABLE IF NOT EXISTS ui_wallet_transactions ( 72 address TEXT NOT NULL, 73 sort_key INTEGER NOT NULL, 74 kind TEXT NOT NULL, 75 signature TEXT NOT NULL, 76 block_height INTEGER NOT NULL, 77 timestamp_ms INTEGER NOT NULL, 78 block_finalizer TEXT NOT NULL, 79 blinded INTEGER NOT NULL, 80 transaction_json BLOB NOT NULL, 81 PRIMARY KEY (address, signature) 82 ); 83 84 CREATE INDEX IF NOT EXISTS idx_ui_wallet_transactions_address_kind_sort 85 ON ui_wallet_transactions(address, kind, sort_key DESC); 86 87 CREATE INDEX IF NOT EXISTS idx_ui_wallet_transactions_address_sort 88 ON ui_wallet_transactions(address, sort_key DESC); 89 90 CREATE TABLE IF NOT EXISTS ui_revealed_transactions ( 91 height INTEGER NOT NULL, 92 commitment TEXT PRIMARY KEY, 93 included_by TEXT NOT NULL, 94 transaction_json BLOB NOT NULL 95 ); 96 97 CREATE INDEX IF NOT EXISTS idx_ui_revealed_transactions_height 98 ON ui_revealed_transactions(height); 99 100 CREATE TABLE IF NOT EXISTS ui_burn_leader_ranks ( 101 block_hash TEXT NOT NULL, 102 rank INTEGER NOT NULL, 103 ticket_id TEXT NOT NULL, 104 owner TEXT NOT NULL, 105 amount INTEGER NOT NULL, 106 eligible_from_height INTEGER NOT NULL, 107 eligible_until_height INTEGER NOT NULL, 108 PRIMARY KEY (block_hash, rank) 109 ); 110 111 CREATE TABLE IF NOT EXISTS ui_burn_leader_rank_blocks ( 112 block_hash TEXT PRIMARY KEY 113 ); 114 "#; 115 116 const UI_CACHE_SCHEMA_VERSION: u32 = 1; 117 118 #[derive(Clone, Debug, Eq, PartialEq, Serialize)] 119 #[serde(rename_all = "camelCase")] 120 pub struct BlockMetricRow { 121 pub height: u64, 122 pub block_hash: String, 123 pub timestamp_ms: u64, 124 pub block_time_ms: Option<u64>, 125 pub mine_difficulty_bits: u32, 126 pub circulating_supply: Amount, 127 pub known_wallet_addresses: u64, 128 pub transaction_count: u64, 129 pub transfer_count: u64, 130 pub burn_count: u64, 131 pub mine_count: u64, 132 pub burned_amount: Amount, 133 pub total_burned_amount: Amount, 134 pub fees_amount: Amount, 135 pub reward_amount: Amount, 136 pub vdf_rounds: u64, 137 pub finalizer_rank: u32, 138 } 139 140 #[derive(Clone, Debug, Eq, PartialEq)] 141 pub struct UiLeaderboardEntry { 142 pub address: String, 143 pub amount: Amount, 144 pub count: u64, 145 } 146 147 #[derive(Clone, Debug, Eq, PartialEq)] 148 pub struct WalletTransactionProjection { 149 pub sort_key: u64, 150 pub kind: String, 151 pub block_height: u64, 152 pub timestamp_ms: u64, 153 pub block_finalizer: String, 154 pub blinded: bool, 155 pub transaction: Transaction, 156 } 157 158 #[derive(Clone, Debug)] 159 pub struct SqliteUiDataStore { 160 path: PathBuf, 161 } 162 163 impl SqliteUiDataStore { 164 pub fn open(path: impl AsRef<Path>) -> Result<Self> { 165 let path = path.as_ref().to_path_buf(); 166 if let Some(parent) = path.parent() { 167 fs::create_dir_all(parent).with_context(|| { 168 format!( 169 "failed to create chain database directory {}", 170 parent.display() 171 ) 172 })?; 173 } 174 175 let store = Self { path }; 176 store.with_connection_mut(|connection| { 177 connection 178 .execute_batch(SCHEMA) 179 .context("failed to initialize UI data database schema")?; 180 ensure_block_metrics_column( 181 connection, 182 "known_wallet_addresses", 183 "INTEGER NOT NULL DEFAULT 0", 184 )?; 185 Ok(()) 186 })?; 187 Ok(store) 188 } 189 190 pub fn path(&self) -> &Path { 191 &self.path 192 } 193 194 pub(crate) fn load_ui_chain_index(&self, tip_hash: &str) -> Result<Option<UiChainIndex>> { 195 self.with_connection(|connection| { 196 let meta = connection 197 .query_row( 198 "SELECT schema_version, tip_hash FROM ui_cache_meta WHERE id = 1", 199 [], 200 |row| Ok((row.get::<_, u32>(0)?, row.get::<_, String>(1)?)), 201 ) 202 .optional() 203 .context("failed to load UI chain index metadata")?; 204 let Some((schema_version, stored_tip_hash)) = meta else { 205 return Ok(None); 206 }; 207 if schema_version != UI_CACHE_SCHEMA_VERSION || stored_tip_hash != tip_hash { 208 return Ok(None); 209 } 210 211 Ok(Some(UiChainIndex { 212 tip_hash: Some(stored_tip_hash), 213 outputs: load_ui_output_index(connection)?, 214 revealed_by_height: load_ui_revealed_transactions(connection)?, 215 burn_leader_ranks_by_hash: load_ui_burn_leader_ranks(connection)?, 216 })) 217 }) 218 } 219 220 pub(crate) fn is_projected_to(&self, tip_hash: &str) -> Result<bool> { 221 self.with_connection(|connection| { 222 let projected = connection 223 .query_row( 224 "SELECT schema_version, tip_hash FROM ui_cache_meta WHERE id = 1", 225 [], 226 |row| Ok((row.get::<_, u32>(0)?, row.get::<_, String>(1)?)), 227 ) 228 .optional() 229 .context("failed to load UI data projection metadata")? 230 .is_some_and(|(schema_version, stored_tip_hash)| { 231 schema_version == UI_CACHE_SCHEMA_VERSION && stored_tip_hash == tip_hash 232 }); 233 Ok(projected) 234 }) 235 } 236 237 pub fn project_snapshot(&self, snapshot: &ChainSnapshot, keep_metrics: bool) -> Result<()> { 238 let updated_at_ms = unix_ms(); 239 let ui_index = build_ui_chain_index(snapshot); 240 let utxos = Ledger::from_persisted_snapshot(snapshot.clone()) 241 .context("failed to rebuild ledger for UI UTXO projection")? 242 .all_utxos(); 243 let wallet_transactions = wallet_transactions_from_snapshot(snapshot); 244 let metrics = if keep_metrics { 245 Some(metrics_from_snapshot(snapshot)?) 246 } else { 247 None 248 }; 249 250 self.with_connection_mut(|connection| { 251 let transaction = connection 252 .transaction() 253 .context("failed to start UI data projection transaction")?; 254 match metrics { 255 Some(metrics) => replace_metrics(&transaction, &metrics)?, 256 None => clear_metrics_in_transaction(&transaction)?, 257 } 258 replace_ui_chain_index(&transaction, &ui_index, updated_at_ms)?; 259 replace_ui_utxos(&transaction, &utxos)?; 260 replace_ui_wallet_transactions(&transaction, &wallet_transactions)?; 261 transaction 262 .commit() 263 .context("failed to commit UI data projection transaction")?; 264 Ok(()) 265 }) 266 } 267 268 pub fn replace_metrics_for_snapshot(&self, snapshot: &ChainSnapshot) -> Result<()> { 269 let metrics = metrics_from_snapshot(snapshot)?; 270 self.with_connection_mut(|connection| { 271 let transaction = connection 272 .transaction() 273 .context("failed to start metrics transaction")?; 274 replace_metrics(&transaction, &metrics)?; 275 transaction 276 .commit() 277 .context("failed to commit metrics transaction")?; 278 Ok(()) 279 }) 280 } 281 282 pub fn clear_metrics(&self) -> Result<()> { 283 self.with_connection_mut(|connection| { 284 connection 285 .execute("DELETE FROM block_metrics", []) 286 .context("failed to delete block metrics")?; 287 Ok(()) 288 }) 289 } 290 291 pub fn clear_all(&self) -> Result<()> { 292 self.with_connection_mut(|connection| { 293 let transaction = connection 294 .transaction() 295 .context("failed to start UI data reset transaction")?; 296 clear_metrics_in_transaction(&transaction)?; 297 clear_ui_chain_index_in_transaction(&transaction)?; 298 transaction 299 .commit() 300 .context("failed to commit UI data reset transaction")?; 301 Ok(()) 302 }) 303 } 304 305 pub fn load_metrics(&self) -> Result<Vec<BlockMetricRow>> { 306 self.with_connection(|connection| { 307 let mut statement = connection 308 .prepare( 309 r#" 310 SELECT height, block_hash, timestamp_ms, block_time_ms, mine_difficulty_bits, 311 circulating_supply, known_wallet_addresses, transaction_count, transfer_count, burn_count, 312 mine_count, burned_amount, total_burned_amount, fees_amount, reward_amount, 313 vdf_rounds, finalizer_rank 314 FROM block_metrics 315 ORDER BY height ASC 316 "#, 317 ) 318 .context("failed to prepare block metrics query")?; 319 let rows = statement 320 .query_map([], |row| { 321 Ok(BlockMetricRow { 322 height: row.get(0)?, 323 block_hash: row.get(1)?, 324 timestamp_ms: row.get(2)?, 325 block_time_ms: row.get(3)?, 326 mine_difficulty_bits: row.get(4)?, 327 circulating_supply: row.get(5)?, 328 known_wallet_addresses: row.get(6)?, 329 transaction_count: row.get(7)?, 330 transfer_count: row.get(8)?, 331 burn_count: row.get(9)?, 332 mine_count: row.get(10)?, 333 burned_amount: row.get(11)?, 334 total_burned_amount: row.get(12)?, 335 fees_amount: row.get(13)?, 336 reward_amount: row.get(14)?, 337 vdf_rounds: row.get(15)?, 338 finalizer_rank: row.get(16)?, 339 }) 340 }) 341 .context("failed to load block metrics")?; 342 rows.collect::<std::result::Result<Vec<_>, _>>() 343 .context("failed to read block metrics rows") 344 }) 345 } 346 347 pub fn load_recent_metrics(&self, limit: usize) -> Result<Vec<BlockMetricRow>> { 348 self.with_connection(|connection| { 349 let mut statement = connection 350 .prepare( 351 r#" 352 SELECT height, block_hash, timestamp_ms, block_time_ms, mine_difficulty_bits, 353 circulating_supply, known_wallet_addresses, transaction_count, transfer_count, burn_count, 354 mine_count, burned_amount, total_burned_amount, fees_amount, reward_amount, 355 vdf_rounds, finalizer_rank 356 FROM block_metrics 357 ORDER BY height DESC 358 LIMIT ?1 359 "#, 360 ) 361 .context("failed to prepare recent block metrics query")?; 362 let rows = statement 363 .query_map([limit as u64], |row| { 364 Ok(BlockMetricRow { 365 height: row.get(0)?, 366 block_hash: row.get(1)?, 367 timestamp_ms: row.get(2)?, 368 block_time_ms: row.get(3)?, 369 mine_difficulty_bits: row.get(4)?, 370 circulating_supply: row.get(5)?, 371 known_wallet_addresses: row.get(6)?, 372 transaction_count: row.get(7)?, 373 transfer_count: row.get(8)?, 374 burn_count: row.get(9)?, 375 mine_count: row.get(10)?, 376 burned_amount: row.get(11)?, 377 total_burned_amount: row.get(12)?, 378 fees_amount: row.get(13)?, 379 reward_amount: row.get(14)?, 380 vdf_rounds: row.get(15)?, 381 finalizer_rank: row.get(16)?, 382 }) 383 }) 384 .context("failed to load recent block metrics")?; 385 let mut rows = rows 386 .collect::<std::result::Result<Vec<_>, _>>() 387 .context("failed to read recent block metrics rows")?; 388 rows.reverse(); 389 Ok(rows) 390 }) 391 } 392 393 pub fn load_leaderboards(&self, limit: usize) -> Result<UiLeaderboards> { 394 self.with_connection(|connection| { 395 Ok(UiLeaderboards { 396 balances: load_balance_leaderboard(connection, limit)?, 397 miners: load_transaction_leaderboard(connection, "mine", limit)?, 398 burners: load_transaction_leaderboard(connection, "burn", limit)?, 399 }) 400 }) 401 } 402 403 pub fn load_wallet_utxos(&self, address: &str) -> Result<Vec<(OutPoint, TxOutput)>> { 404 self.with_connection(|connection| load_wallet_utxos(connection, address)) 405 } 406 407 pub fn load_outputs( 408 &self, 409 outpoints: &BTreeSet<OutPoint>, 410 ) -> Result<BTreeMap<OutPoint, TxOutput>> { 411 self.with_connection(|connection| load_outputs(connection, outpoints)) 412 } 413 414 pub fn load_wallet_transactions( 415 &self, 416 address: &str, 417 kinds: &[&str], 418 offset: usize, 419 limit: usize, 420 ) -> Result<(Vec<WalletTransactionProjection>, usize)> { 421 self.with_connection(|connection| { 422 load_wallet_transactions(connection, address, kinds, offset, limit) 423 }) 424 } 425 426 fn with_connection<T>(&self, work: impl FnOnce(&Connection) -> Result<T>) -> Result<T> { 427 let connection = self.open_connection()?; 428 connection 429 .execute_batch( 430 r#" 431 PRAGMA busy_timeout = 5000; 432 PRAGMA synchronous = NORMAL; 433 "#, 434 ) 435 .context("failed to configure UI data database connection")?; 436 work(&connection) 437 } 438 439 fn with_connection_mut<T>(&self, work: impl FnOnce(&mut Connection) -> Result<T>) -> Result<T> { 440 let mut connection = self.open_connection()?; 441 connection 442 .execute_batch( 443 r#" 444 PRAGMA journal_mode = WAL; 445 PRAGMA busy_timeout = 5000; 446 PRAGMA synchronous = NORMAL; 447 "#, 448 ) 449 .context("failed to configure UI data database connection")?; 450 work(&mut connection) 451 } 452 453 fn open_connection(&self) -> Result<Connection> { 454 Connection::open(&self.path) 455 .with_context(|| format!("failed to open UI data database {}", self.path.display())) 456 } 457 } 458 459 #[derive(Clone, Debug, Default, Eq, PartialEq)] 460 pub struct UiLeaderboards { 461 pub balances: Vec<UiLeaderboardEntry>, 462 pub miners: Vec<UiLeaderboardEntry>, 463 pub burners: Vec<UiLeaderboardEntry>, 464 } 465 466 fn load_balance_leaderboard( 467 connection: &Connection, 468 limit: usize, 469 ) -> Result<Vec<UiLeaderboardEntry>> { 470 let mut statement = connection 471 .prepare( 472 r#" 473 SELECT address, SUM(amount) AS total_amount, COUNT(*) AS output_count 474 FROM ui_utxos 475 GROUP BY address 476 HAVING total_amount > 0 477 ORDER BY total_amount DESC, address ASC 478 LIMIT ?1 479 "#, 480 ) 481 .context("failed to prepare balance leaderboard query")?; 482 let rows = statement 483 .query_map([limit as u64], |row| { 484 Ok(UiLeaderboardEntry { 485 address: row.get(0)?, 486 amount: row.get(1)?, 487 count: row.get(2)?, 488 }) 489 }) 490 .context("failed to load balance leaderboard")?; 491 rows.collect::<std::result::Result<Vec<_>, _>>() 492 .context("failed to read balance leaderboard rows") 493 } 494 495 fn load_transaction_leaderboard( 496 connection: &Connection, 497 kind: &str, 498 limit: usize, 499 ) -> Result<Vec<UiLeaderboardEntry>> { 500 let mut statement = connection 501 .prepare( 502 r#" 503 SELECT address, transaction_json 504 FROM ui_wallet_transactions 505 WHERE kind = ?1 506 "#, 507 ) 508 .with_context(|| format!("failed to prepare {kind} leaderboard query"))?; 509 let rows = statement 510 .query_map([kind], |row| { 511 Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?)) 512 }) 513 .with_context(|| format!("failed to load {kind} leaderboard"))?; 514 let mut entries = BTreeMap::<String, UiLeaderboardEntry>::new(); 515 for row in rows { 516 let (address, transaction_json) = 517 row.with_context(|| format!("failed to read {kind} leaderboard row"))?; 518 let transaction = 519 serde_json::from_slice::<Transaction>(&transaction_json).with_context(|| { 520 format!( 521 "failed to parse {kind} leaderboard transaction JSON with {} bytes", 522 transaction_json.len() 523 ) 524 })?; 525 let amount = match transaction { 526 Transaction::Mine { .. } => MINE_REWARD, 527 Transaction::Burn { amount, .. } => amount, 528 Transaction::Transfer { .. } => 0, 529 }; 530 let entry = entries 531 .entry(address.clone()) 532 .or_insert_with(|| UiLeaderboardEntry { 533 address, 534 amount: 0, 535 count: 0, 536 }); 537 entry.amount = entry 538 .amount 539 .checked_add(amount) 540 .with_context(|| format!("{kind} leaderboard amount overflow"))?; 541 entry.count = entry 542 .count 543 .checked_add(1) 544 .with_context(|| format!("{kind} leaderboard count overflow"))?; 545 } 546 let mut entries = entries.into_values().collect::<Vec<_>>(); 547 entries.sort_by(|left, right| { 548 right 549 .amount 550 .cmp(&left.amount) 551 .then_with(|| right.count.cmp(&left.count)) 552 .then_with(|| left.address.cmp(&right.address)) 553 }); 554 entries.truncate(limit); 555 Ok(entries) 556 } 557 558 fn replace_metrics( 559 transaction: &rusqlite::Transaction<'_>, 560 metrics: &[BlockMetricRow], 561 ) -> Result<()> { 562 clear_metrics_in_transaction(transaction)?; 563 for metric in metrics { 564 transaction 565 .execute( 566 r#" 567 INSERT INTO block_metrics ( 568 height, block_hash, timestamp_ms, block_time_ms, mine_difficulty_bits, 569 circulating_supply, known_wallet_addresses, transaction_count, transfer_count, burn_count, 570 mine_count, burned_amount, total_burned_amount, fees_amount, reward_amount, vdf_rounds, 571 finalizer_rank 572 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17) 573 "#, 574 params![ 575 metric.height, 576 metric.block_hash, 577 metric.timestamp_ms, 578 metric.block_time_ms, 579 metric.mine_difficulty_bits, 580 metric.circulating_supply, 581 metric.known_wallet_addresses, 582 metric.transaction_count, 583 metric.transfer_count, 584 metric.burn_count, 585 metric.mine_count, 586 metric.burned_amount, 587 metric.total_burned_amount, 588 metric.fees_amount, 589 metric.reward_amount, 590 metric.vdf_rounds, 591 metric.finalizer_rank, 592 ], 593 ) 594 .with_context(|| format!("failed to insert metrics for block {}", metric.height))?; 595 } 596 Ok(()) 597 } 598 599 fn ensure_block_metrics_column( 600 connection: &Connection, 601 name: &str, 602 definition: &str, 603 ) -> Result<()> { 604 let mut statement = connection 605 .prepare("PRAGMA table_info(block_metrics)") 606 .context("failed to inspect block_metrics schema")?; 607 let columns = statement 608 .query_map([], |row| row.get::<_, String>(1)) 609 .context("failed to query block_metrics columns")? 610 .collect::<std::result::Result<Vec<_>, _>>() 611 .context("failed to read block_metrics columns")?; 612 if columns.iter().any(|column| column == name) { 613 return Ok(()); 614 } 615 connection 616 .execute( 617 &format!("ALTER TABLE block_metrics ADD COLUMN {name} {definition}"), 618 [], 619 ) 620 .with_context(|| format!("failed to add block_metrics.{name} column"))?; 621 Ok(()) 622 } 623 624 fn clear_metrics_in_transaction(transaction: &rusqlite::Transaction<'_>) -> Result<()> { 625 transaction 626 .execute("DELETE FROM block_metrics", []) 627 .context("failed to clear old block metrics")?; 628 Ok(()) 629 } 630 631 fn replace_ui_chain_index( 632 transaction: &rusqlite::Transaction<'_>, 633 index: &UiChainIndex, 634 updated_at_ms: u64, 635 ) -> Result<()> { 636 clear_ui_chain_index_in_transaction(transaction)?; 637 let Some(tip_hash) = &index.tip_hash else { 638 return Ok(()); 639 }; 640 transaction 641 .execute( 642 r#" 643 INSERT INTO ui_cache_meta (id, schema_version, tip_hash, updated_at_ms) 644 VALUES (1, ?1, ?2, ?3) 645 "#, 646 params![UI_CACHE_SCHEMA_VERSION, tip_hash, updated_at_ms], 647 ) 648 .context("failed to persist UI chain index metadata")?; 649 for (outpoint, output) in &index.outputs { 650 transaction 651 .execute( 652 r#" 653 INSERT INTO ui_output_index (txid, output_index, address, amount) 654 VALUES (?1, ?2, ?3, ?4) 655 "#, 656 params![outpoint.txid, outpoint.index, output.address, output.amount], 657 ) 658 .with_context(|| { 659 format!( 660 "failed to persist UI output index row {}:{}", 661 outpoint.txid, outpoint.index 662 ) 663 })?; 664 } 665 for (height, revealed_transactions) in &index.revealed_by_height { 666 for revealed in revealed_transactions { 667 let transaction_json = serde_json::to_vec(&revealed.transaction) 668 .context("failed to serialize UI revealed transaction")?; 669 transaction 670 .execute( 671 r#" 672 INSERT INTO ui_revealed_transactions (height, commitment, included_by, transaction_json) 673 VALUES (?1, ?2, ?3, ?4) 674 "#, 675 params![ 676 height, 677 revealed.commitment, 678 revealed.included_by, 679 transaction_json 680 ], 681 ) 682 .with_context(|| { 683 format!( 684 "failed to persist UI revealed transaction {}", 685 revealed.commitment 686 ) 687 })?; 688 } 689 } 690 for (block_hash, ranks) in &index.burn_leader_ranks_by_hash { 691 transaction 692 .execute( 693 "INSERT INTO ui_burn_leader_rank_blocks (block_hash) VALUES (?1)", 694 params![block_hash], 695 ) 696 .with_context(|| format!("failed to persist UI burn leader rank block {block_hash}"))?; 697 for rank in ranks { 698 transaction 699 .execute( 700 r#" 701 INSERT INTO ui_burn_leader_ranks ( 702 block_hash, rank, ticket_id, owner, amount, eligible_from_height, eligible_until_height 703 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) 704 "#, 705 params![ 706 block_hash, 707 rank.rank, 708 rank.ticket_id, 709 rank.owner, 710 rank.amount, 711 rank.eligible_from_height, 712 rank.eligible_until_height, 713 ], 714 ) 715 .with_context(|| { 716 format!( 717 "failed to persist UI burn leader rank {} for block {}", 718 rank.rank, block_hash 719 ) 720 })?; 721 } 722 } 723 Ok(()) 724 } 725 726 fn replace_ui_utxos( 727 transaction: &rusqlite::Transaction<'_>, 728 utxos: &[(OutPoint, TxOutput)], 729 ) -> Result<()> { 730 transaction 731 .execute("DELETE FROM ui_utxos", []) 732 .context("failed to clear old UI UTXO index")?; 733 for (outpoint, output) in utxos { 734 transaction 735 .execute( 736 r#" 737 INSERT INTO ui_utxos (txid, output_index, address, amount) 738 VALUES (?1, ?2, ?3, ?4) 739 "#, 740 params![outpoint.txid, outpoint.index, output.address, output.amount], 741 ) 742 .with_context(|| { 743 format!( 744 "failed to persist UI UTXO row {}:{}", 745 outpoint.txid, outpoint.index 746 ) 747 })?; 748 } 749 Ok(()) 750 } 751 752 fn replace_ui_wallet_transactions( 753 transaction: &rusqlite::Transaction<'_>, 754 rows: &[(String, WalletTransactionProjection)], 755 ) -> Result<()> { 756 transaction 757 .execute("DELETE FROM ui_wallet_transactions", []) 758 .context("failed to clear old UI wallet transaction index")?; 759 for (address, row) in rows { 760 let transaction_json = serde_json::to_vec(&row.transaction) 761 .context("failed to serialize UI wallet transaction")?; 762 transaction 763 .execute( 764 r#" 765 INSERT INTO ui_wallet_transactions ( 766 address, sort_key, kind, signature, block_height, timestamp_ms, block_finalizer, blinded, 767 transaction_json 768 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) 769 "#, 770 params![ 771 address, 772 row.sort_key, 773 row.kind, 774 row.transaction.signature(), 775 row.block_height, 776 row.timestamp_ms, 777 row.block_finalizer, 778 row.blinded, 779 transaction_json, 780 ], 781 ) 782 .with_context(|| { 783 format!( 784 "failed to persist UI wallet transaction {} for {}", 785 row.transaction.signature(), 786 address 787 ) 788 })?; 789 } 790 Ok(()) 791 } 792 793 fn clear_ui_chain_index_in_transaction(transaction: &rusqlite::Transaction<'_>) -> Result<()> { 794 transaction 795 .execute("DELETE FROM ui_cache_meta", []) 796 .context("failed to clear old UI cache metadata")?; 797 transaction 798 .execute("DELETE FROM ui_output_index", []) 799 .context("failed to clear old UI output index")?; 800 transaction 801 .execute("DELETE FROM ui_utxos", []) 802 .context("failed to clear old UI UTXO index")?; 803 transaction 804 .execute("DELETE FROM ui_wallet_transactions", []) 805 .context("failed to clear old UI wallet transaction index")?; 806 transaction 807 .execute("DELETE FROM ui_revealed_transactions", []) 808 .context("failed to clear old UI revealed transaction index")?; 809 transaction 810 .execute("DELETE FROM ui_burn_leader_ranks", []) 811 .context("failed to clear old UI burn leader rank index")?; 812 transaction 813 .execute("DELETE FROM ui_burn_leader_rank_blocks", []) 814 .context("failed to clear old UI burn leader rank block index")?; 815 Ok(()) 816 } 817 818 fn load_ui_output_index(connection: &Connection) -> Result<BTreeMap<OutPoint, TxOutput>> { 819 let mut statement = connection 820 .prepare( 821 r#" 822 SELECT txid, output_index, address, amount 823 FROM ui_output_index 824 ORDER BY txid, output_index 825 "#, 826 ) 827 .context("failed to prepare UI output index query")?; 828 let rows = statement 829 .query_map([], |row| { 830 Ok(( 831 OutPoint { 832 txid: row.get(0)?, 833 index: row.get(1)?, 834 }, 835 TxOutput { 836 address: row.get(2)?, 837 amount: row.get(3)?, 838 }, 839 )) 840 }) 841 .context("failed to load UI output index")?; 842 rows.collect::<std::result::Result<BTreeMap<_, _>, _>>() 843 .context("failed to read UI output index rows") 844 } 845 846 fn load_outputs( 847 connection: &Connection, 848 outpoints: &BTreeSet<OutPoint>, 849 ) -> Result<BTreeMap<OutPoint, TxOutput>> { 850 if outpoints.is_empty() { 851 return Ok(BTreeMap::new()); 852 } 853 let mut outputs = BTreeMap::new(); 854 let mut statement = connection 855 .prepare( 856 r#" 857 SELECT address, amount 858 FROM ui_output_index 859 WHERE txid = ?1 AND output_index = ?2 860 "#, 861 ) 862 .context("failed to prepare narrow UI output lookup")?; 863 for outpoint in outpoints { 864 let output = statement 865 .query_row(params![outpoint.txid, outpoint.index], |row| { 866 Ok(TxOutput { 867 address: row.get(0)?, 868 amount: row.get(1)?, 869 }) 870 }) 871 .optional() 872 .with_context(|| { 873 format!( 874 "failed to load UI output {}:{}", 875 outpoint.txid, outpoint.index 876 ) 877 })?; 878 if let Some(output) = output { 879 outputs.insert(outpoint.clone(), output); 880 } 881 } 882 Ok(outputs) 883 } 884 885 fn load_wallet_utxos(connection: &Connection, address: &str) -> Result<Vec<(OutPoint, TxOutput)>> { 886 let mut statement = connection 887 .prepare( 888 r#" 889 SELECT txid, output_index, address, amount 890 FROM ui_utxos 891 WHERE address = ?1 892 ORDER BY amount DESC, txid ASC, output_index ASC 893 "#, 894 ) 895 .context("failed to prepare UI wallet UTXO query")?; 896 let rows = statement 897 .query_map([address], |row| { 898 Ok(( 899 OutPoint { 900 txid: row.get(0)?, 901 index: row.get(1)?, 902 }, 903 TxOutput { 904 address: row.get(2)?, 905 amount: row.get(3)?, 906 }, 907 )) 908 }) 909 .context("failed to load UI wallet UTXOs")?; 910 rows.collect::<std::result::Result<Vec<_>, _>>() 911 .context("failed to read UI wallet UTXO rows") 912 } 913 914 fn load_wallet_transactions( 915 connection: &Connection, 916 address: &str, 917 kinds: &[&str], 918 offset: usize, 919 limit: usize, 920 ) -> Result<(Vec<WalletTransactionProjection>, usize)> { 921 if kinds.is_empty() { 922 return Ok((Vec::new(), 0)); 923 } 924 if wallet_transaction_kinds_cover_all(kinds) { 925 let total = connection 926 .query_row( 927 "SELECT COUNT(*) FROM ui_wallet_transactions WHERE address = ?1", 928 params![address], 929 |row| row.get::<_, u64>(0), 930 ) 931 .context("failed to count UI wallet transactions")? as usize; 932 let mut statement = connection 933 .prepare( 934 r#" 935 SELECT sort_key, kind, block_height, timestamp_ms, block_finalizer, blinded, transaction_json 936 FROM ui_wallet_transactions 937 WHERE address = ?1 938 ORDER BY sort_key DESC 939 LIMIT ?2 OFFSET ?3 940 "#, 941 ) 942 .context("failed to prepare UI wallet transactions query")?; 943 let rows = read_wallet_transaction_rows( 944 statement.query_map(params![address, limit as i64, offset as i64], |row| { 945 wallet_transaction_projection_from_row(row) 946 })?, 947 )?; 948 return Ok((rows, total)); 949 } 950 let placeholders = std::iter::repeat_n("?", kinds.len()) 951 .collect::<Vec<_>>() 952 .join(", "); 953 let count_sql = format!( 954 "SELECT COUNT(*) FROM ui_wallet_transactions WHERE address = ? AND kind IN ({placeholders})" 955 ); 956 let mut count_params = Vec::<Value>::with_capacity(kinds.len() + 1); 957 count_params.push(Value::Text(address.to_string())); 958 for kind in kinds { 959 count_params.push(Value::Text((*kind).to_string())); 960 } 961 let total = connection 962 .query_row(&count_sql, params_from_iter(count_params.iter()), |row| { 963 row.get::<_, u64>(0) 964 }) 965 .context("failed to count UI wallet transactions")? as usize; 966 967 let query_sql = format!( 968 r#" 969 SELECT sort_key, kind, block_height, timestamp_ms, block_finalizer, blinded, transaction_json 970 FROM ui_wallet_transactions 971 WHERE address = ? AND kind IN ({placeholders}) 972 ORDER BY sort_key DESC 973 LIMIT ? OFFSET ? 974 "# 975 ); 976 let mut query_params = Vec::<Value>::with_capacity(kinds.len() + 3); 977 query_params.push(Value::Text(address.to_string())); 978 for kind in kinds { 979 query_params.push(Value::Text((*kind).to_string())); 980 } 981 query_params.push(Value::Integer(limit as i64)); 982 query_params.push(Value::Integer(offset as i64)); 983 let mut statement = connection 984 .prepare(&query_sql) 985 .context("failed to prepare UI wallet transactions query")?; 986 let rows = read_wallet_transaction_rows( 987 statement 988 .query_map(params_from_iter(query_params.iter()), |row| { 989 wallet_transaction_projection_from_row(row) 990 }) 991 .context("failed to load UI wallet transactions")?, 992 )?; 993 Ok((rows, total)) 994 } 995 996 fn wallet_transaction_kinds_cover_all(kinds: &[&str]) -> bool { 997 ["transfer", "mine", "burn"] 998 .into_iter() 999 .all(|kind| kinds.contains(&kind)) 1000 } 1001 1002 fn wallet_transaction_projection_from_row( 1003 row: &rusqlite::Row<'_>, 1004 ) -> rusqlite::Result<WalletTransactionProjection> { 1005 let transaction_json = row.get::<_, Vec<u8>>(6)?; 1006 let transaction = 1007 serde_json::from_slice::<Transaction>(&transaction_json).map_err(|error| { 1008 rusqlite::Error::FromSqlConversionFailure( 1009 transaction_json.len(), 1010 rusqlite::types::Type::Blob, 1011 Box::new(error), 1012 ) 1013 })?; 1014 Ok(WalletTransactionProjection { 1015 sort_key: row.get(0)?, 1016 kind: row.get(1)?, 1017 block_height: row.get(2)?, 1018 timestamp_ms: row.get(3)?, 1019 block_finalizer: row.get(4)?, 1020 blinded: row.get::<_, u64>(5)? != 0, 1021 transaction, 1022 }) 1023 } 1024 1025 fn read_wallet_transaction_rows( 1026 rows: rusqlite::MappedRows< 1027 '_, 1028 impl FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<WalletTransactionProjection>, 1029 >, 1030 ) -> Result<Vec<WalletTransactionProjection>> { 1031 rows.collect::<std::result::Result<Vec<_>, _>>() 1032 .context("failed to read UI wallet transaction rows") 1033 } 1034 1035 fn load_ui_revealed_transactions( 1036 connection: &Connection, 1037 ) -> Result<BTreeMap<u64, Vec<RevealedBlindedTransaction>>> { 1038 let mut statement = connection 1039 .prepare( 1040 r#" 1041 SELECT height, commitment, included_by, transaction_json 1042 FROM ui_revealed_transactions 1043 ORDER BY height, commitment 1044 "#, 1045 ) 1046 .context("failed to prepare UI revealed transaction query")?; 1047 let rows = statement 1048 .query_map([], |row| { 1049 let transaction_json = row.get::<_, Vec<u8>>(3)?; 1050 let transaction = 1051 serde_json::from_slice::<Transaction>(&transaction_json).map_err(|error| { 1052 rusqlite::Error::FromSqlConversionFailure( 1053 transaction_json.len(), 1054 rusqlite::types::Type::Blob, 1055 Box::new(error), 1056 ) 1057 })?; 1058 Ok(RevealedBlindedTransaction { 1059 height: row.get(0)?, 1060 commitment: row.get(1)?, 1061 included_by: row.get(2)?, 1062 transaction, 1063 }) 1064 }) 1065 .context("failed to load UI revealed transactions")?; 1066 let mut by_height = BTreeMap::<u64, Vec<RevealedBlindedTransaction>>::new(); 1067 for revealed in rows { 1068 let revealed = revealed.context("failed to read UI revealed transaction row")?; 1069 by_height.entry(revealed.height).or_default().push(revealed); 1070 } 1071 Ok(by_height) 1072 } 1073 1074 fn load_ui_burn_leader_ranks( 1075 connection: &Connection, 1076 ) -> Result<BTreeMap<String, Vec<BurnLeaderRank>>> { 1077 let mut blocks_statement = connection 1078 .prepare("SELECT block_hash FROM ui_burn_leader_rank_blocks ORDER BY block_hash") 1079 .context("failed to prepare UI burn leader rank block query")?; 1080 let blocks = blocks_statement 1081 .query_map([], |row| row.get::<_, String>(0)) 1082 .context("failed to load UI burn leader rank blocks")?; 1083 let mut by_block_hash = BTreeMap::<String, Vec<BurnLeaderRank>>::new(); 1084 for block_hash in blocks { 1085 by_block_hash.insert( 1086 block_hash.context("failed to read UI burn leader rank block row")?, 1087 Vec::new(), 1088 ); 1089 } 1090 1091 let mut statement = connection 1092 .prepare( 1093 r#" 1094 SELECT block_hash, rank, ticket_id, owner, amount, eligible_from_height, eligible_until_height 1095 FROM ui_burn_leader_ranks 1096 ORDER BY block_hash, rank 1097 "#, 1098 ) 1099 .context("failed to prepare UI burn leader rank query")?; 1100 let rows = statement 1101 .query_map([], |row| { 1102 Ok(( 1103 row.get::<_, String>(0)?, 1104 BurnLeaderRank { 1105 rank: row.get(1)?, 1106 ticket_id: row.get(2)?, 1107 owner: row.get(3)?, 1108 amount: row.get(4)?, 1109 eligible_from_height: row.get(5)?, 1110 eligible_until_height: row.get(6)?, 1111 }, 1112 )) 1113 }) 1114 .context("failed to load UI burn leader ranks")?; 1115 for row in rows { 1116 let (block_hash, rank) = row.context("failed to read UI burn leader rank row")?; 1117 by_block_hash.entry(block_hash).or_default().push(rank); 1118 } 1119 Ok(by_block_hash) 1120 } 1121 1122 fn wallet_transactions_from_snapshot( 1123 snapshot: &ChainSnapshot, 1124 ) -> Vec<(String, WalletTransactionProjection)> { 1125 let revealed_by_height = revealed_blinded_transactions(snapshot) 1126 .unwrap_or_default() 1127 .into_iter() 1128 .fold( 1129 BTreeMap::<u64, Vec<RevealedBlindedTransaction>>::new(), 1130 |mut by_height, revealed| { 1131 by_height.entry(revealed.height).or_default().push(revealed); 1132 by_height 1133 }, 1134 ); 1135 let mut rows = Vec::new(); 1136 for block in &snapshot.blocks { 1137 for (index, transaction) in block.transactions.iter().rev().enumerate() { 1138 push_wallet_transaction_projection( 1139 &mut rows, 1140 transaction, 1141 block, 1142 block.height as u128 * 10_000 + index as u128, 1143 false, 1144 ); 1145 } 1146 if let Some(revealed) = revealed_by_height.get(&block.height) { 1147 for (index, revealed) in revealed.iter().rev().enumerate() { 1148 push_wallet_transaction_projection( 1149 &mut rows, 1150 &revealed.transaction, 1151 block, 1152 block.height as u128 * 10_000 + 5_000 + index as u128, 1153 false, 1154 ); 1155 } 1156 } 1157 } 1158 rows 1159 } 1160 1161 fn push_wallet_transaction_projection( 1162 rows: &mut Vec<(String, WalletTransactionProjection)>, 1163 transaction: &Transaction, 1164 block: &Block, 1165 sort_key: u128, 1166 blinded: bool, 1167 ) { 1168 let kind = transaction_kind(transaction).to_string(); 1169 let projection = WalletTransactionProjection { 1170 sort_key: sort_key.min(u128::from(u64::MAX)) as u64, 1171 kind, 1172 block_height: block.height, 1173 timestamp_ms: block.timestamp_ms, 1174 block_finalizer: block.miner.clone(), 1175 blinded, 1176 transaction: transaction.clone(), 1177 }; 1178 for address in wallet_transaction_addresses(transaction) { 1179 rows.push((address, projection.clone())); 1180 } 1181 } 1182 1183 fn wallet_transaction_addresses(transaction: &Transaction) -> Vec<String> { 1184 match transaction { 1185 Transaction::Transfer { .. } => { 1186 let mut addresses = vec![transaction.sender().to_string()]; 1187 if let Some(to) = transaction.to() { 1188 if to != transaction.sender() { 1189 addresses.push(to.to_string()); 1190 } 1191 } 1192 addresses 1193 } 1194 Transaction::Burn { .. } => vec![transaction.sender().to_string()], 1195 Transaction::Mine { recipient, .. } => vec![recipient.clone()], 1196 } 1197 } 1198 1199 fn transaction_kind(transaction: &Transaction) -> &'static str { 1200 match transaction { 1201 Transaction::Transfer { .. } => "transfer", 1202 Transaction::Burn { .. } => "burn", 1203 Transaction::Mine { .. } => "mine", 1204 } 1205 } 1206 1207 fn metrics_from_snapshot(snapshot: &ChainSnapshot) -> Result<Vec<BlockMetricRow>> { 1208 let ledger = Ledger::from_persisted_snapshot(snapshot.clone()) 1209 .context("failed to rebuild ledger for metrics")?; 1210 let genesis = snapshot 1211 .blocks 1212 .first() 1213 .cloned() 1214 .context("cannot compute metrics for empty chain snapshot")?; 1215 let mut running_ledger = Ledger::from_persisted_snapshot(ChainSnapshot { 1216 genesis_allocations: snapshot.genesis_allocations.clone(), 1217 vdf_rounds: snapshot.vdf_rounds, 1218 launch_profile: snapshot.launch_profile.clone(), 1219 blocks: vec![genesis], 1220 }) 1221 .context("failed to rebuild genesis ledger for metrics")?; 1222 let revealed = revealed_blinded_transactions(snapshot)?.into_iter().fold( 1223 BTreeMap::<u64, Vec<crate::domain::RevealedBlindedTransaction>>::new(), 1224 |mut by_height, revealed| { 1225 by_height.entry(revealed.height).or_default().push(revealed); 1226 by_height 1227 }, 1228 ); 1229 let mut known_wallet_addresses = snapshot 1230 .genesis_allocations 1231 .keys() 1232 .cloned() 1233 .collect::<BTreeSet<_>>(); 1234 let mut total_burned_amount = 0_u64; 1235 let mut rows = Vec::with_capacity(snapshot.blocks.len()); 1236 let mut previous_timestamp_ms = None; 1237 let mut active_blinded = BTreeMap::<String, BlindedTransaction>::new(); 1238 let mut metric_utxos = metric_genesis_utxos(snapshot); 1239 let mut metric_locked_blinded_inputs = BTreeMap::<String, Amount>::new(); 1240 let reveal_bundle_slots_by_height = ledger 1241 .burn_leader_ranks_for_blocks(snapshot.blocks.iter().map(|block| block.height)) 1242 .map(|ranks_by_height| { 1243 ranks_by_height 1244 .into_iter() 1245 .map(|(height, ranks)| { 1246 ( 1247 height, 1248 reveal_committee_slot_count_for_height( 1249 height, 1250 ranks.len(), 1251 ranks.iter().map(|rank| rank.owner.as_str()), 1252 ), 1253 ) 1254 }) 1255 .collect::<BTreeMap<_, _>>() 1256 }) 1257 .unwrap_or_default(); 1258 1259 for block in &snapshot.blocks { 1260 let revealed_transactions = revealed.get(&block.height).cloned().unwrap_or_default(); 1261 let mut transfer_count = 0_u64; 1262 let mut burn_count = 0_u64; 1263 let mut mine_count = 0_u64; 1264 let mut burned_amount = 0_u64; 1265 let mut burned_fee_amount = 0_u64; 1266 let mut fees_amount = 0_u64; 1267 1268 known_wallet_addresses.insert(block.miner.clone()); 1269 for signature in &block.reveal_bundle_section.signatures { 1270 known_wallet_addresses.insert(signature.member.clone()); 1271 } 1272 for transaction in &block.transactions { 1273 collect_transaction_addresses(transaction, &mut known_wallet_addresses); 1274 metric_apply_public_transaction(transaction, &mut metric_utxos)?; 1275 fees_amount = fees_amount 1276 .checked_add(transaction.fee()) 1277 .context("block metric fees overflow")?; 1278 match transaction { 1279 Transaction::Transfer { .. } => transfer_count += 1, 1280 Transaction::Burn { amount, .. } => { 1281 burn_count += 1; 1282 burned_amount = burned_amount 1283 .checked_add(*amount) 1284 .context("block metric burns overflow")?; 1285 } 1286 Transaction::Mine { .. } => { 1287 mine_count += 1; 1288 } 1289 } 1290 } 1291 for revealed in &revealed_transactions { 1292 let transaction = &revealed.transaction; 1293 known_wallet_addresses.insert(revealed.included_by.clone()); 1294 collect_transaction_addresses(transaction, &mut known_wallet_addresses); 1295 metric_index_transaction_outputs(&mut metric_utxos, transaction); 1296 metric_index_blinded_fee_outputs( 1297 &mut metric_utxos, 1298 &revealed.commitment, 1299 &revealed.included_by, 1300 block, 1301 transaction.fee(), 1302 ); 1303 fees_amount = fees_amount 1304 .checked_add(transaction.fee()) 1305 .context("block metric fees overflow")?; 1306 let committer_fee = blinded_fee_share(transaction.fee(), BLINDED_COMMITTER_FEE_BPS); 1307 let included_reveal_bundle_count = block.included_reveal_bundle_count(); 1308 let available_reveal_bundle_slots = reveal_bundle_slots_by_height 1309 .get(&block.height) 1310 .copied() 1311 .unwrap_or(REVEAL_COMMITTEE_SIZE); 1312 let reveal_finalizer_fee = blinded_reveal_finalizer_fee( 1313 transaction.fee(), 1314 included_reveal_bundle_count, 1315 available_reveal_bundle_slots, 1316 ); 1317 let reveal_bundle_signer_fees = 1318 blinded_fee_share(transaction.fee(), BLINDED_REVEAL_BUNDLE_SIGNER_FEE_BPS) 1319 .saturating_mul(included_reveal_bundle_count as u64); 1320 let distributed_fee = committer_fee 1321 .saturating_add(reveal_finalizer_fee) 1322 .saturating_add(reveal_bundle_signer_fees); 1323 burned_fee_amount = burned_fee_amount 1324 .checked_add(transaction.fee().saturating_sub(distributed_fee)) 1325 .context("block metric burned fees overflow")?; 1326 match transaction { 1327 Transaction::Transfer { .. } => transfer_count += 1, 1328 Transaction::Burn { amount, .. } => { 1329 burn_count += 1; 1330 burned_amount = burned_amount 1331 .checked_add(*amount) 1332 .context("block metric burns overflow")?; 1333 } 1334 Transaction::Mine { .. } => { 1335 mine_count += 1; 1336 } 1337 } 1338 } 1339 let revealed_commitments = block 1340 .all_blinded_reveals() 1341 .into_iter() 1342 .map(|reveal| reveal.commitment.clone()) 1343 .collect::<std::collections::BTreeSet<_>>(); 1344 let mut expired_blinded_fee_values = Vec::new(); 1345 active_blinded.retain(|commitment, transaction| { 1346 if revealed_commitments.contains(commitment) { 1347 metric_locked_blinded_inputs.remove(commitment); 1348 return false; 1349 } 1350 if block.height >= transaction.expires_at_height { 1351 if !transaction.inputs.is_empty() { 1352 expired_blinded_fee_values.push(transaction.fee); 1353 metric_index_expired_blinded_change( 1354 &mut metric_utxos, 1355 commitment, 1356 transaction, 1357 metric_locked_blinded_inputs 1358 .remove(commitment) 1359 .unwrap_or_default(), 1360 ); 1361 } 1362 return false; 1363 } 1364 true 1365 }); 1366 let expired_blinded_fees = 1367 expired_blinded_fee_values 1368 .into_iter() 1369 .try_fold(0_u64, |total, fee| { 1370 total 1371 .checked_add(fee) 1372 .context("block metric expiry fees overflow") 1373 })?; 1374 fees_amount = fees_amount 1375 .checked_add(expired_blinded_fees) 1376 .context("block metric expiry fees overflow")?; 1377 burned_fee_amount = burned_fee_amount 1378 .checked_add(expired_blinded_fees) 1379 .context("block metric expired burned fees overflow")?; 1380 for transaction in &block.blinded_transactions { 1381 for input in &transaction.inputs { 1382 known_wallet_addresses.insert(input.owner.clone()); 1383 } 1384 let locked_total = metric_spend_blinded_inputs(transaction, &mut metric_utxos)?; 1385 metric_locked_blinded_inputs.insert(transaction.commitment.clone(), locked_total); 1386 active_blinded.insert(transaction.commitment.clone(), transaction.clone()); 1387 } 1388 total_burned_amount = total_burned_amount 1389 .checked_add(burned_amount) 1390 .and_then(|amount| amount.checked_add(burned_fee_amount)) 1391 .context("total burned metric overflows")?; 1392 1393 if block.height > 0 { 1394 running_ledger 1395 .apply_preverified_block_at(block.clone(), u64::MAX) 1396 .with_context(|| format!("failed to replay block {} for metrics", block.height))?; 1397 } 1398 metric_index_block_reward(&mut metric_utxos, block); 1399 let circulating_supply = ledger_circulating_supply(&running_ledger)? 1400 .checked_add(metric_locked_supply(&metric_locked_blinded_inputs)?) 1401 .context("circulating supply metric overflows")?; 1402 let block_time_ms = 1403 previous_timestamp_ms.map(|previous| block.timestamp_ms.saturating_sub(previous)); 1404 previous_timestamp_ms = Some(block.timestamp_ms); 1405 rows.push(BlockMetricRow { 1406 height: block.height, 1407 block_hash: block.hash.clone(), 1408 timestamp_ms: block.timestamp_ms, 1409 block_time_ms, 1410 mine_difficulty_bits: ledger.mine_difficulty_bits_at_height(block.height), 1411 circulating_supply, 1412 known_wallet_addresses: known_wallet_addresses.len() as u64, 1413 transaction_count: (block.transactions.len() + revealed_transactions.len()) as u64, 1414 transfer_count, 1415 burn_count, 1416 mine_count, 1417 burned_amount, 1418 total_burned_amount, 1419 fees_amount, 1420 reward_amount: block.reward, 1421 vdf_rounds: block.vdf_rounds, 1422 finalizer_rank: block.finalizer_rank, 1423 }); 1424 } 1425 Ok(rows) 1426 } 1427 1428 fn ledger_circulating_supply(ledger: &Ledger) -> Result<Amount> { 1429 ledger 1430 .status() 1431 .balances 1432 .values() 1433 .try_fold(0_u64, |total, amount| { 1434 total 1435 .checked_add(*amount) 1436 .context("circulating supply metric overflows") 1437 }) 1438 } 1439 1440 fn metric_locked_supply(locked: &BTreeMap<String, Amount>) -> Result<Amount> { 1441 locked.values().try_fold(0_u64, |total, amount| { 1442 total 1443 .checked_add(*amount) 1444 .context("circulating supply metric overflows") 1445 }) 1446 } 1447 1448 fn metric_genesis_utxos(snapshot: &ChainSnapshot) -> BTreeMap<OutPoint, TxOutput> { 1449 snapshot 1450 .genesis_allocations 1451 .iter() 1452 .filter(|(_, amount)| **amount > 0) 1453 .map(|(address, amount)| { 1454 ( 1455 metric_genesis_allocation_outpoint(address), 1456 TxOutput { 1457 address: address.clone(), 1458 amount: *amount, 1459 }, 1460 ) 1461 }) 1462 .collect() 1463 } 1464 1465 fn blinded_fee_share(fee: Amount, bps: u64) -> Amount { 1466 ((fee as u128 * bps as u128) / BLINDED_FEE_BPS_DENOMINATOR as u128) as Amount 1467 } 1468 1469 fn metric_apply_public_transaction( 1470 transaction: &Transaction, 1471 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1472 ) -> Result<()> { 1473 metric_spend_transaction_inputs(transaction, utxos)?; 1474 metric_index_transaction_outputs(utxos, transaction); 1475 Ok(()) 1476 } 1477 1478 fn metric_spend_transaction_inputs( 1479 transaction: &Transaction, 1480 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1481 ) -> Result<Amount> { 1482 let inputs = match transaction { 1483 Transaction::Transfer { inputs, .. } | Transaction::Burn { inputs, .. } => inputs, 1484 Transaction::Mine { .. } => return Ok(0), 1485 }; 1486 metric_spend_inputs(inputs, utxos) 1487 } 1488 1489 fn metric_spend_blinded_inputs( 1490 transaction: &BlindedTransaction, 1491 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1492 ) -> Result<Amount> { 1493 metric_spend_inputs(&transaction.inputs, utxos) 1494 } 1495 1496 fn metric_spend_inputs( 1497 inputs: &[TxInput], 1498 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1499 ) -> Result<Amount> { 1500 inputs.iter().try_fold(0_u64, |total, input| { 1501 let output = utxos.remove(&input.outpoint).with_context(|| { 1502 format!( 1503 "metric replay spends missing output {}:{}", 1504 input.outpoint.txid, input.outpoint.index 1505 ) 1506 })?; 1507 total 1508 .checked_add(output.amount) 1509 .context("metric replay input total overflows") 1510 }) 1511 } 1512 1513 fn metric_index_transaction_outputs( 1514 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1515 transaction: &Transaction, 1516 ) { 1517 let outputs = match transaction { 1518 Transaction::Transfer { outputs, .. } => outputs.clone(), 1519 Transaction::Burn { change, .. } => change.clone(), 1520 Transaction::Mine { recipient, .. } => vec![TxOutput { 1521 address: recipient.clone(), 1522 amount: MINE_REWARD, 1523 }], 1524 }; 1525 for (index, output) in outputs.iter().enumerate() { 1526 utxos.insert( 1527 OutPoint { 1528 txid: transaction.signature().to_string(), 1529 index: index as u32, 1530 }, 1531 output.clone(), 1532 ); 1533 } 1534 } 1535 1536 fn metric_index_blinded_fee_outputs( 1537 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1538 commitment: &str, 1539 included_by: &str, 1540 block: &Block, 1541 fee: Amount, 1542 ) { 1543 if fee == 0 { 1544 return; 1545 } 1546 let committer_fee = blinded_fee_share(fee, BLINDED_COMMITTER_FEE_BPS); 1547 if committer_fee > 0 { 1548 utxos.insert( 1549 metric_blinded_committer_fee_outpoint(commitment), 1550 TxOutput { 1551 address: included_by.to_string(), 1552 amount: committer_fee, 1553 }, 1554 ); 1555 } 1556 let reveal_bundle_signer_fee = blinded_fee_share(fee, BLINDED_REVEAL_BUNDLE_SIGNER_FEE_BPS); 1557 if reveal_bundle_signer_fee > 0 { 1558 for signature in &block.reveal_bundle_section.signatures { 1559 utxos.insert( 1560 metric_blinded_reveal_bundle_signer_fee_outpoint(commitment, signature.slot), 1561 TxOutput { 1562 address: signature.member.clone(), 1563 amount: reveal_bundle_signer_fee, 1564 }, 1565 ); 1566 } 1567 } 1568 } 1569 1570 fn metric_index_expired_blinded_change( 1571 utxos: &mut BTreeMap<OutPoint, TxOutput>, 1572 commitment: &str, 1573 transaction: &BlindedTransaction, 1574 locked_total: Amount, 1575 ) { 1576 let Some(first_input) = transaction.inputs.first() else { 1577 return; 1578 }; 1579 let change = locked_total.saturating_sub(transaction.fee); 1580 if change == 0 { 1581 return; 1582 } 1583 utxos.insert( 1584 metric_blinded_expiry_change_outpoint(commitment), 1585 TxOutput { 1586 address: first_input.owner.clone(), 1587 amount: change, 1588 }, 1589 ); 1590 } 1591 1592 fn metric_index_block_reward(utxos: &mut BTreeMap<OutPoint, TxOutput>, block: &Block) { 1593 if block.reward == 0 { 1594 return; 1595 } 1596 utxos.insert( 1597 metric_reward_outpoint(&block.hash), 1598 TxOutput { 1599 address: block.miner.clone(), 1600 amount: block.reward, 1601 }, 1602 ); 1603 } 1604 1605 fn metric_genesis_allocation_outpoint(address: &str) -> OutPoint { 1606 OutPoint { 1607 txid: hex_hash(format!("iuna-genesis-allocation:{address}")), 1608 index: 0, 1609 } 1610 } 1611 1612 fn metric_reward_outpoint(block_hash: &str) -> OutPoint { 1613 OutPoint { 1614 txid: block_hash.to_string(), 1615 index: u32::MAX, 1616 } 1617 } 1618 1619 fn metric_blinded_committer_fee_outpoint(commitment: &str) -> OutPoint { 1620 OutPoint { 1621 txid: commitment.to_string(), 1622 index: u32::MAX - 1, 1623 } 1624 } 1625 1626 fn metric_blinded_reveal_bundle_signer_fee_outpoint(commitment: &str, slot: u8) -> OutPoint { 1627 OutPoint { 1628 txid: commitment.to_string(), 1629 index: u32::MAX - 3 - u32::from(slot), 1630 } 1631 } 1632 1633 fn metric_blinded_expiry_change_outpoint(commitment: &str) -> OutPoint { 1634 OutPoint { 1635 txid: commitment.to_string(), 1636 index: 0, 1637 } 1638 } 1639 1640 fn collect_transaction_addresses(transaction: &Transaction, addresses: &mut BTreeSet<String>) { 1641 match transaction { 1642 Transaction::Transfer { 1643 inputs, outputs, .. 1644 } => { 1645 collect_input_addresses(inputs, addresses); 1646 collect_output_addresses(outputs, addresses); 1647 } 1648 Transaction::Burn { inputs, change, .. } => { 1649 collect_input_addresses(inputs, addresses); 1650 collect_output_addresses(change, addresses); 1651 } 1652 Transaction::Mine { recipient, .. } => { 1653 addresses.insert(recipient.clone()); 1654 } 1655 } 1656 } 1657 1658 fn collect_input_addresses(inputs: &[TxInput], addresses: &mut BTreeSet<String>) { 1659 for input in inputs { 1660 addresses.insert(input.owner.clone()); 1661 } 1662 } 1663 1664 fn collect_output_addresses(outputs: &[TxOutput], addresses: &mut BTreeSet<String>) { 1665 for output in outputs { 1666 addresses.insert(output.address.clone()); 1667 } 1668 } 1669 1670 fn unix_ms() -> u64 { 1671 SystemTime::now() 1672 .duration_since(UNIX_EPOCH) 1673 .unwrap_or_default() 1674 .as_millis() as u64 1675 } 1676 1677 #[cfg(test)] 1678 mod tests;