[index] current unified balances (#366)

* [index] concurrent unified balances

* fixup! [index] concurrent unified balances

rust/Cargo.toml

@@ -61,6 +61,7 @@ google-cloud-googleapis = "0.10.0"
 google-cloud-pubsub = "0.18.0"
 hex = "0.4.3"
 itertools = "0.12.1"
+lazy_static = "1.4.0"
 jemallocator = { version = "0.5.0", features = [
     "profiling",
     "unprefixed_malloc_on_supported_platforms",
@@ -88,6 +89,7 @@ strum = { version = "0.24.1", features = ["derive"] }
 tempfile = "3.3.0"
 toml = "0.7.4"
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }
+tiny-keccak = { version = "2.0.2", features = ["keccak", "sha3"] }
 tokio = { version = "1.35.1", features = ["full"] }
 tonic = { version = "0.11.0", features = [
     "tls",

rust/processor/Cargo.toml

@@ -37,6 +37,7 @@ google-cloud-pubsub = { workspace = true }
 hex = { workspace = true }
 itertools = { workspace = true }
 kanal = { workspace = true }
+lazy_static = { workspace = true }
 num_cpus = { workspace = true }
 once_cell = { workspace = true }
 prometheus = { workspace = true }
@@ -58,6 +59,7 @@ url = { workspace = true }
 # Postgres SSL support
 native-tls = { workspace = true }
 postgres-native-tls = { workspace = true }
+tiny-keccak = { workspace = true }
 tokio-postgres = { workspace = true }
 
 [target.'cfg(unix)'.dependencies]

rust/processor/migrations/2024-05-04-025823_current_unified_fungible_asset_balance/down.sql

@@ -0,0 +1,4 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE IF EXISTS current_unified_fungible_asset_balances;
+DROP INDEX IF EXISTS cufab_owner_at_index;
+DROP INDEX IF EXISTS cufab_insat_index;

rust/processor/migrations/2024-05-04-025823_current_unified_fungible_asset_balance/up.sql

@@ -0,0 +1,21 @@
+-- current fungible asset balances
+CREATE TABLE IF NOT EXISTS current_unified_fungible_asset_balances (
+    storage_id VARCHAR(66) PRIMARY KEY NOT NULL,
+    owner_address VARCHAR(66) NOT NULL,
+    asset_type VARCHAR(66) NOT NULL,
+    coin_type VARCHAR(1000),
+    is_primary BOOLEAN,
+    is_frozen BOOLEAN NOT NULL,
+    amount_v1 NUMERIC,
+    amount_v2 NUMERIC,
+    amount NUMERIC GENERATED ALWAYS AS (COALESCE(amount_v1, 0) + COALESCE(amount_v2, 0)) STORED,
+    last_transaction_version_v1 BIGINT,
+    last_transaction_version_v2 BIGINT,
+    last_transaction_version BIGINT GENERATED ALWAYS AS (GREATEST(last_transaction_version_v1, last_transaction_version_v2)) STORED,
+    last_transaction_timestamp_v1 TIMESTAMP,
+    last_transaction_timestamp_v2 TIMESTAMP,
+    last_transaction_timestamp TIMESTAMP GENERATED ALWAYS AS (GREATEST(last_transaction_timestamp_v1, last_transaction_timestamp_v2)) STORED,
+    inserted_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS cufab_owner_at_index ON current_unified_fungible_asset_balances (owner_address, asset_type);
+CREATE INDEX IF NOT EXISTS cufab_insat_index ON current_unified_fungible_asset_balances (inserted_at);

rust/processor/src/models/ans_models/ans_utils.rs

@@ -220,7 +220,7 @@ impl AnsWriteResource {
         ans_v2_contract_address: &str,
         txn_version: i64,
     ) -> anyhow::Result<Option<Self>> {
-        let type_str = MoveResource::get_outer_type_from_resource(write_resource);
+        let type_str = MoveResource::get_outer_type_from_write_resource(write_resource);
         let data = write_resource.data.as_str();
 
         match type_str.clone() {

rust/processor/src/models/coin_models/coin_activities.rs

@@ -22,11 +22,13 @@ use crate::{
         },
         user_transactions_models::signatures::Signature,
     },
-    processors::coin_processor::APTOS_COIN_TYPE_STR,
     schema::coin_activities,
     utils::{
         counters::PROCESSOR_UNKNOWN_TYPE_COUNT,
-        util::{get_entry_function_from_user_request, standardize_address, u64_to_bigdecimal},
+        util::{
+            get_entry_function_from_user_request, standardize_address, u64_to_bigdecimal,
+            APTOS_COIN_TYPE_STR,
+        },
     },
 };
 use ahash::AHashMap;

rust/processor/src/models/coin_models/coin_supply.rs

@@ -7,7 +7,8 @@
 
 use crate::{
     models::default_models::move_tables::TableItem,
-    processors::coin_processor::APTOS_COIN_TYPE_STR, schema::coin_supply, utils::util::hash_str,
+    schema::coin_supply,
+    utils::util::{hash_str, APTOS_COIN_TYPE_STR},
 };
 use anyhow::Context;
 use aptos_protos::transaction::v1::WriteTableItem;

rust/processor/src/models/coin_models/coin_utils.rs

@@ -8,8 +8,8 @@ use crate::{
     models::default_models::move_resources::MoveResource,
     utils::util::{deserialize_from_string, hash_str, standardize_address, truncate_str},
 };
-use anyhow::{Context, Result};
-use aptos_protos::transaction::v1::{move_type::Content, MoveType, WriteResource};
+use anyhow::{bail, Context, Result};
+use aptos_protos::transaction::v1::{move_type::Content, DeleteResource, MoveType, WriteResource};
 use bigdecimal::BigDecimal;
 use once_cell::sync::Lazy;
 use regex::Regex;
@@ -229,6 +229,8 @@ impl CoinInfoType {
 pub enum CoinResource {
     CoinInfoResource(CoinInfoResource),
     CoinStoreResource(CoinStoreResource),
+    CoinInfoDeletion,
+    CoinStoreDeletion,
 }
 
 impl CoinResource {
@@ -262,11 +264,27 @@ impl CoinResource {
         ))
     }
 
+    fn from_delete_resource_internal(data_type: &str, txn_version: i64) -> Result<CoinResource> {
+        match data_type {
+            x if x == format!("{}::coin::CoinInfo", COIN_ADDR) => {
+                Ok(CoinResource::CoinInfoDeletion)
+            },
+            x if x == format!("{}::coin::CoinStore", COIN_ADDR) => {
+                Ok(CoinResource::CoinStoreDeletion)
+            },
+            _ => bail!(
+                "Resource unsupported! Call is_resource_supported first. version {} type {}",
+                txn_version,
+                data_type
+            ),
+        }
+    }
+
     pub fn from_write_resource(
         write_resource: &WriteResource,
         txn_version: i64,
     ) -> Result<Option<CoinResource>> {
-        let type_str = MoveResource::get_outer_type_from_resource(write_resource);
+        let type_str = MoveResource::get_outer_type_from_write_resource(write_resource);
         if !CoinResource::is_resource_supported(type_str.as_str()) {
             return Ok(None);
         }
@@ -282,6 +300,20 @@ impl CoinResource {
             txn_version,
         )?))
     }
+
+    pub fn from_delete_resource(
+        delete_resource: &DeleteResource,
+        txn_version: i64,
+    ) -> Result<Option<CoinResource>> {
+        let type_str = MoveResource::get_outer_type_from_delete_resource(delete_resource);
+        if !CoinResource::is_resource_supported(type_str.as_str()) {
+            return Ok(None);
+        }
+        Ok(Some(Self::from_delete_resource_internal(
+            &type_str,
+            txn_version,
+        )?))
+    }
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

rust/processor/src/models/default_models/move_resources.rs

@@ -116,7 +116,7 @@ impl MoveResource {
         }
     }
 
-    pub fn get_outer_type_from_resource(write_resource: &WriteResource) -> String {
+    pub fn get_outer_type_from_write_resource(write_resource: &WriteResource) -> String {
         let move_struct_tag =
             Self::convert_move_struct_tag(write_resource.r#type.as_ref().unwrap());
 
@@ -127,6 +127,18 @@ impl MoveResource {
             move_struct_tag.name,
         )
     }
+
+    pub fn get_outer_type_from_delete_resource(delete_resource: &DeleteResource) -> String {
+        let move_struct_tag =
+            Self::convert_move_struct_tag(delete_resource.r#type.as_ref().unwrap());
+
+        format!(
+            "{}::{}::{}",
+            move_struct_tag.get_address(),
+            move_struct_tag.module,
+            move_struct_tag.name,
+        )
+    }
 }
 
 impl MoveStructTag {