Fix fa indexing when secondary store gets burnt

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

@@ -42,9 +42,9 @@ pub type EventToCoinType = AHashMap<EventGuidResource, CoinType>;
 pub struct FungibleAssetActivity {
     pub transaction_version: i64,
     pub event_index: i64,
-    pub owner_address: String,
+    pub owner_address: Option<String>,
     pub storage_id: String,
-    pub asset_type: String,
+    pub asset_type: Option<String>,
     pub is_frozen: Option<bool>,
     pub amount: Option<BigDecimal>,
     pub type_: String,
@@ -72,40 +72,67 @@ impl FungibleAssetActivity {
         if let Some(fa_event) =
             &FungibleAssetEvent::from_event(event_type.as_str(), &event.data, txn_version)?
         {
-            let storage_id = standardize_address(&event.key.as_ref().unwrap().account_address);
+            let (storage_id, is_frozen, amount) = match fa_event {
+                FungibleAssetEvent::WithdrawEvent(inner) => (
+                    standardize_address(&event.key.as_ref().unwrap().account_address),
+                    None,
+                    Some(inner.amount.clone()),
+                ),
+                FungibleAssetEvent::DepositEvent(inner) => (
+                    standardize_address(&event.key.as_ref().unwrap().account_address),
+                    None,
+                    Some(inner.amount.clone()),
+                ),
+                FungibleAssetEvent::FrozenEvent(inner) => (
+                    standardize_address(&event.key.as_ref().unwrap().account_address),
+                    Some(inner.frozen),
+                    None,
+                ),
+                FungibleAssetEvent::WithdrawEventV2(inner) => (
+                    standardize_address(&inner.store),
+                    None,
+                    Some(inner.amount.clone()),
+                ),
+                FungibleAssetEvent::DepositEventV2(inner) => (
+                    standardize_address(&inner.store),
+                    None,
+                    Some(inner.amount.clone()),
+                ),
+                FungibleAssetEvent::FrozenEventV2(inner) => {
+                    (standardize_address(&inner.store), Some(inner.frozen), None)
+                },
+            };
 
             // The event account address will also help us find fungible store which tells us where to find
             // the metadata
-            if let Some(object_metadata) = object_aggregated_data_mapping.get(&storage_id) {
-                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();
+            let maybe_object_metadata = object_aggregated_data_mapping.get(&storage_id);
+            // The ObjectCore might not exist in the transaction if the object got deleted
+            let maybe_owner_address = maybe_object_metadata
+                .map(|metadata| &metadata.object.object_core)
+                .map(|object_core| object_core.get_owner_address());
+            // The FungibleStore might not exist in the transaction if it's a secondary store that got burnt
+            let maybe_asset_type = maybe_object_metadata
+                .and_then(|metadata| metadata.fungible_asset_store.as_ref())
+                .map(|fa| fa.metadata.get_reference_address());
 
-                let (is_frozen, amount) = match fa_event {
-                    FungibleAssetEvent::WithdrawEvent(inner) => (None, Some(inner.amount.clone())),
-                    FungibleAssetEvent::DepositEvent(inner) => (None, Some(inner.amount.clone())),
-                    FungibleAssetEvent::FrozenEvent(inner) => (Some(inner.frozen), None),
-                };
-
-                return Ok(Some(Self {
-                    transaction_version: txn_version,
-                    event_index,
-                    owner_address: object_core.get_owner_address(),
-                    storage_id: storage_id.clone(),
-                    asset_type: asset_type.clone(),
-                    is_frozen,
-                    amount,
-                    type_: event_type.clone(),
-                    is_gas_fee: false,
-                    gas_fee_payer_address: None,
-                    is_transaction_success: true,
-                    entry_function_id_str: entry_function_id_str.clone(),
-                    block_height,
-                    token_standard: TokenStandard::V2.to_string(),
-                    transaction_timestamp: txn_timestamp,
-                    storage_refund_amount: BigDecimal::zero(),
-                }));
-            }
+            return Ok(Some(Self {
+                transaction_version: txn_version,
+                event_index,
+                owner_address: maybe_owner_address,
+                storage_id: storage_id.clone(),
+                asset_type: maybe_asset_type,
+                is_frozen,
+                amount,
+                type_: event_type.clone(),
+                is_gas_fee: false,
+                gas_fee_payer_address: None,
+                is_transaction_success: true,
+                entry_function_id_str: entry_function_id_str.clone(),
+                block_height,
+                token_standard: TokenStandard::V2.to_string(),
+                transaction_timestamp: txn_timestamp,
+                storage_refund_amount: BigDecimal::zero(),
+            }));
         }
         Ok(None)
     }
@@ -122,35 +149,49 @@ impl FungibleAssetActivity {
         if let Some(inner) =
             CoinEvent::from_event(event.type_str.as_str(), &event.data, txn_version)?
         {
-            let amount = match inner {
-                CoinEvent::WithdrawCoinEvent(inner) => inner.amount,
-                CoinEvent::DepositCoinEvent(inner) => inner.amount,
-            };
-            let event_key = event.key.as_ref().context("event must have a key")?;
-            let event_move_guid = EventGuidResource {
-                addr: standardize_address(event_key.account_address.as_str()),
-                creation_num: event_key.creation_number as i64,
+            let (owner_address, amount, coin_type_option) = match inner {
+                CoinEvent::WithdrawCoinEvent(inner) => (
+                    standardize_address(&event.key.as_ref().unwrap().account_address),
+                    inner.amount.clone(),
+                    None,
+                ),
+                CoinEvent::DepositCoinEvent(inner) => (
+                    standardize_address(&event.key.as_ref().unwrap().account_address),
+                    inner.amount.clone(),
+                    None,
+                ),
             };
-            // Given this mapping only contains coin type < 1000 length, we should not assume that the mapping exists.
-            // If it doesn't exist, skip.
-            let coin_type = match event_to_coin_type.get(&event_move_guid) {
-                Some(coin_type) => coin_type.clone(),
-                None => {
-                    tracing::warn!(
+            let coin_type = if let Some(coin_type) = coin_type_option {
+                coin_type
+            } else {
+                let event_key = event.key.as_ref().context("event must have a key")?;
+                let event_move_guid = EventGuidResource {
+                    addr: standardize_address(event_key.account_address.as_str()),
+                    creation_num: event_key.creation_number as i64,
+                };
+                // Given this mapping only contains coin type < 1000 length, we should not assume that the mapping exists.
+                // If it doesn't exist, skip.
+                match event_to_coin_type.get(&event_move_guid) {
+                    Some(coin_type) => coin_type.clone(),
+                    None => {
+                        tracing::warn!(
                         "Could not find event in resources (CoinStore), version: {}, event guid: {:?}, mapping: {:?}",
                         txn_version, event_move_guid, event_to_coin_type
                     );
-                    return Ok(None);
-                },
+                        return Ok(None);
+                    },
+                }
             };
+
             let storage_id =
-                CoinInfoType::get_storage_id(coin_type.as_str(), event_move_guid.addr.as_str());
+                CoinInfoType::get_storage_id(coin_type.as_str(), owner_address.as_str());
+
             Ok(Some(Self {
                 transaction_version: txn_version,
                 event_index,
-                owner_address: event_move_guid.addr,
+                owner_address: Some(owner_address),
                 storage_id,
-                asset_type: coin_type,
+                asset_type: Some(coin_type),
                 is_frozen: None,
                 amount: Some(amount),
                 type_: event.type_str.clone(),
@@ -195,9 +236,9 @@ impl FungibleAssetActivity {
         Self {
             transaction_version,
             event_index: v1_activity.event_index.unwrap(),
-            owner_address: v1_activity.owner_address,
+            owner_address: Some(v1_activity.owner_address),
             storage_id,
-            asset_type: v1_activity.coin_type,
+            asset_type: Some(v1_activity.coin_type),
             is_frozen: None,
             amount: Some(v1_activity.amount),
             type_: v1_activity.activity_type,

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

@@ -274,6 +274,26 @@ pub struct FrozenEvent {
     pub frozen: bool,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct DepositEventV2 {
+    pub store: String,
+    #[serde(deserialize_with = "deserialize_from_string")]
+    pub amount: BigDecimal,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct WithdrawEventV2 {
+    pub store: String,
+    #[serde(deserialize_with = "deserialize_from_string")]
+    pub amount: BigDecimal,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct FrozenEventV2 {
+    pub store: String,
+    pub frozen: bool,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum V2FungibleAssetResource {
     FungibleAssetMetadata(FungibleAssetMetadata),
@@ -338,6 +358,9 @@ pub enum FungibleAssetEvent {
     DepositEvent(DepositEvent),
     WithdrawEvent(WithdrawEvent),
     FrozenEvent(FrozenEvent),
+    DepositEventV2(DepositEventV2),
+    WithdrawEventV2(WithdrawEventV2),
+    FrozenEventV2(FrozenEventV2),
 }
 
 impl FungibleAssetEvent {
@@ -352,6 +375,15 @@ impl FungibleAssetEvent {
             "0x1::fungible_asset::FrozenEvent" => {
                 serde_json::from_str(data).map(|inner| Some(Self::FrozenEvent(inner)))
             },
+            "0x1::fungible_asset::Deposit" => {
+                serde_json::from_str(data).map(|inner| Some(Self::DepositEventV2(inner)))
+            },
+            "0x1::fungible_asset::Withdraw" => {
+                serde_json::from_str(data).map(|inner| Some(Self::WithdrawEventV2(inner)))
+            },
+            "0x1::fungible_asset::Frozen" => {
+                serde_json::from_str(data).map(|inner| Some(Self::FrozenEventV2(inner)))
+            },
             _ => Ok(None),
         }
         .context(format!(

rust/processor/src/db/postgres/migrations/2024-07-10-203513_make_asset_type_nullable/down.sql

@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE IF EXISTS fungible_asset_activities ALTER COLUMN asset_type SET NOT NULL;
+ALTER TABLE IF EXISTS fungible_asset_activities ALTER COLUMN owner_address SET NOT NULL;

rust/processor/src/db/postgres/migrations/2024-07-10-203513_make_asset_type_nullable/up.sql

@@ -0,0 +1,3 @@
+-- Your SQL goes here
+ALTER TABLE IF EXISTS fungible_asset_activities ALTER COLUMN asset_type DROP NOT NULL;
+ALTER TABLE IF EXISTS fungible_asset_activities ALTER COLUMN owner_address DROP NOT NULL;

rust/processor/src/db/postgres/schema.rs

@@ -771,11 +771,11 @@ diesel::table! {
         transaction_version -> Int8,
         event_index -> Int8,
         #[max_length = 66]
-        owner_address -> Varchar,
+        owner_address -> Nullable<Varchar>,
         #[max_length = 66]
         storage_id -> Varchar,
         #[max_length = 1000]
-        asset_type -> Varchar,
+        asset_type -> Nullable<Varchar>,
         is_frozen -> Nullable<Bool>,
         amount -> Nullable<Numeric>,
         #[sql_name = "type"]