Settlement Engine

Cargo.toml

@@ -1,4 +1,5 @@
 [workspace]
+resolver = "2"
 
 members = [
     "crates/config",

Dockerfile

@@ -8,4 +8,3 @@ RUN apt-get update
 RUN apt-get upgrade -y
 RUN apt-get install -y libssl-dev ca-certificates
 COPY --from=builder /usr/local/cargo/bin/reflux /app/reflux
-

README.md

@@ -1,6 +1,7 @@
 ## reflux
 
-Backend of solver which helps in seamless cross-chain asset consolidation. It aggregates user balances, automates routing, and suggests optimal transactions.
+Backend of solver which helps in seamless cross-chain asset consolidation. It aggregates user balances, automates
+routing, and suggests optimal transactions.
 
 #### Installation
 
@@ -18,4 +19,4 @@ Once build is copleted, just run the server and test with the endpoints
 
 ### Dependencies graph
 
-![image](./graph.png)
+![image](./assets/dependency-graph.png)

bin/reflux/src/main.rs

@@ -1,21 +1,20 @@
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
 use axum::http::Method;
 use clap::Parser;
 use log::{debug, error, info};
-use tokio::signal;
-use tokio::sync::broadcast;
 use tower_http::cors::{Any, CorsLayer};
 
 use account_aggregation::service::AccountAggregationService;
 use api::service_controller::ServiceController;
 use config::Config;
-use routing_engine::engine::RoutingEngine;
-use routing_engine::estimator::LinearRegressionEstimator;
 use routing_engine::{BungeeClient, CoingeckoClient, Indexer};
-use storage::mongodb_client::MongoDBClient;
+use routing_engine::estimator::LinearRegressionEstimator;
+use routing_engine::routing_engine::RoutingEngine;
 use storage::{ControlFlow, MessageQueue, RedisClient};
+use storage::mongodb_client::MongoDBClient;
 
 #[derive(Parser, Debug)]
 struct Args {
@@ -98,6 +97,8 @@ async fn run_solver(config: Config) {
 
     // Initialize routing engine
     let buckets = config.buckets.clone();
+    let chain_configs = config.chains.clone();
+    let token_configs = config.tokens.clone();
     let redis_client = RedisClient::build(&config.infra.redis_url)
         .await
         .expect("Failed to instantiate redis client");
@@ -106,15 +107,15 @@ async fn run_solver(config: Config) {
         buckets,
         redis_client.clone(),
         config.solver_config,
+        chain_configs,
+        token_configs,
     ));
 
     // Subscribe to cache update messages
     let cache_update_topic = config.indexer_config.indexer_update_topic.clone();
     let routing_engine_clone = Arc::clone(&routing_engine);
 
-    let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1);
-
-    let cache_update_handle = tokio::spawn(async move {
+    tokio::task::spawn_blocking(move || {
         let redis_client = redis_client.clone();
         if let Err(e) = redis_client.subscribe(&cache_update_topic, move |_msg| {
             info!("Received cache update notification");
@@ -126,33 +127,46 @@ async fn run_solver(config: Config) {
         }) {
             error!("Failed to subscribe to cache update topic: {}", e);
         }
-
-        // Listen for shutdown signal
-        let _ = shutdown_rx.recv().await;
     });
 
+    let token_chain_map: HashMap<String, HashMap<u32, bool>> = config
+        .tokens
+        .iter()
+        .map(|(token, token_config)| {
+            let chain_supported = token_config
+                .by_chain
+                .iter()
+                .map(|(chain_id, chain_config)| (*chain_id, chain_config.is_enabled))
+                .collect();
+            (token.clone(), chain_supported)
+        })
+        .collect();
+
     // API service controller
-    let service_controller = ServiceController::new(account_service, routing_engine);
+    let chain_supported: Vec<(u32, String)> =
+        config.chains.iter().map(|(id, chain)| (*id, chain.name.clone())).collect();
+    let token_supported: Vec<String> =
+        config.tokens.iter().map(|(_, token_config)| token_config.symbol.clone()).collect();
+    let service_controller = ServiceController::new(
+        account_service,
+        routing_engine,
+        token_chain_map,
+        chain_supported,
+        token_supported,
+    );
 
-    let cors = CorsLayer::new().allow_origin(Any).allow_methods([
-        Method::GET,
-        Method::POST,
-        Method::PATCH,
-    ]);
+    let cors = CorsLayer::new()
+        .allow_origin(Any)
+        .allow_methods([Method::GET, Method::POST, Method::PATCH])
+        .allow_headers(Any);
 
     let app = service_controller.router().layer(cors);
 
     let listener = tokio::net::TcpListener::bind(format!("{}:{}", app_host, app_port))
         .await
         .expect("Failed to bind port");
 
-    axum::serve(listener, app.into_make_service())
-        .with_graceful_shutdown(shutdown_signal(shutdown_tx.clone()))
-        .await
-        .unwrap();
-
-    let _ = shutdown_tx.send(());
-    let _ = cache_update_handle.abort();
+    axum::serve(listener, app.into_make_service()).await.unwrap();
 
     info!("Server stopped.");
 }
@@ -170,9 +184,9 @@ async fn run_indexer(config: Config) {
         .expect("Failed to Instantiate Bungee Client");
 
     let token_price_provider = CoingeckoClient::new(
-        &config.coingecko.base_url,
-        &config.coingecko.api_key,
-        &redis_provider,
+        config.coingecko.base_url.clone(),
+        config.coingecko.api_key.clone(),
+        redis_provider.clone(),
         Duration::from_secs(config.coingecko.expiry_sec),
     );
 
@@ -190,25 +204,3 @@ async fn run_indexer(config: Config) {
         Err(e) => error!("Indexer Job Failed: {}", e),
     };
 }
-
-async fn shutdown_signal(shutdown_tx: broadcast::Sender<()>) {
-    let ctrl_c = async {
-        signal::ctrl_c().await.expect("Unable to handle ctrl+c");
-    };
-    #[cfg(unix)]
-    let terminate = async {
-        signal::unix::signal(signal::unix::SignalKind::terminate())
-            .expect("Failed to install signal handler")
-            .recv()
-            .await;
-    };
-    #[cfg(not(unix))]
-    let terminate = std::future::pending::<()>();
-    tokio::select! {
-        _ = ctrl_c => {},
-        _ = terminate => {},
-    }
-
-    info!("signal received, starting graceful shutdown");
-    let _ = shutdown_tx.send(());
-}