IndexerRule parsing of registration function calls, minor logging cle…

…anup. (#78)
near · May 25, 2023 · 544e423 · 544e423
1 parent f0971ad
commit 544e423
Show file tree

Hide file tree

Showing 5 changed files with 53 additions and 17 deletions.
diff --git a/indexer/Cargo.lock b/indexer/Cargo.lock
diff --git a/indexer/queryapi_coordinator/Cargo.toml b/indexer/queryapi_coordinator/Cargo.toml
@@ -36,4 +36,5 @@ aws-types = "0.53.0"
 aws-credential-types = "0.53.0"
 aws-sdk-s3 = "0.23.0"
 aws-sdk-sqs = "0.23.0"
-tracing-subscriber = "0.2.4"
+tracing-subscriber = "0.2.4"
+unescape = "0.1.0"
diff --git a/indexer/queryapi_coordinator/src/indexer_registry.rs b/indexer/queryapi_coordinator/src/indexer_registry.rs
@@ -10,6 +10,7 @@ use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use tokio::sync::MutexGuard;
 use tokio::task::JoinHandle;
+use unescape::unescape;
 
 use crate::indexer_reducer;
 use crate::indexer_reducer::FunctionCallInfo;
@@ -130,7 +131,6 @@ fn index_and_process_register_calls(
             let new_indexer_function = build_indexer_function_from_args(
                 parse_indexer_function_args(&update),
                 update.signer_id,
-                &registry_calls_rule,
             );
 
             match new_indexer_function {
@@ -260,7 +260,6 @@ fn build_function_invocation_from_args(
 fn build_indexer_function_from_args(
     args: Option<Value>,
     signer_id: String,
-    indexer_rule: &IndexerRule,
 ) -> Option<IndexerFunction> {
     match args {
         None => None,
@@ -272,26 +271,51 @@ fn build_indexer_function_from_args(
             let function_name = args["function_name"].as_str();
             match function_name {
                 None => {
-                    tracing::error!(
+                    tracing::warn!(
                         "Unable to parse function_name from indexer function: {:?}",
                         &args
                     );
                     return None;
                 }
-                Some(function_name) => build_indexer_function(
-                    &args,
-                    function_name.to_string(),
-                    account_id,
-                    indexer_rule,
-                ),
+                Some(function_name) => {
+                    match unescape(&args["filter_json"].to_string()) {
+                        Some(filter_string) => {
+                            let filter_json_strip_quotes = &filter_string[1..filter_string.len() - 1];
+                            match serde_json::from_str(&filter_json_strip_quotes) {
+                                Ok(filter_json) => match serde_json::from_value(filter_json) {
+                                    Ok(indexer_rule) => build_indexer_function(
+                                        &args,
+                                        function_name.to_string(),
+                                        account_id,
+                                        &indexer_rule,
+                                    ),
+                                    Err(e) => {
+                                        tracing::warn!("Error parsing filter into indexer_rule for account {} function {}: {}, {}", account_id, function_name, e, filter_string);
+                                        None
+                                    }
+                                },
+                                Err(e) => {
+                                    tracing::warn!("Error parsing indexer_rule filter for account {} function {}: {}, {}", account_id, function_name, e, filter_string);
+                                    None
+                                }
+                            }
+                        },
+                        None => {
+                            tracing::warn!(
+                                "Unable to unescape filter_json from registration args: {:?}",
+                                &args
+                            );
+                            None
+                        }
+                    }
+                }
             }
         }
     }
 }
 
 fn parse_indexer_function_args(update: &FunctionCallInfo) -> Option<Value> {
     if let Ok(mut args_json) = serde_json::from_str(&update.args) {
-        escape_json(&mut args_json);
         return Some(args_json);
     } else {
         tracing::error!(

diff --git a/indexer/queryapi_coordinator/src/main.rs b/indexer/queryapi_coordinator/src/main.rs
@@ -169,6 +169,14 @@ async fn handle_streamer_message(
             let mut indexer_function_messages: Vec<IndexerQueueMessage> = Vec::new();
 
             for indexer_rule_match in indexer_rule_matches.iter() {
+                tracing::debug!(
+                    target: INDEXER,
+                    "Matched filter {:?} for function {} {}",
+                    indexer_function.indexer_rule.matching_rule,
+                    indexer_function.account_id,
+                    indexer_function.function_name,
+                );
+
                 let msg = IndexerQueueMessage {
                     chain_id: indexer_rule_match.chain_id.clone(),
                     indexer_rule_id: indexer_rule_match.indexer_rule_id.unwrap_or(0),
@@ -189,11 +197,11 @@ async fn handle_streamer_message(
 
             stream::iter(indexer_function_messages.into_iter())
                 .chunks(10)
-                .for_each(|alert_queue_messages_batch| async {
+                .for_each(|indexer_queue_messages_batch| async {
                     match opts::send_to_indexer_queue(
                         context.queue_client,
                         context.queue_url.to_string(),
-                        alert_queue_messages_batch,
+                        indexer_queue_messages_batch,
                     )
                     .await
                     {

diff --git a/indexer/queryapi_coordinator/src/opts.rs b/indexer/queryapi_coordinator/src/opts.rs
@@ -227,10 +227,6 @@ pub async fn send_to_indexer_queue(
     queue_url: String,
     indexer_queue_messages: Vec<IndexerQueueMessage>,
 ) -> anyhow::Result<()> {
-    tracing::info!(
-        target: "queryapi_coordinator",
-        "Sending indexer tasks to the queue: {queue_url}",
-    );
 
     let message_bodies: Vec<SendMessageBatchRequestEntry> = indexer_queue_messages
         .into_iter()