Updates: Coderabbit suggestions and validations

Author: parmesant

src/alerts/mod.rs

@@ -535,10 +535,10 @@ pub struct AlertRequest {
 
 impl AlertRequest {
     pub async fn into(self) -> Result<AlertConfig, AlertError> {
-        // let mut targets = Vec::new();
-        // for id in self.targets {
-        //     targets.push(TARGETS.get_target_by_id(id).await?);
-        // }
+        // Validate that all target IDs exist
+        for id in &self.targets {
+            TARGETS.get_target_by_id(id).await?;
+        }
         let config = AlertConfig {
             version: AlertVerison::from(CURRENT_ALERTS_VERSION),
             id: Ulid::new(),
@@ -575,10 +575,10 @@ pub struct AlertConfig {
 
 impl AlertConfig {
     pub async fn modify(&mut self, alert: AlertRequest) -> Result<(), AlertError> {
-        // let mut targets = Vec::new();
-        // for id in alert.targets {
-        //     targets.push(id);
-        // }
+        // Validate that all target IDs exist
+        for id in &alert.targets {
+            TARGETS.get_target_by_id(id).await?;
+        }
         self.title = alert.title;
         self.stream = alert.stream;
         self.alert_type = alert.alert_type;
@@ -853,6 +853,10 @@ pub enum AlertError {
     FromStrError(#[from] FromStrError),
     #[error("Invalid Target ID- {0}")]
     InvalidTargetID(String),
+    #[error("Target already exists")]
+    DuplicateTargetConfig,
+    #[error("Can't delete a Target which is being used")]
+    TargetInUse,
 }
 
 impl actix_web::ResponseError for AlertError {
@@ -871,6 +875,8 @@ impl actix_web::ResponseError for AlertError {
             Self::InvalidAlertModifyRequest => StatusCode::BAD_REQUEST,
             Self::FromStrError(_) => StatusCode::BAD_REQUEST,
             Self::InvalidTargetID(_) => StatusCode::BAD_REQUEST,
+            Self::DuplicateTargetConfig => StatusCode::BAD_REQUEST,
+            Self::TargetInUse => StatusCode::CONFLICT,
         }
     }
 

src/alerts/target.rs

@@ -24,6 +24,7 @@ use std::{
 
 use async_trait::async_trait;
 use base64::Engine;
+use bytes::Bytes;
 use chrono::Utc;
 use http::{header::AUTHORIZATION, HeaderMap, HeaderValue};
 use humantime_serde::re::humantime;
@@ -35,7 +36,7 @@ use tracing::{error, trace, warn};
 use ulid::Ulid;
 use url::Url;
 
-use crate::{alerts::AlertError, parseable::PARSEABLE};
+use crate::{alerts::AlertError, parseable::PARSEABLE, storage::object_storage::target_json_path};
 
 use super::ALERTS;
 
@@ -65,7 +66,12 @@ impl TargetConfigs {
 
     pub async fn update(&self, target: Target) -> Result<(), AlertError> {
         let id = target.id;
-        self.target_configs.write().await.insert(id, target);
+        self.target_configs.write().await.insert(id, target.clone());
+        let path = target_json_path(&id);
+
+        let store = PARSEABLE.storage.get_object_store();
+        let target_bytes = serde_json::to_vec(&target)?;
+        store.put_object(&path, Bytes::from(target_bytes)).await?;
         Ok(())
     }
 
@@ -81,24 +87,38 @@ impl TargetConfigs {
     }
 
     pub async fn get_target_by_id(&self, target_id: &Ulid) -> Result<Target, AlertError> {
-        self.target_configs
+        let target = self
+            .target_configs
             .read()
             .await
             .get(target_id)
             .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
-            .cloned()
+            .cloned()?;
+
+        Ok(target)
     }
 
     pub async fn delete(&self, target_id: &Ulid) -> Result<Target, AlertError> {
-        self.target_configs
+        // ensure that the target is not being used by any alert
+        for (_, alert) in ALERTS.alerts.read().await.iter() {
+            if alert.targets.contains(target_id) {
+                return Err(AlertError::TargetInUse);
+            }
+        }
+        let target = self
+            .target_configs
             .write()
             .await
             .remove(target_id)
-            .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
+            .ok_or(AlertError::InvalidTargetID(target_id.to_string()))?;
+        let path = target_json_path(&target.id);
+        let store = PARSEABLE.storage.get_object_store();
+        store.delete_object(&path).await?;
+        Ok(target)
     }
 }
 
-#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq)]
 #[serde(rename_all = "camelCase")]
 #[serde(untagged)]
 pub enum Retry {
@@ -125,9 +145,19 @@ pub struct Target {
 }
 
 impl Target {
-    pub async fn validate(&self) {
+    pub async fn validate(&self) -> Result<(), AlertError> {
         // just check for liveness
         // what if the target is not live yet but is added by the user?
+        let targets = TARGETS.list().await?;
+        for target in targets {
+            if target.target == self.target
+                && target.timeout.interval == self.timeout.interval
+                && target.timeout.times == self.timeout.times
+            {
+                return Err(AlertError::DuplicateTargetConfig);
+            }
+        }
+        Ok(())
     }
 
     pub fn call(&self, context: Context) {

src/handlers/http/targets.rs

@@ -2,16 +2,11 @@ use actix_web::{
     web::{self, Json, Path},
     HttpRequest, Responder,
 };
-use bytes::Bytes;
 use ulid::Ulid;
 
-use crate::{
-    alerts::{
-        target::{Target, TARGETS},
-        AlertError,
-    },
-    parseable::PARSEABLE,
-    storage::object_storage::target_json_path,
+use crate::alerts::{
+    target::{Target, TARGETS},
+    AlertError,
 };
 
 // POST /targets
@@ -20,13 +15,7 @@ pub async fn post(
     Json(target): Json<Target>,
 ) -> Result<impl Responder, AlertError> {
     // should check for duplicacy and liveness (??)
-    target.validate().await;
-
-    let path = target_json_path(target.id);
-
-    let store = PARSEABLE.storage.get_object_store();
-    let target_bytes = serde_json::to_vec(&target)?;
-    store.put_object(&path, Bytes::from(target_bytes)).await?;
+    target.validate().await?;
 
     // add to the map
     TARGETS.update(target.clone()).await?;
@@ -64,14 +53,9 @@ pub async fn update(
 
     // esnure that the supplied target id is assigned to the target config
     target.id = target_id;
-    // should check for duplicacy and liveness (??)
-    target.validate().await;
 
-    let path = target_json_path(target.id);
-
-    let store = PARSEABLE.storage.get_object_store();
-    let target_bytes = serde_json::to_vec(&target)?;
-    store.put_object(&path, Bytes::from(target_bytes)).await?;
+    // should check for duplicacy and liveness (??)
+    target.validate().await?;
 
     // add to the map
     TARGETS.update(target.clone()).await?;

src/storage/mod.rs

@@ -59,6 +59,7 @@ pub const STREAM_ROOT_DIRECTORY: &str = ".stream";
 pub const PARSEABLE_ROOT_DIRECTORY: &str = ".parseable";
 pub const SCHEMA_FILE_NAME: &str = ".schema";
 pub const ALERTS_ROOT_DIRECTORY: &str = ".alerts";
+pub const SETTINGS_ROOT_DIRECTORY: &str = ".settings";
 pub const TARGETS_ROOT_DIRECTORY: &str = ".targets";
 pub const MANIFEST_FILE: &str = "manifest.json";
 

src/storage/object_storage.rs

@@ -79,6 +79,7 @@ use crate::parseable::PARSEABLE;
 use crate::query::QUERY_SESSION_STATE;
 use crate::stats::FullStats;
 use crate::storage::StreamType;
+use crate::storage::SETTINGS_ROOT_DIRECTORY;
 use crate::storage::TARGETS_ROOT_DIRECTORY;
 use crate::utils::DATASET_STATS_STREAM_NAME;
 
@@ -446,7 +447,8 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
     }
 
     async fn get_targets(&self) -> Result<Vec<Target>, ObjectStorageError> {
-        let targets_path = RelativePathBuf::from(TARGETS_ROOT_DIRECTORY);
+        let targets_path =
+            RelativePathBuf::from_iter([SETTINGS_ROOT_DIRECTORY, TARGETS_ROOT_DIRECTORY]);
         let targets = self
             .get_objects(
                 Some(&targets_path),
@@ -1360,8 +1362,12 @@ pub fn alert_json_path(alert_id: Ulid) -> RelativePathBuf {
 
 /// TODO: Needs to be updated for distributed mode
 #[inline(always)]
-pub fn target_json_path(target_id: Ulid) -> RelativePathBuf {
-    RelativePathBuf::from_iter([TARGETS_ROOT_DIRECTORY, &format!("{target_id}.json")])
+pub fn target_json_path(target_id: &Ulid) -> RelativePathBuf {
+    RelativePathBuf::from_iter([
+        SETTINGS_ROOT_DIRECTORY,
+        TARGETS_ROOT_DIRECTORY,
+        &format!("{target_id}.json"),
+    ])
 }
 
 #[inline(always)]