// file: kb_lib/src/non_trade_event_materialization.rs //! Materialization of useful non-trade DEX events. //! //! This service persists liquidity, pool lifecycle, fee, reward and pool //! administration events from already decoded DEX events. It deliberately does //! not feed trade, metric or candle materialization. /// Result of non-trade event materialization for one transaction. #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] pub struct NonTradeEventMaterializationResult { /// Number of liquidity events inserted or refreshed. pub liquidity_event_count: usize, /// Number of pool lifecycle events inserted or refreshed. pub pool_lifecycle_event_count: usize, /// Number of fee events inserted or refreshed. pub fee_event_count: usize, /// Number of reward events inserted or refreshed. pub reward_event_count: usize, /// Number of pool administration events inserted or refreshed. pub pool_admin_event_count: usize, /// Number of orderbook or limit-order events inserted or refreshed. pub orderbook_event_count: usize, /// Number of launch-surface specific events inserted or refreshed. pub launch_event_count: usize, /// Number of token-account events inserted or refreshed. pub token_account_event_count: usize, } /// Materializes useful non-trade decoded DEX events. #[derive(Debug, Clone)] pub struct NonTradeEventMaterializationService { database: std::sync::Arc, } #[derive(Debug, Clone)] struct NonTradeDecodedEventContext { dex_id: std::option::Option, pool_id: std::option::Option, pair_id: std::option::Option, pair: std::option::Option, } impl NonTradeEventMaterializationService { /// Creates a new non-trade event materialization service. pub fn new(database: std::sync::Arc) -> Self { return Self { database }; } /// Materializes useful non-trade events for one persisted transaction signature. pub async fn record_transaction_by_signature( &self, signature: &str, ) -> Result { let transaction_result = crate::query_chain_transactions_get_by_signature(self.database.as_ref(), signature) .await; let transaction_option = match transaction_result { Ok(transaction_option) => transaction_option, Err(error) => return Err(error), }; let transaction = match transaction_option { Some(transaction) => transaction, None => { return Err(crate::Error::InvalidState(format!( "cannot materialize non-trade events for unknown transaction '{}'", signature ))); }, }; if transaction_has_effective_error(&transaction) { tracing::debug!( signature = %transaction.signature, "skipping non-trade materialization for failed transaction" ); return Ok(crate::NonTradeEventMaterializationResult::default()); } let transaction_id = match transaction.id { Some(transaction_id) => transaction_id, None => { return Err(crate::Error::InvalidState(format!( "transaction '{}' has no internal id", transaction.signature ))); }, }; let decoded_events_result = crate::query_dex_decoded_events_list_by_transaction_id( self.database.as_ref(), transaction_id, ) .await; let decoded_events = match decoded_events_result { Ok(decoded_events) => decoded_events, Err(error) => return Err(error), }; let mut result = crate::NonTradeEventMaterializationResult::default(); for decoded_event in &decoded_events { let payload_result = serde_json::from_str::(decoded_event.payload_json.as_str()); let payload = match payload_result { Ok(payload) => payload, Err(error) => { tracing::warn!( signature = %transaction.signature, event_kind = %decoded_event.event_kind, error = %error, "skipping non-trade materialization for invalid decoded payload" ); continue; }, }; if crate::is_dex_pool_lifecycle_event_kind(decoded_event.event_kind.as_str()) { let cleanup_result = self.delete_stale_pool_admin_event_for_lifecycle(decoded_event).await; match cleanup_result { Ok(_) => {}, Err(error) => return Err(error), } } if crate::is_dex_liquidity_event_kind(decoded_event.event_kind.as_str()) && !decoded_event.event_kind.ends_with(".lp_change_event") { let materialized = self .materialize_liquidity_event( &transaction, transaction_id, decoded_event, &payload, ) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.liquidity_event_count += 1; } }, Err(error) => return Err(error), } } if crate::is_dex_pool_lifecycle_event_kind(decoded_event.event_kind.as_str()) { let materialized = self .materialize_pool_lifecycle_event(&transaction, transaction_id, decoded_event) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.pool_lifecycle_event_count += 1; } }, Err(error) => return Err(error), } } if crate::is_dex_fee_event_kind(decoded_event.event_kind.as_str()) { let materialized = self .materialize_fee_event(&transaction, transaction_id, decoded_event, &payload) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.fee_event_count += 1; } }, Err(error) => return Err(error), } } if crate::is_dex_reward_event_kind(decoded_event.event_kind.as_str()) { let materialized = self .materialize_reward_event(&transaction, transaction_id, decoded_event, &payload) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.reward_event_count += 1; } }, Err(error) => return Err(error), } } if crate::is_dex_admin_event_kind(decoded_event.event_kind.as_str()) && !crate::is_dex_pool_lifecycle_event_kind(decoded_event.event_kind.as_str()) { let materialized = self .materialize_pool_admin_event( &transaction, transaction_id, decoded_event, &payload, ) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.pool_admin_event_count += 1; } }, Err(error) => return Err(error), } } if crate::is_dex_orderbook_event_kind(decoded_event.event_kind.as_str()) { let materialized = self .materialize_orderbook_event( &transaction, transaction_id, decoded_event, &payload, ) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.orderbook_event_count += 1; } }, Err(error) => return Err(error), } } if is_token_account_event_materializable(decoded_event.event_kind.as_str()) { let materialized = self .materialize_token_account_event( &transaction, transaction_id, decoded_event, &payload, ) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.token_account_event_count += 1; } }, Err(error) => return Err(error), } } if is_launchpad_launch_event_materializable(decoded_event.event_kind.as_str()) { let materialized = self .materialize_launch_event(&transaction, transaction_id, decoded_event, &payload) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.launch_event_count += 1; } }, Err(error) => return Err(error), } } } for decoded_event in &decoded_events { if !decoded_event.event_kind.ends_with(".lp_change_event") { continue; } let payload_result = serde_json::from_str::(decoded_event.payload_json.as_str()); let payload = match payload_result { Ok(payload) => payload, Err(error) => { tracing::warn!( signature = %transaction.signature, event_kind = %decoded_event.event_kind, error = %error, "skipping postponed lp_change_event materialization for invalid decoded payload" ); continue; }, }; let materialized = self .materialize_liquidity_event(&transaction, transaction_id, decoded_event, &payload) .await; match materialized { Ok(was_materialized) => { if was_materialized { result.liquidity_event_count += 1; } }, Err(error) => return Err(error), } } return Ok(result); } async fn materialize_pool_lifecycle_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let dto = crate::PoolLifecycleEventDto::new( transaction_id, Some(decoded_event_id), context.dex_id, context.pool_id, context.pair_id, transaction.signature.clone(), transaction.slot, decoded_event.protocol_name.clone(), decoded_event.program_id.clone(), decoded_event.event_kind.clone(), decoded_event.pool_account.clone(), decoded_event.token_a_mint.clone(), decoded_event.token_b_mint.clone(), decoded_event.payload_json.clone(), ); let upsert_result = crate::query_pool_lifecycle_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn materialize_fee_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let actor_wallet = extract_first_string( payload, &[ "actorWallet", "actor_wallet", "receiver", "recipient", "owner", "payer", "authority", "user", ], ); let fee_token_mint = extract_first_string( payload, &[ "feeTokenMint", "fee_token_mint", "tokenMint", "token_mint", "mint", "quoteMint", "quote_mint", ], ); let fee_amount_raw = extract_first_amount_string( payload, &[ "feeAmountRaw", "fee_amount_raw", "feeAmount", "fee_amount", "protocolFeeAmount", "protocol_fee_amount", "fundFeeAmount", "fund_fee_amount", "creatorFeeAmount", "creator_fee_amount", "amount0RequestedRaw", "amount_0_requested_raw", "amount1RequestedRaw", "amount_1_requested_raw", "tokenAAmount", "tokenBAmount", "amount", ], ); let dto = crate::FeeEventDto::new( transaction_id, Some(decoded_event_id), context.dex_id, context.pool_id, context.pair_id, transaction.signature.clone(), transaction.slot, decoded_event.protocol_name.clone(), decoded_event.program_id.clone(), decoded_event.event_kind.clone(), decoded_event.pool_account.clone(), actor_wallet, fee_token_mint, fee_amount_raw, decoded_event.payload_json.clone(), ); let upsert_result = crate::query_fee_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn materialize_reward_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let actor_wallet = extract_first_string( payload, &[ "actorWallet", "actor_wallet", "receiver", "recipient", "owner", "payer", "authority", "user", ], ); let reward_token_mint = extract_first_string( payload, &["rewardTokenMint", "reward_token_mint", "tokenMint", "token_mint", "mint"], ); let reward_amount_raw = extract_first_amount_string( payload, &[ "rewardAmountRaw", "reward_amount_raw", "rewardAmount", "reward_amount", "emissionAmount", "emission_amount", "amount", ], ); let dto = crate::RewardEventDto::new( transaction_id, Some(decoded_event_id), context.dex_id, context.pool_id, context.pair_id, transaction.signature.clone(), transaction.slot, decoded_event.protocol_name.clone(), decoded_event.program_id.clone(), decoded_event.event_kind.clone(), decoded_event.pool_account.clone(), actor_wallet, reward_token_mint, reward_amount_raw, decoded_event.payload_json.clone(), ); let upsert_result = crate::query_reward_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn delete_stale_pool_admin_event_for_lifecycle( &self, decoded_event: &crate::DexDecodedEventDto, ) -> Result<(), crate::Error> { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(()), }; let delete_result = crate::query_pool_admin_events_delete_by_decoded_event_id( self.database.as_ref(), decoded_event_id, ) .await; let deleted_count = match delete_result { Ok(deleted_count) => deleted_count, Err(error) => return Err(error), }; if deleted_count > 0 { tracing::debug!( decoded_event_id = decoded_event_id, event_kind = %decoded_event.event_kind, deleted_count = deleted_count, "removed stale pool admin materialization for lifecycle event" ); } return Ok(()); } async fn materialize_pool_admin_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let actor_wallet = extract_first_string( payload, &["actorWallet", "actor_wallet", "authority", "admin", "owner", "payer", "user"], ); let admin_action = match extract_first_string( payload, &["adminAction", "admin_action", "action", "configAction", "config_action"], ) { Some(admin_action) => Some(admin_action), None => Some(decoded_event.event_kind.clone()), }; let dto = crate::PoolAdminEventDto::new( transaction_id, Some(decoded_event_id), context.dex_id, context.pool_id, context.pair_id, transaction.signature.clone(), transaction.slot, decoded_event.protocol_name.clone(), decoded_event.program_id.clone(), decoded_event.event_kind.clone(), decoded_event.pool_account.clone(), actor_wallet, admin_action, decoded_event.payload_json.clone(), ); let upsert_result = crate::query_pool_admin_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn materialize_token_account_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let token_account = match extract_first_string( payload, &[ "supportMintAssociated", "support_mint_associated", "tokenAccount", "token_account", ], ) { Some(token_account) => Some(token_account), None => extract_account_string(payload, 2), }; let token_mint = match extract_first_string(payload, &["tokenMint", "token_mint", "mint"]) { Some(token_mint) => Some(token_mint), None => extract_account_string(payload, 1), }; let owner_wallet = match extract_first_string(payload, &["owner", "authority", "payer"]) { Some(owner_wallet) => Some(owner_wallet), None => extract_account_string(payload, 0), }; let instruction_index = match extract_first_i64( payload, &["instructionIndex", "instruction_index", "outerInstructionIndex"], ) { Some(instruction_index) => instruction_index, None => decoded_event_id, }; let dto = crate::TokenAccountEventDto::new( Some(transaction_id), Some(decoded_event_id), context.dex_id, context.pool_id, context.pair_id, transaction.signature.clone(), instruction_index, transaction.slot, decoded_event.protocol_name.clone(), Some(decoded_event.program_id.clone()), decoded_event.event_kind.clone(), token_account, token_mint, owner_wallet, Some("create_support_mint_associated".to_string()), Some(decoded_event.payload_json.clone()), ); let upsert_result = crate::query_token_account_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn materialize_launch_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let actor_wallet = extract_first_string( payload, &[ "actorWallet", "actor_wallet", "beneficiary", "owner", "payer", "authority", "user", "creator", "platformVestingWallet", "platform_vesting_wallet", ], ); let event_role = launchpad_launch_event_role(decoded_event.event_kind.as_str()); let related_account = extract_first_string( payload, &[ "platformVestingRecord", "platform_vesting_record", "platformGlobalAccess", "platform_global_access", "vestingRecord", "vesting_record", "config", "platformConfig", "platform_config", "poolState", "pool_state", "poolAccount", ], ); let related_mint = extract_first_string( payload, &[ "baseMint", "base_mint", "baseTokenMint", "base_token_mint", "tokenMint", "token_mint", "mint", ], ); let slot_i64 = match transaction.slot { Some(slot) => { let converted = i64::try_from(slot); match converted { Ok(converted) => Some(converted), Err(error) => { return Err(crate::Error::Db(format!( "cannot convert launch event slot '{}' to i64: {}", slot, error ))); }, } }, None => None, }; let input = crate::LaunchEventUpsertInput { transaction_id, decoded_event_id, dex_id: context.dex_id, pool_id: context.pool_id, pair_id: context.pair_id, signature: transaction.signature.clone(), slot: slot_i64, protocol_name: decoded_event.protocol_name.clone(), program_id: decoded_event.program_id.clone(), event_kind: decoded_event.event_kind.clone(), pool_account: decoded_event.pool_account.clone(), actor_wallet, event_role, related_account, related_mint, payload_json: payload.clone(), }; let upsert_result = crate::query_launch_events_upsert(self.database.as_ref(), &input).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn materialize_liquidity_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self .resolve_liquidity_context(transaction, transaction_id, decoded_event) .await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let dex_id = match context.dex_id { Some(dex_id) => dex_id, None => { let annotate_result = self .annotate_decoded_event_payload( decoded_event, "skipLiquidityReason", "missing_dex_catalog", ) .await; if let Err(error) = annotate_result { return Err(error); } return Ok(false); }, }; let pool_id = match context.pool_id { Some(pool_id) => pool_id, None => { let annotate_result = self .annotate_decoded_event_payload( decoded_event, "skipLiquidityReason", "missing_pool_catalog", ) .await; if let Err(error) = annotate_result { return Err(error); } return Ok(false); }, }; let pair = match context.pair { Some(pair) => pair, None => { let annotate_result = self .annotate_decoded_event_payload( decoded_event, "skipLiquidityReason", "missing_pair_catalog", ) .await; if let Err(error) = annotate_result { return Err(error); } return Ok(false); }, }; let pair_id = match pair.id { Some(pair_id) => Some(pair_id), None => None, }; let event_kind = if crate::is_dex_position_open_event_kind(decoded_event.event_kind.as_str()) { crate::LiquidityEventKind::PositionOpen } else if crate::is_dex_position_close_event_kind(decoded_event.event_kind.as_str()) { crate::LiquidityEventKind::PositionClose } else if decoded_event.event_kind.ends_with(".lp_change_event") { let change_type = extract_first_u64(payload, &["changeType", "change_type"]); match change_type { Some(1) => crate::LiquidityEventKind::Remove, _ => crate::LiquidityEventKind::Add, } } else if crate::is_dex_liquidity_remove_event_kind(decoded_event.event_kind.as_str()) { crate::LiquidityEventKind::Remove } else { crate::LiquidityEventKind::Add }; let actor_wallet = extract_first_string( payload, &[ "actorWallet", "actor_wallet", "user", "owner", "payer", "authority", "liquidityProvider", "liquidity_provider", ], ); let base_amount = extract_first_amount_string( payload, &[ "baseAmountRaw", "base_amount_raw", "baseAmount", "base_amount", "amountBase", "amount_base", "tokenAAmount", "token_a_amount", "token0AmountRaw", "token_0_amount_raw", "token0Amount", "token_0_amount", "amount0Raw", "amount0_raw", "amount0", "amount0RequestedRaw", "amount_0_requested_raw", "amountA", "amount_a", ], ); let quote_amount = extract_first_amount_string( payload, &[ "quoteAmountRaw", "quote_amount_raw", "quoteAmount", "quote_amount", "amountQuote", "amount_quote", "tokenBAmount", "token_b_amount", "token1AmountRaw", "token_1_amount_raw", "token1Amount", "token_1_amount", "amount1Raw", "amount1_raw", "amount1", "amount1RequestedRaw", "amount_1_requested_raw", "amountB", "amount_b", ], ); let lp_amount = extract_first_amount_string( payload, &[ "lpAmountRaw", "lp_amount_raw", "lpAmount", "lp_amount", "liquidity", "liquidity_raw", "liquidityRaw", "liquidityAmount", "liquidity_amount", ], ); let amounts_are_complete = base_amount.is_some() && quote_amount.is_some(); let base_amount_value = match base_amount { Some(base_amount_value) => base_amount_value, None => "0".to_string(), }; let quote_amount_value = match quote_amount { Some(quote_amount_value) => quote_amount_value, None => "0".to_string(), }; let dto = crate::LiquidityEventDto::new( dex_id, pool_id, pair_id, transaction.signature.clone(), decoded_event_id, transaction.slot, event_kind, actor_wallet, pair.base_token_id, pair.quote_token_id, None, base_amount_value, quote_amount_value, lp_amount, ) .with_decoded_event_metadata( Some(transaction_id), Some(decoded_event_id), Some(decoded_event.program_id.clone()), Some(decoded_event.event_kind.clone()), Some(decoded_event.payload_json.clone()), amounts_are_complete, ); let upsert_result = crate::query_liquidity_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn materialize_orderbook_event( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, payload: &serde_json::Value, ) -> Result { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(false), }; let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let order_action = normalize_orderbook_action(decoded_event.event_kind.as_str()); let actor_wallet = extract_first_string( payload, &["actorWallet", "actor_wallet", "owner", "authority", "payer", "user"], ); let order_account = match extract_first_string( payload, &["orderAccount", "order_account", "limitOrder", "limit_order", "order"], ) { Some(order_account) => Some(order_account), None => fallback_order_account(decoded_event.event_kind.as_str(), payload), }; let amount_raw = extract_first_amount_string( payload, &[ "amountRaw", "amount_raw", "amount", "decreasedAmountRaw", "decreased_amount_raw", "decreasedAmount", "increasedAmountRaw", "increased_amount_raw", "increasedAmount", ], ); let amount_min_raw = extract_first_amount_string( payload, &["amountMinRaw", "amount_min_raw", "amountMin", "amount_min"], ); let tick_index = extract_first_i64(payload, &["tickIndex", "tick_index"]); let zero_for_one = extract_first_bool(payload, &["zeroForOne", "zero_for_one"]); let dto = crate::OrderbookEventDto::new( transaction_id, Some(decoded_event_id), context.dex_id, context.pool_id, context.pair_id, transaction.signature.clone(), transaction.slot, decoded_event.protocol_name.clone(), decoded_event.program_id.clone(), decoded_event.event_kind.clone(), order_action, decoded_event.pool_account.clone(), decoded_event.market_account.clone(), actor_wallet, order_account, decoded_event.token_a_mint.clone(), decoded_event.token_b_mint.clone(), amount_raw, amount_min_raw, tick_index, zero_for_one, decoded_event.payload_json.clone(), ); let upsert_result = crate::query_orderbook_events_upsert(self.database.as_ref(), &dto).await; match upsert_result { Ok(_) => return Ok(true), Err(error) => return Err(error), } } async fn annotate_decoded_event_payload( &self, decoded_event: &crate::DexDecodedEventDto, reason_key: &str, reason_value: &str, ) -> Result<(), crate::Error> { let decoded_event_id = match decoded_event.id { Some(decoded_event_id) => decoded_event_id, None => return Ok(()), }; let payload_result = serde_json::from_str::( decoded_event.payload_json.as_str(), ); let mut object = match payload_result { Ok(serde_json::Value::Object(object)) => object, Ok(other) => { let mut object = serde_json::Map::new(); object.insert("rawPayload".to_string(), other); object }, Err(_) => serde_json::Map::new(), }; let existing_reason = match object.get(reason_key).and_then(serde_json::Value::as_str) { Some(existing_reason) => existing_reason.trim().to_string(), None => std::string::String::new(), }; if existing_reason.is_empty() { object.insert( reason_key.to_string(), serde_json::Value::String(reason_value.to_string()), ); } if reason_key == "skipLiquidityReason" { object.insert( "skipCatalogReason".to_string(), serde_json::Value::String(reason_value.to_string()), ); } let payload_json = serde_json::Value::Object(object).to_string(); let update_result = crate::query_dex_decoded_events_update_payload_json_by_id( self.database.as_ref(), decoded_event_id, payload_json.as_str(), ) .await; match update_result { Ok(_) => return Ok(()), Err(error) => return Err(error), } } async fn resolve_liquidity_context( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, ) -> Result { let context = self.resolve_decoded_event_context(decoded_event).await; let context = match context { Ok(context) => context, Err(error) => return Err(error), }; let context = if context.pool_id.is_some() && context.pair.is_some() { context } else { let ensured_context = self.ensure_liquidity_context_from_decoded_event(decoded_event, context).await; match ensured_context { Ok(ensured_context) => ensured_context, Err(error) => return Err(error), } }; if context.pool_id.is_some() && context.pair.is_some() { return Ok(context); } let sibling_context = self .resolve_liquidity_context_from_transaction_siblings(transaction_id, decoded_event) .await; let context = match sibling_context { Ok(Some(sibling_context)) => return Ok(sibling_context), Ok(None) => context, Err(error) => return Err(error), }; let inferred_context = self .resolve_liquidity_context_from_transaction_token_balances( transaction, transaction_id, decoded_event, context.clone(), ) .await; match inferred_context { Ok(Some(inferred_context)) => return Ok(inferred_context), Ok(None) => return Ok(context), Err(error) => return Err(error), } } async fn resolve_liquidity_context_from_transaction_siblings( &self, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, ) -> Result, crate::Error> { let decoded_events_result = crate::query_dex_decoded_events_list_by_transaction_id( self.database.as_ref(), transaction_id, ) .await; let decoded_events = match decoded_events_result { Ok(decoded_events) => decoded_events, Err(error) => return Err(error), }; let decoded_event_id = decoded_event.id; for sibling in &decoded_events { if sibling.id == decoded_event_id { continue; } if sibling.protocol_name != decoded_event.protocol_name { continue; } if sibling.pool_account.is_none() { continue; } let sibling_context = self.resolve_decoded_event_context(sibling).await; let sibling_context = match sibling_context { Ok(sibling_context) => sibling_context, Err(error) => return Err(error), }; let sibling_context = if sibling_context.pool_id.is_some() && sibling_context.pair.is_some() { sibling_context } else { let ensured_context = self .ensure_liquidity_context_from_decoded_event(sibling, sibling_context) .await; match ensured_context { Ok(ensured_context) => ensured_context, Err(error) => return Err(error), } }; if sibling_context.pool_id.is_some() && sibling_context.pair.is_some() { return Ok(Some(sibling_context)); } } return Ok(None); } async fn resolve_liquidity_context_from_transaction_token_balances( &self, transaction: &crate::ChainTransactionDto, transaction_id: i64, decoded_event: &crate::DexDecodedEventDto, context: NonTradeDecodedEventContext, ) -> Result, crate::Error> { let dex_id = match context.dex_id { Some(dex_id) => dex_id, None => return Ok(None), }; let token_mints_by_account = token_mints_by_account_from_transaction(transaction); if token_mints_by_account.is_empty() { return Ok(None); } let decoded_events_result = crate::query_dex_decoded_events_list_by_transaction_id( self.database.as_ref(), transaction_id, ) .await; let decoded_events = match decoded_events_result { Ok(decoded_events) => decoded_events, Err(error) => return Err(error), }; let target_payload_result = serde_json::from_str::( decoded_event.payload_json.as_str(), ); let target_payload = match target_payload_result { Ok(target_payload) => target_payload, Err(_) => serde_json::Value::Object(serde_json::Map::new()), }; for candidate in &decoded_events { if candidate.protocol_name != decoded_event.protocol_name { continue; } if !candidate.event_kind.starts_with("raydium_clmm.") { continue; } let candidate_payload_result = serde_json::from_str::( candidate.payload_json.as_str(), ); let candidate_payload = match candidate_payload_result { Ok(candidate_payload) => candidate_payload, Err(_) => serde_json::Value::Object(serde_json::Map::new()), }; let pool_account = match candidate.pool_account.clone() { Some(pool_account) => Some(pool_account), None => extract_first_string( &candidate_payload, &["poolState", "pool_state", "poolAccount", "pool_account"], ), }; let pool_account = match pool_account { Some(pool_account) => pool_account, None => continue, }; let direct_token_a = match candidate.token_a_mint.clone() { Some(token_a_mint) => Some(token_a_mint), None => extract_first_string( &candidate_payload, &[ "tokenMint0", "token_mint0", "tokenMintA", "token_mint_a", "baseMint", "base_mint", ], ), }; let direct_token_b = match candidate.token_b_mint.clone() { Some(token_b_mint) => Some(token_b_mint), None => extract_first_string( &candidate_payload, &[ "tokenMint1", "token_mint1", "tokenMintB", "token_mint_b", "quoteMint", "quote_mint", ], ), }; let inferred_pair = infer_raydium_clmm_pair_mints_from_payload_accounts( &candidate_payload, &token_mints_by_account, ); let token_a_mint = match direct_token_a { Some(token_a_mint) => Some(token_a_mint), None => match inferred_pair.as_ref() { Some(pair) => Some(pair.0.clone()), None => None, }, }; let token_b_mint = match direct_token_b { Some(token_b_mint) => Some(token_b_mint), None => match inferred_pair.as_ref() { Some(pair) => Some(pair.1.clone()), None => None, }, }; let token_a_mint = match token_a_mint { Some(token_a_mint) => token_a_mint, None => continue, }; let token_b_mint = match token_b_mint { Some(token_b_mint) => token_b_mint, None => continue, }; if token_a_mint == token_b_mint { continue; } let synthetic_payload = merge_payload_with_inferred_pool_context( &target_payload, pool_account.as_str(), token_a_mint.as_str(), token_b_mint.as_str(), candidate.event_kind.as_str(), ); let mut synthetic_event = decoded_event.clone(); synthetic_event.pool_account = Some(pool_account); synthetic_event.token_a_mint = Some(token_a_mint); synthetic_event.token_b_mint = Some(token_b_mint); synthetic_event.payload_json = synthetic_payload.to_string(); let synthetic_context = NonTradeDecodedEventContext { dex_id: Some(dex_id), pool_id: None, pair_id: None, pair: None, }; let ensured = self .ensure_liquidity_context_from_decoded_event(&synthetic_event, synthetic_context) .await; let ensured = match ensured { Ok(ensured) => ensured, Err(error) => return Err(error), }; if ensured.pool_id.is_some() && ensured.pair.is_some() { return Ok(Some(ensured)); } } return Ok(None); } async fn ensure_liquidity_context_from_decoded_event( &self, decoded_event: &crate::DexDecodedEventDto, context: NonTradeDecodedEventContext, ) -> Result { let dex_id = match context.dex_id { Some(dex_id) => dex_id, None => return Ok(context), }; if context.pool_id.is_some() && context.pair.is_some() { return Ok(context); } if decoded_event.pool_account.is_none() || decoded_event.token_a_mint.is_none() || decoded_event.token_b_mint.is_none() { return Ok(context); } let materialization_input_result = crate::dex_pool_materialization::DexPoolMaterializationInput::from_decoded_event( decoded_event, dex_id, crate::PoolKind::Amm, crate::PoolStatus::Active, crate::dex_pool_materialization::DexPoolTokenOrder::AlreadyBaseQuote, None, None, None, ); let materialization_input = match materialization_input_result { Ok(materialization_input) => materialization_input, Err(_) => return Ok(context), }; let materialization_result = crate::dex_pool_materialization::materialize_dex_pool( self.database.as_ref(), &materialization_input, ) .await; if let Err(error) = materialization_result { return Err(error); } let refreshed_context = self.resolve_decoded_event_context(decoded_event).await; match refreshed_context { Ok(refreshed_context) => return Ok(refreshed_context), Err(error) => return Err(error), } } async fn resolve_decoded_event_context( &self, decoded_event: &crate::DexDecodedEventDto, ) -> Result { let dex_result = crate::query_dexs_get_by_code( self.database.as_ref(), decoded_event.protocol_name.as_str(), ) .await; let dex_id = match dex_result { Ok(Some(dex)) => dex.id, Ok(None) => None, Err(error) => return Err(error), }; let pool_address = match decoded_event.pool_account.clone() { Some(pool_address) => pool_address, None => { return Ok(NonTradeDecodedEventContext { dex_id, pool_id: None, pair_id: None, pair: None, }); }, }; let pool_result = crate::query_pools_get_by_address(self.database.as_ref(), pool_address.as_str()).await; let pool = match pool_result { Ok(Some(pool)) => pool, Ok(None) => { return Ok(NonTradeDecodedEventContext { dex_id, pool_id: None, pair_id: None, pair: None, }); }, Err(error) => return Err(error), }; let pool_id = match pool.id { Some(pool_id) => pool_id, None => { return Ok(NonTradeDecodedEventContext { dex_id, pool_id: None, pair_id: None, pair: None, }); }, }; let pair_result = crate::query_pairs_get_by_pool_id(self.database.as_ref(), pool_id).await; let pair = match pair_result { Ok(pair) => pair, Err(error) => return Err(error), }; let pair_id = match pair.as_ref() { Some(pair) => pair.id, None => None, }; return Ok(NonTradeDecodedEventContext { dex_id, pool_id: Some(pool_id), pair_id, pair, }); } } #[derive(Debug, Clone)] struct MaterializationAccountKeyInfo { index: i64, address: std::string::String, } fn token_mints_by_account_from_transaction( transaction: &crate::ChainTransactionDto, ) -> std::collections::HashMap { let transaction_json = serde_json::from_str::( transaction.transaction_json.as_str(), ); let transaction_json = match transaction_json { Ok(transaction_json) => transaction_json, Err(_) => return std::collections::HashMap::new(), }; let meta_value = match transaction.meta_json.as_deref() { Some(meta_json) => match serde_json::from_str::(meta_json) { Ok(meta_value) => Some(meta_value), Err(_) => None, }, None => transaction_json.get("meta").cloned(), }; let account_keys = materialization_account_keys(&transaction_json, meta_value.as_ref()); let mut mints = std::collections::HashMap::new(); collect_token_mints_by_account_side( meta_value.as_ref(), account_keys.as_slice(), "preTokenBalances", &mut mints, ); collect_token_mints_by_account_side( meta_value.as_ref(), account_keys.as_slice(), "postTokenBalances", &mut mints, ); return mints; } fn materialization_account_keys( transaction_json: &serde_json::Value, meta_value: std::option::Option<&serde_json::Value>, ) -> std::vec::Vec { let mut account_keys = std::vec::Vec::new(); let values = transaction_json .get("transaction") .and_then(|value| return value.get("message")) .and_then(|value| return value.get("accountKeys")) .and_then(serde_json::Value::as_array); if let Some(values) = values { let mut index = 0usize; for value in values { let address = if let Some(address) = value.as_str() { Some(address.to_string()) } else { value.get("pubkey").and_then(serde_json::Value::as_str).map(str::to_string) }; if let Some(address) = address { account_keys.push(MaterializationAccountKeyInfo { index: index as i64, address, }); } index += 1; } } append_materialization_loaded_addresses(&mut account_keys, meta_value, "writable"); append_materialization_loaded_addresses(&mut account_keys, meta_value, "readonly"); return account_keys; } fn append_materialization_loaded_addresses( account_keys: &mut std::vec::Vec, meta_value: std::option::Option<&serde_json::Value>, key: &str, ) { let addresses = meta_value .and_then(|value| return value.get("loadedAddresses")) .and_then(|value| return value.get(key)) .and_then(serde_json::Value::as_array); let addresses = match addresses { Some(addresses) => addresses, None => return, }; for value in addresses { let address = match value.as_str() { Some(address) => address, None => continue, }; let index = account_keys.len() as i64; account_keys.push(MaterializationAccountKeyInfo { index, address: address.to_string(), }); } } fn collect_token_mints_by_account_side( meta_value: std::option::Option<&serde_json::Value>, account_keys: &[MaterializationAccountKeyInfo], key: &str, mints: &mut std::collections::HashMap, ) { let values = meta_value .and_then(|value| return value.get(key)) .and_then(serde_json::Value::as_array); let values = match values { Some(values) => values, None => return, }; for value in values { let mint = match value.get("mint").and_then(serde_json::Value::as_str) { Some(mint) => mint, None => continue, }; let account_index = value.get("accountIndex").and_then(serde_json::Value::as_i64); let account_index = match account_index { Some(account_index) => account_index, None => continue, }; let account_address = materialization_account_address_by_index(account_keys, account_index); let account_address = match account_address { Some(account_address) => account_address, None => continue, }; mints.insert(account_address, mint.to_string()); } } fn materialization_account_address_by_index( account_keys: &[MaterializationAccountKeyInfo], account_index: i64, ) -> std::option::Option { for account_key in account_keys { if account_key.index == account_index { return Some(account_key.address.clone()); } } return None; } fn infer_raydium_clmm_pair_mints_from_payload_accounts( payload: &serde_json::Value, token_mints_by_account: &std::collections::HashMap, ) -> std::option::Option<(std::string::String, std::string::String)> { let accounts = payload.get("accounts").and_then(serde_json::Value::as_array); let accounts = match accounts { Some(accounts) => accounts, None => return None, }; let instruction_name = payload .get("instructionName") .and_then(serde_json::Value::as_str); let instruction_name = match instruction_name { Some(instruction_name) => instruction_name, None => "", }; let candidate_pairs = raydium_clmm_token_account_candidate_pairs(instruction_name); for pair in candidate_pairs { let inferred = infer_mints_from_account_pair( accounts, pair.0, pair.1, token_mints_by_account, ); if let Some(inferred) = inferred { return Some(inferred); } } return infer_distinct_mints_from_accounts(accounts, token_mints_by_account); } fn raydium_clmm_token_account_candidate_pairs( instruction_name: &str, ) -> std::vec::Vec<(usize, usize)> { if instruction_name == "open_position" { return vec![(12, 13), (10, 11), (18, 19), (20, 21), (7, 8), (8, 9)]; } if instruction_name == "open_position_v2" { return vec![(12, 13), (13, 14), (18, 19), (20, 21), (10, 11), (7, 8)]; } if instruction_name == "increase_liquidity" { return vec![(9, 10), (7, 8), (13, 14), (14, 15), (5, 6)]; } if instruction_name == "decrease_liquidity" { return vec![(5, 6), (9, 10), (14, 15), (15, 16), (7, 8)]; } if instruction_name == "decrease_liquidity_v2" { return vec![(14, 15), (5, 6), (9, 10)]; } if instruction_name == "increase_liquidity_v2" { return vec![(13, 14), (9, 10), (7, 8)]; } return vec![(12, 13), (13, 14), (9, 10), (7, 8), (5, 6), (10, 11), (14, 15), (18, 19), (20, 21)]; } fn infer_mints_from_account_pair( accounts: &[serde_json::Value], left_index: usize, right_index: usize, token_mints_by_account: &std::collections::HashMap, ) -> std::option::Option<(std::string::String, std::string::String)> { let left_account = accounts.get(left_index).and_then(serde_json::Value::as_str); let right_account = accounts.get(right_index).and_then(serde_json::Value::as_str); let left_account = match left_account { Some(left_account) => left_account, None => return None, }; let right_account = match right_account { Some(right_account) => right_account, None => return None, }; let left_mint = token_mints_by_account.get(left_account); let right_mint = token_mints_by_account.get(right_account); let left_mint = match left_mint { Some(left_mint) => left_mint, None => return None, }; let right_mint = match right_mint { Some(right_mint) => right_mint, None => return None, }; if left_mint == right_mint { return None; } return Some((left_mint.clone(), right_mint.clone())); } fn infer_distinct_mints_from_accounts( accounts: &[serde_json::Value], token_mints_by_account: &std::collections::HashMap, ) -> std::option::Option<(std::string::String, std::string::String)> { let mut first: std::option::Option = None; let mut second: std::option::Option = None; for account in accounts { let account = match account.as_str() { Some(account) => account, None => continue, }; let mint = token_mints_by_account.get(account); let mint = match mint { Some(mint) => mint, None => continue, }; if first.is_none() { first = Some(mint.clone()); continue; } if first.as_ref() == Some(mint) { continue; } second = Some(mint.clone()); break; } match (first, second) { (Some(first), Some(second)) => return Some((first, second)), _ => return None, } } fn merge_payload_with_inferred_pool_context( payload: &serde_json::Value, pool_account: &str, token_a_mint: &str, token_b_mint: &str, context_source_event_kind: &str, ) -> serde_json::Value { let mut object = match payload.clone() { serde_json::Value::Object(object) => object, other => { let mut object = serde_json::Map::new(); object.insert("rawPayload".to_string(), other); object }, }; object.insert("poolState".to_string(), serde_json::Value::String(pool_account.to_string())); object.insert("poolAccount".to_string(), serde_json::Value::String(pool_account.to_string())); object.insert("tokenMint0".to_string(), serde_json::Value::String(token_a_mint.to_string())); object.insert("tokenMint1".to_string(), serde_json::Value::String(token_b_mint.to_string())); object.insert("inferredPoolContext".to_string(), serde_json::Value::Bool(true)); object.insert( "inferredPoolContextSourceEventKind".to_string(), serde_json::Value::String(context_source_event_kind.to_string()), ); return serde_json::Value::Object(object); } fn is_token_account_event_materializable(event_kind: &str) -> bool { return event_kind.ends_with(".create_support_mint_associated"); } fn extract_account_string( payload: &serde_json::Value, index: usize, ) -> std::option::Option { let accounts_option = payload.get("accounts"); let accounts = match accounts_option { Some(accounts) => accounts.as_array(), None => None, }; let accounts = match accounts { Some(accounts) => accounts, None => return None, }; let value = accounts.get(index); match value { Some(value) => match value.as_str() { Some(value) => return Some(value.to_string()), None => return None, }, None => return None, } } fn is_launchpad_launch_event_materializable(event_kind: &str) -> bool { if event_kind.contains("raydium_launchpad.buy_exact_in") { return true; } if event_kind.contains("raydium_launchpad.buy_exact_out") { return true; } if event_kind.contains("raydium_launchpad.sell_exact_in") { return true; } if event_kind.contains("raydium_launchpad.sell_exact_out") { return true; } if event_kind.contains("raydium_launchpad.claim_vested_event") { return true; } if event_kind.contains("raydium_launchpad.claim_vested_token") { return true; } if event_kind.contains("raydium_launchpad.create_platform_vesting_account") { return true; } if event_kind.contains("raydium_launchpad.create_vesting_account") { return true; } if event_kind.contains("raydium_launchpad.create_vesting_event") { return true; } if event_kind.contains("raydium_launchpad.migrate_to_amm") { return true; } if event_kind.contains("raydium_launchpad.migrate_to_cpswap") { return true; } return false; } fn launchpad_launch_event_role(event_kind: &str) -> std::string::String { if event_kind.contains("buy_exact_in") { return "swap_instruction_buy_exact_in".to_string(); } if event_kind.contains("buy_exact_out") { return "swap_instruction_buy_exact_out".to_string(); } if event_kind.contains("sell_exact_in") { return "swap_instruction_sell_exact_in".to_string(); } if event_kind.contains("sell_exact_out") { return "swap_instruction_sell_exact_out".to_string(); } if event_kind.contains("claim_vested") { return "vesting_claim".to_string(); } if event_kind.contains("vesting") { return "vesting".to_string(); } if event_kind.contains("migrate_to_amm") { return "migration_to_amm".to_string(); } if event_kind.contains("migrate_to_cpswap") { return "migration_to_cpswap".to_string(); } return "launch".to_string(); } fn normalize_orderbook_action(event_kind: &str) -> std::string::String { if event_kind.contains(".open_limit_order") { return "open_limit_order".to_string(); } if event_kind.contains(".increase_limit_order") { return "increase_limit_order".to_string(); } if event_kind.contains(".decrease_limit_order") { return "decrease_limit_order".to_string(); } if event_kind.contains(".close_limit_order") { return "close_limit_order".to_string(); } if event_kind.contains(".settle_limit_order") { return "settle_limit_order".to_string(); } if event_kind.contains("order_place") { return "order_place".to_string(); } if event_kind.contains("order_cancel") { return "order_cancel".to_string(); } if event_kind.contains("settle_funds") { return "settle_funds".to_string(); } return event_kind.to_string(); } fn fallback_order_account( event_kind: &str, payload: &serde_json::Value, ) -> std::option::Option { if event_kind.contains(".close_limit_order") { return extract_account_at(payload, 2); } if event_kind.contains(".open_limit_order") || event_kind.contains(".increase_limit_order") || event_kind.contains(".decrease_limit_order") { return extract_account_at(payload, 3); } return None; } fn extract_account_at( value: &serde_json::Value, index: usize, ) -> std::option::Option { if let Some(object) = value.as_object() { let accounts = object.get("accounts"); if let Some(accounts) = accounts { if let Some(array) = accounts.as_array() { let candidate = array.get(index); if let Some(candidate) = candidate { if let Some(text) = candidate.as_str() { let trimmed = text.trim(); if !trimmed.is_empty() { return Some(trimmed.to_string()); } } } } } for nested_value in object.values() { let nested = extract_account_at(nested_value, index); if nested.is_some() { return nested; } } } return None; } fn extract_first_i64( value: &serde_json::Value, candidate_keys: &[&str], ) -> std::option::Option { if let Some(object) = value.as_object() { for candidate_key in candidate_keys { let candidate_value = object.get(*candidate_key); if let Some(candidate_value) = candidate_value { if let Some(number) = candidate_value.as_i64() { return Some(number); } if let Some(number) = candidate_value.as_u64() { let converted = i64::try_from(number); if let Ok(converted) = converted { return Some(converted); } } if let Some(text) = candidate_value.as_str() { let parsed = text.parse::(); if let Ok(parsed) = parsed { return Some(parsed); } } } } for nested_value in object.values() { let nested = extract_first_i64(nested_value, candidate_keys); if nested.is_some() { return nested; } } } return None; } fn extract_first_bool( value: &serde_json::Value, candidate_keys: &[&str], ) -> std::option::Option { if let Some(object) = value.as_object() { for candidate_key in candidate_keys { let candidate_value = object.get(*candidate_key); if let Some(candidate_value) = candidate_value { if let Some(flag) = candidate_value.as_bool() { return Some(flag); } if let Some(number) = candidate_value.as_i64() { return Some(number != 0); } if let Some(text) = candidate_value.as_str() { if text == "true" || text == "1" { return Some(true); } if text == "false" || text == "0" { return Some(false); } } } } for nested_value in object.values() { let nested = extract_first_bool(nested_value, candidate_keys); if nested.is_some() { return nested; } } } return None; } fn transaction_has_effective_error(transaction: &crate::ChainTransactionDto) -> bool { let err_json = match transaction.err_json.as_ref() { Some(err_json) => err_json.trim(), None => return false, }; if err_json.is_empty() { return false; } if err_json == "null" { return false; } return true; } fn extract_first_u64( value: &serde_json::Value, candidate_keys: &[&str], ) -> std::option::Option { if let Some(object) = value.as_object() { for candidate_key in candidate_keys { let candidate_value = object.get(*candidate_key); if let Some(candidate_value) = candidate_value { if let Some(number) = candidate_value.as_u64() { return Some(number); } if let Some(text) = candidate_value.as_str() { let parsed = text.parse::(); if let Ok(parsed) = parsed { return Some(parsed); } } } } } return None; } fn extract_first_amount_string( value: &serde_json::Value, candidate_keys: &[&str], ) -> std::option::Option { let text = extract_first_string(value, candidate_keys); if text.is_some() { return text; } return extract_first_number_as_string(value, candidate_keys); } fn extract_first_string( value: &serde_json::Value, candidate_keys: &[&str], ) -> std::option::Option { if let Some(object) = value.as_object() { for candidate_key in candidate_keys { let value_option = object.get(*candidate_key); let candidate = match value_option { Some(candidate) => candidate, None => continue, }; if let Some(text) = candidate.as_str() { let trimmed = text.trim(); if !trimmed.is_empty() { return Some(trimmed.to_string()); } } } for nested_value in object.values() { let nested = extract_first_string(nested_value, candidate_keys); if nested.is_some() { return nested; } } return None; } if let Some(array) = value.as_array() { for nested_value in array { let nested = extract_first_string(nested_value, candidate_keys); if nested.is_some() { return nested; } } } return None; } fn extract_first_number_as_string( value: &serde_json::Value, candidate_keys: &[&str], ) -> std::option::Option { if let Some(object) = value.as_object() { for candidate_key in candidate_keys { let value_option = object.get(*candidate_key); let candidate = match value_option { Some(candidate) => candidate, None => continue, }; if let Some(number) = candidate.as_i64() { return Some(number.to_string()); } if let Some(number) = candidate.as_u64() { return Some(number.to_string()); } if let Some(number) = candidate.as_f64() { return Some(number.to_string()); } } for nested_value in object.values() { let nested = extract_first_number_as_string(nested_value, candidate_keys); if nested.is_some() { return nested; } } return None; } if let Some(array) = value.as_array() { for nested_value in array { let nested = extract_first_number_as_string(nested_value, candidate_keys); if nested.is_some() { return nested; } } } return None; } #[cfg(test)] mod tests { #[test] fn blank_or_null_err_json_is_not_effective_failure() { let mut transaction = crate::ChainTransactionDto::new( "sig-non-trade-effective-error".to_string(), Some(1), None, None, None, None, None, "{}".to_string(), ); assert!(!super::transaction_has_effective_error(&transaction)); transaction.err_json = Some("".to_string()); assert!(!super::transaction_has_effective_error(&transaction)); transaction.err_json = Some("null".to_string()); assert!(!super::transaction_has_effective_error(&transaction)); transaction.err_json = Some("{\"InstructionError\":[0,\"Custom\"]}".to_string()); assert!(super::transaction_has_effective_error(&transaction)); } #[test] fn extracts_nested_liquidity_amounts() { let payload = serde_json::json!({ "event": { "baseAmountRaw": "100", "quoteAmountRaw": 25, "owner": "Owner111111111111111111111111111111111111" } }); assert_eq!( super::extract_first_amount_string(&payload, &["baseAmountRaw"]), Some("100".to_string()) ); assert_eq!( super::extract_first_amount_string(&payload, &["quoteAmountRaw"]), Some("25".to_string()) ); assert_eq!( super::extract_first_string(&payload, &["owner"]), Some("Owner111111111111111111111111111111111111".to_string()) ); } }