feat: Implement client pooling for Webhook-based notifiers (#281)

* feat: Move generic client storage to a separate module

* feat: Add NotificationClientPool for managing HTTP clients in notifications

* test: Add unit tests for NotificationClientPool functionality

* feat: Extend NotificationClientPool to manage SMTP clients and improve error handling

* feat: make HTTP retry configuration serializable

* feat: add retry policy to TriggerTypeConfig and update related tests

* feat: add retry policy to TriggerTypeConfig and update related notification implementations

* feat: add client pool to NotificationService

* chore: formatting

* feat: add default serde attribute to retry_policy in TriggerTypeConfig variants

* fix: minor

* feat: refactor notification services to use a shared HTTP client

* feat: update WebhookNotifier to use a retryable HTTP client with TransientErrorRetryStrategy

* test: update integration tests to include retryable and non-retryable error scenarios

* feat: update TelegramNotifier to use POST method and JSON payload for webhook notifications

* minor

* feat: refactor EmailNotifier to accept an SMTP client and update notification service to use client pool

* feat: update NotificationClientPool to use ClientWithMiddleware

* test: refactor test to use test http client helpers

* feat: enhance SmtpConfig with additional traits and implement SMTP client creation in NotificationClientPool

* test: adjust error context assertions and add additional test

* test: add missing tests

* test: update retry expectations in notification tests to use dynamic retry count

* refactor: improve documentation for HTTP and SMTP client methods in NotificationClientPool

* minor

* refactor: remove unnecessary WebhookPayload struct and simplify payload handling

* test: remove redundant test

* Update src/services/notification/mod.rs

Co-authored-by: Nicolas Molina <[email protected]>

* refactor: improve error message for missing retry policy in trigger configuration

* fix: pass source error to trigger execution error

* docs: update notification protocols section

* docs: clarify base_for_backoff parameter description in retry policy

---------

Co-authored-by: Nicolas Molina <[email protected]>

Author: isSerge

docs/modules/ROOT/pages/index.adoc

@@ -202,6 +202,20 @@ The monitor implements protocol security validations across different components
 
 ===== Notification Protocols
 
+====== Webhook Notifications
+* *HTTPS Recommended*: URLs should use HTTPS protocol
+* *Authentication Recommended*: Including either:
+** `X-API-Key` header
+** `Authorization` header
+* *Optional Secret*: Can include a secret for HMAC authentication
+** When a secret is provided, the monitor will:
+*** Generate a timestamp in milliseconds
+*** Create an HMAC-SHA256 signature of the payload and timestamp
+*** Add the signature in the `X-Signature` header
+*** Add the timestamp in the `X-Timestamp` header
+** The signature is computed as: `HMAC-SHA256(secret, payload + timestamp)`
+* *Warning*: Non-HTTPS URLs or missing authentication headers will trigger security warnings
+
 ====== Slack Notifications
 * *HTTPS Recommended*: Webhook URLs should start with `https://hooks.slack.com/`
 * *Warning*: Non-HTTPS URLs will trigger security warnings
@@ -210,6 +224,14 @@ The monitor implements protocol security validations across different components
 * *HTTPS Recommended*: Webhook URLs should start with `https://discord.com/api/webhooks/`
 * *Warning*: Non-HTTPS URLs will trigger security warnings
 
+====== Telegram Notifications
+*   **Protocol:** `POST` request with a `application/json` payload to the `sendMessage` method.
+*   **Endpoint:** `https://api.telegram.org/bot<token>/sendMessage`
+*   **Security:**
+** **HTTPS Required:** The API endpoint uses HTTPS.
+**   Authentication is handled via the **Bot Token** in the URL. Keep this token secure.
+*   **Formatting:** Messages are sent with `parse_mode` set to `MarkdownV2`. Special characters in the message title and body are automatically escaped to prevent formatting errors.
+
 ====== Email Notifications
 * *Secure Ports Recommended*: The following ports are considered secure:
 ** 465: SMTPS (SMTP over SSL)
@@ -218,19 +240,27 @@ The monitor implements protocol security validations across different components
 * *Warning*: Using other ports will trigger security warnings
 * *Valid Format*: Email addresses must follow RFC 5322 format
 
-====== Webhook Notifications
-* *HTTPS Recommended*: URLs should use HTTPS protocol
-* *Authentication Recommended*: Including either:
-** `X-API-Key` header
-** `Authorization` header
-* *Optional Secret*: Can include a secret for HMAC authentication
-** When a secret is provided, the monitor will:
-*** Generate a timestamp in milliseconds
-*** Create an HMAC-SHA256 signature of the payload and timestamp
-*** Add the signature in the `X-Signature` header
-*** Add the timestamp in the `X-Timestamp` header
-** The signature is computed as: `HMAC-SHA256(secret, payload + timestamp)`
-* *Warning*: Non-HTTPS URLs or missing authentication headers will trigger security warnings
+====== Notifcations Retry Policy
+
+Following notification protocols support retry policies:
+
+* Slack
+* Discord
+* Telegram
+* Webhook
+
+Default retry policy is using exponential backoff with the following parameters:
+[cols="1,1,1"]
+|===
+| Parameter | Default Value | Description
+| `max_retries` | `3` | Maximum number of retries before giving up
+| `base_for_backoff` | `2` | Base duration for exponential backoff calculations in seconds
+| `initial_backoff` | `250` | Initial backoff duration in milliseconds
+| `max_backoff` | `10` | Maximum backoff duration in seconds
+| `jitter` | `Full` | Jitter strategy to apply to the backoff duration, currently supports `Full` and `None`
+|===
+
+These parameters can be overridden by providing custom `HttpRetryConfig` struct in `retry_policy` field in trigger configuration.
 
 ===== Script Security
 

docs/modules/ROOT/pages/rpc.adoc

@@ -147,7 +147,7 @@ pub struct HttpRetryConfig {
     /// Maximum backoff duration for retries.
     pub max_backoff: Duration,
     /// Base for the exponential backoff calculation (e.g., 2).
-    pub base_for_backoff: u64, // Added field
+    pub base_for_backoff: u64,
     /// Jitter to apply to the backoff duration.
     pub jitter: Jitter,
 }

src/bootstrap/mod.rs

@@ -391,7 +391,7 @@ pub fn create_trigger_handler<S: TriggerExecutionServiceTrait + Send + Sync + 's
 					let filtered_matches = run_trigger_filters(&block.processing_results, &block.network_slug, &trigger_scripts).await;
 					for monitor_match in &filtered_matches {
 						if let Err(e) = handle_match(monitor_match.clone(), &*trigger_service, &trigger_scripts).await {
-							TriggerError::execution_error(e.to_string(), None, None);
+							TriggerError::execution_error(e.to_string(), Some(e.into()), None);
 						}
 					}
 				} => {}

src/models/config/trigger_config.rs

@@ -254,7 +254,12 @@ impl ConfigLoader for Trigger {
 
 		match &self.trigger_type {
 			TriggerType::Slack => {
-				if let TriggerTypeConfig::Slack { slack_url, message } = &self.config {
+				if let TriggerTypeConfig::Slack {
+					slack_url,
+					message,
+					retry_policy: _,
+				} = &self.config
+				{
 					// Validate webhook URL
 					if !slack_url.starts_with("https://hooks.slack.com/") {
 						return Err(ConfigError::validation_error(
@@ -698,6 +703,7 @@ mod tests {
 	use crate::models::NotificationMessage;
 	use crate::models::{core::Trigger, ScriptLanguage, SecretString};
 	use crate::utils::tests::builders::trigger::TriggerBuilder;
+	use crate::utils::HttpRetryConfig;
 	use std::{fs::File, io::Write, os::unix::fs::PermissionsExt};
 	use tempfile::TempDir;
 	use tracing_test::traced_test;
@@ -1440,6 +1446,7 @@ mod tests {
 					title: "Test".to_string(),
 					body: "x".repeat(TELEGRAM_MAX_BODY_LENGTH + 1), // Exceeds max length
 				},
+				retry_policy: HttpRetryConfig::default(),
 			},
 		};
 		assert!(max_body_length.validate().is_err());
@@ -1458,6 +1465,7 @@ mod tests {
 					title: "Test".to_string(),
 					body: "z".repeat(DISCORD_MAX_BODY_LENGTH + 1), // Exceeds max length
 				},
+				retry_policy: HttpRetryConfig::default(),
 			},
 		};
 		assert!(max_body_length.validate().is_err());

src/models/core/trigger.rs

@@ -1,4 +1,7 @@
-use crate::models::{core::ScriptLanguage, SecretValue};
+use crate::{
+	models::{core::ScriptLanguage, SecretValue},
+	utils::HttpRetryConfig,
+};
 use email_address::EmailAddress;
 use serde::{Deserialize, Serialize};
 
@@ -56,6 +59,9 @@ pub enum TriggerTypeConfig {
 		slack_url: SecretValue,
 		/// Notification message
 		message: NotificationMessage,
+		/// Retry policy for HTTP requests
+		#[serde(default)]
+		retry_policy: HttpRetryConfig,
 	},
 	/// Email notification configuration
 	Email {
@@ -86,6 +92,9 @@ pub enum TriggerTypeConfig {
 		headers: Option<std::collections::HashMap<String, String>>,
 		/// Notification message
 		message: NotificationMessage,
+		/// Retry policy for HTTP requests
+		#[serde(default)]
+		retry_policy: HttpRetryConfig,
 	},
 	/// Telegram notification configuration
 	Telegram {
@@ -97,13 +106,19 @@ pub enum TriggerTypeConfig {
 		disable_web_preview: Option<bool>,
 		/// Notification message
 		message: NotificationMessage,
+		/// Retry policy for HTTP requests
+		#[serde(default)]
+		retry_policy: HttpRetryConfig,
 	},
 	/// Discord notification configuration
 	Discord {
 		/// Discord webhook URL
 		discord_url: SecretValue,
 		/// Notification message
 		message: NotificationMessage,
+		/// Retry policy for HTTP requests
+		#[serde(default)]
+		retry_policy: HttpRetryConfig,
 	},
 	/// Script execution configuration
 	Script {
@@ -118,3 +133,16 @@ pub enum TriggerTypeConfig {
 		timeout_ms: u32,
 	},
 }
+
+impl TriggerTypeConfig {
+	/// Get the retry policy for the trigger type, if applicable.
+	pub fn get_retry_policy(&self) -> Option<HttpRetryConfig> {
+		match self {
+			Self::Slack { retry_policy, .. } => Some(retry_policy.clone()),
+			Self::Discord { retry_policy, .. } => Some(retry_policy.clone()),
+			Self::Webhook { retry_policy, .. } => Some(retry_policy.clone()),
+			Self::Telegram { retry_policy, .. } => Some(retry_policy.clone()),
+			_ => None,
+		}
+	}
+}

src/services/blockchain/pool.rs

@@ -10,6 +10,7 @@
 //! The pool uses a fast path for existing clients and a slow path for
 //! creating new ones, optimizing performance while maintaining safety.
 
+use crate::utils::client_storage::ClientStorage;
 use crate::{
 	models::{BlockChainType, Network},
 	services::blockchain::{
@@ -21,7 +22,6 @@ use anyhow::Context;
 use async_trait::async_trait;
 use futures::future::BoxFuture;
 use std::{any::Any, collections::HashMap, sync::Arc};
-use tokio::sync::RwLock;
 
 /// Trait for the client pool.
 #[async_trait]
@@ -41,22 +41,6 @@ pub trait ClientPoolTrait: Send + Sync {
 	) -> Result<Arc<Self::StellarClient>, anyhow::Error>;
 }
 
-/// Generic client storage that can hold any type of blockchain client
-///
-/// Clients are stored in a thread-safe way using a HashMap and an RwLock.
-/// The HashMap is indexed by the network slug and the value is an Arc of the client.
-pub struct ClientStorage<T> {
-	clients: Arc<RwLock<HashMap<String, Arc<T>>>>,
-}
-
-impl<T> ClientStorage<T> {
-	pub fn new() -> Self {
-		Self {
-			clients: Arc::new(RwLock::new(HashMap::new())),
-		}
-	}
-}
-
 /// Main client pool manager that handles multiple blockchain types.
 ///
 /// Provides type-safe access to cached blockchain clients. Clients are created

src/services/notification/discord.rs

@@ -4,9 +4,10 @@
 //! via incoming webhooks, supporting message templates with variable substitution.
 
 use async_trait::async_trait;
+use reqwest_middleware::ClientWithMiddleware;
 use serde::Serialize;
 use serde_json;
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
 use crate::{
 	models::TriggerTypeConfig,
@@ -77,22 +78,26 @@ impl DiscordNotifier {
 	/// * `url` - Discord webhook URL
 	/// * `title` - Message title
 	/// * `body_template` - Message template with variables
+	/// * `http_client` - HTTP client with middleware for retries
 	pub fn new(
 		url: String,
 		title: String,
 		body_template: String,
+		http_client: Arc<ClientWithMiddleware>,
 	) -> Result<Self, NotificationError> {
+		let config = WebhookConfig {
+			url,
+			url_params: None,
+			title,
+			body_template,
+			method: Some("POST".to_string()),
+			secret: None,
+			headers: None,
+			payload_fields: None,
+		};
+
 		Ok(Self {
-			inner: WebhookNotifier::new(WebhookConfig {
-				url,
-				url_params: None,
-				title,
-				body_template,
-				method: Some("POST".to_string()),
-				secret: None,
-				headers: None,
-				payload_fields: None,
-			})?,
+			inner: WebhookNotifier::new(config, http_client)?,
 		})
 	}
 
@@ -112,13 +117,18 @@ impl DiscordNotifier {
 	///
 	/// # Arguments
 	/// * `config` - Trigger configuration containing Discord parameters
+	/// * `http_client` - HTTP client with middleware for retries
 	///
 	/// # Returns
 	/// * `Result<Self, NotificationError>` - Notifier instance if config is Discord type
-	pub fn from_config(config: &TriggerTypeConfig) -> Result<Self, NotificationError> {
+	pub fn from_config(
+		config: &TriggerTypeConfig,
+		http_client: Arc<ClientWithMiddleware>,
+	) -> Result<Self, NotificationError> {
 		if let TriggerTypeConfig::Discord {
 			discord_url,
 			message,
+			..
 		} = config
 		{
 			let webhook_config = WebhookConfig {
@@ -133,7 +143,7 @@ impl DiscordNotifier {
 			};
 
 			Ok(Self {
-				inner: WebhookNotifier::new(webhook_config)?,
+				inner: WebhookNotifier::new(webhook_config, http_client)?,
 			})
 		} else {
 			let msg = format!("Invalid discord configuration: {:?}", config);
@@ -173,7 +183,10 @@ impl Notifier for DiscordNotifier {
 
 #[cfg(test)]
 mod tests {
-	use crate::models::{NotificationMessage, SecretString, SecretValue};
+	use crate::{
+		models::{NotificationMessage, SecretString, SecretValue},
+		utils::{tests::create_test_http_client, HttpRetryConfig},
+	};
 
 	use super::*;
 
@@ -182,6 +195,7 @@ mod tests {
 			"https://non-existent-url-discord-webhook.com".to_string(),
 			"Alert".to_string(),
 			body_template.to_string(),
+			create_test_http_client(),
 		)
 		.unwrap()
 	}
@@ -195,6 +209,7 @@ mod tests {
 				title: "Test Alert".to_string(),
 				body: "Test message ${value}".to_string(),
 			},
+			retry_policy: HttpRetryConfig::default(),
 		}
 	}
 
@@ -242,8 +257,8 @@ mod tests {
 	#[test]
 	fn test_from_config_with_discord_config() {
 		let config = create_test_discord_config();
-
-		let notifier = DiscordNotifier::from_config(&config);
+		let http_client = create_test_http_client();
+		let notifier = DiscordNotifier::from_config(&config, http_client);
 		assert!(notifier.is_ok());
 
 		let notifier = notifier.unwrap();
@@ -252,24 +267,6 @@ mod tests {
 		assert_eq!(notifier.inner.body_template, "Test message ${value}");
 	}
 
-	#[test]
-	fn test_from_config_invalid_type() {
-		// Create a Slack config instead of Discord
-		let config = TriggerTypeConfig::Slack {
-			slack_url: SecretValue::Plain(SecretString::new("random.url".to_string())),
-			message: NotificationMessage {
-				title: "Test Slack".to_string(),
-				body: "This is a test message".to_string(),
-			},
-		};
-
-		let notifier = DiscordNotifier::from_config(&config);
-		assert!(notifier.is_err());
-
-		let error = notifier.unwrap_err();
-		assert!(matches!(error, NotificationError::ConfigError { .. }));
-	}
-
 	////////////////////////////////////////////////////////////
 	// notify tests
 	////////////////////////////////////////////////////////////