Merge branch 'main' into tests/e2e-test-main

.env.test

@@ -25,7 +25,11 @@ SQS_JOB_HANDLE_FAILURE_QUEUE_URL="http://sqs.us-east-1.localhost.localstack.clou
 SQS_WORKER_TRIGGER_QUEUE_URL="http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/madara_orchestrator_worker_trigger_queue"
 
 ##### SNS #####
+
 ALERTS="sns"
+AWS_SNS_REGION="us-east-1"
+AWS_SNS_ARN="arn:aws:sns:us-east-1:000000000000:madara-orchestrator-arn"
+AWS_SNS_ARN_NAME="madara-orchestrator-arn"
 
 ##### DATABASE #####
 
@@ -57,6 +61,7 @@ STARKNET_SOLIDITY_CORE_CONTRACT_ADDRESS="0xc662c410C0ECf747543f5bA90660f6ABeBD9C
 DEFAULT_L1_CORE_CONTRACT_ADDRESS="0xc662c410C0ECf747543f5bA90660f6ABeBD9C8c4"
 TEST_DUMMY_CONTRACT_ADDRESS="0xE5b6F5e695BA6E4aeD92B68c4CC8Df1160D69A81"
 STARKNET_OPERATOR_ADDRESS="0x2C169DFe5fBbA12957Bdd0Ba47d9CEDbFE260CA7"
+ETHEREUM_BLAST_RPC_URL="https://eth-mainnet.public.blastapi.io"
 
 ##### E2E test vars #####
 

.github/workflows/coverage.yml

@@ -1,15 +1,10 @@
 name: Rust Test & Coverage
 
 on:
-  pull_request_target:
-    branches:
-      - main
-    types: [opened, synchronize, reopened]
-  push:
-    branches-ignore:
-      - main
   workflow_call:
-  workflow_dispatch:
+    secrets:
+      ETHEREUM_BLAST_RPC_URL:
+        required: true
 
 jobs:
   coverage:
@@ -19,7 +14,7 @@ jobs:
       localstack:
         image: localstack/localstack
         env:
-          SERVICES: s3, sqs
+          SERVICES: s3, sqs, sns
           DEFAULT_REGION: us-east-1
           AWS_ACCESS_KEY_ID: "AWS_ACCESS_KEY_ID"
           AWS_SECRET_ACCESS_KEY: "AWS_SECRET_ACCESS_KEY"
@@ -74,7 +69,7 @@ jobs:
       - name: Run llvm-cov tests
         env:
           ETHEREUM_BLAST_RPC_URL: ${{ secrets.ETHEREUM_BLAST_RPC_URL }}
-        run: cargo llvm-cov nextest --release --lcov --output-path lcov.info --test-threads=1
+        run: RUST_LOG=debug RUST_BACKTRACE=1 cargo llvm-cov nextest --release --lcov --output-path lcov.info --test-threads=1
 
       - name: Coveralls
         uses: coverallsapp/github-action@v2

.github/workflows/linters-cargo.yml

@@ -4,7 +4,6 @@ name: Task - Linters Cargo
 on:
   workflow_dispatch:
   workflow_call:
-  push:
 
 jobs:
   cargo-lint:

.github/workflows/linters.yml

@@ -4,7 +4,6 @@ name: Task - Linters
 on:
   workflow_dispatch:
   workflow_call:
-  push:
 
 jobs:
   prettier:

.github/workflows/pull-request.yml

@@ -2,7 +2,8 @@
 name: Workflow - Pull Request
 
 on:
-  pull_request_target:
+  workflow_dispatch:
+  pull_request:
     branches: [main]
   push:
     branches: [main]
@@ -24,3 +25,4 @@ jobs:
     name: Run coverage
     uses: ./.github/workflows/coverage.yml
     needs: rust_build
+    secrets: inherit

.github/workflows/rust-build.yml

@@ -4,7 +4,6 @@ name: Task - Build Rust
 on:
   workflow_dispatch:
   workflow_call:
-  push:
 
 jobs:
   rust_build:

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Added
 
+- alerts module.
 - Tests for Settlement client.
 - Worker queues to listen for trigger events.
 - Tests for prover client.
@@ -30,13 +31,15 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Changed
 
+- refactor AWS config usage and clean .env files
 - GitHub's coverage CI yml file for localstack and db testing.
 - Orchestrator :Moved TestConfigBuilder to `config.rs` in tests folder.
 - `.env` file requires two more variables which are queue urls for processing
   and verification.
 
 ## Removed
 
+- revert CI changes from settlement client PR.
 - `init_config` from all the tests.
 - `fetch_from_test` argument
 

crates/da-clients/ethereum/src/config.rs

@@ -19,14 +19,14 @@ pub struct EthereumDaConfig {
 impl DaConfig<EthereumDaClient> for EthereumDaConfig {
     fn new_from_env() -> Self {
         Self {
-            rpc_url: get_env_var_or_panic("ETHEREUM_RPC_URL"),
+            rpc_url: get_env_var_or_panic("SETTLEMENT_RPC_URL"),
             memory_pages_contract: get_env_var_or_panic("MEMORY_PAGES_CONTRACT_ADDRESS"),
             private_key: get_env_var_or_panic("PRIVATE_KEY"),
         }
     }
     async fn build_client(&self) -> EthereumDaClient {
         let client =
-            RpcClient::new_http(Url::from_str(self.rpc_url.as_str()).expect("Failed to parse ETHEREUM_RPC_URL"));
+            RpcClient::new_http(Url::from_str(self.rpc_url.as_str()).expect("Failed to parse SETTLEMENT_RPC_URL"));
         let provider = ProviderBuilder::<_, Ethereum>::new().on_client(client);
 
         EthereumDaClient { provider }

crates/orchestrator/Cargo.toml

@@ -20,6 +20,7 @@ async-trait = { workspace = true }
 aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }
 aws-sdk-sqs = { workspace = true }
+aws-sdk-sns = { version = "1.40.0", features = ["behavior-version-latest"] }
 axum = { workspace = true, features = ["macros"] }
 axum-macros = { workspace = true }
 bincode = { workspace = true }

crates/orchestrator/src/alerts/aws_sns/mod.rs

@@ -0,0 +1,27 @@
+use crate::alerts::Alerts;
+use async_trait::async_trait;
+use aws_sdk_sns::config::Region;
+use aws_sdk_sns::Client;
+use utils::env_utils::get_env_var_or_panic;
+
+pub struct AWSSNS {
+    client: Client,
+}
+
+impl AWSSNS {
+    /// To create a new SNS client
+    pub async fn new() -> Self {
+        let sns_region = get_env_var_or_panic("AWS_SNS_REGION");
+        let config = aws_config::from_env().region(Region::new(sns_region)).load().await;
+        AWSSNS { client: Client::new(&config) }
+    }
+}
+
+#[async_trait]
+impl Alerts for AWSSNS {
+    async fn send_alert_message(&self, message_body: String) -> color_eyre::Result<()> {
+        let topic_arn = get_env_var_or_panic("AWS_SNS_ARN");
+        self.client.publish().topic_arn(topic_arn).message(message_body).send().await?;
+        Ok(())
+    }
+}

crates/orchestrator/src/alerts/mod.rs

@@ -0,0 +1,11 @@
+use async_trait::async_trait;
+use mockall::automock;
+
+pub mod aws_sns;
+
+#[automock]
+#[async_trait]
+pub trait Alerts: Send + Sync {
+    /// To send an alert message to our alert service
+    async fn send_alert_message(&self, message_body: String) -> color_eyre::Result<()>;
+}

crates/orchestrator/src/config.rs

@@ -3,10 +3,13 @@ use alloy::providers::RootProvider;
 use std::str::FromStr;
 use std::sync::Arc;
 
-use crate::data_storage::aws_s3::config::{AWSS3Config, AWSS3ConfigType, S3LocalStackConfig};
+use crate::alerts::aws_sns::AWSSNS;
+use crate::alerts::Alerts;
+use crate::data_storage::aws_s3::config::AWSS3Config;
 use crate::data_storage::aws_s3::AWSS3;
 use crate::data_storage::{DataStorage, DataStorageConfig};
 use arc_swap::{ArcSwap, Guard};
+use aws_config::SdkConfig;
 use da_client_interface::{DaClient, DaConfig};
 use dotenvy::dotenv;
 use ethereum_da_client::config::EthereumDaConfig;
@@ -45,6 +48,8 @@ pub struct Config {
     queue: Box<dyn QueueProvider>,
     /// Storage client
     storage: Box<dyn DataStorage>,
+    /// Alerts client
+    alerts: Box<dyn Alerts>,
 }
 
 /// Initializes the app config
@@ -57,24 +62,42 @@ pub async fn init_config() -> Config {
     ));
 
     // init database
-    let database = Box::new(MongoDb::new(MongoDbConfig::new_from_env()).await);
+    let database = build_database_client().await;
+
+    // init AWS
+    let aws_config = aws_config::load_from_env().await;
 
     // init the queue
-    let queue = Box::new(SqsQueue {});
+    // TODO: we use omniqueue for now which doesn't support loading AWS config
+    // from `SdkConfig`. We can later move to using `aws_sdk_sqs`. This would require
+    // us stop using the generic omniqueue abstractions for message ack/nack
+    let queue = build_queue_client(&aws_config);
 
     let da_client = build_da_client().await;
 
     let settings_provider = DefaultSettingsProvider {};
     let settlement_client = build_settlement_client(&settings_provider).await;
     let prover_client = build_prover_service(&settings_provider);
 
-    let storage_client = build_storage_client().await;
+    let storage_client = build_storage_client(&aws_config).await;
+
+    let alerts_client = build_alert_client().await;
 
-    Config::new(Arc::new(provider), da_client, prover_client, settlement_client, database, queue, storage_client)
+    Config::new(
+        Arc::new(provider),
+        da_client,
+        prover_client,
+        settlement_client,
+        database,
+        queue,
+        storage_client,
+        alerts_client,
+    )
 }
 
 impl Config {
     /// Create a new config
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         starknet_client: Arc<JsonRpcClient<HttpTransport>>,
         da_client: Box<dyn DaClient>,
@@ -83,8 +106,9 @@ impl Config {
         database: Box<dyn Database>,
         queue: Box<dyn QueueProvider>,
         storage: Box<dyn DataStorage>,
+        alerts: Box<dyn Alerts>,
     ) -> Self {
-        Self { starknet_client, da_client, prover_client, settlement_client, database, queue, storage }
+        Self { starknet_client, da_client, prover_client, settlement_client, database, queue, storage, alerts }
     }
 
     /// Returns the starknet client
@@ -121,6 +145,11 @@ impl Config {
     pub fn storage(&self) -> &dyn DataStorage {
         self.storage.as_ref()
     }
+
+    /// Returns the alerts client
+    pub fn alerts(&self) -> &dyn Alerts {
+        self.alerts.as_ref()
+    }
 }
 
 /// The app config. It can be accessed from anywhere inside the service.
@@ -186,12 +215,29 @@ pub async fn build_settlement_client(
     }
 }
 
-pub async fn build_storage_client() -> Box<dyn DataStorage + Send + Sync> {
+pub async fn build_storage_client(aws_config: &SdkConfig) -> Box<dyn DataStorage + Send + Sync> {
     match get_env_var_or_panic("DATA_STORAGE").as_str() {
-        "s3" => Box::new(AWSS3::new(AWSS3ConfigType::WithoutEndpoint(AWSS3Config::new_from_env())).await),
-        "s3_localstack" => {
-            Box::new(AWSS3::new(AWSS3ConfigType::WithEndpoint(S3LocalStackConfig::new_from_env())).await)
-        }
+        "s3" => Box::new(AWSS3::new(AWSS3Config::new_from_env(), aws_config)),
         _ => panic!("Unsupported Storage Client"),
     }
 }
+
+pub async fn build_alert_client() -> Box<dyn Alerts + Send + Sync> {
+    match get_env_var_or_panic("ALERTS").as_str() {
+        "sns" => Box::new(AWSSNS::new().await),
+        _ => panic!("Unsupported Alert Client"),
+    }
+}
+pub fn build_queue_client(_aws_config: &SdkConfig) -> Box<dyn QueueProvider + Send + Sync> {
+    match get_env_var_or_panic("QUEUE_PROVIDER").as_str() {
+        "sqs" => Box::new(SqsQueue {}),
+        _ => panic!("Unsupported Queue Client"),
+    }
+}
+
+pub async fn build_database_client() -> Box<dyn Database + Send + Sync> {
+    match get_env_var_or_panic("DATABASE").as_str() {
+        "mongodb" => Box::new(MongoDb::new(MongoDbConfig::new_from_env()).await),
+        _ => panic!("Unsupported Database Client"),
+    }
+}