Store Real Time Streamer Messages in Redis

block-server/handler.js

@@ -1,6 +1,6 @@
 'use strict';
-const AWS = require('aws-sdk');
-const S3= new AWS.S3();
+import { S3 } from '@aws-sdk/client-s3';
+const S3 = new S3();
 
 const NETWORK = process.env.NETWORK || 'mainnet';
 

indexer/queryapi_coordinator/src/historical_block_processing.rs

@@ -151,6 +151,7 @@ pub(crate) async fn process_historical_messages(
                     redis_connection_manager,
                     storage::generate_historical_storage_key(&indexer_function.get_full_name()),
                     serde_json::to_string(&indexer_function)?,
+                    None,
                 )
                 .await?;
             }

indexer/queryapi_coordinator/src/main.rs

@@ -6,6 +6,7 @@ use tokio::sync::{Mutex, MutexGuard};
 use indexer_rules_engine::types::indexer_rule_match::{ChainId, IndexerRuleMatch};
 use near_lake_framework::near_indexer_primitives::types::{AccountId, BlockHeight};
 use near_lake_framework::near_indexer_primitives::{types, StreamerMessage};
+use utils::serialize_to_camel_case_json_string;
 
 use crate::indexer_types::IndexerFunction;
 use indexer_types::{IndexerQueueMessage, IndexerRegistry};
@@ -146,6 +147,19 @@ async fn handle_streamer_message(
 
     let block_height: BlockHeight = context.streamer_message.block.header.height;
 
+    // Cache streamer message block and shards for use in real time processing
+    storage::set(
+        context.redis_connection_manager,
+        format!(
+            "{}{}",
+            storage::STREAMER_MESSAGE_HASH_KEY_BASE,
+            block_height
+        ),
+        &serialize_to_camel_case_json_string(&context.streamer_message)?,
+        Some(60),
+    )
+    .await?;
+
     let spawned_indexers = indexer_registry::index_registry_changes(
         block_height,
         &mut indexer_registry_locked,
@@ -206,6 +220,7 @@ async fn handle_streamer_message(
                     context.redis_connection_manager,
                     storage::generate_real_time_storage_key(&indexer_function.get_full_name()),
                     serde_json::to_string(indexer_function)?,
+                    None,
                 )
                 .await?;
                 storage::xadd(

indexer/queryapi_coordinator/src/utils.rs

@@ -1,3 +1,5 @@
+use serde_json::Value;
+
 pub(crate) async fn stats(redis_connection_manager: storage::ConnectionManager) {
     let interval_secs = 10;
     let mut previous_processed_blocks: u64 =
@@ -49,3 +51,52 @@ pub(crate) async fn stats(redis_connection_manager: storage::ConnectionManager)
         tokio::time::sleep(std::time::Duration::from_secs(interval_secs)).await;
     }
 }
+
+pub(crate) fn serialize_to_camel_case_json_string(
+    streamer_message: &near_lake_framework::near_indexer_primitives::StreamerMessage,
+) -> anyhow::Result<String, serde_json::Error> {
+    // Serialize the Message object to a JSON string
+    let json_str = serde_json::to_string(&streamer_message)?;
+
+    // Deserialize the JSON string to a Value Object
+    let mut message_value: Value = serde_json::from_str(&json_str)?;
+
+    // Convert keys to Camel Case
+    to_camel_case_keys(&mut message_value);
+
+    return serde_json::to_string(&message_value);
+}
+
+fn to_camel_case_keys(message_value: &mut Value) {
+    // Only process if subfield contains objects
+    match message_value {
+        Value::Object(map) => {
+            for key in map.keys().cloned().collect::<Vec<String>>() {
+                // Generate Camel Case Key
+                let new_key = key
+                    .split("_")
+                    .enumerate()
+                    .map(|(i, str)| {
+                        if i > 0 {
+                            return str[..1].to_uppercase() + &str[1..];
+                        }
+                        return str.to_owned();
+                    })
+                    .collect::<Vec<String>>()
+                    .join("");
+
+                // Recursively process inner fields and update map with new key
+                if let Some(mut val) = map.remove(&key) {
+                    to_camel_case_keys(&mut val);
+                    map.insert(new_key, val);
+                }
+            }
+        }
+        Value::Array(vec) => {
+            for val in vec {
+                to_camel_case_keys(val);
+            }
+        }
+        _ => {}
+    }
+}

indexer/storage/src/lib.rs

@@ -2,7 +2,9 @@ pub use redis::{self, aio::ConnectionManager, FromRedisValue, ToRedisArgs};
 
 const STORAGE: &str = "storage_alertexer";
 
+pub const LAKE_BUCKET_PREFIX: &str = "near-lake-data-";
 pub const STREAMS_SET_KEY: &str = "streams";
+pub const STREAMER_MESSAGE_HASH_KEY_BASE: &str = "streamer:message:";
 
 pub async fn get_redis_client(redis_connection_str: &str) -> redis::Client {
     redis::Client::open(redis_connection_str).expect("can create redis client")
@@ -47,13 +49,23 @@ pub async fn set(
     redis_connection_manager: &ConnectionManager,
     key: impl ToRedisArgs + std::fmt::Debug,
     value: impl ToRedisArgs + std::fmt::Debug,
+    expiration: Option<usize>,
 ) -> anyhow::Result<()> {
-    redis::cmd("SET")
-        .arg(&key)
-        .arg(&value)
-        .query_async(&mut redis_connection_manager.clone())
+    let mut cmd = redis::cmd("SET");
+    cmd.arg(&key).arg(&value);
+
+    // Add expiration arguments if present
+    let exp_to_print: String;
+    if let Some(expiration) = expiration {
+        cmd.arg("EX").arg(expiration);
+        exp_to_print = format!("EX {}", expiration);
+    } else {
+        exp_to_print = "".to_string();
+    }
+
+    cmd.query_async(&mut redis_connection_manager.clone())
         .await?;
-    tracing::debug!(target: STORAGE, "SET: {:?}: {:?}", key, value,);
+    tracing::debug!(target: STORAGE, "SET: {:?}: {:?} {:?}", key, value, exp_to_print);
     Ok(())
 }
 
@@ -113,7 +125,7 @@ pub async fn push_receipt_to_watching_list(
     receipt_id: &str,
     cache_value: &[u8],
 ) -> anyhow::Result<()> {
-    set(redis_connection_manager, receipt_id, cache_value).await?;
+    set(redis_connection_manager, receipt_id, cache_value, None).await?;
     // redis::cmd("INCR")
     //     .arg(format!("receipts_{}", transaction_hash))
     //     .query_async(&mut redis_connection_manager.clone())
@@ -161,7 +173,13 @@ pub async fn update_last_indexed_block(
     redis_connection_manager: &ConnectionManager,
     block_height: u64,
 ) -> anyhow::Result<()> {
-    set(redis_connection_manager, "last_indexed_block", block_height).await?;
+    set(
+        redis_connection_manager,
+        "last_indexed_block",
+        block_height,
+        None,
+    )
+    .await?;
     redis::cmd("INCR")
         .arg("blocks_processed")
         .query_async(&mut redis_connection_manager.clone())

runner/package.json

@@ -42,8 +42,8 @@
     "typescript": "^5.1.6"
   },
   "dependencies": {
+    "@aws-sdk/client-s3": "^3.414.0",
     "@near-lake/primitives": "^0.1.0",
-    "aws-sdk": "^2.1402.0",
     "express": "^4.18.2",
     "node-fetch": "^2.6.11",
     "node-sql-parser": "^4.10.0",