event v2 indexing.

python/processors/nft_orderbooks/parsers/okx_parser.py

@@ -19,6 +19,8 @@
         "okx_listing_utils::CancelListingEvent",
     ]
 )
+DEPOSITE_EVENT_V1 = "0x3::token::DepositEvent"
+DEPOSITE_EVENT_V2 = "0x3::token::Deposit"
 
 
 def parse_marketplace_events(
@@ -141,8 +143,15 @@ def parse_marketplace_events(
 def get_token_data_from_deposit_events(user_transaction) -> Dict[str, TokenDataIdType]:
     # Extract deposit events, which contain token metadata
     deposit_events: Dict[str, TokenDataIdType] = {}
-    for event in user_transaction.events:
-        if event.type_str != "0x3::token::DepositEvent":
+    for idx, event in enumerate(user_transaction.events):
+        if event.type_str != DEPOSITE_EVENT_V1 and event.type_str != DEPOSITE_EVENT_V2:
+            continue
+        # Current event is either DEPOSITE_EVENT_V1 or DEPOSITE_EVENT_V2.
+        if (
+            idx > 0
+            # skip if prior event is V2 deposit event.
+            and user_transaction.events[idx - 1].type_str == DEPOSITE_EVENT_V2
+        ):
             continue
         account_address = standardize_address(event_utils.get_account_address(event))
         data = json.loads(event.data)

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

@@ -89,20 +89,18 @@ impl FungibleAssetActivity {
                     None,
                 ),
                 FungibleAssetEvent::WithdrawEventV2(inner) => (
-                    inner.store.get_reference_address(),
+                    standardize_address(&inner.store),
                     None,
                     Some(inner.amount.clone()),
                 ),
                 FungibleAssetEvent::DepositEventV2(inner) => (
-                    inner.store.get_reference_address(),
+                    standardize_address(&inner.store),
                     None,
                     Some(inner.amount.clone()),
                 ),
-                FungibleAssetEvent::FrozenEventV2(inner) => (
-                    inner.store.get_reference_address(),
-                    Some(inner.frozen),
-                    None,
-                ),
+                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

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

@@ -206,21 +206,21 @@ pub struct FrozenEvent {
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct DepositEventV2 {
-    pub store: ResourceReference,
+    pub store: String,
     #[serde(deserialize_with = "deserialize_from_string")]
     pub amount: BigDecimal,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct WithdrawEventV2 {
-    pub store: ResourceReference,
+    pub store: String,
     #[serde(deserialize_with = "deserialize_from_string")]
     pub amount: BigDecimal,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct FrozenEventV2 {
-    pub store: ResourceReference,
+    pub store: String,
     pub frozen: bool,
 }
 

rust/processor/src/models/mod.rs

@@ -23,5 +23,51 @@ pub(crate) fn should_skip(index: usize, event: &Event, events: &[Event]) -> bool
     let len = event.type_str.len();
     index > 0
         && event.type_str.ends_with("Event")
-        && events[index - 1].type_str[..len - 5] == event.type_str[..len - 5]
+        && events[index - 1]
+            .type_str
+            .starts_with(&event.type_str[..len - 5])
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    // Tests are to make sure
+    // - only when previous processed event is event v2, current event v1 can be skipped
+
+    #[test]
+    fn test_should_skip_intended() {
+        let event1 = Event {
+            type_str: "0x1::coin::Deposit<0x1::aptos_coin::AptosCoin>".to_string(),
+            ..Event::default()
+        };
+        let event2 = Event {
+            type_str: "0x1::coin::DepositEvent".to_string(),
+            ..Event::default()
+        };
+        let events = vec![event1, event2];
+        assert!(!should_skip(0, &events[0], &events));
+        assert!(should_skip(1, &events[1], &events));
+    }
+
+    #[test]
+    fn test_should_not_break_for_length() {
+        let events = [
+            Event {
+                type_str: "Test0000000000000000000000000000000Event".to_string(),
+                ..Event::default()
+            },
+            Event {
+                type_str: "TEvent".to_string(),
+                ..Event::default()
+            },
+            Event {
+                type_str: "Test0000000000000000000000000000000Event".to_string(),
+                ..Event::default()
+            },
+        ];
+        assert!(!should_skip(0, &events[0], &events));
+        // Note, it is intentional that the second event is skipped.
+        assert!(should_skip(1, &events[1], &events));
+        assert!(!should_skip(2, &events[2], &events));
+    }
 }

rust/processor/src/models/stake_models/delegator_activities.rs

@@ -5,6 +5,7 @@
 
 use super::stake_utils::StakeEvent;
 use crate::{
+    models::should_skip,
     schema::delegated_staking_activities,
     utils::{
         counters::PROCESSOR_UNKNOWN_TYPE_COUNT,
@@ -57,6 +58,9 @@ impl DelegatedStakingActivity {
             if let Some(staking_event) =
                 StakeEvent::from_event(event.type_str.as_str(), &event.data, txn_version)?
             {
+                if should_skip(index, event, events) {
+                    continue;
+                }
                 let activity = match staking_event {
                     StakeEvent::AddStakeEvent(inner) => DelegatedStakingActivity {
                         transaction_version: txn_version,

rust/processor/src/models/stake_models/proposal_votes.rs

@@ -51,12 +51,12 @@ impl ProposalVote {
 
         if let TxnData::User(user_txn) = txn_data {
             for (index, event) in user_txn.events.iter().enumerate() {
-                if should_skip(index, event, &user_txn.events) {
-                    continue;
-                };
                 if let Some(StakeEvent::GovernanceVoteEvent(ev)) =
                     StakeEvent::from_event(event.type_str.as_str(), &event.data, txn_version)?
                 {
+                    if should_skip(index, event, &user_txn.events) {
+                        continue;
+                    };
                     proposal_votes.push(Self {
                         transaction_version: txn_version,
                         proposal_id: ev.proposal_id as i64,

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

@@ -584,7 +584,7 @@ impl V2TokenEvent {
             "0x4::collection::MintEvent" => {
                 serde_json::from_str(data).map(|inner| Some(Self::MintEvent(inner)))
             },
-            "0x4::token::MutationEvent" => {
+            "0x4::token::MutationEvent" | "0x4::token::Mutation" => {
                 serde_json::from_str(data).map(|inner| Some(Self::TokenMutationEvent(inner)))
             },
             "0x4::collection::Burn" => {
@@ -593,7 +593,7 @@ impl V2TokenEvent {
             "0x4::collection::BurnEvent" => {
                 serde_json::from_str(data).map(|inner| Some(Self::BurnEvent(inner)))
             },
-            "0x1::object::TransferEvent" => {
+            "0x1::object::TransferEvent" | "0x1::object::Transfer" => {
                 serde_json::from_str(data).map(|inner| Some(Self::TransferEvent(inner)))
             },
             _ => Ok(None),

rust/processor/src/processors/fungible_asset_processor.rs

@@ -18,7 +18,6 @@ use crate::{
             ObjectAggregatedData, ObjectAggregatedDataMapping, ObjectWithMetadata,
         },
         should_skip,
-        token_v2_models::v2_token_utils::TokenV2,
     },
     schema,
     utils::{

rust/processor/src/processors/token_v2_processor.rs

@@ -8,6 +8,7 @@ use crate::{
         object_models::v2_object_utils::{
             ObjectAggregatedData, ObjectAggregatedDataMapping, ObjectWithMetadata,
         },
+        should_skip,
         token_models::tokens::{TableHandleToOwner, TableMetadataForToken},
         token_v2_models::{
             v2_collections::{CollectionV2, CurrentCollectionV2, CurrentCollectionV2PK},
@@ -721,6 +722,9 @@ async fn parse_v2_token(
             // This needs to be here because we need the metadata above for token activities
             // and burn / transfer events need to come before the next section
             for (index, event) in user_txn.events.iter().enumerate() {
+                if should_skip(index, event, user_txn.events.as_slice()) {
+                    continue;
+                }
                 if let Some(burn_event) = Burn::from_event(event, txn_version).unwrap() {
                     tokens_burned.insert(burn_event.get_token_address(), burn_event);
                 }