[EI-397] Separate fungible asset and token indexing (#354)

* Re-index fungible token

* Test move script

* Schema changes

* Comments

* on conflict, update

* Fix is_fungible_v2 column

rust/processor/migrations/2024-04-18-173631_fungible_token/down.sql

@@ -0,0 +1,4 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE fungible_asset_metadata
+DROP COLUMN supply_v2,
+DROP COLUMN maximum_v2;

rust/processor/migrations/2024-04-18-173631_fungible_token/up.sql

@@ -0,0 +1,12 @@
+-- Your SQL goes here
+ALTER TABLE fungible_asset_metadata
+ADD COLUMN supply_v2 NUMERIC,
+ADD COLUMN maximum_v2 NUMERIC;
+
+ALTER TABLE current_token_datas_v2
+ALTER COLUMN supply DROP NOT NULL,
+ALTER COLUMN decimals DROP NOT NULL;
+
+ALTER TABLE token_datas_v2
+ALTER COLUMN supply DROP NOT NULL,
+ALTER COLUMN decimals DROP NOT NULL;

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

@@ -5,10 +5,7 @@
 #![allow(clippy::extra_unused_lifetimes)]
 #![allow(clippy::unused_unit)]
 
-use super::{
-    v2_fungible_asset_utils::{FeeStatement, FungibleAssetEvent},
-    v2_fungible_metadata::FungibleAssetMetadataModel,
-};
+use super::v2_fungible_asset_utils::{FeeStatement, FungibleAssetEvent};
 use crate::{
     models::{
         coin_models::{
@@ -19,7 +16,7 @@ use crate::{
         token_v2_models::v2_token_utils::TokenStandard,
     },
     schema::fungible_asset_activities,
-    utils::{database::PgPoolConnection, util::standardize_address},
+    utils::util::standardize_address,
 };
 use ahash::AHashMap;
 use anyhow::Context;
@@ -70,7 +67,6 @@ impl FungibleAssetActivity {
         event_index: i64,
         entry_function_id_str: &Option<String>,
         object_aggregated_data_mapping: &ObjectAggregatedDataMapping,
-        conn: &mut PgPoolConnection<'_>,
     ) -> anyhow::Result<Option<Self>> {
         let event_type = event.type_str.clone();
         if let Some(fa_event) =
@@ -84,17 +80,6 @@ impl FungibleAssetActivity {
                 let object_core = &object_metadata.object.object_core;
                 let fungible_asset = object_metadata.fungible_asset_store.as_ref().unwrap();
                 let asset_type = fungible_asset.metadata.get_reference_address();
-                // If it's a fungible token, return early
-                if !FungibleAssetMetadataModel::is_address_fungible_asset(
-                    conn,
-                    &asset_type,
-                    object_aggregated_data_mapping,
-                    txn_version,
-                )
-                .await
-                {
-                    return Ok(None);
-                }
 
                 let (is_frozen, amount) = match fa_event {
                     FungibleAssetEvent::WithdrawEvent(inner) => (None, Some(inner.amount.clone())),

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

@@ -7,7 +7,6 @@
 
 use super::{
     v2_fungible_asset_activities::EventToCoinType, v2_fungible_asset_utils::FungibleAssetStore,
-    v2_fungible_metadata::FungibleAssetMetadataModel,
 };
 use crate::{
     models::{
@@ -16,7 +15,7 @@ use crate::{
         token_v2_models::v2_token_utils::TokenStandard,
     },
     schema::{current_fungible_asset_balances, fungible_asset_balances},
-    utils::{database::PgPoolConnection, util::standardize_address},
+    utils::util::standardize_address,
 };
 use ahash::AHashMap;
 use aptos_protos::transaction::v1::WriteResource;
@@ -71,7 +70,6 @@ impl FungibleAssetBalance {
         txn_version: i64,
         txn_timestamp: chrono::NaiveDateTime,
         object_metadatas: &ObjectAggregatedDataMapping,
-        conn: &mut PgPoolConnection<'_>,
     ) -> anyhow::Result<Option<(Self, CurrentFungibleAssetBalance)>> {
         if let Some(inner) = &FungibleAssetStore::from_write_resource(write_resource, txn_version)?
         {
@@ -81,17 +79,6 @@ impl FungibleAssetBalance {
                 let object = &object_data.object.object_core;
                 let owner_address = object.get_owner_address();
                 let asset_type = inner.metadata.get_reference_address();
-                // If it's a fungible token, return early
-                if !FungibleAssetMetadataModel::is_address_fungible_asset(
-                    conn,
-                    &asset_type,
-                    object_metadatas,
-                    txn_version,
-                )
-                .await
-                {
-                    return Ok(None);
-                }
                 let is_primary = Self::is_primary(&owner_address, &asset_type, &storage_id);
 
                 let coin_balance = Self {

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

@@ -13,12 +13,12 @@ use crate::{
         token_v2_models::v2_token_utils::TokenStandard,
     },
     schema::fungible_asset_metadata,
-    utils::{database::PgPoolConnection, util::standardize_address},
+    utils::util::standardize_address,
 };
 use ahash::AHashMap;
 use aptos_protos::transaction::v1::WriteResource;
+use bigdecimal::BigDecimal;
 use diesel::prelude::*;
-use diesel_async::RunQueryDsl;
 use field_count::FieldCount;
 use serde::{Deserialize, Serialize};
 
@@ -44,26 +44,8 @@ pub struct FungibleAssetMetadataModel {
     pub supply_aggregator_table_key_v1: Option<String>,
     pub token_standard: String,
     pub is_token_v2: Option<bool>,
-}
-
-#[derive(Debug, Deserialize, Identifiable, Queryable, Serialize)]
-#[diesel(primary_key(asset_type))]
-#[diesel(table_name = fungible_asset_metadata)]
-pub struct FungibleAssetMetadataQuery {
-    pub asset_type: String,
-    pub creator_address: String,
-    pub name: String,
-    pub symbol: String,
-    pub decimals: i32,
-    pub icon_uri: Option<String>,
-    pub project_uri: Option<String>,
-    pub last_transaction_version: i64,
-    pub last_transaction_timestamp: chrono::NaiveDateTime,
-    pub supply_aggregator_table_handle_v1: Option<String>,
-    pub supply_aggregator_table_key_v1: Option<String>,
-    pub token_standard: String,
-    pub inserted_at: chrono::NaiveDateTime,
-    pub is_token_v2: Option<bool>,
+    pub supply_v2: Option<BigDecimal>,
+    pub maximum_v2: Option<BigDecimal>,
 }
 
 impl FungibleAssetMetadataModel {
@@ -81,7 +63,16 @@ 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 is_token_v2 = object_metadata.token.is_some();
+                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)
+                    };
 
                 return Ok(Some(Self {
                     asset_type: asset_type.clone(),
@@ -96,7 +87,9 @@ impl FungibleAssetMetadataModel {
                     supply_aggregator_table_handle_v1: None,
                     supply_aggregator_table_key_v1: None,
                     token_standard: TokenStandard::V2.to_string(),
-                    is_token_v2: Some(is_token_v2),
+                    is_token_v2: None,
+                    supply_v2,
+                    maximum_v2,
                 }));
             }
         }
@@ -135,7 +128,9 @@ impl FungibleAssetMetadataModel {
                         supply_aggregator_table_handle_v1: supply_aggregator_table_handle,
                         supply_aggregator_table_key_v1: supply_aggregator_table_key,
                         token_standard: TokenStandard::V1.to_string(),
-                        is_token_v2: Some(false),
+                        is_token_v2: None,
+                        supply_v2: None,
+                        maximum_v2: None,
                     }))
                 } else {
                     Ok(None)
@@ -144,54 +139,4 @@ impl FungibleAssetMetadataModel {
             _ => Ok(None),
         }
     }
-
-    /// A fungible asset can also be a token. We will make a best effort guess at whether this is a fungible token.
-    /// 1. If metadata is present without token object, then it's not a token
-    /// 2. If metadata is not present, we will do a lookup in the db.
-    pub async fn is_address_fungible_asset(
-        conn: &mut PgPoolConnection<'_>,
-        asset_type: &str,
-        object_aggregated_data_mapping: &ObjectAggregatedDataMapping,
-        txn_version: i64,
-    ) -> bool {
-        // 1. If metadata is present without token object, then it's not a token
-        if let Some(object_data) = object_aggregated_data_mapping.get(asset_type) {
-            if object_data.fungible_asset_metadata.is_some() {
-                return object_data.token.is_none();
-            }
-        }
-        // 2. If metadata is not present, we will do a lookup in the db.
-        match FungibleAssetMetadataQuery::get_by_asset_type(conn, asset_type).await {
-            Ok(metadata) => {
-                if let Some(is_token_v2) = metadata.is_token_v2 {
-                    return !is_token_v2;
-                }
-
-                // If is_token_v2 is null, then the metadata is a v1 coin info, and it's not a token
-                true
-            },
-            Err(_) => {
-                tracing::error!(
-                    transaction_version = txn_version,
-                    lookup_key = asset_type,
-                    "Missing fungible_asset_metadata for asset_type: {}. You probably should backfill db.",
-                    asset_type,
-                );
-                // Default
-                true
-            },
-        }
-    }
-}
-
-impl FungibleAssetMetadataQuery {
-    pub async fn get_by_asset_type(
-        conn: &mut PgPoolConnection<'_>,
-        asset_type: &str,
-    ) -> diesel::QueryResult<Self> {
-        fungible_asset_metadata::table
-            .filter(fungible_asset_metadata::asset_type.eq(asset_type))
-            .first::<Self>(conn)
-            .await
-    }
 }

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

@@ -52,6 +52,32 @@ pub struct ObjectAggregatedData {
     pub token_identifier: Option<TokenIdentifiers>,
 }
 
+impl Default for ObjectAggregatedData {
+    fn default() -> Self {
+        Self {
+            object: ObjectWithMetadata {
+                object_core: ObjectCore {
+                    allow_ungated_transfer: false,
+                    guid_creation_num: BigDecimal::default(),
+                    owner: String::default(),
+                },
+                state_key_hash: String::default(),
+            },
+            transfer_events: Vec::new(),
+            fungible_asset_metadata: None,
+            fungible_asset_supply: None,
+            fungible_asset_store: None,
+            aptos_collection: None,
+            fixed_supply: None,
+            property_map: None,
+            token: None,
+            unlimited_supply: None,
+            concurrent_supply: None,
+            token_identifier: None,
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct ObjectCore {
     pub allow_ungated_transfer: bool,