Feat : alerts module

.env.example

@@ -1,50 +1,65 @@
 HOST=
 PORT=
 DATABASE_URL=
-MADARA_RPC_URL=
-DA_LAYER=
-SETTLEMENT_LAYER=
-
-# Ethereum
-ETHEREUM_PRIVATE_KEY=
-ETHEREUM_RPC_URL=
-MEMORY_PAGES_CONTRACT_ADDRESS=
-STARKNET_SOLIDITY_CORE_CONTRACT_ADDRESS=
-
+##### ORCHESTRATOR #####
 
-# Starknet
-STARKNET_PUBLIC_KEY=
-STARNET_PRIVATE_KEY=
-STARKNET_RPC_URL=
-STARKNET_CAIRO_CORE_CONTRACT_ADDRESS=
+HOST=
+PORT=
 
-# MongoDB connection string
-MONGODB_CONNECTION_STRING=
+##### AWS CONFIG #####
 
-# AWS
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
-AWS_DEFAULT_REGION=
+AWS_REGION=
+AWS_ENDPOINT_URL=
+
+##### STORAGE #####
 
-# SQS
+DATA_STORAGE=
+AWS_S3_BUCKET_NAME=
+
+##### QUEUE #####
+
+QUEUE_PROVIDER=
 SQS_JOB_PROCESSING_QUEUE_URL=
 SQS_JOB_VERIFICATION_QUEUE_URL=
 SQS_JOB_HANDLE_FAILURE_QUEUE_URL=
 SQS_WORKER_TRIGGER_QUEUE_URL=
 
-# S3
-AWS_S3_BUCKET_NAME=
-AWS_S3_BUCKET_REGION=
+##### DATABASE #####
 
+DATABASE=
+MONGODB_CONNECTION_STRING=
 
-# Ethereum Settlement
-DEFAULT_SETTLEMENT_CLIENT_RPC=
-DEFAULT_L1_CORE_CONTRACT_ADDRESS=
+##### PROVER #####
 
-# Sharp Services
+PROVER_SERVICE=
 SHARP_CUSTOMER_ID=
+SHARP_URL=
+# [IMP!!!] These are test certificates (they don't work)
 SHARP_USER_CRT=
 SHARP_USER_KEY=
 SHARP_SERVER_CRT=
 SHARP_PROOF_LAYOUT=
 
+##### ON CHAIN CONFIG #####
+
+DA_LAYER=
+SETTLEMENT_LAYER=
+ETHEREUM_RPC_URL=
+MADARA_RPC_URL=
+MEMORY_PAGES_CONTRACT_ADDRESS=
+PRIVATE_KEY=
+ETHEREUM_PRIVATE_KEY=
+STARKNET_SOLIDITY_CORE_CONTRACT_ADDRESS=
+
+##### ALERTS #####
+
+AWS_SNS_REGION=
+AWS_SNS_ARN=
+AWS_SNS_ARN_NAME=
+
+##### SETTLEMENT CLIENT #####
+
+DEFAULT_SETTLEMENT_CLIENT_RPC=
+DEFAULT_L1_CORE_CONTRACT_ADDRESS=

.env.test

@@ -10,6 +10,9 @@ SQS_JOB_VERIFICATION_QUEUE_URL="http://sqs.us-east-1.localhost.localstack.cloud:
 SQS_JOB_HANDLE_FAILURE_QUEUE_URL=
 SQS_WORKER_TRIGGER_QUEUE_URL=
 AWS_DEFAULT_REGION="localhost"
+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"
 
 ##### On chain config #####
 
@@ -27,6 +30,7 @@ DA_LAYER="ethereum"
 PROVER_SERVICE="sharp"
 SETTLEMENT_LAYER="ethereum"
 DATA_STORAGE="s3"
+ALERTS="sns"
 MONGODB_CONNECTION_STRING="mongodb://localhost:27017"
 DEFAULT_SETTLEMENT_CLIENT_RPC="http://localhost:3000"
 

.github/workflows/coverage.yml

@@ -14,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"

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.

crates/orchestrator/Cargo.toml

@@ -19,6 +19,7 @@ async-std = "1.12.0"
 async-trait = { workspace = true }
 aws-config = { version = "1.1.7", features = ["behavior-version-latest"] }
 aws-sdk-s3 = { version = "1.38.0", features = ["behavior-version-latest"] }
+aws-sdk-sns = { version = "1.40.0", features = ["behavior-version-latest"] }
 aws-sdk-sqs = "1.36.0"
 axum = { workspace = true, features = ["macros"] }
 axum-macros = { 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

@@ -1,5 +1,7 @@
 use std::sync::Arc;
 
+use crate::alerts::aws_sns::AWSSNS;
+use crate::alerts::Alerts;
 use crate::data_storage::aws_s3::config::{AWSS3Config, AWSS3ConfigType};
 use crate::data_storage::aws_s3::AWSS3;
 use crate::data_storage::{DataStorage, DataStorageConfig};
@@ -42,6 +44,8 @@ pub struct Config {
     queue: Box<dyn QueueProvider>,
     /// Storage client
     storage: Box<dyn DataStorage>,
+    /// Alerts client
+    alerts: Box<dyn Alerts>,
 }
 
 /// Initializes the app config
@@ -67,11 +71,23 @@ pub async fn init_config() -> Config {
 
     let storage_client = build_storage_client().await;
 
-    Config::new(Arc::new(provider), da_client, prover_client, settlement_client, database, queue, storage_client)
+    let alerts_client = build_alert_client().await;
+
+    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>,
@@ -80,8 +96,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
@@ -118,6 +135,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.
@@ -183,3 +205,10 @@ pub async fn build_storage_client() -> Box<dyn DataStorage + Send + Sync> {
         _ => 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"),
+    }
+}

crates/orchestrator/src/jobs/mod.rs

@@ -139,7 +139,8 @@ pub async fn create_job(
     }
 
     let job_handler = factory::get_job_handler(&job_type).await;
-    let job_item = job_handler.create_job(config.as_ref(), internal_id, metadata).await?;
+    let job_item = job_handler.create_job(config.as_ref(), internal_id.clone(), metadata.clone()).await?;
+
     config.database().create_job(job_item.clone()).await.map_err(|e| JobError::Other(OtherError(e)))?;
 
     add_job_to_process_queue(job_item.id).await.map_err(|e| JobError::Other(OtherError(e)))?;
@@ -173,6 +174,7 @@ pub async fn process_job(id: Uuid) -> Result<(), JobError> {
 
     let job_handler = factory::get_job_handler(&job.job_type).await;
     let external_id = job_handler.process_job(config.as_ref(), &mut job).await?;
+
     let metadata = increment_key_in_metadata(&job.metadata, JOB_PROCESS_ATTEMPT_METADATA_KEY)?;
 
     // Fetching the job again because update status above will update the job version
@@ -239,16 +241,13 @@ pub async fn verify_job(id: Uuid) -> Result<(), JobError> {
                 );
                 add_job_to_process_queue(job.id).await.map_err(|e| JobError::Other(OtherError(e)))?;
                 return Ok(());
-            } else {
-                // TODO: send alert
             }
         }
         JobVerificationStatus::Pending => {
             log::info!("Inclusion is still pending for job {}. Pushing back to queue.", job.id);
             let verify_attempts = get_u64_from_metadata(&job.metadata, JOB_VERIFICATION_ATTEMPT_METADATA_KEY)
                 .map_err(|e| JobError::Other(OtherError(e)))?;
             if verify_attempts >= job_handler.max_verification_attempts() {
-                // TODO: send alert
                 log::info!("Verification attempts exceeded for job {}. Marking as timed out.", job.id);
                 config
                     .database()

crates/orchestrator/src/lib.rs

@@ -1,3 +1,5 @@
+/// Contains the trait implementations for alerts
+pub mod alerts;
 /// Config of the service. Contains configurations for DB, Queues and other services.
 pub mod config;
 pub mod constants;