chore(relayer): refactor event sources

Author: kiseln

omni-relayer/Cargo.toml

@@ -34,9 +34,9 @@ solana-transaction-status = "2.1.9"
 solana-rpc-client-api = "2.1.9"
 
 omni-types = { git = "https://github.com/near-one/omni-bridge", package = "omni-types", rev = "ffbdc544c0d7f670942e8bf532c7e023c1c6bfec", features = ["abi", "__abi-generate"] }
-bridge-indexer-types = { git = "ssh://git@github.com/near-one/bridge-indexer-rs.git", package = "bridge-indexer-types", rev = "7e99a29f783eb3a15e4f2296a0759773118c1c6d" }
+bridge-indexer-types = { git = "ssh://git@github.com/near-one/bridge-indexer-rs.git", package = "bridge-indexer-types", rev = "7e99a29f783eb3a15e4f2296a0759773118c1c6d", optional = true }
 
-near-lake-framework = "0.7.14"
+near-lake-framework = { version = "0.7.14", optional = true }
 near-jsonrpc-client = "0.20.0"
 near-jsonrpc-primitives = "0.34.2"
 near-primitives = "0.34.2"
@@ -46,7 +46,7 @@ near-sdk = { version = "5.24.1", features = ["unit-testing"] }
 hex = "0.4.3"
 alloy = { version = "1.1", features = ["node-bindings", "providers", "rpc-types", "sol-types", "provider-ws", "signers", "signer-local"] }
 
-mongodb = "3.2.2"
+mongodb = { version = "3.2.2", optional = true }
 redis = { version = "0.32.5", features = ["aio", "tokio-comp", "connection-manager", "json"] }
 async-nats = "0.45"
 reqwest = "0.12"
@@ -65,6 +65,12 @@ light-client = { git = "https://github.com/Near-One/bridge-sdk-rs", package = "l
 
 omni-connector = { git = "https://github.com/Near-One/bridge-sdk-rs", package = "omni-connector", rev = "19f6cc07f1802497aabee9c943358a16c3a6d901" }
 
+[features]
+default = ["nats-ingestion"]
+nats-ingestion = ["dep:bridge-indexer-types"]
+mongo-ingestion = ["dep:mongodb", "dep:bridge-indexer-types"]
+native-indexers = ["dep:near-lake-framework"]
+
 # The profile that 'dist' will build with
 [profile.dist]
 inherits = "release"

omni-relayer/src/main.rs

@@ -6,12 +6,12 @@ use config::Network;
 use near_sdk::base64::{Engine, engine::general_purpose};
 use omni_types::ChainKind;
 use reqwest::Url;
-use solana_sdk::signature::Signature;
 use tracing::{error, info, warn};
 use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
 
 mod config;
 mod startup;
+mod types;
 mod utils;
 mod workers;
 
@@ -21,27 +21,35 @@ struct CliArgs {
     #[clap(short, long, default_value = "config.toml")]
     config: String,
     /// Start block for Near indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
     near_start_block: Option<u64>,
     /// Start block for Ethereum indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
     eth_start_block: Option<u64>,
     /// Start block for Base indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
     base_start_block: Option<u64>,
     /// Start block for Arbitrum indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
     arb_start_block: Option<u64>,
     /// Start block for Bnb indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
     bnb_start_block: Option<u64>,
     /// Start block for Pol indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
     pol_start_block: Option<u64>,
     /// Start signature for Solana indexer
+    #[cfg(feature = "native-indexers")]
     #[clap(long)]
-    solana_start_signature: Option<Signature>,
+    solana_start_signature: Option<solana_sdk::signature::Signature>,
     /// Start timestamp for bridge indexer
+    #[cfg(feature = "mongo-ingestion")]
     #[clap(long)]
     start_timestamp: Option<u32>,
 }
@@ -144,11 +152,11 @@ async fn main() -> Result<()> {
 
     let jsonrpc_client = near_jsonrpc_client::JsonRpcClient::connect(config.near.rpc_url.clone());
 
-    let near_omni_signer = startup::near::get_signer(&config, config::NearSignerType::Omni)?;
+    let near_omni_signer = utils::near::get_signer(&config, config::NearSignerType::Omni)?;
     let omni_connector = Arc::new(startup::build_omni_connector(&config, &near_omni_signer).await?);
 
     let (near_fast_signer, fast_connector) = if config.is_fast_relayer_enabled() {
-        let near_fast_signer = startup::near::get_signer(&config, config::NearSignerType::Fast)?;
+        let near_fast_signer = utils::near::get_signer(&config, config::NearSignerType::Fast)?;
 
         (
             Some(near_fast_signer.clone()),
@@ -176,40 +184,47 @@ async fn main() -> Result<()> {
 
     let mut handles = Vec::new();
 
+    #[cfg(feature = "nats-ingestion")]
     if config.is_nats_enabled() {
         handles.push(tokio::spawn({
             let config = config.clone();
             let mut redis_connection_manager = redis_connection_manager.clone();
             let nats_client = nats_client.clone().unwrap();
             async move {
-                startup::bridge_indexer::start_indexer_nats(
+                startup::nats_ingestion::start_indexer_nats(
                     &config,
                     &mut redis_connection_manager,
                     nats_client,
                 )
                 .await
             }
         }));
-    } else if config.is_bridge_indexer_enabled() {
+    }
+
+    #[cfg(feature = "mongo-ingestion")]
+    if config.is_bridge_indexer_enabled() {
         handles.push(tokio::spawn({
             let config = config.clone();
             let mut redis_connection_manager = redis_connection_manager.clone();
             async move {
-                startup::bridge_indexer::start_indexer(
+                startup::mongo_ingestion::start_indexer(
                     &config,
                     &mut redis_connection_manager,
                     args.start_timestamp,
                 )
                 .await
             }
         }));
-    } else {
+    }
+
+    #[cfg(feature = "native-indexers")]
+    {
         handles.push(tokio::spawn({
             let config = config.clone();
             let mut redis_connection_manager = redis_connection_manager.clone();
             let jsonrpc_client = jsonrpc_client.clone();
             async move {
-                startup::near::start_indexer(
+                startup::native_indexers::near::start_indexer(
                     &config,
                     &mut redis_connection_manager,
                     jsonrpc_client,
@@ -223,7 +238,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::evm::start_indexer(
+                    startup::native_indexers::evm::start_indexer(
                         &config,
                         &mut redis_connection_manager,
                         ChainKind::Eth,
@@ -238,7 +253,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::evm::start_indexer(
+                    startup::native_indexers::evm::start_indexer(
                         &config,
                         &mut redis_connection_manager,
                         ChainKind::Base,
@@ -253,7 +268,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::evm::start_indexer(
+                    startup::native_indexers::evm::start_indexer(
                         &config,
                         &mut redis_connection_manager,
                         ChainKind::Arb,
@@ -268,7 +283,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::evm::start_indexer(
+                    startup::native_indexers::evm::start_indexer(
                         &config,
                         &mut redis_connection_manager,
                         ChainKind::Bnb,
@@ -283,7 +298,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::evm::start_indexer(
+                    startup::native_indexers::evm::start_indexer(
                         &config,
                         &mut redis_connection_manager,
                         ChainKind::Pol,
@@ -298,7 +313,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::solana::start_indexer(
+                    startup::native_indexers::solana::start_indexer(
                         &config,
                         &mut redis_connection_manager,
                         args.solana_start_signature,
@@ -310,7 +325,7 @@ async fn main() -> Result<()> {
                 let config = config.clone();
                 let mut redis_connection_manager = redis_connection_manager.clone();
                 async move {
-                    startup::solana::process_signature(&config, &mut redis_connection_manager).await
+                    startup::native_indexers::solana::process_signature(&config, &mut redis_connection_manager).await
                 }
             }));
         }