Add ConcurrentFungibleAssetSupply and ConcurrentFungibleAssetBalance (#…

…338)

* Add concurrent supply and balance

* Change Aggregator name

* Minor name change

rust/processor/src/models/fungible_asset_models/v2_fungible_asset_balances.rs

@@ -165,6 +165,13 @@ impl FungibleAssetBalance {
  let asset_type = inner.metadata.get_reference_address();
  let is_primary = Self::is_primary(&owner_address, &asset_type, &storage_id);
 
+ let concurrent_balance = object_data
+ .concurrent_fungible_asset_balance
+ .as_ref()
+ .map(|concurrent_fungible_asset_balance| {
+ concurrent_fungible_asset_balance.balance.value.clone()
+ });
+
  let coin_balance = Self {
  transaction_version: txn_version,
  write_set_change_index,
@@ -173,7 +180,9 @@ impl FungibleAssetBalance {
  asset_type: asset_type.clone(),
  is_primary,
  is_frozen: inner.frozen,
- amount: inner.balance.clone(),
+ amount: concurrent_balance
+ .clone()
+ .unwrap_or_else(|| inner.balance.clone()),
  transaction_timestamp: txn_timestamp,
  token_standard: TokenStandard::V2.to_string(),
  };
@@ -183,7 +192,7 @@ impl FungibleAssetBalance {
  asset_type: asset_type.clone(),
  is_primary,
  is_frozen: inner.frozen,
- amount: inner.balance.clone(),
+ amount: concurrent_balance.unwrap_or_else(|| inner.balance.clone()),
  last_transaction_version: txn_version,
  last_transaction_timestamp: txn_timestamp,
  token_standard: TokenStandard::V2.to_string(),

rust/processor/src/models/fungible_asset_models/v2_fungible_asset_utils.rs

@@ -9,7 +9,7 @@ use crate::{
  coin_models::coin_utils::COIN_ADDR, default_models::move_resources::MoveResource,
  token_models::token_utils::URI_LENGTH, token_v2_models::v2_token_utils::ResourceReference,
  },
- utils::util::{deserialize_from_string, truncate_str},
+ utils::util::{deserialize_from_string, truncate_str, Aggregator},
 };
 use anyhow::{Context, Result};
 use aptos_protos::transaction::v1::WriteResource;
@@ -187,6 +187,76 @@ impl FungibleAssetSupply {
  }
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct ConcurrentFungibleAssetSupply {
+ pub current: Aggregator,
+}
+
+impl ConcurrentFungibleAssetSupply {
+ pub fn from_write_resource(
+ write_resource: &WriteResource,
+ txn_version: i64,
+ ) -> anyhow::Result<Option<Self>> {
+ let type_str: String = MoveResource::get_outer_type_from_write_resource(write_resource);
+ if !V2FungibleAssetResource::is_resource_supported(type_str.as_str()) {
+ return Ok(None);
+ }
+ let resource = MoveResource::from_write_resource(
+ write_resource,
+ 0, // Placeholder, this isn't used anyway
+ txn_version,
+ 0, // Placeholder, this isn't used anyway
+ );
+
+ if let V2FungibleAssetResource::ConcurrentFungibleAssetSupply(inner) =
+ V2FungibleAssetResource::from_resource(
+ &type_str,
+ resource.data.as_ref().unwrap(),
+ txn_version,
+ )?
+ {
+ Ok(Some(inner))
+ } else {
+ Ok(None)
+ }
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct ConcurrentFungibleAssetBalance {
+ pub balance: Aggregator,
+}
+
+impl ConcurrentFungibleAssetBalance {
+ pub fn from_write_resource(
+ write_resource: &WriteResource,
+ txn_version: i64,
+ ) -> anyhow::Result<Option<Self>> {
+ let type_str: String = MoveResource::get_outer_type_from_write_resource(write_resource);
+ if !V2FungibleAssetResource::is_resource_supported(type_str.as_str()) {
+ return Ok(None);
+ }
+ let resource = MoveResource::from_write_resource(
+ write_resource,
+ 0, // Placeholder, this isn't used anyway
+ txn_version,
+ 0, // Placeholder, this isn't used anyway
+ );
+
+ if let V2FungibleAssetResource::ConcurrentFungibleAssetBalance(inner) =
+ V2FungibleAssetResource::from_resource(
+ &type_str,
+ resource.data.as_ref().unwrap(),
+ txn_version,
+ )?
+ {
+ Ok(Some(inner))
+ } else {
+ Ok(None)
+ }
+ }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct DepositEvent {
  #[serde(deserialize_with = "deserialize_from_string")]
@@ -209,14 +279,18 @@ pub enum V2FungibleAssetResource {
  FungibleAssetMetadata(FungibleAssetMetadata),
  FungibleAssetStore(FungibleAssetStore),
  FungibleAssetSupply(FungibleAssetSupply),
+ ConcurrentFungibleAssetSupply(ConcurrentFungibleAssetSupply),
+ ConcurrentFungibleAssetBalance(ConcurrentFungibleAssetBalance),
 }
 
 impl V2FungibleAssetResource {
  pub fn is_resource_supported(data_type: &str) -> bool {
  [
  format!("{}::fungible_asset::Supply", COIN_ADDR),
+ format!("{}::fungible_asset::ConcurrentSupply", COIN_ADDR),
  format!("{}::fungible_asset::Metadata", COIN_ADDR),
  format!("{}::fungible_asset::FungibleStore", COIN_ADDR),
+ format!("{}::fungible_asset::ConcurrentFungibleBalance", COIN_ADDR),
  ]
  .contains(&data_type.to_string())
  }
@@ -231,6 +305,10 @@ impl V2FungibleAssetResource {
  serde_json::from_value(data.clone())
  .map(|inner| Some(Self::FungibleAssetSupply(inner)))
  },
+ x if x == format!("{}::fungible_asset::ConcurrentSupply", COIN_ADDR) => {
+ serde_json::from_value(data.clone())
+ .map(|inner| Some(Self::ConcurrentFungibleAssetSupply(inner)))
+ },
  x if x == format!("{}::fungible_asset::Metadata", COIN_ADDR) => {
  serde_json::from_value(data.clone())
  .map(|inner| Some(Self::FungibleAssetMetadata(inner)))
@@ -239,6 +317,10 @@ impl V2FungibleAssetResource {
  serde_json::from_value(data.clone())
  .map(|inner| Some(Self::FungibleAssetStore(inner)))
  },
+ x if x == format!("{}::fungible_asset::ConcurrentFungibleBalance", COIN_ADDR) => {
+ serde_json::from_value(data.clone())
+ .map(|inner| Some(Self::ConcurrentFungibleAssetBalance(inner)))
+ },
  _ => Ok(None),
  }
  .context(format!(
@@ -312,4 +394,6 @@ mod tests {
  panic!("Wrong type")
  }
  }
+
+ // TODO: Add similar tests for ConcurrentFungibleAssetSupply.
 }

rust/processor/src/models/fungible_asset_models/v2_fungible_metadata.rs

@@ -63,16 +63,23 @@ impl FungibleAssetMetadataModel {
  let asset_type = standardize_address(&write_resource.address.to_string());
  if let Some(object_metadata) = object_metadatas.get(&asset_type) {
  let object = &object_metadata.object.object_core;
- let fungible_asset_supply = object_metadata.fungible_asset_supply.as_ref();
- let (maximum_v2, supply_v2) =
- if let Some(fungible_asset_supply) = fungible_asset_supply {
- (
- fungible_asset_supply.get_maximum(),
- Some(fungible_asset_supply.current.clone()),
- )
- } else {
- (None, None)
- };
+ let (maximum_v2, supply_v2) = if let Some(fungible_asset_supply) =
+ object_metadata.fungible_asset_supply.as_ref()
+ {
+ (
+ fungible_asset_supply.get_maximum(),
+ Some(fungible_asset_supply.current.clone()),
+ )
+ } else if let Some(concurrent_fungible_asset_supply) =
+ object_metadata.concurrent_fungible_asset_supply.as_ref()
+ {
+ (
+ Some(concurrent_fungible_asset_supply.current.max_value.clone()),
+ Some(concurrent_fungible_asset_supply.current.value.clone()),
+ )
+ } else {
+ (None, None)
+ };
 
  return Ok(Some(Self {
  asset_type: asset_type.clone(),

rust/processor/src/models/object_models/v2_object_utils.rs

@@ -9,7 +9,8 @@ use crate::{
  models::{
  default_models::move_resources::MoveResource,
  fungible_asset_models::v2_fungible_asset_utils::{
- FungibleAssetMetadata, FungibleAssetStore, FungibleAssetSupply,
+ ConcurrentFungibleAssetBalance, ConcurrentFungibleAssetSupply, FungibleAssetMetadata,
+ FungibleAssetStore, FungibleAssetSupply,
  },
  token_v2_models::v2_token_utils::{
  AptosCollection, ConcurrentSupply, FixedSupply, PropertyMapModel, TokenIdentifiers,
@@ -41,7 +42,9 @@ pub struct ObjectAggregatedData {
  // Fungible asset structs
  pub fungible_asset_metadata: Option<FungibleAssetMetadata>,
  pub fungible_asset_supply: Option<FungibleAssetSupply>,
+ pub concurrent_fungible_asset_supply: Option<ConcurrentFungibleAssetSupply>,
  pub fungible_asset_store: Option<FungibleAssetStore>,
+ pub concurrent_fungible_asset_balance: Option<ConcurrentFungibleAssetBalance>,
  // Token v2 structs
  pub aptos_collection: Option<AptosCollection>,
  pub fixed_supply: Option<FixedSupply>,
@@ -66,6 +69,8 @@ impl Default for ObjectAggregatedData {
  transfer_events: Vec::new(),
  fungible_asset_metadata: None,
  fungible_asset_supply: None,
+ concurrent_fungible_asset_supply: None,
+ concurrent_fungible_asset_balance: None,
  fungible_asset_store: None,
  aptos_collection: None,
  fixed_supply: None,

rust/processor/src/models/token_v2_models/v2_token_utils.rs

@@ -13,8 +13,7 @@ use crate::{
  },
  utils::util::{
  deserialize_from_string, deserialize_token_object_property_map_from_bcs_hexstring,
- standardize_address, truncate_str, AggregatorSnapshotU64, AggregatorU64,
- DerivedStringSnapshot,
+ standardize_address, truncate_str, Aggregator, AggregatorSnapshot, DerivedStringSnapshot,
  },
 };
 use ahash::{AHashMap, AHashSet};
@@ -253,8 +252,8 @@ impl UnlimitedSupply {
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct ConcurrentSupply {
- pub current_supply: AggregatorU64,
- pub total_minted: AggregatorU64,
+ pub current_supply: Aggregator,
+ pub total_minted: Aggregator,
 }
 
 impl ConcurrentSupply {
@@ -310,7 +309,7 @@ impl MintEvent {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Mint {
  collection: String,
- pub index: AggregatorSnapshotU64,
+ pub index: AggregatorSnapshot,
  token: String,
 }
 

rust/processor/src/processors/fungible_asset_processor.rs

@@ -11,7 +11,8 @@ use crate::{
  CurrentUnifiedFungibleAssetBalance, FungibleAssetBalance,
  },
  v2_fungible_asset_utils::{
- FeeStatement, FungibleAssetMetadata, FungibleAssetStore, FungibleAssetSupply,
+ ConcurrentFungibleAssetBalance, ConcurrentFungibleAssetSupply, FeeStatement,
+ FungibleAssetMetadata, FungibleAssetStore, FungibleAssetSupply,
  },
  v2_fungible_metadata::{FungibleAssetMetadataMapping, FungibleAssetMetadataModel},
  },
@@ -495,6 +496,26 @@ async fn parse_v2_coin(
  {
  aggregated_data.fungible_asset_supply = Some(fungible_asset_supply);
  }
+ if let Some(concurrent_fungible_asset_supply) =
+ ConcurrentFungibleAssetSupply::from_write_resource(
+ write_resource,
+ txn_version,
+ )
+ .unwrap()
+ {
+ aggregated_data.concurrent_fungible_asset_supply =
+ Some(concurrent_fungible_asset_supply);
+ }
+ if let Some(concurrent_fungible_asset_balance) =
+ ConcurrentFungibleAssetBalance::from_write_resource(
+ write_resource,
+ txn_version,
+ )
+ .unwrap()
+ {
+ aggregated_data.concurrent_fungible_asset_balance =
+ Some(concurrent_fungible_asset_balance);
+ }
  }
  } else if let Change::DeleteResource(delete_resource) = wsc.change.as_ref().unwrap() {
  if let Some((balance, current_balance, event_to_coin)) =

rust/processor/src/processors/nft_metadata_processor.rs

@@ -250,6 +250,8 @@ async fn parse_v2_token(
  token: None,
  fungible_asset_metadata: None,
  fungible_asset_supply: None,
+ concurrent_fungible_asset_supply: None,
+ concurrent_fungible_asset_balance: None,
  fungible_asset_store: None,
  token_identifier: None,
  },

rust/processor/src/processors/objects_processor.rs

@@ -205,6 +205,8 @@ impl ProcessorTrait for ObjectsProcessor {
  property_map: None,
  transfer_events: vec![],
  fungible_asset_supply: None,
+ concurrent_fungible_asset_supply: None,
+ concurrent_fungible_asset_balance: None,
  token_identifier: None,
  });
  }

rust/processor/src/utils/util.rs

@@ -457,15 +457,15 @@ pub fn get_name_from_unnested_move_type(move_type: &str) -> &str {
 
 /* COMMON STRUCTS */
 #[derive(Serialize, Deserialize, Debug, Clone)]
-pub struct AggregatorU64 {
+pub struct Aggregator {
  #[serde(deserialize_with = "deserialize_from_string")]
  pub value: BigDecimal,
  #[serde(deserialize_with = "deserialize_from_string")]
  pub max_value: BigDecimal,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
-pub struct AggregatorSnapshotU64 {
+pub struct AggregatorSnapshot {
  #[serde(deserialize_with = "deserialize_from_string")]
  pub value: BigDecimal,
 }