[parquet-sdk][token_v2] migrate token_v2_pending_claims

rust/processor/src/db/common/models/mod.rs

@@ -3,3 +3,4 @@ pub mod ans_models;
 pub mod default_models;
 pub mod event_models;
 pub mod fungible_asset_models;
+pub mod token_v2_models;

rust/processor/src/db/common/models/token_v2_models/mod.rs

@@ -0,0 +1 @@
+pub mod raw_token_claims;

rust/processor/src/db/common/models/token_v2_models/raw_token_claims.rs

@@ -0,0 +1,215 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+// This is required because a diesel macro makes clippy sad
+#![allow(clippy::extra_unused_lifetimes)]
+#![allow(clippy::unused_unit)]
+
+use crate::{
+    db::postgres::models::{
+        token_models::{token_utils::TokenWriteSet, tokens::TableHandleToOwner},
+        token_v2_models::v2_token_activities::TokenActivityHelperV1,
+    },
+    utils::util::standardize_address,
+};
+use ahash::AHashMap;
+use aptos_protos::transaction::v1::{DeleteTableItem, WriteTableItem};
+use bigdecimal::{BigDecimal, Zero};
+use serde::{Deserialize, Serialize};
+
+// Map to keep track of the metadata of token offers that were claimed. The key is the token data id of the offer.
+// Potentially it'd also be useful to keep track of offers that were canceled.
+pub type TokenV1Claimed = AHashMap<String, TokenActivityHelperV1>;
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct RawCurrentTokenPendingClaim {
+    pub token_data_id_hash: String,
+    pub property_version: BigDecimal,
+    pub from_address: String,
+    pub to_address: String,
+    pub collection_data_id_hash: String,
+    pub creator_address: String,
+    pub collection_name: String,
+    pub name: String,
+    pub amount: BigDecimal,
+    pub table_handle: String,
+    pub last_transaction_version: i64,
+    pub last_transaction_timestamp: chrono::NaiveDateTime,
+    pub token_data_id: String,
+    pub collection_id: String,
+}
+
+impl Ord for RawCurrentTokenPendingClaim {
+    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
+        self.token_data_id_hash
+            .cmp(&other.token_data_id_hash)
+            .then(self.property_version.cmp(&other.property_version))
+            .then(self.from_address.cmp(&other.from_address))
+            .then(self.to_address.cmp(&other.to_address))
+    }
+}
+
+impl PartialOrd for RawCurrentTokenPendingClaim {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl RawCurrentTokenPendingClaim {
+    /// Token claim is stored in a table in the offerer's account. The key is token_offer_id (token_id + to address)
+    /// and value is token (token_id + amount)
+    pub fn from_write_table_item(
+        table_item: &WriteTableItem,
+        txn_version: i64,
+        txn_timestamp: chrono::NaiveDateTime,
+        table_handle_to_owner: &TableHandleToOwner,
+    ) -> anyhow::Result<Option<Self>> {
+        let table_item_data = table_item.data.as_ref().unwrap();
+
+        let maybe_offer = match TokenWriteSet::from_table_item_type(
+            table_item_data.key_type.as_str(),
+            &table_item_data.key,
+            txn_version,
+        )? {
+            Some(TokenWriteSet::TokenOfferId(inner)) => Some(inner),
+            _ => None,
+        };
+        if let Some(offer) = &maybe_offer {
+            let maybe_token = match TokenWriteSet::from_table_item_type(
+                table_item_data.value_type.as_str(),
+                &table_item_data.value,
+                txn_version,
+            )? {
+                Some(TokenWriteSet::Token(inner)) => Some(inner),
+                _ => None,
+            };
+            if let Some(token) = &maybe_token {
+                let table_handle = standardize_address(&table_item.handle.to_string());
+
+                let maybe_table_metadata = table_handle_to_owner.get(&table_handle);
+
+                if let Some(table_metadata) = maybe_table_metadata {
+                    let token_id = offer.token_id.clone();
+                    let token_data_id_struct = token_id.token_data_id;
+                    let collection_data_id_hash =
+                        token_data_id_struct.get_collection_data_id_hash();
+                    let token_data_id_hash = token_data_id_struct.to_hash();
+                    // Basically adding 0x prefix to the previous 2 lines. This is to be consistent with Token V2
+                    let collection_id = token_data_id_struct.get_collection_id();
+                    let token_data_id = token_data_id_struct.to_id();
+                    let collection_name = token_data_id_struct.get_collection_trunc();
+                    let name = token_data_id_struct.get_name_trunc();
+
+                    return Ok(Some(Self {
+                        token_data_id_hash,
+                        property_version: token_id.property_version,
+                        from_address: table_metadata.get_owner_address(),
+                        to_address: offer.get_to_address(),
+                        collection_data_id_hash,
+                        creator_address: token_data_id_struct.get_creator_address(),
+                        collection_name,
+                        name,
+                        amount: token.amount.clone(),
+                        table_handle,
+                        last_transaction_version: txn_version,
+                        last_transaction_timestamp: txn_timestamp,
+                        token_data_id,
+                        collection_id,
+                    }));
+                } else {
+                    tracing::warn!(
+                        transaction_version = txn_version,
+                        table_handle = table_handle,
+                        "Missing table handle metadata for TokenClaim. {:?}",
+                        table_handle_to_owner
+                    );
+                }
+            } else {
+                tracing::warn!(
+                    transaction_version = txn_version,
+                    value_type = table_item_data.value_type,
+                    value = table_item_data.value,
+                    "Expecting token as value for key = token_offer_id",
+                );
+            }
+        }
+        Ok(None)
+    }
+
+    pub fn from_delete_table_item(
+        table_item: &DeleteTableItem,
+        txn_version: i64,
+        txn_timestamp: chrono::NaiveDateTime,
+        table_handle_to_owner: &TableHandleToOwner,
+        tokens_claimed: &TokenV1Claimed,
+    ) -> anyhow::Result<Option<Self>> {
+        let table_item_data = table_item.data.as_ref().unwrap();
+
+        let maybe_offer = match TokenWriteSet::from_table_item_type(
+            table_item_data.key_type.as_str(),
+            &table_item_data.key,
+            txn_version,
+        )? {
+            Some(TokenWriteSet::TokenOfferId(inner)) => Some(inner),
+            _ => None,
+        };
+        if let Some(offer) = &maybe_offer {
+            let table_handle = standardize_address(&table_item.handle.to_string());
+            let token_data_id = offer.token_id.token_data_id.to_id();
+
+            // Try to find owner from write resources
+            let mut maybe_owner_address = table_handle_to_owner
+                .get(&table_handle)
+                .map(|table_metadata| table_metadata.get_owner_address());
+
+            // If table handle isn't in TableHandleToOwner, try to find owner from token v1 claim events
+            if maybe_owner_address.is_none() {
+                if let Some(token_claimed) = tokens_claimed.get(&token_data_id) {
+                    maybe_owner_address = token_claimed.from_address.clone();
+                }
+            }
+
+            let owner_address = maybe_owner_address.unwrap_or_else(|| {
+                panic!(
+                    "Missing table handle metadata for claim. \
+                        Version: {}, table handle for PendingClaims: {}, all metadata: {:?} \
+                        Missing token data id in token claim event. \
+                        token_data_id: {}, all token claim events: {:?}",
+                    txn_version, table_handle, table_handle_to_owner, token_data_id, tokens_claimed
+                )
+            });
+
+            let token_id = offer.token_id.clone();
+            let token_data_id_struct = token_id.token_data_id;
+            let collection_data_id_hash = token_data_id_struct.get_collection_data_id_hash();
+            let token_data_id_hash = token_data_id_struct.to_hash();
+            // Basically adding 0x prefix to the previous 2 lines. This is to be consistent with Token V2
+            let collection_id = token_data_id_struct.get_collection_id();
+            let token_data_id = token_data_id_struct.to_id();
+            let collection_name = token_data_id_struct.get_collection_trunc();
+            let name = token_data_id_struct.get_name_trunc();
+
+            return Ok(Some(Self {
+                token_data_id_hash,
+                property_version: token_id.property_version,
+                from_address: owner_address,
+                to_address: offer.get_to_address(),
+                collection_data_id_hash,
+                creator_address: token_data_id_struct.get_creator_address(),
+                collection_name,
+                name,
+                amount: BigDecimal::zero(),
+                table_handle,
+                last_transaction_version: txn_version,
+                last_transaction_timestamp: txn_timestamp,
+                token_data_id,
+                collection_id,
+            }));
+        }
+        Ok(None)
+    }
+}
+
+pub trait CurrentTokenPendingClaimConvertible {
+    fn from_raw(raw_item: RawCurrentTokenPendingClaim) -> Self;
+}

rust/processor/src/db/parquet/models/mod.rs

@@ -3,4 +3,5 @@ pub mod ans_models;
 pub mod default_models;
 pub mod event_models;
 pub mod fungible_asset_models;
+pub mod token_v2_models;
 pub mod transaction_metadata_model;

rust/processor/src/db/parquet/models/token_v2_models/mod.rs

@@ -0,0 +1 @@
+pub mod token_claims;

rust/processor/src/db/parquet/models/token_v2_models/token_claims.rs

@@ -0,0 +1,80 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+// This is required because a diesel macro makes clippy sad
+#![allow(clippy::extra_unused_lifetimes)]
+#![allow(clippy::unused_unit)]
+
+use crate::{
+    bq_analytics::generic_parquet_processor::{GetTimeStamp, HasVersion, NamedTable},
+    db::common::models::token_v2_models::raw_token_claims::{
+        CurrentTokenPendingClaimConvertible, RawCurrentTokenPendingClaim,
+    },
+};
+use allocative_derive::Allocative;
+use bigdecimal::ToPrimitive;
+use field_count::FieldCount;
+use parquet_derive::ParquetRecordWriter;
+use serde::{Deserialize, Serialize};
+
+#[derive(
+    Allocative, Clone, Debug, Default, Deserialize, FieldCount, ParquetRecordWriter, Serialize,
+)]
+pub struct CurrentTokenPendingClaim {
+    pub token_data_id_hash: String,
+    pub property_version: u64,
+    pub from_address: String,
+    pub to_address: String,
+    pub collection_data_id_hash: String,
+    pub creator_address: String,
+    pub collection_name: String,
+    pub name: String,
+    pub amount: String, // String format of BigDecimal
+    pub table_handle: String,
+    pub last_transaction_version: i64,
+    #[allocative(skip)]
+    pub last_transaction_timestamp: chrono::NaiveDateTime,
+    pub token_data_id: String,
+    pub collection_id: String,
+}
+
+impl NamedTable for CurrentTokenPendingClaim {
+    const TABLE_NAME: &'static str = "current_token_pending_claims";
+}
+
+impl HasVersion for CurrentTokenPendingClaim {
+    fn version(&self) -> i64 {
+        self.last_transaction_version
+    }
+}
+
+impl GetTimeStamp for CurrentTokenPendingClaim {
+    fn get_timestamp(&self) -> chrono::NaiveDateTime {
+        self.last_transaction_timestamp
+    }
+}
+
+impl CurrentTokenPendingClaimConvertible for CurrentTokenPendingClaim {
+    // TODO: consider returning a Result
+    fn from_raw(raw_item: RawCurrentTokenPendingClaim) -> Self {
+        Self {
+            token_data_id_hash: raw_item.token_data_id_hash,
+            property_version: raw_item
+                .property_version
+                .to_u64()
+                .expect("Failed to convert property_version to u64"),
+            from_address: raw_item.from_address,
+            to_address: raw_item.to_address,
+            collection_data_id_hash: raw_item.collection_data_id_hash,
+            creator_address: raw_item.creator_address,
+            collection_name: raw_item.collection_name,
+            name: raw_item.name,
+            amount: raw_item.amount.to_string(), // (assuming amount is non-critical)
+            table_handle: raw_item.table_handle,
+            last_transaction_version: raw_item.last_transaction_version,
+            last_transaction_timestamp: raw_item.last_transaction_timestamp,
+            token_data_id: raw_item.token_data_id,
+            collection_id: raw_item.collection_id,
+        }
+    }
+}