diff --git a/Cargo.lock b/Cargo.lock
index a55e8ea60..780bca8a5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7499,8 +7499,8 @@ dependencies = [
  "config",
  "dotenvy",
  "ecstore",
- "http",
  "lazy_static",
+ "once_cell",
  "rdkafka",
  "reqwest",
  "rumqttc",
@@ -7508,6 +7508,7 @@ dependencies = [
  "serde_json",
  "serde_with",
  "smallvec",
+ "snap",
  "strum",
  "thiserror 2.0.12",
  "tokio",
diff --git a/Cargo.toml b/Cargo.toml
index 4b0e1af84..fe5fe27ce 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -109,6 +109,7 @@ nix = { version = "0.30.1", features = ["fs"] }
 num_cpus = { version = "1.16.0" }
 nvml-wrapper = "0.10.0"
 object_store = "0.11.2"
+once_cell = "1.21.3"
 opentelemetry = { version = "0.29.1" }
 opentelemetry-appender-tracing = { version = "0.29.1", features = [
     "experimental_use_tracing_span_context",
@@ -162,6 +163,7 @@ serde_with = "3.12.0"
 sha2 = "0.10.9"
 smallvec = { version = "1.15.0", features = ["serde"] }
 snafu = "0.8.5"
+snap = "1.1.1"
 socket2 = "0.5.9"
 strum = { version = "0.27.1", features = ["derive"] }
 sysinfo = "0.34.2"
diff --git a/crates/event/Cargo.toml b/crates/event/Cargo.toml
index 752ea7faa..8256f54ff 100644
--- a/crates/event/Cargo.toml
+++ b/crates/event/Cargo.toml
@@ -17,8 +17,8 @@ async-trait = { workspace = true }
 config = { workspace = true }
 common = { workspace = true }
 ecstore = { workspace = true }
-http = { workspace = true }
 lazy_static = { workspace = true }
+once_cell = { workspace = true }
 reqwest = { workspace = true, optional = true }
 rumqttc = { workspace = true, optional = true }
 serde = { workspace = true }
@@ -31,6 +31,8 @@ thiserror = { workspace = true }
 tokio = { workspace = true, features = ["sync", "net", "macros", "signal", "rt-multi-thread"] }
 tokio-util = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde"] }
+snap = { workspace = true }
+
 
 # Only enable kafka features and related dependencies on Linux
 [target.'cfg(target_os = "linux")'.dependencies]
diff --git a/crates/event/examples/full.rs b/crates/event/examples/full.rs
index c9e17af94..9d03771d9 100644
--- a/crates/event/examples/full.rs
+++ b/crates/event/examples/full.rs
@@ -16,7 +16,17 @@ async fn setup_notification_system() -> Result<(), NotifierError> {
             auth_token: Some("your-auth-token".into()),
             custom_headers: Some(HashMap::new()),
             max_retries: 3,
-            timeout: 30,
+            timeout: Some(30),
+            retry_interval: Some(5),
+            client_cert: None,
+            client_key: None,
+            common: rustfs_event::AdapterCommon {
+                identifier: "webhook".into(),
+                comment: "webhook".into(),
+                enable: true,
+                queue_dir: "./deploy/logs/event_queue".into(),
+                queue_limit: 100,
+            },
         })],
     };
 
diff --git a/crates/event/examples/simple.rs b/crates/event/examples/simple.rs
index 84cc32c12..f72713fd3 100644
--- a/crates/event/examples/simple.rs
+++ b/crates/event/examples/simple.rs
@@ -25,7 +25,17 @@ async fn main() -> Result<(), Box<dyn error::Error>> {
             auth_token: Some("secret-token".to_string()),
             custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])),
             max_retries: 3,
-            timeout: 10,
+            timeout: Some(30),
+            retry_interval: Some(5),
+            client_cert: None,
+            client_key: None,
+            common: rustfs_event::AdapterCommon {
+                identifier: "webhook".to_string(),
+                comment: "webhook".to_string(),
+                enable: true,
+                queue_dir: "./deploy/logs/event_queue".to_string(),
+                queue_limit: 100,
+            },
         })],
     };
 
diff --git a/crates/event/examples/webhook.rs b/crates/event/examples/webhook.rs
index 4cdf02c66..3af970c79 100644
--- a/crates/event/examples/webhook.rs
+++ b/crates/event/examples/webhook.rs
@@ -1,3 +1,4 @@
+use axum::routing::get;
 use axum::{extract::Json, http::StatusCode, routing::post, Router};
 use serde_json::Value;
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -5,7 +6,9 @@ use std::time::{SystemTime, UNIX_EPOCH};
 #[tokio::main]
 async fn main() {
     // 构建应用
-    let app = Router::new().route("/webhook", post(receive_webhook));
+    let app = Router::new()
+        .route("/webhook", post(receive_webhook))
+        .route("/webhook", get(receive_webhook));
     // 启动服务器
     let listener = tokio::net::TcpListener::bind("0.0.0.0:3020").await.unwrap();
     println!("Server running on http://0.0.0.0:3020");
diff --git a/crates/event/src/adapter/kafka.rs b/crates/event/src/adapter/kafka.rs
index 1ad0ddbd3..a60afaeee 100644
--- a/crates/event/src/adapter/kafka.rs
+++ b/crates/event/src/adapter/kafka.rs
@@ -1,20 +1,22 @@
-use crate::Error;
 use crate::Event;
 use crate::KafkaConfig;
 use crate::{ChannelAdapter, ChannelAdapterType};
+use crate::{Error, QueueStore};
 use async_trait::async_trait;
 use rdkafka::error::KafkaError;
 use rdkafka::producer::{FutureProducer, FutureRecord};
 use rdkafka::types::RDKafkaErrorCode;
 use rdkafka::util::Timeout;
+use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::sleep;
 
 /// Kafka adapter for sending events to a Kafka topic.
 pub struct KafkaAdapter {
     producer: FutureProducer,
-    topic: String,
-    max_retries: u32,
+    store: Option<Arc<QueueStore<Event>>>,
+    config: KafkaConfig,
 }
 
 impl KafkaAdapter {
@@ -26,11 +28,21 @@ impl KafkaAdapter {
             .set("message.timeout.ms", config.timeout.to_string())
             .create()?;
 
-        Ok(Self {
-            producer,
-            topic: config.topic.clone(),
-            max_retries: config.max_retries,
-        })
+        // create a queue store if enabled
+        let store = if !config.queue_dir.is_empty() {
+            let store_path = PathBuf::from(&config.queue_dir);
+            let store = QueueStore::new(store_path, config.queue_limit, Some(".kafka".to_string()));
+            if let Err(e) = store.open() {
+                tracing::error!("Unable to open queue storage: {}", e);
+                None
+            } else {
+                Some(Arc::new(store))
+            }
+        } else {
+            None
+        };
+
+        Ok(Self { config, producer, store })
     }
     /// Sends an event to the Kafka topic with retry logic.
     async fn send_with_retry(&self, event: &Event) -> Result<(), Error> {
diff --git a/crates/event/src/adapter/webhook.rs b/crates/event/src/adapter/webhook.rs
index 4cf523bbb..1de3e40d7 100644
--- a/crates/event/src/adapter/webhook.rs
+++ b/crates/event/src/adapter/webhook.rs
@@ -1,15 +1,20 @@
-use crate::Error;
-use crate::Event;
+use crate::store::queue::Store;
 use crate::WebhookConfig;
 use crate::{ChannelAdapter, ChannelAdapterType};
+use crate::{Error, QueueStore};
+use crate::{Event, DEFAULT_RETRY_INTERVAL};
 use async_trait::async_trait;
-use reqwest::{Client, RequestBuilder};
+use reqwest::{self, Client, Identity, RequestBuilder};
+use std::fs;
+use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::sleep;
 
 /// Webhook adapter for sending events to a webhook endpoint.
 pub struct WebhookAdapter {
     config: WebhookConfig,
+    store: Option<Arc<QueueStore<Event>>>,
     client: Client,
 }
 
@@ -17,17 +22,154 @@ impl WebhookAdapter {
     /// Creates a new Webhook adapter.
     pub fn new(config: WebhookConfig) -> Self {
         let mut builder = Client::builder();
-        if config.timeout > 0 {
-            builder = builder.timeout(Duration::from_secs(config.timeout));
+        if config.timeout.is_some() {
+            // Set the timeout for the client
+            match config.timeout {
+                Some(t) => builder = builder.timeout(Duration::from_secs(t)),
+                None => tracing::warn!("Timeout is not set, using default timeout"),
+            }
         }
-        let client = builder.build().expect("Failed to build reqwest client");
-        Self { config, client }
+        let client = if let (Some(cert_path), Some(key_path)) = (&config.client_cert, &config.client_key) {
+            let cert_path = PathBuf::from(cert_path);
+            let key_path = PathBuf::from(key_path);
+
+            // Check if the certificate file exists
+            if !cert_path.exists() || !key_path.exists() {
+                tracing::warn!("Certificate files not found, falling back to default client");
+                builder.build()
+            } else {
+                // Try to read and load the certificate
+                match (fs::read(&cert_path), fs::read(&key_path)) {
+                    (Ok(cert_data), Ok(key_data)) => {
+                        // Create an identity
+                        let mut pem_data = cert_data;
+                        pem_data.extend_from_slice(&key_data);
+
+                        match Identity::from_pem(&pem_data) {
+                            Ok(identity) => {
+                                tracing::info!("Successfully loaded client certificate");
+                                builder.identity(identity).build()
+                            }
+                            Err(e) => {
+                                tracing::warn!("Failed to create identity from PEM: {}, falling back to default client", e);
+                                builder.build()
+                            }
+                        }
+                    }
+                    _ => {
+                        tracing::warn!("Failed to read certificate files, falling back to default client");
+                        builder.build()
+                    }
+                }
+            }
+        } else {
+            builder.build()
+        }
+        .expect("Failed to create HTTP client");
+
+        // create a queue store if enabled
+        let store = if !config.common.queue_dir.len() > 0 {
+            let store_path = PathBuf::from(&config.common.queue_dir);
+            let store = QueueStore::new(store_path, config.common.queue_limit, Some(".webhook".to_string()));
+            if let Err(e) = store.open() {
+                tracing::error!("Unable to open queue storage: {}", e);
+                None
+            } else {
+                Some(Arc::new(store))
+            }
+        } else {
+            None
+        };
+
+        Self { config, store, client }
     }
+
+    /// Handle backlog events in storage
+    pub async fn process_backlog(&self) -> Result<(), Error> {
+        if let Some(store) = &self.store {
+            let keys = store.list();
+            for key in keys {
+                match store.get_multiple(&key) {
+                    Ok(events) => {
+                        for event in events {
+                            if let Err(e) = self.send_with_retry(&event).await {
+                                tracing::error!("Processing of backlog events failed: {}", e);
+                                continue;
+                            }
+                        }
+                        // Deleted after successful processing
+                        let _ = store.del(&key);
+                    }
+                    Err(e) => {
+                        tracing::error!("Failed to read events from storage: {}", e);
+                        // delete the broken entries
+                        let _ = store.del(&key);
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    ///Send events to the webhook endpoint with retry logic
+    async fn send_with_retry(&self, event: &Event) -> Result<(), Error> {
+        let retry_interval = match self.config.retry_interval {
+            Some(t) => Duration::from_secs(t),
+            None => Duration::from_secs(DEFAULT_RETRY_INTERVAL), // Default to 3 seconds if not set
+        };
+        let mut attempts = 0;
+
+        loop {
+            attempts += 1;
+            match self.send_request(event).await {
+                Ok(_) => return Ok(()),
+                Err(e) => {
+                    if attempts <= self.config.max_retries {
+                        tracing::warn!("Send to webhook fails and will be retried after 3 seconds:{}", e);
+                        sleep(retry_interval).await;
+                    } else if let Some(store) = &self.store {
+                        // store in a queue for later processing
+                        tracing::warn!("The maximum number of retries is reached, and the event is stored in a queue:{}", e);
+                        if let Err(store_err) = store.put(event.clone()) {
+                            tracing::error!("Events cannot be stored to a queue:{}", store_err);
+                        }
+                        return Err(e);
+                    } else {
+                        return Err(e);
+                    }
+                }
+            }
+        }
+    }
+
+    /// Send a single HTTP request
+    async fn send_request(&self, event: &Event) -> Result<(), Error> {
+        // Send a request
+        let response = self.build_request(event).send().await?;
+
+        // Check the response status
+        if !response.status().is_success() {
+            return Err(Error::Custom(format!("Webhook request failed, status code:{}", response.status())));
+        }
+
+        Ok(())
+    }
+
     /// Builds the request to send the event.
     fn build_request(&self, event: &Event) -> RequestBuilder {
-        let mut request = self.client.post(&self.config.endpoint).json(event);
+        let mut request = self
+            .client
+            .post(&self.config.endpoint)
+            .json(event)
+            .header("Content-Type", "application/json");
         if let Some(token) = &self.config.auth_token {
-            request = request.header("Authorization", format!("Bearer {}", token));
+            let tokens: Vec<&str> = token.split_whitespace().collect();
+            match tokens.len() {
+                2 => request = request.header("Authorization", token),
+                1 => request = request.header("Authorization", format!("Bearer {}", token)),
+                _ => tracing::warn!("Invalid auth token format, skipping Authorization header"),
+            }
         }
         if let Some(headers) = &self.config.custom_headers {
             for (key, value) in headers {
@@ -45,21 +187,10 @@ impl ChannelAdapter for WebhookAdapter {
     }
 
     async fn send(&self, event: &Event) -> Result<(), Error> {
-        let mut attempt = 0;
-        tracing::info!("Attempting to send webhook request: {:?}", event);
-        loop {
-            match self.build_request(event).send().await {
-                Ok(response) => {
-                    response.error_for_status().map_err(Error::Http)?;
-                    return Ok(());
-                }
-                Err(e) if attempt < self.config.max_retries => {
-                    attempt += 1;
-                    tracing::warn!("Webhook attempt {} failed: {}. Retrying...", attempt, e);
-                    sleep(Duration::from_secs(2u64.pow(attempt))).await;
-                }
-                Err(e) => return Err(Error::Http(e)),
-            }
-        }
+        // Deal with the backlog of events first
+        let _ = self.process_backlog().await;
+
+        // Send the current event
+        self.send_with_retry(event).await
     }
 }
diff --git a/crates/event/src/config.rs b/crates/event/src/config.rs
index 996db2d3d..77d35539c 100644
--- a/crates/event/src/config.rs
+++ b/crates/event/src/config.rs
@@ -2,17 +2,64 @@ use config::{Config, File, FileFormat};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::env;
+use std::path::Path;
+use tracing::info;
 
+/// The default configuration file name
 const DEFAULT_CONFIG_FILE: &str = "event";
 
-/// Configuration for the webhook adapter.
+/// The prefix for the configuration file
+pub(crate) const STORE_PREFIX: &str = "rustfs";
+
+/// The default retry interval for the webhook adapter
+pub const DEFAULT_RETRY_INTERVAL: u64 = 3;
+
+/// The default maximum retry count for the webhook adapter
+pub const DEFAULT_MAX_RETRIES: u32 = 3;
+
+/// Add a common field for the adapter configuration
 #[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AdapterCommon {
+    /// Adapter identifier for unique identification
+    pub identifier: String,
+    /// Adapter description information
+    pub comment: String,
+    /// Whether to enable this adapter
+    #[serde(default)]
+    pub enable: bool,
+    #[serde(default = "default_queue_dir")]
+    pub queue_dir: String,
+    #[serde(default = "default_queue_limit")]
+    pub queue_limit: u64,
+}
+
+impl Default for AdapterCommon {
+    fn default() -> Self {
+        Self {
+            identifier: String::new(),
+            comment: String::new(),
+            enable: false,
+            queue_dir: default_queue_dir(),
+            queue_limit: default_queue_limit(),
+        }
+    }
+}
+
+/// Configuration for the webhook adapter.
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct WebhookConfig {
+    #[serde(flatten)]
+    pub common: AdapterCommon,
     pub endpoint: String,
     pub auth_token: Option<String>,
     pub custom_headers: Option<HashMap<String, String>>,
     pub max_retries: u32,
-    pub timeout: u64,
+    pub retry_interval: Option<u64>,
+    pub timeout: Option<u64>,
+    #[serde(default)]
+    pub client_cert: Option<String>,
+    #[serde(default)]
+    pub client_key: Option<String>,
 }
 
 impl WebhookConfig {
@@ -26,25 +73,41 @@ impl WebhookConfig {
     ///
     /// ```
     /// use rustfs_event::WebhookConfig;
+    /// use rustfs_event::AdapterCommon;
+    /// use rustfs_event::DEFAULT_RETRY_INTERVAL;
     ///
     /// let config = WebhookConfig {
-    ///     endpoint: "http://example.com/webhook".to_string(),
+    ///     common: AdapterCommon::default(),
+    ///     endpoint: "https://example.com/webhook".to_string(),
     ///     auth_token: Some("my_token".to_string()),
     ///     custom_headers: None,
     ///     max_retries: 3,
-    ///     timeout: 5000,
+    ///     retry_interval: Some(DEFAULT_RETRY_INTERVAL),
+    ///     timeout: Some(5000),
+    ///     client_cert: None,
+    ///     client_key: None,
     /// };
     ///
     /// assert!(config.validate().is_ok());
     pub fn validate(&self) -> Result<(), String> {
+        // If not enabled, the other fields are not validated
+        if !self.common.enable {
+            return Ok(());
+        }
+
         // verify that endpoint cannot be empty
         if self.endpoint.trim().is_empty() {
             return Err("Webhook endpoint cannot be empty".to_string());
         }
 
         // verification timeout must be reasonable
-        if self.timeout == 0 {
-            return Err("Webhook timeout must be greater than 0".to_string());
+        if self.timeout.is_some() {
+            match self.timeout {
+                Some(timeout) if timeout > 0 => {
+                    info!("Webhook timeout is set to {}", timeout);
+                }
+                _ => return Err("Webhook timeout must be greater than 0".to_string()),
+            }
         }
 
         // Verify that the maximum number of retry is reasonable
@@ -52,22 +115,82 @@ impl WebhookConfig {
             return Err("Maximum retry count cannot exceed 10".to_string());
         }
 
+        // Verify the queue directory path
+        if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() {
+            return Err("Queue directory path should be absolute".to_string());
+        }
+
+        // The authentication certificate and key must appear in pairs
+        if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some())
+        {
+            return Err("Certificate and key must be specified as a pair".to_string());
+        }
+
         Ok(())
     }
+
+    /// Create a new webhook configuration
+    pub fn new(identifier: impl Into<String>, endpoint: impl Into<String>) -> Self {
+        Self {
+            common: AdapterCommon {
+                identifier: identifier.into(),
+                comment: String::new(),
+                enable: true,
+                queue_dir: default_queue_dir(),
+                queue_limit: default_queue_limit(),
+            },
+            endpoint: endpoint.into(),
+            ..Default::default()
+        }
+    }
 }
 
 /// Configuration for the Kafka adapter.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct KafkaConfig {
+    #[serde(flatten)]
+    pub common: AdapterCommon,
     pub brokers: String,
     pub topic: String,
     pub max_retries: u32,
     pub timeout: u64,
 }
 
+impl Default for KafkaConfig {
+    fn default() -> Self {
+        Self {
+            common: AdapterCommon::default(),
+            brokers: String::new(),
+            topic: String::new(),
+            max_retries: 3,
+            timeout: 5000,
+        }
+    }
+}
+
+impl KafkaConfig {
+    /// Create a new Kafka configuration
+    pub fn new(identifier: impl Into<String>, brokers: impl Into<String>, topic: impl Into<String>) -> Self {
+        Self {
+            common: AdapterCommon {
+                identifier: identifier.into(),
+                comment: String::new(),
+                enable: true,
+                queue_dir: default_queue_dir(),
+                queue_limit: default_queue_limit(),
+            },
+            brokers: brokers.into(),
+            topic: topic.into(),
+            ..Default::default()
+        }
+    }
+}
+
 /// Configuration for the MQTT adapter.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct MqttConfig {
+    #[serde(flatten)]
+    pub common: AdapterCommon,
     pub broker: String,
     pub port: u16,
     pub client_id: String,
@@ -75,6 +198,37 @@ pub struct MqttConfig {
     pub max_retries: u32,
 }
 
+impl Default for MqttConfig {
+    fn default() -> Self {
+        Self {
+            common: AdapterCommon::default(),
+            broker: String::new(),
+            port: 1883,
+            client_id: String::new(),
+            topic: String::new(),
+            max_retries: 3,
+        }
+    }
+}
+
+impl MqttConfig {
+    /// Create a new MQTT configuration
+    pub fn new(identifier: impl Into<String>, broker: impl Into<String>, topic: impl Into<String>) -> Self {
+        Self {
+            common: AdapterCommon {
+                identifier: identifier.into(),
+                comment: String::new(),
+                enable: true,
+                queue_dir: default_queue_dir(),
+                queue_limit: default_queue_limit(),
+            },
+            broker: broker.into(),
+            topic: topic.into(),
+            ..Default::default()
+        }
+    }
+}
+
 /// Configuration for the adapter.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "type")]
@@ -104,18 +258,18 @@ pub enum AdapterConfig {
 /// ```
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct NotifierConfig {
-    #[serde(default = "default_store_path")]
+    #[serde(default = "default_queue_dir")]
     pub store_path: String,
-    #[serde(default = "default_channel_capacity")]
-    pub channel_capacity: usize,
+    #[serde(default = "default_queue_limit")]
+    pub channel_capacity: u64,
     pub adapters: Vec<AdapterConfig>,
 }
 
 impl Default for NotifierConfig {
     fn default() -> Self {
         Self {
-            store_path: default_store_path(),
-            channel_capacity: default_channel_capacity(),
+            store_path: default_queue_dir(),
+            channel_capacity: default_queue_limit(),
             adapters: Vec::new(),
         }
     }
@@ -150,7 +304,7 @@ impl NotifierConfig {
                 DEFAULT_CONFIG_FILE.to_string()
             } else {
                 // Use the provided path
-                let path = std::path::Path::new(&path);
+                let path = Path::new(&path);
                 if path.extension().is_some() {
                     // If path has extension, use it as is (extension will be added by Config::builder)
                     path.with_extension("").to_string_lossy().into_owned()
@@ -209,17 +363,87 @@ impl NotifierConfig {
 }
 
 /// Provide temporary directories as default storage paths
-fn default_store_path() -> String {
-    env::var("EVENT_STORE_PATH").unwrap_or_else(|e| {
-        tracing::info!("Failed to get `EVENT_STORE_PATH` failed err: {}", e.to_string());
+fn default_queue_dir() -> String {
+    env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| {
+        tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string());
         env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string()
     })
 }
 
 /// Provides the recommended default channel capacity for high concurrency systems
-fn default_channel_capacity() -> usize {
+fn default_queue_limit() -> u64 {
     env::var("EVENT_CHANNEL_CAPACITY")
         .unwrap_or_else(|_| "10000".to_string())
         .parse()
         .unwrap_or(10000) // Default to 10000 if parsing fails
 }
+
+/// Event Notifier Configuration
+/// This struct contains the configuration for the event notifier system,
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct EventNotifierConfig {
+    /// A collection of webhook configurations, with the key being a unique identifier
+    #[serde(default)]
+    pub webhook: HashMap<String, WebhookConfig>,
+    /// A collection of Kafka configurations, with the key being a unique identifier
+    #[serde(default)]
+    pub kafka: HashMap<String, KafkaConfig>,
+    ///MQTT configuration collection, with the key being a unique identifier
+    #[serde(default)]
+    pub mqtt: HashMap<String, MqttConfig>,
+}
+
+impl EventNotifierConfig {
+    /// Create a new default configuration
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Load the configuration from the file
+    pub fn event_load_config(_config_dir: Option<String>) -> EventNotifierConfig {
+        // The existing implementation remains the same, but returns EventNotifierConfig
+        // ...
+
+        EventNotifierConfig::default()
+    }
+
+    /// Deserialization configuration
+    pub fn unmarshal(data: &[u8]) -> common::error::Result<EventNotifierConfig> {
+        let m: EventNotifierConfig = serde_json::from_slice(data)?;
+        Ok(m)
+    }
+
+    /// Serialization configuration
+    pub fn marshal(&self) -> common::error::Result<Vec<u8>> {
+        let data = serde_json::to_vec(&self)?;
+        Ok(data)
+    }
+
+    /// Convert this configuration to a list of adapter configurations
+    pub fn to_adapter_configs(&self) -> Vec<AdapterConfig> {
+        let mut adapters = Vec::new();
+
+        // Add all enabled webhook configurations
+        for (_, webhook) in &self.webhook {
+            if webhook.common.enable {
+                adapters.push(AdapterConfig::Webhook(webhook.clone()));
+            }
+        }
+
+        // Add all enabled Kafka configurations
+        for (_, kafka) in &self.kafka {
+            if kafka.common.enable {
+                adapters.push(AdapterConfig::Kafka(kafka.clone()));
+            }
+        }
+
+        // Add all enabled MQTT configurations
+        for (_, mqtt) in &self.mqtt {
+            if mqtt.common.enable {
+                adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
+            }
+        }
+
+        adapters
+    }
+}
diff --git a/crates/event/src/error.rs b/crates/event/src/error.rs
index ebdaf899d..b62e1f760 100644
--- a/crates/event/src/error.rs
+++ b/crates/event/src/error.rs
@@ -37,6 +37,8 @@ pub enum Error {
     ConfigError(String),
     #[error("Configuration loading error: {0}")]
     Config(#[from] ConfigError),
+    #[error("create adapter failed error: {0}")]
+    AdapterCreationFailed(String),
 }
 
 impl Error {
diff --git a/crates/event/src/global.rs b/crates/event/src/global.rs
index ab5b727db..37d92d169 100644
--- a/crates/event/src/global.rs
+++ b/crates/event/src/global.rs
@@ -174,7 +174,7 @@ async fn get_system() -> Result<Arc<Mutex<NotifierSystem>>, Error> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{AdapterConfig, NotifierConfig, WebhookConfig};
+    use crate::{AdapterCommon, AdapterConfig, NotifierConfig, WebhookConfig};
     use std::collections::HashMap;
 
     #[tokio::test]
@@ -205,11 +205,21 @@ mod tests {
             adapters: vec![
                 // assuming that the empty adapter will cause failure
                 AdapterConfig::Webhook(WebhookConfig {
+                    common: AdapterCommon {
+                        identifier: "empty".to_string(),
+                        comment: "empty".to_string(),
+                        enable: true,
+                        queue_dir: "".to_string(),
+                        queue_limit: 10,
+                    },
                     endpoint: "http://localhost:8080/webhook".to_string(),
                     auth_token: Some("secret-token".to_string()),
                     custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])),
                     max_retries: 3,
-                    timeout: 10,
+                    timeout: Some(10),
+                    retry_interval: Some(5),
+                    client_cert: None,
+                    client_key: None,
                 }),
             ], // assuming that the empty adapter will cause failure
             ..Default::default()
diff --git a/crates/event/src/lib.rs b/crates/event/src/lib.rs
index 20b576dc6..16350e9c9 100644
--- a/crates/event/src/lib.rs
+++ b/crates/event/src/lib.rs
@@ -6,6 +6,7 @@ mod event;
 mod global;
 mod notifier;
 mod store;
+mod target;
 
 pub use adapter::create_adapters;
 #[cfg(all(feature = "kafka", target_os = "linux"))]
@@ -17,13 +18,15 @@ pub use adapter::webhook::WebhookAdapter;
 pub use adapter::ChannelAdapter;
 pub use adapter::ChannelAdapterType;
 pub use bus::event_bus;
+pub use config::AdapterCommon;
+pub use config::EventNotifierConfig;
 #[cfg(all(feature = "kafka", target_os = "linux"))]
 pub use config::KafkaConfig;
 #[cfg(feature = "mqtt")]
 pub use config::MqttConfig;
 #[cfg(feature = "webhook")]
 pub use config::WebhookConfig;
-pub use config::{AdapterConfig, NotifierConfig};
+pub use config::{AdapterConfig, NotifierConfig, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL};
 pub use error::Error;
 
 pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source};
@@ -32,6 +35,7 @@ pub use notifier::NotifierSystem;
 pub use store::event::EventStore;
 
 pub use store::get_event_notifier_config;
+pub use store::queue::QueueStore;
 
 pub use store::EventSys;
 pub use store::GLOBAL_EventSys;
diff --git a/crates/event/src/notifier.rs b/crates/event/src/notifier.rs
index 5ab17d371..db3bbc9c5 100644
--- a/crates/event/src/notifier.rs
+++ b/crates/event/src/notifier.rs
@@ -21,7 +21,7 @@ impl NotifierSystem {
     /// Creates a new `NotificationSystem` instance.
     #[instrument(skip(config))]
     pub async fn new(config: NotifierConfig) -> Result<Self, Error> {
-        let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity);
+        let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity.try_into().unwrap());
         let store = Arc::new(EventStore::new(&config.store_path).await?);
         let shutdown = CancellationToken::new();
 
diff --git a/crates/event/src/store/manager.rs b/crates/event/src/store/manager.rs
new file mode 100644
index 000000000..52fe11956
--- /dev/null
+++ b/crates/event/src/store/manager.rs
@@ -0,0 +1,236 @@
+use crate::store::{CONFIG_FILE, EVENT};
+use crate::{adapter, ChannelAdapter, EventNotifierConfig, WebhookAdapter};
+use common::error::{Error, Result};
+use ecstore::config::com::{read_config, save_config, CONFIG_PREFIX};
+use ecstore::disk::RUSTFS_META_BUCKET;
+use ecstore::store::ECStore;
+use ecstore::store_api::ObjectOptions;
+use ecstore::utils::path::SLASH_SEPARATOR;
+use ecstore::StorageAPI;
+use once_cell::sync::Lazy;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing::instrument;
+
+/// Global storage API access point
+pub static GLOBAL_STORE_API: Lazy<Mutex<Option<Arc<ECStore>>>> = Lazy::new(|| Mutex::new(None));
+
+/// Global event system configuration
+pub static GLOBAL_EVENT_CONFIG: Lazy<Mutex<Option<EventNotifierConfig>>> = Lazy::new(|| Mutex::new(None));
+
+/// EventManager Responsible for managing all operations of the event system
+#[derive(Debug)]
+pub struct EventManager {
+    api: Arc<ECStore>,
+}
+
+impl EventManager {
+    /// Create a new Event Manager
+    pub async fn new(api: Arc<ECStore>) -> Self {
+        // Update the global access point at the same time
+        {
+            let mut global_api = GLOBAL_STORE_API.lock().await;
+            *global_api = Some(api.clone());
+        }
+
+        Self { api }
+    }
+
+    /// Initialize the Event Manager
+    ///
+    /// # Returns
+    /// If it succeeds, it returns configuration information, and if it fails, it returns an error
+    #[instrument(skip_all)]
+    pub async fn init(&self) -> Result<EventNotifierConfig> {
+        tracing::info!("Event system configuration initialization begins");
+
+        let cfg = match read_event_config(self.api.clone()).await {
+            Ok(cfg) => cfg,
+            Err(err) => {
+                tracing::error!("Failed to initialize the event system configuration:{:?}", err);
+                return Err(err);
+            }
+        };
+
+        *GLOBAL_EVENT_CONFIG.lock().await = Some(cfg.clone());
+        tracing::info!("The initialization of the event system configuration is complete");
+
+        Ok(cfg)
+    }
+
+    /// Create a new configuration
+    ///
+    /// # Parameters
+    /// - `cfg`: The configuration to be created
+    ///
+    /// # Returns
+    /// The result of the operation
+    pub async fn create_config(&self, cfg: &EventNotifierConfig) -> Result<()> {
+        // Check whether the configuration already exists
+        if let Ok(_) = read_event_config(self.api.clone()).await {
+            return Err(Error::msg("The configuration already exists, use the update action"));
+        }
+
+        save_event_config(self.api.clone(), cfg).await?;
+        *GLOBAL_EVENT_CONFIG.lock().await = Some(cfg.clone());
+
+        Ok(())
+    }
+
+    /// Update the configuration
+    ///
+    /// # Parameters
+    /// - `cfg`: The configuration to be updated
+    ///
+    /// # Returns
+    /// The result of the operation
+    pub async fn update_config(&self, cfg: &EventNotifierConfig) -> Result<()> {
+        // Read the existing configuration first to merge
+        let current_cfg = read_event_config(self.api.clone()).await.unwrap_or_default();
+
+        // This is where the merge logic can be implemented
+        let merged_cfg = self.merge_configs(current_cfg, cfg.clone());
+
+        save_event_config(self.api.clone(), &merged_cfg).await?;
+        *GLOBAL_EVENT_CONFIG.lock().await = Some(merged_cfg);
+
+        Ok(())
+    }
+
+    /// Merge the two configurations
+    fn merge_configs(&self, current: EventNotifierConfig, new: EventNotifierConfig) -> EventNotifierConfig {
+        let mut merged = current;
+
+        // Merge webhook configurations
+        for (id, config) in new.webhook {
+            merged.webhook.insert(id, config);
+        }
+
+        // Merge Kafka configurations
+        for (id, config) in new.kafka {
+            merged.kafka.insert(id, config);
+        }
+
+        // Merge MQTT configurations
+        for (id, config) in new.mqtt {
+            merged.mqtt.insert(id, config);
+        }
+
+        merged
+    }
+
+    /// Delete the configuration
+    pub async fn delete_config(&self) -> Result<()> {
+        let config_file = get_event_config_file();
+        self.api
+            .delete_object(
+                RUSTFS_META_BUCKET,
+                &config_file,
+                ObjectOptions {
+                    delete_prefix: true,
+                    delete_prefix_object: true,
+                    ..Default::default()
+                },
+            )
+            .await?;
+
+        // Reset the global configuration to default
+        // let _ = GLOBAL_EventSysConfig.set(self.read_config().await?);
+
+        Ok(())
+    }
+
+    /// Read the configuration
+    pub async fn read_config(&self) -> Result<EventNotifierConfig> {
+        read_event_config(self.api.clone()).await
+    }
+
+    /// Create all enabled adapters
+    pub async fn create_adapters(&self) -> Result<Vec<Arc<dyn ChannelAdapter>>> {
+        let config = match GLOBAL_EVENT_CONFIG.lock().await.clone() {
+            Some(cfg) => cfg,
+            None => return Err(Error::msg("The global configuration is not initialized")),
+        };
+
+        let mut adapters: Vec<Arc<dyn ChannelAdapter>> = Vec::new();
+
+        // Create a webhook adapter
+        for (_, webhook_config) in &config.webhook {
+            if webhook_config.common.enable {
+                #[cfg(feature = "webhook")]
+                {
+                    adapters.push(Arc::new(WebhookAdapter::new(webhook_config.clone())));
+                }
+
+                #[cfg(not(feature = "webhook"))]
+                {
+                    return Err(Error::msg("The webhook feature is not enabled"));
+                }
+            }
+        }
+
+        // Create a Kafka adapter
+        for (_, kafka_config) in &config.kafka {
+            if kafka_config.common.enable {
+                #[cfg(all(feature = "kafka", target_os = "linux"))]
+                {
+                    match KafkaAdapter::new(kafka_config.clone()) {
+                        Ok(adapter) => adapters.push(Arc::new(adapter)),
+                        Err(e) => tracing::error!("Failed to create a Kafka adapter:{}", e),
+                    }
+                }
+
+                #[cfg(any(not(feature = "kafka"), not(target_os = "linux")))]
+                {
+                    return Err(Error::msg("Kafka functionality is not enabled or is not a Linux environment"));
+                }
+            }
+        }
+
+        // Create an MQTT adapter
+        for (_, mqtt_config) in &config.mqtt {
+            if mqtt_config.common.enable {
+                #[cfg(feature = "mqtt")]
+                {
+                    // Implement MQTT adapter creation logic
+                    // ...
+                }
+
+                #[cfg(not(feature = "mqtt"))]
+                {
+                    return Err(Error::msg("MQTT The feature is not enabled"));
+                }
+            }
+        }
+
+        Ok(adapters)
+    }
+}
+
+/// Get the Global Storage API
+pub async fn get_global_event_config() -> Option<EventNotifierConfig> {
+    GLOBAL_EVENT_CONFIG.lock().await.clone()
+}
+
+/// Read event configuration
+async fn read_event_config(api: Arc<ECStore>) -> Result<EventNotifierConfig> {
+    let config_file = get_event_config_file();
+    let data = read_config(api, &config_file).await?;
+
+    EventNotifierConfig::unmarshal(&data)
+}
+
+/// Save the event configuration
+async fn save_event_config(api: Arc<ECStore>, config: &EventNotifierConfig) -> Result<()> {
+    let config_file = get_event_config_file();
+    let data = config.marshal()?;
+
+    save_config(api, &config_file, data).await?;
+    Ok(())
+}
+
+/// Get the event profile path
+fn get_event_config_file() -> String {
+    // "event/config.json".to_string()
+    format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE)
+}
diff --git a/crates/event/src/store/mod.rs b/crates/event/src/store/mod.rs
index d1cec06d9..be58ec0ff 100644
--- a/crates/event/src/store/mod.rs
+++ b/crates/event/src/store/mod.rs
@@ -1,4 +1,6 @@
 pub(crate) mod event;
+mod manager;
+pub(crate) mod queue;
 
 use crate::NotifierConfig;
 use common::error::Result;
diff --git a/crates/event/src/store/queue.rs b/crates/event/src/store/queue.rs
new file mode 100644
index 000000000..bde850748
--- /dev/null
+++ b/crates/event/src/store/queue.rs
@@ -0,0 +1,449 @@
+use common::error::{Error, Result};
+use serde::{de::DeserializeOwned, Serialize};
+use snap::raw::{Decoder, Encoder};
+use std::collections::HashMap;
+use std::io::Read;
+use std::marker::PhantomData;
+use std::path::PathBuf;
+use std::sync::{Arc, RwLock};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::{fs, io};
+use uuid::Uuid;
+
+/// Keys in storage
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Key {
+    /// Key name
+    pub name: String,
+    /// Whether or not to compress
+    pub compress: bool,
+    /// filename extension
+    pub extension: String,
+    /// Number of items
+    pub item_count: usize,
+}
+
+impl Key {
+    /// Create a new key
+    pub fn new(name: impl Into<String>, extension: impl Into<String>, compress: bool) -> Self {
+        Self {
+            name: name.into(),
+            compress,
+            extension: extension.into(),
+            item_count: 1,
+        }
+    }
+
+    /// Convert to string form
+    pub fn to_string(&self) -> String {
+        let mut key_str = self.name.clone();
+        if self.item_count > 1 {
+            key_str = format!("{}:{}", self.item_count, self.name);
+        }
+
+        let compress_ext = if self.compress { ".snappy" } else { "" };
+        format!("{}{}{}", key_str, self.extension, compress_ext)
+    }
+}
+
+/// Parse key from file name
+pub fn parse_key(filename: &str) -> Key {
+    let compress = filename.ends_with(".snappy");
+    let filename = if compress {
+        &filename[..filename.len() - 7] // 移除 ".snappy"
+    } else {
+        filename
+    };
+
+    let mut parts = filename.splitn(2, '.');
+    let name_part = parts.next().unwrap_or("");
+    let extension = parts
+        .next()
+        .map_or_else(|| String::new(), |ext| format!(".{}", ext))
+        .to_string();
+
+    let mut name = name_part.to_string();
+    let mut item_count = 1;
+
+    if let Some(pos) = name_part.find(':') {
+        if let Ok(count) = name_part[..pos].parse::<usize>() {
+            item_count = count;
+            name = name_part[pos + 1..].to_string();
+        }
+    }
+
+    Key {
+        name,
+        compress,
+        extension,
+        item_count,
+    }
+}
+
+/// Store the characteristics of the project
+pub trait Store<T>: Send + Sync
+where
+    T: Serialize + DeserializeOwned + Send + Sync,
+{
+    /// Store a single item
+    fn put(&self, item: T) -> Result<Key>;
+
+    /// Store multiple projects
+    fn put_multiple(&self, items: Vec<T>) -> Result<Key>;
+
+    /// Get a single item
+    fn get(&self, key: &Key) -> Result<T>;
+
+    /// Get multiple items
+    fn get_multiple(&self, key: &Key) -> Result<Vec<T>>;
+
+    /// Get the raw bytes
+    fn get_raw(&self, key: &Key) -> Result<Vec<u8>>;
+
+    /// Stores raw bytes
+    fn put_raw(&self, data: &[u8]) -> Result<Key>;
+
+    /// Gets the number of items in storage
+    fn len(&self) -> usize;
+
+    /// Whether it is empty or not
+    fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Lists all keys
+    fn list(&self) -> Vec<Key>;
+
+    /// Delete the key
+    fn del(&self, key: &Key) -> Result<()>;
+
+    /// Open Storage
+    fn open(&self) -> Result<()>;
+
+    /// Delete the storage
+    fn delete(&self) -> Result<()>;
+}
+
+const DEFAULT_LIMIT: u64 = 100000;
+const DEFAULT_EXT: &str = ".unknown";
+const COMPRESS_EXT: &str = ".snappy";
+
+/// Queue storage implementation
+pub struct QueueStore<T> {
+    /// Project Limitations
+    entry_limit: u64,
+    /// Storage directory
+    directory: PathBuf,
+    /// filename extension
+    file_ext: String,
+    /// Item mapping: key -> modified time (Unix nanoseconds)
+    entries: Arc<RwLock<HashMap<String, i64>>>,
+    /// Type tags
+    _phantom: PhantomData<T>,
+}
+
+impl<T> QueueStore<T>
+where
+    T: Serialize + DeserializeOwned + Send + Sync,
+{
+    /// Create a new queue store
+    pub fn new(directory: impl Into<PathBuf>, limit: u64, ext: Option<String>) -> Self {
+        let limit = if limit == 0 { DEFAULT_LIMIT } else { limit };
+        let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string());
+
+        Self {
+            directory: directory.into(),
+            entry_limit: limit,
+            file_ext: ext,
+            entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))),
+            _phantom: PhantomData,
+        }
+    }
+
+    /// Lists all files in the directory, sorted by modification time (oldest takes precedence.)）
+    fn list_files(&self) -> Result<Vec<fs::DirEntry>> {
+        let mut files = Vec::new();
+
+        for entry in fs::read_dir(&self.directory)? {
+            let entry = entry?;
+            let metadata = entry.metadata()?;
+            if metadata.is_file() {
+                files.push(entry);
+            }
+        }
+
+        // Sort by modification time
+        files.sort_by(|a, b| {
+            let a_time = a
+                .metadata()
+                .map(|m| m.modified())
+                .unwrap_or(Ok(UNIX_EPOCH))
+                .unwrap_or(UNIX_EPOCH);
+            let b_time = b
+                .metadata()
+                .map(|m| m.modified())
+                .unwrap_or(Ok(UNIX_EPOCH))
+                .unwrap_or(UNIX_EPOCH);
+            a_time.cmp(&b_time)
+        });
+
+        Ok(files)
+    }
+
+    /// Write the object to a file
+    fn write_object(&self, key: &Key, item: &T) -> Result<()> {
+        // 序列化对象
+        let data = serde_json::to_vec(item)?;
+        self.write_bytes(key, &data)
+    }
+
+    /// Write multiple objects to a file
+    fn write_multiple_objects(&self, key: &Key, items: &[T]) -> Result<()> {
+        let mut data = Vec::new();
+        for item in items {
+            let item_data = serde_json::to_vec(item)?;
+            data.extend_from_slice(&item_data);
+            data.push(b'\n');
+        }
+        self.write_bytes(key, &data)
+    }
+
+    /// Write bytes to a file
+    fn write_bytes(&self, key: &Key, data: &[u8]) -> Result<()> {
+        let path = self.directory.join(key.to_string());
+
+        let file_data = if key.compress {
+            // 使用 snap 压缩数据
+            let mut encoder = Encoder::new();
+            encoder
+                .compress_vec(data)
+                .map_err(|e| Error::msg(format!("Compression failed:{}", e)))?
+        } else {
+            data.to_vec()
+        };
+
+        // Make sure the directory exists
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent)?;
+        }
+
+        // Write to the file
+        fs::write(&path, &file_data)?;
+
+        // Update the item mapping
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
+
+        let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?;
+        entries.insert(key.to_string(), now);
+
+        Ok(())
+    }
+
+    /// Read bytes from a file
+    fn read_bytes(&self, key: &Key) -> Result<Vec<u8>> {
+        let path = self.directory.join(key.to_string());
+        let data = fs::read(&path)?;
+
+        if data.is_empty() {
+            return Err(Error::msg("The file is empty"));
+        }
+
+        if key.compress {
+            // Use Snap to extract the data
+            let mut decoder = Decoder::new();
+            decoder
+                .decompress_vec(&data)
+                .map_err(|e| Error::msg(format!("Failed to decompress:{}", e)))
+        } else {
+            Ok(data)
+        }
+    }
+}
+
+impl<T> Store<T> for QueueStore<T>
+where
+    T: Serialize + DeserializeOwned + Send + Sync + Clone,
+{
+    fn open(&self) -> Result<()> {
+        // Create a directory (if it doesn't exist)
+        fs::create_dir_all(&self.directory)?;
+
+        // Read existing files
+        let files = self.list_files()?;
+
+        let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?;
+
+        for file in files {
+            if let Ok(meta) = file.metadata() {
+                if let Ok(modified) = meta.modified() {
+                    if let Ok(since_epoch) = modified.duration_since(UNIX_EPOCH) {
+                        entries.insert(file.file_name().to_string_lossy().to_string(), since_epoch.as_nanos() as i64);
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn delete(&self) -> Result<()> {
+        fs::remove_dir_all(&self.directory)?;
+        Ok(())
+    }
+
+    fn put(&self, item: T) -> Result<Key> {
+        {
+            let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?;
+            if entries.len() as u64 >= self.entry_limit {
+                return Err(Error::msg("The storage limit has been reached"));
+            }
+        }
+
+        // generate a new uuid
+        let uuid = Uuid::new_v4();
+        let key = Key::new(uuid.to_string(), &self.file_ext, true);
+
+        self.write_object(&key, &item)?;
+
+        Ok(key)
+    }
+
+    fn put_multiple(&self, items: Vec<T>) -> Result<Key> {
+        if items.is_empty() {
+            return Err(Error::msg("The list of items is empty"));
+        }
+
+        {
+            let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?;
+            if entries.len() as u64 >= self.entry_limit {
+                return Err(Error::msg("The storage limit has been reached"));
+            }
+        }
+
+        // Generate a new UUID
+        let uuid = Uuid::new_v4();
+        let mut key = Key::new(uuid.to_string(), &self.file_ext, true);
+        key.item_count = items.len();
+
+        self.write_multiple_objects(&key, &items)?;
+
+        Ok(key)
+    }
+
+    fn get(&self, key: &Key) -> Result<T> {
+        let items = self.get_multiple(key)?;
+        if items.is_empty() {
+            return Err(Error::msg("Item not found"));
+        }
+
+        Ok(items[0].clone())
+    }
+
+    fn get_multiple(&self, key: &Key) -> Result<Vec<T>> {
+        let data = self.get_raw(key)?;
+
+        let mut items = Vec::with_capacity(key.item_count);
+        let mut reader = io::Cursor::new(&data);
+
+        // Try to read each JSON object
+        let mut buffer = Vec::new();
+
+        // if the read fails try parsing it once
+        if reader.read_to_end(&mut buffer).is_err() {
+            // Try to parse the entire data as a single object
+            match serde_json::from_slice::<T>(&data) {
+                Ok(item) => {
+                    items.push(item);
+                    return Ok(items);
+                }
+                Err(_) => {
+                    // An attempt was made to resolve to an array of objects
+                    match serde_json::from_slice::<Vec<T>>(&data) {
+                        Ok(array_items) => return Ok(array_items),
+                        Err(e) => return Err(Error::msg(format!("Failed to parse the data:{}", e))),
+                    }
+                }
+            }
+        }
+
+        // Read JSON objects by row
+        for line in buffer.split(|&b| b == b'\n') {
+            if !line.is_empty() {
+                match serde_json::from_slice::<T>(line) {
+                    Ok(item) => items.push(item),
+                    Err(e) => tracing::warn!("Failed to parse row data:{}", e),
+                }
+            }
+        }
+
+        if items.is_empty() {
+            return Err(Error::msg("Failed to resolve any items"));
+        }
+
+        Ok(items)
+    }
+
+    fn get_raw(&self, key: &Key) -> Result<Vec<u8>> {
+        let data = self.read_bytes(key)?;
+
+        // Delete the wrong file
+        if data.is_empty() {
+            let _ = self.del(key);
+            return Err(Error::msg("the file is empty"));
+        }
+
+        Ok(data)
+    }
+
+    fn put_raw(&self, data: &[u8]) -> Result<Key> {
+        {
+            let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?;
+            if entries.len() as u64 >= self.entry_limit {
+                return Err(Error::msg("the storage limit has been reached"));
+            }
+        }
+
+        // Generate a new UUID
+        let uuid = Uuid::new_v4();
+        let key = Key::new(uuid.to_string(), &self.file_ext, true);
+
+        self.write_bytes(&key, data)?;
+
+        Ok(key)
+    }
+
+    fn len(&self) -> usize {
+        self.entries.read().map(|e| e.len()).unwrap_or(0)
+    }
+
+    fn list(&self) -> Vec<Key> {
+        let entries = match self.entries.read() {
+            Ok(guard) => guard,
+            Err(_) => return Vec::new(),
+        };
+
+        // Convert entries to vectors and sort by timestamp
+        let mut entries_vec: Vec<_> = entries.iter().collect();
+        entries_vec.sort_by(|a, b| a.1.cmp(b.1));
+
+        // Parsing key
+        entries_vec.iter().map(|(filename, _)| parse_key(filename)).collect()
+    }
+
+    fn del(&self, key: &Key) -> Result<()> {
+        let path = self.directory.join(key.to_string());
+
+        // Delete the file
+        if let Err(e) = fs::remove_file(&path) {
+            if e.kind() != io::ErrorKind::NotFound {
+                return Err(e.into());
+            }
+        }
+
+        // Remove the entry from the map
+        let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?;
+        entries.remove(&key.to_string());
+
+        Ok(())
+    }
+}
diff --git a/crates/event/src/target.rs b/crates/event/src/target.rs
new file mode 100644
index 000000000..e6e586dc6
--- /dev/null
+++ b/crates/event/src/target.rs
@@ -0,0 +1,76 @@
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use std::fmt;
+
+///TargetID - Holds the identity and name string of the notification target
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct TargetID {
+    /// Destination ID
+    pub id: String,
+    /// Target name
+    pub name: String,
+}
+
+impl TargetID {
+    /// Create a new TargetID
+    pub fn new(id: impl Into<String>, name: impl Into<String>) -> Self {
+        Self {
+            id: id.into(),
+            name: name.into(),
+        }
+    }
+
+    /// Convert to ARN
+    pub fn to_arn(&self, region: &str) -> ARN {
+        ARN {
+            target_id: self.clone(),
+            region: region.to_string(),
+        }
+    }
+
+    /// The parsed string is TargetID
+    pub fn parse(s: &str) -> Result<Self, String> {
+        let tokens: Vec<&str> = s.split(':').collect();
+        if tokens.len() != 2 {
+            return Err(format!("Invalid TargetID format '{}'", s));
+        }
+
+        Ok(Self {
+            id: tokens[0].to_string(),
+            name: tokens[1].to_string(),
+        })
+    }
+}
+
+impl fmt::Display for TargetID {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}:{}", self.id, self.name)
+    }
+}
+
+impl Serialize for TargetID {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for TargetID {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        TargetID::parse(&s).map_err(serde::de::Error::custom)
+    }
+}
+
+/// ARN - Amazon Resource Name structure
+#[derive(Debug, Clone)]
+pub struct ARN {
+    /// Destination ID
+    pub target_id: TargetID,
+    /// region
+    pub region: String,
+}
diff --git a/crates/event/tests/integration.rs b/crates/event/tests/integration.rs
index 73166fb30..b4ede35c1 100644
--- a/crates/event/tests/integration.rs
+++ b/crates/event/tests/integration.rs
@@ -1,4 +1,4 @@
-use rustfs_event::{AdapterConfig, ChannelAdapterType, NotifierSystem, WebhookConfig};
+use rustfs_event::{AdapterCommon, AdapterConfig, ChannelAdapterType, NotifierSystem, WebhookConfig};
 use rustfs_event::{Bucket, Event, EventBuilder, Identity, Metadata, Name, Object, Source};
 use rustfs_event::{ChannelAdapter, WebhookAdapter};
 use std::collections::HashMap;
@@ -7,11 +7,21 @@ use std::sync::Arc;
 #[tokio::test]
 async fn test_webhook_adapter() {
     let adapter = WebhookAdapter::new(WebhookConfig {
+        common: AdapterCommon {
+            identifier: "webhook".to_string(),
+            comment: "webhook".to_string(),
+            enable: true,
+            queue_dir: "./deploy/logs/event_queue".to_string(),
+            queue_limit: 100,
+        },
         endpoint: "http://localhost:8080/webhook".to_string(),
         auth_token: None,
         custom_headers: None,
         max_retries: 1,
-        timeout: 5,
+        timeout: Some(5),
+        retry_interval: Some(5),
+        client_cert: None,
+        client_key: None,
     });
 
     // create an s3 metadata object
@@ -71,20 +81,28 @@ async fn test_notification_system() {
         store_path: "./test_events".to_string(),
         channel_capacity: 100,
         adapters: vec![AdapterConfig::Webhook(WebhookConfig {
+            common: Default::default(),
             endpoint: "http://localhost:8080/webhook".to_string(),
             auth_token: None,
             custom_headers: None,
             max_retries: 1,
-            timeout: 5,
+            timeout: Some(5),
+            retry_interval: Some(5),
+            client_cert: None,
+            client_key: None,
         })],
     };
     let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await.unwrap()));
     let adapters: Vec<Arc<dyn ChannelAdapter>> = vec![Arc::new(WebhookAdapter::new(WebhookConfig {
+        common: Default::default(),
         endpoint: "http://localhost:8080/webhook".to_string(),
         auth_token: None,
         custom_headers: None,
         max_retries: 1,
-        timeout: 5,
+        timeout: Some(5),
+        retry_interval: Some(5),
+        client_cert: None,
+        client_key: None,
     }))];
 
     // create an s3 metadata object
diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs
index 43f8ef9d1..e70617bd5 100644
--- a/ecstore/src/store.rs
+++ b/ecstore/src/store.rs
@@ -1671,7 +1671,7 @@ impl StorageAPI for ECStore {
         let object = utils::path::encode_dir_object(object);
         let object = object.as_str();
 
-        // 查询在哪个 pool
+        // Query which pool it is in
         let (mut pinfo, errs) = self
             .get_pool_info_existing_with_opts(bucket, object, &opts)
             .await
diff --git a/rustfs/src/admin/handlers/sts.rs b/rustfs/src/admin/handlers/sts.rs
index 025866cde..a4780e699 100644
--- a/rustfs/src/admin/handlers/sts.rs
+++ b/rustfs/src/admin/handlers/sts.rs
@@ -50,7 +50,7 @@ impl Operation for AssumeRoleHandle {
         let (cred, _owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &user.access_key).await?;
 
-        // // TODO: 判断权限, 不允许sts访问
+        // // TODO: 判断权限，不允许 sts 访问
         if cred.is_temp() || cred.is_service_account() {
             return Err(s3_error!(InvalidRequest, "AccessDenied"));
         }
@@ -68,11 +68,11 @@ impl Operation for AssumeRoleHandle {
         let body: AssumeRoleRequest = from_bytes(&bytes).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?;
 
         if body.action.as_str() != ASSUME_ROLE_ACTION {
-            return Err(s3_error!(InvalidArgument, "not suport action"));
+            return Err(s3_error!(InvalidArgument, "not support action"));
         }
 
         if body.version.as_str() != ASSUME_ROLE_VERSION {
-            return Err(s3_error!(InvalidArgument, "not suport version"));
+            return Err(s3_error!(InvalidArgument, "not support version"));
         }
 
         let mut claims = cred.claims.unwrap_or_default();