Migrate FA processor to SDK

rust/Cargo.toml

@@ -30,8 +30,8 @@ testing-transactions = { path = "testing-transactions" }
 
 ahash = { version = "0.8.7", features = ["serde"] }
 anyhow = "1.0.86"
-aptos-indexer-processor-sdk = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "9ecd252ccff53023664562001dd04c2886488c0d" }
-aptos-indexer-processor-sdk-server-framework = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "9ecd252ccff53023664562001dd04c2886488c0d" }
+aptos-indexer-processor-sdk = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "95d76e07dd66a20a0e50a9e6c559885bff7ab52b" }
+aptos-indexer-processor-sdk-server-framework = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "95d76e07dd66a20a0e50a9e6c559885bff7ab52b" }
 aptos-protos = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "5c48aee129b5a141be2792ffa3d9bd0a1a61c9cb" }
 aptos-system-utils = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "202bdccff2b2d333a385ae86a4fcf23e89da9f62" }
 aptos-indexer-test-transactions = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "7246f0536d599789d7dd5e9fb776554abbe11eac" }

rust/processor/src/processors/fungible_asset_processor.rs

@@ -169,7 +169,7 @@ async fn insert_to_db(
     Ok(())
 }
 
-fn insert_fungible_asset_activities_query(
+pub fn insert_fungible_asset_activities_query(
     items_to_insert: Vec<FungibleAssetActivity>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -186,7 +186,7 @@ fn insert_fungible_asset_activities_query(
     )
 }
 
-fn insert_fungible_asset_metadata_query(
+pub fn insert_fungible_asset_metadata_query(
     items_to_insert: Vec<FungibleAssetMetadataModel>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -222,7 +222,7 @@ fn insert_fungible_asset_metadata_query(
     )
 }
 
-fn insert_fungible_asset_balances_query(
+pub fn insert_fungible_asset_balances_query(
     items_to_insert: Vec<FungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -239,7 +239,7 @@ fn insert_fungible_asset_balances_query(
     )
 }
 
-fn insert_current_fungible_asset_balances_query(
+pub fn insert_current_fungible_asset_balances_query(
     items_to_insert: Vec<CurrentFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -269,7 +269,7 @@ fn insert_current_fungible_asset_balances_query(
     )
 }
 
-fn insert_current_unified_fungible_asset_balances_v1_query(
+pub fn insert_current_unified_fungible_asset_balances_v1_query(
     items_to_insert: Vec<CurrentUnifiedFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -298,7 +298,7 @@ fn insert_current_unified_fungible_asset_balances_v1_query(
     )
 }
 
-fn insert_current_unified_fungible_asset_balances_v2_query(
+pub fn insert_current_unified_fungible_asset_balances_v2_query(
     items_to_insert: Vec<CurrentUnifiedFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -328,7 +328,7 @@ fn insert_current_unified_fungible_asset_balances_v2_query(
     )
 }
 
-fn insert_coin_supply_query(
+pub fn insert_coin_supply_query(
     items_to_insert: Vec<CoinSupply>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -451,7 +451,7 @@ impl ProcessorTrait for FungibleAssetProcessor {
 }
 
 /// V2 coin is called fungible assets and this flow includes all data from V1 in coin_processor
-async fn parse_v2_coin(
+pub async fn parse_v2_coin(
     transactions: &[Transaction],
 ) -> (
     Vec<FungibleAssetActivity>,

rust/processor/src/worker.rs

@@ -114,6 +114,18 @@ bitflags! {
     }
 }
 
+impl TableFlags {
+    pub fn from_set(set: &HashSet<String>) -> Self {
+        let mut flags = TableFlags::empty();
+        for table in set {
+            if let Some(flag) = TableFlags::from_name(table) {
+                flags |= flag;
+            }
+        }
+        flags
+    }
+}
+
 pub struct Worker {
     pub db_pool: ArcDbPool,
     pub processor_config: ProcessorConfig,

rust/sdk-processor/src/config/indexer_processor_config.rs

@@ -2,18 +2,25 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use super::{db_config::DbConfig, processor_config::ProcessorConfig};
-use crate::processors::events_processor::EventsProcessor;
+use crate::processors::{
+    events_processor::EventsProcessor, fungible_asset_processor::FungibleAssetProcessor,
+};
 use anyhow::Result;
 use aptos_indexer_processor_sdk::aptos_indexer_transaction_stream::TransactionStreamConfig;
 use aptos_indexer_processor_sdk_server_framework::RunnableConfig;
 use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct IndexerProcessorConfig {
     pub processor_config: ProcessorConfig,
     pub transaction_stream_config: TransactionStreamConfig,
     pub db_config: DbConfig,
+
+    // String vector for deprecated tables to skip db writes
+    #[serde(default)]
+    pub deprecated_tables: HashSet<String>,
 }
 
 #[async_trait::async_trait]
@@ -24,6 +31,10 @@ impl RunnableConfig for IndexerProcessorConfig {
                 let events_processor = EventsProcessor::new(self.clone()).await?;
                 events_processor.run_processor().await
             },
+            ProcessorConfig::FungibleAssetProcessor(_) => {
+                let fungible_asset_processor = FungibleAssetProcessor::new(self.clone()).await?;
+                fungible_asset_processor.run_processor().await
+            },
         }
     }
 

rust/sdk-processor/src/config/processor_config.rs

@@ -1,4 +1,6 @@
-use crate::processors::events_processor::EventsProcessorConfig;
+use crate::processors::{
+    events_processor::EventsProcessorConfig, fungible_asset_processor::FungibleAssetProcessorConfig,
+};
 use serde::{Deserialize, Serialize};
 
 /// This enum captures the configs for all the different processors that are defined.
@@ -35,6 +37,7 @@ use serde::{Deserialize, Serialize};
 )]
 pub enum ProcessorConfig {
     EventsProcessor(EventsProcessorConfig),
+    FungibleAssetProcessor(FungibleAssetProcessorConfig),
 }
 
 impl ProcessorConfig {

rust/sdk-processor/src/processors/events_processor.rs

@@ -94,8 +94,15 @@ impl EventsProcessor {
             .await?;
         check_or_update_chain_id(grpc_chain_id as i64, self.db_pool.clone()).await?;
 
-        let ProcessorConfig::EventsProcessor(events_processor_config) =
-            self.config.processor_config;
+        let events_processor_config = match self.config.processor_config {
+            ProcessorConfig::EventsProcessor(events_processor_config) => events_processor_config,
+            _ => {
+                return Err(anyhow::anyhow!(
+                    "Invalid processor config for EventsProcessor: {:?}",
+                    self.config.processor_config
+                ))
+            },
+        };
         let channel_size = events_processor_config.channel_size;
 
         // Define processor steps
@@ -125,12 +132,9 @@ impl EventsProcessor {
         loop {
             match buffer_receiver.recv().await {
                 Ok(txn_context) => {
-                    if txn_context.data.is_empty() {
-                        continue;
-                    }
                     debug!(
                         "Finished processing events from versions [{:?}, {:?}]",
-                        txn_context.start_version, txn_context.end_version,
+                        txn_context.metadata.start_version, txn_context.metadata.end_version,
                     );
                 },
                 Err(e) => {