feat: SSE (#1487)

Add backend for SSE so that it sends alert 
message only if state = triggered

Author: parmesant

Cargo.lock

@@ -30,6 +30,7 @@ parquet = "57.1.0"
 
 # Web server and HTTP-related
 actix-cors = "0.7.0"
+actix-web-lab = "0.24.3"
 actix-web = { version = "4.9.0", features = ["rustls-0_22"] }
 actix-web-httpauth = "0.8"
 actix-web-prometheus = { version = "0.1" }
@@ -83,7 +84,7 @@ tokio = { version = "^1.43", default-features = false, features = [
     "fs",
     "rt-multi-thread",
 ] }
-tokio-stream = { version = "0.1", features = ["fs"] }
+tokio-stream = { version = "0.1.17", features = ["fs"] }
 tokio-util = { version = "0.7" }
 
 # # Logging and Metrics

Cargo.toml

@@ -58,11 +58,10 @@ use crate::alerts::alert_types::ThresholdAlert;
 use crate::alerts::target::{NotificationConfig, TARGETS};
 use crate::handlers::http::fetch_schema;
 use crate::metastore::MetastoreError;
-// use crate::handlers::http::query::create_streams_for_distributed;
-// use crate::option::Mode;
 use crate::parseable::{PARSEABLE, StreamNotFound};
 use crate::query::{QUERY_SESSION, resolve_stream_names};
-use crate::rbac::map::SessionKey;
+use crate::rbac::map::{SessionKey, sessions};
+use crate::sse::{SSE_HANDLER, SSEAlertInfo, SSEEvent};
 use crate::storage;
 use crate::storage::ObjectStorageError;
 use crate::sync::alert_runtime;
@@ -606,12 +605,39 @@ impl AlertConfig {
 
     pub async fn trigger_notifications(&self, message: String) -> Result<(), AlertError> {
         let mut context = self.get_context();
-        context.message = message;
+        context.message.clone_from(&message);
+
         for target_id in &self.targets {
             let target = TARGETS.get_target_by_id(target_id).await?;
             trace!("Target (trigger_notifications)-\n{target:?}");
             target.call(context.clone());
         }
+
+        // get active sessions
+        let active_sessions = sessions().get_active_sessions();
+        let mut broadcast_to = vec![];
+        for session in active_sessions {
+            if user_auth_for_query(&session, &self.query).await.is_ok()
+                && let SessionKey::SessionId(id) = &session
+            {
+                broadcast_to.push(*id);
+            }
+        }
+
+        if self.state.eq(&AlertState::Triggered)
+            && let Ok(msg) = &serde_json::to_string(&SSEEvent {
+                criticality: crate::sse::Criticality::Error,
+                message: crate::sse::Message::AlertEvent(SSEAlertInfo {
+                    id: self.id,
+                    state: self.state,
+                    name: self.title.clone(),
+                }),
+            })
+            && !broadcast_to.is_empty()
+        {
+            SSE_HANDLER.broadcast(msg, Some(&broadcast_to)).await;
+        }
+
         Ok(())
     }
 

src/alerts/mod.rs

@@ -42,6 +42,7 @@ pub mod prism;
 pub mod query;
 pub mod rbac;
 mod response;
+pub mod sse;
 mod static_schema;
 mod stats;
 pub mod storage;

src/lib.rs

@@ -28,6 +28,7 @@ use super::{
     user,
 };
 use chrono::{DateTime, Utc};
+use itertools::Itertools;
 use once_cell::sync::{Lazy, OnceCell};
 use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use std::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
@@ -168,6 +169,10 @@ pub struct Sessions {
 }
 
 impl Sessions {
+    pub fn get_active_sessions(&self) -> Vec<SessionKey> {
+        self.active_sessions.keys().cloned().collect_vec()
+    }
+
     // only checks if the session is expired or not
     pub fn is_session_expired(&self, key: &SessionKey) -> bool {
         // fetch userid from session key