DPLT-1076 create node worker to process real time streams (#159)

indexer/queryapi_coordinator/src/main.rs

@@ -200,16 +200,21 @@ async fn handle_streamer_message(
                     set_provisioned_flag(&mut indexer_registry_locked, &indexer_function);
                 }
 
+                storage::sadd(
+                    context.redis_connection_manager,
+                    storage::INDEXER_SET_KEY,
+                    indexer_function.get_full_name(),
+                )
+                .await?;
                 storage::set(
                     context.redis_connection_manager,
-                    &format!("{}:storage", indexer_function.get_full_name()),
+                    storage::generate_storage_key(&indexer_function.get_full_name()),
                     serde_json::to_string(indexer_function)?,
                 )
                 .await?;
-
-                storage::add_to_registered_stream(
+                storage::xadd(
                     context.redis_connection_manager,
-                    &format!("{}:stream", indexer_function.get_full_name()),
+                    storage::generate_stream_key(&indexer_function.get_full_name()),
                     &[("block_height", block_height)],
                 )
                 .await?;

indexer/storage/src/lib.rs

@@ -2,12 +2,20 @@ pub use redis::{self, aio::ConnectionManager, FromRedisValue, ToRedisArgs};
 
 const STORAGE: &str = "storage_alertexer";
 
-const STREAMS_SET_KEY: &str = "streams";
+pub const INDEXER_SET_KEY: &str = "indexers";
 
 pub async fn get_redis_client(redis_connection_str: &str) -> redis::Client {
     redis::Client::open(redis_connection_str).expect("can create redis client")
 }
 
+pub fn generate_storage_key(name: &str) -> String {
+    format!("{}:storage", name)
+}
+
+pub fn generate_stream_key(name: &str) -> String {
+    format!("{}:stream", name)
+}
+
 pub async fn connect(redis_connection_str: &str) -> anyhow::Result<ConnectionManager> {
     Ok(get_redis_client(redis_connection_str)
         .await
@@ -53,7 +61,7 @@ pub async fn get<V: FromRedisValue + std::fmt::Debug>(
     Ok(value)
 }
 
-async fn sadd(
+pub async fn sadd(
     redis_connection_manager: &ConnectionManager,
     key: impl ToRedisArgs + std::fmt::Debug,
     value: impl ToRedisArgs + std::fmt::Debug,
@@ -69,16 +77,16 @@ async fn sadd(
     Ok(())
 }
 
-async fn xadd(
+pub async fn xadd(
     redis_connection_manager: &ConnectionManager,
-    stream_key: &str,
+    stream_key: impl ToRedisArgs + std::fmt::Debug,
     fields: &[(&str, impl ToRedisArgs + std::fmt::Debug)],
 ) -> anyhow::Result<()> {
-    tracing::debug!(target: STORAGE, "XADD: {}, {:?}", stream_key, fields);
+    tracing::debug!(target: STORAGE, "XADD: {:?}, {:?}", stream_key, fields);
 
     // TODO: Remove stream cap when we finally start processing it
     redis::cmd("XTRIM")
-        .arg(stream_key)
+        .arg(&stream_key)
         .arg("MAXLEN")
         .arg(100)
         .query_async(&mut redis_connection_manager.clone())
@@ -97,16 +105,6 @@ async fn xadd(
     Ok(())
 }
 
-pub async fn add_to_registered_stream(
-    redis_connection_manager: &ConnectionManager,
-    key: &str,
-    fields: &[(&str, impl ToRedisArgs + std::fmt::Debug)],
-) -> anyhow::Result<()> {
-    sadd(redis_connection_manager, STREAMS_SET_KEY, key).await?;
-    xadd(redis_connection_manager, key, fields).await?;
-
-    Ok(())
-}
 /// Sets the key `receipt_id: &str` with value `transaction_hash: &str` to the Redis storage.
 /// Increments the counter `receipts_{transaction_hash}` by one.
 /// The counter holds how many Receipts related to the Transaction are in watching list

runner/.eslintrc.js

@@ -0,0 +1,32 @@
+module.exports = {
+  parser: '@typescript-eslint/parser',
+  env: {
+    es2021: true,
+    node: true,
+  },
+  overrides: [
+    {
+      files: ['.eslintrc.js', 'jest.config.js'],
+      parser: 'espree',
+      extends: ['standard'],
+      rules: {
+        semi: ['error', 'always'],
+        'comma-dangle': ['error', 'only-multiline'],
+      },
+    },
+    {
+      files: ['./src/**/*'],
+      parserOptions: {
+        project: './tsconfig.json',
+      },
+      extends: [
+        'standard-with-typescript',
+      ],
+      rules: {
+        '@typescript-eslint/semi': ['error', 'always'],
+        '@typescript-eslint/comma-dangle': ['error', 'only-multiline'],
+        '@typescript-eslint/strict-boolean-expressions': 'off',
+      },
+    },
+  ],
+};

runner/.gitignore

@@ -0,0 +1,2 @@
+**/dist
+/node_modules

runner/jest.config.js

@@ -0,0 +1,4 @@
+module.exports = {
+  preset: 'ts-jest',
+  testEnvironment: 'node'
+};