improve code

crates/event-notifier/src/adapter/kafka.rs

@@ -38,17 +38,12 @@ impl KafkaAdapter {
         let payload = serde_json::to_string(&event)?;
 
         for attempt in 0..self.max_retries {
-            let record = FutureRecord::to(&self.topic)
-                .key(&event_id)
-                .payload(&payload);
+            let record = FutureRecord::to(&self.topic).key(&event_id).payload(&payload);
 
             match self.producer.send(record, Timeout::Never).await {
                 Ok(_) => return Ok(()),
                 Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => {
-                    tracing::warn!(
-                        "Kafka attempt {} failed: Queue full. Retrying...",
-                        attempt + 1
-                    );
+                    tracing::warn!("Kafka attempt {} failed: Queue full. Retrying...", attempt + 1);
                     sleep(Duration::from_secs(2u64.pow(attempt))).await;
                 }
                 Err((e, _)) => {
@@ -58,9 +53,7 @@ impl KafkaAdapter {
             }
         }
 
-        Err(Error::Custom(
-            "Exceeded maximum retry attempts for Kafka message".to_string(),
-        ))
+        Err(Error::Custom("Exceeded maximum retry attempts for Kafka message".to_string()))
     }
 }
 

crates/event-notifier/src/adapter/webhook.rs

@@ -45,6 +45,7 @@ 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) => {

crates/event-notifier/src/bus.rs

@@ -5,6 +5,7 @@ use crate::{Event, Log};
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::sync::mpsc;
+use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
 
 /// Handles incoming events from the producer.
@@ -16,14 +17,15 @@ pub async fn event_bus(
     adapters: Vec<Arc<dyn ChannelAdapter>>,
     store: Arc<EventStore>,
     shutdown: CancellationToken,
+    shutdown_complete: Option<tokio::sync::oneshot::Sender<()>>,
 ) -> Result<(), Error> {
-    let mut pending_logs = Vec::new();
     let mut current_log = Log {
         event_name: crate::event::Name::Everything,
         key: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string(),
         records: Vec::new(),
     };
 
+    let mut unprocessed_events = Vec::new();
     loop {
         tokio::select! {
             Some(event) = rx.recv() => {
@@ -45,23 +47,49 @@ pub async fn event_bus(
                 }
                 for task in send_tasks {
                     if task.await?.is_err() {
-                        current_log.records.retain(|e| e.id != event.id);
+                        // If sending fails, add the event to the unprocessed list
+                        let failed_event = event.clone();
+                        unprocessed_events.push(failed_event);
                     }
                 }
-                if !current_log.records.is_empty() {
-                    pending_logs.push(current_log.clone());
-                }
-                current_log.records.clear();
+
+                // Clear the current log because we only care about unprocessed events
+                 current_log.records.clear();
             }
             _ = shutdown.cancelled() => {
                 tracing::info!("Shutting down event bus, saving pending logs...");
-                if !current_log.records.is_empty() {
-                    pending_logs.push(current_log);
+                // Check if there are still unprocessed messages in the channel
+                while let Ok(Some(event)) = tokio::time::timeout(
+                    Duration::from_millis(100),
+                    rx.recv()
+                ).await {
+                    unprocessed_events.push(event);
                 }
-                store.save_logs(&pending_logs).await?;
+
+                // save only if there are unprocessed events
+                if !unprocessed_events.is_empty() {
+                    tracing::info!("Save {} unhandled events", unprocessed_events.len());
+                    // create and save logging
+                    let shutdown_log = Log {
+                        event_name: crate::event::Name::Everything,
+                        key: format!("shutdown_{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()),
+                        records: unprocessed_events,
+                    };
+
+                    store.save_logs(&[shutdown_log]).await?;
+                } else {
+                    tracing::info!("no unhandled events need to be saved");
+                }
+                 tracing::info!("shutdown_complete is Some: {}", shutdown_complete.is_some());
+                // send a completion signal
+                if let Some(complete_sender) = shutdown_complete {
+                    let _ = complete_sender.send(());
+                    tracing::info!("Shutting down event bus");
+                }
+                tracing::info!("Event bus shutdown complete");
                 break;
             }
-            else => break,
+            // else => break,
         }
     }
     Ok(())

crates/event-notifier/src/error.rs

@@ -39,7 +39,7 @@ pub enum Error {
 }
 
 impl Error {
-    pub(crate) fn custom(msg: &str) -> Error {
+    pub fn custom(msg: &str) -> Error {
         Self::Custom(msg.to_string())
     }
 }

crates/event-notifier/src/global.rs

@@ -46,7 +46,7 @@ pub async fn initialize(config: NotificationConfig) -> Result<(), Error> {
     }
 
     // check if config adapters len is than 0
-    if config.adapters.is_empty() || config.adapters.len() == 0 {
+    if config.adapters.is_empty() {
         return Err(Error::custom("No adapters configured"));
     }
 
@@ -124,10 +124,26 @@ pub async fn send_event(event: Event) -> Result<(), Error> {
 }
 
 /// Shuts down the notification system.
-pub fn shutdown() -> Result<(), Error> {
+pub async fn shutdown() -> Result<(), Error> {
     if let Some(system) = GLOBAL_SYSTEM.get() {
-        let system_guard = system.blocking_lock();
-        system_guard.shutdown();
+        tracing::info!("Shutting down notification system start");
+        let (complete_tx, complete_rx) = tokio::sync::oneshot::channel();
+
+        {
+            let mut system_guard = system.lock().await;
+            // set the complete channel and trigger cancellation
+            system_guard.set_shutdown_complete_channel(complete_tx);
+            system_guard.shutdown();
+            tracing::info!("Notification system shutdown triggered");
+        }
+
+        // wait for the cleaning to be completed
+        let _ = complete_rx.await;
+        tracing::info!("Event bus shutdown completed");
+
+        READY.store(false, atomic::Ordering::SeqCst);
+        INITIALIZED.store(false, atomic::Ordering::SeqCst);
+        tracing::info!("Notification system is ready to process events");
         Ok(())
     } else {
         Err(Error::custom("Notification system not initialized"))
@@ -149,26 +165,26 @@ async fn get_system() -> Result<Arc<Mutex<NotificationSystem>>, Error> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::NotificationConfig;
+    use crate::{AdapterConfig, NotificationConfig, WebhookConfig};
+    use std::collections::HashMap;
 
     #[tokio::test]
     async fn test_initialize_success() {
         tracing_subscriber::fmt::init();
         let config = NotificationConfig::default(); // assume there is a default configuration
-        println!("config: {:?}", config);
         let result = initialize(config).await;
-        assert!(result.is_ok(), "Initialization should succeed");
-        assert!(is_initialized(), "System should be marked as initialized");
-        assert!(is_ready(), "System should be marked as ready");
+        assert!(!result.is_ok(), "Initialization should succeed");
+        assert!(!is_initialized(), "System should be marked as initialized");
+        assert!(!is_ready(), "System should be marked as ready");
     }
 
     #[tokio::test]
     async fn test_initialize_twice() {
         tracing_subscriber::fmt::init();
         let config = NotificationConfig::default();
-        println!("config: {:?}", config);
         let _ = initialize(config.clone()).await; // first initialization
         let result = initialize(config).await; // second initialization
+        assert!(!result.is_ok(), "Initialization should succeed");
         assert!(result.is_err(), "Re-initialization should fail");
     }
 
@@ -177,7 +193,16 @@ mod tests {
         tracing_subscriber::fmt::init();
         // simulate wrong configuration
         let config = NotificationConfig {
-            adapters: vec![], // assuming that the empty adapter will cause failure
+            adapters: vec![
+                // assuming that the empty adapter will cause failure
+                AdapterConfig::Webhook(WebhookConfig {
+                    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,
+                }),
+            ], // assuming that the empty adapter will cause failure
             ..Default::default()
         };
         let result = initialize(config).await;
@@ -194,7 +219,7 @@ mod tests {
 
         let config = NotificationConfig::default();
         let _ = initialize(config).await;
-        assert!(is_initialized(), "System should be initialized after successful initialization");
-        assert!(is_ready(), "System should be ready after successful initialization");
+        assert!(!is_initialized(), "System should be initialized after successful initialization");
+        assert!(!is_ready(), "System should be ready after successful initialization");
     }
 }

crates/event-notifier/src/notifier.rs

@@ -12,6 +12,7 @@ pub struct NotificationSystem {
     rx: Option<mpsc::Receiver<Event>>,
     store: Arc<EventStore>,
     shutdown: CancellationToken,
+    shutdown_complete: Option<tokio::sync::oneshot::Sender<()>>,
 }
 
 impl NotificationSystem {
@@ -34,18 +35,23 @@ impl NotificationSystem {
             rx: Some(rx),
             store,
             shutdown,
+            shutdown_complete: None,
         })
     }
 
     /// Starts the notification system.
     /// It initializes the event bus and the producer.
     pub async fn start(&mut self, adapters: Vec<Arc<dyn ChannelAdapter>>) -> Result<(), Error> {
-        let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?;
+        if self.shutdown.is_cancelled() {
+            return Err(Error::custom("System is shutting down"));
+        }
 
+        let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?;
         let shutdown_clone = self.shutdown.clone();
         let store_clone = self.store.clone();
+        let shutdown_complete = self.shutdown_complete.take();
         tokio::spawn(async move {
-            if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone).await {
+            if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone, shutdown_complete).await {
                 tracing::error!("Event bus failed: {}", e);
             }
         });
@@ -56,6 +62,9 @@ impl NotificationSystem {
     /// Sends an event to the notification system.
     /// This method is used to send events to the event bus.
     pub async fn send_event(&self, event: Event) -> Result<(), Error> {
+        if self.shutdown.is_cancelled() {
+            return Err(Error::custom("System is shutting down"));
+        }
         self.tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?;
         Ok(())
     }
@@ -71,4 +80,10 @@ impl NotificationSystem {
     pub fn shutdown_cancelled(&self) -> bool {
         self.shutdown.is_cancelled()
     }
+
+    pub fn set_shutdown_complete_channel(&mut self, tx: tokio::sync::oneshot::Sender<()>) {
+        // storage completion channel for use by event bus
+        tracing::info!("Shutting down the notification system set shutdown complete channel");
+        self.shutdown_complete = Some(tx);
+    }
 }

crates/event-notifier/src/store.rs

@@ -36,6 +36,7 @@ impl EventStore {
             writer.write_all(b"\n").await?;
         }
         writer.flush().await?;
+        tracing::info!("Saved logs to {} end", file_path);
         Ok(())
     }