|
| 1 | +use ahash::AHashMap; |
| 2 | +use aptos_indexer_processor_sdk::{ |
| 3 | + aptos_protos::transaction::v1::Transaction, |
| 4 | + traits::{ |
| 5 | + async_step::AsyncRunType, AsyncStep, NamedStep, PollableAsyncRunType, PollableAsyncStep, |
| 6 | + Processable, |
| 7 | + }, |
| 8 | + types::transaction_context::{Context, TransactionContext}, |
| 9 | + utils::errors::ProcessorError, |
| 10 | +}; |
| 11 | +use async_trait::async_trait; |
| 12 | +use diesel::associations::Identifiable; |
| 13 | +use processor::db::common::models::{ |
| 14 | + coin_models::coin_supply::CoinSupply, |
| 15 | + fungible_asset_models::{ |
| 16 | + v2_fungible_asset_activities::FungibleAssetActivity, |
| 17 | + v2_fungible_asset_balances::{ |
| 18 | + CurrentFungibleAssetBalance, CurrentFungibleAssetBalancePK, |
| 19 | + CurrentUnifiedFungibleAssetBalance, FungibleAssetBalance, |
| 20 | + }, |
| 21 | + v2_fungible_metadata::FungibleAssetMetadataModel, |
| 22 | + }, |
| 23 | +}; |
| 24 | +use std::{collections::BTreeSet, time::Duration}; |
| 25 | + |
| 26 | +pub struct FungibleAssetDeduper |
| 27 | +where |
| 28 | + Self: Sized + Send + 'static, |
| 29 | +{ |
| 30 | + // The duration to collect and dedup data before releasing it |
| 31 | + poll_interval: Duration, |
| 32 | + merged_cfab: AHashMap<CurrentFungibleAssetBalancePK, CurrentFungibleAssetBalance>, |
| 33 | + contexts: BTreeSet<Context>, |
| 34 | +} |
| 35 | + |
| 36 | +impl FungibleAssetDeduper |
| 37 | +where |
| 38 | + Self: Sized + Send + 'static, |
| 39 | +{ |
| 40 | + pub fn new(poll_interval: Duration) -> Self { |
| 41 | + Self { |
| 42 | + poll_interval, |
| 43 | + merged_cfab: AHashMap::new(), |
| 44 | + contexts: BTreeSet::new(), |
| 45 | + } |
| 46 | + } |
| 47 | +} |
| 48 | + |
| 49 | +#[async_trait] |
| 50 | +impl Processable for FungibleAssetDeduper { |
| 51 | + type Input = ( |
| 52 | + Vec<FungibleAssetActivity>, |
| 53 | + Vec<FungibleAssetMetadataModel>, |
| 54 | + Vec<FungibleAssetBalance>, |
| 55 | + Vec<CurrentFungibleAssetBalance>, |
| 56 | + Vec<CurrentUnifiedFungibleAssetBalance>, |
| 57 | + Vec<CoinSupply>, |
| 58 | + ); |
| 59 | + // type Output = ( |
| 60 | + // &'static [FungibleAssetActivity], |
| 61 | + // &'static [FungibleAssetMetadataModel], |
| 62 | + // &'static [FungibleAssetBalance], |
| 63 | + // &'static [CurrentFungibleAssetBalance], |
| 64 | + // &'static [CurrentUnifiedFungibleAssetBalance], |
| 65 | + // &'static [CoinSupply], |
| 66 | + // ); |
| 67 | + type Output = ( |
| 68 | + Vec<FungibleAssetActivity>, |
| 69 | + Vec<FungibleAssetMetadataModel>, |
| 70 | + Vec<FungibleAssetBalance>, |
| 71 | + Vec<CurrentFungibleAssetBalance>, |
| 72 | + Vec<CurrentUnifiedFungibleAssetBalance>, |
| 73 | + Vec<CoinSupply>, |
| 74 | + ); |
| 75 | + type RunType = PollableAsyncRunType; |
| 76 | + |
| 77 | + async fn process( |
| 78 | + &mut self, |
| 79 | + items: TransactionContext<Self::Input>, |
| 80 | + ) -> Result<Option<TransactionContext<Self::Output>>, ProcessorError> { |
| 81 | + let (_, _, _, cfab, _, _) = &items.data[0]; |
| 82 | + |
| 83 | + // Update transaction contexts |
| 84 | + for context in items.context.iter() { |
| 85 | + self.contexts.insert(context.clone()); |
| 86 | + } |
| 87 | + |
| 88 | + // Dedup |
| 89 | + for balance in cfab.iter() { |
| 90 | + self.merged_cfab |
| 91 | + .insert(balance.id().to_string(), balance.clone()); |
| 92 | + } |
| 93 | + |
| 94 | + Ok(None) |
| 95 | + } |
| 96 | +} |
| 97 | + |
| 98 | +#[async_trait] |
| 99 | +impl PollableAsyncStep for FungibleAssetDeduper { |
| 100 | + fn poll_interval(&self) -> Duration { |
| 101 | + self.poll_interval |
| 102 | + } |
| 103 | + |
| 104 | + async fn poll( |
| 105 | + &mut self, |
| 106 | + ) -> Result<Option<Vec<TransactionContext<Self::Output>>>, ProcessorError> { |
| 107 | + // let faa: &[FungibleAssetActivity] = &[]; |
| 108 | + // let fam: &[FungibleAssetMetadataModel] = &[]; |
| 109 | + // let fab: &[FungibleAssetBalance] = &[]; |
| 110 | + // let cfab: &[CurrentFungibleAssetBalance] = &[]; |
| 111 | + // let cufab: &[CurrentUnifiedFungibleAssetBalance] = &[]; |
| 112 | + // let cs: &[CoinSupply] = &[]; |
| 113 | + // let cfab: AHashMap<String, CurrentFungibleAssetBalance> = |
| 114 | + // std::mem::take(&mut self.merged_cfab); |
| 115 | + // let cfab = cfab.values().cloned().collect::<Vec<_>>(); |
| 116 | + let transaction_context = TransactionContext::new_with_contexts( |
| 117 | + vec![( |
| 118 | + vec![], |
| 119 | + vec![], |
| 120 | + vec![], |
| 121 | + std::mem::take(&mut self.merged_cfab) |
| 122 | + .values() |
| 123 | + .cloned() |
| 124 | + .collect::<Vec<_>>(), |
| 125 | + vec![], |
| 126 | + vec![], |
| 127 | + )], |
| 128 | + self.contexts.clone(), |
| 129 | + ); |
| 130 | + Ok(Some(vec![transaction_context])) |
| 131 | + } |
| 132 | +} |
| 133 | + |
| 134 | +impl NamedStep for FungibleAssetDeduper { |
| 135 | + fn name(&self) -> String { |
| 136 | + "FungibleAssetDeduper".to_string() |
| 137 | + } |
| 138 | +} |
0 commit comments