[schema] refactor slow wallet state

src/extract_snapshot.rs

@@ -21,6 +21,7 @@ use log::{error, info, warn};
 use crate::{
     scan::FrameworkVersion,
     schema_account_state::{WarehouseAccState, WarehouseTime},
+    util::COIN_DECIMAL_PRECISION,
 };
 
 // uses libra-compatibility to parse the v5 manifest files, and decode v5 format bytecode into current version data structures (v6+);
@@ -56,11 +57,14 @@ pub async fn extract_v5_snapshot(archive_path: &Path) -> Result<Vec<WarehouseAcc
                 }
 
                 if let Ok(b) = acc.get_resource::<BalanceResourceV5>() {
-                    s.balance = b.coin()
+                    s.balance = b.coin() as f64 / COIN_DECIMAL_PRECISION as f64;
                 }
                 if let Ok(sw) = acc.get_resource::<SlowWalletResourceV5>() {
-                    s.slow_wallet_unlocked = sw.unlocked;
-                    s.slow_wallet_transferred = sw.transferred;
+                    s.slow_wallet_acc = true;
+                    s.slow_wallet_unlocked =
+                        Some(sw.unlocked as f64 / COIN_DECIMAL_PRECISION as f64);
+                    s.slow_wallet_transferred =
+                        Some(sw.transferred as f64 / COIN_DECIMAL_PRECISION as f64);
                 }
 
                 if let Ok(tower) = acc.get_resource::<TowerStateResource>() {
@@ -113,12 +117,14 @@ pub async fn extract_current_snapshot(archive_path: &Path) -> Result<Vec<Warehou
             }
 
             if let Some(b) = el.get_resource::<LibraCoinStoreResource>()? {
-                s.balance = b.coin();
+                s.balance = b.coin() as f64 / COIN_DECIMAL_PRECISION as f64;
             }
 
             if let Some(sw) = el.get_resource::<SlowWalletResource>()? {
-                s.slow_wallet_unlocked = sw.unlocked;
-                s.slow_wallet_transferred = sw.transferred;
+                s.slow_wallet_acc = true;
+                s.slow_wallet_unlocked = Some(sw.unlocked as f64 / COIN_DECIMAL_PRECISION as f64);
+                s.slow_wallet_transferred =
+                    Some(sw.transferred as f64 / COIN_DECIMAL_PRECISION as f64);
             }
 
             // Infer if it is a donor voice account

src/json_rescue_v5_extract.rs

@@ -1,10 +1,8 @@
 use crate::{
     scan::FrameworkVersion,
-    schema_transaction::{
-        EntryFunctionArgs, RelationLabel, WarehouseEvent, WarehouseTxMaster,
-        COIN_DECIMAL_PRECISION, LEGACY_REBASE_MULTIPLIER,
-    },
+    schema_transaction::{EntryFunctionArgs, RelationLabel, WarehouseEvent, WarehouseTxMaster},
     unzip_temp::decompress_tar_archive,
+    util::{COIN_DECIMAL_PRECISION, LEGACY_REBASE_MULTIPLIER},
 };
 use chrono::DateTime;
 use diem_crypto::HashValue;

src/neo4j_init.rs

@@ -52,6 +52,8 @@ pub static INDEX_EXCHANGE_LINK_LEDGER: &str = "
 pub static INDEX_LIFETIME: &str = "
     CREATE INDEX link_ledger IF NOT EXISTS FOR ()-[r:Lifetime]->() ON (r.amount)
     ";
+
+pub static INDEX_SNAPSHOT: &str = "CREATE INDEX snapshot_account_id IF NOT EXISTS FOR (n:Snapshot) ON (n.address, n.epoch, n.version)";
 /// get the testing neo4j connection
 pub async fn get_neo4j_localhost_pool(port: u16) -> Result<Graph> {
     let uri = format!("127.0.0.1:{port}");
@@ -89,6 +91,7 @@ pub async fn maybe_create_indexes(graph: &Graph) -> Result<()> {
         INDEX_EXCHANGE_LEDGER,
         INDEX_EXCHANGE_LINK_LEDGER,
         INDEX_LIFETIME,
+        INDEX_SNAPSHOT,
     ])
     .await?;
     txn.commit().await?;

src/schema_account_state.rs

@@ -16,9 +16,10 @@ pub struct WarehouseAccState {
     pub address: AccountAddress,
     pub time: WarehouseTime,
     pub sequence_num: u64,
-    pub balance: u64,
-    pub slow_wallet_unlocked: u64,
-    pub slow_wallet_transferred: u64,
+    pub balance: f64,
+    pub slow_wallet_unlocked: Option<f64>,
+    pub slow_wallet_transferred: Option<f64>,
+    pub slow_wallet_acc: bool,
     pub donor_voice_acc: bool,
     pub miner_height: Option<u64>,
 }
@@ -28,9 +29,10 @@ impl Default for WarehouseAccState {
         Self {
             address: AccountAddress::ZERO,
             sequence_num: 0,
-            balance: 0,
-            slow_wallet_unlocked: 0,
-            slow_wallet_transferred: 0,
+            balance: 0.0,
+            slow_wallet_unlocked: None,
+            slow_wallet_transferred: None,
+            slow_wallet_acc: false,
             donor_voice_acc: false,
             miner_height: None,
             time: WarehouseTime::default(),
@@ -55,27 +57,42 @@ impl WarehouseAccState {
 impl WarehouseAccState {
     /// creates one transaction record in the cypher query map format
     /// Note original data was in an RFC rfc3339 with Z for UTC, Cypher seems to prefer with offsets +00000
-    pub fn to_cypher_object_template(&self) -> String {
+    pub fn acc_state_to_cypher_map(&self) -> String {
+        let slow_wallet_unlocked_literal = match self.slow_wallet_unlocked {
+            Some(n) => n.to_string(),
+            None => "NULL".to_string(),
+        };
+        let slow_wallet_transferred_literal = match self.slow_wallet_transferred {
+            Some(n) => n.to_string(),
+            None => "NULL".to_string(),
+        };
+
+        let miner_height_literal = match self.miner_height {
+            Some(n) => n.to_string(),
+            None => "NULL".to_string(),
+        };
+
         format!(
-            r#"{{address: "{}", balance: {}, version: {}, epoch: {},sequence_num: {}, slow_unlocked: {}, slow_transfer: {}, framework_version: "{}", donor_voice: {}, miner_height: {}}}"#,
+            r#"{{address: "{}", balance: {}, version: {}, epoch: {},sequence_num: {}, slow_unlocked: {}, slow_transfer: {}, framework_version: "{}", slow_wallet: {}, donor_voice: {}, miner_height: {}}}"#,
             self.address.to_hex_literal(),
             self.balance,
             self.time.version,
             self.time.epoch,
             self.sequence_num,
-            self.slow_wallet_unlocked,
-            self.slow_wallet_transferred,
+            slow_wallet_unlocked_literal,
+            slow_wallet_transferred_literal,
             self.time.framework_version,
+            self.slow_wallet_acc,
             self.donor_voice_acc,
-            self.miner_height.unwrap_or(0)
+            miner_height_literal
         )
     }
 
     /// create a cypher query string for the map object
     pub fn to_cypher_map(list: &[Self]) -> String {
         let mut list_literal = "".to_owned();
         for el in list {
-            let s = el.to_cypher_object_template();
+            let s = el.acc_state_to_cypher_map();
             list_literal.push_str(&s);
             list_literal.push(',');
         }
@@ -92,16 +109,32 @@ UNWIND tx_data AS tx
 MERGE (addr:Account {{address: tx.address}})
 MERGE (snap:Snapshot {{
     address: tx.address,
-    balance: tx.balance,
     epoch: tx.epoch,
-    framework_version: tx.framework_version,
-    version: tx.version,
-    sequence_num: tx.sequence_num,
-    slow_unlocked: tx.slow_unlocked,
-    slow_transfer: tx.slow_transfer,
-    donor_voice: tx.donor_voice,
-    miner_height: coalesce(tx.miner_height, 0)
+    version: tx.version
 }})
+
+SET
+  snap.balance = tx.balance,
+  snap.framework_version = tx.framework_version,
+  snap.sequence_num = tx.sequence_num,
+  snap.slow_wallet = tx.slow_wallet,
+  snap.donor_voice = tx.donor_voice
+
+// Conditionally add `tx.miner_height` if it exists
+FOREACH (_ IN CASE WHEN tx.miner_height IS NOT NULL THEN [1] ELSE [] END |
+    SET snap.miner_height = tx.miner_height
+)
+
+// Conditionally add `tx.slow_unlocked` if it exists
+FOREACH (_ IN CASE WHEN tx.slow_unlocked IS NOT NULL THEN [1] ELSE [] END |
+    SET snap.slow_unlocked = tx.slow_unlocked
+)
+
+// Conditionally add `tx.slow_transfer` if it exists
+FOREACH (_ IN CASE WHEN tx.slow_transfer IS NOT NULL THEN [1] ELSE [] END |
+    SET snap.slow_transfer = tx.slow_transfer
+)
+
 MERGE (addr)-[rel:State {{version: tx.version}}]->(snap)
 
 RETURN COUNT(snap) AS merged_snapshots

src/schema_transaction.rs

@@ -1,4 +1,6 @@
-use crate::{cypher_templates::to_cypher_object, scan::FrameworkVersion};
+use crate::{
+    cypher_templates::to_cypher_object, scan::FrameworkVersion, util::COIN_DECIMAL_PRECISION,
+};
 
 use chrono::{DateTime, Utc};
 use diem_crypto::HashValue;
@@ -12,13 +14,6 @@ use libra_backwards_compatibility::sdk::{
 use libra_types::{exports::AccountAddress, move_resource::coin_register_event::CoinRegisterEvent};
 use serde::{Deserialize, Serialize};
 
-// TODO check decimal precision
-/// Conversion of coins from V5 to V6
-pub const LEGACY_REBASE_MULTIPLIER: u64 = 35;
-/// Decimal precision
-// TODO: duplication, this is probably defined in libra-framework somewhere
-pub const COIN_DECIMAL_PRECISION: u64 = 1000000;
-
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub enum RelationLabel {
     Unknown, // undefined tx

src/util.rs

@@ -3,6 +3,13 @@ use diem_types::account_address::AccountAddress;
 use log::error;
 use serde::{Deserialize, Deserializer};
 
+// TODO check decimal precision
+/// Conversion of coins from V5 to V6
+pub const LEGACY_REBASE_MULTIPLIER: u64 = 35;
+/// Decimal precision
+// TODO: duplication, this is probably defined in libra-framework somewhere
+pub const COIN_DECIMAL_PRECISION: u64 = 1000000;
+
 /// Helper function to parse "YYYY-MM-DD" into `DateTime<Utc>`
 pub fn parse_date(date_str: &str) -> DateTime<Utc> {
     let datetime_str = format!("{date_str}T00:00:00Z"); // Append time and UTC offset

tests/test_extract_state.rs

@@ -5,7 +5,7 @@ use libra_forensic_db::extract_snapshot::{extract_current_snapshot, extract_v5_s
 use support::fixtures::{v5_state_manifest_fixtures_path, v7_state_manifest_fixtures_path};
 
 #[tokio::test]
-async fn test_extract_v5_manifest() -> Result<()> {
+async fn test_extract_v5_from_manifest() -> Result<()> {
     let archive_path = v5_state_manifest_fixtures_path();
     assert!(archive_path.exists());
     let s = extract_v5_snapshot(&archive_path).await?;
@@ -14,9 +14,9 @@ async fn test_extract_v5_manifest() -> Result<()> {
     let first = s.first().unwrap();
 
     assert!(&first.address.to_hex_literal() == "0x407d4d486fdc4e796504135e545be77");
-    assert!(first.balance == 100135989588);
-    assert!(first.slow_wallet_unlocked == 140001000000);
-    assert!(first.slow_wallet_transferred == 15999000000);
+    assert!(first.balance == 100135.989588);
+    assert!(first.slow_wallet_unlocked == Some(140001.000000));
+    assert!(first.slow_wallet_transferred == Some(15999.000000));
     assert!(first.sequence_num == 7);
 
     Ok(())